enh: rsma stream state adaption

include/util/taoserror.h

@@ -757,9 +757,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_RSMA_INVALID_SCHEMA            TAOS_DEF_ERROR_CODE(0, 0x3153)
 #define TSDB_CODE_RSMA_STREAM_STATE_OPEN         TAOS_DEF_ERROR_CODE(0, 0x3154)
 #define TSDB_CODE_RSMA_STREAM_STATE_COMMIT       TAOS_DEF_ERROR_CODE(0, 0x3155)
-#define TSDB_CODE_RSMA_FS_REF                    TAOS_DEF_ERROR_CODE(0, 0x3156)
-#define TSDB_CODE_RSMA_FS_SYNC                   TAOS_DEF_ERROR_CODE(0, 0x3157)
-#define TSDB_CODE_RSMA_FS_UPDATE                 TAOS_DEF_ERROR_CODE(0, 0x3158)
+#define TSDB_CODE_RSMA_FS_SYNC                   TAOS_DEF_ERROR_CODE(0, 0x3156)
+#define TSDB_CODE_RSMA_RESULT                    TAOS_DEF_ERROR_CODE(0, 0x3157)
 
 //index
 #define TSDB_CODE_INDEX_REBUILDING               TAOS_DEF_ERROR_CODE(0, 0x3200)

source/dnode/vnode/CMakeLists.txt

@@ -32,7 +32,6 @@ target_sources(
     # sma
     "src/sma/smaEnv.c"
     "src/sma/smaUtil.c"
-    "src/sma/smaFS.c"
     "src/sma/smaOpen.c"
     "src/sma/smaCommit.c"
     "src/sma/smaRollup.c"

source/dnode/vnode/src/inc/sma.h

@@ -105,17 +105,16 @@ struct SRSmaFS {
 
 struct SRSmaStat {
   SSma            *pSma;
-  int64_t          commitAppliedVer;  // vnode applied version for async commit
-  int64_t          refId;             // shared by fetch tasks
-  volatile int64_t nBufItems;         // number of items in queue buffer
-  SRWLatch         lock;              // r/w lock for rsma fs(e.g. qtaskinfo)
-  volatile int32_t nFetchAll;         // active number of fetch all
-  volatile int8_t  triggerStat;       // shared by fetch tasks
-  volatile int8_t  commitStat;        // 0 not in committing, 1 in committing
-  volatile int8_t  delFlag;           // 0 no deleted SRSmaInfo, 1 has deleted SRSmaInfo
-  SRSmaFS          fs;                // for recovery/snapshot r/w
-  SHashObj        *infoHash;          // key: suid, value: SRSmaInfo
-  tsem_t           notEmpty;          // has items in queue buffer
+  int64_t          refId;        // shared by fetch tasks
+  volatile int64_t nBufItems;    // number of items in queue buffer
+  SRWLatch         lock;         // r/w lock for rsma fs(e.g. qtaskinfo)
+  volatile int32_t nFetchAll;    // active number of fetch all
+  volatile int8_t  triggerStat;  // shared by fetch tasks
+  volatile int8_t  commitStat;   // 0 not in committing, 1 in committing
+  volatile int8_t  delFlag;      // 0 no deleted SRSmaInfo, 1 has deleted SRSmaInfo
+  SRSmaFS          fs;           // for recovery/snapshot r/w
+  SHashObj        *infoHash;     // key: suid, value: SRSmaInfo
+  tsem_t           notEmpty;     // has items in queue buffer
 };
 
 struct SSmaStat {
@@ -156,12 +155,9 @@ struct SRSmaInfo {
   int16_t   padding;
   T_REF_DECLARE()
   SRSmaInfoItem items[TSDB_RETENTION_L2];
-  void         *taskInfo[TSDB_RETENTION_L2];   // qTaskInfo_t
-  STaosQueue   *queue;                         // buffer queue of SubmitReq
-  STaosQall    *qall;                          // buffer qall of SubmitReq
-  void         *iTaskInfo[TSDB_RETENTION_L2];  // immutable qTaskInfo_t
-  STaosQueue   *iQueue;                        // immutable buffer queue of SubmitReq
-  STaosQall    *iQall;                         // immutable buffer qall of SubmitReq
+  void         *taskInfo[TSDB_RETENTION_L2];  // qTaskInfo_t
+  STaosQueue   *queue;                        // buffer queue of SubmitReq
+  STaosQall    *qall;                         // buffer qall of SubmitReq
 };
 
 #define RSMA_INFO_HEAD_LEN     offsetof(SRSmaInfo, items)
@@ -191,6 +187,12 @@ typedef enum {
   RSMA_EXEC_COMMIT = 3,    // triggered by commit
 } ERsmaExecType;
 
+#define TD_SMA_LOOPS_CHECK(n, limit) \
+  if (++(n) > limit) {               \
+    sched_yield();                   \
+    (n) = 0;                         \
+  }
+
 // sma
 int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
 void    tdDestroySmaEnv(SSmaEnv *pSmaEnv);
@@ -213,27 +215,12 @@ int32_t smaPreClose(SSma *pSma);
 
 // rsma
 void   *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree);
-int32_t tdRSmaFSOpen(SSma *pSma, int64_t version, int8_t rollback);
-void    tdRSmaFSClose(SRSmaFS *fs);
-int32_t tdRSmaFSPrepareCommit(SSma *pSma, SRSmaFS *pFSNew);
-int32_t tdRSmaFSCommit(SSma *pSma);
-int32_t tdRSmaFSFinishCommit(SSma *pSma);
-int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFS);
-int32_t tdRSmaFSTakeSnapshot(SSma *pSma, SRSmaFS *pFS);
-int32_t tdRSmaFSRef(SSma *pSma, SRSmaFS *pFS);
-void    tdRSmaFSUnRef(SSma *pSma, SRSmaFS *pFS);
-int32_t tdRSmaFSUpsertQTaskFile(SSma *pSma, SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t nSize);
-int32_t tdRSmaFSRollback(SSma *pSma);
 int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback);
 int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);
 int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type);
