[5318] Implemented timeouts for CommandMgr.

src/bin/dhcp4/tests/ctrl_dhcp4_srv_unittest.cc

@@ -6,6 +6,7 @@
 
 #include <config.h>
 
+#include <asiolink/interval_timer.h>
 #include <asiolink/io_service.h>
 #include <cc/command_interpreter.h>
 #include <config/command_mgr.h>
@@ -326,8 +327,8 @@ public:
     }
 
     /// @brief Command handler which generates long response
-   static  ConstElementPtr longResponseHandler(const std::string&,
-                                               const ConstElementPtr&) {
+   static ConstElementPtr longResponseHandler(const std::string&,
+                                              const ConstElementPtr&) {
         ElementPtr arguments = Element::createList();
         std::string arg = "responseresponseresponseresponseresponseresponse"
             "response";
@@ -1301,5 +1302,54 @@ TEST_F(CtrlChannelDhcpv4SrvTest, longResponse) {
     EXPECT_EQ(reference_response, response.str());
 }
 
+// This test verifies that the server signals timeout if the transmission
+// takes too long.
+TEST_F(CtrlChannelDhcpv4SrvTest, connectionTimeout) {
+    createUnixChannelServer();
+
+    // Server's response will be assigned to this variable.
+    std::string response;
+
+    // It is useful to create a thread and run the server and the client
+    // at the same time and independently.
+    std::thread th([this, &response]() {
+
+        // IO service will be stopped automatically when this object goes
+        // out of scope and is destroyed. This is useful because we use
+        // asserts which may break the thread in various exit points.
+        IOServiceWork work(getIOService());
+
+        // Create the client and connect it to the server.
+        boost::scoped_ptr<UnixControlClient> client(new UnixControlClient());
+        ASSERT_TRUE(client);
+        ASSERT_TRUE(client->connectToServer(socket_path_));
+
+        // Send partial command. The server will be waiting for the remaining
+        // part to be sent and will eventually signal a timeout.
+        std::string command = "{ \"command\": \"foo\" ";
+        ASSERT_TRUE(client->sendCommand(command));
+
+        // Let's wait up to 10s for the server's response. The response
+        // should arrive sooner assuming that the timeout mechanism for
+        // the server is working properly.
+        const unsigned int timeout = 10;
+        ASSERT_TRUE(client->getResponse(response, 10));
+
+        // Explicitly close the client's connection.
+        client->disconnectFromServer();
+    });
+
+    // Run the server until stopped.
+    getIOService()->run();
+
+    // Wait for the thread to return.
+    th.join();
+
+    // Check that the server has signalled a timeout.
+    EXPECT_EQ("{ \"result\": 1, \"text\": \"Connection over control channel"
+              " timed out\" }", response);
+}
+
+
 
 } // End of anonymous namespace

src/lib/config/command_mgr.cc

@@ -5,6 +5,7 @@
 // file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 #include <asiolink/asio_wrapper.h>
+#include <asiolink/interval_timer.h>
 #include <asiolink/io_service.h>
 #include <asiolink/unix_domain_socket.h>
 #include <asiolink/unix_domain_socket_acceptor.h>
@@ -30,6 +31,11 @@ namespace {
 /// @brief Maximum size of the data chunk sent/received over the socket.
 const size_t BUF_SIZE = 8192;
 
+/// @brief Specifies connection timeout in milliseconds.
+///
+/// @todo Make it configurable.
+const unsigned CONNECTION_TIMEOUT = 5000;
+
 class ConnectionPool;
 
 /// @brief Represents a single connection over control socket.
@@ -49,10 +55,12 @@ public:
     /// for data transmission.
     /// @param connection_pool Reference to the connection pool to which this
     /// connection belongs.
-    Connection(const boost::shared_ptr<UnixDomainSocket>& socket,
+    Connection(const IOServicePtr& io_service,
+               const boost::shared_ptr<UnixDomainSocket>& socket,
                ConnectionPool& connection_pool)
-        : socket_(socket), buf_(), response_(), connection_pool_(connection_pool),
-          feed_(), response_in_progress_(false) {
+        : socket_(socket), timeout_timer_(*io_service), buf_(), response_(),
+          connection_pool_(connection_pool), feed_(),
+          response_in_progress_(false) {
 
         LOG_INFO(command_logger, COMMAND_SOCKET_CONNECTION_OPENED)
             .arg(socket_->getNative());
@@ -62,6 +70,17 @@ public:
         isc::dhcp::IfaceMgr::instance().addExternalSocket(socket_->getNative(), 0);
         // Initialize state model for receiving and preparsing commands.
         feed_.initModel();
+
+        // Start timer for detecting timeouts.
+        timeout_timer_.setup(boost::bind(&Connection::timeoutHandler, this),
+                             CONNECTION_TIMEOUT, IntervalTimer::ONE_SHOT);
+    }
+
+    /// @brief Destructor.
+    ///
+    /// Cancels timeout timer if one is scheduled.
+    ~Connection() {
+        timeout_timer_.cancel();
     }
 
     /// @brief Close current connection.
@@ -77,6 +96,7 @@ public:
 
             isc::dhcp::IfaceMgr::instance().deleteExternalSocket(socket_->getNative());
             socket_->close();
+            timeout_timer_.cancel();
         }
     }
 
@@ -118,11 +138,18 @@ public:
     /// @param bytes_transferred Number of bytes sent.
     void sendHandler(const boost::system::error_code& ec,
                      size_t bytes_trasferred);
+
+    /// @brief Handler invoked when timeout has occurred.
+    void timeoutHandler();
+
 private:
 
     /// @brief Pointer to the socket used for transmission.
     boost::shared_ptr<UnixDomainSocket> socket_;
 
+    /// @brief Interval timer used to detect connection timeouts.
+    IntervalTimer timeout_timer_;
+
     /// @brief Buffer used for received data.
     std::array<char, BUF_SIZE> buf_;
 
@@ -299,6 +326,15 @@ Connection::sendHandler(const boost::system::error_code& ec,
     connection_pool_.stop(shared_from_this());
 }
 
+void
+Connection::timeoutHandler() {
+    ConstElementPtr rsp = createAnswer(CONTROL_RESULT_ERROR, "Connection over"
+                                       " control channel timed out");
+    response_ = rsp->str();
+    doSend();
+}
+
+
 }
 
 namespace isc {
@@ -400,7 +436,8 @@ CommandMgrImpl::doAccept() {
     acceptor_->asyncAccept(*socket_, [this](const boost::system::error_code& ec) {
         if (!ec) {
             // New connection is arriving. Start asynchronous transmission.
-            ConnectionPtr connection(new Connection(socket_, connection_pool_));
+            ConnectionPtr connection(new Connection(io_service_, socket_,
+                                                    connection_pool_));
             connection_pool_.start(connection);
 
         } else if (ec.value() != boost::asio::error::operation_aborted) {