Merge pull request #18232 from taosdata/fix/asan-timer-memory-error

fix(sync): fix asan error, use rid in hbdata and syncnode

source/libs/sync/inc/syncEnv.h

@@ -57,6 +57,11 @@ void       syncNodeRemove(int64_t rid);
 SSyncNode* syncNodeAcquire(int64_t rid);
 void       syncNodeRelease(SSyncNode* pNode);
 
+int64_t           syncHbTimerDataAdd(SSyncHbTimerData* pData);
+void              syncHbTimerDataRemove(int64_t rid);
+SSyncHbTimerData* syncHbTimerDataAcquire(int64_t rid);
+void              syncHbTimerDataRelease(SSyncHbTimerData* pData);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/sync/inc/syncInt.h

@@ -57,10 +57,11 @@ typedef struct SRaftId {
 } SRaftId;
 
 typedef struct SSyncHbTimerData {
-  SSyncNode*  pSyncNode;
+  int64_t     syncNodeRid;
   SSyncTimer* pTimer;
   SRaftId     destId;
   uint64_t    logicClock;
+  int64_t     rid;
 } SSyncHbTimerData;
 
 typedef struct SSyncTimer {
@@ -70,7 +71,7 @@ typedef struct SSyncTimer {
   uint64_t          counter;
   int32_t           timerMS;
   SRaftId           destId;
-  SSyncHbTimerData  hbData;
+  int64_t           hbDataRid;
 } SSyncTimer;
 
 typedef struct SElectTimerParam {
@@ -189,6 +190,8 @@ typedef struct SSyncNode {
   int64_t leaderTime;
   int64_t lastReplicateTime;
 
+  bool isStart;
+
 } SSyncNode;
 
 // open/close --------------
@@ -198,6 +201,7 @@ void       syncNodeStartStandBy(SSyncNode* pSyncNode);
 void       syncNodeClose(SSyncNode* pSyncNode);
 void       syncNodePreClose(SSyncNode* pSyncNode);
 int32_t    syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak);
+void       syncHbTimerDataFree(SSyncHbTimerData* pData);
 
 // on message ---------------------
 int32_t syncNodeOnTimeout(SSyncNode* ths, const SRpcMsg* pMsg);

source/libs/sync/src/syncEnv.c

@@ -20,6 +20,7 @@
 
 static SSyncEnv gSyncEnv = {0};
 static int32_t  gNodeRefId = -1;
+static int32_t  gHbDataRefId = -1;
 static void     syncEnvTick(void *param, void *tmrId);
 
 SSyncEnv *syncEnv() { return &gSyncEnv; }
@@ -50,6 +51,13 @@ int32_t syncInit() {
     return -1;
   }
 
+  gHbDataRefId = taosOpenRef(200, (RefFp)syncHbTimerDataFree);
+  if (gHbDataRefId < 0) {
+    sError("failed to init hb-data ref");
+    syncCleanUp();
+    return -1;
+  }
+
   sDebug("sync rsetId:%d is open", gNodeRefId);
   return 0;
 }
@@ -64,6 +72,12 @@ void syncCleanUp() {
     taosCloseRef(gNodeRefId);
     gNodeRefId = -1;
   }
