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

refact worker

This commit is contained in:
Shengliang Guan 2022-03-22 16:51:30 +08:00
parent f48fddc7c4
commit 2af212f42d
14 changed files with 84 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
include/util/tworker.h
View File

@ -74,31 +74,31 @@ typedef struct {
int32_t maxNum; int32_t maxNum;
FItem fp; FItem fp;
void *param; void *param;
} SQWorkerAllCfg; } SSingleWorkerCfg;
typedef struct { typedef struct {
const char *name; const char *name;
STaosQueue *queue; STaosQueue *queue;
SQWorkerPool pool; SQWorkerPool pool;
} SQWorkerAll; } SSingleWorker;
typedef struct { typedef struct {
const char *name; const char *name;
int32_t maxNum; int32_t maxNum;
FItems fp; FItems fp;
void *param; void *param;
} SWWorkerAllCfg; } SMultiWorkerCfg;
typedef struct { typedef struct {
const char *name; const char *name;
STaosQueue *queue; STaosQueue *queue;
SWWorkerPool pool; SWWorkerPool pool;
} SWWorkerAll; } SMultiWorker;
int32_t tQWorkerAllInit(SQWorkerAll *pWorker, const SQWorkerAllCfg *pCfg); int32_t tSingleWorkerInit(SSingleWorker *pWorker, const SSingleWorkerCfg *pCfg);
void tQWorkerAllCleanup(SQWorkerAll *pWorker); void tSingleWorkerCleanup(SSingleWorker *pWorker);
int32_t tWWorkerAllInit(SWWorkerAll *pWorker, const SWWorkerAllCfg *pCfg); int32_t tMultiWorkerInit(SMultiWorker *pWorker, const SMultiWorkerCfg *pCfg);
void tWWorkerAllCleanup(SWWorkerAll *pWorker); void tMultiWorkerCleanup(SMultiWorker *pWorker);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/dnode/mgmt/bnode/inc/bmInt.h
View File

@ -28,7 +28,7 @@ typedef struct SBnodeMgmt {
SDnode *pDnode; SDnode *pDnode;
SMgmtWrapper *pWrapper; SMgmtWrapper *pWrapper;
const char *path; const char *path;
SWWorkerAll writeWorker; SMultiWorker writeWorker;
} SBnodeMgmt; } SBnodeMgmt;
// bmInt.c // bmInt.c

8
source/dnode/mgmt/bnode/src/bmWorker.c
View File

