From 183e444e959e7e584cf904a68a93c259504f8b4d Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sun, 15 Mar 2020 13:44:07 +0800 Subject: [PATCH] change tqueue --- src/dnode/inc/dnodeWrite.h | 2 +- src/dnode/src/dnodeMgmt.c | 8 ++--- src/dnode/src/dnodeRead.c | 47 ++++++++++++++--------------- src/dnode/src/dnodeWrite.c | 60 +++++++++++++++++++------------------- src/util/inc/tlog.h | 2 +- src/util/inc/tqueue.h | 16 +++++----- src/util/src/tqueue.c | 53 ++++++++++++++++++++------------- src/util/src/tsocket.c | 2 +- 8 files changed, 102 insertions(+), 88 deletions(-) diff --git a/src/dnode/inc/dnodeWrite.h b/src/dnode/inc/dnodeWrite.h index 2b1edf9e40..f5904555fb 100644 --- a/src/dnode/inc/dnodeWrite.h +++ b/src/dnode/inc/dnodeWrite.h @@ -23,7 +23,7 @@ extern "C" { int32_t dnodeInitWrite(); void dnodeCleanupWrite(); void dnodeWrite(void *pMsg); -void * dnodeAllocateWriteWorker(); +void * dnodeAllocateWriteWorker(void *pVnode); void dnodeFreeWriteWorker(void *worker); #ifdef __cplusplus diff --git a/src/dnode/src/dnodeMgmt.c b/src/dnode/src/dnodeMgmt.c index e8fe94f358..a4cb752413 100644 --- a/src/dnode/src/dnodeMgmt.c +++ b/src/dnode/src/dnodeMgmt.c @@ -152,8 +152,8 @@ static int32_t dnodeOpenVnode(int32_t vgId) { vnodeObj.status = TSDB_VN_STATUS_NOT_READY; vnodeObj.refCount = 1; vnodeObj.version = 0; - vnodeObj.wworker = dnodeAllocateWriteWorker(); - vnodeObj.rworker = dnodeAllocateReadWorker(); + vnodeObj.wworker = dnodeAllocateWriteWorker(&vnodeObj); + vnodeObj.rworker = dnodeAllocateReadWorker(&vnodeObj); vnodeObj.wal = NULL; vnodeObj.tsdb = pTsdb; vnodeObj.replica = NULL; @@ -217,8 +217,8 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) { vnodeObj.status = TSDB_VN_STATUS_NOT_READY; vnodeObj.refCount = 1; vnodeObj.version = 0; - vnodeObj.wworker = dnodeAllocateWriteWorker(); - vnodeObj.rworker = dnodeAllocateReadWorker(); + vnodeObj.wworker = dnodeAllocateWriteWorker(&vnodeObj); + vnodeObj.rworker = dnodeAllocateReadWorker(&vnodeObj); vnodeObj.wal = NULL; vnodeObj.tsdb = pTsdb; vnodeObj.replica = NULL; diff --git a/src/dnode/src/dnodeRead.c b/src/dnode/src/dnodeRead.c index dacc93ffc2..1bbc65ef5c 100644 --- a/src/dnode/src/dnodeRead.c +++ b/src/dnode/src/dnodeRead.c @@ -33,16 +33,15 @@ typedef struct { void *pCont; int32_t contLen; SRpcMsg rpcMsg; - void *pVnode; SRpcContext *pRpcContext; // RPC message context } SReadMsg; static void *dnodeProcessReadQueue(void *param); -static void dnodeProcessReadResult(SReadMsg *pRead); +static void dnodeProcessReadResult(void *pVnode, SReadMsg *pRead); static void dnodeHandleIdleReadWorker(); -static void dnodeProcessQueryMsg(SReadMsg *pMsg); -static void dnodeProcessRetrieveMsg(SReadMsg *pMsg); -static void(*dnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(SReadMsg *pNode); +static void dnodeProcessQueryMsg(void *pVnode, SReadMsg *pMsg); +static void dnodeProcessRetrieveMsg(void *pVnode, SReadMsg *pMsg); +static void(*dnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(void *pVnode, SReadMsg *pNode); // module global variable static taos_qset readQset; @@ -93,15 +92,14 @@ void dnodeRead(void *rpcMsg) { } // put message into queue - SReadMsg readMsg; - readMsg.rpcMsg = *pMsg; - readMsg.pCont = pCont; - readMsg.contLen = contLen; - readMsg.pRpcContext = pRpcContext; - readMsg.pVnode = pVnode; + SReadMsg *pReadMsg = taosAllocateQitem(sizeof(SReadMsg)); + pReadMsg->rpcMsg = *pMsg; + pReadMsg->pCont = pCont; + pReadMsg->contLen = contLen; + pReadMsg->pRpcContext = pRpcContext; taos_queue queue = dnodeGetVnodeRworker(pVnode); - taosWriteQitem(queue, &readMsg); + taosWriteQitem(queue, 0, pReadMsg); // next vnode leftLen -= contLen; @@ -111,11 +109,11 @@ void dnodeRead(void *rpcMsg) { } } -void *dnodeAllocateReadWorker() { +void *dnodeAllocateReadWorker(void *pVnode) { taos_queue *queue = taosOpenQueue(sizeof(SReadMsg)); if (queue == NULL) return NULL; - taosAddIntoQset(readQset, queue); + taosAddIntoQset(readQset, queue, pVnode); // spawn a thread to process queue if (threads < maxThreads) { @@ -140,22 +138,25 @@ void dnodeFreeReadWorker(void *rqueue) { static void *dnodeProcessReadQueue(void *param) { taos_qset qset = (taos_qset)param; - SReadMsg readMsg; + SReadMsg *pReadMsg; + int type; + void *pVnode; while (1) { - if (taosReadQitemFromQset(qset, &readMsg) <= 0) { + if (taosReadQitemFromQset(qset, &type, &pReadMsg, &pVnode) == 0) { dnodeHandleIdleReadWorker(); continue; } terrno = 0; - if (dnodeProcessReadMsgFp[readMsg.rpcMsg.msgType]) { - (*dnodeProcessReadMsgFp[readMsg.rpcMsg.msgType]) (&readMsg); + if (dnodeProcessReadMsgFp[pReadMsg->rpcMsg.msgType]) { + (*dnodeProcessReadMsgFp[pReadMsg->rpcMsg.msgType]) (pVnode, pReadMsg); } else { terrno = TSDB_CODE_MSG_NOT_PROCESSED; } - dnodeProcessReadResult(&readMsg); + dnodeProcessReadResult(pVnode, pReadMsg); + taosFreeQitem(pReadMsg); } return NULL; @@ -173,11 +174,11 @@ static void dnodeHandleIdleReadWorker() { } } -static void dnodeProcessReadResult(SReadMsg *pRead) { +static void dnodeProcessReadResult(void *pVnode, SReadMsg *pRead) { SRpcContext *pRpcContext = pRead->pRpcContext; int32_t code = 0; - dnodeReleaseVnode(pRead->pVnode); + dnodeReleaseVnode(pVnode); if (pRpcContext) { if (terrno) { @@ -204,10 +205,10 @@ static void dnodeProcessReadResult(SReadMsg *pRead) { rpcFreeCont(pRead->rpcMsg.pCont); // free the received message } -static void dnodeProcessQueryMsg(SReadMsg *pMsg) { +static void dnodeProcessQueryMsg(void *pVnode, SReadMsg *pMsg) { } -static void dnodeProcessRetrieveMsg(SReadMsg *pMsg) { +static void dnodeProcessRetrieveMsg(void *pVnode, SReadMsg *pMsg) { } diff --git a/src/dnode/src/dnodeWrite.c b/src/dnode/src/dnodeWrite.c index 9d38a5c207..9e4acd6e1a 100644 --- a/src/dnode/src/dnodeWrite.c +++ b/src/dnode/src/dnodeWrite.c @@ -33,7 +33,6 @@ typedef struct _write { void *pCont; int32_t contLen; SRpcMsg rpcMsg; - void *pVnode; // pointer to vnode SRpcContext *pRpcContext; // RPC message context } SWriteMsg; @@ -49,20 +48,20 @@ typedef struct _thread_obj { SWriteWorker *writeWorker; } SWriteWorkerPool; -static void (*dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(SWriteMsg *); +static void (*dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(void *, SWriteMsg *); static void *dnodeProcessWriteQueue(void *param); static void dnodeHandleIdleWorker(SWriteWorker *pWorker); -static void dnodeProcessWriteResult(SWriteMsg *pWrite); -static void dnodeProcessSubmitMsg(SWriteMsg *pMsg); -static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg); -static void dnodeProcessDropTableMsg(SWriteMsg *pMsg); +static void dnodeProcessWriteResult(void *pVnode, SWriteMsg *pWrite); +static void dnodeProcessSubmitMsg(void *pVnode, SWriteMsg *pMsg); +static void dnodeProcessCreateTableMsg(void *pVnode, SWriteMsg *pMsg); +static void dnodeProcessDropTableMsg(void *pVnode, SWriteMsg *pMsg); SWriteWorkerPool wWorkerPool; int32_t dnodeInitWrite() { - dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeProcessSubmitMsg; + dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeProcessSubmitMsg; dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeProcessCreateTableMsg; - dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE] = dnodeProcessDropTableMsg; + dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE] = dnodeProcessDropTableMsg; wWorkerPool.max = tsNumOfCores; wWorkerPool.writeWorker = (SWriteWorker *)calloc(sizeof(SWriteWorker), wWorkerPool.max); @@ -107,15 +106,14 @@ void dnodeWrite(void *rpcMsg) { } // put message into queue - SWriteMsg writeMsg; - writeMsg.rpcMsg = *pMsg; - writeMsg.pCont = pCont; - writeMsg.contLen = contLen; - writeMsg.pRpcContext = pRpcContext; - writeMsg.pVnode = pVnode; // pVnode shall be saved for usage later + SWriteMsg *pWriteMsg = taosAllocateQitem(sizeof(SWriteMsg)); + pWriteMsg->rpcMsg = *pMsg; + pWriteMsg->pCont = pCont; + pWriteMsg->contLen = contLen; + pWriteMsg->pRpcContext = pRpcContext; taos_queue queue = dnodeGetVnodeWworker(pVnode); - taosWriteQitem(queue, &writeMsg); + taosWriteQitem(queue, 0, pWriteMsg); // next vnode leftLen -= contLen; @@ -123,7 +121,7 @@ void dnodeWrite(void *rpcMsg) { } } -void *dnodeAllocateWriteWorker() { +void *dnodeAllocateWriteWorker(void *pVnode) { SWriteWorker *pWorker = wWorkerPool.writeWorker + wWorkerPool.nextId; if (pWorker->qset == NULL) { @@ -140,9 +138,9 @@ void *dnodeAllocateWriteWorker() { } } - taos_queue *queue = taosOpenQueue(sizeof(SWriteMsg)); + taos_queue *queue = taosOpenQueue(); if (queue) { - taosAddIntoQset(pWorker->qset, queue); + taosAddIntoQset(pWorker->qset, queue, pVnode); wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max; } @@ -158,11 +156,13 @@ void dnodeFreeWriteWorker(void *wqueue) { static void *dnodeProcessWriteQueue(void *param) { SWriteWorker *pWorker = (SWriteWorker *)param; taos_qall qall; - SWriteMsg writeMsg; + SWriteMsg *pWriteMsg; int32_t numOfMsgs; + int type; + void *pVnode; while (1) { - numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, &qall); + numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, &qall, &pVnode); if (numOfMsgs <=0) { dnodeHandleIdleWorker(pWorker); // thread exit if no queues anymore continue; @@ -170,7 +170,7 @@ static void *dnodeProcessWriteQueue(void *param) { for (int32_t i=0; iwhandle, writeMsg.rpcMsg.msgType, writeMsg.pCont, writeMsg.contLen); } @@ -181,16 +181,16 @@ static void *dnodeProcessWriteQueue(void *param) { // browse all items, and process them one by one taosResetQitems(qall); for (int32_t i = 0; i < numOfMsgs; ++i) { - taosGetQitem(qall, &writeMsg); + taosGetQitem(qall, &type, &pWriteMsg); terrno = 0; - if (dnodeProcessWriteMsgFp[writeMsg.rpcMsg.msgType]) { - (*dnodeProcessWriteMsgFp[writeMsg.rpcMsg.msgType]) (&writeMsg); + if (dnodeProcessWriteMsgFp[pWriteMsg->rpcMsg.msgType]) { + (*dnodeProcessWriteMsgFp[pWriteMsg->rpcMsg.msgType]) (pVnode, pWriteMsg); } else { terrno = TSDB_CODE_MSG_NOT_PROCESSED; } - dnodeProcessWriteResult(&writeMsg); + dnodeProcessWriteResult(pVnode, pWriteMsg); } // free the Qitems; @@ -200,11 +200,11 @@ static void *dnodeProcessWriteQueue(void *param) { return NULL; } -static void dnodeProcessWriteResult(SWriteMsg *pWrite) { +static void dnodeProcessWriteResult(void *pVnode, SWriteMsg *pWrite) { SRpcContext *pRpcContext = pWrite->pRpcContext; int32_t code = 0; - dnodeReleaseVnode(pWrite->pVnode); + dnodeReleaseVnode(pVnode); if (pRpcContext) { if (terrno) { @@ -244,14 +244,14 @@ static void dnodeHandleIdleWorker(SWriteWorker *pWorker) { } } -static void dnodeProcessSubmitMsg(SWriteMsg *pMsg) { +static void dnodeProcessSubmitMsg(void *param, SWriteMsg *pMsg) { } -static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) { +static void dnodeProcessCreateTableMsg(void *param, SWriteMsg *pMsg) { } -static void dnodeProcessDropTableMsg(SWriteMsg *pMsg) { +static void dnodeProcessDropTableMsg(void *param, SWriteMsg *pMsg) { } diff --git a/src/util/inc/tlog.h b/src/util/inc/tlog.h index 350a2c700e..e1c5f8a06a 100644 --- a/src/util/inc/tlog.h +++ b/src/util/inc/tlog.h @@ -78,7 +78,7 @@ void taosResetLogFile(); // utility log function #define pError(...) \ if (uDebugFlag & DEBUG_ERROR) { \ - tprintf("ERROR UTL ", 255, __VA_ARGS__); \ + tprintf("ERROR UTL ", uDebugFlag, __VA_ARGS__); \ } #define pWarn(...) \ if (uDebugFlag & DEBUG_WARN) { \ diff --git a/src/util/inc/tqueue.h b/src/util/inc/tqueue.h index 97408110d4..4920d08e29 100644 --- a/src/util/inc/tqueue.h +++ b/src/util/inc/tqueue.h @@ -24,24 +24,26 @@ typedef void* taos_queue; typedef void* taos_qset; typedef void* taos_qall; -taos_queue taosOpenQueue(int itemSize); +taos_queue taosOpenQueue(); void taosCloseQueue(taos_queue); -int taosWriteQitem(taos_queue, void *item); -int taosReadQitem(taos_queue, void *item); +void *taosAllocateQitem(int size); +void taosFreeQitem(void *item); +int taosWriteQitem(taos_queue, int type, void *item); +int taosReadQitem(taos_queue, int *type, void **pitem); int taosReadAllQitems(taos_queue, taos_qall *); -int taosGetQitem(taos_qall, void *item); +int taosGetQitem(taos_qall, int *type, void **pitem); void taosResetQitems(taos_qall); void taosFreeQitems(taos_qall); taos_qset taosOpenQset(); void taosCloseQset(); -int taosAddIntoQset(taos_qset, taos_queue); +int taosAddIntoQset(taos_qset, taos_queue, void *ahandle); void taosRemoveFromQset(taos_qset, taos_queue); int taosGetQueueNumber(taos_qset); -int taosReadQitemFromQset(taos_qset, void *item); -int taosReadAllQitemsFromQset(taos_qset, taos_qall *); +int taosReadQitemFromQset(taos_qset, int *type, void **pitem, void **handle); +int taosReadAllQitemsFromQset(taos_qset, taos_qall *, void **handle); int taosGetQueueItemsNumber(taos_queue param); int taosGetQsetItemsNumber(taos_qset param); diff --git a/src/util/src/tqueue.c b/src/util/src/tqueue.c index cb218126a8..ed69d7e7c4 100644 --- a/src/util/src/tqueue.c +++ b/src/util/src/tqueue.c @@ -19,6 +19,7 @@ #include "tqueue.h" typedef struct _taos_qnode { + int type; struct _taos_qnode *next; char item[]; } STaosQnode; @@ -30,6 +31,7 @@ typedef struct _taos_q { struct _taos_qnode *tail; struct _taos_q *next; // for queue set struct _taos_qset *qset; // for queue set + void *ahandle; // for queue set pthread_mutex_t mutex; } STaosQueue; @@ -48,7 +50,7 @@ typedef struct _taos_qall { int32_t numOfItems; } STaosQall; -taos_queue taosOpenQueue(int itemSize) { +taos_queue taosOpenQueue() { STaosQueue *queue = (STaosQueue *) calloc(sizeof(STaosQueue), 1); if (queue == NULL) { @@ -57,8 +59,6 @@ taos_queue taosOpenQueue(int itemSize) { } pthread_mutex_init(&queue->mutex, NULL); - queue->itemSize = (int32_t)itemSize; - return queue; } @@ -83,16 +83,24 @@ void taosCloseQueue(taos_queue param) { free(queue); } -int taosWriteQitem(taos_queue param, void *item) { - STaosQueue *queue = (STaosQueue *)param; +void *taosAllocateQitem(int size) { + STaosQnode *pNode = (STaosQnode *)calloc(sizeof(STaosQnode) + size, 1); + if (pNode == NULL) return NULL; + return (void *)pNode->item; +} - STaosQnode *pNode = (STaosQnode *)calloc(sizeof(STaosQnode) + queue->itemSize, 1); - if ( pNode == NULL ) { - terrno = TSDB_CODE_NO_RESOURCE; - return -1; - } +void taosFreeQitem(void *param) { + if (param == NULL) return; + + char *temp = (char *)param; + temp -= sizeof(STaosQnode); + free(temp); +} - memcpy(pNode->item, item, queue->itemSize); +int taosWriteQitem(taos_queue param, int type, void *item) { + STaosQueue *queue = (STaosQueue *)param; + STaosQnode *pNode = (STaosQnode *)((char *)item - sizeof(STaosQnode)); + pNode->type = type; pthread_mutex_lock(&queue->mutex); @@ -112,7 +120,7 @@ int taosWriteQitem(taos_queue param, void *item) { return 0; } -int taosReadQitem(taos_queue param, void *item) { +int taosReadQitem(taos_queue param, int *type, void **pitem) { STaosQueue *queue = (STaosQueue *)param; STaosQnode *pNode = NULL; int code = 0; @@ -121,11 +129,11 @@ int taosReadQitem(taos_queue param, void *item) { if (queue->head) { pNode = queue->head; - memcpy(item, pNode->item, queue->itemSize); + *pitem = pNode->item; + *type = pNode->type; queue->head = pNode->next; if (queue->head == NULL) queue->tail = NULL; - free(pNode); queue->numOfItems--; if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, 1); code = 1; @@ -168,7 +176,7 @@ int taosReadAllQitems(taos_queue param, taos_qall *res) { return code; } -int taosGetQitem(taos_qall param, void *item) { +int taosGetQitem(taos_qall param, int *type, void **pitem) { STaosQall *qall = (STaosQall *)param; STaosQnode *pNode; int num = 0; @@ -178,7 +186,7 @@ int taosGetQitem(taos_qall param, void *item) { qall->current = pNode->next; if (pNode) { - memcpy(item, pNode->item, qall->itemSize); + *pitem = pNode->item; num = 1; } @@ -221,7 +229,7 @@ void taosCloseQset(taos_qset param) { free(qset); } -int taosAddIntoQset(taos_qset p1, taos_queue p2) { +int taosAddIntoQset(taos_qset p1, taos_queue p2, void *ahandle) { STaosQueue *queue = (STaosQueue *)p2; STaosQset *qset = (STaosQset *)p1; @@ -230,6 +238,7 @@ int taosAddIntoQset(taos_qset p1, taos_queue p2) { pthread_mutex_lock(&qset->mutex); queue->next = qset->head; + queue->ahandle = ahandle; qset->head = queue; qset->numOfQueues++; @@ -283,7 +292,7 @@ int taosGetQueueNumber(taos_qset param) { return ((STaosQset *)param)->numOfQueues; } -int taosReadQitemFromQset(taos_qset param, void *item) { +int taosReadQitemFromQset(taos_qset param, int *type, void **pitem, void **phandle) { STaosQset *qset = (STaosQset *)param; STaosQnode *pNode = NULL; int code = 0; @@ -301,11 +310,12 @@ int taosReadQitemFromQset(taos_qset param, void *item) { if (queue->head) { pNode = queue->head; - memcpy(item, pNode->item, queue->itemSize); + *pitem = pNode->item; + *type = pNode->type; + *phandle = queue->ahandle; queue->head = pNode->next; if (queue->head == NULL) queue->tail = NULL; - free(pNode); queue->numOfItems--; atomic_sub_fetch_32(&qset->numOfItems, 1); code = 1; @@ -318,7 +328,7 @@ int taosReadQitemFromQset(taos_qset param, void *item) { return code; } -int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res) { +int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) { STaosQset *qset = (STaosQset *)param; STaosQueue *queue; STaosQall *qall = NULL; @@ -346,6 +356,7 @@ int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res) { qall->numOfItems = queue->numOfItems; qall->itemSize = queue->itemSize; code = qall->numOfItems; + *phandle = queue->ahandle; queue->head = NULL; queue->tail = NULL; diff --git a/src/util/src/tsocket.c b/src/util/src/tsocket.c index 7ab004646e..30c3ada637 100644 --- a/src/util/src/tsocket.c +++ b/src/util/src/tsocket.c @@ -330,7 +330,7 @@ int taosOpenTcpClientSocket(char *destIp, uint16_t destPort, char *clientIp) { struct sockaddr_in serverAddr, clientAddr; int ret; - pTrace("open tcp client socket:%s:%d", destIp, destPort); + pTrace("open tcp client socket:%s:%d, local Ip:%s", destIp, destPort, clientIp); sockFd = (int)socket(PF_INET, SOCK_STREAM, IPPROTO_TCP); -- GitLab