//=============================================================================
/**
 *  @file    Message_Queue_Test.cpp
 *
 *  $Id: Message_Queue_Test.cpp 93638 2011-03-24 13:16:05Z johnnyw $
 *
 *    This is:
 *    0) a test that ensures key ACE_Message_Queue features are
 *       working properly, including timeouts and priorities
 *    1) a simple test of the ACE_Message_Queue that illustrates how to
 *       use the forward and reverse iterators
 *    2) a simple performance measurement test for both single-threaded
 *       (null synch), thread-safe ACE_Message_Queues, and
 *       ACE_Message_Queue_Vx, which wraps VxWorks message queues
 *    3) a test/usage example of ACE_Message_Queue_Vx
 *    4) a test of the message counting in a message queue under load.
 *
 *
 *  @author Irfan Pyarali <irfan@cs.wustl.edu>
 *  @author David L. Levine <levine@cs.wustl.edu>
 *  @author and Douglas C. Schmidt <schmidt@vanderbilt.edu>
 */
//=============================================================================


#include "test_config.h"
#include "ace/Atomic_Op.h"
#include "ace/Thread_Manager.h"
#include "ace/Message_Queue.h"
#include "ace/Message_Queue_NT.h"
#include "ace/Message_Queue_Vx.h"
#include "ace/Synch_Traits.h"
#include "ace/Null_Mutex.h"
#include "ace/Null_Condition.h"
#include "ace/High_Res_Timer.h"
#include "ace/Task.h"
#include "ace/OS_NS_stdio.h"
#include "ace/OS_NS_string.h"
#include "ace/OS_NS_sys_time.h"
#include "ace/OS_NS_unistd.h"



const ACE_TCHAR usage[] = ACE_TEXT ("usage: Message_Queue_Test <number of messages>\n");

typedef ACE_Message_Queue<ACE_NULL_SYNCH> QUEUE;
typedef ACE_Message_Queue_Iterator<ACE_NULL_SYNCH> ITERATOR;
typedef ACE_Message_Queue_Reverse_Iterator<ACE_NULL_SYNCH> REVERSE_ITERATOR;

#if defined (ACE_HAS_WINCE)
static const int MESSAGE_FACTOR = 10000;
#else
static const int MESSAGE_FACTOR = 100000;
#endif
static const int MAX_MESSAGES = 10000;
static const int MAX_MESSAGE_SIZE = 32;
static const char test_message[] = "ACE_Message_Queue Test Message";

static int max_messages = MAX_MESSAGES;

// Dynamically allocate to avoid a static.
static ACE_High_Res_Timer *timer = 0;

#if defined (ACE_HAS_THREADS)
typedef ACE_Message_Queue<ACE_MT_SYNCH> SYNCH_QUEUE;

struct Queue_Wrapper
{
  // = TITLE
  //     Container for data passed to sender and receiver in
  //     performance test.
  //
  // = DESCRIPTION
  //     For use in multithreaded performance test.

  ACE_Message_Queue_Base *q_;
  // The message queue.

  ACE_Message_Block **send_block_;
  // Pointer to messages blocks for sender to send to reciever.

  Queue_Wrapper (void)
    : q_ (0), send_block_ (0)
  {
  }
  // Default constructor.
};

// For the message counting test, there are two tasks, producer and consumer.
// Each will spawn a number of threads, and the two tasks share a queue.
class Counting_Test_Producer : public ACE_Task<ACE_MT_SYNCH>
{
public:
  Counting_Test_Producer (ACE_Message_Queue<ACE_MT_SYNCH> *queue)
    : ACE_Task<ACE_MT_SYNCH> (0, queue), sequence_ (0), produced_ (0) {}
  virtual int svc (void);

  ACE_Atomic_Op<ACE_Thread_Mutex, long> sequence_;
  ACE_Atomic_Op<ACE_Thread_Mutex, long> produced_;
};

class Counting_Test_Consumer : public ACE_Task<ACE_MT_SYNCH>
{
public:
  Counting_Test_Consumer (ACE_Message_Queue<ACE_MT_SYNCH> *queue)
    : ACE_Task<ACE_MT_SYNCH> (0, queue), consumed_ (0) {}
  virtual int svc (void);

