本文为《MongoDB内核源码实现、性能调优、最佳运维实践系列》模块一：MongoDB网络传输层模块源码 第二篇，您可点击此处查看第一篇：MongoDB网络传输模块源码实现及性能调优实践-体验内核性能极致设计。

• transport_layer套接字处理及传输层管理子模块
• session会话子模块
• Ticket数据收发子模块
• service_entry_point服务入口点子模块
• service_state_machine状态机子模块(该《模块在网络传输层模块源码实现三》中分析)
• service_executor线程模型子模块(该《模块在网络传输层模块源码实现四》中分析)

transport_layer套接字处理及传输层管理子模块功能包括套接字相关初始化处理、结合asio库实现异步accept处理、不同线程模型管理及初始化等，该模块的源码实现主要由以下几个文件实现：

上图是套接字处理及传输层管理子模块源码实现的相关文件，其中mock和test文件主要用于模拟测试等，所以真正核心的代码实现只有下表的几个文件，对应源码文件功能说明如下表所示：

该子模块核心代码主要由TransportLayerManager类和TransportLayerASIO类相关接口实现。

TransportLayerManager类主要成员及接口如下：

//网络会话链接，消息处理管理相关的类，在createWithConfig构造该类存入_tls  
class TransportLayerManager final : public TransportLayer {  
    //以下四个接口真正实现在TransportLayerASIO类中具体实现  
    Ticket sourceMessage(...) override;  
    Ticket sinkMessage(...) override;      
    Status wait(Ticket&& ticket) override;  
    void asyncWait(...) override;  
    //配置初始化实现  
    std::unique_ptr<TransportLayer> createWithConfig(...);  
  
    //createWithConfig中赋值,对应TransportLayerASIO,  
    //实际上容器中就一个成员,就是TransportLayerASIO  
    std::vector<std::unique_ptr<TransportLayer>> _tls;  
};

TransportLayerManager类包含一个_tls成员，该类最核心的createWithConfig接口代码实现如下：

createWithConfig函数根据配置文件来确定对应的TransportLayer，如果net.transportLayer配置为”asio”，则选用TransportLayerASIO类来进行底层的网络IO处理，如果配置为”legacy”，则选用TransportLayerLegacy。”legacy”模式当前已淘汰，本文只分析”asio”模式实现。

“asio”模式包含两种线程模型：adaptive(动态线程模型)和synchronous(同步线程模型)。adaptive模式线程设计采用动态线程方式，线程数和MongoDB压力直接相关，如果MongoDB压力大，则线程数增加；如果MongoDB压力变小，则线程数自动减少。同步线程模式也就是一个链接一个线程模型，线程数的多少和链接数的多少成正比，链接数越多则线程数也越大。

MongoDB内核实现中通过opts.transportMode来标记asio的线程模型，这两种模型对应标记如下：

说明：adaptive线程模型被标记为KAsynchronous，synchronous被标记为KSynchronous是有原因的，adaptive动态线程模型网络IO处理借助epoll异步实现，而synchronous一个链接一个线程模型网络IO处理是同步读写操作。MongoDB网络线程模型具体实现及各种优缺点可以参考：MongoDB网络传输处理源码实现及性能调优-体验内核性能极致设计

TransportLayerASIO类核心成员及接口如下：

class TransportLayerASIO final : public TransportLayer {  
    //以下四个接口主要和套接字数据读写相关  
    Ticket sourceMessage(...);  
    Ticket sinkMessage(...);  
    Status wait(Ticket&& ticket);  
    void asyncWait(Ticket&& ticket, TicketCallback callback);  
    void end(const SessionHandle& session);  
    //新链接处理  
    void _acceptConnection(GenericAcceptor& acceptor);  
      
