ACE/tests/Thread_Pool_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Thread_Pool_Test.cpp
   5  *
   6  *   This test program illustrates how the <ACE_Task>
   7  *   synchronization mechanisms work in conjunction with the
   8  *   <ACE_Thread_Manager>.  If the <manual> flag is set input comes
   9  *   from stdin until the user enters a return -- otherwise, the
  10  *   input is generated automatically.  All worker threads shutdown
  11  *   when (1) they receive a message block of length 0 or (2) the
  12  *   queue is deactivated.
  13  *
  14  *  @author Karlheinz Dorn <Karlheinz.Dorn@med.siemens.de>
  15  *  @author Douglas C. Schmidt <d.schmidt@vanderbilt.edu>
  16  *  @author and Prashant Jain <pjain@cs.wustl.edu>
  17  */
  18 //=============================================================================
  19
  20
  21 #include "test_config.h"
  22 #include "ace/Task.h"
  23
  24 #if defined (ACE_HAS_THREADS)
  25 #include "ace/Lock_Adapter_T.h"
  26 #include "ace/OS_NS_stdio.h"
  27 #include "ace/OS_NS_string.h"
  28 #include "ace/OS_NS_unistd.h"
  29
  30 // Number of iterations to run the test.
  31 static size_t n_iterations = 100;
  32
  33 // Controls whether the input is generated "manually" or automatically.
  34 static int manual = 0;
  35
  36 /**
  37  * @class Thread_Pool
  38  *
  39  * @brief Defines a thread pool abstraction based on the <ACE_Task>.
  40  */
  41 class Thread_Pool : public ACE_Task<ACE_MT_SYNCH>
  42 {
  43 public:
  44   /// Create the thread pool containing <n_threads>.
  45   Thread_Pool (int n_threads);
  46
  47   /// Destructor...
  48   ~Thread_Pool () override;
  49
  50   /**
  51    * Activate the task's thread pool, produce the messages that are
  52    * consumed by the threads in the thread pool, and demonstate how to
  53    * shutdown using the <ACE_Message_Queue::deactivate> method.
  54    */
  55   int test_queue_deactivation_shutdown ();
  56
  57   /**
  58    * Activate the task's thread pool, produce the messages that are,
  59    * produce the messages that are consumed by the threads in the
  60    * thread pool, and demonstrate how to shutdown by enqueueing
  61    * "empty" messages into the queue.
  62    */
  63   int test_empty_message_shutdown ();
  64
  65   /// Iterate <n_iterations> time printing off a message and "waiting"
  66   /// for all other threads to complete this iteration.
  67   int svc () override;
  68
  69   /// Allows the producer to pass messages to the <Thread_Pool>.
  70   int put (ACE_Message_Block *mb,
  71                    ACE_Time_Value *tv = 0) override;
  72
  73 private:
  74   //FUZZ: disable check_for_lack_ACE_OS
  75   /// Spawn the threads in the pool.
  76   int open (void * = 0) override;
  77   //FUZZ: enable check_for_lack_ACE_OS
  78
  79   /// Close hook.
  80   int close (u_long) override;
  81
  82   /// Serialize access to <ACE_Message_Block> reference count, which
  83   /// will be decremented by multiple threads.
  84   ACE_Lock_Adapter<ACE_Thread_Mutex> lock_adapter_;
  85
  86   /// Number of threads to spawn.
  87   int n_threads_;
  88 };
  89
  90 Thread_Pool::~Thread_Pool ()
  91 {
  92 }
  93
  94 int
  95 Thread_Pool::close (u_long)
  96 {
  97   ACE_DEBUG ((LM_DEBUG,
  98               ACE_TEXT ("(%t) worker thread closing down\n")));
  99   return 0;
 100 }
 101
 102 Thread_Pool::Thread_Pool (int n_threads)
 103   : n_threads_ (n_threads)
 104 {
 105 }
 106
 107 // Simply enqueue the Message_Block into the end of the queue.
 108
 109 int
 110 Thread_Pool::put (ACE_Message_Block *mb,
 111                   ACE_Time_Value *tv)
 112 {
 113   return this->putq (mb, tv);
 114 }
 115
 116 // Iterate <n_iterations> printing off a message and "waiting" for all
 117 // other threads to complete this iteration.
 118
 119 int
 120 Thread_Pool::svc ()
 121 {
 122   // Keep looping, reading a message out of the queue, until we get a
 123   // message with a length == 0, which signals us to quit.
 124
 125   for (int count = 1; ; count++)
 126     {
 127       ACE_Message_Block *mb = 0;
 128
 129       int result = this->getq (mb);
 130
 131       ACE_TEST_ASSERT (result != -1 || errno == ESHUTDOWN);
 132
 133       if (result == -1 && errno == ESHUTDOWN)
 134         {
 135           // The queue has been deactivated, so let's bail out.
 136           ACE_DEBUG ((LM_DEBUG,
 137                       ACE_TEXT ("(%t) in iteration %d, queue len = %d, ")
 138                       ACE_TEXT ("queue deactivated, exiting\n"),
 139                       count,
 140                       this->msg_queue ()->message_count ()));
 141
 142           break;
 143         }
 144
 145       size_t length = mb->length ();
 146
 147       if (length > 0)
 148         ACE_DEBUG ((LM_DEBUG,
 149                     ACE_TEXT ("(%t) in iteration %d, queue len = %d, ")
 150                     ACE_TEXT ("length = %d, text = \"%*s\"\n"),
 151                     count,
 152                     this->msg_queue ()->message_count (),
 153                     length,
 154                     length - 1,
 155                     mb->rd_ptr ()));
 156
 157       // We're responsible for deallocating this.
 158       mb->release ();
 159
 160       if (length == 0)
 161         {
 162           ACE_DEBUG ((LM_DEBUG,
 163                       ACE_TEXT ("(%t) in iteration %d, queue len = %d, ")
 164                       ACE_TEXT ("got \"empty\" message, exiting\n"),
 165                       count,
 166                       this->msg_queue ()->message_count ()));
 167           break;
 168         }
 169     }
 170
 171   // Note that the <ACE_Task::svc_run> method automatically removes us
 172   // from the <ACE_Thread_Manager> when the thread exits.
 173   return 0;
 174 }
 175
 176 int
 177 Thread_Pool::open (void *)
 178 {
 179   ACE_DEBUG ((LM_DEBUG,
 180               ACE_TEXT ("(%t) producer start, dumping the Thread_Pool\n")));
 181   this->dump ();
 182
 183   // Create a pool of worker threads.
 184   if (this->activate (THR_NEW_LWP,
 185                       this->n_threads_) == -1)
 186     ACE_ERROR_RETURN ((LM_ERROR,
 187                        ACE_TEXT ("%p\n"),
 188                        ACE_TEXT ("activate failed")),
 189                       -1);
 190   return 0;
 191 }
 192
 193 // Activate the task's thread pool, produce the messages that are
 194 // consumed by the threads in the thread pool, and demonstate how to
 195 // shutdown using the <ACE_Message_Queue::deactivate> method.
 196
 197 int
 198 Thread_Pool::test_queue_deactivation_shutdown ()
 199 {
 200   if (this->open () == -1)
 201     return -1;
 202
 203   ACE_Message_Block *mb = 0;
 204
 205   // Run the main loop that generates messages and enqueues them into
 206   // the pool of threads managed by <ACE_Task>.
 207
 208   for (size_t count = 0;
 209        ;
 210        count++)
 211     {
 212       ssize_t n = 0;
 213
 214       // Allocate a new message.
 215       ACE_NEW_RETURN (mb,
 216                       ACE_Message_Block (BUFSIZ,
 217                                          ACE_Message_Block::MB_DATA,
 218                                          0,
 219                                          0,
 220                                          0,
 221                                          &this->lock_adapter_),
 222                       -1);
 223
 224       if (manual)
 225         {
 226           ACE_DEBUG ((LM_DEBUG,
 227                       ACE_TEXT ("(%t) enter a new message for ")
 228                       ACE_TEXT ("the task pool...")));
 229           n = ACE_OS::read (ACE_STDIN,
 230                             mb->wr_ptr (),
 231                             mb->size ());
 232         }
 233       else
 234         {
 235           static size_t count = 0;
 236
 237           ACE_OS::snprintf (reinterpret_cast<ACE_TCHAR *> (mb->wr_ptr ()), BUFSIZ,
 238                            ACE_SIZE_T_FORMAT_SPECIFIER,
 239                            count);
 240           n = ACE_OS::strlen (mb->rd_ptr ());
 241
 242           if (count == n_iterations)
 243             n = 1; // Indicate that we need to shut down.
 244           else
 245             count++;
 246
 247           if (count == 0 || (count % 20 == 0))
 248             ACE_OS::sleep (1);
 249         }
 250
 251       if (n > 1)
 252         {
 253           // Send a normal message to the waiting threads and continue
 254           // producing.
 255           mb->wr_ptr (n * sizeof (ACE_TCHAR));
 256
 257           // Pass the message to the Thread_Pool.
 258           if (this->put (mb) == -1)
 259             ACE_ERROR ((LM_ERROR,
 260                         ACE_TEXT (" (%t) %p\n"),
 261                         ACE_TEXT ("put")));
 262         }
 263       else
 264         {
 265           // Release the <Message_Block> since we're shutting down and
 266           // don't need it anymore.
 267
 268           mb->release ();
 269           // Deactivate the message queue and return.
 270           ACE_DEBUG ((LM_DEBUG,
 271                       ACE_TEXT ("\n(%t) deactivating queue for %d threads, ")
 272                       ACE_TEXT ("dump of task:\n"),
 273                       this->thr_count ()));
 274           this->dump ();
 275
 276           // Deactivate the queue.
 277           return this->msg_queue ()->deactivate ();
 278         }
 279     }
 280 }
 281
 282 // Activate the task's thread pool, produce the messages that are,
 283 // produce the messages that are consumed by the threads in the thread
 284 // pool, and demonstrate how to shutdown by enqueueing "empty"
 285 // messages into the queue.
 286
 287 int
 288 Thread_Pool::test_empty_message_shutdown ()
 289 {
 290   if (this->open () == -1)
 291     return -1;
 292
 293   ACE_Message_Block *mb = 0;
 294
 295   // Run the main loop that generates messages and enqueues them into
 296   // the pool of threads managed by <ACE_Task>.
 297
 298   for (size_t count = 0;
 299        ;
 300        count++)
 301     {
 302       ssize_t n = 0;
 303
 304       // Allocate a new message.
 305       ACE_NEW_RETURN (mb,
 306                       ACE_Message_Block (BUFSIZ,
 307                                          ACE_Message_Block::MB_DATA,
 308                                          0,
 309                                          0,
 310                                          0,
 311                                          &this->lock_adapter_),
 312                       -1);
 313
 314       if (manual)
 315         {
 316           ACE_DEBUG ((LM_DEBUG,
 317                       ACE_TEXT ("(%t) enter a new message for ")
 318                       ACE_TEXT ("the task pool...")));
 319           n = ACE_OS::read (ACE_STDIN,
 320                             mb->wr_ptr (),
 321                             mb->size ());
 322         }
 323       else
 324         {
 325           static size_t count = 0;
 326
 327           ACE_OS::snprintf (reinterpret_cast<ACE_TCHAR *> (mb->wr_ptr ()), BUFSIZ,
 328                            ACE_SIZE_T_FORMAT_SPECIFIER,
 329                            count);
 330           n = ACE_OS::strlen (mb->rd_ptr ());
 331
 332           if (count == n_iterations)
 333             n = 1; // Indicate that we need to shut down.
 334           else
 335             count++;
 336
 337           if (count == 0 || (count % 20 == 0))
 338             ACE_OS::sleep (1);
 339         }
 340
 341       if (n > 1)
 342         {
 343           // Send a normal message to the waiting threads and continue
 344           // producing.
 345           mb->wr_ptr (n * sizeof (ACE_TCHAR));
 346
 347           // Pass the message to the Thread_Pool.
 348           if (this->put (mb) == -1)
 349             ACE_ERROR ((LM_ERROR,
 350                         ACE_TEXT (" (%t) %p\n"),
 351                         ACE_TEXT ("put")));
 352         }
 353       else
 354         {
 355           // Send a shutdown message to the waiting threads and return.
 356           ACE_DEBUG ((LM_DEBUG,
 357                       ACE_TEXT ("\n(%t) sending shutdown message to %d threads, ")
 358                       ACE_TEXT ("dump of task:\n"),
 359                       this->thr_count ()));
 360           this->dump ();
 361
 362           size_t i = 0;
 363
 364           // Enqueue an empty message to flag each consumer thread to
 365           // inform it to shutdown.
 366           for (i = this->thr_count ();
 367                i > 0;
 368                i--)
 369             {
 370               ACE_DEBUG ((LM_DEBUG,
 371                           ACE_TEXT ("(%t) end of input, ")
 372                           ACE_TEXT ("enqueueing \"empty\" message %d\n"),
 373                           i));
 374
 375               // Note the use of reference counting to avoid copying
 376               // the message contents.
 377               ACE_Message_Block *dup = mb->duplicate ();
 378
 379               if (this->put (dup) == -1)
 380                 ACE_ERROR ((LM_ERROR,
 381                             ACE_TEXT (" (%t) %p\n"),
 382                             ACE_TEXT ("put")));
 383             }
 384
 385           mb->release ();
 386
 387           ACE_DEBUG ((LM_DEBUG,
 388                       ACE_TEXT ("\n(%t) end loop, dump of task:\n")));
 389           this->dump ();
 390
 391           return 0;
 392         }
 393     }
 394 }
 395
 396 #endif /* ACE_HAS_THREADS */
 397
 398 int
 399 run_main (int, ACE_TCHAR *[])
 400 {
 401   ACE_START_TEST (ACE_TEXT ("Thread_Pool_Test"));
 402
 403 #if defined (ACE_HAS_THREADS)
 404   int n_threads = ACE_MAX_THREADS;
 405
 406   // Create the worker tasks.
 407   Thread_Pool thread_pool (n_threads);
 408
 409   ACE_DEBUG ((LM_DEBUG,
 410               ACE_TEXT ("(%t) running test with %d threads\n"),
 411               n_threads));
 412
 413   ACE_DEBUG ((LM_DEBUG,
 414               ACE_TEXT ("(%t) starting empty message shutdown test\n")));
 415
 416   // Activate the task's thread pool, produce the messages that are,
 417   // produce the messages that are consumed by the threads in the
 418   // thread pool, and demonstrate how to shutdown by enqueueing
 419   // "empty" messages into the queue.
 420   if (thread_pool.test_empty_message_shutdown () == -1)
 421     return 1;
 422
 423   ACE_DEBUG ((LM_DEBUG,
 424               ACE_TEXT ("(%t) waiting for worker tasks to finish...\n")));
 425   // Wait for all the threads to reach their exit point, at which
 426   // point the barrier in the destructor of the <ACE_Task> portion of
 427   // <Thread_Pool> will return.
 428   if (thread_pool.wait () == -1)
 429     return 1;
 430
 431   ACE_DEBUG ((LM_DEBUG,
 432               ACE_TEXT ("(%t) starting queue deactivation shutdown test\n")));
 433
 434   // Activate the task's thread pool, produce the messages that are
 435   // consumed by the threads in the thread pool, and demonstate how to
 436   // shutdown using the <ACE_Message_Queue::deactivate> method.
 437   if (thread_pool.test_queue_deactivation_shutdown () == -1)
 438     return 1;
 439
 440   ACE_DEBUG ((LM_DEBUG,
 441               ACE_TEXT ("(%t) waiting for worker tasks to finish...\n")));
 442   // Wait for all the threads to reach their exit point, at which
 443   // point the barrier in the destructor of the <ACE_Task> portion of
 444   // <Thread_Pool> will return.
 445   if (thread_pool.wait () == -1)
 446     return 1;
 447
 448   ACE_DEBUG ((LM_DEBUG,
 449               ACE_TEXT ("(%t) all worker tasks destroyed, exiting test...\n")));
 450 #else
 451   ACE_ERROR ((LM_INFO,
 452               ACE_TEXT ("threads not supported on this platform\n")));
 453 #endif /* ACE_HAS_THREADS */
 454   ACE_END_TEST;
 455   return 0;
 456 }