SequoiaDB 系列之六：源码分析之coord节点

好久不见。

在上一篇SequoiaDB 系列之五：源码分析之main函数，有讲述进程开始运行时，会根据自身的角色，来初始化不同的CB（控制块，control block）。

在之前的一篇SequoiaDB 系列之四：架构简析中，我们简单过了一遍SequoiaDB的架构和各个节点的角色。

今天来看看SequoiaDB的coord角色。

首先，需要有个大致的轮廓：

coord节点主要承担代理的角色。作为SequoiaDB集群对外的接头人，它转发消息给其它节点，组合（combine）不同节点返回的数据，把结果返回给client。

catalog节点主要存储meta数据，比如集群中有哪些组，每个组的状态；每个组上有哪些节点，有哪些集合（Collection）,哪些集合是主子表等等。

data节点主要是管理存储的数据，它接受coord转发过来的CRUD等操作，并记录同步日志（最终一致性）。

在注册CB的函数中：

void _pmdController::registerCB( SDB_ROLE dbrole )
{if ( SDB_ROLE_DATA == dbrole ){...}else if ( SDB_ROLE_COORD == dbrole ){PMD_REGISTER_CB( sdbGetTransCB() ) ;      // TRANSPMD_REGISTER_CB( sdbGetCoordCB() ) ;      // COORDPMD_REGISTER_CB( sdbGetFMPCB () ) ;       // FMP}...// 每个节点都会注册的控制块PMD_REGISTER_CB( sdbGetDMSCB() ) ;           // DMSPMD_REGISTER_CB( sdbGetRTNCB() ) ;           // RTNPMD_REGISTER_CB( sdbGetSQLCB() ) ;           // SQLPMD_REGISTER_CB( sdbGetAggrCB() ) ;          // AGGRPMD_REGISTER_CB( sdbGetPMDController() ) ;   // CONTROLLER
}

coord注册这几个CB之后，就开始注册和启动服务：

具体函数在_KRCB::init()中，不再表述。_KRCB::init()会根据节点的角色，启动不同的服务。

客户端连接到coord，coord便会启动一个线程，为该连接服务。

 1 INT32 pmdTcpListenerEntryPoint ( pmdEDUCB *cb, void *pData )
 2 {
 3    ...
 4 
 5    while ( !cb->isDisconnected() && !pListerner->isClosed() )
 6    {
 7       SOCKET s ;
 8       rc = pListerner->accept ( &s, NULL, NULL ) ;
 9       if ( SDB_TIMEOUT == rc || SDB_TOO_MANY_OPEN_FD == rc )
10       {
11          rc = SDB_OK ;
12          continue ;
13       }
14       if ( rc && PMD_IS_DB_DOWN )
15       {
16          rc = SDB_OK ;
17          goto done ;
18       }
19       else if ( rc )
20       {
21          PD_LOG ( PDERROR, "Failed to accept socket in TcpListener(rc=%d)",
22                   rc ) ;
23          if ( pListerner->isClosed() )
24          {
25             break ;
26          }
27          else
28          {
29             continue ;
30          }
31       }
32 
33       cb->incEventCount() ;
34       ++mondbcb->numConnects ;
35       void *pData = NULL ;
36       *((SOCKET *) &pData) = s ;
37       if ( !krcb->isActive() )
38       {
39          ossSocket newsock ( &s ) ;
40          newsock.close () ;
41          continue ;
42       }
43 
44       rc = eduMgr->startEDU ( EDU_TYPE_AGENT, pData, &agentEDU ) ;
45       if ( rc )
46       {
47          PD_LOG( ( rc == SDB_QUIESCED ? PDWARNING : PDERROR ),
48                  "Failed to start edu, rc: %d", rc ) ;
49          ossSocket newsock ( &s ) ;
50          newsock.close () ;
51          continue ;
52       }
53    } //while ( ! cb->isDisconnected() )
54 
55    ...
56 }

服务线程监听到client的连接，启动一个EDU_TYPE_AGENT类型的线程，单独为client服务。

下面讲述coord节点的最主要的功能——消息转发

coord的启动初，会初始化一些必要的全局变量。在SequoiaDB中，会初始化很多command，拿创建集合空间来说，在文件SequoiaDB/engine/rtn/rtnCoord.cpp 中

1 RTN_COORD_CMD_BEGIN
2    ...
3 
4    RTN_COORD_CMD_ADD( COORD_CMD_LISTCOLLECTIONSPACES, rtnCoordCMDListCollectionSpace )
5 
6    ...
7 RTN_COORD_OP_END

嗯，上面的代码有点MFC中消息映射的感觉。

来看看 RTN_COORD_CMD_ADD 宏：

1 #define RTN_COORD_CMD_ADD( cmdName, cmdClass )  {\
2        rtnCoordCommand *pObj = SDB_OSS_NEW cmdClass();\
3        _cmdMap.insert ( COORD_CMD_MAP::value_type (cmdName, pObj ));}

