taosdata / TDengine / #4897

coverage: 65.717% (-0.2%) from 65.929%
#4897

push

travis-ci

web-flow 
fix: [6622889291] Fix invalid rowSize. (#34043)

186011 of 283047 relevant lines covered (65.72%)

113853896.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.77
/source/util/src/tworker.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 









16



#define _DEFAULT_SOURCE










17



#include "tworker.h"










18



#include "taoserror.h"










19



#include "tcompare.h"










20



#include "tgeosctx.h"










21



#include "tlog.h"










22



#include "ttrace.h"










23



#include "tcurl.h"










24












25



#define QUEUE_THRESHOLD (1000 * 1000)










26












27



typedef void *(*ThreadFp)(void *param);










28












29



int32_t tQWorkerInit(SQWorkerPool *pool) {




5,792,989✔





30



  int32_t code = taosOpenQset(&pool->qset);




5,792,989✔





31



  if (code) return code;




5,792,989✔





32












33



  pool->workers = taosMemoryCalloc(pool->max, sizeof(SQueueWorker));




5,792,989✔





34



  if (pool->workers == NULL) {




5,792,989✔





35



    taosCloseQset(pool->qset);




×







36



    return terrno;




×







37



  }










38












39



  (void)taosThreadMutexInit(&pool->mutex, NULL);




5,792,989✔





40












41



  for (int32_t i = 0; i < pool->max; ++i) {




41,028,197✔





42



    SQueueWorker *worker = pool->workers + i;




35,235,208✔





43



    worker->id = i;




35,235,208✔





44



    worker->pool = pool;




35,235,208✔





45



  }










46












47



  uInfo("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max);




5,792,989✔





48



  return 0;




5,792,989✔





49



}










50












51



void tQWorkerCleanup(SQWorkerPool *pool) {




5,792,989✔





52



  for (int32_t i = 0; i < pool->max; ++i) {




41,028,197✔





53



    SQueueWorker *worker = pool->workers + i;




35,235,208✔





54



    if (taosCheckPthreadValid(worker->thread)) {




35,235,208✔





55



      taosQsetThreadResume(pool->qset);




35,235,208✔





56



    }










57



  }










58












59



  for (int32_t i = 0; i < pool->max; ++i) {




41,028,197✔





60



    SQueueWorker *worker = pool->workers + i;




35,235,208✔





61



    if (taosCheckPthreadValid(worker->thread)) {




35,235,208✔





62



      uInfo("worker:%s:%d is stopping", pool->name, worker->id);




35,235,208✔





63



      (void)taosThreadJoin(worker->thread, NULL);




35,235,208✔





64



      taosThreadClear(&worker->thread);




35,235,208✔





65



      uInfo("worker:%s:%d is stopped", pool->name, worker->id);




35,235,208✔





66



    }










67



  }










68












69



  taosMemoryFreeClear(pool->workers);




5,792,989✔





70



  taosCloseQset(pool->qset);




5,792,989✔





71



  (void)taosThreadMutexDestroy(&pool->mutex);




5,792,989✔





72












73



  uInfo("worker:%s is closed", pool->name);




5,792,989✔





74



}




5,792,989✔





75












76



static void *tQWorkerThreadFp(SQueueWorker *worker) {




35,235,208✔





77



  SQWorkerPool *pool = worker->pool;




35,235,208✔





78



  SQueueInfo    qinfo = {0};




35,235,208✔





79



  void         *msg = NULL;




35,234,886✔





80



  int32_t       code = 0;




35,234,886✔





81












82



  int32_t ret = taosBlockSIGPIPE();




35,234,886✔





83



  if (ret < 0) {




35,233,364✔





84



    uError("worker:%s:%d failed to block SIGPIPE", pool->name, worker->id);




×







85



  }










86












87



  setThreadName(pool->name);




35,233,364✔





88



  worker->pid = taosGetSelfPthreadId();




35,235,208✔





89



  uInfo("worker:%s:%d is running, thread:%08" PRId64, pool->name, worker->id, worker->pid);




35,233,030✔





90












91



  while (1) {










92



    if (taosReadQitemFromQset(pool->qset, (void **)&msg, &qinfo) == 0) {




437,054,950✔





93



      uInfo("worker:%s:%d qset:%p, got no message and exiting, thread:%08" PRId64, pool->name, worker->id, pool->qset,




35,223,905✔





94



            worker->pid);










95



      break;




35,231,822✔





96



    }










97












98



    if (qinfo.timestamp != 0) {




401,821,122✔





99



      int64_t cost = taosGetTimestampUs() - qinfo.timestamp;




401,823,816✔





100



      if (cost > QUEUE_THRESHOLD) {




401,823,816✔





101



        uWarn("worker:%s,message has been queued for too long, cost: %" PRId64 "s", pool->name, cost / QUEUE_THRESHOLD);




789,908✔





102



      }










103



    }










104












105



    if (qinfo.fp != NULL) {




401,822,170✔





106



      qinfo.workerId = worker->id;




401,819,868✔





107



      qinfo.threadNum = pool->num;




401,819,854✔





108



      (*((FItem)qinfo.fp))(&qinfo, msg);




401,824,126✔





109



    }










110












111



    taosUpdateItemSize(qinfo.queue, 1);




401,820,774✔





112



  }










113












114



  DestoryThreadLocalRegComp();




35,231,822✔





115












116



  return NULL;




35,232,637✔





117



}










118












119



STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) {




5,792,989✔





120



  int32_t     code;










121



  STaosQueue *queue;




5,790,593✔





122












123



  code = taosOpenQueue(&queue);




5,792,989✔





124



  if (code) {




5,792,989✔





125



    terrno = code;




×







126



    return NULL;




×







127



  }










128












129



  (void)taosThreadMutexLock(&pool->mutex);




5,792,989✔





130



  taosSetQueueFp(queue, fp, NULL);




5,792,989✔





131



  code = taosAddIntoQset(pool->qset, queue, ahandle);




5,792,989✔





132



  if (code) {




5,792,989✔





133



    taosCloseQueue(queue);




×







134



    (void)taosThreadMutexUnlock(&pool->mutex);




×







135



    terrno = code;




×







136



    return NULL;




×







137



  }










138












139



  // spawn a thread to process queue










140



  if (pool->num < pool->max) {




5,792,989✔





141



    do {










142



      SQueueWorker *worker = pool->workers + pool->num;




35,235,208✔





143












144



      TdThreadAttr thAttr;




35,218,472✔





145



      (void)taosThreadAttrInit(&thAttr);




35,235,208✔





146



      (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);




35,235,208✔





147












148



      if (taosThreadCreate(&worker->thread, &thAttr, (ThreadFp)tQWorkerThreadFp, worker) != 0) {




35,235,208✔





149



        taosCloseQueue(queue);




×







150



        terrno = TSDB_CODE_OUT_OF_MEMORY;




×







151



        queue = NULL;




×







152



        break;




×







153



      }










154












155



      (void)taosThreadAttrDestroy(&thAttr);




35,235,208✔





156



      pool->num++;




35,235,208✔





157



      uInfo("worker:%s:%d is launched, total:%d", pool->name, worker->id, pool->num);




35,235,208✔





158



    } while (pool->num < pool->min);




35,235,208✔





159



  }










160












161



  (void)taosThreadMutexUnlock(&pool->mutex);




5,792,989✔





162



  uInfo("worker:%s, queue:%p is allocated, ahandle:%p", pool->name, queue, ahandle);




5,792,989✔





163












164



  return queue;




5,792,989✔





165



}










166












167



void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) {




5,792,989✔





168



  uInfo("worker:%s, queue:%p is freed", pool->name, queue);




5,792,989✔





169



  taosCloseQueue(queue);




5,792,989✔





170



}




5,792,989✔





171












172



int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool) {




×







173



  int32_t code;










174












175



  code = taosOpenQset(&pool->qset);




×







176



  if (code) {




×







177



    return terrno = code;




×







178



  }










179












180



  pool->workers = taosArrayInit(2, sizeof(SQueueWorker *));




×







181



  if (pool->workers == NULL) {




×







182



    taosCloseQset(pool->qset);




×







183



    return terrno;




×







184



  }










185












186



  (void)taosThreadMutexInit(&pool->mutex, NULL);




×







187












188



  uInfo("worker:%s is initialized as auto", pool->name);




×







189



  return 0;




×







190



}










191












192



