SERVER-21458 thread request timeout in to ConnectionPool::get in NetworkInterfaceASIO

4 files changed, 68 insertions, 12 deletions

diff --git a/src/mongo/executor/SConscript b/src/mongo/executor/SConscript
index 4e9a12e0aaa..1ec87aa4fd5 100644
--- a/src/mongo/executor/SConscript
+++ b/src/mongo/executor/SConscript
@@ -79,6 +79,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/util/net/hostandport',
+        'remote_command',
     ],
 )
 
diff --git a/src/mongo/executor/connection_pool.cpp b/src/mongo/executor/connection_pool.cpp
index a6b39841826..522bf440c38 100644
--- a/src/mongo/executor/connection_pool.cpp
+++ b/src/mongo/executor/connection_pool.cpp
@@ -32,6 +32,7 @@
 #include "mongo/executor/connection_pool.h"
 
 #include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/executor/remote_command_request.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
@@ -288,7 +289,12 @@ void ConnectionPool::SpecificPool::getConnection(const HostAndPort& hostAndPort,
                                                  Milliseconds timeout,
                                                  stdx::unique_lock<stdx::mutex> lk,
                                                  GetConnectionCallback cb) {
-    auto expiration = _parent->_factory->now() + timeout;
+    // We need some logic here to handle kNoTimeout, which is defined as -1 Milliseconds. If we just
+    // added the timeout, we would get a time 1MS in the past, which would immediately timeout - the
+    // exact opposite of what we want.
+    auto expiration = (timeout == RemoteCommandRequest::kNoTimeout)
+        ? RemoteCommandRequest::kNoExpirationDate
+        : _parent->_factory->now() + timeout;
 
     _requests.push(make_pair(expiration, std::move(cb)));
 
diff --git a/src/mongo/executor/network_interface_asio.cpp b/src/mongo/executor/network_interface_asio.cpp
index 011aadfc100..776b372e9bd 100644
--- a/src/mongo/executor/network_interface_asio.cpp
+++ b/src/mongo/executor/network_interface_asio.cpp
@@ -185,9 +185,9 @@ void NetworkInterfaceASIO::startCommand(const TaskExecutor::CallbackHandle& cbHa
 
     LOG(2) << "startCommand: " << request.toString();
 
-    auto startTime = now();
+    auto getConnectionStartTime = now();
 
-    auto nextStep = [this, startTime, cbHandle, request, onFinish](
+    auto nextStep = [this, getConnectionStartTime, cbHandle, request, onFinish](
         StatusWith<ConnectionPool::ConnectionHandle> swConn) {
 
         if (!swConn.isOK()) {
@@ -244,15 +244,32 @@ void NetworkInterfaceASIO::startCommand(const TaskExecutor::CallbackHandle& cbHa
         op->_request = std::move(request);
         op->_onFinish = std::move(onFinish);
         op->_connectionPoolHandle = std::move(swConn.getValue());
-        op->_start = startTime;
+        op->_start = getConnectionStartTime;
 
         // This ditches the lock and gets us onto the strand (so we're
         // threadsafe)
-        op->_strand.post([this, op] {
+        op->_strand.post([this, op, getConnectionStartTime] {
             // Set timeout now that we have the correct request object
             if (op->_request.timeout != RemoteCommandRequest::kNoTimeout) {
-                op->_timeoutAlarm =
-                    op->_owner->_timerFactory->make(&op->_strand, op->_request.timeout);
+                // Subtract the time it took to get the connection from the pool from the request
+                // timeout.
+                auto getConnectionDuration = now() - getConnectionStartTime;
+                if (getConnectionDuration >= op->_request.timeout) {
+                    // We only assume that the request timer is guaranteed to fire *after* the
+                    // timeout duration - but make no stronger assumption. It is thus possible that
+                    // we have already exceeded the timeout. In this case we timeout the operation
+                    // manually.
+                    return _completeOperation(op,
+                                              {ErrorCodes::ExceededTimeLimit,
+                                               "Remote command timed out while waiting to get a "
+                                               "connection from the pool."});
+                }
+
+                // The above conditional guarantees that the adjusted timeout will never underflow.
+                invariant(op->_request.timeout > getConnectionDuration);
+                auto adjustedTimeout = op->_request.timeout - getConnectionDuration;
+
+                op->_timeoutAlarm = op->_owner->_timerFactory->make(&op->_strand, adjustedTimeout);
 
                 std::shared_ptr<AsyncOp::AccessControl> access;
                 std::size_t generation;
@@ -290,9 +307,7 @@ void NetworkInterfaceASIO::startCommand(const TaskExecutor::CallbackHandle& cbHa
         });
     };
 
-    // TODO: thread some higher level timeout through, rather than 5 minutes,
-    // once we make timeouts pervasive in this api.
-    _connectionPool.get(request.target, Minutes(5), nextStep);
+    _connectionPool.get(request.target, request.timeout, nextStep);
 }
 
 void NetworkInterfaceASIO::cancelCommand(const TaskExecutor::CallbackHandle& cbHandle) {
diff --git a/src/mongo/executor/network_interface_asio_integration_test.cpp b/src/mongo/executor/network_interface_asio_integration_test.cpp
index 9739615e1f2..cb85f273764 100644
--- a/src/mongo/executor/network_interface_asio_integration_test.cpp
+++ b/src/mongo/executor/network_interface_asio_integration_test.cpp
@@ -57,8 +57,7 @@ namespace {
 
 class NetworkInterfaceASIOIntegrationTest : public mongo::unittest::Test {
 public:
-    void setUp() override {
-        NetworkInterfaceASIO::Options options{};
+    void startNet(NetworkInterfaceASIO::Options options = NetworkInterfaceASIO::Options()) {
         options.streamFactory = stdx::make_unique<AsyncStreamFactory>();
         options.timerFactory = stdx::make_unique<AsyncTimerFactoryASIO>();
         options.connectionPoolOptions.maxConnections = 256u;
@@ -143,10 +142,12 @@ private:
 };
 
 TEST_F(NetworkInterfaceASIOIntegrationTest, Ping) {
+    startNet();
     assertCommandOK("admin", BSON("ping" << 1));
 }
 
 TEST_F(NetworkInterfaceASIOIntegrationTest, Timeouts) {
+    startNet();
     // This sleep command will take 10 seconds, so we should time out client side first given
     // our timeout of 100 milliseconds.
     assertCommandFailsOnClient("admin",
@@ -241,6 +242,7 @@ private:
 };
 
 TEST_F(NetworkInterfaceASIOIntegrationTest, StressTest) {
+    startNet();
     const std::size_t numOps = 10000;
     std::vector<Deferred<Status>> ops;
 
@@ -299,6 +301,38 @@ TEST_F(NetworkInterfaceASIOIntegrationTest, StressTest) {
     ASSERT_OK(res);
 }
 
+// Hook that intentionally never finishes
+class HangingHook : public executor::NetworkConnectionHook {
+    Status validateHost(const HostAndPort&, const RemoteCommandResponse&) final {
+        return Status::OK();
+    }
+
+    StatusWith<boost::optional<RemoteCommandRequest>> makeRequest(
+        const HostAndPort& remoteHost) final {
+        return {boost::make_optional(RemoteCommandRequest(remoteHost,
+                                                          "admin",
+                                                          BSON("sleep" << 1 << "lock"
+                                                                       << "none"
+                                                                       << "secs" << 100000000),
+                                                          BSONObj()))};
+    }
+
+    Status handleReply(const HostAndPort& remoteHost, RemoteCommandResponse&& response) final {
+        MONGO_UNREACHABLE;
+    }
+};
+
+
+// Test that we time out a command if the connection hook hangs.
+TEST_F(NetworkInterfaceASIOIntegrationTest, HookHangs) {
+    NetworkInterfaceASIO::Options options;
+    options.networkConnectionHook = stdx::make_unique<HangingHook>();
+    startNet(std::move(options));
+
+    assertCommandFailsOnClient(
+        "admin", BSON("ping" << 1), Seconds(1), ErrorCodes::ExceededTimeLimit);
+}
+
 }  // namespace
 }  // namespace executor
 }  // namespace mongo