@ -64,15 +64,15 @@ static void bmProcessQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs
} }
int32_t bmProcessWriteMsg(SBnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t bmProcessWriteMsg(SBnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SWWorkerAll *pWorker = &pMgmt->writeWorker; SMultiWorker *pWorker = &pMgmt->writeWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
return taosWriteQitem(pWorker->queue, pMsg); return taosWriteQitem(pWorker->queue, pMsg);
} }
int32_t bmStartWorker(SBnodeMgmt *pMgmt) { int32_t bmStartWorker(SBnodeMgmt *pMgmt) {
SWWorkerAllCfg cfg = {.maxNum = 1, .name = "bnode-write", .fp = (FItems)bmProcessQueue, .param = pMgmt}; SMultiWorkerCfg cfg = {.maxNum = 1, .name = "bnode-write", .fp = (FItems)bmProcessQueue, .param = pMgmt};
if (tWWorkerAllInit(&pMgmt->writeWorker, &cfg) != 0) { if (tMultiWorkerInit(&pMgmt->writeWorker, &cfg) != 0) {
dError("failed to start bnode write worker since %s", terrstr()); dError("failed to start bnode write worker since %s", terrstr());
return -1; return -1;
} }
@ -80,4 +80,4 @@ int32_t bmStartWorker(SBnodeMgmt *pMgmt) {
return 0; return 0;
} }
void bmStopWorker(SBnodeMgmt *pMgmt) { tWWorkerAllCleanup(&pMgmt->writeWorker); } void bmStopWorker(SBnodeMgmt *pMgmt) { tMultiWorkerCleanup(&pMgmt->writeWorker); }

4
source/dnode/mgmt/dnode/inc/dmInt.h
View File

@ -31,8 +31,8 @@ typedef struct SDnodeMgmt {
SArray *dnodeEps; SArray *dnodeEps;
TdThread *threadId; TdThread *threadId;
SRWLatch latch; SRWLatch latch;
SQWorkerAll mgmtWorker; SSingleWorker mgmtWorker;
SQWorkerAll statusWorker; SSingleWorker statusWorker;
const char *path; const char *path;
SDnode *pDnode; SDnode *pDnode;
SMgmtWrapper *pWrapper; SMgmtWrapper *pWrapper;

14
source/dnode/mgmt/dnode/src/dmWorker.c
View File

@ -100,16 +100,16 @@ static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
} }
int32_t dmStartWorker(SDnodeMgmt *pMgmt) { int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
SQWorkerAllCfg mgmtCfg = { SSingleWorkerCfg mgmtCfg = {
.minNum = 1, .maxNum = 1, .name = "dnode-mgmt", .fp = (FItem)dmProcessQueue, .param = pMgmt}; .minNum = 1, .maxNum = 1, .name = "dnode-mgmt", .fp = (FItem)dmProcessQueue, .param = pMgmt};
if (tQWorkerAllInit(&pMgmt->mgmtWorker, &mgmtCfg) != 0) { if (tSingleWorkerInit(&pMgmt->mgmtWorker, &mgmtCfg) != 0) {
dError("failed to start dnode mgmt worker since %s", terrstr()); dError("failed to start dnode mgmt worker since %s", terrstr());
return -1; return -1;
} }
SQWorkerAllCfg statusCfg = { SSingleWorkerCfg statusCfg = {
.minNum = 1, .maxNum = 1, .name = "dnode-status", .fp = (FItem)dmProcessQueue, .param = pMgmt}; .minNum = 1, .maxNum = 1, .name = "dnode-status", .fp = (FItem)dmProcessQueue, .param = pMgmt};
if (tQWorkerAllInit(&pMgmt->statusWorker, &statusCfg) != 0) { if (tSingleWorkerInit(&pMgmt->statusWorker, &statusCfg) != 0) {
dError("failed to start dnode status worker since %s", terrstr()); dError("failed to start dnode status worker since %s", terrstr());
return -1; return -1;
} }
@ -129,8 +129,8 @@ int32_t dmStartThread(SDnodeMgmt *pMgmt) {
} }
void dmStopWorker(SDnodeMgmt *pMgmt) { void dmStopWorker(SDnodeMgmt *pMgmt) {
tQWorkerAllCleanup(&pMgmt->mgmtWorker); tSingleWorkerCleanup(&pMgmt->mgmtWorker);
tQWorkerAllCleanup(&pMgmt->statusWorker); tSingleWorkerCleanup(&pMgmt->statusWorker);
if (pMgmt->threadId != NULL) { if (pMgmt->threadId != NULL) {
taosDestoryThread(pMgmt->threadId); taosDestoryThread(pMgmt->threadId);
@ -139,7 +139,7 @@ void dmStopWorker(SDnodeMgmt *pMgmt) {
} }
int32_t dmProcessMgmtMsg(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t dmProcessMgmtMsg(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SQWorkerAll *pWorker = &pMgmt->mgmtWorker; SSingleWorker *pWorker = &pMgmt->mgmtWorker;
if (pMsg->rpcMsg.msgType == TDMT_MND_STATUS_RSP) { if (pMsg->rpcMsg.msgType == TDMT_MND_STATUS_RSP) {
pWorker = &pMgmt->statusWorker; pWorker = &pMgmt->statusWorker;
} }

6
source/dnode/mgmt/mnode/inc/mmInt.h
View File

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

18
source/dnode/mgmt/mnode/src/mmWorker.c
View File

@ -44,7 +44,7 @@ static void mmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
} }
static int32_t mmPutMsgToWorker(SMnodeMgmt *pMgmt, SQWorkerAll *pWorker, SNodeMsg *pMsg) { static int32_t mmPutMsgToWorker(SMnodeMgmt *pMgmt, SSingleWorker *pWorker, SNodeMsg *pMsg) {
dTrace("msg:%p, put into worker %s", pMsg, pWorker->name); dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
return taosWriteQitem(pWorker->queue, pMsg); return taosWriteQitem(pWorker->queue, pMsg);
} }
@ -61,7 +61,7 @@ int32_t mmProcessReadMsg(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
return mmPutMsgToWorker(pMgmt, &pMgmt->readWorker, pMsg); return mmPutMsgToWorker(pMgmt, &pMgmt->readWorker, pMsg);
} }
static int32_t mmPutRpcMsgToWorker(SMnodeMgmt *pMgmt, SQWorkerAll *pWorker, SRpcMsg *pRpc) { static int32_t mmPutRpcMsgToWorker(SMnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) {
SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg)); SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
if (pMsg == NULL) { if (pMsg == NULL) {
return -1; return -1;
@ -91,19 +91,19 @@ int32_t mmPutMsgToReadQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
} }
int32_t mmStartWorker(SMnodeMgmt *pMgmt) { int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
SQWorkerAllCfg cfg = {.minNum = 0, .maxNum = 1, .name = "mnode-read", .fp = (FItem)mmProcessQueue, .param = pMgmt}; SSingleWorkerCfg cfg = {.minNum = 0, .maxNum = 1, .name = "mnode-read", .fp = (FItem)mmProcessQueue, .param = pMgmt};
if (tQWorkerAllInit(&pMgmt->readWorker, &cfg) != 0) { if (tSingleWorkerInit(&pMgmt->readWorker, &cfg) != 0) {
dError("failed to start mnode-read worker since %s", terrstr()); dError("failed to start mnode-read worker since %s", terrstr());
return -1; return -1;
} }
if (tQWorkerAllInit(&pMgmt->writeWorker, &cfg) != 0) { if (tSingleWorkerInit(&pMgmt->writeWorker, &cfg) != 0) {
dError("failed to start mnode-write worker since %s", terrstr()); dError("failed to start mnode-write worker since %s", terrstr());
return -1; return -1;
} }
if (tQWorkerAllInit(&pMgmt->syncWorker, &cfg) != 0) { if (tSingleWorkerInit(&pMgmt->syncWorker, &cfg) != 0) {
dError("failed to start mnode sync-worker since %s", terrstr()); dError("failed to start mnode sync-worker since %s", terrstr());
return -1; return -1;
} }
@ -112,7 +112,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
} }
void mmStopWorker(SMnodeMgmt *pMgmt) { void mmStopWorker(SMnodeMgmt *pMgmt) {
tQWorkerAllCleanup(&pMgmt->readWorker); tSingleWorkerCleanup(&pMgmt->readWorker);
tQWorkerAllCleanup(&pMgmt->writeWorker); tSingleWorkerCleanup(&pMgmt->writeWorker);
tQWorkerAllCleanup(&pMgmt->syncWorker); tSingleWorkerCleanup(&pMgmt->syncWorker);
} }

4
source/dnode/mgmt/qnode/inc/qmInt.h
View File

@ -28,8 +28,8 @@ typedef struct SQnodeMgmt {
SDnode *pDnode; SDnode *pDnode;
SMgmtWrapper *pWrapper; SMgmtWrapper *pWrapper;
const char *path; const char *path;
SQWorkerAll queryWorker; SSingleWorker queryWorker;
SQWorkerAll fetchWorker; SSingleWorker fetchWorker;
} SQnodeMgmt; } SQnodeMgmt;
// qmInt.c // qmInt.c

