2 //=============================================================================
4 * @file Message_Queue_Test.cpp
7 * 0) a test that ensures key ACE_Message_Queue features are
8 * working properly, including timeouts and priorities
9 * 1) a simple test of the ACE_Message_Queue that illustrates how to
10 * use the forward and reverse iterators
11 * 2) a simple performance measurement test for both single-threaded
12 * (null synch), thread-safe ACE_Message_Queues, and
13 * ACE_Message_Queue_Vx, which wraps VxWorks message queues
14 * 3) a test/usage example of ACE_Message_Queue_Vx
15 * 4) a test of the message counting in a message queue under load.
17 * @author Irfan Pyarali <irfan@cs.wustl.edu>
18 * @author David L. Levine <levine@cs.wustl.edu>
19 * @author and Douglas C. Schmidt <schmidt@vanderbilt.edu>
21 //=============================================================================
24 #include "test_config.h"
25 #include "ace/Atomic_Op.h"
26 #include "ace/Thread_Manager.h"
27 #include "ace/Message_Queue.h"
28 #include "ace/Message_Queue_NT.h"
29 #include "ace/Message_Queue_Vx.h"
30 #include "ace/Synch_Traits.h"
31 #include "ace/Null_Mutex.h"
32 #include "ace/Null_Condition.h"
33 #include "ace/High_Res_Timer.h"
35 #include "ace/OS_NS_stdio.h"
36 #include "ace/OS_NS_string.h"
37 #include "ace/OS_NS_sys_time.h"
38 #include "ace/OS_NS_unistd.h"
41 const ACE_TCHAR usage
[] = ACE_TEXT ("usage: Message_Queue_Test <number of messages>\n");
43 using QUEUE
= ACE_Message_Queue
<ACE_NULL_SYNCH
>;
44 using ITERATOR
= ACE_Message_Queue_Iterator
<ACE_NULL_SYNCH
>;
45 using REVERSE_ITERATOR
= ACE_Message_Queue_Reverse_Iterator
<ACE_NULL_SYNCH
>;
47 static const int MESSAGE_FACTOR
= 100000;
48 static const int MAX_MESSAGES
= 10000;
49 static const int MAX_MESSAGE_SIZE
= 32;
50 static const char test_message
[] = "ACE_Message_Queue Test Message";
52 static int max_messages
= MAX_MESSAGES
;
54 // Dynamically allocate to avoid a static.
55 static ACE_High_Res_Timer
*timer
= 0;
57 #if defined (ACE_HAS_THREADS)
58 using SYNCH_QUEUE
= ACE_Message_Queue
<ACE_MT_SYNCH
>;
63 // Container for data passed to sender and receiver in
67 // For use in multithreaded performance test.
69 ACE_Message_Queue_Base
*q_
;
72 ACE_Message_Block
**send_block_
;
73 // Pointer to messages blocks for sender to send to reciever.
76 : q_ (0), send_block_ (0)
79 // Default constructor.
82 // For the message counting test, there are two tasks, producer and consumer.
83 // Each will spawn a number of threads, and the two tasks share a queue.
84 class Counting_Test_Producer
: public ACE_Task
<ACE_MT_SYNCH
>
87 Counting_Test_Producer (ACE_Message_Queue
<ACE_MT_SYNCH
> *queue
)
88 : ACE_Task
<ACE_MT_SYNCH
> (0, queue
), sequence_ (0), produced_ (0) {}
91 ACE_Atomic_Op
<ACE_Thread_Mutex
, long> sequence_
;
92 ACE_Atomic_Op
<ACE_Thread_Mutex
, long> produced_
;
95 class Counting_Test_Consumer
: public ACE_Task
<ACE_MT_SYNCH
>
98 Counting_Test_Consumer (ACE_Message_Queue
<ACE_MT_SYNCH
> *queue
)
99 : ACE_Task
<ACE_MT_SYNCH
> (0, queue
), consumed_ (0) {}
102 ACE_Atomic_Op
<ACE_Thread_Mutex
, long> consumed_
;
106 Counting_Test_Producer::svc ()
108 // Going to produce a lot of blocks. Since we don't necessarily want them
109 // all consumed, there's no arrangement with the consumer to be sure that
110 // the same number produced will be consumed; the test check will compare
111 // the number produced, consumed, and remaining to be sure it ends up
113 // Also, to be sure there's not just 1 producer and 1 consumer pinging
114 // back and forth, make the producers randomly delay between blocks.
115 ACE_OS::srand (static_cast<unsigned int> (ACE_OS::time ()));
116 int multiple
= ACE_OS::rand () % 10;
117 int delay_ms
= (ACE_OS::rand () % 10) / 2;
118 // The delay usually causes the test to time out in the automated
119 // regression testing. I just left it here in case it's needed someday.
121 long count
= MESSAGE_FACTOR
* (multiple
? multiple
: 1);
123 // Some of the threads enqueue single blocks, others sequences.
124 long lsequence
= ++(this->sequence_
);
125 int seq
= static_cast<int> (lsequence
);
126 ACE_DEBUG ((LM_DEBUG
,
127 ACE_TEXT ("(%t) Producer will enqueue %B blocks in seq of %d, ")
128 ACE_TEXT ("%d msec delay\n"),
133 ACE_Message_Block
*first
= 0, *prev
= 0, *b
= 0;
134 ACE_Time_Value
delay (0, delay_ms
);
135 ACE_Time_Value
timeout (10);
136 while (produced
< count
)
138 ACE_NEW_NORETURN (b
, ACE_Message_Block (1));
141 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("(%t) Producer out of memory\n")));
146 for (int s
= 1; s
< seq
; ++s
)
148 ACE_NEW_NORETURN (b
, ACE_Message_Block (1));
159 while (first
->next () != 0)
167 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("(%t) Producer out of memory\n")));
170 // To be sure we can keep going on slow or completed consumers, but not
171 // delay excessively if the consumers have stopped, limit the time
172 // spent waiting to 10 seconds.
173 ACE_Time_Value block
= ACE_OS::gettimeofday ();
175 if (this->putq (first
, &block
) == -1)
177 ACE_DEBUG ((LM_DEBUG
,
178 ACE_TEXT ("(%t) Producer cannot putq; giving up\n")));
179 while (first
->next () != 0)
190 ACE_OS::sleep (delay
);
192 this->produced_
+= produced
;
193 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("(%t) Producer done\n")));
198 Counting_Test_Consumer::svc ()
200 // Consume lots of blocks and release them. To mimic a thread with work
201 // to do, put a small random delay between dequeuing the blocks. Consume
202 // a calculated number of blocks then stop; the test checker will determine
203 // if the number consumed plus the number remaining is correct for the
205 unsigned int seed
= static_cast<unsigned int> (ACE_OS::time ());
207 int multiple
= ACE_OS::rand_r (&seed
) % 10;
208 int delay_ms
= ACE_OS::rand_r (&seed
) % 10;
209 // The delay usually causes the test to time out in the automated
210 // regression testing. I just left it here in case it's needed someday.
212 long count
= MESSAGE_FACTOR
* (multiple
? multiple
: 1);
214 ACE_DEBUG ((LM_DEBUG
,
215 ACE_TEXT ("(%t) Consumer will dequeue %B blocks, ")
216 ACE_TEXT ("%d msec delay\n"),
219 ACE_Message_Block
*b
= 0;
220 ACE_Time_Value
delay (0, delay_ms
);
221 ACE_Time_Value
timeout (2);
222 while (consumed
< count
)
224 // To be sure we can wait in the case of an empty queue, but not
225 // delay excessively if the producers have stopped, limit the time
226 // spent waiting to 2 seconds.
227 ACE_Time_Value block
= ACE_OS::gettimeofday ();
229 if (this->getq (b
, &block
) == -1)
231 if (errno
== EWOULDBLOCK
)
232 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("(%t) Consumer timed out\n")));
234 ACE_ERROR ((LM_ERROR
,
235 ACE_TEXT ("(%t) Consumer %p\n"),
242 ACE_OS::sleep (delay
);
244 this->consumed_
+= consumed
;
245 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("(%t) Consumer done\n")));
252 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("Starting counting test\n")));
254 ACE_Message_Queue
<ACE_MT_SYNCH
> q (2 * 1024 * 1024); // 2MB high water
255 Counting_Test_Producer
p (&q
);
256 Counting_Test_Consumer
c (&q
);
257 // Activate consumers first; if the producers fail to start, consumers will
259 if (c
.activate (THR_NEW_LWP
| THR_JOINABLE
| THR_INHERIT_SCHED
, 5) == -1)
260 ACE_ERROR_RETURN ((LM_ERROR
,
261 ACE_TEXT ("Consumers %p\n"),
262 ACE_TEXT ("activate")),
264 if (p
.activate (THR_NEW_LWP
| THR_JOINABLE
| THR_INHERIT_SCHED
, 5) == -1)
266 ACE_ERROR ((LM_ERROR
,
267 ACE_TEXT ("Producers %p\n"),
268 ACE_TEXT ("activate")));
272 // Producers and consumers are both running; wait for them then
273 // check the results.
276 // This compare relies on the flush() method counting blocks as it
277 // walks the chain releasing them, and doesn't rely on the count.
279 long q_count
= static_cast<long> (q
.message_count ());
280 long remaining
= q
.flush ();
281 ACE_DEBUG ((LM_DEBUG
,
282 ACE_TEXT ("Queue message_count is %b; %b flushed\n"),
284 (ssize_t
)remaining
));
285 if (q_count
!= remaining
)
288 ACE_ERROR ((LM_ERROR
,
289 ACE_TEXT ("message_count and flushed should be equal!\n")));
291 long expected
= p
.produced_
.value () - c
.consumed_
.value ();
292 ACE_DEBUG ((LM_DEBUG
,
293 ACE_TEXT ("Produced %b, consumed %b; diff %b\n"),
294 (ssize_t
)p
.produced_
.value (),
295 (ssize_t
)c
.consumed_
.value (),
297 if (expected
!= remaining
)
300 ACE_ERROR ((LM_ERROR
,
301 ACE_TEXT ("Producer-consumer diff is %b; should be %b\n"),
303 (ssize_t
)remaining
));
305 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("Ending counting test\n")));
309 #endif /* ACE_HAS_THREADS */
314 const int ITERATIONS
= 5;
315 ACE_TCHAR buffer
[ITERATIONS
][BUFSIZ
];
316 // Use queue size from of 32 Kb (more if using wide-char), instead of the
317 // default of 16 Kb (defined by ACE_Message_Queue_Base::DEFAULT_HWM),
318 // so that the test runs on machines with 8Kb pagesizes.
320 // QUEUE queue (32 * 1024 * sizeof (ACE_TCHAR));
321 QUEUE
queue (sizeof(buffer
));
325 for (i
= 0; i
< ITERATIONS
; i
++)
327 ACE_OS::snprintf (buffer
[i
], BUFSIZ
, ACE_TEXT ("%d"), i
+ 1);
329 ACE_Message_Block
*entry
= 0;
330 ACE_NEW_RETURN (entry
,
331 ACE_Message_Block ((char *) buffer
[i
],
335 if (queue
.is_full ())
336 ACE_ERROR_RETURN ((LM_ERROR
,
337 ACE_TEXT ("QUEUE:: the message queue is full on iteration %u!\n"),
341 if (queue
.enqueue (entry
) == -1)
342 ACE_ERROR_RETURN ((LM_ERROR
,
343 ACE_TEXT ("QUEUE::enqueue\n")),
347 ACE_DEBUG ((LM_DEBUG
,
348 ACE_TEXT ("\nForward Iterations\n")));
350 ITERATOR
iterator (queue
);
352 for (ACE_Message_Block
*entry
= 0;
353 iterator
.next (entry
) != 0;
355 ACE_DEBUG ((LM_DEBUG
,
360 ACE_DEBUG ((LM_DEBUG
,
361 ACE_TEXT ("\nReverse Iterations\n")));
363 REVERSE_ITERATOR
iterator (queue
);
365 for (ACE_Message_Block
*entry
= 0;
366 iterator
.next (entry
) != 0;
368 ACE_DEBUG ((LM_DEBUG
,
373 ACE_DEBUG ((LM_DEBUG
,
374 ACE_TEXT ("\nForward Iterations\n")));
376 QUEUE::ITERATOR
iterator (queue
);
378 for (ACE_Message_Block
*entry
= 0;
379 iterator
.next (entry
) != 0;
381 ACE_DEBUG ((LM_DEBUG
,
386 ACE_DEBUG ((LM_DEBUG
,
387 ACE_TEXT ("\nReverse Iterations\n")));
389 QUEUE::REVERSE_ITERATOR
iterator (queue
);
391 for (ACE_Message_Block
*entry
= 0;
392 iterator
.next (entry
) != 0;
394 ACE_DEBUG ((LM_DEBUG
,
402 #if defined (ACE_HAS_THREADS)
405 chained_block_test ()
408 const char * s
= "123456789"; // Will be length 10 when copied to block
409 const size_t slen
= 10;
410 const size_t num_blks
= 10;
411 ACE_Message_Block b
[num_blks
];
415 for (i
= 0; i
< num_blks
; ++i
)
421 // Test enqueueing single and chained blocks and be sure they end up with
422 // the proper enqueued block count and sizes. Then be sure they are dequeued
423 // in the proper order.
426 // b[3] and b[4] are unchained.
431 q
.enqueue_tail (&b
[3]);
432 q
.enqueue_tail (&b
[4]);
433 int num
= q
.enqueue_head (&b
[0]);
436 ACE_ERROR ((LM_ERROR
, ACE_TEXT ("Chained enqueue expected 5; has %d\n"),
440 num
= q
.enqueue_tail (&b
[5]);
443 ACE_ERROR ((LM_ERROR
, ACE_TEXT ("Chained enqueue expected 9; has %d\n"),
447 num
= q
.enqueue_tail (&b
[9]);
450 ACE_ERROR ((LM_ERROR
, ACE_TEXT ("Chained enqueue expected 10; has %d\n"),
455 msgs
= q
.message_count ();
456 bytes
= q
.message_bytes ();
457 if (msgs
!= 10 || bytes
!= 100)
459 ACE_ERROR ((LM_ERROR
,
460 ACE_TEXT ("Chained enqueue totals: %d msgs, %d bytes; ")
461 ACE_TEXT ("should be 10 msgs, 100 bytes\n"),
462 (int)msgs
, (int)bytes
));
466 // Now see if we can dequeue them, checking the order.
467 ACE_Time_Value
nowait (ACE_OS::gettimeofday ());
468 ACE_Message_Block
*bp
;
470 for (i
= 0; i
< num_blks
; ++i
)
472 qstat
= q
.dequeue_head (bp
, &nowait
);
475 ACE_ERROR ((LM_ERROR
,
476 ACE_TEXT ("Checking chained blocks, pass %d: %p\n"),
477 (int)i
, ACE_TEXT ("dequeue_head")));
480 else if (bp
!= &b
[i
])
482 ACE_ERROR ((LM_ERROR
,
483 ACE_TEXT ("Checking chained blocks, pass %d: ")
484 ACE_TEXT ("block out of order\n"),
491 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("Chained block test OK\n")));
496 single_thread_performance_test (int queue_type
= 0)
498 const char test_message
[] =
499 "ACE_Message_Queue Test Message";
500 const ACE_TCHAR
*message
=
501 ACE_TEXT ("ACE_Message_Queue<ACE_NULL_SYNCH>, single thread");
504 // Create a message queue.
505 ACE_Message_Queue_Base
*msgq
= 0;
508 ACE_NEW_RETURN (msgq
,
511 #if defined (ACE_VXWORKS)
514 ACE_NEW_RETURN (msgq
,
515 ACE_Message_Queue_Vx (max_messages
,
518 message
= "ACE_Message_Queue_Vx, single thread test";
520 #elif defined (ACE_WIN32) && defined (ACE_HAS_WIN32_OVERLAPPED_IO)
523 ACE_NEW_RETURN (msgq
,
524 ACE_Message_Queue_NT
,
526 message
= ACE_TEXT ("ACE_Message_Queue_NT, single thread test");
528 #endif /* ACE_VXWORKS */
530 // Create the messages. Allocate off the heap in case messages
531 // is large relative to the amount of stack space available.
532 ACE_Message_Block
**send_block
= 0;
533 ACE_NEW_RETURN (send_block
,
534 ACE_Message_Block
*[max_messages
],
537 for (i
= 0; i
< max_messages
; ++i
)
538 ACE_NEW_RETURN (send_block
[i
],
539 ACE_Message_Block (test_message
,
543 ACE_Message_Block
**receive_block_p
= 0;
544 ACE_NEW_RETURN (receive_block_p
,
545 ACE_Message_Block
*[max_messages
],
548 #if defined (ACE_VXWORKS)
549 // Set up blocks to receive the messages. Allocate these off the
550 // heap in case messages is large relative to the amount of
551 // stack space available.
552 ACE_Message_Block
*receive_block
= 0;
553 ACE_NEW_RETURN (receive_block
,
554 ACE_Message_Block
[max_messages
],
557 for (i
= 0; i
< max_messages
; ++i
)
559 receive_block
[i
].init (MAX_MESSAGE_SIZE
);
561 // For VxWorks Message Queues, the receive block pointer must be
562 // assigned. It will be used by dequeue_head ().
563 receive_block_p
[i
] = &receive_block
[i
];
565 #endif /* ACE_VXWORKS */
569 // Send/receive the messages.
570 for (i
= 0; i
< max_messages
; ++i
)
572 if (msgq
->enqueue_tail (send_block
[i
]) == -1)
573 ACE_ERROR_RETURN ((LM_ERROR
,
575 ACE_TEXT ("enqueue")),
578 if (msgq
->dequeue_head (receive_block_p
[i
]) == -1)
579 ACE_ERROR_RETURN ((LM_ERROR
,
581 ACE_TEXT ("dequeue_head")),
588 timer
->elapsed_time (tv
);
590 ACE_TEXT ("%s: %u messages took %u msec (%f msec/message)\n"),
594 (double) tv
.msec () / max_messages
));
597 delete [] receive_block_p
;
598 #if defined (ACE_VXWORKS)
599 delete [] receive_block
;
600 #endif /* ACE_VXWORKS */
602 for (i
= 0; i
< max_messages
; ++i
)
603 delete send_block
[i
];
604 delete [] send_block
;
613 Queue_Wrapper
*queue_wrapper
=
614 reinterpret_cast<Queue_Wrapper
*> (arg
);
617 ACE_Message_Block
**receive_block_p
= 0;
618 ACE_NEW_RETURN (receive_block_p
,
619 ACE_Message_Block
*[max_messages
],
622 #if defined (ACE_VXWORKS)
623 // Set up blocks to receive the messages. Allocate these off the
624 // heap in case messages is large relative to the amount of stack
626 ACE_Message_Block
*receive_block
;
627 ACE_NEW_RETURN (receive_block
,
628 ACE_Message_Block
[max_messages
],
631 for (i
= 0; i
< max_messages
; ++i
)
633 receive_block
[i
].init (MAX_MESSAGE_SIZE
);
635 // For VxWorks Message Queues, the receive block pointer must be
636 // assigned. It will be used by <dequeue_head>.
637 receive_block_p
[i
] = &receive_block
[i
];
639 #endif /* ACE_VXWORKS */
641 for (i
= 0; i
< max_messages
; ++i
)
642 if (queue_wrapper
->q_
->dequeue_head (receive_block_p
[i
]) == -1)
643 ACE_ERROR_RETURN ((LM_ERROR
,
645 ACE_TEXT ("dequeue_head")),
649 delete [] receive_block_p
;
650 #if defined (ACE_VXWORKS)
651 delete [] receive_block
;
652 #endif /* ACE_VXWORKS */
660 Queue_Wrapper
*queue_wrapper
=
661 reinterpret_cast<Queue_Wrapper
*> (arg
);
666 // Send the messages.
667 for (i
= 0; i
< max_messages
; ++i
)
668 if (queue_wrapper
->q_
->
669 enqueue_tail (queue_wrapper
->send_block_
[i
]) == -1)
670 ACE_ERROR_RETURN ((LM_ERROR
,
672 ACE_TEXT ("enqueue")),
679 performance_test (int queue_type
= 0)
681 Queue_Wrapper queue_wrapper
;
682 const ACE_TCHAR
*message
=
683 ACE_TEXT ("ACE_Message_Queue<ACE_SYNCH>");
686 // Create the messages. Allocate off the heap in case messages is
687 // large relative to the amount of stack space available. Allocate
688 // it here instead of in the sender, so that we can delete it after
689 // the _receiver_ is done.
690 ACE_Message_Block
**send_block
= 0;
691 ACE_NEW_RETURN (send_block
,
692 ACE_Message_Block
*[max_messages
],
695 for (i
= 0; i
< max_messages
; ++i
)
696 ACE_NEW_RETURN (send_block
[i
],
697 ACE_Message_Block (test_message
,
701 queue_wrapper
.send_block_
= send_block
;
704 ACE_NEW_RETURN (queue_wrapper
.q_
,
707 #if defined (ACE_VXWORKS)
710 ACE_NEW_RETURN (queue_wrapper
.q_
,
711 ACE_Message_Queue_Vx (max_messages
,
714 message
= "ACE_Message_Queue_Vx";
716 #elif defined (ACE_WIN32) && defined (ACE_HAS_WIN32_OVERLAPPED_IO)
719 ACE_NEW_RETURN (queue_wrapper
.q_
,
720 ACE_Message_Queue_NT
,
722 message
= ACE_TEXT ("ACE_Message_Queue_NT");
724 #endif /* ACE_VXWORKS */
726 if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC
) sender
,
729 ACE_ERROR_RETURN ((LM_ERROR
,
731 ACE_TEXT ("spawning sender thread")),
734 if (ACE_Thread_Manager::instance ()->spawn ((ACE_THR_FUNC
) receiver
,
737 ACE_ERROR_RETURN ((LM_ERROR
,
739 ACE_TEXT ("spawning receiver thread")),
742 ACE_Thread_Manager::instance ()->wait ();
744 timer
->elapsed_time (tv
);
745 ACE_DEBUG ((LM_INFO
, ACE_TEXT ("%s: %u messages took %u msec (%f msec/message)\n"),
749 (double) tv
.msec () / max_messages
));
752 delete queue_wrapper
.q_
;
753 queue_wrapper
.q_
= 0;
755 for (i
= 0; i
< max_messages
; ++i
)
756 delete send_block
[i
];
757 delete [] send_block
;
762 // Ensure that the timedout dequeue_head() sets errno code properly.
772 ACE_ERROR ((LM_ERROR
,
773 ACE_TEXT ("New queue is not empty!\n")));
778 ACE_Message_Block
*b
;
779 ACE_Time_Value
tv (ACE_OS::gettimeofday ()); // Now
781 if (mq
.dequeue_head (b
, &tv
) != -1)
783 ACE_ERROR ((LM_ERROR
,
784 ACE_TEXT ("Dequeued from empty queue!\n")));
787 else if (errno
!= EWOULDBLOCK
)
789 ACE_ERROR ((LM_ERROR
,
791 ACE_TEXT ("Dequeue timeout should be EWOULDBLOCK, got")));
796 ACE_DEBUG ((LM_DEBUG
,
797 ACE_TEXT ("Timed dequeue test: OK\n")));
798 status
= 0; // All is well
804 #endif /* ACE_HAS_THREADS */
806 // Check to make sure that dequeue_prio() respects FIFO ordering.
807 // @@ At some point, this function should be enhanced to do a more
813 const char S1
[] = "first";
814 const char S2
[] = "second";
815 const int PRIORITY
= 50;
819 ACE_Message_Block
mb1 (S1
, sizeof S1
, PRIORITY
);
820 ACE_Message_Block
mb2 (S2
, sizeof S2
, PRIORITY
);
822 mq
.enqueue_prio (&mb1
);
823 mq
.enqueue_prio (&mb2
);
825 ACE_Message_Block
*mb1p
= 0;
826 ACE_Message_Block
*mb2p
= 0;
828 mq
.dequeue_prio (mb1p
);
829 mq
.dequeue_prio (mb2p
);
831 ACE_DEBUG ((LM_DEBUG
, "message 1 = %C\nmessage 2 = %C\n",
835 if (ACE_OS::strcmp (mb1p
->rd_ptr (), S1
) == 0
836 && ACE_OS::strcmp (mb2p
->rd_ptr (), S2
) == 0)
849 int flushed_messages
;
852 flushed_messages
= mq1
.close ();
854 if (flushed_messages
!= 0)
856 ACE_ERROR ((LM_ERROR
,
857 ACE_TEXT ("Closing queue should flush 0 messages, close() reports - %d\n"),
863 // There was a bug that return previous queue state instead of
864 // number of flushed messages. Thus, insert 2 messages != ACTIVATE
866 ACE_Message_Block
*pMB1
;
867 ACE_Message_Block
*pMB2
;
868 ACE_NEW_NORETURN (pMB1
, ACE_Message_Block (1));
869 ACE_NEW_NORETURN (pMB2
, ACE_Message_Block (1));
871 mq2
.enqueue_head (pMB1
);
872 mq2
.enqueue_head (pMB2
);
873 flushed_messages
= mq2
.close ();
875 if (flushed_messages
!= 2)
877 ACE_ERROR ((LM_ERROR
,
878 ACE_TEXT ("Closing queue should flush 2 messages, close() reports - %d\n"),
887 run_main (int argc
, ACE_TCHAR
*argv
[])
889 ACE_START_TEST (ACE_TEXT ("Message_Queue_Test"));
893 if (!ACE_OS::strcmp (argv
[1], ACE_TEXT ("-?")))
895 ACE_ERROR ((LM_ERROR
,
901 max_messages
= ACE_OS::atoi (argv
[1]);
905 int status
= prio_test ();
907 // The iterator test occasionally causes a page fault or a hang on
910 status
= iterator_test ();
912 ACE_NEW_RETURN (timer
,
917 status
= close_test ();
919 #if defined (ACE_HAS_THREADS)
921 status
= timeout_test ();
924 status
= chained_block_test ();
927 status
= single_thread_performance_test ();
929 # if defined (ACE_VXWORKS) || defined (ACE_HAS_WIN32_OVERLAPPED_IO)
930 // Test ACE_Message_Queue_Vx. or ACE_Message_Queue_NT
932 status
= single_thread_performance_test (1);
933 # endif /* ACE_VXWORKS */
936 status
= performance_test ();
938 # if defined (ACE_VXWORKS) || defined (ACE_HAS_WIN32_OVERLAPPED_IO)
939 // Test ACE_Message_Queue_Vx or ACE_Message_Queue_NT
941 status
= performance_test (1);
942 # endif /* ACE_VXWORKS */
944 if (counting_test () != 0)
946 #endif /* ACE_HAS_THREADS */
949 ACE_ERROR ((LM_ERROR
,
951 ACE_TEXT ("test failed")));