diff --git a/src/dnode/inc/dnodeVRead.h b/src/dnode/inc/dnodeVRead.h index b3c3df80b2e201ff9f4b40ebfcceafba57de1366..5b17693146cb68ab66943593bc2f9cd6587cc225 100644 --- a/src/dnode/inc/dnodeVRead.h +++ b/src/dnode/inc/dnodeVRead.h @@ -24,7 +24,7 @@ int32_t dnodeInitVRead(); void dnodeCleanupVRead(); void dnodeDispatchToVReadQueue(SRpcMsg *pMsg); void * dnodeAllocVReadQueue(void *pVnode); -void dnodeFreeVReadQueue(void *rqueue); +void dnodeFreeVReadQueue(void *pRqueue); #ifdef __cplusplus } diff --git a/src/dnode/src/dnodeVRead.c b/src/dnode/src/dnodeVRead.c index 34df11adcce72a6a21718b81c1895c30c66bea1c..e1c1b1585d02cc75228dbe065884e744b0d9ab5c 100644 --- a/src/dnode/src/dnodeVRead.c +++ b/src/dnode/src/dnodeVRead.c @@ -35,7 +35,7 @@ typedef struct { pthread_mutex_t mutex; } SVReadWorkerPool; -static void *dnodeProcessReadQueue(void *param); +static void *dnodeProcessReadQueue(void *pWorker); // module global variable static SVReadWorkerPool tsVReadWP; @@ -47,7 +47,7 @@ int32_t dnodeInitVRead() { tsVReadWP.min = tsNumOfCores; tsVReadWP.max = tsNumOfCores * tsNumOfThreadsPerCore; if (tsVReadWP.max <= tsVReadWP.min * 2) tsVReadWP.max = 2 * tsVReadWP.min; - tsVReadWP.worker = (SVReadWorker *)calloc(sizeof(SVReadWorker), tsVReadWP.max); + tsVReadWP.worker = calloc(sizeof(SVReadWorker), tsVReadWP.max); pthread_mutex_init(&tsVReadWP.mutex, NULL); if (tsVReadWP.worker == NULL) return -1; @@ -85,7 +85,7 @@ void dnodeCleanupVRead() { void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) { int32_t queuedMsgNum = 0; int32_t leftLen = pMsg->contLen; - char * pCont = (char *)pMsg->pCont; + char * pCont = pMsg->pCont; while (leftLen > 0) { SMsgHead *pHead = (SMsgHead *)pCont; @@ -146,8 +146,8 @@ void *dnodeAllocVReadQueue(void *pVnode) { return queue; } -void dnodeFreeVReadQueue(void *rqueue) { - taosCloseQueue(rqueue); +void dnodeFreeVReadQueue(void *pRqueue) { + taosCloseQueue(pRqueue); } void dnodeSendRpcVReadRsp(void *pVnode, SVReadMsg *pRead, int32_t code) { @@ -159,14 +159,12 @@ void dnodeSendRpcVReadRsp(void *pVnode, SVReadMsg *pRead, int32_t code) { }; rpcSendResponse(&rpcRsp); - vnodeRelease(pVnode); } void dnodeDispatchNonRspMsg(void *pVnode, SVReadMsg *pRead, int32_t code) { - vnodeRelease(pVnode); } -static void *dnodeProcessReadQueue(void *param) { +static void *dnodeProcessReadQueue(void *pWorker) { SVReadMsg *pRead; int32_t qtype; void * pVnode; @@ -193,7 +191,7 @@ static void *dnodeProcessReadQueue(void *param) { } } - taosFreeQitem(pRead); + vnodeFreeFromRQueue(pVnode, pRead); } return NULL; diff --git a/src/inc/dnode.h b/src/inc/dnode.h index d71fdd5e0a12511d84366a661f6c9dd8a73e8c9a..eef4490800a4191c2dee55c450cf99b8381bb64d 100644 --- a/src/inc/dnode.h +++ b/src/inc/dnode.h @@ -57,7 +57,7 @@ void *dnodeAllocVWriteQueue(void *pVnode); void dnodeFreeVWriteQueue(void *pWqueue); void dnodeSendRpcVWriteRsp(void *pVnode, void *pWrite, int32_t code); void *dnodeAllocVReadQueue(void *pVnode); -void dnodeFreeVReadQueue(void *rqueue); +void dnodeFreeVReadQueue(void *pRqueue); int32_t dnodeAllocateMPeerQueue(); void dnodeFreeMPeerQueue(); diff --git a/src/inc/vnode.h b/src/inc/vnode.h index 563d035898758a69fb57d6e1e5d8dc7e99886af4..4e8389498b5ced87f4ff07dddac817614f5af368 100644 --- a/src/inc/vnode.h +++ b/src/inc/vnode.h @@ -23,12 +23,12 @@ extern "C" { #include "twal.h" typedef enum _VN_STATUS { - TAOS_VN_STATUS_INIT, - TAOS_VN_STATUS_READY, - TAOS_VN_STATUS_CLOSING, - TAOS_VN_STATUS_UPDATING, - TAOS_VN_STATUS_RESET, -} EVnStatus; + TAOS_VN_STATUS_INIT = 0, + TAOS_VN_STATUS_READY = 1, + TAOS_VN_STATUS_CLOSING = 2, + TAOS_VN_STATUS_UPDATING = 3, + TAOS_VN_STATUS_RESET = 4, +} EVnodeStatus; typedef struct { int32_t len; @@ -81,7 +81,8 @@ void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes); int32_t vnodeInitResources(); void vnodeCleanupResources(); -int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam); +int32_t vnodeWriteToRQueue(void *pVnode, void *pCont, int32_t contLen, int8_t qtype, void *rparam); +void vnodeFreeFromRQueue(void *pVnode, SVReadMsg *pRead); int32_t vnodeProcessRead(void *pVnode, SVReadMsg *pRead); #ifdef __cplusplus diff --git a/src/vnode/inc/vnodeInt.h b/src/vnode/inc/vnodeInt.h index c9689ce8b266acd88b0b6212637768934e4833ed..021831a644e1da3d1de882e14852d1fa5e58bc1d 100644 --- a/src/vnode/inc/vnodeInt.h +++ b/src/vnode/inc/vnodeInt.h @@ -37,8 +37,9 @@ extern int32_t vDebugFlag; typedef struct { int32_t vgId; // global vnode group ID int32_t refCount; // reference count - int32_t queuedMsg; - int32_t delay; + int32_t queuedWMsg; + int32_t queuedRMsg; + int32_t delayMs; int8_t status; int8_t role; int8_t accessState; diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 7447acc488c2e05298b1096331745420e3fc9739..3f20efb150a8607a39fb7df11f387a46bc97f011 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -378,9 +378,9 @@ int32_t vnodeClose(int32_t vgId) { return 0; } -void vnodeRelease(void *pVnodeRaw) { - if (pVnodeRaw == NULL) return; - SVnodeObj *pVnode = pVnodeRaw; +void vnodeRelease(void *vparam) { + if (vparam == NULL) return; + SVnodeObj *pVnode = vparam; int32_t code = 0; int32_t vgId = pVnode->vgId; @@ -643,18 +643,19 @@ static void vnodeNotifyRole(void *ahandle, int8_t role) { pVnode->role = role; dnodeSendStatusMsgToMnode(); - if (pVnode->role == TAOS_SYNC_ROLE_MASTER) + if (pVnode->role == TAOS_SYNC_ROLE_MASTER) { cqStart(pVnode->cq); - else + } else { cqStop(pVnode->cq); + } } static void vnodeCtrlFlow(void *ahandle, int32_t mseconds) { SVnodeObj *pVnode = ahandle; - if (pVnode->delay != mseconds) { - vInfo("vgId:%d, sync flow control, mseconds:%d", pVnode->vgId, mseconds); + if (pVnode->delayMs != mseconds) { + pVnode->delayMs = mseconds; + vDebug("vgId:%d, sync flow control, mseconds:%d", pVnode->vgId, mseconds); } - pVnode->delay = mseconds; } static int vnodeResetTsdb(SVnodeObj *pVnode) { diff --git a/src/vnode/src/vnodeRead.c b/src/vnode/src/vnodeRead.c index 4cc8819bcba07847cb3fa1b09f0663b6368b148b..e10d62a0c9f283eded9969547fad779468d3cdaa 100644 --- a/src/vnode/src/vnodeRead.c +++ b/src/vnode/src/vnodeRead.c @@ -41,8 +41,8 @@ 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 *pRead) { - SVnodeObj *pVnode = (SVnodeObj *)param; +int32_t vnodeProcessRead(void *vparam, SVReadMsg *pRead) { + SVnodeObj *pVnode = vparam; int32_t msgType = pRead->msgType; if (vnodeProcessReadMsgFp[msgType] == NULL) { @@ -53,8 +53,8 @@ int32_t vnodeProcessRead(void *param, SVReadMsg *pRead) { return (*vnodeProcessReadMsgFp[msgType])(pVnode, pRead); } -static int32_t vnodeCheckRead(void *param) { - SVnodeObj *pVnode = param; +static int32_t vnodeCheckRead(void *vparam) { + SVnodeObj *pVnode = vparam; if (pVnode->status != TAOS_VN_STATUS_READY) { vDebug("vgId:%d, vnode status is %s, refCount:%d pVnode:%p", pVnode->vgId, vnodeStatus[pVnode->status], pVnode->refCount, pVnode); @@ -76,6 +76,16 @@ static int32_t vnodeCheckRead(void *param) { return TSDB_CODE_SUCCESS; } +void vnodeFreeFromRQueue(void *vparam, SVReadMsg *pRead) { + SVnodeObj *pVnode = vparam; + + atomic_sub_fetch_32(&pVnode->queuedRMsg, 1); + vTrace("vgId:%d, free from vrqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedRMsg); + + taosFreeQitem(pRead); + vnodeRelease(pVnode); +} + int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam) { SVnodeObj *pVnode = vparam; @@ -108,7 +118,8 @@ int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qt pRead->qtype = qtype; atomic_add_fetch_32(&pVnode->refCount, 1); - vTrace("vgId:%d, get vnode rqueue, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode); + atomic_add_fetch_32(&pVnode->queuedRMsg, 1); + vTrace("vgId:%d, write into vrqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedRMsg); taosWriteQitem(pVnode->rqueue, qtype, pRead); return TSDB_CODE_SUCCESS; diff --git a/src/vnode/src/vnodeWrite.c b/src/vnode/src/vnodeWrite.c index 2234b4f8eda7caa8622b60be8a9018d6aae40ad1..2d2be602ed217df502101e423613c6631b400b9f 100644 --- a/src/vnode/src/vnodeWrite.c +++ b/src/vnode/src/vnodeWrite.c @@ -77,7 +77,7 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara // assign version pHead->version = pVnode->version + 1; - if (pVnode->delay) usleep(pVnode->delay * 1000); + if (pVnode->delayMs) taosMsleep(pVnode->delayMs); } else { // from wal or forward // for data from WAL or forward, version may be smaller @@ -245,13 +245,13 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar atomic_add_fetch_32(&pVnode->refCount, 1); - int32_t queued = atomic_add_fetch_32(&pVnode->queuedMsg, 1); + int32_t queued = atomic_add_fetch_32(&pVnode->queuedWMsg, 1); if (queued > MAX_QUEUED_MSG_NUM) { vDebug("vgId:%d, too many msg:%d in vwqueue, flow control", pVnode->vgId, queued); taosMsleep(1); } - vTrace("vgId:%d, write into vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedMsg); + vTrace("vgId:%d, write into vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedWMsg); taosWriteQitem(pVnode->wqueue, qtype, pWrite); return TSDB_CODE_SUCCESS; @@ -260,8 +260,8 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar void vnodeFreeFromWQueue(void *vparam, SVWriteMsg *pWrite) { SVnodeObj *pVnode = vparam; - atomic_sub_fetch_32(&pVnode->queuedMsg, 1); - vTrace("vgId:%d, free from vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedMsg); + atomic_sub_fetch_32(&pVnode->queuedWMsg, 1); + vTrace("vgId:%d, free from vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedWMsg); taosFreeQitem(pWrite); vnodeRelease(pVnode);