Merge pull request #12569 from taosdata/fix/query_limit

fix: mem size limit

source/common/src/tglobal.c

@@ -433,7 +433,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
 
   tsRpcQueueMemoryAllowed = tsTotalMemoryKB * 1024 * 0.1;
   tsRpcQueueMemoryAllowed = TRANGE(tsRpcQueueMemoryAllowed, TSDB_MAX_WAL_SIZE * 10L, TSDB_MAX_WAL_SIZE * 10000L);
-  if (cfgAddInt64(pCfg, "rpcQueueMemoryAllowed", tsRpcQueueMemoryAllowed, 1, INT64_MAX, 0) != 0) return -1;
+  if (cfgAddInt64(pCfg, "rpcQueueMemoryAllowed", tsRpcQueueMemoryAllowed, TSDB_MAX_WAL_SIZE * 10L, INT64_MAX, 0) != 0)
+    return -1;
 
   if (cfgAddBool(pCfg, "monitor", tsEnableMonitor, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "monitorInterval", tsMonitorInterval, 1, 200000, 0) != 0) return -1;

source/dnode/mgmt/test/sut/src/sut.cpp

@@ -30,6 +30,7 @@ void Testbase::InitLog(const char* path) {
   tsdbDebugFlag = 0;
   tsLogEmbedded = 1;
   tsAsyncLog = 0;
+  tsRpcQueueMemoryAllowed = 1024 * 1024 * 64;
 
   taosRemoveDir(path);
   taosMkDir(path);
@@ -82,7 +83,7 @@ SRpcMsg* Testbase::SendReq(tmsg_t msgType, void* pCont, int32_t contLen) {
   return client.SendReq(&rpcMsg);
 }
 
-int32_t Testbase::SendShowReq(int8_t showType, const char *tb, const char* db) {
+int32_t Testbase::SendShowReq(int8_t showType, const char* tb, const char* db) {
   if (showRsp != NULL) {
     rpcFreeCont(showRsp);
     showRsp = NULL;

source/util/src/tqueue.c

@@ -33,11 +33,11 @@ typedef struct STaosQnode {
 } STaosQnode;
 
 typedef struct STaosQueue {
-  STaosQnode   *head;
-  STaosQnode   *tail;
-  STaosQueue   *next;     // for queue set
-  STaosQset    *qset;     // for queue set
-  void         *ahandle;  // for queue set
+  STaosQnode *  head;
+  STaosQnode *  tail;
+  STaosQueue *  next;     // for queue set
+  STaosQset *   qset;     // for queue set
+  void *        ahandle;  // for queue set
   FItem         itemFp;
   FItems        itemsFp;
   TdThreadMutex mutex;
@@ -46,8 +46,8 @@ typedef struct STaosQueue {
 } STaosQueue;
 
 typedef struct STaosQset {
-  STaosQueue   *head;
-  STaosQueue   *current;
+  STaosQueue *  head;
+  STaosQueue *  current;
   TdThreadMutex mutex;
   tsem_t        sem;
   int32_t       numOfQueues;
@@ -85,7 +85,7 @@ void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp) {
 void taosCloseQueue(STaosQueue *queue) {
   if (queue == NULL) return;
   STaosQnode *pTemp;
-  STaosQset  *qset;
+  STaosQset * qset;
 
   taosThreadMutexLock(&queue->mutex);
   STaosQnode *pNode = queue->head;
@@ -152,7 +152,7 @@ void *taosAllocateQitem(int32_t size, EQItype itype) {
 
   if (itype == RPC_QITEM) {
     int64_t alloced = atomic_add_fetch_64(&tsRpcQueueMemoryUsed, size);
-    if (alloced > tsRpcQueueMemoryUsed) {
+    if (alloced > tsRpcQueueMemoryAllowed) {
       taosMemoryFree(pNode);
       terrno = TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE;
       return NULL;

source/util/test/procTest.cpp

@@ -16,11 +16,11 @@
 
 typedef struct STestMsg {
   uint16_t msgType;
-  void    *pCont;
+  void *   pCont;
   int      contLen;
   int32_t  code;
-  void    *handle;         // rpc handle returned to app
-  void    *ahandle;        // app handle set by client
+  void *   handle;         // rpc handle returned to app
+  void *   ahandle;        // app handle set by client
   int      noResp;         // has response or not(default 0, 0: resp, 1: no resp);
   int      persistHandle;  // persist handle or not
 } STestMsg;
@@ -37,7 +37,7 @@ class UtilTesProc : public ::testing::Test {
     head.msgType = 2;
     head.noResp = 3;
     head.persistHandle = 4;
-
+    tsRpcQueueMemoryAllowed = 1024 * 1024 * 64;
     taosRemoveDir("/tmp/td");
     taosMkDir("/tmp/td");
     tstrncpy(tsLogDir, "/tmp/td", PATH_MAX);
@@ -64,18 +64,18 @@ TEST_F(UtilTesProc, 00_Init_Cleanup) {
 
   shm.size = 1023;
   SProcCfg  cfg = {(ProcConsumeFp)NULL,
-                   (ProcMallocFp)taosAllocateQitem,
-                   (ProcFreeFp)taosFreeQitem,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryMalloc,
-                   (ProcConsumeFp)NULL,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryMalloc,
-                   shm,
-                   &shm,
-                   "1234"};
+                  (ProcMallocFp)taosAllocateQitem,
+                  (ProcFreeFp)taosFreeQitem,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryMalloc,
+                  (ProcConsumeFp)NULL,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryMalloc,
+                  shm,
+                  &shm,
+                  "1234"};
   SProcObj *proc = taosProcInit(&cfg);
   ASSERT_EQ(proc, nullptr);
 
@@ -105,18 +105,18 @@ TEST_F(UtilTesProc, 01_Push_Pop_Child) {
   shm.size = 3000;
   ASSERT_EQ(taosCreateShm(&shm, 1235, shm.size), 0);
   SProcCfg  cfg = {(ProcConsumeFp)ConsumeChild1,
-                   (ProcMallocFp)taosAllocateQitem,
-                   (ProcFreeFp)taosFreeQitem,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   (ProcConsumeFp)NULL,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   shm,
-                   (void *)((int64_t)1235),
-                   "1235_c"};
+                  (ProcMallocFp)taosAllocateQitem,
+                  (ProcFreeFp)taosFreeQitem,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  (ProcConsumeFp)NULL,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  shm,
+                  (void *)((int64_t)1235),
+                  "1235_c"};
   SProcObj *cproc = taosProcInit(&cfg);
   ASSERT_NE(cproc, nullptr);
 
@@ -163,18 +163,18 @@ TEST_F(UtilTesProc, 02_Push_Pop_Parent) {
   shm.size = 3000;
   ASSERT_EQ(taosCreateShm(&shm, 1236, shm.size), 0);
   SProcCfg  cfg = {(ProcConsumeFp)NULL,
-                   (ProcMallocFp)taosAllocateQitem,
-                   (ProcFreeFp)taosFreeQitem,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   (ProcConsumeFp)ConsumeParent1,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   shm,
-                   (void *)((int64_t)1236),
-                   "1236_c"};
+                  (ProcMallocFp)taosAllocateQitem,
+                  (ProcFreeFp)taosFreeQitem,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  (ProcConsumeFp)ConsumeParent1,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  shm,
+                  (void *)((int64_t)1236),
+                  "1236_c"};
   SProcObj *cproc = taosProcInit(&cfg);
   ASSERT_NE(cproc, nullptr);
 
@@ -217,18 +217,18 @@ TEST_F(UtilTesProc, 03_Handle) {
   shm.size = 3000;
   ASSERT_EQ(taosCreateShm(&shm, 1237, shm.size), 0);
   SProcCfg  cfg = {(ProcConsumeFp)ConsumeChild3,
-                   (ProcMallocFp)taosAllocateQitem,
-                   (ProcFreeFp)taosFreeQitem,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   (ProcConsumeFp)NULL,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   (ProcMallocFp)taosMemoryMalloc,
-                   (ProcFreeFp)taosMemoryFree,
-                   shm,
-                   (void *)((int64_t)1235),
-                   "1237_p"};
+                  (ProcMallocFp)taosAllocateQitem,
+                  (ProcFreeFp)taosFreeQitem,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  (ProcConsumeFp)NULL,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  (ProcMallocFp)taosMemoryMalloc,
+                  (ProcFreeFp)taosMemoryFree,
+                  shm,
+                  (void *)((int64_t)1235),
+                  "1237_p"};
   SProcObj *cproc = taosProcInit(&cfg);
   ASSERT_NE(cproc, nullptr);
 
@@ -236,7 +236,8 @@ TEST_F(UtilTesProc, 03_Handle) {
     int32_t i = 0;
     for (i = 0; i < 20; ++i) {
       head.handle = (void *)((int64_t)i);
-      ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, (void *)((int64_t)i), i, PROC_FUNC_REQ), 0);
+      ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, (void *)((int64_t)i), i, PROC_FUNC_REQ),
+                0);
     }
 
     cfg.isChild = true;
@@ -247,7 +248,7 @@ TEST_F(UtilTesProc, 03_Handle) {
     taosProcCleanup(pproc);
 
     int64_t ref = 0;
-    
+
     ref = taosProcRemoveHandle(cproc, (void *)((int64_t)3));
     EXPECT_EQ(ref, 3);
     ref = taosProcRemoveHandle(cproc, (void *)((int64_t)5));