ACE/tests/Thread_Manager_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Thread_Manager_Test.cpp
   5  *
   6  *    This program tests the group management mechanisms provided by
   7  *    the <ACE_Thread_Manager>, including the group signal handling,
   8  *    group suspension and resumption, and cooperative thread
   9  *    cancellation mechanisms.
  10  *
  11  *  @author Prashant Jain <pjain@cs.wustl.edu> and Douglas C. Schmidt <d.schmidt@vanderbilt.edu>
  12  */
  13 //=============================================================================
  14
  15 #include "test_config.h"
  16 #include "ace/Thread_Manager.h"
  17 #include "ace/Signal.h"
  18 #include "ace/Task.h"
  19 #include "ace/OS_NS_unistd.h"
  20 #include "ace/OS_NS_sys_time.h"
  21
  22 #if defined (ACE_HAS_THREADS)
  23 #include "ace/Barrier.h"
  24
  25 // Each thread keeps track of whether it has been signalled by using a
  26 // global array.  It must be dynamically allocated to allow sizing at
  27 // runtime, based on the number of threads.
  28 static ACE_thread_t *signalled = 0;
  29 static size_t n_threads = ACE_MAX_THREADS;
  30
  31 // Helper function that looks for an existing entry in the signalled
  32 // array.  Also finds the position of the first unused entry in the
  33 // array, and updates if requested with the t_id.
  34 extern "C" int
  35 been_signalled (const ACE_thread_t t_id,
  36                 const u_int update = 0)
  37 {
  38   u_int unused_slot = n_threads;
  39
  40   for (u_int i = 0; i < n_threads; ++i)
  41     {
  42       if (ACE_OS::thr_equal (signalled[i], t_id))
  43         // Already signalled.
  44         return 1;
  45
  46       if (update  &&
  47           unused_slot == n_threads  &&
  48           ACE_OS::thr_equal (signalled[i],
  49                              ACE_OS::NULL_thread))
  50         unused_slot = i;
  51     }
  52
  53   if (update  &&  unused_slot < n_threads)
  54     // Update the array using the first unused_slot.
  55     signalled[unused_slot] = t_id;
  56
  57   return 0;
  58 }
  59
  60 // Synchronize starts of threads, so that they all start before the
  61 // main thread cancels them.  To avoid creating a static object, it is
  62 // dynamically allocated, before spawning any threads.
  63 static ACE_Barrier *thread_start = 0;
  64
  65 extern "C" void
  66 handler (int /* signum */)
  67 {
  68   if (signalled)
  69     {
  70       // No printout here, to be safe.  Signal handlers must not
  71       // acquire locks, etc.
  72       const ACE_thread_t t_id = ACE_OS::thr_self ();
  73
  74       // Update the signalled indication.
  75       (void) been_signalled (t_id, 1u /* update */);
  76     }
  77 }
  78
  79 static void *
  80 worker (int iterations)
  81 {
  82 #if !defined (ACE_LACKS_UNIX_SIGNALS)
  83   // Cache this thread's ID.
  84   const ACE_thread_t t_id = ACE_OS::thr_self ();
  85 #endif /* ! ACE_LACKS_UNIX_SIGNAL */
  86
  87   ACE_Thread_Manager *thr_mgr = ACE_Thread_Manager::instance ();
  88
  89   // After setting up the signal catcher, block on the start barrier.
  90   thread_start->wait ();
  91
  92   ACE_DEBUG ((LM_DEBUG,
  93               ACE_TEXT ("(%t) worker starting loop\n")));
  94
  95   for (int i = 0; i < iterations; i++)
  96     {
  97       if ((i % 1000) == 0)
  98         {
  99 #if !defined (ACE_LACKS_UNIX_SIGNALS)
 100           if (been_signalled (t_id))
 101             {
 102               ACE_DEBUG ((LM_DEBUG,
 103                           ACE_TEXT ("(%t) had received signal\n")));
 104
 105               // Only test for cancellation after we've been signaled,
 106               // to avoid race conditions for suspend() and resume().
 107               if (thr_mgr->testcancel (ACE_Thread::self ()) != 0)
 108                 {
 109                   char thr_id[BUFSIZ];
 110                   // Test out the ACE_OS::thr_id() method.
 111                   ACE_OS::thr_id (thr_id, sizeof thr_id);
 112                   ACE_DEBUG ((LM_DEBUG,
 113                               ACE_TEXT ("(%C) has been cancelled ")
 114                               ACE_TEXT ("before iteration %d!\n"),
 115                               thr_id,
 116                               i));
 117                   break;
 118                 }
 119             }
 120 #else
 121           if (thr_mgr->testcancel (ACE_Thread::self ()) != 0)
 122             {
 123               char thr_id[BUFSIZ];
 124               // Test out the ACE_OS::thr_id() method.
 125               ACE_OS::thr_id (thr_id, sizeof thr_id);
 126               ACE_DEBUG ((LM_DEBUG,
 127                           ACE_TEXT ("(%C) has been cancelled ")
 128                           ACE_TEXT ("before iteration %d!\n"),
 129                           thr_id,
 130                           i));
 131               break;
 132             }
 133 #endif /* ! ACE_LACKS_UNIX_SIGNAL */
 134           ACE_OS::sleep (1);
 135         }
 136     }
 137
 138   // Destructor removes thread from Thread_Manager.
 139   return 0;
 140 }
 141
 142 static const int DEFAULT_ITERATIONS = 10000;
 143
 144 // Define a ACE_Task that will serve in the tests related to tasks.
 145
 146 class ThrMgr_Task : public ACE_Task_Base {
 147 public:
 148   ThrMgr_Task (ACE_Thread_Manager *);
 149
 150   int svc () override;
 151
 152   static int errors;
 153 };
 154
 155 int ThrMgr_Task::errors = 0;
 156
 157 // Just be sure to set the ACE_Thread_Manager correctly.
 158 ThrMgr_Task::ThrMgr_Task (ACE_Thread_Manager *mgr)
 159   : ACE_Task_Base (mgr)
 160 {
 161 }
 162
 163 // svc just waits til it's been cancelled, then exits.
 164 int
 165 ThrMgr_Task::svc ()
 166 {
 167   ACE_DEBUG ((LM_DEBUG,
 168               ACE_TEXT ("Task 0x%x, thread %t waiting to be cancelled\n"),
 169               this));
 170   ACE_thread_t me = ACE_Thread::self ();
 171   for (int i = 0; i < 30 && !this->thr_mgr ()->testcancel (me); ++i)
 172     ACE_OS::sleep (1);
 173
 174   if (this->thr_mgr ()->testcancel (me))
 175     {
 176       ACE_DEBUG ((LM_DEBUG,
 177                   ACE_TEXT ("Task 0x%x, thread %t cancelled; exiting\n"),
 178                   this));
 179     }
 180   else
 181     {
 182       ACE_ERROR ((LM_ERROR,
 183                   ACE_TEXT ("Task 0x%x, thread %t was not cancelled\n"),
 184                   this));
 185       ++ThrMgr_Task::errors;
 186     }
 187   return 0;
 188 }
 189
 190
 191 // This function tests the task-based record keeping functions.
 192 static int
 193 test_task_record_keeping (ACE_Thread_Manager *mgr)
 194 {
 195   int status = 0;
 196
 197   ThrMgr_Task t1 (mgr), t2 (mgr);
 198   int t1_grp (20), t2_grp (30);
 199
 200   // Start two tasks, with multiple threads per task. Make sure that
 201   // task_all_list() works.
 202   if (-1 == t1.activate (THR_JOINABLE, 2, 0,
 203                          ACE_DEFAULT_THREAD_PRIORITY, t1_grp))
 204     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("activate")), 1);
 205   if (-1 == t2.activate (THR_JOINABLE, 3, 0,
 206                          ACE_DEFAULT_THREAD_PRIORITY, t2_grp))
 207     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("activate")), 1);
 208
 209   ACE_Task_Base *task_list[10];
 210   int num_tasks = mgr->task_all_list (task_list, 10);
 211   if (2 != num_tasks)
 212     {
 213       ACE_ERROR ((LM_ERROR, ACE_TEXT ("Expected 2 tasks; got %d\n"),
 214                   num_tasks));
 215       status = 1;
 216     }
 217   else
 218     {
 219       ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Got %d tasks - correct\n"), num_tasks));
 220       if (((task_list[0] == &t1 && task_list[1] == &t2)
 221            || (task_list[1] == &t1 && task_list[0] == &t2))
 222           && task_list[0] != task_list[1])
 223         ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("The Task IDs are correct\n")));
 224       else
 225         ACE_ERROR ((LM_ERROR, ACE_TEXT ("But Task ID values are wrong!\n")));
 226     }
 227   ACE_DEBUG ((LM_DEBUG, "Canceling grp %d\n", t1_grp));
 228   if (-1 == mgr->cancel_grp (t1_grp))
 229     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("cancel_grp")),
 230                       1);
 231   ACE_DEBUG ((LM_DEBUG, "Canceling grp %d\n", t2_grp));
 232   if (-1 == mgr->cancel_grp (t2_grp))
 233     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("cancel_grp")),
 234                       1);
 235
 236   mgr->wait ();
 237   if (ThrMgr_Task::errors > 0 && status == 0)
 238     status = 1;
 239
 240   return status;
 241 }
 242
 243 #endif /* ACE_HAS_THREADS */
 244
 245 int
 246 run_main (int, ACE_TCHAR *[])
 247 {
 248   ACE_START_TEST (ACE_TEXT ("Thread_Manager_Test"));
 249   int status = 0;
 250
 251 #if defined (ACE_HAS_THREADS)
 252   size_t n_iterations = DEFAULT_ITERATIONS;
 253
 254   u_int i;
 255
 256   // Dynamically allocate signalled so that we can control when it
 257   // gets deleted.  Specifically, we need to delete it before the main
 258   // thread's TSS is cleaned up.
 259   ACE_NEW_RETURN (signalled,
 260                   ACE_thread_t[n_threads],
 261                   1);
 262   // Initialize each ACE_thread_t to avoid Purify UMR's.
 263   for (i = 0; i < n_threads; ++i)
 264     signalled[i] = ACE_OS::NULL_thread;
 265
 266   // And similarly, dynamically allocate the thread_start barrier.
 267   ACE_NEW_RETURN (thread_start,
 268                   ACE_Barrier (n_threads + 1),
 269                   -1);
 270
 271   // Register a signal handler.
 272   ACE_Sig_Action sa ((ACE_SignalHandler) handler, SIGINT);
 273   ACE_UNUSED_ARG (sa);
 274
 275   ACE_Thread_Manager *thr_mgr = ACE_Thread_Manager::instance ();
 276
 277 #if 0
 278   // Assign thread (VxWorks task) names to test that feature.
 279   ACE_hthread_t *thread_name = 0;
 280   ACE_NEW_RETURN (thread_name,
 281                   ACE_hthread_t[n_threads],
 282                   -1);
 283
 284   // And test the ability to specify stack size.
 285   size_t *stack_size = 0;
 286   ACE_NEW_RETURN (stack_size,
 287                   size_t[n_threads],
 288                   -1);
 289
 290   for (i = 0; i < n_threads; ++i)
 291     {
 292       if (i < n_threads - 1)
 293         {
 294           ACE_NEW_RETURN (thread_name[i], char[32], -1);
 295           ACE_OS::snprintf (thread_name[i], 32, ACE_TEXT ("thread%u"), i);
 296         }
 297       else
 298         // Pass an ACE_thread_t pointer of 0 for the last thread name.
 299         thread_name[n_threads - 1] = 0;
 300
 301       stack_size[i] = 40000;
 302     }
 303 #endif /* ACE_HAS_VXTHREADS */
 304
 305   int grp_id = thr_mgr->spawn_n
 306                  (
 307                   n_threads,
 308                   (ACE_THR_FUNC) worker,
 309                   reinterpret_cast <void *> (n_iterations),
 310                   THR_BOUND
 311                   , ACE_DEFAULT_THREAD_PRIORITY
 312                   , -1
 313 #if 0
 314                   , 0
 315                   , thread_name
 316                   , 0
 317                   , stack_size
 318 #endif /* ACE_HAS_VXTHREADS */
 319                   );
 320
 321   ACE_TEST_ASSERT (grp_id != -1);
 322   thread_start->wait ();
 323
 324   // Wait for 1 second and then suspend every thread in the group.
 325   ACE_OS::sleep (1);
 326   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) suspending group\n")));
 327   if (thr_mgr->suspend_grp (grp_id) == -1)
 328     {
 329       // Pthreads w/o UNIX 98 extensions doesn't support suspend/resume,
 330       // so it's allowed to ENOTSUP as long as the config indicates this.
 331       if (errno == ENOTSUP)
 332         {
 333 #if defined (ACE_HAS_PTHREADS) && (defined (ACE_HAS_PTHREAD_SUSPEND) || \
 334                                    defined (ACE_HAS_PTHREAD_SUSPEND_NP))
 335           ACE_ERROR ((LM_ERROR,
 336                       ACE_TEXT ("suspend_grp: ENOTSUP but config ")
 337                       ACE_TEXT ("says it should work.\n")));
 338 #else
 339           ACE_DEBUG((LM_DEBUG,
 340                      ACE_TEXT (" OK: suspend_grp isn't supported with ")
 341                      ACE_TEXT ("Pthreads\n")));
 342 #endif /* ACE_HAS_PTHREADS && should be able to suspend */
 343         }
 344       else
 345         {
 346           ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("suspend_grp")));
 347         }
 348     }
 349
 350   // Wait for 1 more second and then resume every thread in the
 351   // group.
 352   ACE_OS::sleep (ACE_Time_Value (1));
 353
 354   ACE_DEBUG ((LM_DEBUG,
 355               ACE_TEXT ("(%t) resuming group\n")));
 356
 357   if (thr_mgr->resume_grp (grp_id) == -1)
 358     {
 359       // Pthreads w/o UNIX 98 extensions doesn't support suspend/resume,
 360       // so it's allowed to ENOTSUP as long as the config indicates this.
 361       if (errno == ENOTSUP)
 362         {
 363 #if defined (ACE_HAS_PTHREADS) && (defined (ACE_HAS_PTHREAD_CONTINUE) || \
 364                                    defined (ACE_HAS_PTHREAD_CONTINUE_NP) || \
 365                                    defined (ACE_HAS_PTHREAD_RESUME_NP))
 366           ACE_ERROR ((LM_ERROR,
 367                       ACE_TEXT ("resume_grp: ENOTSUP but config ")
 368                       ACE_TEXT ("says it should work.\n")));
 369 #else
 370           ACE_DEBUG((LM_DEBUG,
 371                      ACE_TEXT (" OK: resume_grp isn't supported with ")
 372                      ACE_TEXT ("Pthreads\n")));
 373 #endif /* ACE_HAS_PTHREADS && should be able to continue/resume */
 374         }
 375       else
 376         {
 377           ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("resume_grp")));
 378         }
 379     }
 380
 381   // Wait for 1 more second and then send a SIGINT to every thread in
 382   // the group.
 383   ACE_OS::sleep (ACE_Time_Value (1));
 384
 385   ACE_DEBUG ((LM_DEBUG,
 386               ACE_TEXT ("(%t) signaling group\n")));
 387
 388 #if defined (ACE_HAS_WTHREADS)
 389   thr_mgr->kill_grp (grp_id,
 390                      SIGINT);
 391 #elif !defined (ACE_HAS_PTHREADS_DRAFT4) && !defined(ACE_LACKS_PTHREAD_KILL)
 392   ACE_TEST_ASSERT (thr_mgr->kill_grp (grp_id,
 393                                  SIGINT) != -1);
 394 #else
 395   if (thr_mgr->kill_grp (grp_id,
 396                          SIGINT) == -1)
 397     ACE_TEST_ASSERT (errno == ENOTSUP);
 398 #endif /* ACE_HAS_WTHREADS */
 399
 400   // Wait and then cancel all the threads.
 401   ACE_OS::sleep (ACE_Time_Value (1));
 402
 403   ACE_DEBUG ((LM_DEBUG,
 404               ACE_TEXT ("(%t) cancelling group\n")));
 405
 406   ACE_TEST_ASSERT (thr_mgr->cancel_grp (grp_id) != -1);
 407
 408   // Perform a barrier wait until all the threads have shut down.
 409   // But, wait for a limited time, just in case.
 410   ACE_Time_Value const max_wait (600);
 411   ACE_Time_Value const wait_time (ACE_OS::gettimeofday () + max_wait);
 412   if (thr_mgr->wait (&wait_time) == -1)
 413     {
 414       if (errno == ETIME)
 415         ACE_ERROR ((LM_ERROR,
 416                     ACE_TEXT ("maximum wait time of %d msec exceeded\n"),
 417                                max_wait.msec ()));
 418       else
 419         ACE_ERROR ((LM_ERROR,
 420                     ACE_TEXT ("%p\n"), ACE_TEXT ("wait")));
 421       status = -1;
 422     }
 423
 424   ACE_DEBUG ((LM_DEBUG,
 425               ACE_TEXT ("(%t) main thread finished\n")));
 426
 427 #if 0
 428   for (i = 0; i < n_threads - 1; ++i)
 429     {
 430       delete [] thread_name[i];
 431       // Don't delete the last thread_name, because it points to the
 432       // name in the TCB.  It was initially 0.
 433     }
 434   delete [] thread_name;
 435   delete [] stack_size;
 436 #endif /* ACE_HAS_VXTHREADS */
 437
 438   delete thread_start;
 439   thread_start = 0;
 440   delete [] signalled;
 441   signalled = 0;
 442
 443   // Now test task record keeping
 444   if (test_task_record_keeping (thr_mgr) != 0)
 445     status = -1;
 446
 447 #else
 448   ACE_ERROR ((LM_INFO,
 449               ACE_TEXT ("threads not supported on this platform\n")));
 450 #endif /* ACE_HAS_THREADS */
 451
 452   ACE_END_TEST;
 453   return status;
 454 }