void tAutoQWorkerCleanup(SAutoQWorkerPool *pool) {




×







193



  int32_t size = taosArrayGetSize(pool->workers);




×







194



  for (int32_t i = 0; i < size; ++i) {




×







195



    SQueueWorker *worker = taosArrayGetP(pool->workers, i);




×







196



    if (taosCheckPthreadValid(worker->thread)) {




×







197



      taosQsetThreadResume(pool->qset);




×







198



    }










199



  }










200












201



  for (int32_t i = 0; i < size; ++i) {




×







202



    SQueueWorker *worker = taosArrayGetP(pool->workers, i);




×







203



    if (taosCheckPthreadValid(worker->thread)) {




×







204



      uInfo("worker:%s:%d is stopping", pool->name, worker->id);




×







205



      (void)taosThreadJoin(worker->thread, NULL);




×







206



      taosThreadClear(&worker->thread);




×







207



      uInfo("worker:%s:%d is stopped", pool->name, worker->id);




×







208



    }










209



    taosMemoryFree(worker);




×







210



  }










211












212



  taosArrayDestroy(pool->workers);




×







213



  taosCloseQset(pool->qset);




×







214



  (void)taosThreadMutexDestroy(&pool->mutex);




×







215












216



  uInfo("worker:%s is closed", pool->name);




×







217



}




×







218












219



static void *tAutoQWorkerThreadFp(SQueueWorker *worker) {




×







220



  SAutoQWorkerPool *pool = worker->pool;




×







221



  SQueueInfo        qinfo = {0};




×







222



  void             *msg = NULL;




×







223



  int32_t           code = 0;




×







224












225



  int32_t ret = taosBlockSIGPIPE();




×







226



  if (ret < 0) {




×







227



    uError("worker:%s:%d failed to block SIGPIPE", pool->name, worker->id);




×







228



  }










229












230



  setThreadName(pool->name);




×







231



  worker->pid = taosGetSelfPthreadId();




×







232



  uInfo("worker:%s:%d is running, thread:%08" PRId64, pool->name, worker->id, worker->pid);




×







233












234



  while (1) {










235



    if (taosReadQitemFromQset(pool->qset, (void **)&msg, &qinfo) == 0) {




×







236



      uInfo("worker:%s:%d qset:%p, got no message and exiting, thread:%08" PRId64, pool->name, worker->id, pool->qset,




×







237



            worker->pid);










238



      break;




×







239



    }










240












241



    if (qinfo.timestamp != 0) {




×







242



      int64_t cost = taosGetTimestampUs() - qinfo.timestamp;




×







243



      if (cost > QUEUE_THRESHOLD) {




×







244



        uWarn("worker:%s,message has been queued for too long, cost: %" PRId64 "s", pool->name, cost / QUEUE_THRESHOLD);




×







245



      }










246



    }










247












248



    if (qinfo.fp != NULL) {




×







249



      qinfo.workerId = worker->id;




×







250



      qinfo.threadNum = taosArrayGetSize(pool->workers);




×







251



      (*((FItem)qinfo.fp))(&qinfo, msg);




×







252



    }










253












254



    taosUpdateItemSize(qinfo.queue, 1);




×







255



  }










256



  DestoryThreadLocalRegComp();




×







257



  closeThreadNotificationConn();




×







258












259



  return NULL;




×







260



}










261












262



STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp, int32_t minNum) {




×







263



  int32_t     code;










264



  STaosQueue *queue;




×







265












266



  code = taosOpenQueue(&queue);




×







267



  if (code) {




×







268



    terrno = code;




×







269



    return NULL;




×







270



  }










271












272



  (void)taosThreadMutexLock(&pool->mutex);




×







273



  taosSetQueueFp(queue, fp, NULL);




×







274












275



  code = taosAddIntoQset(pool->qset, queue, ahandle);




×







276



  if (code) {




×







277



    taosCloseQueue(queue);




×







278



    (void)taosThreadMutexUnlock(&pool->mutex);




×







279



    terrno = code;




×







280



    return NULL;




×







281



  }










282












283



  int32_t queueNum = taosGetQueueNumber(pool->qset);




×







284



  int32_t curWorkerNum = taosArrayGetSize(pool->workers);




×







285



  int32_t dstWorkerNum = ceilf(queueNum * pool->ratio);




×







286












287



  if (dstWorkerNum < minNum) {




×







288



    dstWorkerNum = minNum;




×







289



  }










290












291



  // spawn a thread to process queue










292



  while (curWorkerNum < dstWorkerNum) {




×







293



    SQueueWorker *worker = taosMemoryCalloc(1, sizeof(SQueueWorker));




×







294



    if (worker == NULL || taosArrayPush(pool->workers, &worker) == NULL) {




×







295



      uError("worker:%s:%d failed to create", pool->name, curWorkerNum);




×







296



      taosMemoryFree(worker);




×







297



      taosCloseQueue(queue);




×







298



      (void)taosThreadMutexUnlock(&pool->mutex);




×







299



      terrno = TSDB_CODE_OUT_OF_MEMORY;




×







300



      return NULL;




×







301



    }










302



    worker->id = curWorkerNum;




×







303



    worker->pool = pool;




×







304












305



    TdThreadAttr thAttr;




×







306



    (void)taosThreadAttrInit(&thAttr);




×







307



    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);




×







308












309



    if (taosThreadCreate(&worker->thread, &thAttr, (ThreadFp)tAutoQWorkerThreadFp, worker) != 0) {




×







310



      uError("worker:%s:%d failed to create thread, total:%d", pool->name, worker->id, curWorkerNum);




×







311



      void *tmp = taosArrayPop(pool->workers);




×







312



      taosMemoryFree(worker);




×







313



      taosCloseQueue(queue);




×







314



      terrno = TSDB_CODE_OUT_OF_MEMORY;




×







315



      return NULL;




×







316



    }










317












318



    (void)taosThreadAttrDestroy(&thAttr);




×







319



    int32_t numOfThreads = taosArrayGetSize(pool->workers);




×







320



    uInfo("worker:%s:%d is launched, total:%d, expect:%d", pool->name, worker->id, numOfThreads, dstWorkerNum);




×







321












322



    curWorkerNum++;




×







323



  }










324












325



  (void)taosThreadMutexUnlock(&pool->mutex);




×







326



  uInfo("worker:%s, queue:%p is allocated, ahandle:%p", pool->name, queue, ahandle);




×







327












328



  return queue;




×







329



}










330












331



void tAutoQWorkerFreeQueue(SAutoQWorkerPool *pool, STaosQueue *queue) {




×







332



  uInfo("worker:%s, queue:%p is freed", pool->name, queue);




×







333



  taosCloseQueue(queue);




×







334



}




×







335












336



int32_t tWWorkerInit(SWWorkerPool *pool) {




18,131,815✔





337



  pool->nextId = 0;




18,131,815✔





338



  pool->workers = taosMemoryCalloc(pool->max, sizeof(SWWorker));




18,132,178✔





339



  if (pool->workers == NULL) {




18,132,565✔





340



    return terrno;




×







341



  }










342












343



  (void)taosThreadMutexInit(&pool->mutex, NULL);




18,131,769✔





344












345



  for (int32_t i = 0; i < pool->max; ++i) {




43,214,164✔





346



    SWWorker *worker = pool->workers + i;




25,081,883✔





347



    worker->id = i;




25,082,438✔





348



    worker->qall = NULL;




25,082,535✔





349



    worker->qset = NULL;




25,082,083✔





350



    worker->pool = pool;




25,081,954✔





351



  }










352












353



  uInfo("worker:%s is initialized, max:%d", pool->name, pool->max);




18,132,391✔





354



  return 0;




18,133,443✔





355



}










356












357



void tWWorkerCleanup(SWWorkerPool *pool) {




18,133,033✔





358



  for (int32_t i = 0; i < pool->max; ++i) {




43,215,697✔





359



    SWWorker *worker = pool->workers + i;




25,083,028✔





360



    if (taosCheckPthreadValid(worker->thread)) {




25,082,664✔





361



      if (worker->qset) {




20,203,705✔





362



        taosQsetThreadResume(worker->qset);




20,203,705✔





363



      }










364



    }










365



  }










366












367



  for (int32_t i = 0; i < pool->max; ++i) {




43,212,823✔





368



    SWWorker *worker = pool->workers + i;




25,082,664✔





369



    if (taosCheckPthreadValid(worker->thread)) {




25,082,664✔





370



      uInfo("worker:%s:%d is stopping", pool->name, worker->id);




20,203,705✔





371



      (void)taosThreadJoin(worker->thread, NULL);




20,203,705✔





372



      taosThreadClear(&worker->thread);




20,203,705✔





373



      taosFreeQall(worker->qall);




20,203,705✔





374



      taosCloseQset(worker->qset);




20,203,705✔





375



      uInfo("worker:%s:%d is stopped", pool->name, worker->id);




20,203,705✔





376



    }










377



  }










378












379



  taosMemoryFreeClear(pool->workers);




18,133,443✔





380



  (void)taosThreadMutexDestroy(&pool->mutex);




18,133,443✔





381












382



  uInfo("worker:%s is closed", pool->name);




18,133,443✔





383



}




18,132,228✔





384












385



