tcp_server.cc 8.72 KB
Newer Older
1
// Copyright (C) 2011  Internet Systems Consortium, Inc. ("ISC")
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
//
// 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 <config.h>

17
#include <netinet/in.h>
18 19
#include <sys/socket.h>
#include <unistd.h>             // for some IPC/network system calls
20
#include <errno.h>
21

22
#include <boost/shared_array.hpp>
23

24
#include <log/dummylog.h>
25

26
#include <util/buffer.h>
27

28
#include <asio.hpp>
29
#include <asiolink/dummy_io_cb.h>
30 31 32
#include <asiolink/tcp_endpoint.h>
#include <asiolink/tcp_socket.h>
#include <asiolink/tcp_server.h>
33 34 35 36 37 38 39 40


using namespace asio;
using asio::ip::udp;
using asio::ip::tcp;

using namespace std;
using namespace isc::dns;
41
using namespace isc::util;
42 43

namespace asiolink {
44 45

/// The following functions implement the \c TCPServer class.
46 47
///
/// The constructor
48 49
TCPServer::TCPServer(io_service& io_service,
                     const ip::address& addr, const uint16_t port, 
50
                     const SimpleCallback* checkin,
51 52
                     const DNSLookup* lookup,
                     const DNSAnswer* answer) :
53
    io_(io_service), done_(false),
54 55
    checkin_callback_(checkin), lookup_callback_(lookup),
    answer_callback_(answer)
56 57 58 59
{
    tcp::endpoint endpoint(addr, port);
    acceptor_.reset(new tcp::acceptor(io_service));
    acceptor_->open(endpoint.protocol());
60
    // Set v6-only (we use a separate instantiation for v4,
61 62 63 64 65 66 67 68 69 70 71
    // otherwise asio will bind to both v4 and v6
    if (addr.is_v6()) {
        acceptor_->set_option(ip::v6_only(true));
    }
    acceptor_->set_option(tcp::acceptor::reuse_address(true));
    acceptor_->bind(endpoint);
    acceptor_->listen();
}

void
TCPServer::operator()(error_code ec, size_t length) {
72
    /// Because the coroutine reentry block is implemented as
73 74
    /// a switch statement, inline variable declarations are not
    /// permitted.  Certain variables used below can be declared here.
75

76
    boost::array<const_buffer,2> bufs;
Jelte Jansen's avatar
Jelte Jansen committed
77
    OutputBuffer lenbuf(TCP_MESSAGE_LENGTHSIZE);
78

79 80
    CORO_REENTER (this) {
        do {
81
            /// Create a socket to listen for connections
82
            socket_.reset(new tcp::socket(acceptor_->get_io_service()));
83

84
            /// Wait for new connections. In the event of non-fatal error,
85 86 87
            /// try again
            do {
                CORO_YIELD acceptor_->async_accept(*socket_, *this);
hanfeng's avatar
hanfeng committed
88

89 90
                // Abort on fatal errors
                // TODO: Log error?
hanfeng's avatar
hanfeng committed
91 92 93
                if (ec) {
                    using namespace asio::error;
                    if (ec.value() != would_block && ec.value() != try_again &&
94 95
                        ec.value() != connection_aborted &&
                        ec.value() != interrupted) {
hanfeng's avatar
hanfeng committed
96 97 98
                        return;
                    }
                }
99
            } while (ec);
100 101 102 103 104

            /// Fork the coroutine by creating a copy of this one and
            /// scheduling it on the ASIO service queue.  The parent
            /// will continue listening for DNS connections while the
            /// handles the one that has just arrived.
105 106
            CORO_FORK io_.post(TCPServer(*this));
        } while (is_parent());
107

108 109
        /// Instantiate the data buffer that will be used by the
        /// asynchronous read call.
110
        data_.reset(new char[MAX_LENGTH]);
111

112
        /// Read the message, in two parts.  First, the message length:
113
        CORO_YIELD async_read(*socket_, asio::buffer(data_.get(),
114
                              TCP_MESSAGE_LENGTHSIZE), *this);
115
        if (ec) {
116
            socket_->close();
117 118
            CORO_YIELD return;
        }
119

120 121
        /// Now read the message itself. (This is done in a different scope
        /// to allow inline variable declarations.)
122
        CORO_YIELD {
123
            InputBuffer dnsbuffer(data_.get(), length);
124 125 126 127
            uint16_t msglen = dnsbuffer.readUint16();
            async_read(*socket_, asio::buffer(data_.get(), msglen), *this);
        }

128
        if (ec) {
129
            socket_->close();
130 131 132
            CORO_YIELD return;
        }

133

134
        // Create an \c IOMessage object to store the query.
135 136 137 138
        //
        // (XXX: It would be good to write a factory function
        // that would quickly generate an IOMessage object without
        // all these calls to "new".)
139
        peer_.reset(new TCPEndpoint(socket_->remote_endpoint()));
Stephen Morris's avatar
Stephen Morris committed
140 141 142 143 144

        // The TCP socket class has been extended with asynchronous functions
        // and takes as a template parameter a completion callback class.  As
        // TCPServer does not use these extended functions (only those defined
        // in the IOSocket base class) - but needs a TCPSocket to get hold of
145 146 147
        // the underlying Boost TCP socket - DummyIOCallback is used.  This
        // provides the appropriate operator() but is otherwise functionless.
        iosock_.reset(new TCPSocket<DummyIOCallback>(*socket_));
148
        io_message_.reset(new IOMessage(data_.get(), length, *iosock_, *peer_));
149
        bytes_ = length;
150

151 152 153 154 155 156 157 158
        // Perform any necessary operations prior to processing the incoming
        // packet (e.g., checking for queued configuration messages).
        //
        // (XXX: it may be a performance issue to have this called for
        // every single incoming packet; we may wish to throttle it somehow
        // in the future.)
        if (checkin_callback_ != NULL) {
            (*checkin_callback_)(*io_message_);
159 160
        }

161 162
        // If we don't have a DNS Lookup provider, there's no point in
        // continuing; we exit the coroutine permanently.
163
        if (lookup_callback_ == NULL) {
164
            socket_->close();
165 166 167
            CORO_YIELD return;
        }

168 169
        // Reset or instantiate objects that will be needed by the
        // DNS lookup and the write call.
170
        respbuf_.reset(new OutputBuffer(0));
171
        query_message_.reset(new Message(Message::PARSE));
172
        answer_message_.reset(new Message(Message::RENDER));
173

174 175 176 177
        // Schedule a DNS lookup, and yield.  When the lookup is
        // finished, the coroutine will resume immediately after
        // this point.
        CORO_YIELD io_.post(AsyncLookup<TCPServer>(*this));
178

179 180
        // The 'done_' flag indicates whether we have an answer
        // to send back.  If not, exit the coroutine permanently.
181
        if (!done_) {
182 183 184
            // TODO: should we keep the connection open for a short time
            // to see if new requests come in?
            socket_->close();
185
            CORO_YIELD return;
186
        }
187

188 189 190
        if (ec) {
            CORO_YIELD return;
        }
191 192
        // Call the DNS answer provider to render the answer into
        // wire format
193 194
        (*answer_callback_)(*io_message_, query_message_,
                            answer_message_, respbuf_);
195

196
        // Set up the response, beginning with two length bytes.
Jelte Jansen's avatar
Jelte Jansen committed
197 198
        lenbuf.writeUint16(respbuf_->getLength());
        bufs[0] = buffer(lenbuf.getData(), lenbuf.getLength());
199
        bufs[1] = buffer(respbuf_->getData(), respbuf_->getLength());
200 201 202 203 204

        // 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).
205
        CORO_YIELD async_write(*socket_, bufs, *this);
206

207 208 209
        // TODO: should we keep the connection open for a short time
        // to see if new requests come in?
        socket_->close();
210 211 212
    }
}

213 214
/// Call the DNS lookup provider.  (Expected to be called by the
/// AsyncLookup<TCPServer> handler.)
215
void
216
TCPServer::asyncLookup() {
217 218
    (*lookup_callback_)(*io_message_, query_message_,
                        answer_message_, respbuf_, this);
219 220
}

221
void TCPServer::stop() {
222 223
    /// we use close instead of cancel, with the same reason
    /// with udp server stop, refer to the udp server code
224

225
    acceptor_->close();
226 227 228 229 230
    // User may stop the server even when it hasn't started to
    // run, in that that socket_ is empty
    if (socket_) {
        socket_->close();
    }
231
}
232 233 234
/// 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.
235
void
236 237
TCPServer::resume(const bool done) {
    done_ = done;
238 239 240
    io_.post(*this);
}

241 242
} // namespace asiolink