diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index 843798817d..9dd1a745d3 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -232,6 +232,7 @@ typedef struct { SAsyncPool* transCreateAsyncPool(uv_loop_t* loop, int sz, void* arg, AsyncCB cb); void transDestroyAsyncPool(SAsyncPool* pool); int transAsyncSend(SAsyncPool* pool, queue* mq); +bool transAsyncPoolIsEmpty(SAsyncPool* pool); #define TRANS_DESTROY_ASYNC_POOL_MSG(pool, msgType, freeFunc) \ do { \ diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 07a698f883..f94a7f3c37 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -70,6 +70,8 @@ typedef struct SCliThrd { SCvtAddr cvtAddr; + SCliMsg* stopMsg; + bool quit; } SCliThrd; @@ -761,14 +763,17 @@ void cliConnCb(uv_connect_t* req, int status) { } static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd) { + if (!transAsyncPoolIsEmpty(pThrd->asyncPool)) { + pThrd->stopMsg = pMsg; + return; + } + pThrd->stopMsg = NULL; pThrd->quit = true; tDebug("cli work thread %p start to quit", pThrd); destroyCmsg(pMsg); destroyConnPool(pThrd->pool); uv_timer_stop(&pThrd->timer); uv_walk(pThrd->loop, cliWalkCb, NULL); - - // uv_stop(pThrd->loop); } static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) { int64_t refId = (int64_t)(pMsg->msg.info.handle); @@ -925,6 +930,7 @@ static void cliAsyncCb(uv_async_t* handle) { if (count >= 2) { tTrace("cli process batch size:%d", count); } + if (pThrd->stopMsg != NULL) cliHandleQuit(pThrd->stopMsg, pThrd); } static void* cliWorkThread(void* arg) { diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index c89bbd408b..c3cba3118c 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -228,6 +228,14 @@ int transAsyncSend(SAsyncPool* pool, queue* q) { } return uv_async_send(async); } +bool transAsyncPoolIsEmpty(SAsyncPool* pool) { + for (int i = 0; i < pool->nAsync; i++) { + uv_async_t* async = &(pool->asyncs[i]); + SAsyncItem* item = async->data; + if (!QUEUE_IS_EMPTY(&item->qmsg)) return false; + } + return true; +} void transCtxInit(STransCtx* ctx) { // init transCtx