  ACE_Atomic_Op<ACE_Thread_Mutex, long> consumed_;
};

int
Counting_Test_Producer::svc (void)
{
  // Going to produce a lot of blocks. Since we don't necessarily want them
  // all consumed, there's no arrangement with the consumer to be sure that
  // the same number produced will be consumed; the test check will compare
  // the number produced, consumed, and remaining to be sure it ends up
  // correct.
  // Also, to be sure there's not just 1 producer and 1 consumer pinging
  // back and forth, make the producers randomly delay between blocks.
  ACE_OS::srand (static_cast<unsigned int> (ACE_OS::time ()));
  int multiple = ACE_OS::rand () % 10;
  int delay_ms = (ACE_OS::rand () % 10) / 2;
  // The delay usually causes the test to time out in the automated
  // regression testing. I just left it here in case it's needed someday.
  delay_ms = 0;
  long count = MESSAGE_FACTOR * (multiple ? multiple : 1);
  long produced = 0;
  // Some of the threads enqueue single blocks, others sequences.
  long lsequence = ++(this->sequence_);
  int seq = static_cast<int> (lsequence);
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%t) Producer will enqueue %B blocks in seq of %d, ")
              ACE_TEXT ("%d msec delay\n"),
              (size_t)count,
              seq,
              delay_ms));

  ACE_Message_Block *first = 0, *prev = 0, *b = 0;
  ACE_Time_Value delay (0, delay_ms);
  ACE_Time_Value timeout (10);
  while (produced < count)
    {
      ACE_NEW_NORETURN (b, ACE_Message_Block (1));
      if (b == 0)
        {
          ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Producer out of memory\n")));
          break;
        }
      first = b;
      prev = first;
      for (int s = 1; s < seq; ++s)
        {
          ACE_NEW_NORETURN (b, ACE_Message_Block (1));
          if (b == 0)
            break;
          prev->next (b);
          b->prev (prev);
          prev = b;
        }
      if (b == 0)
        {
          if (first != b)
            {
              while (first->next () != 0)
                {
                  b = first->next ();
                  first->release ();
                  first = b;
                }
              first->release ();
            }
          ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Producer out of memory\n")));
          break;
        }
      // To be sure we can keep going on slow or completed consumers, but not
      // delay excessively if the consumers have stopped, limit the time
      // spent waiting to 10 seconds.
      ACE_Time_Value block = ACE_OS::gettimeofday ();
      block += timeout;
      if (this->putq (first, &block) == -1)
        {
          ACE_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("(%t) Producer cannot putq; giving up\n")));
          while (first->next () != 0)
            {
              b = first->next ();
              first->release ();
              first = b;
            }
          first->release ();
          break;
        }
      produced += seq;
      if (delay_ms)
        ACE_OS::sleep (delay);
    }
  this->produced_ += produced;
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Producer done\n")));
  return 0;
}

int
Counting_Test_Consumer::svc (void)
{
  // Consume lots of blocks and release them. To mimic a thread with work
  // to do, put a small random delay between dequeuing the blocks. Consume
  // a calculated number of blocks then stop; the test checker will determine
  // if the number consumed plus the number remaining is correct for the
  // number produced.
  unsigned int seed = static_cast<unsigned int> (ACE_OS::time ());

  int multiple = ACE_OS::rand_r (&seed) % 10;
  int delay_ms = ACE_OS::rand_r (&seed) % 10;
  // The delay usually causes the test to time out in the automated
  // regression testing. I just left it here in case it's needed someday.
  delay_ms = 0;
  long count = MESSAGE_FACTOR * (multiple ? multiple : 1);
  long consumed = 0;
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%t) Consumer will dequeue %B blocks, ")
              ACE_TEXT ("%d msec delay\n"),
              (size_t)count,
              delay_ms));
  ACE_Message_Block *b = 0;
  ACE_Time_Value delay (0, delay_ms);
  ACE_Time_Value timeout (2);
  while (consumed < count)
    {
      // To be sure we can wait in the case of an empty queue, but not
      // delay excessively if the producers have stopped, limit the time
      // spent waiting to 2 seconds.
      ACE_Time_Value block = ACE_OS::gettimeofday ();
      block += timeout;
      if (this->getq (b, &block) == -1)
        {
          if (errno == EWOULDBLOCK)
            ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Consumer timed out\n")));
          else
            ACE_ERROR ((LM_ERROR,
                        ACE_TEXT ("(%t) Consumer %p\n"),
                        ACE_TEXT ("getq")));
          break;
        }
      ++consumed;
      b->release ();
      if (delay_ms)
        ACE_OS::sleep (delay);
    }
  this->consumed_ += consumed;
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Consumer done\n")));
  return 0;
}

