xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch '2.0' of https://github.com/taosdata/TDengine into 2.0

This commit is contained in:
hzcheng 2020-02-19 13:45:30 +00:00
parent f4959b038a 99849ec745
commit f20b026a14
20 changed files with 359 additions and 285 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/CMakeLists.txt
View File

@ -4,11 +4,11 @@ PROJECT(TDengine)
ADD_SUBDIRECTORY(os) ADD_SUBDIRECTORY(os)
ADD_SUBDIRECTORY(util) ADD_SUBDIRECTORY(util)
ADD_SUBDIRECTORY(rpc) ADD_SUBDIRECTORY(rpc)
ADD_SUBDIRECTORY(client) #ADD_SUBDIRECTORY(client)
ADD_SUBDIRECTORY(kit) #ADD_SUBDIRECTORY(kit)
ADD_SUBDIRECTORY(plugins) #ADD_SUBDIRECTORY(plugins)
ADD_SUBDIRECTORY(sdb) #ADD_SUBDIRECTORY(sdb)
ADD_SUBDIRECTORY(mnode) #ADD_SUBDIRECTORY(mnode)
ADD_SUBDIRECTORY(dnode) #ADD_SUBDIRECTORY(dnode)
ADD_SUBDIRECTORY(vnode) #ADD_SUBDIRECTORY(vnode)
ADD_SUBDIRECTORY(connector/jdbc) #ADD_SUBDIRECTORY(connector/jdbc)

4
src/client/CMakeLists.txt
View File

@ -4,8 +4,10 @@ PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc) INCLUDE_DIRECTORIES(inc)
INCLUDE_DIRECTORIES(jni) INCLUDE_DIRECTORIES(jni)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc) INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_ENTERPRISE_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc) INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
AUX_SOURCE_DIRECTORY(./src SRC) AUX_SOURCE_DIRECTORY(src SRC)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM)) IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/jni/linux) INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/jni/linux)

10
src/client/inc/tsclient.h
View File

@ -30,6 +30,7 @@ extern "C" {
#include "taosdef.h" #include "taosdef.h"
#include "tsqlfunction.h" #include "tsqlfunction.h"
#include "tutil.h" #include "tutil.h"
#include "trpc.h"
#define TSC_GET_RESPTR_BASE(res, _queryinfo, col, ord) \ #define TSC_GET_RESPTR_BASE(res, _queryinfo, col, ord) \
(res->data + tscFieldInfoGetOffset(_queryinfo, col) * res->numOfRows) (res->data + tscFieldInfoGetOffset(_queryinfo, col) * res->numOfRows)
@ -324,6 +325,7 @@ typedef struct _sql_obj {
int64_t stime; int64_t stime;
uint32_t queryId; uint32_t queryId;
void * thandle; void * thandle;
SRpcIpSet ipSet;
void * pStream; void * pStream;
void * pSubscription; void * pSubscription;
char * sqlstr; char * sqlstr;
@ -371,12 +373,6 @@ typedef struct _sstream {
struct _sstream *prev, *next; struct _sstream *prev, *next;
} SSqlStream; } SSqlStream;
typedef struct {
char numOfIps;
uint32_t ip[TSDB_MAX_MGMT_IPS];
char ipstr[TSDB_MAX_MGMT_IPS][TSDB_IPv4ADDR_LEN];
} SIpStrList;
// tscSql API // tscSql API
int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion); int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion);
@ -461,7 +457,7 @@ extern void * tscQhandle;
extern int tscKeepConn[]; extern int tscKeepConn[];
extern int tsInsertHeadSize; extern int tsInsertHeadSize;
extern int tscNumOfThreads; extern int tscNumOfThreads;
extern SIpStrList tscMgmtIpList; extern SRpcIpSet tscMgmtIpList;
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int numOfRows); typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int numOfRows);

1
src/client/src/tscAsync.c
View File

@ -25,6 +25,7 @@
#include "tscSQLParser.h" #include "tscSQLParser.h"
#include "tutil.h" #include "tutil.h"
#include "tnote.h" #include "tnote.h"
#include "tsched.h"
static void tscProcessFetchRow(SSchedMsg *pMsg); static void tscProcessFetchRow(SSchedMsg *pMsg);
static void tscAsyncQueryRowsForNextVnode(void *param, TAOS_RES *tres, int numOfRows); static void tscAsyncQueryRowsForNextVnode(void *param, TAOS_RES *tres, int numOfRows);

2
src/client/src/tscProfile.c
View File

