Skip to content

TDengine
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
TDengine
TDengine
提交 9a540919 编写于 作者: M Minghao Li
浏览文件
操作

refactor(sync): pre-commit integration

上级 49a4e83d
隐藏空白更改
内联 并排
Showing with 43 addition and 38 deletion
source/dnode/vnode/src/vnd/vnodeSync.c
浏览文件 @ 9a540919
...@@ -501,49 +501,54 @@ static void vnodeSyncReconfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReCon ...@@ -501,49 +501,54 @@ static void vnodeSyncReconfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReCon
} }
static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
SVnode *pVnode = pFsm->data; if (cbMeta.isWeak == 0) {
vTrace("vgId:%d, commit-cb is excuted, fsm:%p, index:%" PRId64 ", isWeak:%d, code:%d, state:%d %s, msgtype:%d %s", SVnode *pVnode = pFsm->data;
syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, vTrace("vgId:%d, commit-cb is excuted, fsm:%p, index:%" PRId64 ", isWeak:%d, code:%d, state:%d %s, msgtype:%d %s",
syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType)); syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state,
syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType));
if (cbMeta.code == 0 && cbMeta.isWeak == 0) {
SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen}; if (cbMeta.code == 0) {
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen); rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info); memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
rpcMsg.info.conn.applyIndex = cbMeta.index; syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
rpcMsg.info.conn.applyTerm = cbMeta.term; rpcMsg.info.conn.applyIndex = cbMeta.index;
tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg); rpcMsg.info.conn.applyTerm = cbMeta.term;
} else { tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info}; } else {
vError("vgId:%d, sync commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync), pMsg->msgType, SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info};
TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code)); vError("vgId:%d, sync commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync),
if (rsp.info.handle != NULL) { pMsg->msgType, TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code));
tmsgSendRsp(&rsp); if (rsp.info.handle != NULL) {
tmsgSendRsp(&rsp);
}
} }
} }
} }
static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
SVnode *pVnode = pFsm->data; if (cbMeta.isWeak == 1) {
vTrace("vgId:%d, pre-commit-cb is excuted, fsm:%p, index:%" PRId64 ", isWeak:%d, code:%d, state:%d %s, msgtype:%d %s", SVnode *pVnode = pFsm->data;
syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, vTrace("vgId:%d, pre-commit-cb is excuted, fsm:%p, index:%" PRId64
syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType)); ", isWeak:%d, code:%d, state:%d %s, msgtype:%d %s",
syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state,
if (cbMeta.code == 0 && cbMeta.isWeak == 1) { syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType));
SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); if (cbMeta.code == 0) {
memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen); SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info); rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
rpcMsg.info.conn.applyIndex = cbMeta.index; memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
rpcMsg.info.conn.applyTerm = cbMeta.term; syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg); rpcMsg.info.conn.applyIndex = cbMeta.index;
} else { rpcMsg.info.conn.applyTerm = cbMeta.term;
SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info}; tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
vError("vgId:%d, sync pre-commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync), } else {
pMsg->msgType, TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code)); SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info};
if (rsp.info.handle != NULL) { vError("vgId:%d, sync pre-commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync),
tmsgSendRsp(&rsp); pMsg->msgType, TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code));
if (rsp.info.handle != NULL) {
tmsgSendRsp(&rsp);
}
} }
} }
} }
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或