-int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
+// int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
 int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer, int8_t rollback);
-void    tdRSmaQTaskInfoGetFileName(SVnode *pVnode, int64_t suid, int8_t level, int64_t version, char *outputName);
-void    tdRSmaQTaskInfoGetFullName(SVnode *pVnode, int64_t suid, int8_t level, int64_t version, STfs *pTfs,
-                                   char *outputName);
-void    tdRSmaQTaskInfoGetFullPath(SVnode *pVnode, int8_t level, STfs *pTfs, char *outputName);
-void    tdRSmaQTaskInfoGetFullPathEx(SVnode *pVnode, tb_uid_t suid, int8_t level, STfs *pTfs, char *outputName);
+void    tdRSmaQTaskInfoGetFullPath(SVnode *pVnode, tb_uid_t suid, int8_t level, STfs *pTfs, char *outputName);
 
 static FORCE_INLINE void tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   int32_t ref = T_REF_INC(pRSmaInfo);
@@ -244,8 +231,6 @@ static FORCE_INLINE void tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   smaTrace("vgId:%d, unref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
 }
 
-void tdRSmaGetFileName(SVnode *pVnode, STfs *pTfs, const char *fname, int64_t suid, int8_t level, int64_t version,
-                       char *outputName);
 void tdRSmaGetDirName(SVnode *pVnode, STfs *pTfs, bool endWithSep, char *outputName);
 
 #ifdef __cplusplus

source/dnode/vnode/src/inc/vnodeInt.h

@@ -319,7 +319,6 @@ int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData);
 // SRSmaSnapWriter ========================================
 int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter);
 int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
-int32_t rsmaSnapWriterPrepareClose(SRSmaSnapWriter* pWriter);
 int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback);
 
 typedef struct {

source/dnode/vnode/src/sma/smaCommit.c

@@ -108,9 +108,6 @@ int32_t smaFinishCommit(SSma *pSma) {
   int32_t lino = 0;
   SVnode *pVnode = pSma->pVnode;
 
-  code = tdRSmaFSFinishCommit(pSma);
-  TSDB_CHECK_CODE(code, lino, _exit);
-
   if (VND_RSMA1(pVnode) && (code = tsdbFinishCommit(VND_RSMA1(pVnode))) < 0) {
     TSDB_CHECK_CODE(code, lino, _exit);
   }
@@ -150,18 +147,7 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
   atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED);
   if (isCommit) {
     while (atomic_val_compare_exchange_8(RSMA_COMMIT_STAT(pRSmaStat), 0, 1) != 0) {
-      ++nLoops;
-      if (nLoops > 1000) {
-        sched_yield();
-        nLoops = 0;
-      }
-    }
-
-    pRSmaStat->commitAppliedVer = pSma->pVnode->state.applied;
-    if (ASSERTS(pRSmaStat->commitAppliedVer >= -1, "commit applied version %" PRIi64 " < -1",
-                pRSmaStat->commitAppliedVer)) {
-      code = TSDB_CODE_APP_ERROR;
-      TSDB_CHECK_CODE(code, lino, _exit);
+      TD_SMA_LOOPS_CHECK(nLoops, 1000)
     }
   }
   // step 2: wait for all triggered fetch tasks to finish
@@ -173,11 +159,7 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
     } else {
       smaDebug("vgId:%d, rsma commit%d, fetch tasks are not all finished yet", SMA_VID(pSma), isCommit);
     }