@ -279,7 +279,7 @@ void tscKillConnection(STscObj *pObj) {
SSqlObj *pSql = pObj->sqlList; SSqlObj *pSql = pObj->sqlList;
while (pSql) { while (pSql) {
taosStopRpcConn(pSql->thandle); //taosStopRpcConn(pSql->thandle);
pSql = pSql->next; pSql = pSql->next;
} }

102
src/client/src/tscServer.c
View File

@ -31,10 +31,14 @@
#define TSC_MGMT_VNODE 999 #define TSC_MGMT_VNODE 999
SIpStrList tscMgmtIpList; SRpcIpSet tscMgmtIpList;
int tsMasterIndex = 0; int tsMasterIndex = 0;
int tsSlaveIndex = 1; int tsSlaveIndex = 1;
//temp
SRpcIpSet tscMgmtIpSet;
SRpcIpSet tscDnodeIpSet;
int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo) = {0}; int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo) = {0};
int (*tscProcessMsgRsp[TSDB_SQL_MAX])(SSqlObj *pSql); int (*tscProcessMsgRsp[TSDB_SQL_MAX])(SSqlObj *pSql);
@ -53,7 +57,7 @@ void tscPrintMgmtIp() {
tscError("invalid mgmt IP list:%d", tscMgmtIpList.numOfIps); tscError("invalid mgmt IP list:%d", tscMgmtIpList.numOfIps);
} else { } else {
for (int i = 0; i < tscMgmtIpList.numOfIps; ++i) { for (int i = 0; i < tscMgmtIpList.numOfIps; ++i) {
tscTrace("mgmt index:%d ip:%s", i, tscMgmtIpList.ipstr[i]); tscTrace("mgmt index:%d ip:%s", i, tscMgmtIpList.ipStr[i]);
} }
} }
} }
@ -62,7 +66,7 @@ void tscSetMgmtIpListFromCluster(SIpList *pIpList) {
tscMgmtIpList.numOfIps = pIpList->numOfIps; tscMgmtIpList.numOfIps = pIpList->numOfIps;
if (memcmp(tscMgmtIpList.ip, pIpList->ip, pIpList->numOfIps * 4) != 0) { if (memcmp(tscMgmtIpList.ip, pIpList->ip, pIpList->numOfIps * 4) != 0) {
for (int i = 0; i < pIpList->numOfIps; ++i) { for (int i = 0; i < pIpList->numOfIps; ++i) {
tinet_ntoa(tscMgmtIpList.ipstr[i], pIpList->ip[i]); tinet_ntoa(tscMgmtIpList.ipStr[i], pIpList->ip[i]);
tscMgmtIpList.ip[i] = pIpList->ip[i]; tscMgmtIpList.ip[i] = pIpList->ip[i];
} }
tscTrace("cluster mgmt IP list:"); tscTrace("cluster mgmt IP list:");
@ -73,9 +77,9 @@ void tscSetMgmtIpListFromCluster(SIpList *pIpList) {
void tscSetMgmtIpListFromEdge() { void tscSetMgmtIpListFromEdge() {
if (tscMgmtIpList.numOfIps != 2) { if (tscMgmtIpList.numOfIps != 2) {
tscMgmtIpList.numOfIps = 2; tscMgmtIpList.numOfIps = 2;
strcpy(tscMgmtIpList.ipstr[0], tsMasterIp); strcpy(tscMgmtIpList.ipStr[0], tsMasterIp);
tscMgmtIpList.ip[0] = inet_addr(tsMasterIp); tscMgmtIpList.ip[0] = inet_addr(tsMasterIp);
strcpy(tscMgmtIpList.ipstr[1], tsMasterIp); strcpy(tscMgmtIpList.ipStr[1], tsMasterIp);
tscMgmtIpList.ip[1] = inet_addr(tsMasterIp); tscMgmtIpList.ip[1] = inet_addr(tsMasterIp);
tscTrace("edge mgmt IP list:"); tscTrace("edge mgmt IP list:");
tscPrintMgmtIp(); tscPrintMgmtIp();
@ -168,7 +172,7 @@ void tscProcessActivityTimer(void *handle, void *tmrId) {
if (tscShouldFreeHeatBeat(pObj->pHb)) { if (tscShouldFreeHeatBeat(pObj->pHb)) {
tscTrace("%p free HB object and release connection, pConn:%p", pObj, pObj->pHb->thandle); tscTrace("%p free HB object and release connection, pConn:%p", pObj, pObj->pHb->thandle);
taosCloseRpcConn(pObj->pHb->thandle); //taosCloseRpcConn(pObj->pHb->thandle);
tscFreeSqlObj(pObj->pHb); tscFreeSqlObj(pObj->pHb);
tscCloseTscObj(pObj); tscCloseTscObj(pObj);
@ -178,6 +182,7 @@ void tscProcessActivityTimer(void *handle, void *tmrId) {
tscProcessSql(pObj->pHb); tscProcessSql(pObj->pHb);
} }
void tscGetConnToMgmt(SSqlObj *pSql, uint8_t *pCode) { void tscGetConnToMgmt(SSqlObj *pSql, uint8_t *pCode) {
STscObj *pTscObj = pSql->pTscObj; STscObj *pTscObj = pSql->pTscObj;
if (pSql->retry < tscGetMgmtConnMaxRetryTimes()) { if (pSql->retry < tscGetMgmtConnMaxRetryTimes()) {
@ -187,23 +192,24 @@ void tscGetConnToMgmt(SSqlObj *pSql, uint8_t *pCode) {
if (pSql->cmd.command > TSDB_SQL_READ && pSql->index == 0) pSql->index = 1; if (pSql->cmd.command > TSDB_SQL_READ && pSql->index == 0) pSql->index = 1;
void *thandle = taosGetConnFromCache(tscConnCache, tscMgmtIpList.ip[pSql->index], TSC_MGMT_VNODE, pTscObj->user); void *thandle = taosGetConnFromCache(tscConnCache, tscMgmtIpList.ip[pSql->index], TSC_MGMT_VNODE, pTscObj->user);
if (thandle == NULL) {
SRpcConnInit connInit;
memset(&connInit, 0, sizeof(connInit));
connInit.cid = 0;
connInit.sid = 0;
connInit.meterId = pSql->pTscObj->user;
connInit.peerId = 0;
connInit.shandle = pTscMgmtConn;
connInit.ahandle = pSql;
connInit.peerPort = tsMgmtShellPort;
connInit.spi = 1;
connInit.encrypt = 0;
connInit.secret = pSql->pTscObj->pass;
connInit.peerIp = tscMgmtIpList.ipstr[pSql->index]; // if (thandle == NULL) {
thandle = taosOpenRpcConn(&connInit, pCode); // SRpcConnInit connInit;
} // memset(&connInit, 0, sizeof(connInit));
// connInit.cid = 0;
// connInit.sid = 0;
// connInit.meterId = pSql->pTscObj->user;
// connInit.peerId = 0;
// connInit.shandle = pTscMgmtConn;
// connInit.ahandle = pSql;
// connInit.peerPort = tsMgmtShellPort;
// connInit.spi = 1;
// connInit.encrypt = 0;
// connInit.secret = pSql->pTscObj->pass;
//
// connInit.peerIp = tscMgmtIpList.ipstr[pSql->index];
// thandle = taosOpenRpcConn(&connInit, pCode);
// }
pSql->thandle = thandle; pSql->thandle = thandle;
pSql->ip = tscMgmtIpList.ip[pSql->index]; pSql->ip = tscMgmtIpList.ip[pSql->index];
@ -267,23 +273,23 @@ void tscGetConnToVnode(SSqlObj *pSql, uint8_t *pCode) {
void *thandle = void *thandle =
taosGetConnFromCache(tscConnCache, pVPeersDesc[pSql->index].ip, pVPeersDesc[pSql->index].vnode, pTscObj->user); taosGetConnFromCache(tscConnCache, pVPeersDesc[pSql->index].ip, pVPeersDesc[pSql->index].vnode, pTscObj->user);
if (thandle == NULL) { // if (thandle == NULL) {
SRpcConnInit connInit; // SRpcConnInit connInit;
tinet_ntoa(ipstr, pVPeersDesc[pSql->index].ip); // tinet_ntoa(ipstr, pVPeersDesc[pSql->index].ip);
memset(&connInit, 0, sizeof(connInit)); // memset(&connInit, 0, sizeof(connInit));
connInit.cid = vidIndex; // connInit.cid = vidIndex;
connInit.sid = 0; // connInit.sid = 0;
connInit.spi = 0; // connInit.spi = 0;
connInit.encrypt = 0; // connInit.encrypt = 0;
connInit.meterId = pSql->pTscObj->user; // connInit.meterId = pSql->pTscObj->user;
connInit.peerId = htonl((pVPeersDesc[pSql->index].vnode << TSDB_SHELL_VNODE_BITS)); // connInit.peerId = htonl((pVPeersDesc[pSql->index].vnode << TSDB_SHELL_VNODE_BITS));
connInit.shandle = pVnodeConn; // connInit.shandle = pVnodeConn;
connInit.ahandle = pSql; // connInit.ahandle = pSql;
connInit.peerIp = ipstr; // connInit.peerIp = ipstr;
connInit.peerPort = tsVnodeShellPort; // connInit.peerPort = tsVnodeShellPort;
thandle = taosOpenRpcConn(&connInit, pCode); // thandle = taosOpenRpcConn(&connInit, pCode);
vidIndex = (vidIndex + 1) % tscNumOfThreads; // vidIndex = (vidIndex + 1) % tscNumOfThreads;
} // }
pSql->thandle = thandle; pSql->thandle = thandle;
pSql->ip = pVPeersDesc[pSql->index].ip; pSql->ip = pVPeersDesc[pSql->index].ip;
@ -291,6 +297,8 @@ void tscGetConnToVnode(SSqlObj *pSql, uint8_t *pCode) {
tscTrace("%p vnode:%d ip:%p index:%d is picked up, pConn:%p", pSql, pVPeersDesc[pSql->index].vnode, tscTrace("%p vnode:%d ip:%p index:%d is picked up, pConn:%p", pSql, pVPeersDesc[pSql->index].vnode,
pVPeersDesc[pSql->index].ip, pSql->index, pSql->thandle); pVPeersDesc[pSql->index].ip, pSql->index, pSql->thandle);
//TODO fetch from vpeerdesc
pSql->ipSet = tscMgmtIpSet;
break; break;
} }
@ -326,25 +334,29 @@ int tscSendMsgToServer(SSqlObj *pSql) {
size_t totalLen = pSql->cmd.payloadLen + tsRpcHeadSize + sizeof(STaosDigest); size_t totalLen = pSql->cmd.payloadLen + tsRpcHeadSize + sizeof(STaosDigest);
// the memory will be released by taosProcessResponse, so no memory leak here // the memory will be released by taosProcessResponse, so no memory leak here
char *buf = malloc(totalLen); char *pStart = rpcMallocCont(pSql->cmd.payloadLen);
if (NULL == buf) { if (NULL == pStart) {
tscError("%p msg:%s malloc fail", pSql, taosMsg[pSql->cmd.msgType]); tscError("%p msg:%s malloc fail", pSql, taosMsg[pSql->cmd.msgType]);
return TSDB_CODE_CLI_OUT_OF_MEMORY; return TSDB_CODE_CLI_OUT_OF_MEMORY;
} }
memcpy(buf, pSql->cmd.payload, totalLen); memcpy(pStart, pSql->cmd.payload + tsRpcHeadSize, pSql->cmd.payloadLen);
tscTrace("%p msg:%s is sent to server", pSql, taosMsg[pSql->cmd.msgType]); tscTrace("%p msg:%s is sent to server", pSql, taosMsg[pSql->cmd.msgType]);
char *pStart = taosBuildReqHeader(pSql->thandle, pSql->cmd.msgType, buf);
if (pStart) { if (pStart) {
/* /*
* this SQL object may be released by other thread due to the completion of this query even before the log * this SQL object may be released by other thread due to the completion of this query even before the log
* is dumped to log file. So the signature needs to be kept in a local variable. * is dumped to log file. So the signature needs to be kept in a local variable.
*/ */
uint64_t signature = (uint64_t)pSql->signature; uint64_t signature = (uint64_t)pSql->signature;
if (tscUpdateVnodeMsg[pSql->cmd.command]) (*tscUpdateVnodeMsg[pSql->cmd.command])(pSql, buf); //if (tscUpdateVnodeMsg[pSql->cmd.command]) (*tscUpdateVnodeMsg[pSql->cmd.command])(pSql, pStart);
int ret;
if (pSql->cmd.command < TSDB_SQL_MGMT)
ret = rpcSendRequest(pTscMgmtConn, pSql->cmd.msgType, pStart, pSql->cmd.payloadLen, pSql);
else
ret = rpcSendRequest(pVnodeConn, pSql->cmd.msgType, pStart, pSql->cmd.payloadLen, pSql);
int ret = taosSendMsgToPeerH(pSql->thandle, pStart, pSql->cmd.payloadLen, pSql);
if (ret >= 0) { if (ret >= 0) {
code = 0; code = 0;
} }

16
src/client/src/tscSql.c
View File

@ -64,15 +64,15 @@ TAOS *taos_connect_imp(const char *ip, const char *user, const char *pass, const
} }
if (ip && ip[0]) { if (ip && ip[0]) {
tscMgmtIpList.numOfIps = 4; tscMgmtIpList.numOfIps = 3;
strcpy(tscMgmtIpList.ipstr[0], ip); strcpy(tscMgmtIpList.ipStr[0], ip);
tscMgmtIpList.ip[0] = inet_addr(ip); tscMgmtIpList.ip[0] = inet_addr(ip);
strcpy(tscMgmtIpList.ipstr[1], ip); strcpy(tscMgmtIpList.ipStr[1], tsMasterIp);
tscMgmtIpList.ip[1] = inet_addr(ip); tscMgmtIpList.ip[1] = inet_addr(tsMasterIp);
strcpy(tscMgmtIpList.ipstr[2], tsMasterIp); strcpy(tscMgmtIpList.ipStr[2], tsSecondIp);
tscMgmtIpList.ip[2] = inet_addr(tsMasterIp); tscMgmtIpList.ip[2] = inet_addr(tsSecondIp);
strcpy(tscMgmtIpList.ipstr[3], tsSecondIp); tscMgmtIpList.index = 0;
tscMgmtIpList.ip[3] = inet_addr(tsSecondIp); tscMgmtIpList.port = tsMgmtShellPort;
} }
pObj = (STscObj *)malloc(sizeof(STscObj)); pObj = (STscObj *)malloc(sizeof(STscObj));

2
src/client/src/tscStream.c
View File

@ -19,7 +19,7 @@
#include "ttime.h" #include "ttime.h"
#include "ttimer.h" #include "ttimer.h"
#include "tutil.h" #include "tutil.h"
#include "tsched.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "tscUtil.h" #include "tscUtil.h"
#include "tsclient.h" #include "tsclient.h"

47
src/client/src/tscSystem.c
View File

@ -24,8 +24,9 @@
#include "ttime.h" #include "ttime.h"
#include "ttimer.h" #include "ttimer.h"
#include "tutil.h" #include "tutil.h"
#include "tsched.h"
#include "tsclient.h" #include "tsclient.h"
// global, not configurable // global, not configurable
void * pVnodeConn; void * pVnodeConn;
void * pVMeterConn; void * pVMeterConn;
@ -95,17 +96,16 @@ void taos_init_imp() {
taosInitNote(tsNumOfLogLines / 10, 1, (char*)"tsc_note"); taosInitNote(tsNumOfLogLines / 10, 1, (char*)"tsc_note");
} }
tscMgmtIpList.numOfIps = 2; tscMgmtIpList.index = 0;
strcpy(tscMgmtIpList.ipstr[0], tsMasterIp); tscMgmtIpList.port = tsMgmtShellPort;
tscMgmtIpList.numOfIps = 1;
strcpy(tscMgmtIpList.ipStr[0], tsMasterIp);
tscMgmtIpList.ip[0] = inet_addr(tsMasterIp); tscMgmtIpList.ip[0] = inet_addr(tsMasterIp);
strcpy(tscMgmtIpList.ipstr[1], tsMasterIp);
tscMgmtIpList.ip[1] = inet_addr(tsMasterIp);
if (tsSecondIp[0]) { if (tsSecondIp[0]) {
tscMgmtIpList.numOfIps = 3; tscMgmtIpList.numOfIps = 2;
strcpy(tscMgmtIpList.ipstr[2], tsSecondIp); strcpy(tscMgmtIpList.ipStr[1], tsSecondIp);
tscMgmtIpList.ip[2] = inet_addr(tsSecondIp); tscMgmtIpList.ip[1] = inet_addr(tsSecondIp);
} }
tscInitMsgs(); tscInitMsgs();
@ -132,42 +132,23 @@ void taos_init_imp() {
rpcInit.label = "TSC-vnode"; rpcInit.label = "TSC-vnode";
rpcInit.numOfThreads = tscNumOfThreads; rpcInit.numOfThreads = tscNumOfThreads;
rpcInit.fp = tscProcessMsgFromServer; rpcInit.fp = tscProcessMsgFromServer;
rpcInit.bits = 20; rpcInit.sessions = tsMaxVnodeConnections;
rpcInit.numOfChanns = tscNumOfThreads;
rpcInit.sessionsPerChann = tsMaxVnodeConnections / tscNumOfThreads;
rpcInit.idMgmt = TAOS_ID_FREE;
rpcInit.noFree = 0;
rpcInit.connType = TAOS_CONN_SOCKET_TYPE_C(); rpcInit.connType = TAOS_CONN_SOCKET_TYPE_C();
rpcInit.qhandle = tscQhandle; pVnodeConn = rpcOpen(&rpcInit);
pVnodeConn = taosOpenRpc(&rpcInit);
if (pVnodeConn == NULL) { if (pVnodeConn == NULL) {
tscError("failed to init connection to vnode"); tscError("failed to init connection to vnode");
return; return;
} }
for (int i = 0; i < tscNumOfThreads; ++i) {
int retVal = taosOpenRpcChann(pVnodeConn, i, rpcInit.sessionsPerChann);
if (0 != retVal) {
tError("TSC-vnode, failed to open rpc chann");
taosCloseRpc(pVnodeConn);
return;
}
}
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localIp = tsLocalIp; rpcInit.localIp = tsLocalIp;
rpcInit.localPort = 0; rpcInit.localPort = 0;
rpcInit.label = "TSC-mgmt"; rpcInit.label = "TSC-mgmt";
rpcInit.numOfThreads = 1; rpcInit.numOfThreads = 1;
rpcInit.fp = tscProcessMsgFromServer; rpcInit.fp = tscProcessMsgFromServer;
rpcInit.bits = 20; rpcInit.sessions = tsMaxMgmtConnections;
rpcInit.numOfChanns = 1;
rpcInit.sessionsPerChann = tsMaxMgmtConnections;
rpcInit.idMgmt = TAOS_ID_FREE;
rpcInit.noFree = 0;
rpcInit.connType = TAOS_CONN_SOCKET_TYPE_C(); rpcInit.connType = TAOS_CONN_SOCKET_TYPE_C();
rpcInit.qhandle = tscQhandle; pTscMgmtConn = rpcOpen(&rpcInit);
pTscMgmtConn = taosOpenRpc(&rpcInit);
if (pTscMgmtConn == NULL) { if (pTscMgmtConn == NULL) {
tscError("failed to init connection to mgmt"); tscError("failed to init connection to mgmt");
return; return;
@ -183,7 +164,7 @@ void taos_init_imp() {
if (tscCacheHandle == NULL) tscCacheHandle = taosInitDataCache(tsMaxMeterConnections / 2, tscTmr, refreshTime); if (tscCacheHandle == NULL) tscCacheHandle = taosInitDataCache(tsMaxMeterConnections / 2, tscTmr, refreshTime);
tscConnCache = taosOpenConnCache(tsMaxMeterConnections * 2, taosCloseRpcConn, tscTmr, tsShellActivityTimer * 1000); tscConnCache = taosOpenConnCache(tsMaxMeterConnections * 2, NULL/*taosCloseRpcConn*/, tscTmr, tsShellActivityTimer * 1000);
initialized = 1; initialized = 1;
tscTrace("client is initialized successfully"); tscTrace("client is initialized successfully");

2
src/inc/taoserror.h
View File

@ -26,7 +26,7 @@ extern "C" {
#ifdef TAOS_ERROR_C #ifdef TAOS_ERROR_C
#define TAOS_DEFINE_ERROR(name, mod, code, msg) {.val = (0x80000000 | ((mod)<<16) | (code)), .str=(msg)}, #define TAOS_DEFINE_ERROR(name, mod, code, msg) {.val = (0x80000000 | ((mod)<<16) | (code)), .str=(msg)},
#else #else
#define TAOS_DEFINE_ERROR(name, mod, code, msg) const int32_t name = (0x80000000 | ((mod)<<16) | (code)); #define TAOS_DEFINE_ERROR(name, mod, code, msg) static const int32_t name = (0x80000000 | ((mod)<<16) | (code));
#endif #endif
#define TAOS_SYSTEM_ERROR(code) (0x80ff0000 | (code)) #define TAOS_SYSTEM_ERROR(code) (0x80ff0000 | (code))

1
src/inc/taosmsg.h
View File

@ -688,6 +688,7 @@ typedef struct {
typedef struct { typedef struct {
int32_t dnode; //the ID of dnode int32_t dnode; //the ID of dnode
int32_t vnode; //the index of vnode int32_t vnode; //the index of vnode
uint32_t ip;
} SVPeerDesc; } SVPeerDesc;
typedef struct { typedef struct {

49
src/inc/trpc.h
View File

@ -21,6 +21,7 @@ extern "C" {
#include <stdbool.h> #include <stdbool.h>
#include <stdint.h> #include <stdint.h>
#include "taosdef.h"
#define TAOS_CONN_UDPS 0 #define TAOS_CONN_UDPS 0
#define TAOS_CONN_UDPC 1 #define TAOS_CONN_UDPC 1
@ -37,38 +38,46 @@ extern "C" {
extern int tsRpcHeadSize; extern int tsRpcHeadSize;
typedef struct {
char *localIp; // local IP used
uint16_t localPort; // local port
char *label; // for debug purpose
int numOfThreads; // number of threads to handle connections
void *(*fp)(int8_t type, void *pCont, int32_t contLen, void *handle, int32_t index); // function to process the incoming msg
int sessions; // number of sessions allowed
int connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
int idleTime; // milliseconds, 0 means idle timer is disabled
char *meterId; // meter ID
char spi; // security parameter index
char encrypt; // encrypt algorithm
char *secret; // key for authentication
char *ckey; // ciphering key
int (*afp) (char *meterId, char *spi, char *encrypt, uint8_t *secret, uint8_t *ckey); // call back to retrieve auth info
} SRpcInit;
typedef struct { typedef struct {
int16_t index; int16_t index;
int16_t numOfIps; int16_t numOfIps;
uint16_t port; uint16_t port;
uint32_t ip[TSDB_MAX_MPEERS]; uint32_t ip[TSDB_MAX_MPEERS];
char ipStr[TSDB_MAX_MPEERS][TSDB_IPv4ADDR_LEN];
} SRpcIpSet; } SRpcIpSet;
typedef struct {
char *localIp; // local IP used
uint16_t localPort; // local port
char *label; // for debug purpose
int numOfThreads; // number of threads to handle connections
int sessions; // number of sessions allowed
int connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
int idleTime; // milliseconds, 0 means idle timer is disabled
// the following is for client security only
char *meterId; // meter ID
char spi; // security parameter index
char encrypt; // encrypt algorithm
char *secret; // key for authentication
char *ckey; // ciphering key
// call back to process incoming msg
void (*cfp)(char type, void *pCont, int contLen, void *ahandle, int32_t code);
// call back to process notify the ipSet changes
void (*ufp)(void *ahandle, SRpcIpSet ipSet);
// call back to retrieve the client auth info
int (*afp)(char *meterId, char *spi, char *encrypt, uint8_t *secret, uint8_t *ckey);
} SRpcInit;
void *rpcOpen(SRpcInit *pRpc); void *rpcOpen(SRpcInit *pRpc);
void rpcClose(void *); void rpcClose(void *);
void *rpcMallocCont(int contLen); void *rpcMallocCont(int contLen);
void rpcFreeCont(void *pCont); void rpcFreeCont(void *pCont);
void rpcSendRequest(void *thandle, SRpcIpSet ipSet, char msgType, void *pCont, int contLen, void *ahandle); void rpcSendRequest(void *thandle, SRpcIpSet ipSet, char msgType, void *pCont, int contLen, void *ahandle);
void rpcSendResponse(void *pConn, void *pCont, int contLen); void rpcSendResponse(void *pConn, int32_t code, void *pCont, int contLen);
void rpcSendSimpleRsp(void *pConn, int32_t code); void rpcSendRedirectRsp(void *pConn, SRpcIpSet ipSet);
#ifdef __cplusplus #ifdef __cplusplus

3
src/rpc/CMakeLists.txt
View File

@ -23,3 +23,6 @@ ENDIF ()
ADD_LIBRARY(trpc ${SRC}) ADD_LIBRARY(trpc ${SRC})
TARGET_LINK_LIBRARIES(trpc tutil) TARGET_LINK_LIBRARIES(trpc tutil)
ADD_SUBDIRECTORY(test)

8
src/rpc/inc/thaship.h
View File

@ -16,10 +16,18 @@
#ifndef _rpc_hash_ip_header_ #ifndef _rpc_hash_ip_header_
#define _rpc_hash_ip_header_ #define _rpc_hash_ip_header_
#ifdef __cplusplus
extern "C" {
#endif
void *taosOpenIpHash(int maxSessions); void *taosOpenIpHash(int maxSessions);
void taosCloseIpHash(void *handle); void taosCloseIpHash(void *handle);
void *taosAddIpHash(void *handle, void *pData, uint32_t ip, uint16_t port); void *taosAddIpHash(void *handle, void *pData, uint32_t ip, uint16_t port);
void taosDeleteIpHash(void *handle, uint32_t ip, uint16_t port); void taosDeleteIpHash(void *handle, uint32_t ip, uint16_t port);
void *taosGetIpHash(void *handle, uint32_t ip, uint16_t port); void *taosGetIpHash(void *handle, uint32_t ip, uint16_t port);
#ifdef __cplusplus
}
#endif
#endif #endif

8
src/rpc/inc/ttcpclient.h
View File

@ -16,6 +16,10 @@
#ifndef _taos_tcp_client_header_ #ifndef _taos_tcp_client_header_
#define _taos_tcp_client_header_ #define _taos_tcp_client_header_
#ifdef __cplusplus
extern "C" {
#endif
#include "taosdef.h" #include "taosdef.h"
void *taosInitTcpClient(char *ip, uint16_t port, char *label, int num, void *fp, void *shandle); void *taosInitTcpClient(char *ip, uint16_t port, char *label, int num, void *fp, void *shandle);
@ -24,4 +28,8 @@ void *taosOpenTcpClientConnection(void *shandle, void *thandle, char *ip, uint16
void taosCloseTcpClientConnection(void *chandle); void taosCloseTcpClientConnection(void *chandle);
int taosSendTcpClientData(uint32_t ip, uint16_t port, char *data, int len, void *chandle); int taosSendTcpClientData(uint32_t ip, uint16_t port, char *data, int len, void *chandle);
#ifdef __cplusplus
}
#endif
#endif #endif

8
src/rpc/inc/ttcpserver.h
View File

@ -16,6 +16,10 @@
#ifndef _taos_tcp_server_header_ #ifndef _taos_tcp_server_header_
#define _taos_tcp_server_header_ #define _taos_tcp_server_header_
#ifdef __cplusplus
extern "C" {
#endif
#include "taosdef.h" #include "taosdef.h"
void *taosInitTcpServer(char *ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle); void *taosInitTcpServer(char *ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle);
@ -23,4 +27,8 @@ void taosCleanUpTcpServer(void *param);
void taosCloseTcpServerConnection(void *param); void taosCloseTcpServerConnection(void *param);
int taosSendTcpServerData(uint32_t ip, uint16_t port, char *data, int len, void *chandle); int taosSendTcpServerData(uint32_t ip, uint16_t port, char *data, int len, void *chandle);
#ifdef __cplusplus
}
#endif
#endif #endif

8
src/rpc/inc/tudp.h
View File

@ -16,6 +16,10 @@
#ifndef _taos_udp_header_ #ifndef _taos_udp_header_
#define _taos_udp_header_ #define _taos_udp_header_
#ifdef __cplusplus
extern "C" {
#endif
#include "taosdef.h" #include "taosdef.h"
void *taosInitUdpServer(char *ip, uint16_t port, char *label, int, void *fp, void *shandle); void *taosInitUdpServer(char *ip, uint16_t port, char *label, int, void *fp, void *shandle);
@ -30,4 +34,8 @@ void taosSendMsgHdr(void *hdr, int fd);
void taosInitMsgHdr(void **hdr, void *dest, int maxPkts); void taosInitMsgHdr(void **hdr, void *dest, int maxPkts);
void taosSetMsgHdrData(void *hdr, char *data, int dataLen); void taosSetMsgHdrData(void *hdr, char *data, int dataLen);
#ifdef __cplusplus
}
#endif
#endif #endif

290
src/rpc/src/trpc.c
View File

@ -30,18 +30,19 @@
#include "lz4.h" #include "lz4.h"
#include "tconncache.h" #include "tconncache.h"
#include "trpc.h" #include "trpc.h"
#include "taoserror.h"
#define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHeader) + sizeof(SRpcDigest)) #define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHeader) + sizeof(SRpcDigest))
#define rpcHeaderFromCont(cont) ((SRpcHeader *) (cont - sizeof(SRpcHeader))) #define rpcHeaderFromCont(cont) ((SRpcHeader *) (cont - sizeof(SRpcHeader)))
#define rpcContFromHeader(msg) ( msg + sizeof(SRpcHeader)) #define rpcContFromHeader(msg) (msg + sizeof(SRpcHeader))
#define rpcMsgLenFromCont(contLen) ( contLen + sizeof(SRpcHeader)) #define rpcMsgLenFromCont(contLen) (contLen + sizeof(SRpcHeader))
#define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHeader)) #define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHeader))
#define rpcIsReq(type) (type & 1U) #define rpcIsReq(type) (type & 1U)
typedef struct { typedef struct {
int sessions; int sessions;
int numOfThreads; int numOfThreads;
int idleTime; // milliseconds; int idleTime; // milliseconds;
char localIp[TSDB_IPv4ADDR_LEN]; char localIp[TSDB_IPv4ADDR_LEN];
uint16_t localPort; uint16_t localPort;
int connType; int connType;
@ -53,27 +54,29 @@ typedef struct {
char *secret; // key for authentication char *secret; // key for authentication
char *ckey; // ciphering key char *ckey; // ciphering key
void *(*fp)(char type, void *pCont, int contLen, void *handle, int index); void (*cfp)(char type, void *pCont, int contLen, void *ahandle, int32_t code);
int (*afp)(char *meterId, char *spi, char *encrypt, uint8_t *secret, uint8_t *ckey); // FP to retrieve auth info int (*afp)(char *meterId, char *spi, char *encrypt, uint8_t *secret, uint8_t *ckey);
struct _RpcConn *connList; void (*ufp)(void *ahandle, SRpcIpSet ipSet);
void *idPool;
void *tmrCtrl; void *idPool; // handle to ID pool
void *hash; void *tmrCtrl; // handle to timer
void *hash; // handle returned by hash utility
void *shandle; // returned handle from lower layer during initialization void *shandle; // returned handle from lower layer during initialization
void *pCache; // connection cache void *pCache; // connection cache
pthread_mutex_t mutex; pthread_mutex_t mutex;
struct _RpcConn *connList; // connection list
} SRpcInfo; } SRpcInfo;
typedef struct { typedef struct {
SRpcIpSet ipSet; SRpcIpSet ipSet;
void *ahandle; void *ahandle; // handle provided by app
SRpcInfo *pRpc; SRpcInfo *pRpc; // associated SRpcInfo
char msgType; char msgType; // message type
char *pCont; char *pCont; // content provided by app
int contLen; int contLen; // content length
int numOfRetry; int numOfRetry; // number of retry for different servers
int32_t code; int32_t code; // error code
char msg[]; char msg[0]; // RpcHeader starts from here
} SRpcReqContext; } SRpcReqContext;
typedef struct _RpcConn { typedef struct _RpcConn {
@ -124,6 +127,7 @@ typedef struct {
char empty[1]; // reserved char empty[1]; // reserved
uint8_t msgType; // message type uint8_t msgType; // message type
int32_t msgLen; // message length including the header iteslf int32_t msgLen; // message length including the header iteslf
int32_t code;
uint8_t content[0]; // message body starts from here uint8_t content[0]; // message body starts from here
} SRpcHeader; } SRpcHeader;
@ -174,25 +178,25 @@ void (*taosCloseConn[])(void *chandle) = {
}; };
static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerIpStr, uint16_t peerPort); static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerIpStr, uint16_t peerPort);
static void rpcCloseConn(void *thandle); static void rpcCloseConn(void *thandle);
static SRpcConn *rpcSetConnToServer(void *shandle, SRpcIpSet ipSet); static SRpcConn *rpcSetConnToServer(SRpcInfo *pRpc, SRpcIpSet ipSet);
static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc); static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc);
static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, char *meterId, char *hashstr); static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, char *meterId, char *hashstr);
static SRpcConn *rpcGetConnObj(SRpcInfo *pRpc, int sid, char *meterId, char *hashstr); static SRpcConn *rpcGetConnObj(SRpcInfo *pRpc, int sid, char *meterId, char *hashstr);
static void rpcSendReqToServer(SRpcConn *pConn, SRpcReqContext *pContext); static void rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext);
static void rpcSendQuickRsp(SRpcConn *pConn, char code); static void rpcSendQuickRsp(SRpcConn *pConn, char code);
static void rpcSendErrorMsgToPeer(SRpcInfo *pRpc, char *pMsg, int32_t code, uint32_t ip, uint16_t port, void *chandle); static void rpcSendErrorMsgToPeer(SRpcInfo *pRpc, char *pMsg, int32_t code, uint32_t ip, uint16_t port, void *chandle);
static void rpcSendDataToPeer(SRpcConn *pConn, void *data, int dataLen); static void rpcSendMsgToPeer(SRpcConn *pConn, void *data, int dataLen);
static void *rpcProcessDataFromPeer(void *data, int dataLen, uint32_t ip, uint16_t port, void *shandle, void *chandle); static void *rpcProcessMsgFromPeer(void *data, int dataLen, uint32_t ip, uint16_t port, void *shandle, void *chandle);
static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHeader *pHeader); static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHeader *pHeader);
static void rpcProcessConnError(void *param, void *id); static void rpcProcessConnError(void *param, void *id);
static void rpcProcessRetryTimer(void *, void *); static void rpcProcessRetryTimer(void *, void *);
static void rpcProcessIdleTimer(void *param, void *tmrId); static void rpcProcessIdleTimer(void *param, void *tmrId);
static void rpcProcessProgressTimer(void *param, void *tmrId); static void rpcProcessProgressTimer(void *param, void *tmrId);
static void rpcFreeMsg(void *msg); static void rpcFreeOutMsg(void *msg);
static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen); static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen);
static SRpcHeader *rpcDecompressRpcMsg(SRpcHeader *pHeader); static SRpcHeader *rpcDecompressRpcMsg(SRpcHeader *pHeader);
static int rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen); static int rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen);
@ -208,7 +212,6 @@ void *rpcOpen(SRpcInit *pInit) {
if (pRpc == NULL) return NULL; if (pRpc == NULL) return NULL;
strcpy(pRpc->label, pInit->label); strcpy(pRpc->label, pInit->label);
pRpc->fp = pInit->fp;
pRpc->connType = pInit->connType; pRpc->connType = pInit->connType;
pRpc->idleTime = pInit->idleTime; pRpc->idleTime = pInit->idleTime;
pRpc->numOfThreads = pInit->numOfThreads; pRpc->numOfThreads = pInit->numOfThreads;
@ -224,10 +227,12 @@ void *rpcOpen(SRpcInit *pInit) {
pRpc->spi = pInit->spi; pRpc->spi = pInit->spi;
strcpy(pRpc->secret, pInit->secret); strcpy(pRpc->secret, pInit->secret);
strcpy(pRpc->ckey, pInit->ckey); strcpy(pRpc->ckey, pInit->ckey);
pRpc->ufp = pInit->ufp;
pRpc->cfp = pInit->cfp;
pRpc->afp = pInit->afp; pRpc->afp = pInit->afp;
pRpc->shandle = (*taosInitConn[pRpc->connType])(pRpc->localIp, pRpc->localPort, pRpc->label, pRpc->shandle = (*taosInitConn[pRpc->connType])(pRpc->localIp, pRpc->localPort, pRpc->label,
pRpc->numOfThreads, rpcProcessDataFromPeer, pRpc); pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc);
if (pRpc->shandle == NULL) { if (pRpc->shandle == NULL) {
tError("%s failed to init network, %s:%d", pRpc->label, pRpc->localIp, pRpc->localPort); tError("%s failed to init network, %s:%d", pRpc->label, pRpc->localIp, pRpc->localPort);
rpcClose(pRpc); rpcClose(pRpc);
@ -318,7 +323,6 @@ void rpcFreeCont(void *cont) {
void rpcSendRequest(void *shandle, SRpcIpSet ipSet, char type, void *pCont, int contLen, void *ahandle) { void rpcSendRequest(void *shandle, SRpcIpSet ipSet, char type, void *pCont, int contLen, void *ahandle) {
SRpcInfo *pRpc = (SRpcInfo *)shandle; SRpcInfo *pRpc = (SRpcInfo *)shandle;
SRpcConn *pConn;
SRpcReqContext *pContext; SRpcReqContext *pContext;
contLen = rpcCompressRpcMsg(pCont, contLen); contLen = rpcCompressRpcMsg(pCont, contLen);
@ -330,22 +334,23 @@ void rpcSendRequest(void *shandle, SRpcIpSet ipSet, char type, void *pCont, int
pContext->pCont = pCont; pContext->pCont = pCont;
pContext->msgType = type; pContext->msgType = type;
pConn = rpcSetConnToServer(shandle, ipSet); rpcSendReqToServer(pRpc, pContext);
pContext->code = terrno;
if (pConn == NULL) taosTmrStart(rpcProcessConnError, 0, pContext, pRpc->tmrCtrl);
rpcSendReqToServer(pConn, pContext);
return; return;
} }
void rpcSendResponse(void *handle, void *pCont, int contLen) { void rpcSendResponse(void *handle, int32_t code, void *pCont, int contLen) {
int msgLen = 0; int msgLen = 0;
SRpcConn *pConn = (SRpcConn *)handle; SRpcConn *pConn = (SRpcConn *)handle;
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
SRpcHeader *pHeader = rpcHeaderFromCont(pCont); SRpcHeader *pHeader = rpcHeaderFromCont(pCont);
char *msg = (char *)pHeader; char *msg = (char *)pHeader;
if ( pCont == NULL ) {
pCont = rpcMallocCont(0);
contLen = 0;
}
contLen = rpcCompressRpcMsg(pCont, contLen); contLen = rpcCompressRpcMsg(pCont, contLen);
msgLen = rpcMsgLenFromCont(contLen); msgLen = rpcMsgLenFromCont(contLen);
@ -367,13 +372,14 @@ void rpcSendResponse(void *handle, void *pCont, int contLen) {
pHeader->sourceId = pConn->ownId; pHeader->sourceId = pConn->ownId;
pHeader->destId = pConn->peerId; pHeader->destId = pConn->peerId;
pHeader->uid = 0; pHeader->uid = 0;
pHeader->code = htonl(code);
memcpy(pHeader->meterId, pConn->meterId, tListLen(pHeader->meterId)); memcpy(pHeader->meterId, pConn->meterId, tListLen(pHeader->meterId));
// set pConn parameters // set pConn parameters
pConn->inType = 0; pConn->inType = 0;
// response message is released until new response is sent // response message is released until new response is sent
rpcFreeMsg(pConn->pRspMsg); rpcFreeOutMsg(pConn->pRspMsg);
pConn->pRspMsg = msg; pConn->pRspMsg = msg;
pConn->rspMsgLen = msgLen; pConn->rspMsgLen = msgLen;
if (pHeader->content[0] == TSDB_CODE_ACTION_IN_PROGRESS) pConn->inTranId--; if (pHeader->content[0] == TSDB_CODE_ACTION_IN_PROGRESS) pConn->inTranId--;
@ -381,28 +387,21 @@ void rpcSendResponse(void *handle, void *pCont, int contLen) {
pthread_mutex_unlock(&pRpc->mutex); pthread_mutex_unlock(&pRpc->mutex);
taosTmrStopA(&pConn->pTimer); taosTmrStopA(&pConn->pTimer);
rpcSendDataToPeer(pConn, msg, msgLen); rpcSendMsgToPeer(pConn, msg, msgLen);
return; return;
} }
void rpcSendSimpleRsp(void *thandle, int32_t code) { void rpcSendRedirectRsp(void *thandle, SRpcIpSet ipSet) {
char *pMsg; char *pMsg;
STaosRsp *pRsp; int msgLen = sizeof(SRpcIpSet);
int msgLen = sizeof(STaosRsp);
if (thandle == NULL) {
tError("connection is gone, response could not be sent");
return;
}
pMsg = rpcMallocCont(msgLen); pMsg = rpcMallocCont(msgLen);
if (pMsg == NULL) return; if (pMsg == NULL) return;
pRsp = (STaosRsp *)pMsg; memcpy(pMsg, &ipSet, sizeof(ipSet));
pRsp->code = htonl(code);
rpcSendResponse(thandle, pMsg, msgLen); rpcSendResponse(thandle, TSDB_CODE_REDIRECT, pMsg, msgLen);
return; return;
} }
@ -442,34 +441,33 @@ static void rpcCloseConn(void *thandle) {
pthread_mutex_lock(&pRpc->mutex); pthread_mutex_lock(&pRpc->mutex);
if (taosCloseConn[pRpc->connType]) (*taosCloseConn[pRpc->connType])(pConn->chandle); if (pConn->meterId[0]) {
pConn->meterId[0] = 0;
if (taosCloseConn[pRpc->connType]) (*taosCloseConn[pRpc->connType])(pConn->chandle);
taosTmrStopA(&pConn->pTimer); taosTmrStopA(&pConn->pTimer);
taosTmrStopA(&pConn->pIdleTimer); taosTmrStopA(&pConn->pIdleTimer);
if ( pRpc->connType == TAOS_CONN_UDPS || TAOS_CONN_TCPS) { if ( pRpc->connType == TAOS_CONN_UDPS || TAOS_CONN_TCPS) {
char hashstr[40] = {0}; char hashstr[40] = {0};
sprintf(hashstr, "%x:%x:%x", pConn->peerIp, pConn->peerUid, pConn->peerId); sprintf(hashstr, "%x:%x:%x", pConn->peerIp, pConn->peerUid, pConn->peerId);
taosDeleteStrHash(pRpc->hash, hashstr); taosDeleteStrHash(pRpc->hash, hashstr);
rpcFreeMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg rpcFreeOutMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg
pConn->pRspMsg = NULL;
pConn->inType = 0;
pConn->inTranId = 0;
} else {
pConn->outType = 0;
pConn->outTranId = 0;
pConn->pReqMsg = NULL;
}
taosFreeId(pRpc->idPool, pConn->sid);
pConn->pContext = NULL;
tTrace("%s pConn:%p, rpc connection is closed", pRpc->label, pConn);
} }
taosFreeId(pRpc->idPool, pConn->sid);
// reset the link parameters
pConn->meterId[0] = 0;
pConn->outType = 0;
pConn->inType = 0;
pConn->inTranId = 0;
pConn->outTranId = 0;
pConn->pReqMsg = NULL;
pConn->reqMsgLen = 0;
pConn->pRspMsg = NULL;
pConn->rspMsgLen = 0;
pConn->pContext = NULL;
tTrace("%s pConn:%p, rpc connection is closed", pRpc->label, pConn);
pthread_mutex_unlock(&pRpc->mutex); pthread_mutex_unlock(&pRpc->mutex);
} }
@ -553,13 +551,14 @@ static SRpcConn *rpcGetConnObj(SRpcInfo *pRpc, int sid, char *meterId, char *has
return pConn; return pConn;
} }
SRpcConn *rpcSetConnToServer(void *shandle, SRpcIpSet ipSet) { SRpcConn *rpcSetConnToServer(SRpcInfo *pRpc, SRpcIpSet ipSet) {
SRpcInfo *pRpc = (SRpcInfo *)shandle; SRpcConn *pConn;
SRpcConn *pConn = taosGetConnFromCache(pRpc->pCache, ipSet.ip[ipSet.index], ipSet.port, pRpc->meterId);
pConn = taosGetConnFromCache(pRpc->pCache, ipSet.ip[ipSet.index], ipSet.port, pRpc->meterId);
if ( pConn == NULL ) { if ( pConn == NULL ) {
pConn = rpcOpenConn(pRpc, ipSet.ipStr[ipSet.index], ipSet.port); char ipstr[20] = {0};
tinet_ntoa(ipstr, ipSet.ip[ipSet.index]);
pConn = rpcOpenConn(pRpc, ipstr, ipSet.port);
} }
return pConn; return pConn;
@ -585,7 +584,7 @@ static int rpcProcessReqHeader(SRpcConn *pConn, SRpcHeader *pHeader) {
} else if (pConn->inType == 0) { } else if (pConn->inType == 0) {
tTrace("%s pConn:%p, %s is already processed, tranId:%d", pRpc->label, pConn, tTrace("%s pConn:%p, %s is already processed, tranId:%d", pRpc->label, pConn,
taosMsg[pHeader->msgType], pConn->inTranId); taosMsg[pHeader->msgType], pConn->inTranId);
rpcSendDataToPeer(pConn, pConn->pRspMsg, pConn->rspMsgLen); // resend the response rpcSendMsgToPeer(pConn, pConn->pRspMsg, pConn->rspMsgLen); // resend the response
} else { } else {
tTrace("%s pConn:%p, mismatched message %s and tranId", pRpc->label, pConn, taosMsg[pHeader->msgType]); tTrace("%s pConn:%p, mismatched message %s and tranId", pRpc->label, pConn, taosMsg[pHeader->msgType]);
} }
@ -658,13 +657,14 @@ static int rpcProcessHeader(SRpcInfo *pRpc, SRpcConn **ppConn, void *data, int d
SRpcHeader *pHeader = (SRpcHeader *)data; SRpcHeader *pHeader = (SRpcHeader *)data;
sid = htonl(pHeader->destId); sid = htonl(pHeader->destId);
pHeader->code = htonl(pHeader->code);
pHeader->msgLen = (int32_t)htonl((uint32_t)pHeader->msgLen);
if (pHeader->msgType >= TSDB_MSG_TYPE_MAX || pHeader->msgType <= 0) { if (pHeader->msgType >= TSDB_MSG_TYPE_MAX || pHeader->msgType <= 0) {
tTrace("%s sid:%d, invalid message type:%d", pRpc->label, sid, pHeader->msgType); tTrace("%s sid:%d, invalid message type:%d", pRpc->label, sid, pHeader->msgType);
return TSDB_CODE_INVALID_MSG_TYPE; return TSDB_CODE_INVALID_MSG_TYPE;
} }
pHeader->msgLen = (int32_t)htonl((uint32_t)pHeader->msgLen);
if (dataLen != pHeader->msgLen) { if (dataLen != pHeader->msgLen) {
tTrace("%s sid:%d, %s has invalid length, dataLen:%d, msgLen:%d", pRpc->label, sid, tTrace("%s sid:%d, %s has invalid length, dataLen:%d, msgLen:%d", pRpc->label, sid,
taosMsg[pHeader->msgType], dataLen, pHeader->msgLen); taosMsg[pHeader->msgType], dataLen, pHeader->msgLen);
@ -708,7 +708,7 @@ static int rpcProcessHeader(SRpcInfo *pRpc, SRpcConn **ppConn, void *data, int d
return code; return code;
} }
static void *rpcProcessDataFromPeer(void *data, int dataLen, uint32_t ip, uint16_t port, void *shandle, void *chandle) { static void *rpcProcessMsgFromPeer(void *data, int dataLen, uint32_t ip, uint16_t port, void *shandle, void *chandle) {
SRpcHeader *pHeader = (SRpcHeader *)data; SRpcHeader *pHeader = (SRpcHeader *)data;
SRpcInfo *pRpc = (SRpcInfo *)shandle; SRpcInfo *pRpc = (SRpcInfo *)shandle;
SRpcConn *pConn = NULL; SRpcConn *pConn = NULL;
@ -764,42 +764,33 @@ static void *rpcProcessDataFromPeer(void *data, int dataLen, uint32_t ip, uint16
static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHeader *pHeader) { static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHeader *pHeader) {
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
int msgLen = rpcContFromHeader(pHeader->msgLen);
pHeader = rpcDecompressRpcMsg(pHeader); pHeader = rpcDecompressRpcMsg(pHeader);
int contLen = rpcContLenFromMsg(pHeader->msgLen);
uint8_t *pCont = pHeader->content;
if ( rpcIsReq(pHeader->msgType) ) { if ( rpcIsReq(pHeader->msgType) ) {
taosTmrReset(rpcProcessProgressTimer, tsRpcTimer/2, pConn, pRpc->tmrCtrl, &pConn->pTimer); taosTmrReset(rpcProcessProgressTimer, tsRpcTimer/2, pConn, pRpc->tmrCtrl, &pConn->pTimer);
(*(pRpc->fp))(pHeader->msgType, pHeader->content, msgLen, pConn, 0); (*(pRpc->cfp))(pHeader->msgType, pCont, contLen, pConn, 0);
} else { } else {
// it's a response // it's a response
STaosRsp *pRsp = (STaosRsp *)pHeader->content; int32_t code = pHeader->code;
int32_t code = htonl(pRsp->code);
SRpcReqContext *pContext = pConn->pContext; SRpcReqContext *pContext = pConn->pContext;
pConn->pContext = NULL; pConn->pContext = NULL;
taosAddConnIntoCache(pRpc->pCache, pConn, pConn->peerIp, pConn->peerPort, pConn->meterId); taosAddConnIntoCache(pRpc->pCache, pConn, pConn->peerIp, pConn->peerPort, pConn->meterId);
if (code == TSDB_CODE_NO_MASTER) { if (code == TSDB_CODE_REDIRECT) {
pContext->code = code; memcpy(&pContext->ipSet, pHeader->content, sizeof(pContext->ipSet));
taosTmrStart(rpcProcessConnError, 0, pContext, pRpc->tmrCtrl); rpcSendReqToServer(pRpc, pContext);
} else { } else {
rpcFreeMsg(rpcHeaderFromCont(pContext->pCont)); // free the request msg rpcFreeOutMsg(rpcHeaderFromCont(pContext->pCont)); // free the request msg
(*(pRpc->fp))(pHeader->msgType, pHeader->content, msgLen, pContext->ahandle, pContext->ipSet.index); (*(pRpc->cfp))(pHeader->msgType, pCont, contLen, pContext->ahandle, pContext->ipSet.index);
} }
} }
} }
static void rpcSendQuickRsp(SRpcConn *pConn, char code) { static void rpcSendQuickRsp(SRpcConn *pConn, char code) {
char msg[RPC_MSG_OVERHEAD + sizeof(STaosRsp)]; char msg[RPC_MSG_OVERHEAD];
SRpcHeader *pHeader; SRpcHeader *pHeader;
int msgLen;
STaosRsp *pRsp;
pRsp = (STaosRsp *)rpcContFromHeader(msg);
pRsp->code = htonl(code);
msgLen = sizeof(STaosRsp);
// set msg header // set msg header
memset(msg, 0, sizeof(SRpcHeader)); memset(msg, 0, sizeof(SRpcHeader));
@ -814,14 +805,14 @@ static void rpcSendQuickRsp(SRpcConn *pConn, char code) {
pHeader->destId = pConn->peerId; pHeader->destId = pConn->peerId;
pHeader->uid = 0; pHeader->uid = 0;
memcpy(pHeader->meterId, pConn->meterId, tListLen(pHeader->meterId)); memcpy(pHeader->meterId, pConn->meterId, tListLen(pHeader->meterId));
pHeader->code = htonl(code);
rpcSendDataToPeer(pConn, msg, msgLen); rpcSendMsgToPeer(pConn, msg, 0);
} }
static void rpcSendErrorMsgToPeer(SRpcInfo *pRpc, char *pMsg, int32_t code, uint32_t ip, uint16_t port, void *chandle) { static void rpcSendErrorMsgToPeer(SRpcInfo *pRpc, char *pMsg, int32_t code, uint32_t ip, uint16_t port, void *chandle) {
SRpcHeader *pRecvHeader, *pReplyHeader; SRpcHeader *pRecvHeader, *pReplyHeader;
char msg[sizeof(SRpcHeader) + sizeof(SRpcDigest) + sizeof(STaosRsp)]; char msg[sizeof(SRpcHeader) + sizeof(SRpcDigest) + sizeof(uint32_t) ];
STaosRsp *pRsp;
uint32_t timeStamp; uint32_t timeStamp;
int msgLen; int msgLen;
@ -839,13 +830,12 @@ static void rpcSendErrorMsgToPeer(SRpcInfo *pRpc, char *pMsg, int32_t code, uint
pReplyHeader->destId = pRecvHeader->sourceId; pReplyHeader->destId = pRecvHeader->sourceId;
memcpy(pReplyHeader->meterId, pRecvHeader->meterId, tListLen(pReplyHeader->meterId)); memcpy(pReplyHeader->meterId, pRecvHeader->meterId, tListLen(pReplyHeader->meterId));
pRsp = (STaosRsp *)pReplyHeader->content; pReplyHeader->code = htonl(code);
pRsp->code = htonl(code); msgLen = sizeof(SRpcHeader);
msgLen = sizeof(STaosRsp);
char *pContent = pRsp->more;
if (code == TSDB_CODE_INVALID_TIME_STAMP) { if (code == TSDB_CODE_INVALID_TIME_STAMP) {
// include a time stamp if client's time is not synchronized well // include a time stamp if client's time is not synchronized well
uint8_t *pContent = pReplyHeader->content;
timeStamp = taosGetTimestampSec(); timeStamp = taosGetTimestampSec();
memcpy(pContent, &timeStamp, sizeof(timeStamp)); memcpy(pContent, &timeStamp, sizeof(timeStamp));
msgLen += sizeof(timeStamp); msgLen += sizeof(timeStamp);
@ -857,13 +847,19 @@ static void rpcSendErrorMsgToPeer(SRpcInfo *pRpc, char *pMsg, int32_t code, uint
return; return;
} }
static void rpcSendReqToServer(SRpcConn *pConn, SRpcReqContext *pContext) { static void rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) {
SRpcHeader *pHeader = rpcHeaderFromCont(pContext->pCont); SRpcHeader *pHeader = rpcHeaderFromCont(pContext->pCont);
SRpcInfo *pRpc = pConn->pRpc;
char *msg = (char *)pHeader; char *msg = (char *)pHeader;
int msgLen = rpcMsgLenFromCont(pContext->contLen); int msgLen = rpcMsgLenFromCont(pContext->contLen);
char msgType = pContext->msgType; char msgType = pContext->msgType;
SRpcConn *pConn = rpcSetConnToServer(pRpc, pContext->ipSet);
if (pConn == NULL) {
pContext->code = terrno;
taosTmrStart(rpcProcessConnError, 0, pContext, pRpc->tmrCtrl);
return;
}
pthread_mutex_lock(&pRpc->mutex); pthread_mutex_lock(&pRpc->mutex);
// set the message header // set the message header
@ -889,37 +885,37 @@ static void rpcSendReqToServer(SRpcConn *pConn, SRpcReqContext *pContext) {
pthread_mutex_unlock(&pRpc->mutex); pthread_mutex_unlock(&pRpc->mutex);
rpcSendDataToPeer(pConn, msg, msgLen); rpcSendMsgToPeer(pConn, msg, msgLen);
taosTmrReset(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl, &pConn->pTimer); taosTmrReset(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl, &pConn->pTimer);
} }
static void rpcSendDataToPeer(SRpcConn *pConn, void *data, int dataLen) { static void rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) {
int writtenLen = 0; int writtenLen = 0;
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
SRpcHeader *pHeader = (SRpcHeader *)data; SRpcHeader *pHeader = (SRpcHeader *)msg;
dataLen = rpcAddAuthPart(pConn, data, dataLen); msgLen = rpcAddAuthPart(pConn, msg, msgLen);
if ( rpcIsReq(pHeader->msgType)) { if ( rpcIsReq(pHeader->msgType)) {
if (pHeader->msgType < TSDB_MSG_TYPE_HEARTBEAT || (rpcDebugFlag & 16)) if (pHeader->msgType < TSDB_MSG_TYPE_HEARTBEAT || (rpcDebugFlag & 16))
tTrace("%s pConn:%p, %s is sent to %s:%hu, len:%d source:0x%08x dest:0x%08x tranId:%d", tTrace("%s pConn:%p, %s is sent to %s:%hu, len:%d source:0x%08x dest:0x%08x tranId:%d",
pRpc->label, pConn, taosMsg[pHeader->msgType], pConn->peerIpstr, pRpc->label, pConn, taosMsg[pHeader->msgType], pConn->peerIpstr,
pConn->peerPort, dataLen, pHeader->sourceId, pHeader->destId, pHeader->tranId); pConn->peerPort, msgLen, pHeader->sourceId, pHeader->destId, pHeader->tranId);
} else { } else {
if (pHeader->msgType < TSDB_MSG_TYPE_HEARTBEAT || (rpcDebugFlag & 16)) if (pHeader->msgType < TSDB_MSG_TYPE_HEARTBEAT || (rpcDebugFlag & 16))
tTrace( "%s pConn:%p, %s is sent to %s:%hu, code:%u len:%d source:0x%08x dest:0x%08x tranId:%d", tTrace( "%s pConn:%p, %s is sent to %s:%hu, code:%u len:%d source:0x%08x dest:0x%08x tranId:%d",
pRpc->label, pConn, taosMsg[pHeader->msgType], pConn->peerIpstr, pConn->peerPort, pRpc->label, pConn, taosMsg[pHeader->msgType], pConn->peerIpstr, pConn->peerPort,
(uint8_t)pHeader->content[0], dataLen, pHeader->sourceId, pHeader->destId, pHeader->tranId); (uint8_t)pHeader->content[0], msgLen, pHeader->sourceId, pHeader->destId, pHeader->tranId);
} }
writtenLen = (*taosSendData[pRpc->connType])(pConn->peerIp, pConn->peerPort, (char *)pHeader, dataLen, pConn->chandle); writtenLen = (*taosSendData[pRpc->connType])(pConn->peerIp, pConn->peerPort, (char *)pHeader, msgLen, pConn->chandle);
if (writtenLen != dataLen) { if (writtenLen != msgLen) {
tError("%s pConn:%p, failed to send, dataLen:%d writtenLen:%d, reason:%s", pRpc->label, pConn, tError("%s pConn:%p, failed to send, dataLen:%d writtenLen:%d, reason:%s", pRpc->label, pConn,
dataLen, writtenLen, strerror(errno)); msgLen, writtenLen, strerror(errno));
} }
tDump(data, dataLen); tDump(msg, msgLen);
} }
static void rpcProcessConnError(void *param, void *id) { static void rpcProcessConnError(void *param, void *id) {
@ -927,34 +923,20 @@ static void rpcProcessConnError(void *param, void *id) {
SRpcInfo *pRpc = pContext->pRpc; SRpcInfo *pRpc = pContext->pRpc;
if ( pContext->numOfRetry >= pContext->ipSet.numOfIps ) { if ( pContext->numOfRetry >= pContext->ipSet.numOfIps ) {
rpcFreeMsg(rpcHeaderFromCont(pContext->pCont)); // free the request msg rpcFreeOutMsg(rpcHeaderFromCont(pContext->pCont)); // free the request msg
char *rsp = calloc(1, RPC_MSG_OVERHEAD + sizeof(STaosRsp)); (*(pRpc->cfp))(pContext->msgType+1, NULL, 0, pContext->ahandle, pContext->code);
if ( rsp ) {
STaosRsp *pRsp = (STaosRsp *)(rsp+sizeof(SRpcHeader));
pRsp->code = pContext->code;
(*(pRpc->fp))(pContext->msgType+1, pRsp, sizeof(STaosRsp), pContext->ahandle, 0);
} else {
tError("%s failed to malloc RSP", pRpc->label);
}
} else { } else {
// move to next IP // move to next IP
pContext->ipSet.index++; pContext->ipSet.index++;
pContext->ipSet.index = pContext->ipSet.index % pContext->ipSet.numOfIps; pContext->ipSet.index = pContext->ipSet.index % pContext->ipSet.numOfIps;
rpcSendReqToServer(pRpc, pContext);
SRpcConn *pConn = rpcSetConnToServer(pContext->pRpc, pContext->ipSet);
pContext->code = terrno;
if (pConn == NULL) taosTmrStart(rpcProcessConnError, 0, pContext, pRpc->tmrCtrl);
rpcSendReqToServer(pConn, pContext);
} }
} }
static void rpcProcessRetryTimer(void *param, void *tmrId) { static void rpcProcessRetryTimer(void *param, void *tmrId) {
SRpcConn *pConn = (SRpcConn *)param; SRpcConn *pConn = (SRpcConn *)param;
int reportDisc = 0;
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
if (pRpc == NULL) return; // it means it is already released int reportDisc = 0;
pthread_mutex_lock(&pRpc->mutex); pthread_mutex_lock(&pRpc->mutex);
@ -966,7 +948,7 @@ static void rpcProcessRetryTimer(void *param, void *tmrId) {
if (pConn->retry < 4) { if (pConn->retry < 4) {
tTrace("%s pConn:%p, re-send msg:%s to %s:%hu", pRpc->label, tTrace("%s pConn:%p, re-send msg:%s to %s:%hu", pRpc->label,
taosMsg[pConn->outType], pConn->peerIpstr, pConn->peerPort); taosMsg[pConn->outType], pConn->peerIpstr, pConn->peerPort);
rpcSendDataToPeer(pConn, pConn->pReqMsg, pConn->reqMsgLen); rpcSendMsgToPeer(pConn, pConn->pReqMsg, pConn->reqMsgLen);
taosTmrReset(rpcProcessRetryTimer, tsRpcTimer<<pConn->retry, pConn, pRpc->tmrCtrl, &pConn->pTimer); taosTmrReset(rpcProcessRetryTimer, tsRpcTimer<<pConn->retry, pConn, pRpc->tmrCtrl, &pConn->pTimer);
} else { } else {
// close the connection // close the connection
@ -990,18 +972,13 @@ static void rpcProcessRetryTimer(void *param, void *tmrId) {
static void rpcProcessIdleTimer(void *param, void *tmrId) { static void rpcProcessIdleTimer(void *param, void *tmrId) {
SRpcConn *pConn = (SRpcConn *)param; SRpcConn *pConn = (SRpcConn *)param;
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
assert(pRpc);
pthread_mutex_lock(&pRpc->mutex); if (pConn->meterId[0]) {
if (pConn->inType == 0 && pConn->meterId[0]) {
tTrace("%s pConn:%p, close the connection since no activity", pRpc->label, pConn); tTrace("%s pConn:%p, close the connection since no activity", pRpc->label, pConn);
rpcCloseConn(pConn); rpcCloseConn(pConn);
} else { } else {
tTrace("%s pConn:%p, idle timer:%p not processed", pRpc->label, pConn, tmrId); tTrace("%s pConn:%p, idle timer:%p not processed", pRpc->label, pConn, tmrId);
} }
pthread_mutex_unlock(&pRpc->mutex);
} }
static void rpcProcessProgressTimer(void *param, void *tmrId) { static void rpcProcessProgressTimer(void *param, void *tmrId) {
@ -1021,22 +998,27 @@ static void rpcProcessProgressTimer(void *param, void *tmrId) {
pthread_mutex_unlock(&pRpc->mutex); pthread_mutex_unlock(&pRpc->mutex);
} }
static void rpcFreeMsg(void *msg) { static void rpcFreeOutMsg(void *msg) {
if ( msg == NULL ) return; if ( msg == NULL ) return;
char *req = ((char *)msg) - sizeof(SRpcReqContext); char *req = ((char *)msg) - sizeof(SRpcReqContext);
free(req); free(req);
} }
typedef struct {
int32_t reserved;
int32_t contLen;
} SRpcComp;
static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) { static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
SRpcHeader *pHeader = rpcHeaderFromCont(pCont); SRpcHeader *pHeader = rpcHeaderFromCont(pCont);
int32_t overhead = sizeof(int32_t) * 2;
int32_t finalLen = 0; int32_t finalLen = 0;
int overhead = sizeof(SRpcComp);
if (!NEEDTO_COMPRESSS_MSG(contLen)) { if (!NEEDTO_COMPRESSS_MSG(contLen)) {
return contLen; return contLen;
} }
char *buf = malloc (contLen + overhead+8); // 16 extra bytes char *buf = malloc (contLen + overhead + 8); // 8 extra bytes
if (buf == NULL) { if (buf == NULL) {
tError("failed to allocate memory for rpc msg compression, contLen:%d, reason:%s", contLen, strerror(errno)); tError("failed to allocate memory for rpc msg compression, contLen:%d, reason:%s", contLen, strerror(errno));
return contLen; return contLen;
@ -1049,20 +1031,15 @@ static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
* The first four bytes is set to 0, the second four bytes are utilized to keep the original length of message * The first four bytes is set to 0, the second four bytes are utilized to keep the original length of message
*/ */
if (compLen < contLen - overhead) { if (compLen < contLen - overhead) {
//tDump(pCont, contLen); SRpcComp *pComp = (SRpcComp *)pCont;
int32_t *pLen = (int32_t *)pCont; pComp->reserved = 0;
pComp->contLen = htonl(contLen);
*pLen = 0; // first 4 bytes must be zero
pLen = (int32_t *)(pCont + sizeof(int32_t));
*pLen = htonl(contLen); // contLen is encoded in second 4 bytes
memcpy(pCont + overhead, buf, compLen); memcpy(pCont + overhead, buf, compLen);
pHeader->comp = 1; pHeader->comp = 1;
tTrace("compress rpc msg, before:%d, after:%d", contLen, compLen); tTrace("compress rpc msg, before:%d, after:%d", contLen, compLen);
finalLen = compLen + overhead; finalLen = compLen + overhead;
//tDump(pCont, contLen);
} else { } else {
finalLen = contLen; finalLen = contLen;
} }
@ -1072,16 +1049,15 @@ static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
} }
static SRpcHeader *rpcDecompressRpcMsg(SRpcHeader *pHeader) { static SRpcHeader *rpcDecompressRpcMsg(SRpcHeader *pHeader) {
int overhead = sizeof(int32_t) * 2; int overhead = sizeof(SRpcComp);
SRpcHeader *pNewHeader = NULL; SRpcHeader *pNewHeader = NULL;
uint8_t *pCont = pHeader->content; uint8_t *pCont = pHeader->content;
SRpcComp *pComp = (SRpcComp *)pHeader->content;
if (pHeader->comp) { if (pHeader->comp) {
// decompress the content // decompress the content
assert(GET_INT32_VAL(pHeader->content) == 0); assert(pComp->reserved == 0);
int contLen = htonl(pComp->contLen);
// contLen is original message length before compression applied
int contLen = htonl(GET_INT32_VAL(pCont + sizeof(int32_t)));
// prepare the temporary buffer to decompress message // prepare the temporary buffer to decompress message
char *buf = rpcMallocCont(contLen); char *buf = rpcMallocCont(contLen);

15
src/rpc/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,15 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(../inc)
AUX_SOURCE_DIRECTORY(./ TEST_SRC)
ADD_EXECUTABLE(rpcTest ${TEST_SRC})
TARGET_LINK_LIBRARIES(rpcTest trpc)
ENDIF ()

46
src/rpc/test/unittest.c Normal file
View File

@ -0,0 +1,46 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
//#define _DEFAULT_SOURCE
#include "os.h"
#include "tlog.h"
#include "trpc.h"
#include <stdint.h>
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.fp = NULL;
rpcInit.sessions = 1000;
rpcInit.connType = TAOS_CONN_SOCKET_TYPE_S();
rpcInit.idleTime = 2000;
void *pConn = rpcOpen(&rpcInit);
if (pConn != NULL) {
dPrint("conection is opened");
} else {
dError("failed to initialize rpc");
}
return 0;
}