MongoDB transport_layer网络传输层模块源码实现三

class ServiceStateMachine::ThreadGuard {  
  ......  
public:  
  // create a ThreadGuard which will take ownership of the SSM in this thread.  
  //ThreadGuard初始化，标记ssm所有权属于本线程  
  explicit ThreadGuard(ServiceStateMachine* ssm) : _ssm{ssm} {  
      //获取ssm状态机所有权  
      auto owned = _ssm->_owned.compareAndSwap(Ownership::kUnowned, Ownership::kOwned);  
      //如果是一个链接一个线程模型，则条件满足  
      if (owned == Ownership::kStatic) {   
      //一个链接一个线程模型,由于链接始终由同一个线程处理，因此该链接对应的ssm始终归属于同一个线程  
          dassert(haveClient());  
          dassert(Client::getCurrent() == _ssm->_dbClientPtr);  
          //标识归属权已确定  
          _haveTakenOwnership = true;  
          return;  
      }  
        ......
      //adaptive 动态线程模式走下面的模式  
 
      //把当前线程的线程名零时保存起来  
      auto oldThreadName = getThreadName();   
      //ssm保存的线程名和当前线程名不同  
      if (oldThreadName != _ssm->_threadName) {  
          //即将修改线程名了，把修改前的线程名保存到_oldThreadName  
          _ssm->_oldThreadName = getThreadName().toString();  
          //把运行本ssm状态机的线程名改为conn-x线程  
          setThreadName(_ssm->_threadName); //把当前线程改名为_threadName  
      }  
 
      //设置本线程对应client信息,一个链接对应一个client,标识本client当前归属于本线程处理  
      Client::setCurrent(std::move(_ssm->_dbClient));  
        //本状态机ssm所有权有了，归属于运行本ssm的线程  
      _haveTakenOwnership = true;  
  }  
  ......  
  //重新赋值  
  ThreadGuard& operator=(ThreadGuard&& other) {  
      if (this != &other) {  
          _ssm = other._ssm;  
          _haveTakenOwnership = other._haveTakenOwnership;  
          //原来的other所有权失效  
          other._haveTakenOwnership = false;  
      }  
  //返回  
      return *this;  
  };  
 
  //析构函数  
  ~ThreadGuard() {  
  //ssm所有权已确定，则析构的时候，调用release处理，恢复线程原有线程名  
      if (_haveTakenOwnership)  
          release();  
  }  
 
  //一个链接一个线程模型，标记_owned为kStatic，也就是线程名永远不会改变  
  void markStaticOwnership() {  
      dassert(static_cast<bool>(*this));  
      _ssm->_owned.store(Ownership::kStatic);  
  }  
 
  //恢复原有线程名，同时把client信息从调度线程归还给状态机  
  void release() {  
        auto owned = _ssm->_owned.load();  
        //adaptive异步线程池模式满足if条件，表示SSM固定归属于某个线程  
      if (owned != Ownership::kStatic) {  
      //本线程拥有currentClient信息，于是把它归还给SSM状态机  
          if (haveClient()) {  
              _ssm->_dbClient = Client::releaseCurrent();  
          }  
            //恢复到以前的线程名  
          if (!_ssm->_oldThreadName.empty()) {  
              //恢复到老线程名  
              setThreadName(_ssm->_oldThreadName);   
          }  
      }  
      //状态机状态进入end，则调用对应回收hook处理  
      if (_ssm->state() == State::Ended) {  
          //链接关闭的回收处理 ServiceStateMachine::setCleanupHook  
          auto cleanupHook = std::move(_ssm->_cleanupHook);  
          if (cleanupHook)  
              cleanupHook();  
          return;  
      }  
 
      //归属权失效  
      _haveTakenOwnership = false;  
      //归属状态变为未知  
      if (owned == Ownership::kOwned) {  
          _ssm->_owned.store(Ownership::kUnowned);  
      }  
  }  
 
private:  
  //本线程守护当前对应的ssm  
  ServiceStateMachine* _ssm;  
  //默认false，标识该状态机ssm不归属于任何线程  
  bool _haveTakenOwnership = false;  
}

