Refactoring of the asio_link module as preparation for adding a resolver.

1) The asio_link object no longer directly depends on AuthSrv*; instead
the caller must provide a pair of callback objects, CheckinProvider and
DNSProvider, which are called at appropriate times to check for outstanding
configuration messages and to process a DNS message.

2) In hopes of making it less painful to write the ASIO handlers
when we add the code to send requests to upstream authoritative
servers, I rewrote the TCPServer and UDPServer classes to use
the "stackless coroutine" pattern described at:
http://blog.think-async.com/2010/03/potted-guide-to-stackless-coroutines.html

The resulting ASIO code should be functionally identical to the
previous code, but it is shorter and (IMHO) easier to read:
instead of several different asynchronous response handlers,
there's a single function for TCP and another for UDP, and the
I/O operations are all laid out in logical order.

Next step will be to move asio_link into src/lib, but I'm leaving
it here for now to make it easier to read the diff.


git-svn-id: svn://bind10.isc.org/svn/bind10/branches/trac327@2934 e5f2f494-b856-4b98-b285-d166d9295462

src/bin/auth/asio_link.cc

@@ -32,8 +32,8 @@
 
 #include <asio_link.h>
 
-#include <auth/auth_srv.h>
-#include <auth/common.h>
+#include "coroutine.h"
+#include "yield.h"
 
 using namespace asio;
 using asio::ip::udp;