static void *tWWorkerThreadFp(SWWorker *worker) {




20,203,705✔





386



  SWWorkerPool *pool = worker->pool;




20,203,705✔





387



  SQueueInfo    qinfo = {0};




20,203,705✔





388



  void         *msg = NULL;




20,203,705✔





389



  int32_t       code = 0;




20,203,705✔





390



  int32_t       numOfMsgs = 0;




20,203,705✔





391












392



  int32_t ret = taosBlockSIGPIPE();




20,203,705✔





393



  if (ret < 0) {




20,199,303✔





394



    uError("worker:%s:%d failed to block SIGPIPE", pool->name, worker->id);




×







395



  }










396












397



  setThreadName(pool->name);




20,199,303✔





398



  worker->pid = taosGetSelfPthreadId();




20,203,705✔





399



  uInfo("worker:%s:%d is running, thread:%08" PRId64, pool->name, worker->id, worker->pid);




20,202,938✔





400












401



  while (1) {










402



    numOfMsgs = taosReadAllQitemsFromQset(worker->qset, worker->qall, &qinfo);




1,567,145,163✔





403



    if (numOfMsgs == 0) {




1,567,066,501✔





404



      uInfo("worker:%s:%d qset:%p, got no message and exiting, thread:%08" PRId64, pool->name, worker->id, worker->qset,




20,198,679✔





405



            worker->pid);










406



      break;




20,203,340✔





407



    }










408












409



    if (qinfo.timestamp != 0) {




1,546,867,822✔





410



      int64_t cost = taosGetTimestampUs() - qinfo.timestamp;




1,546,822,903✔





411



      if (cost > QUEUE_THRESHOLD) {




1,546,822,903✔





412



        uWarn("worker:%s,message has been queued for too long, cost: %" PRId64 "s", pool->name, cost / QUEUE_THRESHOLD);




27,207✔





413



      }










414



    }










415












416



    if (qinfo.fp != NULL) {




1,546,811,811✔





417



      qinfo.workerId = worker->id;




1,546,853,734✔





418



      qinfo.threadNum = pool->num;




1,546,872,921✔





419



      (*((FItems)qinfo.fp))(&qinfo, worker->qall, numOfMsgs);




1,546,884,707✔





420



    }










421



    taosUpdateItemSize(qinfo.queue, numOfMsgs);




1,546,800,381✔





422



  }










423












424



  return NULL;




20,203,340✔





425



}










426












427



STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp) {




21,642,693✔





428



  (void)taosThreadMutexLock(&pool->mutex);




21,642,693✔





429



  SWWorker   *worker = pool->workers + pool->nextId;




21,643,241✔





430



  int32_t     code = -1;




21,642,983✔





431



  STaosQueue *queue;




21,638,652✔





432












433



  code = taosOpenQueue(&queue);




21,643,241✔





434



  if (code) goto _OVER;




21,643,156✔





435












436



  taosSetQueueFp(queue, NULL, fp);




21,643,156✔





437



  if (worker->qset == NULL) {




21,643,075✔





438



    code = taosOpenQset(&worker->qset);




20,203,636✔





439



    if (code) goto _OVER;




20,202,924✔





440












441



    code = taosAddIntoQset(worker->qset, queue, ahandle);




20,202,924✔





442



    if (code) goto _OVER;




20,202,924✔





443



    code = taosAllocateQall(&worker->qall);




20,202,924✔





444



    if (code) goto _OVER;




20,203,705✔





445












446



    TdThreadAttr thAttr;




20,199,374✔





447



    (void)taosThreadAttrInit(&thAttr);




20,203,705✔





448



    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);




20,202,413✔





449



    code = taosThreadCreate(&worker->thread, &thAttr, (ThreadFp)tWWorkerThreadFp, worker);




20,203,317✔





450



    if ((code)) goto _OVER;




20,203,705✔





451












452



    uInfo("worker:%s:%d is launched, max:%d", pool->name, worker->id, pool->max);




20,203,705✔





453



    pool->nextId = (pool->nextId + 1) % pool->max;




20,203,705✔





454












455



    (void)taosThreadAttrDestroy(&thAttr);




20,203,705✔





456



    pool->num++;




20,203,705✔





457



    if (pool->num > pool->max) pool->num = pool->max;




20,203,705✔





458



  } else {










459



    code = taosAddIntoQset(worker->qset, queue, ahandle);




1,439,536✔





460



    if (code) goto _OVER;




1,439,536✔





461



    pool->nextId = (pool->nextId + 1) % pool->max;




1,439,536✔





462



  }










463












464



_OVER:




21,643,241✔





465



  (void)taosThreadMutexUnlock(&pool->mutex);




21,643,241✔





466












467



  if (code) {




21,643,032✔





468



    if (queue != NULL) taosCloseQueue(queue);




×







469



    if (worker->qset != NULL) taosCloseQset(worker->qset);




×







470



    if (worker->qall != NULL) taosFreeQall(worker->qall);




×







471



    terrno = code;




×







472



    return NULL;




×







473



  } else {










474



    while (worker->pid <= 0) taosMsleep(10);




44,435,685✔





475












476



    taosQueueSetThreadId(queue, worker->pid);




21,643,241✔





477



    uInfo("worker:%s, queue:%p is allocated, ahandle:%p thread:%08" PRId64, pool->name, queue, ahandle, worker->pid);




21,641,998✔





478



    return queue;




21,643,241✔





479



  }










480



}










481












482



void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) {




21,643,241✔





483



  uInfo("worker:%s, queue:%p is freed", pool->name, queue);




21,643,241✔





484



  taosCloseQueue(queue);




21,643,241✔





485



}




21,643,241✔





486












487



int32_t tSingleWorkerInit(SSingleWorker *pWorker, const SSingleWorkerCfg *pCfg) {




6,947,105✔





488



  int32_t code;










489



  pWorker->poolType = pCfg->poolType;




6,947,105✔





490



  pWorker->name = pCfg->name;




6,947,105✔





491



  pWorker->stopNoWaitQueue = pCfg->stopNoWaitQueue;




6,947,105✔





492












493



  switch (pCfg->poolType) {




6,947,105✔





494



    case QWORKER_POOL: {




5,792,989✔





495



      SQWorkerPool *pPool = taosMemoryCalloc(1, sizeof(SQWorkerPool));




5,792,989✔





496



      if (!pPool) {




5,792,989✔





497



        return terrno;




×







498



      }










499



      pPool->name = pCfg->name;




5,792,989✔





500



      pPool->min = pCfg->min;




5,792,989✔





501



      pPool->max = pCfg->max;




5,792,989✔





502



      pWorker->pool = pPool;




5,792,989✔





503



      if ((code = tQWorkerInit(pPool))) {




5,792,989✔





504



        return (terrno = code);




×







505



      }










506












507



      pWorker->queue = tQWorkerAllocQueue(pPool, pCfg->param, pCfg->fp);




5,792,989✔





508



      if (pWorker->queue == NULL) {




5,792,989✔





509



        return terrno;




×







510



      }










511



    } break;




5,792,989✔





512



    case QUERY_AUTO_QWORKER_POOL: {




1,154,116✔





513



      SQueryAutoQWorkerPool *pPool = taosMemoryCalloc(1, sizeof(SQueryAutoQWorkerPool));




1,154,116✔





514



      if (!pPool) {




1,154,116✔





515



        return terrno;




×







516



      }










517



      pPool->name = pCfg->name;




1,154,116✔





518



      pPool->min = pCfg->min;




1,154,116✔





519



      pPool->max = pCfg->max;




1,154,116✔





520



      pPool->stopNoWaitQueue = pCfg->stopNoWaitQueue;




1,154,116✔





521



      pWorker->pool = pPool;




1,154,116✔





522












523



      code = tQueryAutoQWorkerInit(pPool);




1,154,116✔





524



      if (code) return code;




1,154,116✔





525












526



      pWorker->queue = tQueryAutoQWorkerAllocQueue(pPool, pCfg->param, pCfg->fp);




1,154,116✔





527



      if (!pWorker->queue) {




1,154,116✔





528



        return terrno;




×







529



      }










530



    } break;




1,154,116✔





531



    default:




×







532



      return TSDB_CODE_INVALID_PARA;




×







533



  }










534



  return 0;




6,947,105✔





535



}










536












537



void tSingleWorkerCleanup(SSingleWorker *pWorker) {




6,947,105✔





538



  if (pWorker->queue == NULL) return;




6,947,105✔





539



  if (!pWorker->stopNoWaitQueue) {




6,947,105✔





540



    while (!taosQueueEmpty(pWorker->queue)) {




7,474,920✔





541



      taosMsleep(10);




631,178✔





542



    }










543



  }










544












545



  switch (pWorker->poolType) {




6,947,105✔





546



    case QWORKER_POOL:




5,792,989✔





547



      tQWorkerCleanup(pWorker->pool);




5,792,989✔





548



      tQWorkerFreeQueue(pWorker->pool, pWorker->queue);




5,792,989✔





549



      taosMemoryFree(pWorker->pool);




5,792,989✔





550



      break;




5,792,989✔





551



    case QUERY_AUTO_QWORKER_POOL:




1,154,116✔





552



      tQueryAutoQWorkerCleanup(pWorker->pool);




1,154,116✔





553



      tQueryAutoQWorkerFreeQueue(pWorker->pool, pWorker->queue);




1,154,116✔





554



      taosMemoryFree(pWorker->pool);




1,154,116✔





555



      break;




1,154,116✔





556



    default:




×







557



      break;




×







558



  }










559



}










