mongodb 源码实现系列 - 网络传输层模块实现三

{"type":"doc","content":[{"type":"heading","attrs":{"align":"center","level":1},"content":[{"type":"text","text":"transport_layer网络传输层模块源码实现三"}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":"关于作者"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 前滴滴出行技术专家，现任OPPO文档数据库mongodb负责人，负责oppo千万级峰值TPS/十万亿级数据量文档数据库mongodb研发和运维工作，一直专注于分布式缓存、高性能服务端、数据库、中间件等相关研发。Github账号地址:"},{"type":"link","attrs":{"href":"https://github.com/y123456yz","title":null},"content":[{"type":"text","marks":[{"type":"underline"}],"text":"https://github.com/y123456yz"}]}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":"1.说明"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"   在之前的"},{"type":"link","attrs":{"href":"https://xie.infoq.cn/article/d4460b807c9835c6d80707410","title":null},"content":[{"type":"text","marks":[{"type":"underline"}],"text":"Mongodb网络模块源码实现及性能调优(一)"}]},{"type":"text","text":"和"},{"type":"link","attrs":{"href":"https://xie.infoq.cn/article/1cac5adcd1b4f3fe512a1457a","title":null},"content":[{"type":"text","marks":[{"type":"underline"}],"text":"<>"}]},{"type":"text","text":"一文中分析了如何阅读百万级大工程源码、Asio网络库实现、线程模型、transport_layer套接字处理及传输层管理子模块、session会话子模块、Ticket数据收发子模块、service_entry_point服务入口点子模块。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 本文将继续分析网络传输层模块中service_state_machine状态机调度子模块内核源码实现。"}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":"2. service_state_machine状态机调度子模块"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" service_state_machine状态机处理模块主要复杂一次完整请求的状态转换，确保请求可以按照指定的流程顺利进行，最终实现客户端的正常mongodb访问。该模块核心代码实现主要由以下三个源码文件实现（test为测试相关，可以忽略）："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/ca/cafbb6a9c23c1f755c26eccae7469504.png","alt":null,"title":null,"style":null,"href":null,"fromPaste":true,"pastePass":true}},{"type":"heading","attrs":{"align":null,"level":2},"content":[{"type":"text","text":"2.1核心代码实现"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 在service_entry_point服务入口点子模块分析中，当接收到一个新的链接后，在ServiceEntryPointImpl::startSession(...)回调函数中会构造一个ServiceStateMachine  ssm类，从而实现了新链接、session、ssm的一一映射关系。其中，ServiceStateMachine 类实现对ThreadGuard(线程守护)有较多的依赖，因此本文从这两个类核心代码实现来分析整个状态机调度模块的内部设计细节。"}]},{"type":"heading","attrs":{"align":null,"level":3},"content":[{"type":"text","text":"2.1.1 ThreadGuard线程守护类核心代码实现"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" ThreadGuard也就是”线程守护”类，该类主要用于工作线程名的管理维护、ssm归属管理、该ssm对应session链接的回收处理等。该类核心成员及接口实现如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.class ServiceStateMachine::ThreadGuard {  \n2.    ......  \n3.public:  \n4.    // create a ThreadGuard which will take ownership of the SSM in this thread.  \n5.    //ThreadGuard初始化，标记ssm所有权属于本线程  \n6.    explicit ThreadGuard(ServiceStateMachine* ssm) : _ssm{ssm} {  \n7.        //获取ssm状态机所有权  \n8.        auto owned = _ssm->_owned.compareAndSwap(Ownership::kUnowned, Ownership::kOwned);  \n9.        //如果是一个链接一个线程模型，则条件满足  \n10.        if (owned == Ownership::kStatic) {   \n11.        //一个链接一个线程模型,由于链接始终由同一个线程处理，因此该链接对应的ssm始终归属于同一个线程  \n12.            dassert(haveClient());  \n13.            dassert(Client::getCurrent() == _ssm->_dbClientPtr);  \n14.            //标识归属权已确定  \n15.            _haveTakenOwnership = true;  \n16.            return;  \n17.        }  \n18.   ......\n19.     //adaptive 动态线程模式走下面的模式  \n20.  \n21.        //把当前线程的线程名零时保存起来  \n22.        auto oldThreadName = getThreadName();   \n23.     //ssm保存的线程名和当前线程名不同  \n24.        if (oldThreadName != _ssm->_threadName) {  \n25.         //即将修改线程名了，把修改前的线程名保存到_oldThreadName  \n26.            _ssm->_oldThreadName = getThreadName().toString();  \n27.           //把运行本ssm状态机的线程名改为conn-x线程  \n28.        setThreadName(_ssm->_threadName); //把当前线程改名为_threadName  \n29.        }  \n30.  \n31.        //设置本线程对应client信息,一个链接对应一个client,标识本client当前归属于本线程处理  \n32.        Client::setCurrent(std::move(_ssm->_dbClient));  \n33.     //本状态机ssm所有权有了，归属于运行本ssm的线程  \n34.        _haveTakenOwnership = true;  \n35.    }  \n36.    ......  \n37.    //重新赋值  \n38.    ThreadGuard& operator=(ThreadGuard&& other) {  \n39.        if (this != &other) {  \n40.            _ssm = other._ssm;  \n41.            _haveTakenOwnership = other._haveTakenOwnership;  \n42.         //原来的other所有权失效  \n43.            other._haveTakenOwnership = false;  \n44.        }  \n45.    //返回  \n46.        return *this;  \n47.    };  \n48.  \n49.    //析构函数  \n50.    ~ThreadGuard() {  \n51.    //ssm所有权已确定，则析构的时候，调用release处理，恢复线程原有线程名  \n52.        if (_haveTakenOwnership)  \n53.            release();  \n54.    }  \n55.  \n56.    //一个链接一个线程模型，标记_owned为kStatic，也就是线程名永远不会改变  \n57.    void markStaticOwnership() {  \n58.        dassert(static_cast(*this));  \n59.        _ssm->_owned.store(Ownership::kStatic);  \n60.    }  \n61.  \n62.    //恢复原有线程名，同时把client信息从调度线程归还给状态机  \n63.    void release() {  \n64.         auto owned = _ssm->_owned.load();  \n65.     //adaptive异步线程池模式满足if条件，表示SSM固定归属于某个线程  \n66.        if (owned != Ownership::kStatic) {  \n67.        //本线程拥有currentClient信息，于是把它归还给SSM状态机  \n68.            if (haveClient()) {  \n69.                _ssm->_dbClient = Client::releaseCurrent();  \n70.            }  \n71.         //恢复到以前的线程名  \n72.            if (!_ssm->_oldThreadName.empty()) {  \n73.         //恢复到老线程名  \n74.                setThreadName(_ssm->_oldThreadName);   \n75.            }  \n76.        }  \n77.        //状态机状态进入end，则调用对应回收hook处理  \n78.        if (_ssm->state() == State::Ended) {  \n79.            //链接关闭的回收处理 ServiceStateMachine::setCleanupHook  \n80.            auto cleanupHook = std::move(_ssm->_cleanupHook);  \n81.            if (cleanupHook)  \n82.                cleanupHook();  \n83.            return;  \n84.        }  \n85.  \n86.     //归属权失效  \n87.        _haveTakenOwnership = false;  \n88.        //归属状态变为未知  \n89.        if (owned == Ownership::kOwned) {  \n90.            _ssm->_owned.store(Ownership::kUnowned);  \n91.        }  \n92.    }  \n93.  \n94.private:  \n95.    //本线程守护当前对应的ssm  \n96.    ServiceStateMachine* _ssm;  \n97.    //默认false，标识该状态机ssm不归属于任何线程  \n98.    bool _haveTakenOwnership = false;  \n99.}  "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 从上面的代码分析可以看出线程守护类作用比较明确，就是守护当前线程的归属状态，并记录状态机ssm不同状态变化前后的线程名。此外，状态机ssm对应的session链接如果进入end状态，则该链接的资源回收释放也由该类完成。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 查看mongod或者mongos实例，如果启动实例的时候配置了”serviceExecutor: adaptive”会发现这些进程下面有很多线程名为”conn-x”和”worker-x”线程，同时同一个线程线程名可能发生改变，这个过程就是由ThreadGuard线程守护类来实现。synchronous一个链接一个线程模型只有”conn-x”线程，adaptive线程模型将同时存在有线程名为”conn-x”和”worker-x”的线程，具体原理讲在后面继续分析，如下图："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/25/25646a361dc83b3f64d23275231e9cb3.png","alt":null,"title":null,"style":null,"href":null,"fromPaste":true,"pastePass":true}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","marks":[{"type":"strong"}],"text":" 说明："},{"type":"text","text":"synchronous线程模式对应worker初始线程名为”conn-x”，adaptive线程模型对应worker初始线程名为”worker-x”。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" ThreadGuard线程守护类和状态机ssm(service_state_machine)关联紧密，客户端请求处理的内部ssm状态转换也和该类密切关联，请看后续分析。"}]},{"type":"heading","attrs":{"align":null,"level":3},"content":[{"type":"text","text":"2.1.2 ServiceStateMachine 类核心代码实现"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" service_state_machine状态机处理模块核心代码实现通过ServiceStateMachine类完成，该类的核心结构成员和函数接口如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.//一个新链接对应一个ServiceStateMachine保存到ServiceEntryPointImpl._sessions中  \n2.class ServiceStateMachine : public std::enable_shared_from_this {  \n3.   ......  \n4.public:  \n5.   ......  \n6.   static std::shared_ptr create(...);  \n7.   ServiceStateMachine(...);  \n8.   //状态机所属状态\n9.   enum class State {  \n10.       //ServiceStateMachine::ServiceStateMachine构造函数初始状态  \n11.       Created,        \n12.       //ServiceStateMachine::_runNextInGuard开始进入接收网络数据状态  \n13.       //标记本次客户端请求已完成(已经发送DB获取的数据给客户端)，等待调度进行该链接的  \n14.       //下一轮请求，该状态对应处理流程为_sourceMessage  \n15.       Source,         \n16.       //等待获取客户端的数据  \n17.       SourceWait,     \n18.       //接收到一个完整mongodb报文后进入该状态  \n19.       Process,        \n20.       //等待数据发送成功  \n21.       SinkWait,       \n22.       //接收或者发送数据异常、链接关闭，则进入该状态  _cleanupSession  \n23.       EndSession,     \n24.       //session回收处理进入该状态  \n25.       Ended           \n26.   };  \n27.   //所有权状态，主要用来判断是否需要在状态转换中跟新线程名，只对动态线程模型生效  \n28.   enum class Ownership {   \n29.       //该状态表示本状态机SSM处于非活跃状态  \n30.       kUnowned,    \n31.       //该状态标识本状态机SSM归属于某个工作worker线程，处于活跃调度运行状态  \n32.       kOwned,   \n33.       //表示SSM固定归属于某个线程  \n34.       kStatic   \n35.   };  \n36.     \n37.   ......  \n38.private:  \n39.   //ThreadGuard可以理解为线程守护，后面在ThreadGuard类中单独说明  \n40.   class ThreadGuard;  \n41.   friend class ThreadGuard;  \n42.     \n43.   ......  \n44.   //获取session信息  \n45.   const transport::SessionHandle& _session()  \n46.   //以下两个接口为任务task调度相关接口  \n47.   void _scheduleNextWithGuard(...);  \n48.   void _runNextInGuard(ThreadGuard guard);  \n49.   //接收到一个完整mongodb报文后的处理  \n50.   inline void _processMessage(ThreadGuard guard);  \n51.   //以下四个接口完成底层数据读写及其对应回调处理  \n52.   void _sourceCallback(Status status);  \n53.   void _sinkCallback(Status status);  \n54.   void _sourceMessage(ThreadGuard guard);  \n55.   void _sinkMessage(ThreadGuard guard, Message toSink);  \n56.     \n57.   //一次客户端请求，当前mongodb服务端所处的状态  \n58.   AtomicWord _state{State::Created};  \n59.   //服务入口，ServiceEntryPointMongod ServiceEntryPointMongos mongod及mongos入口点  \n60.   ServiceEntryPoint* _sep;  \n61.   //synchronous及adaptive模式，也就是线程模型是一个链接一个线程还是动态线程池  \n62.   transport::Mode _transportMode;  \n63.   //ServiceContextMongoD(mongod)或者ServiceContextNoop(mongos)服务上下文  \n64.   ServiceContext* const _serviceContext;  \n65.   //也就是本ssm对应的session信息，默认对应ASIOSession   \n66.   transport::SessionHandle _sessionHandle;   \n67.   //根据session构造对应client信息,ServiceStateMachine::ServiceStateMachine赋值  \n68.   //也就是本次请求对应的客户端信息  \n69.   ServiceContext::UniqueClient _dbClient;  \n70.   //指向上面的_dbClient  \n71.   const Client* _dbClientPtr;  \n72. //该SSM当前处理线程的线程名，因为adaptive线程模型一次请求中的不同状态会修改线程名\n73.   const std::string _threadName;  \n74.   //修改线程名之前的线程名称  \n75.   std::string _oldThreadName;  \n76.   //ServiceEntryPointImpl::startSession->ServiceStateMachine::setCleanupHook中设置赋值  \n77.   //session链接回收处理  \n78.   stdx::function _cleanupHook;  \n79.   //接收处理的message信息  一个完整的报文就记录在该msg中  \n80.   Message _inMessage;   \n81.   //默认初始化kUnowned,标识本SSM状态机处于非活跃状态，  \n82.   //主要用来判断是否需要在状态转换中跟新线程名，只对动态线程模型生效  \n83.   AtomicWord _owned{Ownership::kUnowned};  \n84.}"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 该类核心成员功能说明如下表："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/aa/aa7515fad797cf7a5b492591037682a5.png","alt":null,"title":null,"style":null,"href":null,"fromPaste":true,"pastePass":true}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 我们知道，链接、session、SSM状态机一一对应，他们也拥有对应的归属权，这里的归属权指的就是当前SSM归属于那个线程，也就是当前SSM状态机调度模块由那个线程实现。归属权通过Ownership类标记，该类保护如下几种状态，每种状态功能说明如下："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/c7/c7faf19544f54a3006dfc684ad9ff2bd.png","alt":null,"title":null,"style":null,"href":null,"fromPaste":true,"pastePass":true}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Mongodb服务端接收到客户端请求后的数据接收、协议解析、从db层获取数据、发送数据给客户端都是通过SSM状态机进行有序的状态转换处理，SSM调度处理过程中保护多个状态，每个状态对应一个状态码，具体状态码及其功能说明如下表所示："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/19/191bf0a1e1a516a07b13108d5103a6a3.png","alt":null,"title":null,"style":null,"href":null,"fromPaste":true,"pastePass":true}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 以上是SSM处理请求过程的状态码信息，状态转换的具体实现过程请参考后面的核心代码分析。listerner线程接收到新的客户端链接后会调用通过service"},{"type":"text","marks":[{"type":"italic"}],"text":"entry"},{"type":"text","text":"point服务入口点子模块的ssm->start()接口进入SSM状态机调度模块，该接口相关的源码实现如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.//ServiceEntryPointImpl::startSession中执行  启动SSM状态机  \n2.void ServiceStateMachine::start(Ownership ownershipModel) {  \n3.    //直接调用_scheduleNextWithGuard接口  \n4.    _scheduleNextWithGuard(   \n5.    //listener线程暂时性的变为conn线程名，在_scheduleNextWithGuard中任  \n6.    //务入队完成后，在下面的_scheduleNextWithGuard调用guard.release()恢复listener线程名  \n7.        ThreadGuard(this), transport::ServiceExecutor::kEmptyFlags, ownershipModel);  \n8.}  \n9.  \n10.void ServiceStateMachine::_scheduleNextWithGuard(...) {  \n11.    //该任务func实际上由worker线程运行,worker线程从asio库的全局队列获取任务调度执行  \n12.    auto func = [ ssm = shared_from_this(), ownershipModel ] {  \n13.     //构造ThreadGuard  \n14.        ThreadGuard guard(ssm.get());    \n15.     //说明是sync mode,即一个链接一个线程模式, 归属权明确，属于指定线程  \n16.     if (ownershipModel == Ownership::kStatic)   \n17.         guard.markStaticOwnership();  \n18.     //对应:ServiceStateMachine::_runNextInGuard，在这里面完成状态调度转换  \n19.        ssm->_runNextInGuard(std::move(guard));    \n20.    };  \n21.    //恢复之前的线程名，如果该SSM进入Ended状态，则开始资源回收处理  \n22.    guard.release();  \n23.    //ServiceExecutorAdaptive::schedule(adaptive)   ServiceExecutorSynchronous::schedule(synchronous)  \n24.    //第一次进入该函数的时候在这里面创建新线程，不是第一次则把task任务入队调度  \n25.    Status status = _serviceContext->getServiceExecutor()->schedule(std::move(func), flags);  \n26.    if (status.isOK()) {  \n27.        return;  \n28.    }  \n29.    //异常处理  \n30.    ......  \n31.}  "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"     ServiceStateMachine::start()接口调用ServiceStateMachine::_scheduleNextWithGuard()来启动状态机运行。_scheduleNextWithGuard()接口最核心的作用就是调用service_executor服务运行子模块(线程模型子模块)的schedule()接口来把状态机调度任务入队到ASIO网络库的一个全局队列(adaptive动态线程模型)，如果是一个链接一个线程模型，则任务入队到线程级私有队列。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"      adaptive线程模型，任务入队以及工作线程调度任务执行的流程将在后续的线程模型子模块中分析，也可以参考："},{"type":"link","attrs":{"href":"https://my.oschina.net/u/4087916/blog/4295038","title":null},"content":[{"type":"text","text":"<>"}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"      此外，_scheduleNextWithGuard()入队到全局队列的任务就是本模块后面需要分析的SSM状态机任务，这些task任务通过本函数接口的func (...)进行封装，然后通过线程模型子模块入队到一个全局队列。Func(...)这个task任务中会直接调用_runNextInGuard()接口来进行状态转换处理，该接口也就是入队到ASIO全局队列的任务，核心代码功能如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.void ServiceStateMachine::_runNextInGuard(ThreadGuard guard) {  \n2.    //获取当前SSM状态  \n3.    auto curState = state();  \n4.    // If this is the first run of the SSM, then update its state to Source  \n5.    //如果是第一次运行该SSM，则状态为Created，到这里标记可以准备接收数据了  \n6.    if (curState == State::Created) {   \n7.        //进入Source等待接收数据  \n8.        curState = State::Source;  \n9.        _state.store(curState);  \n10.    }  \n11.    //各状态对应处理  \n12.    try {  \n13.        switch (curState) {   \n14.         //接收数据  \n15.            case State::Source:    \n16.                _sourceMessage(std::move(guard));  \n17.                break;  \n18.            //以及接收到完整的一个mongodb报文，可以内部处理(解析+命令处理+应答客户端)  \n19.            case State::Process:  \n20.                _processMessage(std::move(guard));  \n21.                break;  \n22.            //链接异常或者已经关闭，则开始回收处理  \n23.            case State::EndSession:  \n24.                _cleanupSession(std::move(guard));  \n25.                break;  \n26.            default:  \n27.                MONGO_UNREACHABLE;  \n28.        }  \n29. return;\n30.    } catch (...) {  \n31.        //异常打印  \n32.    }  \n33.    //异常处理  \n34.    ......  \n35.    //进入EndSession状态  \n36.    _state.store(State::EndSession);  \n37.    _cleanupSession(std::move(guard));  \n38.} "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 从上面的代码实现可以看出，真正入队到全局队列中的任务类型只有三个，分别是："}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"1) 接收mongodb数据的task任务，简称为readTask。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"2) 接收到一个完整mongodb数据后的后续处理(包括协议解析、命令处理、DB获取数据、发送数据给客户端等)，简称为dealTask。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"3) 接收或者发送数据异常、链接关闭等引起的后续资源释放，简称为cleanTask。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 下面针对这三种task任务核心代码实现进行分析："}]},{"type":"bulletedlist","content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","marks":[{"type":"strong"}],"text":" readTask任务核心代码实现"}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" readTask任务核心代码实现由_sourceMessage()接口实现，具体代码如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.//接收客户端数据  \n2.void ServiceStateMachine::_sourceMessage(ThreadGuard guard) {  \n3.    ......  \n4.    //获取本session接收数据的ticket，也就是ASIOSourceTicket  \n5.    auto ticket = _session()->sourceMessage(&_inMessage);   \n6.    //进入等等接收数据状态  \n7.    _state.store(State::SourceWait);    \n8.    //release恢复worker线程原有的线程名，synchronous线程模型为\"conn-xx\"，adaptive对应worker线程名为\"conn-xx\"  \n9.    guard.release();  \n10.    //线程模型默认同步方式，也就是一个链接一个线程  \n11.    if (_transportMode == transport::Mode::kSynchronous) {  \n12.         //同步方式，读取到一个完整mongodb报文后执行_sourceCallback回调  \n13.         _sourceCallback([this](auto ticket) {  \n14.            MONGO_IDLE_THREAD_BLOCK;  \n15.            return _session()->getTransportLayer()->wait(std::move(ticket));  \n16.        }(std::move(ticket)));   \n17.    } else if (_transportMode == transport::Mode::kAsynchronous) {  \n18.        //adaptive线程模型，异步读取一个mongodb报文后执行_sourceCallback回调  \n19.        _session()->getTransportLayer()->asyncWait(   \n20.            ////TransportLayerASIO::ASIOSourceTicket::_bodyCallback读取到一个完整报文后执行该回调  \n21.            std::move(ticket), [this](Status status) { _sourceCallback(status); });  \n22.    }  \n23.}  \n24.  \n25.//接收到一个完整mongodb报文后的回调处理  \n26.void ServiceStateMachine::_sourceCallback(Status status) {  \n27.    //构造ThreadGuard，修改执行本SSM接口的线程名为conn-xx  \n28.    ThreadGuard guard(this);   \n29.  \n30.    //状态检查  \n31.    dassert(state() == State::SourceWait);  \n32.    //获取链接session远端信息  \n33.    auto remote = _session()->remote();   \n34.    if (status.isOK()) {  \n35.    //等待调度，进入处理消息阶段  _processMessage  \n36.        _state.store(State::Process);  \n37.        //注意kMayRecurse标识State::Process阶段的处理还是由本线程执行，这是一个递归标记  \n38.        return _scheduleNextWithGuard(std::move(guard), ServiceExecutor::kMayRecurse);  \n39.    }  \n40.    ......  \n41.    //异常流程调用  \n42.    _runNextInGuard(std::move(guard));  \n43.}  "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" SSM调度的第一个任务就是readTask任务，从上面的源码分析可以看出，该任务就是通过ticket数据分发模块从ASIO网络库读取一个完整长度的mongodb报文，然后执行"},{"type":"text","marks":[{"type":"italic"}],"text":"sourceCallback回调。进入该回调函数后，即刻设置SSM状态为State::Process状态，然后调用"},{"type":"text","text":"scheduleNextWithGuard(...)把dealTask任务入队到ASIO的全局队列(adaptive线程模型)，或者入队到线程级私有队列(synchronous线程模型)等待worker线程调度执行。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 这里有个细节，在把dealTask入队的时候，携带了kMayRecurse标记，该标记标识该任务可以递归调用，也就是该任务可以由当前线程继续执行，这样也就可以保证同一个请求的taskRead任务和dealTask任务由同一个线程处理。任务递归调度，可以参考后面的线程模型子模块源码实现。"}]},{"type":"bulletedlist","content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","marks":[{"type":"strong"}],"text":" dealTask任务核心代码实现"}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 当读取到一个完整长度的mongodb报文后，就会把dealTask任务入队到全局队列，然后由worker线程调度执行该task任务。dealTask任务的核心代码实现如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.//dealTask处理  \n2.void ServiceStateMachine::_processMessage(ThreadGuard guard) {  \n3.    ......  \n4.    //入口流量计数  \n5.    networkCounter.hitLogicalIn(_inMessage.size());  \n6.    //获取一个唯一的UniqueOperationContext，一个客户端对应一个UniqueOperationContext  \n7.    auto opCtx = Client::getCurrent()->makeOperationContext();  \n8.    //ServiceEntryPointMongod::handleRequest  ServiceEntryPointMongos::handleRequest请求处理  \n9.    //command处理、DB访问后的数据通过dbresponse返回  \n10.    DbResponse dbresponse = _sep->handleRequest(opCtx.get(), _inMessage);  \n11.    //释放opCtx，这样currentop就看不到了  \n12.    opCtx.reset();  \n13.    //需要发送给客户端的应答数据  \n14.    Message& toSink = dbresponse.response;  \n15.    //应答数据存在  \n16.    if (!toSink.empty()) {    \n17.        ......  \n18.        //发送数据 ServiceStateMachine::_sinkMessage()  \n19.        _sinkMessage(std::move(guard), std::move(toSink));  \n20.  \n21.    } else {  \n22.       //如果不需要应答客户端的处理  \n23.       ......  \n24.    }  \n25.}  \n26.  \n27.//调用Sinkticket发送数据  \n28.void ServiceStateMachine::_sinkMessage(ThreadGuard guard, Message toSink) {  \n29.    //获取发送数据的ASIOSinkTicket  \n30.    auto ticket = _session()->sinkMessage(toSink);  \n31.    //进入sink发送等待状态  \n32.    _state.store(State::SinkWait);  \n33.    //恢复原有的worker线程名  \n34.    guard.release();  \n35.    //synchronous线程模型，同步发送  \n36.    if (_transportMode == transport::Mode::kSynchronous) {  \n37.     //最终在ASIOSinkTicket同步发送数据成功后执行_sinkCallback  \n38.        _sinkCallback(_session()->getTransportLayer()->wait(std::move(ticket)));  \n39.    } else if (_transportMode == transport::Mode::kAsynchronous) {  \n40.        //最终在ASIOSinkTicket异步发送数据成功后执行_sinkCallback  \n41.        _session()->getTransportLayer()->asyncWait(  \n42.            std::move(ticket), [this](Status status) { _sinkCallback(status); });  \n43.    }  \n44.}  \n45.  \n46.//sink数据发送  \n47.void ServiceStateMachine::_sinkCallback(Status status) {  \n48.    //SSM归属于新的guard，同时修改当前线程名为conn-xx  \n49.    ThreadGuard guard(this);  \n50.    //状态检查  \n51.    dassert(state() == State::SinkWait);  \n52.    if (!status.isOK()) {  \n53.        //进入EndSession状态  \n54.        _state.store(State::EndSession);  \n55.     //异常情况调用  \n56.        return _runNextInGuard(std::move(guard));  \n57.    } else if (_inExhaust) { //_inExhaust方式   \n58.        //注意这里的状态是process   _processMessage   还需要继续进行Process处理  \n59.        _state.store(State::Process);   \n60.    } else {   \n61.        //正常流程始终进入该分支 _sourceMessage    这里继续进行递归接收数据处理  \n62.        //注意这里的状态是Source,继续接收客户端请求  \n63.        _state.store(State::Source);  \n64.    }  \n65.    //本链接对应的一次mongo访问已经应答完成，需要继续要一次调度了  \n66.    return _scheduleNextWithGuard(std::move(guard),  \n67.                                  ServiceExecutor::kDeferredTask |  \n68.                                      ServiceExecutor::kMayYieldBeforeSchedule);  \n69.}  "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" readTask通过ticket数据分发子模块读取一个完整长度的mongodb报文后，开始dealTask任务逻辑，该任务也就是_processMessage(...)。该接口中核心实现就是调用mongod和mongos实例对应的服务入口类的handleRequest(...)接口来完成后续的command命令处理、DB层数据访问等，访问到的数据存储到DbResponse中，最终在通过_sinkMessage(...)把数据发送出去。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 真正的mongodb内部处理流程实际上就是通过该dealTask任务完成，该任务也是处理整个请求中资源耗费最重的一个环节。在该task任务中，当数据成功发送给客户端后，该session链接对应的SSM状态机进入State::Source状态，继续等待worker线程调度来完成后续该链接的新请求。"}]},{"type":"bulletedlist","content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","marks":[{"type":"strong"}],"text":" cleanTask任务"}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 在数据读写过程、客户端链接关闭、访问DB数据层等任何一个环节异常，则会进入State::EndSession状态。该状态对应得任务实现相对比较简单，具体代码实现如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.//session对应链接回收处理  \n2.void ServiceStateMachine::_cleanupSession(ThreadGuard guard) {  \n3.    //进入这个状态后在~ThreadGuard::release中进行ssm _cleanupHook处理，该hook在ServiceEntryPointImpl::startSession  \n4.    _state.store(State::Ended);  \n5.    //清空message buffer  \n6.    _inMessage.reset();  \n7.    //释放链接对应client资源  \n8.    Client::releaseCurrent();  \n9.}  "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 进入该状态后直接由本线程进行session资源回收和client资源释放处理，而无需状态机调度worker线程来回收。"}]},{"type":"heading","attrs":{"align":null,"level":2},"content":[{"type":"text","text":"2.2关于worker线程名和guardthread线程守护类"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 前面得分析我们知道，当线程模型为adaptive动态线程模型的时候，mongod和mongos实例对应的子线程中有很多名为“conn-xx”和”worker-xx”的线程，而且同一个线程可能一会儿线程名为“conn-xx”，下一次又变为了”worker-xx”。这个线程名的初始命名和线程名更改与ServiceStateMachine状态机调度类、guardthread线程守护类、worker线程模型等都有关系。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Worker线程由ServiceExecutor线程模型子模块创建，请参考后续线程模型子模块相关章节。默认初始化线程名为”conn-x”，初始化代码实现如下："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.//ServiceStateMachine::create调用,ServiceStateMachine类初始化构造  \n2.ServiceStateMachine::ServiceStateMachine(...)  \n3.    ......  \n4.    //线程名初始化：conn-xx，xx代码session id  \n5.    _threadName{str::stream() << \"conn-\" << _session()->id()} {}   \n6.}  \n7.  \n8.class Session {  \n9.    ......  \n10.    //sessionID，自增生成  \n11.    const Id _id;  \n12.}  \n13.  \n14.//全局sessionIdCounter计数器，初始化为0  \n15.AtomicUInt64 sessionIdCounter(0);  \n16.  \n17.//session id自增  \n18.Session::Session() : _id(sessionIdCounter.addAndFetch(1)) {} "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" SSM状态处理过程中，会把一个完整的请求过程= readTask任务+ dealTask任务，这两个任务都是通过SSM状态机和ServiceExecutor线程模型子模块的worker线程配合调度完成，在任务处理过程中处理同一个任务的线程可能会有多次线程名更改，这个就是结合guardthread线程守护类来完成，以一个线程名切换更改伪代码实现为例："}]},{"type":"codeblock","attrs":{"lang":"cpp"},"content":[{"type":"text","text":"1.worker_thread_run_task(...)  \n2.{  \n3.    //如果是adaptive线程模型，当前worker线程名为\"worker-xx\"  \n4.    print(threadName)  \n5.    //业务逻辑处理1  \n6.    ......  \n7.      \n8.    //初始化构造ThreadGuard，这里面修改线程名为_ssm->_threadName,也就是\"conn-xx\"，  \n9.    //同时保存原来的线程名\"worker-xx\"到_ssm->_oldThreadName中  \n10.    ThreadGuard guard(this);  \n11.    //如果是adaptive线程模型，线程名打印内容为\"conn-xx\"  \n12.    print(threadName)  \n13.    //业务逻辑处理2  \n14.    ......  \n15.    //恢复_ssm->_oldThreadName保存的线程名\"worker-xx\"  \n16.    guard.release();  \n17.      \n18.    //如果是adaptive线程模型，线程名恢复为\"worker-xx\"  \n19.    print(threadName)  \n20.}  "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 从上面的伪代码可以看出，adaptive线程模型对应worker线程名为”worker”，在进入ThreadGuard guard(this)流程后，线程名更改为”conn-xx”线程，当guard.release()释放后恢复原有”worker-xx”线程名。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 结合前面的SSM状态处理流程，adaptive线程模型可以得到如下总结：底层网络IO数据读写过程，worker线程名会改为”worker-xx”，其他非网络IO的mongodb内部逻辑处理线程名为”conn-xx”。所以，如果查看mongod或者mongos进程所有线程名的时候，如果发现线程名为”worker-xx”，说明当前线程在处理网络IO；如果发现线程名为”conn-xx”，则说明当前线程在处理内部逻辑处理，对于mongod实例可以理解为主要处理磁盘IO。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 由于synchronous同步线程模型，同一链接对应的所有客户端请求至始至终都有同一线程处理，所以整个处理线程名不会改变，也没必要修改线程名，整个过程都是”conn-xx”线程名。"}]},{"type":"heading","attrs":{"align":null,"level":2},"content":[{"type":"text","text":"2.3该模块函数接口总结大全"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"      前面分析了主要核心接口源码实现，很多其他接口没有一一列举详细分析，该模块u所有接口功能总结如下，更多接口代码实现详见"},{"type":"link","attrs":{"href":"https://github.com/y123456yz/reading-and-annotate-mongodb-3.6","title":null},"content":[{"type":"text","marks":[{"type":"underline"}],"text":"Mongodb内核源码详细注释分析"}]},{"type":"text","text":"："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/d7/d7d7d677becaaa54eaba8397110a1276.png","alt":null,"title":null,"style":null,"href":null,"fromPaste":true,"pastePass":true}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/ba/ba6ed1479ab80046e2e5c1b717a7bf24.png","alt":null,"title":null,"style":null,"href":null,"fromPaste":true,"pastePass":true}},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":"3. 总结"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"   "},{"type":"link","attrs":{"href":"https://xie.infoq.cn/article/d4460b807c9835c6d80707410","title":null},"content":[{"type":"text","text":"Mongodb网络模块源码实现及性能调优(一)"}],"marks":[{"type":"underline"}]},{"type":"text","marks":[{"type":"underline"}],"text":" "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://xie.infoq.cn/article/1cac5adcd1b4f3fe512a1457a","title":null},"content":[{"type":"text","marks":[{"type":"underline"}],"text":"<>"}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 本文主要分析了service_state_machine状态机子模块，该模块把session对应的客户端请求转换为readTask任务、dealTask任务和cleanTask任务，前两个任务通过worker线程完成调度处理，cleanTask任务在内部处理异常或者链接关闭的时候由本线程直接执行，而不是通过worker线程调度执行。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 这三个任务处理过程会分别对应到Created、Source、SourceWait、Process、SinkWait、EndSession、Ended七种状态的一种或者多种，具体详见前面的状态码分析。一个正常的客户端请求状态转换过程如下:"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"1) 链接刚建立的第一次请求状态转换过程："}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Created->Source -> SourceWait -> Process -> SinkWait -> Source"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"2) 该链接后续的请求状态转换过程："}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"  Source -> SourceWait -> Process -> SinkWait -> Source"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 此外，SSM状态机调度模块通过ServiceStateMachine::_scheduleNextWithGuard(...)接口和线程模型子模块关联起来。SSM通过该接口完成worker线程初始创建、task任务入队处理，下期将分析"},{"type":"text","marks":[{"type":"strong"}],"text":"<<网络线程模型子模块>>"},{"type":"text","text":"详细源码实现。"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","marks":[{"type":"strong"}],"text":"说明："}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" 该模块更多接口实现细节详见Mongodb内核源码注释："},{"type":"link","attrs":{"href":"https://github.com/y123456yz/reading-and-annotate-mongodb-3.6","title":null},"content":[{"type":"text","marks":[{"type":"underline"}],"text":"Mongodb内核源码详细注释分析"}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}}]}