From 0ad9c4cebd4a5eb4fcb62768e940d84b9ad8615c Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Mon, 23 May 2022 20:40:37 +0800
Subject: [PATCH] fix: init once

---
 source/dnode/vnode/src/inc/tq.h |  2 +-
 source/dnode/vnode/src/tq/tq.c  | 22 +++++++++++----
 source/libs/wal/src/walMgmt.c   | 47 ++++++++++++++++++++-------------
 3 files changed, 47 insertions(+), 24 deletions(-)

diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h
index 40b490da47..ad3f8cc869 100644
--- a/source/dnode/vnode/src/inc/tq.h
+++ b/source/dnode/vnode/src/inc/tq.h
@@ -104,7 +104,7 @@ typedef struct {
   tmr_h  timer;
 } STqMgmt;
 
-static STqMgmt tqMgmt;
+static STqMgmt tqMgmt = {0};
 
 // init once
 int  tqInit();
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index f7b4ba93a6..bd48ed9b4c 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -16,22 +16,34 @@
 #include "tq.h"
 
 int32_t tqInit() {
-  int8_t old = atomic_val_compare_exchange_8(&tqMgmt.inited, 0, 1);
+  int8_t old;
+  while (1) {
+    old = atomic_val_compare_exchange_8(&tqMgmt.inited, 0, 2);
+    if (old != 2) break;
+  }
+
   if (old == 0) {
     tqMgmt.timer = taosTmrInit(10000, 100, 10000, "TQ");
     if (tqMgmt.timer == NULL) {
       atomic_store_8(&tqMgmt.inited, 0);
       return -1;
     }
+    atomic_store_8(&tqMgmt.inited, 1);
   }
   return 0;
 }
 
 void tqCleanUp() {
-  int8_t old = atomic_val_compare_exchange_8(&tqMgmt.inited, 1, 2);
-  if (old != 1) return;
-  taosTmrCleanUp(tqMgmt.timer);
-  atomic_store_8(&tqMgmt.inited, 0);
+  int8_t old;
+  while (1) {
+    old = atomic_val_compare_exchange_8(&tqMgmt.inited, 1, 2);
+    if (old != 2) break;
+  }
+
+  if (old == 1) {
+    taosTmrCleanUp(tqMgmt.timer);
+    atomic_store_8(&tqMgmt.inited, 0);
+  }
 }
 
 STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
diff --git a/source/libs/wal/src/walMgmt.c b/source/libs/wal/src/walMgmt.c
index f2f423ddf5..71cd6de73f 100644
--- a/source/libs/wal/src/walMgmt.c
+++ b/source/libs/wal/src/walMgmt.c
@@ -36,31 +36,42 @@ static void     walFreeObj(void *pWal);
 int64_t walGetSeq() { return (int64_t)atomic_load_32(&tsWal.seq); }
 
 int32_t walInit() {
-  int8_t old = atomic_val_compare_exchange_8(&tsWal.inited, 0, 1);
-  if (old == 1) return 0;
-
-  tsWal.refSetId = taosOpenRef(TSDB_MIN_VNODES, walFreeObj);
-
-  int32_t code = walCreateThread();
-  if (code != 0) {
-    wError("failed to init wal module since %s", tstrerror(code));
-    atomic_store_8(&tsWal.inited, 0);
-    return code;
+  int8_t old;
+  while (1) {
+    old = atomic_val_compare_exchange_8(&tsWal.inited, 0, 2);
+    if (old != 2) break;
+  }
+
+  if (old == 0) {
+    tsWal.refSetId = taosOpenRef(TSDB_MIN_VNODES, walFreeObj);
+
+    int32_t code = walCreateThread();
+    if (code != 0) {
+      wError("failed to init wal module since %s", tstrerror(code));
+      atomic_store_8(&tsWal.inited, 0);
+      return code;
+    }
+
+    wInfo("wal module is initialized, rsetId:%d", tsWal.refSetId);
+    atomic_store_8(&tsWal.inited, 1);
   }
 
-  wInfo("wal module is initialized, rsetId:%d", tsWal.refSetId);
   return 0;
 }
 
 void walCleanUp() {
-  int8_t old = atomic_val_compare_exchange_8(&tsWal.inited, 1, 2);
-  if (old != 1) {
-    return;
+  int8_t old;
+  while (1) {
+    old = atomic_val_compare_exchange_8(&tsWal.inited, 1, 2);
+    if (old != 2) break;
+  }
+
+  if (old == 1) {
+    walStopThread();
+    taosCloseRef(tsWal.refSetId);
+    wInfo("wal module is cleaned up");
+    atomic_store_8(&tsWal.inited, 0);
   }
-  walStopThread();
-  taosCloseRef(tsWal.refSetId);
-  wInfo("wal module is cleaned up");
-  atomic_store_8(&tsWal.inited, 0);
 }
 
 SWal *walOpen(const char *path, SWalCfg *pCfg) {