xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mnode/qnode support QueryAutoQWorkerPool

This commit is contained in:
wangjiaming0909 2024-06-28 15:11:44 +08:00
parent 26adfdd48d
commit 5040ff5bd9
16 changed files with 137 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/dnode/mnode/mnode.h
View File

@ -109,7 +109,7 @@ int64_t mndGetRoleTimeMs(SMnode *pMnode);
* @param pMsg The request msg.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t mndProcessRpcMsg(SRpcMsg *pMsg);
int32_t mndProcessRpcMsg(SRpcMsg *pMsg, SQueueInfo* pQueueInfo);
int32_t mndProcessSyncMsg(SRpcMsg *pMsg);
int32_t mndPreProcessQueryMsg(SRpcMsg *pMsg);
void mndPostProcessQueryMsg(SRpcMsg *pMsg);

2
include/dnode/qnode/qnode.h
View File

@ -60,7 +60,7 @@ int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad);
* @param pQnode The qnode object.
* @param pMsg The request message
*/
int32_t qndProcessQueryMsg(SQnode *pQnode, int64_t ts, SRpcMsg *pMsg);
int32_t qndProcessQueryMsg(SQnode *pQnode, SQueueInfo* pInfo, SRpcMsg *pMsg);
#ifdef __cplusplus
}

2
include/util/tqueue.h
View File