560












561



int32_t tMultiWorkerInit(SMultiWorker *pWorker, const SMultiWorkerCfg *pCfg) {




16,898,663✔





562



  SWWorkerPool *pPool = &pWorker->pool;




16,898,663✔





563



  pPool->name = pCfg->name;




16,898,663✔





564



  pPool->max = pCfg->max;




16,899,292✔





565












566



  int32_t code = tWWorkerInit(pPool);




16,899,292✔





567



  if (code) return code;




16,899,292✔





568












569



  pWorker->queue = tWWorkerAllocQueue(pPool, pCfg->param, pCfg->fp);




16,899,292✔





570



  if (pWorker->queue == NULL) {




16,899,292✔





571



    return terrno;




×







572



  }










573












574



  pWorker->name = pCfg->name;




16,899,292✔





575



  return 0;




16,899,292✔





576



}










577












578



void tMultiWorkerCleanup(SMultiWorker *pWorker) {




16,898,600✔





579



  if (pWorker->queue == NULL) return;




16,898,600✔





580












581



  while (!taosQueueEmpty(pWorker->queue)) {




19,690,177✔





582



    taosMsleep(10);




2,792,311✔





583



  }










584












585



  tWWorkerCleanup(&pWorker->pool);




16,899,292✔





586



  tWWorkerFreeQueue(&pWorker->pool, pWorker->queue);




16,899,292✔





587



}










588












589



static int32_t tQueryAutoQWorkerAddWorker(SQueryAutoQWorkerPool *pool);










590



static int32_t tQueryAutoQWorkerBeforeBlocking(void *p);










591



static int32_t tQueryAutoQWorkerRecoverFromBlocking(void *p);










592



static void    tQueryAutoQWorkerWaitingCheck(SQueryAutoQWorkerPool *pPool);










593



static bool    tQueryAutoQWorkerTryRecycleWorker(SQueryAutoQWorkerPool *pPool, SQueryAutoQWorker *pWorker);










594












595



#define GET_ACTIVE_N(int64_val)  (int32_t)((int64_val) >> 32)










596



#define GET_RUNNING_N(int64_val) (int32_t)(int64_val & 0xFFFFFFFF)










597












598



static int32_t atomicFetchSubActive(int64_t *ptr, int32_t val) {




×







599



  int64_t acutalSubVal = val;




×







600



  acutalSubVal <<= 32;




×







601



  int64_t newVal64 = atomic_fetch_sub_64(ptr, acutalSubVal);




×







602



  return GET_ACTIVE_N(newVal64);




×







603



}










604












605



static int32_t atomicFetchSubRunning(int64_t *ptr, int32_t val) { return GET_RUNNING_N(atomic_fetch_sub_64(ptr, val)); }




2,147,483,647✔





606












607



static int32_t atomicFetchAddActive(int64_t *ptr, int32_t val) {




171,113,362✔





608



  int64_t actualAddVal = val;




171,113,362✔





609



  actualAddVal <<= 32;




171,113,362✔





610



  int64_t newVal64 = atomic_fetch_add_64(ptr, actualAddVal);




171,113,362✔





611



  return GET_ACTIVE_N(newVal64);




171,113,362✔





612



}










613












614



static int32_t atomicFetchAddRunning(int64_t *ptr, int32_t val) { return GET_RUNNING_N(atomic_fetch_add_64(ptr, val)); }




×







615












616



static bool atomicCompareExchangeActive(int64_t *ptr, int32_t *expectedVal, int32_t newVal) {




7,079✔





617



  int64_t oldVal64 = *expectedVal, newVal64 = newVal;




7,079✔





618



  int32_t running = GET_RUNNING_N(*ptr);




7,079✔





619



  oldVal64 <<= 32;




7,079✔





620



  newVal64 <<= 32;




7,079✔





621



  oldVal64 |= running;




7,079✔





622



  newVal64 |= running;




7,079✔





623



  int64_t actualNewVal64 = atomic_val_compare_exchange_64(ptr, oldVal64, newVal64);




7,079✔





624



  if (actualNewVal64 == oldVal64) {




7,079✔





625



    return true;




7,079✔





626



  } else {










627



    *expectedVal = GET_ACTIVE_N(actualNewVal64);




×







628



    return false;




×







629



  }










630



}










631












632



static int64_t atomicCompareExchangeRunning(int64_t *ptr, int32_t *expectedVal, int32_t newVal) {




×







633



  int64_t oldVal64 = *expectedVal, newVal64 = newVal;




×







634



  int64_t activeShifted = GET_ACTIVE_N(*ptr);




×







635



  activeShifted <<= 32;




×







636



  oldVal64 |= activeShifted;




×







637



  newVal64 |= activeShifted;




×







638



  int64_t actualNewVal64 = atomic_val_compare_exchange_64(ptr, oldVal64, newVal64);




×







639



  if (actualNewVal64 == oldVal64) {




×







640



    return true;




×







641



  } else {










642



    *expectedVal = GET_RUNNING_N(actualNewVal64);




×







643



    return false;




×







644



  }










645



}










646












647



static int64_t atomicCompareExchangeActiveAndRunning(int64_t *ptr, int32_t *expectedActive, int32_t newActive,




2,147,483,647✔





648



                                                     int32_t *expectedRunning, int32_t newRunning) {










649



  int64_t oldVal64 = *expectedActive, newVal64 = newActive;




2,147,483,647✔





650



  oldVal64 <<= 32;




2,147,483,647✔





651



  oldVal64 |= *expectedRunning;




2,147,483,647✔





652



  newVal64 <<= 32;




2,147,483,647✔





653



  newVal64 |= newRunning;




2,147,483,647✔





654



  int64_t actualNewVal64 = atomic_val_compare_exchange_64(ptr, oldVal64, newVal64);




2,147,483,647✔





655



  if (actualNewVal64 == oldVal64) {




2,147,483,647✔





656



    return true;




2,147,483,647✔





657



  } else {










658



    *expectedActive = GET_ACTIVE_N(actualNewVal64);




837,057✔





659



    *expectedRunning = GET_RUNNING_N(actualNewVal64);




837,057✔





660



    return false;




837,057✔





661



  }










662



}










663












664



static void *tQueryAutoQWorkerThreadFp(SQueryAutoQWorker *worker) {




171,498,795✔





665



  SQueryAutoQWorkerPool *pool = worker->pool;




171,498,795✔





666



  SQueueInfo             qinfo = {0};




171,515,886✔





667



  void                  *msg = NULL;




171,518,126✔





668



  int32_t                code = 0;




171,521,323✔





669












670



  int32_t ret = taosBlockSIGPIPE();




171,521,323✔





671



  if (ret < 0) {




171,508,149✔





672



    uError("worker:%s:%d failed to block SIGPIPE", pool->name, worker->id);




×







673



  }










674












675



  setThreadName(pool->name);




171,508,149✔





676



  worker->pid = taosGetSelfPthreadId();




171,513,773✔





677



  uDebug("worker:%s:%d is running, thread:%08" PRId64, pool->name, worker->id, worker->pid);




171,513,493✔





678












679



  while (1) {










680



    if (taosReadQitemFromQset(pool->qset, (void **)&msg, &qinfo) == 0) {




2,147,483,647✔





681



      uInfo("worker:%s:%d qset:%p, got no message and exiting, thread:%08" PRId64, pool->name, worker->id, pool->qset,




171,311,250✔





682



            worker->pid);










683



      break;




171,384,743✔





684



    }










685












686



    if (pool->exit) {




2,147,483,647✔





687



      uInfo("worker:%s:%d exit, thread:%08" PRId64, pool->name, worker->id, worker->pid);




×







688



      break;




×







689



    }










690












691



    if (qinfo.timestamp != 0) {




2,147,483,647✔





692



      int64_t cost = taosGetTimestampUs() - qinfo.timestamp;




2,147,483,647✔





693



      if (cost > QUEUE_THRESHOLD) {




2,147,483,647✔





694



        uWarn("worker:%s,message has been queued for too long, cost: %" PRId64 "s", pool->name, cost / QUEUE_THRESHOLD);




×







695



      }










696



    }










697












698



    tQueryAutoQWorkerWaitingCheck(pool);




2,147,483,647✔





699












700



    if (qinfo.fp != NULL) {




2,147,483,647✔





701



      qinfo.workerId = worker->id;




2,147,483,647✔





702



      qinfo.threadNum = pool->num;




2,147,483,647✔





703



      qinfo.workerCb = pool->pCb;




2,147,483,647✔





704



      (*((FItem)qinfo.fp))(&qinfo, msg);




2,147,483,647✔





705



    }










706












707



    taosUpdateItemSize(qinfo.queue, 1);




2,147,483,647✔





708



    if (!tQueryAutoQWorkerTryRecycleWorker(pool, worker)) {




2,147,483,647✔





709



      uDebug("worker:%s:%d exited", pool->name, worker->id);




112,535✔





710



      break;




112,535✔





711



    }










712



  }










713












714



  DestoryThreadLocalRegComp();




171,497,278✔





715



  closeThreadNotificationConn();




171,467,201✔





716












717



  return NULL;




171,470,413✔





718



}