static int
counting_test (void)
{
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Starting counting test\n")));

  ACE_Message_Queue<ACE_MT_SYNCH> q (2 * 1024 * 1024);  // 2MB high water
  Counting_Test_Producer p (&q);
  Counting_Test_Consumer c (&q);
  // Activate consumers first; if the producers fail to start, consumers will
  // stop quicker.
  if (c.activate (THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED, 5) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("Consumers %p\n"),
                       ACE_TEXT ("activate")),
                      -1);
  if (p.activate (THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED, 5) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("Producers %p\n"),
                  ACE_TEXT ("activate")));
      c.wait ();
      return -1;
    }
  // Producers and consumers are both running; wait for them then
  // check the results.
  p.wait ();
  c.wait ();
  // This compare relies on the flush() method counting blocks as it
  // walks the chain releasing them, and doesn't rely on the count.
  int status = 0;
  long q_count = static_cast<long> (q.message_count ());
  long remaining = q.flush ();
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("Queue message_count is %b; %b flushed\n"),
              (ssize_t)q_count,
              (ssize_t)remaining));
  if (q_count != remaining)
    {
      status = -1;
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("message_count and flushed should be equal!\n")));
    }
  long expected = p.produced_.value () - c.consumed_.value ();
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("Produced %b, consumed %b; diff %b\n"),
              (ssize_t)p.produced_.value (),
              (ssize_t)c.consumed_.value (),
              (ssize_t)expected));
  if (expected != remaining)
    {
      status = -1;
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("Producer-consumer diff is %b; should be %b\n"),
                  (ssize_t)expected,
                  (ssize_t)remaining));
    }
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Ending counting test\n")));
  return status;
}

#endif /* ACE_HAS_THREADS */

static int
iterator_test (void)
{
  const int ITERATIONS = 5;
  ACE_TCHAR buffer[ITERATIONS][BUFSIZ];
  // Use queue size from of 32 Kb (more if using wide-char), instead of the
  // default of 16 Kb (defined by ACE_Message_Queue_Base::DEFAULT_HWM),
  // so that the test runs on machines with 8Kb pagesizes.

  //  QUEUE queue (32 * 1024 * sizeof (ACE_TCHAR));
  QUEUE queue (sizeof(buffer));

  int i;

  for (i = 0; i < ITERATIONS; i++)
    {
      ACE_OS::sprintf (buffer[i],
                       ACE_TEXT ("%d"),
                       i + 1);

      ACE_Message_Block *entry = 0;
      ACE_NEW_RETURN (entry,
                      ACE_Message_Block ((char *) buffer[i],
                                         sizeof buffer[i]),
                      -1);

      if (queue.is_full ())
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("QUEUE:: the message queue is full on iteration %u!\n"),
                           i + 1),
                          -1);

      if (queue.enqueue (entry) == -1)
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("QUEUE::enqueue\n")),
                          -1);
    }

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\nForward Iterations\n")));
  {
    ITERATOR iterator (queue);

    for (ACE_Message_Block *entry = 0;
         iterator.next (entry) != 0;
         iterator.advance ())
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("%s\n"),
                  entry->base ()));
  }

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\nReverse Iterations\n")));
  {
    REVERSE_ITERATOR iterator (queue);

    for (ACE_Message_Block *entry = 0;
         iterator.next (entry) != 0;
         iterator.advance ())
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("%s\n"),
                  entry->base ()));
  }

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\nForward Iterations\n")));
  {
    QUEUE::ITERATOR iterator (queue);

    for (ACE_Message_Block *entry = 0;
         iterator.next (entry) != 0;
         iterator.advance ())
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("%s\n"),
                  entry->base ()));
  }

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\nReverse Iterations\n")));
  {
    QUEUE::REVERSE_ITERATOR iterator (queue);

    for (ACE_Message_Block *entry = 0;
         iterator.next (entry) != 0;
         iterator.advance ())
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("%s\n"),
                  entry->base ()));
  }

  return 0;
}

