diff --git a/include/util/tqueue.h b/include/util/tqueue.h index 63ba460d392c7cec868c7533613be1985d8a2f11..cfa5a65c2a14543e6690fcf0b886d77696857cb0 100644 --- a/include/util/tqueue.h +++ b/include/util/tqueue.h @@ -15,6 +15,7 @@ #ifndef _TD_UTIL_QUEUE_H #define _TD_UTIL_QUEUE_H +#include "os.h" #ifdef __cplusplus extern "C" { @@ -40,13 +41,13 @@ shall be used to set up the protection. typedef struct STaosQueue STaosQueue; typedef struct STaosQset STaosQset; typedef struct STaosQall STaosQall; -typedef void (*FProcessItem)(void *ahandle, void *pItem); -typedef void (*FProcessItems)(void *ahandle, STaosQall *qall, int32_t numOfItems); +typedef void (*FItem)(void *ahandle, void *pItem); +typedef void (*FItems)(void *ahandle, STaosQall *qall, int32_t numOfItems); STaosQueue *taosOpenQueue(); void taosCloseQueue(STaosQueue *queue); -void taosSetQueueFp(STaosQueue *queue, FProcessItem itemFp, FProcessItems itemsFp); -void *taosAllocateQitem(int32_t size); +void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp); +void * taosAllocateQitem(int32_t size); void taosFreeQitem(void *pItem); int32_t taosWriteQitem(STaosQueue *queue, void *pItem); int32_t taosReadQitem(STaosQueue *queue, void **ppItem); @@ -66,8 +67,11 @@ int32_t taosAddIntoQset(STaosQset *qset, STaosQueue *queue, void *ahandle); void taosRemoveFromQset(STaosQset *qset, STaosQueue *queue); int32_t taosGetQueueNumber(STaosQset *qset); -int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FProcessItem *itemFp); -int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FProcessItems *itemsFp); +int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp); +int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FItems *itemsFp); + +int32_t taosReadQitemFromQsetByThread(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp, int32_t threadId); +void taosResetQsetThread(STaosQset *qset, void *pItem); int32_t taosGetQueueItemsNumber(STaosQueue *queue); int32_t taosGetQsetItemsNumber(STaosQset *qset); diff --git a/include/util/tworker.h b/include/util/tworker.h index 27f03bd2b69e6d6add97cdc2e5ab05ad2fdd5eca..771c7c943359b08e664f3d7de45fe6a4ecf6ada3 100644 --- a/include/util/tworker.h +++ b/include/util/tworker.h @@ -15,57 +15,61 @@ #ifndef _TD_UTIL_WORKER_H #define _TD_UTIL_WORKER_H - #include "tqueue.h" #ifdef __cplusplus extern "C" { #endif -typedef struct SWorkerPool SWorkerPool; -typedef struct SMWorkerPool SMWorkerPool; +typedef struct SQWorkerPool SQWorkerPool; +typedef struct SWWorkerPool SWWorkerPool; -typedef struct SWorker { - int32_t id; // worker ID - pthread_t thread; // thread - SWorkerPool *pool; -} SWorker; +typedef struct SQWorker { + int32_t id; // worker ID + pthread_t thread; // thread + SQWorkerPool *pool; +} SQWorker, SFWorker; -typedef struct SWorkerPool { +typedef struct SQWorkerPool { int32_t max; // max number of workers int32_t min; // min number of workers int32_t num; // current number of workers - STaosQset *qset; - const char *name; - SWorker *workers; + STaosQset * qset; + const char * name; + SQWorker * workers; pthread_mutex_t mutex; -} SWorkerPool; +} SQWorkerPool, SFWorkerPool; -typedef struct SMWorker { +typedef struct SWWorker { int32_t id; // worker id pthread_t thread; // thread - STaosQall *qall; - STaosQset *qset; // queue set - SMWorkerPool *pool; -} SMWorker; + STaosQall * qall; + STaosQset * qset; // queue set + SWWorkerPool *pool; +} SWWorker; -typedef struct SMWorkerPool { +typedef struct SWWorkerPool { int32_t max; // max number of workers int32_t nextId; // from 0 to max-1, cyclic - const char *name; - SMWorker *workers; + const char * name; + SWWorker * workers; pthread_mutex_t mutex; -} SMWorkerPool; +} SWWorkerPool; + +int32_t tQWorkerInit(SQWorkerPool *pool); +void tQWorkerCleanup(SQWorkerPool *pool); +STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp); +void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue); -int32_t tWorkerInit(SWorkerPool *pool); -void tWorkerCleanup(SWorkerPool *pool); -STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp); -void tWorkerFreeQueue(SWorkerPool *pool, STaosQueue *queue); +int32_t tFWorkerInit(SFWorkerPool *pool); +void tFWorkerCleanup(SFWorkerPool *pool); +STaosQueue *tFWorkerAllocQueue(SFWorkerPool *pool, void *ahandle, FItem fp); +void tFWorkerFreeQueue(SFWorkerPool *pool, STaosQueue *queue); -int32_t tMWorkerInit(SMWorkerPool *pool); -void tMWorkerCleanup(SMWorkerPool *pool); -STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems fp); -void tMWorkerFreeQueue(SMWorkerPool *pool, STaosQueue *queue); +int32_t tWWorkerInit(SWWorkerPool *pool); +void tWWorkerCleanup(SWWorkerPool *pool); +STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp); +void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue); #ifdef __cplusplus } diff --git a/source/dnode/mgmt/impl/inc/dndEnv.h b/source/dnode/mgmt/impl/inc/dndEnv.h index 7ab3f46fdbe36c0dbd7c85b079c45ceb68825ee0..4214dca11d930c93240fccc58f58882e1c3c47b3 100644 --- a/source/dnode/mgmt/impl/inc/dndEnv.h +++ b/source/dnode/mgmt/impl/inc/dndEnv.h @@ -31,8 +31,8 @@ typedef struct { SDnode *pDnode; STaosQueue *queue; union { - SWorkerPool pool; - SMWorkerPool mpool; + SQWorkerPool pool; + SWWorkerPool mpool; }; } SDnodeWorker; @@ -109,10 +109,10 @@ typedef struct { int32_t openVnodes; int32_t totalVnodes; SRWLatch latch; - SWorkerPool queryPool; - SWorkerPool fetchPool; - SMWorkerPool syncPool; - SMWorkerPool writePool; + SQWorkerPool queryPool; + SFWorkerPool fetchPool; + SWWorkerPool syncPool; + SWWorkerPool writePool; } SVnodesMgmt; typedef struct { diff --git a/source/dnode/mgmt/impl/src/dndVnodes.c b/source/dnode/mgmt/impl/src/dndVnodes.c index f487859fd6e0b02e5a21785f8401a0cc88b7b8c9..dcb73d13c78c1a058fdbc13b3facd64b09fa1f3a 100644 --- a/source/dnode/mgmt/impl/src/dndVnodes.c +++ b/source/dnode/mgmt/impl/src/dndVnodes.c @@ -253,7 +253,7 @@ static int32_t dndGetVnodesFromFile(SDnode *pDnode, SWrapperCfg **ppCfgs, int32_ } for (int32_t i = 0; i < vnodesNum; ++i) { - cJSON *vnode = cJSON_GetArrayItem(vnodes, i); + cJSON * vnode = cJSON_GetArrayItem(vnodes, i); SWrapperCfg *pCfg = &pCfgs[i]; cJSON *vgId = cJSON_GetObjectItem(vnode, "vgId"); @@ -382,7 +382,7 @@ static void *dnodeOpenVnodeFunc(void *param) { dndReportStartup(pDnode, "open-vnodes", stepDesc); SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId}; - SVnode *pImpl = vnodeOpen(pCfg->path, &cfg); + SVnode * pImpl = vnodeOpen(pCfg->path, &cfg); if (pImpl == NULL) { dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex); pThread->failed++; @@ -910,27 +910,27 @@ static int32_t dndInitVnodeWorkers(SDnode *pDnode) { int32_t maxWriteThreads = TMAX(pDnode->env.numOfCores, 1); int32_t maxSyncThreads = TMAX(pDnode->env.numOfCores / 2, 1); - SWorkerPool *pPool = &pMgmt->queryPool; - pPool->name = "vnode-query"; - pPool->min = minQueryThreads; - pPool->max = maxQueryThreads; - if (tWorkerInit(pPool) != 0) return -1; + SQWorkerPool *pQPool = &pMgmt->queryPool; + pQPool->name = "vnode-query"; + pQPool->min = minQueryThreads; + pQPool->max = maxQueryThreads; + if (tQWorkerInit(pQPool) != 0) return -1; - pPool = &pMgmt->fetchPool; - pPool->name = "vnode-fetch"; - pPool->min = minFetchThreads; - pPool->max = maxFetchThreads; - if (tWorkerInit(pPool) != 0) return -1; + SFWorkerPool *pFPool = &pMgmt->fetchPool; + pFPool->name = "vnode-fetch"; + pFPool->min = minFetchThreads; + pFPool->max = maxFetchThreads; + if (tFWorkerInit(pFPool) != 0) return -1; - SMWorkerPool *pMPool = &pMgmt->writePool; - pMPool->name = "vnode-write"; - pMPool->max = maxWriteThreads; - if (tMWorkerInit(pMPool) != 0) return -1; + SWWorkerPool *pWPool = &pMgmt->writePool; + pWPool->name = "vnode-write"; + pWPool->max = maxWriteThreads; + if (tWWorkerInit(pWPool) != 0) return -1; - pMPool = &pMgmt->syncPool; - pMPool->name = "vnode-sync"; - pMPool->max = maxSyncThreads; - if (tMWorkerInit(pMPool) != 0) return -1; + pWPool = &pMgmt->syncPool; + pWPool->name = "vnode-sync"; + pWPool->max = maxSyncThreads; + if (tWWorkerInit(pWPool) != 0) return -1; dDebug("vnode workers is initialized"); return 0; @@ -938,21 +938,21 @@ static int32_t dndInitVnodeWorkers(SDnode *pDnode) { static void dndCleanupVnodeWorkers(SDnode *pDnode) { SVnodesMgmt *pMgmt = &pDnode->vmgmt; - tWorkerCleanup(&pMgmt->fetchPool); - tWorkerCleanup(&pMgmt->queryPool); - tMWorkerCleanup(&pMgmt->writePool); - tMWorkerCleanup(&pMgmt->syncPool); + tFWorkerCleanup(&pMgmt->fetchPool); + tQWorkerCleanup(&pMgmt->queryPool); + tWWorkerCleanup(&pMgmt->writePool); + tWWorkerCleanup(&pMgmt->syncPool); dDebug("vnode workers is closed"); } static int32_t dndAllocVnodeQueue(SDnode *pDnode, SVnodeObj *pVnode) { SVnodesMgmt *pMgmt = &pDnode->vmgmt; - pVnode->pWriteQ = tMWorkerAllocQueue(&pMgmt->writePool, pVnode, (FProcessItems)dndProcessVnodeWriteQueue); - pVnode->pApplyQ = tMWorkerAllocQueue(&pMgmt->writePool, pVnode, (FProcessItems)dndProcessVnodeApplyQueue); - pVnode->pSyncQ = tMWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FProcessItems)dndProcessVnodeSyncQueue); - pVnode->pFetchQ = tWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FProcessItem)dndProcessVnodeFetchQueue); - pVnode->pQueryQ = tWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FProcessItem)dndProcessVnodeQueryQueue); + pVnode->pWriteQ = tWWorkerAllocQueue(&pMgmt->writePool, pVnode, (FItems)dndProcessVnodeWriteQueue); + pVnode->pApplyQ = tWWorkerAllocQueue(&pMgmt->writePool, pVnode, (FItems)dndProcessVnodeApplyQueue); + pVnode->pSyncQ = tWWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FItems)dndProcessVnodeSyncQueue); + pVnode->pFetchQ = tFWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItem)dndProcessVnodeFetchQueue); + pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)dndProcessVnodeQueryQueue); if (pVnode->pApplyQ == NULL || pVnode->pWriteQ == NULL || pVnode->pSyncQ == NULL || pVnode->pFetchQ == NULL || pVnode->pQueryQ == NULL) { @@ -965,11 +965,11 @@ static int32_t dndAllocVnodeQueue(SDnode *pDnode, SVnodeObj *pVnode) { static void dndFreeVnodeQueue(SDnode *pDnode, SVnodeObj *pVnode) { SVnodesMgmt *pMgmt = &pDnode->vmgmt; - tWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ); - tWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ); - tMWorkerFreeQueue(&pMgmt->writePool, pVnode->pWriteQ); - tMWorkerFreeQueue(&pMgmt->writePool, pVnode->pApplyQ); - tMWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ); + tQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ); + tFWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ); + tWWorkerFreeQueue(&pMgmt->writePool, pVnode->pWriteQ); + tWWorkerFreeQueue(&pMgmt->writePool, pVnode->pApplyQ); + tWWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ); pVnode->pWriteQ = NULL; pVnode->pApplyQ = NULL; pVnode->pSyncQ = NULL; diff --git a/source/dnode/mgmt/impl/src/dndWorker.c b/source/dnode/mgmt/impl/src/dndWorker.c index e0db262f89b9cc3181174b0c3341757f23b313bb..5ccf6640c04789911932a1aabd2717f7efd3762a 100644 --- a/source/dnode/mgmt/impl/src/dndWorker.c +++ b/source/dnode/mgmt/impl/src/dndWorker.c @@ -31,28 +31,28 @@ int32_t dndInitWorker(SDnode *pDnode, SDnodeWorker *pWorker, EWorkerType type, c pWorker->pDnode = pDnode; if (pWorker->type == DND_WORKER_SINGLE) { - SWorkerPool *pPool = &pWorker->pool; + SQWorkerPool *pPool = &pWorker->pool; pPool->name = name; pPool->min = minNum; pPool->max = maxNum; - if (tWorkerInit(pPool) != 0) { + if (tQWorkerInit(pPool) != 0) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } - pWorker->queue = tWorkerAllocQueue(pPool, pDnode, (FProcessItem)queueFp); + pWorker->queue = tQWorkerAllocQueue(pPool, pDnode, (FItem)queueFp); if (pWorker->queue == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } } else if (pWorker->type == DND_WORKER_MULTI) { - SMWorkerPool *pPool = &pWorker->mpool; + SWWorkerPool *pPool = &pWorker->mpool; pPool->name = name; pPool->max = maxNum; - if (tMWorkerInit(pPool) != 0) { + if (tWWorkerInit(pPool) != 0) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } - pWorker->queue = tMWorkerAllocQueue(pPool, pDnode, (FProcessItems)queueFp); + pWorker->queue = tWWorkerAllocQueue(pPool, pDnode, (FItems)queueFp); if (pWorker->queue == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -70,11 +70,11 @@ void dndCleanupWorker(SDnodeWorker *pWorker) { } if (pWorker->type == DND_WORKER_SINGLE) { - tWorkerCleanup(&pWorker->pool); - tWorkerFreeQueue(&pWorker->pool, pWorker->queue); + tQWorkerCleanup(&pWorker->pool); + tQWorkerFreeQueue(&pWorker->pool, pWorker->queue); } else if (pWorker->type == DND_WORKER_MULTI) { - tMWorkerCleanup(&pWorker->mpool); - tMWorkerFreeQueue(&pWorker->mpool, pWorker->queue); + tWWorkerCleanup(&pWorker->mpool); + tWWorkerFreeQueue(&pWorker->mpool, pWorker->queue); } else { } } diff --git a/source/util/src/tqueue.c b/source/util/src/tqueue.c index 5cb149d53c20b94a4dbbe6d4ed7f9adb9082029f..8125f550d05df47d6f1b9880ee82c82e83a307c0 100644 --- a/source/util/src/tqueue.c +++ b/source/util/src/tqueue.c @@ -13,35 +13,36 @@ * along with this program. If not, see . */ -#include "os.h" - -#include "taoserror.h" +#define _DEFAULT_SOURCE #include "tqueue.h" +#include "taoserror.h" #include "ulog.h" typedef struct STaosQnode STaosQnode; typedef struct STaosQnode { STaosQnode *next; + STaosQueue *queue; char item[]; } STaosQnode; typedef struct STaosQueue { int32_t itemSize; int32_t numOfItems; - STaosQnode *head; - STaosQnode *tail; - STaosQueue *next; // for queue set - STaosQset *qset; // for queue set - void *ahandle; // for queue set - FProcessItem itemFp; - FProcessItems itemsFp; + int32_t threadId; + STaosQnode * head; + STaosQnode * tail; + STaosQueue * next; // for queue set + STaosQset * qset; // for queue set + void * ahandle; // for queue set + FItem itemFp; + FItems itemsFp; pthread_mutex_t mutex; } STaosQueue; typedef struct STaosQset { - STaosQueue *head; - STaosQueue *current; + STaosQueue * head; + STaosQueue * current; pthread_mutex_t mutex; int32_t numOfQueues; int32_t numOfItems; @@ -56,19 +57,23 @@ typedef struct STaosQall { } STaosQall; STaosQueue *taosOpenQueue() { - STaosQueue *queue = calloc(sizeof(STaosQueue), 1); + STaosQueue *queue = calloc(1, sizeof(STaosQueue)); if (queue == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } - pthread_mutex_init(&queue->mutex, NULL); + if (pthread_mutex_init(&queue->mutex, NULL) != 0) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } - uTrace("queue:%p is opened", queue); + queue->threadId = -1; + uDebug("queue:%p is opened", queue); return queue; } -void taosSetQueueFp(STaosQueue *queue, FProcessItem itemFp, FProcessItems itemsFp) { +void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp) { if (queue == NULL) return; queue->itemFp = itemFp; queue->itemsFp = itemsFp; @@ -77,7 +82,7 @@ void taosSetQueueFp(STaosQueue *queue, FProcessItem itemFp, FProcessItems itemsF void taosCloseQueue(STaosQueue *queue) { if (queue == NULL) return; STaosQnode *pTemp; - STaosQset *qset; + STaosQset * qset; pthread_mutex_lock(&queue->mutex); STaosQnode *pNode = queue->head; @@ -85,7 +90,9 @@ void taosCloseQueue(STaosQueue *queue) { qset = queue->qset; pthread_mutex_unlock(&queue->mutex); - if (queue->qset) taosRemoveFromQset(qset, queue); + if (queue->qset) { + taosRemoveFromQset(qset, queue); + } while (pNode) { pTemp = pNode; @@ -96,7 +103,7 @@ void taosCloseQueue(STaosQueue *queue) { pthread_mutex_destroy(&queue->mutex); free(queue); - uTrace("queue:%p is closed", queue); + uDebug("queue:%p is closed", queue); } bool taosQueueEmpty(STaosQueue *queue) { @@ -120,19 +127,23 @@ int32_t taosQueueSize(STaosQueue *queue) { } void *taosAllocateQitem(int32_t size) { - STaosQnode *pNode = (STaosQnode *)calloc(sizeof(STaosQnode) + size, 1); + STaosQnode *pNode = calloc(1, sizeof(STaosQnode) + size); + + if (pNode == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } - if (pNode == NULL) return NULL; uTrace("item:%p, node:%p is allocated", pNode->item, pNode); return (void *)pNode->item; } -void taosFreeQitem(void *param) { - if (param == NULL) return; +void taosFreeQitem(void *pItem) { + if (pItem == NULL) return; - char *temp = (char *)param; + char *temp = pItem; temp -= sizeof(STaosQnode); - uTrace("item:%p, node:%p is freed", param, temp); + uTrace("item:%p, node:%p is freed", pItem, temp); free(temp); } @@ -175,7 +186,7 @@ int32_t taosReadQitem(STaosQueue *queue, void **ppItem) { queue->numOfItems--; if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, 1); code = 1; - uDebug("item:%p is read out from queue:%p, items:%d", *ppItem, queue, queue->numOfItems); + uTrace("item:%p is read out from queue:%p, items:%d", *ppItem, queue, queue->numOfItems); } pthread_mutex_unlock(&queue->mutex); @@ -183,7 +194,7 @@ int32_t taosReadQitem(STaosQueue *queue, void **ppItem) { return code; } -STaosQall *taosAllocateQall() { return calloc(sizeof(STaosQall), 1); } +STaosQall *taosAllocateQall() { return calloc(1, sizeof(STaosQall)); } void taosFreeQall(STaosQall *qall) { free(qall); } @@ -238,7 +249,7 @@ int32_t taosGetQitem(STaosQall *qall, void **ppItem) { void taosResetQitems(STaosQall *qall) { qall->current = qall->start; } STaosQset *taosOpenQset() { - STaosQset *qset = (STaosQset *)calloc(sizeof(STaosQset), 1); + STaosQset *qset = calloc(sizeof(STaosQset), 1); if (qset == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; @@ -247,7 +258,7 @@ STaosQset *taosOpenQset() { pthread_mutex_init(&qset->mutex, NULL); tsem_init(&qset->sem, 0, 0); - uTrace("qset:%p is opened", qset); + uDebug("qset:%p is opened", qset); return qset; } @@ -268,7 +279,7 @@ void taosCloseQset(STaosQset *qset) { pthread_mutex_destroy(&qset->mutex); tsem_destroy(&qset->sem); free(qset); - uTrace("qset:%p is closed", qset); + uDebug("qset:%p is closed", qset); } // tsem_post 'qset->sem', so that reader threads waiting for it @@ -338,12 +349,12 @@ void taosRemoveFromQset(STaosQset *qset, STaosQueue *queue) { pthread_mutex_unlock(&qset->mutex); - uTrace("queue:%p is removed from qset:%p", queue, qset); + uDebug("queue:%p is removed from qset:%p", queue, qset); } int32_t taosGetQueueNumber(STaosQset *qset) { return qset->numOfQueues; } -int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FProcessItem *itemFp) { +int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp) { STaosQnode *pNode = NULL; int32_t code = 0; @@ -365,6 +376,7 @@ int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FP *ppItem = pNode->item; if (ahandle) *ahandle = queue->ahandle; if (itemFp) *itemFp = queue->itemFp; + queue->head = pNode->next; if (queue->head == NULL) queue->tail = NULL; queue->numOfItems--; @@ -382,7 +394,7 @@ int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FP return code; } -int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FProcessItems *itemsFp) { +int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FItems *itemsFp) { STaosQueue *queue; int32_t code = 0; @@ -411,7 +423,9 @@ int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahand queue->tail = NULL; queue->numOfItems = 0; atomic_sub_fetch_32(&qset->numOfItems, qall->numOfItems); - for (int32_t j = 1; j < qall->numOfItems; ++j) tsem_wait(&qset->sem); + for (int32_t j = 1; j < qall->numOfItems; ++j) { + tsem_wait(&qset->sem); + } } pthread_mutex_unlock(&queue->mutex); @@ -423,6 +437,65 @@ int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahand return code; } +int32_t taosReadQitemFromQsetByThread(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp, int32_t threadId) { + STaosQnode *pNode = NULL; + int32_t code = -1; + + tsem_wait(&qset->sem); + + pthread_mutex_lock(&qset->mutex); + + for (int32_t i = 0; i < qset->numOfQueues; ++i) { + if (qset->current == NULL) qset->current = qset->head; + STaosQueue *queue = qset->current; + if (queue) qset->current = queue->next; + if (queue == NULL) break; + if (queue->head == NULL) continue; + if (queue->threadId != -1 && queue->threadId != threadId) { + code = 0; + continue; + } + + pthread_mutex_lock(&queue->mutex); + + if (queue->head) { + pNode = queue->head; + pNode->queue = queue; + queue->threadId = threadId; + *ppItem = pNode->item; + + if (ahandle) *ahandle = queue->ahandle; + if (itemFp) *itemFp = queue->itemFp; + + queue->head = pNode->next; + if (queue->head == NULL) queue->tail = NULL; + queue->numOfItems--; + atomic_sub_fetch_32(&qset->numOfItems, 1); + code = 1; + uTrace("item:%p is read out from queue:%p, items:%d", *ppItem, queue, queue->numOfItems); + } + + pthread_mutex_unlock(&queue->mutex); + if (pNode) break; + } + + pthread_mutex_unlock(&qset->mutex); + + return code; +} + +void taosResetQsetThread(STaosQset *qset, void *pItem) { + if (pItem == NULL) return; + STaosQnode *pNode = (STaosQnode *)((char *)pItem - sizeof(STaosQnode)); + + pthread_mutex_lock(&qset->mutex); + pNode->queue->threadId = -1; + for (int32_t i = 0; i < pNode->queue->numOfItems; ++i) { + tsem_post(&qset->sem); + } + pthread_mutex_unlock(&qset->mutex); +} + int32_t taosGetQueueItemsNumber(STaosQueue *queue) { if (!queue) return 0; diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c index ed74041712ce0022eb66ac5c9331b88fda961222..97440f8dae3ccdbcaa9dc75a8e2f5c3b2da02cf7 100644 --- a/source/util/src/tworker.c +++ b/source/util/src/tworker.c @@ -14,38 +14,46 @@ */ #define _DEFAULT_SOURCE -#include "os.h" -#include "ulog.h" -#include "tqueue.h" #include "tworker.h" +#include "taoserror.h" +#include "ulog.h" -typedef void* (*ThreadFp)(void *param); +typedef void *(*ThreadFp)(void *param); -int32_t tWorkerInit(SWorkerPool *pool) { +int32_t tQWorkerInit(SQWorkerPool *pool) { pool->qset = taosOpenQset(); - pool->workers = calloc(sizeof(SWorker), pool->max); - pthread_mutex_init(&pool->mutex, NULL); - for (int i = 0; i < pool->max; ++i) { - SWorker *worker = pool->workers + i; + pool->workers = calloc(sizeof(SQWorker), pool->max); + if (pool->workers == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + if (pthread_mutex_init(&pool->mutex, NULL)) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + for (int32_t i = 0; i < pool->max; ++i) { + SQWorker *worker = pool->workers + i; worker->id = i; worker->pool = pool; } - uInfo("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max); + uDebug("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max); return 0; } -void tWorkerCleanup(SWorkerPool *pool) { - for (int i = 0; i < pool->max; ++i) { - SWorker *worker = pool->workers + i; +void tQWorkerCleanup(SQWorkerPool *pool) { + for (int32_t i = 0; i < pool->max; ++i) { + SQWorker *worker = pool->workers + i; if (worker == NULL) continue; if (taosCheckPthreadValid(worker->thread)) { taosQsetThreadResume(pool->qset); } } - for (int i = 0; i < pool->max; ++i) { - SWorker *worker = pool->workers + i; + for (int32_t i = 0; i < pool->max; ++i) { + SQWorker *worker = pool->workers + i; if (worker == NULL) continue; if (taosCheckPthreadValid(worker->thread)) { pthread_join(worker->thread, NULL); @@ -56,15 +64,15 @@ void tWorkerCleanup(SWorkerPool *pool) { taosCloseQset(pool->qset); pthread_mutex_destroy(&pool->mutex); - uInfo("worker:%s is closed", pool->name); + uDebug("worker:%s is closed", pool->name); } -static void *tWorkerThreadFp(SWorker *worker) { - SWorkerPool *pool = worker->pool; - FProcessItem fp = NULL; +static void *tQWorkerThreadFp(SQWorker *worker) { + SQWorkerPool *pool = worker->pool; + FItem fp = NULL; - void *msg = NULL; - void *ahandle = NULL; + void * msg = NULL; + void * ahandle = NULL; int32_t code = 0; taosBlockSIGPIPE(); @@ -77,7 +85,7 @@ static void *tWorkerThreadFp(SWorker *worker) { break; } - if (fp) { + if (fp != NULL) { (*fp)(ahandle, msg); } } @@ -85,11 +93,12 @@ static void *tWorkerThreadFp(SWorker *worker) { return NULL; } -STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp) { +STaosQueue *tWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp, ThreadFp threadFp) { pthread_mutex_lock(&pool->mutex); STaosQueue *queue = taosOpenQueue(); if (queue == NULL) { pthread_mutex_unlock(&pool->mutex); + terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } @@ -99,14 +108,18 @@ STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp) // spawn a thread to process queue if (pool->num < pool->max) { do { - SWorker *worker = pool->workers + pool->num; + SQWorker *worker = pool->workers + pool->num; pthread_attr_t thAttr; pthread_attr_init(&thAttr); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); - if (pthread_create(&worker->thread, &thAttr, (ThreadFp)tWorkerThreadFp, worker) != 0) { + if (pthread_create(&worker->thread, &thAttr, threadFp, worker) != 0) { uError("worker:%s:%d failed to create thread to process since %s", pool->name, worker->id, strerror(errno)); + taosCloseQueue(queue); + terrno = TSDB_CODE_OUT_OF_MEMORY; + queue = NULL; + break; } pthread_attr_destroy(&thAttr); @@ -121,19 +134,73 @@ STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp) return queue; } -void tWorkerFreeQueue(SWorkerPool *pool, STaosQueue *queue) { +STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) { + return tWorkerAllocQueue(pool, ahandle, fp, (ThreadFp)tQWorkerThreadFp); +} + +void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) { taosCloseQueue(queue); uDebug("worker:%s, queue:%p is freed", pool->name, queue); } -int32_t tMWorkerInit(SMWorkerPool *pool) { +int32_t tFWorkerInit(SFWorkerPool *pool) { return tQWorkerInit((SQWorkerPool *)pool); } + +void tFWorkerCleanup(SFWorkerPool *pool) { tQWorkerCleanup(pool); } + +static void *tFWorkerThreadFp(SQWorker *worker) { + SQWorkerPool *pool = worker->pool; + + FItem fp = NULL; + void * msg = NULL; + void * ahandle = NULL; + int32_t code = 0; + + taosBlockSIGPIPE(); + setThreadName(pool->name); + uDebug("worker:%s:%d is running", pool->name, worker->id); + + while (1) { + code = taosReadQitemFromQsetByThread(pool->qset, (void **)&msg, &ahandle, &fp, worker->id); + + if (code < 0) { + uDebug("worker:%s:%d qset:%p, got no message and exiting", pool->name, worker->id, pool->qset); + break; + } else if (code == 0) { + // uTrace("worker:%s:%d qset:%p, got no message and continue", pool->name, worker->id, pool->qset); + continue; + } + + if (fp != NULL) { + (*fp)(ahandle, msg); + } + + taosResetQsetThread(pool->qset, msg); + } + + return NULL; +} + +STaosQueue *tFWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) { + return tWorkerAllocQueue(pool, ahandle, fp, (ThreadFp)tFWorkerThreadFp); +} + +void tFWorkerFreeQueue(SFWorkerPool *pool, STaosQueue *queue) { tQWorkerFreeQueue(pool, queue); } + +int32_t tWWorkerInit(SWWorkerPool *pool) { pool->nextId = 0; - pool->workers = calloc(sizeof(SMWorker), pool->max); - if (pool->workers == NULL) return -1; + pool->workers = calloc(sizeof(SWWorker), pool->max); + if (pool->workers == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + if (pthread_mutex_init(&pool->mutex, NULL) != 0) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } - pthread_mutex_init(&pool->mutex, NULL); for (int32_t i = 0; i < pool->max; ++i) { - SMWorker *worker = pool->workers + i; + SWWorker *worker = pool->workers + i; worker->id = i; worker->qall = NULL; worker->qset = NULL; @@ -144,16 +211,18 @@ int32_t tMWorkerInit(SMWorkerPool *pool) { return 0; } -void tMWorkerCleanup(SMWorkerPool *pool) { +void tWWorkerCleanup(SWWorkerPool *pool) { for (int32_t i = 0; i < pool->max; ++i) { - SMWorker *worker = pool->workers + i; + SWWorker *worker = pool->workers + i; if (taosCheckPthreadValid(worker->thread)) { - if (worker->qset) taosQsetThreadResume(worker->qset); + if (worker->qset) { + taosQsetThreadResume(worker->qset); + } } } for (int32_t i = 0; i < pool->max; ++i) { - SMWorker *worker = pool->workers + i; + SWWorker *worker = pool->workers + i; if (taosCheckPthreadValid(worker->thread)) { pthread_join(worker->thread, NULL); taosFreeQall(worker->qall); @@ -167,12 +236,12 @@ void tMWorkerCleanup(SMWorkerPool *pool) { uInfo("worker:%s is closed", pool->name); } -static void *tWriteWorkerThreadFp(SMWorker *worker) { - SMWorkerPool *pool = worker->pool; - FProcessItems fp = NULL; +static void *tWWorkerThreadFp(SWWorker *worker) { + SWWorkerPool *pool = worker->pool; + FItems fp = NULL; - void *msg = NULL; - void *ahandle = NULL; + void * msg = NULL; + void * ahandle = NULL; int32_t numOfMsgs = 0; int32_t qtype = 0; @@ -187,7 +256,7 @@ static void *tWriteWorkerThreadFp(SMWorker *worker) { break; } - if (fp) { + if (fp != NULL) { (*fp)(ahandle, worker->qall, numOfMsgs); } } @@ -195,13 +264,14 @@ static void *tWriteWorkerThreadFp(SMWorker *worker) { return NULL; } -STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems fp) { +STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp) { pthread_mutex_lock(&pool->mutex); - SMWorker *worker = pool->workers + pool->nextId; + SWWorker *worker = pool->workers + pool->nextId; STaosQueue *queue = taosOpenQueue(); if (queue == NULL) { pthread_mutex_unlock(&pool->mutex); + terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } @@ -221,17 +291,19 @@ STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems taosCloseQset(worker->qset); taosCloseQueue(queue); pthread_mutex_unlock(&pool->mutex); + terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } pthread_attr_t thAttr; pthread_attr_init(&thAttr); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); - if (pthread_create(&worker->thread, &thAttr, (ThreadFp)tWriteWorkerThreadFp, worker) != 0) { + if (pthread_create(&worker->thread, &thAttr, (ThreadFp)tWWorkerThreadFp, worker) != 0) { uError("worker:%s:%d failed to create thread to process since %s", pool->name, worker->id, strerror(errno)); taosFreeQall(worker->qall); taosCloseQset(worker->qset); taosCloseQueue(queue); + terrno = TSDB_CODE_OUT_OF_MEMORY; queue = NULL; } else { uDebug("worker:%s:%d is launched, max:%d", pool->name, worker->id, pool->max); @@ -250,7 +322,7 @@ STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems return queue; } -void tMWorkerFreeQueue(SMWorkerPool *pool, STaosQueue *queue) { +void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) { taosCloseQueue(queue); uDebug("worker:%s, queue:%p is freed", pool->name, queue); }