summaryrefslogtreecommitdiff
path: root/src/mongo/executor
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/executor')
-rw-r--r--src/mongo/executor/connection_pool.cpp247
-rw-r--r--src/mongo/executor/network_interface_asio_auth.cpp1
-rw-r--r--src/mongo/executor/task_executor.h2
-rw-r--r--src/mongo/executor/task_executor_test_common.cpp8
4 files changed, 159 insertions, 99 deletions
diff --git a/src/mongo/executor/connection_pool.cpp b/src/mongo/executor/connection_pool.cpp
index cf8aadd4d3a..9bbf528725e 100644
--- a/src/mongo/executor/connection_pool.cpp
+++ b/src/mongo/executor/connection_pool.cpp
@@ -60,6 +60,45 @@ namespace executor {
*/
class ConnectionPool::SpecificPool {
public:
+ /**
+ * These active client methods must be used whenever entering a specific pool outside of the
+ * shutdown background task. The presence of an active client will bump a counter on the
+ * specific pool which will prevent the shutdown thread from deleting it.
+ *
+ * The complexity comes from the need to hold a lock when writing to the
+ * _activeClients param on the specific pool. Because the code beneath the client needs to lock
+ * and unlock the parent mutex (and can leave unlocked), we want to start the client with the
+ * lock acquired, move it into the client, then re-acquire to decrement the counter on the way
+ * out.
+ *
+ * It's used like:
+ *
+ * pool.runWithActiveClient([](stdx::unique_lock<stdx::mutex> lk){ codeToBeProtected(); });
+ */
+ template <typename Callback>
+ void runWithActiveClient(Callback&& cb) {
+ runWithActiveClient(stdx::unique_lock<stdx::mutex>(_parent->_mutex),
+ std::forward<Callback>(cb));
+ }
+
+ template <typename Callback>
+ void runWithActiveClient(stdx::unique_lock<stdx::mutex> lk, Callback&& cb) {
+ invariant(lk.owns_lock());
+
+ _activeClients++;
+
+ const auto guard = MakeGuard([&] {
+ invariant(!lk.owns_lock());
+ stdx::lock_guard<stdx::mutex> lk(_parent->_mutex);
+ _activeClients--;
+ });
+
+ {
+ decltype(lk) localLk(std::move(lk));
+ cb(std::move(localLk));
+ }
+ }
+
SpecificPool(ConnectionPool* parent, const HostAndPort& hostAndPort);
~SpecificPool();
@@ -149,6 +188,7 @@ private:
std::unique_ptr<TimerInterface> _requestTimer;
Date_t _requestTimerExpiration;
+ size_t _activeClients;
size_t _generation;
bool _inFulfillRequests;
bool _inSpawnConnections;
@@ -206,8 +246,11 @@ void ConnectionPool::dropConnections(const HostAndPort& hostAndPort) {
if (iter == _pools.end())
return;
- iter->second.get()->processFailure(
- Status(ErrorCodes::PooledConnectionsDropped, "Pooled connections dropped"), std::move(lk));
+ iter->second->runWithActiveClient(std::move(lk), [&](decltype(lk) lk) {
+ iter->second->processFailure(
+ Status(ErrorCodes::PooledConnectionsDropped, "Pooled connections dropped"),
+ std::move(lk));
+ });
}
void ConnectionPool::get(const HostAndPort& hostAndPort,
@@ -229,7 +272,9 @@ void ConnectionPool::get(const HostAndPort& hostAndPort,
invariant(pool);
- pool->getConnection(hostAndPort, timeout, std::move(lk), std::move(cb));
+ pool->runWithActiveClient(std::move(lk), [&](decltype(lk) lk) {
+ pool->getConnection(hostAndPort, timeout, std::move(lk), std::move(cb));
+ });
}
void ConnectionPool::appendConnectionStats(ConnectionPoolStats* stats) const {
@@ -264,13 +309,16 @@ void ConnectionPool::returnConnection(ConnectionInterface* conn) {
invariant(iter != _pools.end());
- iter->second.get()->returnConnection(conn, std::move(lk));
+ iter->second->runWithActiveClient(std::move(lk), [&](decltype(lk) lk) {
+ iter->second->returnConnection(conn, std::move(lk));
+ });
}
ConnectionPool::SpecificPool::SpecificPool(ConnectionPool* parent, const HostAndPort& hostAndPort)
: _parent(parent),
_hostAndPort(hostAndPort),
_requestTimer(parent->_factory->makeTimer()),
+ _activeClients(0),
_generation(0),
_inFulfillRequests(false),
_inSpawnConnections(false),
@@ -362,47 +410,48 @@ void ConnectionPool::SpecificPool::returnConnection(ConnectionInterface* connPtr
// Unlock in case refresh can occur immediately
lk.unlock();
- connPtr->refresh(_parent->_options.refreshTimeout,
- [this](ConnectionInterface* connPtr, Status status) {
- connPtr->indicateUsed();
-
- stdx::unique_lock<stdx::mutex> lk(_parent->_mutex);
-
- auto conn = takeFromProcessingPool(connPtr);
-
- // If the host and port were dropped, let this lapse
- if (conn->getGeneration() != _generation) {
- spawnConnections(lk);
- return;
- }
-
- // If we're in shutdown, we don't need refreshed connections
- if (_state == State::kInShutdown)
- return;
-
- // If the connection refreshed successfully, throw it back in the ready
- // pool
- if (status.isOK()) {
- addToReady(lk, std::move(conn));
- spawnConnections(lk);
- return;
- }
-
- // If we've exceeded the time limit, start a new connect, rather than
- // failing all operations. We do this because the various callers have
- // their own time limit which is unrelated to our internal one.
- if (status.code() == ErrorCodes::NetworkInterfaceExceededTimeLimit) {
- log() << "Pending connection to host " << _hostAndPort
- << " did not complete within the connection timeout,"
- << " retrying with a new connection;" << openConnections(lk)
- << " connections to that host remain open";
- spawnConnections(lk);
- return;
- }
-
- // Otherwise pass the failure on through
- processFailure(status, std::move(lk));
- });
+ connPtr->refresh(
+ _parent->_options.refreshTimeout, [this](ConnectionInterface* connPtr, Status status) {
+ connPtr->indicateUsed();
+
+ runWithActiveClient([&](stdx::unique_lock<stdx::mutex> lk) {
+ auto conn = takeFromProcessingPool(connPtr);
+
+ // If the host and port were dropped, let this lapse
+ if (conn->getGeneration() != _generation) {
+ spawnConnections(lk);
+ return;
+ }
+
+ // If we're in shutdown, we don't need refreshed connections
+ if (_state == State::kInShutdown)
+ return;
+
+ // If the connection refreshed successfully, throw it back in
+ // the ready pool
+ if (status.isOK()) {
+ addToReady(lk, std::move(conn));
+ spawnConnections(lk);
+ return;
+ }
+
+ // If we've exceeded the time limit, start a new connect,
+ // rather than failing all operations. We do this because the
+ // various callers have their own time limit which is unrelated
+ // to our internal one.
+ if (status.code() == ErrorCodes::NetworkInterfaceExceededTimeLimit) {
+ log() << "Pending connection to host " << _hostAndPort
+ << " did not complete within the connection timeout,"
+ << " retrying with a new connection;" << openConnections(lk)
+ << " connections to that host remain open";
+ spawnConnections(lk);
+ return;
+ }
+
+ // Otherwise pass the failure on through
+ processFailure(status, std::move(lk));
+ });
+ });
lk.lock();
} else {
// If it's fine as it is, just put it in the ready queue
@@ -425,25 +474,25 @@ void ConnectionPool::SpecificPool::addToReady(stdx::unique_lock<stdx::mutex>& lk
connPtr->setTimeout(_parent->_options.refreshRequirement, [this, connPtr]() {
OwnedConnection conn;
- stdx::unique_lock<stdx::mutex> lk(_parent->_mutex);
-
- if (!_readyPool.count(connPtr)) {
- // We've already been checked out. We don't need to refresh
- // ourselves.
- return;
- }
+ runWithActiveClient([&](stdx::unique_lock<stdx::mutex> lk) {
+ if (!_readyPool.count(connPtr)) {
+ // We've already been checked out. We don't need to refresh
+ // ourselves.
+ return;
+ }
- conn = takeFromPool(_readyPool, connPtr);
+ conn = takeFromPool(_readyPool, connPtr);
- // If we're in shutdown, we don't need to refresh connections
- if (_state == State::kInShutdown)
- return;
+ // If we're in shutdown, we don't need to refresh connections
+ if (_state == State::kInShutdown)
+ return;
- _checkedOutPool[connPtr] = std::move(conn);
+ _checkedOutPool[connPtr] = std::move(conn);
- connPtr->indicateSuccess();
+ connPtr->indicateSuccess();
- returnConnection(connPtr, std::move(lk));
+ returnConnection(connPtr, std::move(lk));
+ });
});
fulfillRequests(lk);
@@ -586,26 +635,26 @@ void ConnectionPool::SpecificPool::spawnConnections(stdx::unique_lock<stdx::mute
_parent->_options.refreshTimeout, [this](ConnectionInterface* connPtr, Status status) {
connPtr->indicateUsed();
- stdx::unique_lock<stdx::mutex> lk(_parent->_mutex);
-
- auto conn = takeFromProcessingPool(connPtr);
-
- if (conn->getGeneration() != _generation) {
- // If the host and port was dropped, let the
- // connection lapse
- spawnConnections(lk);
- } else if (status.isOK()) {
- addToReady(lk, std::move(conn));
- spawnConnections(lk);
- } else if (status.code() == ErrorCodes::NetworkInterfaceExceededTimeLimit) {
- // If we've exceeded the time limit, restart the connect, rather than
- // failing all operations. We do this because the various callers
- // have their own time limit which is unrelated to our internal one.
- spawnConnections(lk);
- } else {
- // If the setup failed, cascade the failure edge
- processFailure(status, std::move(lk));
- }
+ runWithActiveClient([&](stdx::unique_lock<stdx::mutex> lk) {
+ auto conn = takeFromProcessingPool(connPtr);
+
+ if (conn->getGeneration() != _generation) {
+ // If the host and port was dropped, let the
+ // connection lapse
+ spawnConnections(lk);
+ } else if (status.isOK()) {
+ addToReady(lk, std::move(conn));
+ spawnConnections(lk);
+ } else if (status.code() == ErrorCodes::NetworkInterfaceExceededTimeLimit) {
+ // If we've exceeded the time limit, restart the connect, rather than
+ // failing all operations. We do this because the various callers
+ // have their own time limit which is unrelated to our internal one.
+ spawnConnections(lk);
+ } else {
+ // If the setup failed, cascade the failure edge
+ processFailure(status, std::move(lk));
+ }
+ });
});
// Note that this assumes that the refreshTimeout is sound for the
// setupTimeout
@@ -641,7 +690,7 @@ void ConnectionPool::SpecificPool::shutdown() {
// If we have processing connections, wait for them to finish or timeout
// before shutdown
- if (_processingPool.size() || _droppedProcessingPool.size()) {
+ if (_processingPool.size() || _droppedProcessingPool.size() || _activeClients) {
_requestTimer->setTimeout(Seconds(1), [this]() { shutdown(); });
return;
@@ -693,27 +742,27 @@ void ConnectionPool::SpecificPool::updateStateInLock() {
// We set a timer for the most recent request, then invoke each timed
// out request we couldn't service
_requestTimer->setTimeout(timeout, [this]() {
- stdx::unique_lock<stdx::mutex> lk(_parent->_mutex);
-
- auto now = _parent->_factory->now();
-
- while (_requests.size()) {
- auto& x = _requests.top();
-
- if (x.first <= now) {
- auto cb = std::move(x.second);
- _requests.pop();
-
- lk.unlock();
- cb(Status(ErrorCodes::NetworkInterfaceExceededTimeLimit,
- "Couldn't get a connection within the time limit"));
- lk.lock();
- } else {
- break;
+ runWithActiveClient([&](stdx::unique_lock<stdx::mutex> lk) {
+ auto now = _parent->_factory->now();
+
+ while (_requests.size()) {
+ auto& x = _requests.top();
+
+ if (x.first <= now) {
+ auto cb = std::move(x.second);
+ _requests.pop();
+
+ lk.unlock();
+ cb(Status(ErrorCodes::NetworkInterfaceExceededTimeLimit,
+ "Couldn't get a connection within the time limit"));
+ lk.lock();
+ } else {
+ break;
+ }
}
- }
- updateStateInLock();
+ updateStateInLock();
+ });
});
} else if (_checkedOutPool.size()) {
// If we have no requests, but someone's using a connection, we just
diff --git a/src/mongo/executor/network_interface_asio_auth.cpp b/src/mongo/executor/network_interface_asio_auth.cpp
index 1f2039dbd1e..571a1be1f81 100644
--- a/src/mongo/executor/network_interface_asio_auth.cpp
+++ b/src/mongo/executor/network_interface_asio_auth.cpp
@@ -38,6 +38,7 @@
#include "mongo/db/auth/authorization_manager_global.h"
#include "mongo/db/auth/internal_user_auth.h"
#include "mongo/db/commands.h"
+#include "mongo/db/commands/feature_compatibility_version_command_parser.h"
#include "mongo/db/server_options.h"
#include "mongo/db/wire_version.h"
#include "mongo/rpc/factory.h"
diff --git a/src/mongo/executor/task_executor.h b/src/mongo/executor/task_executor.h
index 2d558512f91..ebf7c447ef0 100644
--- a/src/mongo/executor/task_executor.h
+++ b/src/mongo/executor/task_executor.h
@@ -196,6 +196,8 @@ public:
/**
* Schedules "work" to be run by the executor no sooner than "when".
*
+ * If "when" is <= now(), then it schedules the "work" to be run ASAP.
+ *
* Returns a handle for waiting on or canceling the callback, or
* ErrorCodes::ShutdownInProgress.
*
diff --git a/src/mongo/executor/task_executor_test_common.cpp b/src/mongo/executor/task_executor_test_common.cpp
index 57c12813250..d7a384231b8 100644
--- a/src/mongo/executor/task_executor_test_common.cpp
+++ b/src/mongo/executor/task_executor_test_common.cpp
@@ -328,14 +328,22 @@ COMMON_EXECUTOR_TEST(ScheduleWorkAt) {
Status status1 = getDetectableErrorStatus();
Status status2 = getDetectableErrorStatus();
Status status3 = getDetectableErrorStatus();
+ Status status4 = getDetectableErrorStatus();
+
const Date_t now = net->now();
const TaskExecutor::CallbackHandle cb1 = unittest::assertGet(executor.scheduleWorkAt(
now + Milliseconds(100), stdx::bind(setStatus, stdx::placeholders::_1, &status1)));
+ const TaskExecutor::CallbackHandle cb4 = unittest::assertGet(executor.scheduleWorkAt(
+ now - Milliseconds(50), stdx::bind(setStatus, stdx::placeholders::_1, &status4)));
unittest::assertGet(executor.scheduleWorkAt(
now + Milliseconds(5000), stdx::bind(setStatus, stdx::placeholders::_1, &status3)));
const TaskExecutor::CallbackHandle cb2 = unittest::assertGet(executor.scheduleWorkAt(
now + Milliseconds(200),
stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, &status2)));
+
+ executor.wait(cb4);
+ ASSERT_OK(status4);
+
const Date_t startTime = net->now();
net->enterNetwork();
net->runUntil(startTime + Milliseconds(200));
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 15:39:24 +0000