+
+  if (gHbDataRefId != -1) {
+    sDebug("sync rsetId:%d is closed", gHbDataRefId);
+    taosCloseRef(gHbDataRefId);
+    gHbDataRefId = -1;
+  }
 }
 
 int64_t syncNodeAdd(SSyncNode *pNode) {
@@ -88,6 +102,26 @@ SSyncNode *syncNodeAcquire(int64_t rid) {
 
 void syncNodeRelease(SSyncNode *pNode) { taosReleaseRef(gNodeRefId, pNode->rid); }
 
+int64_t syncHbTimerDataAdd(SSyncHbTimerData *pData) {
+  pData->rid = taosAddRef(gHbDataRefId, pData);
+  if (pData->rid < 0) return -1;
+  return pData->rid;
+}
+
+void syncHbTimerDataRemove(int64_t rid) { taosRemoveRef(gHbDataRefId, rid); }
+
+SSyncHbTimerData *syncHbTimerDataAcquire(int64_t rid) {
+  SSyncHbTimerData *pData = taosAcquireRef(gHbDataRefId, rid);
+  if (pData == NULL) {
+    sError("failed to acquire hb-timer-data from refId:%" PRId64, rid);
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+  }
+
+  return pData;
+}
+
+void syncHbTimerDataRelease(SSyncHbTimerData *pData) { taosReleaseRef(gHbDataRefId, pData->rid); }
+
 #if 0
 void syncEnvStartTimer() {
   taosTmrReset(gSyncEnv.FpEnvTickTimer, gSyncEnv.envTickTimerMS, &gSyncEnv, gSyncEnv.pTimerManager,

source/libs/sync/src/syncMain.c

@@ -91,6 +91,7 @@ void syncStart(int64_t rid) {
 void syncStop(int64_t rid) {
   SSyncNode* pSyncNode = syncNodeAcquire(rid);
   if (pSyncNode != NULL) {
+    pSyncNode->isStart = false;
     syncNodeRelease(pSyncNode);
     syncNodeRemove(rid);
   }
@@ -665,13 +666,20 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa
 static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
   int32_t ret = 0;
   if (syncIsInit()) {
-    SSyncHbTimerData* pData = &pSyncTimer->hbData;
-    pData->pSyncNode = pSyncNode;
+    SSyncHbTimerData* pData = syncHbTimerDataAcquire(pSyncTimer->hbDataRid);
+    if (pData == NULL) {
+      pData = taosMemoryMalloc(sizeof(SSyncHbTimerData));
+      pData->rid = syncHbTimerDataAdd(pData);
+    }
+    pSyncTimer->hbDataRid = pData->rid;
+
+    pData->syncNodeRid = pSyncNode->rid;
     pData->pTimer = pSyncTimer;
     pData->destId = pSyncTimer->destId;
     pData->logicClock = pSyncTimer->logicClock;
 
-    taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, pData, syncEnv()->pTimerManager, &pSyncTimer->pTimer);
+    taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, (void*)(pData->rid), syncEnv()->pTimerManager,
+                 &pSyncTimer->pTimer);
   } else {
     sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId);
   }
@@ -683,6 +691,8 @@ static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
   atomic_add_fetch_64(&pSyncTimer->logicClock, 1);
   taosTmrStop(pSyncTimer->pTimer);
   pSyncTimer->pTimer = NULL;
+  syncHbTimerDataRemove(pSyncTimer->hbDataRid);
+  pSyncTimer->hbDataRid = -1;
   return ret;
 }
 
@@ -960,6 +970,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
   // snapshotting
   atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);
 
+  pSyncNode->isStart = true;
   sNTrace(pSyncNode, "sync open");
 
   return pSyncNode;
@@ -1027,6 +1038,8 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
   syncNodeStopHeartbeatTimer(pSyncNode);
 }
 
+void syncHbTimerDataFree(SSyncHbTimerData* pData) { taosMemoryFree(pData); }
+
 void syncNodeClose(SSyncNode* pSyncNode) {
   if (pSyncNode == NULL) {
     return;
@@ -1929,19 +1942,36 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
 }
 
 static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
-  SSyncHbTimerData* pData = (SSyncHbTimerData*)param;
-  SSyncNode*        pSyncNode = pData->pSyncNode;
-  SSyncTimer*       pSyncTimer = pData->pTimer;
+  int64_t hbDataRid = (int64_t)param;
 
+  SSyncHbTimerData* pData = syncHbTimerDataAcquire(hbDataRid);
+  if (pData == NULL) {
+    return;
+  }
+
+  SSyncNode* pSyncNode = syncNodeAcquire(pData->syncNodeRid);
   if (pSyncNode == NULL) {
+    syncHbTimerDataRelease(pData);
+    return;
+  }
+
+  SSyncTimer* pSyncTimer = pData->pTimer;
+
+  if (!pSyncNode->isStart) {
+    syncNodeRelease(pSyncNode);
+    syncHbTimerDataRelease(pData);
     return;
   }
 
   if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
+    syncNodeRelease(pSyncNode);
+    syncHbTimerDataRelease(pData);
     return;
   }
 
   if (pSyncNode->pRaftStore == NULL) {
+    syncNodeRelease(pSyncNode);
+    syncHbTimerDataRelease(pData);
     return;
   }
 
@@ -1978,6 +2008,9 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
              msgLogicClock);
     }
   }
+
+  syncHbTimerDataRelease(pData);
+  syncNodeRelease(pSyncNode);
 }
 
 static int32_t syncNodeEqNoop(SSyncNode* pNode) {