TAO/tao/CSD_ThreadPool/CSD_TP_Task.h

   1 // -*- C++ -*-
   2
   3 //=============================================================================
   4 /**
   5  *  @file    CSD_TP_Task.h
   6  *
   7  *  @author  Tim Bradley <bradley_t@ociweb.com>
   8  */
   9 //=============================================================================
  10
  11 #ifndef TAO_CSD_TP_TASK_H
  12 #define TAO_CSD_TP_TASK_H
  13
  14 #include /**/ "ace/pre.h"
  15
  16 #include "tao/CSD_ThreadPool/CSD_TP_Export.h"
  17
  18 #include "tao/CSD_ThreadPool/CSD_TP_Queue.h"
  19 #include "tao/PortableServer/PortableServer.h"
  20 #include "tao/Condition.h"
  21
  22 #if !defined (ACE_LACKS_PRAGMA_ONCE)
  23 # pragma once
  24 #endif /* ACE_LACKS_PRAGMA_ONCE */
  25
  26 #include "ace/Task.h"
  27 #include "ace/Synch.h"
  28 #include "ace/Containers_T.h"
  29 #include "ace/Vector_T.h"
  30
  31 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  32
  33 namespace TAO
  34 {
  35   namespace CSD
  36   {
  37     /// Typedef for the number of threads.
  38     typedef unsigned long Thread_Counter;
  39
  40     /**
  41      * @class TP_Task
  42      *
  43      * @brief Active Object managing a queue of request objects.
  44      *
  45      * There are two types of "users" of a TP_Task object:
  46      *
  47      *    1) The TP_Strategy object that "owns" this task object.
  48      *    2) The worker threads that "run" this task object as an
  49      *       "active object".
  50      *
  51      * The TP_Strategy object that "owns" this task object dictates
  52      * when the worker threads are activated and when they are shutdown.  It
  53      * also injects requests into this task's queue via calls to the
  54      * add_request() method.  It is also the TP_Strategy object that
  55      * dictates the number of worker threads to be activated via a call to
  56      * the set_num_threads() method.
  57      *
  58      * The active object pattern is implemented via the use of the
  59      * the ACE_Task_Base base class, and each worker thread will
  60      * invoke this task's svc() method, and when the svc() returns, the
  61      * worker thread will invoke this task's close() method (with the
  62      * flag argument equal to 0).
  63      *
  64      * @note I just wanted to document an idea...  When the pool consists
  65      *       of only one worker thread, we could care less about checking
  66      *       if target servant objects are busy or not.  The simple fact
  67      *       that only one thread will be dispatching all requests means
  68      *       that servant objects will never be busy when the thread
  69      *       tests to see if a request is "ready_for_dispatch()".  I'm
  70      *       just wondering if this knowledge can be applied to the
  71      *       implementation such that the "pool with one worker thread" case
  72      *       performs more efficiently.  This is STP vs SSTP.
  73      *
  74      */
  75     class TAO_CSD_TP_Export TP_Task : public ACE_Task_Base
  76     {
  77     public:
  78       /// Default Constructor.
  79       TP_Task();
  80
  81       /// Virtual Destructor.
  82       virtual ~TP_Task();
  83
  84       /// Put a request object on to the request queue.
  85       /// Returns true if successful, false otherwise (it has been "rejected").
  86       bool add_request(TP_Request* request);
  87
  88       /// Activate the worker threads
  89       virtual int open(void* args = 0);
  90
  91       /// The "mainline" executed by each worker thread.
  92       virtual int svc();
  93
  94       /// Multi-purpose: argument value is used to differentiate purpose.
  95       ///
  96       /// 0) Invoked by each worker thread after its invocation of the
  97       ///    svc() method has completed (ie, returned).
  98       /// 1) Invoked by the strategy object to shutdown all worker threads.
  99       virtual int close(u_long flag = 0);
 100
 101       /// Cancel all requests that are targeted for the provided servant.
 102       void cancel_servant (PortableServer::Servant servant);
 103
 104     private:
 105       typedef TAO_SYNCH_MUTEX         LockType;
 106       typedef TAO_Condition<LockType> ConditionType;
 107
 108       /// Lock to protect the "state" (all of the data members) of this object.
 109       LockType lock_;
 110
 111       /// Condition used to signal worker threads that they may be able to
 112       /// find a request in the queue_ that needs to be dispatched to a
 113       /// servant that is currently "not busy".
 114       /// This condition will be signal()'ed each time a new request is
 115       /// added to the queue_, and also when a servant has become "not busy".
 116       ConditionType work_available_;
 117
 118       /// This condition will be signal()'ed each time the num_threads_
 119       /// data member has its value changed.  This is used to keep the
 120       /// close(1) invocation (ie, a shutdown request) blocked until all
 121       /// of the worker threads have stopped running.
 122       ConditionType active_workers_;
 123
 124       /// Flag used to indicate when this task will (or will not) accept
 125       /// requests via the the add_request() method.
 126       bool accepting_requests_;
 127
 128       /// Flag used to initiate a shutdown request to all worker threads.
 129       bool shutdown_initiated_;
 130
 131       /// Complete shutdown needed to be deferred because the thread calling
 132       /// close(1) was also one of the ThreadPool threads
 133       bool deferred_shutdown_initiated_;
 134
 135       /// Flag used to avoid multiple open() calls.
 136       bool opened_;
 137
 138       /// The number of currently active worker threads.
 139       Thread_Counter num_threads_;
 140
 141       /// The queue of pending servant requests (a.k.a. the "request queue").
 142       TP_Queue queue_;
 143
 144       typedef ACE_Vector <ACE_thread_t> Thread_Ids;
 145
 146       /// The list of ids for the threads launched by this task.
 147       Thread_Ids activated_threads_;
 148
 149       enum { MAX_THREADPOOL_TASK_WORKER_THREADS = 50 };
 150     };
 151   }
 152 }
 153
 154 TAO_END_VERSIONED_NAMESPACE_DECL
 155
 156 #if defined (__ACE_INLINE__)
 157 # include "tao/CSD_ThreadPool/CSD_TP_Task.inl"
 158 #endif /* __ACE_INLINE__ */
 159
 160 #include /**/ "ace/post.h"
 161
 162 #endif /* TAO_CSD_TP_TASK_H */