ACE/examples/Threads/manual_event.cpp

   1 // The test shows the use of an ACE_Manual_Event to create a
   2 // Pseudo_Barrier. Multiple threads are created which do the
   3 // following:
   4 //
   5 //      1. work
   6 //      2. synch with other threads
   7 //      3. more work
   8 //
   9 // ACE_Manual_Event is use to synch with other
  10 // threads. ACE_Manual_Event::signal() is used for broadcasting.
  11
  12 #include "ace/OS_NS_unistd.h"
  13 #include "ace/OS_main.h"
  14 #include "ace/Service_Config.h"
  15 #include "ace/Thread_Mutex.h"
  16 #include "ace/Manual_Event.h"
  17 #include "ace/Thread_Manager.h"
  18 #include "ace/Atomic_Op.h"
  19
  20
  21 #if defined (ACE_HAS_THREADS)
  22 static ACE_Atomic_Op <ACE_Thread_Mutex, int> amount_of_work = 0;
  23
  24 class Pseudo_Barrier
  25   // = TITLE
  26   //    A barrier class using ACE manual-reset events.
  27   //
  28   // = DESCRIPTION
  29   //    This is *not* a real barrier.
  30   //    Pseudo_Barrier is more like a ``one shot'' barrier.
  31   //    All waiters after the Nth waiter are allowed to go.
  32   //    The barrier does not reset after the Nth waiter.
  33   //    For an example of a real barrier, please see class ACE_Barrier.
  34 {
  35 public:
  36   Pseudo_Barrier (u_long count);
  37
  38   //FUZZ: disable check_for_lack_ACE_OS
  39   int wait ();
  40   //FUZZ: enable check_for_lack_ACE_OS
  41
  42 private:
  43   ACE_Atomic_Op <ACE_Thread_Mutex, int> counter_;
  44   ACE_Manual_Event event_;
  45 };
  46
  47 Pseudo_Barrier::Pseudo_Barrier (u_long count)
  48   : counter_ (count)
  49 {
  50 }
  51
  52 int
  53 Pseudo_Barrier::wait ()
  54 {
  55   if (--this->counter_ == 0)
  56     return this->event_.signal ();
  57   else
  58     return this->event_.wait ();
  59 }
  60
  61 static void *
  62 worker (void *arg)
  63 {
  64   Pseudo_Barrier &thread_barrier = *(Pseudo_Barrier *) arg;
  65
  66   // work
  67   ACE_DEBUG ((LM_DEBUG, "(%t) working (%d secs)\n", ++::amount_of_work));
  68   ACE_OS::sleep (::amount_of_work.value ());
  69
  70   // synch with everybody else
  71   ACE_DEBUG ((LM_DEBUG, "(%t) waiting to synch with others\n"));
  72   thread_barrier.wait ();
  73
  74   // more work
  75   ACE_DEBUG ((LM_DEBUG, "(%t) more work (%d secs)\n", ++::amount_of_work));
  76   ACE_OS::sleep (::amount_of_work.value ());
  77
  78   ACE_DEBUG ((LM_DEBUG, "(%t) dying\n"));
  79
  80   return 0;
  81 }
  82
  83 int
  84 ACE_TMAIN (int argc, ACE_TCHAR **argv)
  85 {
  86   int n_threads = argc == 2 ? ACE_OS::atoi (argv[1]) : 5;
  87
  88   ACE_Thread_Manager &tm = *ACE_Thread_Manager::instance ();
  89
  90   // synch object shared by all threads
  91   Pseudo_Barrier thread_barrier (n_threads);
  92
  93   // create workers
  94   if (tm.spawn_n (n_threads, (ACE_THR_FUNC) worker, &thread_barrier) == -1)
  95     ACE_ERROR_RETURN ((LM_ERROR, "thread creates for worker failed"), -1);
  96
  97   // wait for all workers to exit
  98   if (tm.wait () == -1)
  99     ACE_ERROR_RETURN ((LM_ERROR, "thread wait failed"), -1);
 100   else
 101     ACE_DEBUG ((LM_ERROR, "graceful exit\n"));
 102
 103   return 0;
 104 }
 105
 106 #else
 107 int
 108 ACE_TMAIN (int, ACE_TCHAR *[])
 109 {
 110   ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n"));
 111   return 0;
 112 }
 113 #endif /* ACE_HAS_THREADS */