宏主要是new一个对象，再把对象插入到_cmdMap中，这样在程序初始化时候，便会有一系列的command对象存储在_cmdMap中。

另外，对SequoiaDB而言，所有的command操作，都是在查询操作的基础上做的，服务端用一些方法区别开是真正的查询，还是command。SequoiaDB的命令，是以$开头的字符串。

前提简述完毕，现在假设client连接上了coord，coord也创建了一个线程，为这个client服务。

  1 INT32 _pmdLocalSession::run()
  2    {
  3       INT32 rc                = SDB_OK ;
  4       UINT32 msgSize          = 0 ;
  5       CHAR *pBuff             = NULL ;
  6       INT32 buffSize          = 0 ;
  7       pmdEDUMgr *pmdEDUMgr    = NULL ;
  8 
  9       if ( !_pEDUCB )
 10       {
 11          rc = SDB_SYS ;
 12          goto error ;
 13       }
 14 
 15       pmdEDUMgr = _pEDUCB->getEDUMgr() ;
 16 
 17       while ( !_pEDUCB->isDisconnected() && !_socket.isClosed() )
 18       {
 19          _pEDUCB->resetInterrupt() ;
 20          _pEDUCB->resetInfo( EDU_INFO_ERROR ) ;
 21          _pEDUCB->resetLsn() ;
 22 
 23          rc = recvData( (CHAR*)&msgSize, sizeof(UINT32) ) ;                         // 收取数据包的前四个字节，代表该数据包有多大
 24          if ( rc )
 25          {
 26             if ( SDB_APP_FORCED != rc )
 27             {
 28                PD_LOG( PDERROR, "Session[%s] failed to recv msg size, "
 29                        "rc: %d", sessionName(), rc ) ;
 30             }
 31             break ;
 32          }
 33 
 34          if ( msgSize == (UINT32)MSG_SYSTEM_INFO_LEN )  // 如果包长度是 MSG_SYSTEM_INFO_LEN （-1），则这是一个请求系统信息包，coord会返回本机的字节序列给client
 35          {                                              // 每个连接的第一个包，一定是长度标记为 MSG_SYSTEM_INFO_LEN 的包，否则字节序不正确，所有的数据都不能保证能正确解析（万一数据库运行在大端机上呢）
 36             rc = _recvSysInfoMsg( msgSize, &pBuff, buffSize ) ;
 37             if ( rc )
 38             {
 39                break ;
 40             }
 41             rc = _processSysInfoRequest( pBuff ) ;
 42             if ( rc )
 43             {
 44                break ;
 45             }
 46 
 47             _setHandshakeReceived() ;
 48          }
 49          else if ( msgSize < sizeof(MsgHeader) || msgSize > SDB_MAX_MSG_LENGTH )  // 对包的大小做出了限制，包长超过某值或者小于某值的包，都会导致连接中断
 50          {
 51             PD_LOG( PDERROR, "Session[%s] recv msg size[%d] is less than "
 52                     "MsgHeader size[%d] or more than max msg size[%d]",
 53                     sessionName(), msgSize, sizeof(MsgHeader),
 54                     SDB_MAX_MSG_LENGTH ) ;
 55             rc = SDB_INVALIDARG ;
 56             break ;
 57          }
 58          else
 59          {
 60             pBuff = getBuff( msgSize + 1 ) ;
 61             if ( !pBuff )
 62             {
 63                rc = SDB_OOM ;
 64                break ;
 65             }
 66             buffSize = getBuffLen() ;
 67             *(UINT32*)pBuff = msgSize ;
 68             rc = recvData( pBuff + sizeof(UINT32),
 69                            msgSize - sizeof(UINT32),
 70                            PMD_RECV_DATA_AFTER_LENGTH_TIMEOUT ) ;                // 到此处，说明程序可以愉快的接受client的发送的数据包了
 71             if ( rc )
 72             {
 73                if ( SDB_APP_FORCED != rc )
 74                {
 75                   PD_LOG( PDERROR, "Session[%s] failed to recv msg[len: %u], "
 76                           "rc: %d", sessionName(), msgSize - sizeof(UINT32),
 77                           rc ) ;
 78                }
 79                break ;
 80             }
 81  
 82             _pEDUCB->incEventCount() ;
 83             pBuff[ msgSize ] = 0 ;
 84             if ( SDB_OK != ( rc = pmdEDUMgr->activateEDU( _pEDUCB ) ) )
 85             {
 86                PD_LOG( PDERROR, "Session[%s] activate edu failed, rc: %d",
 87                        sessionName(), rc ) ;
 88                break ;
 89             }
 90             rc = _processMsg( (MsgHeader*)pBuff ) ;                                     // 收到数据包，开始处理，该函数在结合代码讲解
 91             if ( rc )
 92             {
 93                break ;
 94             }
 95             if ( SDB_OK != ( rc = pmdEDUMgr->waitEDU( _pEDUCB ) ) )
 96             {
 97                PD_LOG( PDERROR, "Session[%s] wait edu failed, rc: %d",
 98                        sessionName(), rc ) ;
 99                break ;
100             }
101          }
102       } // end while
103 
104    done:
105       disconnect() ;
106       return rc ;
107    error:
108       goto done ;
109    }

