summaryrefslogtreecommitdiff
path: root/src/mongo/executor/connection_pool.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/executor/connection_pool.cpp')
-rw-r--r--src/mongo/executor/connection_pool.cpp181
1 files changed, 54 insertions, 127 deletions
diff --git a/src/mongo/executor/connection_pool.cpp b/src/mongo/executor/connection_pool.cpp
index 91fb862bb90..5900f65a650 100644
--- a/src/mongo/executor/connection_pool.cpp
+++ b/src/mongo/executor/connection_pool.cpp
@@ -137,8 +137,8 @@ std::string ConnectionPool::ConnectionControls::toString() const {
}
std::string ConnectionPool::HostState::toString() const {
- return "{{ requests: {}, ready: {}, pending: {}, active: {}, leased: {}, isExpired: {} }}"_format(
- requests, ready, pending, active, leased, health.isExpired);
+ return "{{ requests: {}, ready: {}, pending: {}, active: {}, isExpired: {} }}"_format(
+ requests, ready, pending, active, health.isExpired);
}
/**
@@ -162,7 +162,7 @@ public:
const auto minConns = getPool()->_options.minConnections;
const auto maxConns = getPool()->_options.maxConnections;
- data.target = stats.requests + stats.active + stats.leased;
+ data.target = stats.requests + stats.active;
if (data.target < minConns) {
data.target = minConns;
} else if (data.target > maxConns) {
@@ -275,7 +275,7 @@ public:
* Gets a connection from the specific pool. Sinks a unique_lock from the
* parent to preserve the lock on _mutex
*/
- Future<ConnectionHandle> getConnection(Milliseconds timeout, bool lease);
+ Future<ConnectionHandle> getConnection(Milliseconds timeout);
/**
* Triggers the shutdown procedure. This function sets isShutdown to true
@@ -298,11 +298,6 @@ public:
size_t inUseConnections() const;
/**
- * Returns the number of leased connections from the pool.
- */
- size_t leasedConnections() const;
-
- /**
* Returns the number of available connections in the pool.
*/
size_t availableConnections() const;
@@ -325,7 +320,7 @@ public:
/**
* Returns the total number of connections currently open that belong to
* this pool. This is the sum of refreshingConnections, availableConnections,
- * inUseConnections, and leasedConnections.
+ * and inUseConnections.
*/
size_t openConnections() const;
@@ -366,20 +361,14 @@ private:
using OwnedConnection = std::shared_ptr<ConnectionInterface>;
using OwnershipPool = stdx::unordered_map<ConnectionInterface*, OwnedConnection>;
using LRUOwnershipPool = LRUCache<OwnershipPool::key_type, OwnershipPool::mapped_type>;
- struct Request {
- Date_t expiration;
- Promise<ConnectionHandle> promise;
- // Whether or not the requested connection should be "leased".
- bool lease;
- };
-
+ using Request = std::pair<Date_t, Promise<ConnectionHandle>>;
struct RequestComparator {
bool operator()(const Request& a, const Request& b) {
- return a.expiration > b.expiration;
+ return a.first > b.first;
}
};
- ConnectionHandle makeHandle(ConnectionInterface* connection, bool isLeased);
+ ConnectionHandle makeHandle(ConnectionInterface* connection);
/**
* Establishes connections until the ControllerInterface's target is met.
@@ -392,11 +381,11 @@ private:
void fulfillRequests();
- void returnConnection(ConnectionInterface* connPtr, bool isLeased);
+ void returnConnection(ConnectionInterface* connPtr);
// This internal helper is used both by get and by _fulfillRequests and differs in that it
// skips some bookkeeping that the other callers do on their own
- ConnectionHandle tryGetConnection(bool lease);
+ ConnectionHandle tryGetConnection();
template <typename OwnershipPoolType>
typename OwnershipPoolType::mapped_type takeFromPool(
@@ -425,7 +414,6 @@ private:
OwnershipPool _processingPool;
OwnershipPool _droppedProcessingPool;
OwnershipPool _checkedOutPool;
- OwnershipPool _leasedPool;
std::vector<Request> _requests;
Date_t _lastActiveTime;
@@ -560,37 +548,21 @@ void ConnectionPool::mutateTags(
pool->mutateTags(mutateFunc);
}
-void ConnectionPool::retrieve_forTest(RetrieveConnection retrieve, GetConnectionCallback cb) {
- // We kick ourselves onto the executor queue to prevent us from deadlocking with our own thread
- auto getConnectionFunc =
- [this, retrieve = std::move(retrieve), cb = std::move(cb)](Status&&) mutable {
- retrieve().thenRunOn(_factory->getExecutor()).getAsync(std::move(cb));
- };
- _factory->getExecutor()->schedule(std::move(getConnectionFunc));
-}
-
void ConnectionPool::get_forTest(const HostAndPort& hostAndPort,
Milliseconds timeout,
GetConnectionCallback cb) {
- auto getConnectionFunc = [this, hostAndPort, timeout]() mutable {
- return get(hostAndPort, transport::kGlobalSSLMode, timeout);
- };
- retrieve_forTest(getConnectionFunc, std::move(cb));
-}
-
-void ConnectionPool::lease_forTest(const HostAndPort& hostAndPort,
- Milliseconds timeout,
- GetConnectionCallback cb) {
- auto getConnectionFunc = [this, hostAndPort, timeout]() mutable {
- return lease(hostAndPort, transport::kGlobalSSLMode, timeout);
+ // We kick ourselves onto the executor queue to prevent us from deadlocking with our own thread
+ auto getConnectionFunc = [this, hostAndPort, timeout, cb = std::move(cb)](Status&&) mutable {
+ get(hostAndPort, transport::kGlobalSSLMode, timeout)
+ .thenRunOn(_factory->getExecutor())
+ .getAsync(std::move(cb));
};
- retrieve_forTest(getConnectionFunc, std::move(cb));
+ _factory->getExecutor()->schedule(std::move(getConnectionFunc));
}
-SemiFuture<ConnectionPool::ConnectionHandle> ConnectionPool::_get(const HostAndPort& hostAndPort,
- transport::ConnectSSLMode sslMode,
- Milliseconds timeout,
- bool lease) {
+SemiFuture<ConnectionPool::ConnectionHandle> ConnectionPool::get(const HostAndPort& hostAndPort,
+ transport::ConnectSSLMode sslMode,
+ Milliseconds timeout) {
stdx::lock_guard lk(_mutex);
auto& pool = _pools[hostAndPort];
@@ -602,7 +574,7 @@ SemiFuture<ConnectionPool::ConnectionHandle> ConnectionPool::_get(const HostAndP
invariant(pool);
- auto connFuture = pool->getConnection(timeout, lease);
+ auto connFuture = pool->getConnection(timeout);
pool->updateState();
return std::move(connFuture).semi();
@@ -618,7 +590,6 @@ void ConnectionPool::appendConnectionStats(ConnectionPoolStats* stats) const {
auto& pool = kv.second;
ConnectionStatsPer hostStats{pool->inUseConnections(),
pool->availableConnections(),
- pool->leasedConnections(),
pool->createdConnections(),
pool->refreshingConnections(),
pool->refreshedConnections()};
@@ -654,7 +625,6 @@ ConnectionPool::SpecificPool::~SpecificPool() {
if (shouldInvariantOnPoolCorrectness()) {
invariant(_requests.empty());
invariant(_checkedOutPool.empty());
- invariant(_leasedPool.empty());
}
}
@@ -666,10 +636,6 @@ size_t ConnectionPool::SpecificPool::availableConnections() const {
return _readyPool.size();
}
-size_t ConnectionPool::SpecificPool::leasedConnections() const {
- return _leasedPool.size();
-}
-
size_t ConnectionPool::SpecificPool::refreshingConnections() const {
return _processingPool.size();
}
@@ -683,7 +649,7 @@ size_t ConnectionPool::SpecificPool::createdConnections() const {
}
size_t ConnectionPool::SpecificPool::openConnections() const {
- return _checkedOutPool.size() + _readyPool.size() + _processingPool.size() + _leasedPool.size();
+ return _checkedOutPool.size() + _readyPool.size() + _processingPool.size();
}
size_t ConnectionPool::SpecificPool::requestsPending() const {
@@ -691,7 +657,7 @@ size_t ConnectionPool::SpecificPool::requestsPending() const {
}
Future<ConnectionPool::ConnectionHandle> ConnectionPool::SpecificPool::getConnection(
- Milliseconds timeout, bool lease) {
+ Milliseconds timeout) {
// Reset our activity timestamp
auto now = _parent->_factory->now();
@@ -699,7 +665,7 @@ Future<ConnectionPool::ConnectionHandle> ConnectionPool::SpecificPool::getConnec
// If we do not have requests, then we can fulfill immediately
if (_requests.size() == 0) {
- auto conn = tryGetConnection(lease);
+ auto conn = tryGetConnection();
if (conn) {
LOGV2_DEBUG(22559,
@@ -725,24 +691,23 @@ Future<ConnectionPool::ConnectionHandle> ConnectionPool::SpecificPool::getConnec
const auto expiration = now + timeout;
auto pf = makePromiseFuture<ConnectionHandle>();
- _requests.push_back({expiration, std::move(pf.promise), lease});
+ _requests.push_back(make_pair(expiration, std::move(pf.promise)));
std::push_heap(begin(_requests), end(_requests), RequestComparator{});
return std::move(pf.future);
}
-auto ConnectionPool::SpecificPool::makeHandle(ConnectionInterface* connection, bool isLeased)
- -> ConnectionHandle {
- auto deleter = [this, anchor = shared_from_this(), isLeased](ConnectionInterface* connection) {
+auto ConnectionPool::SpecificPool::makeHandle(ConnectionInterface* connection) -> ConnectionHandle {
+ auto deleter = [this, anchor = shared_from_this()](ConnectionInterface* connection) {
stdx::lock_guard lk(_parent->_mutex);
- returnConnection(connection, isLeased);
+ returnConnection(connection);
_lastActiveTime = _parent->_factory->now();
updateState();
};
return ConnectionHandle(connection, std::move(deleter));
}
-ConnectionPool::ConnectionHandle ConnectionPool::SpecificPool::tryGetConnection(bool lease) {
+ConnectionPool::ConnectionHandle ConnectionPool::SpecificPool::tryGetConnection() {
while (_readyPool.size()) {
// _readyPool is an LRUCache, so its begin() object is the MRU item.
auto iter = _readyPool.begin();
@@ -764,15 +729,12 @@ ConnectionPool::ConnectionHandle ConnectionPool::SpecificPool::tryGetConnection(
auto connPtr = conn.get();
- if (lease) {
- _leasedPool[connPtr] = std::move(conn);
- } else {
- _checkedOutPool[connPtr] = std::move(conn);
- }
+ // check out the connection
+ _checkedOutPool[connPtr] = std::move(conn);
// pass it to the user
connPtr->resetToUnknown();
- auto handle = makeHandle(connPtr, lease);
+ auto handle = makeHandle(connPtr);
return handle;
}
@@ -808,14 +770,15 @@ void ConnectionPool::SpecificPool::finishRefresh(ConnectionInterface* connPtr, S
return;
}
- // If the error can be contained to one connection, drop the one connection.
- if (status.code() == ErrorCodes::ConnectionError) {
- LOGV2_DEBUG(6832901,
+ // Pass a failure on through
+ if (!status.isOK()) {
+ LOGV2_DEBUG(22563,
kDiagnosticLogLevel,
- "Dropping single connection",
+ "Connection failed to {hostAndPort} due to {error}",
+ "Connection failed",
"hostAndPort"_attr = _hostAndPort,
- "error"_attr = redact(status),
- "numOpenConns"_attr = openConnections());
+ "error"_attr = redact(status));
+ processFailure(status);
return;
}
@@ -829,18 +792,6 @@ void ConnectionPool::SpecificPool::finishRefresh(ConnectionInterface* connPtr, S
return;
}
- // Pass a failure on through
- if (!status.isOK()) {
- LOGV2_DEBUG(22563,
- kDiagnosticLogLevel,
- "Connection failed to {hostAndPort} due to {error}",
- "Connection failed",
- "hostAndPort"_attr = _hostAndPort,
- "error"_attr = redact(status));
- processFailure(status);
- return;
- }
-
LOGV2_DEBUG(22565,
kDiagnosticLogLevel,
"Finishing connection refresh for {hostAndPort}",
@@ -853,10 +804,10 @@ void ConnectionPool::SpecificPool::finishRefresh(ConnectionInterface* connPtr, S
fulfillRequests();
}
-void ConnectionPool::SpecificPool::returnConnection(ConnectionInterface* connPtr, bool isLeased) {
+void ConnectionPool::SpecificPool::returnConnection(ConnectionInterface* connPtr) {
auto needsRefreshTP = connPtr->getLastUsed() + _parent->_controller->toRefreshTimeout();
- auto conn = takeFromPool(isLeased ? _leasedPool : _checkedOutPool, connPtr);
+ auto conn = takeFromPool(_checkedOutPool, connPtr);
invariant(conn);
if (_health.isShutdown) {
@@ -870,29 +821,7 @@ void ConnectionPool::SpecificPool::returnConnection(ConnectionInterface* connPtr
}
if (auto status = conn->getStatus(); !status.isOK()) {
- // Our error handling here is determined by the MongoDB SDAM specification for handling
- // application errors on established connections. In particular, if a network error occurs,
- // we must close all idle sockets in the connection pool for the server: "if one socket is
- // bad, it is likely that all are." However, if the error is just a network _timeout_ error,
- // we don't drop the connections because the timeout may indicate a slow operation rather
- // than an unavailable server. Additionally, if we can isolate the error to a single
- // socket/connection based on it's type, we won't drop other connections/sockets.
- //
- // See the spec for additional details:
- // https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#application-errors
- bool isSingleConnectionError = status.code() == ErrorCodes::ConnectionError;
- if (ErrorCodes::isNetworkError(status) && !isSingleConnectionError &&
- !ErrorCodes::isNetworkTimeoutError(status)) {
- LOGV2_DEBUG(7719500,
- kDiagnosticLogLevel,
- "Connection failed to {hostAndPort} due to {error}",
- "Connection failed",
- "hostAndPort"_attr = _hostAndPort,
- "error"_attr = redact(status));
- processFailure(status);
- return;
- }
- // Otherwise, drop the one connection.
+ // TODO: alert via some callback if the host is bad
LOGV2(22566,
"Ending connection to host {hostAndPort} due to bad connection status: {error}; "
"{numOpenConns} connections to that host remain open",
@@ -913,7 +842,8 @@ void ConnectionPool::SpecificPool::returnConnection(ConnectionInterface* connPtr
if (shouldRefreshConnection) {
auto controls = _parent->_controller->getControls(_id);
- if (openConnections() >= controls.targetConnections) {
+ if (_readyPool.size() + _processingPool.size() + _checkedOutPool.size() >=
+ controls.targetConnections) {
// If we already have minConnections, just let the connection lapse
LOGV2(22567,
"Ending idle connection to host {hostAndPort} because the pool meets "
@@ -980,7 +910,7 @@ void ConnectionPool::SpecificPool::addToReady(OwnedConnection conn) {
connPtr->indicateSuccess();
- returnConnection(connPtr, false);
+ returnConnection(connPtr);
});
connPtr->setTimeout(_parent->_controller->toRefreshTimeout(), std::move(returnConnectionFunc));
}
@@ -1047,7 +977,7 @@ void ConnectionPool::SpecificPool::processFailure(const Status& status) {
}
for (auto& request : _requests) {
- request.promise.setError(status);
+ request.second.setError(status);
}
LOGV2_DEBUG(22573,
@@ -1069,14 +999,14 @@ void ConnectionPool::SpecificPool::fulfillRequests() {
// deadlock).
//
// None of the heap manipulation code throws, but it's something to keep in mind.
- auto conn = tryGetConnection(_requests.front().lease);
+ auto conn = tryGetConnection();
if (!conn) {
break;
}
// Grab the request and callback
- auto promise = std::move(_requests.front().promise);
+ auto promise = std::move(_requests.front().second);
std::pop_heap(begin(_requests), end(_requests), RequestComparator{});
_requests.pop_back();
@@ -1184,8 +1114,7 @@ void ConnectionPool::SpecificPool::updateHealth() {
const auto now = _parent->_factory->now();
// We're expired if we have no sign of connection use and are past our expiry
- _health.isExpired = _requests.empty() && _checkedOutPool.empty() && _leasedPool.empty() &&
- (_hostExpiration <= now);
+ _health.isExpired = _requests.empty() && _checkedOutPool.empty() && (_hostExpiration <= now);
// We're failed until we get new requests or our timer triggers
if (_health.isFailed) {
@@ -1203,7 +1132,7 @@ void ConnectionPool::SpecificPool::updateEventTimer() {
}
// If our expiration comes before our next event, then it is the next event
- if (_requests.empty() && _checkedOutPool.empty() && _leasedPool.empty()) {
+ if (_requests.empty() && _checkedOutPool.empty()) {
_hostExpiration = _lastActiveTime + _parent->_controller->hostTimeout();
if ((_hostExpiration > now) && (_hostExpiration < nextEventTime)) {
nextEventTime = _hostExpiration;
@@ -1211,8 +1140,8 @@ void ConnectionPool::SpecificPool::updateEventTimer() {
}
// If a request would timeout before the next event, then it is the next event
- if (_requests.size() && (_requests.front().expiration < nextEventTime)) {
- nextEventTime = _requests.front().expiration;
+ if (_requests.size() && (_requests.front().first < nextEventTime)) {
+ nextEventTime = _requests.front().first;
}
// Clamp next event time to be either now or in the future. Next event time
@@ -1240,12 +1169,12 @@ void ConnectionPool::SpecificPool::updateEventTimer() {
_health.isFailed = false;
- while (_requests.size() && (_requests.front().expiration <= now)) {
+ while (_requests.size() && (_requests.front().first <= now)) {
std::pop_heap(begin(_requests), end(_requests), RequestComparator{});
auto& request = _requests.back();
- request.promise.setError(Status(ErrorCodes::NetworkInterfaceExceededTimeLimit,
- "Couldn't get a connection within the time limit"));
+ request.second.setError(Status(ErrorCodes::NetworkInterfaceExceededTimeLimit,
+ "Couldn't get a connection within the time limit"));
_requests.pop_back();
// Since we've failed a request, we've interacted with external users
@@ -1269,7 +1198,6 @@ void ConnectionPool::SpecificPool::updateController() {
refreshingConnections(),
availableConnections(),
inUseConnections(),
- leasedConnections(),
};
LOGV2_DEBUG(22578,
kDiagnosticLogLevel,
@@ -1308,7 +1236,6 @@ void ConnectionPool::SpecificPool::updateController() {
if (shouldInvariantOnPoolCorrectness()) {
invariant(pool->_checkedOutPool.empty());
invariant(pool->_requests.empty());
- invariant(pool->_leasedPool.empty());
}
pool->triggerShutdown(Status(ErrorCodes::ConnectionPoolExpired,
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 14:59:50 +0000