    //adaptive线程模型网络IO上下文处理  
    std::shared_ptr<asio::io_context> _workerIOContext;   
    //accept接收客户端链接对应的IO上下文  
    std::unique_ptr<asio::io_context> _acceptorIOContext;    
    //bindIp配置中的ip地址列表，用于bind监听，accept客户端请求  
    std::vector<std::pair<SockAddr, GenericAcceptor>> _acceptors;  
    //listener线程负责接收客户端新链接  
    stdx::thread _listenerThread;  
    //服务类型，也就是本实例是mongos还是mongod  
    //mongos对应ServiceEntryPointMongod,mongod对应ServiceEntryPointMongos  
    ServiceEntryPoint* const _sep = nullptr;  
    //当前运行状态  
    AtomicWord<bool> _running{false};  
    //listener处理相关的配置信息  
    Options _listenerOptions;  
}

从上面的类结构可以看出，该类主要通过listenerThread线程完成bind绑定及listen监听操作，同时部分接口实现新连接上的数据读写。

套接字初始化代码实现如下：

Status TransportLayerASIO::setup() {  
    std::vector<std::string> listenAddrs;  
    //如果没有配置bindIp,则默认监听"127.0.0.1：27017"
    if (_listenerOptions.ipList.empty()) {  
        listenAddrs = {"127.0.0.1"};  
    } else {  
        //配置文件中的bindIp:1.1.1.1,2.2.2.2，以逗号分隔符获取ip列表存入ipList  
        boost::split(listenAddrs, _listenerOptions.ipList, boost::is_any_of(","), boost::token_compress_on);  
    }  
    //遍历ip地址列表  
    for (auto& ip : listenAddrs) {  
        //根据IP和端口构造对应SockAddr结构  
        const auto addrs = SockAddr::createAll(  
            ip, _listenerOptions.port, _listenerOptions.enableIPv6 ? AF_UNSPEC : AF_INET);  
        ......  
        //根据addr构造endpoint  
        asio::generic::stream_protocol::endpoint endpoint(addr.raw(), addr.addressSize);  
        //_acceptorIOContext和_acceptors关联  
        GenericAcceptor acceptor(*_acceptorIOContext);  
        //epoll注册，也就是fd和epoll关联  
        //basic_socket_acceptor::open  
        acceptor.open(endpoint.protocol());   
         //SO_REUSEADDR配置 basic_socket_acceptor::set_option  
        acceptor.set_option(GenericAcceptor::reuse_address(true));  
        //非阻塞设置 basic_socket_acceptor::non_blocking  
        acceptor.non_blocking(true, ec);    
        //bind绑定    
        acceptor.bind(endpoint, ec);   
        if (ec) {  
            return errorCodeToStatus(ec);  
        }  
    }  
}

从上面的分析可以看出，代码实现首先解析出配置文件中bindIP中的ip:port列表，然后遍历列表绑定所有服务端需要监听的ip:port，每个ip:port对应一个GenericAcceptor ，所有acceptor和全局accept IO上下文_acceptorIOContext关联，同时bind()绑定所有ip:port。

Bind()绑定所有配置文件中的Ip:port后，然后通过TransportLayerASIO::start()完成后续处理，该接口代码实现如下：

//_initAndListen中调用执行   
Status TransportLayerASIO::start() { //listen线程处理  
    ......  
    //这里专门起一个线程做listen相关的accept事件处理  
    _listenerThread = stdx::thread([this] {  
        //修改线程名  
        setThreadName("listener");   
        //该函数中循环处理accept事件  
        while (_running.load()) {  
            asio::io_context::work work(*_acceptorIOContext);   
            try {  
                //accept事件调度处理  
                 _acceptorIOContext->run();    
            } catch (...) { //异常处理  
                severe() << "Uncaught exception in the listener: " << exceptionToStatus();  
                fassertFailed(40491);  
            }  
        }  
    });   
   遍历_acceptors，进行listen监听处理  
   for (auto& acceptor : _acceptors) {   
        acceptor.second.listen(serverGlobalParams.listenBacklog);  
        //异步accept回调注册在该函数中  
        _acceptConnection(acceptor.second);       
    }  
}

从上面的TransportLayerASIO::start()接口可以看出，MongoDB特地创建了一个listener线程用于客户端accept事件处理，然后借助ASIO网络库的_acceptorIOContext->run()接口来调度，当有新链接到来的时候，就会执行相应的accept回调处理，accept回调注册到io_context的流程由acceptConnection()完成，该接口核心源码实现如下：

