xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(sync): add term in rpcMsg

This commit is contained in:
Minghao Li 2022-07-06 16:52:39 +08:00
parent fcb2197200
commit b73914fb0d
3 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
include/libs/transport/trpc.h
View File

@ -27,7 +27,7 @@ extern "C" {
#define TAOS_CONN_SERVER 0 #define TAOS_CONN_SERVER 0
#define TAOS_CONN_CLIENT 1 #define TAOS_CONN_CLIENT 1
#define IsReq(pMsg) (pMsg->msgType & 1U) #define IsReq(pMsg) (pMsg->msgType & 1U)
extern int32_t tsRpcHeadSize; extern int32_t tsRpcHeadSize;
@ -35,12 +35,13 @@ typedef struct {
uint32_t clientIp; uint32_t clientIp;
uint16_t clientPort; uint16_t clientPort;
int64_t applyIndex; int64_t applyIndex;
uint64_t applyTerm;
char user[TSDB_USER_LEN]; char user[TSDB_USER_LEN];
} SRpcConnInfo; } SRpcConnInfo;
typedef struct SRpcHandleInfo { typedef struct SRpcHandleInfo {
// rpc info // rpc info
void * handle; // rpc handle returned to app void *handle; // rpc handle returned to app
int64_t refId; // refid, used by server int64_t refId; // refid, used by server
int32_t noResp; // has response or not(default 0, 0: resp, 1: no resp); int32_t noResp; // has response or not(default 0, 0: resp, 1: no resp);
int32_t persistHandle; // persist handle or not int32_t persistHandle; // persist handle or not
@ -53,7 +54,7 @@ typedef struct SRpcHandleInfo {
void *node; // node mgmt handle void *node; // node mgmt handle
// resp info // resp info
void * rsp; void *rsp;
int32_t rspLen; int32_t rspLen;
// conn info // conn info
@ -62,7 +63,7 @@ typedef struct SRpcHandleInfo {
typedef struct SRpcMsg { typedef struct SRpcMsg {
tmsg_t msgType; tmsg_t msgType;
void * pCont; void *pCont;
int32_t contLen; int32_t contLen;
int32_t code; int32_t code;
SRpcHandleInfo info; SRpcHandleInfo info;
@ -74,7 +75,7 @@ typedef bool (*RpcRfp)(int32_t code, tmsg_t msgType);
typedef struct SRpcInit { typedef struct SRpcInit {
char localFqdn[TSDB_FQDN_LEN]; char localFqdn[TSDB_FQDN_LEN];
uint16_t localPort; // local port uint16_t localPort; // local port
char * label; // for debug purpose char *label; // for debug purpose
int32_t numOfThreads; // number of threads to handle connections int32_t numOfThreads; // number of threads to handle connections
int32_t sessions; // number of sessions allowed int32_t sessions; // number of sessions allowed
int8_t connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS int8_t connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
@ -99,12 +100,12 @@ typedef struct {
typedef struct { typedef struct {
int32_t msgType; int32_t msgType;
void * val; void *val;
int32_t (*clone)(void *src, void **dst); int32_t (*clone)(void *src, void **dst);
} SRpcBrokenlinkVal; } SRpcBrokenlinkVal;
typedef struct { typedef struct {
SHashObj * args; SHashObj *args;
SRpcBrokenlinkVal brokenVal; SRpcBrokenlinkVal brokenVal;
void (*freeFunc)(const void *arg); void (*freeFunc)(const void *arg);
} SRpcCtx; } SRpcCtx;

1
source/dnode/vnode/src/vnd/vnodeSync.c
View File

@ -411,6 +411,7 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c
memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen); memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info); syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
rpcMsg.info.conn.applyIndex = cbMeta.index; rpcMsg.info.conn.applyIndex = cbMeta.index;
rpcMsg.info.conn.applyTerm = cbMeta.term;
tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg); tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
} }

1
source/libs/sync/src/syncMain.c
View File

@ -818,6 +818,7 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
int32_t code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg, &retIndex); int32_t code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg, &retIndex);
if (code == 0) { if (code == 0) {
pMsg->info.conn.applyIndex = retIndex; pMsg->info.conn.applyIndex = retIndex;
pMsg->info.conn.applyTerm = pSyncNode->pRaftStore->currentTerm;
rpcFreeCont(rpcMsg.pCont); rpcFreeCont(rpcMsg.pCont);
syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum); syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum);
ret = 1; ret = 1;