719












720



static bool tQueryAutoQWorkerTrySignalWaitingAfterBlock(void *p) {




2,147,483,647✔





721



  SQueryAutoQWorkerPool *pPool = p;




2,147,483,647✔





722



  bool                   ret = false;




2,147,483,647✔





723



  int32_t                waiting = pPool->waitingAfterBlockN;




2,147,483,647✔





724



  while (waiting > 0) {




2,147,483,647✔





725



    int32_t waitingNew = atomic_val_compare_exchange_32(&pPool->waitingAfterBlockN, waiting, waiting - 1);




23,548✔





726



    if (waitingNew == waiting) {




23,548✔





727



      (void)taosThreadMutexLock(&pPool->waitingAfterBlockLock);




23,548✔





728



      (void)taosThreadCondSignal(&pPool->waitingAfterBlockCond);




23,548✔





729



      (void)taosThreadMutexUnlock(&pPool->waitingAfterBlockLock);




23,548✔





730



      ret = true;




18,529✔





731



      break;




18,529✔





732



    }










733



    waiting = waitingNew;




×







734



  }










735



  return ret;




2,147,483,647✔





736



}










737












738



static bool tQueryAutoQWorkerTrySignalWaitingBeforeProcess(void *p) {




2,147,483,647✔





739



  SQueryAutoQWorkerPool *pPool = p;




2,147,483,647✔





740



  bool                   ret = false;




2,147,483,647✔





741



  int32_t                waiting = pPool->waitingBeforeProcessMsgN;




2,147,483,647✔





742



  while (waiting > 0) {




2,147,483,647✔





743



    int32_t waitingNew = atomic_val_compare_exchange_32(&pPool->waitingBeforeProcessMsgN, waiting, waiting - 1);




7,079✔





744



    if (waitingNew == waiting) {




7,079✔





745



      (void)taosThreadMutexLock(&pPool->waitingBeforeProcessMsgLock);




7,079✔





746



      (void)taosThreadCondSignal(&pPool->waitingBeforeProcessMsgCond);




7,079✔





747



      (void)taosThreadMutexUnlock(&pPool->waitingBeforeProcessMsgLock);




7,079✔





748



      ret = true;




×







749



      break;




×







750



    }










751



    waiting = waitingNew;




×







752



  }










753



  return ret;




2,147,483,647✔





754



}










755












756



static bool tQueryAutoQWorkerTryDecActive(void *p, int32_t minActive) {




2,147,483,647✔





757



  SQueryAutoQWorkerPool *pPool = p;




2,147,483,647✔





758



  bool                   ret = false;




2,147,483,647✔





759



  int64_t                val64 = pPool->activeRunningN;




2,147,483,647✔





760



  int32_t                active = GET_ACTIVE_N(val64), running = GET_RUNNING_N(val64);




2,147,483,647✔





761



  while (active > minActive) {




2,147,483,647✔





762



    if (atomicCompareExchangeActiveAndRunning(&pPool->activeRunningN, &active, active - 1, &running, running - 1))




169,854,455✔





763



      return true;




169,841,431✔





764



  }










765



  return false;




2,147,483,647✔





766



}










767












768



static void tQueryAutoQWorkerWaitingCheck(SQueryAutoQWorkerPool *pPool) {




2,147,483,647✔





769



  while (1) {




×







770



    int64_t val64 = pPool->activeRunningN;




2,147,483,647✔





771



    int32_t running = GET_RUNNING_N(val64), active = GET_ACTIVE_N(val64);




2,147,483,647✔





772



    while (running < pPool->num) {




2,147,483,647✔





773



      if (atomicCompareExchangeActiveAndRunning(&pPool->activeRunningN, &active, active, &running, running + 1)) {




2,147,483,647✔





774



        return;




2,147,483,647✔





775



      }










776



    }










777



    if (atomicCompareExchangeActive(&pPool->activeRunningN, &active, active - 1)) {




2,154✔





778



      break;




7,079✔





779



    }










780



  }










781



  // to wait for process










782



  (void)taosThreadMutexLock(&pPool->waitingBeforeProcessMsgLock);




7,079✔





783



  (void)atomic_fetch_add_32(&pPool->waitingBeforeProcessMsgN, 1);




7,079✔





784



  if (!pPool->exit) (void)taosThreadCondWait(&pPool->waitingBeforeProcessMsgCond, &pPool->waitingBeforeProcessMsgLock);




7,079✔





785



  // recovered from waiting










786



  (void)taosThreadMutexUnlock(&pPool->waitingBeforeProcessMsgLock);




7,079✔





787



  return;




7,079✔





788



}










789












790



bool tQueryAutoQWorkerTryRecycleWorker(SQueryAutoQWorkerPool *pPool, SQueryAutoQWorker *pWorker) {




2,147,483,647✔





791



  if (tQueryAutoQWorkerTrySignalWaitingAfterBlock(pPool) || tQueryAutoQWorkerTrySignalWaitingBeforeProcess(pPool) ||




2,147,483,647✔





792



      tQueryAutoQWorkerTryDecActive(pPool, pPool->num)) {




2,147,483,647✔





793



    (void)taosThreadMutexLock(&pPool->poolLock);




89,480,385✔





794



    if (pPool->exit) {




89,465,304✔





795



      (void)taosThreadMutexUnlock(&pPool->poolLock);




1,778✔





796



      return false;




1,778✔





797



    }










798












799



    SListNode *pNode = listNode(pWorker);




89,463,526✔





800



    SListNode *tNode = tdListPopNode(pPool->workers, pNode);




89,463,526✔





801



    // reclaim some workers










802



    if (pWorker->id >= pPool->maxInUse) {




89,463,526✔





803



      while (listNEles(pPool->exitedWorkers) > pPool->maxInUse - pPool->num) {




1,149✔





804



        SListNode         *head = tdListPopHead(pPool->exitedWorkers);




×







805



        SQueryAutoQWorker *pWorker = (SQueryAutoQWorker *)head->data;




×







806



        if (pWorker && taosCheckPthreadValid(pWorker->thread)) {




×







807



          (void)taosThreadJoin(pWorker->thread, NULL);




×







808



          taosThreadClear(&pWorker->thread);




×







809



        }










810



        taosMemoryFree(head);




×







811



      }










812



      tdListAppendNode(pPool->exitedWorkers, pNode);




1,149✔





813



      (void)taosThreadMutexUnlock(&pPool->poolLock);




1,149✔





814



      return false;




1,149✔





815



    }










816












817



    // put back to backup pool










818



    tdListAppendNode(pPool->backupWorkers, pNode);




89,462,377✔





819



    (void)taosThreadMutexUnlock(&pPool->poolLock);




89,462,377✔





820












821



    // start to wait at backup cond










822



    (void)taosThreadMutexLock(&pPool->backupLock);




89,462,377✔





823



    (void)atomic_fetch_add_32(&pPool->backupNum, 1);




89,462,377✔





824



    if (!pPool->exit) (void)taosThreadCondWait(&pPool->backupCond, &pPool->backupLock);




89,462,377✔





825



    (void)taosThreadMutexUnlock(&pPool->backupLock);




89,462,377✔





826












827



    // recovered from backup










828



    (void)taosThreadMutexLock(&pPool->poolLock);




89,461,433✔





829



    if (pPool->exit) {




89,462,377✔





830



      (void)taosThreadMutexUnlock(&pPool->poolLock);




109,608✔





831



      return false;




109,608✔





832



    }










833



    SListNode *tNode1 = tdListPopNode(pPool->backupWorkers, pNode);




89,352,769✔





834



    tdListAppendNode(pPool->workers, pNode);




89,352,769✔





835



    (void)taosThreadMutexUnlock(&pPool->poolLock);




89,352,769✔





836












837



    return true;




89,352,769✔





838



  } else {










839



    (void)atomicFetchSubRunning(&pPool->activeRunningN, 1);




2,100,159,367✔





840



    return true;




2,100,191,719✔





841



  }










842



}










843












844