//accept新连接到来的回调注册 
void TransportLayerASIO::_acceptConnection(GenericAcceptor& acceptor) {  
      //新链接到来时候的回调函数，服务端接收到新连接都会执行该回调
    //注意这里面是递归执行，保证所有accept事件都会一次处理完毕
    auto acceptCb = [this, &acceptor](const std::error_code& ec, GenericSocket peerSocket) mutable {  
        if (!_running.load())  
            return;  
  
        ......  
        //每个新的链接都会new一个新的ASIOSession  
        std::shared_ptr<ASIOSession> session(new ASIOSession(this, std::move(peerSocket)));  
        //新的链接处理ServiceEntryPointImpl::startSession，  
        //和ServiceEntryPointImpl服务入口点模块关联起来  
        _sep->startSession(std::move(session));  
        //递归，直到处理完所有的网络accept事件  
        _acceptConnection(acceptor);   
    };  
    //accept新连接到来后服务端的回调处理在这里注册  
    acceptor.async_accept(*_workerIOContext, std::move(acceptCb));  
}

TransportLayerASIO::_acceptConnection的新连接处理过程借助ASIO库实现，通过acceptor.async_accept实现所有监听的acceptor回调异步注册。

当服务端接收到客户端新连接事件通知后，会触发执行acceptCb()回调，该回调中底层ASIO库通过epoll_wait获取到所有的accept事件，每获取到一个accept事件就代表一个新的客户端链接，然后调用ServiceEntryPointImpl::startSession()接口处理这个新的链接事件，整个过程递归执行，保证一次可以处理所有的客户端accept请求信息。

每个链接都会构造一个唯一的session信息，该session就代表一个唯一的新连接，链接和session一一对应。此外，最终会调用ServiceEntryPointImpl::startSession()进行真正的accept()处理，从而获取到一个新的链接。

注意：TransportLayerASIO::_acceptConnection()中实现了TransportLayerASIO类和ServiceEntryPointImpl类的关联，这两个类在该接口实现了关联。

此外，从前面的TransportLayerASIO类结构中可以看出，该类还包含如下四个接口：sourceMessage(…)、sinkMessage(…)、wait(Ticket&& ticket)、asyncWait(Ticket&&ticket, TicketCallback callback)，这四个接口入参都和Ticket数据分发子模块相关联，具体核心代码实现如下：

//根据asioSession, expiration, message三个信息构造数据接收类ASIOSourceTicket  
Ticket TransportLayerASIO::sourceMessage(...) {  
    ......  
    auto asioSession = checked_pointer_cast<ASIOSession>(session);  
    //根据asioSession, expiration, message三个信息构造ASIOSourceTicket  
    auto ticket = stdx::make_unique<ASIOSourceTicket>(asioSession, expiration, message);  
    return {this, std::move(ticket)};  
}  
  
//根据asioSession, expiration, message三个信息构造数据发送类ASIOSinkTicket  
Ticket TransportLayerASIO::sinkMessage(...) {  
    auto asioSession = checked_pointer_cast<ASIOSession>(session);  
    auto ticket = stdx::make_unique<ASIOSinkTicket>(asioSession, expiration, message);  
    return {this, std::move(ticket)};  
}  
  
//同步接收或者发送，最终调用ASIOSourceTicket::fill 或者 ASIOSinkTicket::fill  
Status TransportLayerASIO::wait(Ticket&& ticket) {  
    //获取对应Ticket，接收对应ASIOSourceTicket，发送对应ASIOSinkTicket  
    auto ownedASIOTicket = getOwnedTicketImpl(std::move(ticket));  
    auto asioTicket = checked_cast<ASIOTicket*>(ownedASIOTicket.get());  
    ......  
    //调用对应fill接口 同步接收ASIOSourceTicket::fill 或者 同步发送ASIOSinkTicket::fill  
    asioTicket->fill(true, [&waitStatus](Status result) { waitStatus = result; });  
    return waitStatus;  
}  
//异步接收或者发送，最终调用ASIOSourceTicket::fill 或者 ASIOSinkTicket::fill  
void TransportLayerASIO::asyncWait(Ticket&& ticket, TicketCallback callback) {  
    //获取对应数据收发的Ticket，接收对应ASIOSourceTicket，发送对应ASIOSinkTicket  
    auto ownedASIOTicket = std::shared_ptr<TicketImpl>(getOwnedTicketImpl(std::move(ticket)));  
    auto asioTicket = checked_cast<ASIOTicket*>(ownedASIOTicket.get());  
  
   //调用对应ASIOTicket::fill  
    asioTicket->fill(  
        false,   [ callback = std::move(callback),  
        ownedASIOTicket = std::move(ownedASIOTicket) ](Status status) { callback(status); });  
}