#if defined (ACE_HAS_THREADS)

static int
chained_block_test (void)
{

  QUEUE q;
  const char * s = "123456789";      // Will be length 10 when copied to block
  const size_t slen = 10;
  const size_t num_blks = 10;
  ACE_Message_Block b[num_blks];
  size_t i;
  int status = 0;

  for (i = 0; i < num_blks; ++i)
    {
      b[i].init (slen);
      b[i].copy (s);
    }

  // Test enqueueing single and chained blocks and be sure they end up with
  // the proper enqueued block count and sizes. Then be sure they are dequeued
  // in the proper order.
  b[0].next (&b[1]);
  b[1].next (&b[2]);
  // b[3] and b[4] are unchained.
  b[5].next (&b[6]);
  b[6].next (&b[7]);
  b[7].next (&b[8]);
  // b[9] is unchained
  q.enqueue_tail (&b[3]);
  q.enqueue_tail (&b[4]);
  int num = q.enqueue_head (&b[0]);
  if (num != 5)
    {
      ACE_ERROR ((LM_ERROR, ACE_TEXT ("Chained enqueue expected 5; has %d\n"),
                  num));
      status = -1;
    }
  num = q.enqueue_tail (&b[5]);
  if (num != 9)
    {
      ACE_ERROR ((LM_ERROR, ACE_TEXT ("Chained enqueue expected 9; has %d\n"),
                  num));
      status = -1;
    }
  num = q.enqueue_tail (&b[9]);
  if (num != 10)
    {
      ACE_ERROR ((LM_ERROR, ACE_TEXT ("Chained enqueue expected 10; has %d\n"),
                  num));
      status = -1;
    }
  size_t msgs, bytes;
  msgs = q.message_count ();
  bytes = q.message_bytes ();
  if (msgs != 10 || bytes != 100)
    {
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("Chained enqueue totals: %d msgs, %d bytes; ")
                  ACE_TEXT ("should be 10 msgs, 100 bytes\n"),
                  (int)msgs, (int)bytes));
      status = -1;
    }

  // Now see if we can dequeue them, checking the order.
  ACE_Time_Value nowait (ACE_OS::gettimeofday ());
  ACE_Message_Block *bp;
  int qstat;
  for (i = 0; i < num_blks; ++i)
    {
      qstat = q.dequeue_head (bp, &nowait);
      if (qstat == -1)
        {
          ACE_ERROR ((LM_ERROR,
                      ACE_TEXT ("Checking chained blocks, pass %d: %p\n"),
                      (int)i, ACE_TEXT ("dequeue_head")));
          status = -1;
        }
      else if (bp != &b[i])
        {
          ACE_ERROR ((LM_ERROR,
                      ACE_TEXT ("Checking chained blocks, pass %d: ")
                      ACE_TEXT ("block out of order\n"),
                      (int)i));
          status = -1;
        }
    }

  if (status == 0)
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Chained block test OK\n")));
  return status;
}

