xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: init once

This commit is contained in:
Liu Jicong 2022-05-23 20:40:37 +08:00
parent 3629958b43
commit 0ad9c4cebd
3 changed files with 47 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/vnode/src/inc/tq.h
View File

@ -104,7 +104,7 @@ typedef struct {
tmr_h timer; tmr_h timer;
} STqMgmt; } STqMgmt;
static STqMgmt tqMgmt; static STqMgmt tqMgmt = {0};
// init once // init once
int tqInit(); int tqInit();

18
source/dnode/vnode/src/tq/tq.c
View File

@ -16,23 +16,35 @@
#include "tq.h" #include "tq.h"
int32_t tqInit() { 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) { if (old == 0) {
tqMgmt.timer = taosTmrInit(10000, 100, 10000, "TQ"); tqMgmt.timer = taosTmrInit(10000, 100, 10000, "TQ");
if (tqMgmt.timer == NULL) { if (tqMgmt.timer == NULL) {
atomic_store_8(&tqMgmt.inited, 0); atomic_store_8(&tqMgmt.inited, 0);
return -1; return -1;
} }
atomic_store_8(&tqMgmt.inited, 1);
} }
return 0; return 0;
} }
void tqCleanUp() { void tqCleanUp() {
int8_t old = atomic_val_compare_exchange_8(&tqMgmt.inited, 1, 2); int8_t old;
if (old != 1) return; while (1) {
old = atomic_val_compare_exchange_8(&tqMgmt.inited, 1, 2);
if (old != 2) break;
}
if (old == 1) {
taosTmrCleanUp(tqMgmt.timer); taosTmrCleanUp(tqMgmt.timer);
atomic_store_8(&tqMgmt.inited, 0); atomic_store_8(&tqMgmt.inited, 0);
} }
}
STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) { STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
STQ* pTq = taosMemoryMalloc(sizeof(STQ)); STQ* pTq = taosMemoryMalloc(sizeof(STQ));

21
source/libs/wal/src/walMgmt.c
View File

@ -36,9 +36,13 @@ static void walFreeObj(void *pWal);
int64_t walGetSeq() { return (int64_t)atomic_load_32(&tsWal.seq); } int64_t walGetSeq() { return (int64_t)atomic_load_32(&tsWal.seq); }
int32_t walInit() { int32_t walInit() {
int8_t old = atomic_val_compare_exchange_8(&tsWal.inited, 0, 1); int8_t old;
if (old == 1) return 0; 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); tsWal.refSetId = taosOpenRef(TSDB_MIN_VNODES, walFreeObj);
int32_t code = walCreateThread(); int32_t code = walCreateThread();
@ -49,19 +53,26 @@ int32_t walInit() {
} }
wInfo("wal module is initialized, rsetId:%d", tsWal.refSetId); wInfo("wal module is initialized, rsetId:%d", tsWal.refSetId);
atomic_store_8(&tsWal.inited, 1);
}
return 0; return 0;
} }
void walCleanUp() { void walCleanUp() {
int8_t old = atomic_val_compare_exchange_8(&tsWal.inited, 1, 2); int8_t old;
if (old != 1) { while (1) {
return; old = atomic_val_compare_exchange_8(&tsWal.inited, 1, 2);
if (old != 2) break;
} }
if (old == 1) {
walStopThread(); walStopThread();
taosCloseRef(tsWal.refSetId); taosCloseRef(tsWal.refSetId);
wInfo("wal module is cleaned up"); wInfo("wal module is cleaned up");
atomic_store_8(&tsWal.inited, 0); atomic_store_8(&tsWal.inited, 0);
} }
}
SWal *walOpen(const char *path, SWalCfg *pCfg) { SWal *walOpen(const char *path, SWalCfg *pCfg) {
SWal *pWal = taosMemoryMalloc(sizeof(SWal)); SWal *pWal = taosMemoryMalloc(sizeof(SWal));