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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
|
// Copyright (C) 2011 Internet Systems Consortium, Inc. ("ISC")
//
// Permission to use, copy, modify, and/or distribute this software for any
// purpose with or without fee is hereby granted, provided that the above
// copyright notice and this permission notice appear in all copies.
//
// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
// AND FITNESS. IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
// PERFORMANCE OF THIS SOFTWARE.
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h> // for some IPC/network system calls
#include <errno.h>
#include <boost/shared_array.hpp>
#include <config.h>
#include <log/dummylog.h>
#include <asio.hpp>
#include <asio/error.hpp>
#include <asiolink/dummy_io_cb.h>
#include <asiolink/udp_endpoint.h>
#include <asiolink/udp_server.h>
#include <asiolink/udp_socket.h>
#include <dns/opcode.h>
using namespace asio;
using asio::ip::udp;
using isc::log::dlog;
using namespace std;
using namespace isc::dns;
using namespace isc::util;
namespace asiolink {
/*
* Some of the member variables here are shared_ptrs and some are
* auto_ptrs. There will be one instance of Data for the lifetime
* of packet. The variables that are state only for a single packet
* use auto_ptr, as it is more lightweight. In the case of shared
* configuration (eg. the callbacks, socket), we use shared_ptrs.
*/
struct UDPServer::Data {
/*
* Constructor from parameters passed to UDPServer constructor.
* This instance will not be used to retrieve and answer the actual
* query, it will only hold parameters until we wait for the
* first packet. But we do initialize the socket in here.
*/
Data(io_service& io_service, const ip::address& addr, const uint16_t port,
SimpleCallback* checkin, DNSLookup* lookup, DNSAnswer* answer) :
io_(io_service), done_(false),
checkin_callback_(checkin),lookup_callback_(lookup),
answer_callback_(answer)
{
// We must use different instantiations for v4 and v6;
// otherwise ASIO will bind to both
udp proto = addr.is_v4() ? udp::v4() : udp::v6();
socket_.reset(new udp::socket(io_service, proto));
socket_->set_option(socket_base::reuse_address(true));
if (addr.is_v6()) {
socket_->set_option(asio::ip::v6_only(true));
}
socket_->bind(udp::endpoint(addr, port));
}
/*
* Copy constructor. Default one would probably do, but it is unnecessary
* to copy many of the member variables every time we fork to handle
* another packet.
*
* We also allocate data for receiving the packet here.
*/
Data(const Data& other) :
io_(other.io_), socket_(other.socket_), done_(false),
checkin_callback_(other.checkin_callback_),
lookup_callback_(other.lookup_callback_),
answer_callback_(other.answer_callback_)
{
// Instantiate the data buffer and endpoint that will
// be used by the asynchronous receive call.
data_.reset(new char[MAX_LENGTH]);
sender_.reset(new udp::endpoint());
}
// The ASIO service object
asio::io_service& io_;
// Class member variables which are dynamic, and changes to which
// need to accessible from both sides of a coroutine fork or from
// outside of the coroutine (i.e., from an asynchronous I/O call),
// should be declared here as pointers and allocated in the
// constructor or in the coroutine. This allows state information
// to persist when an individual copy of the coroutine falls out
// scope while waiting for an event, *so long as* there is another
// object that is referencing the same data. As a side-benefit, using
// pointers also reduces copy overhead for coroutine objects.
//
// Note: Currently these objects are allocated by "new" in the
// constructor, or in the function operator while processing a query.
// Repeated allocations from the heap for every incoming query is
// clearly a performance issue; this must be optimized in the future.
// The plan is to have a structure pre-allocate several "Data"
// objects which can be pulled off a free list and placed on an in-use
// list whenever a query comes in. This will serve the dual purpose
// of improving performance and guaranteeing that state information
// will *not* be destroyed when any one instance of the coroutine
// falls out of scope while waiting for an event.
//
// Socket used to for listen for queries. Created in the
// constructor and stored in a shared_ptr because socket objects
// are not copyable.
boost::shared_ptr<asio::ip::udp::socket> socket_;
// The ASIO-internal endpoint object representing the client
std::auto_ptr<asio::ip::udp::endpoint> sender_;
// \c IOMessage and \c Message objects to be passed to the
// DNS lookup and answer providers
std::auto_ptr<asiolink::IOMessage> io_message_;
// The original query as sent by the client
isc::dns::MessagePtr query_message_;
// The response message we are building
isc::dns::MessagePtr answer_message_;
// The buffer into which the response is written
isc::util::OutputBufferPtr respbuf_;
// The buffer into which the query packet is written
boost::shared_array<char> data_;
// State information that is entirely internal to a given instance
// of the coroutine can be declared here.
size_t bytes_;
bool done_;
// Callback functions provided by the caller
const SimpleCallback* checkin_callback_;
const DNSLookup* lookup_callback_;
const DNSAnswer* answer_callback_;
std::auto_ptr<IOEndpoint> peer_;
std::auto_ptr<IOSocket> iosock_;
};
/// The following functions implement the \c UDPServer class.
///
/// The constructor. It just creates new internal state object
/// and lets it handle the initialization.
UDPServer::UDPServer(io_service& io_service, const ip::address& addr,
const uint16_t port, SimpleCallback* checkin, DNSLookup* lookup,
DNSAnswer* answer) :
data_(new Data(io_service, addr, port, checkin, lookup, answer))
{ }
/// The function operator is implemented with the "stackless coroutine"
/// pattern; see internal/coroutine.h for details.
void
UDPServer::operator()(error_code ec, size_t length) {
/// Because the coroutine reentry block is implemented as
/// a switch statement, inline variable declarations are not
/// permitted. Certain variables used below can be declared here.
CORO_REENTER (this) {
do {
/*
* This is preparation for receiving a packet. We get a new
* state object for the lifetime of the next packet to come.
* It allocates the buffers to receive data into.
*/
data_.reset(new Data(*data_));
do {
// Begin an asynchronous receive, then yield.
// When the receive event is posted, the coroutine
// will resume immediately after this point.
CORO_YIELD data_->socket_->async_receive_from(
buffer(data_->data_.get(), MAX_LENGTH), *data_->sender_,
*this);
// Abort on fatal errors
// TODO: add log
if (ec) {
using namespace asio::error;
if (ec.value() != would_block && ec.value() != try_again &&
ec.value() != interrupted) {
return;
}
}
} while (ec || length == 0);
data_->bytes_ = length;
/*
* We fork the coroutine now. One (the child) will keep
* the current state and handle the packet, then die and
* drop ownership of the state. The other (parent) will just
* go into the loop again and replace the current state with
* a new one for a new object.
*
* Actually, both of the coroutines will be a copy of this
* one, but that's just internal implementation detail.
*/
CORO_FORK data_->io_.post(UDPServer(*this));
} while (is_parent());
// Create an \c IOMessage object to store the query.
//
// (XXX: It would be good to write a factory function
// that would quickly generate an IOMessage object without
// all these calls to "new".)
data_->peer_.reset(new UDPEndpoint(*data_->sender_));
// The UDP socket class has been extended with asynchronous functions
// and takes as a template parameter a completion callback class. As
// UDPServer does not use these extended functions (only those defined
// in the IOSocket base class) - but needs a UDPSocket to get hold of
// the underlying Boost UDP socket - DummyIOCallback is used. This
// provides the appropriate operator() but is otherwise functionless.
data_->iosock_.reset(
new UDPSocket<DummyIOCallback>(*data_->socket_));
data_->io_message_.reset(new IOMessage(data_->data_.get(),
data_->bytes_, *data_->iosock_, *data_->peer_));
// Perform any necessary operations prior to processing an incoming
// query (e.g., checking for queued configuration messages).
//
// (XXX: it may be a performance issue to check in for every single
// incoming query; we may wish to throttle this in the future.)
if (data_->checkin_callback_ != NULL) {
(*data_->checkin_callback_)(*data_->io_message_);
}
// If we don't have a DNS Lookup provider, there's no point in
// continuing; we exit the coroutine permanently.
if (data_->lookup_callback_ == NULL) {
CORO_YIELD return;
}
// Instantiate objects that will be needed by the
// asynchronous DNS lookup and/or by the send call.
data_->respbuf_.reset(new OutputBuffer(0));
data_->query_message_.reset(new Message(Message::PARSE));
data_->answer_message_.reset(new Message(Message::RENDER));
// Schedule a DNS lookup, and yield. When the lookup is
// finished, the coroutine will resume immediately after
// this point.
CORO_YIELD data_->io_.post(AsyncLookup<UDPServer>(*this));
// The 'done_' flag indicates whether we have an answer
// to send back. If not, exit the coroutine permanently.
if (!data_->done_) {
CORO_YIELD return;
}
// Call the DNS answer provider to render the answer into
// wire format
(*data_->answer_callback_)(*data_->io_message_, data_->query_message_,
data_->answer_message_, data_->respbuf_);
// Begin an asynchronous send, and then yield. When the
// send completes, we will resume immediately after this point
// (though we have nothing further to do, so the coroutine
// will simply exit at that time).
CORO_YIELD data_->socket_->async_send_to(
buffer(data_->respbuf_->getData(), data_->respbuf_->getLength()),
*data_->sender_, *this);
}
}
/// Call the DNS lookup provider. (Expected to be called by the
/// AsyncLookup<UDPServer> handler.)
void
UDPServer::asyncLookup() {
(*data_->lookup_callback_)(*data_->io_message_,
data_->query_message_, data_->answer_message_, data_->respbuf_, this);
}
/// Stop the UDPServer
void
UDPServer::stop() {
/// Using close instead of cancel, because cancel
/// will only cancel the asynchornized event already submitted
/// to io service, the events post to io service after
/// cancel still can be scheduled by io service, if
/// the socket is cloesed, all the asynchronized event
/// for it won't be scheduled by io service not matter it is
/// submit to io serice before or after close call. And we will
//. get bad_descriptor error
data_->socket_->close();
}
/// Post this coroutine on the ASIO service queue so that it will
/// resume processing where it left off. The 'done' parameter indicates
/// whether there is an answer to return to the client.
void
UDPServer::resume(const bool done) {
data_->done_ = done;
data_->io_.post(*this);
}
bool
UDPServer::hasAnswer() {
return (data_->done_);
}
} // namespace asiolink
|