diff --git a/src/inc/trpc.h b/src/inc/trpc.h index 710e9bc5e6a0f2e7f56fc447b1a74ccc8204ec1f..48280061fe249758d2d8dc30cbc059eaf0080351 100644 --- a/src/inc/trpc.h +++ b/src/inc/trpc.h @@ -27,6 +27,7 @@ extern "C" { #define TAOS_CONN_CLIENT 1 extern int tsRpcHeadSize; +extern int tsRpcMaxUdpSize; typedef struct { int8_t inUse; diff --git a/src/rpc/inc/rpcHead.h b/src/rpc/inc/rpcHead.h index 09c66d33ae34cd5374994d3615930d330ec99660..9bbcd60fc4d386b9afc1b52c82b9cbe645193f7a 100644 --- a/src/rpc/inc/rpcHead.h +++ b/src/rpc/inc/rpcHead.h @@ -26,6 +26,8 @@ extern "C" { #define RPC_CONN_TCPC 3 #define RPC_CONN_TCP 2 +extern int tsRpcOverhead; + typedef struct { void *msg; int msgLen; diff --git a/src/rpc/src/rpcClient.c b/src/rpc/src/rpcClient.c index b616dae82f17bdf910ccbb405ddd84c0973eb45d..b362b1ba4494ad6dc0ba1b077f10ad06d3c18afb 100644 --- a/src/rpc/src/rpcClient.c +++ b/src/rpc/src/rpcClient.c @@ -236,6 +236,7 @@ static void *taosReadTcpData(void *param) { STcpFd *pFdObj; struct epoll_event events[maxTcpEvents]; SRecvInfo recvInfo; + SRpcHead rpcHead; while (1) { pthread_mutex_lock(&pTcp->mutex); @@ -260,37 +261,24 @@ static void *taosReadTcpData(void *param) { continue; } - void *buffer = malloc(1024); - if (NULL == buffer) { - tTrace("%s TCP malloc(size:1024) fail\n", pTcp->label); - taosCleanUpTcpFdObj(pFdObj); - continue; - } - - int headLen = taosReadMsg(pFdObj->fd, buffer, sizeof(SRpcHead)); + int headLen = taosReadMsg(pFdObj->fd, &rpcHead, sizeof(SRpcHead)); if (headLen != sizeof(SRpcHead)) { tError("%s read error, headLen:%d", pTcp->label, headLen); - tfree(buffer); taosCleanUpTcpFdObj(pFdObj); continue; } - int dataLen = (int32_t)htonl((uint32_t)((SRpcHead *)buffer)->msgLen); - if (dataLen > 1024) { - void *b = realloc(buffer, (size_t)dataLen); - if (NULL == b) { - tTrace("%s TCP malloc(size:%d) fail\n", pTcp->label, dataLen); - tfree(buffer); - taosCleanUpTcpFdObj(pFdObj); - continue; - } - buffer = b; + int32_t msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen); + char *buffer = (char *)malloc((size_t)msgLen + tsRpcOverhead); + if (NULL == buffer) { + tTrace("%s TCP malloc(size:%d) fail\n", pTcp->label, msgLen); + taosCleanUpTcpFdObj(pFdObj); + continue; } - int leftLen = dataLen - headLen; - int retLen = taosReadMsg(pFdObj->fd, buffer + headLen, leftLen); - - // tTrace("%s TCP data is received, ip:%s port:%u len:%d", pTcp->label, pFdObj->ipstr, pFdObj->port, dataLen); + char *msg = buffer + tsRpcOverhead; + int32_t leftLen = msgLen - headLen; + int32_t retLen = taosReadMsg(pFdObj->fd, msg + headLen, leftLen); if (leftLen != retLen) { tError("%s read error, leftLen:%d retLen:%d", pTcp->label, leftLen, retLen); @@ -299,8 +287,11 @@ static void *taosReadTcpData(void *param) { continue; } - recvInfo.msg = buffer; - recvInfo.msgLen = dataLen; + // tTrace("%s TCP data is received, ip:%s:%u len:%d", pTcp->label, pFdObj->ipstr, pFdObj->port, msgLen); + + memcpy(msg, &rpcHead, sizeof(SRpcHead)); + recvInfo.msg = msg; + recvInfo.msgLen = msgLen; recvInfo.ip = pFdObj->ip; recvInfo.port = pFdObj->port; recvInfo.shandle = pTcp->shandle; diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c index 05141b0235e73656d7f71cde54d0e05db4b40a27..5b1f24f6647ca534803d3e4c67c5fe54c0fe9d01 100755 --- a/src/rpc/src/rpcMain.c +++ b/src/rpc/src/rpcMain.c @@ -122,10 +122,12 @@ typedef struct _RpcConn { } SRpcConn; int tsRpcProgressTime = 10; // milliseocnds +int tsRpcMaxUdpSize = 15000; // bytes; // not configurable int tsRpcMaxRetry; int tsRpcHeadSize; +int tsRpcOverhead; // server:0 client:1 tcp:2 udp:0 #define RPC_CONN_UDPS 0 @@ -188,7 +190,7 @@ static void rpcProcessRetryTimer(void *, void *); static void rpcProcessIdleTimer(void *param, void *tmrId); static void rpcProcessProgressTimer(void *param, void *tmrId); -static void rpcFreeOutMsg(void *msg); +static void rpcFreeMsg(void *msg); static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen); static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead); static int rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen); @@ -201,6 +203,7 @@ void *rpcOpen(SRpcInit *pInit) { tsRpcMaxRetry = tsRpcMaxTime * 1000 / tsRpcProgressTime; tsRpcHeadSize = RPC_MSG_OVERHEAD; + tsRpcOverhead = sizeof(SRpcReqContext); pRpc = (SRpcInfo *)calloc(1, sizeof(SRpcInfo)); if (pRpc == NULL) return NULL; @@ -313,8 +316,8 @@ void *rpcMallocCont(int contLen) { void rpcFreeCont(void *cont) { if ( cont ) { - char *msg = ((char *)cont) - sizeof(SRpcHead); - free(msg); + char *temp = ((char *)cont) - sizeof(SRpcHead) - sizeof(SRpcReqContext); + free(temp); } } @@ -351,7 +354,7 @@ void rpcSendRequest(void *shandle, SRpcIpSet *pIpSet, char type, void *pCont, in pContext->oldInUse = pIpSet->inUse; pContext->connType = RPC_CONN_UDPC; - if (contLen > 16000) pContext->connType = RPC_CONN_TCPC; + if (contLen > tsRpcMaxUdpSize) pContext->connType = RPC_CONN_TCPC; // connection type is application specific. // for TDengine, all the query, show commands shall have TCP connection @@ -406,7 +409,7 @@ void rpcSendResponse(void *handle, int32_t code, void *pCont, int contLen) { pConn->inType = 0; // response message is released until new response is sent - rpcFreeOutMsg(pConn->pRspMsg); + rpcFreeMsg(pConn->pRspMsg); pConn->pRspMsg = msg; pConn->rspMsgLen = msgLen; if (pHead->content[0] == TSDB_CODE_ACTION_IN_PROGRESS) pConn->inTranId--; @@ -442,6 +445,13 @@ void rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) { strcpy(pInfo->user, pConn->user); } +static void rpcFreeMsg(void *msg) { + if ( msg ) { + char *temp = (char *)msg - sizeof(SRpcReqContext); + free(temp); + } +} + static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerIpStr, uint16_t peerPort, int8_t connType) { SRpcConn *pConn; @@ -490,7 +500,7 @@ static void rpcCloseConn(void *thandle) { char hashstr[40] = {0}; sprintf(hashstr, "%x:%x:%x:%d", pConn->peerIp, pConn->peerUid, pConn->peerId, pConn->connType); taosDeleteStrHash(pRpc->hash, hashstr); - rpcFreeOutMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg + rpcFreeMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg pConn->pRspMsg = NULL; pConn->inType = 0; pConn->inTranId = 0; @@ -822,7 +832,7 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) { } } - if ( terrno ) free (pRecv->msg); + if (terrno) rpcFreeMsg(pRecv->msg); return pConn; } @@ -855,7 +865,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) { if ( pRpc->ufp && (pContext->ipSet.inUse != pContext->oldInUse || pContext->redirect) ) (*pRpc->ufp)(pContext->ahandle, &pContext->ipSet); // notify the update of ipSet (*pRpc->cfp)(pHead->msgType, pCont, contLen, pContext->ahandle, code); - rpcFreeOutMsg(rpcHeadFromCont(pContext->pCont)); // free the request msg + rpcFreeCont(pContext->pCont); // free the request msg } } } @@ -996,8 +1006,8 @@ static void rpcProcessConnError(void *param, void *id) { tTrace("%s connection error happens", pRpc->label); if ( pContext->numOfTry >= pContext->ipSet.numOfIps ) { - rpcFreeOutMsg(rpcHeadFromCont(pContext->pCont)); // free the request msg (*(pRpc->cfp))(pContext->msgType+1, NULL, 0, pContext->ahandle, pContext->code); + rpcFreeCont(pContext->pCont); // free the request msg } else { // move to next IP pContext->ipSet.inUse++; @@ -1127,7 +1137,8 @@ static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead) { int contLen = htonl(pComp->contLen); // prepare the temporary buffer to decompress message - pNewHead = (SRpcHead *)malloc(contLen + RPC_MSG_OVERHEAD); + char *temp = (char *)malloc(contLen + RPC_MSG_OVERHEAD); + pNewHead = (SRpcHead *)(temp + sizeof(SRpcReqContext)); // reserve SRpcReqContext if (pNewHead) { int compLen = rpcContLenFromMsg(pHead->msgLen) - overhead; @@ -1136,7 +1147,7 @@ static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead) { memcpy(pNewHead, pHead, sizeof(SRpcHead)); pNewHead->msgLen = rpcMsgLenFromCont(origLen); - free(pHead); // free the compressed message buffer + rpcFreeMsg(pHead); // free the compressed message buffer pHead = pNewHead; tTrace("decompress rpc msg, compLen:%d, after:%d", compLen, contLen); } else { diff --git a/src/rpc/src/rpcServer.c b/src/rpc/src/rpcServer.c index 010a9ae69bbc7b1794d25e277d5e8266cc140f3f..1aadabc5f73f3a124261ca5af1c794dbd297d9d3 100644 --- a/src/rpc/src/rpcServer.c +++ b/src/rpc/src/rpcServer.c @@ -191,8 +191,9 @@ static void taosProcessTcpData(void *param) { int i, fdNum; SFdObj * pFdObj; struct epoll_event events[maxEvents]; - SRecvInfo recvInfo; + SRecvInfo recvInfo; pThreadObj = (SThreadObj *)param; + SRpcHead rpcHead; while (1) { pthread_mutex_lock(&pThreadObj->threadMutex); @@ -219,24 +220,24 @@ static void taosProcessTcpData(void *param) { continue; } - void *buffer = malloc(1024); - int headLen = taosReadMsg(pFdObj->fd, buffer, sizeof(SRpcHead)); - + int32_t headLen = taosReadMsg(pFdObj->fd, &rpcHead, sizeof(SRpcHead)); if (headLen != sizeof(SRpcHead)) { tError("%s read error, headLen:%d, errno:%d", pThreadObj->label, headLen, errno); taosCleanUpFdObj(pFdObj); - tfree(buffer); continue; } - int dataLen = (int32_t)htonl((uint32_t)((SRpcHead *)buffer)->msgLen); - if (dataLen > 1024) buffer = realloc(buffer, (size_t)dataLen); - - int leftLen = dataLen - headLen; - int retLen = taosReadMsg(pFdObj->fd, buffer + headLen, leftLen); + int32_t msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen); + char *buffer = malloc(msgLen + tsRpcOverhead); + if ( NULL == buffer) { + tError("%s TCP malloc(size:%d) fail\n", pThreadObj->label, msgLen); + taosCleanUpFdObj(pFdObj); + continue; + } - // tTrace("%s TCP data is received, ip:%s port:%u len:%d", - // pThreadObj->label, pFdObj->ipstr, pFdObj->port, dataLen); + char *msg = buffer + tsRpcOverhead; + int32_t leftLen = msgLen - headLen; + int32_t retLen = taosReadMsg(pFdObj->fd, msg + headLen, leftLen); if (leftLen != retLen) { tError("%s read error, leftLen:%d retLen:%d", pThreadObj->label, leftLen, retLen); @@ -245,8 +246,11 @@ static void taosProcessTcpData(void *param) { continue; } - recvInfo.msg = buffer; - recvInfo.msgLen = dataLen; + // tTrace("%s TCP data is received, ip:%s:%u len:%d", pTcp->label, pFdObj->ipstr, pFdObj->port, msgLen); + + memcpy(msg, &rpcHead, sizeof(SRpcHead)); + recvInfo.msg = msg; + recvInfo.msgLen = msgLen; recvInfo.ip = pFdObj->ip; recvInfo.port = pFdObj->port; recvInfo.shandle = pThreadObj->shandle; diff --git a/src/rpc/src/rpcUdp.c b/src/rpc/src/rpcUdp.c index 245a98b2c7383c72629b64d179ce535bfb870faf..471b1bcc723b8e1557b1ab158a7570c39bf1f466 100644 --- a/src/rpc/src/rpcUdp.c +++ b/src/rpc/src/rpcUdp.c @@ -40,12 +40,12 @@ typedef struct { char label[12]; // copy from udpConnSet; pthread_t thread; pthread_mutex_t mutex; - void * tmrCtrl; // copy from UdpConnSet; - void * hash; - void * shandle; // handle passed by upper layer during server initialization - void * pSet; - void *(*processData)(SRecvInfo *pRecv); - char buffer[RPC_MAX_UDP_SIZE]; // buffer to receive data + void *tmrCtrl; // copy from UdpConnSet; + void *hash; + void *shandle; // handle passed by upper layer during server initialization + void *pSet; + void *(*processData)(SRecvInfo *pRecv); + char *buffer; // buffer to receive data } SUdpConn; typedef struct { @@ -96,15 +96,15 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v pSet->fp = fp; strcpy(pSet->label, label); - // if ( tsUdpDelay ) { - char udplabel[12]; - sprintf(udplabel, "%s.b", label); - pSet->tmrCtrl = taosTmrInit(RPC_MAX_UDP_CONNS * threads, 5, 5000, udplabel); - if (pSet->tmrCtrl == NULL) { - tError("%s failed to initialize tmrCtrl") taosCleanUpUdpConnection(pSet); - return NULL; + if ( tsUdpDelay ) { + char udplabel[12]; + sprintf(udplabel, "%s.b", label); + pSet->tmrCtrl = taosTmrInit(RPC_MAX_UDP_CONNS * threads, 5, 5000, udplabel); + if (pSet->tmrCtrl == NULL) { + tError("%s failed to initialize tmrCtrl") taosCleanUpUdpConnection(pSet); + return NULL; + } } - // } pthread_attr_init(&thAttr); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); @@ -120,6 +120,13 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v return NULL; } + pConn->buffer = malloc(RPC_MAX_UDP_SIZE); + if (NULL == pConn->buffer) { + tError("%s failed to malloc recv buffer", label); + taosCleanUpUdpConnection(pSet); + return NULL; + } + struct sockaddr_in sin; unsigned int addrlen = sizeof(sin); if (getsockname(pConn->fd, (struct sockaddr *)&sin, &addrlen) == 0 && sin.sin_family == AF_INET && @@ -128,14 +135,6 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v } strcpy(pConn->label, label); - - if (pthread_create(&pConn->thread, &thAttr, taosRecvUdpData, pConn) != 0) { - tError("%s failed to create thread to process UDP data, reason:%s", label, strerror(errno)); - taosCloseSocket(pConn->fd); - taosCleanUpUdpConnection(pSet); - return NULL; - } - pConn->shandle = shandle; pConn->processData = fp; pConn->index = i; @@ -146,6 +145,14 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v pthread_mutex_init(&pConn->mutex, NULL); pConn->tmrCtrl = pSet->tmrCtrl; } + + if (pthread_create(&pConn->thread, &thAttr, taosRecvUdpData, pConn) != 0) { + tError("%s failed to create thread to process UDP data, reason:%s", label, strerror(errno)); + taosCloseSocket(pConn->fd); + taosCleanUpUdpConnection(pSet); + return NULL; + } + ++pSet->threads; } @@ -164,6 +171,7 @@ void taosCleanUpUdpConnection(void *handle) { for (int i = 0; i < pSet->threads; ++i) { pConn = pSet->udpConn + i; pConn->signature = NULL; + free(pConn->buffer); pthread_cancel(pConn->thread); taosCloseSocket(pConn->fd); if (pConn->hash) { @@ -210,7 +218,7 @@ static void *taosRecvUdpData(void *param) { tTrace("%s UDP thread is created, index:%d", pConn->label, pConn->index); while (1) { - dataLen = recvfrom(pConn->fd, pConn->buffer, sizeof(pConn->buffer), 0, (struct sockaddr *)&sourceAdd, &addLen); + dataLen = recvfrom(pConn->fd, pConn->buffer, RPC_MAX_UDP_SIZE, 0, (struct sockaddr *)&sourceAdd, &addLen); tTrace("%s msg is recv from 0x%x:%hu len:%d", pConn->label, sourceAdd.sin_addr.s_addr, ntohs(sourceAdd.sin_port), dataLen); @@ -235,9 +243,15 @@ static void *taosRecvUdpData(void *param) { break; } - char *data = malloc((size_t)msgLen); - memcpy(data, msg, (size_t)msgLen); - recvInfo.msg = data; + char *tmsg = malloc((size_t)msgLen + tsRpcOverhead); + if (NULL == tmsg) { + tError("%s failed to allocate memory, size:%d", pConn->label, msgLen); + break; + } + + tmsg += tsRpcOverhead; // overhead for SRpcReqContext + memcpy(tmsg, msg, (size_t)msgLen); + recvInfo.msg = tmsg; recvInfo.msgLen = msgLen; recvInfo.ip = sourceAdd.sin_addr.s_addr; recvInfo.port = port;