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.h6
-rw-r--r--src/mongo/transport/transport_layer_asio.cpp26
-rw-r--r--src/mongo/transport/transport_layer_asio.h8
-rw-r--r--src/mongo/transport/transport_layer_asio_test.cpp114
5 files changed, 85 insertions, 71 deletions
diff --git a/src/mongo/transport/asio_utils.cpp b/src/mongo/transport/asio_utils.cpp
index 8e8368ff586..61d9a56dea2 100644
--- a/src/mongo/transport/asio_utils.cpp
+++ b/src/mongo/transport/asio_utils.cpp
@@ -55,6 +55,8 @@ 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.h b/src/mongo/transport/message_compressor_manager.h
index 9f1a1de35e3..bccbebdb316 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 an isMaster request. This function will append the result
+ * Called by a client constructing a "hello" 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 an isMaster response (received after calling
+ * Called by a client that has received a "hello" 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 an isMaster request.
+ * Called by a server that has received a "hello" request.
*
* If no compressors are configured that match those requested by the client, then it will
* not append anything to the BSONObjBuilder output.
diff --git a/src/mongo/transport/transport_layer_asio.cpp b/src/mongo/transport/transport_layer_asio.cpp
index 85b5e816930..53076c84099 100644
--- a/src/mongo/transport/transport_layer_asio.cpp
+++ b/src/mongo/transport/transport_layer_asio.cpp
@@ -104,6 +104,7 @@ 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;
@@ -296,8 +297,11 @@ TransportLayerASIO::Options::Options(const ServerGlobalParams* params,
maxConns(params->maxConns) {
}
-TransportLayerASIO::TimerService::TimerService()
- : _reactor(std::make_shared<TransportLayerASIO::ASIOReactor>()) {}
+TransportLayerASIO::TimerService::TimerService(Options opt)
+ : _reactor(std::make_shared<TransportLayerASIO::ASIOReactor>()) {
+ if (opt.spawn)
+ _spawn = std::move(opt.spawn);
+}
TransportLayerASIO::TimerService::~TimerService() {
stop();
@@ -315,10 +319,13 @@ void TransportLayerASIO::TimerService::start() {
auto lk = stdx::lock_guard(_mutex);
auto precondition = State::kInitialized;
if (_state.compareAndSwap(&precondition, State::kStarted)) {
- _thread = stdx::thread([reactor = _reactor] {
+ _thread = _spawn([reactor = _reactor] {
LOGV2_INFO(5490002, "Started a new thread for the timer service");
reactor->run();
- LOGV2_INFO(5490003, "Returning from the timer service thread");
+
+ if (!serverGlobalParams.quiet.load()) {
+ LOGV2_INFO(5490003, "Returning from the timer service thread");
+ }
});
}
}
@@ -733,6 +740,9 @@ 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",
@@ -1295,9 +1305,10 @@ void TransportLayerASIO::_runListener() noexcept {
Status TransportLayerASIO::start() {
stdx::unique_lock lk(_mutex);
-
- // Make sure we haven't shutdown already
- invariant(!_isShutdown);
+ if (_isShutdown) {
+ LOGV2(6986801, "Cannot start an already shutdown TransportLayer");
+ return ShutdownStatus;
+ }
if (_listenerOptions.isIngress()) {
_listener.thread = stdx::thread([this] { _runListener(); });
@@ -1316,7 +1327,6 @@ 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 c0ecf4b9b80..e5a77fcb242 100644
--- a/src/mongo/transport/transport_layer_asio.h
+++ b/src/mongo/transport/transport_layer_asio.h
@@ -127,7 +127,12 @@ public:
*/
class TimerService {
public:
- TimerService();
+ using Spawn = std::function<stdx::thread(std::function<void()>)>;
+ struct Options {
+ Spawn spawn;
+ };
+ explicit TimerService(Options opt);
+ TimerService() : TimerService(Options{}) {}
~TimerService();
/**
@@ -162,6 +167,7 @@ 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 d9803d00613..b3f63792a26 100644
--- a/src/mongo/transport/transport_layer_asio_test.cpp
+++ b/src/mongo/transport/transport_layer_asio_test.cpp
@@ -60,7 +60,6 @@
#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"
@@ -528,20 +527,21 @@ TEST(TransportLayerASIO, EgressConnectionResetByPeerDuringSessionCtor) {
// `fp` pauses the `ASIOSession` constructor immediately prior to its
// `setsockopt` sequence, to allow time for the peer reset to propagate.
- FailPoint& fp = transport::transportLayerASIOSessionPauseBeforeSetSocketOption;
+ auto fp = std::make_unique<FailPointEnableBlock>(
+ "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();
- sleepFor(Seconds{1});
- fp.setMode(FailPoint::off);
+ fp.reset();
});
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,45 +600,50 @@ 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:
- static ThreadCounter& get() {
- static StaticImmortal<ThreadCounter> instance;
- return *instance;
+ 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();
+ }};
+ };
}
- int64_t count() const {
- const auto count = _count.load();
- invariant(count > 0);
- return count;
+ int64_t created() const {
+ stdx::lock_guard lk(_core->mutex);
+ return _core->created;
}
- void onCreateThread() {
- _count.fetchAndAdd(1);
+ int64_t started() const {
+ stdx::lock_guard lk(_core->mutex);
+ return _core->started;
}
- void onDestroyThread() {
- _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); });
}
private:
- AtomicWord<int64_t> _count;
- };
-
- struct ThreadToken {
- ThreadToken() {
- ThreadCounter::get().onCreateThread();
- }
-
- ~ThreadToken() {
- ThreadCounter::get().onDestroyThread();
- }
+ 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>();
};
void setUp() override {
@@ -658,46 +663,32 @@ public:
}
};
-const auto getThreadToken =
- ThreadContext::declareDecoration<TransportLayerASIOWithServiceContextTest::ThreadToken>();
-
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceDoesNotSpawnThreadsBeforeStart) {
- 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);
+ ThreadCounter counter;
+ { transport::TransportLayerASIO::TimerService service{{counter.makeSpawnFunc()}}; }
+ ASSERT_EQ(counter.created(), 0);
}
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceOneShotStart) {
- const auto beforeThreadCount = ThreadCounter::get().count();
- transport::TransportLayerASIO::TimerService service;
+ ThreadCounter counter;
+ transport::TransportLayerASIO::TimerService service{{counter.makeSpawnFunc()}};
service.start();
- 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);
+ 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());
- // 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(afterThreadCount, ThreadCounter::get().count());
+ ASSERT_EQ(counter.created(), 1) << "Redundant start should spawn only once";
}
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceDoesNotStartAfterStop) {
- const auto beforeThreadCount = ThreadCounter::get().count();
- transport::TransportLayerASIO::TimerService service;
+ ThreadCounter counter;
+ transport::TransportLayerASIO::TimerService service{{counter.makeSpawnFunc()}};
service.stop();
service.start();
- const auto afterThreadCount = ThreadCounter::get().count();
- // The test would fail if `start` proceeds to spawn a thread for `service`.
- ASSERT_EQ(beforeThreadCount, afterThreadCount);
+ ASSERT_EQ(counter.created(), 0) << "Stop then start should not spawn";
}
TEST_F(TransportLayerASIOWithServiceContextTest, TimerServiceCanStopMoreThanOnce) {
@@ -715,6 +706,11 @@ 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 11:10:11 +0000