diff --git a/include/common/tglobal.h b/include/common/tglobal.h index bd5e74387e..6db2ffb7ea 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -139,7 +139,7 @@ extern int32_t tsTtlPushInterval; extern int32_t tsGrantHBInterval; extern int32_t tsUptimeInterval; -#define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize) +//#define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize) int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc); diff --git a/include/libs/transport/trpc.h b/include/libs/transport/trpc.h index 655c903c0b..f8a08288de 100644 --- a/include/libs/transport/trpc.h +++ b/include/libs/transport/trpc.h @@ -82,6 +82,9 @@ typedef struct SRpcInit { int8_t connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS int32_t idleTime; // milliseconds, 0 means idle timer is disabled + const int32_t compressSize; // -1: no compress, 0 : all data compressed, size: compress data if larger than size + const int8_t encryption; // encrypt or not + // the following is for client app ecurity only char *user; // user name @@ -115,10 +118,9 @@ typedef struct { } SRpcCtx; int32_t rpcInit(); +void rpcCleanup(); -void rpcCleanup(); void *rpcOpen(const SRpcInit *pRpc); - void rpcClose(void *); void rpcCloseImpl(void *); void *rpcMallocCont(int32_t contLen); diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index c6f3066be7..b83f84e3f2 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -200,15 +200,13 @@ typedef enum { ConnNormal, ConnAcquire, ConnRelease, ConnBroken, ConnInPool } Co #define TRANS_MSG_OVERHEAD (sizeof(STransMsgHead)) #define transHeadFromCont(cont) ((STransMsgHead*)((char*)cont - sizeof(STransMsgHead))) -#define transContFromHead(msg) (msg + sizeof(STransMsgHead)) +#define transContFromHead(msg) (((char*)msg) + sizeof(STransMsgHead)) #define transMsgLenFromCont(contLen) (contLen + sizeof(STransMsgHead)) #define transContLenFromMsg(msgLen) (msgLen - sizeof(STransMsgHead)); #define transIsReq(type) (type & 1U) #define transLabel(trans) ((STrans*)trans)->label -void transFreeMsg(void* msg); -// typedef struct SConnBuffer { char* buf; int len; @@ -415,6 +413,10 @@ void transThreadOnce(); void transInit(); void transCleanup(); +void transFreeMsg(void* msg); +int32_t transCompressMsg(char* msg, int32_t len); +int32_t transDecompressMsg(char** msg, int32_t len); + int32_t transOpenRefMgt(int size, void (*func)(void*)); void transCloseRefMgt(int32_t refMgt); int64_t transAddExHandle(int32_t refMgt, void* p); diff --git a/source/libs/transport/inc/transportInt.h b/source/libs/transport/inc/transportInt.h index 6aeeffa192..b9167501e2 100644 --- a/source/libs/transport/inc/transportInt.h +++ b/source/libs/transport/inc/transportInt.h @@ -16,9 +16,7 @@ #ifndef _TD_TRANSPORT_INT_H_ #define _TD_TRANSPORT_INT_H_ -#ifdef USE_UV #include -#endif #include "lz4.h" #include "os.h" #include "taoserror.h" @@ -34,8 +32,6 @@ extern "C" { #endif -#ifdef USE_UV - void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle); void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle); @@ -51,19 +47,20 @@ typedef struct { char label[TSDB_LABEL_LEN]; char user[TSDB_UNI_LEN]; // meter ID + int32_t compressSize; // -1: no compress, 0 : all data compressed, size: compress data if larger than size + int8_t encryption; // encrypt or not + void (*cfp)(void* parent, SRpcMsg*, SEpSet*); bool (*retry)(int32_t code, tmsg_t msgType); bool (*startTimer)(int32_t code, tmsg_t msgType); - int index; + int index; void* parent; void* tcphandle; // returned handle from TCP initialization int64_t refId; TdThreadMutex mutex; } SRpcInfo; -#endif // USE_LIBUV - #ifdef __cplusplus } #endif diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index 9e0a8f2a10..e5a783292d 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -45,6 +45,10 @@ void* rpcOpen(const SRpcInit* pInit) { if (pInit->label) { tstrncpy(pRpc->label, pInit->label, TSDB_LABEL_LEN); } + + pRpc->compressSize = pInit->compressSize; + pRpc->encryption = pInit->encryption; + // register callback handle pRpc->cfp = pInit->cfp; pRpc->retry = pInit->rfp; @@ -130,9 +134,6 @@ void* rpcReallocCont(void* ptr, int32_t contLen) { return st + TRANS_MSG_OVERHEAD; } -int32_t rpcReportProgress(void* pConn, char* pCont, int32_t contLen) { return -1; } -void rpcCancelRequest(int64_t rid) { return; } - int rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid) { return transSendRequest(shandle, pEpSet, pMsg, NULL); } diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index bd99a23267..819eb51f1f 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -319,13 +319,18 @@ void cliHandleResp(SCliConn* conn) { } STransMsgHead* pHead = NULL; - if (transDumpFromBuffer(&conn->readBuf, (char**)&pHead) <= 0) { + + int32_t msgLen = transDumpFromBuffer(&conn->readBuf, (char**)&pHead); + if (msgLen <= 0) { tDebug("%s conn %p recv invalid packet ", CONN_GET_INST_LABEL(conn), conn); return; } + + if (transDecompressMsg((char**)&pHead, msgLen) < 0) { + tDebug("%s conn %p recv invalid packet, failed to decompress", CONN_GET_INST_LABEL(conn), conn); + } pHead->code = htonl(pHead->code); pHead->msgLen = htonl(pHead->msgLen); - if (cliRecvReleaseReq(conn, pHead)) { return; } @@ -553,7 +558,7 @@ static void addConnToPool(void* pool, SCliConn* conn) { if (conn->list->size >= 50) { STaskArg* arg = taosMemoryCalloc(1, sizeof(STaskArg)); arg->param1 = conn; - arg->param2 = thrd; + arg->param2 = NULL; STrans* pTransInst = thrd->pTransInst; conn->task = transDQSched(thrd->timeoutQueue, doCloseIdleConn, arg, CONN_PERSIST_TIME(pTransInst->idleTime)); @@ -772,20 +777,19 @@ void cliSend(SCliConn* pConn) { memcpy(pHead->user, pTransInst->user, strlen(pTransInst->user)); pHead->traceId = pMsg->info.traceId; pHead->magicNum = htonl(TRANS_MAGIC_NUM); + if (pHead->persist == 1) { + CONN_SET_PERSIST_BY_APP(pConn); + } STraceId* trace = &pMsg->info.traceId; tGDebug("%s conn %p %s is sent to %s, local info %s, len:%d", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pHead->msgType), pConn->dst, pConn->src, pMsg->contLen); - if (pHead->persist == 1) { - CONN_SET_PERSIST_BY_APP(pConn); - } - if (pTransInst->startTimer != NULL && pTransInst->startTimer(0, pMsg->msgType)) { uv_timer_t* timer = taosArrayGetSize(pThrd->timerList) > 0 ? *(uv_timer_t**)taosArrayPop(pThrd->timerList) : NULL; if (timer == NULL) { - tDebug("no avaiable timer, create"); timer = taosMemoryCalloc(1, sizeof(uv_timer_t)); + tDebug("no available timer, create a timer %p", timer); uv_timer_init(pThrd->loop, timer); } timer->data = pConn; @@ -795,6 +799,11 @@ void cliSend(SCliConn* pConn) { uv_timer_start((uv_timer_t*)pConn->timer, cliReadTimeoutCb, TRANS_READ_TIMEOUT, 0); } + if (pTransInst->compressSize != -1 && pTransInst->compressSize > pMsg->contLen) { + msgLen = transCompressMsg(pMsg->pCont, pMsg->contLen) + sizeof(STransMsgHead); + pHead->msgLen = (int32_t)htonl((uint32_t)msgLen); + } + uv_buf_t wb = uv_buf_init((char*)pHead, msgLen); uv_write_t* req = transReqQueuePush(&pConn->wreqQueue); @@ -1275,17 +1284,13 @@ FORCE_INLINE int cliRBChoseIdx(STrans* pTransInst) { } static FORCE_INLINE void doDelayTask(void* param) { STaskArg* arg = param; - SCliMsg* pMsg = arg->param1; - SCliThrd* pThrd = arg->param2; + cliHandleReq((SCliMsg*)arg->param1, (SCliThrd*)arg->param2); taosMemoryFree(arg); - - cliHandleReq(pMsg, pThrd); } static void doCloseIdleConn(void* param) { STaskArg* arg = param; SCliConn* conn = arg->param1; - SCliThrd* pThrd = arg->param2; tTrace("%s conn %p idle, close it", CONN_GET_INST_LABEL(conn), conn); conn->task = NULL; cliDestroyConn(conn, true); diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index e86314d4bd..eaa2e6c661 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -23,52 +23,64 @@ static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT; static int32_t refMgt; static int32_t instMgt; -bool transCompressMsg(char* msg, int32_t len, int32_t* flen) { - return false; - // SRpcHead* pHead = rpcHeadFromCont(pCont); - bool succ = false; - int overhead = sizeof(STransCompMsg); - if (!NEEDTO_COMPRESSS_MSG(len)) { - return succ; - } +int32_t transCompressMsg(char* msg, int32_t len) { + int32_t ret = 0; + int compHdr = sizeof(STransCompMsg); + STransMsgHead* pHead = transHeadFromCont(msg); - char* buf = taosMemoryMalloc(len + overhead + 8); // 8 extra bytes + char* buf = taosMemoryMalloc(len + compHdr + 8); // 8 extra bytes if (buf == NULL) { tError("failed to allocate memory for rpc msg compression, contLen:%d", len); - *flen = len; - return succ; + ret = len; + return ret; } - int32_t clen = LZ4_compress_default(msg, buf, len, len + overhead); - tDebug("compress rpc msg, before:%d, after:%d, overhead:%d", len, clen, overhead); + int32_t clen = LZ4_compress_default(msg, buf, len, len + compHdr); + tDebug("compress rpc msg, before:%d, after:%d, overhead:%d", len, clen, compHdr); /* * only the compressed size is less than the value of contLen - overhead, the compression is applied * The first four bytes is set to 0, the second four bytes are utilized to keep the original length of message */ - if (clen > 0 && clen < len - overhead) { + if (clen > 0 && clen < len - compHdr) { STransCompMsg* pComp = (STransCompMsg*)msg; pComp->reserved = 0; pComp->contLen = htonl(len); - memcpy(msg + overhead, buf, clen); + memcpy(msg + compHdr, buf, clen); tDebug("compress rpc msg, before:%d, after:%d", len, clen); - *flen = clen + overhead; - succ = true; + ret = clen + compHdr; + pHead->comp = 1; } else { - *flen = len; - succ = false; + ret = len; + pHead->comp = 0; } taosMemoryFree(buf); - return succ; + return ret; } -bool transDecompressMsg(char* msg, int32_t len, int32_t* flen) { - // impl later - return false; - STransCompMsg* pComp = (STransCompMsg*)msg; +int32_t transDecompressMsg(char** msg, int32_t len) { + STransMsgHead* pHead = (STransMsgHead*)(*msg); + if (pHead->comp == 0) return 0; - int overhead = sizeof(STransCompMsg); - int clen = 0; - return false; + char* pCont = transContFromHead(pHead); + STransCompMsg* pComp = (STransCompMsg*)pCont; + int32_t oriLen = htonl(pComp->contLen); + + char* buf = taosMemoryCalloc(1, oriLen + sizeof(STransMsgHead)); + STransMsgHead* pNewHead = (STransMsgHead*)buf; + + int32_t decompLen = LZ4_decompress_safe(pCont + sizeof(STransCompMsg), pNewHead->content, + len - sizeof(STransMsgHead) - sizeof(STransCompMsg), oriLen); + memcpy((char*)pNewHead, (char*)pHead, sizeof(STransMsgHead)); + + pNewHead->msgLen = htonl(oriLen + sizeof(STransMsgHead)); + + taosMemoryFree(pHead); + + *msg = buf; + if (decompLen != oriLen) { + return -1; + } + return 0; } void transFreeMsg(void* msg) { diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index c63f8f398b..99855d7557 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -186,16 +186,22 @@ static void uvHandleActivityTimeout(uv_timer_t* handle) { static bool uvHandleReq(SSvrConn* pConn) { STrans* pTransInst = pConn->pTransInst; - STransMsgHead* msg = NULL; - int msgLen = transDumpFromBuffer(&pConn->readBuf, (char**)&msg); + STransMsgHead* pHead = NULL; + + int msgLen = transDumpFromBuffer(&pConn->readBuf, (char**)&pHead); if (msgLen <= 0) { tError("%s conn %p read invalid packet", transLabel(pTransInst), pConn); return false; } - STransMsgHead* pHead = (STransMsgHead*)msg; + if (transDecompressMsg((char**)&pHead, msgLen) < 0) { + tDebug("%s conn %p recv invalid packet, failed to decompress", transLabel(pTransInst), pConn); + return false; + } + pHead->code = htonl(pHead->code); pHead->msgLen = htonl(pHead->msgLen); + memcpy(pConn->user, pHead->user, strlen(pHead->user)); if (uvRecvReleaseReq(pConn, pHead)) { @@ -399,17 +405,22 @@ static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) { pHead->release = smsg->type == Release ? 1 : 0; pHead->code = htonl(pMsg->code); + pHead->msgLen = htonl(pMsg->contLen + sizeof(STransMsgHead)); char* msg = (char*)pHead; int32_t len = transMsgLenFromCont(pMsg->contLen); - STrans* pTransInst = pConn->pTransInst; + STrans* pTransInst = pConn->pTransInst; + if (pTransInst->compressSize != -1 && pTransInst->compressSize > pMsg->contLen) { + len = transCompressMsg(pMsg->pCont, pMsg->contLen) + sizeof(STransMsgHead); + pHead->msgLen = (int32_t)htonl((uint32_t)len); + } + STraceId* trace = &pMsg->info.traceId; tGDebug("%s conn %p %s is sent to %s, local info:%s, len:%d", transLabel(pTransInst), pConn, TMSG_INFO(pHead->msgType), pConn->dst, pConn->src, pMsg->contLen); - pHead->msgLen = htonl(len); - wb->base = msg; + wb->base = (char*)pHead; wb->len = len; }