ACE/examples/APG/ThreadPools/ThreadPool.cpp

   1 #include "ace/config-lite.h"
   2 #if defined (ACE_HAS_THREADS)
   3
   4 #include "ace/OS_NS_string.h"
   5 #include "ace/OS_NS_time.h"
   6 #include "ace/Task.h"
   7 #include "ace/Containers.h"
   8 #include "ace/Synch.h"
   9 #include "ace/SString.h"
  10 #include "ace/Method_Request.h"
  11 #include "ace/Future.h"
  12 #include "ace/Activation_Queue.h"
  13 #include "ace/Condition_T.h"
  14
  15 class Worker;
  16
  17 class IManager
  18 {
  19 public:
  20   virtual ~IManager () { }
  21
  22   virtual int return_to_work (Worker *worker) = 0;
  23 };
  24
  25 // Listing 2 code/ch16
  26 class Worker : public ACE_Task<ACE_MT_SYNCH>
  27 {
  28 public:
  29   Worker (IManager *manager) : manager_(manager) { }
  30
  31   virtual int svc ()
  32   {
  33     ACE_Thread_ID id;
  34     thread_id_ = id;
  35     while (1)
  36       {
  37         ACE_Message_Block *mb = 0;
  38         if (this->getq (mb) == -1)
  39           ACE_ERROR_BREAK
  40             ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("getq")));
  41         if (mb->msg_type () == ACE_Message_Block::MB_HANGUP)
  42           {
  43             ACE_DEBUG ((LM_INFO,
  44                         ACE_TEXT ("(%t) Shutting down\n")));
  45             mb->release ();
  46             break;
  47           }
  48         // Process the message.
  49         process_message (mb);
  50         // Return to work.
  51         this->manager_->return_to_work (this);
  52       }
  53
  54     return 0;
  55   }
  56   // Listing 2
  57
  58   const ACE_Thread_ID& thread_id ()
  59   {
  60     return this->thread_id_;
  61   }
  62
  63 private:
  64   void process_message (ACE_Message_Block *mb)
  65   {
  66     ACE_TRACE ("Worker::process_message");
  67     int msgId;
  68     ACE_OS::memcpy (&msgId, mb->rd_ptr (), sizeof(int));
  69     mb->release ();
  70
  71     ACE_DEBUG ((LM_DEBUG,
  72                 ACE_TEXT ("(%t) Started processing message %d\n"),
  73                 msgId));
  74     ACE_OS::sleep (3);
  75     ACE_DEBUG ((LM_DEBUG,
  76                 ACE_TEXT ("(%t) Finished processing message %d\n"),
  77                 msgId));
  78   }
  79
  80   IManager *manager_;
  81   ACE_Thread_ID thread_id_;
  82 };
  83
  84 // Listing 1 code/ch16
  85 class Manager: public ACE_Task<ACE_MT_SYNCH>, private IManager
  86 {
  87 public:
  88   enum {POOL_SIZE = 5, MAX_TIMEOUT = 5};
  89
  90   Manager ()
  91     : shutdown_(0), workers_lock_(), workers_cond_(workers_lock_)
  92   {
  93     ACE_TRACE ("Manager::Manager");
  94   }
  95
  96   int svc ()
  97   {
  98     ACE_TRACE ("Manager::svc");
  99
 100     ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%t) Manager started\n")));
 101
 102     // Create pool.
 103     create_worker_pool ();
 104
 105     while (!done ())
 106       {
 107         ACE_Message_Block *mb = 0;
 108         ACE_Time_Value tv ((long)MAX_TIMEOUT);
 109         tv += ACE_OS::time (0);
 110
 111         // Get a message request.
 112         if (this->getq (mb, &tv) < 0)
 113           {
 114             shut_down ();
 115             break;
 116           }
 117
 118         // Choose a worker.
 119         Worker *worker = 0;
 120         {
 121           ACE_GUARD_RETURN (ACE_Thread_Mutex,
 122                             worker_mon, this->workers_lock_, -1);
 123
 124           while (this->workers_.is_empty ())
 125             workers_cond_.wait ();
 126
 127           this->workers_.dequeue_head (worker);
 128         }
 129
 130         // Ask the worker to do the job.
 131         worker->putq (mb);
 132       }
 133
 134     return 0;
 135   }
 136
 137   int shut_down ();
 138
 139   const ACE_Thread_ID& thread_id (Worker *worker);
 140
 141   virtual int return_to_work (Worker *worker)
 142   {
 143     ACE_GUARD_RETURN (ACE_Thread_Mutex,
 144                       worker_mon, this->workers_lock_, -1);
 145     ACE_DEBUG ((LM_DEBUG,
 146                 ACE_TEXT ("(%t) Worker %t returning to work.\n")));
 147     this->workers_.enqueue_tail (worker);
 148     this->workers_cond_.signal ();
 149
 150     return 0;
 151   }
 152
 153 private:
 154   int create_worker_pool ()
 155   {
 156     ACE_GUARD_RETURN (ACE_Thread_Mutex,
 157                       worker_mon,
 158                       this->workers_lock_,
 159                       -1);
 160     for (int i = 0; i < POOL_SIZE; i++)
 161       {
 162         Worker *worker = 0;
 163         ACE_NEW_RETURN (worker, Worker (this), -1);
 164         this->workers_.enqueue_tail (worker);
 165         worker->activate ();
 166       }
 167
 168     return 0;
 169   }
 170
 171   int done ();
 172
 173 private:
 174   int shutdown_;
 175   ACE_Thread_Mutex workers_lock_;
 176   ACE_Condition<ACE_Thread_Mutex> workers_cond_;
 177   ACE_Unbounded_Queue<Worker* > workers_;
 178 };
 179 // Listing 1
 180
 181 int Manager::done ()
 182 {
 183   return (shutdown_ == 1);
 184 }
 185
 186 int
 187 Manager::shut_down ()
 188 {
 189   ACE_TRACE ("Manager::shut_down");
 190   ACE_Unbounded_Queue<Worker* >::ITERATOR iter =
 191     this->workers_.begin ();
 192   Worker **worker_ptr = 0;
 193   do
 194     {
 195       iter.next (worker_ptr);
 196       Worker *worker = (*worker_ptr);
 197       ACE_Thread_ID id = thread_id (worker);
 198       char buf [65];
 199       id.to_string (buf);
 200       ACE_DEBUG ((LM_DEBUG,
 201                  ACE_TEXT ("(%t) Attempting shutdown of %C\n"),
 202                  buf));
 203
 204       // Send the hangup message.
 205       ACE_Message_Block *mb = 0;
 206       ACE_NEW_RETURN
 207         (mb,
 208          ACE_Message_Block(0,
 209                            ACE_Message_Block::MB_HANGUP),
 210          -1);
 211       worker->putq (mb);
 212
 213       // Wait for the exit.
 214       worker->wait ();
 215
 216       ACE_ASSERT (worker->msg_queue ()->is_empty ());
 217       ACE_DEBUG ((LM_DEBUG,
 218                   ACE_TEXT ("(%t) Worker %C shut down.\n"),
 219                   buf));
 220       delete worker;
 221     }
 222   while (iter.advance ());
 223
 224   shutdown_ = 1;
 225
 226   return 0;
 227 }
 228
 229 const ACE_Thread_ID&
 230 Manager::thread_id (Worker *worker)
 231 {
 232   return worker->thread_id ();
 233 }
 234
 235 int ACE_TMAIN (int, ACE_TCHAR *[])
 236 {
 237   Manager tp;
 238   tp.activate ();
 239
 240   // Wait for a moment every time you send a message.
 241   ACE_Time_Value tv;
 242   tv.msec (100);
 243
 244   ACE_Message_Block *mb = 0;
 245   for (int i = 0; i < 30; i++)
 246     {
 247       ACE_NEW_RETURN
 248         (mb, ACE_Message_Block(sizeof(int)), -1);
 249
 250       ACE_OS::memcpy (mb->wr_ptr (), &i, sizeof(int));
 251
 252       ACE_OS::sleep (tv);
 253
 254       // Add a new work item.
 255       tp.putq (mb);
 256     }
 257
 258   ACE_Thread_Manager::instance ()->wait ();
 259   return 0;
 260 }
 261
 262 #else
 263 #include "ace/OS_main.h"
 264 #include "ace/OS_NS_stdio.h"
 265
 266 int ACE_TMAIN (int, ACE_TCHAR *[])
 267 {
 268   ACE_OS::puts (ACE_TEXT ("This example requires threads."));
 269   return 0;
 270 }
 271
 272 #endif /* ACE_HAS_THREADS */