summaryrefslogtreecommitdiff
path: root/src/mongo/transport
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/transport')
-rw-r--r--src/mongo/transport/asio_utils.cpp2
-rw-r--r--src/mongo/transport/message_compressor_manager.cpp5
-rw-r--r--src/mongo/transport/message_compressor_manager.h8
-rw-r--r--src/mongo/transport/service_state_machine.cpp4
-rw-r--r--src/mongo/transport/transport_layer_asio.cpp32
-rw-r--r--src/mongo/transport/transport_layer_asio.h8
-rw-r--r--src/mongo/transport/transport_layer_asio_test.cpp114
7 files changed, 75 insertions, 98 deletions
diff --git a/src/mongo/transport/asio_utils.cpp b/src/mongo/transport/asio_utils.cpp
index 61d9a56dea2..8e8368ff586 100644
--- a/src/mongo/transport/asio_utils.cpp
+++ b/src/mongo/transport/asio_utils.cpp
@@ -55,8 +55,6 @@ Status errorCodeToStatus(const std::error_code& ec) {
return {ErrorCodes::HostUnreachable, "Connection reset by peer"};
} else if (ec == asio::error::network_reset) {
return {ErrorCodes::HostUnreachable, "Connection reset by network"};
- } else if (ec == asio::error::in_progress) {
- return {ErrorCodes::ConnectionError, "Socket operation in progress"};
}
// If the ec.category() is a mongoErrorCategory() then this error was propogated from
diff --git a/src/mongo/transport/message_compressor_manager.cpp b/src/mongo/transport/message_compressor_manager.cpp
index bc4b8d66795..b6a60cdcebd 100644
--- a/src/mongo/transport/message_compressor_manager.cpp
+++ b/src/mongo/transport/message_compressor_manager.cpp
@@ -319,11 +319,6 @@ void MessageCompressorManager::serverNegotiate(
}
}
-const std::vector<MessageCompressorBase*>& MessageCompressorManager::getNegotiatedCompressors()
- const {
- return _negotiated;
-}
-
MessageCompressorManager& MessageCompressorManager::forSession(
const transport::SessionHandle& session) {
return getForSession(session.get());
diff --git a/src/mongo/transport/message_compressor_manager.h b/src/mongo/transport/message_compressor_manager.h
index 9cf7d931106..9f1a1de35e3 100644
--- a/src/mongo/transport/message_compressor_manager.h
+++ b/src/mongo/transport/message_compressor_manager.h
@@ -62,14 +62,14 @@ public:
MessageCompressorManager& operator=(MessageCompressorManager&&) = default;
/*
- * Called by a client constructing a "hello" request. This function will append the result
+ * Called by a client constructing an isMaster request. This function will append the result
* of _registry->getCompressorNames() to the BSONObjBuilder as a BSON array. If no compressors
* are configured, it won't append anything.
*/
void clientBegin(BSONObjBuilder* output);
/*
- * Called by a client that has received a "hello" response (received after calling
+ * Called by a client that has received an isMaster response (received after calling
* clientBegin) and wants to finish negotiating compression.
*
* This looks for a BSON array called "compression" with the server's list of
@@ -79,7 +79,7 @@ public:
void clientFinish(const BSONObj& input);
/*
- * Called by a server that has received a "hello" request.
+ * Called by a server that has received an isMaster request.
*
* If no compressors are configured that match those requested by the client, then it will
* not append anything to the BSONObjBuilder output.
@@ -121,8 +121,6 @@ public:
StatusWith<Message> decompressMessage(const Message& msg,
MessageCompressorId* compressorId = nullptr);
- const std::vector<MessageCompressorBase*>& getNegotiatedCompressors() const;
-
static MessageCompressorManager& forSession(const transport::SessionHandle& session);
private:
diff --git a/src/mongo/transport/service_state_machine.cpp b/src/mongo/transport/service_state_machine.cpp
index b3a6ec73724..ce330e8b060 100644
--- a/src/mongo/transport/service_state_machine.cpp
+++ b/src/mongo/transport/service_state_machine.cpp
@@ -373,9 +373,7 @@ Future<void> ServiceStateMachine::Impl::processMessage() {
// Format our response, if we have one
Message& toSink = dbresponse.response;
if (!toSink.empty()) {
- tassert(ErrorCodes::InternalError,
- "Attempted to respond to fire-and-forget request",
- !OpMsg::isFlagSet(_inMessage, OpMsg::kMoreToCome));
+ invariant(!OpMsg::isFlagSet(_inMessage, OpMsg::kMoreToCome));
invariant(!OpMsg::isFlagSet(toSink, OpMsg::kChecksumPresent));
// Update the header for the response message.
diff --git a/src/mongo/transport/transport_layer_asio.cpp b/src/mongo/transport/transport_layer_asio.cpp
index ca35c4a5337..85b5e816930 100644
--- a/src/mongo/transport/transport_layer_asio.cpp
+++ b/src/mongo/transport/transport_layer_asio.cpp
@@ -104,7 +104,6 @@ boost::optional<Status> maybeTcpFastOpenStatus;
MONGO_FAIL_POINT_DEFINE(transportLayerASIOasyncConnectTimesOut);
MONGO_FAIL_POINT_DEFINE(transportLayerASIOhangBeforeAccept);
-MONGO_FAIL_POINT_DEFINE(transportLayerASIOasyncConnectReturnsConnectionError);
#ifdef MONGO_CONFIG_SSL
SSLConnectionContext::~SSLConnectionContext() = default;
@@ -297,11 +296,8 @@ TransportLayerASIO::Options::Options(const ServerGlobalParams* params,
maxConns(params->maxConns) {
}
-TransportLayerASIO::TimerService::TimerService(Options opt)
- : _reactor(std::make_shared<TransportLayerASIO::ASIOReactor>()) {
- if (opt.spawn)
- _spawn = std::move(opt.spawn);
-}
+TransportLayerASIO::TimerService::TimerService()
+ : _reactor(std::make_shared<TransportLayerASIO::ASIOReactor>()) {}
TransportLayerASIO::TimerService::~TimerService() {
stop();
@@ -319,13 +315,10 @@ void TransportLayerASIO::TimerService::start() {
auto lk = stdx::lock_guard(_mutex);
auto precondition = State::kInitialized;
if (_state.compareAndSwap(&precondition, State::kStarted)) {
- _thread = _spawn([reactor = _reactor] {
+ _thread = stdx::thread([reactor = _reactor] {
LOGV2_INFO(5490002, "Started a new thread for the timer service");
reactor->run();
-
- if (!serverGlobalParams.quiet.load()) {
- LOGV2_INFO(5490003, "Returning from the timer service thread");
- }
+ LOGV2_INFO(5490003, "Returning from the timer service thread");
});
}
}
@@ -680,7 +673,7 @@ StatusWith<TransportLayerASIO::ASIOSessionHandle> TransportLayerASIO::_doSyncCon
"connect (sync) TCP fast open",
logv2::LogSeverity::Info(),
ec);
- if (ec && tcpFastOpenIsConfigured) {
+ if (tcpFastOpenIsConfigured) {
return errorCodeToStatus(ec);
}
ec = std::error_code();
@@ -740,9 +733,6 @@ Future<SessionHandle> TransportLayerASIO::asyncConnect(
const ReactorHandle& reactor,
Milliseconds timeout,
std::shared_ptr<const SSLConnectionContext> transientSSLContext) {
- if (MONGO_unlikely(transportLayerASIOasyncConnectReturnsConnectionError.shouldFail()))
- return Status{ErrorCodes::ConnectionError, "Failing asyncConnect due to fail-point"};
-
if (transientSSLContext) {
uassert(ErrorCodes::InvalidSSLConfiguration,
"Specified transient SSL context but connection SSL mode is not set",
@@ -840,7 +830,7 @@ Future<SessionHandle> TransportLayerASIO::asyncConnect(
"connect (async) TCP fast open",
logv2::LogSeverity::Info(),
ec);
- if (ec && tcpFastOpenIsConfigured) {
+ if (tcpFastOpenIsConfigured) {
return futurize(ec);
}
#endif
@@ -1173,7 +1163,7 @@ Status TransportLayerASIO::setup() {
"acceptor TCP fast open",
logv2::LogSeverity::Info(),
ec);
- if (ec && tcpFastOpenIsConfigured) {
+ if (tcpFastOpenIsConfigured) {
return errorCodeToStatus(ec);
}
ec = std::error_code();
@@ -1305,10 +1295,9 @@ void TransportLayerASIO::_runListener() noexcept {
Status TransportLayerASIO::start() {
stdx::unique_lock lk(_mutex);
- if (_isShutdown) {
- LOGV2(6986801, "Cannot start an already shutdown TransportLayer");
- return ShutdownStatus;
- }
+
+ // Make sure we haven't shutdown already
+ invariant(!_isShutdown);
if (_listenerOptions.isIngress()) {
_listener.thread = stdx::thread([this] { _runListener(); });
@@ -1327,6 +1316,7 @@ void TransportLayerASIO::shutdown() {
// We were already stopped
return;
}
+
lk.unlock();
_timerService->stop();
lk.lock();
diff --git a/src/mongo/transport/transport_layer_asio.h b/src/mongo/transport/transport_layer_asio.h
index e5a77fcb242..c0ecf4b9b80 100644
--- a/src/mongo/transport/transport_layer_asio.h
+++ b/src/mongo/transport/transport_layer_asio.h
@@ -127,12 +127,7 @@ public:
*/
class TimerService {
public:
- using Spawn = std::function<stdx::thread(std::function<void()>)>;
- struct Options {
- Spawn spawn;
- };
- explicit TimerService(Options opt);
- TimerService() : TimerService(Options{}) {}
+ TimerService();
~TimerService();
/**
@@ -167,7 +162,6 @@ public:
enum class State { kInitialized, kStarted, kStopped };
AtomicWord<State> _state;
- Spawn _spawn = [](std::function<void()> f) { return stdx::thread{std::move(f)}; };
stdx::thread _thread;
};
diff --git a/src/mongo/transport/transport_layer_asio_test.cpp b/src/mongo/transport/transport_layer_asio_test.cpp
index b3f63792a26..d9803d00613 100644
--- a/src/mongo/transport/transport_layer_asio_test.cpp
+++ b/src/mongo/transport/transport_layer_asio_test.cpp
@@ -60,6 +60,7 @@
#include "mongo/util/scopeguard.h"
#include "mongo/util/static_immortal.h"
#include "mongo/util/synchronized_value.h"
+#include "mongo/util/thread_context.h"
#include "mongo/util/time_support.h"
#include "mongo/util/waitable.h"
@@ -527,21 +528,20 @@ TEST(TransportLayerASIO, EgressConnectionResetByPeerDuringSessionCtor) {
// `fp` pauses the `ASIOSession` constructor immediately prior to its
// `setsockopt` sequence, to allow time for the peer reset to propagate.
- auto fp = std::make_unique<FailPointEnableBlock>(
- "transportLayerASIOSessionPauseBeforeSetSocketOption");
+ FailPoint& fp = transport::transportLayerASIOSessionPauseBeforeSetSocketOption;
Acceptor server(ioContext);
server.setOnAccept([&](std::shared_ptr<Acceptor::Connection> conn) {
- LOGV2(7598701, "waiting for the client to reach the fail-point");
- (*fp)->waitForTimesEntered(fp->initialTimesEntered() + 1);
LOGV2(6101604, "handling a connection by resetting it");
conn->socket.set_option(asio::socket_base::linger(true, 0));
conn->socket.close();
- fp.reset();
+ sleepFor(Seconds{1});
+ fp.setMode(FailPoint::off);
});
JoinThread ioThread{[&] { ioContext.run(); }};
ScopeGuard ioContextStop = [&] { ioContext.stop(); };
+ fp.setMode(FailPoint::alwaysOn);
LOGV2(6101602, "Connecting", "port"_attr = server.port());
using namespace unittest::match;
// On MacOS, calling `setsockopt` on a peer-reset connection yields an
@@ -600,50 +600,45 @@ TEST(TransportLayerASIO, ConfirmSocketSetOptionOnResetConnections) {
class TransportLayerASIOWithServiceContextTest : public ServiceContextTest {
public:
+ /**
+ * `ThreadCounter` and `ThreadToken` allow tracking the number of active (running) threads.
+ * For each thread, a `ThreadToken` is created. The token notifies `ThreadCounter` about
+ * creation and destruction of its associated thread. This allows maintaining the number of
+ * active threads at any point during the execution of this unit-test.
+ */
class ThreadCounter {
public:
- std::function<stdx::thread(std::function<void()>)> makeSpawnFunc() {
- return [core = _core](std::function<void()> cb) {
- {
- stdx::lock_guard lk(core->mutex);
- ++core->created;
- core->cv.notify_all();
- }
- return stdx::thread{[core, cb = std::move(cb)]() mutable {
- {
- stdx::lock_guard lk(core->mutex);
- ++core->started;
- core->cv.notify_all();
- }
- cb();
- }};
- };
+ static ThreadCounter& get() {
+ static StaticImmortal<ThreadCounter> instance;
+ return *instance;
}
- int64_t created() const {
- stdx::lock_guard lk(_core->mutex);
- return _core->created;
+ int64_t count() const {
+ const auto count = _count.load();
+ invariant(count > 0);
+ return count;
}
- int64_t started() const {
- stdx::lock_guard lk(_core->mutex);
- return _core->started;
+ void onCreateThread() {
+ _count.fetchAndAdd(1);
}
- template <typename Pred>
- void waitForStarted(const Pred& pred) const {
- stdx::unique_lock lk(_core->mutex);
- _core->cv.wait(lk, [&] { return pred(_core->started); });
+ void onDestroyThread() {
+ _count.fetchAndAdd(-1);
}
private:
- struct Core {
- mutable stdx::mutex mutex; // NOLINT
- mutable stdx::condition_variable cv;
- int64_t created = 0;
- int64_t started = 0;
- };
- std::shared_ptr<Core> _core = std::make_shared<Core>();
+ AtomicWord<int64_t> _count;
+ };
+
+ struct ThreadToken {
+ ThreadToken() {
+ ThreadCounter::get().onCreateThread();
+ }
+
+ ~ThreadToken() {
+ ThreadCounter::get().onDestroyThread();
+ }
};
void setUp() override {
@@ -663,32 +658,46 @@ public:
}
};
+const auto getThreadToken =
+ ThreadContext::declareDecoration<TransportLayerASIOWithServiceContextTest::ThreadToken>();
+
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceDoesNotSpawnThreadsBeforeStart) {
- ThreadCounter counter;
- { transport::TransportLayerASIO::TimerService service{{counter.makeSpawnFunc()}}; }
- ASSERT_EQ(counter.created(), 0);
+ const auto beforeThreadCount = ThreadCounter::get().count();
+ transport::TransportLayerASIO::TimerService service;
+ // Note that the following is a best-effort and not deterministic as we don't have control over
+ // when threads may start running and advance the thread count.
+ const auto afterThreadCount = ThreadCounter::get().count();
+ ASSERT_EQ(beforeThreadCount, afterThreadCount);
}
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceOneShotStart) {
- ThreadCounter counter;
- transport::TransportLayerASIO::TimerService service{{counter.makeSpawnFunc()}};
+ const auto beforeThreadCount = ThreadCounter::get().count();
+ transport::TransportLayerASIO::TimerService service;
service.start();
- LOGV2(5490004, "Awaiting timer thread start", "threads"_attr = counter.started());
- counter.waitForStarted([](auto n) { return n > 0; });
- LOGV2(5490005, "Awaited timer thread start", "threads"_attr = counter.started());
+ LOGV2(5490004, "Waiting for the timer thread to start", "threads"_attr = beforeThreadCount);
+ while (ThreadCounter::get().count() == beforeThreadCount) {
+ sleepFor(Milliseconds(1));
+ }
+ const auto afterThreadCount = ThreadCounter::get().count();
+ LOGV2(5490005, "Returned from waiting for the timer thread", "threads"_attr = afterThreadCount);
+ // Start the service a few times and verify that the thread count has not changed. Note that the
+ // following is a best-effort and not deterministic as we don't have control over when threads
+ // may start running and advance the thread count.
service.start();
service.start();
service.start();
- ASSERT_EQ(counter.created(), 1) << "Redundant start should spawn only once";
+ ASSERT_EQ(afterThreadCount, ThreadCounter::get().count());
}
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceDoesNotStartAfterStop) {
- ThreadCounter counter;
- transport::TransportLayerASIO::TimerService service{{counter.makeSpawnFunc()}};
+ const auto beforeThreadCount = ThreadCounter::get().count();
+ transport::TransportLayerASIO::TimerService service;
service.stop();
service.start();
- ASSERT_EQ(counter.created(), 0) << "Stop then start should not spawn";
+ const auto afterThreadCount = ThreadCounter::get().count();
+ // The test would fail if `start` proceeds to spawn a thread for `service`.
+ ASSERT_EQ(beforeThreadCount, afterThreadCount);
}
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceCanStopMoreThanOnce) {
@@ -706,11 +715,6 @@ TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceCanStopMoreThanOnce
}
}
-TEST_F(TransportLayerASIOWithServiceContextTest, TransportStartAfterShutDown) {
- tla().shutdown();
- ASSERT_EQ(tla().start(), transport::TransportLayer::ShutdownStatus);
-}
-
#ifdef MONGO_CONFIG_SSL
#ifndef _WIN32
// TODO SERVER-62035: enable the following on Windows.
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 13:32:18 +0000