static int
single_thread_performance_test (int queue_type = 0)
{
  const char test_message[] =
    "ACE_Message_Queue Test Message";
  const ACE_TCHAR *message =
    ACE_TEXT ("ACE_Message_Queue<ACE_NULL_SYNCH>, single thread");
  int i = 0;

  // Create a message queue.
  ACE_Message_Queue_Base *msgq = 0;

  if (queue_type == 0)
    ACE_NEW_RETURN (msgq,
                    QUEUE,
                    -1);
#if defined (ACE_VXWORKS)
  else
    {
      ACE_NEW_RETURN (msgq,
                      ACE_Message_Queue_Vx (max_messages,
                                            MAX_MESSAGE_SIZE),
                      -1);
      message = "ACE_Message_Queue_Vx, single thread test";
    }
#elif defined (ACE_WIN32) && defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  else
    {
      ACE_NEW_RETURN (msgq,
                      ACE_Message_Queue_NT,
                      -1);
      message = ACE_TEXT ("ACE_Message_Queue_NT, single thread test");
    }
#endif /* ACE_VXWORKS */

  // Create the messages.  Allocate off the heap in case messages
  // is large relative to the amount of stack space available.
  ACE_Message_Block **send_block = 0;
  ACE_NEW_RETURN (send_block,
                  ACE_Message_Block *[max_messages],
                  -1);

  for (i = 0; i < max_messages; ++i)
    ACE_NEW_RETURN (send_block[i],
                    ACE_Message_Block (test_message,
                                       MAX_MESSAGE_SIZE),
                    -1);

  ACE_Message_Block **receive_block_p = 0;
  ACE_NEW_RETURN (receive_block_p,
                  ACE_Message_Block *[max_messages],
                  -1);

#if defined (ACE_VXWORKS)
  // Set up blocks to receive the messages.  Allocate these off the
  // heap in case messages is large relative to the amount of
  // stack space available.
  ACE_Message_Block *receive_block = 0;
  ACE_NEW_RETURN (receive_block,
                  ACE_Message_Block[max_messages],
                  -1);

  for (i = 0; i < max_messages; ++i)
    {
      receive_block[i].init (MAX_MESSAGE_SIZE);

      // For VxWorks Message Queues, the receive block pointer must be
      // assigned.  It will be used by dequeue_head ().
      receive_block_p[i] = &receive_block[i];
    }
#endif /* ACE_VXWORKS */

  timer->start ();

  // Send/receive the messages.
  for (i = 0; i < max_messages; ++i)
    {
      if (msgq->enqueue_tail (send_block[i]) == -1)
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("%p\n"),
                           ACE_TEXT ("enqueue")),
                          -1);

      if (msgq->dequeue_head (receive_block_p[i]) == -1)
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("%p\n"),
                           ACE_TEXT ("dequeue_head")),
                          -1);
    }

  timer->stop ();

  ACE_Time_Value tv;
  timer->elapsed_time (tv);
  ACE_DEBUG ((LM_INFO,
              ACE_TEXT ("%s: %u messages took %u msec (%f msec/message)\n"),
              message,
              max_messages,
              tv.msec (),
              (double) tv.msec () / max_messages));
  timer->reset ();

  delete [] receive_block_p;
#if defined (ACE_VXWORKS)
  delete [] receive_block;
#endif /* ACE_VXWORKS */

  for (i = 0; i < max_messages; ++i)
    delete send_block[i];
  delete [] send_block;
  delete msgq;

  return 0;
}

static void *
receiver (void *arg)
{
  Queue_Wrapper *queue_wrapper =
    reinterpret_cast<Queue_Wrapper *> (arg);
  int i;

  ACE_Message_Block **receive_block_p = 0;
  ACE_NEW_RETURN (receive_block_p,
                  ACE_Message_Block *[max_messages],
                  (void *) -1);

#if defined (ACE_VXWORKS)
  // Set up blocks to receive the messages.  Allocate these off the
  // heap in case messages is large relative to the amount of stack
  // space available.
  ACE_Message_Block *receive_block;
  ACE_NEW_RETURN (receive_block,
                  ACE_Message_Block[max_messages],
                  (void *) -1);

  for (i = 0; i < max_messages; ++i)
    {
      receive_block[i].init (MAX_MESSAGE_SIZE);

      // For VxWorks Message Queues, the receive block pointer must be
      // assigned.  It will be used by <dequeue_head>.
      receive_block_p[i] = &receive_block[i];
    }
#endif /* ACE_VXWORKS */

  for (i = 0; i < max_messages; ++i)
    if (queue_wrapper->q_->dequeue_head (receive_block_p[i]) == -1)
      ACE_ERROR_RETURN ((LM_ERROR,
                         ACE_TEXT ("%p\n"),
                         ACE_TEXT ("dequeue_head")),
                        0);
  timer->stop ();

  delete [] receive_block_p;
#if defined (ACE_VXWORKS)
  delete [] receive_block;
#endif /* ACE_VXWORKS */

  return 0;
}

