MongoDB Command命令处理模块源码实现一

//dealTask处理  
void ServiceStateMachine::_processMessage(ThreadGuard guard) {  
    ......
    //command处理、DB访问后的数据通过dbresponse返回  
    DbResponse dbresponse = _sep->handleRequest(opCtx.get(), _inMessage);  
    ......
}

//mongos实例启动初始化  
static ExitCode runMongosServer() {  
    ......  
    //mongos实例对应sep为ServiceEntryPointMongos  
    auto sep = stdx::make_unique<ServiceEntryPointMongos>(getGlobalServiceContext());  
    getGlobalServiceContext()->setServiceEntryPoint(std::move(sep));  
    ......  
}  
  
//mongod实例启动初始化  
ExitCode _initAndListen(int listenPort) {  
    ......  
    //mongod实例对应sep为ServiceEntryPointMongod  
    serviceContext->setServiceEntryPoint(  
        stdx::make_unique<ServiceEntryPointMongod>(serviceContext));  
    ......  
}  
  
//SSM状态机初始化  
ServiceStateMachine::ServiceStateMachine(...)  
    : _state{State::Created},  
      //mongod和mongos实例的服务入口通过这里赋值给_seq成员变量  
      _sep{svcContext->getServiceEntryPoint()},  
      ......  
}

OP_MSG {  
    //MongoDB报文头部  
    MsgHeader header;            
    //位图，用于标识报文是否需要校验 是否需要应答等  
    uint32 flagBits;           // message flags  
    //报文内容，例如find write等命令内容通过bson格式存在于该结构中  
    Sections[] sections;       // data sections  
    //报文CRC校验  
    optional<uint32> checksum; // optional CRC-32C checksum  
}  

namespace MSGHEADER {  
//header头部各个字段信息  
struct Layout {  
    //整个message长度，包括header长度和body长度  
    int32_t messageLength;     
    //requestID 该请求id信息  
    int32_t requestID;         
    //getResponseToMsgId解析  
    int32_t responseTo;        
    //操作类型：OP_UPDATE、OP_INSERT、OP_QUERY、OP_DELETE、OP_MSG等  
    int32_t opCode;  
};  
  
//ConstView实现header头部数据解析  
class ConstView {   
public:  
    ......  
    //初始化构造  
    ConstView(const char* data) : _data(data) {}  
    //获取_data地址  
    const char* view2ptr() const {  
        return data().view();  
    }  
    //TransportLayerASIO::ASIOSourceTicket::_headerCallback调用  
    //解析header头部的messageLength字段  
    int32_t getMessageLength() const {  
        return data().read<LittleEndian<int32_t>>(offsetof(Layout, messageLength));  
    }  
    //解析header头部的requestID字段  
    int32_t getRequestMsgId() const {  
        return data().read<LittleEndian<int32_t>>(offsetof(Layout, requestID));  
    }  
    //解析header头部的getResponseToMsgId字段  
    int32_t getResponseToMsgId() const {  
        return data().read<LittleEndian<int32_t>>(offsetof(Layout, responseTo));  
    }  
    //解析header头部的opCode字段  
    int32_t getOpCode() const {  
        return data().read<LittleEndian<int32_t>>(offsetof(Layout, opCode));  
    }  
  
protected:  
    //MongoDB报文数据起始地址  
    const view_type& data() const {  
        return _data;  
    }  
private:  
    //数据部分  
    view_type _data;  
};  
  
//View填充header头部数据  
class View : public ConstView {  
public:  
    ......  
    //构造初始化  
    View(char* data) : ConstView(data) {}  
    //header起始地址  
    char* view2ptr() {  
        return data().view();  
    }  
    //以下四个接口进行header填充  
    //填充header头部messageLength字段  
    void setMessageLength(int32_t value) {  
        data().write(tagLittleEndian(value), offsetof(Layout, messageLength));  
    }  
    //填充header头部requestID字段  
    void setRequestMsgId(int32_t value) {  
        data().write(tagLittleEndian(value), offsetof(Layout, requestID));  
    }  
    //填充header头部responseTo字段  
    void setResponseToMsgId(int32_t value) {  
        data().write(tagLittleEndian(value), offsetof(Layout, responseTo));  
    }  
    //填充header头部opCode字段  
    void setOpCode(int32_t value) {  
        data().write(tagLittleEndian(value), offsetof(Layout, opCode));  
    }  
private:  
    //指向header起始地址  
    view_type data() const {  
        return const_cast<char*>(ConstView::view2ptr());  
    }  
};  
}

