From eabc5511d13d9135d5d09f324999c97b82a2a131 Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sun, 22 Mar 2020 15:53:21 +0800 Subject: [PATCH] change the code to accomodate the changes by tqueue --- src/dnode/src/dnodeRead.c | 19 ++++---- src/dnode/src/dnodeWrite.c | 28 ++++++------ src/rpc/test/rserver.c | 32 ++++++++----- src/util/inc/tqueue.h | 11 +++-- src/util/src/tqueue.c | 92 +++++++++++++++++--------------------- src/util/src/tsocket.c | 4 +- 6 files changed, 96 insertions(+), 90 deletions(-) diff --git a/src/dnode/src/dnodeRead.c b/src/dnode/src/dnodeRead.c index 11cb845798..a870a9e159 100644 --- a/src/dnode/src/dnodeRead.c +++ b/src/dnode/src/dnodeRead.c @@ -90,7 +90,7 @@ void dnodeRead(SRpcMsg *pMsg) { while (leftLen > 0) { SMsgHead *pHead = (SMsgHead *) pCont; - pHead->vgId = 1;//htonl(pHead->vgId); + pHead->vgId = 1; //htonl(pHead->vgId); pHead->contLen = pMsg->contLen; //htonl(pHead->contLen); void *pVnode = dnodeGetVnode(pHead->vgId); @@ -101,22 +101,19 @@ void dnodeRead(SRpcMsg *pMsg) { } // put message into queue - SReadMsg readMsg; - readMsg.rpcMsg = *pMsg; - readMsg.pCont = pCont; - readMsg.contLen = pHead->contLen; - readMsg.pRpcContext = pRpcContext; - readMsg.pVnode = pVnode; + SReadMsg *pRead = (SReadMsg *)taosAllocateQitem(sizeof(SReadMsg)); + pRead->rpcMsg = *pMsg; + pRead->pCont = pCont; + pRead->contLen = pHead->contLen; + pRead->pRpcContext = pRpcContext; taos_queue queue = dnodeGetVnodeRworker(pVnode); - taosWriteQitem(queue, 0, pReadMsg); + taosWriteQitem(queue, TAOS_QTYPE_RPC, pRead); // next vnode leftLen -= pHead->contLen; pCont -= pHead->contLen; queuedMsgNum++; - - dnodeReleaseVnode(pVnode); } if (queuedMsgNum == 0) { @@ -179,6 +176,8 @@ static void *dnodeProcessReadQueue(void *param) { dnodeProcessReadResult(pVnode, pReadMsg); taosFreeQitem(pReadMsg); + + dnodeReleaseVnode(pVnode); } return NULL; diff --git a/src/dnode/src/dnodeWrite.c b/src/dnode/src/dnodeWrite.c index 8e5bcc5844..5dd075e4df 100644 --- a/src/dnode/src/dnodeWrite.c +++ b/src/dnode/src/dnodeWrite.c @@ -115,15 +115,14 @@ void dnodeWrite(SRpcMsg *pMsg) { } // put message into queue - SWriteMsg writeMsg; - writeMsg.rpcMsg = *pMsg; - writeMsg.pCont = pCont; - writeMsg.contLen = pHead->contLen; - writeMsg.pRpcContext = pRpcContext; - writeMsg.pVnode = pVnode; // pVnode shall be saved for usage later + SWriteMsg *pWrite = (SWriteMsg *)taosAllocateQitem(sizeof(SWriteMsg)); + pWrite->rpcMsg = *pMsg; + pWrite->pCont = pCont; + pWrite->contLen = pHead->contLen; + pWrite->pRpcContext = pRpcContext; taos_queue queue = dnodeGetVnodeWworker(pVnode); - taosWriteQitem(queue, &writeMsg); + taosWriteQitem(queue, TAOS_QTYPE_RPC, pWrite); // next vnode leftLen -= pHead->contLen; @@ -145,14 +144,14 @@ void dnodeWrite(SRpcMsg *pMsg) { void *dnodeAllocateWriteWorker(void *pVnode) { SWriteWorker *pWorker = wWorkerPool.writeWorker + wWorkerPool.nextId; - taos_queue *queue = taosOpenQueue(sizeof(SWriteMsg)); + taos_queue *queue = taosOpenQueue(); if (queue == NULL) return NULL; if (pWorker->qset == NULL) { pWorker->qset = taosOpenQset(); if (pWorker->qset == NULL) return NULL; - taosAddIntoQset(pWorker->qset, queue); + taosAddIntoQset(pWorker->qset, queue, pVnode); wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max; pthread_attr_t thAttr; @@ -164,7 +163,7 @@ void *dnodeAllocateWriteWorker(void *pVnode) { taosCloseQset(pWorker->qset); } } else { - taosAddIntoQset(pWorker->qset, queue); + taosAddIntoQset(pWorker->qset, queue, pVnode); wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max; } @@ -185,8 +184,10 @@ static void *dnodeProcessWriteQueue(void *param) { int type; void *pVnode; + qall = taosAllocateQall(); + while (1) { - numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, &qall, &pVnode); + numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, qall, &pVnode); if (numOfMsgs <=0) { dnodeHandleIdleWorker(pWorker); // thread exit if no queues anymore continue; @@ -215,12 +216,13 @@ static void *dnodeProcessWriteQueue(void *param) { } dnodeProcessWriteResult(pVnode, pWriteMsg); + taosFreeQitem(pWriteMsg); } - // free the Qitems; - taosFreeQitems(qall); } + taosFreeQall(qall); + return NULL; } diff --git a/src/rpc/test/rserver.c b/src/rpc/test/rserver.c index d9e5da51a6..6c5b320809 100644 --- a/src/rpc/test/rserver.c +++ b/src/rpc/test/rserver.c @@ -28,10 +28,13 @@ void *qhandle = NULL; void processShellMsg() { static int num = 0; taos_qall qall; - SRpcMsg rpcMsg; + SRpcMsg *pRpcMsg, rpcMsg; + int type; + + qall = taosAllocateQall(); while (1) { - int numOfMsgs = taosReadAllQitems(qhandle, &qall); + int numOfMsgs = taosReadAllQitems(qhandle, qall); if (numOfMsgs <= 0) { usleep(1000); continue; @@ -40,10 +43,10 @@ void processShellMsg() { tTrace("%d shell msgs are received", numOfMsgs); for (int i=0; i=0) { - if ( write(dataFd, rpcMsg.pCont, rpcMsg.contLen) <0 ) { + if ( write(dataFd, pRpcMsg->pCont, pRpcMsg->contLen) <0 ) { tPrint("failed to write data file, reason:%s", strerror(errno)); } } @@ -62,19 +65,22 @@ void processShellMsg() { taosResetQitems(qall); for (int i=0; ipCont); + rpcMsg.pCont = rpcMallocCont(msgSize); rpcMsg.contLen = msgSize; - rpcMsg.handle = rpcMsg.handle; + rpcMsg.handle = pRpcMsg->handle; rpcMsg.code = 1; rpcSendResponse(&rpcMsg); + + taosFreeQitem(pRpcMsg); } - taosFreeQitems(qall); } + taosFreeQall(qall); /* SRpcIpSet ipSet; ipSet.numOfIps = 1; @@ -108,8 +114,13 @@ int retrieveAuthInfo(char *meterId, char *spi, char *encrypt, char *secret, char } void processRequestMsg(SRpcMsg *pMsg) { - tTrace("request is received, type:%d, contLen:%d", pMsg->msgType, pMsg->contLen); - taosWriteQitem(qhandle, pMsg); + SRpcMsg *pTemp; + + pTemp = taosAllocateQitem(sizeof(SRpcMsg)); + memcpy(pTemp, pMsg, sizeof(SRpcMsg)); + + tTrace("request is received, type:%d, contLen:%d, item:%p", pMsg->msgType, pMsg->contLen, pTemp); + taosWriteQitem(qhandle, TAOS_QTYPE_RPC, pTemp); } int main(int argc, char *argv[]) { @@ -143,6 +154,7 @@ int main(int argc, char *argv[]) { commit = atoi(argv[++i]); } else if (strcmp(argv[i], "-d")==0 && i < argc-1) { rpcDebugFlag = atoi(argv[++i]); + ddebugFlag = rpcDebugFlag; uDebugFlag = rpcDebugFlag; } else { printf("\nusage: %s [options] \n", argv[0]); diff --git a/src/util/inc/tqueue.h b/src/util/inc/tqueue.h index 4920d08e29..c45eb10518 100644 --- a/src/util/inc/tqueue.h +++ b/src/util/inc/tqueue.h @@ -20,6 +20,10 @@ extern "C" { #endif +#define TAOS_QTYPE_RPC 0 +#define TAOS_QTYPE_FWD 1 +#define TAOS_QTYPE_WAL 2 + typedef void* taos_queue; typedef void* taos_qset; typedef void* taos_qall; @@ -31,10 +35,11 @@ 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 *); +taos_qall taosAllocateQall(); +void taosFreeQall(taos_qall); +int taosReadAllQitems(taos_queue, taos_qall); int taosGetQitem(taos_qall, int *type, void **pitem); void taosResetQitems(taos_qall); -void taosFreeQitems(taos_qall); taos_qset taosOpenQset(); void taosCloseQset(); @@ -43,7 +48,7 @@ void taosRemoveFromQset(taos_qset, taos_queue); int taosGetQueueNumber(taos_qset); int taosReadQitemFromQset(taos_qset, int *type, void **pitem, void **handle); -int taosReadAllQitemsFromQset(taos_qset, taos_qall *, 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 f1db24b8ce..1ac05556d6 100644 --- a/src/util/src/tqueue.c +++ b/src/util/src/tqueue.c @@ -92,6 +92,8 @@ void *taosAllocateQitem(int size) { void taosFreeQitem(void *param) { if (param == NULL) return; + //pTrace("item:%p is freed", param); + char *temp = (char *)param; temp -= sizeof(STaosQnode); free(temp); @@ -115,6 +117,8 @@ int taosWriteQitem(taos_queue param, int type, void *item) { queue->numOfItems++; if (queue->qset) atomic_add_fetch_32(&queue->qset->numOfItems, 1); + //pTrace("item:%p is put into queue, items:%d", item, queue->numOfItems); + pthread_mutex_unlock(&queue->mutex); return 0; @@ -137,6 +141,7 @@ int taosReadQitem(taos_queue param, int *type, void **pitem) { queue->numOfItems--; if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, 1); code = 1; + //pTrace("item:%p is read out from queue, items:%d", *pitem, queue->numOfItems); } pthread_mutex_unlock(&queue->mutex); @@ -144,35 +149,38 @@ int taosReadQitem(taos_queue param, int *type, void **pitem) { return code; } -int taosReadAllQitems(taos_queue param, taos_qall *res) { +void *taosAllocateQall() { + void *p = malloc(sizeof(STaosQall)); + return p; +} + +void taosFreeQall(void *param) { + free(param); +} + +int taosReadAllQitems(taos_queue param, taos_qall p2) { STaosQueue *queue = (STaosQueue *)param; - STaosQall *qall = NULL; + STaosQall *qall = (STaosQall *)p2; int code = 0; pthread_mutex_lock(&queue->mutex); if (queue->head) { - qall = (STaosQall *) calloc(sizeof(STaosQall), 1); - if ( qall == NULL ) { - terrno = TSDB_CODE_NO_RESOURCE; - code = -1; - } else { - qall->current = queue->head; - qall->start = queue->head; - qall->numOfItems = queue->numOfItems; - qall->itemSize = queue->itemSize; - code = qall->numOfItems; - - queue->head = NULL; - queue->tail = NULL; - queue->numOfItems = 0; - if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, qall->numOfItems); - } + memset(qall, 0, sizeof(STaosQall)); + qall->current = queue->head; + qall->start = queue->head; + qall->numOfItems = queue->numOfItems; + qall->itemSize = queue->itemSize; + code = qall->numOfItems; + + queue->head = NULL; + queue->tail = NULL; + queue->numOfItems = 0; + if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, qall->numOfItems); } pthread_mutex_unlock(&queue->mutex); - *res = qall; return code; } @@ -189,6 +197,7 @@ int taosGetQitem(taos_qall param, int *type, void **pitem) { *pitem = pNode->item; *type = pNode->type; num = 1; + //pTrace("item:%p is fetched", *pitem); } return num; @@ -199,19 +208,6 @@ void taosResetQitems(taos_qall param) { qall->current = qall->start; } -void taosFreeQitems(taos_qall param) { - STaosQall *qall = (STaosQall *)param; - STaosQnode *pNode; - - while (qall->current) { - pNode = qall->current; - qall->current = pNode->next; - free(pNode); - } - - free(qall); -} - taos_qset taosOpenQset() { STaosQset *qset = (STaosQset *) calloc(sizeof(STaosQset), 1); @@ -329,10 +325,10 @@ int taosReadQitemFromQset(taos_qset param, int *type, void **pitem, void **phand return code; } -int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) { +int taosReadAllQitemsFromQset(taos_qset param, taos_qall p2, void **phandle) { STaosQset *qset = (STaosQset *)param; STaosQueue *queue; - STaosQall *qall = NULL; + STaosQall *qall = (STaosQall *)p2; int code = 0; for(int i=0; inumOfQueues; ++i) { @@ -347,23 +343,17 @@ int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) { pthread_mutex_lock(&queue->mutex); if (queue->head) { - qall = (STaosQall *) calloc(sizeof(STaosQall), 1); - if (qall == NULL) { - terrno = TSDB_CODE_NO_RESOURCE; - code = -1; - } else { - qall->current = queue->head; - qall->start = queue->head; - qall->numOfItems = queue->numOfItems; - qall->itemSize = queue->itemSize; - code = qall->numOfItems; - *phandle = queue->ahandle; + qall->current = queue->head; + qall->start = queue->head; + qall->numOfItems = queue->numOfItems; + qall->itemSize = queue->itemSize; + code = qall->numOfItems; + *phandle = queue->ahandle; - queue->head = NULL; - queue->tail = NULL; - queue->numOfItems = 0; - atomic_sub_fetch_32(&qset->numOfItems, qall->numOfItems); - } + queue->head = NULL; + queue->tail = NULL; + queue->numOfItems = 0; + atomic_sub_fetch_32(&qset->numOfItems, qall->numOfItems); } pthread_mutex_unlock(&queue->mutex); @@ -371,8 +361,6 @@ int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) { if (code != 0) break; } - *res = qall; - return code; } diff --git a/src/util/src/tsocket.c b/src/util/src/tsocket.c index 30c3ada637..86be428af7 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, local Ip:%s", destIp, destPort, clientIp); + // pTrace("open tcp client socket:%s:%d, local Ip:%s", destIp, destPort, clientIp); sockFd = (int)socket(PF_INET, SOCK_STREAM, IPPROTO_TCP); @@ -362,7 +362,7 @@ int taosOpenTcpClientSocket(char *destIp, uint16_t destPort, char *clientIp) { ret = connect(sockFd, (struct sockaddr *)&serverAddr, sizeof(serverAddr)); if (ret != 0) { - pError("failed to connect socket, ip:%s, port:%hu, reason: %s", destIp, destPort, strerror(errno)); + //pError("failed to connect socket, ip:%s, port:%hu, reason: %s", destIp, destPort, strerror(errno)); taosCloseSocket(sockFd); sockFd = -1; } -- GitLab