-    ++nLoops;
-    if (nLoops > 1000) {
-      sched_yield();
-      nLoops = 0;
-    }
+    TD_SMA_LOOPS_CHECK(nLoops, 1000);
   }
 
   /**
@@ -189,40 +171,17 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
           (void *)taosGetSelfPthreadId());
   nLoops = 0;
   while (atomic_load_64(&pRSmaStat->nBufItems) > 0) {
-    ++nLoops;
-    if (nLoops > 1000) {
-      sched_yield();
-      nLoops = 0;
-    }
+    TD_SMA_LOOPS_CHECK(nLoops, 1000);
   }
 
   if (!isCommit) goto _exit;
 
-  smaInfo("vgId:%d, rsma commit, all items are consumed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
-  code = tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));
+  // code = tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));
   TSDB_CHECK_CODE(code, lino, _exit);
 
   smaInfo("vgId:%d, rsma commit, operator state committed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
 
-#if 0  // consuming task of qTaskInfo clone 
-  // step 4:  swap queue/qall and iQueue/iQall
-  // lock
-  taosWLockLatch(SMA_ENV_LOCK(pEnv));
-
-  void *pIter = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), NULL);
-
-  while (pIter) {
-    SRSmaInfo *pInfo = *(SRSmaInfo **)pIter;
-    TSWAP(pInfo->iQall, pInfo->qall);
-    TSWAP(pInfo->iQueue, pInfo->queue);
-    TSWAP(pInfo->iTaskInfo[0], pInfo->taskInfo[0]);
-    TSWAP(pInfo->iTaskInfo[1], pInfo->taskInfo[1]);
-    pIter = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), pIter);
-  }
-
-  // unlock
-  taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
-#endif
+  smaInfo("vgId:%d, rsma commit, all items are consumed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
 
   // all rsma results are written completely
   STsdb *pTsdb = NULL;
@@ -258,9 +217,6 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
     goto _exit;
   }
 
-  code = tdRSmaFSCommit(pSma);
-  TSDB_CHECK_CODE(code, lino, _exit);
-
   code = tsdbCommit(VND_RSMA1(pVnode), pInfo);
   TSDB_CHECK_CODE(code, lino, _exit);
 
@@ -310,20 +266,6 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
 
         continue;
       }
-#if 0
-    if (pRSmaInfo->taskInfo[0]) {
-      if (pRSmaInfo->iTaskInfo[0]) {
-        SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)pRSmaInfo->iTaskInfo[0];
-        tdFreeRSmaInfo(pSma, pRSmaInfo, false);
-        pRSmaInfo->iTaskInfo[0] = NULL;
-      }
-    } else {
-      TSWAP(pRSmaInfo->taskInfo[0], pRSmaInfo->iTaskInfo[0]);
-    }
-
-    taosHashPut(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t), pIter, sizeof(pIter));
-    smaDebug("vgId:%d, rsma async post commit, migrated from iRsmaInfoHash for table:%" PRIi64, SMA_VID(pSma), *pSuid);
-#endif
     }
 
     // unlock

source/dnode/vnode/src/sma/smaEnv.c

@@ -30,7 +30,6 @@ static int32_t tdRsmaStartExecutor(const SSma *pSma);
 static int32_t tdRsmaStopExecutor(const SSma *pSma);
 static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
 static void   *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
-static void   *tdFreeTSmaStat(STSmaStat *pStat);
 static void    tdDestroyRSmaStat(void *pRSmaStat);
 
 /**
@@ -63,19 +62,15 @@ int32_t smaInit() {
 
     int32_t type = (8 == POINTER_BYTES) ? TSDB_DATA_TYPE_UBIGINT : TSDB_DATA_TYPE_UINT;
     smaMgmt.refHash = taosHashInit(64, taosGetDefaultHashFunction(type), true, HASH_ENTRY_LOCK);
-    if (!smaMgmt.refHash) {
-      taosCloseRef(smaMgmt.rsetId);
-      atomic_store_8(&smaMgmt.inited, 0);
-      smaError("failed to init sma tmr hanle since %s", terrstr());
-      return TSDB_CODE_FAILED;
-    }
-
     // init fetch timer handle
     smaMgmt.tmrHandle = taosTmrInit(10000, 100, 10000, "RSMA");
-    if (!smaMgmt.tmrHandle) {
+
+    if (!smaMgmt.refHash || !smaMgmt.tmrHandle) {
       taosCloseRef(smaMgmt.rsetId);
-      taosHashCleanup(smaMgmt.refHash);
-      smaMgmt.refHash = NULL;
+      if (smaMgmt.refHash) {
+        taosHashCleanup(smaMgmt.refHash);
+        smaMgmt.refHash = NULL;
+      }
       atomic_store_8(&smaMgmt.inited, 0);
       smaError("failed to init sma tmr handle since %s", terrstr());
       return TSDB_CODE_FAILED;
@@ -143,10 +138,6 @@ static int32_t tdNewSmaEnv(SSma *pSma, int8_t smaType, SSmaEnv **ppEnv) {
 }
 
 static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, SSmaEnv **ppEnv) {
-  if (!ppEnv) {
-    terrno = TSDB_CODE_INVALID_PTR;
-    return TSDB_CODE_FAILED;
-  }
 
   if (!(*ppEnv)) {
     if (tdNewSmaEnv(pSma, smaType, ppEnv) != TSDB_CODE_SUCCESS) {
@@ -196,10 +187,6 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pS
   int32_t code = 0;
   int32_t lino = 0;
 
-  if (ASSERTS(pSmaStat != NULL, "pSmaStat is NULL")) {
-    terrno = TSDB_CODE_RSMA_INVALID_ENV;
-    TSDB_CHECK_CODE(code, lino, _exit);
-  }
 
   if (*pSmaStat) {  // no lock
     return code;    // success, return directly
@@ -255,15 +242,13 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pS
       taosInitRWLatch(RSMA_FS_LOCK(pRSmaStat));
     } else if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
       // TODO
-    } else {
-      ASSERTS(0, "unknown smaType:%" PRIi8, smaType);
-      code = TSDB_CODE_APP_ERROR;
-      TSDB_CHECK_CODE(code, lino, _exit);
     }
   }
 _exit:
   if (code) {
     smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pSma), __func__, lino, tstrerror(code));
+  } else {
+    smaDebug("vgId:%d, %s succeed, type:%" PRIi8, SMA_VID(pSma), __func__, smaType);
   }
   return code;
 }
@@ -277,12 +262,6 @@ static void tdDestroyTSmaStat(STSmaStat *pStat) {
   }
 }
 
-static void *tdFreeTSmaStat(STSmaStat *pStat) {
-  tdDestroyTSmaStat(pStat);
-  taosMemoryFreeClear(pStat);
-  return NULL;
-}
-
 static void tdDestroyRSmaStat(void *pRSmaStat) {
   if (pRSmaStat) {
     SRSmaStat *pStat = (SRSmaStat *)pRSmaStat;
@@ -300,11 +279,7 @@ static void tdDestroyRSmaStat(void *pRSmaStat) {
       } else {
         smaDebug("vgId:%d, rsma fetch tasks are not all finished yet", SMA_VID(pSma));
       }
-      ++nLoops;
-      if (nLoops > 1000) {
-        sched_yield();
-        nLoops = 0;
-      }
+      TD_SMA_LOOPS_CHECK(nLoops, 1000);
     }
 
     // step 3:
@@ -313,10 +288,7 @@ static void tdDestroyRSmaStat(void *pRSmaStat) {
     // step 4: destroy the rsma info and associated fetch tasks
     taosHashCleanup(RSMA_INFO_HASH(pStat));
 
-    // step 5:
-    tdRSmaFSClose(RSMA_FS(pStat));
-
-    // step 6: free pStat
+    // step 5: free pStat
     tsem_destroy(&(pStat->notEmpty));
     taosMemoryFreeClear(pStat);
   }
@@ -354,10 +326,7 @@ static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
         smaDebug("vgId:%d, remove refId:%" PRIi64 " from rsmaRef:%" PRIi32 " succeed", vid, refId, smaMgmt.rsetId);
       }
     } else {
-      ASSERTS(0, "unknown smaType:%" PRIi8, smaType);
-      terrno = TSDB_CODE_APP_ERROR;
-      smaError("%s failed at line %d since %s", __func__, __LINE__, terrstr());
-      return -1;
+      smaError("%s failed at line %d since Unknown type", __func__, __LINE__);
     }
   }
   return 0;
@@ -375,11 +344,6 @@ int32_t tdLockSma(SSma *pSma) {
 }
 
 int32_t tdUnLockSma(SSma *pSma) {
-  if (ASSERTS(SMA_LOCKED(pSma), "pSma %p is not locked:%d", pSma, pSma->locked)) {
-    terrno = TSDB_CODE_APP_ERROR;
-    smaError("vgId:%d, failed to unlock since %s", SMA_VID(pSma), tstrerror(terrno));
-    return -1;
-  }
 
   pSma->locked = false;
   int code = taosThreadMutexUnlock(&pSma->mutex);