add client

source/libs/transport/src/trans.c

@@ -42,7 +42,18 @@ void* rpcOpen(const SRpcInit* pInit) {
   return pRpc;
 }
 void  rpcClose(void* arg) { return; }
-void* rpcMallocCont(int contLen) { return NULL; }
+void* rpcMallocCont(int contLen) {
+  int size = contLen + RPC_MSG_OVERHEAD;
+
+  char* start = (char*)calloc(1, (size_t)size);
+  if (start == NULL) {
+    tError("failed to malloc msg, size:%d", size);
+    return NULL;
+  } else {
+    tTrace("malloc mem:%p size:%d", start, size);
+  }
+  return start + sizeof(SRpcReqContext) + sizeof(SRpcHead);
+}
 void  rpcFreeCont(void* cont) { return; }
 void* rpcReallocCont(void* ptr, int contLen) { return NULL; }
 

source/libs/transport/src/transCli.c

@@ -55,6 +55,10 @@ static void* clientThread(void* arg);
 static void clientWriteCb(uv_write_t* req, int status) {
   // impl later
 }
+static void clientFailedCb(uv_handle_t* handle) {
+  // impl later
+  tDebug("close handle");
+}
 static void clientReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
   // impl later
 }
@@ -68,8 +72,10 @@ static void clientConnCb(struct uv_connect_s* req, int status) {
   if (status != 0) {
     // call user fp later
     tError("failed to connect server(%s, %d), errmsg: %s", fqdn, port, uv_strerror(status));
+    uv_close((uv_handle_t*)req->handle, clientFailedCb);
     return;
   }
+  assert(pConn->stream == req->handle);
 
   // impl later
 }
@@ -123,19 +129,6 @@ static void clientAsyncCb(uv_async_t* handle) {
 
 static void* clientThread(void* arg) {
   SCliThrdObj* pThrd = (SCliThrdObj*)arg;
-
-  QUEUE_INIT(&pThrd->msg);
-  pthread_mutex_init(&pThrd->msgMtx, NULL);
-
-  // QUEUE_INIT(&pThrd->clientCache);
-
-  pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t));
-  uv_loop_init(pThrd->loop);
-
-  pThrd->cliAsync = malloc(sizeof(uv_async_t));
-  uv_async_init(pThrd->loop, pThrd->cliAsync, clientAsyncCb);
-  pThrd->cliAsync->data = pThrd;
-
   uv_run(pThrd->loop, UV_RUN_DEFAULT);
 }
 
@@ -146,14 +139,25 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
   cli->pThreadObj = (SCliThrdObj**)calloc(cli->numOfThreads, sizeof(SCliThrdObj*));
 
   for (int i = 0; i < cli->numOfThreads; i++) {
-    SCliThrdObj* thrd = (SCliThrdObj*)calloc(1, sizeof(SCliThrdObj));
+    SCliThrdObj* pThrd = (SCliThrdObj*)calloc(1, sizeof(SCliThrdObj));
+    QUEUE_INIT(&pThrd->msg);
+    pthread_mutex_init(&pThrd->msgMtx, NULL);
+
+    // QUEUE_INIT(&pThrd->clientCache);
+
+    pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t));
+    uv_loop_init(pThrd->loop);
+
+    pThrd->cliAsync = malloc(sizeof(uv_async_t));
+    uv_async_init(pThrd->loop, pThrd->cliAsync, clientAsyncCb);
+    pThrd->cliAsync->data = pThrd;
 
-    thrd->shandle = shandle;
-    int err = pthread_create(&thrd->thread, NULL, clientThread, (void*)(thrd));
+    pThrd->shandle = shandle;
+    int err = pthread_create(&pThrd->thread, NULL, clientThread, (void*)(pThrd));
     if (err == 0) {
       tDebug("sucess to create tranport-client thread %d", i);
     }
-    cli->pThreadObj[i] = thrd;
+    cli->pThreadObj[i] = pThrd;
   }
   return cli;
 }

source/libs/transport/test/rclient.c

@@ -34,7 +34,8 @@ typedef struct {
 
 static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
   SInfo *pInfo = (SInfo *)pMsg->ahandle;
-  tDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen, pMsg->code);
+  tDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen,
+         pMsg->code);
 
   if (pEpSet) pInfo->epSet = *pEpSet;
 
@@ -185,7 +186,8 @@ int main(int argc, char *argv[]) {
   // float usedTime = (endTime - startTime) / 1000.0f;  // mseconds
 
   // tInfo("it takes %.3f mseconds to send %d requests to server", usedTime, numOfReqs * appThreads);
-  // tInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0 * numOfReqs * appThreads / usedTime, msgSize);
+  // tInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0 * numOfReqs * appThreads / usedTime,
+  // msgSize);
 
   int ch = getchar();
   UNUSED(ch);