Merge pull request #10894 from taosdata/feature/shm

refact queue

include/util/tqueue.h

@@ -70,10 +70,7 @@ int32_t    taosGetQueueNumber(STaosQset *qset);
 
 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

@@ -27,33 +27,33 @@ typedef struct SWWorkerPool SWWorkerPool;
 
 typedef struct SQWorker {
   int32_t       id;      // worker ID
-  TdThread     thread;  // thread
+  TdThread      thread;  // thread
   SQWorkerPool *pool;
-} SQWorker, SFWorker;
+} SQWorker;
 
 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;
-  SQWorker       *workers;
+  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;
+  SQWorker     *workers;
   TdThreadMutex mutex;
-} SQWorkerPool, SFWorkerPool;
+} SQWorkerPool;
 
 typedef struct SWWorker {
   int32_t       id;      // worker id
-  TdThread     thread;  // thread
+  TdThread      thread;  // thread
   STaosQall    *qall;
   STaosQset    *qset;  // queue set
   SWWorkerPool *pool;
 } SWWorker;
 
 typedef struct SWWorkerPool {
-  int32_t         max;     // max number of workers
-  int32_t         nextId;  // from 0 to max-1, cyclic
-  const char     *name;
-  SWWorker       *workers;
+  int32_t       max;     // max number of workers
+  int32_t       nextId;  // from 0 to max-1, cyclic
+  const char   *name;
+  SWWorker     *workers;
   TdThreadMutex mutex;
 } SWWorkerPool;
 
@@ -62,16 +62,43 @@ void        tQWorkerCleanup(SQWorkerPool *pool);
 STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp);
 void        tQWorkerFreeQueue(SQWorkerPool *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     tWWorkerInit(SWWorkerPool *pool);
 void        tWWorkerCleanup(SWWorkerPool *pool);
 STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp);
 void        tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue);
 
+typedef struct {
+  const char *name;
+  int32_t     minNum;
+  int32_t     maxNum;
+  FItem       fp;
+  void       *param;
+} SQWorkerAllCfg;
+
+typedef struct {
+  const char  *name;
+  STaosQueue  *queue;
+  SQWorkerPool pool;
+} SQWorkerAll;
+
+typedef struct {
+  const char *name;
+  int32_t     maxNum;
+  FItems      fp;
+  void       *param;
+} SWWorkerAllCfg;
+
+typedef struct {
+  const char  *name;
+  STaosQueue  *queue;
+  SWWorkerPool pool;
+} SWWorkerAll;
+
+int32_t tQWorkerAllInit(SQWorkerAll *pWorker, const SQWorkerAllCfg *pCfg);
+void    tQWorkerAllCleanup(SQWorkerAll *pWorker);
+int32_t tWWorkerAllInit(SWWorkerAll *pWorker, const SWWorkerAllCfg *pCfg);
+void    tWWorkerAllCleanup(SWWorkerAll *pWorker);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mgmt/mnode/inc/mmInt.h

