diff --git a/include/util/tqueue.h b/include/util/tqueue.h index 9f09bd2930..eb887596d0 100644 --- a/include/util/tqueue.h +++ b/include/util/tqueue.h @@ -72,40 +72,6 @@ struct STaosQnode { char item[]; }; -struct STaosQueue { - STaosQnode *head; - STaosQnode *tail; - STaosQueue *next; // for queue set - STaosQset *qset; // for queue set - void *ahandle; // for queue set - FItem itemFp; - FItems itemsFp; - TdThreadMutex mutex; - int64_t memOfItems; - int32_t numOfItems; - int64_t threadId; - int64_t memLimit; - int64_t itemLimit; -}; - -struct STaosQset { - STaosQueue *head; - STaosQueue *current; - TdThreadMutex mutex; - tsem_t sem; - int32_t numOfQueues; - int32_t numOfItems; -}; - -struct STaosQall { - STaosQnode *current; - STaosQnode *start; - int32_t numOfItems; - int64_t memOfItems; - int32_t unAccessedNumOfItems; - int64_t unAccessMemOfItems; -}; - STaosQueue *taosOpenQueue(); void taosCloseQueue(STaosQueue *queue); void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp); @@ -140,6 +106,8 @@ int32_t taosGetQueueNumber(STaosQset *qset); int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, SQueueInfo *qinfo); int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, SQueueInfo *qinfo); void taosResetQsetThread(STaosQset *qset, void *pItem); +void taosQueueSetThreadId(STaosQueue *pQueue, int64_t threadId); +int64_t taosQueueGetThreadId(STaosQueue *pQueue); #ifdef __cplusplus } diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 9b74456da2..a893b27896 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -876,12 +876,13 @@ int32_t tmqHandleAllDelayedTask(tmq_t* pTmq) { STaosQall* qall = taosAllocateQall(); taosReadAllQitems(pTmq->delayedTask, qall); - if (qall->numOfItems == 0) { + int32_t numOfItems = taosQallItemSize(qall); + if (numOfItems == 0) { taosFreeQall(qall); return TSDB_CODE_SUCCESS; } - tscDebug("consumer:0x%" PRIx64 " handle delayed %d tasks before poll data", pTmq->consumerId, qall->numOfItems); + tscDebug("consumer:0x%" PRIx64 " handle delayed %d tasks before poll data", pTmq->consumerId, numOfItems); int8_t* pTaskType = NULL; taosGetQitem(qall, (void**)&pTaskType); @@ -1839,7 +1840,7 @@ static void updateVgInfo(SMqClientVg* pVg, STqOffsetVal* reqOffset, STqOffsetVal } static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout) { - tscDebug("consumer:0x%" PRIx64 " start to handle the rsp, total:%d", tmq->consumerId, tmq->qall->numOfItems); + tscDebug("consumer:0x%" PRIx64 " start to handle the rsp, total:%d", tmq->consumerId, taosQallItemSize(tmq->qall)); while (1) { SMqRspWrapper* pRspWrapper = NULL; diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c index be88e8b3fd..3dfc2bd96f 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c @@ -194,26 +194,26 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal) while (pVnode->refCount > 0) taosMsleep(10); dInfo("vgId:%d, wait for vnode write queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue, - pVnode->pWriteW.queue->threadId); + taosQueueGetThreadId(pVnode->pWriteW.queue)); tMultiWorkerCleanup(&pVnode->pWriteW); dInfo("vgId:%d, wait for vnode sync queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue, - pVnode->pSyncW.queue->threadId); + taosQueueGetThreadId(pVnode->pSyncW.queue)); tMultiWorkerCleanup(&pVnode->pSyncW); dInfo("vgId:%d, wait for vnode sync rd queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue, - pVnode->pSyncRdW.queue->threadId); + taosQueueGetThreadId(pVnode->pSyncRdW.queue)); tMultiWorkerCleanup(&pVnode->pSyncRdW); dInfo("vgId:%d, wait for vnode apply queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue, - pVnode->pApplyW.queue->threadId); + taosQueueGetThreadId(pVnode->pApplyW.queue)); tMultiWorkerCleanup(&pVnode->pApplyW); dInfo("vgId:%d, wait for vnode query queue:%p is empty", pVnode->vgId, pVnode->pQueryQ); while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10); dInfo("vgId:%d, wait for vnode fetch queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ, - pVnode->pFetchQ->threadId); + taosQueueGetThreadId(pVnode->pFetchQ)); while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10); tqNotifyClose(pVnode->pImpl->pTq); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index 8b80527447..a6abe5ab4d 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -365,16 +365,16 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { } dInfo("vgId:%d, write-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue, - pVnode->pWriteW.queue->threadId); + taosQueueGetThreadId(pVnode->pWriteW.queue)); dInfo("vgId:%d, sync-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue, - pVnode->pSyncW.queue->threadId); + taosQueueGetThreadId(pVnode->pSyncW.queue)); dInfo("vgId:%d, sync-rd-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue, - pVnode->pSyncRdW.queue->threadId); + taosQueueGetThreadId(pVnode->pSyncRdW.queue)); dInfo("vgId:%d, apply-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue, - pVnode->pApplyW.queue->threadId); + taosQueueGetThreadId(pVnode->pApplyW.queue)); dInfo("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ); dInfo("vgId:%d, fetch-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ, - pVnode->pFetchQ->threadId); + taosQueueGetThreadId(pVnode->pFetchQ)); dInfo("vgId:%d, stream-queue:%p is alloced", pVnode->vgId, pVnode->pStreamQ); return 0; } diff --git a/source/util/src/tarray.c b/source/util/src/tarray.c index 88eb51d500..d9686d77f8 100644 --- a/source/util/src/tarray.c +++ b/source/util/src/tarray.c @@ -89,12 +89,14 @@ static int32_t taosArrayResize(SArray* pArray) { int32_t taosArrayEnsureCap(SArray* pArray, size_t newCap) { if (newCap > pArray->capacity) { float factor = BOUNDARY_BIG_FACTOR; - if(newCap * pArray->elemSize > BOUNDARY_SIZE){ + if (newCap * pArray->elemSize > BOUNDARY_SIZE) { factor = BOUNDARY_SMALL_FACTOR; } + size_t tsize = (pArray->capacity * factor); while (newCap > tsize) { - tsize = (tsize * factor); + size_t newSize = (tsize * factor); + tsize = (newSize == tsize) ? (tsize + 2) : newSize; } pArray->pData = taosMemoryRealloc(pArray->pData, tsize * pArray->elemSize); diff --git a/source/util/src/tqueue.c b/source/util/src/tqueue.c index 2cc13be6ba..7a4eb09b99 100644 --- a/source/util/src/tqueue.c +++ b/source/util/src/tqueue.c @@ -21,6 +21,40 @@ int64_t tsRpcQueueMemoryAllowed = 0; int64_t tsRpcQueueMemoryUsed = 0; +struct STaosQueue { + STaosQnode *head; + STaosQnode *tail; + STaosQueue *next; // for queue set + STaosQset *qset; // for queue set + void *ahandle; // for queue set + FItem itemFp; + FItems itemsFp; + TdThreadMutex mutex; + int64_t memOfItems; + int32_t numOfItems; + int64_t threadId; + int64_t memLimit; + int64_t itemLimit; +}; + +struct STaosQset { + STaosQueue *head; + STaosQueue *current; + TdThreadMutex mutex; + tsem_t sem; + int32_t numOfQueues; + int32_t numOfItems; +}; + +struct STaosQall { + STaosQnode *current; + STaosQnode *start; + int32_t numOfItems; + int64_t memOfItems; + int32_t unAccessedNumOfItems; + int64_t unAccessMemOfItems; +}; + void taosSetQueueMemoryCapacity(STaosQueue *queue, int64_t cap) { queue->memLimit = cap; } void taosSetQueueCapacity(STaosQueue *queue, int64_t size) { queue->itemLimit = size; } @@ -497,6 +531,12 @@ int64_t taosQallUnAccessedMemSize(STaosQall *qall) { return qall->unAccessMemOfI void taosResetQitems(STaosQall *qall) { qall->current = qall->start; } int32_t taosGetQueueNumber(STaosQset *qset) { return qset->numOfQueues; } +void taosQueueSetThreadId(STaosQueue* pQueue, int64_t threadId) { + pQueue->threadId = threadId; +} + +int64_t taosQueueGetThreadId(STaosQueue *pQueue) { return pQueue->threadId; } + #if 0 void taosResetQsetThread(STaosQset *qset, void *pItem) { diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c index 3e591c7d7f..138d4bc1f4 100644 --- a/source/util/src/tworker.c +++ b/source/util/src/tworker.c @@ -417,9 +417,9 @@ _OVER: return NULL; } else { while (worker->pid <= 0) taosMsleep(10); - queue->threadId = worker->pid; - uInfo("worker:%s, queue:%p is allocated, ahandle:%p thread:%08" PRId64, pool->name, queue, ahandle, - queue->threadId); + + taosQueueSetThreadId(queue, worker->pid); + uInfo("worker:%s, queue:%p is allocated, ahandle:%p thread:%08" PRId64, pool->name, queue, ahandle, worker->pid); return queue; } }