diff --git a/src/rpc/src/trpc.c b/src/rpc/src/trpc.c index e3db06d952972e4e4069e7feade2afec3ae8ef8c..0333f2b3bf4e550043ceb6a4a67b6bf7f72962da 100755 --- a/src/rpc/src/trpc.c +++ b/src/rpc/src/trpc.c @@ -68,14 +68,16 @@ typedef struct { } SRpcInfo; typedef struct { - SRpcIpSet ipSet; - void *ahandle; // handle provided by app SRpcInfo *pRpc; // associated SRpcInfo + SRpcIpSet ipSet; // ip list provided by app + void *ahandle; // handle provided by app char msgType; // message type - char *pCont; // content provided by app - int contLen; // content length - int numOfRetry; // number of retry for different servers + uint8_t *pCont; // content provided by app + int32_t contLen; // content length int32_t code; // error code + int16_t numOfTry; // number of try for different servers + int8_t oldIndex; // server IP index passed by app + int8_t redirect; // flag to indicate redirect char msg[0]; // RpcHeader starts from here } SRpcReqContext; @@ -131,6 +133,11 @@ typedef struct { uint8_t content[0]; // message body starts from here } SRpcHeader; +typedef struct { + int32_t reserved; + int32_t contLen; +} SRpcComp; + typedef struct { uint32_t timeStamp; uint8_t auth[TSDB_AUTH_LEN]; @@ -333,6 +340,7 @@ void rpcSendRequest(void *shandle, SRpcIpSet ipSet, char type, void *pCont, int pContext->contLen = contLen; pContext->pCont = pCont; pContext->msgType = type; + pContext->oldIndex = ipSet.index; rpcSendReqToServer(pRpc, pContext); @@ -779,11 +787,15 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHeader *pHeader) { taosAddConnIntoCache(pRpc->pCache, pConn, pConn->peerIp, pConn->peerPort, pConn->meterId); if (code == TSDB_CODE_REDIRECT) { + pContext->redirect = 1; + pContext->numOfTry = 0; memcpy(&pContext->ipSet, pHeader->content, sizeof(pContext->ipSet)); rpcSendReqToServer(pRpc, pContext); } else { rpcFreeOutMsg(rpcHeaderFromCont(pContext->pCont)); // free the request msg - (*(pRpc->cfp))(pHeader->msgType, pCont, contLen, pContext->ahandle, pContext->ipSet.index); + if ( pContext->ipSet.index != pContext->oldIndex || pContext->redirect ) + (*pRpc->ufp)(pContext->ahandle, pContext->ipSet); + (*pRpc->cfp)(pHeader->msgType, pCont, contLen, pContext->ahandle, pContext->ipSet.index); } } } @@ -853,6 +865,7 @@ static void rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) { int msgLen = rpcMsgLenFromCont(pContext->contLen); char msgType = pContext->msgType; + pContext->numOfTry++; SRpcConn *pConn = rpcSetConnToServer(pRpc, pContext->ipSet); if (pConn == NULL) { pContext->code = terrno; @@ -922,7 +935,7 @@ static void rpcProcessConnError(void *param, void *id) { SRpcReqContext *pContext = (SRpcReqContext *)param; SRpcInfo *pRpc = pContext->pRpc; - if ( pContext->numOfRetry >= pContext->ipSet.numOfIps ) { + if ( pContext->numOfTry >= pContext->ipSet.numOfIps ) { rpcFreeOutMsg(rpcHeaderFromCont(pContext->pCont)); // free the request msg (*(pRpc->cfp))(pContext->msgType+1, NULL, 0, pContext->ahandle, pContext->code); } else { @@ -1004,11 +1017,6 @@ static void rpcFreeOutMsg(void *msg) { free(req); } -typedef struct { - int32_t reserved; - int32_t contLen; -} SRpcComp; - static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) { SRpcHeader *pHeader = rpcHeaderFromCont(pCont); int32_t finalLen = 0; diff --git a/src/rpc/test/unittest.c b/src/rpc/test/rclient.c similarity index 65% rename from src/rpc/test/unittest.c rename to src/rpc/test/rclient.c index 262c93a62b95432ab81b68fa76d184446ea4cc8c..a139a591c028de44dc8ee2bb09869c29b3b8475b 100644 --- a/src/rpc/test/unittest.c +++ b/src/rpc/test/rclient.c @@ -19,6 +19,10 @@ #include "trpc.h" #include +void processMsg(char type, void *pCont, int contLen, void *ahandle, int32_t code) { + dPrint("response is received, type:%d, contLen:%d code:%d, ahandle:%p", type, contLen, code, ahandle); +} + int32_t main(int32_t argc, char *argv[]) { dPrint("unit test for rpc module"); @@ -28,18 +32,28 @@ int32_t main(int32_t argc, char *argv[]) { rpcInit.localPort = 7000; rpcInit.label = "unittest"; rpcInit.numOfThreads = 1; - rpcInit.fp = NULL; + rpcInit.cfp = processMsg; rpcInit.sessions = 1000; - rpcInit.connType = TAOS_CONN_SOCKET_TYPE_S(); + rpcInit.connType = TAOS_CONN_UDPC; rpcInit.idleTime = 2000; - void *pConn = rpcOpen(&rpcInit); - if (pConn != NULL) { - dPrint("conection is opened"); - } else { + void *pRpc = rpcOpen(&rpcInit); + if (pRpc == NULL) { dError("failed to initialize rpc"); + return -1; } + SRpcIpSet ipSet; + ipSet.numOfIps = 2; + ipSet.index = 0; + ipSet.ip[0] = inet_addr("127.0.0.1"); + ipSet.ip[1] = inet_addr("192.168.0.1"); + + void *cont = rpcMallocCont(100); + rpcSendRequest(pRpc, ipSet, 1, cont, 100, 1); + + getchar(); + return 0; } diff --git a/src/rpc/test/rserver.c b/src/rpc/test/rserver.c new file mode 100644 index 0000000000000000000000000000000000000000..1fc7b6e336bb00e17e2e7d9a5141d3fae38d3c1d --- /dev/null +++ b/src/rpc/test/rserver.c @@ -0,0 +1,64 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +//#define _DEFAULT_SOURCE +#include "os.h" +#include "tlog.h" +#include "trpc.h" +#include + +void processMsg(char type, void *pCont, int contLen, void *ahandle, int32_t code) { + dPrint("request is received, type:%d, contLen:%d", type, contLen); + void *rsp = rpcMallocCont(128); + rpcSendResponse(ahandle, 1, rsp, 128); + rpcFreeCont(pCont); +} + +int32_t main(int32_t argc, char *argv[]) { + dPrint("unit test for rpc module"); + + SRpcInit rpcInit; + memset(&rpcInit, 0, sizeof(rpcInit)); + rpcInit.localIp = "0.0.0.0"; + rpcInit.localPort = 7000; + rpcInit.label = "unittest"; + rpcInit.numOfThreads = 1; + rpcInit.cfp = processMsg; + rpcInit.sessions = 1000; + rpcInit.connType = TAOS_CONN_UDPS; + rpcInit.idleTime = 2000; + + void *pRpc = rpcOpen(&rpcInit); + if (pRpc == NULL) { + dError("failed to initialize rpc"); + return -1; + } + +/* + SRpcIpSet ipSet; + ipSet.numOfIps = 2; + ipSet.index = 0; + ipSet.ip[0] = inet_addr("127.0.0.1"); + ipSet.ip[1] = inet_addr("192.168.0.1"); +*/ + + dPrint("server is running..."); + + getchar(); + + return 0; +} + +