static void *
sender (void *arg)
{
  Queue_Wrapper *queue_wrapper =
    reinterpret_cast<Queue_Wrapper *> (arg);
  int i;

  timer->start ();

  // Send the messages.
  for (i = 0; i < max_messages; ++i)
    if (queue_wrapper->q_->
        enqueue_tail (queue_wrapper->send_block_[i]) == -1)
      ACE_ERROR_RETURN ((LM_ERROR,
                         ACE_TEXT ("%p\n"),
                         ACE_TEXT ("enqueue")),
                        0);
  return 0;
}

static
int
performance_test (int queue_type = 0)
{
  Queue_Wrapper queue_wrapper;
  const ACE_TCHAR *message =
    ACE_TEXT ("ACE_Message_Queue<ACE_SYNCH>");
  int i = 0;

  // Create the messages.  Allocate off the heap in case messages is
  // large relative to the amount of stack space available.  Allocate
  // it here instead of in the sender, so that we can delete it after
  // the _receiver_ is done.
  ACE_Message_Block **send_block = 0;
  ACE_NEW_RETURN (send_block,
                  ACE_Message_Block *[max_messages],
                  -1);

  for (i = 0; i < max_messages; ++i)
    ACE_NEW_RETURN (send_block[i],
                    ACE_Message_Block (test_message,
                                       MAX_MESSAGE_SIZE),
                    -1);

  queue_wrapper.send_block_ = send_block;

  if (queue_type == 0)
    ACE_NEW_RETURN (queue_wrapper.q_,
                    SYNCH_QUEUE,
                    -1);
#if defined (ACE_VXWORKS)
  else
    {
      ACE_NEW_RETURN (queue_wrapper.q_,
                      ACE_Message_Queue_Vx (max_messages,
                                            MAX_MESSAGE_SIZE),
                      -1);
      message = "ACE_Message_Queue_Vx";
    }
#elif defined (ACE_WIN32) && defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  else
    {
      ACE_NEW_RETURN (queue_wrapper.q_,
                      ACE_Message_Queue_NT,
                      -1);
      message = ACE_TEXT ("ACE_Message_Queue_NT");
    }
#endif /* ACE_VXWORKS */

  if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) sender,
                                              &queue_wrapper,
                                              THR_BOUND) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("%p\n"),
                       ACE_TEXT ("spawning sender thread")),
                      -1);

  if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC) receiver,
                                              &queue_wrapper,
                                              THR_BOUND) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("%p\n"),
                       ACE_TEXT ("spawning receiver thread")),
                      -1);

  ACE_Thread_Manager::instance ()->wait ();
  ACE_Time_Value tv;
  timer->elapsed_time (tv);
  ACE_DEBUG ((LM_INFO, ACE_TEXT ("%s: %u messages took %u msec (%f msec/message)\n"),
              message,
              max_messages,
              tv.msec (),
              (double) tv.msec () / max_messages));
  timer->reset ();

  delete queue_wrapper.q_;
  queue_wrapper.q_ = 0;

  for (i = 0; i < max_messages; ++i)
    delete send_block[i];
  delete [] send_block;

  return 0;
}

// Ensure that the timedout dequeue_head() sets errno code properly.

static int
timeout_test (void)
{
  SYNCH_QUEUE mq;
  int status = 0;

  if (!mq.is_empty ())
    {
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("New queue is not empty!\n")));
      status = 1;
    }
  else
    {
      ACE_Message_Block *b;
      ACE_Time_Value tv (ACE_OS::gettimeofday ());   // Now

      if (mq.dequeue_head (b, &tv) != -1)
        {
          ACE_ERROR ((LM_ERROR,
                      ACE_TEXT ("Dequeued from empty queue!\n")));
          status = 1;
        }
      else if (errno != EWOULDBLOCK)
        {
          ACE_ERROR ((LM_ERROR,
                      ACE_TEXT ("%p\n"),
                      ACE_TEXT ("Dequeue timeout should be EWOULDBLOCK, got")));
          status = 1;
        }
      else
        {
          ACE_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("Timed dequeue test: OK\n")));
          status = 0;     // All is well
        }
    }

  return status;
}
#endif /* ACE_HAS_THREADS */

