mongodb 源码实现系列 - command命令处理模块源码实现一

文章来源：infoq杨亚洲(专注mongodb及高性能中间件)

关于作者

前滴滴出行技术专家，现任OPPO文档数据库mongodb负责人，负责oppo千万级峰值TPS/十万亿级数据量文档数据库mongodb内核研发及运维工作，一直专注于分布式缓存、高性能服务端、数据库、中间件等相关研发。后续持续分享《MongoDB内核源码设计、性能优化、最佳运维实践》，Github账号地址:https://github.com/y123456yz

1.?背景

???? <<transport_layer网络传输层模块源码实现>>中分享了mongodb内核底层网络IO处理相关实现，包括套接字初始化、一个完整mongodb报文的读取、获取到DB数据发送给客户端等。Mongodb支持多种增、删、改、查、聚合处理、cluster处理等操作，每个操作在内核实现中对应一个command，每个command有不同的功能，mongodb内核如何进行command源码处理将是本文分析的重点

此外，mongodb提供了mongostat工具来监控当前集群的各种操作统计。Mongostat监控统计如下图所示：

其中，insert、delete、update、query这四项统计比较好理解，分别对应增、删、改、查。但是，comand、getmore不是很好理解，command代表什么统计？getMore代表什么统计？，这两项相对比较难理解。

此外，通过本文字分析，我们将搞明白这六项统计的具体含义，同时弄清这六项统计由那些操作进行计数。

Command命令处理模块分为：mongos操作命令、mongod操作命令、mongodb集群内部命令，具体定义如下:

①　mongos操作命令，客户端可以通过mongos访问集群相关的命令。

②　mongod操作命令：客户端可以通过mongod复制集和cfg server访问集群的相关命令。

③　mongodb集群内部命令：mongos、mongod、mongo-cfg集群实例之间交互的命令。

???? Command命令处理模块核心代码实现如下：

???? 《command命令处理模块源码实现》相关文章重点分析命令处理模块核心代码实现，也就是上面截图中的命令处理源码文件实现。

2.?<<transport_layer网络传输层模块源码实现>>衔接回顾

<<transport_layer网络传输层模块源码实现三>>一文中，我们对service_state_machine状态机调度子模块进行了分析，该模块中的dealTask任务进行mongodb内部业务逻辑处理，其核心实现如下：


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

上面的sep对应mongod或者mongos实例的服务入口实现，该seq成员分别在如下代码中初始化为ServiceEntryPointMongod和ServiceEntryPointMongod类实现。SSM状态机的_seq成员初始化赋值核心代码实现如下：


1.//mongos实例启动初始化??
2.static?ExitCode?runMongosServer()?{??
3.????......??
4.????//mongos实例对应sep为ServiceEntryPointMongos??
5.????auto?sep?=?stdx::make_unique<ServiceEntryPointMongos>(getGlobalServiceContext());??
6.????getGlobalServiceContext()->setServiceEntryPoint(std::move(sep));??
7.????......??
8.}??
9.??
10.//mongod实例启动初始化??
11.ExitCode?_initAndListen(int?listenPort)?{??
12.????......??
13.????//mongod实例对应sep为ServiceEntryPointMongod??
14.????serviceContext->setServiceEntryPoint(??
15.????????stdx::make_unique<ServiceEntryPointMongod>(serviceContext));??
16.????......??
17.}??
18.??
19.//SSM状态机初始化??
20.ServiceStateMachine::ServiceStateMachine(...)??
21.????:?_state{State::Created},??
22.??????//mongod和mongos实例的服务入口通过这里赋值给_seq成员变量??
23.??????_sep{svcContext->getServiceEntryPoint()},??
24.??????......??
25.}?

通过上面的几个核心接口把mongos和mongod实例的服务入口与状态机SSM(ServiceStateMachine)联系起来，最终和下面的command命令处理模块关联。

? dealTask进行一次mongodb请求的内部逻辑处理，该处理由_sep->handleRequest()接口实现。由于mongos和mongod服务入口分别由ServiceEntryPointMongos和ServiceEntryPointMongod两个类实现，因此dealTask也就演变为如下接口处理：

①　mongos实例：ServiceEntryPointMongos::handleRequest(...)

②　Mongod实例:：ServiceEntryPointMongod::handleRequest(...)

这两个接口入参都是OperationContext和Message，分别对应操作上下文、请求原始数据内容。下文会分析Message解析实现、OperationContext服务上下文实现将在后续章节分析。

Mongod和mongos实例服务入口类都继承自网络传输模块中的ServiceEntryPointImpl类，如下图所示：

Tips: mongos和mongod服务入口类为何要继承网络传输模块服务入口类？

原因是一个请求对应一个链接session，该session对应的请求又和SSM状态机唯一对应。所有客户端请求对应的SSM状态机信息全部保存再ServiceEntryPointImpl._sessions成员中，而command命令处理模块为SSM状态机任务中的dealTask任务，通过该继承关系，ServiceEntryPointMongod和ServiceEntryPointMongos子类也就可以和状态机及任务处理关联起来，同时也可以获取当前请求对应的session链接信息。