@@ -28,9 +28,9 @@ typedef struct SMnodeMgmt {
   SDnode       *pDnode;
   SMgmtWrapper *pWrapper;
   const char   *path;
-  SDnodeWorker  readWorker;
-  SDnodeWorker  writeWorker;
-  SDnodeWorker  syncWorker;
+  SQWorkerAll   readWorker;
+  SQWorkerAll   writeWorker;
+  SQWorkerAll   syncWorker;
   SReplica      replicas[TSDB_MAX_REPLICA];
   int8_t        replica;
   int8_t        selfIndex;

source/dnode/mgmt/mnode/src/mmWorker.c

@@ -42,9 +42,9 @@ static void mmProcessQueue(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
   taosFreeQitem(pMsg);
 }
 
-static int32_t mmPutMsgToWorker(SMnodeMgmt *pMgmt, SDnodeWorker *pWorker, SNodeMsg *pMsg) {
+static int32_t mmPutMsgToWorker(SMnodeMgmt *pMgmt, SQWorkerAll *pWorker, SNodeMsg *pMsg) {
   dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
-  return dndWriteMsgToWorker(pWorker, pMsg);
+  return taosWriteQitem(pWorker->queue, pMsg);
 }
 
 int32_t mmProcessWriteMsg(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
@@ -59,7 +59,7 @@ int32_t mmProcessReadMsg(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
   return mmPutMsgToWorker(pMgmt, &pMgmt->readWorker, pMsg);
 }
 
-static int32_t mmPutRpcMsgToWorker(SMnodeMgmt *pMgmt, SDnodeWorker *pWorker, SRpcMsg *pRpc) {
+static int32_t mmPutRpcMsgToWorker(SMnodeMgmt *pMgmt, SQWorkerAll *pWorker, SRpcMsg *pRpc) {
   SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
   if (pMsg == NULL) {
     return -1;
@@ -68,7 +68,7 @@ static int32_t mmPutRpcMsgToWorker(SMnodeMgmt *pMgmt, SDnodeWorker *pWorker, SRp
   dTrace("msg:%p, is created and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType));
   pMsg->rpcMsg = *pRpc;
 
-  int32_t code = dndWriteMsgToWorker(pWorker, pMsg);
+  int32_t code = taosWriteQitem(pWorker->queue, pMsg);
   if (code != 0) {
     dTrace("msg:%p, is freed", pMsg);
     taosFreeQitem(pMsg);
@@ -89,18 +89,20 @@ int32_t mmPutMsgToReadQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
 }
 
 int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
-  if (dndInitWorker(pMgmt, &pMgmt->readWorker, DND_WORKER_SINGLE, "mnode-read", 0, 1, mmProcessQueue) != 0) {
-    dError("failed to start mnode read worker since %s", terrstr());
+  SQWorkerAllCfg cfg = {.minNum = 0, .maxNum = 1, .name = "mnode-read", .fp = (FItem)mmProcessQueue, .param = pMgmt};
+
+  if (tQWorkerAllInit(&pMgmt->readWorker, &cfg) != 0) {
+    dError("failed to start mnode-read worker since %s", terrstr());
     return -1;
   }
 
-  if (dndInitWorker(pMgmt, &pMgmt->writeWorker, DND_WORKER_SINGLE, "mnode-write", 0, 1, mmProcessQueue) != 0) {
-    dError("failed to start mnode write worker since %s", terrstr());
+  if (tQWorkerAllInit(&pMgmt->writeWorker, &cfg) != 0) {
+    dError("failed to start mnode-write worker since %s", terrstr());
     return -1;
   }
 
-  if (dndInitWorker(pMgmt, &pMgmt->syncWorker, DND_WORKER_SINGLE, "mnode-sync", 0, 1, mmProcessQueue) != 0) {
-    dError("failed to start mnode sync worker since %s", terrstr());
+  if (tQWorkerAllInit(&pMgmt->syncWorker, &cfg) != 0) {
+    dError("failed to start mnode sync-worker since %s", terrstr());
     return -1;
   }
 
@@ -108,7 +110,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
 }
 
 void mmStopWorker(SMnodeMgmt *pMgmt) {
-  dndCleanupWorker(&pMgmt->readWorker);
-  dndCleanupWorker(&pMgmt->writeWorker);
-  dndCleanupWorker(&pMgmt->syncWorker);
+  tQWorkerAllCleanup(&pMgmt->readWorker);
+  tQWorkerAllCleanup(&pMgmt->writeWorker);
+  tQWorkerAllCleanup(&pMgmt->syncWorker);
 }

source/dnode/mgmt/vnode/inc/vmInt.h

@@ -30,7 +30,7 @@ typedef struct SVnodesMgmt {
   SVnodesStat   state;
   STfs         *pTfs;
   SQWorkerPool  queryPool;
-  SFWorkerPool  fetchPool;
+  SQWorkerPool  fetchPool;
   SWWorkerPool  syncPool;
   SWWorkerPool  writePool;
   const char   *path;

source/dnode/mgmt/vnode/src/vmWorker.c

@@ -262,7 +262,7 @@ int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
   pVnode->pWriteQ = tWWorkerAllocQueue(&pMgmt->writePool, pVnode, (FItems)vmProcessWriteQueue);
   pVnode->pApplyQ = tWWorkerAllocQueue(&pMgmt->writePool, pVnode, (FItems)vmProcessApplyQueue);
   pVnode->pSyncQ = tWWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FItems)vmProcessSyncQueue);
-  pVnode->pFetchQ = tFWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItem)vmProcessFetchQueue);
+  pVnode->pFetchQ = tQWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItem)vmProcessFetchQueue);
   pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue);
 
   if (pVnode->pApplyQ == NULL || pVnode->pWriteQ == NULL || pVnode->pSyncQ == NULL || pVnode->pFetchQ == NULL ||
@@ -277,7 +277,7 @@ int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
 
 void vmFreeQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
   tQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ);
-  tFWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);
+  tQWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);
   tWWorkerFreeQueue(&pMgmt->writePool, pVnode->pWriteQ);
   tWWorkerFreeQueue(&pMgmt->writePool, pVnode->pApplyQ);
   tWWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ);
@@ -303,11 +303,11 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
   pQPool->max = maxQueryThreads;
   if (tQWorkerInit(pQPool) != 0) return -1;
 
-  SFWorkerPool *pFPool = &pMgmt->fetchPool;
+  SQWorkerPool *pFPool = &pMgmt->fetchPool;
   pFPool->name = "vnode-fetch";
   pFPool->min = minFetchThreads;
   pFPool->max = maxFetchThreads;
-  if (tFWorkerInit(pFPool) != 0) return -1;
+  if (tQWorkerInit(pFPool) != 0) return -1;
 
   SWWorkerPool *pWPool = &pMgmt->writePool;
   pWPool->name = "vnode-write";
@@ -330,7 +330,7 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
 
 void vmStopWorker(SVnodesMgmt *pMgmt) {
   dndCleanupWorker(&pMgmt->mgmtWorker);
-  tFWorkerCleanup(&pMgmt->fetchPool);
+  tQWorkerCleanup(&pMgmt->fetchPool);
   tQWorkerCleanup(&pMgmt->queryPool);
   tWWorkerCleanup(&pMgmt->writePool);
   tWWorkerCleanup(&pMgmt->syncPool);

source/util/src/tqueue.c

@@ -68,7 +68,6 @@ STaosQueue *taosOpenQueue() {
     return NULL;
   }
 
-  queue->threadId = -1;
   uDebug("queue:%p is opened", queue);
   return queue;
 }
@@ -437,59 +436,11 @@ 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);
-
-  taosThreadMutexLock(&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;
-    }
-
-    taosThreadMutexLock(&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);
-    }
-
-    taosThreadMutexUnlock(&queue->mutex);
-    if (pNode) break;
-  }
-
-  taosThreadMutexUnlock(&qset->mutex);
-
-  return code;
-}
-
 void taosResetQsetThread(STaosQset *qset, void *pItem) {
   if (pItem == NULL) return;
   STaosQnode *pNode = (STaosQnode *)((char *)pItem - sizeof(STaosQnode));
 
   taosThreadMutexLock(&qset->mutex);
-  pNode->queue->threadId = -1;
   for (int32_t i = 0; i < pNode->queue->numOfItems; ++i) {
     tsem_post(&qset->sem);
   }

source/util/src/tworker.c

@@ -22,7 +22,7 @@ typedef void *(*ThreadFp)(void *param);
 
 int32_t tQWorkerInit(SQWorkerPool *pool) {
   pool->qset = taosOpenQset();
-  pool->workers = calloc(sizeof(SQWorker), pool->max);
+  pool->workers = calloc(pool->max, sizeof(SQWorker));
   if (pool->workers == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -93,7 +93,7 @@ static void *tQWorkerThreadFp(SQWorker *worker) {
   return NULL;
 }
 
-STaosQueue *tWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp, ThreadFp threadFp) {
+STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) {
   taosThreadMutexLock(&pool->mutex);
   STaosQueue *queue = taosOpenQueue();
   if (queue == NULL) {
@@ -114,7 +114,7 @@ STaosQueue *tWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp, Threa
       taosThreadAttrInit(&thAttr);
       taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
 
-      if (taosThreadCreate(&worker->thread, &thAttr, threadFp, worker) != 0) {
+      if (taosThreadCreate(&worker->thread, &thAttr, (ThreadFp)tQWorkerThreadFp, 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;
@@ -134,58 +134,11 @@ STaosQueue *tWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp, Threa
   return 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 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)tQWorkerThreadFp);
-}
-
-void tFWorkerFreeQueue(SFWorkerPool *pool, STaosQueue *queue) { tQWorkerFreeQueue(pool, queue); }
-
 int32_t tWWorkerInit(SWWorkerPool *pool) {
   pool->nextId = 0;
   pool->workers = calloc(pool->max, sizeof(SWWorker));
@@ -326,3 +279,60 @@ void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) {
   taosCloseQueue(queue);
   uDebug("worker:%s, queue:%p is freed", pool->name, queue);
 }
+
+int32_t tQWorkerAllInit(SQWorkerAll *pWorker, const SQWorkerAllCfg *pCfg) {
+  SQWorkerPool *pPool = &pWorker->pool;
+  pPool->name = pCfg->name;
+  pPool->min = pCfg->minNum;
+  pPool->max = pCfg->maxNum;
+  if (tQWorkerInit(pPool) != 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  pWorker->queue = tQWorkerAllocQueue(pPool, pCfg->param, pCfg->fp);
+  if (pWorker->queue == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  pWorker->name = pCfg->name;
+  return 0;
+}
+
+void tQWorkerAllCleanup(SQWorkerAll *pWorker) {
+  if (pWorker->queue == NULL) return;
+
+  while (!taosQueueEmpty(pWorker->queue)) {
+    taosMsleep(10);
+  }
+
+  tQWorkerCleanup(&pWorker->pool);
+  tQWorkerFreeQueue(&pWorker->pool, pWorker->queue);
+}
+
+int32_t tWWorkerAllInit(SWWorkerAll *pWorker, const SWWorkerAllCfg *pCfg) {
+  SWWorkerPool *pPool = &pWorker->pool;
+  pPool->name = pCfg->name;
+  pPool->max = pCfg->maxNum;
+  if (tWWorkerInit(pPool) != 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  pWorker->queue = tWWorkerAllocQueue(pPool, pCfg->param, pCfg->fp);
+  if (pWorker->queue == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  pWorker->name = pCfg->name;
+  return 0;
+}
+
+void tWWorkerAllCleanup(SWWorkerAll *pWorker) {
+  if (pWorker->queue == NULL) return;
+
+  while (!taosQueueEmpty(pWorker->queue)) {
+    taosMsleep(10);
+  }
+
+  tWWorkerCleanup(&pWorker->pool);
+  tWWorkerFreeQueue(&pWorker->pool, pWorker->queue);
+}