上面四个接口中的前两个接口主要通过Session, expiration, message这三个参数来获取对应的Ticket 信息，实际上MongoDB内核实现中把接收数据的Ticket和发送数据的Ticket分别用不同的继承类ASIOSourceTicket和ASIOSinkTicket来区分，三个参数的作用如下表所示：

数据收发包括同步收发和异步收发，同步收发通过TransportLayerASIO::wait()实现，异步收发通过TransportLayerASIO::asyncWait()实现。

注意：以上四个接口把TransportLayerASIO类和Ticket 数据收发类的关联。

transport_layer套接字处理及传输层管理子模块主要由transport_layer_manager和transport_layer_asio两个核心类组成，这两个类的核心接口功能总结如下表所示：

Transport_layer_manager中初始化TransportLayer和serviceExecutor，net.TransportLayer配置可以为legacy和asio，其中legacy已经淘汰，当前内核只支持asio模式。asio配置对应的TransportLayer由TransportLayerASIO实现，对应的serviceExecutor线程模型可以是adaptive动态线程模型，也可以是synchronous同步线程模型。

套接字创建、bind()绑定、listen()监听、accept事件注册等都由本类实现，同时数据分发Ticket模块也与本模块关联，一起配合完成整个后续Ticket模块模块的同步及异步数据读写流程。此外，本模块还通过ServiceEntryPoint服务入口子模块联动，保证了套接字初始化、accept事件注册完成后，服务入口子模块能有序的进行新连接接收处理。

接下来继续分析本模块相关联的ServiceEntryPoint服务入口子模块和Ticket数据分发子模块实现。

service_entry_point服务入口点子模块主要负责如下功能：新连接处理、Session会话管理、接收到一个完整报文后的回调处理(含报文解析、认证、引擎层处理等)。

该模块的源码实现主要包含以下几个文件：

service_entry_point开头的代码文件都和本模块相关，其中service_entry_point_utils*负责工作线程创建，service_entry_point_impl*完成新链接回调处理及sesseion会话管理。

服务入口子模块相关代码实现比较简洁，主要由ServiceEntryPointImpl类和service_entry_point_utils中的线程创建函数组成。

ServiceEntryPointImpl类主要成员和接口如下：

class ServiceEntryPointImpl : public ServiceEntryPoint {  
    MONGO_DISALLOW_COPYING(ServiceEntryPointImpl);  
public:  
    //构造函数  
    explicit ServiceEntryPointImpl(ServiceContext* svcCtx);     
    //以下三个接口进行session会话处理控制  
    void startSession(transport::SessionHandle session) final;  
    void endAllSessions(transport::Session::TagMask tags) final;  
    bool shutdown(Milliseconds timeout) final;  
    //session会话统计  
    Stats sessionStats() const final;  
    ......  
private:  
    //该list结构管理所有的ServiceStateMachine信息  
    using SSMList = stdx::list<std::shared_ptr<ServiceStateMachine>>;  
    //SSMList对应的迭代器  
    using SSMListIterator = SSMList::iterator;  
    //赋值ServiceEntryPointImpl::ServiceEntryPointImpl  
    //对应ServiceContextMongoD(mongod)或者ServiceContextNoop(mongos)类  
    ServiceContext* const _svcCtx;   
    //该成员变量在代码中没有使用  
    AtomicWord<std::size_t> _nWorkers;  
    //锁  
    mutable stdx::mutex _sessionsMutex;  
    //一个新链接对应一个ssm保存到ServiceEntryPointImpl._sessions中  
    SSMList _sessions;  
    //最大链接数控制  
    size_t _maxNumConnections{DEFAULT_MAX_CONN};  
    //当前的总链接数，不包括关闭的链接  
    AtomicWord<size_t> _currentConnections{0};  
    //所有的链接，包括已经关闭的链接  
    AtomicWord<size_t> _createdConnections{0};  
};

