SERVER-24162 Integrate TransportLayer

Expand the transport layer as needed to replace uses of abstract message port for ingress
networking.

84 files changed, 2186 insertions, 819 deletions

diff --git a/src/mongo/SConscript b/src/mongo/SConscript
index a13e463dd2b..a1a1a609e46 100644
--- a/src/mongo/SConscript
+++ b/src/mongo/SConscript
@@ -254,6 +254,8 @@ mongodLibDeps = [
     "db/repl/storage_interface_impl",
     "executor/network_interface_factory",
     's/commands/shared_cluster_commands',
+    "transport/transport_layer_legacy",
+    "transport/service_entry_point_utils",
     "util/clock_sources",
     "util/ntservice",
 ]
@@ -266,6 +268,7 @@ mongod = env.Program(
     source=[
         "db/db.cpp",
         "db/mongod_options_init.cpp",
+        "db/service_entry_point_mongod.cpp",
     ],
     LIBDEPS=mongodLibDeps,
 )
@@ -297,6 +300,7 @@ env.Install(
             's/client/sharding_connection_hook_for_mongos.cpp',
             's/mongos_options_init.cpp',
             's/server.cpp',
+            's/service_entry_point_mongos.cpp',
         ],
         LIBDEPS=[
             'db/conn_pool_options',
@@ -309,6 +313,8 @@ env.Install(
             's/coreshard',
             's/mongoscore',
             's/sharding_initialization',
+            'transport/service_entry_point_utils',
+            'transport/transport_layer_legacy',
             'util/clock_sources',
             'util/ntservice',
             'util/options_parser/options_parser_init',
diff --git a/src/mongo/base/error_codes.err b/src/mongo/base/error_codes.err
index fa413cfd6dd..ab0fbc2c2f7 100644
--- a/src/mongo/base/error_codes.err
+++ b/src/mongo/base/error_codes.err
@@ -170,6 +170,7 @@ error_code("OperatorNotSupportedOnView", 168)
 error_code("CommandOnShardedViewNotSupportedOnMongos", 169)
 error_code("TooManyMatchingDocuments", 170)
 error_code("CannotIndexParallelArrays", 171)
+error_code("TransportSessionNotFound", 172)
 
 # Non-sequential error codes (for compatibility only)
 error_code("SocketException", 9001)
diff --git a/src/mongo/client/SConscript b/src/mongo/client/SConscript
index 78601d0b1b6..3a0d63c4320 100644
--- a/src/mongo/client/SConscript
+++ b/src/mongo/client/SConscript
@@ -256,7 +256,7 @@ env.CppUnitTest(
         'clientdriver',
         '$BUILD_DIR/mongo/db/auth/authorization_manager_mock_init',
         '$BUILD_DIR/mongo/db/service_context_noop_init',
-        '$BUILD_DIR/mongo/util/net/message_server_port',
+        '$BUILD_DIR/mongo/transport/transport_layer_legacy',
         '$BUILD_DIR/mongo/util/net/network',
     ],
 )
diff --git a/src/mongo/client/scoped_db_conn_test.cpp b/src/mongo/client/scoped_db_conn_test.cpp
index 9ee72a63953..ba54e4ec733 100644
--- a/src/mongo/client/scoped_db_conn_test.cpp
+++ b/src/mongo/client/scoped_db_conn_test.cpp
@@ -41,11 +41,14 @@
 #include "mongo/rpc/request_interface.h"
 #include "mongo/stdx/mutex.h"
 #include "mongo/stdx/thread.h"
+#include "mongo/transport/service_entry_point.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/transport/transport_layer_legacy.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/fail_point_service.h"
 #include "mongo/util/log.h"
 #include "mongo/util/net/listen.h"
-#include "mongo/util/net/message_server.h"
 #include "mongo/util/net/socket_exception.h"
 #include "mongo/util/quick_exit.h"
 #include "mongo/util/time_support.h"
@@ -72,12 +75,30 @@ namespace {
 const string TARGET_HOST = "localhost:27017";
 const int TARGET_PORT = 27017;
 
-class DummyMessageHandler final : public MessageHandler {
+class DummyServiceEntryPoint : public ServiceEntryPoint {
+    MONGO_DISALLOW_COPYING(DummyServiceEntryPoint);
+
 public:
-    virtual void connected(AbstractMessagingPort* p) {}
+    DummyServiceEntryPoint() {}
+
+    virtual ~DummyServiceEntryPoint() {
+        for (auto& t : _threads) {
+            t.join();
+        }
+    }
+
+    void startSession(transport::Session&& session) override {
+        _threads.emplace_back(&DummyServiceEntryPoint::run, this, std::move(session));
+    }
+
+private:
+    void run(transport::Session&& session) {
+        Message inMessage;
+        if (!session.sourceMessage(&inMessage).wait().isOK()) {
+            return;
+        }
 
-    virtual void process(Message& m, AbstractMessagingPort* port) {
-        auto request = rpc::makeRequest(&m);
+        auto request = rpc::makeRequest(&inMessage);
         auto reply = rpc::makeReplyBuilder(request->getProtocol());
 
         BSONObjBuilder commandResponse;
@@ -92,10 +113,14 @@ public:
                             .setMetadata(rpc::makeEmptyMetadata())
                             .done();
 
-        port->reply(m, response);
+        response.header().setResponseToMsgId(inMessage.header().getId());
+
+        if (!session.sinkMessage(response).wait().isOK()) {
+            return;
+        }
     }
 
-    virtual void close() {}
+    std::vector<stdx::thread> _threads;
 };
 
 // TODO: Take this out and make it as a reusable class in a header file. The only
@@ -130,15 +155,15 @@ public:
      * @param messageHandler the message handler to use for this server. Ownership
      *     of this object is passed to this server.
      */
-    void run(std::shared_ptr<MessageHandler> messsageHandler) {
-        if (_server != NULL) {
+    void run(std::shared_ptr<ServiceEntryPoint> serviceEntryPoint) {
+        if (_server) {
             return;
         }
 
-        MessageServer::Options options;
+        transport::TransportLayerLegacy::Options options;
         options.port = _port;
 
-        _server.reset(createServer(options, std::move(messsageHandler), getGlobalServiceContext()));
+        _server = stdx::make_unique<transport::TransportLayerLegacy>(options, serviceEntryPoint);
         _serverThread = stdx::thread(runServer, _server.get());
     }
 
@@ -164,23 +189,23 @@ public:
             sleepmillis(500);
             connCount = Listener::globalTicketHolder.used();
         }
-
+        _server->shutdown();
         _server.reset();
     }
 
     /**
      * Helper method for running the server on a separate thread.
      */
-    static void runServer(MessageServer* server) {
-        server->setupSockets();
-        server->run();
+    static void runServer(transport::TransportLayerLegacy* server) {
+        server->setup();
+        server->start();
     }
 
 private:
     const int _port;
 
     stdx::thread _serverThread;
-    unique_ptr<MessageServer> _server;
+    unique_ptr<transport::TransportLayerLegacy> _server;
 };
 
 /**
@@ -190,9 +215,9 @@ class DummyServerFixture : public unittest::Test {
 public:
     void setUp() {
         _maxPoolSizePerHost = globalConnPool.getMaxPoolSize();
-        _dummyServer = new DummyServer(TARGET_PORT);
+        _dummyServer = stdx::make_unique<DummyServer>(TARGET_PORT);
 
-        auto dummyHandler = std::make_shared<DummyMessageHandler>();
+        auto dummyHandler = std::make_shared<DummyServiceEntryPoint>();
         _dummyServer->run(std::move(dummyHandler));
         DBClientConnection conn;
         Timer timer;
@@ -212,7 +237,7 @@ public:
 
     void tearDown() {
         ScopedDbConnection::clearPool();
-        delete _dummyServer;
+        _dummyServer.reset();
 
         globalConnPool.setMaxPoolSize(_maxPoolSizePerHost);
     }
@@ -272,12 +297,12 @@ protected:
     }
 
 private:
-    static void runServer(MessageServer* server) {
-        server->setupSockets();
-        server->run();
+    static void runServer(transport::TransportLayerLegacy* server) {
+        server->setup();
+        server->start();
     }
 
-    DummyServer* _dummyServer;
+    std::unique_ptr<DummyServer> _dummyServer;
     uint32_t _maxPoolSizePerHost;
 };
 
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 92f97280964..ba8c1e855bd 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -100,6 +100,7 @@ env.Library(
         "dbmessage.cpp",
     ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/transport/transport_layer_common',
         '$BUILD_DIR/mongo/util/net/network',
     ]
 )
@@ -402,7 +403,6 @@ env.Library(
         "$BUILD_DIR/mongo/rpc/command_reply",
         "$BUILD_DIR/mongo/rpc/command_request",
         "$BUILD_DIR/mongo/rpc/metadata",
-        "$BUILD_DIR/mongo/util/net/message_server_port",
         "$BUILD_DIR/mongo/util/net/miniwebserver",
         "$BUILD_DIR/mongo/util/processinfo",
         "$BUILD_DIR/mongo/util/signal_handlers",
@@ -473,6 +473,8 @@ env.Library(
         '$BUILD_DIR/mongo/util/decorable',
         '$BUILD_DIR/mongo/util/fail_point',
         '$BUILD_DIR/mongo/util/net/hostandport',
+        '$BUILD_DIR/mongo/transport/transport_layer_common',
+        '$BUILD_DIR/mongo/transport/transport_layer_manager',
     ],
 )
 
diff --git a/src/mongo/db/auth/sasl_commands.cpp b/src/mongo/db/auth/sasl_commands.cpp
index f006cded7d0..0fefd3b6cc9 100644
--- a/src/mongo/db/auth/sasl_commands.cpp
+++ b/src/mongo/db/auth/sasl_commands.cpp
@@ -187,10 +187,10 @@ Status doSaslStep(const ClientBasic* client,
     status = session->step(payload, &responsePayload);
 
     if (!status.isOK()) {
-        const SockAddr clientAddr = client->port()->remoteAddr();
         log() << session->getMechanism() << " authentication failed for "
               << session->getPrincipalId() << " on " << session->getAuthenticationDatabase()
-              << " from client " << clientAddr.getAddr() << " ; " << status.toString() << std::endl;
+              << " from client " << client->getRemote().toString() << " ; " << status.toString()
+              << std::endl;
 
         sleepmillis(saslGlobalParams.authFailedDelay);
         // All the client needs to know is that authentication has failed.
diff --git a/src/mongo/db/client.cpp b/src/mongo/db/client.cpp
index 807a7650aa5..2a665268811 100644
--- a/src/mongo/db/client.cpp
+++ b/src/mongo/db/client.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/lasterror.h"
 #include "mongo/db/service_context.h"
 #include "mongo/stdx/thread.h"
+#include "mongo/transport/session.h"
 #include "mongo/util/concurrency/thread_name.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/mongoutils/str.h"
@@ -63,16 +64,16 @@ void Client::initThreadIfNotAlready() {
     initThreadIfNotAlready(getThreadName().c_str());
 }
 
-void Client::initThread(const char* desc, AbstractMessagingPort* mp) {
-    initThread(desc, getGlobalServiceContext(), mp);
+void Client::initThread(const char* desc, transport::Session* session) {
+    initThread(desc, getGlobalServiceContext(), session);
 }
 
-void Client::initThread(const char* desc, ServiceContext* service, AbstractMessagingPort* mp) {
+void Client::initThread(const char* desc, ServiceContext* service, transport::Session* session) {
     invariant(currentClient.getMake()->get() == nullptr);
 
     std::string fullDesc;
-    if (mp != NULL) {
-        fullDesc = str::stream() << desc << mp->connectionId();
+    if (session) {
+        fullDesc = str::stream() << desc << session->id();
     } else {
         fullDesc = desc;
     }
@@ -80,7 +81,7 @@ void Client::initThread(const char* desc, ServiceContext* service, AbstractMessa
     setThreadName(fullDesc.c_str());
 
     // Create the client obj, attach to thread
-    *currentClient.get() = service->makeClient(fullDesc, mp);
+    *currentClient.get() = service->makeClient(fullDesc, session);
 }
 
 void Client::destroy() {
@@ -98,11 +99,11 @@ int64_t generateSeed(const std::string& desc) {
 }
 }  // namespace
 
-Client::Client(std::string desc, ServiceContext* serviceContext, AbstractMessagingPort* p)
-    : ClientBasic(serviceContext, p),
+Client::Client(std::string desc, ServiceContext* serviceContext, transport::Session* session)
+    : ClientBasic(serviceContext, session),
       _desc(std::move(desc)),
       _threadId(stdx::this_thread::get_id()),
-      _connectionId(p ? p->connectionId() : 0),
+      _connectionId(session ? session->id() : 0),
       _prng(generateSeed(_desc)) {}
 
 void Client::reportState(BSONObjBuilder& builder) {
diff --git a/src/mongo/db/client.h b/src/mongo/db/client.h
index 407a0bac868..a7523981e47 100644
--- a/src/mongo/db/client.h
+++ b/src/mongo/db/client.h
@@ -51,6 +51,10 @@ class AbstractMessagingPort;
 class Collection;
 class OperationContext;
 
+namespace transport {
+class Session;
+}  // namespace transport
+
 typedef long long ConnectionId;
 
 /** the database's concept of an outside "client" */
@@ -59,16 +63,16 @@ public:
     /**
      * Creates a Client object and stores it in TLS for the current thread.
      *
-     * An unowned pointer to a AbstractMessagingPort may optionally be provided. If 'mp' is
-     * non-null, then it will be used to augment the thread name, and for reporting purposes.
+     * An unowned pointer to a transport::Session may optionally be provided. If 'session'
+     * is non-null, then it will be used to augment the thread name, and for reporting purposes.
      *
-     * If provided, 'mp' must outlive the newly-created Client object. Client::destroy() may be used
-     * to help enforce that the Client does not outlive 'mp'.
+     * If provided, 'session' must outlive the newly-created Client object. Client::destroy() may
+     * be used to help enforce that the Client does not outlive 'session.'
      */
-    static void initThread(const char* desc, AbstractMessagingPort* mp = 0);
+    static void initThread(const char* desc, transport::Session* session = nullptr);
     static void initThread(const char* desc,
                            ServiceContext* serviceContext,
-                           AbstractMessagingPort* mp);
+                           transport::Session* session);
 
     /**
      * Inits a thread if that thread has not already been init'd, setting the thread name to
@@ -159,7 +163,7 @@ public:
 
 private:
     friend class ServiceContext;
-    Client(std::string desc, ServiceContext* serviceContext, AbstractMessagingPort* p = 0);
+    Client(std::string desc, ServiceContext* serviceContext, transport::Session* session = nullptr);
 
 
     // Description for the client (e.g. conn8)
diff --git a/src/mongo/db/client_basic.cpp b/src/mongo/db/client_basic.cpp
index 06e619597c3..578ebdf668d 100644
--- a/src/mongo/db/client_basic.cpp
+++ b/src/mongo/db/client_basic.cpp
@@ -32,8 +32,8 @@
 
 namespace mongo {
 
-ClientBasic::ClientBasic(ServiceContext* serviceContext, AbstractMessagingPort* messagingPort)
-    : _serviceContext(serviceContext), _messagingPort(messagingPort) {}
+ClientBasic::ClientBasic(ServiceContext* serviceContext, transport::Session* session)
+    : _serviceContext(serviceContext), _session(session) {}
 
 ClientBasic::~ClientBasic() = default;
 
diff --git a/src/mongo/db/client_basic.h b/src/mongo/db/client_basic.h
index 88d7ef4afe8..09f31b5da42 100644
--- a/src/mongo/db/client_basic.h
+++ b/src/mongo/db/client_basic.h
@@ -31,6 +31,7 @@
 #include <memory>
 
 #include "mongo/base/disallow_copying.h"
+#include "mongo/transport/session.h"
 #include "mongo/util/decorable.h"
 #include "mongo/util/net/abstract_message_port.h"
 #include "mongo/util/net/hostandport.h"
@@ -58,11 +59,12 @@ public:
     }
 
     bool hasRemote() const {
-        return _messagingPort;
+        return _session;
     }
+
     HostAndPort getRemote() const {
-        verify(_messagingPort);
-        return _messagingPort->remote();
+        verify(_session);
+        return _session->remote();
     }
 
     /**
@@ -73,20 +75,20 @@ public:
     }
 
     /**
-     * Returns the AbstractMessagePort to which this client session is bound, if any.
+     * Returns the Session to which this client is bound, if any.
      */
-    AbstractMessagingPort* port() const {
-        return _messagingPort;
+    transport::Session* session() const {
+        return _session;
     }
 
     static ClientBasic* getCurrent();
 
 protected:
-    ClientBasic(ServiceContext* serviceContext, AbstractMessagingPort* messagingPort);
+    ClientBasic(ServiceContext* serviceContext, transport::Session* session);
     ~ClientBasic();
 
 private:
     ServiceContext* const _serviceContext;
-    AbstractMessagingPort* const _messagingPort;
+    transport::Session* const _session;
 };
 }
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 919510c4b79..87c472dc519 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -71,6 +71,7 @@ env.Library(
         '$BUILD_DIR/mongo/s/coreshard',
         '$BUILD_DIR/mongo/s/write_ops/batch_write_types',
         '$BUILD_DIR/mongo/scripting/scripting_common',
+        '$BUILD_DIR/mongo/transport/transport_layer_common',
         '$BUILD_DIR/mongo/util/cmdline_utils/cmdline_utils',
         '$BUILD_DIR/mongo/util/foundation',
         '$BUILD_DIR/mongo/util/ntservice',
diff --git a/src/mongo/db/commands/authentication_commands.cpp b/src/mongo/db/commands/authentication_commands.cpp
index 6fc6268a807..2f05971ef56 100644
--- a/src/mongo/db/commands/authentication_commands.cpp
+++ b/src/mongo/db/commands/authentication_commands.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/server_options.h"
 #include "mongo/platform/random.h"
 #include "mongo/stdx/memory.h"
+#include "mongo/transport/session.h"
 #include "mongo/util/concurrency/mutex.h"
 #include "mongo/util/log.h"
 #include "mongo/util/md5.hpp"
@@ -314,17 +315,17 @@ Status CmdAuthenticate::_authenticateX509(OperationContext* txn,
 
     ClientBasic* client = ClientBasic::getCurrent();
     AuthorizationSession* authorizationSession = AuthorizationSession::get(client);
-    std::string clientSubjectName = client->port()->getX509SubjectName();
+    auto clientName = client->session()->getX509SubjectName();
 
     if (!getSSLManager()->getSSLConfiguration().hasCA) {
         return Status(ErrorCodes::AuthenticationFailed,
                       "Unable to verify x.509 certificate, as no CA has been provided.");
-    } else if (user.getUser() != clientSubjectName) {
+    } else if (user.getUser() != clientName) {
         return Status(ErrorCodes::AuthenticationFailed,
                       "There is no x.509 client certificate matching the user.");
     } else {
         // Handle internal cluster member auth, only applies to server-server connections
-        if (getSSLManager()->getSSLConfiguration().isClusterMember(clientSubjectName)) {
+        if (getSSLManager()->getSSLConfiguration().isClusterMember(clientName)) {
             int clusterAuthMode = serverGlobalParams.clusterAuthMode.load();
             if (clusterAuthMode == ServerGlobalParams::ClusterAuthMode_undefined ||
                 clusterAuthMode == ServerGlobalParams::ClusterAuthMode_keyFile) {
diff --git a/src/mongo/db/commands/server_status.cpp b/src/mongo/db/commands/server_status.cpp
index bbe7e9930a7..e567a15b18b 100644
--- a/src/mongo/db/commands/server_status.cpp
+++ b/src/mongo/db/commands/server_status.cpp
@@ -47,9 +47,9 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/platform/process_id.h"
+#include "mongo/transport/transport_layer.h"
 #include "mongo/util/log.h"
 #include "mongo/util/net/hostname_canonicalization_worker.h"
-#include "mongo/util/net/listen.h"
 #include "mongo/util/net/ssl_manager.h"
 #include "mongo/util/processinfo.h"
 #include "mongo/util/ramlog.h"
@@ -228,9 +228,10 @@ public:
 
     BSONObj generateSection(OperationContext* txn, const BSONElement& configElement) const {
         BSONObjBuilder bb;
-        bb.append("current", Listener::globalTicketHolder.used());
-        bb.append("available", Listener::globalTicketHolder.available());
-        bb.append("totalCreated", Listener::globalConnectionNumber.load());
+        auto stats = txn->getServiceContext()->getTransportLayer()->sessionStats();
+        bb.append("current", static_cast<int>(stats.numOpenSessions));
+        bb.append("available", static_cast<int>(stats.numAvailableSessions));
+        bb.append("totalCreated", static_cast<int>(stats.numCreatedSessions));
         return bb.obj();
     }
 
diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index a10b0be5a39..56dff7cecaf 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -96,6 +96,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/service_context_d.h"
 #include "mongo/db/service_context_d.h"
+#include "mongo/db/service_entry_point_mongod.h"
 #include "mongo/db/startup_warnings_mongod.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/db/stats/snapshots.h"
@@ -114,6 +115,7 @@
 #include "mongo/stdx/future.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/stdx/thread.h"
+#include "mongo/transport/transport_layer_legacy.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/cmdline_utils/censor_cmdline.h"
 #include "mongo/util/concurrency/task.h"
@@ -124,7 +126,6 @@
 #include "mongo/util/log.h"
 #include "mongo/util/net/hostname_canonicalization_worker.h"
 #include "mongo/util/net/listen.h"
-#include "mongo/util/net/message_server.h"
 #include "mongo/util/net/ssl_manager.h"
 #include "mongo/util/ntservice.h"
 #include "mongo/util/options_parser/startup_options.h"
@@ -169,65 +170,6 @@ ntservice::NtServiceDefaultStrings defaultServiceStrings = {
 
 Timer startupSrandTimer;
 
-class MyMessageHandler : public MessageHandler {
-public:
-    virtual void connected(AbstractMessagingPort* p) {
-        Client::initThread("conn", p);
-    }
-
-    virtual void process(Message& m, AbstractMessagingPort* port) {
-        while (true) {
-            if (inShutdown()) {
-                log() << "got request after shutdown()" << endl;
-                break;
-            }
-
-            DbResponse dbresponse;
-            {
-                auto opCtx = getGlobalServiceContext()->makeOperationContext(&cc());
-                assembleResponse(opCtx.get(), m, dbresponse, port->remote());
-
-                // opCtx must go out of scope here so that the operation cannot show up in currentOp
-                // results after the response reaches the client
-            }
-
-            if (!dbresponse.response.empty()) {
-                port->reply(m, dbresponse.response, dbresponse.responseToMsgId);
-                if (dbresponse.exhaustNS.size() > 0) {
-                    MsgData::View header = dbresponse.response.header();
-                    QueryResult::View qr = header.view2ptr();
-                    long long cursorid = qr.getCursorId();
-                    if (cursorid) {
-                        verify(dbresponse.exhaustNS.size() && dbresponse.exhaustNS[0]);
-                        string ns = dbresponse.exhaustNS;  // before reset() free's it...
-                        m.reset();
-                        BufBuilder b(512);
-                        b.appendNum((int)0 /*size set later*/);
-                        b.appendNum(header.getId());
-                        b.appendNum(header.getResponseToMsgId());
-                        b.appendNum((int)dbGetMore);
-                        b.appendNum((int)0);
-                        b.appendStr(ns);
-                        b.appendNum((int)0);  // ntoreturn
-                        b.appendNum(cursorid);
-
-                        MsgData::View header = b.buf();
-                        header.setLen(b.len());
-                        m.setData(b.release());
-                        DEV log() << "exhaust=true sending more";
-                        continue;  // this goes back to top loop
-                    }
-                }
-            }
-            break;
-        }
-    }
-
-    virtual void close() {
-        Client::destroy();
-    }
-};
-
 static void logStartup(OperationContext* txn) {
     BSONObjBuilder toLog;
     stringstream id;
@@ -530,7 +472,7 @@ static void _initWireSpec() {
 }
 
 
-static void _initAndListen(int listenPort) {
+static ExitCode _initAndListen(int listenPort) {
     Client::initThread("initandlisten");
 
     _initWireSpec();
@@ -570,20 +512,18 @@ static void _initAndListen(int listenPort) {
 
     checked_cast<ServiceContextMongoD*>(getGlobalServiceContext())->createLockFile();
 
-    // Due to SERVER-15389, we must setupSockets first thing at startup in order to avoid
-    // obtaining too high a file descriptor for our calls to select().
-    MessageServer::Options options;
+    transport::TransportLayerLegacy::Options options;
     options.port = listenPort;
     options.ipList = serverGlobalParams.bind_ip;
 
-    auto handler = std::make_shared<MyMessageHandler>();
-    MessageServer* server = createServer(options, std::move(handler), getGlobalServiceContext());
-
-    // This is what actually creates the sockets, but does not yet listen on them because we
-    // do not want connections to just hang if recovery takes a very long time.
-    if (!server->setupSockets()) {
-        error() << "Failed to set up sockets during startup.";
-        return;
+    // Create, start, and attach the TL
+    auto transportLayer = stdx::make_unique<transport::TransportLayerLegacy>(
+        options,
+        std::make_shared<ServiceEntryPointMongod>(getGlobalServiceContext()->getTransportLayer()));
+    auto res = transportLayer->setup();
+    if (!res.isOK()) {
+        error() << "Failed to set up listener: " << res.toString();
+        return EXIT_NET_ERROR;
     }
 
     std::shared_ptr<DbWebServer> dbWebServer;
@@ -594,7 +534,7 @@ static void _initAndListen(int listenPort) {
                                           new RestAdminAccess()));
         if (!dbWebServer->setupSockets()) {
             error() << "Failed to set up sockets for HTTP interface during startup.";
-            return;
+            return EXIT_NET_ERROR;
         }
     }
 
@@ -674,7 +614,7 @@ static void _initAndListen(int listenPort) {
     }
 
     if (mmapv1GlobalOptions.journalOptions & MMAPV1Options::JournalRecoverOnly)
-        return;
+        return EXIT_NET_ERROR;
 
     if (mongodGlobalParams.scriptingEnabled) {
         ScriptEngine::setup();
@@ -799,14 +739,19 @@ static void _initAndListen(int listenPort) {
     // MessageServer::run will return when exit code closes its socket and we don't need the
     // operation context anymore
     startupOpCtx.reset();
-    server->run();
+
+    auto start = getGlobalServiceContext()->addAndStartTransportLayer(std::move(transportLayer));
+    if (!start.isOK()) {
+        error() << "Failed to start the listener: " << start.toString();
+        return EXIT_NET_ERROR;
+    }
+
+    return waitForShutdown();
 }
 
 ExitCode initAndListen(int listenPort) {
     try {
-        _initAndListen(listenPort);
-
-        return inShutdown() ? EXIT_CLEAN : EXIT_NET_ERROR;
+        return _initAndListen(listenPort);
     } catch (DBException& e) {
         log() << "exception in initAndListen: " << e.toString() << ", terminating" << endl;
         return EXIT_UNCAUGHT;
@@ -1001,7 +946,6 @@ static void reportEventToSystemImpl(const char* msg) {
 // registerShutdownTask is called below. It must not depend on the
 // prior execution of mongo initializers or the existence of threads.
 static void shutdownTask() {
-
     auto serviceContext = getGlobalServiceContext();
 
     Client::initThreadIfNotAlready();
@@ -1014,6 +958,7 @@ static void shutdownTask() {
         txn = uniqueTxn.get();
     }
 
+    getGlobalServiceContext()->getTransportLayer()->shutdown();
     log(LogComponent::kNetwork) << "shutdown: going to close listening sockets..." << endl;
     ListeningSockets::get()->closeAll();
 
@@ -1037,12 +982,6 @@ static void shutdownTask() {
     // will do it for us when the process terminates.
 
     stdx::packaged_task<void()> dryOutTask([] {
-
-        // Walk all open sockets and close them. This should unblock any that are sitting in
-        // recv. Other sockets on which there are active operations should see the inShutdown flag
-        // as true when they complete the operation.
-        Listener::closeMessagingPorts(0);
-
         // There isn't currently a way to wait on the TicketHolder to have all its tickets back,
         // unfortunately. So, busy wait in this detached thread.
         while (true) {
diff --git a/src/mongo/db/dbmessage.cpp b/src/mongo/db/dbmessage.cpp
index 67a5548dbfb..f16ec2b6276 100644
--- a/src/mongo/db/dbmessage.cpp
+++ b/src/mongo/db/dbmessage.cpp
@@ -30,7 +30,9 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/dbmessage.h"
+#include "mongo/db/operation_context.h"
 #include "mongo/platform/strnlen.h"
+#include "mongo/transport/session.h"
 
 namespace mongo {
 
@@ -173,20 +175,23 @@ OpQueryReplyBuilder::OpQueryReplyBuilder() : _buffer(32768) {
     _buffer.skip(sizeof(QueryResult::Value));
 }
 
-void OpQueryReplyBuilder::send(AbstractMessagingPort* destination,
+void OpQueryReplyBuilder::send(transport::Session* session,
                                int queryResultFlags,
-                               Message& requestMsg,
+                               const Message& requestMsg,
                                int nReturned,
                                int startingFrom,
                                long long cursorId) {
     Message response;
     putInMessage(&response, queryResultFlags, nReturned, startingFrom, cursorId);
-    destination->reply(requestMsg, response, requestMsg.header().getId());
+
+    response.header().setId(nextMessageId());
+    response.header().setResponseToMsgId(requestMsg.header().getId());
+
+    uassertStatusOK(session->sinkMessage(response).wait());
 }
 
-void OpQueryReplyBuilder::sendCommandReply(AbstractMessagingPort* destination,
-                                           Message& requestMsg) {
-    send(destination, /*queryFlags*/ 0, requestMsg, /*nReturned*/ 1);
+void OpQueryReplyBuilder::sendCommandReply(transport::Session* session, const Message& requestMsg) {
+    send(session, /*queryFlags*/ 0, requestMsg, /*nReturned*/ 1);
 }
 
 void OpQueryReplyBuilder::putInMessage(
@@ -202,7 +207,7 @@ void OpQueryReplyBuilder::putInMessage(
 }
 
 void replyToQuery(int queryResultFlags,
-                  AbstractMessagingPort* p,
+                  transport::Session* session,
                   Message& requestMsg,
                   const void* data,
                   int size,
@@ -211,15 +216,19 @@ void replyToQuery(int queryResultFlags,
                   long long cursorId) {
     OpQueryReplyBuilder reply;
     reply.bufBuilderForResults().appendBuf(data, size);
-    reply.send(p, queryResultFlags, requestMsg, nReturned, startingFrom, cursorId);
+    reply.send(session, queryResultFlags, requestMsg, nReturned, startingFrom, cursorId);
 }
 
 void replyToQuery(int queryResultFlags,
-                  AbstractMessagingPort* p,
+                  transport::Session* session,
                   Message& requestMsg,
                   const BSONObj& responseObj) {
-    replyToQuery(
-        queryResultFlags, p, requestMsg, (void*)responseObj.objdata(), responseObj.objsize(), 1);
+    replyToQuery(queryResultFlags,
+                 session,
+                 requestMsg,
+                 (void*)responseObj.objdata(),
+                 responseObj.objsize(),
+                 1);
 }
 
 void replyToQuery(int queryResultFlags, Message& m, DbResponse& dbresponse, BSONObj obj) {
diff --git a/src/mongo/db/dbmessage.h b/src/mongo/db/dbmessage.h
index 7f1a3a2d0f2..7fabf117421 100644
--- a/src/mongo/db/dbmessage.h
+++ b/src/mongo/db/dbmessage.h
@@ -39,6 +39,12 @@
 
 namespace mongo {
 
+class OperationContext;
+
+namespace transport {
+class Session;
+}  // namespace transport
+
 /* db response format
 
    Query or GetMore: // see struct QueryResult
@@ -352,9 +358,9 @@ public:
     /**
      * Finishes the reply and sends the message out to 'destination'.
      */
-    void send(AbstractMessagingPort* destination,
+    void send(transport::Session* session,
               int queryResultFlags,
-              Message& requestMsg,  // should be const but MessagePort::reply takes non-const.
+              const Message& requestMsg,
               int nReturned,
               int startingFrom = 0,
               long long cursorId = 0);
@@ -362,14 +368,14 @@ public:
     /**
      * Similar to send() but used for replying to a command.
      */
-    void sendCommandReply(AbstractMessagingPort* destination, Message& requestMsg);
+    void sendCommandReply(transport::Session* session, const Message& requestMsg);
 
 private:
     BufBuilder _buffer;
 };
 
 void replyToQuery(int queryResultFlags,
-                  AbstractMessagingPort* p,
+                  transport::Session* session,
                   Message& requestMsg,
                   const void* data,
                   int size,
@@ -377,10 +383,9 @@ void replyToQuery(int queryResultFlags,
                   int startingFrom = 0,
                   long long cursorId = 0);
 
-
 /* object reply helper. */
 void replyToQuery(int queryResultFlags,
-                  AbstractMessagingPort* p,
+                  transport::Session* session,
                   Message& requestMsg,
                   const BSONObj& responseObj);
 
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index b6a2981dde0..b92ec300cf8 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -398,6 +398,7 @@ env.Library('repl_coordinator_impl',
                      '$BUILD_DIR/mongo/db/service_context',
                      '$BUILD_DIR/mongo/rpc/command_status',
                      '$BUILD_DIR/mongo/rpc/metadata',
+                     '$BUILD_DIR/mongo/transport/transport_layer_common',
                      '$BUILD_DIR/mongo/util/fail_point',
                      'collection_cloner',
                      'data_replicator',
@@ -627,6 +628,7 @@ env.Library('replica_set_messages',
                 '$BUILD_DIR/mongo/client/connection_string',
                 '$BUILD_DIR/mongo/db/common',
                 '$BUILD_DIR/mongo/db/server_options_core',
+                '$BUILD_DIR/mongo/transport/transport_layer_common',
                 '$BUILD_DIR/mongo/util/net/hostandport',
                 'optime',
                 'read_concern_args',
diff --git a/src/mongo/db/repl/repl_set_request_votes.cpp b/src/mongo/db/repl/repl_set_request_votes.cpp
index 4b34c351769..c01628466c0 100644
--- a/src/mongo/db/repl/repl_set_request_votes.cpp
+++ b/src/mongo/db/repl/repl_set_request_votes.cpp
@@ -35,6 +35,8 @@
 #include "mongo/db/repl/repl_set_request_votes_args.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/executor/network_interface.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/transport_layer.h"
 #include "mongo/util/scopeguard.h"
 
 namespace mongo {
@@ -64,16 +66,17 @@ private:
 
         // We want to keep request vote connection open when relinquishing primary.
         // Tag it here.
-        unsigned originalTag = 0;
-        AbstractMessagingPort* mp = txn->getClient()->port();
-        if (mp) {
-            originalTag = mp->getTag();
-            mp->setTag(originalTag | executor::NetworkInterface::kMessagingPortKeepOpen);
+        transport::Session::TagMask originalTag = 0;
+        transport::Session* session = txn->getClient()->session();
+        if (session) {
+            originalTag = session->getTags();
+            session->replaceTags(originalTag | transport::Session::kKeepOpen);
         }
+
         // Untag the connection on exit.
-        ON_BLOCK_EXIT([mp, originalTag]() {
-            if (mp) {
-                mp->setTag(originalTag);
+        ON_BLOCK_EXIT([session, originalTag]() {
+            if (session) {
+                session->replaceTags(originalTag);
             }
         });
 
diff --git a/src/mongo/db/repl/replication_coordinator_external_state.h b/src/mongo/db/repl/replication_coordinator_external_state.h
index 582b8f45970..245b37cca7f 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state.h
+++ b/src/mongo/db/repl/replication_coordinator_external_state.h
@@ -192,8 +192,8 @@ public:
     virtual HostAndPort getClientHostAndPort(const OperationContext* txn) = 0;
 
     /**
-     * Closes all connections except those marked with the keepOpen property, which should
-     * just be connections used for heartbeating.
+     * Closes all connections in the given TransportLayer except those marked with the
+     * keepOpen property, which should just be connections used for heartbeating.
      * This is used during stepdown, and transition out of primary.
      */
     virtual void closeConnections() = 0;
diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
index 81ed9c16bf3..f140580fbcc 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
@@ -76,6 +76,8 @@
 #include "mongo/stdx/functional.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/stdx/thread.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/transport_layer.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/concurrency/thread_pool.h"
 #include "mongo/util/exit.h"
@@ -449,7 +451,7 @@ HostAndPort ReplicationCoordinatorExternalStateImpl::getClientHostAndPort(
 }
 
 void ReplicationCoordinatorExternalStateImpl::closeConnections() {
-    Listener::closeMessagingPorts(executor::NetworkInterface::kMessagingPortKeepOpen);
+    getGlobalServiceContext()->getTransportLayer()->endAllSessions(transport::Session::kKeepOpen);
 }
 
 void ReplicationCoordinatorExternalStateImpl::killAllUserOperations(OperationContext* txn) {
diff --git a/src/mongo/db/repl/replication_info.cpp b/src/mongo/db/repl/replication_info.cpp
index cc844eaf35c..fd36395c017 100644
--- a/src/mongo/db/repl/replication_info.cpp
+++ b/src/mongo/db/repl/replication_info.cpp
@@ -234,9 +234,10 @@ public:
         // Tag connections to avoid closing them on stepdown.
         auto hangUpElement = cmdObj["hangUpOnStepDown"];
         if (!hangUpElement.eoo() && !hangUpElement.trueValue()) {
-            AbstractMessagingPort* mp = txn->getClient()->port();
-            if (mp) {
-                mp->setTag(mp->getTag() | executor::NetworkInterface::kMessagingPortKeepOpen);
+            auto session = txn->getClient()->session();
+            if (session) {
+                session->replaceTags(session->getTags() |
+                                     executor::NetworkInterface::kMessagingPortKeepOpen);
             }
         }
         appendReplicationInfo(txn, result, 0);
diff --git a/src/mongo/db/repl/replset_commands.cpp b/src/mongo/db/repl/replset_commands.cpp
index 1b96dac313f..f591d64c11b 100644
--- a/src/mongo/db/repl/replset_commands.cpp
+++ b/src/mongo/db/repl/replset_commands.cpp
@@ -59,6 +59,8 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/executor/network_interface.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/transport_layer.h"
 #include "mongo/util/fail_point_service.h"
 #include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
@@ -726,17 +728,17 @@ public:
 
         /* we want to keep heartbeat connections open when relinquishing primary.
            tag them here. */
-        AbstractMessagingPort* mp = txn->getClient()->port();
-        unsigned originalTag = 0;
-        if (mp) {
-            originalTag = mp->getTag();
-            mp->setTag(originalTag | executor::NetworkInterface::kMessagingPortKeepOpen);
+        transport::Session::TagMask originalTag = 0;
+        transport::Session* session = txn->getClient()->session();
+        if (session) {
+            originalTag = session->getTags();
+            session->replaceTags(originalTag | transport::Session::kKeepOpen);
         }
 
         // Unset the tag on block exit
-        ON_BLOCK_EXIT([mp, originalTag]() {
-            if (mp) {
-                mp->setTag(originalTag);
+        ON_BLOCK_EXIT([session, originalTag]() {
+            if (session) {
+                session->replaceTags(originalTag);
             }
         });
 
diff --git a/src/mongo/db/s/collection_metadata_test.cpp b/src/mongo/db/s/collection_metadata_test.cpp
index 013c135cc84..359cb3b4fc1 100644
--- a/src/mongo/db/s/collection_metadata_test.cpp
+++ b/src/mongo/db/s/collection_metadata_test.cpp
@@ -50,7 +50,7 @@ class NoChunkFixture : public ShardingCatalogTestFixture {
 protected:
     void setUp() {
         ShardingCatalogTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
         configTargeter()->setFindHostReturnValue(configHost);
 
         OID epoch = OID::gen();
@@ -318,7 +318,7 @@ class SingleChunkFixture : public ShardingCatalogTestFixture {
 protected:
     void setUp() {
         ShardingCatalogTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
         configTargeter()->setFindHostReturnValue(configHost);
 
         OID epoch = OID::gen();
@@ -574,7 +574,7 @@ class SingleChunkMinMaxCompoundKeyFixture : public ShardingCatalogTestFixture {
 protected:
     void setUp() {
         ShardingCatalogTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
         configTargeter()->setFindHostReturnValue(configHost);
 
         OID epoch = OID::gen();
@@ -642,7 +642,7 @@ class TwoChunksWithGapCompoundKeyFixture : public ShardingCatalogTestFixture {
 protected:
     void setUp() {
         ShardingCatalogTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
         configTargeter()->setFindHostReturnValue(configHost);
 
         OID epoch = OID::gen();
@@ -839,7 +839,7 @@ class ThreeChunkWithRangeGapFixture : public ShardingCatalogTestFixture {
 protected:
     void setUp() {
         ShardingCatalogTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
         configTargeter()->setFindHostReturnValue(configHost);
 
         OID epoch = OID::gen();
diff --git a/src/mongo/db/s/metadata_loader_test.cpp b/src/mongo/db/s/metadata_loader_test.cpp
index c96b77a1590..e33e3142d1a 100644
--- a/src/mongo/db/s/metadata_loader_test.cpp
+++ b/src/mongo/db/s/metadata_loader_test.cpp
@@ -51,7 +51,7 @@ public:
 protected:
     void setUp() override {
         ShardingCatalogTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
         configTargeter()->setFindHostReturnValue(configHost);
         _maxCollVersion = ChunkVersion(1, 0, OID::gen());
         _loader.reset(new MetadataLoader);
diff --git a/src/mongo/db/service_context.cpp b/src/mongo/db/service_context.cpp
index 03dc9bcaf8d..648fd546721 100644
--- a/src/mongo/db/service_context.cpp
+++ b/src/mongo/db/service_context.cpp
@@ -34,6 +34,9 @@
 #include "mongo/db/client.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/stdx/memory.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/transport/transport_layer_manager.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/mongoutils/str.h"
 #include "mongo/util/system_clock_source.h"
@@ -115,7 +118,8 @@ Status validateStorageOptions(
 }
 
 ServiceContext::ServiceContext()
-    : _tickSource(stdx::make_unique<SystemTickSource>()),
+    : _transportLayerManager(stdx::make_unique<transport::TransportLayerManager>()),
+      _tickSource(stdx::make_unique<SystemTickSource>()),
       _fastClockSource(stdx::make_unique<SystemClockSource>()),
       _preciseClockSource(stdx::make_unique<SystemClockSource>()) {}
 
@@ -125,8 +129,8 @@ ServiceContext::~ServiceContext() {
 }
 
 ServiceContext::UniqueClient ServiceContext::makeClient(std::string desc,
-                                                        AbstractMessagingPort* p) {
-    std::unique_ptr<Client> client(new Client(std::move(desc), this, p));
+                                                        transport::Session* session) {
+    std::unique_ptr<Client> client(new Client(std::move(desc), this, session));
     auto observer = _clientObservers.cbegin();
     try {
         for (; observer != _clientObservers.cend(); ++observer) {
@@ -150,6 +154,14 @@ ServiceContext::UniqueClient ServiceContext::makeClient(std::string desc,
     return UniqueClient(client.release());
 }
 
+transport::TransportLayer* ServiceContext::getTransportLayer() const {
+    return _transportLayerManager.get();
+}
+
+Status ServiceContext::addAndStartTransportLayer(std::unique_ptr<transport::TransportLayer> tl) {
+    return _transportLayerManager->addAndStartTransportLayer(std::move(tl));
+}
+
 TickSource* ServiceContext::getTickSource() const {
     return _tickSource.get();
 }
diff --git a/src/mongo/db/service_context.h b/src/mongo/db/service_context.h
index fe6a4432d70..174a52ebe52 100644
--- a/src/mongo/db/service_context.h
+++ b/src/mongo/db/service_context.h
@@ -48,6 +48,12 @@ class Client;
 class OperationContext;
 class OpObserver;
 
+namespace transport {
+class Session;
+class TransportLayer;
+class TransportLayerManager;
+}  // namespace transport
+
 /**
  * Classes that implement this interface can receive notification on killOp.
  *
@@ -207,9 +213,9 @@ public:
      *
      * The "desc" string is used to set a descriptive name for the client, used in logging.
      *
-     * If supplied, "p" is the communication channel used for communicating with the client.
+     * If supplied, "session" is the transport::Session used for communicating with the client.
      */
-    UniqueClient makeClient(std::string desc, AbstractMessagingPort* p = nullptr);
+    UniqueClient makeClient(std::string desc, transport::Session* session = nullptr);
 
     /**
      * Creates a new OperationContext on "client".
@@ -299,6 +305,26 @@ public:
     void registerKillOpListener(KillOpListenerInterface* listener);
 
     //
+    // Transport.
+    //
+
+    /**
+     * Get the master TransportLayer. Routes to all other TransportLayers that
+     * may be in use within this service.
+     *
+     * See TransportLayerManager for more details.
+     */
+    transport::TransportLayer* getTransportLayer() const;
+
+    /**
+     * Add a new TransportLayer to this service context. The new TransportLayer will
+     * be added to the TransportLayerManager accessible via getTransportLayer().
+     *
+     * It additionally calls start() on the TransportLayer after adding it.
+     */
+    Status addAndStartTransportLayer(std::unique_ptr<transport::TransportLayer> tl);
+
+    //
     // Global OpObserver.
     //
 
@@ -370,6 +396,11 @@ private:
 
 
     /**
+     * The TransportLayerManager.
+     */
+    std::unique_ptr<transport::TransportLayerManager> _transportLayerManager;
+
+    /**
      * Vector of registered observers.
      */
     std::vector<std::unique_ptr<ClientObserver>> _clientObservers;
diff --git a/src/mongo/db/service_entry_point_mongod.cpp b/src/mongo/db/service_entry_point_mongod.cpp
new file mode 100644
index 00000000000..86f782a0bad
--- /dev/null
+++ b/src/mongo/db/service_entry_point_mongod.cpp
@@ -0,0 +1,153 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kNetwork
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/service_entry_point_mongod.h"
+
+#include <vector>
+
+#include "mongo/db/client.h"
+#include "mongo/db/dbmessage.h"
+#include "mongo/db/instance.h"
+#include "mongo/stdx/thread.h"
+#include "mongo/transport/service_entry_point_utils.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/ticket.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/util/exit.h"
+#include "mongo/util/log.h"
+#include "mongo/util/net/message.h"
+#include "mongo/util/net/socket_exception.h"
+#include "mongo/util/net/thread_idle_callback.h"
+#include "mongo/util/quick_exit.h"
+
+namespace mongo {
+namespace {
+
+// Set up proper headers for formatting an exhaust request, if we need to
+bool setExhaustMessage(Message* m, const DbResponse& dbresponse) {
+    MsgData::View header = dbresponse.response.header();
+    QueryResult::View qr = header.view2ptr();
+    long long cursorid = qr.getCursorId();
+
+    if (!cursorid) {
+        return false;
+    }
+
+    verify(dbresponse.exhaustNS.size() && dbresponse.exhaustNS[0]);
+
+    auto ns = dbresponse.exhaustNS;  // reset() will free this
+
+    m->reset();
+
+    BufBuilder b(512);
+    b.appendNum(static_cast<int>(0) /* size set later in appendData() */);
+    b.appendNum(header.getId());
+    b.appendNum(header.getResponseToMsgId());
+    b.appendNum(static_cast<int>(dbGetMore));
+    b.appendNum(static_cast<int>(0));
+    b.appendStr(ns);
+    b.appendNum(static_cast<int>(0));  // ntoreturn
+    b.appendNum(cursorid);
+
+    MsgData::View(b.buf()).setLen(b.len());
+    m->setData(b.release());
+
+    return true;
+}
+
+}  // namespace
+
+using transport::Session;
+using transport::TransportLayer;
+
+ServiceEntryPointMongod::ServiceEntryPointMongod(TransportLayer* tl) : _tl(tl) {}
+
+void ServiceEntryPointMongod::startSession(Session&& session) {
+    launchWrappedServiceEntryWorkerThread(std::move(session),
+                                          [this](Session* session) { _sessionLoop(session); });
+}
+
+void ServiceEntryPointMongod::_sessionLoop(Session* session) {
+    Message inMessage;
+    bool inExhaust = false;
+    int64_t counter = 0;
+
+    while (true) {
+        // 1. Source a Message from the client (unless we are exhausting)
+        if (!inExhaust) {
+            inMessage.reset();
+            auto status = session->sourceMessage(&inMessage).wait();
+
+            if (ErrorCodes::isInterruption(status.code())) {
+                break;
+            }
+
+            uassertStatusOK(status);
+        }
+
+        // 2. Pass sourced Message up to mongod
+        DbResponse dbresponse;
+        {
+            auto opCtx = cc().makeOperationContext();
+            assembleResponse(opCtx.get(), inMessage, dbresponse, session->remote());
+
+            // opCtx must go out of scope here so that the operation cannot show
+            // up in currentOp results after the response reaches the client
+        }
+
+        // 3. Format our response, if we have one
+        Message& toSink = dbresponse.response;
+        if (!toSink.empty()) {
+            toSink.header().setId(nextMessageId());
+            toSink.header().setResponseToMsgId(inMessage.header().getId());
+
+            // If this is an exhaust cursor, don't source more Messages
+            if (dbresponse.exhaustNS.size() > 0 && setExhaustMessage(&inMessage, dbresponse)) {
+                log() << "we are in exhaust";
+                inExhaust = true;
+            } else {
+                inExhaust = false;
+            }
+
+            // 4. Sink our response to the client
+            uassertStatusOK(session->sinkMessage(toSink).wait());
+        } else {
+            inExhaust = false;
+        }
+
+        if ((counter++ & 0xf) == 0) {
+            markThreadIdle();
+        }
+    }
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/service_entry_point_mongod.h b/src/mongo/db/service_entry_point_mongod.h
new file mode 100644
index 00000000000..31ed05098c8
--- /dev/null
+++ b/src/mongo/db/service_entry_point_mongod.h
@@ -0,0 +1,63 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include <vector>
+
+#include "mongo/base/disallow_copying.h"
+#include "mongo/transport/service_entry_point.h"
+
+namespace mongo {
+
+namespace transport {
+class Session;
+class TransportLayer;
+}  // namespace transport
+
+/**
+ * The entry point from the TransportLayer into Mongod. startSession() spawns and
+ * detaches a new thread for each incoming connection (transport::Session).
+ */
+class ServiceEntryPointMongod final : public ServiceEntryPoint {
+    MONGO_DISALLOW_COPYING(ServiceEntryPointMongod);
+
+public:
+    explicit ServiceEntryPointMongod(transport::TransportLayer* tl);
+
+    virtual ~ServiceEntryPointMongod() = default;
+
+    void startSession(transport::Session&& session) override;
+
+private:
+    void _sessionLoop(transport::Session* session);
+
+    transport::TransportLayer* _tl;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index 1f0411b13cf..8b0b8321aaa 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -98,6 +98,7 @@ env.Library(
         '$BUILD_DIR/mongo/s/catalog/replset/sharding_catalog_client_impl',
         '$BUILD_DIR/mongo/s/catalog/replset/sharding_catalog_manager_impl',
         '$BUILD_DIR/mongo/s/coreshard',
+        '$BUILD_DIR/mongo/transport/transport_layer_mock',
         '$BUILD_DIR/mongo/util/clock_source_mock',
         '$BUILD_DIR/mongo/util/net/message_port_mock',
         'mongoscore',
diff --git a/src/mongo/s/catalog/replset/sharding_catalog_add_shard_test.cpp b/src/mongo/s/catalog/replset/sharding_catalog_add_shard_test.cpp
index 53aa481fdbc..ac7d4dd759e 100644
--- a/src/mongo/s/catalog/replset/sharding_catalog_add_shard_test.cpp
+++ b/src/mongo/s/catalog/replset/sharding_catalog_add_shard_test.cpp
@@ -78,8 +78,7 @@ protected:
      */
     void setUp() override {
         ShardingCatalogTestFixture::setUp();
-
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
 
         configTargeter()->setConnectionStringReturnValue(_configConnStr);
 
diff --git a/src/mongo/s/catalog/replset/sharding_catalog_drop_coll_test.cpp b/src/mongo/s/catalog/replset/sharding_catalog_drop_coll_test.cpp
index b81a5bbc68f..f5609adef83 100644
--- a/src/mongo/s/catalog/replset/sharding_catalog_drop_coll_test.cpp
+++ b/src/mongo/s/catalog/replset/sharding_catalog_drop_coll_test.cpp
@@ -58,8 +58,8 @@ class DropColl2ShardTest : public ShardingCatalogTestFixture {
 public:
     void setUp() override {
         ShardingCatalogTestFixture::setUp();
+        setRemote(_clientHost);
 
-        getMessagingPort()->setRemote(_clientHost);
         configTargeter()->setFindHostReturnValue(_configHost);
         configTargeter()->setConnectionStringReturnValue(_configCS);
 
diff --git a/src/mongo/s/catalog/replset/sharding_catalog_remove_shard_test.cpp b/src/mongo/s/catalog/replset/sharding_catalog_remove_shard_test.cpp
index 59cf084d98d..36fcb6eba63 100644
--- a/src/mongo/s/catalog/replset/sharding_catalog_remove_shard_test.cpp
+++ b/src/mongo/s/catalog/replset/sharding_catalog_remove_shard_test.cpp
@@ -124,7 +124,7 @@ TEST_F(RemoveShardTest, RemoveShardCantRemoveLastShard) {
 TEST_F(RemoveShardTest, RemoveShardStartDraining) {
     string shardName = "shardToRemove";
     const HostAndPort clientHost{"client1:12345"};
-    getMessagingPort()->setRemote(clientHost);
+    setRemote(clientHost);
 
     auto future = launchAsync([&] {
         auto result = assertGet(catalogClient()->removeShard(operationContext(), shardName));
@@ -288,7 +288,7 @@ TEST_F(RemoveShardTest, RemoveShardStillDrainingDatabasesRemaining) {
 TEST_F(RemoveShardTest, RemoveShardCompletion) {
     string shardName = "shardToRemove";
     const HostAndPort clientHost{"client1:12345"};
-    getMessagingPort()->setRemote(clientHost);
+    setRemote(clientHost);
 
     auto future = launchAsync([&] {
         auto result = assertGet(catalogClient()->removeShard(operationContext(), shardName));
diff --git a/src/mongo/s/catalog/replset/sharding_catalog_shard_collection_test.cpp b/src/mongo/s/catalog/replset/sharding_catalog_shard_collection_test.cpp
index d1d36756984..57e35290ced 100644
--- a/src/mongo/s/catalog/replset/sharding_catalog_shard_collection_test.cpp
+++ b/src/mongo/s/catalog/replset/sharding_catalog_shard_collection_test.cpp
@@ -87,7 +87,7 @@ public:
         ShardingCatalogTestFixture::setUp();
         configTargeter()->setFindHostReturnValue(configHost);
         configTargeter()->setConnectionStringReturnValue(configCS);
-        getMessagingPort()->setRemote(clientHost);
+        setRemote(clientHost);
     }
 
     void expectGetDatabase(const DatabaseType& expectedDb) {
diff --git a/src/mongo/s/chunk_manager_tests.cpp b/src/mongo/s/chunk_manager_tests.cpp
index 280b9921d69..5fe831b50a9 100644
--- a/src/mongo/s/chunk_manager_tests.cpp
+++ b/src/mongo/s/chunk_manager_tests.cpp
@@ -68,7 +68,7 @@ class ChunkManagerFixture : public ShardingCatalogTestFixture {
 public:
     void setUp() override {
         ShardingCatalogTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("FakeRemoteClient:34567"));
+        setRemote(HostAndPort("FakeRemoteClient:34567"));
         configTargeter()->setFindHostReturnValue(configHost);
     }
 
diff --git a/src/mongo/s/commands/request.cpp b/src/mongo/s/commands/request.cpp
index 050483d2969..6605db10424 100644
--- a/src/mongo/s/commands/request.cpp
+++ b/src/mongo/s/commands/request.cpp
@@ -42,6 +42,7 @@
 #include "mongo/s/cluster_last_error_info.h"
 #include "mongo/s/commands/strategy.h"
 #include "mongo/s/grid.h"
+#include "mongo/transport/session.h"
 #include "mongo/util/log.h"
 #include "mongo/util/timer.h"
 
@@ -49,8 +50,8 @@ namespace mongo {
 
 using std::string;
 
-Request::Request(Message& m, AbstractMessagingPort* p)
-    : _clientInfo(&cc()), _m(m), _d(m), _p(p), _id(_m.header().getId()), _didInit(false) {
+Request::Request(Message& m)
+    : _clientInfo(&cc()), _m(m), _d(m), _id(_m.header().getId()), _didInit(false) {
     ClusterLastErrorInfo::get(_clientInfo).newRequest();
 }
 
@@ -122,4 +123,8 @@ void Request::process(OperationContext* txn, int attempt) {
            << " op: " << op << " attempt: " << attempt;
 }
 
+transport::Session* Request::session() const {
+    return _clientInfo->session();
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/commands/request.h b/src/mongo/s/commands/request.h
index 5664474e845..ace3e77a720 100644
--- a/src/mongo/s/commands/request.h
+++ b/src/mongo/s/commands/request.h
@@ -37,11 +37,15 @@ namespace mongo {
 class Client;
 class OperationContext;
 
+namespace transport {
+class Session;
+}  // namespace transport
+
 class Request {
     MONGO_DISALLOW_COPYING(Request);
 
 public:
-    Request(Message& m, AbstractMessagingPort* p);
+    Request(Message& m);
 
     const char* getns() const {
         return _d.getns();
@@ -71,9 +75,8 @@ public:
     DbMessage& d() {
         return _d;
     }
-    AbstractMessagingPort* p() const {
-        return _p;
-    }
+
+    transport::Session* session() const;
 
     void process(OperationContext* txn, int attempt = 0);
 
@@ -84,7 +87,6 @@ private:
 
     Message& _m;
     DbMessage _d;
-    AbstractMessagingPort* const _p;
 
     int32_t _id;
 
diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp
index fcba8883592..69d02a3e0e9 100644
--- a/src/mongo/s/commands/strategy.cpp
+++ b/src/mongo/s/commands/strategy.cpp
@@ -176,7 +176,7 @@ void Strategy::queryOp(OperationContext* txn, Request& request) {
 
         BSONObj explainObj = explainBuilder.done();
         replyToQuery(0,  // query result flags
-                     request.p(),
+                     request.session(),
                      request.m(),
                      static_cast<const void*>(explainObj.objdata()),
                      explainObj.objsize(),
@@ -202,7 +202,7 @@ void Strategy::queryOp(OperationContext* txn, Request& request) {
         obj.appendSelfToBufBuilder(reply.bufBuilderForResults());
         numResults++;
     }
-    reply.send(request.p(),
+    reply.send(request.session(),
                0,  // query result flags
                request.m(),
                numResults,
@@ -265,7 +265,7 @@ void Strategy::clientCommandOp(OperationContext* txn, Request& request) {
                 BSONObjBuilder builder(reply.bufBuilderForResults());
                 runAgainstRegistered(txn, q.ns, cmdObj, builder, q.queryOptions);
             }
-            reply.sendCommandReply(request.p(), request.m());
+            reply.sendCommandReply(request.session(), request.m());
             return;
         } catch (const StaleConfigException& e) {
             if (loops <= 0)
@@ -288,7 +288,7 @@ void Strategy::clientCommandOp(OperationContext* txn, Request& request) {
                 BSONObjBuilder builder(reply.bufBuilderForResults());
                 Command::appendCommandStatus(builder, e.toStatus());
             }
-            reply.sendCommandReply(request.p(), request.m());
+            reply.sendCommandReply(request.session(), request.m());
             return;
         }
     }
@@ -325,7 +325,7 @@ bool Strategy::handleSpecialNamespaces(OperationContext* txn, Request& request,
     }
 
     BSONObj x = reply.done();
-    replyToQuery(0, request.p(), request.m(), x);
+    replyToQuery(0, request.session(), request.m(), x);
     return true;
 }
 
@@ -369,7 +369,7 @@ void Strategy::getMore(OperationContext* txn, Request& request) {
     const NamespaceString nss(ns);
     auto statusGetDb = grid.catalogCache()->getDatabase(txn, nss.db().toString());
     if (statusGetDb == ErrorCodes::NamespaceNotFound) {
-        replyToQuery(ResultFlag_CursorNotFound, request.p(), request.m(), 0, 0, 0);
+        replyToQuery(ResultFlag_CursorNotFound, request.session(), request.m(), 0, 0, 0);
         return;
     }
 
@@ -384,7 +384,7 @@ void Strategy::getMore(OperationContext* txn, Request& request) {
 
     auto cursorResponse = ClusterFind::runGetMore(txn, getMoreRequest);
     if (cursorResponse == ErrorCodes::CursorNotFound) {
-        replyToQuery(ResultFlag_CursorNotFound, request.p(), request.m(), 0, 0, 0);
+        replyToQuery(ResultFlag_CursorNotFound, request.session(), request.m(), 0, 0, 0);
         return;
     }
     uassertStatusOK(cursorResponse.getStatus());
@@ -399,7 +399,7 @@ void Strategy::getMore(OperationContext* txn, Request& request) {
     }
 
     replyToQuery(0,
-                 request.p(),
+                 request.session(),
                  request.m(),
                  buffer.buf(),
                  buffer.len(),
diff --git a/src/mongo/s/query/async_results_merger_test.cpp b/src/mongo/s/query/async_results_merger_test.cpp
index b6893c076ea..e2957554ad2 100644
--- a/src/mongo/s/query/async_results_merger_test.cpp
+++ b/src/mongo/s/query/async_results_merger_test.cpp
@@ -67,8 +67,8 @@ public:
 
     void setUp() override {
         ShardingTestFixture::setUp();
+        setRemote(HostAndPort("ClientHost", 12345));
 
-        getMessagingPort()->setRemote(HostAndPort("ClientHost", 12345));
         configTargeter()->setFindHostReturnValue(kTestConfigShardHost);
 
         std::vector<ShardType> shards;
diff --git a/src/mongo/s/server.cpp b/src/mongo/s/server.cpp
index d4912eead77..f4fda4103ee 100644
--- a/src/mongo/s/server.cpp
+++ b/src/mongo/s/server.cpp
@@ -77,12 +77,14 @@
 #include "mongo/s/mongos_options.h"
 #include "mongo/s/query/cluster_cursor_cleanup_job.h"
 #include "mongo/s/query/cluster_cursor_manager.h"
+#include "mongo/s/service_entry_point_mongos.h"
 #include "mongo/s/sharding_egress_metadata_hook_for_mongos.h"
 #include "mongo/s/sharding_initialization.h"
 #include "mongo/s/sharding_uptime_reporter.h"
 #include "mongo/s/version_mongos.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/stdx/thread.h"
+#include "mongo/transport/transport_layer_legacy.h"
 #include "mongo/util/admin_access.h"
 #include "mongo/util/cmdline_utils/censor_cmdline.h"
 #include "mongo/util/concurrency/thread_name.h"
@@ -92,7 +94,6 @@
 #include "mongo/util/log.h"
 #include "mongo/util/net/hostname_canonicalization_worker.h"
 #include "mongo/util/net/message.h"
-#include "mongo/util/net/message_server.h"
 #include "mongo/util/net/socket_exception.h"
 #include "mongo/util/net/ssl_manager.h"
 #include "mongo/util/ntservice.h"
@@ -160,58 +161,6 @@ static BSONObj buildErrReply(const DBException& ex) {
     return errB.obj();
 }
 
-class ShardedMessageHandler : public MessageHandler {
-public:
-    virtual ~ShardedMessageHandler() {}
-
-    virtual void connected(AbstractMessagingPort* p) {
-        Client::initThread("conn", getGlobalServiceContext(), p);
-    }
-
-    virtual void process(Message& m, AbstractMessagingPort* p) {
-        verify(p);
-        Request r(m, p);
-        auto txn = cc().makeOperationContext();
-
-        try {
-            r.init(txn.get());
-            r.process(txn.get());
-        } catch (const AssertionException& ex) {
-            LOG(ex.isUserAssertion() ? 1 : 0) << "Assertion failed"
-                                              << " while processing "
-                                              << networkOpToString(m.operation()) << " op"
-                                              << " for " << r.getnsIfPresent() << causedBy(ex);
-
-            if (r.expectResponse()) {
-                m.header().setId(r.id());
-                replyToQuery(ResultFlag_ErrSet, p, m, buildErrReply(ex));
-            }
-
-            // We *always* populate the last error for now
-            LastError::get(cc()).setLastError(ex.getCode(), ex.what());
-        } catch (const DBException& ex) {
-            log() << "Exception thrown"
-                  << " while processing " << networkOpToString(m.operation()) << " op"
-                  << " for " << r.getnsIfPresent() << causedBy(ex);
-
-            if (r.expectResponse()) {
-                m.header().setId(r.id());
-                replyToQuery(ResultFlag_ErrSet, p, m, buildErrReply(ex));
-            }
-
-            // We *always* populate the last error for now
-            LastError::get(cc()).setLastError(ex.getCode(), ex.what());
-        }
-
-        // Release connections back to pool, if any still cached
-        ShardConnection::releaseMyConnections();
-    }
-
-    virtual void close() {
-        Client::destroy();
-    }
-};
-
 }  // namespace mongo
 
 using namespace mongo;
@@ -278,6 +227,19 @@ static ExitCode runMongosServer() {
 
     _initWireSpec();
 
+    transport::TransportLayerLegacy::Options opts;
+    opts.port = serverGlobalParams.port;
+    opts.ipList = serverGlobalParams.bind_ip;
+
+    auto sep =
+        std::make_shared<ServiceEntryPointMongos>(getGlobalServiceContext()->getTransportLayer());
+
+    auto transportLayer = stdx::make_unique<transport::TransportLayerLegacy>(opts, sep);
+    auto res = transportLayer->setup();
+    if (!res.isOK()) {
+        return EXIT_NET_ERROR;
+    }
+
     // Add sharding hooks to both connection pools - ShardingConnectionHook includes auth hooks
     globalConnPool.addHook(new ShardingConnectionHookForMongos(false));
     shardConnectionPool.addHook(new ShardingConnectionHookForMongos(true));
@@ -352,19 +314,13 @@ static ExitCode runMongosServer() {
 
     PeriodicTask::startRunningPeriodicTasks();
 
-    MessageServer::Options opts;
-    opts.port = serverGlobalParams.port;
-    opts.ipList = serverGlobalParams.bind_ip;
-
-    auto handler = std::make_shared<ShardedMessageHandler>();
-    MessageServer* server = createServer(opts, std::move(handler), getGlobalServiceContext());
-    if (!server->setupSockets()) {
+    auto start = getGlobalServiceContext()->addAndStartTransportLayer(std::move(transportLayer));
+    if (!start.isOK()) {
         return EXIT_NET_ERROR;
     }
-    server->run();
 
-    // MessageServer::run will return when exit code closes its socket
-    return inShutdown() ? EXIT_CLEAN : EXIT_NET_ERROR;
+    // Block until shutdown.
+    return waitForShutdown();
 }
 
 MONGO_INITIALIZER_GENERAL(ForkServer, ("EndStartupOptionHandling"), ("default"))
diff --git a/src/mongo/s/service_entry_point_mongos.cpp b/src/mongo/s/service_entry_point_mongos.cpp
new file mode 100644
index 00000000000..fe7d92fd1ad
--- /dev/null
+++ b/src/mongo/s/service_entry_point_mongos.cpp
@@ -0,0 +1,141 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kNetwork
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/s/service_entry_point_mongos.h"
+
+#include <vector>
+
+#include "mongo/db/lasterror.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/server_options.h"
+#include "mongo/db/service_context.h"
+#include "mongo/s/client/shard_connection.h"
+#include "mongo/s/commands/request.h"
+#include "mongo/stdx/thread.h"
+#include "mongo/transport/service_entry_point_utils.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/util/exit.h"
+#include "mongo/util/log.h"
+#include "mongo/util/net/message.h"
+#include "mongo/util/net/socket_exception.h"
+#include "mongo/util/net/thread_idle_callback.h"
+#include "mongo/util/quick_exit.h"
+#include "mongo/util/scopeguard.h"
+
+namespace mongo {
+
+namespace {
+
+BSONObj buildErrReply(const DBException& ex) {
+    BSONObjBuilder errB;
+    errB.append("$err", ex.what());
+    errB.append("code", ex.getCode());
+    if (!ex._shard.empty()) {
+        errB.append("shard", ex._shard);
+    }
+    return errB.obj();
+}
+
+}  // namespace
+
+using transport::Session;
+using transport::TransportLayer;
+
+ServiceEntryPointMongos::ServiceEntryPointMongos(TransportLayer* tl) : _tl(tl) {}
+
+void ServiceEntryPointMongos::startSession(Session&& session) {
+    launchWrappedServiceEntryWorkerThread(std::move(session),
+                                          [this](Session* session) { _sessionLoop(session); });
+}
+
+void ServiceEntryPointMongos::_sessionLoop(Session* session) {
+    Message message;
+    int64_t counter = 0;
+
+    while (true) {
+        // Release any cached egress connections for client back to pool before destroying
+        auto guard = MakeGuard(ShardConnection::releaseMyConnections);
+
+        message.reset();
+
+        // 1. Source a Message from the client
+        {
+            auto status = session->sourceMessage(&message).wait();
+
+            if (ErrorCodes::isInterruption(status.code())) {
+                break;
+            }
+
+            uassertStatusOK(status);
+        }
+
+        // 2. Build a sharding request
+        Request r(message);
+        auto txn = cc().makeOperationContext();
+
+        try {
+            r.init(txn.get());
+            r.process(txn.get());
+        } catch (const AssertionException& ex) {
+            LOG(ex.isUserAssertion() ? 1 : 0) << "Assertion failed"
+                                              << " while processing "
+                                              << networkOpToString(message.operation()) << " op"
+                                              << " for " << r.getnsIfPresent() << causedBy(ex);
+            if (r.expectResponse()) {
+                message.header().setId(r.id());
+                replyToQuery(ResultFlag_ErrSet, session, message, buildErrReply(ex));
+            }
+
+            // We *always* populate the last error for now
+            LastError::get(cc()).setLastError(ex.getCode(), ex.what());
+        } catch (const DBException& ex) {
+            log() << "Exception thrown"
+                  << " while processing " << networkOpToString(message.operation()) << " op"
+                  << " for " << r.getnsIfPresent() << causedBy(ex);
+
+            if (r.expectResponse()) {
+                message.header().setId(r.id());
+                replyToQuery(ResultFlag_ErrSet, session, message, buildErrReply(ex));
+            }
+
+            // We *always* populate the last error for now
+            LastError::get(cc()).setLastError(ex.getCode(), ex.what());
+        }
+
+        if ((counter++ & 0xf) == 0) {
+            markThreadIdle();
+        }
+    }
+}
+
+}  // namespace mongo
diff --git a/src/mongo/s/service_entry_point_mongos.h b/src/mongo/s/service_entry_point_mongos.h
new file mode 100644
index 00000000000..d96540b2b1e
--- /dev/null
+++ b/src/mongo/s/service_entry_point_mongos.h
@@ -0,0 +1,63 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include <vector>
+
+#include "mongo/base/disallow_copying.h"
+#include "mongo/transport/service_entry_point.h"
+
+namespace mongo {
+
+namespace transport {
+class Session;
+class TransportLayer;
+}  // namespace transport
+
+/**
+ * The entry point from the TransportLayer into Mongos. startSession() spawns and
+ * detaches a new thread for each incoming connection (transport::Session).
+ */
+class ServiceEntryPointMongos final : public ServiceEntryPoint {
+    MONGO_DISALLOW_COPYING(ServiceEntryPointMongos);
+
+public:
+    ServiceEntryPointMongos(transport::TransportLayer* tl);
+
+    virtual ~ServiceEntryPointMongos() = default;
+
+    void startSession(transport::Session&& session) override;
+
+private:
+    void _sessionLoop(transport::Session* session);
+
+    transport::TransportLayer* _tl;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/s/sharding_test_fixture.cpp b/src/mongo/s/sharding_test_fixture.cpp
index 5d8c3e2285e..f099ac2e67b 100644
--- a/src/mongo/s/sharding_test_fixture.cpp
+++ b/src/mongo/s/sharding_test_fixture.cpp
@@ -65,6 +65,8 @@
 #include "mongo/s/write_ops/batched_command_request.h"
 #include "mongo/s/write_ops/batched_command_response.h"
 #include "mongo/stdx/memory.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/transport/transport_layer_mock.h"
 #include "mongo/util/clock_source_mock.h"
 #include "mongo/util/tick_source_mock.h"
 
@@ -92,8 +94,12 @@ void ShardingTestFixture::setUp() {
     _service->setFastClockSource(stdx::make_unique<ClockSourceMock>());
     _service->setPreciseClockSource(stdx::make_unique<ClockSourceMock>());
     _service->setTickSource(stdx::make_unique<TickSourceMock>());
-    _messagePort = stdx::make_unique<MessagingPortMock>();
-    _client = _service->makeClient("ShardingTestFixture", _messagePort.get());
+    auto tlMock = stdx::make_unique<transport::TransportLayerMock>();
+    _transportLayer = tlMock.get();
+    _service->addAndStartTransportLayer(std::move(tlMock));
+    _transportSession =
+        stdx::make_unique<transport::Session>(HostAndPort{}, HostAndPort{}, _transportLayer);
+    _client = _service->makeClient("ShardingTestFixture", _transportSession.get());
     _opCtx = _client->makeOperationContext();
 
     // Set up executor pool used for most operations.
@@ -184,6 +190,7 @@ void ShardingTestFixture::tearDown() {
     grid.catalogClient(_opCtx.get())->shutDown(_opCtx.get());
     grid.clearForUnitTests();
 
+    _transportSession.reset();
     _opCtx.reset();
     _client.reset();
     _service.reset();
@@ -238,10 +245,6 @@ executor::TaskExecutor* ShardingTestFixture::executor() const {
     return _executor;
 }
 
-MessagingPortMock* ShardingTestFixture::getMessagingPort() const {
-    return _messagePort.get();
-}
-
 DistLockManagerMock* ShardingTestFixture::distLock() const {
     invariant(_distLockManager);
     return _distLockManager;
@@ -508,6 +511,10 @@ void ShardingTestFixture::expectCount(const HostAndPort& configHost,
     });
 }
 
+void ShardingTestFixture::setRemote(const HostAndPort& remote) {
+    *_transportSession = transport::Session{remote, HostAndPort{}, _transportLayer};
+}
+
 void ShardingTestFixture::checkReadConcern(const BSONObj& cmdObj,
                                            const Timestamp& expectedTS,
                                            long long expectedTerm) const {
diff --git a/src/mongo/s/sharding_test_fixture.h b/src/mongo/s/sharding_test_fixture.h
index 2a930bbefc4..9365a18dd66 100644
--- a/src/mongo/s/sharding_test_fixture.h
+++ b/src/mongo/s/sharding_test_fixture.h
@@ -60,6 +60,10 @@ class NetworkInterfaceMock;
 class TaskExecutor;
 }  // namespace executor
 
+namespace transport {
+class TransportLayerMock;
+}  // namepsace transport
+
 /**
  * Sets up the mocked out objects for testing the replica-set backed catalog manager and catalog
  * client.
@@ -97,7 +101,7 @@ protected:
 
     executor::TaskExecutor* executor() const;
 
-    MessagingPortMock* getMessagingPort() const;
+    transport::Session* getTransportSession() const;
 
     DistLockManagerMock* distLock() const;
 
@@ -196,6 +200,8 @@ protected:
 
     void shutdownExecutor();
 
+    void setRemote(const HostAndPort& remote);
+
     /**
      * Checks that the given command has the expected settings for read after opTime.
      */
@@ -207,7 +213,8 @@ private:
     std::unique_ptr<ServiceContext> _service;
     ServiceContext::UniqueClient _client;
     ServiceContext::UniqueOperationContext _opCtx;
-    std::unique_ptr<MessagingPortMock> _messagePort;
+    transport::TransportLayerMock* _transportLayer;
+    std::unique_ptr<transport::Session> _transportSession;
 
     RemoteCommandTargeterFactoryMock* _targeterFactory;
     RemoteCommandTargeterMock* _configTargeter;
diff --git a/src/mongo/s/write_ops/batch_write_exec_test.cpp b/src/mongo/s/write_ops/batch_write_exec_test.cpp
index dea991b80ef..d79e45a1fa9 100644
--- a/src/mongo/s/write_ops/batch_write_exec_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec_test.cpp
@@ -66,7 +66,7 @@ public:
 
     void setUp() override {
         ShardingTestFixture::setUp();
-        getMessagingPort()->setRemote(HostAndPort("ClientHost", 12345));
+        setRemote(HostAndPort("ClientHost", 12345));
 
         // Set up the RemoteCommandTargeter for the config shard.
         configTargeter()->setFindHostReturnValue(kTestConfigShardHost);
diff --git a/src/mongo/tools/bridge.cpp b/src/mongo/tools/bridge.cpp
index 7109c174a99..d37751fe537 100644
--- a/src/mongo/tools/bridge.cpp
+++ b/src/mongo/tools/bridge.cpp
@@ -320,17 +320,17 @@ public:
         log() << "Setting random seed: " << mongoBridgeGlobalParams.seed;
     }
 
-    void accepted(AbstractMessagingPort* mp) override final {
+    void accepted(std::unique_ptr<AbstractMessagingPort> mp) override final {
         {
             stdx::lock_guard<stdx::mutex> lk(_portsMutex);
             if (_inShutdown.load()) {
                 mp->shutdown();
                 return;
             }
-            _ports.insert(mp);
+            _ports.insert(mp.get());
         }
 
-        Forwarder f(mp, &_settingsMutex, &_settings, _seedSource.nextInt64());
+        Forwarder f(mp.release(), &_settingsMutex, &_settings, _seedSource.nextInt64());
         stdx::thread t(f);
         t.detach();
     }
diff --git a/src/mongo/transport/SConscript b/src/mongo/transport/SConscript
index 82045d5c0e8..adedade05a3 100644
--- a/src/mongo/transport/SConscript
+++ b/src/mongo/transport/SConscript
@@ -13,13 +13,10 @@ env.CppUnitTest(
 )
 
 env.Library(
-    target='service_entry_point_test_suite',
+    target='transport_layer_common',
     source=[
-        'service_entry_point_test_suite.cpp',
-        ## move these somewhere better:
         'session.cpp',
         'ticket.cpp',
-        'transport_layer.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/unittest/unittest',
@@ -28,10 +25,63 @@ env.Library(
     ],
 )
 
+env.Library(
+    target='transport_layer_manager',
+    source=[
+        'transport_layer_manager.cpp',
+    ],
+    LIBDEPS=[
+        'transport_layer_common',
+    ],
+)
+
+env.Library(
+    target='transport_layer_mock',
+    source=[
+        'transport_layer_mock.cpp',
+    ],
+    LIBDEPS=[
+        'transport_layer_common',
+    ],
+)
+
+env.Library(
+    target='transport_layer_legacy',
+    source=[
+        'transport_layer_legacy.cpp',
+    ],
+    LIBDEPS=[
+        'transport_layer_common',
+        '$BUILD_DIR/mongo/db/service_context',
+        '$BUILD_DIR/mongo/db/stats/counters',
+    ],
+)
+
+env.Library(
+    target='service_entry_point_test_suite',
+    source=[
+        'service_entry_point_test_suite.cpp',
+    ],
+    LIBDEPS=[
+        'transport_layer_common',
+        '$BUILD_DIR/mongo/unittest/unittest',
+    ],
+)
+
+env.Library(
+    target='service_entry_point_utils',
+    source=[
+        'service_entry_point_utils.cpp',
+    ],
+    LIBDEPS=[
+        'transport_layer_common',
+    ],
+)
+
 env.CppUnitTest(
     target='service_entry_point_mock_test',
     source=[
-        'service_entry_point_mock.cpp', ## move?
+        'service_entry_point_mock.cpp',
         'service_entry_point_mock_test.cpp',
     ],
     LIBDEPS=[
diff --git a/src/mongo/transport/service_entry_point_mock.cpp b/src/mongo/transport/service_entry_point_mock.cpp
index 6cfe3606043..8d4530d7f13 100644
--- a/src/mongo/transport/service_entry_point_mock.cpp
+++ b/src/mongo/transport/service_entry_point_mock.cpp
@@ -100,16 +100,12 @@ void ServiceEntryPointMock::run(transport::Session&& session) {
         }
 
         // sourceMessage()
-        auto sourceTicket = _tl->sourceMessage(session, &inMessage);
-        auto sourceRes = _tl->wait(std::move(sourceTicket));
-        if (!sourceRes.isOK()) {
+        if (!session.sourceMessage(&inMessage).wait().isOK()) {
             break;
         }
 
         // sinkMessage()
-        auto sinkTicket = _tl->sinkMessage(session, _outMessage);
-        auto sinkRes = _tl->wait(std::move(sinkTicket));
-        if (!sinkRes.isOK()) {
+        if (!session.sinkMessage(_outMessage).wait().isOK()) {
             break;
         }
     }
diff --git a/src/mongo/transport/service_entry_point_test_suite.cpp b/src/mongo/transport/service_entry_point_test_suite.cpp
index e4f93ae3344..e2a9c7f336c 100644
--- a/src/mongo/transport/service_entry_point_test_suite.cpp
+++ b/src/mongo/transport/service_entry_point_test_suite.cpp
@@ -57,7 +57,6 @@ namespace mongo {
 using namespace transport;
 using namespace stdx::placeholders;
 
-using SessionId = Session::SessionId;
 using TicketCallback = TransportLayer::TicketCallback;
 
 namespace {
@@ -107,7 +106,7 @@ ServiceEntryPointTestSuite::MockTicket::MockTicket(const Session& session,
 ServiceEntryPointTestSuite::MockTicket::MockTicket(const Session& session, Date_t expiration)
     : _sessionId(session.id()), _expiration(expiration) {}
 
-SessionId ServiceEntryPointTestSuite::MockTicket::sessionId() const {
+Session::Id ServiceEntryPointTestSuite::MockTicket::sessionId() const {
     return _sessionId;
 }
 
@@ -140,20 +139,35 @@ Ticket ServiceEntryPointTestSuite::MockTLHarness::sinkMessage(const Session& ses
     return _sinkMessage(session, message, expiration);
 }
 
-Status ServiceEntryPointTestSuite::MockTLHarness::wait(Ticket ticket) {
+Status ServiceEntryPointTestSuite::MockTLHarness::wait(Ticket&& ticket) {
     return _wait(std::move(ticket));
 }
 
-void ServiceEntryPointTestSuite::MockTLHarness::asyncWait(Ticket ticket, TicketCallback callback) {
+void ServiceEntryPointTestSuite::MockTLHarness::asyncWait(Ticket&& ticket,
+                                                          TicketCallback callback) {
     return _asyncWait(std::move(ticket), std::move(callback));
 }
 
+std::string ServiceEntryPointTestSuite::MockTLHarness::getX509SubjectName(const Session& session) {
+    return "mock";
+}
+
+void ServiceEntryPointTestSuite::MockTLHarness::registerTags(const Session& session) {}
+
+TransportLayer::Stats ServiceEntryPointTestSuite::MockTLHarness::sessionStats() {
+    return Stats();
+}
+
 void ServiceEntryPointTestSuite::MockTLHarness::end(const Session& session) {
     return _end(session);
 }
 
-void ServiceEntryPointTestSuite::MockTLHarness::endAllSessions() {
-    return _endAllSessions();
+void ServiceEntryPointTestSuite::MockTLHarness::endAllSessions(Session::TagMask tags) {
+    return _endAllSessions(tags);
+}
+
+Status ServiceEntryPointTestSuite::MockTLHarness::start() {
+    return _start();
 }
 
 void ServiceEntryPointTestSuite::MockTLHarness::shutdown() {
@@ -180,13 +194,13 @@ Status ServiceEntryPointTestSuite::MockTLHarness::_waitOnceThenError(transport::
 Ticket ServiceEntryPointTestSuite::MockTLHarness::_defaultSource(const Session& s,
                                                                  Message* m,
                                                                  Date_t d) {
-    return Ticket(stdx::make_unique<ServiceEntryPointTestSuite::MockTicket>(s, m, d));
+    return Ticket(this, stdx::make_unique<ServiceEntryPointTestSuite::MockTicket>(s, m, d));
 }
 
 Ticket ServiceEntryPointTestSuite::MockTLHarness::_defaultSink(const Session& s,
                                                                const Message&,
                                                                Date_t d) {
-    return Ticket(stdx::make_unique<ServiceEntryPointTestSuite::MockTicket>(s, d));
+    return Ticket(this, stdx::make_unique<ServiceEntryPointTestSuite::MockTicket>(s, d));
 }
 
 Ticket ServiceEntryPointTestSuite::MockTLHarness::_sinkThenErrorOnWait(const Session& s,
@@ -308,6 +322,7 @@ void ServiceEntryPointTestSuite::interruptingSessionTest() {
     auto resumeAFuture = resumeA.get_future();
 
     stdx::promise<void> testComplete;
+
     auto testFuture = testComplete.get_future();
 
     _tl->_resetHooks();
@@ -376,10 +391,11 @@ void ServiceEntryPointTestSuite::burstStressTest(int numSessions,
                                                  Milliseconds delay) {
     AtomicWord<int> ended{0};
     stdx::promise<void> allSessionsComplete;
+
     auto allCompleteFuture = allSessionsComplete.get_future();
 
     stdx::mutex cyclesLock;
-    std::unordered_map<SessionId, int> completedCycles;
+    std::unordered_map<Session::Id, int> completedCycles;
 
     _tl->_resetHooks();
 
diff --git a/src/mongo/transport/service_entry_point_test_suite.h b/src/mongo/transport/service_entry_point_test_suite.h
index a1999efd873..828fee5fe3c 100644
--- a/src/mongo/transport/service_entry_point_test_suite.h
+++ b/src/mongo/transport/service_entry_point_test_suite.h
@@ -99,7 +99,7 @@ private:
         MockTicket(MockTicket&&) = default;
         MockTicket& operator=(MockTicket&&) = default;
 
-        transport::Session::SessionId sessionId() const override;
+        transport::Session::Id sessionId() const override;
 
         Date_t expiration() const override;
 
@@ -107,7 +107,7 @@ private:
 
     private:
         boost::optional<Message*> _message;
-        SessionId _sessionId;
+        transport::Session::Id _sessionId;
         Date_t _expiration;
     };
 
@@ -127,10 +127,15 @@ private:
             const transport::Session& session,
             const Message& message,
             Date_t expiration = transport::Ticket::kNoExpirationDate) override;
-        Status wait(transport::Ticket ticket) override;
-        void asyncWait(transport::Ticket ticket, TicketCallback callback) override;
+        Status wait(transport::Ticket&& ticket) override;
+        void asyncWait(transport::Ticket&& ticket, TicketCallback callback) override;
+        std::string getX509SubjectName(const transport::Session& session) override;
+        void registerTags(const transport::Session& session) override;
+        Stats sessionStats() override;
         void end(const transport::Session& session) override;
-        void endAllSessions() override;
+        void endAllSessions(
+            transport::Session::TagMask tags = transport::Session::kEmptyTagMask) override;
+        Status start() override;
         void shutdown() override;
 
         ServiceEntryPointTestSuite::MockTicket* getMockTicket(const transport::Ticket& ticket);
@@ -143,7 +148,9 @@ private:
         stdx::function<Status(transport::Ticket)> _wait;
         stdx::function<void(transport::Ticket, TicketCallback)> _asyncWait;
         stdx::function<void(const transport::Session&)> _end;
-        stdx::function<void(void)> _endAllSessions = [] {};
+        stdx::function<void(transport::Session::TagMask tags)> _endAllSessions =
+            [](transport::Session::TagMask tags) {};
+        stdx::function<Status(void)> _start = [] { return Status::OK(); };
         stdx::function<void(void)> _shutdown = [] {};
 
         // Pre-set hook methods
diff --git a/src/mongo/transport/service_entry_point_utils.cpp b/src/mongo/transport/service_entry_point_utils.cpp
new file mode 100644
index 00000000000..efe934967fb
--- /dev/null
+++ b/src/mongo/transport/service_entry_point_utils.cpp
@@ -0,0 +1,150 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kDefault
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/transport/service_entry_point_utils.h"
+
+#include "mongo/db/client.h"
+#include "mongo/db/server_options.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/debug_util.h"
+#include "mongo/util/log.h"
+#include "mongo/util/net/socket_exception.h"
+#include "mongo/util/quick_exit.h"
+
+#ifdef __linux__  // TODO: consider making this ifndef _WIN32
+#include <sys/resource.h>
+#endif
+
+#if !defined(__has_feature)
+#define __has_feature(x) 0
+#endif
+
+namespace mongo {
+
+namespace {
+
+struct Context {
+    Context(transport::Session session, stdx::function<void(transport::Session*)> task)
+        : session(std::move(session)), task(std::move(task)) {}
+
+    transport::Session session;
+    stdx::function<void(transport::Session*)> task;
+};
+
+void* runFunc(void* ptr) {
+    std::unique_ptr<Context> ctx(static_cast<Context*>(ptr));
+
+    auto tl = ctx->session.getTransportLayer();
+    Client::initThread("conn", &ctx->session);
+    setThreadName(std::string(str::stream() << "conn" << ctx->session.id()));
+
+    try {
+        ctx->task(&ctx->session);
+    } catch (const AssertionException& e) {
+        log() << "AssertionException handling request, closing client connection: " << e;
+    } catch (const SocketException& e) {
+        log() << "SocketException handling request, closing client connection: " << e;
+    } catch (const DBException& e) {
+        // must be right above std::exception to avoid catching subclasses
+        log() << "DBException handling request, closing client connection: " << e;
+    } catch (const std::exception& e) {
+        error() << "Uncaught std::exception: " << e.what() << ", terminating";
+        quickExit(EXIT_UNCAUGHT);
+    }
+
+    tl->end(ctx->session);
+
+    if (!serverGlobalParams.quiet) {
+        auto conns = tl->sessionStats().numOpenSessions;
+        const char* word = (conns == 1 ? " connection" : " connections");
+        log() << "end connection " << ctx->session.remote() << " (" << conns << word
+              << " now open)";
+    }
+
+    Client::destroy();
+
+    return nullptr;
+}
+}  // namespace
+
+void launchWrappedServiceEntryWorkerThread(transport::Session&& session,
+                                           stdx::function<void(transport::Session*)> task) {
+    auto ctx = stdx::make_unique<Context>(std::move(session), std::move(task));
+
+    try {
+#ifndef __linux__  // TODO: consider making this ifdef _WIN32
+        stdx::thread(stdx::bind(runFunc, ctx.get())).detach();
+        ctx.release();
+#else
+        pthread_attr_t attrs;
+        pthread_attr_init(&attrs);
+        pthread_attr_setdetachstate(&attrs, PTHREAD_CREATE_DETACHED);
+
+        static const size_t STACK_SIZE =
+            1024 * 1024;  // if we change this we need to update the warning
+
+        struct rlimit limits;
+        invariant(getrlimit(RLIMIT_STACK, &limits) == 0);
+        if (limits.rlim_cur > STACK_SIZE) {
+            size_t stackSizeToSet = STACK_SIZE;
+#if !__has_feature(address_sanitizer)
+            if (kDebugBuild)
+                stackSizeToSet /= 2;
+#endif
+            pthread_attr_setstacksize(&attrs, stackSizeToSet);
+        } else if (limits.rlim_cur < 1024 * 1024) {
+            warning() << "Stack size set to " << (limits.rlim_cur / 1024) << "KB. We suggest 1MB";
+        }
+
+
+        pthread_t thread;
+        int failed = pthread_create(&thread, &attrs, runFunc, ctx.get());
+
+        pthread_attr_destroy(&attrs);
+
+        if (failed) {
+            log() << "pthread_create failed: " << errnoWithDescription(failed);
+            throw std::system_error(
+                std::make_error_code(std::errc::resource_unavailable_try_again));
+        }
+        ctx.release();
+#endif  // __linux__
+
+    } catch (...) {
+        log() << "failed to create service entry worker thread for " << ctx->session.remote();
+    }
+}
+
+}  // namespace mongo
diff --git a/src/mongo/transport/transport_layer.cpp b/src/mongo/transport/service_entry_point_utils.h
index 20030e64661..42be4079f9c 100644
--- a/src/mongo/transport/transport_layer.cpp
+++ b/src/mongo/transport/service_entry_point_utils.h
@@ -26,19 +26,17 @@
  *    it in the license file.
  */
 
-#include "mongo/platform/basic.h"
+#pragma once
 
-#include "mongo/transport/transport_layer.h"
+#include "mongo/stdx/functional.h"
 
 namespace mongo {
-namespace transport {
 
-TransportLayer::TransportLayer() = default;
-TransportLayer::~TransportLayer() = default;
+namespace transport {
+class Session;
+}  // namespace transport
 
-TicketImpl* TransportLayer::getTicketImpl(const Ticket& ticket) {
-    return ticket.impl();
-}
+void launchWrappedServiceEntryWorkerThread(transport::Session&& session,
+                                           stdx::function<void(transport::Session*)> task);
 
-}  // namespace transport
 }  // namespace mongo
diff --git a/src/mongo/transport/session.cpp b/src/mongo/transport/session.cpp
index b9f382a6965..ebf237f8bd4 100644
--- a/src/mongo/transport/session.cpp
+++ b/src/mongo/transport/session.cpp
@@ -43,11 +43,14 @@ AtomicUInt64 sessionIdCounter(0);
 }  // namespace
 
 Session::Session(HostAndPort remote, HostAndPort local, TransportLayer* tl)
-    : _id(sessionIdCounter.addAndFetch(1)), _remote(remote), _local(local), _tl(tl) {}
+    : _id(sessionIdCounter.addAndFetch(1)),
+      _remote(std::move(remote)),
+      _local(std::move(local)),
+      _tags(kEmptyTagMask),
+      _tl(tl) {}
 
 Session::~Session() {
     if (_tl != nullptr) {
-        invariant(_tl);
         _tl->end(*this);
     }
 }
@@ -75,16 +78,21 @@ Session& Session::operator=(Session&& other) {
     return *this;
 }
 
-Session::SessionId Session::id() const {
-    return _id;
+void Session::replaceTags(TagMask tags) {
+    _tags = tags;
+    _tl->registerTags(*this);
 }
 
-const HostAndPort& Session::remote() const {
-    return _remote;
+Ticket Session::sourceMessage(Message* message, Date_t expiration) {
+    return _tl->sourceMessage(*this, message, expiration);
 }
 
-const HostAndPort& Session::local() const {
-    return _local;
+Ticket Session::sinkMessage(const Message& message, Date_t expiration) {
+    return _tl->sinkMessage(*this, message, expiration);
+}
+
+std::string Session::getX509SubjectName() const {
+    return _tl->getX509SubjectName(*this);
 }
 
 }  // namespace transport
diff --git a/src/mongo/transport/session.h b/src/mongo/transport/session.h
index 4692f22437d..b340b0fecf8 100644
--- a/src/mongo/transport/session.h
+++ b/src/mongo/transport/session.h
@@ -29,7 +29,10 @@
 #pragma once
 
 #include "mongo/base/disallow_copying.h"
+#include "mongo/transport/session_id.h"
+#include "mongo/transport/ticket.h"
 #include "mongo/util/net/hostandport.h"
+#include "mongo/util/net/message.h"
 
 namespace mongo {
 namespace transport {
@@ -47,7 +50,14 @@ public:
     /**
      * Type to indicate the internal id for this session.
      */
-    using SessionId = uint64_t;
+    using Id = SessionId;
+
+    /**
+     * Tags for groups of connections.
+     */
+    using TagMask = uint32_t;
+    static constexpr TagMask kEmptyTagMask = 0;
+    static constexpr TagMask kKeepOpen = 1;
 
     /**
      * Construct a new session.
@@ -68,24 +78,72 @@ public:
     /**
      * Return the id for this session.
      */
-    SessionId id() const;
+    Id id() const {
+        return _id;
+    }
 
     /**
      * Return the remote host for this session.
      */
-    const HostAndPort& remote() const;
+    const HostAndPort& remote() const {
+        return _remote;
+    }
 
     /**
      * Return the local host information for this session.
      */
-    const HostAndPort& local() const;
+    const HostAndPort& local() const {
+        return _local;
+    }
+
+    /**
+     * Return the X509 subject name for this connection (SSL only).
+     */
+    std::string getX509SubjectName() const;
+
+    /**
+     * Set this session's tags. This Session will register
+     * its new tags with its TransportLayer.
+     */
+    void replaceTags(TagMask tags);
+
+    /**
+     * Get this session's tags.
+     */
+    TagMask getTags() const {
+        return _tags;
+    }
+
+    /**
+     * Source (receive) a new Message for this Session.
+     *
+     * This method will forward to sourceMessage on this Session's transport layer.
+     */
+    Ticket sourceMessage(Message* message, Date_t expiration = Ticket::kNoExpirationDate);
+
+    /**
+     * Sink (send) a new Message for this Session. This method should be used
+     * to send replies to a given host.
+     *
+     * This method will forward to sinkMessage on this Session's transport layer.
+     */
+    Ticket sinkMessage(const Message& message, Date_t expiration = Ticket::kNoExpirationDate);
+
+    /**
+     * The TransportLayer for this Session.
+     */
+    TransportLayer* getTransportLayer() const {
+        return _tl;
+    }
 
 private:
-    SessionId _id;
+    Id _id;
 
     HostAndPort _remote;
     HostAndPort _local;
 
+    TagMask _tags;
+
     TransportLayer* _tl;
 };
 
diff --git a/src/mongo/transport/session_id.h b/src/mongo/transport/session_id.h
new file mode 100644
index 00000000000..b1395ef1429
--- /dev/null
+++ b/src/mongo/transport/session_id.h
@@ -0,0 +1,39 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+
+namespace mongo {
+namespace transport {
+
+using SessionId = uint64_t;
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/transport/ticket.cpp b/src/mongo/transport/ticket.cpp
index 76bce1dc51e..77a290087dd 100644
--- a/src/mongo/transport/ticket.cpp
+++ b/src/mongo/transport/ticket.cpp
@@ -30,30 +30,27 @@
 
 #include "mongo/transport/ticket.h"
 #include "mongo/transport/ticket_impl.h"
+#include "mongo/transport/transport_layer.h"
 
 namespace mongo {
 namespace transport {
 
 const Date_t Ticket::kNoExpirationDate{Date_t::max()};
 
-Ticket::Ticket(std::unique_ptr<TicketImpl> ticket) : _ticket(std::move(ticket)) {}
+Ticket::Ticket(TransportLayer* tl, std::unique_ptr<TicketImpl> ticket)
+    : _tl(tl), _ticket(std::move(ticket)) {}
 
 Ticket::~Ticket() = default;
 
 Ticket::Ticket(Ticket&&) = default;
 Ticket& Ticket::operator=(Ticket&&) = default;
 
-Session::SessionId Ticket::sessionId() const {
-    return _ticket->sessionId();
+Status Ticket::wait()&& {
+    return _tl->wait(std::move(*this));
 }
 
-Date_t Ticket::expiration() const {
-    return _ticket->expiration();
-}
-
-// TODO should this actually be const?
-TicketImpl* Ticket::impl() const {
-    return _ticket.get();
+void Ticket::asyncWait(TicketCallback cb)&& {
+    return _tl->asyncWait(std::move(*this), std::move(cb));
 }
 
 }  // namespace transport
diff --git a/src/mongo/transport/ticket.h b/src/mongo/transport/ticket.h
index 875a66b3602..dd8927085aa 100644
--- a/src/mongo/transport/ticket.h
+++ b/src/mongo/transport/ticket.h
@@ -28,14 +28,18 @@
 
 #pragma once
 
+#include <memory>
+
 #include "mongo/base/disallow_copying.h"
-#include "mongo/transport/session.h"
+#include "mongo/stdx/functional.h"
+#include "mongo/transport/session_id.h"
+#include "mongo/transport/ticket_impl.h"
 #include "mongo/util/time_support.h"
 
 namespace mongo {
 namespace transport {
 
-class TicketImpl;
+class TransportLayer;
 
 /**
  * A Ticket represents some work to be done within the TransportLayer.
@@ -46,16 +50,16 @@ class Ticket {
     MONGO_DISALLOW_COPYING(Ticket);
 
 public:
-    friend class TransportLayer;
+    using TicketCallback = stdx::function<void(Status)>;
 
-    using SessionId = Session::SessionId;
+    friend class TransportLayer;
 
     /**
      * Indicates that there is no expiration time by when a ticket needs to complete.
      */
     static const Date_t kNoExpirationDate;
 
-    Ticket(std::unique_ptr<TicketImpl> ticket);
+    Ticket(TransportLayer* tl, std::unique_ptr<TicketImpl> ticket);
     ~Ticket();
 
     /**
@@ -67,20 +71,41 @@ public:
     /**
      * Return this ticket's session id.
      */
-    SessionId sessionId() const;
+    SessionId sessionId() const {
+        return _ticket->sessionId();
+    }
 
     /**
      * Return this ticket's expiration date.
      */
-    Date_t expiration() const;
+    Date_t expiration() const {
+        return _ticket->expiration();
+    }
+
+    /**
+     * Wait for this ticket to be filled.
+     *
+     * This is this-rvalue qualified because it consumes the ticket
+     */
+    Status wait() &&;
+
+    /**
+     * Asynchronously wait for this ticket to be filled.
+     *
+     * This is this-rvalue qualified because it consumes the ticket
+     */
+    void asyncWait(TicketCallback cb) &&;
 
 protected:
     /**
      * Return a non-owning pointer to the underlying TicketImpl type
      */
-    TicketImpl* impl() const;
+    TicketImpl* impl() const {
+        return _ticket.get();
+    }
 
 private:
+    TransportLayer* _tl;
     std::unique_ptr<TicketImpl> _ticket;
 };
 
diff --git a/src/mongo/transport/ticket_impl.h b/src/mongo/transport/ticket_impl.h
index 378f9fb2999..4cbbc0a22d0 100644
--- a/src/mongo/transport/ticket_impl.h
+++ b/src/mongo/transport/ticket_impl.h
@@ -28,7 +28,8 @@
 
 #pragma once
 
-#include "mongo/transport/session.h"
+#include "mongo/base/disallow_copying.h"
+#include "mongo/transport/session_id.h"
 #include "mongo/util/time_support.h"
 
 namespace mongo {
@@ -43,8 +44,6 @@ class TicketImpl {
     MONGO_DISALLOW_COPYING(TicketImpl);
 
 public:
-    using SessionId = Session::SessionId;
-
     virtual ~TicketImpl() = default;
 
     TicketImpl(TicketImpl&&) = default;
diff --git a/src/mongo/transport/transport_layer.h b/src/mongo/transport/transport_layer.h
index 3028506c6f0..dde6248eaf5 100644
--- a/src/mongo/transport/transport_layer.h
+++ b/src/mongo/transport/transport_layer.h
@@ -57,7 +57,30 @@ class TransportLayer {
     MONGO_DISALLOW_COPYING(TransportLayer);
 
 public:
-    virtual ~TransportLayer();
+    /**
+     * Stats for sessions open in the Transport Layer.
+     */
+    struct Stats {
+        /**
+         * Returns the number of sessions currently open in the transport layer.
+         */
+        size_t numOpenSessions = 0;
+
+        /**
+         * Returns the total number of sessions that have ever been created by this TransportLayer.
+         */
+        size_t numCreatedSessions = 0;
+
+        /**
+         * Returns the number of available sessions we could still open. Only relevant
+         * when we are operating under a transport::Session limit (for example, in the
+         * legacy implementation, we respect a maximum number of connections). If there
+         * is no session limit, returns std::numeric_limits<int>::max().
+         */
+        size_t numAvailableSessions = 0;
+    };
+
+    virtual ~TransportLayer() = default;
 
     /**
      * Source (receive) a new Message for this Session.
@@ -101,7 +124,7 @@ public:
      * This thread may be used by the TransportLayer to run other Tickets that were
      * enqueued prior to this call.
      */
-    virtual Status wait(Ticket ticket) = 0;
+    virtual Status wait(Ticket&& ticket) = 0;
 
     /**
      * Callback for Tickets that are run via asyncWait().
@@ -115,7 +138,27 @@ public:
      * This thread will not be used by the TransportLayer to perform work. The callback
      * passed to asyncWait() may be run on any thread.
      */
-    virtual void asyncWait(Ticket ticket, TicketCallback callback) = 0;
+    virtual void asyncWait(Ticket&& ticket, TicketCallback callback) = 0;
+
+    /**
+     * Tag this Session within the TransportLayer with the tags currently assigned to the
+     * Session. If endAllSessions() is called with a matching
+     * Session::TagMask, this Session will not be ended.
+     *
+     * Before calling this method, use Session::replaceTags() to set the desired TagMask.
+     */
+    virtual void registerTags(const Session& session) = 0;
+
+    /**
+     * Return the stored X509 subject name for this session. If the session does not
+     * exist in this TransportLayer, returns "".
+     */
+    virtual std::string getX509SubjectName(const Session& session) = 0;
+
+    /**
+     * Returns the number of sessions currently open in the transport layer.
+     */
+    virtual Stats sessionStats() = 0;
 
     /**
      * End the given Session. Tickets for this Session that have already been
@@ -131,11 +174,20 @@ public:
     virtual void end(const Session& session) = 0;
 
     /**
-     * End all active sessions in the TransportLayer. Tickets that have already been
-     * started via wait() or asyncWait() will complete, but may return a failed Status.
-     * This method is synchronous and will not return until all sessions have ended.
+     * End all active sessions in the TransportLayer. Tickets that have already been started via
+     * wait() or asyncWait() will complete, but may return a failed Status.  This method is
+     * asynchronous and will return after all sessions have been notified to end.
+     *
+     * If a TagMask is provided, endAllSessions() will skip over sessions with matching
+     * tags and leave them open.
+     */
+    virtual void endAllSessions(Session::TagMask tags = Session::kEmptyTagMask) = 0;
+
+    /**
+     * Start the TransportLayer. After this point, the TransportLayer will begin accepting active
+     * sessions from new transport::Endpoints.
      */
-    virtual void endAllSessions() = 0;
+    virtual Status start() = 0;
 
     /**
      * Shut the TransportLayer down. After this point, the TransportLayer will
@@ -147,12 +199,21 @@ public:
     virtual void shutdown() = 0;
 
 protected:
-    TransportLayer();
+    TransportLayer() = default;
 
     /**
      * Return the implementation of this Ticket.
      */
-    TicketImpl* getTicketImpl(const Ticket& ticket);
+    TicketImpl* getTicketImpl(const Ticket& ticket) {
+        return ticket.impl();
+    }
+
+    /**
+     * Return the transport layer of this Ticket.
+     */
+    TransportLayer* getTicketTransportLayer(const Ticket& ticket) {
+        return ticket._tl;
+    }
 };
 
 }  // namespace transport
diff --git a/src/mongo/transport/transport_layer_legacy.cpp b/src/mongo/transport/transport_layer_legacy.cpp
new file mode 100644
index 00000000000..b45d6a6ad07
--- /dev/null
+++ b/src/mongo/transport/transport_layer_legacy.cpp
@@ -0,0 +1,307 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kNetwork
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/transport/transport_layer_legacy.h"
+
+#include "mongo/base/checked_cast.h"
+#include "mongo/config.h"
+#include "mongo/db/service_context.h"
+#include "mongo/db/stats/counters.h"
+#include "mongo/stdx/functional.h"
+#include "mongo/transport/service_entry_point.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
+#include "mongo/util/net/abstract_message_port.h"
+#include "mongo/util/net/socket_exception.h"
+
+namespace mongo {
+namespace transport {
+
+TransportLayerLegacy::ListenerLegacy::ListenerLegacy(const TransportLayerLegacy::Options& opts,
+                                                     NewConnectionCb callback)
+    : Listener("", opts.ipList, opts.port, getGlobalServiceContext(), true),
+      _accepted(std::move(callback)) {}
+
+void TransportLayerLegacy::ListenerLegacy::accepted(std::unique_ptr<AbstractMessagingPort> mp) {
+    _accepted(std::move(mp));
+}
+
+TransportLayerLegacy::TransportLayerLegacy(const TransportLayerLegacy::Options& opts,
+                                           std::shared_ptr<ServiceEntryPoint> sep)
+    : _sep(sep),
+      _listener(stdx::make_unique<ListenerLegacy>(
+          opts,
+          stdx::bind(&TransportLayerLegacy::_handleNewConnection, this, stdx::placeholders::_1))),
+      _running(false),
+      _options(opts) {}
+
+TransportLayerLegacy::LegacyTicket::LegacyTicket(const Session& session,
+                                                 Date_t expiration,
+                                                 WorkHandle work)
+    : _sessionId(session.id()), _expiration(expiration), _fill(std::move(work)) {}
+
+Session::Id TransportLayerLegacy::LegacyTicket::sessionId() const {
+    return _sessionId;
+}
+
+Date_t TransportLayerLegacy::LegacyTicket::expiration() const {
+    return _expiration;
+}
+
+Status TransportLayerLegacy::setup() {
+    if (!_listener->setupSockets()) {
+        error() << "Failed to set up sockets during startup.";
+        return {ErrorCodes::InternalError, "Failed to set up sockets"};
+    }
+
+    return Status::OK();
+}
+
+Status TransportLayerLegacy::start() {
+    if (_running.swap(true)) {
+        return {ErrorCodes::InternalError, "TransportLayer is already running"};
+    }
+
+    _listenerThread = stdx::thread([this]() { _listener->initAndListen(); });
+
+    return Status::OK();
+}
+
+TransportLayerLegacy::~TransportLayerLegacy() = default;
+
+Ticket TransportLayerLegacy::sourceMessage(const Session& session,
+                                           Message* message,
+                                           Date_t expiration) {
+    auto sourceCb = [message](AbstractMessagingPort* amp) -> Status {
+        if (!amp->recv(*message)) {
+            return {ErrorCodes::HostUnreachable, "Recv failed"};
+        }
+        return Status::OK();
+    };
+
+    return Ticket(this, stdx::make_unique<LegacyTicket>(session, expiration, std::move(sourceCb)));
+}
+
+std::string TransportLayerLegacy::getX509SubjectName(const Session& session) {
+    {
+        stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+        auto conn = _connections.find(session.id());
+        if (conn == _connections.end()) {
+            // Return empty string if the session is not found
+            return "";
+        }
+
+        return conn->second.x509SubjectName.value_or("");
+    }
+}
+
+TransportLayer::Stats TransportLayerLegacy::sessionStats() {
+    Stats stats;
+    {
+        stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+        stats.numOpenSessions = _connections.size();
+    }
+
+    stats.numAvailableSessions = Listener::globalTicketHolder.available();
+    stats.numCreatedSessions = Listener::globalConnectionNumber.load();
+
+    return stats;
+}
+
+Ticket TransportLayerLegacy::sinkMessage(const Session& session,
+                                         const Message& message,
+                                         Date_t expiration) {
+    auto sinkCb = [&message](AbstractMessagingPort* amp) -> Status {
+        try {
+            amp->say(message);
+            return Status::OK();
+        } catch (const SocketException& e) {
+            return {ErrorCodes::HostUnreachable, e.what()};
+        }
+    };
+
+    return Ticket(this, stdx::make_unique<LegacyTicket>(session, expiration, std::move(sinkCb)));
+}
+
+Status TransportLayerLegacy::wait(Ticket&& ticket) {
+    return _runTicket(std::move(ticket));
+}
+
+void TransportLayerLegacy::asyncWait(Ticket&& ticket, TicketCallback callback) {
+    // Left unimplemented because there is no reasonable way to offer general async waiting besides
+    // offering a background thread that can handle waits for multiple tickets. We may never
+    // implement this for the legacy TL.
+    MONGO_UNREACHABLE;
+}
+
+void TransportLayerLegacy::end(const Session& session) {
+    stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+    auto conn = _connections.find(session.id());
+    if (conn != _connections.end()) {
+        _endSession_inlock(conn);
+    }
+}
+
+void TransportLayerLegacy::registerTags(const Session& session) {
+    stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+    auto conn = _connections.find(session.id());
+    if (conn != _connections.end()) {
+        conn->second.tags = session.getTags();
+    }
+}
+
+void TransportLayerLegacy::_endSession_inlock(
+    decltype(TransportLayerLegacy::_connections.begin()) conn) {
+    conn->second.amp->shutdown();
+
+    // If the amp is in use it means that we're in the middle of fulfilling the ticket in _runTicket
+    // and can't erase it immediately.
+    //
+    // In that case we'll rely on _runTicket to do the removal for us later
+    if (conn->second.inUse) {
+        conn->second.ended = true;
+    } else {
+        Listener::globalTicketHolder.release();
+        _connections.erase(conn);
+    }
+}
+
+void TransportLayerLegacy::endAllSessions(Session::TagMask tags) {
+    log() << "legacy transport layer ending all sessions";
+    {
+        stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+        auto&& conn = _connections.begin();
+        while (conn != _connections.end()) {
+            // If we erase this connection below, we invalidate our iterator, use a placeholder.
+            auto placeholder = conn;
+            placeholder++;
+
+            if (conn->second.tags & tags) {
+                log() << "Skip closing connection for connection # " << conn->second.connectionId;
+            } else {
+                _endSession_inlock(conn);
+            }
+
+            conn = placeholder;
+        }
+    }
+}
+
+void TransportLayerLegacy::shutdown() {
+    _running.store(false);
+    _listener->shutdown();
+    _listenerThread.join();
+    endAllSessions();
+}
+
+Status TransportLayerLegacy::_runTicket(Ticket ticket) {
+    if (!_running.load()) {
+        return {ErrorCodes::ShutdownInProgress, "TransportLayer in shutdown"};
+    }
+
+    if (ticket.expiration() < Date_t::now()) {
+        return {ErrorCodes::ExceededTimeLimit, "Ticket has expired"};
+    }
+
+    AbstractMessagingPort* amp;
+
+    {
+        stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+
+        auto conn = _connections.find(ticket.sessionId());
+        if (conn == _connections.end()) {
+            return {ErrorCodes::TransportSessionNotFound, "No such session in TransportLayer"};
+        }
+
+        // "check out" the port
+        conn->second.inUse = true;
+        amp = conn->second.amp.get();
+    }
+
+    amp->clearCounters();
+
+    auto legacyTicket = checked_cast<LegacyTicket*>(getTicketImpl(ticket));
+    auto res = legacyTicket->_fill(amp);
+
+    networkCounter.hit(amp->getBytesIn(), amp->getBytesOut());
+
+    {
+        stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+
+        auto conn = _connections.find(ticket.sessionId());
+        invariant(conn != _connections.end());
+
+#ifdef MONGO_CONFIG_SSL
+        // If we didn't have an X509 subject name, see if we have one now
+        if (!conn->second.x509SubjectName) {
+            auto name = amp->getX509SubjectName();
+            if (name != "") {
+                conn->second.x509SubjectName = name;
+            }
+        }
+#endif
+        conn->second.inUse = false;
+
+        if (conn->second.ended) {
+            Listener::globalTicketHolder.release();
+            _connections.erase(conn);
+        }
+    }
+
+    return res;
+}
+
+void TransportLayerLegacy::_handleNewConnection(std::unique_ptr<AbstractMessagingPort> amp) {
+    if (!Listener::globalTicketHolder.tryAcquire()) {
+        log() << "connection refused because too many open connections: "
+              << Listener::globalTicketHolder.used();
+        amp->shutdown();
+        return;
+    }
+
+    Session session(amp->remote(), HostAndPort(amp->localAddr().toString(true)), this);
+
+    amp->setLogLevel(logger::LogSeverity::Debug(1));
+
+    {
+        stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);
+        _connections.emplace(std::piecewise_construct,
+                             std::forward_as_tuple(session.id()),
+                             std::forward_as_tuple(std::move(amp), false, session.getTags()));
+    }
+
+    invariant(_sep);
+    _sep->startSession(std::move(session));
+}
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/transport/transport_layer_legacy.h b/src/mongo/transport/transport_layer_legacy.h
new file mode 100644
index 00000000000..ec17c76ca1e
--- /dev/null
+++ b/src/mongo/transport/transport_layer_legacy.h
@@ -0,0 +1,174 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include <unordered_map>
+
+#include "mongo/stdx/memory.h"
+#include "mongo/stdx/mutex.h"
+#include "mongo/stdx/thread.h"
+#include "mongo/transport/ticket_impl.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/util/net/listen.h"
+#include "mongo/util/net/sock.h"
+
+namespace mongo {
+
+class AbstractMessagingPort;
+class ServiceEntryPoint;
+
+namespace transport {
+
+/**
+ * A TransportLayer implementation based on legacy networking primitives (the Listener,
+ * AbstractMessagingPort).
+ */
+class TransportLayerLegacy final : public TransportLayer {
+    MONGO_DISALLOW_COPYING(TransportLayerLegacy);
+
+public:
+    struct Options {
+        int port;            // port to bind to
+        std::string ipList;  // addresses to bind to
+
+        Options() : port(0), ipList("") {}
+    };
+
+    TransportLayerLegacy(const Options& opts, std::shared_ptr<ServiceEntryPoint> sep);
+
+    ~TransportLayerLegacy();
+
+    Status setup();
+    Status start() override;
+
+    Ticket sourceMessage(const Session& session,
+                         Message* message,
+                         Date_t expiration = Ticket::kNoExpirationDate) override;
+
+    Ticket sinkMessage(const Session& session,
+                       const Message& message,
+                       Date_t expiration = Ticket::kNoExpirationDate) override;
+
+    Status wait(Ticket&& ticket) override;
+    void asyncWait(Ticket&& ticket, TicketCallback callback) override;
+
+    void registerTags(const Session& session) override;
+    std::string getX509SubjectName(const Session& session) override;
+
+    Stats sessionStats() override;
+
+    void end(const Session& session) override;
+    void endAllSessions(transport::Session::TagMask tags = Session::kKeepOpen) override;
+    void shutdown() override;
+
+private:
+    void _handleNewConnection(std::unique_ptr<AbstractMessagingPort> amp);
+
+    Status _runTicket(Ticket ticket);
+
+    using NewConnectionCb = stdx::function<void(std::unique_ptr<AbstractMessagingPort>)>;
+    using WorkHandle = stdx::function<Status(AbstractMessagingPort*)>;
+
+    /**
+     * A TicketImpl implementation for this TransportLayer. WorkHandle is a callable that
+     * can be invoked to fill this ticket.
+     */
+    class LegacyTicket : public TicketImpl {
+        MONGO_DISALLOW_COPYING(LegacyTicket);
+
+    public:
+        LegacyTicket(const Session& session, Date_t expiration, WorkHandle work);
+
+        SessionId sessionId() const override;
+        Date_t expiration() const override;
+
+        SessionId _sessionId;
+        Date_t _expiration;
+
+        WorkHandle _fill;
+    };
+
+    /**
+     * This Listener wraps the interface in listen.h so that we may implement our own
+     * version of accepted().
+     */
+    class ListenerLegacy : public Listener {
+    public:
+        ListenerLegacy(const TransportLayerLegacy::Options& opts, NewConnectionCb callback);
+
+        void runListener();
+
+        void accepted(std::unique_ptr<AbstractMessagingPort> mp) override;
+
+        bool useUnixSockets() const override {
+            return true;
+        }
+
+    private:
+        NewConnectionCb _accepted;
+    };
+
+    /**
+     * Connection object, to associate Session ids with AbstractMessagingPorts.
+     */
+    struct Connection {
+        Connection(std::unique_ptr<AbstractMessagingPort> port, bool ended, Session::TagMask tags)
+            : amp(std::move(port)),
+              connectionId(amp->connectionId()),
+              tags(tags),
+              inUse(false),
+              ended(false) {}
+
+        std::unique_ptr<AbstractMessagingPort> amp;
+
+        const long long connectionId;
+
+        boost::optional<std::string> x509SubjectName;
+        Session::TagMask tags;
+        bool inUse;
+        bool ended;
+    };
+
+    std::shared_ptr<ServiceEntryPoint> _sep;
+
+    std::unique_ptr<Listener> _listener;
+    stdx::thread _listenerThread;
+
+    stdx::mutex _connectionsMutex;
+    std::unordered_map<Session::Id, Connection> _connections;
+
+    void _endSession_inlock(decltype(_connections.begin()) conn);
+
+    AtomicWord<bool> _running;
+
+    Options _options;
+};
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/transport/transport_layer_manager.cpp b/src/mongo/transport/transport_layer_manager.cpp
new file mode 100644
index 00000000000..c64023b2067
--- /dev/null
+++ b/src/mongo/transport/transport_layer_manager.cpp
@@ -0,0 +1,129 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/transport/transport_layer_manager.h"
+
+#include "mongo/base/status.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/transport/session.h"
+#include "mongo/util/time_support.h"
+#include <limits>
+
+#include <iostream>
+
+namespace mongo {
+namespace transport {
+
+TransportLayerManager::TransportLayerManager() = default;
+
+Ticket TransportLayerManager::sourceMessage(const Session& session,
+                                            Message* message,
+                                            Date_t expiration) {
+    return session.getTransportLayer()->sourceMessage(session, message, expiration);
+}
+
+Ticket TransportLayerManager::sinkMessage(const Session& session,
+                                          const Message& message,
+                                          Date_t expiration) {
+    return session.getTransportLayer()->sinkMessage(session, message, expiration);
+}
+
+Status TransportLayerManager::wait(Ticket&& ticket) {
+    return getTicketTransportLayer(ticket)->wait(std::move(ticket));
+}
+
+void TransportLayerManager::asyncWait(Ticket&& ticket, TicketCallback callback) {
+    return getTicketTransportLayer(ticket)->asyncWait(std::move(ticket), std::move(callback));
+}
+
+std::string TransportLayerManager::getX509SubjectName(const Session& session) {
+    return session.getX509SubjectName();
+}
+
+template <typename Callable>
+void TransportLayerManager::_foreach(Callable&& cb) {
+    {
+        stdx::lock_guard<stdx::mutex> lk(_tlsMutex);
+        for (auto&& tl : _tls) {
+            cb(tl.get());
+        }
+    }
+}
+
+TransportLayer::Stats TransportLayerManager::sessionStats() {
+    Stats stats;
+
+    _foreach([&](TransportLayer* tl) {
+        Stats s = tl->sessionStats();
+
+        stats.numOpenSessions += s.numOpenSessions;
+        stats.numCreatedSessions += s.numCreatedSessions;
+        if (std::numeric_limits<size_t>::max() - stats.numAvailableSessions <
+            s.numAvailableSessions) {
+            stats.numAvailableSessions = std::numeric_limits<size_t>::max();
+        } else {
+            stats.numAvailableSessions += s.numAvailableSessions;
+        }
+    });
+
+    return stats;
+}
+
+void TransportLayerManager::registerTags(const Session& session) {
+    session.getTransportLayer()->registerTags(session);
+}
+
+void TransportLayerManager::end(const Session& session) {
+    session.getTransportLayer()->end(session);
+}
+
+void TransportLayerManager::endAllSessions(Session::TagMask tags) {
+    _foreach([&tags](TransportLayer* tl) { tl->endAllSessions(tags); });
+}
+
+Status TransportLayerManager::start() {
+    return Status::OK();
+}
+
+void TransportLayerManager::shutdown() {
+    _foreach([](TransportLayer* tl) { tl->shutdown(); });
+}
+
+Status TransportLayerManager::addAndStartTransportLayer(std::unique_ptr<TransportLayer> tl) {
+    auto ptr = tl.get();
+    {
+        stdx::lock_guard<stdx::mutex> lk(_tlsMutex);
+        _tls.emplace_back(std::move(tl));
+    }
+    return ptr->start();
+}
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/transport/transport_layer_manager.h b/src/mongo/transport/transport_layer_manager.h
new file mode 100644
index 00000000000..391bdf4ebea
--- /dev/null
+++ b/src/mongo/transport/transport_layer_manager.h
@@ -0,0 +1,90 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include <vector>
+
+#include "mongo/base/status.h"
+#include "mongo/stdx/mutex.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/ticket.h"
+#include "mongo/transport/ticket_impl.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/util/net/message.h"
+#include "mongo/util/time_support.h"
+
+namespace mongo {
+namespace transport {
+
+/**
+ * This TransportLayerManager is a TransportLayer implementation that holds other
+ * TransportLayers. Mongod and Mongos can treat this like the "only" TransportLayer
+ * and not be concerned with which other TransportLayer implementations it holds
+ * underneath.
+ */
+class TransportLayerManager final : public TransportLayer {
+    MONGO_DISALLOW_COPYING(TransportLayerManager);
+
+public:
+    TransportLayerManager();
+
+    Ticket sourceMessage(const Session& session,
+                         Message* message,
+                         Date_t expiration = Ticket::kNoExpirationDate) override;
+    Ticket sinkMessage(const Session& session,
+                       const Message& message,
+                       Date_t expiration = Ticket::kNoExpirationDate) override;
+
+    Status wait(Ticket&& ticket) override;
+    void asyncWait(Ticket&& ticket, TicketCallback callback) override;
+
+    std::string getX509SubjectName(const Session& session) override;
+    void registerTags(const Session& session) override;
+
+    Stats sessionStats() override;
+
+    void end(const Session& session) override;
+    void endAllSessions(Session::TagMask tags = Session::kEmptyTagMask) override;
+
+    Status start() override;
+
+    void shutdown() override;
+
+    Status addAndStartTransportLayer(std::unique_ptr<TransportLayer> tl);
+
+private:
+    template <typename Callable>
+    void _foreach(Callable&& cb);
+
+    stdx::mutex _tlsMutex;
+    std::vector<std::unique_ptr<TransportLayer>> _tls;
+};
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/transport/transport_layer_mock.cpp b/src/mongo/transport/transport_layer_mock.cpp
new file mode 100644
index 00000000000..c550570b401
--- /dev/null
+++ b/src/mongo/transport/transport_layer_mock.cpp
@@ -0,0 +1,95 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include <memory>
+
+#include "mongo/base/status.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/ticket.h"
+#include "mongo/transport/ticket_impl.h"
+#include "mongo/transport/transport_layer_mock.h"
+#include "mongo/util/time_support.h"
+
+#include "mongo/stdx/memory.h"
+
+namespace mongo {
+namespace transport {
+
+Session::Id TransportLayerMock::MockTicket::sessionId() const {
+    return Session::Id{};
+}
+
+Date_t TransportLayerMock::MockTicket::expiration() const {
+    return Date_t::now();
+}
+
+Ticket TransportLayerMock::sourceMessage(const Session& session,
+                                         Message* message,
+                                         Date_t expiration) {
+    return Ticket(this, stdx::make_unique<MockTicket>());
+}
+
+Ticket TransportLayerMock::sinkMessage(const Session& session,
+                                       const Message& message,
+                                       Date_t expiration) {
+    return Ticket(this, stdx::make_unique<MockTicket>());
+}
+
+Status TransportLayerMock::wait(Ticket&& ticket) {
+    return Status::OK();
+}
+
+void TransportLayerMock::asyncWait(Ticket&& ticket, TicketCallback callback) {
+    callback(Status::OK());
+}
+
+std::string TransportLayerMock::getX509SubjectName(const Session& session) {
+    return session.getX509SubjectName();
+}
+
+TransportLayer::Stats TransportLayerMock::sessionStats() {
+    return Stats();
+}
+
+void TransportLayerMock::registerTags(const Session& session) {}
+
+void TransportLayerMock::end(const Session& session) {}
+
+void TransportLayerMock::endAllSessions(Session::TagMask tags) {}
+
+Status TransportLayerMock::start() {
+    return Status::OK();
+}
+
+void TransportLayerMock::shutdown() {}
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/transport/transport_layer_mock.h b/src/mongo/transport/transport_layer_mock.h
new file mode 100644
index 00000000000..400d6b02258
--- /dev/null
+++ b/src/mongo/transport/transport_layer_mock.h
@@ -0,0 +1,90 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include "mongo/base/status.h"
+#include "mongo/transport/session.h"
+#include "mongo/transport/ticket.h"
+#include "mongo/transport/ticket_impl.h"
+#include "mongo/transport/transport_layer.h"
+#include "mongo/util/net/message.h"
+#include "mongo/util/time_support.h"
+
+namespace mongo {
+namespace transport {
+
+/**
+ * This TransportLayerMock is a noop TransportLayer implementation.
+ */
+class TransportLayerMock : public TransportLayer {
+    MONGO_DISALLOW_COPYING(TransportLayerMock);
+
+public:
+    TransportLayerMock() = default;
+
+    Ticket sourceMessage(const Session& session,
+                         Message* message,
+                         Date_t expiration = Ticket::kNoExpirationDate) override;
+    Ticket sinkMessage(const Session& session,
+                       const Message& message,
+                       Date_t expiration = Ticket::kNoExpirationDate) override;
+
+    Status wait(Ticket&& ticket) override;
+    void asyncWait(Ticket&& ticket, TicketCallback callback) override;
+
+    std::string getX509SubjectName(const Session& session) override;
+    void registerTags(const Session& session) override;
+
+    Stats sessionStats() override;
+
+    void end(const Session& session) override;
+    void endAllSessions(Session::TagMask tags = Session::kEmptyTagMask) override;
+
+    Status start() override;
+    void shutdown() override;
+
+private:
+    /**
+     * A class for Tickets issued from the TransportLayerMock. These will
+     * route to the appropriate TransportLayer when run with wait() or
+     * asyncWait().
+     */
+    class MockTicket : public TicketImpl {
+        MONGO_DISALLOW_COPYING(MockTicket);
+
+    public:
+        MockTicket() = default;
+
+        Session::Id sessionId() const override;
+        Date_t expiration() const override;
+    };
+};
+
+}  // namespace transport
+}  // namespace mongo
diff --git a/src/mongo/util/exit.cpp b/src/mongo/util/exit.cpp
index 44cdf07f68b..7955a3483c6 100644
--- a/src/mongo/util/exit.cpp
+++ b/src/mongo/util/exit.cpp
@@ -32,6 +32,7 @@
 
 #include "mongo/util/exit.h"
 
+#include <boost/optional.hpp>
 #include <stack>
 
 #include "mongo/stdx/condition_variable.h"
@@ -47,6 +48,7 @@ namespace {
 
 stdx::mutex shutdownMutex;
 stdx::condition_variable shutdownTasksComplete;
+boost::optional<ExitCode> shutdownExitCode;
 bool shutdownTasksInProgress = false;
 AtomicUInt32 shutdownFlag;
 std::stack<stdx::function<void()>> shutdownTasks;
@@ -69,6 +71,10 @@ MONGO_COMPILER_NORETURN void logAndQuickExit(ExitCode code) {
     quickExit(code);
 }
 
+void setShutdownFlag() {
+    shutdownFlag.fetchAndAdd(1);
+}
+
 }  // namespace
 
 bool inShutdown() {
@@ -79,6 +85,13 @@ bool inShutdownStrict() {
     return shutdownFlag.load() != 0;
 }
 
+ExitCode waitForShutdown() {
+    stdx::unique_lock<stdx::mutex> lk(shutdownMutex);
+    shutdownTasksComplete.wait(lk, [] { return static_cast<bool>(shutdownExitCode); });
+
+    return shutdownExitCode.get();
+}
+
 void registerShutdownTask(stdx::function<void()> task) {
     stdx::lock_guard<stdx::mutex> lock(shutdownMutex);
     invariant(!inShutdown());
@@ -105,7 +118,7 @@ void shutdown(ExitCode code) {
             logAndQuickExit(code);
         }
 
-        shutdownFlag.fetchAndAdd(1);
+        setShutdownFlag();
         shutdownTasksInProgress = true;
         shutdownTasksThreadId = stdx::this_thread::get_id();
 
@@ -117,6 +130,7 @@ void shutdown(ExitCode code) {
     {
         stdx::lock_guard<stdx::mutex> lock(shutdownMutex);
         shutdownTasksInProgress = false;
+        shutdownExitCode.emplace(code);
     }
 
     shutdownTasksComplete.notify_all();
@@ -133,7 +147,7 @@ void shutdownNoTerminate() {
         if (inShutdown())
             return;
 
-        shutdownFlag.fetchAndAdd(1);
+        setShutdownFlag();
         shutdownTasksInProgress = true;
         shutdownTasksThreadId = stdx::this_thread::get_id();
 
diff --git a/src/mongo/util/exit.h b/src/mongo/util/exit.h
index fa6a8bbb85b..1aed81117a8 100644
--- a/src/mongo/util/exit.h
+++ b/src/mongo/util/exit.h
@@ -46,6 +46,11 @@ bool inShutdown();
 bool inShutdownStrict();
 
 /**
+ * Does not return until all shutdown tasks have run.
+ */
+ExitCode waitForShutdown();
+
+/**
  * Registers a new shutdown task to be called when shutdown or
  * shutdownNoTerminate is called. If this function is invoked after
  * shutdown or shutdownNoTerminate has been called, std::terminate is
diff --git a/src/mongo/util/net/SConscript b/src/mongo/util/net/SConscript
index fa8899561e9..34105547ab6 100644
--- a/src/mongo/util/net/SConscript
+++ b/src/mongo/util/net/SConscript
@@ -95,17 +95,6 @@ env.Library(
 )
 
 env.Library(
-    target="message_server_port",
-    source=[
-        "message_server_port.cpp",
-    ],
-    LIBDEPS=[
-        'network',
-        '$BUILD_DIR/mongo/db/stats/counters',
-    ],
-)
-
-env.Library(
     target='miniwebserver',
     source=[
         'miniwebserver.cpp',
diff --git a/src/mongo/util/net/abstract_message_port.h b/src/mongo/util/net/abstract_message_port.h
index c4ff5503f41..e4884e75133 100644
--- a/src/mongo/util/net/abstract_message_port.h
+++ b/src/mongo/util/net/abstract_message_port.h
@@ -32,6 +32,7 @@
 
 #include "mongo/config.h"
 #include "mongo/logger/log_severity.h"
+#include "mongo/stdx/functional.h"
 #include "mongo/util/net/message.h"
 #include "mongo/util/net/sockaddr.h"
 #include "mongo/util/time_support.h"
@@ -88,6 +89,11 @@ public:
     virtual void say(Message& toSend, int responseTo = 0) = 0;
 
     /**
+     * Sends the message (does not set headers).
+     */
+    virtual void say(const Message& toSend) = 0;
+
+    /**
      * Sends the data over the socket.
      */
     virtual void send(const char* data, int len, const char* context) = 0;
diff --git a/src/mongo/util/net/asio_message_port.cpp b/src/mongo/util/net/asio_message_port.cpp
index 88e37feb2d8..16fab7ea8c1 100644
--- a/src/mongo/util/net/asio_message_port.cpp
+++ b/src/mongo/util/net/asio_message_port.cpp
@@ -54,40 +54,6 @@ const char kGET[] = "GET";
 const int kHeaderLen = sizeof(MSGHEADER::Value);
 const int kInitialMessageSize = 1024;
 
-class ASIOPorts {
-public:
-    void closeAll(AbstractMessagingPort::Tag skipMask) {
-        stdx::lock_guard<stdx::mutex> lock_guard(mutex);
-        for (auto&& port : _portSet) {
-            if (port->getTag() & skipMask) {
-                LOG(3) << "Skip closing connection # " << port->connectionId();
-                continue;
-            }
-            LOG(3) << "Closing connection # " << port->connectionId();
-            port->shutdown();
-        }
-    }
-
-    void insert(ASIOMessagingPort* p) {
-        stdx::lock_guard<stdx::mutex> lock_guard(mutex);
-        _portSet.insert(p);
-    }
-
-    void erase(ASIOMessagingPort* p) {
-        stdx::lock_guard<stdx::mutex> lock_guard(mutex);
-        _portSet.erase(p);
-    }
-
-private:
-    stdx::mutex mutex;
-    std::unordered_set<ASIOMessagingPort*> _portSet;
-};
-
-// We "new" this so it will still be around when other automatic global vars are being destructed
-// during termination. TODO: This is an artifact from MessagingPort and should be removed when we
-// decide where to put networking global state.
-ASIOPorts& asioPorts = *(new ASIOPorts());
-
 #ifdef MONGO_CONFIG_SSL
 struct ASIOSSLContextPair {
     ASIOSSLContext server;
@@ -109,10 +75,6 @@ MONGO_INITIALIZER_WITH_PREREQUISITES(ASIOSSLContextSetup, ("SSLManager"))(Initia
 
 }  // namespace
 
-void ASIOMessagingPort::closeSockets(AbstractMessagingPort::Tag skipMask) {
-    asioPorts.closeAll(skipMask);
-}
-
 ASIOMessagingPort::ASIOMessagingPort(int fd, SockAddr farEnd)
     : _service(1),
       _timer(_service),
@@ -145,7 +107,6 @@ ASIOMessagingPort::ASIOMessagingPort(int fd, SockAddr farEnd)
                                            farEnd.getType() == AF_UNIX ? 0 : IPPROTO_TCP),
             fd) {
 #endif  // MONGO_CONFIG_SSL
-    asioPorts.insert(this);
     _getSocket().non_blocking(true);
     _remote = HostAndPort(farEnd.getAddr(), farEnd.getPort());
     _timer.expires_at(decltype(_timer)::time_point::max());
@@ -177,7 +138,6 @@ ASIOMessagingPort::ASIOMessagingPort(Milliseconds timeout, logger::LogSeverity l
 #else
       _sock(_service) {
 #endif  // MONGO_CONFIG_SSL
-    asioPorts.insert(this);
     if (*_timeout == Milliseconds(0)) {
         _timeout = boost::none;
     }
@@ -187,7 +147,6 @@ ASIOMessagingPort::ASIOMessagingPort(Milliseconds timeout, logger::LogSeverity l
 
 ASIOMessagingPort::~ASIOMessagingPort() {
     shutdown();
-    asioPorts.erase(this);
 }
 
 void ASIOMessagingPort::setTimeout(Milliseconds millis) {
@@ -502,6 +461,11 @@ void ASIOMessagingPort::say(Message& toSend, int responseTo) {
     invariant(!toSend.empty());
     toSend.header().setId(nextMessageId());
     toSend.header().setResponseToMsgId(responseTo);
+    return say(const_cast<const Message&>(toSend));
+}
+
+void ASIOMessagingPort::say(const Message& toSend) {
+    invariant(!toSend.empty());
     auto buf = toSend.buf();
     if (buf) {
         send(buf, MsgData::ConstView(buf).getLen(), nullptr);
diff --git a/src/mongo/util/net/asio_message_port.h b/src/mongo/util/net/asio_message_port.h
index d474b03ef29..d3e5fc33b1c 100644
--- a/src/mongo/util/net/asio_message_port.h
+++ b/src/mongo/util/net/asio_message_port.h
@@ -78,6 +78,7 @@ public:
     void reply(Message& received, Message& response) override;
 
     void say(Message& toSend, int responseTo = 0) override;
+    void say(const Message& toSend) override;
 
     void send(const char* data, int len, const char*) override;
     void send(const std::vector<std::pair<char*, int>>& data, const char*) override;
@@ -118,8 +119,6 @@ public:
 
     bool secure(SSLManagerInterface* ssl, const std::string& remoteHost) override;
 
-    static void closeSockets(AbstractMessagingPort::Tag skipMask = kSkipAllMask);
-
 private:
     void _setTimerCallback();
     asio::error_code _read(char* buf, std::size_t size);
diff --git a/src/mongo/util/net/listen.cpp b/src/mongo/util/net/listen.cpp
index deb30887e5c..65cec09e45a 100644
--- a/src/mongo/util/net/listen.cpp
+++ b/src/mongo/util/net/listen.cpp
@@ -280,7 +280,9 @@ void Listener::initAndListen() {
     }
 
     struct timeval maxSelectTime;
-    while (!inShutdown()) {
+    // The check against _finished allows us to actually stop the listener by signalling it through
+    // the _finished flag.
+    while (!inShutdown() && !_finished.load()) {
         fd_set fds[1];
         FD_ZERO(fds);
 
@@ -601,7 +603,7 @@ void Listener::_accepted(const std::shared_ptr<Socket>& psocket, long long conne
         port = stdx::make_unique<MessagingPort>(psocket);
     }
     port->setConnectionId(connectionId);
-    accepted(port.release());
+    accepted(std::move(port));
 }
 
 // ----- ListeningSockets -------
@@ -648,9 +650,8 @@ void Listener::checkTicketNumbers() {
     globalTicketHolder.resize(want);
 }
 
-void Listener::closeMessagingPorts(AbstractMessagingPort::Tag skipMask) {
-    ASIOMessagingPort::closeSockets(skipMask);
-    MessagingPort::closeSockets(skipMask);
+void Listener::shutdown() {
+    _finished.store(true);
 }
 
 TicketHolder Listener::globalTicketHolder(DEFAULT_MAX_CONN);
diff --git a/src/mongo/util/net/listen.h b/src/mongo/util/net/listen.h
index 0bbd3ad8a0b..f0c66a41d60 100644
--- a/src/mongo/util/net/listen.h
+++ b/src/mongo/util/net/listen.h
@@ -66,7 +66,7 @@ public:
     void initAndListen();  // never returns unless error (start a thread)
 
     /* spawn a thread, etc., then return */
-    virtual void accepted(AbstractMessagingPort* mp) = 0;
+    virtual void accepted(std::unique_ptr<AbstractMessagingPort> mp) = 0;
 
     const int _port;
 
@@ -83,6 +83,8 @@ public:
      */
     void waitUntilListening() const;
 
+    void shutdown();
+
 private:
     std::vector<SockAddr> _mine;
     std::vector<SOCKET> _socks;
@@ -94,6 +96,7 @@ private:
     mutable stdx::condition_variable _readyCondition;  // Used to wait for changes to _ready
     // Boolean that indicates whether this Listener is ready to accept incoming network requests
     bool _ready;
+    AtomicBool _finished{false};
 
     ServiceContext* _ctx;
     bool _setAsServiceCtxDecoration;
@@ -119,13 +122,6 @@ public:
 
     /** makes sure user input is sane */
     static void checkTicketNumbers();
-
-    /**
-     * This will close implementations of AbstractMessagingPort, skipping any that have tags
-     * matching `skipMask`.
-     */
-    static void closeMessagingPorts(
-        AbstractMessagingPort::Tag skipMask = AbstractMessagingPort::kSkipAllMask);
 };
 
 class ListeningSockets {
diff --git a/src/mongo/util/net/message.h b/src/mongo/util/net/message.h
index 37a6c0bb66b..601d02b9e12 100644
--- a/src/mongo/util/net/message.h
+++ b/src/mongo/util/net/message.h
@@ -458,11 +458,16 @@ public:
         return _buf.get();
     }
 
+    const char* buf() const {
+        return _buf.get();
+    }
+
     std::string toString() const;
 
     SharedBuffer sharedBuffer() {
         return _buf;
     }
+
     ConstSharedBuffer sharedBuffer() const {
         return _buf;
     }
diff --git a/src/mongo/util/net/message_port.cpp b/src/mongo/util/net/message_port.cpp
index 5051962ed5d..137d494c4ac 100644
--- a/src/mongo/util/net/message_port.cpp
+++ b/src/mongo/util/net/message_port.cpp
@@ -63,41 +63,6 @@ using std::string;
 
 /* messagingport -------------------------------------------------------------- */
 
-class Ports {
-    std::set<MessagingPort*> ports;
-    stdx::mutex m;
-
-public:
-    Ports() : ports() {}
-    void closeAll(AbstractMessagingPort::Tag skip_mask) {
-        stdx::lock_guard<stdx::mutex> bl(m);
-        for (std::set<MessagingPort*>::iterator i = ports.begin(); i != ports.end(); i++) {
-            if ((*i)->getTag() & skip_mask) {
-                LOG(3) << "Skip closing connection # " << (*i)->connectionId();
-                continue;
-            }
-            LOG(3) << "Closing connection # " << (*i)->connectionId();
-            (*i)->shutdown();
-        }
-    }
-    void insert(MessagingPort* p) {
-        stdx::lock_guard<stdx::mutex> bl(m);
-        ports.insert(p);
-    }
-    void erase(MessagingPort* p) {
-        stdx::lock_guard<stdx::mutex> bl(m);
-        ports.erase(p);
-    }
-};
-
-// we "new" this so it is still be around when other automatic global vars
-// are being destructed during termination.
-Ports& ports = *(new Ports());
-
-void MessagingPort::closeSockets(AbstractMessagingPort::Tag skipMask) {
-    ports.closeAll(skipMask);
-}
-
 MessagingPort::MessagingPort(int fd, const SockAddr& remote)
     : MessagingPort(std::make_shared<Socket>(fd, remote)) {}
 
@@ -108,7 +73,6 @@ MessagingPort::MessagingPort(std::shared_ptr<Socket> sock)
     : _x509SubjectName(), _connectionId(), _tag(), _psock(std::move(sock)) {
     SockAddr sa = _psock->remoteAddr();
     _remoteParsed = HostAndPort(sa.getAddr(), sa.getPort());
-    ports.insert(this);
 }
 
 void MessagingPort::setTimeout(Milliseconds millis) {
@@ -122,7 +86,6 @@ void MessagingPort::shutdown() {
 
 MessagingPort::~MessagingPort() {
     shutdown();
-    ports.erase(this);
 }
 
 bool MessagingPort::recv(Message& m) {
@@ -216,6 +179,7 @@ bool MessagingPort::call(Message& toSend, Message& response) {
     say(toSend);
     bool success = recv(response);
     if (success) {
+        invariant(!response.empty());
         if (response.header().getResponseToMsgId() != toSend.header().getId()) {
             response.reset();
             uasserted(40134, "Response ID did not match the sent message ID.");
@@ -225,9 +189,15 @@ bool MessagingPort::call(Message& toSend, Message& response) {
 }
 
 void MessagingPort::say(Message& toSend, int responseTo) {
-    verify(!toSend.empty());
+    invariant(!toSend.empty());
     toSend.header().setId(nextMessageId());
     toSend.header().setResponseToMsgId(responseTo);
+
+    return say(const_cast<const Message&>(toSend));
+}
+
+void MessagingPort::say(const Message& toSend) {
+    invariant(!toSend.empty());
     auto buf = toSend.buf();
     if (buf) {
         send(buf, MsgData::ConstView(buf).getLen(), "say");
diff --git a/src/mongo/util/net/message_port.h b/src/mongo/util/net/message_port.h
index 18931e20750..41ecc2fce3b 100644
--- a/src/mongo/util/net/message_port.h
+++ b/src/mongo/util/net/message_port.h
@@ -66,6 +66,7 @@ public:
     bool call(Message& toSend, Message& response) override;
 
     void say(Message& toSend, int responseTo = 0) override;
+    void say(const Message& toSend) override;
 
     unsigned remotePort() const override {
         return _psock->remotePort();
@@ -142,11 +143,6 @@ private:
     long long _connectionId;
     AbstractMessagingPort::Tag _tag;
     std::shared_ptr<Socket> _psock;
-
-
-public:
-    static void closeSockets(AbstractMessagingPort::Tag skipMask = kSkipAllMask);
 };
 
-
 }  // namespace mongo
diff --git a/src/mongo/util/net/message_port_mock.cpp b/src/mongo/util/net/message_port_mock.cpp
index 6869723129f..de07ba032ae 100644
--- a/src/mongo/util/net/message_port_mock.cpp
+++ b/src/mongo/util/net/message_port_mock.cpp
@@ -55,6 +55,7 @@ void MessagingPortMock::reply(Message& received, Message& response, int32_t resp
 void MessagingPortMock::reply(Message& received, Message& response) {}
 
 void MessagingPortMock::say(Message& toSend, int responseTo) {}
+void MessagingPortMock::say(const Message& toSend) {}
 
 bool MessagingPortMock::connect(SockAddr& farEnd) {
     return true;
diff --git a/src/mongo/util/net/message_port_mock.h b/src/mongo/util/net/message_port_mock.h
index 61ab5be8958..721998f81f3 100644
--- a/src/mongo/util/net/message_port_mock.h
+++ b/src/mongo/util/net/message_port_mock.h
@@ -55,6 +55,7 @@ public:
     void reply(Message& received, Message& response) override;
 
     void say(Message& toSend, int responseTo = 0) override;
+    void say(const Message& toSend) override;
 
     bool connect(SockAddr& farEnd) override;
 
diff --git a/src/mongo/util/net/message_server.h b/src/mongo/util/net/message_server.h
deleted file mode 100644
index 544c859c6f7..00000000000
--- a/src/mongo/util/net/message_server.h
+++ /dev/null
@@ -1,83 +0,0 @@
-// message_server.h
-
-/*    Copyright 2009 10gen Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    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
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    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 GNU Affero General 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.
- */
-
-/*
-  abstract database server
-  async io core, worker thread system
- */
-
-#pragma once
-
-#include "mongo/platform/basic.h"
-
-namespace mongo {
-
-class ServiceContext;
-
-class MessageHandler {
-public:
-    virtual ~MessageHandler() {}
-
-    /**
-     * Called once when a socket is connected.
-     */
-    virtual void connected(AbstractMessagingPort* p) = 0;
-
-    /**
-     * Called every time a message comes in. Handler is responsible for responding to client.
-     */
-    virtual void process(Message& m, AbstractMessagingPort* p) = 0;
-
-    /**
-     * Called once, either when the client disconnects or when the process is shutting down. After
-     * close() is called, this handler's AbstractMessagingPort pointer (passed in via the
-     * connected() method) is no longer valid.
-     */
-    virtual void close() = 0;
-};
-
-class MessageServer {
-public:
-    struct Options {
-        int port;            // port to bind to
-        std::string ipList;  // addresses to bind to
-
-        Options() : port(0), ipList("") {}
-    };
-
-    virtual ~MessageServer() {}
-    virtual void run() = 0;
-    virtual bool setupSockets() = 0;
-};
-
-// TODO use a factory here to decide between port and asio variations
-MessageServer* createServer(const MessageServer::Options& opts,
-                            std::shared_ptr<MessageHandler> handler,
-                            ServiceContext* ctx);
-}
diff --git a/src/mongo/util/net/message_server_port.cpp b/src/mongo/util/net/message_server_port.cpp
deleted file mode 100644
index e8432f8c510..00000000000
--- a/src/mongo/util/net/message_server_port.cpp
+++ /dev/null
@@ -1,244 +0,0 @@
-// message_server_port.cpp
-
-/*    Copyright 2009 10gen Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    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
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kNetwork
-
-#include "mongo/platform/basic.h"
-
-#include <memory>
-#include <system_error>
-
-#include "mongo/base/disallow_copying.h"
-#include "mongo/config.h"
-#include "mongo/db/lasterror.h"
-#include "mongo/db/server_options.h"
-#include "mongo/db/stats/counters.h"
-#include "mongo/stdx/functional.h"
-#include "mongo/stdx/memory.h"
-#include "mongo/stdx/thread.h"
-#include "mongo/util/concurrency/thread_name.h"
-#include "mongo/util/concurrency/ticketholder.h"
-#include "mongo/util/debug_util.h"
-#include "mongo/util/exit.h"
-#include "mongo/util/log.h"
-#include "mongo/util/mongoutils/str.h"
-#include "mongo/util/net/abstract_message_port.h"
-#include "mongo/util/net/listen.h"
-#include "mongo/util/net/message.h"
-#include "mongo/util/net/message_server.h"
-#include "mongo/util/net/socket_exception.h"
-#include "mongo/util/net/ssl_manager.h"
-#include "mongo/util/net/thread_idle_callback.h"
-#include "mongo/util/quick_exit.h"
-#include "mongo/util/scopeguard.h"
-
-#ifdef __linux__  // TODO: consider making this ifndef _WIN32
-#include <sys/resource.h>
-#endif
-
-#if !defined(__has_feature)
-#define __has_feature(x) 0
-#endif
-
-namespace mongo {
-
-using std::unique_ptr;
-using std::endl;
-
-namespace {
-
-using MessagingPortWithHandler = std::pair<AbstractMessagingPort*, std::shared_ptr<MessageHandler>>;
-
-}  // namespace
-
-class PortMessageServer : public MessageServer, public Listener {
-public:
-    /**
-     * Creates a new message server.
-     *
-     * @param opts
-     * @param handler the handler to use.
-     */
-    PortMessageServer(const MessageServer::Options& opts,
-                      std::shared_ptr<MessageHandler> handler,
-                      ServiceContext* ctx)
-        : Listener("", opts.ipList, opts.port, ctx, true), _handler(std::move(handler)) {}
-
-    virtual void accepted(AbstractMessagingPort* mp) {
-        ScopeGuard sleepAfterClosingPort = MakeGuard(sleepmillis, 2);
-        auto portWithHandler = stdx::make_unique<MessagingPortWithHandler>(mp, _handler);
-
-        if (!Listener::globalTicketHolder.tryAcquire()) {
-            log() << "connection refused because too many open connections: "
-                  << Listener::globalTicketHolder.used() << endl;
-            return;
-        }
-
-        try {
-#ifndef __linux__  // TODO: consider making this ifdef _WIN32
-            {
-                stdx::thread thr(stdx::bind(&handleIncomingMsg, portWithHandler.get()));
-                thr.detach();
-            }
-#else
-            pthread_attr_t attrs;
-            pthread_attr_init(&attrs);
-            pthread_attr_setdetachstate(&attrs, PTHREAD_CREATE_DETACHED);
-
-            static const size_t STACK_SIZE =
-                1024 * 1024;  // if we change this we need to update the warning
-
-            struct rlimit limits;
-            verify(getrlimit(RLIMIT_STACK, &limits) == 0);
-            if (limits.rlim_cur > STACK_SIZE) {
-                size_t stackSizeToSet = STACK_SIZE;
-#if !__has_feature(address_sanitizer)
-                if (kDebugBuild)
-                    stackSizeToSet /= 2;
-#endif
-                pthread_attr_setstacksize(&attrs, stackSizeToSet);
-            } else if (limits.rlim_cur < 1024 * 1024) {
-                warning() << "Stack size set to " << (limits.rlim_cur / 1024)
-                          << "KB. We suggest 1MB" << endl;
-            }
-
-
-            pthread_t thread;
-            int failed = pthread_create(&thread, &attrs, &handleIncomingMsg, portWithHandler.get());
-
-            pthread_attr_destroy(&attrs);
-
-            if (failed) {
-                log() << "pthread_create failed: " << errnoWithDescription(failed) << endl;
-                throw std::system_error(
-                    std::make_error_code(std::errc::resource_unavailable_try_again));
-            }
-#endif  // __linux__
-
-            portWithHandler.release();
-            sleepAfterClosingPort.Dismiss();
-        } catch (...) {
-            Listener::globalTicketHolder.release();
-            log() << "failed to create thread after accepting new connection, closing connection";
-        }
-    }
-
-    virtual bool setupSockets() {
-        return Listener::setupSockets();
-    }
-
-    void run() {
-        initAndListen();
-    }
-
-    virtual bool useUnixSockets() const {
-        return true;
-    }
-
-private:
-    const std::shared_ptr<MessageHandler> _handler;
-
-    /**
-     * Handles incoming messages from a given socket.
-     *
-     * Terminating conditions:
-     * 1. Assertions while handling the request.
-     * 2. Socket is closed.
-     * 3. Server is shutting down (based on inShutdown)
-     *
-     * @param arg this method is in charge of cleaning up the arg object.
-     *
-     * @return NULL
-     */
-    static void* handleIncomingMsg(void* arg) {
-        TicketHolderReleaser connTicketReleaser(&Listener::globalTicketHolder);
-
-        invariant(arg);
-        unique_ptr<MessagingPortWithHandler> portWithHandler(
-            static_cast<MessagingPortWithHandler*>(arg));
-        auto mp = portWithHandler->first;
-        auto handler = portWithHandler->second;
-
-        setThreadName(std::string(str::stream() << "conn" << mp->connectionId()));
-        mp->setLogLevel(logger::LogSeverity::Debug(1));
-
-        Message m;
-        int64_t counter = 0;
-        try {
-            handler->connected(mp);
-            ON_BLOCK_EXIT([handler]() { handler->close(); });
-
-            while (!inShutdown()) {
-                m.reset();
-                mp->clearCounters();
-
-                if (!mp->recv(m)) {
-                    if (!serverGlobalParams.quiet) {
-                        int conns = Listener::globalTicketHolder.used() - 1;
-                        const char* word = (conns == 1 ? " connection" : " connections");
-                        log() << "end connection " << mp->remote().toString() << " (" << conns
-                              << word << " now open)" << endl;
-                    }
-                    break;
-                }
-
-                handler->process(m, mp);
-                networkCounter.hit(mp->getBytesIn(), mp->getBytesOut());
-
-                // Occasionally we want to see if we're using too much memory.
-                if ((counter++ & 0xf) == 0) {
-                    markThreadIdle();
-                }
-            }
-        } catch (AssertionException& e) {
-            log() << "AssertionException handling request, closing client connection: " << e
-                  << endl;
-        } catch (SocketException& e) {
-            log() << "SocketException handling request, closing client connection: " << e << endl;
-        } catch (const DBException&
-                     e) {  // must be right above std::exception to avoid catching subclasses
-            log() << "DBException handling request, closing client connection: " << e << endl;
-        } catch (std::exception& e) {
-            error() << "Uncaught std::exception: " << e.what() << ", terminating" << endl;
-            quickExit(EXIT_UNCAUGHT);
-        }
-        mp->shutdown();
-
-        return NULL;
-    }
-};
-
-
-MessageServer* createServer(const MessageServer::Options& opts,
-                            std::shared_ptr<MessageHandler> handler,
-                            ServiceContext* ctx) {
-    return new PortMessageServer(opts, std::move(handler), ctx);
-}
-
-}  // namespace mongo
diff --git a/src/mongo/util/net/miniwebserver.cpp b/src/mongo/util/net/miniwebserver.cpp
index ec64f174560..78f81f86134 100644
--- a/src/mongo/util/net/miniwebserver.cpp
+++ b/src/mongo/util/net/miniwebserver.cpp
@@ -204,7 +204,7 @@ void MiniWebServer::_accepted(const std::shared_ptr<Socket>& psock, long long co
     }
 }
 
-void MiniWebServer::accepted(AbstractMessagingPort* mp) {
+void MiniWebServer::accepted(std::unique_ptr<AbstractMessagingPort> mp) {
     MONGO_UNREACHABLE;
 }
 
diff --git a/src/mongo/util/net/miniwebserver.h b/src/mongo/util/net/miniwebserver.h
index a2d0690225e..122281555fc 100644
--- a/src/mongo/util/net/miniwebserver.h
+++ b/src/mongo/util/net/miniwebserver.h
@@ -69,7 +69,7 @@ public:
     }
 
     // This is not currently used for the MiniWebServer. See SERVER-24200
-    void accepted(AbstractMessagingPort* mp) override;
+    void accepted(std::unique_ptr<AbstractMessagingPort> mp) override;
 
 private:
     void _accepted(const std::shared_ptr<Socket>& psocket, long long connectionId) override;
diff --git a/src/mongo/util/net/sockaddr.cpp b/src/mongo/util/net/sockaddr.cpp
index 00960f2fd97..5236a76892f 100644
--- a/src/mongo/util/net/sockaddr.cpp
+++ b/src/mongo/util/net/sockaddr.cpp
@@ -46,8 +46,8 @@
 #endif
 #endif
 
+#include "mongo/bson/util/builder.h"
 #include "mongo/util/log.h"
-#include "mongo/util/mongoutils/str.h"
 #include "mongo/util/net/sock.h"
 
 namespace mongo {
@@ -150,10 +150,19 @@ bool SockAddr::isLocalHost() const {
 }
 
 std::string SockAddr::toString(bool includePort) const {
-    std::string out = getAddr();
-    if (includePort && getType() != AF_UNIX && getType() != AF_UNSPEC)
-        out += mongoutils::str::stream() << ':' << getPort();
-    return out;
+    if (includePort && (getType() != AF_UNIX) && (getType() != AF_UNSPEC)) {
+        StringBuilder ss;
+
+        if (getType() == AF_INET6) {
+            ss << '[' << getAddr() << "]:" << getPort();
+        } else {
+            ss << getAddr() << ':' << getPort();
+        }
+
+        return ss.str();
+    } else {
+        return getAddr();
+    }
 }
 
 sa_family_t SockAddr::getType() const {
diff --git a/src/mongo/util/ntservice_test.cpp b/src/mongo/util/ntservice_test.cpp
index 0a0fa6bde64..ad5b7e7ed2a 100644
--- a/src/mongo/util/ntservice_test.cpp
+++ b/src/mongo/util/ntservice_test.cpp
@@ -34,7 +34,6 @@
 #include <string>
 #include <vector>
 
-
 #include "mongo/db/client.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/ntservice.h"
diff --git a/src/mongo/util/tcmalloc_server_status_section.cpp b/src/mongo/util/tcmalloc_server_status_section.cpp
index 7a8eb49e25b..303076ba784 100644
--- a/src/mongo/util/tcmalloc_server_status_section.cpp
+++ b/src/mongo/util/tcmalloc_server_status_section.cpp
@@ -38,6 +38,8 @@
 
 #include "mongo/base/init.h"
 #include "mongo/db/commands/server_status.h"
+#include "mongo/db/service_context.h"
+#include "mongo/transport/transport_layer.h"
 #include "mongo/util/log.h"
 #include "mongo/util/net/listen.h"
 #include "mongo/util/net/thread_idle_callback.h"
@@ -58,7 +60,8 @@ stdx::mutex tcmallocCleanupLock;
  *  favorable times. Ideally would do some milder cleanup or scavenge...
  */
 void threadStateChange() {
-    if (Listener::globalTicketHolder.used() <= kManyClients)
+    if (getGlobalServiceContext()->getTransportLayer()->sessionStats().numOpenSessions <=
+        kManyClients)
         return;
 
 #if MONGO_HAVE_GPERFTOOLS_GET_THREAD_CACHE_SIZE