ACE/tests/Message_Queue_Notifications_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Message_Queue_Notifications_Test.cpp
   5  *
   6  *    There are two tests that test 2 different notification
   7  *    mechanisms in Message Queue.
   8  *
   9  *    The first test illustrates the notification mechanisms in
  10  *    Message_Queue and its integration with Reactor.
  11  *
  12  *    Note the following things about this part of the test:
  13  *
  14  *    1. Multiple threads are not required.
  15  *    2. You do not have to explicitly notify the Reactor
  16  *    3. This code will work the same with any Reactor Implementation
  17  *    4. handle_input, handle_exception, handle_output are the only
  18  *       callbacks supported by this mechanism
  19  *    5. The notification mechanism need not notify the Reactor. You can
  20  *       write your own strategy classes that can do whatever application
  21  *       specific behavior you want.
  22  *
  23  *    The second test also makes sure the high/low water mark
  24  *    signaling mechanism works flawlessly.
  25  *
  26  *  @author Irfan Pyarali <irfan@cs.wustl.edu> and Nanbor Wang <nanbor@cs.wustl.edu>
  27  */
  28 //=============================================================================
  29
  30
  31 #include "test_config.h"
  32 #include "ace/Reactor.h"
  33 #include "ace/Task.h"
  34 #include "ace/Reactor_Notification_Strategy.h"
  35 #include "ace/Atomic_Op.h"
  36 #include "ace/Barrier.h"
  37 #include "ace/Synch_Traits.h"
  38 #include "ace/Null_Condition.h"
  39 #include "ace/Null_Mutex.h"
  40 #include "ace/OS_NS_string.h"
  41 #include "ace/OS_NS_unistd.h"
  42
  43
  44
  45 static int iterations = 10;
  46
  47 static const size_t worker_threads = 2;
  48 static const char * default_message = "ACE RULES";
  49 static const size_t default_high_water_mark = 20;
  50 static const size_t default_low_water_mark = 10;
  51 static const int watermark_iterations = 2 * default_high_water_mark;
  52
  53 /**
  54  * @class Message_Handler
  55  *
  56  * @brief This class implements a notification strategy for the Reactor.
  57  */
  58 class Message_Handler : public ACE_Task<ACE_NULL_SYNCH>
  59 {
  60 public:
  61   /// Constructor.
  62   Message_Handler (ACE_Reactor &reactor);
  63
  64   // = Demuxing hooks.
  65   int handle_input (ACE_HANDLE) override;
  66   int handle_output (ACE_HANDLE fd = ACE_INVALID_HANDLE) override;
  67   int handle_exception (ACE_HANDLE fd = ACE_INVALID_HANDLE) override;
  68
  69 private:
  70   int process_message ();
  71   void make_message ();
  72
  73   ACE_Reactor_Notification_Strategy notification_strategy_;
  74 };
  75
  76 /**
  77  * @class Watermark_Test
  78  *
  79  * @brief This class test the correct functioning of build-in flow
  80  * control machanism in ACE_Task.
  81  */
  82 class Watermark_Test : public ACE_Task<ACE_SYNCH>
  83 {
  84 public:
  85   Watermark_Test ();
  86
  87   int svc () override;
  88
  89   int consumer ();
  90   int producer ();
  91   int put_message (ACE_Time_Value* timeout = 0);
  92   int get_message ();
  93   void print_producer_debug_message ();
  94
  95 private:
  96   const size_t len_;
  97   const size_t hwm_;
  98   const size_t lwm_;
  99   ACE_Atomic_Op <ACE_SYNCH_MUTEX, int> role_;
 100 #if defined (ACE_HAS_THREADS)
 101   ACE_Barrier mq_full_;
 102   ACE_Barrier mq_low_water_mark_hit_;
 103 #endif /* ACE_HAS_THREADS */
 104 };
 105
 106 Message_Handler::Message_Handler (ACE_Reactor &reactor)
 107   // First time handle_input will be called
 108   : notification_strategy_ (&reactor,
 109                             this,
 110                             ACE_Event_Handler::READ_MASK)
 111 {
 112   this->msg_queue ()->notification_strategy (&this->notification_strategy_);
 113   this->make_message ();
 114 }
 115
 116 int
 117 Message_Handler::handle_input (ACE_HANDLE)
 118 {
 119   ACE_DEBUG ((LM_DEBUG,
 120               ACE_TEXT ("Message_Handler::handle_input\n")));
 121
 122   // Next time handle_output will be called.
 123   this->notification_strategy_.mask (ACE_Event_Handler::WRITE_MASK);
 124
 125   return process_message ();
 126 }
 127
 128 int
 129 Message_Handler::handle_output (ACE_HANDLE fd)
 130 {
 131   ACE_DEBUG ((LM_DEBUG,
 132               ACE_TEXT ("Message_Handler::handle_output\n")));
 133   ACE_UNUSED_ARG (fd);
 134
 135   // Next time handle_exception will be called.
 136   this->notification_strategy_.mask (ACE_Event_Handler::EXCEPT_MASK);
 137
 138   return process_message ();
 139 }
 140
 141 int
 142 Message_Handler::handle_exception (ACE_HANDLE fd)
 143 {
 144   ACE_DEBUG ((LM_DEBUG,
 145               ACE_TEXT ("Message_Handler::handle_exception\n")));
 146   ACE_UNUSED_ARG (fd);
 147
 148   // Next time handle_input will be called.
 149   this->notification_strategy_.mask (ACE_Event_Handler::READ_MASK);
 150
 151   return this->process_message ();
 152 }
 153
 154 int
 155 Message_Handler::process_message ()
 156 {
 157   ACE_Message_Block *mb = 0;
 158
 159   if (this->getq (mb,
 160                   (ACE_Time_Value *) &ACE_Time_Value::zero) == -1)
 161     ACE_ERROR_RETURN ((LM_ERROR,
 162                        ACE_TEXT ("%p\n"),
 163                        ACE_TEXT ("dequeue_head")),
 164                       -1);
 165   else
 166     {
 167       ACE_DEBUG ((LM_DEBUG,
 168                   ACE_TEXT ("message received = %s\n"),
 169                   mb->rd_ptr ()));
 170       mb->release ();
 171     }
 172
 173   this->make_message ();
 174   return 0;
 175 }
 176
 177 void
 178 Message_Handler::make_message ()
 179 {
 180   if (--iterations > 0)
 181     {
 182       ACE_Message_Block *mb = 0;
 183       ACE_NEW (mb,
 184                ACE_Message_Block ((char *) ACE_TEXT ("hello")));
 185
 186       ACE_DEBUG ((LM_DEBUG,
 187                   ACE_TEXT ("sending message\n")));
 188       this->putq (mb);
 189     }
 190 }
 191
 192 Watermark_Test::Watermark_Test ()
 193   : len_ (ACE_OS::strlen (default_message) + 1),
 194     hwm_ (this->len_ * default_high_water_mark),
 195     lwm_ (this->len_ * default_low_water_mark),
 196     role_ (0)
 197 #if defined (ACE_HAS_THREADS)
 198     , mq_full_ (worker_threads),
 199     mq_low_water_mark_hit_ (worker_threads)
 200 #endif /* ACE_HAS_THREADS */
 201 {
 202   this->water_marks (ACE_IO_Cntl_Msg::SET_LWM,
 203                      this->lwm_);
 204   this->water_marks (ACE_IO_Cntl_Msg::SET_HWM,
 205                      this->hwm_);
 206 }
 207
 208 int
 209 Watermark_Test::producer ()
 210 {
 211   int i = watermark_iterations;
 212
 213   for (ssize_t hwm = this->hwm_;
 214        hwm >= 0 ;
 215        hwm -= this->len_)
 216     {
 217       this->put_message ();
 218       this->print_producer_debug_message ();
 219       i--;
 220       if (this->msg_queue ()->is_full ())
 221         break;
 222
 223     }
 224   ACE_DEBUG ((LM_DEBUG,
 225               ACE_TEXT ("(%P|%t) Producer: High water mark hit ----\n")));
 226
 227   ACE_MT (this->mq_full_.wait ());
 228
 229   // The following put_message should block until the message queue
 230   // has dropped under the lwm.
 231   this->put_message ();
 232
 233   ACE_TEST_ASSERT (this->msg_queue ()-> message_bytes () <= this->lwm_ + this->len_);
 234
 235   this->print_producer_debug_message ();
 236
 237   for (i--; i >= 0 ; i--)
 238     {
 239       this->put_message ();
 240       this->print_producer_debug_message ();
 241     }
 242
 243   return 0;
 244 }
 245
 246 int
 247 Watermark_Test::consumer ()
 248 {
 249   ACE_MT (this->mq_full_.wait ());
 250
 251   ACE_OS::sleep (1);
 252
 253   // Let producer proceed and block in putq.
 254
 255   for (int i = watermark_iterations; i >= 0; i--)
 256     {
 257       this->get_message ();
 258       ACE_OS::sleep (0);
 259     }
 260
 261   return 0;
 262 }
 263
 264 int
 265 Watermark_Test::get_message ()
 266 {
 267   ACE_Message_Block *mb = 0;
 268
 269   if (this->getq (mb) == -1)
 270     ACE_ERROR_RETURN ((LM_ERROR,
 271                        ACE_TEXT ("%p\n"),
 272                        ACE_TEXT ("dequeue_head")),
 273                       -1);
 274   else
 275     {
 276       ACE_DEBUG ((LM_DEBUG,
 277                   ACE_TEXT ("(%P|%t) Consumer: message size = %3d, ")
 278                   ACE_TEXT ("message count = %3d\n"),
 279                   this->msg_queue ()-> message_bytes (),
 280                   this->msg_queue ()-> message_count ()));
 281       mb->release ();
 282     }
 283
 284   return 0;
 285 }
 286
 287 int
 288 Watermark_Test::put_message (ACE_Time_Value *timeout)
 289 {
 290   ACE_Message_Block *mb = 0;
 291
 292   ACE_NEW_RETURN (mb,
 293                   ACE_Message_Block (default_message,
 294                                      this->len_),
 295                   -1);
 296
 297   return this->putq (mb, timeout);
 298 }
 299
 300 void
 301 Watermark_Test::print_producer_debug_message ()
 302 {
 303   ACE_DEBUG ((LM_DEBUG,
 304               ACE_TEXT ("(%P|%t) Producer: message size = %3d, ")
 305               ACE_TEXT ("message count = %3d\n"),
 306               this->msg_queue ()-> message_bytes (),
 307               this->msg_queue ()-> message_count ()));
 308 }
 309
 310 int
 311 Watermark_Test::svc ()
 312 {
 313   // this->role_ is an Atomic_Op object.
 314   int role = this->role_++;
 315
 316   switch (role)
 317     {
 318     case 0:
 319       this->producer ();
 320       break;
 321     case 1:
 322       this->consumer ();
 323       break;
 324     default:
 325       break;
 326     }
 327   return 0;
 328 }
 329
 330 int
 331 run_main (int, ACE_TCHAR *[])
 332 {
 333   ACE_START_TEST (ACE_TEXT ("Message_Queue_Notifications_Test"));
 334
 335   ACE_DEBUG ((LM_DEBUG,
 336               ACE_TEXT ("Starting message queue reactive notification test...\n")));
 337
 338   ACE_Reactor reactor;
 339   Message_Handler mh (reactor);
 340
 341   while (iterations > 0)
 342     reactor.handle_events ();
 343
 344 #if defined (ACE_HAS_THREADS)
 345   ACE_DEBUG ((LM_DEBUG,
 346               ACE_TEXT ("Starting message queue watermark test...\n")));
 347   Watermark_Test watermark_test;
 348   ACE_DEBUG ((LM_DEBUG,
 349               ACE_TEXT ("High water mark is %d\n")
 350               ACE_TEXT ("Low water mark is %d\n"),
 351               default_high_water_mark,
 352               default_low_water_mark));
 353
 354   watermark_test.activate (THR_NEW_LWP,
 355                            worker_threads);
 356
 357   ACE_Thread_Manager::instance ()->wait ();
 358 #else
 359   ACE_DEBUG ((LM_INFO,
 360               ACE_TEXT ("Message queue watermark test not performed because threads are not supported\n")));
 361 #endif /* ACE_HAS_THREADS */
 362
 363   ACE_END_TEST;
 364   return 0;
 365 }
 366