From 4bc9284a76e1e656f1d9673f90452dfae948353f Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Fri, 6 Nov 2020 09:31:29 +0800 Subject: [PATCH] TD-1915 --- src/dnode/inc/dnodeMPeer.h | 10 +-- src/dnode/inc/dnodeMRead.h | 10 +-- src/dnode/inc/dnodeMWrite.h | 10 +-- src/dnode/src/dnodeMPeer.c | 60 +++++++++--------- src/dnode/src/dnodeMRead.c | 76 +++++++++++------------ src/dnode/src/dnodeMWrite.c | 74 +++++++++++----------- src/dnode/src/dnodeMain.c | 26 ++++---- src/dnode/src/dnodePeer.c | 42 ++++++------- src/dnode/src/dnodeShell.c | 88 +++++++++++++-------------- src/dnode/src/dnodeVWrite.c | 2 +- src/inc/dnode.h | 32 +++++----- src/mnode/src/mnodeMain.c | 12 ++-- src/mnode/src/mnodeSdb.c | 2 +- src/mnode/src/mnodeTable.c | 24 ++++---- src/mnode/src/mnodeVgroup.c | 10 +-- src/plugins/monitor/src/monitorMain.c | 64 +++++++++---------- src/vnode/src/vnodeCfg.c | 1 - src/vnode/src/vnodeMain.c | 30 ++++----- src/vnode/src/vnodeRead.c | 67 ++++++++++---------- src/vnode/src/vnodeWrite.c | 3 +- 20 files changed, 316 insertions(+), 327 deletions(-) diff --git a/src/dnode/inc/dnodeMPeer.h b/src/dnode/inc/dnodeMPeer.h index 9a48703110..00221baa22 100644 --- a/src/dnode/inc/dnodeMPeer.h +++ b/src/dnode/inc/dnodeMPeer.h @@ -20,11 +20,11 @@ extern "C" { #endif -int32_t dnodeInitMnodePeer(); -void dnodeCleanupMnodePeer(); -int32_t dnodeAllocateMnodePqueue(); -void dnodeFreeMnodePqueue(); -void dnodeDispatchToMnodePeerQueue(SRpcMsg *pMsg); +int32_t dnodeInitMPeer(); +void dnodeCleanupMPeer(); +int32_t dnodeAllocateMPeerQueue(); +void dnodeFreeMPeerQueue(); +void dnodeDispatchToMPeerQueue(SRpcMsg *pMsg); #ifdef __cplusplus } diff --git a/src/dnode/inc/dnodeMRead.h b/src/dnode/inc/dnodeMRead.h index 4e93838b79..8a8e71227d 100644 --- a/src/dnode/inc/dnodeMRead.h +++ b/src/dnode/inc/dnodeMRead.h @@ -20,11 +20,11 @@ extern "C" { #endif -int32_t dnodeInitMnodeRead(); -void dnodeCleanupMnodeRead(); -int32_t dnodeAllocateMnodeRqueue(); -void dnodeFreeMnodeRqueue(); -void dnodeDispatchToMnodeReadQueue(SRpcMsg *rpcMsg); +int32_t dnodeInitMRead(); +void dnodeCleanupMRead(); +int32_t dnodeAllocMReadQueue(); +void dnodeFreeMReadQueue(); +void dnodeDispatchToMReadQueue(SRpcMsg *rpcMsg); #ifdef __cplusplus } diff --git a/src/dnode/inc/dnodeMWrite.h b/src/dnode/inc/dnodeMWrite.h index 498fea81c5..6a3d41bc81 100644 --- a/src/dnode/inc/dnodeMWrite.h +++ b/src/dnode/inc/dnodeMWrite.h @@ -20,11 +20,11 @@ extern "C" { #endif -int32_t dnodeInitMnodeWrite(); -void dnodeCleanupMnodeWrite(); -int32_t dnodeAllocateMnodeWqueue(); -void dnodeFreeMnodeWqueue(); -void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg); +int32_t dnodeInitMWrite(); +void dnodeCleanupMWrite(); +int32_t dnodeAllocMWritequeue(); +void dnodeFreeMWritequeue(); +void dnodeDispatchToMWriteQueue(SRpcMsg *pMsg); #ifdef __cplusplus } diff --git a/src/dnode/src/dnodeMPeer.c b/src/dnode/src/dnodeMPeer.c index 0da7ab74e4..5c49dc259c 100644 --- a/src/dnode/src/dnodeMPeer.c +++ b/src/dnode/src/dnodeMPeer.c @@ -35,44 +35,44 @@ typedef struct { typedef struct { int32_t curNum; int32_t maxNum; - SMPeerWorker *peerWorker; + SMPeerWorker *worker; } SMPeerWorkerPool; -static SMPeerWorkerPool tsMPeerPool; +static SMPeerWorkerPool tsMPeerWP; static taos_qset tsMPeerQset; static taos_queue tsMPeerQueue; -static void *dnodeProcessMnodePeerQueue(void *param); +static void *dnodeProcessMPeerQueue(void *param); -int32_t dnodeInitMnodePeer() { +int32_t dnodeInitMPeer() { tsMPeerQset = taosOpenQset(); - tsMPeerPool.maxNum = 1; - tsMPeerPool.curNum = 0; - tsMPeerPool.peerWorker = (SMPeerWorker *)calloc(sizeof(SMPeerWorker), tsMPeerPool.maxNum); + tsMPeerWP.maxNum = 1; + tsMPeerWP.curNum = 0; + tsMPeerWP.worker = (SMPeerWorker *)calloc(sizeof(SMPeerWorker), tsMPeerWP.maxNum); - if (tsMPeerPool.peerWorker == NULL) return -1; - for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) { - SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; + if (tsMPeerWP.worker == NULL) return -1; + for (int32_t i = 0; i < tsMPeerWP.maxNum; ++i) { + SMPeerWorker *pWorker = tsMPeerWP.worker + i; pWorker->workerId = i; dDebug("dnode mpeer worker:%d is created", i); } - dDebug("dnode mpeer is initialized, workers:%d qset:%p", tsMPeerPool.maxNum, tsMPeerQset); + dDebug("dnode mpeer is initialized, workers:%d qset:%p", tsMPeerWP.maxNum, tsMPeerQset); return 0; } -void dnodeCleanupMnodePeer() { - for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) { - SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; +void dnodeCleanupMPeer() { + for (int32_t i = 0; i < tsMPeerWP.maxNum; ++i) { + SMPeerWorker *pWorker = tsMPeerWP.worker + i; if (pWorker->thread) { taosQsetThreadResume(tsMPeerQset); } dDebug("dnode mpeer worker:%d is closed", i); } - for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) { - SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; + for (int32_t i = 0; i < tsMPeerWP.maxNum; ++i) { + SMPeerWorker *pWorker = tsMPeerWP.worker + i; dDebug("dnode mpeer worker:%d start to join", i); if (pWorker->thread) { pthread_join(pWorker->thread, NULL); @@ -84,44 +84,44 @@ void dnodeCleanupMnodePeer() { taosCloseQset(tsMPeerQset); tsMPeerQset = NULL; - taosTFree(tsMPeerPool.peerWorker); + taosTFree(tsMPeerWP.worker); } -int32_t dnodeAllocateMnodePqueue() { +int32_t dnodeAllocateMPeerQueue() { tsMPeerQueue = taosOpenQueue(); if (tsMPeerQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY; taosAddIntoQset(tsMPeerQset, tsMPeerQueue, NULL); - for (int32_t i = tsMPeerPool.curNum; i < tsMPeerPool.maxNum; ++i) { - SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; + for (int32_t i = tsMPeerWP.curNum; i < tsMPeerWP.maxNum; ++i) { + SMPeerWorker *pWorker = tsMPeerWP.worker + i; pWorker->workerId = i; pthread_attr_t thAttr; pthread_attr_init(&thAttr); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); - if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMnodePeerQueue, pWorker) != 0) { + if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMPeerQueue, pWorker) != 0) { dError("failed to create thread to process mpeer queue, reason:%s", strerror(errno)); } pthread_attr_destroy(&thAttr); - tsMPeerPool.curNum = i + 1; - dDebug("dnode mpeer worker:%d is launched, total:%d", pWorker->workerId, tsMPeerPool.maxNum); + tsMPeerWP.curNum = i + 1; + dDebug("dnode mpeer worker:%d is launched, total:%d", pWorker->workerId, tsMPeerWP.maxNum); } dDebug("dnode mpeer queue:%p is allocated", tsMPeerQueue); return TSDB_CODE_SUCCESS; } -void dnodeFreeMnodePqueue() { +void dnodeFreeMPeerQueue() { dDebug("dnode mpeer queue:%p is freed", tsMPeerQueue); taosCloseQueue(tsMPeerQueue); tsMPeerQueue = NULL; } -void dnodeDispatchToMnodePeerQueue(SRpcMsg *pMsg) { +void dnodeDispatchToMPeerQueue(SRpcMsg *pMsg) { if (!mnodeIsRunning() || tsMPeerQueue == NULL) { dnodeSendRedirectMsg(pMsg, false); rpcFreeCont(pMsg->pCont); @@ -133,12 +133,12 @@ void dnodeDispatchToMnodePeerQueue(SRpcMsg *pMsg) { taosWriteQitem(tsMPeerQueue, TAOS_QTYPE_RPC, pPeer); } -static void dnodeFreeMnodePeerMsg(SMnodeMsg *pPeer) { +static void dnodeFreeMPeerMsg(SMnodeMsg *pPeer) { mnodeCleanupMsg(pPeer); taosFreeQitem(pPeer); } -static void dnodeSendRpcMnodePeerRsp(SMnodeMsg *pPeer, int32_t code) { +static void dnodeSendRpcMPeerRsp(SMnodeMsg *pPeer, int32_t code) { if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return; SRpcMsg rpcRsp = { @@ -149,10 +149,10 @@ static void dnodeSendRpcMnodePeerRsp(SMnodeMsg *pPeer, int32_t code) { }; rpcSendResponse(&rpcRsp); - dnodeFreeMnodePeerMsg(pPeer); + dnodeFreeMPeerMsg(pPeer); } -static void *dnodeProcessMnodePeerQueue(void *param) { +static void *dnodeProcessMPeerQueue(void *param) { SMnodeMsg *pPeerMsg; int32_t type; void * unUsed; @@ -165,7 +165,7 @@ static void *dnodeProcessMnodePeerQueue(void *param) { dDebug("msg:%s will be processed in mpeer queue", taosMsg[pPeerMsg->rpcMsg.msgType]); int32_t code = mnodeProcessPeerReq(pPeerMsg); - dnodeSendRpcMnodePeerRsp(pPeerMsg, code); + dnodeSendRpcMPeerRsp(pPeerMsg, code); } return NULL; diff --git a/src/dnode/src/dnodeMRead.c b/src/dnode/src/dnodeMRead.c index 4ad787f26e..8208a66e1b 100644 --- a/src/dnode/src/dnodeMRead.c +++ b/src/dnode/src/dnodeMRead.c @@ -35,46 +35,46 @@ typedef struct { typedef struct { int32_t curNum; int32_t maxNum; - SMReadWorker *readWorker; + SMReadWorker *worker; } SMReadWorkerPool; -static SMReadWorkerPool tsMReadPool; +static SMReadWorkerPool tsMReadWP; static taos_qset tsMReadQset; static taos_queue tsMReadQueue; -static void *dnodeProcessMnodeReadQueue(void *param); +static void *dnodeProcessMReadQueue(void *param); -int32_t dnodeInitMnodeRead() { +int32_t dnodeInitMRead() { tsMReadQset = taosOpenQset(); - tsMReadPool.maxNum = tsNumOfCores * tsNumOfThreadsPerCore / 2; - tsMReadPool.maxNum = MAX(2, tsMReadPool.maxNum); - tsMReadPool.maxNum = MIN(4, tsMReadPool.maxNum); - tsMReadPool.curNum = 0; - tsMReadPool.readWorker = (SMReadWorker *)calloc(sizeof(SMReadWorker), tsMReadPool.maxNum); + tsMReadWP.maxNum = tsNumOfCores * tsNumOfThreadsPerCore / 2; + tsMReadWP.maxNum = MAX(2, tsMReadWP.maxNum); + tsMReadWP.maxNum = MIN(4, tsMReadWP.maxNum); + tsMReadWP.curNum = 0; + tsMReadWP.worker = (SMReadWorker *)calloc(sizeof(SMReadWorker), tsMReadWP.maxNum); - if (tsMReadPool.readWorker == NULL) return -1; - for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) { - SMReadWorker *pWorker = tsMReadPool.readWorker + i; + if (tsMReadWP.worker == NULL) return -1; + for (int32_t i = 0; i < tsMReadWP.maxNum; ++i) { + SMReadWorker *pWorker = tsMReadWP.worker + i; pWorker->workerId = i; dDebug("dnode mread worker:%d is created", i); } - dDebug("dnode mread is initialized, workers:%d qset:%p", tsMReadPool.maxNum, tsMReadQset); + dDebug("dnode mread is initialized, workers:%d qset:%p", tsMReadWP.maxNum, tsMReadQset); return 0; } -void dnodeCleanupMnodeRead() { - for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) { - SMReadWorker *pWorker = tsMReadPool.readWorker + i; +void dnodeCleanupMRead() { + for (int32_t i = 0; i < tsMReadWP.maxNum; ++i) { + SMReadWorker *pWorker = tsMReadWP.worker + i; if (pWorker->thread) { taosQsetThreadResume(tsMReadQset); } dDebug("dnode mread worker:%d is closed", i); } - for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) { - SMReadWorker *pWorker = tsMReadPool.readWorker + i; + for (int32_t i = 0; i < tsMReadWP.maxNum; ++i) { + SMReadWorker *pWorker = tsMReadWP.worker + i; dDebug("dnode mread worker:%d start to join", i); if (pWorker->thread) { pthread_join(pWorker->thread, NULL); @@ -86,43 +86,43 @@ void dnodeCleanupMnodeRead() { taosCloseQset(tsMReadQset); tsMReadQset = NULL; - free(tsMReadPool.readWorker); + free(tsMReadWP.worker); } -int32_t dnodeAllocateMnodeRqueue() { +int32_t dnodeAllocMReadQueue() { tsMReadQueue = taosOpenQueue(); if (tsMReadQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY; taosAddIntoQset(tsMReadQset, tsMReadQueue, NULL); - for (int32_t i = tsMReadPool.curNum; i < tsMReadPool.maxNum; ++i) { - SMReadWorker *pWorker = tsMReadPool.readWorker + i; + for (int32_t i = tsMReadWP.curNum; i < tsMReadWP.maxNum; ++i) { + SMReadWorker *pWorker = tsMReadWP.worker + i; pWorker->workerId = i; pthread_attr_t thAttr; pthread_attr_init(&thAttr); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); - if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMnodeReadQueue, pWorker) != 0) { + if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMReadQueue, pWorker) != 0) { dError("failed to create thread to process mread queue, reason:%s", strerror(errno)); } pthread_attr_destroy(&thAttr); - tsMReadPool.curNum = i + 1; - dDebug("dnode mread worker:%d is launched, total:%d", pWorker->workerId, tsMReadPool.maxNum); + tsMReadWP.curNum = i + 1; + dDebug("dnode mread worker:%d is launched, total:%d", pWorker->workerId, tsMReadWP.maxNum); } dDebug("dnode mread queue:%p is allocated", tsMReadQueue); return TSDB_CODE_SUCCESS; } -void dnodeFreeMnodeRqueue() { +void dnodeFreeMReadQueue() { dDebug("dnode mread queue:%p is freed", tsMReadQueue); taosCloseQueue(tsMReadQueue); tsMReadQueue = NULL; } -void dnodeDispatchToMnodeReadQueue(SRpcMsg *pMsg) { +void dnodeDispatchToMReadQueue(SRpcMsg *pMsg) { if (!mnodeIsRunning() || tsMReadQueue == NULL) { dnodeSendRedirectMsg(pMsg, true); rpcFreeCont(pMsg->pCont); @@ -134,16 +134,16 @@ void dnodeDispatchToMnodeReadQueue(SRpcMsg *pMsg) { taosWriteQitem(tsMReadQueue, TAOS_QTYPE_RPC, pRead); } -static void dnodeFreeMnodeReadMsg(SMnodeMsg *pRead) { +static void dnodeFreeMReadMsg(SMnodeMsg *pRead) { mnodeCleanupMsg(pRead); taosFreeQitem(pRead); } -static void dnodeSendRpcMnodeReadRsp(SMnodeMsg *pRead, int32_t code) { +static void dnodeSendRpcMReadRsp(SMnodeMsg *pRead, int32_t code) { if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return; if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) { // may be a auto create req, should put into write queue - dnodeReprocessMnodeWriteMsg(pRead); + dnodeReprocessMWriteMsg(pRead); return; } @@ -155,23 +155,23 @@ static void dnodeSendRpcMnodeReadRsp(SMnodeMsg *pRead, int32_t code) { }; rpcSendResponse(&rpcRsp); - dnodeFreeMnodeReadMsg(pRead); + dnodeFreeMReadMsg(pRead); } -static void *dnodeProcessMnodeReadQueue(void *param) { - SMnodeMsg *pReadMsg; +static void *dnodeProcessMReadQueue(void *param) { + SMnodeMsg *pRead; int32_t type; void * unUsed; - + while (1) { - if (taosReadQitemFromQset(tsMReadQset, &type, (void **)&pReadMsg, &unUsed) == 0) { + if (taosReadQitemFromQset(tsMReadQset, &type, (void **)&pRead, &unUsed) == 0) { dDebug("qset:%p, mnode read got no message from qset, exiting", tsMReadQset); break; } - dDebug("%p, msg:%s will be processed in mread queue", pReadMsg->rpcMsg.ahandle, taosMsg[pReadMsg->rpcMsg.msgType]); - int32_t code = mnodeProcessRead(pReadMsg); - dnodeSendRpcMnodeReadRsp(pReadMsg, code); + dDebug("%p, msg:%s will be processed in mread queue", pRead->rpcMsg.ahandle, taosMsg[pRead->rpcMsg.msgType]); + int32_t code = mnodeProcessRead(pRead); + dnodeSendRpcMReadRsp(pRead, code); } return NULL; diff --git a/src/dnode/src/dnodeMWrite.c b/src/dnode/src/dnodeMWrite.c index 600688b9fd..e628ea1f81 100644 --- a/src/dnode/src/dnodeMWrite.c +++ b/src/dnode/src/dnodeMWrite.c @@ -36,45 +36,45 @@ typedef struct { typedef struct { int32_t curNum; int32_t maxNum; - SMWriteWorker *writeWorker; + SMWriteWorker *worker; } SMWriteWorkerPool; -static SMWriteWorkerPool tsMWritePool; +static SMWriteWorkerPool tsMWriteWP; static taos_qset tsMWriteQset; static taos_queue tsMWriteQueue; extern void * tsDnodeTmr; -static void *dnodeProcessMnodeWriteQueue(void *param); +static void *dnodeProcessMWriteQueue(void *param); -int32_t dnodeInitMnodeWrite() { +int32_t dnodeInitMWrite() { tsMWriteQset = taosOpenQset(); - tsMWritePool.maxNum = 1; - tsMWritePool.curNum = 0; - tsMWritePool.writeWorker = (SMWriteWorker *)calloc(sizeof(SMWriteWorker), tsMWritePool.maxNum); + tsMWriteWP.maxNum = 1; + tsMWriteWP.curNum = 0; + tsMWriteWP.worker = (SMWriteWorker *)calloc(sizeof(SMWriteWorker), tsMWriteWP.maxNum); - if (tsMWritePool.writeWorker == NULL) return -1; - for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) { - SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; + if (tsMWriteWP.worker == NULL) return -1; + for (int32_t i = 0; i < tsMWriteWP.maxNum; ++i) { + SMWriteWorker *pWorker = tsMWriteWP.worker + i; pWorker->workerId = i; dDebug("dnode mwrite worker:%d is created", i); } - dDebug("dnode mwrite is initialized, workers:%d qset:%p", tsMWritePool.maxNum, tsMWriteQset); + dDebug("dnode mwrite is initialized, workers:%d qset:%p", tsMWriteWP.maxNum, tsMWriteQset); return 0; } -void dnodeCleanupMnodeWrite() { - for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) { - SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; +void dnodeCleanupMWrite() { + for (int32_t i = 0; i < tsMWriteWP.maxNum; ++i) { + SMWriteWorker *pWorker = tsMWriteWP.worker + i; if (pWorker->thread) { taosQsetThreadResume(tsMWriteQset); } dDebug("dnode mwrite worker:%d is closed", i); } - for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) { - SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; + for (int32_t i = 0; i < tsMWriteWP.maxNum; ++i) { + SMWriteWorker *pWorker = tsMWriteWP.worker + i; dDebug("dnode mwrite worker:%d start to join", i); if (pWorker->thread) { pthread_join(pWorker->thread, NULL); @@ -86,43 +86,43 @@ void dnodeCleanupMnodeWrite() { taosCloseQset(tsMWriteQset); tsMWriteQset = NULL; - taosTFree(tsMWritePool.writeWorker); + taosTFree(tsMWriteWP.worker); } -int32_t dnodeAllocateMnodeWqueue() { +int32_t dnodeAllocMWritequeue() { tsMWriteQueue = taosOpenQueue(); if (tsMWriteQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY; taosAddIntoQset(tsMWriteQset, tsMWriteQueue, NULL); - for (int32_t i = tsMWritePool.curNum; i < tsMWritePool.maxNum; ++i) { - SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; + for (int32_t i = tsMWriteWP.curNum; i < tsMWriteWP.maxNum; ++i) { + SMWriteWorker *pWorker = tsMWriteWP.worker + i; pWorker->workerId = i; pthread_attr_t thAttr; pthread_attr_init(&thAttr); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); - if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMnodeWriteQueue, pWorker) != 0) { + if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMWriteQueue, pWorker) != 0) { dError("failed to create thread to process mwrite queue, reason:%s", strerror(errno)); } pthread_attr_destroy(&thAttr); - tsMWritePool.curNum = i + 1; - dDebug("dnode mwrite worker:%d is launched, total:%d", pWorker->workerId, tsMWritePool.maxNum); + tsMWriteWP.curNum = i + 1; + dDebug("dnode mwrite worker:%d is launched, total:%d", pWorker->workerId, tsMWriteWP.maxNum); } dDebug("dnode mwrite queue:%p is allocated", tsMWriteQueue); return TSDB_CODE_SUCCESS; } -void dnodeFreeMnodeWqueue() { +void dnodeFreeMWritequeue() { dDebug("dnode mwrite queue:%p is freed", tsMWriteQueue); taosCloseQueue(tsMWriteQueue); tsMWriteQueue = NULL; } -void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) { +void dnodeDispatchToMWriteQueue(SRpcMsg *pMsg) { if (!mnodeIsRunning() || tsMWriteQueue == NULL) { dnodeSendRedirectMsg(pMsg, true); rpcFreeCont(pMsg->pCont); @@ -137,7 +137,7 @@ void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) { taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite); } -static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) { +static void dnodeFreeMWriteMsg(SMnodeMsg *pWrite) { dDebug("app:%p:%p, msg:%s is freed from mwrite queue:%p", pWrite->rpcMsg.ahandle, pWrite, taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue); @@ -145,12 +145,12 @@ static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) { taosFreeQitem(pWrite); } -void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code) { +void dnodeSendRpcMWriteRsp(void *pMsg, int32_t code) { SMnodeMsg *pWrite = pMsg; if (pWrite == NULL) return; if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return; if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) { - dnodeReprocessMnodeWriteMsg(pWrite); + dnodeReprocessMWriteMsg(pWrite); return; } @@ -162,10 +162,10 @@ void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code) { }; rpcSendResponse(&rpcRsp); - dnodeFreeMnodeWriteMsg(pWrite); + dnodeFreeMWriteMsg(pWrite); } -static void *dnodeProcessMnodeWriteQueue(void *param) { +static void *dnodeProcessMWriteQueue(void *param) { SMnodeMsg *pWrite; int32_t type; void * unUsed; @@ -180,13 +180,13 @@ static void *dnodeProcessMnodeWriteQueue(void *param) { taosMsg[pWrite->rpcMsg.msgType]); int32_t code = mnodeProcessWrite(pWrite); - dnodeSendRpcMnodeWriteRsp(pWrite, code); + dnodeSendRpcMWriteRsp(pWrite, code); } return NULL; } -void dnodeReprocessMnodeWriteMsg(void *pMsg) { +void dnodeReprocessMWriteMsg(void *pMsg) { SMnodeMsg *pWrite = pMsg; if (!mnodeIsRunning() || tsMWriteQueue == NULL) { @@ -194,7 +194,7 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) { taosMsg[pWrite->rpcMsg.msgType], pWrite->retry); dnodeSendRedirectMsg(pMsg, true); - dnodeFreeMnodeWriteMsg(pWrite); + dnodeFreeMWriteMsg(pWrite); } else { dDebug("app:%p:%p, msg:%s is reput into mwrite queue:%p, retry times:%d", pWrite->rpcMsg.ahandle, pWrite, taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue, pWrite->retry); @@ -203,12 +203,12 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) { } } -static void dnodeDoDelayReprocessMnodeWriteMsg(void *param, void *tmrId) { - dnodeReprocessMnodeWriteMsg(param); +static void dnodeDoDelayReprocessMWriteMsg(void *param, void *tmrId) { + dnodeReprocessMWriteMsg(param); } -void dnodeDelayReprocessMnodeWriteMsg(void *pMsg) { +void dnodeDelayReprocessMWriteMsg(void *pMsg) { SMnodeMsg *mnodeMsg = pMsg; void *unUsed = NULL; - taosTmrReset(dnodeDoDelayReprocessMnodeWriteMsg, 300, mnodeMsg, tsDnodeTmr, &unUsed); + taosTmrReset(dnodeDoDelayReprocessMWriteMsg, 300, mnodeMsg, tsDnodeTmr, &unUsed); } diff --git a/src/dnode/src/dnodeMain.c b/src/dnode/src/dnodeMain.c index 5fde4f972b..f4c0ee565e 100644 --- a/src/dnode/src/dnodeMain.c +++ b/src/dnode/src/dnodeMain.c @@ -37,11 +37,11 @@ #include "dnodeShell.h" #include "dnodeTelemetry.h" -static SDnodeRunStatus tsDnodeRunStatus = TSDB_DNODE_RUN_STATUS_STOPPED; +static SRunStatus tsRunStatus = TSDB_RUN_STATUS_STOPPED; static int32_t dnodeInitStorage(); static void dnodeCleanupStorage(); -static void dnodeSetRunStatus(SDnodeRunStatus status); +static void dnodeSetRunStatus(SRunStatus status); static void dnodeCheckDataDirOpenned(char *dir); static int32_t dnodeInitComponents(); static void dnodeCleanupComponents(int32_t stepId); @@ -63,9 +63,9 @@ static const SDnodeComponent tsDnodeComponents[] = { {"check", dnodeInitCheck, dnodeCleanupCheck}, // NOTES: dnodeInitCheck must be behind the dnodeinitStorage component !!! {"vread", dnodeInitVRead, dnodeCleanupVRead}, {"vwrite", dnodeInitVWrite, dnodeCleanupVWrite}, - {"mread", dnodeInitMnodeRead, dnodeCleanupMnodeRead}, - {"mwrite", dnodeInitMnodeWrite, dnodeCleanupMnodeWrite}, - {"mpeer", dnodeInitMnodePeer, dnodeCleanupMnodePeer}, + {"mread", dnodeInitMRead, dnodeCleanupMRead}, + {"mwrite", dnodeInitMWrite, dnodeCleanupMWrite}, + {"mpeer", dnodeInitMPeer, dnodeCleanupMPeer}, {"client", dnodeInitClient, dnodeCleanupClient}, {"server", dnodeInitServer, dnodeCleanupServer}, {"mgmt", dnodeInitMgmt, dnodeCleanupMgmt}, @@ -104,7 +104,7 @@ static int32_t dnodeInitComponents() { } int32_t dnodeInitSystem() { - dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_INITIALIZE); + dnodeSetRunStatus(TSDB_RUN_STATUS_INITIALIZE); tscEmbedded = 1; taosBlockSIGPIPE(); taosResolveCRC(); @@ -137,7 +137,7 @@ int32_t dnodeInitSystem() { } dnodeStartModules(); - dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_RUNING); + dnodeSetRunStatus(TSDB_RUN_STATUS_RUNING); dInfo("TDengine is initialized successfully"); @@ -145,20 +145,20 @@ int32_t dnodeInitSystem() { } void dnodeCleanUpSystem() { - if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_STOPPED) { - dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED); + if (dnodeGetRunStatus() != TSDB_RUN_STATUS_STOPPED) { + dnodeSetRunStatus(TSDB_RUN_STATUS_STOPPED); dnodeCleanupComponents(sizeof(tsDnodeComponents) / sizeof(tsDnodeComponents[0]) - 1); taos_cleanup(); taosCloseLog(); } } -SDnodeRunStatus dnodeGetRunStatus() { - return tsDnodeRunStatus; +SRunStatus dnodeGetRunStatus() { + return tsRunStatus; } -static void dnodeSetRunStatus(SDnodeRunStatus status) { - tsDnodeRunStatus = status; +static void dnodeSetRunStatus(SRunStatus status) { + tsRunStatus = status; } static void dnodeCheckDataDirOpenned(char *dir) { diff --git a/src/dnode/src/dnodePeer.c b/src/dnode/src/dnodePeer.c index c6fc2b9e36..cbdbc54c25 100644 --- a/src/dnode/src/dnodePeer.c +++ b/src/dnode/src/dnodePeer.c @@ -34,8 +34,8 @@ static void (*dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *); static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *); static void (*dnodeProcessRspMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *rpcMsg); static void dnodeProcessRspFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet); -static void *tsDnodeServerRpc = NULL; -static void *tsDnodeClientRpc = NULL; +static void *tsServerRpc = NULL; +static void *tsClientRpc = NULL; int32_t dnodeInitServer() { dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeDispatchToVWriteQueue; @@ -50,11 +50,11 @@ int32_t dnodeInitServer() { dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_MNODE] = dnodeDispatchToMgmtQueue; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = dnodeDispatchToMnodePeerQueue; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = dnodeDispatchToMnodePeerQueue; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_AUTH] = dnodeDispatchToMnodePeerQueue; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_GRANT] = dnodeDispatchToMnodePeerQueue; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_STATUS] = dnodeDispatchToMnodePeerQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = dnodeDispatchToMPeerQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = dnodeDispatchToMPeerQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_AUTH] = dnodeDispatchToMPeerQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_GRANT] = dnodeDispatchToMPeerQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_STATUS] = dnodeDispatchToMPeerQueue; SRpcInit rpcInit; memset(&rpcInit, 0, sizeof(rpcInit)); @@ -66,8 +66,8 @@ int32_t dnodeInitServer() { rpcInit.connType = TAOS_CONN_SERVER; rpcInit.idleTime = tsShellActivityTimer * 1000; - tsDnodeServerRpc = rpcOpen(&rpcInit); - if (tsDnodeServerRpc == NULL) { + tsServerRpc = rpcOpen(&rpcInit); + if (tsServerRpc == NULL) { dError("failed to init inter-dnodes RPC server"); return -1; } @@ -77,9 +77,9 @@ int32_t dnodeInitServer() { } void dnodeCleanupServer() { - if (tsDnodeServerRpc) { - rpcClose(tsDnodeServerRpc); - tsDnodeServerRpc = NULL; + if (tsServerRpc) { + rpcClose(tsServerRpc); + tsServerRpc = NULL; dInfo("inter-dnodes RPC server is closed"); } } @@ -93,7 +93,7 @@ static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { if (pMsg->pCont == NULL) return; - if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_RUNING) { + if (dnodeGetRunStatus() != TSDB_RUN_STATUS_RUNING) { rspMsg.code = TSDB_CODE_APP_NOT_READY; rpcSendResponse(&rspMsg); rpcFreeCont(pMsg->pCont); @@ -131,8 +131,8 @@ int32_t dnodeInitClient() { rpcInit.ckey = "key"; rpcInit.secret = secret; - tsDnodeClientRpc = rpcOpen(&rpcInit); - if (tsDnodeClientRpc == NULL) { + tsClientRpc = rpcOpen(&rpcInit); + if (tsClientRpc == NULL) { dError("failed to init mnode rpc client"); return -1; } @@ -142,9 +142,9 @@ int32_t dnodeInitClient() { } void dnodeCleanupClient() { - if (tsDnodeClientRpc) { - rpcClose(tsDnodeClientRpc); - tsDnodeClientRpc = NULL; + if (tsClientRpc) { + rpcClose(tsClientRpc); + tsClientRpc = NULL; dInfo("dnode inter-dnodes rpc client is closed"); } } @@ -168,15 +168,15 @@ void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)) { } void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) { - rpcSendRequest(tsDnodeClientRpc, epSet, rpcMsg); + rpcSendRequest(tsClientRpc, epSet, rpcMsg); } void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) { SRpcEpSet epSet = {0}; dnodeGetEpSetForPeer(&epSet); - rpcSendRecv(tsDnodeClientRpc, &epSet, rpcMsg, rpcRsp); + rpcSendRecv(tsClientRpc, &epSet, rpcMsg, rpcRsp); } void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet) { - rpcSendRecv(tsDnodeClientRpc, epSet, rpcMsg, rpcRsp); + rpcSendRecv(tsClientRpc, epSet, rpcMsg, rpcRsp); } \ No newline at end of file diff --git a/src/dnode/src/dnodeShell.c b/src/dnode/src/dnodeShell.c index 2f6d844ff1..89f657f789 100644 --- a/src/dnode/src/dnodeShell.c +++ b/src/dnode/src/dnodeShell.c @@ -33,9 +33,9 @@ static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *); static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *); static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey); -static void * tsDnodeShellRpc = NULL; -static int32_t tsDnodeQueryReqNum = 0; -static int32_t tsDnodeSubmitReqNum = 0; +static void * tsShellRpc = NULL; +static int32_t tsQueryReqNum = 0; +static int32_t tsSubmitReqNum = 0; int32_t dnodeInitShell() { dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeDispatchToVWriteQueue; @@ -44,35 +44,35 @@ int32_t dnodeInitShell() { dnodeProcessShellMsgFp[TSDB_MSG_TYPE_UPDATE_TAG_VAL] = dnodeDispatchToVWriteQueue; // the following message shall be treated as mnode write - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_ACCT] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_ACCT] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_ACCT] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_USER] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_USER] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_USER] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DNODE]= dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DNODE] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DB] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DB] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_DB] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TABLE]= dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_TABLE] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_TABLE] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_STREAM]= dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_QUERY] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_STREAM] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_CONN] = dnodeDispatchToMnodeWriteQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONFIG_DNODE]= dnodeDispatchToMnodeWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_ACCT] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_ACCT] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_ACCT] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_USER] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_USER] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_USER] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DNODE]= dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DNODE] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DB] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DB] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_DB] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TABLE]= dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_TABLE] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_TABLE] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_STREAM]= dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_QUERY] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_STREAM] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_CONN] = dnodeDispatchToMWriteQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONFIG_DNODE]= dnodeDispatchToMWriteQueue; // the following message shall be treated as mnode query - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_HEARTBEAT] = dnodeDispatchToMnodeReadQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONNECT] = dnodeDispatchToMnodeReadQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_USE_DB] = dnodeDispatchToMnodeReadQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLE_META] = dnodeDispatchToMnodeReadQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_STABLE_VGROUP]= dnodeDispatchToMnodeReadQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLES_META] = dnodeDispatchToMnodeReadQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_SHOW] = dnodeDispatchToMnodeReadQueue; - dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_RETRIEVE] = dnodeDispatchToMnodeReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_HEARTBEAT] = dnodeDispatchToMReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONNECT] = dnodeDispatchToMReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_USE_DB] = dnodeDispatchToMReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLE_META] = dnodeDispatchToMReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_STABLE_VGROUP]= dnodeDispatchToMReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLES_META] = dnodeDispatchToMReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_SHOW] = dnodeDispatchToMReadQueue; + dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_RETRIEVE] = dnodeDispatchToMReadQueue; int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore; numOfThreads = (int32_t) ((1.0 - tsRatioOfQueryThreads) * numOfThreads / 2.0); @@ -91,8 +91,8 @@ int32_t dnodeInitShell() { rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.afp = dnodeRetrieveUserAuthInfo; - tsDnodeShellRpc = rpcOpen(&rpcInit); - if (tsDnodeShellRpc == NULL) { + tsShellRpc = rpcOpen(&rpcInit); + if (tsShellRpc == NULL) { dError("failed to init shell rpc server"); return -1; } @@ -102,13 +102,13 @@ int32_t dnodeInitShell() { } void dnodeCleanupShell() { - if (tsDnodeShellRpc) { - rpcClose(tsDnodeShellRpc); - tsDnodeShellRpc = NULL; + if (tsShellRpc) { + rpcClose(tsShellRpc); + tsShellRpc = NULL; } } -void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { +static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { SRpcMsg rpcMsg = { .handle = pMsg->handle, .pCont = NULL, @@ -117,7 +117,7 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { if (pMsg->pCont == NULL) return; - if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_RUNING) { + if (dnodeGetRunStatus() != TSDB_RUN_STATUS_RUNING) { dError("RPC %p, shell msg:%s is ignored since dnode not running", pMsg->handle, taosMsg[pMsg->msgType]); rpcMsg.code = TSDB_CODE_APP_NOT_READY; rpcSendResponse(&rpcMsg); @@ -126,9 +126,9 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { } if (pMsg->msgType == TSDB_MSG_TYPE_QUERY) { - atomic_fetch_add_32(&tsDnodeQueryReqNum, 1); + atomic_fetch_add_32(&tsQueryReqNum, 1); } else if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) { - atomic_fetch_add_32(&tsDnodeSubmitReqNum, 1); + atomic_fetch_add_32(&tsSubmitReqNum, 1); } else {} if ( dnodeProcessShellMsgFp[pMsg->msgType] ) { @@ -211,12 +211,12 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid) { } } -SDnodeStatisInfo dnodeGetStatisInfo() { - SDnodeStatisInfo info = {0}; - if (dnodeGetRunStatus() == TSDB_DNODE_RUN_STATUS_RUNING) { +SStatisInfo dnodeGetStatisInfo() { + SStatisInfo info = {0}; + if (dnodeGetRunStatus() == TSDB_RUN_STATUS_RUNING) { info.httpReqNum = httpGetReqCount(); - info.queryReqNum = atomic_exchange_32(&tsDnodeQueryReqNum, 0); - info.submitReqNum = atomic_exchange_32(&tsDnodeSubmitReqNum, 0); + info.queryReqNum = atomic_exchange_32(&tsQueryReqNum, 0); + info.submitReqNum = atomic_exchange_32(&tsSubmitReqNum, 0); } return info; diff --git a/src/dnode/src/dnodeVWrite.c b/src/dnode/src/dnodeVWrite.c index c28ad66b65..9b7497d3ec 100644 --- a/src/dnode/src/dnodeVWrite.c +++ b/src/dnode/src/dnodeVWrite.c @@ -113,7 +113,7 @@ void dnodeDispatchToVWriteQueue(SRpcMsg *pRpcMsg) { void *dnodeAllocVWriteQueue(void *pVnode) { pthread_mutex_lock(&tsVWriteWP.mutex); SVWriteWorker *pWorker = tsVWriteWP.worker + tsVWriteWP.nextId; - void *queue = taosOpenQueue(); + taos_queue *queue = taosOpenQueue(); if (queue == NULL) { pthread_mutex_unlock(&tsVWriteWP.mutex); return NULL; diff --git a/src/inc/dnode.h b/src/inc/dnode.h index 6032d8cc0a..9454b97e38 100644 --- a/src/inc/dnode.h +++ b/src/inc/dnode.h @@ -27,16 +27,16 @@ typedef struct { int32_t queryReqNum; int32_t submitReqNum; int32_t httpReqNum; -} SDnodeStatisInfo; +} SStatisInfo; typedef enum { - TSDB_DNODE_RUN_STATUS_INITIALIZE, - TSDB_DNODE_RUN_STATUS_RUNING, - TSDB_DNODE_RUN_STATUS_STOPPED -} SDnodeRunStatus; + TSDB_RUN_STATUS_INITIALIZE, + TSDB_RUN_STATUS_RUNING, + TSDB_RUN_STATUS_STOPPED +} SRunStatus; -SDnodeRunStatus dnodeGetRunStatus(); -SDnodeStatisInfo dnodeGetStatisInfo(); +SRunStatus dnodeGetRunStatus(); +SStatisInfo dnodeGetStatisInfo(); bool dnodeIsFirstDeploy(); bool dnodeIsMasterEp(char *ep); @@ -59,15 +59,15 @@ void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code); void *dnodeAllocVReadQueue(void *pVnode); void dnodeFreeVReadQueue(void *rqueue); -int32_t dnodeAllocateMnodePqueue(); -void dnodeFreeMnodePqueue(); -int32_t dnodeAllocateMnodeRqueue(); -void dnodeFreeMnodeRqueue(); -int32_t dnodeAllocateMnodeWqueue(); -void dnodeFreeMnodeWqueue(); -void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code); -void dnodeReprocessMnodeWriteMsg(void *pMsg); -void dnodeDelayReprocessMnodeWriteMsg(void *pMsg); +int32_t dnodeAllocateMPeerQueue(); +void dnodeFreeMPeerQueue(); +int32_t dnodeAllocMReadQueue(); +void dnodeFreeMReadQueue(); +int32_t dnodeAllocMWritequeue(); +void dnodeFreeMWritequeue(); +void dnodeSendRpcMWriteRsp(void *pMsg, int32_t code); +void dnodeReprocessMWriteMsg(void *pMsg); +void dnodeDelayReprocessMWriteMsg(void *pMsg); void dnodeSendStatusMsgToMnode(); diff --git a/src/mnode/src/mnodeMain.c b/src/mnode/src/mnodeMain.c index 2bb8a81056..1f5ad42bde 100644 --- a/src/mnode/src/mnodeMain.c +++ b/src/mnode/src/mnodeMain.c @@ -96,9 +96,9 @@ int32_t mnodeStartSystem() { return -1; } - dnodeAllocateMnodeWqueue(); - dnodeAllocateMnodeRqueue(); - dnodeAllocateMnodePqueue(); + dnodeAllocMWritequeue(); + dnodeAllocMReadQueue(); + dnodeAllocateMPeerQueue(); if (mnodeInitComponents() != 0) { return -1; @@ -127,9 +127,9 @@ void mnodeCleanupSystem() { mInfo("starting to clean up mnode"); tsMgmtIsRunning = false; - dnodeFreeMnodeWqueue(); - dnodeFreeMnodeRqueue(); - dnodeFreeMnodePqueue(); + dnodeFreeMWritequeue(); + dnodeFreeMReadQueue(); + dnodeFreeMPeerQueue(); mnodeCleanupTimer(); mnodeCleanupComponents(sizeof(tsMnodeComponents) / sizeof(tsMnodeComponents[0]) - 1); diff --git a/src/mnode/src/mnodeSdb.c b/src/mnode/src/mnodeSdb.c index 4e370ca028..6bc144d1e1 100644 --- a/src/mnode/src/mnodeSdb.c +++ b/src/mnode/src/mnodeSdb.c @@ -295,7 +295,7 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) { if (pOper->writeCb != NULL) { pOper->retCode = (*pOper->writeCb)(pMsg, pOper->retCode); } - dnodeSendRpcMnodeWriteRsp(pMsg, pOper->retCode); + dnodeSendRpcMWriteRsp(pMsg, pOper->retCode); // if ahandle, means this func is called by sdb write if (ahandle == NULL) { diff --git a/src/mnode/src/mnodeTable.c b/src/mnode/src/mnodeTable.c index 28dc2c6cff..00234e33b2 100644 --- a/src/mnode/src/mnodeTable.c +++ b/src/mnode/src/mnodeTable.c @@ -1679,12 +1679,12 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) { pTable->info.tableId, pMsg->rpcMsg.handle); pMsg->retry = 0; - dnodeReprocessMnodeWriteMsg(pMsg); + dnodeReprocessMWriteMsg(pMsg); } else { mDebug("app:%p:%p, table:%s, created in dnode, thandle:%p", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId, pMsg->rpcMsg.handle); - dnodeSendRpcMnodeWriteRsp(pMsg, TSDB_CODE_SUCCESS); + dnodeSendRpcMWriteRsp(pMsg, TSDB_CODE_SUCCESS); } return TSDB_CODE_MND_ACTION_IN_PROGRESS; } else { @@ -2351,14 +2351,14 @@ static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg) { mError("app:%p:%p, table:%s, failed to drop in dnode, vgId:%d sid:%d uid:%" PRIu64 ", reason:%s", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid, tstrerror(rpcMsg->code)); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); + dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code); return; } if (mnodeMsg->pVgroup == NULL) mnodeMsg->pVgroup = mnodeGetVgroup(pTable->vgId); if (mnodeMsg->pVgroup == NULL) { mError("app:%p:%p, table:%s, failed to get vgroup", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, TSDB_CODE_MND_VGROUP_NOT_EXIST); + dnodeSendRpcMWriteRsp(mnodeMsg, TSDB_CODE_MND_VGROUP_NOT_EXIST); return; } @@ -2368,7 +2368,7 @@ static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg) { mnodeDropVgroup(mnodeMsg->pVgroup, NULL); } - dnodeSendRpcMnodeWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS); + dnodeSendRpcMWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS); } /* @@ -2399,7 +2399,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { mnodeSendDropChildTableMsg(mnodeMsg, false); rpcMsg->code = TSDB_CODE_SUCCESS; - dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); + dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code); return; } @@ -2416,7 +2416,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) { mnodeMsg->pTable = NULL; mnodeDestroyChildTable(pTable); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, code); + dnodeSendRpcMWriteRsp(mnodeMsg, code); } } else { if (mnodeMsg->retry++ < 10) { @@ -2425,7 +2425,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, mnodeMsg->retry, pTable->vgId, pTable->tid, pTable->uid, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle); - dnodeDelayReprocessMnodeWriteMsg(mnodeMsg); + dnodeDelayReprocessMWriteMsg(mnodeMsg); } else { mError("app:%p:%p, table:%s, failed to create in dnode, vgId:%d sid:%d uid:%" PRIu64 ", result:%s thandle:%p", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid, @@ -2434,7 +2434,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { SSdbOper oper = {.type = SDB_OPER_GLOBAL, .table = tsChildTableSdb, .pObj = pTable}; sdbDeleteRow(&oper); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); + dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code); } } } @@ -2452,18 +2452,18 @@ static void mnodeProcessAlterTableRsp(SRpcMsg *rpcMsg) { mDebug("app:%p:%p, ctable:%s, altered in dnode, thandle:%p result:%s", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, mnodeMsg->rpcMsg.handle, tstrerror(rpcMsg->code)); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS); + dnodeSendRpcMWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS); } else { if (mnodeMsg->retry++ < 3) { mDebug("app:%p:%p, table:%s, alter table rsp received, need retry, times:%d result:%s thandle:%p", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, mnodeMsg->retry, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle); - dnodeDelayReprocessMnodeWriteMsg(mnodeMsg); + dnodeDelayReprocessMWriteMsg(mnodeMsg); } else { mError("app:%p:%p, table:%s, failed to alter in dnode, result:%s thandle:%p", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); + dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code); } } } diff --git a/src/mnode/src/mnodeVgroup.c b/src/mnode/src/mnodeVgroup.c index 1d1f21c3d7..fada760cb7 100644 --- a/src/mnode/src/mnodeVgroup.c +++ b/src/mnode/src/mnodeVgroup.c @@ -529,7 +529,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) { SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb}; (void)sdbUpdateRow(&desc); - dnodeReprocessMnodeWriteMsg(pMsg); + dnodeReprocessMWriteMsg(pMsg); return TSDB_CODE_MND_ACTION_IN_PROGRESS; // if (pVgroup->status == TAOS_VG_STATUS_CREATING || pVgroup->status == TAOS_VG_STATUS_READY) { // mInfo("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId, @@ -537,7 +537,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) { // pVgroup->status = TAOS_VG_STATUS_READY; // SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb}; // (void)sdbUpdateRow(&desc); - // dnodeReprocessMnodeWriteMsg(pMsg); + // dnodeReprocessMWriteMsg(pMsg); // return TSDB_CODE_MND_ACTION_IN_PROGRESS; // } else { // mError("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d, but vgroup is dropping", pMsg->rpcMsg.ahandle, @@ -969,7 +969,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) { if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) { mnodeMsg->pVgroup = NULL; mnodeDestroyVgroup(pVgroup); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, code); + dnodeSendRpcMWriteRsp(mnodeMsg, code); } } else { SSdbOper oper = { @@ -978,7 +978,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) { .pObj = pVgroup }; sdbDeleteRow(&oper); - dnodeSendRpcMnodeWriteRsp(mnodeMsg, mnodeMsg->code); + dnodeSendRpcMWriteRsp(mnodeMsg, mnodeMsg->code); } } @@ -1040,7 +1040,7 @@ static void mnodeProcessDropVnodeRsp(SRpcMsg *rpcMsg) { code = TSDB_CODE_MND_SDB_ERROR; } - dnodeReprocessMnodeWriteMsg(mnodeMsg); + dnodeReprocessMWriteMsg(mnodeMsg); } static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) { diff --git a/src/plugins/monitor/src/monitorMain.c b/src/plugins/monitor/src/monitorMain.c index 048f839b72..de1e0e233c 100644 --- a/src/plugins/monitor/src/monitorMain.c +++ b/src/plugins/monitor/src/monitorMain.c @@ -27,12 +27,12 @@ #include "monitor.h" #include "taoserror.h" -#define monitorFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }} -#define monitorError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }} -#define monitorWarn(...) { if (monitorDebugFlag & DEBUG_WARN) { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }} -#define monitorInfo(...) { if (monitorDebugFlag & DEBUG_INFO) { taosPrintLog("MON ", 255, __VA_ARGS__); }} -#define monitorDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} -#define monitorTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} +#define mnFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }} +#define mnError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }} +#define mnWarn(...) { if (monitorDebugFlag & DEBUG_WARN) { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }} +#define mnInfo(...) { if (monitorDebugFlag & DEBUG_INFO) { taosPrintLog("MON ", 255, __VA_ARGS__); }} +#define mnDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} +#define mnTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} #define SQL_LENGTH 1030 #define LOG_LEN_STR 100 @@ -91,12 +91,12 @@ int32_t monitorInitSystem() { pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); if (pthread_create(&tsMonitor.thread, &thAttr, monitorThreadFunc, NULL)) { - monitorError("failed to create thread to for monitor module, reason:%s", strerror(errno)); + mnError("failed to create thread to for monitor module, reason:%s", strerror(errno)); return -1; } pthread_attr_destroy(&thAttr); - monitorDebug("monitor thread is launched"); + mnDebug("monitor thread is launched"); monitorStartSystemFp = monitorStartSystem; monitorStopSystemFp = monitorStopSystem; @@ -107,12 +107,12 @@ int32_t monitorStartSystem() { taos_init(); tsMonitor.start = 1; monitorExecuteSQLFp = monitorExecuteSQL; - monitorInfo("monitor module start"); + mnInfo("monitor module start"); return 0; } static void *monitorThreadFunc(void *param) { - monitorDebug("starting to initialize monitor module ..."); + mnDebug("starting to initialize monitor module ..."); while (1) { static int32_t accessTimes = 0; @@ -121,7 +121,7 @@ static void *monitorThreadFunc(void *param) { if (tsMonitor.quiting) { tsMonitor.state = MON_STATE_NOT_INIT; - monitorInfo("monitor thread will quit, for taosd is quiting"); + mnInfo("monitor thread will quit, for taosd is quiting"); break; } else { taosGetDisk(); @@ -132,7 +132,7 @@ static void *monitorThreadFunc(void *param) { } if (dnodeGetDnodeId() <= 0) { - monitorDebug("dnode not initialized, waiting for 3000 ms to start monitor module"); + mnDebug("dnode not initialized, waiting for 3000 ms to start monitor module"); continue; } @@ -140,10 +140,10 @@ static void *monitorThreadFunc(void *param) { tsMonitor.state = MON_STATE_NOT_INIT; tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0); if (tsMonitor.conn == NULL) { - monitorError("failed to connect to database, reason:%s", tstrerror(terrno)); + mnError("failed to connect to database, reason:%s", tstrerror(terrno)); continue; } else { - monitorDebug("connect to database success"); + mnDebug("connect to database success"); } } @@ -155,10 +155,10 @@ static void *monitorThreadFunc(void *param) { taos_free_result(res); if (code != 0) { - monitorError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code)); + mnError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code)); break; } else { - monitorDebug("successfully to exec sql:%s", tsMonitor.sql); + mnDebug("successfully to exec sql:%s", tsMonitor.sql); } } @@ -174,7 +174,7 @@ static void *monitorThreadFunc(void *param) { } } - monitorInfo("monitor thread is stopped"); + mnInfo("monitor thread is stopped"); return NULL; } @@ -238,7 +238,7 @@ void monitorStopSystem() { tsMonitor.start = 0; tsMonitor.state = MON_STATE_NOT_INIT; monitorExecuteSQLFp = NULL; - monitorInfo("monitor module stopped"); + mnInfo("monitor module stopped"); } void monitorCleanUpSystem() { @@ -249,7 +249,7 @@ void monitorCleanUpSystem() { taos_close(tsMonitor.conn); tsMonitor.conn = NULL; } - monitorInfo("monitor module is cleaned up"); + mnInfo("monitor module is cleaned up"); } // unit is MB @@ -257,13 +257,13 @@ static int32_t monitorBuildMemorySql(char *sql) { float sysMemoryUsedMB = 0; bool suc = taosGetSysMemory(&sysMemoryUsedMB); if (!suc) { - monitorDebug("failed to get sys memory info"); + mnDebug("failed to get sys memory info"); } float procMemoryUsedMB = 0; suc = taosGetProcMemory(&procMemoryUsedMB); if (!suc) { - monitorDebug("failed to get proc memory info"); + mnDebug("failed to get proc memory info"); } return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB); @@ -274,7 +274,7 @@ static int32_t monitorBuildCpuSql(char *sql) { float sysCpuUsage = 0, procCpuUsage = 0; bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage); if (!suc) { - monitorDebug("failed to get cpu usage"); + mnDebug("failed to get cpu usage"); } if (sysCpuUsage <= procCpuUsage) { @@ -294,14 +294,14 @@ static int32_t monitorBuildBandSql(char *sql) { float bandSpeedKb = 0; bool suc = taosGetBandSpeed(&bandSpeedKb); if (!suc) { - monitorDebug("failed to get bandwidth speed"); + mnDebug("failed to get bandwidth speed"); } return sprintf(sql, ", %f", bandSpeedKb); } static int32_t monitorBuildReqSql(char *sql) { - SDnodeStatisInfo info = dnodeGetStatisInfo(); + SStatisInfo info = dnodeGetStatisInfo(); return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum); } @@ -309,7 +309,7 @@ static int32_t monitorBuildIoSql(char *sql) { float readKB = 0, writeKB = 0; bool suc = taosGetProcIO(&readKB, &writeKB); if (!suc) { - monitorDebug("failed to get io info"); + mnDebug("failed to get io info"); } return sprintf(sql, ", %f, %f", readKB, writeKB); @@ -332,19 +332,19 @@ static void monitorSaveSystemInfo() { taos_free_result(res); if (code != 0) { - monitorError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql); + mnError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql); } else { - monitorDebug("successfully to save system info, sql:%s", tsMonitor.sql); + mnDebug("successfully to save system info, sql:%s", tsMonitor.sql); } } static void montiorExecSqlCb(void *param, TAOS_RES *result, int32_t code) { int32_t c = taos_errno(result); if (c != TSDB_CODE_SUCCESS) { - monitorError("save %s failed, reason:%s", (char *)param, tstrerror(c)); + mnError("save %s failed, reason:%s", (char *)param, tstrerror(c)); } else { int32_t rows = taos_affected_rows(result); - monitorDebug("save %s succ, rows:%d", (char *)param, rows); + mnDebug("save %s succ, rows:%d", (char *)param, rows); } taos_free_result(result); @@ -380,7 +380,7 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) { pMon->totalConns, pMon->maxConns, pMon->accessState); - monitorDebug("save account info, sql:%s", sql); + mnDebug("save account info, sql:%s", sql); taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "account info"); } @@ -401,13 +401,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) { len += sprintf(sql + len, "', '%s')", tsLocalEp); sql[len++] = 0; - monitorDebug("save log, sql: %s", sql); + mnDebug("save log, sql: %s", sql); taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "log"); } void monitorExecuteSQL(char *sql) { if (tsMonitor.state != MON_STATE_INITED) return; - monitorDebug("execute sql:%s", sql); + mnDebug("execute sql:%s", sql); taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "sql"); } diff --git a/src/vnode/src/vnodeCfg.c b/src/vnode/src/vnodeCfg.c index f0040f8cdf..e8dd44b48f 100644 --- a/src/vnode/src/vnodeCfg.c +++ b/src/vnode/src/vnodeCfg.c @@ -22,7 +22,6 @@ #include "tsdb.h" #include "dnode.h" #include "vnodeInt.h" -#include "vnodeVersion.h" #include "vnodeCfg.h" static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) { diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 128da72623..e60173e066 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -15,17 +15,11 @@ #define _DEFAULT_SOURCE #include "os.h" - -#include "tcache.h" -#include "cJSON.h" -#include "dnode.h" -#include "hash.h" #include "taoserror.h" #include "taosmsg.h" #include "tglobal.h" #include "trpc.h" #include "tsdb.h" -#include "ttimer.h" #include "tutil.h" #include "vnode.h" #include "vnodeInt.h" @@ -34,7 +28,7 @@ #include "vnodeCfg.h" #include "vnodeVersion.h" -static SHashObj*tsDnodeVnodesHash; +static SHashObj*tsVnodesHash; static void vnodeCleanUp(SVnodeObj *pVnode); static int vnodeProcessTsdbStatus(void *arg, int status); static uint32_t vnodeGetFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion); @@ -67,8 +61,8 @@ int32_t vnodeInitResources() { vnodeInitWriteFp(); vnodeInitReadFp(); - tsDnodeVnodesHash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, true); - if (tsDnodeVnodesHash == NULL) { + tsVnodesHash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, true); + if (tsVnodesHash == NULL) { vError("failed to init vnode list"); return TSDB_CODE_VND_OUT_OF_MEMORY; } @@ -77,10 +71,10 @@ int32_t vnodeInitResources() { } void vnodeCleanupResources() { - if (tsDnodeVnodesHash != NULL) { + if (tsVnodesHash != NULL) { vDebug("vnode list is cleanup"); - taosHashCleanup(tsDnodeVnodesHash); - tsDnodeVnodesHash = NULL; + taosHashCleanup(tsVnodesHash); + tsVnodesHash = NULL; } syncCleanUp(); @@ -348,7 +342,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { pVnode->status = TAOS_VN_STATUS_READY; vDebug("vgId:%d, vnode is opened in %s, pVnode:%p", pVnode->vgId, rootDir, pVnode); - taosHashPut(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *)); + taosHashPut(tsVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *)); return TSDB_CODE_SUCCESS; } @@ -440,7 +434,7 @@ void vnodeRelease(void *pVnodeRaw) { tsem_destroy(&pVnode->sem); free(pVnode); - int32_t count = taosHashGetSize(tsDnodeVnodesHash); + int32_t count = taosHashGetSize(tsVnodesHash); vDebug("vgId:%d, vnode is destroyed, vnodes:%d", vgId, count); } @@ -457,7 +451,7 @@ static void vnodeIncRef(void *ptNode) { } void *vnodeAcquire(int32_t vgId) { - SVnodeObj **ppVnode = taosHashGetCB(tsDnodeVnodesHash, &vgId, sizeof(int32_t), vnodeIncRef, NULL, sizeof(void *)); + SVnodeObj **ppVnode = taosHashGetCB(tsVnodesHash, &vgId, sizeof(int32_t), vnodeIncRef, NULL, sizeof(void *)); if (ppVnode == NULL || *ppVnode == NULL) { terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; @@ -496,7 +490,7 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) { } int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) { - SHashMutableIterator *pIter = taosHashCreateIter(tsDnodeVnodesHash); + SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash); while (taosHashIterNext(pIter)) { SVnodeObj **pVnode = taosHashIterGet(pIter); if (pVnode == NULL) continue; @@ -517,7 +511,7 @@ int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) { void vnodeBuildStatusMsg(void *param) { SStatusMsg *pStatus = param; - SHashMutableIterator *pIter = taosHashCreateIter(tsDnodeVnodesHash); + SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash); while (taosHashIterNext(pIter)) { SVnodeObj **pVnode = taosHashIterGet(pIter); @@ -546,7 +540,7 @@ void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes) { static void vnodeCleanUp(SVnodeObj *pVnode) { // remove from hash, so new messages wont be consumed - taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t)); + taosHashRemove(tsVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t)); if (pVnode->status != TAOS_VN_STATUS_INIT) { // it may be in updateing or reset state, then it shall wait diff --git a/src/vnode/src/vnodeRead.c b/src/vnode/src/vnodeRead.c index fb984f6750..b91f5ba0ab 100644 --- a/src/vnode/src/vnodeRead.c +++ b/src/vnode/src/vnodeRead.c @@ -15,13 +15,10 @@ #define _DEFAULT_SOURCE #define _NON_BLOCKING_RETRIEVE 0 - #include "os.h" - #include "tglobal.h" #include "taoserror.h" #include "taosmsg.h" -#include "tcache.h" #include "query.h" #include "trpc.h" #include "tsdb.h" @@ -29,9 +26,9 @@ #include "vnodeInt.h" #include "tqueue.h" -static int32_t (*vnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *pVnode, SVReadMsg *pReadMsg); -static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg); -static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg); +static int32_t (*vnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *pVnode, SVReadMsg *pRead); +static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead); +static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead); static int32_t vnodeNotifyCurrentQhandle(void* handle, void* qhandle, int32_t vgId); void vnodeInitReadFp(void) { @@ -44,16 +41,16 @@ void vnodeInitReadFp(void) { // still required, or there will be a deadlock, so we don’t do any check here, but put the check codes before the // request enters the queue // -int32_t vnodeProcessRead(void *param, SVReadMsg *pReadMsg) { +int32_t vnodeProcessRead(void *param, SVReadMsg *pRead) { SVnodeObj *pVnode = (SVnodeObj *)param; - int32_t msgType = pReadMsg->msgType; + int32_t msgType = pRead->msgType; if (vnodeProcessReadMsgFp[msgType] == NULL) { vDebug("vgId:%d, msgType:%s not processed, no handle", pVnode->vgId, taosMsg[msgType]); return TSDB_CODE_VND_MSG_NOT_PROCESSED; } - return (*vnodeProcessReadMsgFp[msgType])(pVnode, pReadMsg); + return (*vnodeProcessReadMsgFp[msgType])(pVnode, pRead); } static int32_t vnodeCheckRead(void *param) { @@ -180,27 +177,27 @@ static void vnodeBuildNoResultQueryRsp(SRspRet *pRet) { pRsp->completed = true; } -static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { - void * pCont = pReadMsg->pCont; - int32_t contLen = pReadMsg->contLen; - SRspRet *pRet = &pReadMsg->rspRet; +static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { + void * pCont = pRead->pCont; + int32_t contLen = pRead->contLen; + SRspRet *pRet = &pRead->rspRet; SQueryTableMsg *pQueryTableMsg = (SQueryTableMsg *)pCont; memset(pRet, 0, sizeof(SRspRet)); // qHandle needs to be freed correctly - if (pReadMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { - SRetrieveTableMsg *killQueryMsg = (SRetrieveTableMsg *)pReadMsg->pCont; + if (pRead->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { + SRetrieveTableMsg *killQueryMsg = (SRetrieveTableMsg *)pRead->pCont; killQueryMsg->free = htons(killQueryMsg->free); killQueryMsg->qhandle = htobe64(killQueryMsg->qhandle); - vWarn("QInfo:%p connection %p broken, kill query", (void *)killQueryMsg->qhandle, pReadMsg->rpcHandle); - assert(pReadMsg->contLen > 0 && killQueryMsg->free == 1); + vWarn("QInfo:%p connection %p broken, kill query", (void *)killQueryMsg->qhandle, pRead->rpcHandle); + assert(pRead->contLen > 0 && killQueryMsg->free == 1); void **qhandle = qAcquireQInfo(pVnode->qMgmt, (uint64_t)killQueryMsg->qhandle); if (qhandle == NULL || *qhandle == NULL) { vWarn("QInfo:%p invalid qhandle, no matched query handle, conn:%p", (void *)killQueryMsg->qhandle, - pReadMsg->rpcHandle); + pRead->rpcHandle); } else { assert(*qhandle == (void *)killQueryMsg->qhandle); @@ -242,9 +239,9 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { } if (handle != NULL && - vnodeNotifyCurrentQhandle(pReadMsg->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) { + vnodeNotifyCurrentQhandle(pRead->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) { vError("vgId:%d, QInfo:%p, query discarded since link is broken, %p", pVnode->vgId, *handle, - pReadMsg->rpcHandle); + pRead->rpcHandle); pRsp->code = TSDB_CODE_RPC_NETWORK_UNAVAIL; qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true); return pRsp->code; @@ -255,7 +252,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { if (handle != NULL) { vDebug("vgId:%d, QInfo:%p, dnode query msg disposed, create qhandle and returns to app", vgId, *handle); - code = vnodePutItemIntoReadQueue(pVnode, handle, pReadMsg->rpcHandle); + code = vnodePutItemIntoReadQueue(pVnode, handle, pRead->rpcHandle); if (code != TSDB_CODE_SUCCESS) { pRsp->code = code; qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true); @@ -264,7 +261,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { } } else { assert(pCont != NULL); - void **qhandle = (void **)pReadMsg->qhandle; + void **qhandle = (void **)pRead->qhandle; vDebug("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle); @@ -276,14 +273,14 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { // build query rsp, the retrieve request has reached here already if (buildRes) { // update the connection info according to the retrieve connection - pReadMsg->rpcHandle = qGetResultRetrieveMsg(*qhandle); - assert(pReadMsg->rpcHandle != NULL); + pRead->rpcHandle = qGetResultRetrieveMsg(*qhandle); + assert(pRead->rpcHandle != NULL); vDebug("vgId:%d, QInfo:%p, start to build retrieval rsp after query paused, %p", pVnode->vgId, *qhandle, - pReadMsg->rpcHandle); + pRead->rpcHandle); // set the real rsp error code - pReadMsg->code = vnodeDumpQueryResult(&pRead->rspRet, pVnode, qhandle, &freehandle, pReadMsg->rpcHandle); + pRead->code = vnodeDumpQueryResult(&pRead->rspRet, pVnode, qhandle, &freehandle, pRead->rpcHandle); // NOTE: set return code to be TSDB_CODE_QRY_HAS_RSP to notify dnode to return msg to client code = TSDB_CODE_QRY_HAS_RSP; @@ -308,16 +305,16 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { return code; } -static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { - void * pCont = pReadMsg->pCont; - SRspRet *pRet = &pReadMsg->rspRet; +static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { + void * pCont = pRead->pCont; + SRspRet *pRet = &pRead->rspRet; SRetrieveTableMsg *pRetrieve = pCont; pRetrieve->free = htons(pRetrieve->free); pRetrieve->qhandle = htobe64(pRetrieve->qhandle); vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, (void *)pRetrieve->qhandle, - pRetrieve->free, pReadMsg->rpcHandle); + pRetrieve->free, pRead->rpcHandle); memset(pRet, 0, sizeof(SRspRet)); @@ -348,8 +345,8 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { } // register the qhandle to connect to quit query immediate if connection is broken - if (vnodeNotifyCurrentQhandle(pReadMsg->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) { - vError("vgId:%d, QInfo:%p, retrieve discarded since link is broken, %p", pVnode->vgId, *handle, pReadMsg->rpcHandle); + if (vnodeNotifyCurrentQhandle(pRead->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) { + vError("vgId:%d, QInfo:%p, retrieve discarded since link is broken, %p", pVnode->vgId, *handle, pRead->rpcHandle); code = TSDB_CODE_RPC_NETWORK_UNAVAIL; qKillQuery(*handle); qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true); @@ -359,7 +356,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { bool freeHandle = true; bool buildRes = false; - code = qRetrieveQueryResultInfo(*handle, &buildRes, pReadMsg->rpcHandle); + code = qRetrieveQueryResultInfo(*handle, &buildRes, pRead->rpcHandle); if (code != TSDB_CODE_SUCCESS) { // TODO handle malloc failure pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp)); @@ -370,7 +367,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { assert(buildRes == true); #if _NON_BLOCKING_RETRIEVE if (!buildRes) { - assert(pReadMsg->rpcHandle != NULL); + assert(pRead->rpcHandle != NULL); qReleaseQInfo(pVnode->qMgmt, (void **)&handle, false); return TSDB_CODE_QRY_NOT_READY; @@ -378,7 +375,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { #endif // ahandle is the sqlObj pointer - code = vnodeDumpQueryResult(pRet, pVnode, handle, &freeHandle, pReadMsg->rpcHandle); + code = vnodeDumpQueryResult(pRet, pVnode, handle, &freeHandle, pRead->rpcHandle); } // If qhandle is not added into vread queue, the query should be completed already or paused with error. diff --git a/src/vnode/src/vnodeWrite.c b/src/vnode/src/vnodeWrite.c index 3caee2fb0c..f42b359b26 100644 --- a/src/vnode/src/vnodeWrite.c +++ b/src/vnode/src/vnodeWrite.c @@ -19,7 +19,6 @@ #include "taoserror.h" #include "tqueue.h" #include "trpc.h" -#include "tutil.h" #include "tsdb.h" #include "twal.h" #include "tsync.h" @@ -185,7 +184,7 @@ static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pRet) { SDropSTableMsg *pTable = pCont; - int32_t code = TSDB_CODE_SUCCESS; + int32_t code = TSDB_CODE_SUCCESS; vDebug("vgId:%d, stable:%s, start to drop", pVnode->vgId, pTable->tableId); -- GitLab