From 4fc0e3bd55ecac34dc05156fbbe38fd55dfc61dc Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Fri, 4 Nov 2022 17:21:00 +0800 Subject: [PATCH] enh: adjust the number of vnode threads so that one vnode has one write thread --- docs/en/14-reference/12-config/index.md | 24 +++--- docs/zh/14-reference/12-config/index.md | 24 +++--- include/common/tglobal.h | 2 - include/util/tworker.h | 12 +-- source/common/src/tglobal.c | 32 -------- source/dnode/mgmt/mgmt_vnode/inc/vmInt.h | 30 +++---- source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 24 +++--- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 87 ++++++--------------- source/util/src/tworker.c | 2 - 9 files changed, 79 insertions(+), 158 deletions(-) diff --git a/docs/en/14-reference/12-config/index.md b/docs/en/14-reference/12-config/index.md index 769d19e7f9..aadf563a63 100644 --- a/docs/en/14-reference/12-config/index.md +++ b/docs/en/14-reference/12-config/index.md @@ -735,19 +735,17 @@ To prevent system resource from being exhausted by multiple concurrent streams, | 44 | numOfVnodeQueryThreads | No | Yes | | 45 | numOfVnodeStreamThreads | No | Yes | | 46 | numOfVnodeFetchThreads | No | Yes | -| 47 | numOfVnodeWriteThreads | No | Yes | -| 48 | numOfVnodeSyncThreads | No | Yes | -| 49 | numOfVnodeRsmaThreads | No | Yes | -| 50 | numOfQnodeQueryThreads | No | Yes | -| 51 | numOfQnodeFetchThreads | No | Yes | -| 52 | numOfSnodeSharedThreads | No | Yes | -| 53 | numOfSnodeUniqueThreads | No | Yes | -| 54 | rpcQueueMemoryAllowed | No | Yes | -| 55 | logDir | Yes | Yes | -| 56 | minimalLogDirGB | Yes | Yes | -| 57 | numOfLogLines | Yes | Yes | -| 58 | asyncLog | Yes | Yes | -| 59 | logKeepDays | Yes | Yes | +| 47 | numOfVnodeRsmaThreads | No | Yes | +| 48 | numOfQnodeQueryThreads | No | Yes | +| 49 | numOfQnodeFetchThreads | No | Yes | +| 50 | numOfSnodeSharedThreads | No | Yes | +| 51 | numOfSnodeUniqueThreads | No | Yes | +| 52 | rpcQueueMemoryAllowed | No | Yes | +| 53 | logDir | Yes | Yes | +| 54 | minimalLogDirGB | Yes | Yes | +| 55 | numOfLogLines | Yes | Yes | +| 56 | asyncLog | Yes | Yes | +| 57 | logKeepDays | Yes | Yes | | 60 | debugFlag | Yes | Yes | | 61 | tmrDebugFlag | Yes | Yes | | 62 | uDebugFlag | Yes | Yes | diff --git a/docs/zh/14-reference/12-config/index.md b/docs/zh/14-reference/12-config/index.md index d587dd2f2c..54c362de95 100644 --- a/docs/zh/14-reference/12-config/index.md +++ b/docs/zh/14-reference/12-config/index.md @@ -711,19 +711,17 @@ charset 的有效值是 UTF-8。 | 44 | numOfVnodeQueryThreads | 否 | 是 | | | 45 | numOfVnodeStreamThreads | 否 | 是 | | | 46 | numOfVnodeFetchThreads | 否 | 是 | | -| 47 | numOfVnodeWriteThreads | 否 | 是 | | -| 48 | numOfVnodeSyncThreads | 否 | 是 | | -| 49 | numOfVnodeRsmaThreads | 否 | 是 | | -| 50 | numOfQnodeQueryThreads | 否 | 是 | | -| 51 | numOfQnodeFetchThreads | 否 | 是 | | -| 52 | numOfSnodeSharedThreads | 否 | 是 | | -| 53 | numOfSnodeUniqueThreads | 否 | 是 | | -| 54 | rpcQueueMemoryAllowed | 否 | 是 | | -| 55 | logDir | 是 | 是 | | -| 56 | minimalLogDirGB | 是 | 是 | | -| 57 | numOfLogLines | 是 | 是 | | -| 58 | asyncLog | 是 | 是 | | -| 59 | logKeepDays | 是 | 是 | | +| 47 | numOfVnodeRsmaThreads | 否 | 是 | | +| 48 | numOfQnodeQueryThreads | 否 | 是 | | +| 49 | numOfQnodeFetchThreads | 否 | 是 | | +| 50 | numOfSnodeSharedThreads | 否 | 是 | | +| 51 | numOfSnodeUniqueThreads | 否 | 是 | | +| 52 | rpcQueueMemoryAllowed | 否 | 是 | | +| 53 | logDir | 是 | 是 | | +| 54 | minimalLogDirGB | 是 | 是 | | +| 55 | numOfLogLines | 是 | 是 | | +| 56 | asyncLog | 是 | 是 | | +| 57 | logKeepDays | 是 | 是 | | | 60 | debugFlag | 是 | 是 | | | 61 | tmrDebugFlag | 是 | 是 | | | 62 | uDebugFlag | 是 | 是 | | diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 99bbfde3e1..681d1beb79 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -55,8 +55,6 @@ extern int32_t tsNumOfMnodeReadThreads; extern int32_t tsNumOfVnodeQueryThreads; extern int32_t tsNumOfVnodeStreamThreads; extern int32_t tsNumOfVnodeFetchThreads; -extern int32_t tsNumOfVnodeWriteThreads; -extern int32_t tsNumOfVnodeSyncThreads; extern int32_t tsNumOfVnodeRsmaThreads; extern int32_t tsNumOfQnodeQueryThreads; extern int32_t tsNumOfQnodeFetchThreads; diff --git a/include/util/tworker.h b/include/util/tworker.h index 6b439d0c29..8766f87a08 100644 --- a/include/util/tworker.h +++ b/include/util/tworker.h @@ -26,9 +26,9 @@ typedef struct SQWorkerPool SQWorkerPool; typedef struct SWWorkerPool SWWorkerPool; typedef struct SQWorker { - int32_t id; // worker ID - TdThread thread; // thread - int64_t pid; + int32_t id; // worker id + int64_t pid; // thread pid + TdThread thread; // thread id SQWorkerPool *pool; } SQWorker; @@ -44,10 +44,10 @@ typedef struct SQWorkerPool { typedef struct SWWorker { int32_t id; // worker id - TdThread thread; // thread - int64_t pid; + int64_t pid; // thread pid + TdThread thread; // thread id STaosQall *qall; - STaosQset *qset; // queue set + STaosQset *qset; SWWorkerPool *pool; } SWWorker; diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 7f4a826c5e..064249d754 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -50,8 +50,6 @@ int32_t tsNumOfMnodeReadThreads = 1; int32_t tsNumOfVnodeQueryThreads = 4; int32_t tsNumOfVnodeStreamThreads = 2; int32_t tsNumOfVnodeFetchThreads = 4; -int32_t tsNumOfVnodeWriteThreads = 2; -int32_t tsNumOfVnodeSyncThreads = 2; int32_t tsNumOfVnodeRsmaThreads = 2; int32_t tsNumOfQnodeQueryThreads = 4; int32_t tsNumOfQnodeFetchThreads = 1; @@ -374,14 +372,6 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4); if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 4, 1024, 0) != 0) return -1; - tsNumOfVnodeWriteThreads = tsNumOfCores; - tsNumOfVnodeWriteThreads = TMAX(tsNumOfVnodeWriteThreads, 1); - if (cfgAddInt32(pCfg, "numOfVnodeWriteThreads", tsNumOfVnodeWriteThreads, 1, 1024, 0) != 0) return -1; - - tsNumOfVnodeSyncThreads = tsNumOfCores * 2; - tsNumOfVnodeSyncThreads = TMAX(tsNumOfVnodeSyncThreads, 16); - if (cfgAddInt32(pCfg, "numOfVnodeSyncThreads", tsNumOfVnodeSyncThreads, 1, 1024, 0) != 0) return -1; - tsNumOfVnodeRsmaThreads = tsNumOfCores; tsNumOfVnodeRsmaThreads = TMAX(tsNumOfVnodeRsmaThreads, 4); if (cfgAddInt32(pCfg, "numOfVnodeRsmaThreads", tsNumOfVnodeRsmaThreads, 1, 1024, 0) != 0) return -1; @@ -506,22 +496,6 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) { pItem->stype = stype; } - pItem = cfgGetItem(tsCfg, "numOfVnodeWriteThreads"); - if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { - tsNumOfVnodeWriteThreads = numOfCores; - tsNumOfVnodeWriteThreads = TMAX(tsNumOfVnodeWriteThreads, 1); - pItem->i32 = tsNumOfVnodeWriteThreads; - pItem->stype = stype; - } - - pItem = cfgGetItem(tsCfg, "numOfVnodeSyncThreads"); - if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { - tsNumOfVnodeSyncThreads = numOfCores * 2; - tsNumOfVnodeSyncThreads = TMAX(tsNumOfVnodeSyncThreads, 16); - pItem->i32 = tsNumOfVnodeSyncThreads; - pItem->stype = stype; - } - pItem = cfgGetItem(tsCfg, "numOfVnodeRsmaThreads"); if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { tsNumOfVnodeRsmaThreads = numOfCores; @@ -699,8 +673,6 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsNumOfVnodeQueryThreads = cfgGetItem(pCfg, "numOfVnodeQueryThreads")->i32; tsNumOfVnodeStreamThreads = cfgGetItem(pCfg, "numOfVnodeStreamThreads")->i32; tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; - tsNumOfVnodeWriteThreads = cfgGetItem(pCfg, "numOfVnodeWriteThreads")->i32; - tsNumOfVnodeSyncThreads = cfgGetItem(pCfg, "numOfVnodeSyncThreads")->i32; tsNumOfVnodeRsmaThreads = cfgGetItem(pCfg, "numOfVnodeRsmaThreads")->i32; tsNumOfQnodeQueryThreads = cfgGetItem(pCfg, "numOfQnodeQueryThreads")->i32; // tsNumOfQnodeFetchThreads = cfgGetItem(pCfg, "numOfQnodeFetchThreads")->i32; @@ -943,10 +915,6 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { } else if (strcasecmp("numOfVnodeFetchThreads", name) == 0) { tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; */ - } else if (strcasecmp("numOfVnodeWriteThreads", name) == 0) { - tsNumOfVnodeWriteThreads = cfgGetItem(pCfg, "numOfVnodeWriteThreads")->i32; - } else if (strcasecmp("numOfVnodeSyncThreads", name) == 0) { - tsNumOfVnodeSyncThreads = cfgGetItem(pCfg, "numOfVnodeSyncThreads")->i32; } else if (strcasecmp("numOfVnodeRsmaThreads", name) == 0) { tsNumOfVnodeRsmaThreads = cfgGetItem(pCfg, "numOfVnodeRsmaThreads")->i32; } else if (strcasecmp("numOfQnodeQueryThreads", name) == 0) { diff --git a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h index bf1ccc1a7b..b38dc19361 100644 --- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h +++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h @@ -33,10 +33,6 @@ typedef struct SVnodeMgmt { SQWorkerPool queryPool; SQWorkerPool streamPool; SWWorkerPool fetchPool; - SWWorkerPool syncPool; - SWWorkerPool syncCtrlPool; - SWWorkerPool writePool; - SWWorkerPool applyPool; SSingleWorker mgmtWorker; SHashObj *hash; TdThreadRwlock lock; @@ -52,19 +48,19 @@ typedef struct { } SWrapperCfg; typedef struct { - int32_t vgId; - int32_t vgVersion; - int32_t refCount; - int8_t dropped; - char *path; - SVnode *pImpl; - STaosQueue *pWriteQ; - STaosQueue *pSyncQ; - STaosQueue *pSyncCtrlQ; - STaosQueue *pApplyQ; - STaosQueue *pQueryQ; - STaosQueue *pStreamQ; - STaosQueue *pFetchQ; + int32_t vgId; + int32_t vgVersion; + int32_t refCount; + int8_t dropped; + char *path; + SVnode *pImpl; + SMultiWorker pWriteW; + SMultiWorker pSyncW; + SMultiWorker pSyncCtrlW; + SMultiWorker pApplyW; + STaosQueue *pQueryQ; + STaosQueue *pStreamQ; + STaosQueue *pFetchQ; } SVnodeObj; typedef struct { diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c index 0d476e30ab..07ebd72379 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c @@ -90,21 +90,21 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { dInfo("vgId:%d, wait for vnode ref become 0", pVnode->vgId); while (pVnode->refCount > 0) taosMsleep(10); - dInfo("vgId:%d, wait for vnode write queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteQ, - pVnode->pWriteQ->threadId); - while (!taosQueueEmpty(pVnode->pWriteQ)) taosMsleep(10); + dInfo("vgId:%d, wait for vnode write queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue, + pVnode->pWriteW.queue->threadId); + tMultiWorkerCleanup(&pVnode->pWriteW); - dInfo("vgId:%d, wait for vnode sync queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncQ, - pVnode->pSyncQ->threadId); - while (!taosQueueEmpty(pVnode->pSyncQ)) taosMsleep(10); + dInfo("vgId:%d, wait for vnode sync queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue, + pVnode->pSyncW.queue->threadId); + tMultiWorkerCleanup(&pVnode->pSyncW); - dInfo("vgId:%d, wait for vnode sync ctrl queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncCtrlQ, - pVnode->pSyncCtrlQ->threadId); - while (!taosQueueEmpty(pVnode->pSyncCtrlQ)) taosMsleep(10); + dInfo("vgId:%d, wait for vnode sync ctrl queue:%p is empty, thread:%08" PRId64, pVnode->vgId, + pVnode->pSyncCtrlW.queue, pVnode->pSyncCtrlW.queue->threadId); + tMultiWorkerCleanup(&pVnode->pSyncCtrlW); - dInfo("vgId:%d, wait for vnode apply queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyQ, - pVnode->pApplyQ->threadId); - while (!taosQueueEmpty(pVnode->pApplyQ)) taosMsleep(10); + dInfo("vgId:%d, wait for vnode apply queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue, + pVnode->pApplyW.queue->threadId); + tMultiWorkerCleanup(&pVnode->pApplyW); dInfo("vgId:%d, wait for vnode query queue:%p is empty", pVnode->vgId, pVnode->pQueryQ); while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index a5070e09aa..03c242b081 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -188,30 +188,20 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp dDebug("vgId:%d, msg:%p put into vnode-write queue failed since %s", pVnode->vgId, pMsg, terrstr(code)); } else { dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg); - taosWriteQitem(pVnode->pWriteQ, pMsg); -#if 0 // tests for batch writes - if (pMsg->msgType == TDMT_VND_CREATE_TABLE) { - SRpcMsg *pDup = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); - memcpy(pDup, pMsg, sizeof(SRpcMsg)); - pDup->pCont = rpcMallocCont(pMsg->contLen); - memcpy(pDup->pCont, pMsg->pCont, pMsg->contLen); - pDup->info.handle = NULL; - taosWriteQitem(pVnode->pWriteQ, pDup); - } -#endif + taosWriteQitem(pVnode->pWriteW.queue, pMsg); } break; case SYNC_QUEUE: dGTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg); - taosWriteQitem(pVnode->pSyncQ, pMsg); + taosWriteQitem(pVnode->pSyncW.queue, pMsg); break; case SYNC_CTRL_QUEUE: dGTrace("vgId:%d, msg:%p put into vnode-sync-ctrl queue", pVnode->vgId, pMsg); - taosWriteQitem(pVnode->pSyncCtrlQ, pMsg); + taosWriteQitem(pVnode->pSyncCtrlW.queue, pMsg); break; case APPLY_QUEUE: dGTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg); - taosWriteQitem(pVnode->pApplyQ, pMsg); + taosWriteQitem(pVnode->pApplyW.queue, pMsg); break; default: code = -1; @@ -276,13 +266,13 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { if (pVnode != NULL) { switch (qtype) { case WRITE_QUEUE: - size = taosQueueItemSize(pVnode->pWriteQ); + size = taosQueueItemSize(pVnode->pWriteW.queue); break; case SYNC_QUEUE: - size = taosQueueItemSize(pVnode->pSyncQ); + size = taosQueueItemSize(pVnode->pSyncW.queue); break; case APPLY_QUEUE: - size = taosQueueItemSize(pVnode->pApplyQ); + size = taosQueueItemSize(pVnode->pApplyW.queue); break; case QUERY_QUEUE: size = taosQueueItemSize(pVnode->pQueryQ); @@ -306,27 +296,33 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { } int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { - pVnode->pWriteQ = tWWorkerAllocQueue(&pMgmt->writePool, pVnode->pImpl, (FItems)vnodeProposeWriteMsg); - pVnode->pSyncQ = tWWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FItems)vmProcessSyncQueue); - pVnode->pSyncCtrlQ = tWWorkerAllocQueue(&pMgmt->syncCtrlPool, pVnode, (FItems)vmProcessSyncQueue); - pVnode->pApplyQ = tWWorkerAllocQueue(&pMgmt->applyPool, pVnode->pImpl, (FItems)vnodeApplyWriteMsg); + SMultiWorkerCfg wcfg = {.max = 1, .name = "vnode-write", .fp = (FItems)vnodeProposeWriteMsg, .param = pVnode->pImpl}; + SMultiWorkerCfg scfg = {.max = 1, .name = "vnode-sync", .fp = (FItems)vmProcessSyncQueue, .param = pVnode}; + SMultiWorkerCfg sccfg = {.max = 1, .name = "vnode-sync-ctrl", .fp = (FItems)vmProcessSyncQueue, .param = pVnode}; + SMultiWorkerCfg acfg = {.max = 1, .name = "vnode-apply", .fp = (FItems)vnodeApplyWriteMsg, .param = pVnode}; + (void)tMultiWorkerInit(&pVnode->pWriteW, &wcfg); + (void)tMultiWorkerInit(&pVnode->pSyncW, &scfg); + (void)tMultiWorkerInit(&pVnode->pSyncCtrlW, &sccfg); + (void)tMultiWorkerInit(&pVnode->pApplyW, &acfg); + pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue); pVnode->pStreamQ = tQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue); pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue); - if (pVnode->pWriteQ == NULL || pVnode->pSyncQ == NULL || pVnode->pApplyQ == NULL || pVnode->pQueryQ == NULL || - pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL) { + if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncCtrlW.queue == NULL || + pVnode->pApplyW.queue == NULL || pVnode->pQueryQ == NULL || pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } - dInfo("vgId:%d, write-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteQ, - pVnode->pWriteQ->threadId); - dInfo("vgId:%d, sync-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncQ, pVnode->pSyncQ->threadId); - dInfo("vgId:%d, sync-ctrl-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncCtrlQ, - pVnode->pSyncCtrlQ->threadId); - dInfo("vgId:%d, apply-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyQ, - pVnode->pApplyQ->threadId); + dInfo("vgId:%d, write-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue, + pVnode->pWriteW.queue->threadId); + dInfo("vgId:%d, sync-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue, + pVnode->pSyncW.queue->threadId); + dInfo("vgId:%d, sync-ctrl-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncCtrlW.queue, + pVnode->pSyncCtrlW.queue->threadId); + dInfo("vgId:%d, apply-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue, + pVnode->pApplyW.queue->threadId); dInfo("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ); dInfo("vgId:%d, fetch-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ, pVnode->pFetchQ->threadId); @@ -335,16 +331,9 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { } void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { - tWWorkerFreeQueue(&pMgmt->writePool, pVnode->pWriteQ); - tWWorkerFreeQueue(&pMgmt->applyPool, pVnode->pApplyQ); - tWWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ); - tWWorkerFreeQueue(&pMgmt->syncCtrlPool, pVnode->pSyncCtrlQ); tQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ); tQWorkerFreeQueue(&pMgmt->streamPool, pVnode->pStreamQ); tWWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ); - pVnode->pWriteQ = NULL; - pVnode->pSyncQ = NULL; - pVnode->pApplyQ = NULL; pVnode->pQueryQ = NULL; pVnode->pStreamQ = NULL; pVnode->pFetchQ = NULL; @@ -369,26 +358,6 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { pFPool->max = tsNumOfVnodeFetchThreads; if (tWWorkerInit(pFPool) != 0) return -1; - SWWorkerPool *pWPool = &pMgmt->writePool; - pWPool->name = "vnode-write"; - pWPool->max = tsNumOfVnodeWriteThreads; - if (tWWorkerInit(pWPool) != 0) return -1; - - SWWorkerPool *pAPool = &pMgmt->applyPool; - pAPool->name = "vnode-apply"; - pAPool->max = tsNumOfVnodeWriteThreads; - if (tWWorkerInit(pAPool) != 0) return -1; - - SWWorkerPool *pSPool = &pMgmt->syncPool; - pSPool->name = "vnode-sync"; - pSPool->max = tsNumOfVnodeSyncThreads; - if (tWWorkerInit(pSPool) != 0) return -1; - - SWWorkerPool *pSCPool = &pMgmt->syncCtrlPool; - pSCPool->name = "vnode-sync-ctrl"; - pSCPool->max = tsNumOfVnodeSyncThreads; - if (tWWorkerInit(pSCPool) != 0) return -1; - SSingleWorkerCfg mgmtCfg = { .min = 1, .max = 1, @@ -403,10 +372,6 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { } void vmStopWorker(SVnodeMgmt *pMgmt) { - tWWorkerCleanup(&pMgmt->writePool); - tWWorkerCleanup(&pMgmt->applyPool); - tWWorkerCleanup(&pMgmt->syncPool); - tWWorkerCleanup(&pMgmt->syncCtrlPool); tQWorkerCleanup(&pMgmt->queryPool); tQWorkerCleanup(&pMgmt->streamPool); tWWorkerCleanup(&pMgmt->fetchPool); diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c index a5f7da235a..5971033bf8 100644 --- a/source/util/src/tworker.c +++ b/source/util/src/tworker.c @@ -43,7 +43,6 @@ int32_t tQWorkerInit(SQWorkerPool *pool) { void tQWorkerCleanup(SQWorkerPool *pool) { for (int32_t i = 0; i < pool->max; ++i) { SQWorker *worker = pool->workers + i; - // if (worker == NULL) continue; if (taosCheckPthreadValid(worker->thread)) { taosQsetThreadResume(pool->qset); } @@ -51,7 +50,6 @@ void tQWorkerCleanup(SQWorkerPool *pool) { for (int32_t i = 0; i < pool->max; ++i) { SQWorker *worker = pool->workers + i; - // if (worker == NULL) continue; if (taosCheckPthreadValid(worker->thread)) { taosThreadJoin(worker->thread, NULL); taosThreadClear(&worker->thread);