ACE/examples/Reactor/Proactor/post_completions.cpp

   1 //=============================================================================
   2 /**
   3  *  @file     post_completions.cpp
   4  *
   5  *  This program demonstrates how to post fake completions to The
   6  *  Proactor. It also shows the how to specify the particular
   7  *  real-time signals to post completions. The Real-time signal
   8  *  based completion strategy is implemented with
   9  *  ACE_POSIX_SIG_PROACTOR.
  10  *  (So, it can be used only if both ACE_HAS_AIO_CALLS and
  11  *  ACE_HAS_POSIX_REALTIME_SIGNALS are defined.)
  12  *  Since it is faking results, you have to pay by knowing and
  13  *  using platform-specific implementation objects for Asynchronous
  14  *  Result classes.
  15  *  This example shows using an arbitrary result class for faking
  16  *  completions. You can also use the predefined Result classes for
  17  *  faking. The factory methods in the Proactor class create the
  18  *  Result objects.
  19  *  make
  20  *  ./post_completions
  21  *
  22  *  @author  Alexander Babu Arulanthu <alex@cs.wustl.edu>
  23  */
  24 //=============================================================================
  25
  26
  27 #include "ace/OS_NS_unistd.h"
  28 #include "ace/OS_main.h"
  29 #include "ace/Proactor.h"
  30 #include "ace/Task.h"
  31 #include "ace/WIN32_Proactor.h"
  32 #include "ace/POSIX_Proactor.h"
  33 #include "ace/Atomic_Op.h"
  34 #include "ace/Thread_Mutex.h"
  35
  36 // Keep track of how many completions are still expected.
  37 static ACE_Atomic_Op <ACE_SYNCH_MUTEX, size_t> Completions_To_Go;
  38
  39
  40 #if defined (ACE_HAS_WIN32_OVERLAPPED_IO) || defined (ACE_HAS_AIO_CALLS)
  41 // This only works on Win32 platforms and on Unix platforms supporting
  42 // POSIX aio calls.
  43
  44 #if defined (ACE_HAS_AIO_CALLS)
  45 #define RESULT_CLASS ACE_POSIX_Asynch_Result
  46 #elif defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  47 #define RESULT_CLASS ACE_WIN32_Asynch_Result
  48 #endif /* ACE_HAS_AIO_CALLS */
  49
  50 /**
  51  * @class My_Result
  52  *
  53  * @brief Result Object that we will post to the Proactor.
  54  */
  55 class My_Result : public RESULT_CLASS
  56 {
  57 public:
  58   My_Result (ACE_Handler &handler,
  59              const void *act,
  60              int signal_number,
  61              size_t sequence_number)
  62     : RESULT_CLASS (handler.proxy (),
  63                     act,
  64                     ACE_INVALID_HANDLE,
  65                     0, // Offset
  66                     0, // OffsetHigh
  67                     0, // Priority
  68                     signal_number),
  69       sequence_number_ (sequence_number)
  70     {}
  71   // Constructor.
  72
  73   virtual ~My_Result ()
  74     {}
  75   // Destructor.
  76
  77     /**
  78      * This is the method that will be called by the Proactor for
  79      * dispatching the completion. This method generally calls one of
  80      * the call back hood methods defined in the ACE_Handler
  81      * class. But, we will just handle the completions here.
  82      */
  83   void complete (size_t,
  84                  int success,
  85                  const void *completion_key,
  86                  u_long error)
  87     {
  88       this->success_ = success;
  89       this->completion_key_ = completion_key;
  90       this->error_ = error;
  91
  92       size_t to_go = --Completions_To_Go;
  93
  94       // Print the completion details.
  95       ACE_DEBUG ((LM_DEBUG,
  96                   "(%t) Completion sequence number %d, success : %d, error : %d, signal_number : %d, %u more to go\n",
  97                   this->sequence_number_,
  98                   this->success_,
  99                   this->error_,
 100                   this->signal_number (),
 101                   to_go));
 102
 103       // Sleep for a while.
 104       ACE_OS::sleep (4);
 105     }
 106
 107 private:
 108   /// Sequence number for the result object.
 109   size_t sequence_number_;
 110 };
 111
 112 /**
 113  * @class My_Handler
 114  *
 115  * @brief Handler class for faked completions.
 116  */
 117 class My_Handler : public  ACE_Handler
 118 {
 119 public:
 120   /// Constructor.
 121   My_Handler () {}
 122
 123   /// Destructor.
 124   virtual ~My_Handler () {}
 125 };
 126
 127 /**
 128  * @class My_Task:
 129  *
 130  * @brief Contains thread functions which execute event loops. Each
 131  * thread waits for a different signal.
 132  */
 133 class My_Task: public ACE_Task <ACE_NULL_SYNCH>
 134 {
 135 public:
 136   /// Constructor.
 137   My_Task () {}
 138
 139   /// Destructor.
 140   virtual ~My_Task () {}
 141
 142   //FUZZ: disable check_for_lack_ACE_OS
 143   int open (void *proactor)
 144     {
 145   //FUZZ: enable check_for_lack_ACE_OS
 146       // Store the proactor.
 147       this->proactor_ = (ACE_Proactor *) proactor;
 148
 149       // Activate the Task.
 150       this->activate (THR_NEW_LWP, 5);
 151       return 0;
 152     }
 153
 154   int svc ()
 155     {
 156       // Handle events for 13 seconds.
 157       ACE_Time_Value run_time (13);
 158
 159       ACE_DEBUG ((LM_DEBUG, "(%t):Starting svc routine\n"));
 160
 161       if (this->proactor_->handle_events (run_time) == -1)
 162         ACE_ERROR_RETURN ((LM_ERROR, "(%t):%p.\n", "Worker::svc"), -1);
 163
 164       ACE_DEBUG ((LM_DEBUG, "(%t) work complete\n"));
 165
 166       return 0;
 167     }
 168
 169 private:
 170   /// Proactor for this task.
 171   ACE_Proactor *proactor_;
 172 };
 173
 174 int
 175 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 176 {
 177   ACE_UNUSED_ARG (argc);
 178   ACE_UNUSED_ARG (argv);
 179
 180   ACE_DEBUG ((LM_DEBUG,
 181               "(%P | %t):Test starts\n"));
 182
 183   // = Get two POSIX_SIG_Proactors, one with SIGRTMIN and one with
 184   //   SIGRTMAX.
 185
 186   ACE_Proactor  proactor1;
 187   // Proactor1. SIGRTMIN Proactor. (default).
 188
 189   // = Proactor2. SIGRTMAX Proactor.
 190 #if defined (ACE_HAS_AIO_CALLS) && defined (ACE_HAS_POSIX_REALTIME_SIGNALS)
 191
 192   ACE_DEBUG ((LM_DEBUG, "Using ACE_POSIX_SIG_Proactor\n"));
 193
 194   sigset_t signal_set;
 195   // Signal set that we want to mask.
 196
 197   // Clear the signal set.
 198   if (ACE_OS::sigemptyset (&signal_set) == -1)
 199     ACE_ERROR_RETURN ((LM_ERROR,
 200                        "Error:%p\n",
 201                        "sigemptyset failed"),
 202                       1);
 203
 204   // Add the SIGRTMAX to the signal set.
 205   if (ACE_OS::sigaddset (&signal_set, ACE_SIGRTMAX) == -1)
 206     ACE_ERROR_RETURN ((LM_ERROR,
 207                        "Error:%p\n",
 208                        "sigaddset failed"),
 209                       1);
 210
 211   // Make the POSIX Proactor.
 212   ACE_POSIX_SIG_Proactor posix_proactor (signal_set);
 213   // Get the Proactor interface out of it.
 214   ACE_Proactor proactor2 (&posix_proactor);
 215 #else /* ACE_HAS_AIO_CALLS && ACE_HAS_POSIX_REALTIME_SIGNALS */
 216   ACE_Proactor proactor2;
 217 #endif /* ACE_HAS_AIO_CALLS && ACE_HAS_POSIX_REALTIME_SIGNALS */
 218
 219   // = Create Tasks. One pool of threads to handle completions on
 220   //   SIGRTMIN and the other one to handle completions on SIGRTMAX.
 221   My_Task task1, task2;
 222   task1.open (&proactor1);
 223   task2.open (&proactor2);
 224
 225   // Handler for completions.
 226   My_Handler handler;
 227
 228   // = Create a few MyResult objects and post them to Proactor.
 229   const size_t NrCompletions (10);
 230   My_Result *result_objects [NrCompletions];
 231   int signal_number = ACE_SIGRTMAX;
 232   size_t ri = 0;
 233
 234   Completions_To_Go = NrCompletions;
 235
 236   // Creation.
 237   for (ri = 0; ri < NrCompletions; ri++)
 238     {
 239       // Use RTMIN and RTMAX proactor alternatively, to post
 240       // completions.
 241       if (ri % 2)
 242         signal_number = ACE_SIGRTMIN;
 243       else
 244         signal_number = ACE_SIGRTMAX;
 245       // Create the result.
 246       ACE_NEW_RETURN (result_objects [ri],
 247                       My_Result (handler,
 248                                  0,
 249                                  signal_number,
 250                                  ri),
 251                       1);
 252     }
 253   ACE_OS::sleep(5);
 254   // Post all the result objects.
 255   ACE_Proactor *proactor;
 256   for (ri = 0; ri < NrCompletions; ri++)
 257     {
 258       // Use RTMIN and RTMAX Proactor alternatively, to post
 259       // completions.
 260       if (ri % 2)
 261         proactor = &proactor1;
 262       else
 263         proactor = &proactor2;
 264       if (result_objects [ri]->post_completion (proactor->implementation ())
 265           == -1)
 266         ACE_ERROR_RETURN ((LM_ERROR,
 267                            "Test failed\n"),
 268                           1);
 269     }
 270
 271   ACE_Thread_Manager::instance ()->wait ();
 272
 273   int status = 0;
 274   size_t to_go = Completions_To_Go.value ();
 275   if (size_t (0) != to_go)
 276     {
 277       ACE_ERROR ((LM_ERROR,
 278                   "Fail! Expected all completions to finish but %u to go\n",
 279                   to_go));
 280       status = 1;
 281     }
 282
 283   ACE_DEBUG ((LM_DEBUG,
 284               "(%P | %t):Test ends\n"));
 285   return status;
 286 }
 287
 288 #else /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */
 289
 290 int
 291 ACE_TMAIN (int, ACE_TCHAR *[])
 292 {
 293   ACE_DEBUG ((LM_DEBUG,
 294               "This example cannot work with AIOCB_Proactor.\n"));
 295   return 1;
 296 }
 297
 298 #endif /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */
 299