ACE/examples/Reactor/Misc/test_reactors.cpp

   1 // Perform a torture test of multiple ACE_Reactors and ACE_Tasks in
   2 // the same process...  Thanks to Detlef Becker for contributing this.
   3
   4 #include "ace/Reactor.h"
   5 #include "ace/Service_Config.h"
   6 #include "ace/Task.h"
   7 #include "ace/Atomic_Op.h"
   8
   9
  10 #if defined (ACE_HAS_THREADS)
  11
  12 #include "ace/Recursive_Thread_Mutex.h"
  13
  14 static const int NUM_INVOCATIONS = 10;
  15 static const int MAX_TASKS = 20;
  16
  17 class Test_Task : public ACE_Task<ACE_MT_SYNCH>
  18 {
  19 public:
  20   Test_Task ();
  21   ~Test_Task ();
  22
  23   //FUZZ: disable check_for_lack_ACE_OS
  24   virtual int open (void *args = 0);
  25   virtual int close (u_long flags = 0);
  26   //FUZZ: enable check_for_lack_ACE_OS
  27
  28   virtual int svc ();
  29
  30   virtual int handle_input (ACE_HANDLE handle);
  31   virtual int handle_close (ACE_HANDLE fd,
  32                             ACE_Reactor_Mask close_mask);
  33
  34 private:
  35   int handled_;
  36
  37   static int task_count_;
  38 };
  39
  40 int Test_Task::task_count_ = 0;
  41
  42 static ACE_Atomic_Op<ACE_Thread_Mutex, int> done_count = MAX_TASKS * 2;
  43
  44 static ACE_Recursive_Thread_Mutex reclock_;
  45
  46 Test_Task::Test_Task ()
  47   : handled_ (0)
  48 {
  49   ACE_GUARD (ACE_Recursive_Thread_Mutex, ace_mon, reclock_);
  50
  51   Test_Task::task_count_++;
  52   ACE_DEBUG ((LM_DEBUG,
  53               "(%t) TT+ Test_Task::task_count_ = %d\n",
  54               Test_Task::task_count_));
  55 }
  56
  57 Test_Task::~Test_Task ()
  58 {
  59   ACE_GUARD (ACE_Recursive_Thread_Mutex, ace_mon, reclock_);
  60
  61   ACE_DEBUG ((LM_DEBUG,
  62               "(%t) TT- Test_Task::task_count_ = %d\n",
  63               Test_Task::task_count_));
  64 }
  65
  66 int
  67 Test_Task::open (void *args)
  68 {
  69   this->reactor ((ACE_Reactor *) args);
  70   return this->activate (THR_NEW_LWP);
  71 }
  72
  73 int
  74 Test_Task::close (u_long)
  75 {
  76   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, ace_mon, reclock_, -1);
  77
  78   Test_Task::task_count_--;
  79   ACE_DEBUG ((LM_DEBUG,
  80               "(%t) close Test_Task::task_count_ = %d\n",
  81               Test_Task::task_count_));
  82   return 0;
  83 }
  84
  85 int
  86 Test_Task::svc ()
  87 {
  88   for (int i = 0; i < NUM_INVOCATIONS; i++)
  89     {
  90       ACE_OS::thr_yield ();
  91
  92       // ACE_DEBUG ((LM_DEBUG, "(%t) calling notify %d\n", i));
  93
  94       if (this->reactor ()->notify (this, ACE_Event_Handler::READ_MASK) == -1)
  95         ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "notify"), -1);
  96
  97       // ACE_DEBUG ((LM_DEBUG, "(%t) leaving notify %d\n", i));
  98     }
  99
 100   return 0;
 101 }
 102
 103 int
 104 Test_Task::handle_close (ACE_HANDLE,
 105                          ACE_Reactor_Mask)
 106 {
 107   ACE_DEBUG ((LM_DEBUG, "(%t) handle_close\n"));
 108   return 0;
 109 }
 110
 111 int
 112 Test_Task::handle_input (ACE_HANDLE)
 113 {
 114   ACE_DEBUG ((LM_DEBUG, "(%t) handle_input\n"));
 115
 116   this->handled_++;
 117
 118   if (this->handled_ == NUM_INVOCATIONS)
 119     {
 120       done_count--;
 121       ACE_DEBUG ((LM_DEBUG,
 122                   "(%t) handle_input, handled_ = %d, done_count = %d\n",
 123                   this->handled_, done_count.value ()));
 124     }
 125
 126   ACE_OS::thr_yield ();
 127   return -1;
 128 }
 129
 130 static void *
 131 worker (void *args)
 132 {
 133   ACE_Reactor *reactor = (ACE_Reactor *) args;
 134
 135   reactor->owner (ACE_Thread::self ());
 136
 137   ACE_Time_Value timeout (4);
 138
 139   for (;;)
 140     {
 141       //ACE_DEBUG ((LM_DEBUG, "(%t) calling handle_events\n"));
 142
 143       switch (reactor->handle_events (timeout))
 144         {
 145         case -1:
 146           ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "reactor"), 0);
 147           /* NOTREACHED */
 148         case 0:
 149           ACE_ERROR_RETURN ((LM_ERROR, "(%t) timeout\n"), 0);
 150           /* NOTREACHED */
 151         }
 152
 153       // ACE_DEBUG ((LM_DEBUG, "(%t) done with handle_events\n"));
 154     }
 155
 156   ACE_NOTREACHED(return 0);
 157 }
 158
 159 int
 160 ACE_TMAIN (int, ACE_TCHAR *[])
 161 {
 162   ACE_Reactor *react1 = ACE_Reactor::instance ();
 163   ACE_Reactor *react2 = new ACE_Reactor ();
 164   Test_Task tt1[MAX_TASKS];
 165   Test_Task tt2[MAX_TASKS];
 166
 167   for (int i = 0; i < MAX_TASKS; i++)
 168     {
 169       tt1[i].open (react1);
 170       tt2[i].open (react2);
 171     }
 172
 173   if (ACE_Thread_Manager::instance ()->spawn
 174       (ACE_THR_FUNC (worker), (void *) react1, THR_NEW_LWP) == -1)
 175     ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), -1);
 176
 177   else if (ACE_Thread_Manager::instance ()->spawn
 178       (ACE_THR_FUNC (worker), (void *) react2, THR_NEW_LWP) == -1)
 179     ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), -1);
 180
 181   ACE_Thread_Manager::instance ()->wait ();
 182   ACE_DEBUG ((LM_DEBUG, "(%t) done\n"));
 183
 184   return 0;
 185 }
 186
 187 #else
 188 int
 189 ACE_TMAIN (int, ACE_TCHAR *[])
 190 {
 191   ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n"));
 192   return 0;
 193 }
 194 #endif /* ACE_HAS_THREADS */