@@ -43,6 +43,26 @@ using namespace std;
 using namespace isc::dns;
 
 namespace asio_link {
+
+// Constructors and destructors for the callback provider base classes.
+DNSProvider::DNSProvider() {}
+DNSProvider::~DNSProvider() {}
+
+bool
+DNSProvider::operator()(const IOMessage& io_message,
+                        isc::dns::Message& dns_message,
+                        isc::dns::MessageRenderer& renderer) const
+{
+    return (false);
+}
+
+CheckinProvider::CheckinProvider() {}
+CheckinProvider::~CheckinProvider() {}
+
+void
+CheckinProvider::operator()(void) const {}
+
+
 IOAddress::IOAddress(const string& address_str)
     // XXX: we cannot simply construct the address in the initialization list
     // because we'd like to throw our own exception on failure.
@@ -179,289 +199,225 @@ IOMessage::IOMessage(const void* data, const size_t data_size,
 {}
 
 //
-// Helper classes for asynchronous I/O using asio
+// Asynchronous TCP server coroutine
 //
-class TCPClient {
+class TCPServer : public coroutine {
 public:
-    TCPClient(AuthSrv* auth_server, io_service& io_service) :
-        auth_server_(auth_server),
-        socket_(io_service),
-        io_socket_(socket_),
-        response_buffer_(0),
-        responselen_buffer_(TCP_MESSAGE_LENGTHSIZE),
-        response_renderer_(response_buffer_),
-        dns_message_(Message::PARSE),
-        custom_callback_(NULL)
-    {}
-
-    void start() {
-        // Check for queued configuration commands
-        if (auth_server_ != NULL &&
-            auth_server_->configSession()->hasQueuedMsgs()) {
-            auth_server_->configSession()->checkCommand();
-        }
-        async_read(socket_, asio::buffer(data_, TCP_MESSAGE_LENGTHSIZE),
-                   boost::bind(&TCPClient::headerRead, this,
-                               placeholders::error,
-                               placeholders::bytes_transferred));
-    }
-
-    tcp::socket& getSocket() { return (socket_); }
-
-    void headerRead(const asio::error_code& error,
-                    size_t bytes_transferred)
+    explicit TCPServer(io_service& io_service,
+                       const ip::address& addr, const uint16_t port, 
+                       CheckinProvider* checkin = NULL,
+                       DNSProvider* process = NULL) :
+        checkin_callback_(checkin), dns_callback_(process)
     {
-        if (!error) {
-            InputBuffer dnsbuffer(data_, bytes_transferred);
-
-            uint16_t msglen = dnsbuffer.readUint16();
-            async_read(socket_, asio::buffer(data_, msglen),
-                       boost::bind(&TCPClient::requestRead, this,
-                                   placeholders::error,
-                                   placeholders::bytes_transferred));
-        } else {
-            delete this;
-        }
-    }
 
-    void requestRead(const asio::error_code& error,
-                     size_t bytes_transferred)
-    {
-        if (!error) {
-            const TCPEndpoint remote_endpoint(socket_.remote_endpoint());
-            const IOMessage io_message(data_, bytes_transferred, io_socket_,
-                                       remote_endpoint);
-            // currently, for testing purpose only
-            if (custom_callback_ != NULL) {
-                (*custom_callback_)(io_message);
-                start();
-                return;
-            }
-
-            if (auth_server_->processMessage(io_message, dns_message_,
-                                             response_renderer_)) {
-                responselen_buffer_.writeUint16(
-                    response_buffer_.getLength());
-                async_write(socket_,
-                            asio::buffer(
-                                responselen_buffer_.getData(),
-                                responselen_buffer_.getLength()),
-                            boost::bind(&TCPClient::responseWrite, this,
-                                        placeholders::error));
-            } else {
-                delete this;
-            }
-        } else {
-            delete this;
+        tcp::endpoint endpoint(addr, port);
+        acceptor_.reset(new tcp::acceptor(io_service));
+        acceptor_->open(endpoint.protocol());
+        // Set v6-only (we use a different instantiation for v4,
+        // 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 responseWrite(const asio::error_code& error) {
-        if (!error) {
-                async_write(socket_,
-                            asio::buffer(response_buffer_.getData(),
-                                         response_buffer_.getLength()),
-                            boost::bind(&TCPClient::handleWrite, this,
-                                        placeholders::error));
-        } else {
-            delete this;
+    void operator()(error_code ec = error_code(), size_t length = 0) {
+        if (ec) {
+            return;
         }
-    }
 
-    void handleWrite(const asio::error_code& error) {
-        if (!error) {
-            start();            // handle next request, if any.
-      } else {
-            delete this;
-      }
-    }
+        reenter (this) {
+            do {
+                socket_.reset(new tcp::socket(acceptor_->get_io_service()));
+                yield acceptor_->async_accept(*socket_, *this);
+                fork TCPServer(*this)();
+            } while (is_parent());
+
+            // Perform any necessary operations prior to processing
+            // an incoming packet (e.g., checking for queued
+            // configuration messages).
+            if (checkin_callback_ != NULL) {
+                (*checkin_callback_)();
+            }
 
-    // Currently this is for tests only
-    void setCallBack(const IOService::IOCallBack* callback) {
-        custom_callback_ = callback;
-    }
+            // Instantiate the data buffer that will be used by the
+            // asynchronous read calls.
+            // data_.reset(new boost::array<char, MAX_LENGTH>);
+            data_ = boost::shared_ptr<char>(new char[MAX_LENGTH]);
 
-private:
-    AuthSrv* auth_server_;
-    tcp::socket socket_;
-    TCPSocket io_socket_;
-    OutputBuffer response_buffer_;
-    OutputBuffer responselen_buffer_;
-    MessageRenderer response_renderer_;
-    Message dns_message_;
-    enum { MAX_LENGTH = 65535 };
-    static const size_t TCP_MESSAGE_LENGTHSIZE = 2;
-    char data_[MAX_LENGTH];
+            yield async_read(*socket_,
+                             asio::buffer(data_.get(), TCP_MESSAGE_LENGTHSIZE),
+                             *this);
 
-    // currently, for testing purpose only.
-    const IOService::IOCallBack* custom_callback_;
-};
+            yield {
+                InputBuffer dnsbuffer((const void *) data_.get(), length);
+                uint16_t msglen = dnsbuffer.readUint16();
+                async_read(*socket_, asio::buffer(data_.get(), msglen), *this);
+            }
 
-class TCPServer {
-public:
-    TCPServer(AuthSrv* auth_server, io_service& io_service,
-              const ip::address& addr, const uint16_t port) :
-        auth_server_(auth_server), io_service_(io_service),
-        acceptor_(io_service_), listening_(new TCPClient(auth_server_,
-                                                         io_service_)),
-        custom_callback_(NULL)
-    {
-        tcp::endpoint endpoint(addr, port);
-        acceptor_.open(endpoint.protocol());
-        // Set v6-only (we use a different instantiation for v4,
-        // 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();
-        acceptor_.async_accept(listening_->getSocket(),
-                               boost::bind(&TCPServer::handleAccept, this,
-                                           listening_, placeholders::error));
-    }
+            // Stop here if we don't have a DNS callback function
+            if (dns_callback_ == NULL) {
+                yield return;
+            }
 
-    ~TCPServer() { delete listening_; }
+            // Instantiate the objects that will be used by the
+            // asynchronous write calls.
+            dns_message_.reset(new Message(Message::PARSE));
+            response_.reset(new OutputBuffer(0));
+            responselen_buffer_.reset(new OutputBuffer(TCP_MESSAGE_LENGTHSIZE));
+            renderer_.reset(new MessageRenderer(*response_));
+            io_socket_.reset(new TCPSocket(*socket_));
+            io_endpoint_.reset(new TCPEndpoint(socket_->remote_endpoint()));
+            io_message_.reset(new IOMessage(data_.get(), length, *io_socket_,
+                                            *io_endpoint_));
+
+            // Process the DNS message
+            if (! (*dns_callback_)(*io_message_, *dns_message_, *renderer_)) {
+                yield return;
+            }
 
-    void handleAccept(TCPClient* new_client,
-                      const asio::error_code& error)
-    {
-        if (!error) {
-            assert(new_client == listening_);
-            new_client->setCallBack(custom_callback_);
-            new_client->start();
-            listening_ = new TCPClient(auth_server_, io_service_);
-            acceptor_.async_accept(listening_->getSocket(),
-                                   boost::bind(&TCPServer::handleAccept,
-                                               this, listening_,
-                                               placeholders::error));
-        } else {
-            delete new_client;
+            responselen_buffer_->writeUint16(response_->getLength());
+            yield async_write(*socket_,
+                              asio::buffer(responselen_buffer_->getData(),
+                                           responselen_buffer_->getLength()),
+                              *this);
+            yield async_write(*socket_,
+                              asio::buffer(response_->getData(),
+                                           response_->getLength()),
+                              *this);
         }
     }
 
-    // Currently this is for tests only
-    void setCallBack(const IOService::IOCallBack* callback) {
-        custom_callback_ = callback;
-    }
-
 private:
-    AuthSrv* auth_server_;
-    io_service& io_service_;
-    tcp::acceptor acceptor_;
-    TCPClient* listening_;
+    enum { MAX_LENGTH = 65535 };
+    static const size_t TCP_MESSAGE_LENGTHSIZE = 2;
 
-    // currently, for testing purpose only.
-    const IOService::IOCallBack* custom_callback_;
+    // All class member variables which are expected to persist across
+    // multiple invocations of the coroutine must be declared here as
+    // shared pointers, and instantiated in the constructor or in the
+    // coroutine itself.  When the coroutine is deleted, its destructor
+    // will free the memory.
+    boost::shared_ptr<tcp::acceptor> acceptor_;
+    boost::shared_ptr<tcp::socket> socket_;
+    boost::shared_ptr<OutputBuffer> response_;
+    boost::shared_ptr<OutputBuffer> responselen_buffer_;
+    boost::shared_ptr<MessageRenderer> renderer_;
+    boost::shared_ptr<Message> dns_message_;
+    boost::shared_ptr<IOMessage> io_message_;
+    boost::shared_ptr<TCPSocket> io_socket_;
+    boost::shared_ptr<TCPEndpoint> io_endpoint_;
+    boost::shared_ptr<char> data_;
+
+    // Callbacks
+    const CheckinProvider* checkin_callback_;
+    const DNSProvider* dns_callback_;
 };
 
-class UDPServer {
+//
+// Asynchronous UDP server coroutine
+//
+class UDPServer : public coroutine {
 public:
-    UDPServer(AuthSrv* auth_server, io_service& io_service,
-              const ip::address& addr, const uint16_t port) :
-        auth_server_(auth_server),
-        io_service_(io_service),
-        socket_(io_service, addr.is_v6() ? udp::v6() : udp::v4()),
-        io_socket_(socket_),
-        response_buffer_(0),
-        response_renderer_(response_buffer_),
-        dns_message_(Message::PARSE),
-        custom_callback_(NULL)
+    explicit UDPServer(io_service& io_service,
+                       const ip::address& addr, const uint16_t port,
+                       CheckinProvider* checkin = NULL,
+                       DNSProvider* process = NULL) :
+        checkin_callback_(checkin), dns_callback_(process)
     {
-        socket_.set_option(socket_base::reuse_address(true));
-        // Set v6-only (we use a different instantiation for v4,
+        // Wwe use a different instantiation for v4,
         // otherwise asio will bind to both v4 and v6
         if (addr.is_v6()) {
-            socket_.set_option(asio::ip::v6_only(true));
-            socket_.bind(udp::endpoint(addr, port));
+            socket_.reset(new udp::socket(io_service, udp::v6()));
+            socket_->set_option(socket_base::reuse_address(true));
+            socket_->set_option(asio::ip::v6_only(true));
+            socket_->bind(udp::endpoint(udp::v6(), port));
         } else {
-            socket_.bind(udp::endpoint(addr, port));
+            socket_.reset(new udp::socket(io_service, udp::v4()));
+            socket_->set_option(socket_base::reuse_address(true));
+            socket_->bind(udp::endpoint(udp::v6(), port));
         }
-        startReceive();
     }
 
-    void handleRequest(const asio::error_code& error,
-                       size_t bytes_recvd)
-    {
-        // Check for queued configuration commands
-        if (auth_server_ != NULL &&
-            auth_server_->configSession()->hasQueuedMsgs()) {
-            auth_server_->configSession()->checkCommand();
-        }
-        if (!error && bytes_recvd > 0) {
-            const UDPEndpoint remote_endpoint(sender_endpoint_);
-            const IOMessage io_message(data_, bytes_recvd, io_socket_,
-                                       remote_endpoint);
-            // currently, for testing purpose only
-            if (custom_callback_ != NULL) {
-                (*custom_callback_)(io_message);
-                startReceive();
-                return;
+    void operator()(error_code ec = error_code(), size_t length = 0) {
+        reenter (this) for (;;) {
+            // Instantiate the data buffer that will be used by the
+            // asynchronous read calls.
+            // data_.reset(new boost::array<char, MAX_LENGTH>);
+            data_ = boost::shared_ptr<char>(new char[MAX_LENGTH]);
+            sender_.reset(new udp::endpoint());
+            yield socket_->async_receive_from(asio::buffer(data_.get(),
+                                                           MAX_LENGTH),
+                                              *sender_, *this);
+            if (ec || length == 0) {
+                yield continue;
             }
 
-            dns_message_.clear(Message::PARSE);
-            response_renderer_.clear();
-            if (auth_server_->processMessage(io_message, dns_message_,
-                                             response_renderer_)) {
-                socket_.async_send_to(
-                    asio::buffer(response_buffer_.getData(),
-                                        response_buffer_.getLength()),
-                    sender_endpoint_,
-                    boost::bind(&UDPServer::sendCompleted,
-                                this,
-                                placeholders::error,
-                                placeholders::bytes_transferred));
-            } else {
-                startReceive();
+            bytes_ = length;
+            fork UDPServer(*this)();
+            if (is_child()) {
+                // Perform any necessary operations prior to processing
+                // an incoming packet (e.g., checking for queued
+                // configuration messages).
+                if (checkin_callback_ != NULL) {
+                    (*checkin_callback_)();
+                }
+
+                // Stop here if we don't have a DNS callback function
+                if (dns_callback_ == NULL) {
+                    yield return;
+                }
+    
+                // Instantiate the objects that will be used by the
+                // asynchronous write calls.
+                dns_message_.reset(new Message(Message::PARSE));
+                response_.reset(new OutputBuffer(0));
+                renderer_.reset(new MessageRenderer(*response_));
+                io_socket_.reset(new UDPSocket(*socket_));
+                io_endpoint_.reset(new UDPEndpoint(*sender_));
+                io_message_.reset(new IOMessage(data_.get(), bytes_,
+                                                *io_socket_,
+                                                *io_endpoint_));
+
+                // Process the DNS message
+                if (! (*dns_callback_)(*io_message_, *dns_message_, *renderer_))
+                {
+                    yield return;
+                }
+
+                yield socket_->async_send_to(asio::buffer(response_->getData(),
+                                                        response_->getLength()),
+                                             *sender_, *this);
             }
-        } else {
-            startReceive();
         }
     }
 
-    void sendCompleted(const asio::error_code& error UNUSED_PARAM,
-                       size_t bytes_sent UNUSED_PARAM)
-    {
-        // Even if error occurred there's nothing to do.  Simply handle
-        // the next request.
-        startReceive();
-    }
-
-    // Currently this is for tests only
-    void setCallBack(const IOService::IOCallBack* callback) {
-        custom_callback_ = callback;
-    }
-private:
-    void startReceive() {
-        socket_.async_receive_from(
-            asio::buffer(data_, MAX_LENGTH), sender_endpoint_,
-            boost::bind(&UDPServer::handleRequest, this,
-                        placeholders::error,
-                        placeholders::bytes_transferred));
-    }
-
 private:
-    AuthSrv* auth_server_;
-    io_service& io_service_;
-    udp::socket socket_;
-    UDPSocket io_socket_;
-    OutputBuffer response_buffer_;
-    MessageRenderer response_renderer_;
-    Message dns_message_;
-    udp::endpoint sender_endpoint_;
     enum { MAX_LENGTH = 4096 };
-    char data_[MAX_LENGTH];
 
-    // currently, for testing purpose only.
-    const IOService::IOCallBack* custom_callback_;
+    boost::shared_ptr<udp::socket> socket_;
+    boost::shared_ptr<udp::endpoint> sender_;
+    boost::shared_ptr<UDPEndpoint> io_endpoint_;
+    boost::shared_ptr<OutputBuffer> response_;
+    boost::shared_ptr<MessageRenderer> renderer_;
+    boost::shared_ptr<Message> dns_message_;
+    boost::shared_ptr<IOMessage> io_message_;
+    boost::shared_ptr<UDPSocket> io_socket_;
+    boost::shared_ptr<char> data_;
+    size_t bytes_;
+
+    // Callbacks
+    const CheckinProvider* checkin_callback_;
+    const DNSProvider* dns_callback_;
 };
 
 class IOServiceImpl {
 public:
-    IOServiceImpl(AuthSrv* auth_server, const char& port,
-                  const ip::address* v4addr, const ip::address* v6addr);
+    IOServiceImpl(const char& port,
+                  const ip::address* v4addr, const ip::address* v6addr,
+                  CheckinProvider* checkin, DNSProvider* process);
     asio::io_service io_service_;
-    AuthSrv* auth_server_;
 
     typedef boost::shared_ptr<UDPServer> UDPServerPtr;
     typedef boost::shared_ptr<TCPServer> TCPServerPtr;
@@ -469,15 +425,12 @@ public:
     UDPServerPtr udp6_server_;
     TCPServerPtr tcp4_server_;
     TCPServerPtr tcp6_server_;
-
-    // This member is used only for testing at the moment.
-    IOService::IOCallBack callback_;
 };
 
-IOServiceImpl::IOServiceImpl(AuthSrv* auth_server, const char& port,
+IOServiceImpl::IOServiceImpl(const char& port,
                              const ip::address* const v4addr,
-                             const ip::address* const v6addr) :
-    auth_server_(auth_server),
+                             const ip::address* const v6addr,
+                             CheckinProvider* checkin, DNSProvider* process) :
     udp4_server_(UDPServerPtr()), udp6_server_(UDPServerPtr()),
     tcp4_server_(TCPServerPtr()), tcp6_server_(TCPServerPtr())
 {
@@ -500,16 +453,24 @@ IOServiceImpl::IOServiceImpl(AuthSrv* auth_server, const char& port,
 
     try {
         if (v4addr != NULL) {
-            udp4_server_ = UDPServerPtr(new UDPServer(auth_server, io_service_,
-                                                      *v4addr, portnum));
-            tcp4_server_ = TCPServerPtr(new TCPServer(auth_server, io_service_,
-                                                      *v4addr, portnum));
+            udp4_server_ = UDPServerPtr(new UDPServer(io_service_,
+                                                      *v4addr, portnum,
+                                                      checkin, process));
+            (*udp4_server_)();
+            tcp4_server_ = TCPServerPtr(new TCPServer(io_service_,
+                                                      *v4addr, portnum,
+                                                      checkin, process));
+            (*tcp4_server_)();
         }
         if (v6addr != NULL) {
-            udp6_server_ = UDPServerPtr(new UDPServer(auth_server, io_service_,
-                                                      *v6addr, portnum));
-            tcp6_server_ = TCPServerPtr(new TCPServer(auth_server, io_service_,
-                                                      *v6addr, portnum));
+            udp6_server_ = UDPServerPtr(new UDPServer(io_service_,
+                                                      *v6addr, portnum,
+                                                      checkin, process));
+            (*udp6_server_)();
+            tcp6_server_ = TCPServerPtr(new TCPServer(io_service_,
+                                                      *v6addr, portnum,
+                                                      checkin, process));
+            (*tcp6_server_)();
         }
     } catch (const asio::system_error& err) {
         // We need to catch and convert any ASIO level exceptions.
@@ -520,8 +481,8 @@ IOServiceImpl::IOServiceImpl(AuthSrv* auth_server, const char& port,
     }
 }
 
-IOService::IOService(AuthSrv* auth_server, const char& port,
-                     const char& address) :
+IOService::IOService(const char& port, const char& address,
+                     CheckinProvider* checkin, DNSProvider* process) :
     impl_(NULL)
 {
     error_code err;
@@ -531,20 +492,23 @@ IOService::IOService(AuthSrv* auth_server, const char& port,
                   << err.message());
     }
 
-    impl_ = new IOServiceImpl(auth_server, port,
+    impl_ = new IOServiceImpl(port,
                               addr.is_v4() ? &addr : NULL,
-                              addr.is_v6() ? &addr : NULL);
+                              addr.is_v6() ? &addr : NULL,
+                              checkin, process);
 }
 
-IOService::IOService(AuthSrv* auth_server, const char& port,
-                     const bool use_ipv4, const bool use_ipv6) :
+IOService::IOService(const char& port,
+                     const bool use_ipv4, const bool use_ipv6,
+                     CheckinProvider* checkin, DNSProvider* process) :
     impl_(NULL)
 {
     const ip::address v4addr_any = ip::address(ip::address_v4::any());
     const ip::address* const v4addrp = use_ipv4 ? &v4addr_any : NULL; 
     const ip::address v6addr_any = ip::address(ip::address_v6::any());
     const ip::address* const v6addrp = use_ipv6 ? &v6addr_any : NULL;
-    impl_ = new IOServiceImpl(auth_server, port, v4addrp, v6addrp);
+    impl_ = new IOServiceImpl(port, v4addrp, v6addrp,
+                              checkin, process);
 }
 
 IOService::~IOService() {
@@ -566,20 +530,4 @@ IOService::get_io_service() {
     return (impl_->io_service_);
 }
 
-void
-IOService::setCallBack(const IOCallBack callback) {
-    impl_->callback_ = callback;
-    if (impl_->udp4_server_ != NULL) {
-        impl_->udp4_server_->setCallBack(&impl_->callback_);
-    }
-    if (impl_->udp6_server_ != NULL) {
-        impl_->udp6_server_->setCallBack(&impl_->callback_);
-    }
-    if (impl_->tcp4_server_ != NULL) {
-        impl_->tcp4_server_->setCallBack(&impl_->callback_);
-    }
-    if (impl_->tcp6_server_ != NULL) {
-        impl_->tcp6_server_->setCallBack(&impl_->callback_);
-    }
-}
 }

src/bin/auth/asio_link.h

@@ -28,6 +28,9 @@
 
 #include <boost/function.hpp>
 
+#include <dns/message.h>
+#include <dns/messagerenderer.h>
+
 #include <exceptions/exceptions.h>
 
 namespace asio {
@@ -35,8 +38,6 @@ namespace asio {
 class io_service;
 }
 
-class AuthSrv;
-
 /// \namespace asio_link
 /// \brief A wrapper interface for the ASIO library.
 ///
@@ -372,6 +373,71 @@ private:
     const IOEndpoint& remote_endpoint_;
 };
 
+/// \brief The \c DNSProvider class is an abstract base class for a DNS
+/// provider function.
+///
+/// Specific derived class implementations are hidden within the
+/// implementation.  Instances of the derived classes can be called
+/// as functions via the operator() interface.  Pointers to these
+/// instances can then be provided to the \c IOService class
+/// via its constructor.
+class DNSProvider {
+    ///
+    /// \name Constructors and Destructor
+    ///
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private, making this class non-copyable.
+    //@{
+private:
+    DNSProvider(const DNSProvider& source);
+    DNSProvider& operator=(const DNSProvider& source);
+protected:
+    /// \brief The default constructor.
+    ///
+    /// This is intentionally defined as \c protected as this base class
+    /// should never be instantiated (except as part of a derived class).
+    DNSProvider();
+public:
+    /// \brief The destructor
+    virtual ~DNSProvider();
+    //@}
+    virtual bool operator()(const IOMessage& io_message,
+                            isc::dns::Message& dns_message,
+                            isc::dns::MessageRenderer& renderer) const;
+};
+
+/// \brief The \c CheckinProvider class is an abstract base class for a
+/// checkin function.
+///
+/// Specific derived class implementations are hidden within the
+/// implementation.  Instances of the derived classes can be called
+/// as functions via the operator() interface.  Pointers to these
+/// instances can then be provided to the \c IOService class
+/// via its constructor.
+class CheckinProvider {
+    ///
+    /// \name Constructors and Destructor
+    ///
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private, making this class non-copyable.
+    //@{
+private:
+    CheckinProvider(const CheckinProvider& source);
+    CheckinProvider& operator=(const CheckinProvider& source);
+protected:
+    /// \brief The default constructor.
+    ///
+    /// This is intentionally defined as \c protected as this base class
+    /// should never be instantiated (except as part of a derived class).
+    CheckinProvider();
+    //@}
+public:
+    /// \brief The destructor
+    virtual ~CheckinProvider();
+    //@}
+    virtual void operator()(void) const;
+};
+
 /// \brief The \c IOService class is a wrapper for the ASIO \c io_service
 /// class.
 ///
@@ -395,15 +461,16 @@ private:
 public:
     /// \brief The constructor with a specific IP address and port on which
     /// the services listen on.
-    IOService(AuthSrv* auth_server, const char& port, const char& address);
+    IOService(const char& port, const char& address,
+              CheckinProvider* checkin, DNSProvider* process);
     /// \brief The constructor with a specific port on which the services
     /// listen on.
     ///
     /// It effectively listens on "any" IPv4 and/or IPv6 addresses.
     /// IPv4/IPv6 services will be available if and only if \c use_ipv4
     /// or \c use_ipv6 is \c true, respectively.
-    IOService(AuthSrv* auth_server, const char& port,
-              const bool use_ipv4, const bool use_ipv6);
+    IOService(const char& port, const bool use_ipv4, const bool use_ipv6,
+              CheckinProvider* checkin, DNSProvider* process);
     /// \brief The destructor.
     ~IOService();
     //@}
@@ -426,24 +493,10 @@ public:
     /// It will eventually be removed once the wrapper interface is
     /// generalized.
     asio::io_service& get_io_service();
-
-    /// \brief A functor(-like) class that specifies a custom call back
-    /// invoked from the event loop instead of the embedded authoritative
-    /// server callbacks.
-    ///
-    /// Currently, the callback is intended to be used only for testing
-    /// purposes.  But we'll need a generic callback type like this to
-    /// generalize the wrapper interface.
-    typedef boost::function<void(const IOMessage& io_message)> IOCallBack;
-
-    /// \brief Set the custom call back invoked from the event loop.
-    ///
-    /// Right now this method is only for testing, but will eventually be
-    /// generalized.
-    void setCallBack(IOCallBack callback);
 private:
     IOServiceImpl* impl_;
 };
+
 }      // asio_link
 #endif // __ASIO_LINK_H
 

src/bin/auth/auth_srv.cc

@@ -123,12 +123,56 @@ AuthSrvImpl::~AuthSrvImpl() {
     }
 }
 
+// This is a derived class of \c DNSProvider, to serve as a
+// callback in the asio_link module.  It calls
+// AuthSrv::processMessage() on a single DNS message.
+class MessageProcessor : public DNSProvider {
+public:
+    MessageProcessor(AuthSrv* srv) : server_(srv) {}
+    virtual bool operator()(const IOMessage& io_message,
+                            isc::dns::Message& dns_message,
+                            isc::dns::MessageRenderer& renderer) const {
+        return (server_->processMessage(io_message, dns_message, renderer));
+    }
+private:
+    AuthSrv* server_;
+};
+
+// This is a derived class of \c CheckinProvider, to serve
+// as a callback in the asio_link module.  It checks for queued
+// configuration messages, and executes them if found.
+class ConfigChecker : public CheckinProvider {
+public:
+    ConfigChecker(AuthSrv* srv) : server_(srv) {}
+    virtual void operator()(void) const {
+        if (server_->configSession()->hasQueuedMsgs()) {
+            server_->configSession()->checkCommand();
+        }
+    }
+private:
+    AuthSrv* server_;
+};
+
 AuthSrv::AuthSrv(const bool use_cache, AbstractXfroutClient& xfrout_client) :
-    impl_(new AuthSrvImpl(use_cache, xfrout_client))
+    impl_(new AuthSrvImpl(use_cache, xfrout_client)),
+    checkin_provider_(new ConfigChecker(this)),
+    dns_provider_(new MessageProcessor(this))
 {}
 
 AuthSrv::~AuthSrv() {
     delete impl_;
+    delete checkin_provider_;
+    delete dns_provider_;
+}
+
+asio_link::CheckinProvider*
+AuthSrv::getCheckinProvider() {
+    return (checkin_provider_);
+}
+
+asio_link::DNSProvider*
+AuthSrv::getDNSProvider() {
+    return (dns_provider_);
 }
 
 namespace {

src/bin/auth/auth_srv.h

@@ -22,6 +22,8 @@
 #include <cc/data.h>
 #include <config/ccsession.h>
 
+#include <auth/asio_link.h>
+
 namespace isc {
 namespace dns {
 class InputBuffer;
@@ -72,6 +74,8 @@ public:
     isc::data::ConstElementPtr updateConfig(isc::data::ConstElementPtr config);
     isc::config::ModuleCCSession* configSession() const;
     void setConfigSession(isc::config::ModuleCCSession* config_session);
+    asio_link::CheckinProvider* getCheckinProvider();
+    asio_link::DNSProvider* getDNSProvider();
 
     ///
     /// Note: this interface is tentative.  We'll revisit the ASIO and session
@@ -89,6 +93,8 @@ public:
     void setXfrinSession(isc::cc::AbstractSession* xfrin_session);
 private:
     AuthSrvImpl* impl_;
+    asio_link::CheckinProvider* checkin_provider_;
+    asio_link::DNSProvider* dns_provider_;
 };
 
 #endif // __AUTH_SRV_H

src/bin/auth/coroutine.h

@@ -0,0 +1,79 @@
+//
+// coroutine.h
+// ~~~~~~~~~~~
+//
+// Copyright (c) 2003-2010 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#ifndef COROUTINE_HPP
+#define COROUTINE_HPP
+
+class coroutine
+{
+public:
+  coroutine() : value_(0) {}
+  bool is_child() const { return value_ < 0; }
+  bool is_parent() const { return !is_child(); }
+  bool is_complete() const { return value_ == -1; }
+private:
+  friend class coroutine_ref;
+  int value_;
+};
+
+class coroutine_ref
+{
+public:
+  coroutine_ref(coroutine& c) : value_(c.value_), modified_(false) {}
+  coroutine_ref(coroutine* c) : value_(c->value_), modified_(false) {}
+  ~coroutine_ref() { if (!modified_) value_ = -1; }
+  operator int() const { return value_; }
+  int& operator=(int v) { modified_ = true; return value_ = v; }
+private:
+  void operator=(const coroutine_ref&);
+  int& value_;
+  bool modified_;
+};
+
+#define CORO_REENTER(c) \
+  switch (coroutine_ref _coro_value = c) \
+    case -1: if (_coro_value) \
+    { \
+      goto terminate_coroutine; \
+      terminate_coroutine: \
+      _coro_value = -1; \
+      goto bail_out_of_coroutine; \
+      bail_out_of_coroutine: \
+      break; \
+    } \
+    else case 0:
+
+#define CORO_YIELD \
+  for (_coro_value = __LINE__;;) \
+    if (_coro_value == 0) \
+    { \
+      case __LINE__: ; \
+      break; \
+    } \
+    else \
+      switch (_coro_value ? 0 : 1) \
+        for (;;) \
+          case -1: if (_coro_value) \
+            goto terminate_coroutine; \
+          else for (;;) \
+            case 1: if (_coro_value) \
+              goto bail_out_of_coroutine; \
+            else case 0:
+
+#define CORO_FORK \
+  for (_coro_value = -__LINE__;; _coro_value = __LINE__) \
+    if (_coro_value == __LINE__) \
+    { \
+      case -__LINE__: ; \
+      break; \
+    } \
+    else
+
+#endif // COROUTINE_HPP

src/bin/auth/main.cc

@@ -176,6 +176,9 @@ main(int argc, char* argv[]) {
         auth_server->setVerbose(verbose_mode);
         cout << "[b10-auth] Server created." << endl;
 
+        asio_link::CheckinProvider* checkin = auth_server->getCheckinProvider();
+        asio_link::DNSProvider* process = auth_server->getDNSProvider();
+
         if (address != NULL) {
             // XXX: we can only specify at most one explicit address.
             // This also means the server cannot run in the dual address
@@ -183,11 +186,11 @@ main(int argc, char* argv[]) {
             // We don't bother to fix this problem, however.  The -a option
             // is a short term workaround until we support dynamic listening
             // port allocation.
-            io_service = new asio_link::IOService(auth_server, *port,
-                                                  *address);
+            io_service = new asio_link::IOService(*port, *address,
+                                                  checkin, process);
         } else {
-            io_service = new asio_link::IOService(auth_server, *port,
-                                                  use_ipv4, use_ipv6);
+            io_service = new asio_link::IOService(*port, use_ipv4, use_ipv6,
+                                                  checkin, process);
         }
         cout << "[b10-auth] IOService created." << endl;
 
@@ -221,7 +224,7 @@ main(int argc, char* argv[]) {
         cout << "[b10-auth] Server started." << endl;
         io_service->run();
     } catch (const std::exception& ex) {
-        cerr << "[b10-auth] Initialization failed: " << ex.what() << endl;
+        cerr << "[b10-auth] Server failed: " << ex.what() << endl;
         ret = 1;
     }
 

src/bin/auth/tests/asio_link_unittest.cc

@@ -14,15 +14,8 @@
 
 // $Id$
 
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netdb.h>
 
-#include <stdint.h>
-
-#include <functional>
-#include <string>
-#include <vector>
+#include <config.h>
 
 #include <gtest/gtest.h>
 
@@ -90,55 +83,52 @@ TEST(IOSocketTest, dummySockets) {
 }
 
 TEST(IOServiceTest, badPort) {
-    EXPECT_THROW(IOService(NULL, *"65536", true, false), IOError);
-    EXPECT_THROW(IOService(NULL, *"5300.0", true, false), IOError);
-    EXPECT_THROW(IOService(NULL, *"-1", true, false), IOError);
-    EXPECT_THROW(IOService(NULL, *"domain", true, false), IOError);
+    EXPECT_THROW(IOService(*"65536", true, false, NULL, NULL), IOError);
+    EXPECT_THROW(IOService(*"5300.0", true, false, NULL, NULL), IOError);
+    EXPECT_THROW(IOService(*"-1", true, false, NULL, NULL), IOError);
+    EXPECT_THROW(IOService(*"domain", true, false, NULL, NULL), IOError);
 }
 
 TEST(IOServiceTest, badAddress) {
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"192.0.2.1.1"),
-                 IOError);
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"2001:db8:::1"),
-                 IOError);
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"localhost"),
-                 IOError);
+    EXPECT_THROW(IOService(*TEST_PORT, *"192.0.2.1.1", NULL, NULL), IOError);
+    EXPECT_THROW(IOService(*TEST_PORT, *"2001:db8:::1", NULL, NULL), IOError);
+    EXPECT_THROW(IOService(*TEST_PORT, *"localhost", NULL, NULL), IOError);
 }
 
 TEST(IOServiceTest, unavailableAddress) {
     // These addresses should generally be unavailable as a valid local
     // address, although there's no guarantee in theory.
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"255.255.0.0"), IOError);
+    EXPECT_THROW(IOService(*TEST_PORT, *"255.255.0.0", NULL, NULL), IOError);
 
     // Some OSes would simply reject binding attempt for an AF_INET6 socket
     // to an IPv4-mapped IPv6 address.  Even if those that allow it, since
     // the corresponding IPv4 address is the same as the one used in the
     // AF_INET socket case above, it should at least show the same result
     // as the previous one.
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"::ffff:255.255.0.0"), IOError);
+    EXPECT_THROW(IOService(*TEST_PORT, *"::ffff:255.255.0.0", NULL, NULL), IOError);
 }
 
 TEST(IOServiceTest, duplicateBind) {
     // In each sub test case, second attempt should fail due to duplicate bind
 
     // IPv6, "any" address
-    IOService* io_service = new IOService(NULL, *TEST_PORT, false, true);
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, false, true), IOError);
+    IOService* io_service = new IOService(*TEST_PORT, false, true, NULL, NULL);
+    EXPECT_THROW(IOService(*TEST_PORT, false, true, NULL, NULL), IOError);
     delete io_service;
 
     // IPv6, specific address
-    io_service = new IOService(NULL, *TEST_PORT, *TEST_IPV6_ADDR);
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, *TEST_IPV6_ADDR), IOError);
+    io_service = new IOService(*TEST_PORT, *TEST_IPV6_ADDR, NULL, NULL);
+    EXPECT_THROW(IOService(*TEST_PORT, *TEST_IPV6_ADDR, NULL, NULL), IOError);
     delete io_service;
 
     // IPv4, "any" address
-    io_service = new IOService(NULL, *TEST_PORT, true, false);
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, true, false), IOError);
+    io_service = new IOService(*TEST_PORT, true, false, NULL, NULL);
+    EXPECT_THROW(IOService(*TEST_PORT, true, false, NULL, NULL), IOError);
     delete io_service;
 
     // IPv4, specific address
