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