Merge pull request #10017 from taosdata/feature/mnode

refact worker for fetch

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);

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
 }

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 {

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;

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 {
   }
 }

source/util/src/tqueue.c

@@ -13,35 +13,36 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#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;
 

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);
 }