SERVER-48980 Associate ServiceExecutors with Clients

8 files changed, 316 insertions, 108 deletions

diff --git a/src/mongo/transport/SConscript b/src/mongo/transport/SConscript
index 9ac3bf2c329..7932abc662d 100644
--- a/src/mongo/transport/SConscript
+++ b/src/mongo/transport/SConscript
@@ -86,6 +86,7 @@ env.Library(
 tlEnv.Library(
     target='service_executor',
     source=[
+        'service_executor.cpp',
         'service_executor_fixed.cpp',
         'service_executor_reserved.cpp',
         'service_executor_synchronous.cpp',
diff --git a/src/mongo/transport/service_entry_point.h b/src/mongo/transport/service_entry_point.h
index aa970f1ef67..fb1ad895d78 100644
--- a/src/mongo/transport/service_entry_point.h
+++ b/src/mongo/transport/service_entry_point.h
@@ -29,6 +29,8 @@
 
 #pragma once
 
+#include <limits>
+
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/dbmessage.h"
 #include "mongo/transport/session.h"
@@ -81,6 +83,13 @@ public:
     virtual size_t numOpenSessions() const = 0;
 
     /**
+     * Returns the maximum number of sessions that can be open.
+     */
+    virtual size_t maxOpenSessions() const {
+        return std::numeric_limits<size_t>::max();
+    }
+
+    /**
      * Processes a request and fills out a DbResponse.
      */
     virtual Future<DbResponse> handleRequest(OperationContext* opCtx,
diff --git a/src/mongo/transport/service_entry_point_impl.cpp b/src/mongo/transport/service_entry_point_impl.cpp
index ec35471ede6..0090f30a686 100644
--- a/src/mongo/transport/service_entry_point_impl.cpp
+++ b/src/mongo/transport/service_entry_point_impl.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/transport/service_entry_point_impl.h"
 
+#include <fmt/format.h>
 #include <vector>
 
 #include "mongo/db/auth/restriction_environment.h"
@@ -54,10 +55,17 @@
 
 namespace mongo {
 
+using namespace fmt::literals;
+
 bool shouldOverrideMaxConns(const transport::SessionHandle& session,
                             const std::vector<stdx::variant<CIDR, std::string>>& exemptions) {
+    if (exemptions.empty()) {
+        return false;
+    }
+
     const auto& remoteAddr = session->remoteAddr();
     const auto& localAddr = session->localAddr();
+
     boost::optional<CIDR> remoteCIDR;
 
     if (remoteAddr.isValid() && remoteAddr.isIP()) {
@@ -85,8 +93,7 @@ bool shouldOverrideMaxConns(const transport::SessionHandle& session,
     return false;
 }
 
-ServiceEntryPointImpl::ServiceEntryPointImpl(ServiceContext* svcCtx) : _svcCtx(svcCtx) {
-
+size_t getSupportedMax() {
     const auto supportedMax = [] {
 #ifdef _WIN32
         return serverGlobalParams.maxConns;
@@ -117,9 +124,12 @@ ServiceEntryPointImpl::ServiceEntryPointImpl(ServiceContext* svcCtx) : _svcCtx(s
               "limit"_attr = supportedMax);
     }
 
-    _maxNumConnections = supportedMax;
+    return supportedMax;
 }
 
+ServiceEntryPointImpl::ServiceEntryPointImpl(ServiceContext* svcCtx)
+    : _svcCtx(svcCtx), _maxNumConnections(getSupportedMax()) {}
+
 Status ServiceEntryPointImpl::start() {
     if (auto status = transport::ServiceExecutorSynchronous::get(_svcCtx)->start();
         !status.isOK()) {
@@ -149,44 +159,48 @@ void ServiceEntryPointImpl::startSession(transport::SessionHandle session) {
     auto restrictionEnvironment = std::make_unique<RestrictionEnvironment>(remoteAddr, localAddr);
     RestrictionEnvironment::set(session, std::move(restrictionEnvironment));
 
-    SSMListIterator ssmIt;
+    bool canOverrideMaxConns = shouldOverrideMaxConns(session, serverGlobalParams.maxConnsOverride);
 
-    const bool quiet = serverGlobalParams.quiet.load();
-    size_t connectionCount;
-    auto transportMode = transport::ServiceExecutorSynchronous::get(_svcCtx)->transportMode();
+    auto clientName = "conn{}"_format(session->id());
+    auto client = _svcCtx->makeClient(clientName, session);
 
-    auto ssm = ServiceStateMachine::create(_svcCtx, session, transportMode);
-    auto usingMaxConnOverride = false;
     {
-        stdx::lock_guard<decltype(_sessionsMutex)> lk(_sessionsMutex);
-        connectionCount = _sessions.size() + 1;
-        if (connectionCount > _maxNumConnections) {
-            usingMaxConnOverride =
-                shouldOverrideMaxConns(session, serverGlobalParams.maxConnsOverride);
-        }
+        stdx::lock_guard lk(*client);
+        auto seCtx =
+            transport::ServiceExecutorContext{}
+                .setThreadingModel(transport::ServiceExecutorContext::ThreadingModel::kDedicated)
+                .setCanUseReserved(canOverrideMaxConns);
 
-        if (connectionCount <= _maxNumConnections || usingMaxConnOverride) {
-            ssmIt = _sessions.emplace(_sessions.begin(), ssm);
-            _currentConnections.store(connectionCount);
-            _createdConnections.addAndFetch(1);
-        }
+        transport::ServiceExecutorContext::set(client.get(), std::move(seCtx));
     }
 
-    // Checking if we successfully added a connection above. Separated from the lock so we don't log
-    // while holding it.
-    if (connectionCount > _maxNumConnections && !usingMaxConnOverride) {
+    auto ssm = std::make_shared<transport::ServiceStateMachine>(std::move(client));
+
+    const bool quiet = serverGlobalParams.quiet.load();
+
+    size_t connectionCount;
+    auto ssmIt = [&]() -> boost::optional<SSMListIterator> {
+        stdx::lock_guard lk(_sessionsMutex);
+        connectionCount = _currentConnections.load();
+        if (connectionCount > _maxNumConnections && !canOverrideMaxConns) {
+            return boost::none;
+        }
+
+        auto it = _sessions.emplace(_sessions.begin(), ssm);
+        connectionCount = _sessions.size();
+        _currentConnections.store(connectionCount);
+        _createdConnections.addAndFetch(1);
+        return it;
+    }();
+
+    if (!ssmIt) {
         if (!quiet) {
             LOGV2(22942,
                   "Connection refused because there are too many open connections",
                   "connectionCount"_attr = connectionCount);
         }
         return;
-    } else if (auto exec = transport::ServiceExecutorReserved::get(_svcCtx);
-               usingMaxConnOverride && exec) {
-        ssm->setServiceExecutor(exec);
-    }
-
-    if (!quiet) {
+    } else if (!quiet) {
         LOGV2(22943,
               "Connection accepted",
               "remote"_attr = session->remote(),
@@ -199,7 +213,7 @@ void ServiceEntryPointImpl::startSession(transport::SessionHandle session) {
         auto remote = session->remote();
         {
             stdx::lock_guard<decltype(_sessionsMutex)> lk(_sessionsMutex);
-            _sessions.erase(ssmIt);
+            _sessions.erase(*ssmIt);
             connectionCount = _sessions.size();
             _currentConnections.store(connectionCount);
         }
@@ -214,11 +228,7 @@ void ServiceEntryPointImpl::startSession(transport::SessionHandle session) {
         }
     });
 
-    auto ownership = ServiceStateMachine::Ownership::kOwned;
-    if (transportMode == transport::Mode::kSynchronous) {
-        ownership = ServiceStateMachine::Ownership::kStatic;
-    }
-    ssm->start(ownership);
+    ssm->start();
 }
 
 void ServiceEntryPointImpl::endAllSessions(transport::Session::TagMask tags) {
diff --git a/src/mongo/transport/service_entry_point_impl.h b/src/mongo/transport/service_entry_point_impl.h
index 0528a19fbdb..1693469fcda 100644
--- a/src/mongo/transport/service_entry_point_impl.h
+++ b/src/mongo/transport/service_entry_point_impl.h
@@ -77,8 +77,12 @@ public:
         return _currentConnections.load();
     }
 
+    size_t maxOpenSessions() const final {
+        return _maxNumConnections;
+    }
+
 private:
-    using SSMList = std::list<std::shared_ptr<ServiceStateMachine>>;
+    using SSMList = std::list<std::shared_ptr<transport::ServiceStateMachine>>;
     using SSMListIterator = SSMList::iterator;
 
     ServiceContext* const _svcCtx;
@@ -89,7 +93,7 @@ private:
     stdx::condition_variable _shutdownCondition;
     SSMList _sessions;
 
-    size_t _maxNumConnections{DEFAULT_MAX_CONN};
+    const size_t _maxNumConnections{DEFAULT_MAX_CONN};
     AtomicWord<size_t> _currentConnections{0};
     AtomicWord<size_t> _createdConnections{0};
 };
diff --git a/src/mongo/transport/service_executor.cpp b/src/mongo/transport/service_executor.cpp
new file mode 100644
index 00000000000..37c3d03e1b1
--- /dev/null
+++ b/src/mongo/transport/service_executor.cpp
@@ -0,0 +1,135 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kNetwork
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/transport/service_executor.h"
+
+#include <boost/optional.hpp>
+
+#include "mongo/logv2/log.h"
+#include "mongo/transport/service_entry_point.h"
+#include "mongo/transport/service_executor_fixed.h"
+#include "mongo/transport/service_executor_reserved.h"
+#include "mongo/transport/service_executor_synchronous.h"
+
+namespace mongo {
+namespace transport {
+namespace {
+static constexpr auto kDiagnosticLogLevel = 4;
+
+auto getServiceExecutorContext =
+    Client::declareDecoration<boost::optional<ServiceExecutorContext>>();
+}  // namespace
+
+StringData toString(ServiceExecutorContext::ThreadingModel threadingModel) {
+    switch (threadingModel) {
+        case ServiceExecutorContext::ThreadingModel::kDedicated:
+            return "Dedicated"_sd;
+        case ServiceExecutorContext::ThreadingModel::kBorrowed:
+            return "Borrowed"_sd;
+        default:
+            MONGO_UNREACHABLE;
+    }
+}
+
+ServiceExecutorContext* ServiceExecutorContext::get(Client* client) noexcept {
+    auto& serviceExecutorContext = getServiceExecutorContext(client);
+
+    if (!serviceExecutorContext) {
+        // Service worker Clients will never have a ServiceExecutorContext.
+        return nullptr;
+    }
+
+    return &serviceExecutorContext.get();
+}
+
+void ServiceExecutorContext::set(Client* client, ServiceExecutorContext seCtx) noexcept {
+    auto& serviceExecutorContext = getServiceExecutorContext(client);
+    invariant(!serviceExecutorContext);
+
+    seCtx._client = client;
+    seCtx._sep = client->getServiceContext()->getServiceEntryPoint();
+
+    LOGV2_DEBUG(4898000,
+                kDiagnosticLogLevel,
+                "Setting initial ServiceExecutor context for client",
+                "client"_attr = client->desc(),
+                "threadingModel"_attr = seCtx._threadingModel,
+                "canUseReserved"_attr = seCtx._canUseReserved);
+    serviceExecutorContext = std::move(seCtx);
+}
+
+ServiceExecutorContext& ServiceExecutorContext::setThreadingModel(
+    ThreadingModel threadingModel) noexcept {
+    _threadingModel = threadingModel;
+    return *this;
+}
+
+ServiceExecutorContext& ServiceExecutorContext::setCanUseReserved(bool canUseReserved) noexcept {
+    _canUseReserved = canUseReserved;
+    return *this;
+}
+
+ServiceExecutor* ServiceExecutorContext::getServiceExecutor() const noexcept {
+    invariant(_client);
+
+    switch (_threadingModel) {
+        case ThreadingModel::kBorrowed:
+            return ServiceExecutorFixed::get(_client->getServiceContext());
+        case ThreadingModel::kDedicated: {
+            // Continue on.
+        } break;
+        default:
+            MONGO_UNREACHABLE;
+    }
+
+    auto shouldUseReserved = [&] {
+        // This is at best a naive solution. There could be a world where numOpenSessions() changes
+        // very quickly. We are not taking locks on the ServiceEntryPoint, so we may chose to
+        // schedule onto the ServiceExecutorReserved when it is no longer necessary. The upside is
+        // that we will automatically shift to the ServiceExecutorSynchronous after the first
+        // command loop.
+        return _sep->numOpenSessions() > _sep->maxOpenSessions();
+    };
+
+    if (_canUseReserved && shouldUseReserved()) {
+        if (auto exec = transport::ServiceExecutorReserved::get(_client->getServiceContext())) {
+            // We are allowed to use the reserved executor, we should use it, and it exists.
+            return exec;
+        }
+    }
+
+    return transport::ServiceExecutorSynchronous::get(_client->getServiceContext());
+}
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/transport/service_executor.h b/src/mongo/transport/service_executor.h
index b702198e6b5..38116908272 100644
--- a/src/mongo/transport/service_executor.h
+++ b/src/mongo/transport/service_executor.h
@@ -33,20 +33,19 @@
 
 #include "mongo/base/status.h"
 #include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/client.h"
+#include "mongo/db/service_context.h"
 #include "mongo/platform/bitwise_enum_operators.h"
+#include "mongo/transport/service_entry_point.h"
+#include "mongo/transport/session.h"
 #include "mongo/transport/transport_mode.h"
 #include "mongo/util/duration.h"
 #include "mongo/util/functional.h"
 #include "mongo/util/out_of_line_executor.h"
 
 namespace mongo {
-// This needs to be forward declared here because the service_context.h is a circular dependency.
-class ServiceContext;
-
 namespace transport {
 
-class Session;
-
 /*
  * This is the interface for all ServiceExecutors.
  */
@@ -123,6 +122,74 @@ public:
     virtual void appendStats(BSONObjBuilder* bob) const = 0;
 };
 
+/**
+ * ServiceExecutorContext determines which ServiceExecutor is used for each Client.
+ */
+class ServiceExecutorContext {
+public:
+    enum ThreadingModel {
+        kBorrowed,
+        kDedicated,
+    };
+
+    /**
+     * Get a pointer to the ServiceExecutorContext for a given client.
+     *
+     * This function is valid to invoke either on the Client thread or with the Client lock.
+     */
+    static ServiceExecutorContext* get(Client* client) noexcept;
+
+    /**
+     * Set the ServiceExecutorContext for a given client.
+     *
+     * This function may only be invoked once and only while under the Client lock.
+     */
+    static void set(Client* client, ServiceExecutorContext seCtx) noexcept;
+
+    ServiceExecutorContext() = default;
+
+    /**
+     * Set the ThreadingModel for the associated Client's service execution.
+     *
+     * This function is only valid to invoke with the Client lock or before the Client is set.
+     */
+    ServiceExecutorContext& setThreadingModel(ThreadingModel threadingModel) noexcept;
+
+    /**
+     * Set if reserved resources are available for the associated Client's service execution.
+     *
+     * This function is only valid to invoke with the Client lock or before the Client is set.
+     */
+    ServiceExecutorContext& setCanUseReserved(bool canUseReserved) noexcept;
+
+    /**
+     * Get the ThreadingModel for the associated Client.
+     *
+     * This function is valid to invoke either on the Client thread or with the Client lock.
+     */
+    auto getThreadingModel() const noexcept {
+        return _threadingModel;
+    }
+
+    /**
+     * Get an appropriate ServiceExecutor given the current parameters.
+     *
+     * This function is only valid to invoke from the associated Client thread. This function does
+     * not require the Client lock since all writes must also happen from that thread.
+     */
+    ServiceExecutor* getServiceExecutor() const noexcept;
+
+private:
+    friend StringData toString(ThreadingModel threadingModel);
+
+    Client* _client = nullptr;
+    ServiceEntryPoint* _sep = nullptr;
+
+    ThreadingModel _threadingModel = ThreadingModel::kDedicated;
+    bool _canUseReserved = false;
+};
+
+
 }  // namespace transport
 
 ENABLE_BITMASK_OPERATORS(transport::ServiceExecutor::ScheduleFlags)
diff --git a/src/mongo/transport/service_state_machine.cpp b/src/mongo/transport/service_state_machine.cpp
index b71e6c94abd..5124542291b 100644
--- a/src/mongo/transport/service_state_machine.cpp
+++ b/src/mongo/transport/service_state_machine.cpp
@@ -36,7 +36,6 @@
 #include <memory>
 
 #include "mongo/config.h"
-#include "mongo/db/client.h"
 #include "mongo/db/dbmessage.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/db/traffic_recorder.h"
@@ -45,6 +44,7 @@
 #include "mongo/rpc/op_msg.h"
 #include "mongo/transport/message_compressor_manager.h"
 #include "mongo/transport/service_entry_point.h"
+#include "mongo/transport/service_executor_fixed.h"
 #include "mongo/transport/service_executor_synchronous.h"
 #include "mongo/transport/session.h"
 #include "mongo/transport/transport_layer.h"
@@ -59,6 +59,7 @@
 #include "mongo/util/quick_exit.h"
 
 namespace mongo {
+namespace transport {
 namespace {
 MONGO_FAIL_POINT_DEFINE(doNotSetMoreToCome);
 /**
@@ -160,7 +161,6 @@ Message makeExhaustMessage(Message requestMsg, DbResponse* dbresponse) {
 
     return exhaustMessage;
 }
-
 }  // namespace
 
 using transport::ServiceExecutor;
@@ -194,9 +194,10 @@ public:
 
         // Set up the thread name
         auto oldThreadName = getThreadName();
-        if (oldThreadName != _ssm->_threadName) {
-            _ssm->_oldThreadName = getThreadName().toString();
-            setThreadName(_ssm->_threadName);
+        const auto& threadName = _ssm->_dbClient->desc();
+        if (oldThreadName != threadName) {
+            _oldThreadName = oldThreadName.toString();
+            setThreadName(threadName);
         }
 
         // Swap the current Client so calls to cc() work as expected
@@ -258,8 +259,8 @@ public:
                 _ssm->_dbClient = Client::releaseCurrent();
             }
 
-            if (!_ssm->_oldThreadName.empty()) {
-                setThreadName(_ssm->_oldThreadName);
+            if (!_oldThreadName.empty()) {
+                setThreadName(_oldThreadName);
             }
         }
 
@@ -287,29 +288,22 @@ public:
 private:
     ServiceStateMachine* _ssm;
     bool _haveTakenOwnership = false;
+    std::string _oldThreadName;
 };
 
-std::shared_ptr<ServiceStateMachine> ServiceStateMachine::create(ServiceContext* svcContext,
-                                                                 transport::SessionHandle session,
-                                                                 transport::Mode transportMode) {
-    return std::make_shared<ServiceStateMachine>(svcContext, std::move(session), transportMode);
-}
-
-ServiceStateMachine::ServiceStateMachine(ServiceContext* svcContext,
-                                         transport::SessionHandle session,
-                                         transport::Mode transportMode)
+ServiceStateMachine::ServiceStateMachine(ServiceContext::UniqueClient client)
     : _state{State::Created},
-      _sep{svcContext->getServiceEntryPoint()},
-      _transportMode(transportMode),
-      _serviceContext(svcContext),
-      _sessionHandle(session),
-      _threadName{str::stream() << "conn" << _session()->id()},
-      _dbClient{svcContext->makeClient(_threadName, std::move(session))},
-      _dbClientPtr{_dbClient.get()},
-      _serviceExecutor(transport::ServiceExecutorSynchronous::get(_serviceContext)) {}
+      _serviceContext{client->getServiceContext()},
+      _sep{_serviceContext->getServiceEntryPoint()},
+      _dbClient{std::move(client)},
+      _dbClientPtr{_dbClient.get()} {}
 
 const transport::SessionHandle& ServiceStateMachine::_session() const {
-    return _sessionHandle;
+    return _dbClientPtr->session();
+}
+
+ServiceExecutor* ServiceStateMachine::_executor() {
+    return ServiceExecutorContext::get(_dbClientPtr)->getServiceExecutor();
 }
 
 Future<void> ServiceStateMachine::_sourceMessage(ThreadGuard guard) {
@@ -319,11 +313,12 @@ Future<void> ServiceStateMachine::_sourceMessage(ThreadGuard guard) {
     guard.release();
 
     auto sourceMsgImpl = [&] {
-        if (_transportMode == transport::Mode::kSynchronous) {
+        const auto& transportMode = _executor()->transportMode();
+        if (transportMode == transport::Mode::kSynchronous) {
             MONGO_IDLE_THREAD_BLOCK;
             return Future<Message>::makeReady(_session()->sourceMessage());
         } else {
-            invariant(_transportMode == transport::Mode::kAsynchronous);
+            invariant(transportMode == transport::Mode::kAsynchronous);
             return _session()->asyncSourceMessage();
         }
     };
@@ -346,13 +341,14 @@ Future<void> ServiceStateMachine::_sinkMessage(ThreadGuard guard) {
     auto toSink = std::exchange(_outMessage, {});
 
     auto sinkMsgImpl = [&] {
-        if (_transportMode == transport::Mode::kSynchronous) {
+        const auto& transportMode = _executor()->transportMode();
+        if (transportMode == transport::Mode::kSynchronous) {
             // We don't consider ourselves idle while sending the reply since we are still doing
             // work on behalf of the client. Contrast that with sourceMessage() where we are waiting
             // for the client to send us more work to do.
             return Future<void>::makeReady(_session()->sinkMessage(std::move(toSink)));
         } else {
-            invariant(_transportMode == transport::Mode::kAsynchronous);
+            invariant(transportMode == transport::Mode::kAsynchronous);
             return _session()->asyncSinkMessage(std::move(toSink));
         }
     };
@@ -439,7 +435,7 @@ Future<void> ServiceStateMachine::_processMessage(ThreadGuard guard) {
     invariant(!_inMessage.empty());
 
     TrafficRecorder::get(_serviceContext)
-        .observe(_sessionHandle, _serviceContext->getPreciseClockSource()->now(), _inMessage);
+        .observe(_session(), _serviceContext->getPreciseClockSource()->now(), _inMessage);
 
     auto& compressorMgr = MessageCompressorManager::forSession(_session());
 
@@ -511,8 +507,7 @@ Future<void> ServiceStateMachine::_processMessage(ThreadGuard guard) {
                 }
 
                 TrafficRecorder::get(_serviceContext)
-                    .observe(
-                        _sessionHandle, _serviceContext->getPreciseClockSource()->now(), toSink);
+                    .observe(_session(), _serviceContext->getPreciseClockSource()->now(), toSink);
 
                 _outMessage = std::move(toSink);
             } else {
@@ -524,16 +519,13 @@ Future<void> ServiceStateMachine::_processMessage(ThreadGuard guard) {
         });
 }
 
-void ServiceStateMachine::setServiceExecutor(ServiceExecutor* executor) {
-    _serviceExecutor = executor;
-}
-
-void ServiceStateMachine::start(Ownership ownershipModel) {
-    _serviceExecutor->schedule(GuaranteedExecutor::enforceRunOnce(
-        [this, anchor = shared_from_this(), ownershipModel](Status status) {
+void ServiceStateMachine::start() {
+    _executor()->schedule(
+        GuaranteedExecutor::enforceRunOnce([this, anchor = shared_from_this()](Status status) {
+            // TODO(SERVER-49109) We can't use static ownership in general with
+            // a ServiceExecutorFixed and async commands. ThreadGuard needs to become smarter.
             ThreadGuard guard(shared_from_this().get());
-            if (ownershipModel == Ownership::kStatic)
-                guard.markStaticOwnership();
+            guard.markStaticOwnership();
 
             // If this is the first run of the SSM, then update its state to Source
             if (state() == State::Created) {
@@ -583,7 +575,7 @@ void ServiceStateMachine::_runOnce() {
                 return;
             }
 
-            _serviceExecutor->schedule(GuaranteedExecutor::enforceRunOnce(
+            _executor()->schedule(GuaranteedExecutor::enforceRunOnce(
                 [this, anchor = shared_from_this()](Status status) { _runOnce(); }));
         });
 }
@@ -674,4 +666,5 @@ void ServiceStateMachine::_cleanupSession(ThreadGuard guard) {
     Client::releaseCurrent();
 }
 
+}  // namespace transport
 }  // namespace mongo
diff --git a/src/mongo/transport/service_state_machine.h b/src/mongo/transport/service_state_machine.h
index b04a419a362..7794cdfbfce 100644
--- a/src/mongo/transport/service_state_machine.h
+++ b/src/mongo/transport/service_state_machine.h
@@ -35,6 +35,7 @@
 
 #include "mongo/base/status.h"
 #include "mongo/config.h"
+#include "mongo/db/client.h"
 #include "mongo/db/service_context.h"
 #include "mongo/platform/atomic_word.h"
 #include "mongo/platform/mutex.h"
@@ -48,6 +49,7 @@
 #include "mongo/util/net/ssl_manager.h"
 
 namespace mongo {
+namespace transport {
 
 /*
  * The ServiceStateMachine holds the state of a single client connection and represents the
@@ -64,17 +66,9 @@ public:
     ServiceStateMachine& operator=(ServiceStateMachine&&) = delete;
 
     /*
-     * Creates a new ServiceStateMachine for a given session/service context. If sync is true,
-     * then calls into the transport layer will block while they complete, otherwise they will
-     * be handled asynchronously.
+     * Construct a ServiceStateMachine for a given Client.
      */
-    static std::shared_ptr<ServiceStateMachine> create(ServiceContext* svcContext,
-                                                       transport::SessionHandle session,
-                                                       transport::Mode transportMode);
-
-    ServiceStateMachine(ServiceContext* svcContext,
-                        transport::SessionHandle session,
-                        transport::Mode transportMode);
+    ServiceStateMachine(ServiceContext::UniqueClient client);
 
     /*
      * Any state may transition to EndSession in case of an error, otherwise the valid state
@@ -109,13 +103,7 @@ public:
     /*
      * start() schedules a call to _runOnce() in the future.
      */
-    void start(Ownership ownershipModel);
-
-    /*
-     * Set the executor to be used for the next call to runNext(). This allows switching between
-     * thread models after the SSM has started.
-     */
-    void setServiceExecutor(transport::ServiceExecutor* executor);
+    void start();
 
     /*
      * Gets the current state of connection for testing/diagnostic purposes.
@@ -174,6 +162,11 @@ private:
     const transport::SessionHandle& _session() const;
 
     /*
+     * Gets the transport::ServiceExecutor associated with this connection.
+     */
+    ServiceExecutor* _executor();
+
+    /*
      * This function actually calls into the database and processes a request. It's broken out
      * into its own inline function for better readability.
      */
@@ -210,16 +203,12 @@ private:
 
     AtomicWord<State> _state{State::Created};
 
-    ServiceEntryPoint* _sep;
-    transport::Mode _transportMode;
-
     ServiceContext* const _serviceContext;
+    ServiceEntryPoint* const _sep;
 
     transport::SessionHandle _sessionHandle;
-    const std::string _threadName;
     ServiceContext::UniqueClient _dbClient;
-    const Client* _dbClientPtr;
-    transport::ServiceExecutor* _serviceExecutor;
+    Client* _dbClientPtr;
     std::function<void()> _cleanupHook;
 
     bool _inExhaust = false;
@@ -235,7 +224,6 @@ private:
 #if MONGO_CONFIG_DEBUG_BUILD
     AtomicWord<stdx::thread::id> _owningThread;
 #endif
-    std::string _oldThreadName;
 };
 
 template <typename T>
@@ -266,4 +254,5 @@ T& operator<<(T& stream, const ServiceStateMachine::State& state) {
     return stream;
 }
 
+}  // namespace transport
 }  // namespace mongo