3.?Mongodb协议解析

在《transport_layer网络传输层模块源码实现二》中的数据收发子模块完成了一个完整mongodb报文的接收，一个mongodb报文由Header头部+opCode包体组成，如下图所示：

上图中各个字段说明如下表：

opCode取值比较多，早期版本中OPINSERT、OPDELETE、OPUPDATE、OPQUERY分别针对增删改查请求，Mongodb从3.6版本开始默认使用OPMSG操作作为默认opCode，是一种可扩展的消息格式，旨在包含其他操作码的功能，新版本读写请求协议都对应该操作码。本文以OPMSG操作码对应协议为例进行分析，其他操作码协议分析过程类似，OP_MSG请求协议格式如下：


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

OP_MSG各个字段说明如下表：

一个完整OP_MSG请求格式如下：

除了通用头部header外，客户端命令请求实际上都保存于sections字段中，该字段存放的是请求的原始bson格式数据。BSON是由10gen开发的一个数据格式，目前主要用于MongoDB中，是MongoDB的数据存储格式。BSON基于JSON格式，选择JSON进行改造的原因主要是JSON的通用性及JSON的schemaless的特性。BSON相比JSON具有以下特性：

①　Lightweight(更轻量级)

②　Traversable(易操作)

③　Efficient(高效性能)

本文重点不是分析bson协议格式，bson协议实现细节将在后续章节分享。bson协议更多设计细节详见：http://bsonspec.org/

总结：一个完整mongodb报文由header+body组成，其中header长度固定为16字节，body长度等于messageLength-16。Header部分协议解析由message.cpp和message.h两源码文件实现，body部分对应的OP_MSG类请求解析由op_msg.cpp和op_msg.h两源码文件实现。

4. mongodb报文通用头部解析及封装源码实现

Header头部解析由src/mongo/util/net目录下message.cpp和message.h两文件完成，该类主要完成通用header头部和body部分的解析、封装。因此报文头部核心代码分为以下两类：

①　报文头部内容解析及封装(MSGHEADER命名空间实现)

②　头部和body内容解析及封装(MsgData命名空间实现)

4.1 mongodb报文头部解析及封装核心代码实现

mongodb报文头部解析由namespace MSGHEADER {...}实现，该类主要成员及接口实现如下：


1.namespace?MSGHEADER?{??
2.//header头部各个字段信息??
3.struct?Layout?{??
4.????//整个message长度，包括header长度和body长度??
5.????int32_t?messageLength;?????
6.????//requestID?该请求id信息??
7.????int32_t?requestID;?????????
8.????//getResponseToMsgId解析??
9.????int32_t?responseTo;????????
10.????//操作类型：OP_UPDATE、OP_INSERT、OP_QUERY、OP_DELETE、OP_MSG等??
11.????int32_t?opCode;??
12.};??
13.??
14.//ConstView实现header头部数据解析??
15.class?ConstView?{???
16.public:??
17.????......??
18.????//初始化构造??
19.????ConstView(const?char*?data)?:?_data(data)?{}??
20.????//获取_data地址??
21.????const?char*?view2ptr()?const?{??
22.????????return?data().view();??
23.????}??
24.????//TransportLayerASIO::ASIOSourceTicket::_headerCallback调用??
25.????//解析header头部的messageLength字段??
26.????int32_t?getMessageLength()?const?{??
27.????????return?data().read<LittleEndian<int32_t>>(offsetof(Layout,?messageLength));??
28.????}??
29.????//解析header头部的requestID字段??
30.????int32_t?getRequestMsgId()?const?{??
31.????????return?data().read<LittleEndian<int32_t>>(offsetof(Layout,?requestID));??
32.????}??
33.????//解析header头部的getResponseToMsgId字段??
34.????int32_t?getResponseToMsgId()?const?{??
35.????????return?data().read<LittleEndian<int32_t>>(offsetof(Layout,?responseTo));??
36.????}??
37.????//解析header头部的opCode字段??
38.????int32_t?getOpCode()?const?{??
39.????????return?data().read<LittleEndian<int32_t>>(offsetof(Layout,?opCode));??
40.????}??
41.??
42.protected:??
43.????//mongodb报文数据起始地址??
44.????const?view_type&?data()?const?{??
45.????????return?_data;??
46.????}??
47.private:??
48.????//数据部分??
49.????view_type?_data;??
50.};??
51.??
52.//View填充header头部数据??
53.class?View?:?public?ConstView?{??
54.public:??
55.????......??
56.????//构造初始化??
57.????View(char*?data)?:?ConstView(data)?{}??
58.????//header起始地址??
59.????char*?view2ptr()?{??
60.????????return?data().view();??
61.????}??
62.????//以下四个接口进行header填充??
63.????//填充header头部messageLength字段??
64.????void?setMessageLength(int32_t?value)?{??
65.????????data().write(tagLittleEndian(value),?offsetof(Layout,?messageLength));??
66.????}??
67.????//填充header头部requestID字段??
68.????void?setRequestMsgId(int32_t?value)?{??
69.????????data().write(tagLittleEndian(value),?offsetof(Layout,?requestID));??
70.????}??
71.????//填充header头部responseTo字段??
72.????void?setResponseToMsgId(int32_t?value)?{??
73.????????data().write(tagLittleEndian(value),?offsetof(Layout,?responseTo));??
74.????}??
75.????//填充header头部opCode字段??
76.????void?setOpCode(int32_t?value)?{??
77.????????data().write(tagLittleEndian(value),?offsetof(Layout,?opCode));??
78.????}??
79.private:??
80.????//指向header起始地址??
81.????view_type?data()?const?{??
82.????????return?const_cast<char*>(ConstView::view2ptr());??
83.????}??
84.};??
85.}

