sync refactor

source/libs/sync/src/syncIO.c

@@ -29,7 +29,7 @@ static int32_t  syncIODestroy(SSyncIO *io);
 static int32_t  syncIOStartInternal(SSyncIO *io);
 static int32_t  syncIOStopInternal(SSyncIO *io);
 
-static void *  syncIOConsumerFunc(void *param);
+static void   *syncIOConsumerFunc(void *param);
 static void    syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
 static void    syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
 static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey);
@@ -75,6 +75,7 @@ int32_t syncIOSendMsg(void *clientRpc, const SEpSet *pEpSet, SRpcMsg *pMsg) {
   syncRpcMsgPrint2(logBuf, pMsg);
 
   pMsg->handle = NULL;
+  pMsg->noResp = 1;
   rpcSendRequest(clientRpc, pEpSet, pMsg, NULL);
   return ret;
 }
@@ -234,9 +235,9 @@ static int32_t syncIOStopInternal(SSyncIO *io) {
 }
 
 static void *syncIOConsumerFunc(void *param) {
-  SSyncIO *  io = param;
+  SSyncIO   *io = param;
   STaosQall *qall;
-  SRpcMsg *  pRpcMsg, rpcMsg;
+  SRpcMsg   *pRpcMsg, rpcMsg;
   qall = taosAllocateQall();
 
   while (1) {
@@ -324,19 +325,21 @@ static void *syncIOConsumerFunc(void *param) {
       taosGetQitem(qall, (void **)&pRpcMsg);
       rpcFreeCont(pRpcMsg->pCont);
 
-      if (pRpcMsg->handle != NULL) {
-        int msgSize = 32;
-        memset(&rpcMsg, 0, sizeof(rpcMsg));
-        rpcMsg.msgType = SYNC_RESPONSE;
-        rpcMsg.pCont = rpcMallocCont(msgSize);
-        rpcMsg.contLen = msgSize;
-        snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "give a reply");
-        rpcMsg.handle = pRpcMsg->handle;
-        rpcMsg.code = 0;
-
-        syncRpcMsgPrint2((char *)"syncIOConsumerFunc rpcSendResponse --> ", &rpcMsg);
-        rpcSendResponse(&rpcMsg);
-      }
+      /*
+            if (pRpcMsg->handle != NULL) {
+              int msgSize = 32;
+              memset(&rpcMsg, 0, sizeof(rpcMsg));
+              rpcMsg.msgType = SYNC_RESPONSE;
+              rpcMsg.pCont = rpcMallocCont(msgSize);
+              rpcMsg.contLen = msgSize;
+              snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "give a reply");
+              rpcMsg.handle = pRpcMsg->handle;
+              rpcMsg.code = 0;
+
+              syncRpcMsgPrint2((char *)"syncIOConsumerFunc rpcSendResponse --> ", &rpcMsg);
+              rpcSendResponse(&rpcMsg);
+            }
+      */
 
       taosFreeQitem(pRpcMsg);
     }

source/libs/sync/src/syncRaftLog.c

@@ -111,7 +111,7 @@ SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore) {
 int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
-  walCommit(pWal, index);
+  assert(walCommit(pWal, index) == 0);
   return 0;  // to avoid compiler error
 }