//一个新链接对应一个ServiceStateMachine保存到ServiceEntryPointImpl._sessions中  
class ServiceStateMachine : public std::enable_shared_from_this<ServiceStateMachine> {  
  ......  
public:  
  ......  
  static std::shared_ptr<ServiceStateMachine> create(...);  
  ServiceStateMachine(...);  
  //状态机所属状态
  enum class State {  
      //ServiceStateMachine::ServiceStateMachine构造函数初始状态  
      Created,        
      //ServiceStateMachine::_runNextInGuard开始进入接收网络数据状态  
      //标记本次客户端请求已完成(已经发送DB获取的数据给客户端)，等待调度进行该链接的  
      //下一轮请求，该状态对应处理流程为_sourceMessage  
      Source,         
      //等待获取客户端的数据  
      SourceWait,     
      //接收到一个完整mongodb报文后进入该状态  
      Process,        
      //等待数据发送成功  
      SinkWait,       
      //接收或者发送数据异常、链接关闭，则进入该状态 _cleanupSession  
      EndSession,     
      //session回收处理进入该状态  
      Ended           
  };  
  //所有权状态，主要用来判断是否需要在状态转换中跟新线程名，只对动态线程模型生效  
  enum class Ownership {   
      //该状态表示本状态机SSM处于非活跃状态  
      kUnowned,    
      //该状态标识本状态机SSM归属于某个工作worker线程，处于活跃调度运行状态  
      kOwned,   
      //表示SSM固定归属于某个线程  
      kStatic   
  };  
    
  ......  
private:  
  //ThreadGuard可以理解为线程守护，后面在ThreadGuard类中单独说明  
  class ThreadGuard;  
  friend class ThreadGuard;  
    
  ......  
  //获取session信息  
  const transport::SessionHandle& _session()  
  //以下两个接口为任务task调度相关接口  
  void _scheduleNextWithGuard(...);  
  void _runNextInGuard(ThreadGuard guard);  
  //接收到一个完整mongodb报文后的处理  
  inline void _processMessage(ThreadGuard guard);  
  //以下四个接口完成底层数据读写及其对应回调处理  
  void _sourceCallback(Status status);  
  void _sinkCallback(Status status);  
  void _sourceMessage(ThreadGuard guard);  
  void _sinkMessage(ThreadGuard guard, Message toSink);  
    
  //一次客户端请求，当前mongodb服务端所处的状态  
  AtomicWord<State> _state{State::Created};  
  //服务入口，ServiceEntryPointMongod ServiceEntryPointMongos mongod及mongos入口点  
  ServiceEntryPoint* _sep;  
  //synchronous及adaptive模式，也就是线程模型是一个链接一个线程还是动态线程池  
  transport::Mode _transportMode;  
  //ServiceContextMongoD(mongod)或者ServiceContextNoop(mongos)服务上下文  
  ServiceContext* const _serviceContext;  
  //也就是本ssm对应的session信息，默认对应ASIOSession   
  transport::SessionHandle _sessionHandle;   
  //根据session构造对应client信息,ServiceStateMachine::ServiceStateMachine赋值  
  //也就是本次请求对应的客户端信息  
  ServiceContext::UniqueClient _dbClient;  
  //指向上面的_dbClient  
  const Client* _dbClientPtr;  
  //该SSM当前处理线程的线程名，因为adaptive线程模型一次请求中的不同状态会修改线程名
  const std::string _threadName;  
  //修改线程名之前的线程名称  
  std::string _oldThreadName;  
  //ServiceEntryPointImpl::startSession->ServiceStateMachine::setCleanupHook中设置赋值  
  //session链接回收处理  
  stdx::function<void()> _cleanupHook;  
  //接收处理的message信息 一个完整的报文就记录在该msg中  
  Message _inMessage;   
  //默认初始化kUnowned,标识本SSM状态机处于非活跃状态，  
  //主要用来判断是否需要在状态转换中跟新线程名，只对动态线程模型生效  
  AtomicWord<Ownership> _owned{Ownership::kUnowned};  
}

//ServiceEntryPointImpl::startSession中执行 启动SSM状态机  
void ServiceStateMachine::start(Ownership ownershipModel) {  
  //直接调用_scheduleNextWithGuard接口  
  _scheduleNextWithGuard(   
  //listener线程暂时性的变为conn线程名，在_scheduleNextWithGuard中任  
  //务入队完成后，在下面的_scheduleNextWithGuard调用guard.release()恢复listener线程名  
      ThreadGuard(this), transport::ServiceExecutor::kEmptyFlags, ownershipModel);  
}  
 
