好久不见。
在上一篇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() ) ; // TRANS
PMD_REGISTER_CB( sdbGetCoordCB() ) ; // COORD
PMD_REGISTER_CB( sdbGetFMPCB () ) ; // FMP
}
...
// 每个节点都会注册的控制块
PMD_REGISTER_CB( sdbGetDMSCB() ) ; // DMS
PMD_REGISTER_CB( sdbGetRTNCB() ) ; // RTN
PMD_REGISTER_CB( sdbGetSQLCB() ) ; // SQL
PMD_REGISTER_CB( sdbGetAggrCB() ) ; // AGGR
PMD_REGISTER_CB( sdbGetPMDController() ) ; // CONTROLLER
}
coord注册这几个CB之后，就开始注册和启动服务：
具体函数在_KRCB::init()中，不再表述。_KRCB::init()会根据节点的角色，启动不同的服务。
客户端连接到coord，coord便会启动一个线程，为该连接服务。
INT32 pmdTcpListenerEntryPoint ( pmdEDUCB *cb, void *pData )
{
... while ( !cb->isDisconnected() && !pListerner->isClosed() )
{
SOCKET s ;
rc = pListerner->accept ( &s, NULL, NULL ) ;
if ( SDB_TIMEOUT == rc || SDB_TOO_MANY_OPEN_FD == rc )
{
rc = SDB_OK ;
continue ;
}
if ( rc && PMD_IS_DB_DOWN )
{
rc = SDB_OK ;
goto done ;
}
else if ( rc )
{
PD_LOG ( PDERROR, "Failed to accept socket in TcpListener(rc=%d)",
rc ) ;
if ( pListerner->isClosed() )
{
break ;
}
else
{
continue ;
}
} cb->incEventCount() ;
++mondbcb->numConnects ;
void *pData = NULL ;
*((SOCKET *) &pData) = s ;
if ( !krcb->isActive() )
{
ossSocket newsock ( &s ) ;
newsock.close () ;
continue ;
} rc = eduMgr->startEDU ( EDU_TYPE_AGENT, pData, &agentEDU ) ;
if ( rc )
{
PD_LOG( ( rc == SDB_QUIESCED ? PDWARNING : PDERROR ),
"Failed to start edu, rc: %d", rc ) ;
ossSocket newsock ( &s ) ;
newsock.close () ;
continue ;
}
} //while ( ! cb->isDisconnected() ) ...
}
服务线程监听到client的连接，启动一个EDU_TYPE_AGENT类型的线程，单独为client服务。
下面讲述coord节点的最主要的功能——消息转发
coord的启动初，会初始化一些必要的全局变量。在SequoiaDB中，会初始化很多command，拿创建集合空间来说，在文件SequoiaDB/engine/rtn/rtnCoord.cpp 中
RTN_COORD_CMD_BEGIN
... RTN_COORD_CMD_ADD( COORD_CMD_LISTCOLLECTIONSPACES, rtnCoordCMDListCollectionSpace ) ...
RTN_COORD_OP_END
嗯，上面的代码有点MFC中消息映射的感觉。
来看看 RTN_COORD_CMD_ADD 宏：
#define RTN_COORD_CMD_ADD( cmdName, cmdClass ) {\
rtnCoordCommand *pObj = SDB_OSS_NEW cmdClass();\
_cmdMap.insert ( COORD_CMD_MAP::value_type (cmdName, pObj ));}
宏主要是new一个对象，再把对象插入到_cmdMap中，这样在程序初始化时候，便会有一系列的command对象存储在_cmdMap中。
另外，对SequoiaDB而言，所有的command操作，都是在查询操作的基础上做的，服务端用一些方法区别开是真正的查询，还是command。SequoiaDB的命令，是以$开头的字符串。
前提简述完毕，现在假设client连接上了coord，coord也创建了一个线程，为这个client服务。
INT32 _pmdLocalSession::run()
{
INT32 rc = SDB_OK ;
UINT32 msgSize = ;
CHAR *pBuff = NULL ;
INT32 buffSize = ;
pmdEDUMgr *pmdEDUMgr = NULL ; if ( !_pEDUCB )
{
rc = SDB_SYS ;
goto error ;
} pmdEDUMgr = _pEDUCB->getEDUMgr() ; while ( !_pEDUCB->isDisconnected() && !_socket.isClosed() )
{
_pEDUCB->resetInterrupt() ;
_pEDUCB->resetInfo( EDU_INFO_ERROR ) ;
_pEDUCB->resetLsn() ; rc = recvData( (CHAR*)&msgSize, sizeof(UINT32) ) ; // 收取数据包的前四个字节，代表该数据包有多大
if ( rc )
{
if ( SDB_APP_FORCED != rc )
{
PD_LOG( PDERROR, "Session[%s] failed to recv msg size, "
"rc: %d", sessionName(), rc ) ;
}
break ;
} if ( msgSize == (UINT32)MSG_SYSTEM_INFO_LEN ) // 如果包长度是 MSG_SYSTEM_INFO_LEN （-1），则这是一个请求系统信息包，coord会返回本机的字节序列给client
{ // 每个连接的第一个包，一定是长度标记为 MSG_SYSTEM_INFO_LEN 的包，否则字节序不正确，所有的数据都不能保证能正确解析（万一数据库运行在大端机上呢）
rc = _recvSysInfoMsg( msgSize, &pBuff, buffSize ) ;
if ( rc )
{
break ;
}
rc = _processSysInfoRequest( pBuff ) ;
if ( rc )
{
break ;
} _setHandshakeReceived() ;
}
else if ( msgSize < sizeof(MsgHeader) || msgSize > SDB_MAX_MSG_LENGTH ) // 对包的大小做出了限制，包长超过某值或者小于某值的包，都会导致连接中断
{
PD_LOG( PDERROR, "Session[%s] recv msg size[%d] is less than "
"MsgHeader size[%d] or more than max msg size[%d]",
sessionName(), msgSize, sizeof(MsgHeader),
SDB_MAX_MSG_LENGTH ) ;
rc = SDB_INVALIDARG ;
break ;
}
else
{
pBuff = getBuff( msgSize + ) ;
if ( !pBuff )
{
rc = SDB_OOM ;
break ;
}
buffSize = getBuffLen() ;
*(UINT32*)pBuff = msgSize ;
rc = recvData( pBuff + sizeof(UINT32),
msgSize - sizeof(UINT32),
PMD_RECV_DATA_AFTER_LENGTH_TIMEOUT ) ; // 到此处，说明程序可以愉快的接受client的发送的数据包了
if ( rc )
{
if ( SDB_APP_FORCED != rc )
{
PD_LOG( PDERROR, "Session[%s] failed to recv msg[len: %u], "
"rc: %d", sessionName(), msgSize - sizeof(UINT32),
rc ) ;
}
break ;
} _pEDUCB->incEventCount() ;
pBuff[ msgSize ] = ;
if ( SDB_OK != ( rc = pmdEDUMgr->activateEDU( _pEDUCB ) ) )
{
PD_LOG( PDERROR, "Session[%s] activate edu failed, rc: %d",
sessionName(), rc ) ;
break ;
}
rc = _processMsg( (MsgHeader*)pBuff ) ; // 收到数据包，开始处理，该函数在结合代码讲解
if ( rc )
{
break ;
}
if ( SDB_OK != ( rc = pmdEDUMgr->waitEDU( _pEDUCB ) ) )
{
PD_LOG( PDERROR, "Session[%s] wait edu failed, rc: %d",
sessionName(), rc ) ;
break ;
}
}
} // end while done:
disconnect() ;
return rc ;
error:
goto done ;
}
_processMsg方法：
INT32 _pmdLocalSession::_processMsg( MsgHeader * msg )
{
INT32 rc = SDB_OK ;
const CHAR *pBody = NULL ;
INT32 bodyLen = ;
rtnContextBuf contextBuff ;
INT32 opCode = msg->opCode ; rc = _onMsgBegin( msg ) ; // 对数据包做前期处理，例如改数据包是不是需要返回，（若出错）需不需要回滚，并初始化好回复的数据包头部
if ( SDB_OK == rc )
{
rc = _processor->processMsg( msg, contextBuff, // 我是项目经理，这个包就交给processor处理去吧，我要的是结果。
_replyHeader.contextID, // processor在不同的节点中，指向不同的对象（咦，这不是多态么？），因此也有不同的处理方式
_needReply ) ;
pBody = contextBuff.data() ; // pBody指向要返回的数据，避免拷贝（提高执行效率）
bodyLen = contextBuff.size() ; // 数据长度，不表
_replyHeader.numReturned = contextBuff.recordNum() ; // 返回的数据共有多少条记录
_replyHeader.startFrom = (INT32)contextBuff.getStartFrom() ; // 应该从哪一条开始读
if ( SDB_OK != rc )
{
if ( _needRollback ) // 当执行过程中例如（insert， delete等），出错了，需要把数据复原
{
INT32 rcTmp = rtnTransRollback( eduCB(), getDPSCB() ) ;
if ( rcTmp )
{
PD_LOG( PDERROR, "Session[%s] failed to rollback trans "
"info, rc: %d", sessionName(), rcTmp ) ;
}
_needRollback = FALSE ;
}
}
} if ( _needReply ) // 需要回复，那就再处理一下把
{
if ( rc && bodyLen == ) // 执行过程出错，那就返回出错信息
{
_errorInfo = utilGetErrorBson( rc, _pEDUCB->getInfo(
EDU_INFO_ERROR ) ) ;
pBody = _errorInfo.objdata() ;
bodyLen = (INT32)_errorInfo.objsize() ;
_replyHeader.numReturned = ;
}
_replyHeader.header.opCode = MAKE_REPLY_TYPE(opCode) ; // 填充回复数据包中的字段
_replyHeader.flags = rc ;
_replyHeader.header.messageLength = sizeof( _replyHeader ) +
bodyLen ; INT32 rcTmp = _reply( &_replyHeader, pBody, bodyLen ) ; // 把包发送给client
if ( rcTmp )
{
PD_LOG( PDERROR, "Session[%s] failed to send response, rc: %d",
sessionName(), rcTmp ) ;
disconnect() ;
}
} _onMsgEnd( rc, msg ) ;
rc = SDB_OK ; return rc ;
}
coord节点上的processor，是pmdCoordProcessor的一个实例，是用来做数据转发的，不同于真正做数据处理的pmdDataProcessor。
INT32 _pmdCoordProcessor::processMsg( MsgHeader *msg,
rtnContextBuf &contextBuff,
INT64 &contextID,
BOOLEAN &needReply )
{
... rc = _processCoordMsg( msg, _replyHeader, contextBuff ) ; // 转给另一个函数（_processCoordMsg）处理,下面讲述
if ( SDB_COORD_UNKNOWN_OP_REQ == rc )
{
contextBuff.release() ;
rc = _pmdDataProcessor::processMsg( msg, contextBuff, // 如果上一个函数处理后，返回的错误是一个 SDB_COORD_UNKNOWN_OP_REQ类型，则交给pmdDataProcessor处理
contextID, needReply ) ;
}
...
}
pmdCoordProcessor的处理过程
INT32 _pmdCoordProcessor::_processCoordMsg( MsgHeader *msg,
MsgOpReply &replyHeader,
rtnContextBuf &contextBuff )
{
INT32 rc = SDB_OK ;
if ( NULL != _pErrorObj )
{
SDB_OSS_DEL _pErrorObj ;
_pErrorObj = NULL ;
}
if ( NULL != _pResultBuff )
{
_pResultBuff = NULL ;
}
CoordCB *pCoordcb = _pKrcb->getCoordCB();
rtnCoordProcesserFactory *pProcesserFactory
= pCoordcb->getProcesserFactory(); if ( MSG_AUTH_VERIFY_REQ == msg->opCode )
{
rc = SDB_COORD_UNKNOWN_OP_REQ ;
goto done ;
}
else if ( MSG_BS_INTERRUPTE == msg->opCode ||
MSG_BS_INTERRUPTE_SELF == msg->opCode ||
MSG_BS_DISCONNECT == msg->opCode )
{
}
else if ( !getClient()->isAuthed() ) // 没有用用户和密码登录，就收到了数据包的，就先尝试用默认的用户名和密码，先取得数据库的授权，否则无法做操作
{
rc = getClient()->authenticate( "", "" ) ;
if ( rc )
{
goto done ;
}
} switch ( msg->opCode ) // 开始检查client要做什么样的操作了
{
case MSG_BS_GETMORE_REQ : // get more操作，coord不做处理，先标记成 SDB_COORD_UNKNOWN_OP_REQ，交给其它地方处理
rc = SDB_COORD_UNKNOWN_OP_REQ ;
break ;
case MSG_BS_QUERY_REQ: // 查询操作，这个是重点。所有的command
{
MsgOpQuery *pQueryMsg = ( MsgOpQuery * )msg ;
CHAR *pQueryName = pQueryMsg->name ;
SINT32 queryNameLen = pQueryMsg->nameLength ;
if ( queryNameLen > && '$' == pQueryName[] ) // 如果查询的name字段，是用$开头的字符串，则认为这个是command，要走command处理
{
rtnCoordCommand *pCmdProcesser =
pProcesserFactory->getCommandProcesser( pQueryMsg ) ; // 找到command的对象，上文中有描述所有的command都在初始化的时候，存入_cmdMap中
if ( NULL != pCmdProcesser )
{
rc = pCmdProcesser->execute( ( CHAR *)msg, // 找到了，就开始command处理了
msg->messageLength,
eduCB(),
replyHeader,
&contextBuff ) ;
break ;
}
}
// 如果没有找到，则走入 default代码块
}
default:
{
rtnContextBase *pContext = NULL ;
rtnCoordOperator *pOperator =
pProcesserFactory->getOperator( msg->opCode ) ; // 交给operator处理，operator是类似于command的几个特殊的处理对象，数量比较少，此处不表
rc = pOperator->execute( ( CHAR* )msg, // 转发给对应的operator类实例
msg->messageLength,
eduCB(),
replyHeader,
&contextBuff ) ;
...
}
}
以创建集合空间的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 = ;
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 ) ;这一句上。跟进这一函数：
INT32 rtnCoordCommand::executeOnCataGroup ( CHAR *pBuffer,
netMultiRouteAgent *pRouteAgent,
pmdEDUCB *cb,
rtnContextCoord *pContext,
CoordGroupList *pGroupList,
std::vector<BSONObj> *pReplyObjs )
{
INT32 rc = SDB_OK;
...
retry :
rc = rtnCoordGetCatGroupInfo( cb, isNeedRefresh, catGroupInfo ); // 查询catalog的信息，主要是获取到catalog组的主节点的服务地址
if ( rc )
{
probe = ;
goto error ;
PD_LOG ( PDERROR, "Execute on catalogue node failed, failed to get "
"catalogue group info(rc=%d)", rc );
}
rc = rtnCoordSendRequestToPrimary( pBuffer, catGroupInfo, sendNodes, // 跟了这么久，做了那么多的准备，这一句才是真开始了，有兴趣可以自己看一下 ：)
pRouteAgent, MSG_ROUTE_CAT_SERVICE,
cb );
if ( rc )
{
probe = ;
goto error ;
}
rc = rtnCoordGetReply( cb, sendNodes, replyQue, // 等待并收取远程节点处理的返回信息
MAKE_REPLY_TYPE(((MsgHeader*)pBuffer)->opCode) ) ;
...
}
rtnCoordSendRequestToPrimary就不再详细跟进描述了，根据函数名，大致就可以了解一个大概，是把数据发送到指定组（此处是catalog组）的主节点。
coord上的其它command或者operator也是采用类似的方法来转发消息给其它节点，就不再一一赘述了。
综合全文的讲述，coord处理client请求的流程
发送请求给coord节点
coord先揪出这个请求是做什么
交给对应的command处理
查询（本地缓存或者远程获取的）catalog信息
把消息转成节点间的内部消息
转发给目标节点
然后等待返回数据
再把返回数据交给处理线程
线程把返回结果发送给client
=====>THE END<=====