从上面的header头部解析、填充的实现类可以看出，header头部解析由MSGHEADER::ConstView实现；header头部填充由MSGHEADER::View完成。实际上代码实现上，通过offsetof来进行移位，从而快速定位到头部对应字段。

4.2 mongodb报文头部+body解析封装核心代码实现

Namespace MSGHEADER{...}命名空间只负责header头部的处理，namespace MsgData{...}命名空间相对MSGHEADER命名空间更加完善，除了处理头部解析封装外，还负责body数据起始地址维护、body数据封装、数据长度检查等。MsgData命名空间核心代码实现如下：


1.namespace?MsgData?{??
2.struct?Layout?{??
3.????//数据填充组成：header部分??
4.????MSGHEADER::Layout?header;??
5.????//数据填充组成:?body部分,body先用data占位置??
6.????char?data[4];??
7.};??
8.??
9.//解析header字段信息及body其实地址信息??
10.class?ConstView?{??
11.public:??
12.????//初始化构造??
13.????ConstView(const?char*?storage)?:?_storage(storage)?{}??
14.????//获取数据起始地址??
15.????const?char*?view2ptr()?const?{??
16.????????return?storage().view();??
17.????}??
18.??
19.????//以下四个接口间接执行前面的MSGHEADER中的头部字段解析??
20.????//填充header头部messageLength字段??
21.????int32_t?getLen()?const?{??
22.????????return?header().getMessageLength();??
23.????}??
24.????//填充header头部requestID字段??
25.????int32_t?getId()?const?{??
26.????????return?header().getRequestMsgId();??
27.????}??
28.????//填充header头部responseTo字段??
29.????int32_t?getResponseToMsgId()?const?{??
30.????????return?header().getResponseToMsgId();??
31.????}??
32.????//获取网络数据报文中的opCode字段??
33.????NetworkOp?getNetworkOp()?const?{??
34.????????return?NetworkOp(header().getOpCode());??
35.????}??
36.????//指向body起始地址??
37.????const?char*?data()?const?{??
38.????????return?storage().view(offsetof(Layout,?data));??
39.????}??
40.????//messageLength长度检查，opcode检查??
41.????bool?valid()?const?{??
42.????????if?(getLen()?<=?0?||?getLen()?>?(4?*?BSONObjMaxInternalSize))??
43.????????????return?false;??
44.????????if?(getNetworkOp()?<?0?||?getNetworkOp()?>?30000)??
45.????????????return?false;??
46.????????return?true;??
47.????}??
48.????......??
49.protected:??
50.????//获取_storage??
51.????const?ConstDataView&?storage()?const?{??
52.????????return?_storage;??
53.????}??
54.????//指向header起始地址??
55.????MSGHEADER::ConstView?header()?const?{??
56.????????return?storage().view(offsetof(Layout,?header));??
57.????}??
58.private:??
59.????//mongodb报文存储在这里??
60.????ConstDataView?_storage;??
61.};??
62.??
63.//填充数据，包括Header和body??
64.class?View?:?public?ConstView?{??
65.public:??
66.????//构造初始化??
67.????View(char*?storage)?:?ConstView(storage)?{}??
68.????......??
69.????//获取报文起始地址??
70.????char*?view2ptr()?{??
71.????????return?storage().view();??
72.????}??
73.??
74.????//以下四个接口间接执行前面的MSGHEADER中的头部字段构造??
75.????//以下四个接口完成msg?header赋值??
76.????//填充header头部messageLength字段??
77.????void?setLen(int?value)?{??
78.????????return?header().setMessageLength(value);??
79.????}??
80.????//填充header头部messageLength字段??
81.????void?setId(int32_t?value)?{??
82.????????return?header().setRequestMsgId(value);??
83.????}??
84.????//填充header头部messageLength字段??
85.????void?setResponseToMsgId(int32_t?value)?{??
86.????????return?header().setResponseToMsgId(value);??
87.????}??
88.????//填充header头部messageLength字段??
89.????void?setOperation(int?value)?{??
90.????????return?header().setOpCode(value);??
91.????}??
92.??
93.????using?ConstView::data;??
94.????//指向data??
95.????char*?data()?{??
96.????????return?storage().view(offsetof(Layout,?data));??
97.????}??
98.private:??
99.????//也就是报文起始地址??
100.????DataView?storage()?const?{??
101.????????return?const_cast<char*>(ConstView::view2ptr());??
102.????}??
103.????//指向header头部??
104.????MSGHEADER::View?header()?const?{??
105.????????return?storage().view(offsetof(Layout,?header));??
106.????}??
107.};??
108.??
109.......??
110.//Value为前面的Layout，减4是因为有4字节填充data，所以这个就是header长度??
111.const?int?MsgDataHeaderSize?=?sizeof(Value)?-?4;??
112.??
113.//除去头部后的数据部分长度??
114.inline?int?ConstView::dataLen()?const?{???
115.????return?getLen()?-?MsgDataHeaderSize;??
116.}??
117.}??//?namespace?MsgData??

