refactor code

source/dnode/vnode/src/tq/tqRestore.c

@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "vnd.h"
 #include "tq.h"
+#include "vnd.h"
 
 static int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle);
 static int32_t doSetOffsetForWalReader(SStreamTask* pTask, int32_t vgId);
@@ -171,8 +171,8 @@ int32_t doSetOffsetForWalReader(SStreamTask *pTask, int32_t vgId) {
   // seek the stored version and extract data from WAL
   int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader);
   if (pTask->chkInfo.currentVer < firstVer) {
-    tqWarn("vgId:%d s-task:%s ver:%"PRId64" earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64, vgId,
+    tqWarn("vgId:%d s-task:%s ver:%" PRId64 " earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64,
-           pTask->id.idStr, pTask->chkInfo.currentVer, firstVer, firstVer);
+           vgId, pTask->id.idStr, pTask->chkInfo.currentVer, firstVer, firstVer);
 
     pTask->chkInfo.currentVer = firstVer;
 
@@ -193,7 +193,8 @@ int32_t doSetOffsetForWalReader(SStreamTask *pTask, int32_t vgId) {
       }
 
       // append the data for the stream
-      tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer);
+      tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr,
+              pTask->chkInfo.currentVer);
     }
   }
 
@@ -321,6 +322,7 @@ int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
 
     if (pTask->status.taskStatus != TASK_STATUS__NORMAL) {
       tqDebug("s-task:%s not ready for submit block from wal, status:%s", pTask->id.idStr, pStatus);
+      streamMetaReleaseTask(pStreamMeta, pTask);
       taosThreadMutexUnlock(&pTask->lock);
       continue;
     }
@@ -360,4 +362,3 @@ int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
   taosArrayDestroy(pTaskList);
   return 0;
 }
-

source/libs/stream/src/streamBackendRocksdb.c

@@ -439,7 +439,7 @@ int32_t rebuildDirFromCheckpoint(const char* path, int64_t chkpId, char** dst) {
       if (code != 0) {
         qError("failed to restart stream backend from %s, reason: %s", chkp, tstrerror(TAOS_SYSTEM_ERROR(errno)));
       } else {
-        qInfo("succ to restart stream backend at checkpoint path: %s", chkp);
+        qInfo("start to restart stream backend at checkpoint path: %s", chkp);
       }
 
     } else {
@@ -510,7 +510,11 @@ void* streamBackendInit(const char* streamPath, int64_t chkpId) {
     /*
       list all cf and get prefix
     */
-    streamStateOpenBackendCf(pHandle, (char*)backendPath, cfs, nCf);
+    code = streamStateOpenBackendCf(pHandle, (char*)backendPath, cfs, nCf);
+    if (code != 0) {
+      rocksdb_list_column_families_destroy(cfs, nCf);
+      goto _EXIT;
+    }
   }
   if (cfs != NULL) {
     rocksdb_list_column_families_destroy(cfs, nCf);
@@ -545,16 +549,6 @@ void streamBackendCleanup(void* arg) {
   taosHashCleanup(pHandle->cfInst);
 
   if (pHandle->db) {
-    // char*                   err = NULL;
-    //  rocksdb_flushoptions_t* flushOpt = rocksdb_flushoptions_create();
-    //  rocksdb_flushoptions_set_wait(flushOpt, 1);
-    //  rocksdb_flush(pHandle->db, flushOpt, &err);
-
-    // if (err != NULL) {
-    //   qError("failed to flush db before streamBackend clean up, reason:%s", err);
-    //   taosMemoryFree(err);
-    // }
-    // rocksdb_flushoptions_destroy(flushOpt);
     rocksdb_close(pHandle->db);
   }
   rocksdb_options_destroy(pHandle->dbOpt);
@@ -1480,6 +1474,12 @@ int32_t streamStateOpenBackendCf(void* backend, char* name, char** cfs, int32_t
   if (err != NULL) {
     qError("failed to open rocksdb cf, reason:%s", err);
     taosMemoryFree(err);
+    taosMemoryFree(cfHandle);
+    taosMemoryFree(pCompare);
+    taosMemoryFree(params);
+    taosMemoryFree(cfOpts);
+    // fix other leak
+    return -1;
   } else {
     qDebug("succ to open rocksdb cf");
   }
@@ -2851,7 +2851,7 @@ int32_t streamStatePutBatch_rocksdb(SStreamState* pState, void* pBatch) {
     taosMemoryFree(err);
     return -1;
   } else {
-    qDebug("write batch to backend opt: %p", wrapper->pBackend);
+    qDebug("write batch to backend:%p", wrapper->pBackend);
   }
   return 0;
 }

source/libs/stream/src/streamMeta.c

@@ -107,8 +107,10 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
     pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId);
     if (pMeta->streamBackend == NULL) {
       qError("vgId:%d failed to init stream backend", pMeta->vgId);
+      qInfo("vgId:%d retry to init stream backend", pMeta->vgId);
     }
   }
+
   // if (pMeta->streamBackend == NULL) {
   //   goto _err;
   // }
@@ -172,6 +174,7 @@ int32_t streamMetaReopen(SStreamMeta* pMeta, int64_t chkpId) {
     pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId);
     if (pMeta->streamBackend == NULL) {
       qError("vgId:%d failed to init stream backend", pMeta->vgId);
+      qInfo("vgId:%d retry to init stream backend", pMeta->vgId);
       // return -1;
     }
   }