diff --git a/src/dnode/src/dnodeVMgmt.c b/src/dnode/src/dnodeVMgmt.c index 66c94bf675..41e48411ac 100644 --- a/src/dnode/src/dnodeVMgmt.c +++ b/src/dnode/src/dnodeVMgmt.c @@ -169,7 +169,7 @@ static int32_t dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) { static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) { SAlterVnodeMsg *pAlter = dnodeParseVnodeMsg(rpcMsg); - void *pVnode = vnodeAcquire(pAlter->cfg.vgId); + void *pVnode = vnodeAcquireNotClose(pAlter->cfg.vgId); if (pVnode != NULL) { dDebug("vgId:%d, alter vnode msg is received", pAlter->cfg.vgId); int32_t code = vnodeAlter(pVnode, pAlter); diff --git a/src/dnode/src/dnodeVRead.c b/src/dnode/src/dnodeVRead.c index ea738661ce..41016d7b99 100644 --- a/src/dnode/src/dnodeVRead.c +++ b/src/dnode/src/dnodeVRead.c @@ -63,7 +63,7 @@ void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) { pHead->contLen = htonl(pHead->contLen); assert(pHead->contLen > 0); - void *pVnode = vnodeAcquire(pHead->vgId); + void *pVnode = vnodeAcquireNotClose(pHead->vgId); if (pVnode != NULL) { code = vnodeWriteToRQueue(pVnode, pCont, pHead->contLen, TAOS_QTYPE_RPC, pMsg); if (code == TSDB_CODE_SUCCESS) queuedMsgNum++; diff --git a/src/dnode/src/dnodeVWrite.c b/src/dnode/src/dnodeVWrite.c index 26084a52eb..bbf257ff95 100644 --- a/src/dnode/src/dnodeVWrite.c +++ b/src/dnode/src/dnodeVWrite.c @@ -85,7 +85,7 @@ void dnodeDispatchToVWriteQueue(SRpcMsg *pRpcMsg) { pMsg->vgId = htonl(pMsg->vgId); pMsg->contLen = htonl(pMsg->contLen); - void *pVnode = vnodeAcquire(pMsg->vgId); + void *pVnode = vnodeAcquireNotClose(pMsg->vgId); if (pVnode == NULL) { code = TSDB_CODE_VND_INVALID_VGROUP_ID; } else { diff --git a/src/inc/vnode.h b/src/inc/vnode.h index 39bd2f46c3..576acb624a 100644 --- a/src/inc/vnode.h +++ b/src/inc/vnode.h @@ -68,6 +68,7 @@ int32_t vnodeInitMgmt(); void vnodeCleanupMgmt(); void* vnodeAcquire(int32_t vgId); void vnodeRelease(void *pVnode); +void* vnodeAcquireNotClose(int32_t vgId); void* vnodeGetWal(void *pVnode); int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes); void vnodeBuildStatusMsg(void *pStatus); diff --git a/src/vnode/inc/vnodeInt.h b/src/vnode/inc/vnodeInt.h index d770a38e37..ef05cf4a40 100644 --- a/src/vnode/inc/vnodeInt.h +++ b/src/vnode/inc/vnodeInt.h @@ -41,6 +41,8 @@ typedef struct { int32_t queuedWMsg; int32_t queuedRMsg; int32_t flowctrlLevel; + int8_t preClose; // drop and close switch + int8_t reserved[3]; int64_t sequence; // for topic int8_t status; int8_t role; diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index f4515b6688..62aa91894a 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -93,7 +93,7 @@ int32_t vnodeCreate(SCreateVnodeMsg *pVnodeCfg) { } int32_t vnodeSync(int32_t vgId) { - SVnodeObj *pVnode = vnodeAcquire(vgId); + SVnodeObj *pVnode = vnodeAcquireNotClose(vgId); if (pVnode == NULL) { vDebug("vgId:%d, failed to sync, vnode not find", vgId); return TSDB_CODE_VND_INVALID_VGROUP_ID; @@ -115,7 +115,7 @@ int32_t vnodeSync(int32_t vgId) { } int32_t vnodeDrop(int32_t vgId) { - SVnodeObj *pVnode = vnodeAcquire(vgId); + SVnodeObj *pVnode = vnodeAcquireNotClose(vgId); if (pVnode == NULL) { vDebug("vgId:%d, failed to drop, vnode not find", vgId); return TSDB_CODE_VND_INVALID_VGROUP_ID; @@ -390,15 +390,16 @@ int32_t vnodeOpen(int32_t vgId) { } int32_t vnodeClose(int32_t vgId) { - SVnodeObj *pVnode = vnodeAcquire(vgId); + SVnodeObj *pVnode = vnodeAcquireNotClose(vgId); if (pVnode == NULL) return 0; if (pVnode->dropped) { vnodeRelease(pVnode); return 0; } + pVnode->preClose = 1; + vDebug("vgId:%d, vnode will be closed, pVnode:%p", pVnode->vgId, pVnode); - vnodeRemoveFromHash(pVnode); vnodeRelease(pVnode); vnodeCleanUp(pVnode); diff --git a/src/vnode/src/vnodeMgmt.c b/src/vnode/src/vnodeMgmt.c index 7e6022fc87..a30628680f 100644 --- a/src/vnode/src/vnodeMgmt.c +++ b/src/vnode/src/vnodeMgmt.c @@ -125,6 +125,18 @@ void vnodeRelease(void *vparam) { } } +void *vnodeAcquireNotClose(int32_t vgId) { + SVnodeObj *pVnode = vnodeAcquire(vgId); + if (pVnode != NULL && pVnode->preClose == 1) { + vnodeRelease(pVnode); + terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; + vDebug("vgId:%d, not exist, pre closing", vgId); + return NULL; + } + + return pVnode; +} + static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) { int64_t totalStorage = 0; int64_t compStorage = 0; @@ -187,7 +199,7 @@ void vnodeBuildStatusMsg(void *param) { void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes) { for (int32_t i = 0; i < numOfVnodes; ++i) { pAccess[i].vgId = htonl(pAccess[i].vgId); - SVnodeObj *pVnode = vnodeAcquire(pAccess[i].vgId); + SVnodeObj *pVnode = vnodeAcquireNotClose(pAccess[i].vgId); if (pVnode != NULL) { pVnode->accessState = pAccess[i].accessState; if (pVnode->accessState != TSDB_VN_ALL_ACCCESS) { diff --git a/src/vnode/src/vnodeWrite.c b/src/vnode/src/vnodeWrite.c index 555eda6d13..a7c418711d 100644 --- a/src/vnode/src/vnodeWrite.c +++ b/src/vnode/src/vnodeWrite.c @@ -396,10 +396,13 @@ static int32_t vnodePerformFlowCtrl(SVWriteMsg *pWrite) { } void vnodeWaitWriteCompleted(SVnodeObj *pVnode) { + int32_t extraSleep = 0; while (pVnode->queuedWMsg > 0) { vTrace("vgId:%d, queued wmsg num:%d", pVnode->vgId, pVnode->queuedWMsg); taosMsleep(10); + extraSleep = 1; } - taosMsleep(900); + if (extraSleep) + taosMsleep(900); }