int32_t tQueryAutoQWorkerInit(SQueryAutoQWorkerPool *pool) {




3,797,848✔





845



  int32_t code;










846












847



  pool->exit = false;




3,797,848✔





848












849



  (void)taosThreadMutexInit(&pool->poolLock, NULL);




3,797,848✔





850



  (void)taosThreadMutexInit(&pool->backupLock, NULL);




3,797,848✔





851



  (void)taosThreadMutexInit(&pool->waitingAfterBlockLock, NULL);




3,797,848✔





852



  (void)taosThreadMutexInit(&pool->waitingBeforeProcessMsgLock, NULL);




3,797,848✔





853












854



  (void)taosThreadCondInit(&pool->waitingBeforeProcessMsgCond, NULL);




3,797,848✔





855



  (void)taosThreadCondInit(&pool->waitingAfterBlockCond, NULL);




3,797,848✔





856



  (void)taosThreadCondInit(&pool->backupCond, NULL);




3,797,848✔





857












858



  code = taosOpenQset(&pool->qset);




3,797,848✔





859



  if (code) return terrno = code;




3,797,848✔





860



  pool->workers = tdListNew(sizeof(SQueryAutoQWorker));




3,797,848✔





861



  if (!pool->workers) return terrno;




3,797,848✔





862



  pool->backupWorkers = tdListNew(sizeof(SQueryAutoQWorker));




3,797,848✔





863



  if (!pool->backupWorkers) return terrno;




3,797,848✔





864



  pool->exitedWorkers = tdListNew(sizeof(SQueryAutoQWorker));




3,797,848✔





865



  if (!pool->exitedWorkers) return terrno;




3,797,848✔





866



  pool->maxInUse = pool->max * 2 + 2;




3,797,848✔





867












868



  if (!pool->pCb) {




3,797,848✔





869



    pool->pCb = taosMemoryCalloc(1, sizeof(SQueryAutoQWorkerPoolCB));




3,797,848✔





870



    if (!pool->pCb) return terrno;




3,797,848✔





871



    pool->pCb->pPool = pool;




3,797,848✔





872



    pool->pCb->beforeBlocking = tQueryAutoQWorkerBeforeBlocking;




3,797,848✔





873



    pool->pCb->afterRecoverFromBlocking = tQueryAutoQWorkerRecoverFromBlocking;




3,797,848✔





874



  }










875



  return TSDB_CODE_SUCCESS;




3,797,848✔





876



}










877












878



void tQueryAutoQWorkerCleanup(SQueryAutoQWorkerPool *pPool) {




3,797,883✔





879



  (void)taosThreadMutexLock(&pPool->poolLock);




3,797,883✔





880



  if (pPool->stopNoWaitQueue) {




3,797,883✔





881



    pPool->exit = true;




103,363✔





882



  }










883



  int32_t size = 0;




3,797,883✔





884



  if (pPool->workers) {




3,797,883✔





885



    size = listNEles(pPool->workers);




3,797,848✔





886



  }










887



  if (pPool->backupWorkers) {




3,797,883✔





888



    size += listNEles(pPool->backupWorkers);




3,797,848✔





889



  }










890



  if (pPool->qset) {




3,797,883✔





891



    for (int32_t i = 0; i < size; ++i) {




175,321,625✔





892



      taosQsetThreadResume(pPool->qset);




171,523,777✔





893



    }










894



  }










895



  (void)taosThreadMutexUnlock(&pPool->poolLock);




3,797,883✔





896












897



  (void)taosThreadMutexLock(&pPool->backupLock);




3,797,883✔





898



  (void)taosThreadCondBroadcast(&pPool->backupCond);




3,797,883✔





899



  (void)taosThreadMutexUnlock(&pPool->backupLock);




3,797,883✔





900












901



  (void)taosThreadMutexLock(&pPool->waitingAfterBlockLock);




3,797,883✔





902



  (void)taosThreadCondBroadcast(&pPool->waitingAfterBlockCond);




3,797,883✔





903



  (void)taosThreadMutexUnlock(&pPool->waitingAfterBlockLock);




3,797,883✔





904












905



  (void)taosThreadMutexLock(&pPool->waitingBeforeProcessMsgLock);




3,797,883✔





906



  (void)taosThreadCondBroadcast(&pPool->waitingBeforeProcessMsgCond);




3,797,883✔





907



  (void)taosThreadMutexUnlock(&pPool->waitingBeforeProcessMsgLock);




3,797,883✔





908












909



  int32_t            idx = 0;




3,797,883✔





910



  SQueryAutoQWorker *worker = NULL;




3,797,883✔





911



  while (pPool->workers) {




175,212,052✔





912



    (void)taosThreadMutexLock(&pPool->poolLock);




175,212,017✔





913



    if (listNEles(pPool->workers) == 0) {




175,212,017✔





914



      (void)taosThreadMutexUnlock(&pPool->poolLock);




3,797,848✔





915



      break;




3,797,848✔





916



    }










917



    SListNode *pNode = tdListPopHead(pPool->workers);




171,414,169✔





918



    uDebug("0free worker node:%p, prev:%p, next:%p", pNode, TD_DLIST_NODE_PREV(pNode), TD_DLIST_NODE_NEXT(pNode));




171,414,169✔





919



    worker = pNode ? (SQueryAutoQWorker *)pNode->data : NULL;




171,414,169✔





920



    (void)taosThreadMutexUnlock(&pPool->poolLock);




171,414,169✔





921



    if (worker && taosCheckPthreadValid(worker->thread)) {




171,414,169✔





922



      (void)taosThreadJoin(worker->thread, NULL);




171,414,169✔





923



      taosThreadClear(&worker->thread);




171,414,169✔





924



    }










925



    uDebug("free worker node:%p, prev:%p, next:%p", pNode, TD_DLIST_NODE_PREV(pNode), TD_DLIST_NODE_NEXT(pNode));




171,414,169✔





926












927



    taosMemoryFree(pNode);




171,414,169✔





928



  }










929












930



  while (pPool->backupWorkers) {




3,907,491✔





931



    (void)taosThreadMutexLock(&pPool->poolLock);




3,907,456✔





932



    if (listNEles(pPool->backupWorkers) == 0) {




3,907,456✔





933



      (void)taosThreadMutexUnlock(&pPool->poolLock);




3,797,848✔





934



      break;




3,797,848✔





935



    }










936



    uDebug("backupworker head:%p, prev:%p, next:%p", TD_DLIST_HEAD(pPool->backupWorkers), TD_DLIST_NODE_PREV(TD_DLIST_HEAD(pPool->backupWorkers)), TD_DLIST_NODE_NEXT(TD_DLIST_HEAD(pPool->backupWorkers)));




109,608✔





937



    SListNode *pNode = tdListPopHead(pPool->backupWorkers);




109,608✔





938



    worker = pNode ? (SQueryAutoQWorker *)pNode->data : NULL;




109,608✔





939



    (void)taosThreadMutexUnlock(&pPool->poolLock);




109,608✔





940












941



    if (worker && taosCheckPthreadValid(worker->thread)) {




109,608✔





942



      (void)taosThreadJoin(worker->thread, NULL);




109,608✔





943



      taosThreadClear(&worker->thread);




109,608✔





944



    }










945



    taosMemoryFree(pNode);




109,608✔





946



  }










947












948



  while (pPool->exitedWorkers) {




3,799,032✔





949



    (void)taosThreadMutexLock(&pPool->poolLock);




3,798,997✔





950



    if (listNEles(pPool->exitedWorkers) == 0) {




3,798,997✔





951



      (void)taosThreadMutexUnlock(&pPool->poolLock);




3,797,848✔





952



      break;




3,797,848✔





953



    }










954












955



    SListNode *pNode = tdListPopHead(pPool->exitedWorkers);




1,149✔





956



    worker = pNode ? (SQueryAutoQWorker *)pNode->data : NULL;




1,149✔





957



    (void)taosThreadMutexUnlock(&pPool->poolLock);




1,149✔





958












959



    if (worker && taosCheckPthreadValid(worker->thread)) {




1,149✔





960



      (void)taosThreadJoin(worker->thread, NULL);




1,149✔





961



      taosThreadClear(&worker->thread);




1,149✔





962



    }










963



    taosMemoryFree(pNode);




1,149✔





964



  }










965












966



  (void)taosThreadMutexLock(&pPool->poolLock);




3,797,883✔





967



  pPool->workers = tdListFree(pPool->workers);




3,797,883✔





968



  pPool->backupWorkers = tdListFree(pPool->backupWorkers);




3,797,883✔





969



  pPool->exitedWorkers = tdListFree(pPool->exitedWorkers);




3,797,883✔





970



  taosMemoryFree(pPool->pCb);




3,797,883✔





971



  (void)taosThreadMutexUnlock(&pPool->poolLock);




3,797,883✔





972












973



  (void)taosThreadMutexDestroy(&pPool->poolLock);




3,797,883✔





974



  (void)taosThreadMutexDestroy(&pPool->backupLock);




3,797,883✔





975



  (void)taosThreadMutexDestroy(&pPool->waitingAfterBlockLock);




3,797,883✔





976



  (void)taosThreadMutexDestroy(&pPool->waitingBeforeProcessMsgLock);




3,797,883✔





977












978



  (void)taosThreadCondDestroy(&pPool->backupCond);




3,797,883✔





979



  (void)taosThreadCondDestroy(&pPool->waitingAfterBlockCond);




3,797,883✔





980



  (void)taosThreadCondDestroy(&pPool->waitingBeforeProcessMsgCond);




3,797,883✔





981



  taosCloseQset(pPool->qset);




3,797,883✔





982



}