32
source/dnode/mgmt/qnode/src/qmWorker.c
View File

@ -49,7 +49,7 @@ static void qmProcessFetchQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
} }
static int32_t qmPutMsgToWorker(SQWorkerAll *pWorker, SNodeMsg *pMsg) { static int32_t qmPutMsgToWorker(SSingleWorker *pWorker, SNodeMsg *pMsg) {
dTrace("msg:%p, put into worker %s", pMsg, pWorker->name); dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
return taosWriteQitem(pWorker->queue, pMsg); return taosWriteQitem(pWorker->queue, pMsg);
} }
@ -58,7 +58,7 @@ int32_t qmProcessQueryMsg(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { return qmPutMsgTo
int32_t qmProcessFetchMsg(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { return qmPutMsgToWorker(&pMgmt->fetchWorker, pMsg); } int32_t qmProcessFetchMsg(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { return qmPutMsgToWorker(&pMgmt->fetchWorker, pMsg); }
static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SQWorkerAll *pWorker, SRpcMsg *pRpc) { static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) {
SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg)); SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
if (pMsg == NULL) { if (pMsg == NULL) {
return -1; return -1;
@ -110,24 +110,24 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
int32_t minQueryThreads = TMAX((int32_t)(tsNumOfCores * tsRatioOfQueryCores), 1); int32_t minQueryThreads = TMAX((int32_t)(tsNumOfCores * tsRatioOfQueryCores), 1);
int32_t maxQueryThreads = minQueryThreads; int32_t maxQueryThreads = minQueryThreads;
SQWorkerAllCfg queryCfg = {.minNum = minQueryThreads, SSingleWorkerCfg queryCfg = {.minNum = minQueryThreads,
.maxNum = maxQueryThreads, .maxNum = maxQueryThreads,
.name = "qnode-query", .name = "qnode-query",
.fp = (FItem)qmProcessQueryQueue, .fp = (FItem)qmProcessQueryQueue,
.param = pMgmt}; .param = pMgmt};
if (tQWorkerAllInit(&pMgmt->queryWorker, &queryCfg) != 0) { if (tSingleWorkerInit(&pMgmt->queryWorker, &queryCfg) != 0) {
dError("failed to start qnode-query worker since %s", terrstr()); dError("failed to start qnode-query worker since %s", terrstr());
return -1; return -1;
} }
SQWorkerAllCfg fetchCfg = {.minNum = minFetchThreads, SSingleWorkerCfg fetchCfg = {.minNum = minFetchThreads,
.maxNum = maxFetchThreads, .maxNum = maxFetchThreads,
.name = "qnode-fetch", .name = "qnode-fetch",
.fp = (FItem)qmProcessFetchQueue, .fp = (FItem)qmProcessFetchQueue,
.param = pMgmt}; .param = pMgmt};
if (tQWorkerAllInit(&pMgmt->fetchWorker, &fetchCfg) != 0) { if (tSingleWorkerInit(&pMgmt->fetchWorker, &fetchCfg) != 0) {
dError("failed to start qnode-fetch worker since %s", terrstr()); dError("failed to start qnode-fetch worker since %s", terrstr());
return -1; return -1;
} }
@ -136,6 +136,6 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
} }
void qmStopWorker(SQnodeMgmt *pMgmt) { void qmStopWorker(SQnodeMgmt *pMgmt) {
tQWorkerAllCleanup(&pMgmt->queryWorker); tSingleWorkerCleanup(&pMgmt->queryWorker);
tQWorkerAllCleanup(&pMgmt->fetchWorker); tSingleWorkerCleanup(&pMgmt->fetchWorker);
} }

4
source/dnode/mgmt/snode/inc/smInt.h
View File

@ -30,8 +30,8 @@ typedef struct SSnodeMgmt {
const char *path; const char *path;
SRWLatch latch; SRWLatch latch;
int8_t uniqueWorkerInUse; int8_t uniqueWorkerInUse;
SArray *uniqueWorkers; // SArray<SWWorkerAll*> SArray *uniqueWorkers; // SArray<SMultiWorker*>
SQWorkerAll sharedWorker; SSingleWorker sharedWorker;
} SSnodeMgmt; } SSnodeMgmt;
// smInt.c // smInt.c

34
source/dnode/mgmt/snode/src/smWorker.c
View File

@ -44,22 +44,22 @@ static void smProcessSharedQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
} }
int32_t smStartWorker(SSnodeMgmt *pMgmt) { int32_t smStartWorker(SSnodeMgmt *pMgmt) {
pMgmt->uniqueWorkers = taosArrayInit(0, sizeof(SWWorkerAll *)); pMgmt->uniqueWorkers = taosArrayInit(0, sizeof(SMultiWorker *));
if (pMgmt->uniqueWorkers == NULL) { if (pMgmt->uniqueWorkers == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
for (int32_t i = 0; i < SND_UNIQUE_THREAD_NUM; i++) { for (int32_t i = 0; i < SND_UNIQUE_THREAD_NUM; i++) {
SWWorkerAll *pUniqueWorker = malloc(sizeof(SWWorkerAll)); SMultiWorker *pUniqueWorker = malloc(sizeof(SMultiWorker));
if (pUniqueWorker == NULL) { if (pUniqueWorker == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
SWWorkerAllCfg cfg = {.maxNum = 1, .name = "snode-unique", .fp = smProcessUniqueQueue, .param = pMgmt}; SMultiWorkerCfg cfg = {.maxNum = 1, .name = "snode-unique", .fp = smProcessUniqueQueue, .param = pMgmt};
if (tWWorkerAllInit(pUniqueWorker, &cfg) != 0) { if (tMultiWorkerInit(pUniqueWorker, &cfg) != 0) {
dError("failed to start snode-unique worker since %s", terrstr()); dError("failed to start snode-unique worker since %s", terrstr());
return -1; return -1;
} }
@ -69,13 +69,13 @@ int32_t smStartWorker(SSnodeMgmt *pMgmt) {
} }
} }
SQWorkerAllCfg cfg = {.minNum = SND_SHARED_THREAD_NUM, SSingleWorkerCfg cfg = {.minNum = SND_SHARED_THREAD_NUM,
.maxNum = SND_SHARED_THREAD_NUM, .maxNum = SND_SHARED_THREAD_NUM,
.name = "snode-shared", .name = "snode-shared",
.fp = (FItem)smProcessSharedQueue, .fp = (FItem)smProcessSharedQueue,
.param = pMgmt}; .param = pMgmt};
if (tQWorkerAllInit(&pMgmt->sharedWorker, &cfg)) { if (tSingleWorkerInit(&pMgmt->sharedWorker, &cfg)) {
dError("failed to start snode shared-worker since %s", terrstr()); dError("failed to start snode shared-worker since %s", terrstr());
return -1; return -1;
} }
@ -85,11 +85,11 @@ int32_t smStartWorker(SSnodeMgmt *pMgmt) {
void smStopWorker(SSnodeMgmt *pMgmt) { void smStopWorker(SSnodeMgmt *pMgmt) {
for (int32_t i = 0; i < taosArrayGetSize(pMgmt->uniqueWorkers); i++) { for (int32_t i = 0; i < taosArrayGetSize(pMgmt->uniqueWorkers); i++) {
SWWorkerAll *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, i); SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, i);
tWWorkerAllCleanup(pWorker); tMultiWorkerCleanup(pWorker);
} }
taosArrayDestroy(pMgmt->uniqueWorkers); taosArrayDestroy(pMgmt->uniqueWorkers);
tQWorkerAllCleanup(&pMgmt->sharedWorker); tSingleWorkerCleanup(&pMgmt->sharedWorker);
} }
static FORCE_INLINE int32_t smGetSWIdFromMsg(SRpcMsg *pMsg) { static FORCE_INLINE int32_t smGetSWIdFromMsg(SRpcMsg *pMsg) {
@ -105,7 +105,7 @@ static FORCE_INLINE int32_t smGetSWTypeFromMsg(SRpcMsg *pMsg) {
} }
int32_t smProcessMgmtMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t smProcessMgmtMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SWWorkerAll *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, 0); SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, 0);
if (pWorker == NULL) { if (pWorker == NULL) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
return -1; return -1;
@ -116,8 +116,8 @@ int32_t smProcessMgmtMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
} }
int32_t smProcessUniqueMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t smProcessUniqueMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
int32_t index = smGetSWIdFromMsg(&pMsg->rpcMsg); int32_t index = smGetSWIdFromMsg(&pMsg->rpcMsg);
SWWorkerAll *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, index); SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, index);
if (pWorker == NULL) { if (pWorker == NULL) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
return -1; return -1;
@ -128,7 +128,7 @@ int32_t smProcessUniqueMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
} }
int32_t smProcessSharedMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t smProcessSharedMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SQWorkerAll *pWorker = &pMgmt->sharedWorker; SSingleWorker *pWorker = &pMgmt->sharedWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
return taosWriteQitem(pWorker->queue, pMsg); return taosWriteQitem(pWorker->queue, pMsg);

