ACE/tests/Thread_Creation_Threshold_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Thread_Creation_Threshold_Test.cpp
   5  *
   6  *   This test program stresses how many threads can be
   7  *   consecutively (not simultaneously) created on a platform.
   8  *   Rather than testing exhaustively, it establishes a
   9  *   semi-arbitrary upper limit (MAX_THREAD)of threads.
  10  *
  11  *  @author Chris Cleeland <cleeland@ociweb.com>
  12  */
  13 //=============================================================================
  14
  15
  16 #include "test_config.h"
  17 #include "ace/Task.h"
  18
  19
  20 #if defined (ACE_HAS_THREADS)
  21 #include "ace/Lock_Adapter_T.h"
  22 #include "ace/OS_NS_stdio.h"
  23 #include "ace/OS_NS_string.h"
  24 #include "ace/OS_NS_unistd.h"
  25
  26 // Number of iterations to run the test.
  27 #if defined (ACE_VXWORKS)
  28 const unsigned long MAX_THREAD = 500;
  29 #else
  30  # ifdef ACE_FACE_SAFETY_BASE
  31 const unsigned long MAX_THREAD = 10000;
  32  # else
  33 const unsigned long MAX_THREAD = 100000;
  34  # endif
  35 #endif
  36
  37 namespace
  38 {
  39   // Change this to 'true' if you want lots of debugging messages in the log
  40   const bool PRINT_DEBUG_MSGS = true;
  41 }
  42
  43 /**
  44  * @class Thread_Pool
  45  *
  46  * @brief Defines a thread pool abstraction based on the <ACE_Task>.
  47  */
  48 class Thread_Pool : public ACE_Task<ACE_MT_SYNCH>
  49 {
  50 public:
  51   /// Create the thread pool containing <n_threads>.
  52   Thread_Pool (int n_threads);
  53
  54   /// Destructor...
  55   ~Thread_Pool () override;
  56
  57   /// Iterate <n_iterations> time printing off a message and "waiting"
  58   /// for all other threads to complete this iteration.
  59   int svc () override;
  60
  61   /// Start the threads in the pool.
  62   void start();
  63
  64   unsigned long total_threads()
  65     {
  66       return this->total_activated_threads_.value();
  67     }
  68
  69   /// Number of threads to spawn.
  70   int n_threads_;
  71
  72   /// Returns true iff failed_ == false.
  73   bool operator! ();
  74
  75 private:
  76   //FUZZ: disable check_for_lack_ACE_OS
  77   /// Spawn the threads in the pool.
  78   int open (void * = 0) override;
  79
  80   /// Close hook.
  81   ///FUZZ: enable check_for_lack_ACE_OS
  82   int close (u_long) override;
  83
  84   /// Total number of threads activated through this thread pool ever.
  85   ACE_Atomic_Op<ACE_Thread_Mutex, unsigned long> total_activated_threads_;
  86
  87   /// Flag set only in worker threads to indicate whether they should print
  88   /// debug messages.
  89   bool doprint_;
  90
  91   /// Flag indicating that start() failed.
  92   bool failed_;
  93 };
  94
  95 bool
  96 Thread_Pool::operator!()
  97 {
  98   return ! this->failed_;
  99 }
 100
 101 Thread_Pool::~Thread_Pool ()
 102 {
 103 }
 104
 105 int
 106 Thread_Pool::close (u_long)
 107 {
 108   if (this->doprint_)
 109     ACE_DEBUG ((LM_DEBUG,
 110                 ACE_TEXT ("(%P|%t) worker thread closing down\n")));
 111   return 0;
 112 }
 113
 114 Thread_Pool::Thread_Pool (int n_threads)
 115   : n_threads_ (n_threads)
 116   , total_activated_threads_ (0)
 117   , doprint_ (false)
 118   , failed_ (false)
 119 {
 120 }
 121
 122 void
 123 Thread_Pool::start ()
 124 {
 125   this->failed_ = false;
 126   if (this->open () == -1)
 127     {
 128       ACE_DEBUG ((LM_DEBUG,
 129                   ACE_TEXT ("(%P|%t) thread activation ")
 130                   ACE_TEXT ("failed after %u threads\n"),
 131                   this->total_threads()));
 132       this->failed_ = true;
 133     }
 134 }
 135
 136 int
 137 Thread_Pool::svc ()
 138 {
 139   unsigned long t = ++this->total_activated_threads_;
 140   if (PRINT_DEBUG_MSGS) // change this to 'true' for debugging
 141     this->doprint_ = (t > 65530);
 142
 143   if (this->doprint_)
 144     ACE_DEBUG ((LM_DEBUG, ACE_TEXT("  (%P|%t) thread started.\n")));
 145
 146   // Note that the <ACE_Task::svc_run> method automatically removes us
 147   // from the <ACE_Thread_Manager> when the thread exits.
 148   return 0;
 149 }
 150
 151 namespace {
 152   void no_op() { }
 153 }
 154
 155 int
 156 Thread_Pool::open (void *)
 157 {
 158   if (PRINT_DEBUG_MSGS)
 159     ACE_DEBUG ((LM_DEBUG,
 160                 ACE_TEXT ("(%P|%t) pool start %d threads...\n"),
 161                 this->n_threads_));
 162
 163   if (this->total_activated_threads_ >= (65534 - this->n_threads_))
 164     no_op ();
 165
 166   // Create a pool of worker threads.
 167   if (this->activate (THR_NEW_LWP,
 168                       this->n_threads_) == -1)
 169     ACE_ERROR_RETURN ((LM_ERROR,
 170                        ACE_TEXT ("%p\n"),
 171                        ACE_TEXT ("activate failed")),
 172                       -1);
 173
 174   return 0;
 175 }
 176
 177 #endif /* ACE_HAS_THREADS */
 178
 179
 180 int
 181 run_main (int, ACE_TCHAR *[])
 182 {
 183   ACE_START_TEST (ACE_TEXT ("Thread_Creation_Threshold_Test"));
 184
 185 #if defined (ACE_HAS_THREADS)
 186   ACE_DEBUG ((LM_DEBUG,
 187               ACE_TEXT ("(%P|%t) will try to start and kill up")
 188               ACE_TEXT (" to %u threads sequentially\n"),
 189               MAX_THREAD));
 190   int initial_pool_size = 50;
 191
 192   // Create the worker tasks.
 193   Thread_Pool thread_pool (initial_pool_size);
 194
 195   while (!thread_pool && thread_pool.total_threads() < MAX_THREAD)
 196     {
 197       // Activate the task's thread pool, produce the messages that are,
 198       // produce the messages that are consumed by the threads in the
 199       // thread pool, and demonstrate how to shutdown by enqueueing
 200       // "empty" messages into the queue.
 201       thread_pool.start ();
 202
 203       // Wait for all the threads to reach their exit point, at which
 204       // point the barrier in the destructor of the <ACE_Task> portion of
 205       // <Thread_Pool> will return.
 206       if (thread_pool.wait () == -1)
 207         {
 208           ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("wait")), 1);
 209         }
 210
 211       if (PRINT_DEBUG_MSGS)
 212         ACE_DEBUG ((LM_DEBUG,
 213                     ACE_TEXT ("%u total threads\n"),
 214                     thread_pool.total_threads()));
 215
 216       int& n_threads = thread_pool.n_threads_;
 217       const unsigned long THREAD_THRESHOLD = 63336;
 218       if (thread_pool.total_threads() >= (THREAD_THRESHOLD - n_threads))
 219         n_threads = 1;
 220     }
 221
 222   ACE_DEBUG ((LM_DEBUG,
 223               ACE_TEXT ("%u total threads successfully started and died;")
 224               ACE_TEXT (" expected %u.\n"),
 225               thread_pool.total_threads (),
 226               MAX_THREAD));
 227 #else
 228   ACE_ERROR ((LM_INFO,
 229               ACE_TEXT ("threads not supported on this platform\n")));
 230 #endif /* ACE_HAS_THREADS */
 231   ACE_END_TEST;
 232   return 0;
 233 }