该类的几个接口主要是session相关控制处理，该类中的变量成员说明如下：

ServiceEntryPointImpl类最核心的startSession()接口负责每个新连接到来后的内部回调处理，具体实现如下：

该接口拿到该链接对应的服务端和客户端地址后，记录到该链接对应session中，然后根据该session、transportMode、_svcCtx构建一个服务状态机ssm(ServiceStateMachine)。一个新链接对应一个唯一session，一个session对应一个唯一的服务状态机ssm，这三者保持唯一的一对一关系。

最终，startSession()让服务入口子模块、session会话子模块、ssm状态机子模块关联起来。

service_entry_point_utils源码文件只有launchServiceWorkerThread一个接口，该接口主要负责工作线程创建，并设置每个工作线程的线程栈大小，如果系统默认栈大于1M，则每个工作线程的线程栈大小设置为1M，如果系统栈大小小于1M，则以系统堆栈大小为准，同时warning打印提示。该函数实现如下：

service_entry_point服务入口点子模块主要负责新连接后的回调处理及工作线程创建，该模块和后续的session会话模块、SSM服务状态机模块衔接配合，完成数据收发的正常逻辑转换处理。上面的分析只列出了服务入口点子模块的核心接口实现，下表总结该模块所有的接口功能：

Ticket数据收发模块相关实现主要由ASIOTicket类完成，该类结构如下：

该类保护多个成员变量，这些成员变量功能说明如下：

MongoDB在具体实现上，数据接收和数据发送分开实现，分别是数据接收类ASIOSourceTicket和数据发送类ASIOSinkTicket，这两个类都继承自ASIOTicket类，这两个类的主要结构如下：

//数据接收的ticket  
class TransportLayerASIO::ASIOSourceTicket : public TransportLayerASIO::ASIOTicket {  
public:  
    //初始化构造  
    ASIOSourceTicket(const ASIOSessionHandle& session, Date_t expiration, Message* msg);  
protected:  
    //数据接收Impl  
    void fillImpl() final;  
private:  
    //接收到MongoDB头部数据后的回调处理  
    void _headerCallback(const std::error_code& ec, size_t size);  
    //接收到MongoDB包体数据后的回调处理    
    void _bodyCallback(const std::error_code& ec, size_t size);    
    //存储数据的buffer，网络IO读取到的原始数据内容  
    SharedBuffer _buffer;  
    //数据Message管理，数据来源为_buffer  
    Message* _target;  
};  
//数据发送的ticket  
class TransportLayerASIO::ASIOSinkTicket : public TransportLayerASIO::ASIOTicket {  
 public:  
   //初始化构造  
   ASIOSinkTicket(const ASIOSessionHandle& session, Date_t expiration, const Message& msg);
  	protected:  
	    //数据发送Impl  
	    void fillImpl() final;  
	private:  
    //发送数据完成的回调处理  
   void _sinkCallback(const std::error_code& ec, size_t size);  
    //需要发送的数据message信息  
    Message _msgToSend;  
};

从上面的代码实现可以看出，ASIOSinkTicket 和ASIOSourceTicket 类接口及成员实现几乎意义，只是具体的实现方法不同，下面对ASIOSourceTicket和ASIOSinkTicket 相关核心代码实现进行分析。

数据接收过程核心代码如下：