???? 和MSGHEADER命名空间相比，MsgData这个namespace命名空间接口实现和前面的MSGHEADER命名空间实现大同小异。MsgData不仅仅处理header头部的解析组装，还负责body部分数据头部指针指向、头部长度检查、opCode检查、数据填充等。其中，MsgData命名空间中header头部的解析构造底层依赖MSGHEADER实现。

4.3 Message/DbMessage核心代码实现

在《transport_layer网络传输层模块源码实现二》中，从底层ASIO库接收到的mongodb报文是存放在Message结构中存储，最终存放在ServiceStateMachine._inMessage成员中。

在前面第2章我们知道mongod和mongso实例的服务入口接口handleRequest(...)中都带有Message入参，也就是接收到的Message数据通过该接口处理。Message类主要接口实现如下：


1.//DbMessage._msg成员为该类型??
2.class?Message?{??
3.public:??
4.????//message初始化??
5.????explicit?Message(SharedBuffer?data)?:?_buf(std::move(data))?{}??
6.????//头部header数据??
7.????MsgData::View?header()?const?{??
8.????????verify(!empty());??
9.????????return?_buf.get();??
10.????}??
11.????//获取网络数据报文中的op字段??
12.????NetworkOp?operation()?const?{??
13.????????return?header().getNetworkOp();??
14.????}??
15.????//_buf释放为空??
16.????bool?empty()?const?{??
17.????????return?!_buf;??
18.????}??
19.????//获取报文总长度messageLength??
20.????int?size()?const?{??
21.????????if?(_buf)?{??
22.????????????return?MsgData::ConstView(_buf.get()).getLen();??
23.????????}??
24.????????return?0;??
25.????}??
26.????//body长度??
27.????int?dataSize()?const?{??
28.????????return?size()?-?sizeof(MSGHEADER::Value);??
29.????}??
30.????//buf重置??
31.????void?reset()?{??
32.????????_buf?=?{};??
33.????}??
34.????//?use?to?set?first?buffer?if?empty??
35.????//_buf直接使用buf空间??
36.????void?setData(SharedBuffer?buf)?{??
37.????????verify(empty());??
38.????????_buf?=?std::move(buf);??
39.????}??
40.?????//把msgtxt拷贝到_buf中??
41.????void?setData(int?operation,?const?char*?msgtxt)?{??
42.????????setData(operation,?msgtxt,?strlen(msgtxt)?+?1);??
43.????}??
44.????//根据operation和msgdata构造一个完整mongodb报文??
45.????void?setData(int?operation,?const?char*?msgdata,?size_t?len)?{??
46.????????verify(empty());??
47.????????size_t?dataLen?=?len?+?sizeof(MsgData::Value)?-?4;??
48.????????_buf?=?SharedBuffer::allocate(dataLen);??
49.????????MsgData::View?d?=?_buf.get();??
50.????????if?(len)??
51.????????????memcpy(d.data(),?msgdata,?len);??
52.????????d.setLen(dataLen);??
53.????????d.setOperation(operation);??
54.????}??
55.????......??
56.????//获取_buf对应指针??
57.????const?char*?buf()?const?{??
58.????????return?_buf.get();??
59.????}??
60.??
61.private:??
62.????//存放接收数据的buf??
63.????SharedBuffer?_buf;??
64.};??

Message是操作mongodb收发报文最直接的实现类，该类主要完成一个完整mongodb报文封装。有关mongodb报文头后面的body更多的解析实现在DbMessage类中完成，DbMessage类包含Message类成员msg。实际上，Message报文信息在handleRequest(...)实例服务入口中赋值给DbMessage.msg，报文后续的body处理继续由DbMessage类相关接口完成处理。DbMessage和Message类关系如下:


1.class?DbMessage?{??
2.????......??
3.????//包含Message成员变量??
4.????const?Message&?_msg;??
5.    //mongodb报文起始地址
6.    const char* _nsStart; 
7.    //报文结束地址
8.    const char* _theEnd; 
9.}??
10.??
11.DbMessage::DbMessage(const?Message&?msg)?:?_msg(msg),???
12.??_nsStart(NULL),?_mark(NULL),?_nsLen(0)?{??
13.????//一个mongodb报文(header+body)数据的结束地址??
14.????_theEnd?=?_msg.singleData().data()?+?_msg.singleData().dataLen();??
15.????//报文起始地址?[_nextjsobj,?_theEnd?]之间的数据就是一个完整mongodb报文??
16.????_nextjsobj?=?_msg.singleData().data();??
17.????......??
}

