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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
// $Id$
#include "ace/config-lite.h"
#if defined (ACE_HAS_THREADS)
#include "ace/OS_NS_string.h"
#include "ace/OS_NS_time.h"
#include "ace/Task.h"
#include "ace/Synch.h"
#include "ace/SString.h"
// Listing 2 code/ch16
class Workers : public ACE_Task<ACE_MT_SYNCH>
{
public:
Workers ()
{ }
virtual int svc (void)
{
while (1)
{
ACE_Message_Block *mb = 0;
if (this->getq (mb) == -1)
{
ACE_DEBUG ((LM_INFO,
ACE_TEXT ("(%t) Shutting down\n")));
break;
}
// Process the message.
process_message (mb);
}
return 0;
}
// Listing 2
private:
void process_message (ACE_Message_Block *mb)
{
ACE_TRACE ("Workers::process_message");
int msgId;
ACE_OS::memcpy (&msgId, mb->rd_ptr (), sizeof(int));
mb->release ();
ACE_DEBUG ((LM_DEBUG,
ACE_TEXT ("(%t) Started processing message %d\n"),
msgId));
ACE_OS::sleep (3);
ACE_DEBUG ((LM_DEBUG,
ACE_TEXT ("(%t) Finished processing message %d\n"),
msgId));
}
};
// Listing 1 code/ch16
class Manager : public ACE_Task<ACE_MT_SYNCH>
{
public:
enum {POOL_SIZE = 5, MAX_TIMEOUT = 5};
Manager () : shutdown_(0)
{
ACE_TRACE ("Manager::Manager");
}
int svc (void)
{
ACE_TRACE ("Manager::svc");
ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%t) Manager started\n")));
// Create pool.
Workers pool;
pool.activate (THR_NEW_LWP | THR_JOINABLE, POOL_SIZE);
while (!done ())
{
ACE_Message_Block *mb = 0;
ACE_Time_Value tv ((long)MAX_TIMEOUT);
tv += ACE_OS::time (0);
// Get a message request.
if (this->getq (mb, &tv) < 0)
{
pool.msg_queue ()->deactivate ();
pool.wait ();
break;
}
// Ask the worker pool to do the job.
pool.putq (mb);
}
return 0;
}
private:
int done (void);
int shutdown_;
};
// Listing 1
int Manager::done (void)
{
return (shutdown_ == 1);
}
int ACE_TMAIN (int, ACE_TCHAR *[])
{
Manager tp;
tp.activate ();
// Wait for a moment every time you send a message.
ACE_Time_Value tv;
tv.msec (100);
ACE_Message_Block *mb = 0;
for (int i = 0; i < 30; i++)
{
ACE_NEW_RETURN
(mb, ACE_Message_Block(sizeof(int)), -1);
ACE_OS::memcpy (mb->wr_ptr (), &i, sizeof(int));
ACE_OS::sleep (tv);
// Add a new work item.
tp.putq (mb);
}
ACE_Thread_Manager::instance ()->wait ();
return 0;
}
#else
#include "ace/OS_main.h"
#include "ace/OS_NS_stdio.h"
int ACE_TMAIN (int, ACE_TCHAR *[])
{
ACE_OS::puts (ACE_TEXT ("This example requires threads."));
return 0;
}
#endif /* ACE_HAS_THREADS */
|
