SERVER-21108 check if connection is healthy before returning it from the pool

6 files changed, 37 insertions, 1 deletions

diff --git a/src/mongo/executor/connection_pool.cpp b/src/mongo/executor/connection_pool.cpp
index adcf4f7b5d7..169caef2e38 100644
--- a/src/mongo/executor/connection_pool.cpp
+++ b/src/mongo/executor/connection_pool.cpp
@@ -25,6 +25,8 @@
  *    it in the license file.
  */
 
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kASIO
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/executor/connection_pool.h"
@@ -32,9 +34,9 @@
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
 
-
 // One interesting implementation note herein concerns how setup() and
 // refresh() are invoked outside of the global lock, but setTimeout is not.
 // This implementation detail simplifies mocks, allowing them to return
@@ -457,6 +459,22 @@ void ConnectionPool::SpecificPool::fulfillRequests(stdx::unique_lock<stdx::mutex
         _readyPool.erase(iter);
         conn->cancelTimeout();
 
+        if (!conn->isHealthy()) {
+            log() << "dropping unhealthy pooled connection to " << conn->getHostAndPort().host()
+                  << ":" << conn->getHostAndPort().port();
+
+            if (_readyPool.empty()) {
+                log() << "after drop, pool was empty, going to spawn some connections";
+                // Spawn some more connections to the bad host if we're all out.
+                spawnConnections(lk, conn->getHostAndPort());
+            }
+
+            // Drop the bad connection.
+            conn.reset();
+            // Retry.
+            continue;
+        }
+
         // Grab the request and callback
         auto cb = std::move(_requests.top().second);
         _requests.pop();
diff --git a/src/mongo/executor/connection_pool.h b/src/mongo/executor/connection_pool.h
index bc2a5aee0a8..b1b0549365c 100644
--- a/src/mongo/executor/connection_pool.h
+++ b/src/mongo/executor/connection_pool.h
@@ -208,6 +208,11 @@ public:
      */
     virtual const HostAndPort& getHostAndPort() const = 0;
 
+    /**
+     * Check if the connection is healthy using some implementation defined condition.
+     */
+    virtual bool isHealthy() = 0;
+
 protected:
     /**
      * Making these protected makes the definitions available to override in
diff --git a/src/mongo/executor/connection_pool_asio.cpp b/src/mongo/executor/connection_pool_asio.cpp
index 12e6d0bd062..6a9065faef3 100644
--- a/src/mongo/executor/connection_pool_asio.cpp
+++ b/src/mongo/executor/connection_pool_asio.cpp
@@ -121,6 +121,11 @@ const HostAndPort& ASIOConnection::getHostAndPort() const {
     return _hostAndPort;
 }
 
+bool ASIOConnection::isHealthy() {
+    // Check if the remote host has closed the connection.
+    return _impl->connection().stream().isOpen();
+}
+
 void ASIOConnection::indicateUsed() {
     // It is illegal to attempt to use a connection after calling indicateFailure().
     invariant(_status.isOK() || _status == ConnectionPool::kConnectionStateUnknown);
diff --git a/src/mongo/executor/connection_pool_asio.h b/src/mongo/executor/connection_pool_asio.h
index 44235d23e16..20b728e542e 100644
--- a/src/mongo/executor/connection_pool_asio.h
+++ b/src/mongo/executor/connection_pool_asio.h
@@ -78,6 +78,8 @@ public:
     std::unique_ptr<NetworkInterfaceASIO::AsyncOp> releaseAsyncOp();
     void bindAsyncOp(std::unique_ptr<NetworkInterfaceASIO::AsyncOp> op);
 
+    bool isHealthy() override;
+
 private:
     void indicateUsed() override;
     Date_t getLastUsed() const override;
diff --git a/src/mongo/executor/connection_pool_test_fixture.cpp b/src/mongo/executor/connection_pool_test_fixture.cpp
index 3a3b9843fd3..2d48ad0b5e2 100644
--- a/src/mongo/executor/connection_pool_test_fixture.cpp
+++ b/src/mongo/executor/connection_pool_test_fixture.cpp
@@ -101,6 +101,10 @@ const HostAndPort& ConnectionImpl::getHostAndPort() const {
     return _hostAndPort;
 }
 
+bool ConnectionImpl::isHealthy() {
+    return true;
+}
+
 void ConnectionImpl::clear() {
     _setupQueue.clear();
     _refreshQueue.clear();
diff --git a/src/mongo/executor/connection_pool_test_fixture.h b/src/mongo/executor/connection_pool_test_fixture.h
index 37cb4e504ae..74946ba5cf4 100644
--- a/src/mongo/executor/connection_pool_test_fixture.h
+++ b/src/mongo/executor/connection_pool_test_fixture.h
@@ -87,6 +87,8 @@ public:
 
     const HostAndPort& getHostAndPort() const override;
 
+    bool isHealthy() override;
+
     // Dump all connection callbacks
     static void clear();