_processMsg方法：

 1 INT32 _pmdLocalSession::_processMsg( MsgHeader * msg )
 2    {
 3       INT32 rc          = SDB_OK ;
 4       const CHAR *pBody = NULL ;
 5       INT32 bodyLen     = 0 ;
 6       rtnContextBuf contextBuff ;
 7       INT32 opCode      = msg->opCode ;
 8 
 9       rc = _onMsgBegin( msg ) ;                                     // 对数据包做前期处理，例如改数据包是不是需要返回，（若出错）需不需要回滚，并初始化好回复的数据包头部
10       if ( SDB_OK == rc )
11       {
12          rc = _processor->processMsg( msg, contextBuff,             // 我是项目经理，这个包就交给processor处理去吧，我要的是结果。
13                                       _replyHeader.contextID,       // processor在不同的节点中，指向不同的对象（咦，这不是多态么？），因此也有不同的处理方式
14                                       _needReply ) ;
15          pBody     = contextBuff.data() ;                           // pBody指向要返回的数据，避免拷贝（提高执行效率）
16          bodyLen   = contextBuff.size() ;                           // 数据长度，不表
17          _replyHeader.numReturned = contextBuff.recordNum() ;       // 返回的数据共有多少条记录
18          _replyHeader.startFrom = (INT32)contextBuff.getStartFrom() ; // 应该从哪一条开始读
19          if ( SDB_OK != rc )
20          {
21             if ( _needRollback )                                          // 当执行过程中例如（insert， delete等），出错了，需要把数据复原
22             {
23                INT32 rcTmp = rtnTransRollback( eduCB(), getDPSCB() ) ;
24                if ( rcTmp )
25                {
26                   PD_LOG( PDERROR, "Session[%s] failed to rollback trans "
27                           "info, rc: %d", sessionName(), rcTmp ) ;
28                }
29                _needRollback = FALSE ;
30             }
31          }
32       }
33 
34       if ( _needReply )                                              // 需要回复，那就再处理一下把
35       {
36          if ( rc && bodyLen == 0 )                                   // 执行过程出错，那就返回出错信息
37          {
38             _errorInfo = utilGetErrorBson( rc, _pEDUCB->getInfo(
39                                            EDU_INFO_ERROR ) ) ;
40             pBody = _errorInfo.objdata() ;
41             bodyLen = (INT32)_errorInfo.objsize() ;
42             _replyHeader.numReturned = 1 ;
43          }
44          _replyHeader.header.opCode = MAKE_REPLY_TYPE(opCode) ;            // 填充回复数据包中的字段
45          _replyHeader.flags         = rc ;
46          _replyHeader.header.messageLength = sizeof( _replyHeader ) +
47                                              bodyLen ;
48 
49          INT32 rcTmp = _reply( &_replyHeader, pBody, bodyLen ) ;            // 把包发送给client
50          if ( rcTmp )
51          {
52             PD_LOG( PDERROR, "Session[%s] failed to send response, rc: %d",
53                     sessionName(), rcTmp ) ;
54             disconnect() ;
55          }
56       }
57 
58       _onMsgEnd( rc, msg ) ;
59       rc = SDB_OK ;
60 
61       return rc ;
62    }

