From 652f51fec524eb93bd269f0996b6c06a3b660607 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Sat, 10 Dec 2022 15:09:55 +0800 Subject: [PATCH] fix: control rpc qitem memory --- include/util/tqueue.h | 4 ++-- source/dnode/mgmt/mgmt_mnode/src/mmWorker.c | 2 +- source/dnode/mgmt/mgmt_qnode/src/qmWorker.c | 4 ++-- source/dnode/mgmt/mgmt_snode/src/smWorker.c | 4 ++-- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 2 +- source/dnode/mgmt/node_mgmt/src/dmTransport.c | 2 +- source/util/src/tqueue.c | 6 ++++-- 7 files changed, 13 insertions(+), 11 deletions(-) diff --git a/include/util/tqueue.h b/include/util/tqueue.h index 25a0019106..1f6b205cdf 100644 --- a/include/util/tqueue.h +++ b/include/util/tqueue.h @@ -65,7 +65,7 @@ typedef struct STaosQnode { STaosQnode *next; STaosQueue *queue; int64_t timestamp; - int32_t dataSize; + int64_t dataSize; int32_t size; int8_t itype; int8_t reserved[3]; @@ -104,7 +104,7 @@ typedef struct STaosQall { STaosQueue *taosOpenQueue(); void taosCloseQueue(STaosQueue *queue); void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp); -void *taosAllocateQitem(int32_t size, EQItype itype, int32_t dataSize); +void *taosAllocateQitem(int32_t size, EQItype itype, int64_t dataSize); void taosFreeQitem(void *pItem); void taosWriteQitem(STaosQueue *queue, void *pItem); int32_t taosReadQitem(STaosQueue *queue, void **ppItem); diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c index 5d86e11c57..095857825d 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c @@ -164,7 +164,7 @@ int32_t mmPutMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { memcpy(pMsg, pRpc, sizeof(SRpcMsg)); pRpc->pCont = NULL; - dTrace("msg:%p, is created and will put into %s queue, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType)); + dTrace("msg:%p, is created and will put into %s queue, type:%s len:%d", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType), pRpc->contLen); int32_t code = mmPutMsgToWorker(pMgmt, pWorker, pMsg); if (code != 0) { dTrace("msg:%p, is freed", pMsg); diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c index 579a4e2090..28da0f9c5f 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c @@ -65,12 +65,12 @@ int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { switch (qtype) { case QUERY_QUEUE: - dTrace("msg:%p, is created and will put into qnode-query queue", pMsg); + dTrace("msg:%p, is created and will put into qnode-query queue, len:%d", pMsg, pRpc->contLen); taosWriteQitem(pMgmt->queryWorker.queue, pMsg); return 0; case READ_QUEUE: case FETCH_QUEUE: - dTrace("msg:%p, is created and will put into qnode-fetch queue", pMsg); + dTrace("msg:%p, is created and will put into qnode-fetch queue, len:%d", pMsg, pRpc->contLen); taosWriteQitem(pMgmt->fetchWorker.queue, pMsg); return 0; default: diff --git a/source/dnode/mgmt/mgmt_snode/src/smWorker.c b/source/dnode/mgmt/mgmt_snode/src/smWorker.c index e1cc378396..9bd5be5201 100644 --- a/source/dnode/mgmt/mgmt_snode/src/smWorker.c +++ b/source/dnode/mgmt/mgmt_snode/src/smWorker.c @@ -139,8 +139,8 @@ int32_t smPutMsgToQueue(SSnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SSnode *pSnode = pMgmt->pSnode; if (pSnode == NULL) { - dError("msg:%p failed to put into snode queue since %s, type:%s qtype:%d", pMsg, terrstr(), - TMSG_INFO(pMsg->msgType), qtype); + dError("msg:%p failed to put into snode queue since %s, type:%s qtype:%d len:%d", pMsg, terrstr(), + TMSG_INFO(pMsg->msgType), qtype, pRpc->contLen); taosFreeQitem(pMsg); rpcFreeCont(pRpc->pCont); pRpc->pCont = NULL; diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index c48216ed7d..7e3915f3d1 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -241,7 +241,7 @@ int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { } SMsgHead *pHead = pRpc->pCont; - dTrace("vgId:%d, msg:%p is created, type:%s", pHead->vgId, pMsg, TMSG_INFO(pRpc->msgType)); + dTrace("vgId:%d, msg:%p is created, type:%s len:%d", pHead->vgId, pMsg, TMSG_INFO(pRpc->msgType), pRpc->contLen); pHead->contLen = htonl(pHead->contLen); pHead->vgId = htonl(pHead->vgId); diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 33646d6d67..4fa09a46b7 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -145,7 +145,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { if (pMsg == NULL) goto _OVER; memcpy(pMsg, pRpc, sizeof(SRpcMsg)); - dGTrace("msg:%p, is created, type:%s handle:%p", pMsg, TMSG_INFO(pRpc->msgType), pMsg->info.handle); + dGTrace("msg:%p, is created, type:%s handle:%p len:%d", pMsg, TMSG_INFO(pRpc->msgType), pMsg->info.handle, pRpc->contLen); code = dmProcessNodeMsg(pWrapper, pMsg); diff --git a/source/util/src/tqueue.c b/source/util/src/tqueue.c index 1079dff0bf..42b6358893 100644 --- a/source/util/src/tqueue.c +++ b/source/util/src/tqueue.c @@ -109,7 +109,7 @@ int64_t taosQueueMemorySize(STaosQueue *queue) { return memOfItems; } -void *taosAllocateQitem(int32_t size, EQItype itype, int32_t dataSize) { +void *taosAllocateQitem(int32_t size, EQItype itype, int64_t dataSize) { STaosQnode *pNode = taosMemoryCalloc(1, sizeof(STaosQnode) + size); if (pNode == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; @@ -124,7 +124,9 @@ void *taosAllocateQitem(int32_t size, EQItype itype, int32_t dataSize) { if (itype == RPC_QITEM) { int64_t alloced = atomic_add_fetch_64(&tsRpcQueueMemoryUsed, size + dataSize); if (alloced > tsRpcQueueMemoryAllowed) { - uError("failed to alloc qitem, size:%d alloc:%" PRId64 " allowed:%" PRId64, size + dataSize, alloced, tsRpcQueueMemoryUsed); + uError("failed to alloc qitem, size:%" PRId64 " alloc:%" PRId64 " allowed:%" PRId64, size + dataSize, alloced, + tsRpcQueueMemoryUsed); + atomic_sub_fetch_64(&tsRpcQueueMemoryUsed, size + dataSize); taosMemoryFree(pNode); terrno = TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE; return NULL; -- GitLab