//数据接收的fillImpl接口实现  
void TransportLayerASIO::ASIOSourceTicket::fillImpl() {    
    //获取对应session信息  
    auto session = getSession();  
    if (!session)  
        return;  
    //收到读取MongoDB头部数据，头部数据长度是固定的kHeaderSize字节  
    const auto initBufSize = kHeaderSize;  
    _buffer = SharedBuffer::allocate(initBufSize);  
  
    //调用TransportLayerASIO::ASIOSession::read读取底层数据存入_buffer  
    //读完头部数据后执行对应的_headerCallback回调函数  
    session->read(isSync(),  
                  asio::buffer(_buffer.get(), initBufSize), //先读取头部字段出来  
                  [this](const std::error_code& ec, size_t size) { _headerCallback(ec, size); });  
}  
  
//读取到MongoDB header头部信息后的回调处理  
void TransportLayerASIO::ASIOSourceTicket::_headerCallback(const std::error_code& ec, size_t size) {  
    ......  
    //获取session信息  
    auto session = getSession();  
    if (!session)  
        return;  
    //从_buffer中获取头部信息  
    MSGHEADER::View headerView(_buffer.get());  
    //获取message长度  
    auto msgLen = static_cast<size_t>(headerView.getMessageLength());  
    //长度太小或者太大，直接报错  
    if (msgLen < kHeaderSize || msgLen > MaxMessageSizeBytes) {  
        .......  
        return;  
    }  
    ....  
   //内容还不够一个mongo协议报文，继续读取body长度字节的数据，读取完毕后开始body处理  
   //注意这里是realloc，保证头部和body在同一个buffer中  
    _buffer.realloc(msgLen);   
    MsgData::View msgView(_buffer.get());  
  
    //调用底层TransportLayerASIO::ASIOSession::read读取数据body   
    session->read(isSync(),  
      //数据读取到该buffer                  
      asio::buffer(msgView.data(), msgView.dataLen()),  
      //读取成功后的回调处理  
      [this](const std::error_code& ec, size_t size) { _bodyCallback(ec, size); });  
}  
  
//_headerCallback对header读取后解析header头部获取到对应的msg长度，然后开始body部分处理  
void TransportLayerASIO::ASIOSourceTicket::_bodyCallback(const std::error_code& ec, size_t size) {  
    ......  
    //buffer转存到_target中  
    _target->setData(std::move(_buffer));  
    //流量统计  
    networkCounter.hitPhysicalIn(_target->size());  
    //TransportLayerASIO::ASIOTicket::finishFill    
    finishFill(Status::OK()); //包体内容读完后，开始下一阶段的处理    
    //报文读取完后的下一阶段就是报文内容处理，开始走ServiceStateMachine::_processMessage  
}

MongoDB协议由msg header + msg body组成，一个完整的MongoDB报文内容格式如下：

上图所示各个字段及body内容部分功能说明如下表：

ASIOSourceTicket类的几个核心接口都是围绕这一原则展开，整个MongoDB数据接收流程如下：

1. 读取MongoDB头部header数据，解析出header中的messageLength字段。

2. 检查messageLength字段是否在指定的合理范围，该字段不能小于Header整个头部大小，也不能超过MaxMessageSizeBytes最大长度。

3. Header len检查通过，说明读取header数据完成，于是执行_headerCallback回调。

4. realloc更多的空间来存储body内容。

5. 继续读取body len长度数据，读取body完成后，执行_bodyCallback回调处理。

ASIOSinkTicket发送类相比接收类，没有数据解析相关的流程，因此实现过程会更加简单，具体源码实现如下：

//发送数据成功后的回调处理  
void TransportLayerASIO::ASIOSinkTicket::_sinkCallback(const std::error_code& ec, size_t size) {  
    //发送的网络字节数统计  
    networkCounter.hitPhysicalOut(_msgToSend.size());   
    //执行SSM中对应的状态流程  
    finishFill(ec ? errorCodeToStatus(ec) : Status::OK());  
}  
  
//发送数据的fillImpl  
void TransportLayerASIO::ASIOSinkTicket::fillImpl() {  
    //获取对应session  
    auto session = getSession();  
    if (!session)  
        return;  
  
    //调用底层TransportLayerASIO::ASIOSession::write发送数据，发送成功后执行_sinkCallback回调   
    session->write(isSync(),  
       asio::buffer(_msgToSend.buf(), _msgToSend.size()),  
       //发送数据成功后的callback回调  
       [this](const std::error_code& ec, size_t size) { _sinkCallback(ec, size); });  
}