DbMessage._msg成员为DbMessage?类型，DbMessage的_nsStart和_theEnd成员分别记录完整mongodb报文的起始地址和结束地址，通过这两个指针就可以获取一个完整mongodb报文的全部内容，包括header和body。

注意：DbMessage是早期mongodb版本(version<3.6)中用于报文body解析封装的类，这些类针对opCode=[dbUpdate, dbDelete]这个区间的操作。在mongodb新版本(version>=3.6)中，body解析及封装由op_msg.h和op_msg.cpp代码文件中的clase OpMsgRequest{}完成处理。

4.4 OpMsg报文解析封装核心代码实现

???? ?Mongodb从3.6版本开始默认使用OP_MSG操作作为默认opCode，是一种可扩展的消息格式，旨在包含其他操作码的功能，新版本读写请求协议都对应该操作码。OP_MSG对应mongodb报文body解析封装处理由OpMsg类相关接口完成，OpMsg::parse(Message)从Message中解析出报文body内容，其核心代码实现如下：


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


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

OpMsg类被OpMsgRequest类继承，OpMsgRequest类中核心接口就是解析出OpMsg.body中的库信息和表信息，OpMsgRequest类代码实现如下：


1.//协议解析得时候会用到，见runCommands??
2.struct?OpMsgRequest?:?public?OpMsg?{??
3.????......??
4.????//构造初始化??
5.????explicit?OpMsgRequest(OpMsg&&?generic)?:?OpMsg(std::move(generic))?{}??
6.????//opMsgRequestFromAnyProtocol->OpMsgRequest::parse???
7.????//从message中解析出OpMsg所需成员信息??
8.????static?OpMsgRequest?parse(const?Message&?message)?{??
9.????????//OpMsg::parse??
10.????????return?OpMsgRequest(OpMsg::parse(message));??
11.????}??
12.????//根据db?body?extraFields填充OpMsgRequest??
13.????static?OpMsgRequest?fromDBAndBody(...?{??
14.????????OpMsgRequest?request;??
15.????????request.body?=?([&]?{??
16.????????????//填充request.body??
17.????????????......??
18.????????}());??
19.????????return?request;??
20.????}??
21.????//从body中获取db?name??
22.????StringData?getDatabase()?const?{??
23.????????if?(auto?elem?=?body["$db"])??
24.????????????return?elem.checkAndGetStringData();??
25.????????uasserted(40571,?"OP_MSG?requests?require?a?$db?argument");??
26.????}??
27.????//find??insert?等命令信息??body中的第一个elem就是command?名??
28.????StringData?getCommandName()?const?{??
29.????????return?body.firstElementFieldName();??
30.????}??
};?

OpMsgRequest通过OpMsg::parse(message)解析出OpMsg信息，从而获取到body内容，GetCommandName()接口和getDatabase()则分别从body中获取库DB信息、命令名信息。通过该类相关接口，命令名(find、write、update等)和DB库都获取到了。

OpMsg模块除了OPMSG相关报文解析外，还负责OPMSG报文组装填充，该模块接口功能大全如下表：

5. Mongod实例服务入口核心代码实现

Mongod实例服务入口类ServiceEntryPointMongod继承ServiceEntryPointImpl类，mongod实例的报文解析处理、命令解析、命令执行都由该类负责处理。ServiceEntryPointMongod核心接口可以细分为：opCode解析及回调处理、命令解析及查找、命令执行三个子模块。

5.1 opCode解析及回调处理

???? OpCode操作码解析及其回调处理由ServiceEntryPointMongod::handleRequest(...)接口实现，核心代码实现如下:


1.//mongod服务对于客户端请求的处理????
2.//通过状态机SSM模块的如下接口调用:ServiceStateMachine::_processMessage??
3.DbResponse?ServiceEntryPointMongod::handleRequest(OperationContext*?opCtx,?const?Message&?m)?{??
4.????//获取opCode，3.6版本对应客户端默认使用OP_MSG??
5.????NetworkOp?op?=?m.operation();???
6.????......??
7.????//根据message构造DbMessage??
8.????DbMessage?dbmsg(m);??
9.????//根据操作上下文获取对应的client??
10.????Client&?c?=?*opCtx->getClient();????
11.????......??
12.????//获取库.表信息，注意只有dbUpdate<opCode<dbDelete的opCode请求才通过dbmsg直接获取库和表信息??
13.????const?char*?ns?=?dbmsg.messageShouldHaveNs()???dbmsg.getns()?:?NULL;??
14.????const?NamespaceString?nsString?=?ns???NamespaceString(ns)?:?NamespaceString();??
15.????....??
16.????//CurOp::debug?初始化opDebug，慢日志相关记录??
17.????OpDebug&?debug?=?currentOp.debug();??
18.????//慢日志阀值??
19.????long?long?logThresholdMs?=?serverGlobalParams.slowMS;??
20.????//时mongodb将记录这次慢操作,1为只记录慢操作,即操作时间大于了设置的配置,2表示记录所有操作????
21.????bool?shouldLogOpDebug?=?shouldLog(logger::LogSeverity::Debug(1));??
22.????DbResponse?dbresponse;??
23.????if?(op?==?dbMsg?||?op?==?dbCommand?||?(op?==?dbQuery?&&?isCommand))?{??
24.????????//新版本op=dbMsg，因此走这里??
25.????????//从DB获取数据，获取到的数据通过dbresponse返回??
26.????????dbresponse?=?runCommands(opCtx,?m);?????
27.????}?else?if?(op?==?dbQuery)?{??
28.????????......???
29.????????//早期mongodb版本查询走这里??
30.????????dbresponse?=?receivedQuery(opCtx,?nsString,?c,?m);??
31.????}?else?if?(op?==?dbGetMore)?{????
32.????????//早期mongodb版本查询走这里??
33.????????dbresponse?=?receivedGetMore(opCtx,?m,?currentOp,?&shouldLogOpDebug);??
34.????}?else?{??
35.????????......??
36.????????//早期版本增?删?改走这里处理??
37.?????????if?(op?==?dbInsert)?{??
38.??????????????receivedInsert(opCtx,?nsString,?m);?//插入操作入口???新版本CmdInsert::runImpl??
39.?????????}?else?if?(op?==?dbUpdate)?{??
40.??????????????receivedUpdate(opCtx,?nsString,?m);?//更新操作入口????
41.?????????}?else?if?(op?==?dbDelete)?{??
42.??????????????receivedDelete(opCtx,?nsString,?m);?//删除操作入口????
43.?????????}???
44.????}??
45.????//获取runCommands执行时间，也就是内部处理时间??
46.????debug.executionTimeMicros?=?durationCount<Microseconds>(currentOp.elapsedTimeExcludingPauses());??
47.????......??
48.????//慢日志记录??
49.????if?(shouldLogOpDebug?||?(shouldSample?&&?debug.executionTimeMicros?>?logThresholdMs?*?1000LL))?{??
50.????????Locker::LockerInfo?lockerInfo;????
51.????????//OperationContext::lockState??LockerImpl<>::getLockerInfo??
52.????????opCtx->lockState()->getLockerInfo(&lockerInfo);???
53.??
54.????//OpDebug::report?记录慢日志到日志文件??
55.????????log()?<<?debug.report(&c,?currentOp,?lockerInfo.stats);???
56.????}??
57.????//各种统计信息??
58.????recordCurOpMetrics(opCtx);??
59.}??

Mongod的handleRequest()接口主要完成以下工作：

①　从Message中获取OpCode，早期版本每个命令又对应取值，例如增删改查早期版本分别对应：dbInsert、dbDelete、dbUpdate、dbQuery；Mongodb 3.6开始，默认请求对应OpCode都是OP_MSG，本文默认只分析OpCode=OP_MSG相关的处理。

②　获取本操作对应的Client客户端信息。

③　如果是早期版本，通过Message构造DbMessage，同时解析出库.表信息。

④　根据不同OpCode执行对应回调操作，OP_MSG对应操作为runCommands(...)，获取的数据通过dbresponse返回。

⑤　获取到db层返回的数据后，进行慢日志判断，如果db层数据访问超过阀值，记录慢日志。

⑥　设置debug的各种统计信息。

5.2命令解析及查找

从上面的分析可以看出，接口最后调用runCommands(...)，该接口核心代码实现如下所示：


1.//message解析出对应command执行??
2.DbResponse?runCommands(OperationContext*?opCtx,?const?Message&?message)?{??
3.????//获取message对应的ReplyBuilder，3.6默认对应OpMsgReplyBuilder??
4.????//应答数据通过该类构造??
5.????auto?replyBuilder?=?rpc::makeReplyBuilder(rpc::protocolForMessage(message));??
6.????[&]?{??
7.????????OpMsgRequest?request;??
8.????????try?{??//?Parse.??
9.????????????//协议解析?根据message获取对应OpMsgRequest??
10.????????????request?=?rpc::opMsgRequestFromAnyProtocol(message);??
11.????????}???
12.????}??
13.????try?{??//?Execute.??
14.????????//opCtx初始化??
15.????????curOpCommandSetup(opCtx,?request);??
16.????????//command初始化为Null??
17.????????Command*?c?=?nullptr;??
18.????????//OpMsgRequest::getCommandName查找??
19.????????if?(!(c?=?Command::findCommand(request.getCommandName())))?{???
20.?????????????//没有找到相应的command的后续异常处理??
21.?????????????......??
22.????????}??
23.????????//执行command命令，获取到的数据通过replyBuilder.get()返回??
24.????????execCommandDatabase(opCtx,?c,?request,?replyBuilder.get());??
25.????}??
26.????//OpMsgReplyBuilder::done对数据进行序列化操作??
27.????auto?response?=?replyBuilder->done();??
28.????//responseLength赋值??
29.????CurOp::get(opCtx)->debug().responseLength?=?response.header().dataLen();??
30.????//?返回??
31.????return?DbResponse{std::move(response)};??
}?

RunCommands(...)接口从message中解析出OpMsg信息，然后获取该OpMsg对应的command命令信息，最后执行该命令对应的后续处理操作。主要功能说明如下：

①　获取该OpCode对应replyBuilder，OP_MSG操作对应builder为OpMsgReplyBuilder。

②　根据message解析出OpMsgRequest数据，OpMsgRequest来中包含了真正的命令请求bson信息。

③　opCtx初始化操作。

④　通过request.getCommandName()返回命令信息(如“find”、“update”等字符串)。

⑤　通过Command::findCommand(command name)从CommandMap这个map表中查找是否支持该 command命令。如果没找到说明不支持，如果找到说明支持。

⑥　调用execCommandDatabase(...)执行该命令，并获取命令的执行结果。

⑦　根据command执行结果构造response并返回

5.3命令执行


1.void?execCommandDatabase(...)?{??
2.????......??
3.????//获取dbname??
4.????const?auto?dbname?=?request.getDatabase().toString();??
5.????......??
6.????//mab表存放从bson中解析出的elem信息??
7.????StringMap<int>?topLevelFields;??
8.????//body?elem解析??
9.????for?(auto&&?element?:?request.body)?{??
10.????????//获取bson中的elem信息??
11.????????StringData?fieldName?=?element.fieldNameStringData();??
12.????????//如果elem信息重复，则异常处理??
13.????????......??
14.????}??
15.????//如果是help命令，则给出help提示??
16.????if?(Command::isHelpRequest(helpField))?{??
17.????????//给出help提示??
18.????????Command::generateHelpResponse(opCtx,?replyBuilder,?*command);??
19.????????return;??
20.????}??
21.????//权限认证检查，检查该命令执行权限??
22.????uassertStatusOK(Command::checkAuthorization(command,?opCtx,?request));??
23.????......??
24.??
25.????//该命令执行次数统计??db.serverStatus().metrics.commands可以获取统计信息??
26.????command->incrementCommandsExecuted();??
27.????//真正的命令执行在这里面??
28.????retval?=?runCommandImpl(opCtx,?command,?request,?replyBuilder,?startOperationTime);??
29.????//该命令执行失败次数统计??
30.????if?(!retval)?{??
31.????????command->incrementCommandsFailed();??
32.?????}??
33.?????......??
}

execCommandDatabase(...)最终调用RunCommandImpl(...)进行对应命令的真正处理，该接口核心代码实现如下：


1.bool?runCommandImpl(...)?{??
2.????//获取命令请求内容body??
3.????BSONObj?cmd?=?request.body;??
4.????//获取请求中的DB库信息??
5.????const?std::string?db?=?request.getDatabase().toString();??
6.????//ReadConcern检查??
7.????Status?rcStatus?=?waitForReadConcern(??
8.????????opCtx,?repl::ReadConcernArgs::get(opCtx),?command->allowsAfterClusterTime(cmd));??
9.????//ReadConcern检查不通过，直接异常提示处理??
10.????if?(!rcStatus.isOK())?{??
11.?????????//异常处理??
12.?????????return;??
13.????}??
14.????if?(!command->supportsWriteConcern(cmd))?{??
15.????????//命令不支持WriteConcern，但是对应的请求中却带有WriteConcern配置，直接报错不支持??
16.????????if?(commandSpecifiesWriteConcern(cmd))?{??
17.????????????//异常处理"Command?does?not?support?writeConcern"??
18.????????????......??
19.????????????return?result;??
20.????????}??
21.????//调用Command::publicRun执行不同命令操作??
22.????????result?=?command->publicRun(opCtx,?request,?inPlaceReplyBob);??
23.????}??
24.????//提取WriteConcernOptions信息??
25.????auto?wcResult?=?extractWriteConcern(opCtx,?cmd,?db);??
26.????//提取异常，直接异常处理??
27.????if?(!wcResult.isOK())?{??
28.????????//异常处理??
29.????????......??
30.????????return?result;??
31.????}??
32.????......??
33.????//执行对应的命令Command::publicRun，执行不同命令操作??
34.????result?=?command->publicRun(opCtx,?request,?inPlaceReplyBob);??
35.????......??
36.}

???? RunCommandImpl(...)接口最终调用该接口入参的command，执行?command->publicRun(...)接口，也就是命令模块的公共publicRun。

5.4总结

Mongod服务入口首先从message中解析出opCode操作码，3.6版本对应客户端默认操作码为OP_MSQ，解析出该操作对应OpMsgRequest信息。然后从message原始数据中解析出command命令字符串后，继续通过全局Map表种查找是否支持该命令操作，如果支持则执行该命令；如果不支持，直接异常打印，同时返回。

6. Mongos实例服务入口核心代码实现

????? ?mongos服务入口核心代码实现过程和mongod服务入口代码实现流程几乎相同，mongos实例message解析、OP_MSG操作码处理、command命令查找等流程和上一章节mongod实例处理过程类似，本章节不在详细分析。Mongos实例服务入口处理调用流程如下：

ServiceEntryPointMongos::handleRequest(...)->Strategy::clientCommand(...)-->runCommand(...)->execCommandClient(...)

最后的接口核心代码实现如下：


1.void?runCommand(...)?{??
2.????......??
3.????//获取请求命令name??
4.????auto?const?commandName?=?request.getCommandName();??
5.????//从全局map表中查找??
6.????auto?const?command?=?Command::findCommand(commandName);??
7.????//没有对应的command存在,抛异常说明不支持该命令??
8.????if?(!command)?{???
9.????????......??
10.????????return;??
11.????}???
12.????......??
13.????//执行命令??
14.????execCommandClient(opCtx,?command,?request,?builder);???
15.????......??
16.}??
17.
18.void?execCommandClient(...)??
19.{???
20.????......??
21.????//认证检查，是否有操作该command命令的权限，没有则异常提示??
22.????Status?status?=?Command::checkAuthorization(c,?opCtx,?request);????
23.????if?(!status.isOK())?{??
24.????????Command::appendCommandStatus(result,?status);??
25.????????return;??
26.????}??
27.????//该命令的执行次数自增，代理上面也是要计数的??
28.????c->incrementCommandsExecuted();???
29.????//如果需要command统计，则加1??
30.????if?(c->shouldAffectCommandCounter())?{??
31.????????globalOpCounters.gotCommand();??
32.????}??
33.????......??
34.????//有部分命令不支持writeconcern配置，报错??
35.????bool?supportsWriteConcern?=?c->supportsWriteConcern(request.body);??
36.????//不支持writeconcern又带有该参数的请求，直接异常处理"Command?does?not?support?writeConcern"??
37.????if?(!supportsWriteConcern?&&?!wcResult.getValue().usedDefault)?{??
38.????????......??
39.????????return;??
40.????}??
41.????//执行本命令对应的公共publicRun接口，Command::publicRun??
42.????ok?=?c->publicRun(opCtx,?request,?result);???
43.????......??
44.}??

Tips: mongos和mongod实例服务入口核心代码实现的一点小区别

①　Mongod实例opCode操作码解析、OpMsg解析、command查找及对应命令调用处理都由class ServiceEntryPointMongod{...}类一起完成。

②　mongos实例则把opCode操作码解析交由class ServiceEntryPointMongos{...}类实现，OpMsg解析、command查找及对应命令调用处理放到了clase Strategy{...}类来处理。

7.总结

?? Mongodb报文解析及组装流程总结

①　一个完整mongodb报文由通用报文header头部+body部分组成。

②　Body部分内容，根据报文头部的opCode来决定不同的body内容。

③　3.6版本对应客户端请求opCode默认为OP_MSG，该操作码对应body部分由flagBits + sections + checksum组成，其中sections中存放的是真正的命令请求信息，已bson数据格式保存。

④　Header头部和body报文体封装及解析过程由class Message {...}类实现

⑤　Body中对应command命令名、库名、表名的解析在mongodb(version<3.6)低版本协议中由class DbMessage {...}类实现

⑥　Body中对应command命令名、库名、表名的解析在mongodb(version<3.6)低版本协议中由struct OpMsgRequest{...}结构和struct OpMsg {...}类实现

?? Mongos和mongod实例的服务入口处理流程大同小异，整体处理流程如下：

①　从message解析出opCode操作码，根据不同操作码执行对应操作码回调。

②　根据message解析出OpMsg request信息，mongodb报文的命令信息就存储在该body中，该body已bson格式存储。

③　从body中解析出command命令字符串信息(如“insert”、“update”等)。

④　从全局_commands map表中查找是否支持该命令，如果支持则执行该命令处理，如果不支持则直接报错提示。

⑤　最终找到对应command命令后，执行command的功能run接口。

?? 图形化总结如下:

说明：第3章的协议解析及封装过程实际上应该算是网络处理模块范畴，本文为了分析command命令处理模块方便，把该部分实现归纳到了命令处理模块，这样方便理解。

Tips: 下期继续分享不同command命令执行细节。

8.遗留问题

???? 第1章节中的统计信息，将在command模块核心代码分析完毕后揭晓答案，《mongodb command命令处理模块源码实现二》中继续分析，敬请关注。

发表于: 2020-11-252020-11-25 19:55:18
本文为 InfoQ 中文站特供稿件
首发地址：https://www.infoq.cn/article/ed28fa03590e58fd88e987c82
如有侵权，请联系 cloudcommunity@tencent.com 删除。

扫码

添加站长进交流群

领取专属 10元无门槛券

私享最新 技术干货