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 "ace/Auto_Ptr.h"
  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 ()
  82     : q_ (0), send_block_ (0)
  83   {
  84   }
  85 };
  86
  87 struct MQ_Ex_N_Tester_Wrapper
  88 {
  89   //  = TITLE
  90   //      Container for data passed to sender in the MQ_Ex_N_Tester
  91   //      performance test.
  92   //
  93   //  = DESCRIPTION
  94   //     For use in multithreaded performance test.
  95   MQ_Ex_N_Tester *tester_;
  96   User_Class     *head_send_block_;
  97 };
  98
  99 #endif /* ACE_HAS_THREADS */
 100
 101 // Encapsulates the sent messages creation and destruction
 102 struct Send_Messages
 103 {
 104   Send_Messages (int number_of_messages, int chain_limit):
 105     send_block_ (0),
 106     number_of_messages_ (number_of_messages),
 107     chain_limit_ (chain_limit)
 108   {
 109   }
 110
 111   int create_messages (const char test_message[])
 112   {
 113     int limit = this->number_of_messages_ / this->chain_limit_;
 114     ACE_NEW_RETURN (this->send_block_,
 115                     User_Class *[limit],
 116                     -1);
 117
 118     int i, j;
 119     for (i = 0; i < limit; ++i)
 120       {
 121         User_Class *&temp1 = this->send_block_[i];
 122         ACE_NEW_RETURN (temp1,
 123                         User_Class (test_message),
 124                         -1);
 125         User_Class *tail = temp1;
 126         for (j = 1; j < this->chain_limit_; ++j)
 127           {
 128             User_Class *temp2 = 0;
 129             ACE_NEW_RETURN (temp2,
 130                             User_Class (test_message),
 131                             -1);
 132             tail->next (temp2);
 133             tail = temp2;
 134           }
 135       }
 136     this->head_send_block_ = this->send_block_[0];
 137     return 0;
 138   }
 139
 140   ~Send_Messages ()
 141   {
 142     int j, i = 0;
 143     int limit = this->number_of_messages_ / this->chain_limit_;
 144     for (; i < limit; ++i)
 145       {
 146         User_Class *&temp1 = this->send_block_[i];
 147         for (j = 0; j < this->chain_limit_; ++j)
 148           {
 149             User_Class *temp2 = temp1->next ();
 150             delete temp1;
 151             temp1 = temp2;
 152           }
 153       }
 154     delete [] this->send_block_;
 155   }
 156
 157   User_Class *  head_send_block_;
 158   User_Class ** send_block_;
 159   int           number_of_messages_;
 160   int           chain_limit_;
 161 };
 162
 163 // Encapsulates the received messages creation and destruction
 164 struct Receive_Messages
 165 {
 166   Receive_Messages (int number_of_messages) :
 167     receive_block_ (0),
 168     number_of_messages_ (number_of_messages)
 169   {
 170   }
 171
 172   int create ()
 173   {
 174     ACE_NEW_RETURN (this->receive_block_,
 175                     User_Class *[this->number_of_messages_],
 176                     -1);
 177     return 0;
 178   }
 179
 180   ~Receive_Messages ()
 181   {
 182     delete [] this->receive_block_;
 183   }
 184
 185   User_Class **receive_block_;
 186   int          number_of_messages_;
 187 };
 188
 189 static int
 190 single_thread_performance_test ()
 191 {
 192   const char test_message[] =
 193     "ACE_Message_Queue_Ex Test Message";
 194   const ACE_TCHAR *message =
 195     ACE_TEXT ("ACE_Message_Queue_Ex<ACE_NULL_SYNCH>, single thread");
 196
 197   // Create a message queue.
 198   QUEUE *msgq = 0;
 199
 200   ACE_NEW_RETURN (msgq,
 201                   QUEUE,
 202                   -1);
 203
 204   // Create the messages.  Allocate off the heap in case messages is
 205   // large relative to the amount of stack space available.
 206   User_Class **send_block = 0;
 207   ACE_NEW_RETURN (send_block,
 208                   User_Class *[max_messages],
 209                   -1);
 210
 211   int i = 0;
 212
 213   for (i = 0; i < max_messages; ++i)
 214     ACE_NEW_RETURN (send_block[i],
 215                     User_Class (test_message),
 216                     -1);
 217
 218   User_Class **receive_block_p = 0;
 219   ACE_NEW_RETURN (receive_block_p,
 220                   User_Class *[max_messages],
 221                   -1);
 222
 223   timer->start ();
 224
 225   // Send/receive the messages.
 226   for (i = 0; i < max_messages; ++i)
 227     {
 228       if (msgq->enqueue_tail (send_block[i]) == -1)
 229         ACE_ERROR_RETURN ((LM_ERROR,
 230                            ACE_TEXT ("%p\n"),
 231                            ACE_TEXT ("enqueue")),
 232                           -1);
 233
 234       if (msgq->dequeue_head (receive_block_p[i]) == -1)
 235         ACE_ERROR_RETURN ((LM_ERROR,
 236                            ACE_TEXT ("%p\n"),
 237                            ACE_TEXT ("dequeue_head")),
 238                           -1);
 239     }
 240
 241   timer->stop ();
 242   print_message (message);
 243   timer->reset ();
 244
 245   delete [] receive_block_p;
 246
 247   for (i = 0; i < max_messages; ++i)
 248     delete send_block[i];
 249   delete [] send_block;
 250   delete msgq;
 251
 252   return 0;
 253 }
 254
 255 int
 256 MQ_Ex_N_Tester::single_thread_performance_test ()
 257 {
 258   // Create the messages.  Allocate off the heap in case messages is
 259   // large relative to the amount of stack space available.
 260
 261   if ((0 != this->test_enqueue_tail ()) ||
 262       (0 != this->test_enqueue_head ())    )
 263     {
 264       return -1;
 265     }
 266
 267   return 0;
 268 }
 269
 270 int
 271 MQ_Ex_N_Tester::test_enqueue_tail ()
 272 {
 273   const ACE_TCHAR *message =
 274     ACE_TEXT ("ACE_Message_Queue_Ex_N<ACE_NULL_SYNCH>, test_enqueue_tail");
 275
 276   // Send_Messages creates messages and deletes them when it gets out of scope
 277   Send_Messages messages (max_messages, chain_limit);
 278   if (-1 == messages.create_messages (test_message))
 279     {
 280       return -1;
 281     }
 282   Receive_Messages r_messages (max_messages);
 283   if (-1 == r_messages.create ())
 284     {
 285       return -1;
 286     }
 287
 288   // prepare
 289   int limit = max_messages / chain_limit;
 290   timer->start ();
 291   // Send with just one call
 292   for (int i = 0; i < limit; ++i)
 293     {
 294       if (-1 == this->st_queue_.enqueue_tail (messages.send_block_[i]))
 295         {
 296           ACE_ERROR_RETURN ((LM_ERROR,
 297                              ACE_TEXT ("%p\n"),
 298                              ACE_TEXT ("enqueue_tail_n")),
 299                             -1);
 300         }
 301
 302       for (int j = 0, k = 0; j < chain_limit; ++j, ++k)
 303         {
 304           if (this->st_queue_.dequeue_head (r_messages.receive_block_[k]) == -1)
 305             {
 306               ACE_ERROR_RETURN ((LM_ERROR,
 307                                  ACE_TEXT ("%p\n"),
 308                                  ACE_TEXT ("dequeue_head")),
 309                                 -1);
 310             }
 311         }
 312     }
 313   timer->stop ();
 314
 315   print_message (message);
 316
 317   timer->reset ();
 318
 319   return 0;
 320 }
 321
 322 int
 323 MQ_Ex_N_Tester::test_enqueue_head ()
 324 {
 325   const ACE_TCHAR *message =
 326     ACE_TEXT ("ACE_Message_Queue_Ex_N<ACE_NULL_SYNCH>, test_enqueue_head");
 327
 328   // Send_Messages creates messages and deletes them when it gets out of scope
 329   Send_Messages messages (max_messages, chain_limit);
 330   if (-1 == messages.create_messages (test_message))
 331     {
 332       return -1;
 333     }
 334   Receive_Messages r_messages (max_messages);
 335   if (-1 == r_messages.create ())
 336     {
 337       return -1;
 338     }
 339
 340   // prepare
 341   int i, j, k = 0;
 342
 343   int limit = max_messages / chain_limit;
 344   timer->start ();
 345
 346   // Send/receive the messages.
 347   // Send with just one call
 348   for (i = 0; i < limit; ++i)
 349     {
 350       if (-1 == this->st_queue_.enqueue_head (messages.send_block_[i]))
 351         {
 352           ACE_ERROR_RETURN ((LM_ERROR,
 353                              ACE_TEXT ("%p\n"),
 354                              ACE_TEXT ("enqueue_tail_n")),
 355                             -1);
 356         }
 357
 358       for (j = 0; j < chain_limit; ++j, ++k)
 359         {
 360           if (this->st_queue_.dequeue_head (r_messages.receive_block_[k]) == -1)
 361             {
 362               ACE_ERROR_RETURN ((LM_ERROR,
 363                                  ACE_TEXT ("%p\n"),
 364                                  ACE_TEXT ("dequeue_head")),
 365                                 -1);
 366             }
 367         }
 368     }
 369   timer->stop ();
 370
 371   print_message (message);
 372
 373   timer->reset ();
 374
 375   return 0;
 376 }
 377
 378 #if defined (ACE_HAS_THREADS)
 379
 380 static void *
 381 receiver (void *arg)
 382 {
 383   Queue_Wrapper *queue_wrapper = reinterpret_cast<Queue_Wrapper *> (arg);
 384   int i;
 385
 386   User_Class **receive_block_p = 0;
 387   ACE_NEW_RETURN (receive_block_p,
 388                   User_Class *[max_messages],
 389                   (void *) -1);
 390
 391   for (i = 0; i < max_messages; ++i)
 392     if (queue_wrapper->q_->dequeue_head (receive_block_p[i]) == -1)
 393       ACE_ERROR_RETURN ((LM_ERROR,
 394                          ACE_TEXT ("%p\n"),
 395                          ACE_TEXT ("dequeue_head")),
 396                         0);
 397   timer->stop ();
 398
 399   delete [] receive_block_p;
 400
 401   return 0;
 402 }
 403
 404 static void *
 405 sender (void *arg)
 406 {
 407   Queue_Wrapper *queue_wrapper = reinterpret_cast<Queue_Wrapper *> (arg);
 408   int i;
 409
 410   timer->start ();
 411
 412   // Send the messages.
 413   for (i = 0; i < max_messages; ++i)
 414     if (queue_wrapper->q_->
 415         enqueue_tail (queue_wrapper->send_block_[i]) == -1)
 416       ACE_ERROR_RETURN ((LM_ERROR,
 417                          ACE_TEXT ("%p\n"),
 418                          ACE_TEXT ("enqueue")),
 419                         0);
 420   return 0;
 421 }
 422
 423 static int
 424 performance_test ()
 425 {
 426   Queue_Wrapper queue_wrapper;
 427   const ACE_TCHAR *message =
 428     ACE_TEXT ("ACE_Message_Queue_Ex<ACE_SYNCH>");
 429   int i = 0;
 430
 431   // Create the messages.  Allocate off the heap in case messages is
 432   // large relative to the amount of stack space available.  Allocate
 433   // it here instead of in the sender, so that we can delete it after
 434   // the _receiver_ is done.
 435   User_Class **send_block = 0;
 436   ACE_NEW_RETURN (send_block,
 437                   User_Class *[max_messages],
 438                   -1);
 439
 440   for (i = 0; i < max_messages; ++i)
 441     ACE_NEW_RETURN (send_block[i],
 442                     User_Class (test_message),
 443                     -1);
 444
 445   queue_wrapper.send_block_ = send_block;
 446
 447   ACE_NEW_RETURN (queue_wrapper.q_,
 448                   SYNCH_QUEUE,
 449                   -1);
 450
 451   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) sender,
 452                                               &queue_wrapper,
 453                                               THR_BOUND) == -1)
 454     ACE_ERROR_RETURN ((LM_ERROR,
 455                        ACE_TEXT ("%p\n"),
 456                        ACE_TEXT ("spawning sender thread")),
 457                       -1);
 458
 459   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) receiver,
 460                                               &queue_wrapper,
 461                                               THR_BOUND) == -1)
 462     ACE_ERROR_RETURN ((LM_ERROR,
 463                        ACE_TEXT ("%p\n"),
 464                        ACE_TEXT ("spawning receiver thread")),
 465                       -1);
 466
 467   ACE_Thread_Manager::instance ()->wait ();
 468   print_message (message);
 469   timer->reset ();
 470
 471   delete queue_wrapper.q_;
 472   queue_wrapper.q_ = 0;
 473
 474   for (i = 0; i < max_messages; ++i)
 475     delete send_block[i];
 476   delete [] send_block;
 477
 478   return 0;
 479 }
 480
 481 int
 482 MQ_Ex_N_Tester::performance_test ()
 483 {
 484   const ACE_TCHAR *message =
 485     ACE_TEXT ("ACE_Message_Queue_Ex_N<ACE_SYNCH>");
 486
 487   Send_Messages messages (max_messages, chain_limit);
 488   if (-1 == messages.create_messages (test_message))
 489     {
 490       return -1;
 491     }
 492
 493   MQ_Ex_N_Tester_Wrapper tester_wrapper;
 494   tester_wrapper.head_send_block_ = messages.head_send_block_;
 495   tester_wrapper.tester_ = this;
 496
 497   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) &MQ_Ex_N_Tester::sender,
 498                                               &tester_wrapper,
 499                                               THR_BOUND) == -1)
 500     ACE_ERROR_RETURN ((LM_ERROR,
 501                        ACE_TEXT ("%p\n"),
 502                        ACE_TEXT ("spawning sender thread")),
 503                       -1);
 504
 505   if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) &MQ_Ex_N_Tester::receiver,
 506                                               this,
 507                                               THR_BOUND) == -1)
 508     ACE_ERROR_RETURN ((LM_ERROR,
 509                        ACE_TEXT ("%p\n"),
 510                        ACE_TEXT ("spawning receiver thread")),
 511                       -1);
 512
 513   ACE_Thread_Manager::instance ()->wait ();
 514
 515   print_message (message);
 516
 517   timer->reset ();
 518
 519   return 0;
 520 }
 521
 522 ACE_THR_FUNC_RETURN
 523 MQ_Ex_N_Tester::receiver (void *args)
 524 {
 525   MQ_Ex_N_Tester *tester = reinterpret_cast<MQ_Ex_N_Tester *> (args);
 526
 527   User_Class **receive_block_p = 0;
 528   ACE_NEW_RETURN (receive_block_p,
 529                   User_Class *[max_messages],
 530                   (ACE_THR_FUNC_RETURN) -1);
 531
 532   int i;
 533   tester_barrier.wait ();
 534   for (i = 0; i < max_messages; ++i)
 535     {
 536       if (tester->mt_queue_.dequeue_head (receive_block_p[i]) == -1)
 537         {
 538           ACE_ERROR ((LM_ERROR,
 539                       ACE_TEXT ("%p\n"),
 540                       ACE_TEXT ("dequeue_head")));
 541           return (ACE_THR_FUNC_RETURN) -1;
 542         }
 543     }
 544   timer->stop ();
 545
 546   delete [] receive_block_p;
 547
 548   return 0;
 549 }
 550
 551 ACE_THR_FUNC_RETURN
 552 MQ_Ex_N_Tester::sender (void *args)
 553 {
 554   MQ_Ex_N_Tester_Wrapper *tester_wrapper =
 555     reinterpret_cast<MQ_Ex_N_Tester_Wrapper *> (args);
 556   MQ_Ex_N_Tester *tester = tester_wrapper->tester_;
 557
 558   Send_Messages messages (max_messages, chain_limit);
 559   if (-1 == messages.create_messages (test_message))
 560     {
 561       return (ACE_THR_FUNC_RETURN) -1;
 562     }
 563   int limit = max_messages / chain_limit;
 564   tester_barrier.wait ();
 565   timer->start ();
 566   // Send/receive the messages.
 567   timer->start ();
 568   // Send with just one call
 569   for (int i = 0; i < limit; ++i)
 570     {
 571       if (-1 == tester->mt_queue_.enqueue_tail (messages.send_block_[i]))
 572         {
 573           ACE_ERROR ((LM_ERROR,
 574                       ACE_TEXT ("%p\n"),
 575                       ACE_TEXT ("enqueue_tail_n")));
 576           return (ACE_THR_FUNC_RETURN) -1;
 577         }
 578     }
 579   return 0;
 580 }
 581
 582 #endif /* ACE_HAS_THREADS */
 583
 584 int basic_queue_test (ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>& q)
 585 {
 586     int status = 0;
 587     if (!q.is_empty ())
 588       {
 589         ACE_ERROR ((LM_ERROR, ACE_TEXT ("New queue is not empty!\n")));
 590         status = 1;
 591       }
 592     else
 593       {
 594         User_Class *b;
 595         ACE_Time_Value tv (ACE_OS::gettimeofday ());   // Now
 596         if (q.dequeue_head (b, &tv) != -1)
 597           {
 598             ACE_ERROR ((LM_ERROR, ACE_TEXT ("Dequeued from empty queue!\n")));
 599             status = 1;
 600           }
 601         else if (errno != EWOULDBLOCK)
 602           {
 603             ACE_ERROR ((LM_ERROR,
 604                         ACE_TEXT ("%p\n"),
 605                         ACE_TEXT ("Dequeue timeout should be EWOULDBLOCK, got")));
 606             status = 1;
 607           }
 608         else
 609           {
 610             ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Timed dequeue test: OK\n")));
 611             status = 0;     // All is well
 612           }
 613       }
 614
 615     return status;
 616 }
 617
 618 int queue_priority_test (ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>& q)
 619 {
 620   int status = 0;
 621   if (!q.is_empty ())
 622     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("Prio test queue not empty\n")), 1);
 623
 624   // Set up a few objects with names for how they should come out of the queue.
 625   ACE_Auto_Basic_Ptr<User_Class> b1, b2, b3, b4;
 626   b1.reset (new User_Class ("first"));
 627   b2.reset (new User_Class ("second"));
 628   b3.reset (new User_Class ("third"));
 629   b4.reset (new User_Class ("fourth"));
 630   unsigned long prio =
 631     ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>::DEFAULT_PRIORITY;
 632
 633   prio += 1;
 634   if (-1 == q.enqueue_prio (b2.get (), 0, prio))
 635     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b2")), 1);
 636   if (-1 == q.enqueue_prio (b3.get (), 0, prio))
 637     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b3")), 1);
 638   prio -= 1;
 639   if (-1 == q.enqueue_prio (b4.get (), 0, prio))
 640     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b4")), 1);
 641   prio += 5;
 642   if (-1 == q.enqueue_prio (b1.get (), 0, prio))
 643     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b1")), 1);
 644
 645   User_Class *b = 0;
 646   if (q.dequeue_head (b) == -1)
 647     {
 648       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 1")));
 649       ++status;
 650     }
 651   else
 652     {
 653       if (ACE_OS::strcmp (b->message (), "first") != 0)
 654         {
 655           ACE_ERROR ((LM_ERROR,
 656                       ACE_TEXT ("First dequeued was %C\n"),
 657                       b->message ()));
 658           ++status;
 659         }
 660     }
 661   if (q.dequeue_head (b) == -1)
 662     {
 663       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 2")));
 664       ++status;
 665     }
 666   else
 667     {
 668       if (ACE_OS::strcmp (b->message (), "second") != 0)
 669         {
 670           ACE_ERROR ((LM_ERROR,
 671                       ACE_TEXT ("Second dequeued was %C\n"),
 672                       b->message ()));
 673           ++status;
 674         }
 675     }
 676   if (q.dequeue_head (b) == -1)
 677     {
 678       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 3")));
 679       ++status;
 680     }
 681   else
 682     {
 683       if (ACE_OS::strcmp (b->message (), "third") != 0)
 684         {
 685           ACE_ERROR ((LM_ERROR,
 686                       ACE_TEXT ("Third dequeued was %C\n"),
 687                       b->message ()));
 688           ++status;
 689         }
 690     }
 691   if (q.dequeue_head (b) == -1)
 692     {
 693       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("dequeue 4")));
 694       ++status;
 695     }
 696   else
 697     {
 698       if (ACE_OS::strcmp (b->message (), "fourth") != 0)
 699         {
 700           ACE_ERROR ((LM_ERROR,
 701                       ACE_TEXT ("Fourth dequeued was %C\n"),
 702                       b->message ()));
 703           ++status;
 704         }
 705     }
 706
 707   if (status == 0)
 708     ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Priority queueing test: OK\n")));
 709   return status;
 710 }
 711
 712 int
 713 run_main (int argc, ACE_TCHAR *argv[])
 714 {
 715   ACE_START_TEST (ACE_TEXT ("Message_Queue_Test_Ex"));
 716
 717   if (argc == 2)
 718     {
 719       if (! ACE_OS::strcmp (argv[1], ACE_TEXT ("-?")))
 720         {
 721           ACE_ERROR ((LM_ERROR,
 722                       ACE_TEXT ("%s/n"),
 723                       usage));
 724         }
 725       else
 726         {
 727           max_messages = ACE_OS::atoi (argv[1]);
 728         }
 729     }
 730
 731   int status = 0;
 732
 733   // Be sure that the a timed out get sets the error code properly.
 734   ACE_Message_Queue_Ex<User_Class, ACE_SYNCH> q1;
 735   ACE_Message_Queue_Ex_N<User_Class, ACE_SYNCH> q2;
 736   if (0 != basic_queue_test (q1) ||
 737       0 != basic_queue_test (q2))
 738     {
 739       ++status;
 740     }
 741
 742   // Check priority operations.
 743   if (0 != queue_priority_test (q1))
 744     {
 745       ++status;
 746     }
 747
 748   ACE_NEW_RETURN (timer,
 749                   ACE_High_Res_Timer,
 750                   -1);
 751
 752   status += single_thread_performance_test ();
 753
 754 #if defined (ACE_HAS_THREADS)
 755   status += performance_test ();
 756 #endif /* ACE_HAS_THREADS */
 757
 758   {
 759     MQ_Ex_N_Tester ex_n_tester;
 760     status += ex_n_tester.single_thread_performance_test ();
 761 #if defined (ACE_HAS_THREADS)
 762     status += ex_n_tester.performance_test ();
 763 #endif /* ACE_HAS_THREADS */
 764     }
 765
 766   if (status != 0)
 767     ACE_ERROR ((LM_ERROR,
 768                 ACE_TEXT ("%p\n"),
 769                 ACE_TEXT ("test failed")));
 770   delete timer;
 771   timer = 0;
 772
 773   ACE_END_TEST;
 774   return status;
 775 }