从上面的分析可以看出，Ticket数据收发模块主要调用session会话模块来进行底层数据的读写、解析，当读取或者发送一个完整的MongoDB报文后最终交由SSM服务状态机模块调度处理。

ticket模块主要接口功能总结如下表所示：

前面的分析也可以看出，Ticket数据收发模块会调用session处理模块来进行真正的数据读写，同时接收或者发送完一个完整MongoDB报文后的后续回调处理讲交由SSM服务状态机模块处理。

class TransportLayerASIO::ASIOSession : public Session {  
    //初始化构造  
    ASIOSession(TransportLayerASIO* tl, GenericSocket socket);  
    //获取本session使用的tl  
    TransportLayer* getTransportLayer();  
    //以下四个接口套接字相关，本端/对端地址获取，获取fd,关闭fd等  
    const HostAndPort& remote();  
    const HostAndPort& local();  
    GenericSocket& getSocket();  
    void shutdown();  
  
    //以下四个接口调用asio网络库实现数据的同步收发和异步收发  
    void read(...)  
    void write(...)  
    void opportunisticRead(...)  
    void opportunisticWrite(...)  
  
    //远端地址信息  
    HostAndPort _remote;  
    //本段地址信息  
    HostAndPort _local;  
    //赋值见TransportLayerASIO::_acceptConnection  
    //也就是fd，一个新连接对应一个_socket  
    GenericSocket _socket;  
    //SSL相关不做分析，  
#ifdef MONGO_CONFIG_SSL  
    boost::optional<asio::ssl::stream<decltype(_socket)>> _sslSocket;  
    bool _ranHandshake = false;  
#endif  
  
    //本套接字对应的tl，赋值建TransportLayerASIO::_acceptConnection(...)  
    TransportLayerASIO* const _tl;  
}

该类最核心的三个接口ASIOSession(…)、opportunisticRead(…)、opportunisticWrite(..)分别完成套接字处理、调用asio库接口实现底层数据读和底层数据写。这三个核心接口源码实现如下：

该类初始化的时候完成新连接_socket相关的初始化设置，包括阻塞读写还是非阻塞读写。如果是同步线程模型(一个链接一个线程)，则读写方式位阻塞读写；如果是异步线程模型(adaptive动态线程模型)，则调用asio网络库接口实现异步读写。

此外，该链接_socket对应的客户端ip:port和服务端ip:port也在该初始化类中获取，最终保存到本session的_remote和_local成员中。

数据读取核心代码实现如下:

//读取指定长度数据，然后执行handler回调  
void opportunisticRead(...) {  
    std::error_code ec;  
    //如果是异步线程模型，在ASIOSession构造初始化的时候会设置non_blocking非阻塞模式  
    //异步线程模型这里实际上是非阻塞读取，如果是同步线程模型，则没有non_blocking设置，也就是阻塞读取  
    auto size = asio::read(stream, buffers, ec);    
  
    //如果是异步读，并且read返回would_block或者try_again说明指定长度的数据还没有读取完毕  
    if ((ec == asio::error::would_block || ec == asio::error::try_again) && !sync) {  
        //buffers有大小size，实际读最多读size字节  
        MutableBufferSequence asyncBuffers(buffers);  
        if (size > 0) {  
            asyncBuffers += size; //buffer offset向后移动  
        }  
  
        //继续异步方式读取数据，读取到指定长度数据后执行handler回调处理  
        asio::async_read(stream, asyncBuffers, std::forward<CompleteHandler>(handler));  
    } else {   
        //阻塞方式读取read返回后可以保证读取到了size字节长度的数据  
        //直接read获取到size字节数据，则直接执行handler   
        handler(ec, size);  
    }  
}

opportunisticRead首先调用asio::read(stream, buffers, ec)读取buffers对应size长度的数据，buffers空间大小就是需要读取的数据size大小。如果是同步线程模型，则这里为阻塞式读取，直到读到size字节才会返回；如果是异步线程模型，这这里是非阻塞读取，非阻塞读取当内核网络协议栈数据读取完毕后，如果还没有读到size字节，则继续进行async_read异步读取。