mespace MsgData {  
ruct Layout {  
  //数据填充组成：header部分  
  MSGHEADER::Layout header;  
  //数据填充组成: body部分,body先用data占位置  
  char data[4];  
  

解析header字段信息及body其实地址信息  
ass ConstView {  
blic:  
  //初始化构造  
  ConstView(const char* storage) : _storage(storage) {}  
  //获取数据起始地址  
  const char* view2ptr() const {  
      return storage().view();  
  }  

  //以下四个接口间接执行前面的MSGHEADER中的头部字段解析  
  //填充header头部messageLength字段  
  int32_t getLen() const {  
      return header().getMessageLength();  
  }  
  //填充header头部requestID字段  
  int32_t getId() const {  
      return header().getRequestMsgId();  
  }  
  //填充header头部responseTo字段  
  int32_t getResponseToMsgId() const {  
      return header().getResponseToMsgId();  
  }  
  //获取网络数据报文中的opCode字段  
  NetworkOp getNetworkOp() const {  
      return NetworkOp(header().getOpCode());  
  }  
  //指向body起始地址  
  const char* data() const {  
      return storage().view(offsetof(Layout, data));  
  }  
  //messageLength长度检查，opcode检查  
  bool valid() const {  
      if (getLen() <= 0 || getLen() > (4 * BSONObjMaxInternalSize))  
          return false;  
      if (getNetworkOp() < 0 || getNetworkOp() > 30000)  
          return false;  
      return true;  
  }  
  ......  
otected:  
  //获取_storage  
  const ConstDataView& storage() const {  
      return _storage;  
  }  
  //指向header起始地址  
  MSGHEADER::ConstView header() const {  
      return storage().view(offsetof(Layout, header));  
  }  
ivate:  
  //MongoDB报文存储在这里  
  ConstDataView _storage;  
  

填充数据，包括Header和body  
ass View : public ConstView {  
blic:  
  //构造初始化  
  View(char* storage) : ConstView(storage) {}  
  ......  
  //获取报文起始地址  
  char* view2ptr() {  
      return storage().view();  
  }  

  //以下四个接口间接执行前面的MSGHEADER中的头部字段构造  
  //以下四个接口完成msg header赋值  
  //填充header头部messageLength字段  
  void setLen(int value) {  
      return header().setMessageLength(value);  
  }  
  //填充header头部messageLength字段  
  void setId(int32_t value) {  
      return header().setRequestMsgId(value);  
  }  
  //填充header头部messageLength字段  
  void setResponseToMsgId(int32_t value) {  
      return header().setResponseToMsgId(value);  
  }  
  //填充header头部messageLength字段  
  void setOperation(int value) {  
      return header().setOpCode(value);  
  }  

  using ConstView::data;  
  //指向data  
  char* data() {  
      return storage().view(offsetof(Layout, data));  
  }  
ivate:  
  //也就是报文起始地址  
    DataView storage() const {  
        return const_cast<char*>(ConstView::view2ptr());  
    }  
    //指向header头部  
    MSGHEADER::View header() const {  
        return storage().view(offsetof(Layout, header));  
    }  
};  
  
......  
//Value为前面的Layout，减4是因为有4字节填充data，所以这个就是header长度  
const int MsgDataHeaderSize = sizeof(Value) - 4;  
  
//除去头部后的数据部分长度  
inline int ConstView::dataLen() const {   
    return getLen() - MsgDataHeaderSize;  
}  
}  // namespace MsgData  

//DbMessage._msg成员为该类型  
class Message {  
public:  
    //message初始化  
    explicit Message(SharedBuffer data) : _buf(std::move(data)) {}  
    //头部header数据  
    MsgData::View header() const {  
        verify(!empty());  
        return _buf.get();  
    }  
    //获取网络数据报文中的op字段  
    NetworkOp operation() const {  
        return header().getNetworkOp();  
    }  
    //_buf释放为空  
    bool empty() const {  
        return !_buf;  
    }  
    //获取报文总长度messageLength  
    int size() const {  
        if (_buf) {  
            return MsgData::ConstView(_buf.get()).getLen();  
        }  
        return 0;  
    }  
    //body长度  
    int dataSize() const {  
        return size() - sizeof(MSGHEADER::Value);  
    }  
    //buf重置  
    void reset() {  
        _buf = {};  
    }  
    // use to set first buffer if empty  
    //_buf直接使用buf空间  
    void setData(SharedBuffer buf) {  
        verify(empty());  
        _buf = std::move(buf);  
    }  
     //把msgtxt拷贝到_buf中  
    void setData(int operation, const char* msgtxt) {  
        setData(operation, msgtxt, strlen(msgtxt) + 1);  
    }  
    //根据operation和msgdata构造一个完整MongoDB报文  
    void setData(int operation, const char* msgdata, size_t len) {  
        verify(empty());  
        size_t dataLen = len + sizeof(MsgData::Value) - 4;  
        _buf = SharedBuffer::allocate(dataLen);  
        MsgData::View d = _buf.get();  
        if (len)  
            memcpy(d.data(), msgdata, len);  
        d.setLen(dataLen);  
        d.setOperation(operation);  
    }  
    ......  
    //获取_buf对应指针  
    const char* buf() const {  
        return _buf.get();  
    }  
  
private:  
    //存放接收数据的buf  
    SharedBuffer _buf;  
};  

class DbMessage {  
    ......  
    //包含Message成员变量  
    const Message& _msg;  
    //MongoDB报文起始地址
    const char* _nsStart; 
    //报文结束地址
    const char* _theEnd; 
}  
  
DbMessage::DbMessage(const Message& msg) : _msg(msg),   
  _nsStart(NULL), _mark(NULL), _nsLen(0) {  
    //一个MongoDB报文(header+body)数据的结束地址  
    _theEnd = _msg.singleData().data() + _msg.singleData().dataLen();  
    //报文起始地址 [_nextjsobj, _theEnd ]之间的数据就是一个完整MongoDB报文  
    _nextjsobj = _msg.singleData().data();  
    ......  
}

struct OpMsg {   
      ......  
    //msg解析赋值见OpMsg::parse     
    //各种命令(insert update find等)都存放在该body中  
    BSONObj body;    
    //sequences用法暂时没看懂，感觉没什么用？先跳过  
    std::vector<DocumentSequence> sequences; //赋值见OpMsg::parse  
}  

//从message中解析出OpMsg信息  
OpMsg OpMsg::parse(const Message& message) try {  
    //message不能为空，并且opCode必须为dbMsg  
    invariant(!message.empty());  
    invariant(message.operation() == dbMsg);  
    //获取flagBits  
    const uint32_t flags = OpMsg::flags(message);  
    //flagBits有效性检查,bit 0-15中只能对第0和第1位操作  
    uassert(ErrorCodes::IllegalOpMsgFlag,  
            str::stream() << "Message contains illegal flags value: Ob"  
                          << std::bitset<32>(flags).to_string(),  
            !containsUnknownRequiredFlags(flags));  
  
    //校验码默认4字节  
    constexpr int kCrc32Size = 4;  
    //判断该mongo报文body内容是否启用了校验功能  
    const bool haveChecksum = flags & kChecksumPresent;  
    //如果有启用校验功能，则报文末尾4字节为校验码  
    const int checksumSize = haveChecksum ? kCrc32Size : 0;  
    //sections字段内容  
    BufReader sectionsBuf(message.singleData().data() + sizeof(flags),  
                          message.dataSize() - sizeof(flags) - checksumSize);  
  
    //默认先设置位false  
    bool haveBody = false;  
    OpMsg msg;  
    //解析sections对应命令请求数据  
    while (!sectionsBuf.atEof()) {  
        //BufReader::read读取kind内容，一个字节  
        const auto sectionKind = sectionsBuf.read<Section>();  
        //kind为0对应命令请求body内容，内容通过bson报错  
        switch (sectionKind) {  
            //sections第一个字节是0说明是body  
            case Section::kBody: {  
                //默认只能有一个body  
                uassert(40430, "Multiple body sections in message", !haveBody);  
                haveBody = true;  
                //命令请求的bson信息保存在这里  
                msg.body = sectionsBuf.read<Validated<BSONObj>>();  
                break;  
            }  
  
            //DocSequence暂时没看明白，用到的地方很少，跳过，后续等  
            //该系列文章主流功能分析完成后，从头再回首分析  
            case Section::kDocSequence: {  
                  ......  
            }  
        }  
    }  
    //OP_MSG必须有body内容  
    uassert(40587, "OP_MSG messages must have a body", haveBody);  
    //body和sequence去重判断  
    for (const auto& docSeq : msg.sequences) {  
        ......  
    }  
    return msg;  
}

//协议解析得时候会用到，见runCommands  
struct OpMsgRequest : public OpMsg {  
    ......  
    //构造初始化  
    explicit OpMsgRequest(OpMsg&& generic) : OpMsg(std::move(generic)) {}  
    //opMsgRequestFromAnyProtocol->OpMsgRequest::parse   
    //从message中解析出OpMsg所需成员信息  
    static OpMsgRequest parse(const Message& message) {  
        //OpMsg::parse  
        return OpMsgRequest(OpMsg::parse(message));  
    }  
    //根据db body extraFields填充OpMsgRequest  
    static OpMsgRequest fromDBAndBody(... {  
        OpMsgRequest request;  
        request.body = ([&] {  
            //填充request.body  
            ......  
        }());  
        return request;  
    }  
    //从body中获取db name  
    StringData getDatabase() const {  
        if (auto elem = body["$db"])  
            return elem.checkAndGetStringData();  
        uasserted(40571, "OP_MSG requests require a $db argument");  
    }  
    //find  insert 等命令信息  body中的第一个elem就是command 名  
    StringData getCommandName() const {  
        return body.firstElementFieldName();  
    }  
};

//mongod服务对于客户端请求的处理    
//通过状态机SSM模块的如下接口调用:ServiceStateMachine::_processMessage  
DbResponse ServiceEntryPointMongod::handleRequest(OperationContext* opCtx, const Message& m) {  
    //获取opCode，3.6版本对应客户端默认使用OP_MSG  
    NetworkOp op = m.operation();   
    ......  
    //根据message构造DbMessage  
    DbMessage dbmsg(m);  
    //根据操作上下文获取对应的client  
    Client& c = *opCtx->getClient();    
    ......  
    //获取库.表信息，注意只有dbUpdate<opCode<dbDelete的opCode请求才通过dbmsg直接获取库和表信息  
    const char* ns = dbmsg.messageShouldHaveNs() ? dbmsg.getns() : NULL;  
    const NamespaceString nsString = ns ? NamespaceString(ns) : NamespaceString();  
    ....  
    //CurOp::debug 初始化opDebug，慢日志相关记录  
    OpDebug& debug = currentOp.debug();  
    //慢日志阀值  
    long long logThresholdMs = serverGlobalParams.slowMS;  
    //时MongoDB将记录这次慢操作,1为只记录慢操作,即操作时间大于了设置的配置,2表示记录所有操作    
    bool shouldLogOpDebug = shouldLog(logger::LogSeverity::Debug(1));  
    DbResponse dbresponse;  
    if (op == dbMsg || op == dbCommand || (op == dbQuery && isCommand)) {  
        //新版本op=dbMsg，因此走这里  
        //从DB获取数据，获取到的数据通过dbresponse返回  
        dbresponse = runCommands(opCtx, m);     
    } else if (op == dbQuery) {  
        ......   
        //早期MongoDB版本查询走这里  
        dbresponse = receivedQuery(opCtx, nsString, c, m);  
    } else if (op == dbGetMore) {    
        //早期MongoDB版本查询走这里  
        dbresponse = receivedGetMore(opCtx, m, currentOp, &shouldLogOpDebug);  
    } else {  
        ......  
        //早期版本增 删 改走这里处理  
         if (op == dbInsert) {  
              receivedInsert(opCtx, nsString, m); //插入操作入口   新版本CmdInsert::runImpl  
         } else if (op == dbUpdate) {  
              receivedUpdate(opCtx, nsString, m); //更新操作入口    
         } else if (op == dbDelete) {  
              receivedDelete(opCtx, nsString, m); //删除操作入口    
         }   
    }  
    //获取runCommands执行时间，也就是内部处理时间  
    debug.executionTimeMicros = durationCount<Microseconds>(currentOp.elapsedTimeExcludingPauses());  
    ......  
    //慢日志记录  
    if (shouldLogOpDebug || (shouldSample && debug.executionTimeMicros > logThresholdMs * 1000LL)) {  
        Locker::LockerInfo lockerInfo;    
        //OperationContext::lockState  LockerImpl<>::getLockerInfo  
        opCtx->lockState()->getLockerInfo(&lockerInfo);   
  
    //OpDebug::report 记录慢日志到日志文件  
        log() << debug.report(&c, currentOp, lockerInfo.stats);   
    }  
    //各种统计信息  
    recordCurOpMetrics(opCtx);  
}

//message解析出对应command执行  
DbResponse runCommands(OperationContext* opCtx, const Message& message) {  
    //获取message对应的ReplyBuilder，3.6默认对应OpMsgReplyBuilder  
    //应答数据通过该类构造  
    auto replyBuilder = rpc::makeReplyBuilder(rpc::protocolForMessage(message));  
    [&] {  
        OpMsgRequest request;  
        try {  // Parse.  
            //协议解析 根据message获取对应OpMsgRequest  
            request = rpc::opMsgRequestFromAnyProtocol(message);  
        }   
    }  
    try {  // Execute.  
        //opCtx初始化  
        curOpCommandSetup(opCtx, request);  
        //command初始化为Null  
        Command* c = nullptr;  
        //OpMsgRequest::getCommandName查找  
        if (!(c = Command::findCommand(request.getCommandName()))) {   
             //没有找到相应的command的后续异常处理  
             ......  
        }  
        //执行command命令，获取到的数据通过replyBuilder.get()返回  
        execCommandDatabase(opCtx, c, request, replyBuilder.get());  
    }  
    //OpMsgReplyBuilder::done对数据进行序列化操作  
    auto response = replyBuilder->done();  
    //responseLength赋值  
    CurOp::get(opCtx)->debug().responseLength = response.header().dataLen();  
    // 返回  
    return DbResponse{std::move(response)};  
}  

void execCommandDatabase(...) {  
    ......  
    //获取dbname  
    const auto dbname = request.getDatabase().toString();  
    ......  
    //mab表存放从bson中解析出的elem信息  
    StringMap<int> topLevelFields;  
    //body elem解析  
    for (auto&& element : request.body) {  
        //获取bson中的elem信息  
        StringData fieldName = element.fieldNameStringData();  
        //如果elem信息重复，则异常处理  
        ......  
    }  
    //如果是help命令，则给出help提示  
    if (Command::isHelpRequest(helpField)) {  
        //给出help提示  
        Command::generateHelpResponse(opCtx, replyBuilder, *command);  
        return;  
    }  
    //权限认证检查，检查该命令执行权限  
    uassertStatusOK(Command::checkAuthorization(command, opCtx, request));  
    ......  
  
    //该命令执行次数统计  db.serverStatus().metrics.commands可以获取统计信息  
    command->incrementCommandsExecuted();  
    //真正的命令执行在这里面  
    retval = runCommandImpl(opCtx, command, request, replyBuilder, startOperationTime);  
    //该命令执行失败次数统计  
    if (!retval) {  
        command->incrementCommandsFailed();  
     }  
     ......  
}

bool runCommandImpl(...) {  
    //获取命令请求内容body  
    BSONObj cmd = request.body;  
    //获取请求中的DB库信息  
    const std::string db = request.getDatabase().toString();  
    //ReadConcern检查  
    Status rcStatus = waitForReadConcern(  
        opCtx, repl::ReadConcernArgs::get(opCtx), command->allowsAfterClusterTime(cmd));  
    //ReadConcern检查不通过，直接异常提示处理  
    if (!rcStatus.isOK()) {  
         //异常处理  
         return;  
    }  
    if (!command->supportsWriteConcern(cmd)) {  
        //命令不支持WriteConcern，但是对应的请求中却带有WriteConcern配置，直接报错不支持  
        if (commandSpecifiesWriteConcern(cmd)) {  
            //异常处理"Command does not support writeConcern"  
            ......  
            return result;  
        }  
    //调用Command::publicRun执行不同命令操作  
        result = command->publicRun(opCtx, request, inPlaceReplyBob);  
    }  
    //提取WriteConcernOptions信息  
    auto wcResult = extractWriteConcern(opCtx, cmd, db);  
    //提取异常，直接异常处理  
    if (!wcResult.isOK()) {  
        //异常处理  
        ......  
        return result;  
    }  
    ......  
    //执行对应的命令Command::publicRun，执行不同命令操作  
    result = command->publicRun(opCtx, request, inPlaceReplyBob);  
    ......  
}

void runCommand(...) {  
    ......  
    //获取请求命令name  
    auto const commandName = request.getCommandName();  
    //从全局map表中查找  
    auto const command = Command::findCommand(commandName);  
    //没有对应的command存在,抛异常说明不支持该命令  
    if (!command) {   
        ......  
        return;  
    }   
    ......  
    //执行命令  
    execCommandClient(opCtx, command, request, builder);   
    ......  
}  

void execCommandClient(...)  
{   
    ......  
    //认证检查，是否有操作该command命令的权限，没有则异常提示  
    Status status = Command::checkAuthorization(c, opCtx, request);    
    if (!status.isOK()) {  
        Command::appendCommandStatus(result, status);  
        return;  
    }  
    //该命令的执行次数自增，代理上面也是要计数的  
    c->incrementCommandsExecuted();   
    //如果需要command统计，则加1  
    if (c->shouldAffectCommandCounter()) {  
        globalOpCounters.gotCommand();  
    }  
    ......  
    //有部分命令不支持writeconcern配置，报错  
    bool supportsWriteConcern = c->supportsWriteConcern(request.body);  
    //不支持writeconcern又带有该参数的请求，直接异常处理"Command does not support writeConcern"  
    if (!supportsWriteConcern && !wcResult.getValue().usedDefault) {  
        ......  
        return;  
    }  
    //执行本命令对应的公共publicRun接口，Command::publicRun  
    ok = c->publicRun(opCtx, request, result);   
    ......  
}