From: Michal 'vorner' Vaner Date: Wed, 22 Feb 2012 15:25:56 +0000 (+0100) Subject: ]1599] Changed the server to be synchronous X-Git-Tag: trac2351_base~226^2~116^2~163^2~5 X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=0c0161c6e555e2c8b33b3291043727bc07324dc2;p=thirdparty%2Fkea.git ]1599] Changed the server to be synchronous It compiles, but has some problems in tests. --- diff --git a/src/lib/asiodns/sync_udp_server.cc b/src/lib/asiodns/sync_udp_server.cc index d2a978205b..4974882723 100644 --- a/src/lib/asiodns/sync_udp_server.cc +++ b/src/lib/asiodns/sync_udp_server.cc @@ -1,4 +1,4 @@ -// Copyright (C) 2011 Internet Systems Consortium, Inc. ("ISC") +// Copyright (C) 2012 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 @@ -12,312 +12,162 @@ // OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR // PERFORMANCE OF THIS SOFTWARE. -#include -#include -#include // for some IPC/network system calls -#include - -#include - #include -#include - #include #include + +#include "sync_udp_server.h" +#include "logger.h" + +#include +#include + #include #include #include -#include "sync_udp_server.h" -#include "logger.h" -#include +#include -using namespace asio; -using asio::ip::udp; -using isc::log::dlog; +#include +#include +#include // for some IPC/network system calls +#include using namespace std; -using namespace isc::dns; -using namespace isc::util; +using namespace boost; using namespace isc::asiolink; namespace isc { namespace asiodns { -/* - * 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 SyncUDPServer::Data { - /* - * Constructors 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)); - } - Data(io_service& io_service, int fd, int af, SimpleCallback* checkin, - DNSLookup* lookup, DNSAnswer* answer) : - io_(io_service), done_(false), - checkin_callback_(checkin),lookup_callback_(lookup), - answer_callback_(answer) - { - if (af != AF_INET && af != AF_INET6) { - isc_throw(InvalidParameter, "Address family must be either AF_INET " - "or AF_INET6, not " << af); - } - LOG_DEBUG(logger, DBGLVL_TRACE_BASIC, ASIODNS_FD_ADD_UDP).arg(fd); - try { - socket_.reset(new udp::socket(io_service)); - socket_->assign(af == AF_INET6 ? udp::v6() : udp::v4(), fd); - } catch (const std::exception& exception) { - // Whatever the thing throws, it is something from ASIO and we - // convert it - isc_throw(IOError, exception.what()); - } +SyncUDPServer::SyncUDPServer(asio::io_service& io_service, + const asio::ip::address& addr, + const uint16_t port, + asiolink::SimpleCallback* checkin, + DNSLookup* lookup, DNSAnswer* answer) : + io_(io_service), checkin_callback_(checkin), lookup_callback_(lookup), + answer_callback_(answer) +{ + // We must use different instantiations for v4 and v6; + // otherwise ASIO will bind to both + asio::ip::udp proto = addr.is_v4() ? asio::ip::udp::v4() : + asio::ip::udp::v6(); + socket_.reset(new asio::ip::udp::socket(io_service, proto)); + socket_->set_option(asio::socket_base::reuse_address(true)); + if (addr.is_v6()) { + socket_->set_option(asio::ip::v6_only(true)); } + socket_->bind(asio::ip::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()); +SyncUDPServer::SyncUDPServer(asio::io_service& io_service, const int fd, + const int af, asiolink::SimpleCallback* checkin, + DNSLookup* lookup, DNSAnswer* answer) : + io_(io_service), checkin_callback_(checkin), lookup_callback_(lookup), + answer_callback_(answer) +{ + if (af != AF_INET && af != AF_INET6) { + isc_throw(InvalidParameter, "Address family must be either AF_INET " + "or AF_INET6, not " << af); } + LOG_DEBUG(logger, DBGLVL_TRACE_BASIC, ASIODNS_FD_ADD_UDP).arg(fd); + try { + socket_.reset(new asio::ip::udp::socket(io_service)); + socket_->assign(af == AF_INET6 ? asio::ip::udp::v6() : + asio::ip::udp::v4(), fd); + } catch (const std::exception& exception) { + // Whatever the thing throws, it is something from ASIO and we + // convert it + isc_throw(IOError, exception.what()); + } +} - // 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 socket_; - - // The ASIO-internal endpoint object representing the client - std::auto_ptr sender_; - - // \c IOMessage and \c Message objects to be passed to the - // DNS lookup and answer providers - std::auto_ptr 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 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 peer_; - std::auto_ptr iosock_; -}; - -/// The following functions implement the \c UDPServer class. -/// -/// The constructor. It just creates new internal state object -/// and lets it handle the initialization. -SyncUDPServer::SyncUDPServer(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)) -{ } - -SyncUDPServer::SyncUDPServer(io_service& io_service, int fd, int af, - SimpleCallback* checkin, DNSLookup* lookup, - DNSAnswer* answer) : - data_(new Data(io_service, fd, af, checkin, lookup, answer)) -{ } - -/// The function operator is implemented with the "stackless coroutine" -/// pattern; see internal/coroutine.h for details. void -SyncUDPServer::operator()(asio::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(SyncUDPServer(*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(*data_->socket_)); - - data_->io_message_.reset(new IOMessage(data_->data_.get(), - data_->bytes_, *data_->iosock_, *data_->peer_)); +SyncUDPServer::scheduleRead() { + socket_->async_receive_from(asio::buffer(data_, MAX_LENGTH), sender_, + boost::bind(&SyncUDPServer::handleRead, this, + _1, _2)); +} - // 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_); +void +SyncUDPServer::handleRead(const asio::error_code& ec, const size_t length) { + // Abort on fatal errors + if (ec) { + using namespace asio::error; + if (ec.value() != would_block && ec.value() != try_again && + ec.value() != interrupted) { + return; } + } + // Some kind of interrupt, spurious wakeup, or like that. Just try reading + // again. + if (ec || length == 0) { + scheduleRead(); + return; + } + // OK, we have a real packet of data. Let's dig into it! + + // XXX: This is taken (and ported) from UDPSocket class. What the hell does + // it really mean? + + // 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. + UDPSocket socket(*socket_); + UDPEndpoint endpoint(sender_); + IOMessage message(data_, length, socket, endpoint); + if (checkin_callback_ != NULL) { + (*checkin_callback_)(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; - } + // If we don't have a DNS Lookup provider, there's no point in + // continuing; we exit the coroutine permanently. + if (lookup_callback_ == NULL) { + scheduleRead(); + 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)); + // TODO: Can any of these be put to the object and reused? + isc::util::OutputBufferPtr output(new isc::util::OutputBuffer(0)); + isc::dns::MessagePtr + query(new isc::dns::Message(isc::dns::Message::PARSE)); + isc::dns::MessagePtr + answer(new isc::dns::Message(isc::dns::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(*this)); + // Mark that we don't have an answer yet. + done_ = false; + resume_called_ = false; - // 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 actual lookup + (*lookup_callback_)(message, query, answer, output, this); - // 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_); + if (!resume_called_) { + isc_throw(isc::Unexpected, + "No resume called from the lookup callback"); + } + + if (done_) { + // Good, there's an answer. + // Call the answer callback to render it. + (*answer_callback_)(message, query, answer, output); - // 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); + socket_->send_to(asio::buffer(output->getData(), output->getLength()), + sender_); } + + // And schedule handling another socket. + scheduleRead(); } -/// Call the DNS lookup provider. (Expected to be called by the -/// AsyncLookup handler.) void -SyncUDPServer::asyncLookup() { - (*data_->lookup_callback_)(*data_->io_message_, - data_->query_message_, data_->answer_message_, data_->respbuf_, this); +SyncUDPServer::operator()(asio::error_code, size_t) { + // To start the server, we just schedule reading of data when they + // arrive. + scheduleRead(); } /// Stop the UDPServer @@ -331,7 +181,7 @@ SyncUDPServer::stop() { /// 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(); + socket_->close(); } /// Post this coroutine on the ASIO service queue so that it will @@ -339,13 +189,13 @@ SyncUDPServer::stop() { /// whether there is an answer to return to the client. void SyncUDPServer::resume(const bool done) { - data_->done_ = done; - data_->io_.post(*this); + resume_called_ = true; + done_ = done; } bool SyncUDPServer::hasAnswer() { - return (data_->done_); + return (done_); } } // namespace asiodns diff --git a/src/lib/asiodns/sync_udp_server.h b/src/lib/asiodns/sync_udp_server.h index 2cf70efbfd..57aecdfa58 100644 --- a/src/lib/asiodns/sync_udp_server.h +++ b/src/lib/asiodns/sync_udp_server.h @@ -1,4 +1,4 @@ -// Copyright (C) 2011 Internet Systems Consortium, Inc. ("ISC") +// Copyright (C) 2012 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 @@ -19,25 +19,24 @@ #error "asio.hpp must be included before including this, see asiolink.h as to why" #endif +#include "dns_answer.h" +#include "dns_lookup.h" +#include "dns_server.h" + #include -#include -#include -#include +#include -#include +#include namespace isc { namespace asiodns { -// -// Asynchronous UDP server coroutine -// +/// \brief An UDP server that doesn't asynchronous lookup handlers. /// -/// \brief This class implements the coroutine to handle UDP -/// DNS query event. As such, it is both a \c DNSServer and -/// a \c coroutine -/// -class SyncUDPServer : public virtual DNSServer, public virtual coroutine { +/// That means, the lookup handler must provide the answer right away. +/// This allows for implementation with less overhead, compared with +/// the UDPClass. +class SyncUDPServer : public DNSServer, public boost::noncopyable { public: /// \brief Constructor /// \param io_service the asio::io_service to work with @@ -47,10 +46,10 @@ public: /// \param lookup the callbackprovider for DNS lookup events /// \param answer the callbackprovider for DNS answer events explicit SyncUDPServer(asio::io_service& io_service, - const asio::ip::address& addr, const uint16_t port, - isc::asiolink::SimpleCallback* checkin = NULL, - DNSLookup* lookup = NULL, - DNSAnswer* answer = NULL); + const asio::ip::address& addr, const uint16_t port, + isc::asiolink::SimpleCallback* checkin = NULL, + DNSLookup* lookup = NULL, + DNSAnswer* answer = NULL); /// \brief Constructor /// \param io_service the asio::io_service to work with @@ -62,59 +61,64 @@ public: /// \throw isc::InvalidParameter if af is neither AF_INET nor AF_INET6 /// \throw isc::asiolink::IOError when a low-level error happens, like the /// fd is not a valid descriptor. - SyncUDPServer(asio::io_service& io_service, int fd, int af, - isc::asiolink::SimpleCallback* checkin = NULL, - DNSLookup* lookup = NULL, DNSAnswer* answer = NULL); + SyncUDPServer(asio::io_service& io_service, const int fd, const int af, + isc::asiolink::SimpleCallback* checkin = NULL, + DNSLookup* lookup = NULL, DNSAnswer* answer = NULL); - /// \brief The function operator - void operator()(asio::error_code ec = asio::error_code(), + /// \brief Start the SyncUDPServer. + /// + /// This is the function operator to keep interface with other server + /// classes. They need that because they're coroutines. + virtual void operator()(asio::error_code ec = asio::error_code(), size_t length = 0); /// \brief Calls the lookup callback - void asyncLookup(); + virtual void asyncLookup() { + isc_throw(Unexpected, + "SyncUDPServer doesn't support asyncLookup by design, use " + "UDPServer if you need it."); + } /// \brief Stop the running server /// \note once the server stopped, it can't restart - void stop(); + virtual void stop(); /// \brief Resume operation /// + /// Note that unlike other servers, this one expects it to be called + /// directly from the lookup callback. If it isn't, the server will + /// throw an Unexpected exception (probably to the event loop, which + /// would usually lead to termination of the program, but that's OK, + /// as it would be serious programmer error). + /// /// \param done Set this to true if the lookup action is done and /// we have an answer - void resume(const bool done); + virtual void resume(const bool done); /// \brief Check if we have an answer /// /// \return true if we have an answer - bool hasAnswer(); - - /// \brief Returns the coroutine state value - /// - /// \return the coroutine state value - int value() { return (get_value()); } + virtual bool hasAnswer(); /// \brief Clones the object /// /// \return a newly allocated copy of this object - DNSServer* clone() { - SyncUDPServer* s = new SyncUDPServer(*this); - return (s); + virtual DNSServer* clone() { + isc_throw(Unexpected, "SyncUDPServer can't be cloned."); } - private: - enum { MAX_LENGTH = 4096 }; - - /** - * \brief Internal state and data. - * - * We use the pimple design pattern, but not because we need to hide - * internal data. This class and whole header is for private use anyway. - * It turned out that SyncUDPServer is copied a lot, because it is a coroutine. - * This way the overhead of copying is lower, we copy only one shared - * pointer instead of about 10 of them. - */ - struct Data; - boost::shared_ptr data_; + static const size_t MAX_LENGTH = 4096; + uint8_t data_[MAX_LENGTH]; + std::auto_ptr socket_; + asio::io_service& io_; + asio::ip::udp::endpoint sender_; + const asiolink::SimpleCallback* checkin_callback_; + const DNSLookup* lookup_callback_; + const DNSAnswer* answer_callback_; + bool resume_called_, done_; + + void scheduleRead(); + void handleRead(const asio::error_code& ec, const size_t length); }; } // namespace asiodns diff --git a/src/lib/asiodns/tcp_server.h b/src/lib/asiodns/tcp_server.h index 8ae84f9f73..a75fddbf80 100644 --- a/src/lib/asiodns/tcp_server.h +++ b/src/lib/asiodns/tcp_server.h @@ -67,7 +67,6 @@ public: return (s); } - private: enum { MAX_LENGTH = 65535 }; static const size_t TCP_MESSAGE_LENGTHSIZE = 2;