ACE/tests/Message_Queue_Test_Ex.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Message_Queue_Test_Ex.cpp
   5  *
   6  *    This is:
   7  *    1. A simple test of the ACE_Message_Queue_Ex that executes
   8  *       a performance measurement test for both single-threaded
   9  *       (null synch) and thread-safe ACE_Message_Queue_Ex
  10  *       instantiations.
  11  *    2. An example of using a user-defined class to parameterize
  12  *       ACE_Message_Queue_Ex.
  13  *
  14  *  @author Michael Vitlo <mvitalo@sprynet.com>
  15  *  @author copied the code from: Irfan Pyarali <irfan@cs.wustl.edu> and David L. Levine <levine@cs.wustl.edu>
  16  */
  17 //=============================================================================
  18
  19
  20 #include "test_config.h"
  21 #include "ace/Thread_Manager.h"
  22
  23 #include <memory>
  24 #include "ace/Message_Queue.h"
  25 #include "ace/Synch_Traits.h"
  26 #include "ace/Null_Mutex.h"
  27 #include "ace/Null_Condition.h"
  28 #include "ace/High_Res_Timer.h"
  29 #include "ace/Message_Block.h"
  30 #include "ace/OS_NS_sys_time.h"
  31 #include "ace/Barrier.h"
  32 #include "Message_Queue_Test_Ex.h"    // Declares User_Class
  33
  34 const ACE_TCHAR usage[] =
  35   ACE_TEXT ("usage: Message_Queue_Test_Ex <number of messages>\n");
  36
  37 using QUEUE = ACE_Message_Queue_Ex<User_Class, ACE_NULL_SYNCH>;
  38
  39 static const int MAX_MESSAGES = 10000;
  40 static const char test_message[] = "ACE_Message_Queue_Ex Test Message";
  41
  42 static int max_messages = MAX_MESSAGES;
  43 static int chain_limit = 4;
  44 static ACE_Barrier tester_barrier (2);
  45
  46 // Dynamically allocate to avoid a static.
  47 static ACE_High_Res_Timer *timer = 0;
  48
  49 // Helper printing function
  50 static void
  51 print_message (const ACE_TCHAR *message)
  52 {
  53   ACE_Time_Value tv;
  54   timer->elapsed_time (tv);
  55   ACE_DEBUG ((LM_INFO,
  56               ACE_TEXT ("%s: %u messages took %u msec (%f msec/message)\n"),
  57               message,
  58               max_messages,
  59               tv.msec (),
  60               (double) tv.msec () / max_messages));
  61 }
  62
  63 #if defined (ACE_HAS_THREADS)
  64 using SYNCH_QUEUE = ACE_Message_Queue_Ex<User_Class, ACE_MT_SYNCH>;
  65
  66 /**
  67  * Container for data passed to sender and receiver in
  68  * performance test.
  69  *
  70  * For use in multithreaded performance test.
  71  */
  72 struct Queue_Wrapper
  73 {
  74   /// The message queue.
  75   SYNCH_QUEUE *q_ {};
  76
  77   /// Pointer to messages blocks for sender to send to reciever.
  78   User_Class **send_block_ {};
  79
  80   /// Default constructor.
  81   Queue_Wrapper () = default;
  82 };
  83
  84 /**
  85  * Container for data passed to sender in the MQ_Ex_N_Tester
  86  * performance test.
  87  *
  88  * For use in multithreaded performance test.
  89  */
  90 struct MQ_Ex_N_Tester_Wrapper
  91 {
  92   MQ_Ex_N_Tester *tester_ {};
  93   User_Class     *head_send_block_ {};
  94 };
  95
  96 #endif /* ACE_HAS_THREADS */
  97
  98 // Encapsulates the sent messages creation and destruction
  99 struct Send_Messages
 100 {
 101   Send_Messages (int number_of_messages, int chain_limit):
 102     send_block_ (0),
 103     number_of_messages_ (number_of_messages),
 104     chain_limit_ (chain_limit)
 105   {
 106   }
 107
 108   int create_messages (const char test_message[])
 109   {
 110     int limit = this->number_of_messages_ / this->chain_limit_;
 111     ACE_NEW_RETURN (this->send_block_,
 112                     User_Class *[limit],
 113                     -1);
 114
 115     int i, j;
 116     for (i = 0; i < limit; ++i)
 117       {
 118         User_Class *&temp1 = this->send_block_[i];
 119         ACE_NEW_RETURN (temp1,
 120                         User_Class (test_message),
 121                         -1);
 122         User_Class *tail = temp1;
 123         for (j = 1; j < this->chain_limit_; ++j)
 124           {
 125             User_Class *temp2 = 0;
 126             ACE_NEW_RETURN (temp2,
 127                             User_Class (test_message),
 128                             -1);
 129             tail->next (temp2);
 130             tail = temp2;
 131           }
 132       }
 133     this->head_send_block_ = this->send_block_[0];
 134     return 0;
 135   }
 136
 137   ~Send_Messages ()
 138   {
 139     int j, i = 0;
 140     int limit = this->number_of_messages_ / this->chain_limit_;
 141     for (; i < limit; ++i)
 142       {
 143         User_Class *&temp1 = this->send_block_[i];
 144         for (j = 0; j < this->chain_limit_; ++j)
 145           {
 146             User_Class *temp2 = temp1->next ();
 147             delete temp1;
 148             temp1 = temp2;
 149           }
 150       }
 151     delete [] this->send_block_;
 152   }
 153
 154   User_Class *  head_send_block_;
 155   User_Class ** send_block_;
 156   int           number_of_messages_;
 157   int           chain_limit_;
 158 };
 159
 160 // Encapsulates the received messages creation and destruction
 161 struct Receive_Messages
 162 {
 163   Receive_Messages (int number_of_messages) :
 164     receive_block_ (0),
 165     number_of_messages_ (number_of_messages)
 166   {
 167   }
 168
 169   int create ()
 170   {
 171     ACE_NEW_RETURN (this->receive_block_,
 172                     User_Class *[this->number_of_messages_],
 173                     -1);
 174     return 0;
 175   }
 176
 177   ~Receive_Messages ()
 178   {
 179     delete [] this->receive_block_;
 180   }
 181
 182   User_Class **receive_block_;
 183   int          number_of_messages_;
 184 };
 185
 186 static int
 187 single_thread_performance_test ()
 188 {
 189   const char test_message[] =
 190     "ACE_Message_Queue_Ex Test Message";
 191   const ACE_TCHAR *message =
 192     ACE_TEXT ("ACE_Message_Queue_Ex<ACE_NULL_SYNCH>, single thread");
 193
 194   // Create a message queue.
 195   QUEUE *msgq = 0;
 196
 197   ACE_NEW_RETURN (msgq,
 198                   QUEUE,
 199                   -1);
 200
 201   // Create the messages.  Allocate off the heap in case messages is
 202   // large relative to the amount of stack space available.
 203   User_Class **send_block = 0;
 204   ACE_NEW_RETURN (send_block,
 205                   User_Class *[max_messages],
 206                   -1);
 207
 208   int i = 0;
 209
 210   for (i = 0; i < max_messages; ++i)
 211     ACE_NEW_RETURN (send_block[i],
 212                     User_Class (test_message),
 213                     -1);
 214
 215   User_Class **receive_block_p = 0;
 216   ACE_NEW_RETURN (receive_block_p,
 217                   User_Class *[max_messages],
 218                   -1);
 219
 220   timer->start ();
 221
 222   // Send/receive the messages.
 223   for (i = 0; i < max_messages; ++i)
 224     {
 225       if (msgq->enqueue_tail (send_block[i]) == -1)
 226         ACE_ERROR_RETURN ((LM_ERROR,
 227                            ACE_TEXT ("%p\n"),
 228                            ACE_TEXT ("enqueue")),
 229                           -1);
 230
 231       if (msgq->dequeue_head (receive_block_p[i]) == -1)
 232         ACE_ERROR_RETURN ((LM_ERROR,
 233                            ACE_TEXT ("%p\n"),
 234                            ACE_TEXT ("dequeue_head")),
 235                           -1);
 236     }
 237
 238   timer->stop ();
 239   print_message (message);
 240   timer->reset ();
 241
 242   delete [] receive_block_p;
 243
 244   for (i = 0; i < max_messages; ++i)
 245     delete send_block[i];
 246   delete [] send_block;
 247   delete msgq;
 248
 249   return 0;
 250 }
 251
 252 int
 253 MQ_Ex_N_Tester::single_thread_performance_test ()
 254 {
 255   // Create the messages.  Allocate off the heap in case messages is
 256   // large relative to the amount of stack space available.
 257
 258   if ((0 != this->test_enqueue_tail ()) ||
 259       (0 != this->test_enqueue_head ())    )
 260     {
 261       return -1;
 262     }
 263
 264   return 0;
 265 }
 266
 267 int
 268 MQ_Ex_N_Tester::test_enqueue_tail ()
 269 {
 270   const ACE_TCHAR *message =
 271     ACE_TEXT ("ACE_Message_Queue_Ex_N<ACE_NULL_SYNCH>, test_enqueue_tail");
 272
 273   // Send_Messages creates messages and deletes them when it gets out of scope
 274   Send_Messages messages (max_messages, chain_limit);
 275   if (-1 == messages.create_messages (test_message))
 276     {
 277       return -1;
 278     }
 279   Receive_Messages r_messages (max_messages);
 280   if (-1 == r_messages.create ())
 281     {
 282       return -1;
 283     }
 284
 285   // prepare
 286   int limit = max_messages / chain_limit;
 287   timer->start ();
 288   // Send with just one call
 289   for (int i = 0; i < limit; ++i)
 290     {
 291       if (-1 == this->st_queue_.enqueue_tail (messages.send_block_[i]))
 292         {
 293           ACE_ERROR_RETURN ((LM_ERROR,
 294                              ACE_TEXT ("%p\n"),
 295                              ACE_TEXT ("enqueue_tail_n")),
 296                             -1);
 297         }
 298
 299       for (int j = 0, k = 0; j < chain_limit; ++j, ++k)
 300         {
 301           if (this->st_queue_.dequeue_head (r_messages.receive_block_[k]) == -1)
 302             {
 303               ACE_ERROR_RETURN ((LM_ERROR,
 304                                  ACE_TEXT ("%p\n"),
 305                                  ACE_TEXT ("dequeue_head")),
 306                                 -1);
 307             }
 308         }
 309     }
 310   timer->stop ();
 311
 312   print_message (message);
 313
 314   timer->reset ();
 315
 316   return 0;
 317 }
 318
 319 int
 320 MQ_Ex_N_Tester::test_enqueue_head ()
 321 {
 322   const ACE_TCHAR *message =
 323     ACE_TEXT ("ACE_Message_Queue_Ex_N<ACE_NULL_SYNCH>, test_enqueue_head");
 324
 325   // Send_Messages creates messages and deletes them when it gets out of scope
 326   Send_Messages messages (max_messages, chain_limit);
 327   if (-1 == messages.create_messages (test_message))
 328     {
 329       return -1;
 330     }
 331   Receive_Messages r_messages (max_messages);
 332   if (-1 == r_messages.create ())
 333     {
 334       return -1;
 335     }
 336
 337   // prepare
 338   int i, j, k = 0;
 339
 340   int limit = max_messages / chain_limit;
 341   timer->start ();
 342
 343   // Send/receive the messages.
 344   // Send with just one call
 345   for (i = 0; i < limit; ++i)
 346     {
 347       if (-1 == this->st_queue_.enqueue_head (messages.send_block_[i]))
 348         {
 349           ACE_ERROR_RETURN ((LM_ERROR,
 350                              ACE_TEXT ("%p\n"),
 351                              ACE_TEXT ("enqueue_tail_n")),
 352                             -1);
 353         }
 354
 355       for (j = 0; j < chain_limit; ++j, ++k)
 356         {
 357           if (this->st_queue_.dequeue_head (r_messages.receive_block_[k]) == -1)
 358             {
 359               ACE_ERROR_RETURN ((LM_ERROR,
 360                                  ACE_TEXT ("%p\n"),
 361                                  ACE_TEXT ("dequeue_head")),
 362                                 -1);
 363             }
 364         }
 365     }
 366   timer->stop ();
 367
 368   print_message (message);
 369
 370   timer->reset ();
 371
 372   return 0;
 373 }
 374
 375 #if defined (ACE_HAS_THREADS)
 376
 377 static void *
 378 receiver (void *arg)
 379 {
 380   Queue_Wrapper *queue_wrapper = reinterpret_cast<Queue_Wrapper *> (arg);
 381   int i;
 382
 383   User_Class **receive_block_p = 0;
 384   ACE_NEW_RETURN (receive_block_p,
 385                   User_Class *[max_messages],
 386                   (void *) -1);
 387
 388   for (i = 0; i < max_messages; ++i)
 389     if (queue_wrapper->q_->dequeue_head (receive_block_p[i]) == -1)
 390       ACE_ERROR_RETURN ((LM_ERROR,
 391                          ACE_TEXT ("%p\n"),
 392                          ACE_TEXT ("dequeue_head")),
 393                         0);
 394   timer->stop ();
 395
 396   delete [] receive_block_p;
 397
 398   return 0;
 399 }
 400
 401 static void *
 402 sender (void *arg)
 403 {
 404   Queue_Wrapper *queue_wrapper = reinterpret_cast<Queue_Wrapper *> (arg);
 405   int i;
 406
 407   timer->start ();
 408
 409   // Send the messages.
 410   for (i = 0; i < max_messages; ++i)
 411     if (queue_wrapper->q_->
 412         enqueue_tail (queue_wrapper->send_block_[i]) == -1)
 413       ACE_ERROR_RETURN ((LM_ERROR,
 414                          ACE_TEXT ("%p\n"),
 415                          ACE_TEXT ("enqueue")),
 416                         0);
 417   return 0;
 418 }
 419
 420 static int
 421 performance_test ()
 422 {
 423   Queue_Wrapper queue_wrapper;
 424   const ACE_TCHAR *message =
 425     ACE_TEXT ("ACE_Message_Queue_Ex<ACE_SYNCH>");
 426   int i = 0;
 427
 428   // Create the messages.  Allocate off the heap in case messages is
 429   // large relative to the amount of stack space available.  Allocate
 430   // it here instead of in the sender, so that we can delete it after
 431   // the _receiver_ is done.
 432   User_Class **send_block = 0;
 433   ACE_NEW_RETURN (send_block,
 434                   User_Class *[max_messages],
 435                   -1);
 436
 437   for (i = 0; i < max_messages; ++i)
 438     ACE_NEW_RETURN (send_block[i],
 439                     User_Class (test_message),
 440                     -1);
 441
 442   queue_wrapper.send_block_ = send_block;
 443
 444   ACE_NEW_RETURN (queue_wrapper.q_,
 445                   SYNCH_QUEUE,
 446                   -1);
 447
 448   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) sender,
 449                                               &queue_wrapper,
 450                                               THR_BOUND) == -1)
 451     ACE_ERROR_RETURN ((LM_ERROR,
 452                        ACE_TEXT ("%p\n"),
 453                        ACE_TEXT ("spawning sender thread")),
 454                       -1);
 455
 456   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) receiver,
 457                                               &queue_wrapper,
 458                                               THR_BOUND) == -1)
 459     ACE_ERROR_RETURN ((LM_ERROR,
 460                        ACE_TEXT ("%p\n"),
 461                        ACE_TEXT ("spawning receiver thread")),
 462                       -1);
 463
 464   ACE_Thread_Manager::instance ()->wait ();
 465   print_message (message);
 466   timer->reset ();
 467
 468   delete queue_wrapper.q_;
 469   queue_wrapper.q_ = 0;
 470
 471   for (i = 0; i < max_messages; ++i)
 472     delete send_block[i];
 473   delete [] send_block;
 474
 475   return 0;
 476 }
 477
 478 int
 479 MQ_Ex_N_Tester::performance_test ()
 480 {
 481   const ACE_TCHAR *message =
 482     ACE_TEXT ("ACE_Message_Queue_Ex_N<ACE_SYNCH>");
 483
 484   Send_Messages messages (max_messages, chain_limit);
 485   if (-1 == messages.create_messages (test_message))
 486     {
 487       return -1;
 488     }
 489
 490   MQ_Ex_N_Tester_Wrapper tester_wrapper;
 491   tester_wrapper.head_send_block_ = messages.head_send_block_;
 492   tester_wrapper.tester_ = this;
 493
 494   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) &MQ_Ex_N_Tester::sender,
 495                                               &tester_wrapper,
 496                                               THR_BOUND) == -1)
 497     ACE_ERROR_RETURN ((LM_ERROR,
 498                        ACE_TEXT ("%p\n"),
 499                        ACE_TEXT ("spawning sender thread")),
 500                       -1);
 501
 502   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) &MQ_Ex_N_Tester::receiver,
 503                                               this,
 504                                               THR_BOUND) == -1)
 505     ACE_ERROR_RETURN ((LM_ERROR,
 506                        ACE_TEXT ("%p\n"),
 507                        ACE_TEXT ("spawning receiver thread")),
 508                       -1);
 509
 510   ACE_Thread_Manager::instance ()->wait ();
 511
 512   print_message (message);
 513
 514   timer->reset ();
 515
 516   return 0;
 517 }
 518
 519 ACE_THR_FUNC_RETURN
 520 MQ_Ex_N_Tester::receiver (void *args)
 521 {
 522   MQ_Ex_N_Tester *tester = reinterpret_cast<MQ_Ex_N_Tester *> (args);
 523
 524   User_Class **receive_block_p = 0;
 525   ACE_NEW_RETURN (receive_block_p,
 526                   User_Class *[max_messages],
 527                   (ACE_THR_FUNC_RETURN) -1);
 528
 529   int i;
 530   tester_barrier.wait ();
 531   for (i = 0; i < max_messages; ++i)
 532     {
 533       if (tester->mt_queue_.dequeue_head (receive_block_p[i]) == -1)
 534         {
 535           ACE_ERROR ((LM_ERROR,
 536                       ACE_TEXT ("%p\n"),
 537                       ACE_TEXT ("dequeue_head")));
 538           return (ACE_THR_FUNC_RETURN) -1;
 539         }
 540     }
 541   timer->stop ();
 542
 543   delete [] receive_block_p;
 544
 545   return 0;
 546 }
 547
 548 ACE_THR_FUNC_RETURN
 549 MQ_Ex_N_Tester::sender (void *args)
 550 {
 551   MQ_Ex_N_Tester_Wrapper *tester_wrapper =
 552     reinterpret_cast<MQ_Ex_N_Tester_Wrapper *> (args);
 553   MQ_Ex_N_Tester *tester = tester_wrapper->tester_;
 554
 555   Send_Messages messages (max_messages, chain_limit);
 556   if (-1 == messages.create_messages (test_message))
 557     {
 558       return (ACE_THR_FUNC_RETURN) -1;
 559     }
 560   int limit = max_messages / chain_limit;
 561   tester_barrier.wait ();
 562   timer->start ();
 563   // Send/receive the messages.
 564   timer->start ();
 565   // Send with just one call
 566   for (int i = 0; i < limit; ++i)
 567     {
 568       if (-1 == tester->mt_queue_.enqueue_tail (messages.send_block_[i]))
 569         {
 570           ACE_ERROR ((LM_ERROR,
 571                       ACE_TEXT ("%p\n"),
 572                       ACE_TEXT ("enqueue_tail_n")));
 573           return (ACE_THR_FUNC_RETURN) -1;
 574         }
 575     }
 576   return 0;
 577 }
 578
 579 #endif /* ACE_HAS_THREADS */
 580
 581 int basic_queue_test (ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>& q)
 582 {
 583     int status = 0;
 584     if (!q.is_empty ())
 585       {
 586         ACE_ERROR ((LM_ERROR, ACE_TEXT ("New queue is not empty!\n")));
 587         status = 1;
 588       }
 589     else
 590       {
 591         User_Class *b;
 592         ACE_Time_Value tv (ACE_OS::gettimeofday ());   // Now
 593         if (q.dequeue_head (b, &tv) != -1)
 594           {
 595             ACE_ERROR ((LM_ERROR, ACE_TEXT ("Dequeued from empty queue!\n")));
 596             status = 1;
 597           }
 598         else if (errno != EWOULDBLOCK)
 599           {
 600             ACE_ERROR ((LM_ERROR,
 601                         ACE_TEXT ("%p\n"),
 602                         ACE_TEXT ("Dequeue timeout should be EWOULDBLOCK, got")));
 603             status = 1;
 604           }
 605         else
 606           {
 607             ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Timed dequeue test: OK\n")));
 608             status = 0;     // All is well
 609           }
 610       }
 611
 612     return status;
 613 }
 614
 615 int queue_priority_test (ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>& q)
 616 {
 617   int status = 0;
 618   if (!q.is_empty ())
 619     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("Prio test queue not empty\n")), 1);
 620
 621   // Set up a few objects with names for how they should come out of the queue.
 622   std::unique_ptr<User_Class> b1, b2, b3, b4;
 623   b1.reset (new User_Class ("first"));
 624   b2.reset (new User_Class ("second"));
 625   b3.reset (new User_Class ("third"));
 626   b4.reset (new User_Class ("fourth"));
 627   unsigned long prio =
 628     ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>::DEFAULT_PRIORITY;
 629
 630   prio += 1;
 631   if (-1 == q.enqueue_prio (b2.get (), 0, prio))
 632     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b2")), 1);
 633   if (-1 == q.enqueue_prio (b3.get (), 0, prio))
 634     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b3")), 1);
 635   prio -= 1;
 636   if (-1 == q.enqueue_prio (b4.get (), 0, prio))
 637     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b4")), 1);
 638   prio += 5;
 639   if (-1 == q.enqueue_prio (b1.get (), 0, prio))
 640     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b1")), 1);
 641
 642   User_Class *b = 0;
 643   if (q.dequeue_head (b) == -1)
 644     {
 645       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 1")));
 646       ++status;
 647     }
 648   else
 649     {
 650       if (ACE_OS::strcmp (b->message (), "first") != 0)
 651         {
 652           ACE_ERROR ((LM_ERROR,
 653                       ACE_TEXT ("First dequeued was %C\n"),
 654                       b->message ()));
 655           ++status;
 656         }
 657     }
 658   if (q.dequeue_head (b) == -1)
 659     {
 660       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 2")));
 661       ++status;
 662     }
 663   else
 664     {
 665       if (ACE_OS::strcmp (b->message (), "second") != 0)
 666         {
 667           ACE_ERROR ((LM_ERROR,
 668                       ACE_TEXT ("Second dequeued was %C\n"),
 669                       b->message ()));
 670           ++status;
 671         }
 672     }
 673   if (q.dequeue_head (b) == -1)
 674     {
 675       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 3")));
 676       ++status;
 677     }
 678   else
 679     {
 680       if (ACE_OS::strcmp (b->message (), "third") != 0)
 681         {
 682           ACE_ERROR ((LM_ERROR,
 683                       ACE_TEXT ("Third dequeued was %C\n"),
 684                       b->message ()));
 685           ++status;
 686         }
 687     }
 688   if (q.dequeue_head (b) == -1)
 689     {
 690       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 4")));
 691       ++status;
 692     }
 693   else
 694     {
 695       if (ACE_OS::strcmp (b->message (), "fourth") != 0)
 696         {
 697           ACE_ERROR ((LM_ERROR,
 698                       ACE_TEXT ("Fourth dequeued was %C\n"),
 699                       b->message ()));
 700           ++status;
 701         }
 702     }
 703
 704   if (status == 0)
 705     ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Priority queueing test: OK\n")));
 706   return status;
 707 }
 708
 709 class Queue_Ex_Iterator_No_Lock
 710  : public ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>
 711 {
 712  public:
 713   typedef ACE_Message_Queue_Ex<User_Class, ACE_SYNCH, ACE_System_Time_Policy> MESSAGE_QUEUE_EX_T;
 714
 715   explicit Queue_Ex_Iterator_No_Lock (MESSAGE_QUEUE_EX_T& queue_in)
 716     : ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy> (queue_in.queue ())
 717   {}
 718   virtual ~Queue_Ex_Iterator_No_Lock () = default;
 719
 720   int next (User_Class*& message_inout)
 721   {
 722     if (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_)
 723     {
 724       message_inout =
 725             reinterpret_cast<User_Class*> (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_->base ());
 726       return 1;
 727     }
 728
 729     return 0;
 730   }
 731
 732   int done () const
 733   {
 734     return (!ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_ ? 1 : 0);
 735   }
 736
 737   int advance ()
 738   {
 739     if (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_)
 740     {
 741       ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_ =
 742         ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_->next ();
 743     }
 744
 745     return (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_ ? 1 : 0);
 746   }
 747 };
 748
 749 int queue_iterator_test (ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>& q)
 750 {
 751   int status = 0;
 752   if (!q.is_empty ())
 753     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("Iterator test queue not empty\n")), 1);
 754
 755   // Set up a few objects with names for how they should come out of the queue.
 756   std::unique_ptr<User_Class> b1 = std::make_unique<User_Class> ("first");
 757   std::unique_ptr<User_Class> b2 = std::make_unique<User_Class> ("second");
 758   std::unique_ptr<User_Class> b3 = std::make_unique<User_Class> ("third");
 759   std::unique_ptr<User_Class> b4 = std::make_unique<User_Class> ("fourth");
 760   if (-1 == q.enqueue_tail (b1.get (), 0))
 761     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b1")), 1);
 762   if (-1 == q.enqueue_tail (b2.get (), 0))
 763     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b2")), 1);
 764   if (-1 == q.enqueue_tail (b3.get (), 0))
 765     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b3")), 1);
 766   if (-1 == q.enqueue_tail (b4.get (), 0))
 767     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b4")), 1);
 768
 769   User_Class* b = nullptr;
 770   {
 771     ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, aGuard, q.lock (), 1);
 772     int counter = 0;
 773     for (Queue_Ex_Iterator_No_Lock iterator (q);
 774          iterator.next (b);
 775          iterator.advance ())
 776     {
 777       ACE_ASSERT (b);
 778       ++counter;
 779       if (counter == 1)
 780       {
 781         if (ACE_OS::strcmp (b->message (), "first") != 0)
 782         {
 783           ACE_ERROR ((LM_ERROR, ACE_TEXT ("First message was %C\n"), b->message ()));
 784           ++status;
 785         }
 786       }
 787       else if (counter == 4)
 788       {
 789         if (ACE_OS::strcmp (b->message (), "fourth") != 0)
 790         {
 791           ACE_ERROR ((LM_ERROR, ACE_TEXT ("Fourth message was %C\n"), b->message ()));
 792           ++status;
 793         }
 794       }
 795
 796       b = nullptr;
 797     }
 798   }
 799
 800   while (!q.is_empty ())
 801     q.dequeue_head (b, 0);
 802
 803   if (status == 0)
 804     ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Iterator test: OK\n")));
 805   return status;
 806 }
 807
 808 int
 809 run_main (int argc, ACE_TCHAR *argv[])
 810 {
 811   ACE_START_TEST (ACE_TEXT ("Message_Queue_Test_Ex"));
 812
 813   if (argc == 2)
 814     {
 815       if (! ACE_OS::strcmp (argv[1], ACE_TEXT ("-?")))
 816         {
 817           ACE_ERROR ((LM_ERROR,
 818                       ACE_TEXT ("%s/n"),
 819                       usage));
 820         }
 821       else
 822         {
 823           max_messages = ACE_OS::atoi (argv[1]);
 824         }
 825     }
 826
 827   int status = 0;
 828
 829   // Be sure that the a timed out get sets the error code properly.
 830   ACE_Message_Queue_Ex<User_Class, ACE_SYNCH> q1;
 831   ACE_Message_Queue_Ex_N<User_Class, ACE_SYNCH> q2;
 832   if (0 != basic_queue_test (q1) ||
 833       0 != basic_queue_test (q2))
 834     {
 835       ++status;
 836     }
 837
 838   // Check priority operations.
 839   if (0 != queue_priority_test (q1))
 840     {
 841       ++status;
 842     }
 843
 844   // Check iterator operations.
 845   if (0 != queue_iterator_test (q1))
 846     {
 847       ++status;
 848     }
 849
 850   ACE_NEW_RETURN (timer,
 851                   ACE_High_Res_Timer,
 852                   -1);
 853
 854   status += single_thread_performance_test ();
 855
 856 #if defined (ACE_HAS_THREADS)
 857   status += performance_test ();
 858 #endif /* ACE_HAS_THREADS */
 859
 860   {
 861     MQ_Ex_N_Tester ex_n_tester;
 862     status += ex_n_tester.single_thread_performance_test ();
 863 #if defined (ACE_HAS_THREADS)
 864     status += ex_n_tester.performance_test ();
 865 #endif /* ACE_HAS_THREADS */
 866     }
 867
 868   if (status != 0)
 869     ACE_ERROR ((LM_ERROR,
 870                 ACE_TEXT ("%p\n"),
 871                 ACE_TEXT ("test failed")));
 872   delete timer;
 873   timer = 0;
 874
 875   ACE_END_TEST;
 876   return status;
 877 }