3,797,883✔





983












984



STaosQueue *tQueryAutoQWorkerAllocQueue(SQueryAutoQWorkerPool *pool, void *ahandle, FItem fp) {




10,816,659✔





985



  STaosQueue *queue;




10,797,156✔





986



  int32_t     code = taosOpenQueue(&queue);




10,816,659✔





987



  if (code) {




10,816,659✔





988



    terrno = code;




×







989



    return NULL;




×







990



  }










991












992



  (void)taosThreadMutexLock(&pool->poolLock);




10,816,659✔





993



  taosSetQueueFp(queue, fp, NULL);




10,817,444✔





994



  code = taosAddIntoQset(pool->qset, queue, ahandle);




10,817,444✔





995



  if (code) {




10,816,116✔





996



    taosCloseQueue(queue);




×







997



    queue = NULL;




×







998



    (void)taosThreadMutexUnlock(&pool->poolLock);




×







999



    return NULL;




×







1000



  }










1001



  SQueryAutoQWorker  worker = {0};




10,816,116✔





1002



  SQueryAutoQWorker *pWorker = NULL;




10,816,875✔





1003












1004



  // spawn a thread to process queue










1005



  if (pool->num < pool->max) {




10,816,875✔





1006



    do {










1007



      worker.id = listNEles(pool->workers);




171,113,362✔





1008



      worker.backupIdx = -1;




171,113,362✔





1009



      worker.pool = pool;




171,113,362✔





1010



      SListNode *pNode = tdListAdd(pool->workers, &worker);




171,113,362✔





1011



      if (!pNode) {




171,113,362✔





1012



        taosCloseQueue(queue);




×







1013



        queue = NULL;




×







1014



        terrno = TSDB_CODE_OUT_OF_MEMORY;




×







1015



        break;




×







1016



      }










1017



      pWorker = (SQueryAutoQWorker *)pNode->data;




171,113,362✔





1018












1019



      TdThreadAttr thAttr;




169,688,022✔





1020



      (void)taosThreadAttrInit(&thAttr);




171,113,362✔





1021



      (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);




171,113,362✔





1022












1023



      if (taosThreadCreate(&pWorker->thread, &thAttr, (ThreadFp)tQueryAutoQWorkerThreadFp, pWorker) != 0) {




171,113,362✔





1024



        taosCloseQueue(queue);




×







1025



        queue = NULL;




×







1026



        break;




×







1027



      }










1028












1029



      (void)taosThreadAttrDestroy(&thAttr);




171,113,362✔





1030



      pool->num++;




171,113,362✔





1031



      (void)atomicFetchAddActive(&pool->activeRunningN, 1);




171,113,362✔





1032



      uInfo("worker:%s:%d is launched, total:%d", pool->name, pWorker->id, pool->num);




171,113,362✔





1033



    } while (pool->num < pool->min);




171,113,362✔





1034



  }










1035












1036



  (void)taosThreadMutexUnlock(&pool->poolLock);




10,816,875✔





1037



  uInfo("worker:%s, queue:%p is allocated, ahandle:%p", pool->name, queue, ahandle);




10,817,444✔





1038












1039



  return queue;




10,817,444✔





1040



}










1041












1042



void tQueryAutoQWorkerFreeQueue(SQueryAutoQWorkerPool *pPool, STaosQueue *pQ) { taosCloseQueue(pQ); }




9,602,708✔





1043












1044



static int32_t tQueryAutoQWorkerAddWorker(SQueryAutoQWorkerPool *pool) {




89,459,293✔





1045



  // try backup pool










1046



  int32_t backup = pool->backupNum;




89,459,293✔





1047



  while (backup > 0) {




89,502,074✔





1048



    int32_t backupNew = atomic_val_compare_exchange_32(&pool->backupNum, backup, backup - 1);




89,090,510✔





1049



    if (backupNew == backup) {




89,097,122✔





1050



      (void)taosThreadCondSignal(&pool->backupCond);




89,053,740✔





1051



      return TSDB_CODE_SUCCESS;




89,053,740✔





1052



    }










1053



    backup = backupNew;




43,382✔





1054



  }










1055



  // backup pool is empty, create new










1056



  SQueryAutoQWorker *pWorker = NULL;




411,564✔





1057



  SQueryAutoQWorker  worker = {0};




411,564✔





1058



  worker.pool = pool;




411,564✔





1059



  worker.backupIdx = -1;




411,564✔





1060



  (void)taosThreadMutexLock(&pool->poolLock);




411,564✔





1061



  if (pool->exit) {




411,564✔





1062



    (void)taosThreadMutexUnlock(&pool->poolLock);




×







1063



    return TSDB_CODE_SUCCESS;




×







1064



  }










1065



  worker.id = listNEles(pool->workers);




411,564✔





1066



  SListNode *pNode = tdListAdd(pool->workers, &worker);




411,564✔





1067



  if (!pNode) {




411,564✔





1068



    (void)taosThreadMutexUnlock(&pool->poolLock);




×







1069



    return terrno;




×







1070



  }










1071



  (void)taosThreadMutexUnlock(&pool->poolLock);




411,564✔





1072



  pWorker = (SQueryAutoQWorker *)pNode->data;




411,564✔





1073












1074



  TdThreadAttr thAttr;




410,517✔





1075



  (void)taosThreadAttrInit(&thAttr);




411,564✔





1076



  (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);




411,564✔





1077












1078



  if (taosThreadCreate(&pWorker->thread, &thAttr, (ThreadFp)tQueryAutoQWorkerThreadFp, pWorker) != 0) {




411,564✔





1079



    uError("create queryAutoWorker thread failed, error:%s", tstrerror(terrno));




×







1080



    return terrno;




×







1081



  }










1082



  (void)taosThreadAttrDestroy(&thAttr);




411,564✔





1083












1084



  return TSDB_CODE_SUCCESS;




411,564✔





1085



}










1086












1087



static int32_t tQueryAutoQWorkerBeforeBlocking(void *p) {




169,864,844✔





1088



  SQueryAutoQWorkerPool *pPool = p;




169,864,844✔





1089



  if (tQueryAutoQWorkerTrySignalWaitingAfterBlock(p) || tQueryAutoQWorkerTrySignalWaitingBeforeProcess(p) ||




339,715,557✔





1090



      tQueryAutoQWorkerTryDecActive(p, pPool->num)) {




169,840,396✔





1091



  } else {










1092



    int32_t code = tQueryAutoQWorkerAddWorker(pPool);




89,460,055✔





1093



    if (code != TSDB_CODE_SUCCESS) {




89,464,232✔





1094



      return code;




×







1095



    }










1096



    (void)atomicFetchSubRunning(&pPool->activeRunningN, 1);




89,464,232✔





1097



  }










1098












1099



  return TSDB_CODE_SUCCESS;




169,871,500✔





1100



}










1101












1102



static int32_t tQueryAutoQWorkerRecoverFromBlocking(void *p) {




169,870,542✔





1103



  SQueryAutoQWorkerPool *pPool = p;




169,870,542✔





1104



  int64_t                val64 = pPool->activeRunningN;




169,870,542✔





1105



  int32_t                running = GET_RUNNING_N(val64), active = GET_ACTIVE_N(val64);




169,871,522✔





1106



  while (running < pPool->num) {




169,879,353✔





1107



    if (atomicCompareExchangeActiveAndRunning(&pPool->activeRunningN, &active, active + 1, &running, running + 1)) {




169,856,451✔





1108



      return TSDB_CODE_SUCCESS;




169,848,510✔





1109



    }










1110



  }










1111



  (void)taosThreadMutexLock(&pPool->waitingAfterBlockLock);




22,782✔





1112



  (void)atomic_fetch_add_32(&pPool->waitingAfterBlockN, 1);




23,548✔





1113



  if (!pPool->exit) (void)taosThreadCondWait(&pPool->waitingAfterBlockCond, &pPool->waitingAfterBlockLock);




23,548✔





1114



  (void)taosThreadMutexUnlock(&pPool->waitingAfterBlockLock);




23,548✔





1115



  if (pPool->exit) return TSDB_CODE_QRY_QWORKER_QUIT;




23,548✔





1116



  return TSDB_CODE_SUCCESS;




23,548✔





1117



}










1118












1119



int32_t tDispatchWorkerInit(SDispatchWorkerPool *pPool) {




1,339,045✔





1120



  int32_t code = 0;




1,339,045✔





1121



  pPool->num = 0;




1,339,045✔





1122



  pPool->pWorkers = taosMemCalloc(pPool->max, sizeof(SDispatchWorker));




1,339,045✔





1123



  if (!pPool->pWorkers) return terrno;




1,339,045✔





1124



  (void)taosThreadMutexInit(&pPool->poolLock, NULL);




1,339,045✔





1125



  return code;




1,339,045✔





1126



}










1127












1128