// Check to make sure that dequeue_prio() respects FIFO ordering.
// @@ At some point, this function should be enhanced to do a more
// thorough check...

static int
prio_test (void)
{
  const char S1[] = "first";
  const char S2[] = "second";
  const int PRIORITY = 50;
  QUEUE mq;
  int status;

  ACE_Message_Block mb1 (S1, sizeof S1, PRIORITY);
  ACE_Message_Block mb2 (S2, sizeof S2, PRIORITY);

  mq.enqueue_prio (&mb1);
  mq.enqueue_prio (&mb2);

  ACE_Message_Block *mb1p;
  ACE_Message_Block *mb2p;

  mq.dequeue_prio (mb1p);
  mq.dequeue_prio (mb2p);

  ACE_DEBUG ((LM_DEBUG, "message 1 = %C\nmessage 2 = %C\n",
              mb1p->rd_ptr (),
              mb2p->rd_ptr ()));

  if (ACE_OS_String::strcmp (mb1p->rd_ptr (), S1) == 0
      && ACE_OS_String::strcmp (mb2p->rd_ptr (), S2) == 0)
    status = 0;
  else
    status = 1;

  return status;
}

static int
close_test (void)
{
  int status = 0;

  int flushed_messages;

  QUEUE mq1;
  flushed_messages = mq1.close ();

  if (flushed_messages != 0)
    {
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("Closing queue should flush 0 messages, close() reports - %d\n"),
                  flushed_messages ));
      status = 1;
      return status;
    }

  // There was a bug that return previous queue state instead of
  // number of flushed messages. Thus, insert 2 messages != ACTIVATE
  // queue state
  ACE_Message_Block *pMB1;
  ACE_Message_Block *pMB2;
  ACE_NEW_NORETURN (pMB1, ACE_Message_Block (1));
  ACE_NEW_NORETURN (pMB2, ACE_Message_Block (1));
  QUEUE mq2;
  mq2.enqueue_head (pMB1);
  mq2.enqueue_head (pMB2);
  flushed_messages = mq2.close ();

  if (flushed_messages != 2)
    {
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("Closing queue should flush 2 messages, close() reports - %d\n"),
                  flushed_messages ));
      status = 1;
      return status;
    }
  return status;
}

int
run_main (int argc, ACE_TCHAR *argv[])
{
  ACE_START_TEST (ACE_TEXT ("Message_Queue_Test"));

  if (argc == 2)
    {
      if (!ACE_OS::strcmp (argv[1], ACE_TEXT ("-?")))
        {
          ACE_ERROR ((LM_ERROR,
                      ACE_TEXT ("%s/n"),
                      usage));
        }
      else
        {
          max_messages = ACE_OS::atoi (argv[1]);
        }
    }

  int status = prio_test ();

  // The iterator test occasionally causes a page fault or a hang on
  // VxWorks.
  if (status == 0)
    status = iterator_test ();

  ACE_NEW_RETURN (timer,
                  ACE_High_Res_Timer,
                  -1);

  if (status == 0)
    status = close_test ();

#if defined (ACE_HAS_THREADS)
  if (status == 0)
    status = timeout_test ();

  if (status == 0)
    status = chained_block_test ();

  if (status == 0)
    status = single_thread_performance_test ();

# if defined (ACE_VXWORKS) || defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  // Test ACE_Message_Queue_Vx. or ACE_Message_Queue_NT
  if (status == 0)
    status = single_thread_performance_test (1);
# endif /* ACE_VXWORKS */

  if (status == 0)
    status = performance_test ();

# if defined (ACE_VXWORKS) || defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  // Test ACE_Message_Queue_Vx or ACE_Message_Queue_NT
  if (status == 0)
    status = performance_test (1);
# endif /* ACE_VXWORKS */

  if (counting_test () != 0)
    status = -1;
#endif /* ACE_HAS_THREADS */

  if (status != 0)
    ACE_ERROR ((LM_ERROR,
                ACE_TEXT ("%p\n"),
                ACE_TEXT ("test failed")));
  delete timer;
  timer = 0;



  ACE_END_TEST;
  return status;
}