1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
// $Id$
// This short program copies stdin to stdout via the use of an ASX
// Message_Queue. It illustrates an implementation of the classic
// "bounded buffer" program.
#include "ace/Message_Queue.h"
#include "ace/Thread_Manager.h"
#if defined (ACE_HAS_THREADS)
// Global thread manager.
static ACE_Thread_Manager thr_mgr;
// The producer reads data from the stdin stream, creates a message,
// and then queues the message in the message list, where it is
// removed by the consumer thread. A 0-sized message is enqueued when
// there is no more data to read. The consumer uses this as a flag to
// know when to exit.
static void *
producer (ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue)
{
// Keep reading stdin, until we reach EOF.
for (int n; ; )
{
// Allocate a new message.
ACE_Message_Block *mb;
ACE_NEW_RETURN (mb, ACE_Message_Block (BUFSIZ), 0);
n = ACE_OS::read (ACE_STDIN, mb->rd_ptr (), mb->size ());
if (n <= 0)
{
// Send a shutdown message to the other thread and exit.
mb->length (0);
if (msg_queue->enqueue_tail (mb) == -1)
ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next"));
break;
}
// Send the message to the other thread.
else
{
mb->msg_priority (n);
mb->wr_ptr (n);
if (msg_queue->enqueue_tail (mb) == -1)
ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next"));
}
}
return 0;
}
// The consumer dequeues a message from the ACE_Message_Queue, writes
// the message to the stderr stream, and deletes the message. The
// producer sends a 0-sized message to inform the consumer to stop
// reading and exit.
static void *consumer (ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue)
{
int result = 0;
// Keep looping, reading a message out of the queue, until we timeout
// or get a message with a length == 0, which signals us to quit.
for (;;)
{
ACE_Message_Block *mb;
ACE_Time_Value timeout (ACE_OS::time (0) + 4, 0); // Wait for upto 4 seconds
result = msg_queue->dequeue_head (mb, &timeout);
if (result == -1)
break;
int length = mb->length ();
if (length > 0)
ACE_OS::write (ACE_STDOUT, mb->rd_ptr (), length);
mb->release ();
if (length == 0)
break;
}
if (result == -1 && errno == EWOULDBLOCK)
ACE_ERROR ((LM_ERROR, "(%t) %p\n%a", "timed out waiting for message", 1));
return 0;
}
// Spawn off two threads that copy stdin to stdout.
int
main (int, char *[])
{
// Message list.
ACE_Message_Queue<ACE_MT_SYNCH> msg_queue;
if (thr_mgr.spawn (ACE_THR_FUNC (producer), (void *) &msg_queue,
THR_NEW_LWP | THR_DETACHED) == -1)
ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 1);
else if (thr_mgr.spawn (ACE_THR_FUNC (consumer), (void *) &msg_queue,
THR_NEW_LWP | THR_DETACHED) == -1)
ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 1);
// Wait for producer and consumer threads to exit.
thr_mgr.wait ();
return 0;
}
#else
int
main (int, char *[])
{
ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n"));
return 0;
}
#endif /* ACE_HAS_THREADS */
|