static void *tDispatchWorkerThreadFp(SDispatchWorker *pWorker) {




10,291,309✔





1129



  SDispatchWorkerPool *pPool = pWorker->pool;




10,291,309✔





1130



  SQueueInfo qinfo = {0};




10,299,810✔





1131



  int32_t code = 0;




10,299,810✔





1132



  void *msg = NULL;




10,299,810✔





1133












1134



  int32_t ret = taosBlockSIGPIPE();




10,299,810✔





1135



  if (ret < 0) {




10,298,599✔





1136



    uError("worker:%s:%d failed to block SIGPIPE", pPool->name, pWorker->id);




×







1137



  }










1138












1139



  setThreadName(pPool->name);




10,298,599✔





1140



  pWorker->pid = taosGetSelfPthreadId();




10,299,859✔





1141



  uInfo("worker:%s:%d is running, thread:%d", pPool->name, pWorker->id, pWorker->pid);




10,299,844✔





1142












1143



  while (1) {










1144



    if (taosReadQitemFromQset(pWorker->qset, (void **)&msg, &qinfo) == 0) {




110,569,058✔





1145



      uInfo("worker:%s:%d qset:%p, got no message and exiting, thread:%d", pPool->name, pWorker->id,




10,285,600✔





1146



            pWorker->qset, pWorker->pid);










1147



      break;




10,299,464✔





1148



    }










1149












1150



    if (qinfo.timestamp != 0) {




100,281,726✔





1151



      int64_t cost = taosGetTimestampUs() - qinfo.timestamp;




100,284,129✔





1152



      if (cost > QUEUE_THRESHOLD) {




100,284,129✔





1153



        uWarn("worker:%s,message has been queued for too long, cost: %" PRId64 "s", pPool->name, cost / QUEUE_THRESHOLD);




261,163✔





1154



      }










1155



    }










1156












1157



    if (qinfo.fp != NULL) {




100,284,883✔





1158



      qinfo.workerId = pWorker->id;




100,280,037✔





1159



      qinfo.threadNum = pPool->num;




100,280,412✔





1160



      (*((FItem)qinfo.fp))(&qinfo, msg);




100,282,084✔





1161



    }










1162



  }










1163



  DestoryThreadLocalRegComp();




10,299,464✔





1164



  closeThreadNotificationConn();




10,297,962✔





1165



  return NULL;




10,299,154✔





1166



}










1167












1168



int32_t tDispatchWorkerAllocQueue(SDispatchWorkerPool *pPool, void *ahandle, FItem fp, DispatchFp dispatchFp) {




1,339,045✔





1169



  int32_t code = 0;




1,339,045✔





1170



  SDispatchWorker* pWorker = NULL;




1,339,045✔





1171



  (void)taosThreadMutexLock(&pPool->poolLock);




1,339,045✔





1172



  pPool->dispatchFp = dispatchFp;




1,339,045✔





1173



  for (int32_t i = pPool->num; i < pPool->max; ++i) {




11,638,953✔





1174



    pWorker = pPool->pWorkers + i;




10,299,908✔





1175



    pWorker->id = pPool->num;




10,299,908✔





1176



    pWorker->pool = pPool;




10,299,908✔





1177



    pPool->num++;




10,299,908✔





1178



    code = taosOpenQset(&pWorker->qset);




10,299,908✔





1179



    if (code != 0) break;




10,299,908✔





1180



    code = taosOpenQueue(&pWorker->queue);




10,299,908✔





1181



    if (code != 0) break;




10,299,908✔





1182



    taosSetQueueFp(pWorker->queue, fp, ahandle);




10,299,908✔





1183



    code = taosAddIntoQset(pWorker->qset, pWorker->queue, ahandle);




10,299,908✔





1184



    if (code != 0) break;




10,299,908✔





1185












1186



    TdThreadAttr thAttr;




10,291,308✔





1187



    (void)taosThreadAttrInit(&thAttr);




10,299,908✔





1188



    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);




10,299,908✔





1189












1190



    if (taosThreadCreate(&pWorker->thread, &thAttr, (ThreadFp)tDispatchWorkerThreadFp, pWorker) != 0) {




10,299,908✔





1191



      code = terrno;




×







1192



      (void)taosThreadAttrDestroy(&thAttr);




×







1193



      break;




×







1194



    }










1195



    (void)taosThreadAttrDestroy(&thAttr);




10,299,908✔





1196



    uInfo("worker:%s:%d is launched, threadId:%" PRId64 ", total:%d", pPool->name, pWorker->id, taosGetPthreadId(pWorker->thread), pPool->num);




10,299,908✔





1197



  }










1198












1199



  (void)taosThreadMutexUnlock(&pPool->poolLock);




1,339,045✔





1200



  if (code == 0) uInfo("worker:%s, queue:%p is allocated, ahandle:%p", pPool->name, pWorker->queue, ahandle);




1,339,045✔





1201



  return code;




1,339,045✔





1202



}










1203












1204



static void tDispatchWorkerFreeQueue(SDispatchWorkerPool *pPool) {




1,339,045✔





1205



  (void)taosThreadMutexLock(&pPool->poolLock);




1,339,045✔





1206



  if (!pPool->pWorkers) return;




1,339,045✔





1207



  for (int32_t i = 0; i < pPool->num; ++i) {




11,638,953✔





1208



    SDispatchWorker *pWorker = pPool->pWorkers + i;




10,299,908✔





1209



    if (pWorker->queue) {




10,299,908✔





1210



      taosCloseQueue(pWorker->queue);




10,299,908✔





1211



      pWorker->queue = NULL;




10,299,908✔





1212



    }










1213



    if (pWorker->qset) {




10,299,908✔





1214



      taosCloseQset(pWorker->qset);




10,299,908✔





1215



      pWorker->qset = NULL;




10,299,908✔





1216



    }










1217



  }










1218



  (void)taosThreadMutexUnlock(&pPool->poolLock);




1,339,045✔





1219



}










1220












1221



void tDispatchWorkerCleanup(SDispatchWorkerPool *pPool) {




1,339,045✔





1222



  (void)taosThreadMutexLock(&pPool->poolLock);




1,339,045✔





1223



  pPool->exit = true;




1,339,045✔





1224



  if (pPool->pWorkers) {




1,339,045✔





1225



    for (int32_t i = 0; i < pPool->num; ++i) {




11,638,953✔





1226



      SDispatchWorker *pWorker = pPool->pWorkers + i;




10,299,908✔





1227



      if (pWorker->qset) {




10,299,908✔





1228



        taosQsetThreadResume(pWorker->qset);




10,299,908✔





1229



      }










1230



    }










1231



  }










1232



  (void)taosThreadMutexUnlock(&pPool->poolLock);




1,339,045✔





1233












1234



  if (pPool->pWorkers) {




1,339,045✔





1235



    for (int32_t i = 0; i < pPool->num; ++i) {




11,638,953✔





1236



      SDispatchWorker *pWorker = pPool->pWorkers + i;




10,299,908✔





1237



      if (taosCheckPthreadValid(pWorker->thread)) {




10,299,908✔





1238



        (void)taosThreadJoin(pWorker->thread, NULL);




10,299,908✔





1239



        taosThreadClear(&pWorker->thread);




10,299,908✔





1240



      }










1241



    }










1242



  }










1243



  tDispatchWorkerFreeQueue(pPool);




1,339,045✔





1244



  taosMemoryFreeClear(pPool->pWorkers);




1,339,045✔





1245



  (void)taosThreadMutexDestroy(&pPool->poolLock);




1,339,045✔





1246



}




1,339,045✔





1247












1248



int32_t tAddTaskIntoDispatchWorkerPool(SDispatchWorkerPool *pPool, void *pMsg) {




100,272,711✔





1249



  int32_t code = 0;




100,272,711✔





1250



  int32_t idx = 0;




100,272,711✔





1251



  SDispatchWorker *pWorker = NULL;




100,275,150✔





1252



  (void)taosThreadMutexLock(&pPool->poolLock);




100,275,150✔





1253



  code = pPool->dispatchFp(pPool, pMsg, &idx);




100,287,305✔





1254



  if (code == 0) {




100,287,305✔





1255



    pWorker = pPool->pWorkers + idx;




100,287,305✔





1256



    if (pWorker->queue) {




100,287,305✔





1257



      code = taosWriteQitem(pWorker->queue, pMsg);




100,287,305✔





1258



    } else {










1259



      code = TSDB_CODE_INTERNAL_ERROR;




×







1260



    }










1261



  }










1262



  (void)taosThreadMutexUnlock(&pPool->poolLock);




100,287,305✔





1263



  if (code != 0) {




100,286,493✔





1264



    uError("worker:%s, failed to add task into dispatch worker pool, code:%d", pPool->name, code);




×







1265



  } else {










1266



    uDebug("msg %p dispatch to the %dth worker, threadId:%" PRId64, pMsg, idx, taosGetPthreadId(pWorker->thread));




100,286,493✔





1267



  }










1268



  return code;




100,286,493✔





1269



}
























    

  • 

    •