@ -49,7 +49,7 @@ typedef struct {
int32_t workerId;
int32_t threadNum;
int64_t timestamp;
void *poolCb;
void *workerCb;
} SQueueInfo;
typedef enum {

11
include/util/tworker.h
View File

@ -84,18 +84,25 @@ void tWWorkerCleanup(SWWorkerPool *pool);
STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp);
void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue);
typedef enum SQWorkerPoolType {
QWORKER_POOL = 0,
QUERY_AUTO_QWORKER_POOL,
} SQWorkerPoolType;
typedef struct {
const char *name;
int32_t min;
int32_t max;
FItem fp;
void *param;
SQWorkerPoolType poolType;
} SSingleWorkerCfg;
typedef struct {
const char *name;
STaosQueue *queue;
SQWorkerPool pool;
SQWorkerPoolType poolType; // default to QWORKER_POOL
void *pool;
} SSingleWorker;
typedef struct {
@ -134,6 +141,8 @@ typedef struct SQueryAutoQWorkerPool {
int32_t activeN; // running workers and workers waiting at reading new queue msg
int32_t runningN; // workers processing queue msgs, not include blocking/waitingA/waitingB workers.
TdThreadMutex activeLock;
int32_t waitingAfterBlockN; // workers that recovered from blocking but waiting for too many running workers
TdThreadMutex waitingAfterBlockLock;

3
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
View File

@ -53,7 +53,7 @@ static void mmProcessRpcMsg(SQueueInfo *pInfo, SRpcMsg *pMsg) {
const STraceId *trace = &pMsg->info.traceId;
dGTrace("msg:%p, get from mnode queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
int32_t code = mndProcessRpcMsg(pMsg);
int32_t code = mndProcessRpcMsg(pMsg, pInfo);
if (pInfo->timestamp != 0) {
int64_t cost = taosGetTimestampUs() - pInfo->timestamp;
@ -203,6 +203,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
.name = "mnode-query",
.fp = (FItem)mmProcessRpcMsg,
.param = pMgmt,
.poolType = QUERY_AUTO_QWORKER_POOL,
};
if (tSingleWorkerInit(&pMgmt->queryWorker, &qCfg) != 0) {
dError("failed to start mnode-query worker since %s", terrstr());

3
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
View File

@ -30,7 +30,7 @@ static void qmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
SQnodeMgmt *pMgmt = pInfo->ahandle;
dTrace("msg:%p, get from qnode queue", pMsg);
int32_t code = qndProcessQueryMsg(pMgmt->pQnode, pInfo->timestamp, pMsg);
int32_t code = qndProcessQueryMsg(pMgmt->pQnode, pInfo, pMsg);
if (IsReq(pMsg) && code != TSDB_CODE_ACTION_IN_PROGRESS) {
if (code != 0 && terrno != 0) code = terrno;
qmSendRsp(pMsg, code);
@ -105,6 +105,7 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
.name = "qnode-query",
.fp = (FItem)qmProcessQueue,
.param = pMgmt,
.poolType = QUERY_AUTO_QWORKER_POOL,
};
if (tSingleWorkerInit(&pMgmt->queryWorker, &queryCfg) != 0) {

3
source/dnode/mnode/impl/inc/mndInt.h
View File

@ -54,6 +54,7 @@ extern "C" {
#define SYSTABLE_SCH_COL_NAME_LEN ((TSDB_COL_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
typedef int32_t (*MndMsgFp)(SRpcMsg *pMsg);
typedef int32_t (*MndMsgFpExt)(SRpcMsg *pMsg, SQueueInfo* pInfo);
typedef int32_t (*MndInitFp)(SMnode *pMnode);
typedef void (*MndCleanupFp)(SMnode *pMnode);
typedef int32_t (*ShowRetrieveFp)(SRpcMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
@ -137,11 +138,13 @@ typedef struct SMnode {
SEncryptMgmt encryptMgmt;
SGrantInfo grant;
MndMsgFp msgFp[TDMT_MAX];
MndMsgFpExt msgFpExt[TDMT_MAX];
SMsgCb msgCb;
int64_t ipWhiteVer;
} SMnode;
void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp);
void mndSetMsgHandleExt(SMnode *pMnode, tmsg_t msgType, MndMsgFpExt fp);
int64_t mndGenerateUid(const char *name, int32_t len);
void mndSetRestored(SMnode *pMnode, bool restored);

19
source/dnode/mnode/impl/src/mndMain.c
View File

@ -844,21 +844,29 @@ _OVER:
return -1;
}
int32_t mndProcessRpcMsg(SRpcMsg *pMsg) {
int32_t mndProcessRpcMsg(SRpcMsg *pMsg, SQueueInfo* pQueueInfo) {
SMnode *pMnode = pMsg->info.node;
const STraceId *trace = &pMsg->info.traceId;
int32_t code = TSDB_CODE_SUCCESS;
MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(pMsg->msgType)];
MndMsgFpExt fpExt = NULL;
if (fp == NULL) {
fpExt = pMnode->msgFpExt[TMSG_INDEX(pMsg->msgType)];
if (fpExt == NULL) {
mGError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
return -1;
}
}
if (mndCheckMnodeState(pMsg) != 0) return -1;
mGTrace("msg:%p, start to process in mnode, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));
int32_t code = (*fp)(pMsg);
if (fp)
code = (*fp)(pMsg);
else
code = (*fpExt)(pMsg, pQueueInfo);
mndReleaseRpc(pMnode);
if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
@ -883,6 +891,13 @@ void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) {
}
}
void mndSetMsgHandleExt(SMnode *pMnode, tmsg_t msgType, MndMsgFpExt fp) {
tmsg_t type = TMSG_INDEX(msgType);
if (type < TDMT_MAX) {
pMnode->msgFpExt[type] = fp;
}
}
// Note: uid 0 is reserved
int64_t mndGenerateUid(const char *name, int32_t len) {
int32_t hashval = MurmurHash3_32(name, len);

20
source/dnode/mnode/impl/src/mndQuery.c
View File

@ -30,11 +30,11 @@ void mndPostProcessQueryMsg(SRpcMsg *pMsg) {
qWorkerAbortPreprocessQueryMsg(pMnode->pQuery, pMsg);
}
int32_t mndProcessQueryMsg(SRpcMsg *pMsg) {
int32_t mndProcessQueryMsg(SRpcMsg *pMsg, SQueueInfo* pInfo) {
int32_t code = -1;
SMnode *pMnode = pMsg->info.node;
SReadHandle handle = {.mnd = pMnode, .pMsgCb = &pMnode->msgCb};
SReadHandle handle = {.mnd = pMnode, .pMsgCb = &pMnode->msgCb, .pWorkerCb = pInfo->workerCb};
mTrace("msg:%p, in query queue is processing", pMsg);
switch (pMsg->msgType) {
@ -173,14 +173,14 @@ int32_t mndInitQuery(SMnode *pMnode) {
return -1;
}
mndSetMsgHandle(pMnode, TDMT_SCH_QUERY, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_SCH_MERGE_QUERY, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_SCH_QUERY_CONTINUE, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_SCH_FETCH, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_SCH_MERGE_FETCH, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_SCH_TASK_NOTIFY, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_SCH_DROP_TASK, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_SCH_QUERY_HEARTBEAT, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_MERGE_QUERY, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY_CONTINUE, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_FETCH, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_MERGE_FETCH, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_TASK_NOTIFY, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_DROP_TASK, mndProcessQueryMsg);
mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY_HEARTBEAT, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_MND_BATCH_META, mndProcessBatchMetaMsg);
return 0;

2
source/dnode/qnode/inc/qndInt.h
View File

@ -29,7 +29,7 @@
extern "C" {
#endif
typedef struct SQueueWorker SQHandle;
typedef struct SQWorker SQHandle;
typedef struct SQnode {
int32_t qndId;

7
source/dnode/qnode/src/qnode.c
View File

@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tqueue.h"
#include "executor.h"
#include "qndInt.h"
#include "query.h"
@ -24,6 +25,7 @@ SQnode *qndOpen(const SQnodeOpt *pOption) {
qError("calloc SQnode failed");
return NULL;
}
pQnode->qndId = QNODE_HANDLE;
if (qWorkerInit(NODE_TYPE_QNODE, pQnode->qndId, (void **)&pQnode->pQuery, &pOption->msgCb)) {
taosMemoryFreeClear(pQnode);
@ -72,9 +74,10 @@ int32_t qndPreprocessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) {
return qWorkerPreprocessQueryMsg(pQnode->pQuery, pMsg, false);
}
int32_t qndProcessQueryMsg(SQnode *pQnode, int64_t ts, SRpcMsg *pMsg) {
int32_t qndProcessQueryMsg(SQnode *pQnode, SQueueInfo* pInfo, SRpcMsg *pMsg) {
int32_t code = -1;
SReadHandle handle = {.pMsgCb = &pQnode->msgCb};
int64_t ts = pInfo->timestamp;
SReadHandle handle = {.pMsgCb = &pQnode->msgCb, .pWorkerCb = pInfo->workerCb};
qTrace("message in qnode queue is processing");
switch (pMsg->msgType) {

2
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -747,7 +747,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo* pInfo) {
return 0;
}
SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb, .pWorkerCb = pInfo->poolCb};
SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb, .pWorkerCb = pInfo->workerCb};
initStorageAPI(&handle.api);
switch (pMsg->msgType) {

1
source/libs/audit/src/auditMain.c
View File

@ -22,7 +22,6 @@
#include "ttime.h"
#include "tjson.h"
#include "tglobal.h"
#include "mnode.h"
#include "audit.h"
#include "osMemory.h"

15
source/libs/executor/src/exchangeoperator.c
View File

@ -976,14 +976,21 @@ int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDa
static int32_t exchangeWait(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo) {
SExecTaskInfo* pTask = pOperator->pTaskInfo;
int32_t code = TSDB_CODE_SUCCESS;
if (pTask->pWorkerCb) {
pTask->code = pTask->pWorkerCb->beforeBlocking(pTask->pWorkerCb->pPool);
if (pTask->code != TSDB_CODE_SUCCESS) return pTask->code;
code = pTask->pWorkerCb->beforeBlocking(pTask->pWorkerCb->pPool);
if (code != TSDB_CODE_SUCCESS) {
pTask->code = code;
return pTask->code;
}
}
tsem_wait(&pExchangeInfo->ready);
if (pTask->pWorkerCb) {
pTask->code = pTask->pWorkerCb->afterRecoverFromBlocking(pTask->pWorkerCb->pPool);
if (pTask->code != TSDB_CODE_SUCCESS) return pTask->code;
code = pTask->pWorkerCb->afterRecoverFromBlocking(pTask->pWorkerCb->pPool);
if (code != TSDB_CODE_SUCCESS) {
pTask->code = code;
return pTask->code;
}
}
return TSDB_CODE_SUCCESS;
}

1
source/libs/executor/src/executor.c
View File

@ -794,6 +794,7 @@ void qStopTaskOperators(SExecTaskInfo* pTaskInfo) {
SExchangeOpStopInfo* pStop = taosArrayGet(pTaskInfo->stopInfo.pStopInfo, i);
SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pStop->refId);
if (pExchangeInfo) {
qDebug("%s stop exchange operator", GET_TASKID(pTaskInfo));
tsem_post(&pExchangeInfo->ready);
taosReleaseRef(exchangeObjRefPool, pStop->refId);
}

63
source/util/src/tworker.c
View File

@ -433,28 +433,66 @@ void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) {
}
int32_t tSingleWorkerInit(SSingleWorker *pWorker, const SSingleWorkerCfg *pCfg) {
SQWorkerPool *pPool = &pWorker->pool;
pWorker->poolType = pCfg->poolType;
pWorker->name = pCfg->name;
switch (pCfg->poolType) {
case QWORKER_POOL: {
SQWorkerPool *pPool = taosMemoryCalloc(1, sizeof(SQWorkerPool));
if (!pPool) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pPool->name = pCfg->name;
pPool->min = pCfg->min;
pPool->max = pCfg->max;
pWorker->pool = pPool;
if (tQWorkerInit(pPool) != 0) return -1;
pWorker->queue = tQWorkerAllocQueue(pPool, pCfg->param, pCfg->fp);
if (pWorker->queue == NULL) return -1;
} break;
case QUERY_AUTO_QWORKER_POOL: {
SQueryAutoQWorkerPool *pPool = taosMemoryCalloc(1, sizeof(SQueryAutoQWorkerPool));
if (!pPool) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pPool->name = pCfg->name;
pPool->min = pCfg->min;
pPool->max = pCfg->max;
pWorker->pool = pPool;
if (tQueryAutoQWorkerInit(pPool) != 0) return -1;
pWorker->name = pCfg->name;
pWorker->queue = tQueryAutoQWorkerAllocQueue(pPool, pCfg->param, pCfg->fp);
if (!pWorker->queue) return -1;
} break;
default:
assert(0);
}
return 0;
}
void tSingleWorkerCleanup(SSingleWorker *pWorker) {
if (pWorker->queue == NULL) return;
while (!taosQueueEmpty(pWorker->queue)) {
taosMsleep(10);
}
tQWorkerCleanup(&pWorker->pool);
tQWorkerFreeQueue(&pWorker->pool, pWorker->queue);
switch (pWorker->poolType) {
case QWORKER_POOL:
tQWorkerCleanup(pWorker->pool);
tQWorkerFreeQueue(pWorker->pool, pWorker->queue);
taosMemoryFree(pWorker->pool);
break;
case QUERY_AUTO_QWORKER_POOL:
tQueryAutoQWorkerCleanup(pWorker->pool);
tQueryAutoQWorkerFreeQueue(pWorker->pool, pWorker->queue);
taosMemoryFree(pWorker->pool);
break;
default:
assert(0);
}
}
int32_t tMultiWorkerInit(SMultiWorker *pWorker, const SMultiWorkerCfg *pCfg) {
@ -517,7 +555,7 @@ static void *tQueryAutoQWorkerThreadFp(SQueryAutoQWorker *worker) {
if (qinfo.fp != NULL) {
qinfo.workerId = worker->id;
qinfo.threadNum = pool->num;
qinfo.poolCb = pool->pCb;
qinfo.workerCb = pool->pCb;
(*((FItem)qinfo.fp))(&qinfo, msg);
}
@ -573,33 +611,38 @@ static bool tQueryAutoQWorkerTrySignalWaitingBeforeProcess(void* p) {
static bool tQueryAutoQWorkerTryDecActive(void* p, int32_t minActive) {
SQueryAutoQWorkerPool *pPool = p;
bool ret = false;
taosThreadMutexLock(&pPool->activeLock);
int32_t active = pPool->activeN;
while (active > minActive) {
int32_t activeNew = atomic_val_compare_exchange_32(&pPool->activeN, active, active - 1);
if (activeNew == active) {
atomic_fetch_sub_32(&pPool->runningN, 1);
ret = true;
break;
}
active = activeNew;
}
atomic_fetch_sub_32(&pPool->runningN, 1);
taosThreadMutexUnlock(&pPool->activeLock);
return ret;
}
static int32_t tQueryAutoQWorkerWaitingCheck(SQueryAutoQWorkerPool* pPool) {
taosThreadMutexLock(&pPool->activeLock);
int32_t running = pPool->runningN;
while (running < pPool->num) {
int32_t runningNew = atomic_val_compare_exchange_32(&pPool->runningN, running, running + 1);
if (runningNew == running) {
// to running
taosThreadMutexUnlock(&pPool->activeLock);
return TSDB_CODE_SUCCESS;
}
running = runningNew;
}
atomic_fetch_sub_32(&pPool->activeN, 1);
taosThreadMutexUnlock(&pPool->activeLock);
// to wait for process
taosThreadMutexLock(&pPool->waitingBeforeProcessMsgLock);
atomic_fetch_add_32(&pPool->waitingBeforeProcessMsgN, 1);
atomic_fetch_sub_32(&pPool->activeN, 1);
if (!pPool->exit) taosThreadCondWait(&pPool->waitingBeforeProcessMsgCond, &pPool->waitingBeforeProcessMsgLock);
// recovered from waiting
taosThreadMutexUnlock(&pPool->waitingBeforeProcessMsgLock);
@ -650,7 +693,6 @@ bool tQueryAutoQWorkerTryRecycleWorker(SQueryAutoQWorkerPool* pPool, SQueryAutoQ
return true;
} else {
atomic_fetch_sub_32(&pPool->runningN, 1);
return true;
}
}
@ -670,6 +712,7 @@ int32_t tQueryAutoQWorkerInit(SQueryAutoQWorkerPool *pool) {
(void)taosThreadMutexInit(&pool->backupLock, NULL);
(void)taosThreadMutexInit(&pool->waitingAfterBlockLock, NULL);
(void)taosThreadMutexInit(&pool->waitingBeforeProcessMsgLock, NULL);
(void)taosThreadMutexInit(&pool->activeLock, NULL);
(void)taosThreadCondInit(&pool->waitingBeforeProcessMsgCond, NULL);
(void)taosThreadCondInit(&pool->waitingAfterBlockCond, NULL);
@ -757,6 +800,7 @@ void tQueryAutoQWorkerCleanup(SQueryAutoQWorkerPool *pPool) {
taosThreadMutexDestroy(&pPool->backupLock);
taosThreadMutexDestroy(&pPool->waitingAfterBlockLock);
taosThreadMutexDestroy(&pPool->waitingBeforeProcessMsgLock);
taosThreadMutexDestroy(&pPool->activeLock);
taosThreadCondDestroy(&pPool->backupCond);
taosThreadCondDestroy(&pPool->waitingAfterBlockCond);
@ -866,7 +910,6 @@ static int32_t tQueryAutoQWorkerBeforeBlocking(void *p) {
if (code != TSDB_CODE_SUCCESS) {
return code;
}
atomic_fetch_sub_32(&pPool->runningN, 1);
}
return TSDB_CODE_SUCCESS;