refactor rpc

source/libs/transport/src/transCli.c

@@ -52,7 +52,14 @@ typedef struct SClientObj {
   SCliThrdObj** pThreadObj;
 } SClientObj;
 
+typedef struct SConnList {
+  queue conn;
+} SConnList;
+
 // conn pool
+// add expire timeout and capacity limit
+static void*     connCacheCreate(int size);
+static void*     connCacheDestroy(void* cache);
 static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port);
 static void      addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn);
 
@@ -81,6 +88,53 @@ static void clientProcessData(SCliConn* conn) {
 }
 static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
 
+static void* connCacheCreate(int size) {
+  SHashObj* cache = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
+  return false;
+}
+static void* connCacheDestroy(void* cache) {
+  SConnList* connList = taosHashIterate((SHashObj*)cache, NULL);
+  while (!QUEUE_IS_EMPTY(&connList->conn)) {
+    queue* h = QUEUE_HEAD(&connList->conn);
+    QUEUE_REMOVE(h);
+    SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
+    clientConnDestroy(c);
+  }
+  taosHashClear(cache);
+}
+
+static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port) {
+  char key[128] = {0};
+  tstrncpy(key, ip, strlen(ip));
+  tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));
+
+  SHashObj*  pCache = cache;
+  SConnList* plist = taosHashGet(pCache, key, strlen(key));
+  if (plist == NULL) {
+    SConnList list;
+    plist = &list;
+    QUEUE_INIT(&plist->conn);
+    taosHashPut(pCache, key, strlen(key), plist, sizeof(*plist));
+  }
+
+  if (QUEUE_IS_EMPTY(&plist->conn)) {
+    return NULL;
+  }
+  queue* h = QUEUE_HEAD(&plist->conn);
+  QUEUE_REMOVE(h);
+  return QUEUE_DATA(h, SCliConn, conn);
+}
+static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn) {
+  char key[128] = {0};
+  tstrncpy(key, ip, strlen(ip));
+  tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));
+
+  SHashObj*  pCache = cache;
+  SConnList* plist = taosHashGet(pCache, key, strlen(key));
+  // list already create before
+  assert(plist != NULL);
+  QUEUE_PUSH(&plist->conn, &conn->conn);
+}
 static bool clientReadComplete(SConnBuffer* data) {
   STransMsgHead head;
   int32_t       headLen = sizeof(head);
@@ -206,15 +260,6 @@ static void clientConnCb(uv_connect_t* req, int status) {
   clientWrite(pConn);
 }
 
-static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port) {
-  // impl later
-
-  return NULL;
-}
-static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn) {
-  // impl later
-}
-
 static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
   uint64_t et = taosGetTimestampUs();
   uint64_t el = et - pMsg->st;

source/libs/transport/src/transSrv.c

@@ -210,8 +210,8 @@ static int uvAuthMsg(SConn* pConn, char* msg, int len) {
 
 // refers specifically to query or insert timeout
 static void uvHandleActivityTimeout(uv_timer_t* handle) {
-  // impl later
   SConn* conn = handle->data;
+  tDebug("%p timeout since no activity", conn);
 }
 
 static void uvProcessData(SConn* pConn) {
@@ -232,12 +232,13 @@ static void uvProcessData(SConn* pConn) {
 
   SRpcInfo* pRpc = (SRpcInfo*)p->shandle;
   // auth here
+  // auth should not do in rpc thread
 
-  int8_t code = uvAuthMsg(pConn, (char*)pHead, p->msgLen);
-  if (code != 0) {
-    terrno = code;
-    return;
-  }
+  // int8_t code = uvAuthMsg(pConn, (char*)pHead, p->msgLen);
+  // if (code != 0) {
+  //  terrno = code;
+  //  return;
+  //}
   pHead->code = htonl(pHead->code);
 
   int32_t dlen = 0;
@@ -248,7 +249,7 @@ static void uvProcessData(SConn* pConn) {
   } else {
     // impl later
   }
-  rpcMsg.contLen = rpcContLenFromMsg(pHead->msgLen);
+  rpcMsg.contLen = transContLenFromMsg(pHead->msgLen);
   rpcMsg.pCont = pHead->content;
   rpcMsg.msgType = pHead->msgType;
   rpcMsg.code = pHead->code;
@@ -318,6 +319,9 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
       return;
     }
     uv_buf_t wb = uv_buf_init(conn->writeBuf.buf, conn->writeBuf.len);
+
+    uv_timer_stop(conn->pTimer);
+
     uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
   }
 }

source/libs/transport/test/rclient.c

@@ -40,6 +40,7 @@ static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
   if (pEpSet) pInfo->epSet = *pEpSet;
 
   rpcFreeCont(pMsg->pCont);
+  // tsem_post(&pInfo->rspSem);
   tsem_post(&pInfo->rspSem);
 }
 
@@ -60,6 +61,7 @@ static void *sendRequest(void *param) {
     // tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
     rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg, NULL);
     if (pInfo->num % 20000 == 0) tInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num);
+    // tsem_wait(&pInfo->rspSem);
     tsem_wait(&pInfo->rspSem);
   }