当buffers指定的size字节全部读取完整后，不管是同步模式还是异步模式，最终都会执行handler回调，开始后续的数据解析及处理流程。

发送数据核心代码实现如下:

//发送数据  
void opportunisticWrite(...) {  
    std::error_code ec;  
    //如果是同步模式，则阻塞写，直到全部写成功。异步模式则非阻塞写  
    auto size = asio::write(stream, buffers, ec);   
  
    //异步写如果返回try_again说明数据还没有发送完，则继续异步写发送  
    if ((ec == asio::error::would_block || ec == asio::error::try_again) && !sync) {  
        ConstBufferSequence asyncBuffers(buffers);  
        if (size > 0) {  //buffer中数据指针偏移计数
            asyncBuffers += size;  
        }  
        //异步写发送完成，执行handler回调  
        asio::async_write(stream, asyncBuffers, std::forward<CompleteHandler>(handler));  
    } else {  
        //同步写成功，则直接执行handler回调处理  
        handler(ec, size);  
    }  
}

不管是同步模式还是异步模式发送数据，最终数据发送成功后，都会调用handler()回调来执行后续的流程。

从上面的代码分析可以看出，session会话模块最终直接和asio网络库交互实现数据的读写操作。该模块核心接口功能总结如下表：

《MongoDB网络传输处理源码实现及性能调优-体验内核性能极致设计》一文对MongoDB网络传输模块中的ASIO网络库实现、service_executor服务运行子模块(即线程模型子模块)进行了详细的分析，加上本文的transport_layer套接字处理及传输层管理子模块、session会话子模块、Ticket数据收发子模块、service_entry_point服务入口点子模块。

transport_layer套接字处理及传输层管理子模块主要由transport_layer_manager和transport_layer_asio两个核心类组成。分别完成net相关的配置文件初始化操作，套接字初始化及处理，最终transport_layer_asio的相应接口实现了和ticket数据分发子模块、服务入口点子模块的关联。

服务入口子模块主要由ServiceEntryPointImpl类和service_entry_point_utils中的线程创建函数组成，实现新连接accept处理及控制。该模块通过startSession()让服务入口子模块、session会话子模块、ssm状态机子模块关联起来。

ticket数据收发子模块主要功能如下：调用session子模块进行底层asio库处理、拆分数据接收和数据发送到两个类、完整MongoDB报文读取 、接收或者发送MongoDB报文后的回调处理, 回调处理由SSM服务状态机模块管理，当读取或者发送一个完整的MongoDB报文后最终交由SSM服务状态机模块调度处理。

Session会话模块功能主要如下：负责记录HostAndPort、和底层asio库直接互动，实现数据的同步或者异步收发。一个新连接fd对应一个唯一的session，对fd的操作直接映射为session操作。

到这里，整个MongoDB网络传输层模块分析只差service_state_machine状态机调度子模块，状态机调度子模块相比本文分析的几个子模块更加复杂，因此将在下期《MongoDB网络传输层模块源码分析三》中单独分析。

本文所有源码注释分析详见如下链接：MongoDB网络传输模块详细源码分析

更多文章：

OPPO百万级高并发MongoDB集群性能数十倍提升优化实践

作者：杨亚洲前滴滴出行技术专家，现任OPPO文档数据库MongoDB负责人，负责oppo千万级峰值TPS/十万亿级数据量文档数据库MongoDB内核研发及运维工作，一直专注于分布式缓存、高性能服务端、数据库、中间件等相关研发。Github账号地址:https://github.com/y123456yz

 

 

 

 

 

MongoDB中文手册翻译正在进行中，欢迎更多朋友在自己的空闲时间学习并进行文档翻译，您的翻译将由社区专家进行审阅，并拥有署名权更新到中文用户手册和发布到社区微信内容平台。

更多问题可以添加社区助理小芒果微信（mongoingcom）咨询，进入社区微信交流群请备注“mongo”。