void ServiceStateMachine::_scheduleNextWithGuard(...) {  
  //该任务func实际上由worker线程运行,worker线程从asio库的全局队列获取任务调度执行  
  auto func = [ ssm = shared_from_this(), ownershipModel ] {  
      //构造ThreadGuard  
      ThreadGuard guard(ssm.get());    
      //说明是sync mode,即一个链接一个线程模式, 归属权明确，属于指定线程  
      if (ownershipModel == Ownership::kStatic)   
          guard.markStaticOwnership();  
      //对应:ServiceStateMachine::_runNextInGuard，在这里面完成状态调度转换  
      ssm->_runNextInGuard(std::move(guard));    
  };  
  //恢复之前的线程名，如果该SSM进入Ended状态，则开始资源回收处理  
  guard.release();  
  //ServiceExecutorAdaptive::schedule(adaptive)   ServiceExecutorSynchronous::schedule(synchronous)  
  //第一次进入该函数的时候在这里面创建新线程，不是第一次则把task任务入队调度  
  Status status = _serviceContext->getServiceExecutor()->schedule(std::move(func), flags);  
  if (status.isOK()) {  
      return;  
  }  
  //异常处理  
  ......  
}

void ServiceStateMachine::_runNextInGuard(ThreadGuard guard) {  
  //获取当前SSM状态  
  auto curState = state();  
  // If this is the first run of the SSM, then update its state to Source  
  //如果是第一次运行该SSM，则状态为Created，到这里标记可以准备接收数据了  
  if (curState == State::Created) {   
      //进入Source等待接收数据  
      curState = State::Source;  
      _state.store(curState);  
  }  
  //各状态对应处理  
  try {  
      switch (curState) {   
          //接收数据  
          case State::Source:    
              _sourceMessage(std::move(guard));  
              break;  
          //以及接收到完整的一个mongodb报文，可以内部处理(解析+命令处理+应答客户端)  
          case State::Process:  
              _processMessage(std::move(guard));  
              break;  
          //链接异常或者已经关闭，则开始回收处理  
          case State::EndSession:  
              _cleanupSession(std::move(guard));  
              break;  
          default:  
              MONGO_UNREACHABLE;  
      }  
        return;
  } catch (...) {  
      //异常打印  
  }  
  //异常处理  
  ......  
  //进入EndSession状态  
  _state.store(State::EndSession);  
  _cleanupSession(std::move(guard));  
}

//接收客户端数据  
void ServiceStateMachine::_sourceMessage(ThreadGuard guard) {  
  ......  
  //获取本session接收数据的ticket，也就是ASIOSourceTicket  
  auto ticket = _session()->sourceMessage(&_inMessage);   
  //进入等等接收数据状态  
  _state.store(State::SourceWait);    
  //release恢复worker线程原有的线程名，synchronous线程模型为"conn-xx"，adaptive对应worker线程名为"conn-xx"  
  guard.release();  
  //线程模型默认同步方式，也就是一个链接一个线程  
  if (_transportMode == transport::Mode::kSynchronous) {  
        //同步方式，读取到一个完整mongodb报文后执行_sourceCallback回调  
        _sourceCallback([this](auto ticket) {  
          MONGO_IDLE_THREAD_BLOCK;  
          return _session()->getTransportLayer()->wait(std::move(ticket));  
      }(std::move(ticket)));   
  } else if (_transportMode == transport::Mode::kAsynchronous) {  
      //adaptive线程模型，异步读取一个mongodb报文后执行_sourceCallback回调  
      _session()->getTransportLayer()->asyncWait(   
          ////TransportLayerASIO::ASIOSourceTicket::_bodyCallback读取到一个完整报文后执行该回调  
          std::move(ticket), [this](Status status) { _sourceCallback(status); });  
  }  
}  
 
//接收到一个完整mongodb报文后的回调处理  
void ServiceStateMachine::_sourceCallback(Status status) {  
  //构造ThreadGuard，修改执行本SSM接口的线程名为conn-xx  
  ThreadGuard guard(this);   
 
  //状态检查  
  dassert(state() == State::SourceWait);  
  //获取链接session远端信息  
  auto remote = _session()->remote();   
  if (status.isOK()) {  
  //等待调度，进入处理消息阶段 _processMessage  
      _state.store(State::Process);  
      //注意kMayRecurse标识State::Process阶段的处理还是由本线程执行，这是一个递归标记  
      return _scheduleNextWithGuard(std::move(guard), ServiceExecutor::kMayRecurse);  
  }  
  ......  
  //异常流程调用  
  _runNextInGuard(std::move(guard));  
}