-    io_service = new IOService(NULL, *TEST_PORT, *TEST_IPV4_ADDR);
-    EXPECT_THROW(IOService(NULL, *TEST_PORT, *TEST_IPV4_ADDR), IOError);
+    io_service = new IOService(*TEST_PORT, *TEST_IPV4_ADDR, NULL, NULL);
+    EXPECT_THROW(IOService(*TEST_PORT, *TEST_IPV4_ADDR, NULL, NULL), IOError);
     delete io_service;
 }
 
@@ -168,12 +158,12 @@ resolveAddress(const int family, const int sock_type, const int protocol) {
 // to the service that would run in the IOService object.
 // A mock callback function (an ASIOCallBack object) is registered with the
 // IOService object, so the test code should be able to examine the data
-// receives on the server side.  It then checks the received data matches
+// received on the server side.  It then checks the received data matches
 // expected parameters.
 // If initialization parameters of the IOService should be modified, the test
 // case can do it using the setIOService() method.
 // Note: the set of tests in ASIOLinkTest use actual network services and may
-// involve undesirable side effect such as blocking.
+// involve undesirable side effects such as blocking.
 class ASIOLinkTest : public ::testing::Test {
 protected:
     ASIOLinkTest();
@@ -219,14 +209,14 @@ protected:
     void setIOService(const char& address) {
         delete io_service_;
         io_service_ = NULL;
-        io_service_ = new IOService(NULL, *TEST_PORT, address);
-        io_service_->setCallBack(ASIOCallBack(this));
+        ASIOCallBack* cb = new ASIOCallBack(this);
+        io_service_ = new IOService(*TEST_PORT, address, NULL, cb);
     }
     void setIOService(const bool use_ipv4, const bool use_ipv6) {
         delete io_service_;
         io_service_ = NULL;
-        io_service_ = new IOService(NULL, *TEST_PORT, use_ipv4, use_ipv6);
-        io_service_->setCallBack(ASIOCallBack(this));
+        ASIOCallBack* cb = new ASIOCallBack(this);
+        io_service_ = new IOService(*TEST_PORT, use_ipv4, use_ipv6, NULL, cb);
     }
     void doTest(const int family, const int protocol) {
         if (protocol == IPPROTO_UDP) {
@@ -253,11 +243,15 @@ protected:
                             expected_data, expected_datasize);
     }
 private:
-    class ASIOCallBack : public std::unary_function<IOMessage, void> {
+    class ASIOCallBack : public DNSProvider {
     public:
         ASIOCallBack(ASIOLinkTest* test_obj) : test_obj_(test_obj) {}
-        void operator()(const IOMessage& io_message) const {
+        bool operator()(const IOMessage& io_message,
+                        isc::dns::Message& dns_message UNUSED_PARAM,
+                        isc::dns::MessageRenderer& renderer UNUSED_PARAM) const
+        {
             test_obj_->callBack(io_message);
+            return (true);
         }
     private:
         ASIOLinkTest* test_obj_;

src/bin/auth/unyield.h

@@ -0,0 +1,21 @@
+//
+// unyield.hpp
+// ~~~~~~~~~~~
+//
+// Copyright (c) 2003-2010 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#ifdef reenter
+# undef reenter
+#endif
+
+#ifdef yield
+# undef yield
+#endif
+
+#ifdef fork
+# undef fork
+#endif

src/bin/auth/yield.h

@@ -0,0 +1,23 @@
+//
+// yield.h
+// ~~~~~~~
+//
+// Copyright (c) 2003-2010 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include "coroutine.h"
+
+#ifndef reenter
+# define reenter(c) CORO_REENTER(c)
+#endif
+
+#ifndef yield
+# define yield CORO_YIELD
+#endif
+
+#ifndef fork
+# define fork CORO_FORK
+#endif