ACE/tests/Bug_3878_Regression_Test.cpp

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