2
source/dnode/mgmt/vnode/inc/vmInt.h
View File

@ -36,7 +36,7 @@ typedef struct SVnodesMgmt {
const char *path; const char *path;
SDnode *pDnode; SDnode *pDnode;
SMgmtWrapper *pWrapper; SMgmtWrapper *pWrapper;
SQWorkerAll mgmtWorker; SSingleWorker mgmtWorker;
} SVnodesMgmt; } SVnodesMgmt;
typedef struct { typedef struct {

24
source/dnode/mgmt/vnode/src/vmWorker.c
View File

@ -200,24 +200,16 @@ static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueTyp
return code; return code;
} }
int32_t vmProcessSyncMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { int32_t vmProcessSyncMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { return vmPutNodeMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); }
return vmPutNodeMsgToQueue(pMgmt, pMsg, SYNC_QUEUE);
}
int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { return vmPutNodeMsgToQueue(pMgmt, pMsg, WRITE_QUEUE); }
return vmPutNodeMsgToQueue(pMgmt, pMsg, WRITE_QUEUE);
}
int32_t vmProcessQueryMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { int32_t vmProcessQueryMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { return vmPutNodeMsgToQueue(pMgmt, pMsg, QUERY_QUEUE); }
return vmPutNodeMsgToQueue(pMgmt, pMsg, QUERY_QUEUE);
}
int32_t vmProcessFetchMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { int32_t vmProcessFetchMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { return vmPutNodeMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); }
return vmPutNodeMsgToQueue(pMgmt, pMsg, FETCH_QUEUE);
}
int32_t vmProcessMgmtMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { int32_t vmProcessMgmtMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SQWorkerAll *pWorker = &pMgmt->mgmtWorker; SSingleWorker *pWorker = &pMgmt->mgmtWorker;
dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name); dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
return taosWriteQitem(pWorker->queue, pMsg); return taosWriteQitem(pWorker->queue, pMsg);
} }
@ -357,9 +349,9 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
pWPool->max = maxSyncThreads; pWPool->max = maxSyncThreads;
if (tWWorkerInit(pWPool) != 0) return -1; if (tWWorkerInit(pWPool) != 0) return -1;
SQWorkerAllCfg cfg = { SSingleWorkerCfg cfg = {
.minNum = 1, .maxNum = 1, .name = "vnode-mgmt", .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt}; .minNum = 1, .maxNum = 1, .name = "vnode-mgmt", .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt};
if (tQWorkerAllInit(&pMgmt->mgmtWorker, &cfg) != 0) { if (tSingleWorkerInit(&pMgmt->mgmtWorker, &cfg) != 0) {
dError("failed to start vnode-mgmt worker since %s", terrstr()); dError("failed to start vnode-mgmt worker since %s", terrstr());
return -1; return -1;
} }
@ -369,7 +361,7 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
} }
void vmStopWorker(SVnodesMgmt *pMgmt) { void vmStopWorker(SVnodesMgmt *pMgmt) {
tQWorkerAllCleanup(&pMgmt->mgmtWorker); tSingleWorkerCleanup(&pMgmt->mgmtWorker);
tQWorkerCleanup(&pMgmt->fetchPool); tQWorkerCleanup(&pMgmt->fetchPool);
tQWorkerCleanup(&pMgmt->queryPool); tQWorkerCleanup(&pMgmt->queryPool);
tWWorkerCleanup(&pMgmt->writePool); tWWorkerCleanup(&pMgmt->writePool);

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

@ -284,7 +284,7 @@ void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) {
uDebug("worker:%s, queue:%p is freed", pool->name, queue); uDebug("worker:%s, queue:%p is freed", pool->name, queue);
} }
int32_t tQWorkerAllInit(SQWorkerAll *pWorker, const SQWorkerAllCfg *pCfg) { int32_t tSingleWorkerInit(SSingleWorker *pWorker, const SSingleWorkerCfg *pCfg) {
SQWorkerPool *pPool = &pWorker->pool; SQWorkerPool *pPool = &pWorker->pool;
pPool->name = pCfg->name; pPool->name = pCfg->name;
pPool->min = pCfg->minNum; pPool->min = pCfg->minNum;
@ -302,7 +302,7 @@ int32_t tQWorkerAllInit(SQWorkerAll *pWorker, const SQWorkerAllCfg *pCfg) {
return 0; return 0;
} }
void tQWorkerAllCleanup(SQWorkerAll *pWorker) { void tSingleWorkerCleanup(SSingleWorker *pWorker) {
if (pWorker->queue == NULL) return; if (pWorker->queue == NULL) return;
while (!taosQueueEmpty(pWorker->queue)) { while (!taosQueueEmpty(pWorker->queue)) {
@ -313,7 +313,7 @@ void tQWorkerAllCleanup(SQWorkerAll *pWorker) {
tQWorkerFreeQueue(&pWorker->pool, pWorker->queue); tQWorkerFreeQueue(&pWorker->pool, pWorker->queue);
} }
int32_t tWWorkerAllInit(SWWorkerAll *pWorker, const SWWorkerAllCfg *pCfg) { int32_t tMultiWorkerInit(SMultiWorker *pWorker, const SMultiWorkerCfg *pCfg) {
SWWorkerPool *pPool = &pWorker->pool; SWWorkerPool *pPool = &pWorker->pool;
pPool->name = pCfg->name; pPool->name = pCfg->name;
pPool->max = pCfg->maxNum; pPool->max = pCfg->maxNum;
@ -330,7 +330,7 @@ int32_t tWWorkerAllInit(SWWorkerAll *pWorker, const SWWorkerAllCfg *pCfg) {
return 0; return 0;
} }
void tWWorkerAllCleanup(SWWorkerAll *pWorker) { void tMultiWorkerCleanup(SMultiWorker *pWorker) {
if (pWorker->queue == NULL) return; if (pWorker->queue == NULL) return;
while (!taosQueueEmpty(pWorker->queue)) { while (!taosQueueEmpty(pWorker->queue)) {