diff --git a/include/libs/transport/trpc.h b/include/libs/transport/trpc.h index fdc9368b76..8dfd736df6 100644 --- a/include/libs/transport/trpc.h +++ b/include/libs/transport/trpc.h @@ -71,6 +71,10 @@ typedef struct SRpcInit { // call back to keep conn or not bool (*pfp)(void *parent, tmsg_t msgType); + // to support Send messages multiple times on a link + // + void* (*mfp)(void *parent, tmsg_t msgType); + void *parent; } SRpcInit; @@ -89,6 +93,9 @@ void rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp) int rpcReportProgress(void *pConn, char *pCont, int contLen); void rpcCancelRequest(int64_t rid); +// just release client conn to rpc instance, no close sock +void rpcReleaseHandle(void *handle); + void rpcRefHandle(void *handle, int8_t type); void rpcUnrefHandle(void *handle, int8_t type); diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index f2ac77fe61..99f890d3a0 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -120,6 +120,10 @@ typedef struct { // SEpSet* pSet; // for synchronous API } SRpcReqContext; +typedef SRpcMsg STransMsg; +typedef SRpcInfo STrans; +typedef SRpcConnInfo STransHandleInfo; + typedef struct { SEpSet epSet; // ip list provided by app void* ahandle; // handle provided by app @@ -134,8 +138,8 @@ typedef struct { int8_t connType; // connection type int64_t rid; // refId returned by taosAddRef - SRpcMsg* pRsp; // for synchronous API - tsem_t* pSem; // for synchronous API + STransMsg* pRsp; // for synchronous API + tsem_t* pSem; // for synchronous API int hThrdIdx; char* ip; @@ -249,4 +253,15 @@ void transUnrefSrvHandle(void* handle); void transRefCliHandle(void* handle); void transUnrefCliHandle(void* handle); +void transSendRequest(void* shandle, const char* ip, uint32_t port, STransMsg* pMsg); +void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pMsg, STransMsg* pRsp); +void transSendResponse(const STransMsg* pMsg); +int transGetConnInfo(void* thandle, STransHandleInfo* pInfo); + +void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle); +void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle); + +void transCloseClient(void* arg); +void transCloseServer(void* arg); + #endif diff --git a/source/libs/transport/inc/transportInt.h b/source/libs/transport/inc/transportInt.h index 4e4dcf7aa4..3924a5cf1a 100644 --- a/source/libs/transport/inc/transportInt.h +++ b/source/libs/transport/inc/transportInt.h @@ -64,6 +64,7 @@ typedef struct { void (*cfp)(void* parent, SRpcMsg*, SEpSet*); int (*afp)(void* parent, char* user, char* spi, char* encrypt, char* secret, char* ckey); bool (*pfp)(void* parent, tmsg_t msgType); + void* (*mfp)(void* parent, tmsg_t msgType); int32_t refCount; void* parent; diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index 4d244665c7..015018f73f 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -18,8 +18,9 @@ #include "transComm.h" void* (*taosInitHandle[])(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) = { - taosInitServer, taosInitClient}; -void (*taosCloseHandle[])(void* arg) = {taosCloseServer, taosCloseClient}; + transInitServer, transInitClient}; + +void (*taosCloseHandle[])(void* arg) = {transCloseServer, transCloseClient}; void* rpcOpen(const SRpcInit* pInit) { SRpcInfo* pRpc = calloc(1, sizeof(SRpcInfo)); @@ -34,11 +35,12 @@ void* rpcOpen(const SRpcInit* pInit) { pRpc->cfp = pInit->cfp; pRpc->afp = pInit->afp; pRpc->pfp = pInit->pfp; + pRpc->mfp = pInit->mfp; if (pInit->connType == TAOS_CONN_SERVER) { pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads; } else { - pRpc->numOfThreads = pInit->numOfThreads; + pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads; } pRpc->connType = pInit->connType; @@ -116,6 +118,24 @@ int32_t rpcInit() { return 0; } +void rpcSendRequest(void *shandle, const SEpSet *pEpSet, SRpcMsg* pMsg, int64_t *pRid) { + char* ip = (char*)(pEpSet->eps[pEpSet->inUse].fqdn); + uint32_t port = pEpSet->eps[pEpSet->inUse].port; + transSendRequest(shandle, ip, port, pMsg); +} +void rpcSendRecv(void* shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + char* ip = (char*)(pEpSet->eps[pEpSet->inUse].fqdn); + uint32_t port = pEpSet->eps[pEpSet->inUse].port; + transSendRecv(shandle, ip, port, pMsg, pRsp); +} + +void rpcSendResponse(const SRpcMsg *pMsg) { + transSendResponse(pMsg); +} +int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) { + return transGetConnInfo((void *)thandle, pInfo); +} + void rpcCleanup(void) { // impl later // @@ -129,6 +149,7 @@ void rpcRefHandle(void* handle, int8_t type) { assert(type == TAOS_CONN_SERVER || type == TAOS_CONN_CLIENT); (*taosRefHandle[type])(handle); } + void rpcUnrefHandle(void* handle, int8_t type) { assert(type == TAOS_CONN_SERVER || type == TAOS_CONN_CLIENT); (*taosUnRefHandle[type])(handle); diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 727845b7a9..4af02a982e 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -42,7 +42,7 @@ typedef struct SCliConn { typedef struct SCliMsg { STransConnCtx* ctx; - SRpcMsg msg; + STransMsg msg; queue q; uint64_t st; } SCliMsg; @@ -105,9 +105,9 @@ static void cliHandleExcept(SCliConn* conn); static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliSendQuit(SCliThrdObj* thrd); -static void destroyUserdata(SRpcMsg* userdata); +static void destroyUserdata(STransMsg* userdata); -static int cliRBChoseIdx(SRpcInfo* pTransInst); +static int cliRBChoseIdx(STrans* pTransInst); static void destroyCmsg(SCliMsg* cmsg); static void transDestroyConnCtx(STransConnCtx* ctx); @@ -118,11 +118,11 @@ static void destroyThrdObj(SCliThrdObj* pThrd); #define CONN_HOST_THREAD_INDEX(conn) (conn ? ((SCliConn*)conn)->hThrdIdx : -1) #define CONN_PERSIST_TIME(para) (para * 1000 * 10) -#define CONN_GET_INST_LABEL(conn) (((SRpcInfo*)(((SCliThrdObj*)conn->hostThrd)->pTransInst))->label) +#define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrdObj*)conn->hostThrd)->pTransInst))->label) #define CONN_HANDLE_THREAD_QUIT(conn, thrd) \ do { \ if (thrd->quit) { \ - cliHandleExcept(conn); \ + cliHandleExcept(conn); \ goto _RETURE; \ } \ } while (0) @@ -130,20 +130,25 @@ static void destroyThrdObj(SCliThrdObj* pThrd); #define CONN_HANDLE_BROKEN(conn) \ do { \ if (conn->broken) { \ - cliHandleExcept(conn); \ + cliHandleExcept(conn); \ goto _RETURE; \ } \ } while (0); +#define CONN_SET_PERSIST_BY_APP(conn) \ + do { \ + if (conn->persist == false) { \ + conn->persist = true; \ + transRefCliHandle(conn); \ + } \ + } while (0) +#define CONN_NO_PERSIST_BY_APP(conn) ((conn)->persist == false) + static void* cliWorkThread(void* arg); -static void* cliNotifyApp() {} -static void cliHandleResp(SCliConn* conn) { - SCliMsg* pMsg = conn->data; - STransConnCtx* pCtx = pMsg->ctx; - +void cliHandleResp(SCliConn* conn) { SCliThrdObj* pThrd = conn->hostThrd; - SRpcInfo* pTransInst = pThrd->pTransInst; + STrans* pTransInst = pThrd->pTransInst; STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf); pHead->code = htonl(pHead->code); @@ -152,19 +157,29 @@ static void cliHandleResp(SCliConn* conn) { // buf's mem alread translated to rpcMsg.pCont transClearBuffer(&conn->readBuf); - SRpcMsg rpcMsg = {0}; + STransMsg rpcMsg = {0}; rpcMsg.contLen = transContLenFromMsg(pHead->msgLen); rpcMsg.pCont = transContFromHead((char*)pHead); rpcMsg.code = pHead->code; rpcMsg.msgType = pHead->msgType; - rpcMsg.ahandle = pCtx->ahandle; + rpcMsg.ahandle = NULL; - if (pTransInst->pfp != NULL && (pTransInst->pfp)(pTransInst->parent, rpcMsg.msgType)) { + SCliMsg* pMsg = conn->data; + STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL; + if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(conn)) { + rpcMsg.ahandle = pTransInst->mfp ? (*pTransInst->mfp)(pTransInst->parent, rpcMsg.msgType) : NULL; + } else { + rpcMsg.ahandle = pCtx ? pCtx->ahandle : NULL; + } + // if (rpcMsg.ahandle == NULL) { + // tDebug("%s cli conn %p handle except", CONN_GET_INST_LABEL(conn), conn); + // return; + //} + + if (pTransInst->pfp != NULL && (*pTransInst->pfp)(pTransInst->parent, rpcMsg.msgType)) { rpcMsg.handle = conn; - transRefCliHandle(conn); - - conn->persist = 1; - tDebug("cli conn %p persist by app", conn); + CONN_SET_PERSIST_BY_APP(conn); + tDebug("%s cli conn %p ref by app", CONN_GET_INST_LABEL(conn), conn); } tDebug("%s cli conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pTransInst->label, conn, @@ -173,7 +188,7 @@ static void cliHandleResp(SCliConn* conn) { conn->secured = pHead->secured; - if (pCtx->pSem == NULL) { + if (pCtx == NULL || pCtx->pSem == NULL) { tTrace("%s cli conn %p handle resp", pTransInst->label, conn); (pTransInst->cfp)(pTransInst->parent, &rpcMsg, NULL); } else { @@ -184,8 +199,7 @@ static void cliHandleResp(SCliConn* conn) { uv_read_start((uv_stream_t*)conn->stream, cliAllocBufferCb, cliRecvCb); - // user owns conn->persist = 1 - if (conn->persist == 0) { + if (CONN_NO_PERSIST_BY_APP(conn)) { addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn); } destroyCmsg(conn->data); @@ -196,24 +210,32 @@ static void cliHandleResp(SCliConn* conn) { // uv_timer_start((uv_timer_t*)&pThrd->timer, cliTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0); } } -static void cliHandleExcept(SCliConn* pConn) { + +void cliHandleExcept(SCliConn* pConn) { if (pConn->data == NULL) { - // handle conn except in conn pool - transUnrefCliHandle(pConn); - return; + if (pConn->broken == true || CONN_NO_PERSIST_BY_APP(pConn)) { + transUnrefCliHandle(pConn); + return; + } } SCliThrdObj* pThrd = pConn->hostThrd; - SRpcInfo* pTransInst = pThrd->pTransInst; + STrans* pTransInst = pThrd->pTransInst; SCliMsg* pMsg = pConn->data; - STransConnCtx* pCtx = pMsg->ctx; + STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL; - SRpcMsg rpcMsg = {0}; - rpcMsg.ahandle = pCtx->ahandle; + STransMsg rpcMsg = {0}; rpcMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL; - rpcMsg.msgType = pMsg->msg.msgType + 1; + rpcMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0; + rpcMsg.ahandle = NULL; - if (pCtx->pSem == NULL) { + if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(pConn)) { + rpcMsg.ahandle = pTransInst->mfp ? (*pTransInst->mfp)(pTransInst->parent, rpcMsg.msgType) : NULL; + } else { + rpcMsg.ahandle = pCtx ? pCtx->ahandle : NULL; + } + + if (pCtx == NULL || pCtx->pSem == NULL) { tTrace("%s cli conn %p handle resp", pTransInst->label, pConn); (pTransInst->cfp)(pTransInst->parent, &rpcMsg, NULL); } else { @@ -228,9 +250,9 @@ static void cliHandleExcept(SCliConn* pConn) { transUnrefCliHandle(pConn); } -static void cliTimeoutCb(uv_timer_t* handle) { +void cliTimeoutCb(uv_timer_t* handle) { SCliThrdObj* pThrd = handle->data; - SRpcInfo* pRpc = pThrd->pTransInst; + STrans* pRpc = pThrd->pTransInst; int64_t currentTime = pThrd->nextTimeout; tTrace("%s, cli conn timeout, try to remove expire conn from conn pool", pRpc->label); @@ -252,11 +274,12 @@ static void cliTimeoutCb(uv_timer_t* handle) { pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime); uv_timer_start(handle, cliTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0); } -static void* createConnPool(int size) { + +void* createConnPool(int size) { // thread local, no lock return taosHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK); } -static void* destroyConnPool(void* pool) { +void* destroyConnPool(void* pool) { SConnList* connList = taosHashIterate((SHashObj*)pool, NULL); while (connList != NULL) { while (!QUEUE_IS_EMPTY(&connList->conn)) { @@ -301,7 +324,7 @@ static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) { tstrncpy(key + strlen(key), (char*)(&port), sizeof(port)); tTrace("cli conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap); - SRpcInfo* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst; + STrans* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst; conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime); SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key)); @@ -358,6 +381,7 @@ static SCliConn* cliCreateConn(SCliThrdObj* pThrd) { QUEUE_INIT(&conn->conn); conn->hostThrd = pThrd; + conn->persist = false; conn->broken = false; transRefCliHandle(conn); return conn; @@ -395,16 +419,16 @@ static void cliSendCb(uv_write_t* req, int status) { uv_read_start((uv_stream_t*)pConn->stream, cliAllocBufferCb, cliRecvCb); } -static void cliSend(SCliConn* pConn) { +void cliSend(SCliConn* pConn) { CONN_HANDLE_BROKEN(pConn); SCliMsg* pCliMsg = pConn->data; STransConnCtx* pCtx = pCliMsg->ctx; SCliThrdObj* pThrd = pConn->hostThrd; - SRpcInfo* pTransInst = pThrd->pTransInst; + STrans* pTransInst = pThrd->pTransInst; - SRpcMsg* pMsg = (SRpcMsg*)(&pCliMsg->msg); + STransMsg* pMsg = (STransMsg*)(&pCliMsg->msg); STransMsgHead* pHead = transHeadFromCont(pMsg->pCont); int msgLen = transMsgLenFromCont(pMsg->contLen); @@ -442,7 +466,8 @@ static void cliSend(SCliConn* pConn) { _RETURE: return; } -static void cliConnCb(uv_connect_t* req, int status) { + +void cliConnCb(uv_connect_t* req, int status) { // impl later SCliConn* pConn = req->data; if (status != 0) { @@ -472,11 +497,11 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) { pThrd->quit = true; uv_stop(pThrd->loop); } -static SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { + +SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { SCliConn* conn = NULL; if (pMsg->msg.handle != NULL) { conn = (SCliConn*)(pMsg->msg.handle); - transUnrefCliHandle(conn); if (conn != NULL) { tTrace("%s cli conn %p reused", CONN_GET_INST_LABEL(conn), conn); } @@ -487,13 +512,14 @@ static SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { } return conn; } -static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { + +void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { uint64_t et = taosGetTimestampUs(); uint64_t el = et - pMsg->st; - tTrace("%s cli msg tran time cost: %" PRIu64 "us", ((SRpcInfo*)pThrd->pTransInst)->label, el); + tTrace("%s cli msg tran time cost: %" PRIu64 "us", ((STrans*)pThrd->pTransInst)->label, el); STransConnCtx* pCtx = pMsg->ctx; - SRpcInfo* pTransInst = pThrd->pTransInst; + STrans* pTransInst = pThrd->pTransInst; SCliConn* conn = cliGetConn(pMsg, pThrd); if (conn != NULL) { @@ -514,6 +540,7 @@ static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { tTrace("%s cli conn %p try to connect to %s:%d", pTransInst->label, conn, pMsg->ctx->ip, pMsg->ctx->port); uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, cliConnCb); } + conn->hThrdIdx = pCtx->hThrdIdx; } static void cliAsyncCb(uv_async_t* handle) { @@ -522,7 +549,7 @@ static void cliAsyncCb(uv_async_t* handle) { SCliMsg* pMsg = NULL; // batch process to avoid to lock/unlock frequently - queue wq; + queue wq; pthread_mutex_lock(&item->mtx); QUEUE_MOVE(&item->qmsg, &wq); pthread_mutex_unlock(&item->mtx); @@ -551,10 +578,10 @@ static void* cliWorkThread(void* arg) { uv_run(pThrd->loop, UV_RUN_DEFAULT); } -void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) { +void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) { SCliObj* cli = calloc(1, sizeof(SCliObj)); - SRpcInfo* pRpc = shandle; + STrans* pRpc = shandle; memcpy(cli->label, label, strlen(label)); cli->numOfThreads = numOfThreads; cli->pThreadObj = (SCliThrdObj**)calloc(cli->numOfThreads, sizeof(SCliThrdObj*)); @@ -573,7 +600,7 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, return cli; } -static void destroyUserdata(SRpcMsg* userdata) { +static void destroyUserdata(STransMsg* userdata) { if (userdata->pCont == NULL) { return; } @@ -629,12 +656,20 @@ static void transDestroyConnCtx(STransConnCtx* ctx) { free(ctx); } // -static void cliSendQuit(SCliThrdObj* thrd) { +void cliSendQuit(SCliThrdObj* thrd) { // cli can stop gracefully SCliMsg* msg = calloc(1, sizeof(SCliMsg)); transSendAsync(thrd->asyncPool, &msg->q); } -void taosCloseClient(void* arg) { +int cliRBChoseIdx(STrans* pTransInst) { + int64_t index = pTransInst->index; + if (pTransInst->index++ >= pTransInst->numOfThreads) { + pTransInst->index = 0; + } + return index % pTransInst->numOfThreads; +} + +void transCloseClient(void* arg) { SCliObj* cli = arg; for (int i = 0; i < cli->numOfThreads; i++) { cliSendQuit(cli->pThreadObj[i]); @@ -643,13 +678,6 @@ void taosCloseClient(void* arg) { free(cli->pThreadObj); free(cli); } -static int cliRBChoseIdx(SRpcInfo* pTransInst) { - int64_t index = pTransInst->index; - if (pTransInst->index++ >= pTransInst->numOfThreads) { - pTransInst->index = 0; - } - return index % pTransInst->numOfThreads; -} void transRefCliHandle(void* handle) { if (handle == NULL) { return; @@ -665,17 +693,11 @@ void transUnrefCliHandle(void* handle) { if (ref == 0) { cliDestroyConn((SCliConn*)handle, true); } - - // unref cli handle } -void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid) { - // impl later - char* ip = (char*)(pEpSet->eps[pEpSet->inUse].fqdn); - uint32_t port = pEpSet->eps[pEpSet->inUse].port; - SRpcInfo* pTransInst = (SRpcInfo*)shandle; - - int index = CONN_HOST_THREAD_INDEX(pMsg->handle); +void transSendRequest(void* shandle, const char* ip, uint32_t port, STransMsg* pMsg) { + STrans* pTransInst = (STrans*)shandle; + int index = CONN_HOST_THREAD_INDEX((SCliConn*)pMsg->handle); if (index == -1) { index = cliRBChoseIdx(pTransInst); } @@ -683,6 +705,7 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* if (transCompressMsg(pMsg->pCont, pMsg->contLen, &flen)) { // imp later } + tDebug("send request at thread:%d %p", index, pMsg); STransConnCtx* pCtx = calloc(1, sizeof(STransConnCtx)); pCtx->ahandle = pMsg->ahandle; pCtx->msgType = pMsg->msgType; @@ -701,14 +724,9 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[index]; transSendAsync(thrd->asyncPool, &(cliMsg->q)); } - -void rpcSendRecv(void* shandle, SEpSet* pEpSet, SRpcMsg* pReq, SRpcMsg* pRsp) { - char* ip = (char*)(pEpSet->eps[pEpSet->inUse].fqdn); - uint32_t port = pEpSet->eps[pEpSet->inUse].port; - - SRpcInfo* pTransInst = (SRpcInfo*)shandle; - - int index = CONN_HOST_THREAD_INDEX(pReq->handle); +void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pReq, STransMsg* pRsp) { + STrans* pTransInst = (STrans*)shandle; + int index = CONN_HOST_THREAD_INDEX(pReq->handle); if (index == -1) { index = cliRBChoseIdx(pTransInst); } @@ -734,7 +752,6 @@ void rpcSendRecv(void* shandle, SEpSet* pEpSet, SRpcMsg* pReq, SRpcMsg* pRsp) { tsem_wait(pSem); tsem_destroy(pSem); free(pSem); - - return; } + #endif diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c index 1abca9ad97..c236a69f4e 100644 --- a/source/libs/transport/src/transSrv.c +++ b/source/libs/transport/src/transSrv.c @@ -37,8 +37,7 @@ typedef struct SSrvConn { struct sockaddr_in addr; struct sockaddr_in locaddr; - // SRpcMsg sendMsg; - // del later + char secured; int spi; char info[64]; @@ -49,7 +48,7 @@ typedef struct SSrvConn { typedef struct SSrvMsg { SSrvConn* pConn; - SRpcMsg msg; + STransMsg msg; queue q; } SSrvMsg; @@ -207,20 +206,20 @@ static void uvHandleReq(SSrvConn* pConn) { pConn->inType = pHead->msgType; - SRpcInfo* pRpc = (SRpcInfo*)p->shandle; + STrans* pRpc = (STrans*)p->shandle; pHead->code = htonl(pHead->code); int32_t dlen = 0; if (transDecompressMsg(NULL, 0, NULL)) { // add compress later - // pHead = rpcDecompressRpcMsg(pHead); + // pHead = rpcDecompresSTransMsg(pHead); } else { pHead->msgLen = htonl(pHead->msgLen); // impl later // } - SRpcMsg rpcMsg; + STransMsg rpcMsg; rpcMsg.contLen = transContLenFromMsg(pHead->msgLen); rpcMsg.pCont = pHead->content; rpcMsg.msgType = pHead->msgType; @@ -260,7 +259,7 @@ void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) { } tError("server conn %p read error: %s", conn, uv_err_name(nread)); - if (nread < 0 || nread == UV_EOF) { + if (nread < 0) { conn->broken = true; transUnrefSrvHandle(conn); @@ -318,8 +317,8 @@ static void uvPrepareSendData(SSrvMsg* smsg, uv_buf_t* wb) { // impl later; tTrace("server conn %p prepare to send resp", smsg->pConn); - SSrvConn* pConn = smsg->pConn; - SRpcMsg* pMsg = &smsg->msg; + SSrvConn* pConn = smsg->pConn; + STransMsg* pMsg = &smsg->msg; if (pMsg->pCont == 0) { pMsg->pCont = (void*)rpcMallocCont(0); pMsg->contLen = 0; @@ -547,7 +546,7 @@ static bool addHandleToWorkloop(void* arg) { return false; } - // SRpcInfo* pRpc = pThrd->shandle; + // STrans* pRpc = pThrd->shandle; uv_pipe_init(pThrd->loop, pThrd->pipe, 1); uv_pipe_open(pThrd->pipe, pThrd->fd); @@ -668,7 +667,7 @@ static int transAddAuthPart(SSrvConn* pConn, char* msg, int msgLen) { return msgLen; } -void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) { +void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) { SServerObj* srv = calloc(1, sizeof(SServerObj)); srv->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t)); srv->numOfThreads = numOfThreads; @@ -720,7 +719,7 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, return srv; End: - taosCloseServer(srv); + transCloseServer(srv); return NULL; } @@ -740,7 +739,7 @@ void sendQuitToWorkThrd(SWorkThrdObj* pThrd) { transSendAsync(pThrd->asyncPool, &srvMsg->q); } -void taosCloseServer(void* arg) { +void transCloseServer(void* arg) { // impl later SServerObj* srv = arg; for (int i = 0; i < srv->numOfThreads; i++) { @@ -786,7 +785,7 @@ void transUnrefSrvHandle(void* handle) { } // unref srv handle } -void rpcSendResponse(const SRpcMsg* pMsg) { +void transSendResponse(const STransMsg* pMsg) { if (pMsg->handle == NULL) { return; } @@ -799,14 +798,12 @@ void rpcSendResponse(const SRpcMsg* pMsg) { tTrace("server conn %p start to send resp", pConn); transSendAsync(pThrd->asyncPool, &srvMsg->q); } - -int rpcGetConnInfo(void* thandle, SRpcConnInfo* pInfo) { - SSrvConn* pConn = thandle; - +int transGetConnInfo(void* thandle, STransHandleInfo* pInfo) { + SSrvConn* pConn = thandle; struct sockaddr_in addr = pConn->addr; + pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr); pInfo->clientPort = ntohs(addr.sin_port); - tstrncpy(pInfo->user, pConn->user, sizeof(pInfo->user)); return 0; }