From 5c025c00061405869b87074a9feabb182f517fa9 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 27 Apr 2022 12:45:28 +0800
Subject: [PATCH 1/2] fix(rpc): fix duplicat port error

---
 include/util/taoserror.h             |  1 +
 source/libs/transport/src/trans.c    |  4 ++++
 source/libs/transport/src/transSrv.c | 20 ++++++++++++--------
 source/util/src/terror.c             |  1 +
 4 files changed, 18 insertions(+), 8 deletions(-)

diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index 2c249a2d8d..2def58f748 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -62,6 +62,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_APP_NOT_READY                 TAOS_DEF_ERROR_CODE(0, 0x0014)
 #define TSDB_CODE_RPC_FQDN_ERROR                TAOS_DEF_ERROR_CODE(0, 0x0015)
 #define TSDB_CODE_RPC_INVALID_VERSION           TAOS_DEF_ERROR_CODE(0, 0x0016)
+#define TSDB_CODE_RPC_PORT_EADDRINUSE           TAOS_DEF_ERROR_CODE(0, 0x0017)
 
 //common & util
 #define TSDB_CODE_OUT_OF_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0100)
diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c
index c0da3f9c1f..fecb5d9279 100644
--- a/source/libs/transport/src/trans.c
+++ b/source/libs/transport/src/trans.c
@@ -49,6 +49,10 @@ void* rpcOpen(const SRpcInit* pInit) {
   pRpc->connType = pInit->connType;
   pRpc->idleTime = pInit->idleTime;
   pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
+  if (pRpc->tcphandle == NULL) {
+    taosMemoryFree(pRpc);
+    return NULL;
+  }
   pRpc->parent = pInit->parent;
   if (pInit->user) {
     memcpy(pRpc->user, pInit->user, strlen(pInit->user));
diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c
index c02cb07101..b78edf6cd0 100644
--- a/source/libs/transport/src/transSrv.c
+++ b/source/libs/transport/src/transSrv.c
@@ -93,6 +93,8 @@ typedef struct SServerObj {
   uint32_t    ip;
   uint32_t    port;
   uv_async_t* pAcceptAsync;  // just to quit from from accept thread
+
+  bool inited;
 } SServerObj;
 
 // handle
@@ -143,7 +145,7 @@ static void (*transAsyncHandle[])(SSrvMsg* msg, SWorkThrdObj* thrd) = {uvHandleR
 
 static int32_t exHandlesMgt;
 
-void       uvInitExHandleMgt();
+void       uvInitEnv();
 void       uvOpenExHandleMgt(int size);
 void       uvCloseExHandleMgt();
 int64_t    uvAddExHandle(void* p);
@@ -716,6 +718,7 @@ static bool addHandleToAcceptloop(void* arg) {
   }
   if ((err = uv_listen((uv_stream_t*)&srv->server, 512, uvOnAcceptCb)) != 0) {
     tError("failed to listen: %s", uv_err_name(err));
+    terrno = TSDB_CODE_RPC_PORT_EADDRINUSE;
     return false;
   }
   return true;
@@ -800,7 +803,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
   srv->port = port;
   uv_loop_init(srv->loop);
 
-  taosThreadOnce(&transModuleInit, uvInitExHandleMgt);
+  taosThreadOnce(&transModuleInit, uvInitEnv);
   transSrvInst++;
 
   for (int i = 0; i < srv->numOfThreads; i++) {
@@ -844,15 +847,15 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
     goto End;
     // clear all resource later
   }
-
+  srv->inited = true;
   return srv;
 End:
   transCloseServer(srv);
   return NULL;
 }
 
-void uvInitExHandleMgt() {
-  // init exhandle mgt
+void uvInitEnv() {
+  uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
   uvOpenExHandleMgt(10000);
 }
 void uvOpenExHandleMgt(int size) {
@@ -958,9 +961,10 @@ void transCloseServer(void* arg) {
   SServerObj* srv = arg;
 
   tDebug("send quit msg to accept thread");
-  uv_async_send(srv->pAcceptAsync);
-  taosThreadJoin(srv->thread, NULL);
-
+  if (srv->inited) {
+    uv_async_send(srv->pAcceptAsync);
+    taosThreadJoin(srv->thread, NULL);
+  }
   SRV_RELEASE_UV(srv->loop);
 
   for (int i = 0; i < srv->numOfThreads; i++) {
diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index 61f2015fe5..376146fa95 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -68,6 +68,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RPC_INVALID_TIME_STAMP,       "Client and server's t
 TAOS_DEFINE_ERROR(TSDB_CODE_APP_NOT_READY,                "Database not ready")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_FQDN_ERROR,               "Unable to resolve FQDN")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_INVALID_VERSION,          "Invalid app version")
+TAOS_DEFINE_ERROR(TSDB_CODE_RPC_PORT_EADDRINUSE,           "port already in use")
 
 //common & util
 TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_MEMORY,                "Out of Memory")

From f6910e525a62fbcc1611d3d599d0da7cd11e44db Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 27 Apr 2022 15:00:45 +0800
Subject: [PATCH 2/2] fix(rpc): fix duplicat port error

---
 source/libs/transport/src/transCli.c | 1 -
 1 file changed, 1 deletion(-)

diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c
index ab57f7d017..4115db5796 100644
--- a/source/libs/transport/src/transCli.c
+++ b/source/libs/transport/src/transCli.c
@@ -902,7 +902,6 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
         cliDestroy((uv_handle_t*)pConn->stream);
         return -1;
       }
-
     } else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) {
       if (pResp->contLen == 0) {
         pEpSet->inUse = (pEpSet->inUse++) % pEpSet->numOfEps;