diff options
Diffstat (limited to 'src/mongo/executor/connection_pool_test.cpp')
| -rw-r--r-- | src/mongo/executor/connection_pool_test.cpp | 219 |
1 files changed, 196 insertions, 23 deletions
diff --git a/src/mongo/executor/connection_pool_test.cpp b/src/mongo/executor/connection_pool_test.cpp index 79eaaaf0218..cfc514e562e 100644 --- a/src/mongo/executor/connection_pool_test.cpp +++ b/src/mongo/executor/connection_pool_test.cpp @@ -31,6 +31,8 @@ #include "mongo/executor/connection_pool_test_fixture.h" +#include "mongo/util/duration.h" +#include "mongo/util/net/hostandport.h" #include <algorithm> #include <memory> #include <random> @@ -41,6 +43,7 @@ #include <fmt/ostream.h> #include "mongo/executor/connection_pool.h" +#include "mongo/executor/connection_pool_stats.h" #include "mongo/stdx/future.h" #include "mongo/unittest/thread_assertion_monitor.h" #include "mongo/unittest/unittest.h" @@ -52,6 +55,8 @@ namespace connection_pool_test_details { class ConnectionPoolTest : public unittest::Test { public: + constexpr static Milliseconds kNoTimeout = Milliseconds{-1}; + protected: void setUp() override {} @@ -91,6 +96,12 @@ protected: ExecutorFuture(_executor).getAsync([conn = std::move(conn)](auto) {}); } + void doneWithError(ConnectionPool::ConnectionHandle& conn, Status error) { + dynamic_cast<ConnectionImpl*>(conn.get())->indicateFailure(error); + + ExecutorFuture(_executor).getAsync([conn = std::move(conn)](auto) {}); + } + using StatusWithConn = StatusWith<ConnectionPool::ConnectionHandle>; auto getId(const ConnectionPool::ConnectionHandle& conn) { @@ -148,6 +159,8 @@ TEST_F(ConnectionPoolTest, SameConn) { */ TEST_F(ConnectionPoolTest, ConnectionsAreAcquiredInMRUOrder) { auto pool = makePool(); + std::random_device rd; + std::mt19937 rng(rd()); // Obtain a set of connections constexpr size_t kSize = 100; @@ -170,17 +183,24 @@ TEST_F(ConnectionPoolTest, ConnectionsAreAcquiredInMRUOrder) { } }); + std::uniform_int_distribution<> dist{0, 1}; for (size_t i = 0; i != kSize; ++i) { ConnectionImpl::pushSetup(Status::OK()); - pool->get_forTest(HostAndPort(), - Milliseconds(5000), - [&](StatusWith<ConnectionPool::ConnectionHandle> swConn) { - monitors[i].exec([&]() { - ASSERT(swConn.isOK()); - connections.push_back(std::move(swConn.getValue())); - monitors[i].notifyDone(); - }); - }); + auto cb = [&](StatusWith<ConnectionPool::ConnectionHandle> swConn) { + monitors[i].exec([&]() { + ASSERT(swConn.isOK()); + connections.push_back(std::move(swConn.getValue())); + monitors[i].notifyDone(); + }); + }; + auto timeout = Milliseconds(5000); + + // Randomly lease or check out connection. + if (dist(rng)) { + pool->get_forTest(HostAndPort(), timeout, cb); + } else { + pool->lease_forTest(HostAndPort(), timeout, cb); + } } for (auto& monitor : monitors) { @@ -190,8 +210,6 @@ TEST_F(ConnectionPoolTest, ConnectionsAreAcquiredInMRUOrder) { ASSERT_EQ(connections.size(), kSize); // Shuffle them into a random order - std::random_device rd; - std::mt19937 rng(rd()); std::shuffle(connections.begin(), connections.end(), rng); // Return them to the pool in that random order, recording IDs in a stack @@ -211,18 +229,24 @@ TEST_F(ConnectionPoolTest, ConnectionsAreAcquiredInMRUOrder) { // as the IDs in the stack, since the pool returns them in MRU order. for (size_t i = 0; i != kSize; ++i) { ConnectionImpl::pushSetup(Status::OK()); - pool->get_forTest(HostAndPort(), - Milliseconds(5000), - [&](StatusWith<ConnectionPool::ConnectionHandle> swConn) { - monitors[i].exec([&]() { - ASSERT(swConn.isOK()); - const auto id = verifyAndGetId(swConn); - connections.push_back(std::move(swConn.getValue())); - ASSERT_EQ(id, ids.top()); - ids.pop(); - monitors[i].notifyDone(); - }); - }); + auto cb = [&](StatusWith<ConnectionPool::ConnectionHandle> swConn) { + monitors[i].exec([&]() { + ASSERT(swConn.isOK()); + const auto id = verifyAndGetId(swConn); + connections.push_back(std::move(swConn.getValue())); + ASSERT_EQ(id, ids.top()); + ids.pop(); + monitors[i].notifyDone(); + }); + }; + auto timeout = Milliseconds(5000); + + // Randomly lease or check out connection. + if (dist(rng)) { + pool->get_forTest(HostAndPort(), timeout, cb); + } else { + pool->lease_forTest(HostAndPort(), timeout, cb); + } } for (auto& monitor : monitors) { @@ -384,6 +408,155 @@ TEST_F(ConnectionPoolTest, FailedConnDifferentConn) { } /** + * Verify that a connection returned with an error indicating the remote + * is unavailable drops current generation connections to that remote. + */ +TEST_F(ConnectionPoolTest, FailedHostDropsConns) { + auto pool = makePool(); + + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), 0U); + + constexpr size_t kSize = 100; + std::vector<ConnectionPool::ConnectionHandle> connections; + std::vector<unittest::ThreadAssertionMonitor> monitors(kSize); + + // Ensure that no matter how we leave the test, we mark any + // checked out connections as OK before implicity returning them + // to the pool by destroying the 'connections' vector. Otherwise, + // this test would cause an invariant failure instead of a normal + // test failure if it fails, which would be confusing. + auto drainConnPool = [&] { + while (!connections.empty()) { + try { + ConnectionPool::ConnectionHandle conn = std::move(connections.back()); + connections.pop_back(); + doneWith(conn); + } catch (...) { + } + } + }; + const ScopeGuard guard(drainConnPool); + + auto now = Date_t::now(); + PoolImpl::setNow(now); + + // Check out kSize connections from the pool. + for (size_t i = 0; i != kSize; ++i) { + ConnectionImpl::pushSetup(Status::OK()); + auto cb = [&](StatusWith<ConnectionPool::ConnectionHandle> swConn) { + monitors[i].exec([&]() { + ASSERT(swConn.isOK()); + connections.push_back(std::move(swConn.getValue())); + monitors[i].notifyDone(); + }); + }; + auto timeout = Milliseconds(5000); + pool->get_forTest(HostAndPort(), timeout, cb); + } + + for (auto& monitor : monitors) { + monitor.wait(); + } + + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), kSize); + + // Return one connection with a network error. + ConnectionPool::ConnectionHandle conn = std::move(connections.back()); + connections.pop_back(); + doneWithError(conn, {ErrorCodes::HostUnreachable, "error"}); + + // We should still have all of the connections open, minus the one we just returned with an + // error. + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), kSize - 1); + + // Put the remaining connections back. + drainConnPool(); + + // They should all be discarded since the host should be marked as down + // due to the connection returned with a network error. + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), 0); +} + +/** + * Verify that a connection returned with an error that does _not_ indicate + * the remote is unavailable does _not_ drop current generation connections to that remote. + */ +TEST_F(ConnectionPoolTest, OtherErrorsDontDropConns) { + auto pool = makePool(); + + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), 0U); + + constexpr size_t kSize = 100; + std::vector<ConnectionPool::ConnectionHandle> connections; + + // Ensure that no matter how we leave the test, we mark any + // checked out connections as OK before implicity returning them + // to the pool by destroying the 'connections' vector. Otherwise, + // this test would cause an invariant failure instead of a normal + // test failure if it fails, which would be confusing. + auto drainConnPool = [&] { + while (!connections.empty()) { + try { + ConnectionPool::ConnectionHandle conn = std::move(connections.back()); + connections.pop_back(); + doneWith(conn); + } catch (...) { + } + } + }; + const ScopeGuard guard(drainConnPool); + + auto now = Date_t::now(); + PoolImpl::setNow(now); + + auto checkOutConnections = [&] { + std::vector<unittest::ThreadAssertionMonitor> monitors(kSize); + for (size_t i = 0; i != kSize; ++i) { + ConnectionImpl::pushSetup(Status::OK()); + auto cb = [&](StatusWith<ConnectionPool::ConnectionHandle> swConn) { + monitors[i].exec([&]() { + ASSERT(swConn.isOK()); + connections.push_back(std::move(swConn.getValue())); + monitors[i].notifyDone(); + }); + }; + auto timeout = Milliseconds(5000); + pool->get_forTest(HostAndPort(), timeout, cb); + } + + for (auto& monitor : monitors) { + monitor.wait(); + } + + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), kSize); + }; + + // All three types of error that shouldn't result in us dropping connections - a non-network + // error; a network timeout error, and a network error that we can isolate to a specific + // connection. + std::array<ErrorCodes::Error, 3> errors = { + ErrorCodes::InternalError, ErrorCodes::NetworkTimeout, ErrorCodes::ConnectionError}; + for (size_t i = 0; i < errors.size(); ++i) { + // Check out kSize connections from the pool. + checkOutConnections(); + // Return one connection with a non-network error. + ConnectionPool::ConnectionHandle conn = std::move(connections.back()); + connections.pop_back(); + doneWithError(conn, {errors[i], "error"}); + + // We should still have all of the connections open, minus the one we just returned with an + // error. + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), kSize - 1); + + // Put the remaining connections back. + drainConnPool(); + + // They should all still be open. + ASSERT_EQ(pool->getNumConnectionsPerHost(HostAndPort()), kSize - 1); + } +} + +/** * Verify that providing different host and ports gives you different * connections. */ |