//dealTask处理  
void ServiceStateMachine::_processMessage(ThreadGuard guard) {  
  ......  
  //入口流量计数  
  networkCounter.hitLogicalIn(_inMessage.size());  
  //获取一个唯一的UniqueOperationContext，一个客户端对应一个UniqueOperationContext  
  auto opCtx = Client::getCurrent()->makeOperationContext();  
  //ServiceEntryPointMongod::handleRequest ServiceEntryPointMongos::handleRequest请求处理  
  //command处理、DB访问后的数据通过dbresponse返回  
  DbResponse dbresponse = _sep->handleRequest(opCtx.get(), _inMessage);  
  //释放opCtx，这样currentop就看不到了  
  opCtx.reset();  
  //需要发送给客户端的应答数据  
  Message& toSink = dbresponse.response;  
  //应答数据存在  
  if (!toSink.empty()) {    
      ......  
      //发送数据 ServiceStateMachine::_sinkMessage()  
      _sinkMessage(std::move(guard), std::move(toSink));  
 
  } else {  
      //如果不需要应答客户端的处理  
      ......  
  }  
}  
 
//调用Sinkticket发送数据  
void ServiceStateMachine::_sinkMessage(ThreadGuard guard, Message toSink) {  
  //获取发送数据的ASIOSinkTicket  
  auto ticket = _session()->sinkMessage(toSink);  
  //进入sink发送等待状态  
  _state.store(State::SinkWait);  
  //恢复原有的worker线程名  
  guard.release();  
  //synchronous线程模型，同步发送  
  if (_transportMode == transport::Mode::kSynchronous) {  
      //最终在ASIOSinkTicket同步发送数据成功后执行_sinkCallback  
      _sinkCallback(_session()->getTransportLayer()->wait(std::move(ticket)));  
  } else if (_transportMode == transport::Mode::kAsynchronous) {  
      //最终在ASIOSinkTicket异步发送数据成功后执行_sinkCallback  
      _session()->getTransportLayer()->asyncWait(  
          std::move(ticket), [this](Status status) { _sinkCallback(status); });  
  }  
}  
 
//sink数据发送  
void ServiceStateMachine::_sinkCallback(Status status) {  
  //SSM归属于新的guard，同时修改当前线程名为conn-xx  
  ThreadGuard guard(this);  
  //状态检查  
  dassert(state() == State::SinkWait);  
  if (!status.isOK()) {  
      //进入EndSession状态  
      _state.store(State::EndSession);  
      //异常情况调用  
      return _runNextInGuard(std::move(guard));  
  } else if (_inExhaust) { //_inExhaust方式   
      //注意这里的状态是process   _processMessage   还需要继续进行Process处理  
      _state.store(State::Process);   
  } else {   
      //正常流程始终进入该分支 _sourceMessage   这里继续进行递归接收数据处理  
      //注意这里的状态是Source,继续接收客户端请求  
      _state.store(State::Source);  
  }  
  //本链接对应的一次mongo访问已经应答完成，需要继续要一次调度了  
  return _scheduleNextWithGuard(std::move(guard),  
                                ServiceExecutor::kDeferredTask |  
                                    ServiceExecutor::kMayYieldBeforeSchedule);  
}

//session对应链接回收处理  
void ServiceStateMachine::_cleanupSession(ThreadGuard guard) {  
  //进入这个状态后在~ThreadGuard::release中进行ssm _cleanupHook处理，该hook在ServiceEntryPointImpl::startSession  
  _state.store(State::Ended);  
  //清空message buffer  
  _inMessage.reset();  
  //释放链接对应client资源  
  Client::releaseCurrent();  
}

//ServiceStateMachine::create调用,ServiceStateMachine类初始化构造  
ServiceStateMachine::ServiceStateMachine(...)  
  ......  
  //线程名初始化：conn-xx，xx代码session id  
  _threadName{str::stream() << "conn-" << _session()->id()} {}   
}  
 
class Session {  
  ......  
  //sessionID，自增生成  
  const Id _id;  
}  
 
//全局sessionIdCounter计数器，初始化为0  
AtomicUInt64 sessionIdCounter(0);  
 
//session id自增  
Session::Session() : _id(sessionIdCounter.addAndFetch(1)) {}

worker_thread_run_task(...)  
{  
  //如果是adaptive线程模型，当前worker线程名为"worker-xx"  
  print(threadName)  
  //业务逻辑处理1  
  ......  
     
  //初始化构造ThreadGuard，这里面修改线程名为_ssm->_threadName,也就是"conn-xx"，  
  //同时保存原来的线程名"worker-xx"到_ssm->_oldThreadName中  
  ThreadGuard guard(this);  
  //如果是adaptive线程模型，线程名打印内容为"conn-xx"  
  print(threadName)  
  //业务逻辑处理2  
  ......  
  //恢复_ssm->_oldThreadName保存的线程名"worker-xx"  
  guard.release();  
     
  //如果是adaptive线程模型，线程名恢复为"worker-xx"  
  print(threadName)  
}