From 97c4dfb570d61b23523f8c200d4267d49c6ee19d Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 25 Feb 2023 18:01:17 +0800
Subject: [PATCH] fix: limit session num

---
 source/common/src/tglobal.c                   |  2 +-
 source/dnode/mgmt/node_mgmt/src/dmTransport.c | 20 +++++++++----------
 source/libs/function/src/udfd.c               | 15 +++++++++-----
 3 files changed, 21 insertions(+), 16 deletions(-)

diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index 6eb8a245e4..2275345d7f 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -756,7 +756,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsPrintAuth = cfgGetItem(pCfg, "printAuth")->bval;
 
   tsNumOfRpcThreads = cfgGetItem(pCfg, "numOfRpcThreads")->i32;
-  tsNumOfRpcSessions = cfgGetItem(pCfg, "numofrpcsessions")->i32;
+  tsNumOfRpcSessions = cfgGetItem(pCfg, "numOfRpcSessions")->i32;
   tsTimeToGetAvailableConn = cfgGetItem(pCfg, "timeToGetAvailableConn")->i32;
 
   tsNumOfCommitThreads = cfgGetItem(pCfg, "numOfCommitThreads")->i32;
diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c
index 953a2452dd..ea46b70693 100644
--- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c
+++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c
@@ -93,15 +93,15 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
       break;
   }
 
-/*
-pDnode is null, TD-22618
-at trans.c line 91
-before this line, dmProcessRpcMsg callback is set
-after this line, parent is set
-so when dmProcessRpcMsg is called, pDonde is still null.
-*/
-  if (pDnode != NULL){
-    if(pDnode->status != DND_STAT_RUNNING) {
+  /*
+  pDnode is null, TD-22618
+  at trans.c line 91
+  before this line, dmProcessRpcMsg callback is set
+  after this line, parent is set
+  so when dmProcessRpcMsg is called, pDonde is still null.
+  */
+  if (pDnode != NULL) {
+    if (pDnode->status != DND_STAT_RUNNING) {
       if (pRpc->msgType == TDMT_DND_SERVER_STATUS) {
         dmProcessServerStartupStatus(pDnode, pRpc);
         return;
@@ -113,7 +113,7 @@ so when dmProcessRpcMsg is called, pDonde is still null.
         }
         goto _OVER;
       }
-    }   
+    }
   } else {
     terrno = TSDB_CODE_APP_IS_STARTING;
     goto _OVER;
diff --git a/source/libs/function/src/udfd.c b/source/libs/function/src/udfd.c
index fe6ed7d785..dc6fc3ad74 100644
--- a/source/libs/function/src/udfd.c
+++ b/source/libs/function/src/udfd.c
@@ -606,9 +606,8 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
 }
 static bool udfdRpcRfp(int32_t code, tmsg_t msgType) {
   if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK || code == TSDB_CODE_SYN_NOT_LEADER ||
-      code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED ||
-      code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_MNODE_NOT_FOUND || code == TSDB_CODE_APP_IS_STARTING ||
-      code == TSDB_CODE_APP_IS_STOPPING) {
+      code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED || code == TSDB_CODE_SYN_RESTORING ||
+      code == TSDB_CODE_MNODE_NOT_FOUND || code == TSDB_CODE_APP_IS_STARTING || code == TSDB_CODE_APP_IS_STOPPING) {
     if (msgType == TDMT_SCH_QUERY || msgType == TDMT_SCH_MERGE_QUERY || msgType == TDMT_SCH_FETCH ||
         msgType == TDMT_SCH_MERGE_FETCH) {
       return false;
@@ -673,6 +672,12 @@ int32_t udfdOpenClientRpc() {
   rpcInit.rfp = udfdRpcRfp;
   rpcInit.compressSize = tsCompressMsgSize;
 
+  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3);
+  connLimitNum = TMAX(connLimitNum, 10);
+  connLimitNum = TMIN(connLimitNum, 500);
+  rpcInit.connLimitNum = connLimitNum;
+  rpcInit.timeToGetConn = tsTimeToGetAvailableConn;
+
   global.clientRpc = rpcOpen(&rpcInit);
   if (global.clientRpc == NULL) {
     fnError("failed to init dnode rpc client");
@@ -765,7 +770,7 @@ bool isUdfdUvMsgComplete(SUdfdUvConn *pipe) {
 }
 
 void udfdHandleRequest(SUdfdUvConn *conn) {
-  char *inputBuf = conn->inputBuf;
+  char   *inputBuf = conn->inputBuf;
   int32_t inputLen = conn->inputLen;
 
   uv_work_t  *work = taosMemoryMalloc(sizeof(uv_work_t));
@@ -784,7 +789,7 @@ void udfdHandleRequest(SUdfdUvConn *conn) {
 
 void udfdPipeCloseCb(uv_handle_t *pipe) {
   SUdfdUvConn *conn = pipe->data;
-  SUvUdfWork* pWork = conn->pWorkList;
+  SUvUdfWork  *pWork = conn->pWorkList;
   while (pWork != NULL) {
     pWork->conn = NULL;
     pWork = pWork->pWorkNext;
-- 
GitLab