coord节点上的processor，是pmdCoordProcessor的一个实例，是用来做数据转发的，不同于真正做数据处理的pmdDataProcessor。

 1 INT32 _pmdCoordProcessor::processMsg( MsgHeader *msg,
 2                                          rtnContextBuf &contextBuff,
 3                                          INT64 &contextID,
 4                                          BOOLEAN &needReply )
 5    {
 6       ...
 7 
 8       rc = _processCoordMsg( msg, _replyHeader, contextBuff ) ;                     // 转给另一个函数（_processCoordMsg）处理,下面讲述
 9       if ( SDB_COORD_UNKNOWN_OP_REQ == rc )
10       {
11          contextBuff.release() ;
12          rc = _pmdDataProcessor::processMsg( msg, contextBuff,                     // 如果上一个函数处理后，返回的错误是一个 SDB_COORD_UNKNOWN_OP_REQ类型，则交给pmdDataProcessor处理
13                                              contextID, needReply ) ;
14       }
15       ...
16    }

pmdCoordProcessor的处理过程

 1 INT32 _pmdCoordProcessor::_processCoordMsg( MsgHeader *msg, 
 2                                                MsgOpReply &replyHeader,
 3                                                rtnContextBuf &contextBuff )
 4    {
 5       INT32 rc = SDB_OK ;
 6       if ( NULL != _pErrorObj )
 7       {
 8          SDB_OSS_DEL _pErrorObj ;
 9          _pErrorObj = NULL ;
10       }
11       if ( NULL != _pResultBuff )
12       {
13          _pResultBuff = NULL ;
14       }
15       CoordCB *pCoordcb  = _pKrcb->getCoordCB();
16       rtnCoordProcesserFactory *pProcesserFactory
17                                         = pCoordcb->getProcesserFactory();
18 
19       if ( MSG_AUTH_VERIFY_REQ == msg->opCode )
20       {
21          rc = SDB_COORD_UNKNOWN_OP_REQ ;
22          goto done ;
23       }
24       else if ( MSG_BS_INTERRUPTE == msg->opCode ||
25                 MSG_BS_INTERRUPTE_SELF == msg->opCode ||
26                 MSG_BS_DISCONNECT == msg->opCode )
27       {
28       }
29       else if ( !getClient()->isAuthed() )                        // 没有用用户和密码登录，就收到了数据包的，就先尝试用默认的用户名和密码，先取得数据库的授权，否则无法做操作
30       {
31          rc = getClient()->authenticate( "", "" ) ;
32          if ( rc )
33          {
34             goto done ;
35          }
36       }
37 
38       switch ( msg->opCode )                                      // 开始检查client要做什么样的操作了
39       {
40       case MSG_BS_GETMORE_REQ :                                   // get more操作，coord不做处理，先标记成 SDB_COORD_UNKNOWN_OP_REQ，交给其它地方处理
41          rc = SDB_COORD_UNKNOWN_OP_REQ ;
42          break ;
43       case MSG_BS_QUERY_REQ:                                      // 查询操作，这个是重点。所有的command
44          {
45             MsgOpQuery *pQueryMsg   = ( MsgOpQuery * )msg ;
46             CHAR *pQueryName        = pQueryMsg->name ;
47             SINT32 queryNameLen     = pQueryMsg->nameLength ;
48             if ( queryNameLen > 0 && '$' == pQueryName[0] )       // 如果查询的name字段，是用$开头的字符串，则认为这个是command，要走command处理
49             {
50                rtnCoordCommand *pCmdProcesser = 
51                            pProcesserFactory->getCommandProcesser( pQueryMsg ) ;  // 找到command的对象，上文中有描述所有的command都在初始化的时候，存入_cmdMap中
52                if ( NULL != pCmdProcesser ) 
53                {
54                   rc = pCmdProcesser->execute( ( CHAR *)msg,                    // 找到了，就开始command处理了
55                                                msg->messageLength,
56                                                eduCB(),
57                                                replyHeader,
58                                                &contextBuff ) ;
59                   break ;
60                }
61             }
62             // 如果没有找到，则走入 default代码块
63          }
64       default:
65          {
66             rtnContextBase *pContext = NULL ;
67             rtnCoordOperator *pOperator = 
68                            pProcesserFactory->getOperator( msg->opCode ) ;         // 交给operator处理，operator是类似于command的几个特殊的处理对象，数量比较少，此处不表
69             rc = pOperator->execute( ( CHAR* )msg,                                 // 转发给对应的operator类实例
70                                      msg->messageLength,
71                                      eduCB(),
72                                      replyHeader,
73                                      &contextBuff ) ;
74              ...
75           }
76    }

以创建集合空间的command为例，看看 rtnCoordCMDListCollectionSpace 的 execute做了什么：

INT32 rtnCoordCMDCreateCollectionSpace::execute( CHAR *pReceiveBuffer,SINT32 packSize,pmdEDUCB *cb,MsgOpReply &replyHeader,rtnContextBuf *buf ){...MsgOpQuery *pCreateReq           = (MsgOpQuery *)pReceiveBuffer;                   // 构造一个 MSG_CAT_CREATE_COLLECTION_SPACE_REQ 的数据包pCreateReq->header.routeID.value = 0;pCreateReq->header.TID           = cb->getTID();pCreateReq->header.opCode        = MSG_CAT_CREATE_COLLECTION_SPACE_REQ;            // 数据包的类型
rc = executeOnCataGroup ( (CHAR*)pCreateReq, pRouteAgent,cb, NULL, NULL ) ;if ( rc ){PD_LOG ( PDERROR, "create collectionspace failed, rc = %d", rc ) ;goto error ;}done :replyHeader.flags = rc ;PD_TRACE_EXITRC ( SDB_RTNCOCMDCRCS_EXE, rc ) ;return rc;error :goto done ;}

该函数的主体，构造了另外一个数据包，然后执行 executeOnCataGroup ( (CHAR*)pCreateReq, pRouteAgent, cb, NULL, NULL ) ;这一句上。跟进这一函数：

 1 INT32 rtnCoordCommand::executeOnCataGroup ( CHAR *pBuffer,
 2                                                netMultiRouteAgent *pRouteAgent,
 3                                                pmdEDUCB *cb,
 4                                                rtnContextCoord *pContext,
 5                                                CoordGroupList *pGroupList,
 6                                                std::vector<BSONObj> *pReplyObjs )
 7    {
 8       INT32 rc = SDB_OK;
 9       ...
10    retry :
11       rc = rtnCoordGetCatGroupInfo( cb, isNeedRefresh, catGroupInfo );            // 查询catalog的信息，主要是获取到catalog组的主节点的服务地址
12       if ( rc )
13       {
14          probe = 100 ;
15          goto error ;
16          PD_LOG ( PDERROR, "Execute on catalogue node failed, failed to get "
17                   "catalogue group info(rc=%d)", rc );
18       }
19       rc = rtnCoordSendRequestToPrimary( pBuffer, catGroupInfo, sendNodes,       // 跟了这么久，做了那么多的准备，这一句才是真开始了，有兴趣可以自己看一下 ：)
20                                          pRouteAgent, MSG_ROUTE_CAT_SERVICE,
21                                          cb );
22       if ( rc )
23       {
24          probe = 200 ;
25          goto error ;
26       }
27       rc = rtnCoordGetReply( cb, sendNodes, replyQue,                         // 等待并收取远程节点处理的返回信息
28                              MAKE_REPLY_TYPE(((MsgHeader*)pBuffer)->opCode) ) ;
29       ...
30    }