From a9b32dc3276755c266932ae8bd4d2d5d32fb7bd5 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 2 Jun 2022 18:26:29 +0800 Subject: [PATCH] refactor: adjust vnode queue --- include/common/tmsgcb.h | 4 +- source/common/src/tmsgcb.c | 36 +++---------- source/dnode/mgmt/mgmt_mnode/inc/mmInt.h | 5 +- source/dnode/mgmt/mgmt_mnode/src/mmInt.c | 5 +- source/dnode/mgmt/mgmt_mnode/src/mmWorker.c | 56 ++++++++++----------- source/dnode/mgmt/mgmt_qnode/inc/qmInt.h | 3 +- source/dnode/mgmt/mgmt_qnode/src/qmInt.c | 3 +- source/dnode/mgmt/mgmt_qnode/src/qmWorker.c | 26 +++++----- source/dnode/mgmt/mgmt_vnode/inc/vmInt.h | 41 +++++++-------- source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 9 +--- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 31 +----------- source/libs/qworker/test/qworkerTests.cpp | 33 ++++++------ 12 files changed, 93 insertions(+), 159 deletions(-) diff --git a/include/common/tmsgcb.h b/include/common/tmsgcb.h index 9fa657a2a6..e99377f9b4 100644 --- a/include/common/tmsgcb.h +++ b/include/common/tmsgcb.h @@ -38,7 +38,7 @@ typedef enum { QUEUE_MAX, } EQueueType; -typedef int32_t (*PutToQueueFp)(void* pMgmt, SRpcMsg* pMsg); +typedef int32_t (*PutToQueueFp)(void* pMgmt, EQueueType qtype, SRpcMsg* pMsg); typedef int32_t (*GetQueueSizeFp)(void* pMgmt, int32_t vgId, EQueueType qtype); typedef int32_t (*SendReqFp)(const SEpSet* pEpSet, SRpcMsg* pMsg); typedef void (*SendRspFp)(SRpcMsg* pMsg); @@ -50,7 +50,7 @@ typedef void (*ReportStartup)(const char* name, const char* desc); typedef struct { void* mgmt; void* clientRpc; - PutToQueueFp queueFps[QUEUE_MAX]; + PutToQueueFp putToQueueFp; GetQueueSizeFp qsizeFp; SendReqFp sendReqFp; SendRspFp sendRspFp; diff --git a/source/common/src/tmsgcb.c b/source/common/src/tmsgcb.c index 126a4c023a..b8eec655b1 100644 --- a/source/common/src/tmsgcb.c +++ b/source/common/src/tmsgcb.c @@ -22,41 +22,21 @@ static SMsgCb defaultMsgCb; void tmsgSetDefault(const SMsgCb* msgcb) { defaultMsgCb = *msgcb; } int32_t tmsgPutToQueue(const SMsgCb* msgcb, EQueueType qtype, SRpcMsg* pMsg) { - PutToQueueFp fp = msgcb->queueFps[qtype]; - return (*fp)(msgcb->mgmt, pMsg); + return (*msgcb->putToQueueFp)(msgcb->mgmt, qtype, pMsg); } int32_t tmsgGetQueueSize(const SMsgCb* msgcb, int32_t vgId, EQueueType qtype) { - GetQueueSizeFp fp = msgcb->qsizeFp; - return (*fp)(msgcb->mgmt, vgId, qtype); + return (*msgcb->qsizeFp)(msgcb->mgmt, vgId, qtype); } -int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) { - SendReqFp fp = defaultMsgCb.sendReqFp; - return (*fp)(epSet, pMsg); -} +int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) { return (*defaultMsgCb.sendReqFp)(epSet, pMsg); } -void tmsgSendRsp(SRpcMsg* pMsg) { - SendRspFp fp = defaultMsgCb.sendRspFp; - return (*fp)(pMsg); -} +void tmsgSendRsp(SRpcMsg* pMsg) { return (*defaultMsgCb.sendRspFp)(pMsg); } -void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) { - SendRedirectRspFp fp = defaultMsgCb.sendRedirectRspFp; - (*fp)(pMsg, pNewEpSet); -} +void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) { (*defaultMsgCb.sendRedirectRspFp)(pMsg, pNewEpSet); } -void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) { - RegisterBrokenLinkArgFp fp = defaultMsgCb.registerBrokenLinkArgFp; - (*fp)(pMsg); -} +void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) { (*defaultMsgCb.registerBrokenLinkArgFp)(pMsg); } -void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) { - ReleaseHandleFp fp = defaultMsgCb.releaseHandleFp; - (*fp)(pHandle, type); -} +void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) { (*defaultMsgCb.releaseHandleFp)(pHandle, type); } -void tmsgReportStartup(const char* name, const char* desc) { - ReportStartup fp = defaultMsgCb.reportStartupFp; - (*fp)(name, desc); -} \ No newline at end of file +void tmsgReportStartup(const char* name, const char* desc) { (*defaultMsgCb.reportStartupFp)(name, desc); } \ No newline at end of file diff --git a/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h b/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h index bd034fe7d6..9a0cfdfc93 100644 --- a/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h +++ b/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h @@ -65,10 +65,7 @@ int32_t mmPutNodeMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutNodeMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutNodeMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc); #ifdef __cplusplus } diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmInt.c b/source/dnode/mgmt/mgmt_mnode/src/mmInt.c index 1b973f3045..0f3c06cb3a 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmInt.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmInt.c @@ -105,10 +105,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->path = pInput->path; pMgmt->name = pInput->name; pMgmt->msgCb = pInput->msgCb; - pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)mmPutRpcMsgToQueryQueue; - pMgmt->msgCb.queueFps[READ_QUEUE] = (PutToQueueFp)mmPutRpcMsgToReadQueue; - pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)mmPutRpcMsgToWriteQueue; - pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)mmPutRpcMsgToSyncQueue; + pMgmt->msgCb.putToQueueFp = (PutToQueueFp)mmPutRpcMsgToQueue; pMgmt->msgCb.mgmt = pMgmt; taosThreadRwlockInit(&pMgmt->lock, NULL); diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c index 1de9875d06..460f2242f2 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c @@ -96,40 +96,38 @@ int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { return mmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg); } -static inline int32_t mmPutRpcMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pRpc) { +int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); if (pMsg == NULL) return -1; - - dTrace("msg:%p, create and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType)); memcpy(pMsg, pRpc, sizeof(SRpcMsg)); - taosWriteQitem(pWorker->queue, pMsg); - return 0; -} -int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return mmPutRpcMsgToWorker(&pMgmt->queryWorker, pMsg); -} - -int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return mmPutRpcMsgToWorker(&pMgmt->writeWorker, pMsg); -} - -int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return mmPutRpcMsgToWorker(&pMgmt->readWorker, pMsg); -} - -int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - int32_t code = -1; - if (mmAcquire(pMgmt) == 0) { - code = mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg); - mmRelease(pMgmt); - } - - if (code != 0) { - rpcFreeCont(pMsg->pCont); - pMsg->pCont = NULL; + switch (qtype) { + case WRITE_QUEUE: + dTrace("msg:%p, is created and will put into vnode-write queue", pMsg); + taosWriteQitem(pMgmt->writeWorker.queue, pMsg); + return 0; + case QUERY_QUEUE: + dTrace("msg:%p, is created and will put into vnode-query queue", pMsg); + taosWriteQitem(pMgmt->queryWorker.queue, pMsg); + return 0; + + case READ_QUEUE: + dTrace("msg:%p, is created and will put into vnode-read queue", pMsg); + taosWriteQitem(pMgmt->readWorker.queue, pMsg); + return 0; + case SYNC_QUEUE: + if (mmAcquire(pMgmt) == 0) { + dTrace("msg:%p, is created and will put into vnode-sync queue", pMsg); + taosWriteQitem(pMgmt->syncWorker.queue, pMsg); + mmRelease(pMgmt); + return 0; + } else { + return -1; + } + default: + terrno = TSDB_CODE_INVALID_PARA; + return -1; } - return code; } int32_t mmStartWorker(SMnodeMgmt *pMgmt) { diff --git a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h index 9738fb0c45..54e9da24a4 100644 --- a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h +++ b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h @@ -42,8 +42,7 @@ int32_t qmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg); int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); // qmWorker.c -int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pMsg); int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype); int32_t qmStartWorker(SQnodeMgmt *pMgmt); diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmInt.c b/source/dnode/mgmt/mgmt_qnode/src/qmInt.c index 06c18ab288..3b425a0b49 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmInt.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmInt.c @@ -43,8 +43,7 @@ static int32_t qmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->path = pInput->path; pMgmt->name = pInput->name; pMgmt->msgCb = pInput->msgCb; - pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qmPutRpcMsgToQueryQueue; - pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)qmPutRpcMsgToFetchQueue; + pMgmt->msgCb.putToQueueFp = (PutToQueueFp)qmPutRpcMsgToQueue; pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)qmGetQueueSize; pMgmt->msgCb.mgmt = pMgmt; diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c index e7fc261b67..e36efa83db 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c @@ -68,22 +68,24 @@ int32_t qmPutNodeMsgToMonitorQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) { return qmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg); } -static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) { +int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); if (pMsg == NULL) return -1; - - dTrace("msg:%p, create and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType)); memcpy(pMsg, pRpc, sizeof(SRpcMsg)); - taosWriteQitem(pWorker->queue, pMsg); - return 0; -} - -int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return qmPutRpcMsgToWorker(pMgmt, &pMgmt->queryWorker, pRpc); -} -int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return qmPutRpcMsgToWorker(pMgmt, &pMgmt->fetchWorker, pRpc); + switch (qtype) { + case QUERY_QUEUE: + dTrace("msg:%p, is created and will put into qnode-query queue", pMsg); + taosWriteQitem(pMgmt->queryWorker.queue, pMsg); + return 0; + case READ_QUEUE: + dTrace("msg:%p, is created and will put into qnode-fetch queue", pMsg); + taosWriteQitem(pMgmt->fetchWorker.queue, pMsg); + return 0; + default: + terrno = TSDB_CODE_INVALID_PARA; + return -1; + } } int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { diff --git a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h index f74fd119c5..adc18fdced 100644 --- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h +++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h @@ -44,26 +44,26 @@ typedef struct SVnodeMgmt { } SVnodeMgmt; typedef struct { - int32_t vgId; - int32_t vgVersion; - int8_t dropped; - char path[PATH_MAX + 20]; + int32_t vgId; + int32_t vgVersion; + int8_t dropped; + char path[PATH_MAX + 20]; } SWrapperCfg; typedef struct { - int32_t vgId; - int32_t refCount; - int32_t vgVersion; - int8_t dropped; - int8_t accessState; - char *path; - SVnode *pImpl; - STaosQueue *pWriteQ; - STaosQueue *pSyncQ; - STaosQueue *pApplyQ; - STaosQueue *pQueryQ; - STaosQueue *pFetchQ; - STaosQueue *pMergeQ; + int32_t vgId; + int32_t refCount; + int32_t vgVersion; + int8_t dropped; + int8_t accessState; + char *path; + SVnode *pImpl; + STaosQueue *pWriteQ; + STaosQueue *pSyncQ; + STaosQueue *pApplyQ; + STaosQueue *pQueryQ; + STaosQueue *pFetchQ; + STaosQueue *pMergeQ; } SVnodeObj; typedef struct { @@ -100,13 +100,8 @@ void vmStopWorker(SVnodeMgmt *pMgmt); int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode); void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode); -int32_t vmPutRpcMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToApplyQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype); +int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc); int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c index 5d3479f74b..7e76ee3af2 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c @@ -156,7 +156,7 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) { pMgmt->state.totalVnodes = numOfVnodes; - int32_t threadNum = tsNumOfCores; + int32_t threadNum = 1; int32_t vnodesPerThread = numOfVnodes / threadNum + 1; SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread)); @@ -249,12 +249,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->path = pInput->path; pMgmt->name = pInput->name; pMgmt->msgCb = pInput->msgCb; - pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToWriteQueue; - pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)vmPutRpcMsgToSyncQueue; - pMgmt->msgCb.queueFps[APPLY_QUEUE] = (PutToQueueFp)vmPutRpcMsgToApplyQueue; - pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)vmPutRpcMsgToQueryQueue; - pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)vmPutRpcMsgToFetchQueue; - pMgmt->msgCb.queueFps[MERGE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToMergeQueue; + pMgmt->msgCb.putToQueueFp = (PutToQueueFp)vmPutRpcMsgToQueue; pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)vmGetQueueSize; pMgmt->msgCb.mgmt = pMgmt; taosThreadRwlockInit(&pMgmt->lock, NULL); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index e5953b8383..dbd653a44e 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -304,14 +304,9 @@ int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return 0; } -static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType qtype) { +int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); - if (pMsg == NULL) { - rpcFreeCont(pRpc->pCont); - pRpc->pCont = NULL; - terrno = TSDB_CODE_OUT_OF_MEMORY; - return -1; - } + if (pMsg == NULL) return -1; SMsgHead *pHead = pRpc->pCont; pHead->contLen = htonl(pHead->contLen); @@ -322,28 +317,6 @@ static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType q return vmPutMsgToQueue(pMgmt, pMsg, qtype); } -int32_t vmPutRpcMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, WRITE_QUEUE); -} - -int32_t vmPutRpcMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { return vmPutRpcMsgToQueue(pMgmt, pRpc, SYNC_QUEUE); } - -int32_t vmPutRpcMsgToApplyQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, APPLY_QUEUE); -} - -int32_t vmPutRpcMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, QUERY_QUEUE); -} - -int32_t vmPutRpcMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, FETCH_QUEUE); -} - -int32_t vmPutRpcMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, MERGE_QUEUE); -} - int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { int32_t size = -1; SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId); diff --git a/source/libs/qworker/test/qworkerTests.cpp b/source/libs/qworker/test/qworkerTests.cpp index 16dcd7b6e0..5bb6acee60 100644 --- a/source/libs/qworker/test/qworkerTests.cpp +++ b/source/libs/qworker/test/qworkerTests.cpp @@ -172,8 +172,7 @@ int32_t qwtPutReqToFetchQueue(void *node, struct SRpcMsg *pMsg) { return 0; } - -int32_t qwtPutReqToQueue(void *node, struct SRpcMsg *pMsg) { +int32_t qwtPutReqToQueue(void *node, EQueueType qtype, struct SRpcMsg *pMsg) { taosWLockLatch(&qwtTestQueryQueueLock); struct SRpcMsg *newMsg = (struct SRpcMsg *)taosMemoryCalloc(1, sizeof(struct SRpcMsg)); memcpy(newMsg, pMsg, sizeof(struct SRpcMsg)); @@ -181,7 +180,7 @@ int32_t qwtPutReqToQueue(void *node, struct SRpcMsg *pMsg) { if (qwtTestQueryQueueWIdx >= qwtTestQueryQueueSize) { qwtTestQueryQueueWIdx = 0; } - + qwtTestQueryQueueNum++; if (qwtTestQueryQueueWIdx == qwtTestQueryQueueRIdx) { @@ -702,9 +701,9 @@ void *qwtclientThread(void *param) { while (!qwtTestStop) { qwtTestCaseFinished = false; - + qwtBuildQueryReqMsg(&queryRpc); - qwtPutReqToQueue((void *)0x1, &queryRpc); + qwtPutReqToQueue((void *)0x1, QUERY_QUEUE, &queryRpc); while (!qwtTestCaseFinished) { taosUsleep(1); @@ -874,7 +873,7 @@ TEST(seqTest, normalCase) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -907,10 +906,10 @@ TEST(seqTest, cancelFirst) { stubSetStringToPlan(); stubSetRpcSendResponse(); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -944,10 +943,10 @@ TEST(seqTest, randCase) { stubSetCreateExecTask(); taosSeedRand(taosGetTimestampSec()); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1015,10 +1014,10 @@ TEST(seqTest, multithreadRand) { stubSetGetDataBlock(); taosSeedRand(taosGetTimestampSec()); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1081,7 +1080,7 @@ TEST(rcTest, shortExecshortDelay) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1165,7 +1164,7 @@ TEST(rcTest, longExecshortDelay) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1251,7 +1250,7 @@ TEST(rcTest, shortExeclongDelay) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1332,10 +1331,10 @@ TEST(rcTest, dropTest) { stubSetGetDataBlock(); taosSeedRand(taosGetTimestampSec()); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); -- GitLab