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
150
151
152
153
154
155
156
157
158
159
160
|
// author : Boris Kolpackov <boris@dre.vanderbilt.edu>
// OS primitives
#include "ace/OS_NS_string.h"
#include "ace/OS_NS_stdlib.h"
#include "ace/Refcounted_Auto_Ptr.h"
#include "ace/SOCK_Dgram_Mcast.h"
#include "ace/Synch_Traits.h"
#include "Messaging.hpp"
#include "Protocol.hpp"
namespace ACE_TMCast
{
class LinkFailure : public virtual Message {};
class LinkData : public virtual Message
{
public:
LinkData (Protocol::MessageHeader const* header,
void* payload,
size_t size)
: size_ (size)
{
ACE_OS::memcpy (&header_, header, sizeof (Protocol::MessageHeader));
ACE_OS::memcpy (payload_, payload, size_);
}
Protocol::MessageHeader const&
header () const
{
return header_;
}
void const*
payload () const
{
return payload_;
}
size_t
size () const
{
return size_;
}
private:
Protocol::MessageHeader header_;
char payload_[Protocol::MAX_MESSAGE_SIZE];
size_t size_;
};
typedef
ACE_Refcounted_Auto_Ptr<LinkData, ACE_Null_Mutex>
LinkDataPtr;
class LinkListener
{
private:
class Terminate : public virtual Message {};
public:
LinkListener (ACE_SOCK_Dgram_Mcast& sock, MessageQueue& out)
: sock_(sock), out_ (out)
{
ACE_thread_t unused;
if (ACE_OS::thr_create (&thread_thunk,
this,
THR_JOINABLE,
&unused,
&thread_) != 0) ACE_OS::abort ();
}
~LinkListener ()
{
{
MessageQueueAutoLock lock (control_);
control_.push (MessagePtr (new Terminate));
}
if (ACE_OS::thr_join (thread_, 0) != 0) ACE_OS::abort ();
// cerr << "Link listener is down." << endl;
}
static ACE_THR_FUNC_RETURN
thread_thunk (void* arg)
{
LinkListener* obj = reinterpret_cast<LinkListener*> (arg);
obj->execute ();
return 0;
}
void
execute ()
{
char msg[Protocol::MAX_MESSAGE_SIZE];
ssize_t header_size = sizeof (Protocol::MessageHeader);
// OS::Time timeout (1000000); // one millisecond
ACE_Time_Value timeout (0, 1000); // one millisecond
try
{
while (true)
{
// Check control message queue
{
MessageQueueAutoLock lock (control_);
if (!control_.empty ()) break;
}
ACE_INET_Addr junk;
ssize_t n = sock_.recv (msg,
Protocol::MAX_MESSAGE_SIZE,
junk,
0,
&timeout);
if (n != -1)
{
if (n < header_size) throw false;
Protocol::MessageHeader* header =
reinterpret_cast<Protocol::MessageHeader*> (msg);
MessageQueueAutoLock lock (out_);
out_.push (MessagePtr (new LinkData (header,
msg + header_size,
n - header_size)));
}
}
}
catch (...)
{
MessageQueueAutoLock lock (out_);
out_.push (MessagePtr (new LinkFailure));
}
}
private:
// FUZZ: disable check_for_ACE_Guard
typedef ACE_Guard<ACE_SYNCH_MUTEX> AutoLock;
// FUZZ: enable check_for_ACE_Guard
ACE_hthread_t thread_;
ACE_SOCK_Dgram_Mcast& sock_;
MessageQueue& out_;
MessageQueue control_;
};
}
|
