未验证 提交 7fbc1fc8 编写于 作者: S Shengliang Guan 提交者: GitHub

Merge pull request #16669 from taosdata/fix/rpcAuthFail

fix(rpc): conn auth failure
...@@ -13,28 +13,28 @@ ...@@ -13,28 +13,28 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "hash.h"
#include "lz4.h"
#include "os.h" #include "os.h"
#include "rpcCache.h"
#include "rpcHead.h"
#include "rpcLog.h"
#include "rpcTcp.h"
#include "rpcUdp.h"
#include "taoserror.h"
#include "taosmsg.h"
#include "tglobal.h"
#include "tidpool.h" #include "tidpool.h"
#include "tmd5.h" #include "tmd5.h"
#include "tmempool.h" #include "tmempool.h"
#include "ttimer.h"
#include "tutil.h"
#include "lz4.h"
#include "tref.h" #include "tref.h"
#include "taoserror.h"
#include "tsocket.h"
#include "tglobal.h"
#include "taosmsg.h"
#include "trpc.h" #include "trpc.h"
#include "hash.h" #include "tsocket.h"
#include "rpcLog.h" #include "ttimer.h"
#include "rpcUdp.h" #include "tutil.h"
#include "rpcCache.h"
#include "rpcTcp.h"
#include "rpcHead.h"
#define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHead) + sizeof(SRpcDigest)) #define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHead) + sizeof(SRpcDigest))
#define rpcHeadFromCont(cont) ((SRpcHead *) ((char*)cont - sizeof(SRpcHead))) #define rpcHeadFromCont(cont) ((SRpcHead *)((char *)cont - sizeof(SRpcHead)))
#define rpcContFromHead(msg) (msg + sizeof(SRpcHead)) #define rpcContFromHead(msg) (msg + sizeof(SRpcHead))
#define rpcMsgLenFromCont(contLen) (contLen + sizeof(SRpcHead)) #define rpcMsgLenFromCont(contLen) (contLen + sizeof(SRpcHead))
#define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHead)) #define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHead))
...@@ -62,8 +62,8 @@ typedef struct { ...@@ -62,8 +62,8 @@ typedef struct {
void *idPool; // handle to ID pool void *idPool; // handle to ID pool
void *tmrCtrl; // handle to timer void *tmrCtrl; // handle to timer
SHashObj *hash; // handle returned by hash utility SHashObj *hash; // handle returned by hash utility
void *tcphandle;// returned handle from TCP initialization void *tcphandle; // returned handle from TCP initialization
void *udphandle;// returned handle from UDP initialization void *udphandle; // returned handle from UDP initialization
void *pCache; // connection cache void *pCache; // connection cache
pthread_mutex_t mutex; pthread_mutex_t mutex;
struct SRpcConn *connList; // connection list struct SRpcConn *connList; // connection list
...@@ -98,7 +98,7 @@ typedef struct { ...@@ -98,7 +98,7 @@ typedef struct {
} SRpcReqContext; } SRpcReqContext;
typedef struct SRpcConn { typedef struct SRpcConn {
char info[48];// debug info: label + pConn + ahandle char info[48]; // debug info: label + pConn + ahandle
int sid; // session ID int sid; // session ID
uint32_t ownId; // own link ID uint32_t ownId; // own link ID
uint32_t peerId; // peer link ID uint32_t peerId; // peer link ID
...@@ -144,7 +144,7 @@ int tsRpcOverhead; ...@@ -144,7 +144,7 @@ int tsRpcOverhead;
static int tsRpcRefId = -1; static int tsRpcRefId = -1;
static int32_t tsRpcNum = 0; static int32_t tsRpcNum = 0;
//static pthread_once_t tsRpcInit = PTHREAD_ONCE_INIT; // static pthread_once_t tsRpcInit = PTHREAD_ONCE_INIT;
// server:0 client:1 tcp:2 udp:0 // server:0 client:1 tcp:2 udp:0
#define RPC_CONN_UDPS 0 #define RPC_CONN_UDPS 0
...@@ -153,18 +153,10 @@ static int32_t tsRpcNum = 0; ...@@ -153,18 +153,10 @@ static int32_t tsRpcNum = 0;
#define RPC_CONN_TCPC 3 #define RPC_CONN_TCPC 3
void *(*taosInitConn[])(uint32_t ip, uint16_t port, char *label, int threads, void *fp, void *shandle) = { void *(*taosInitConn[])(uint32_t ip, uint16_t port, char *label, int threads, void *fp, void *shandle) = {
taosInitUdpConnection, taosInitUdpConnection, taosInitUdpConnection, taosInitTcpServer, taosInitTcpClient};
taosInitUdpConnection,
taosInitTcpServer,
taosInitTcpClient
};
void (*taosCleanUpConn[])(void *thandle) = { void (*taosCleanUpConn[])(void *thandle) = {taosCleanUpUdpConnection, taosCleanUpUdpConnection, taosCleanUpTcpServer,
taosCleanUpUdpConnection, taosCleanUpTcpClient};
taosCleanUpUdpConnection,
taosCleanUpTcpServer,
taosCleanUpTcpClient
};
void (*taosStopConn[])(void *thandle) = { void (*taosStopConn[])(void *thandle) = {
taosStopUdpConnection, taosStopUdpConnection,
...@@ -174,11 +166,7 @@ void (*taosStopConn[])(void *thandle) = { ...@@ -174,11 +166,7 @@ void (*taosStopConn[])(void *thandle) = {
}; };
int (*taosSendData[])(uint32_t ip, uint16_t port, void *data, int len, void *chandle) = { int (*taosSendData[])(uint32_t ip, uint16_t port, void *data, int len, void *chandle) = {
taosSendUdpData, taosSendUdpData, taosSendUdpData, taosSendTcpData, taosSendTcpData};
taosSendUdpData,
taosSendTcpData,
taosSendTcpData
};
void *(*taosOpenConn[])(void *shandle, void *thandle, uint32_t ip, uint16_t port) = { void *(*taosOpenConn[])(void *shandle, void *thandle, uint32_t ip, uint16_t port) = {
taosOpenUdpConnection, taosOpenUdpConnection,
...@@ -187,12 +175,7 @@ void *(*taosOpenConn[])(void *shandle, void *thandle, uint32_t ip, uint16_t port ...@@ -187,12 +175,7 @@ void *(*taosOpenConn[])(void *shandle, void *thandle, uint32_t ip, uint16_t port
taosOpenTcpClientConnection, taosOpenTcpClientConnection,
}; };
void (*taosCloseConn[])(void *chandle) = { void (*taosCloseConn[])(void *chandle) = {NULL, NULL, taosCloseTcpConnection, taosCloseTcpConnection};
NULL,
NULL,
taosCloseTcpConnection,
taosCloseTcpConnection
};
static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerFqdn, uint16_t peerPort, int8_t connType); static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerFqdn, uint16_t peerPort, int8_t connType);
static void rpcCloseConn(void *thandle); static void rpcCloseConn(void *thandle);
...@@ -215,7 +198,7 @@ static void rpcProcessIdleTimer(void *param, void *tmrId); ...@@ -215,7 +198,7 @@ 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 rpcFreeMsg(void *msg);
static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen); static int32_t rpcCompressRpcMsg(char *pCont, int32_t contLen);
static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead); static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead);
static int rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen); static int rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen);
static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen); static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen);
...@@ -224,14 +207,44 @@ static void rpcUnlockConn(SRpcConn *pConn); ...@@ -224,14 +207,44 @@ static void rpcUnlockConn(SRpcConn *pConn);
static void rpcAddRef(SRpcInfo *pRpc); static void rpcAddRef(SRpcInfo *pRpc);
static void rpcDecRef(SRpcInfo *pRpc); static void rpcDecRef(SRpcInfo *pRpc);
static bool rpcGenUID(uint32_t *first, uint32_t *second) {
static uint64_t hashId = 0;
static uint32_t tranId = 0;
if (hashId == 0) {
char uid[128] = {0};
taosGetSystemUid(uid);
hashId = MurmurHash3_32(uid, strlen(uid));
}
uint64_t id = 0;
while (true) {
int64_t ts = taosGetTimestampMs();
uint64_t pid = taosGetPId();
int32_t val = atomic_add_fetch_32(&tranId, 1);
if (val >= 0xFFFF) {
atomic_store_32(&tranId, 0);
}
id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
if (id) {
break;
}
}
*first = (id >> 32) & 0xFFFFFFFF;
*second = id & 0xFFFFFFFF;
return true;
}
static void rpcFree(void *p) { static void rpcFree(void *p) {
tTrace("free mem: %p", p); tTrace("free mem: %p", p);
free(p); free(p);
} }
int32_t rpcInit(void) { int32_t rpcInit(void) {
tsProgressTimer = tsRpcTimer/2; tsProgressTimer = tsRpcTimer / 2;
tsRpcMaxRetry = tsRpcMaxTime * 1000/tsProgressTimer; tsRpcMaxRetry = tsRpcMaxTime * 1000 / tsProgressTimer;
tsRpcHeadSize = RPC_MSG_OVERHEAD; tsRpcHeadSize = RPC_MSG_OVERHEAD;
tsRpcOverhead = sizeof(SRpcReqContext); tsRpcOverhead = sizeof(SRpcReqContext);
...@@ -248,22 +261,22 @@ void rpcCleanup(void) { ...@@ -248,22 +261,22 @@ void rpcCleanup(void) {
void *rpcOpen(const SRpcInit *pInit) { void *rpcOpen(const SRpcInit *pInit) {
SRpcInfo *pRpc; SRpcInfo *pRpc;
//pthread_once(&tsRpcInit, rpcInit); // pthread_once(&tsRpcInit, rpcInit);
pRpc = (SRpcInfo *)calloc(1, sizeof(SRpcInfo)); pRpc = (SRpcInfo *)calloc(1, sizeof(SRpcInfo));
if (pRpc == NULL) return NULL; if (pRpc == NULL) return NULL;
if(pInit->label) tstrncpy(pRpc->label, pInit->label, sizeof(pRpc->label)); if (pInit->label) tstrncpy(pRpc->label, pInit->label, sizeof(pRpc->label));
pRpc->connType = pInit->connType; pRpc->connType = pInit->connType;
if (pRpc->connType == TAOS_CONN_CLIENT) { if (pRpc->connType == TAOS_CONN_CLIENT) {
pRpc->numOfThreads = pInit->numOfThreads; pRpc->numOfThreads = pInit->numOfThreads;
} else { } else {
pRpc->numOfThreads = pInit->numOfThreads>TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS:pInit->numOfThreads; pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
} }
pRpc->idleTime = pInit->idleTime; pRpc->idleTime = pInit->idleTime;
pRpc->localPort = pInit->localPort; pRpc->localPort = pInit->localPort;
pRpc->afp = pInit->afp; pRpc->afp = pInit->afp;
pRpc->sessions = pInit->sessions+1; pRpc->sessions = pInit->sessions + 1;
if (pInit->user) tstrncpy(pRpc->user, pInit->user, sizeof(pRpc->user)); if (pInit->user) tstrncpy(pRpc->user, pInit->user, sizeof(pRpc->user));
if (pInit->secret) memcpy(pRpc->secret, pInit->secret, sizeof(pRpc->secret)); if (pInit->secret) memcpy(pRpc->secret, pInit->secret, sizeof(pRpc->secret));
if (pInit->ckey) tstrncpy(pRpc->ckey, pInit->ckey, sizeof(pRpc->ckey)); if (pInit->ckey) tstrncpy(pRpc->ckey, pInit->ckey, sizeof(pRpc->ckey));
...@@ -282,14 +295,14 @@ void *rpcOpen(const SRpcInit *pInit) { ...@@ -282,14 +295,14 @@ void *rpcOpen(const SRpcInit *pInit) {
return NULL; return NULL;
} }
pRpc->idPool = taosInitIdPool(pRpc->sessions-1); pRpc->idPool = taosInitIdPool(pRpc->sessions - 1);
if (pRpc->idPool == NULL) { if (pRpc->idPool == NULL) {
tError("%s failed to init ID pool", pRpc->label); tError("%s failed to init ID pool", pRpc->label);
rpcClose(pRpc); rpcClose(pRpc);
return NULL; return NULL;
} }
pRpc->tmrCtrl = taosTmrInit(pRpc->sessions*2 + 1, 50, 10000, pRpc->label); pRpc->tmrCtrl = taosTmrInit(pRpc->sessions * 2 + 1, 50, 10000, pRpc->label);
if (pRpc->tmrCtrl == NULL) { if (pRpc->tmrCtrl == NULL) {
tError("%s failed to init timers", pRpc->label); tError("%s failed to init timers", pRpc->label);
rpcClose(pRpc); rpcClose(pRpc);
...@@ -305,7 +318,7 @@ void *rpcOpen(const SRpcInit *pInit) { ...@@ -305,7 +318,7 @@ void *rpcOpen(const SRpcInit *pInit) {
} }
} else { } else {
pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime * 20); pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime * 20);
if ( pRpc->pCache == NULL ) { if (pRpc->pCache == NULL) {
tError("%s failed to init connection cache", pRpc->label); tError("%s failed to init connection cache", pRpc->label);
rpcClose(pRpc); rpcClose(pRpc);
return NULL; return NULL;
...@@ -314,10 +327,10 @@ void *rpcOpen(const SRpcInit *pInit) { ...@@ -314,10 +327,10 @@ void *rpcOpen(const SRpcInit *pInit) {
pthread_mutex_init(&pRpc->mutex, NULL); pthread_mutex_init(&pRpc->mutex, NULL);
pRpc->tcphandle = (*taosInitConn[pRpc->connType|RPC_CONN_TCP])(0, pRpc->localPort, pRpc->label, pRpc->tcphandle = (*taosInitConn[pRpc->connType | RPC_CONN_TCP])(0, pRpc->localPort, pRpc->label, pRpc->numOfThreads,
pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc); rpcProcessMsgFromPeer, pRpc);
pRpc->udphandle = (*taosInitConn[pRpc->connType])(0, pRpc->localPort, pRpc->label, pRpc->udphandle =
pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc); (*taosInitConn[pRpc->connType])(0, pRpc->localPort, pRpc->label, pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc);
if (pRpc->tcphandle == NULL || pRpc->udphandle == NULL) { if (pRpc->tcphandle == NULL || pRpc->udphandle == NULL) {
tError("%s failed to init network, port:%d", pRpc->label, pRpc->localPort); tError("%s failed to init network, port:%d", pRpc->label, pRpc->localPort);
...@@ -378,7 +391,7 @@ void *rpcReallocCont(void *ptr, int contLen) { ...@@ -378,7 +391,7 @@ void *rpcReallocCont(void *ptr, int contLen) {
if (ptr == NULL) return rpcMallocCont(contLen); if (ptr == NULL) return rpcMallocCont(contLen);
char *start = ((char *)ptr) - sizeof(SRpcReqContext) - sizeof(SRpcHead); char *start = ((char *)ptr) - sizeof(SRpcReqContext) - sizeof(SRpcHead);
if (contLen == 0 ) { if (contLen == 0) {
free(start); free(start);
return NULL; return NULL;
} }
...@@ -398,7 +411,7 @@ TBOOL rpcSendRequest(void *shandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg, int6 ...@@ -398,7 +411,7 @@ TBOOL rpcSendRequest(void *shandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg, int6
SRpcReqContext *pContext; SRpcReqContext *pContext;
int contLen = rpcCompressRpcMsg(pMsg->pCont, pMsg->contLen); int contLen = rpcCompressRpcMsg(pMsg->pCont, pMsg->contLen);
pContext = (SRpcReqContext *) ((char*)pMsg->pCont-sizeof(SRpcHead)-sizeof(SRpcReqContext)); pContext = (SRpcReqContext *)((char *)pMsg->pCont - sizeof(SRpcHead) - sizeof(SRpcReqContext));
pContext->ahandle = pMsg->ahandle; pContext->ahandle = pMsg->ahandle;
pContext->pRpc = (SRpcInfo *)shandle; pContext->pRpc = (SRpcInfo *)shandle;
pContext->epSet = *pEpSet; pContext->epSet = *pEpSet;
...@@ -414,10 +427,10 @@ TBOOL rpcSendRequest(void *shandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg, int6 ...@@ -414,10 +427,10 @@ TBOOL rpcSendRequest(void *shandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg, int6
// connection type is application specific. // connection type is application specific.
// for TDengine, all the query, show commands shall have TCP connection // for TDengine, all the query, show commands shall have TCP connection
char type = pMsg->msgType; char type = pMsg->msgType;
if (type == TSDB_MSG_TYPE_QUERY || type == TSDB_MSG_TYPE_CM_RETRIEVE if (type == TSDB_MSG_TYPE_QUERY || type == TSDB_MSG_TYPE_CM_RETRIEVE || type == TSDB_MSG_TYPE_FETCH ||
|| type == TSDB_MSG_TYPE_FETCH || type == TSDB_MSG_TYPE_CM_STABLE_VGROUP type == TSDB_MSG_TYPE_CM_STABLE_VGROUP || type == TSDB_MSG_TYPE_CM_TABLES_META ||
|| type == TSDB_MSG_TYPE_CM_TABLES_META || type == TSDB_MSG_TYPE_CM_TABLE_META type == TSDB_MSG_TYPE_CM_TABLE_META || type == TSDB_MSG_TYPE_CM_SHOW || type == TSDB_MSG_TYPE_DM_STATUS ||
|| type == TSDB_MSG_TYPE_CM_SHOW || type == TSDB_MSG_TYPE_DM_STATUS || type == TSDB_MSG_TYPE_CM_ALTER_TABLE) type == TSDB_MSG_TYPE_CM_ALTER_TABLE)
pContext->connType = RPC_CONN_TCPC; pContext->connType = RPC_CONN_TCPC;
pContext->rid = taosAddRef(tsRpcRefId, pContext); pContext->rid = taosAddRef(tsRpcRefId, pContext);
...@@ -433,7 +446,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) { ...@@ -433,7 +446,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
SRpcMsg *pMsg = &rpcMsg; SRpcMsg *pMsg = &rpcMsg;
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
if ( pMsg->pCont == NULL ) { if (pMsg->pCont == NULL) {
pMsg->pCont = rpcMallocCont(0); pMsg->pCont = rpcMallocCont(0);
pMsg->contLen = 0; pMsg->contLen = 0;
} }
...@@ -446,7 +459,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) { ...@@ -446,7 +459,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
rpcLockConn(pConn); rpcLockConn(pConn);
if ( pConn->inType == 0 || pConn->user[0] == 0 ) { if (pConn->inType == 0 || pConn->user[0] == 0) {
tError("%s, connection is already released, rsp wont be sent", pConn->info); tError("%s, connection is already released, rsp wont be sent", pConn->info);
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
...@@ -456,7 +469,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) { ...@@ -456,7 +469,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
// set msg header // set msg header
pHead->version = 1; pHead->version = 1;
pHead->msgType = pConn->inType+1; pHead->msgType = pConn->inType + 1;
pHead->spi = pConn->spi; pHead->spi = pConn->spi;
pHead->encrypt = pConn->encrypt; pHead->encrypt = pConn->encrypt;
pHead->tranId = pConn->inTranId; pHead->tranId = pConn->inTranId;
...@@ -465,7 +478,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) { ...@@ -465,7 +478,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
pHead->linkUid = pConn->linkUid; pHead->linkUid = pConn->linkUid;
pHead->port = htons(pConn->localPort); pHead->port = htons(pConn->localPort);
pHead->code = htonl(pMsg->code); pHead->code = htonl(pMsg->code);
pHead->ahandle = (uint64_t) pConn->ahandle; pHead->ahandle = (uint64_t)pConn->ahandle;
// set pConn parameters // set pConn parameters
pConn->inType = 0; pConn->inType = 0;
...@@ -484,8 +497,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) { ...@@ -484,8 +497,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
rpcSendMsgToPeer(pConn, msg, msgLen); rpcSendMsgToPeer(pConn, msg, msgLen);
// if not set to secured, set it expcet NOT_READY case, since client wont treat it as secured // if not set to secured, set it expcet NOT_READY case, since client wont treat it as secured
if (pConn->secured == 0 && pMsg->code != TSDB_CODE_RPC_NOT_READY) if (pConn->secured == 0 && pMsg->code != TSDB_CODE_RPC_NOT_READY) pConn->secured = 1; // connection shall be secured
pConn->secured = 1; // connection shall be secured
if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg); if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg);
pConn->pReqMsg = NULL; pConn->pReqMsg = NULL;
...@@ -529,7 +541,7 @@ int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) { ...@@ -529,7 +541,7 @@ int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
void rpcSendRecv(void *shandle, SRpcEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { void rpcSendRecv(void *shandle, SRpcEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
SRpcReqContext *pContext; SRpcReqContext *pContext;
pContext = (SRpcReqContext *) ((char*)pMsg->pCont-sizeof(SRpcHead)-sizeof(SRpcReqContext)); pContext = (SRpcReqContext *)((char *)pMsg->pCont - sizeof(SRpcHead) - sizeof(SRpcReqContext));
memset(pRsp, 0, sizeof(SRpcMsg)); memset(pRsp, 0, sizeof(SRpcMsg));
...@@ -569,7 +581,6 @@ int rpcReportProgress(void *handle, char *pCont, int contLen) { ...@@ -569,7 +581,6 @@ int rpcReportProgress(void *handle, char *pCont, int contLen) {
} }
void rpcCancelRequest(int64_t rid) { void rpcCancelRequest(int64_t rid) {
SRpcReqContext *pContext = taosAcquireRef(tsRpcRefId, rid); SRpcReqContext *pContext = taosAcquireRef(tsRpcRefId, rid);
if (pContext == NULL) return; if (pContext == NULL) return;
...@@ -579,7 +590,7 @@ void rpcCancelRequest(int64_t rid) { ...@@ -579,7 +590,7 @@ void rpcCancelRequest(int64_t rid) {
} }
static void rpcFreeMsg(void *msg) { static void rpcFreeMsg(void *msg) {
if ( msg ) { if (msg) {
char *temp = (char *)msg - sizeof(SRpcReqContext); char *temp = (char *)msg - sizeof(SRpcReqContext);
free(temp); free(temp);
tTrace("free mem: %p", temp); tTrace("free mem: %p", temp);
...@@ -606,7 +617,7 @@ static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerFqdn, uint16_t peerPort, ...@@ -606,7 +617,7 @@ static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerFqdn, uint16_t peerPort,
pConn->connType = connType; pConn->connType = connType;
if (taosOpenConn[connType]) { if (taosOpenConn[connType]) {
void *shandle = (connType & RPC_CONN_TCP)? pRpc->tcphandle:pRpc->udphandle; void *shandle = (connType & RPC_CONN_TCP) ? pRpc->tcphandle : pRpc->udphandle;
pConn->chandle = (*taosOpenConn[connType])(shandle, pConn, pConn->peerIp, pConn->peerPort); pConn->chandle = (*taosOpenConn[connType])(shandle, pConn, pConn->peerIp, pConn->peerPort);
if (pConn->chandle == NULL) { if (pConn->chandle == NULL) {
tError("failed to connect to:0x%x:%d", pConn->peerIp, pConn->peerPort); tError("failed to connect to:0x%x:%d", pConn->peerIp, pConn->peerPort);
...@@ -631,9 +642,10 @@ static void rpcReleaseConn(SRpcConn *pConn) { ...@@ -631,9 +642,10 @@ static void rpcReleaseConn(SRpcConn *pConn) {
taosTmrStopA(&pConn->pTimer); taosTmrStopA(&pConn->pTimer);
taosTmrStopA(&pConn->pIdleTimer); taosTmrStopA(&pConn->pIdleTimer);
if ( pRpc->connType == TAOS_CONN_SERVER) { if (pRpc->connType == TAOS_CONN_SERVER) {
char hashstr[40] = {0}; char hashstr[40] = {0};
size_t size = snprintf(hashstr, sizeof(hashstr), "%x:%x:%x:%d", pConn->peerIp, pConn->linkUid, pConn->peerId, pConn->connType); size_t size = snprintf(hashstr, sizeof(hashstr), "%x:%x:%x:%d", pConn->peerIp, pConn->linkUid, pConn->peerId,
pConn->connType);
taosHashRemove(pRpc->hash, hashstr, size); taosHashRemove(pRpc->hash, hashstr, size);
rpcFreeMsg(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->pRspMsg = NULL;
...@@ -684,8 +696,7 @@ static void rpcCloseConn(void *thandle) { ...@@ -684,8 +696,7 @@ static void rpcCloseConn(void *thandle) {
rpcLockConn(pConn); rpcLockConn(pConn);
if (pConn->user[0]) if (pConn->user[0]) rpcReleaseConn(pConn);
rpcReleaseConn(pConn);
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
} }
...@@ -693,6 +704,9 @@ static void rpcCloseConn(void *thandle) { ...@@ -693,6 +704,9 @@ static void rpcCloseConn(void *thandle) {
static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc) { static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc) {
SRpcConn *pConn = NULL; SRpcConn *pConn = NULL;
uint32_t transId, linkUid;
rpcGenUID(&transId, &linkUid);
int sid = taosAllocateId(pRpc->idPool); int sid = taosAllocateId(pRpc->idPool);
if (sid <= 0) { if (sid <= 0) {
tError("%s maximum number of sessions:%d is reached", pRpc->label, pRpc->sessions); tError("%s maximum number of sessions:%d is reached", pRpc->label, pRpc->sessions);
...@@ -702,9 +716,9 @@ static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc) { ...@@ -702,9 +716,9 @@ static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc) {
pConn->pRpc = pRpc; pConn->pRpc = pRpc;
pConn->sid = sid; pConn->sid = sid;
pConn->tranId = (uint16_t)(taosRand() & 0xFFFF); pConn->tranId = transId;
pConn->ownId = htonl(pConn->sid); pConn->ownId = htonl(pConn->sid);
pConn->linkUid = (uint32_t)((int64_t)pConn + (int64_t)getpid() + (int64_t)pConn->tranId); pConn->linkUid = linkUid;
pConn->spi = pRpc->spi; pConn->spi = pRpc->spi;
pConn->encrypt = pRpc->encrypt; pConn->encrypt = pRpc->encrypt;
if (pConn->spi) memcpy(pConn->secret, pRpc->secret, TSDB_KEY_LEN); if (pConn->spi) memcpy(pConn->secret, pRpc->secret, TSDB_KEY_LEN);
...@@ -719,7 +733,8 @@ static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) { ...@@ -719,7 +733,8 @@ static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) {
char hashstr[40] = {0}; char hashstr[40] = {0};
SRpcHead *pHead = (SRpcHead *)pRecv->msg; SRpcHead *pHead = (SRpcHead *)pRecv->msg;
size_t size = snprintf(hashstr, sizeof(hashstr), "%x:%x:%x:%d", pRecv->ip, pHead->linkUid, pHead->sourceId, pRecv->connType); size_t size =
snprintf(hashstr, sizeof(hashstr), "%x:%x:%x:%d", pRecv->ip, pHead->linkUid, pHead->sourceId, pRecv->connType);
// check if it is already allocated // check if it is already allocated
SRpcConn **ppConn = (SRpcConn **)(taosHashGet(pRpc->hash, hashstr, size)); SRpcConn **ppConn = (SRpcConn **)(taosHashGet(pRpc->hash, hashstr, size));
...@@ -769,7 +784,8 @@ static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) { ...@@ -769,7 +784,8 @@ static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) {
} }
taosHashPut(pRpc->hash, hashstr, size, (char *)&pConn, POINTER_BYTES); taosHashPut(pRpc->hash, hashstr, size, (char *)&pConn, POINTER_BYTES);
tDebug("%s %p server connection is allocated, uid:0x%x sid:%d key:%s", pRpc->label, pConn, pConn->linkUid, sid, hashstr); tDebug("%s %p server connection is allocated, uid:0x%x sid:%d key:%s", pRpc->label, pConn, pConn->linkUid, sid,
hashstr);
} }
return pConn; return pConn;
...@@ -809,8 +825,9 @@ static SRpcConn *rpcSetupConnToServer(SRpcReqContext *pContext) { ...@@ -809,8 +825,9 @@ static SRpcConn *rpcSetupConnToServer(SRpcReqContext *pContext) {
SRpcInfo *pRpc = pContext->pRpc; SRpcInfo *pRpc = pContext->pRpc;
SRpcEpSet *pEpSet = &pContext->epSet; SRpcEpSet *pEpSet = &pContext->epSet;
pConn = rpcGetConnFromCache(pRpc->pCache, pEpSet->fqdn[pEpSet->inUse], pEpSet->port[pEpSet->inUse], pContext->connType); pConn =
if ( pConn == NULL || pConn->user[0] == 0) { rpcGetConnFromCache(pRpc->pCache, pEpSet->fqdn[pEpSet->inUse], pEpSet->port[pEpSet->inUse], pContext->connType);
if (pConn == NULL || pConn->user[0] == 0) {
pConn = rpcOpenConn(pRpc, pEpSet->fqdn[pEpSet->inUse], pEpSet->port[pEpSet->inUse], pContext->connType); pConn = rpcOpenConn(pRpc, pEpSet->fqdn[pEpSet->inUse], pEpSet->port[pEpSet->inUse], pContext->connType);
} }
...@@ -827,13 +844,11 @@ static SRpcConn *rpcSetupConnToServer(SRpcReqContext *pContext) { ...@@ -827,13 +844,11 @@ static SRpcConn *rpcSetupConnToServer(SRpcReqContext *pContext) {
} }
static int rpcProcessReqHead(SRpcConn *pConn, SRpcHead *pHead) { static int rpcProcessReqHead(SRpcConn *pConn, SRpcHead *pHead) {
if (pConn->peerId == 0) { if (pConn->peerId == 0) {
pConn->peerId = pHead->sourceId; pConn->peerId = pHead->sourceId;
} else { } else {
if (pConn->peerId != pHead->sourceId) { if (pConn->peerId != pHead->sourceId) {
tDebug("%s, source Id is changed, old:0x%08x new:0x%08x", pConn->info, tDebug("%s, source Id is changed, old:0x%08x new:0x%08x", pConn->info, pConn->peerId, pHead->sourceId);
pConn->peerId, pHead->sourceId);
return TSDB_CODE_RPC_INVALID_VALUE; return TSDB_CODE_RPC_INVALID_VALUE;
} }
} }
...@@ -858,8 +873,7 @@ static int rpcProcessReqHead(SRpcConn *pConn, SRpcHead *pHead) { ...@@ -858,8 +873,7 @@ static int rpcProcessReqHead(SRpcConn *pConn, SRpcHead *pHead) {
} }
if (pConn->inType != 0) { if (pConn->inType != 0) {
tDebug("%s, last session is not finished, inTranId:%d tranId:%d", pConn->info, tDebug("%s, last session is not finished, inTranId:%d tranId:%d", pConn->info, pConn->inTranId, pHead->tranId);
pConn->inTranId, pHead->tranId);
return TSDB_CODE_RPC_LAST_SESSION_NOT_FINISHED; return TSDB_CODE_RPC_LAST_SESSION_NOT_FINISHED;
} }
...@@ -963,20 +977,23 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont ...@@ -963,20 +977,23 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont
if (pHead->msgType >= TSDB_MSG_TYPE_MAX || pHead->msgType <= 0) { if (pHead->msgType >= TSDB_MSG_TYPE_MAX || pHead->msgType <= 0) {
tDebug("%s sid:%d, invalid message type:%d", pRpc->label, sid, pHead->msgType); tDebug("%s sid:%d, invalid message type:%d", pRpc->label, sid, pHead->msgType);
terrno = TSDB_CODE_RPC_INVALID_MSG_TYPE; return NULL; terrno = TSDB_CODE_RPC_INVALID_MSG_TYPE;
return NULL;
} }
if (sid < 0 || sid >= pRpc->sessions) { if (sid < 0 || sid >= pRpc->sessions) {
tDebug("%s sid:%d, sid is out of range, max sid:%d, %s discarded", pRpc->label, sid, tDebug("%s sid:%d, sid is out of range, max sid:%d, %s discarded", pRpc->label, sid, pRpc->sessions,
pRpc->sessions, taosMsg[pHead->msgType]); taosMsg[pHead->msgType]);
terrno = TSDB_CODE_RPC_INVALID_SESSION_ID; return NULL; terrno = TSDB_CODE_RPC_INVALID_SESSION_ID;
return NULL;
} }
// compatibility between old version client and new version server, since 2.4.0.0 // compatibility between old version client and new version server, since 2.4.0.0
if (rpcIsReq(pHead->msgType)){ if (rpcIsReq(pHead->msgType)) {
if((htonl(pHead->msgVer) >> 16 != tsVersion >> 24) || if ((htonl(pHead->msgVer) >> 16 != tsVersion >> 24) ||
((htonl(pHead->msgVer) >> 16 == tsVersion >> 24) && htonl(pHead->msgVer) < ((2 << 16) | (4 << 8)))){ ((htonl(pHead->msgVer) >> 16 == tsVersion >> 24) && htonl(pHead->msgVer) < ((2 << 16) | (4 << 8)))) {
tError("%s sid:%d, invalid client version:%x/%x %s", pRpc->label, sid, htonl(pHead->msgVer), tsVersion, taosMsg[pHead->msgType]); tError("%s sid:%d, invalid client version:%x/%x %s", pRpc->label, sid, htonl(pHead->msgVer), tsVersion,
taosMsg[pHead->msgType]);
terrno = TSDB_CODE_RPC_INVALID_VERSION; terrno = TSDB_CODE_RPC_INVALID_VERSION;
return NULL; return NULL;
} }
...@@ -1016,7 +1033,7 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont ...@@ -1016,7 +1033,7 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont
// decrypt here // decrypt here
} }
if ( rpcIsReq(pHead->msgType) ) { if (rpcIsReq(pHead->msgType)) {
pConn->connType = pRecv->connType; pConn->connType = pRecv->connType;
terrno = rpcProcessReqHead(pConn, pHead); terrno = rpcProcessReqHead(pConn, pHead);
...@@ -1025,7 +1042,7 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont ...@@ -1025,7 +1042,7 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont
// client shall send the request within tsRpcTime again for UDP, double it // client shall send the request within tsRpcTime again for UDP, double it
if (pConn->connType != RPC_CONN_TCPS) if (pConn->connType != RPC_CONN_TCPS)
pConn->pIdleTimer = taosTmrStart(rpcProcessIdleTimer, tsRpcTimer*20, pConn, pRpc->tmrCtrl); pConn->pIdleTimer = taosTmrStart(rpcProcessIdleTimer, tsRpcTimer * 20, pConn, pRpc->tmrCtrl);
} else { } else {
terrno = rpcProcessRspHead(pConn, pHead); terrno = rpcProcessRspHead(pConn, pHead);
*ppContext = pConn->pContext; *ppContext = pConn->pContext;
...@@ -1120,7 +1137,7 @@ static void rpcProcessProbeMsg(SRecvInfo *pRecv, SRpcConn *pConn) { ...@@ -1120,7 +1137,7 @@ static void rpcProcessProbeMsg(SRecvInfo *pRecv, SRpcConn *pConn) {
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
rpcFreeMsg(pRecv->msg); rpcFreeMsg(pRecv->msg);
} else if (pHead->msgType == TSDB_MSG_TYPE_PROBE_CONN_RSP) { } else if (pHead->msgType == TSDB_MSG_TYPE_PROBE_CONN_RSP) {
if(pConn) { if (pConn) {
rpcLockConn(pConn); rpcLockConn(pConn);
// get req content // get req content
SRpcReqContext *pContext = taosAcquireRef(tsRpcRefId, pConn->rid); SRpcReqContext *pContext = taosAcquireRef(tsRpcRefId, pConn->rid);
...@@ -1129,7 +1146,8 @@ static void rpcProcessProbeMsg(SRecvInfo *pRecv, SRpcConn *pConn) { ...@@ -1129,7 +1146,8 @@ static void rpcProcessProbeMsg(SRecvInfo *pRecv, SRpcConn *pConn) {
rpcProcessIncomingMsg(pConn, pHead, pContext); rpcProcessIncomingMsg(pConn, pHead, pContext);
taosReleaseRef(tsRpcRefId, pConn->rid); taosReleaseRef(tsRpcRefId, pConn->rid);
} else { } else {
tInfo("PROBE 0x%" PRIx64 " recv response probe msg but pContext is NULL. pConn->rid=0x%" PRIX64, ahandle, pConn->rid); tInfo("PROBE 0x%" PRIx64 " recv response probe msg but pContext is NULL. pConn->rid=0x%" PRIX64, ahandle,
pConn->rid);
rpcFreeMsg(pRecv->msg); rpcFreeMsg(pRecv->msg);
} }
...@@ -1184,10 +1202,12 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) { ...@@ -1184,10 +1202,12 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE) { if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE) {
rpcCloseConn(pConn); rpcCloseConn(pConn);
} }
if (pHead->msgType + 1 > 1 && pHead->msgType+1 < TSDB_MSG_TYPE_MAX) { if (pHead->msgType + 1 > 1 && pHead->msgType + 1 < TSDB_MSG_TYPE_MAX) {
tDebug("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle, taosMsg[pHead->msgType+1], code); tDebug("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle,
taosMsg[pHead->msgType + 1], code);
} else { } else {
tError("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle, taosMsg[pHead->msgType], code); tError("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle,
taosMsg[pHead->msgType], code);
} }
} }
} else { // msg is passed to app only parsing is ok } else { // msg is passed to app only parsing is ok
...@@ -1211,21 +1231,18 @@ static void rpcNotifyClient(SRpcReqContext *pContext, SRpcMsg *pMsg) { ...@@ -1211,21 +1231,18 @@ static void rpcNotifyClient(SRpcReqContext *pContext, SRpcMsg *pMsg) {
} else { } else {
// for asynchronous API // for asynchronous API
SRpcEpSet *pEpSet = NULL; SRpcEpSet *pEpSet = NULL;
if (pContext->epSet.inUse != pContext->oldInUse || pContext->redirect) if (pContext->epSet.inUse != pContext->oldInUse || pContext->redirect) pEpSet = &pContext->epSet;
pEpSet = &pContext->epSet;
(*pRpc->cfp)(pMsg, pEpSet); (*pRpc->cfp)(pMsg, pEpSet);
} }
// free the request message // free the request message
if(pMsg->msgType != TSDB_MSG_TYPE_PROBE_CONN && pMsg->msgType != TSDB_MSG_TYPE_PROBE_CONN_RSP) { if (pMsg->msgType != TSDB_MSG_TYPE_PROBE_CONN && pMsg->msgType != TSDB_MSG_TYPE_PROBE_CONN_RSP) {
taosRemoveRef(tsRpcRefId, pContext->rid); taosRemoveRef(tsRpcRefId, pContext->rid);
} }
} }
static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqContext *pContext) { static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqContext *pContext) {
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
...@@ -1235,7 +1252,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte ...@@ -1235,7 +1252,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte
rpcMsg.msgType = pHead->msgType; rpcMsg.msgType = pHead->msgType;
rpcMsg.code = pHead->code; rpcMsg.code = pHead->code;
if ( rpcIsReq(pHead->msgType) ) { if (rpcIsReq(pHead->msgType)) {
rpcMsg.ahandle = pConn->ahandle; rpcMsg.ahandle = pConn->ahandle;
rpcMsg.handle = pConn; rpcMsg.handle = pConn;
rpcAddRef(pRpc); // add the refCount for requests rpcAddRef(pRpc); // add the refCount for requests
...@@ -1249,7 +1266,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte ...@@ -1249,7 +1266,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte
if (pHead->msgType == TSDB_MSG_TYPE_PROBE_CONN_RSP) { if (pHead->msgType == TSDB_MSG_TYPE_PROBE_CONN_RSP) {
// probe msg // probe msg
rpcNotifyClient(pContext, &rpcMsg); rpcNotifyClient(pContext, &rpcMsg);
return ; return;
} }
// reset pConn NULL // reset pConn NULL
...@@ -1257,14 +1274,15 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte ...@@ -1257,14 +1274,15 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte
// for UDP, port may be changed by server, the port in epSet shall be used for cache // for UDP, port may be changed by server, the port in epSet shall be used for cache
if (pHead->code != TSDB_CODE_RPC_TOO_SLOW) { if (pHead->code != TSDB_CODE_RPC_TOO_SLOW) {
rpcAddConnIntoCache(pRpc->pCache, pConn, pConn->peerFqdn, pContext->epSet.port[pContext->epSet.inUse], pConn->connType); rpcAddConnIntoCache(pRpc->pCache, pConn, pConn->peerFqdn, pContext->epSet.port[pContext->epSet.inUse],
pConn->connType);
} else { } else {
rpcCloseConn(pConn); rpcCloseConn(pConn);
} }
if (pHead->code == TSDB_CODE_RPC_REDIRECT) { if (pHead->code == TSDB_CODE_RPC_REDIRECT) {
pContext->numOfTry = 0; pContext->numOfTry = 0;
SRpcEpSet *pEpSet = (SRpcEpSet*)pHead->content; SRpcEpSet *pEpSet = (SRpcEpSet *)pHead->content;
if (pEpSet->numOfEps > 0) { if (pEpSet->numOfEps > 0) {
memcpy(&pContext->epSet, pHead->content, sizeof(pContext->epSet)); memcpy(&pContext->epSet, pHead->content, sizeof(pContext->epSet));
tDebug("%s, redirect is received, numOfEps:%d inUse:%d", pConn->info, pContext->epSet.numOfEps, tDebug("%s, redirect is received, numOfEps:%d inUse:%d", pConn->info, pContext->epSet.numOfEps,
...@@ -1277,7 +1295,8 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte ...@@ -1277,7 +1295,8 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte
} }
rpcSendReqToServer(pRpc, pContext); rpcSendReqToServer(pRpc, pContext);
rpcFreeCont(rpcMsg.pCont); rpcFreeCont(rpcMsg.pCont);
} else if (pHead->code == TSDB_CODE_RPC_NOT_READY || pHead->code == TSDB_CODE_APP_NOT_READY || pHead->code == TSDB_CODE_DND_EXITING) { } else if (pHead->code == TSDB_CODE_RPC_NOT_READY || pHead->code == TSDB_CODE_APP_NOT_READY ||
pHead->code == TSDB_CODE_DND_EXITING) {
pContext->code = pHead->code; pContext->code = pHead->code;
rpcProcessConnError(pContext, NULL); rpcProcessConnError(pContext, NULL);
rpcFreeCont(rpcMsg.pCont); rpcFreeCont(rpcMsg.pCont);
...@@ -1295,7 +1314,7 @@ static void rpcSendQuickRsp(SRpcConn *pConn, int32_t code) { ...@@ -1295,7 +1314,7 @@ static void rpcSendQuickRsp(SRpcConn *pConn, int32_t code) {
memset(msg, 0, sizeof(SRpcHead)); memset(msg, 0, sizeof(SRpcHead));
pHead = (SRpcHead *)msg; pHead = (SRpcHead *)msg;
pHead->version = 1; pHead->version = 1;
pHead->msgType = pConn->inType+1; pHead->msgType = pConn->inType + 1;
pHead->spi = pConn->spi; pHead->spi = pConn->spi;
pHead->encrypt = 0; pHead->encrypt = 0;
pHead->tranId = pConn->inTranId; pHead->tranId = pConn->inTranId;
...@@ -1335,7 +1354,7 @@ static void rpcSendReqHead(SRpcConn *pConn) { ...@@ -1335,7 +1354,7 @@ static void rpcSendReqHead(SRpcConn *pConn) {
static void rpcSendErrorMsgToPeer(SRecvInfo *pRecv, int32_t code) { static void rpcSendErrorMsgToPeer(SRecvInfo *pRecv, int32_t code) {
SRpcHead *pRecvHead, *pReplyHead; SRpcHead *pRecvHead, *pReplyHead;
char msg[sizeof(SRpcHead) + sizeof(SRpcDigest) + sizeof(uint32_t) ]; char msg[sizeof(SRpcHead) + sizeof(SRpcDigest) + sizeof(uint32_t)];
uint32_t timeStamp; uint32_t timeStamp;
int msgLen; int msgLen;
...@@ -1395,7 +1414,7 @@ static TBOOL rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) { ...@@ -1395,7 +1414,7 @@ static TBOOL rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) {
pHead->msgType = msgType; pHead->msgType = msgType;
pHead->encrypt = 0; pHead->encrypt = 0;
pConn->tranId++; pConn->tranId++;
if ( pConn->tranId == 0 ) pConn->tranId++; if (pConn->tranId == 0) pConn->tranId++;
pHead->tranId = pConn->tranId; pHead->tranId = pConn->tranId;
pHead->sourceId = pConn->ownId; pHead->sourceId = pConn->ownId;
pHead->destId = pConn->peerId; pHead->destId = pConn->peerId;
...@@ -1410,8 +1429,7 @@ static TBOOL rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) { ...@@ -1410,8 +1429,7 @@ static TBOOL rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) {
pConn->pReqMsg = msg; pConn->pReqMsg = msg;
pConn->reqMsgLen = msgLen; pConn->reqMsgLen = msgLen;
pConn->pContext = pContext; pConn->pContext = pContext;
if(pContext) if (pContext) pConn->rid = pContext->rid;
pConn->rid = pContext->rid;
// save // save
pContext->sendInfo.pConn = pConn; pContext->sendInfo.pConn = pConn;
...@@ -1424,7 +1442,7 @@ static TBOOL rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) { ...@@ -1424,7 +1442,7 @@ static TBOOL rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) {
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
if(ret == BOOL_FALSE) { if (ret == BOOL_FALSE) {
// try next ip again // try next ip again
pContext->code = terrno; pContext->code = terrno;
// in rpcProcessConnError if numOfTry over limit, could call rpcNotifyClient to stop query // in rpcProcessConnError if numOfTry over limit, could call rpcNotifyClient to stop query
...@@ -1442,18 +1460,16 @@ static bool rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) { ...@@ -1442,18 +1460,16 @@ static bool rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) {
msgLen = rpcAddAuthPart(pConn, msg, msgLen); msgLen = rpcAddAuthPart(pConn, msg, msgLen);
if ( rpcIsReq(pHead->msgType)) { if (rpcIsReq(pHead->msgType)) {
tDebug("%s, %s is sent to %s:%hu, len:%d sig:0x%08x:0x%08x:%d", tDebug("%s, %s is sent to %s:%hu, len:%d sig:0x%08x:0x%08x:%d", pConn->info, taosMsg[pHead->msgType],
pConn->info, taosMsg[pHead->msgType], pConn->peerFqdn, pConn->peerPort, pConn->peerFqdn, pConn->peerPort, msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
} else { } else {
if (pHead->code == 0) pConn->secured = 1; // for success response, set link as secured if (pHead->code == 0) pConn->secured = 1; // for success response, set link as secured
tDebug("%s, %s is sent to 0x%x:%hu, code:0x%x len:%d sig:0x%08x:0x%08x:%d", tDebug("%s, %s is sent to 0x%x:%hu, code:0x%x len:%d sig:0x%08x:0x%08x:%d", pConn->info, taosMsg[pHead->msgType],
pConn->info, taosMsg[pHead->msgType], pConn->peerIp, pConn->peerPort, pConn->peerIp, pConn->peerPort, htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
} }
//tTrace("connection type is: %d", pConn->connType); // tTrace("connection type is: %d", pConn->connType);
writtenLen = (*taosSendData[pConn->connType])(pConn->peerIp, pConn->peerPort, pHead, msgLen, pConn->chandle); writtenLen = (*taosSendData[pConn->connType])(pConn->peerIp, pConn->peerPort, pHead, msgLen, pConn->chandle);
if (writtenLen != msgLen) { if (writtenLen != msgLen) {
...@@ -1475,17 +1491,19 @@ static void rpcProcessConnError(void *param, void *id) { ...@@ -1475,17 +1491,19 @@ static void rpcProcessConnError(void *param, void *id) {
} }
if (pContext->numOfTry >= pContext->epSet.numOfEps || pContext->msgType == TSDB_MSG_TYPE_FETCH) { if (pContext->numOfTry >= pContext->epSet.numOfEps || pContext->msgType == TSDB_MSG_TYPE_FETCH) {
rpcMsg.msgType = pContext->msgType+1; rpcMsg.msgType = pContext->msgType + 1;
rpcMsg.ahandle = pContext->ahandle; rpcMsg.ahandle = pContext->ahandle;
rpcMsg.code = pContext->code; rpcMsg.code = pContext->code;
rpcMsg.pCont = NULL; rpcMsg.pCont = NULL;
rpcMsg.contLen = 0; rpcMsg.contLen = 0;
tWarn("%s %p, connection error. notify client query over. numOfTry=%d msgType=%d", pRpc->label, pContext->ahandle, pContext->numOfTry, pContext->msgType); tWarn("%s %p, connection error. notify client query over. numOfTry=%d msgType=%d", pRpc->label, pContext->ahandle,
pContext->numOfTry, pContext->msgType);
rpcNotifyClient(pContext, &rpcMsg); rpcNotifyClient(pContext, &rpcMsg);
} else { } else {
// move to next IP // move to next IP
tWarn("%s %p, connection error. retry to send request again. numOfTry=%d msgType=%d", pRpc->label, pContext->ahandle, pContext->numOfTry, pContext->msgType); tWarn("%s %p, connection error. retry to send request again. numOfTry=%d msgType=%d", pRpc->label,
pContext->ahandle, pContext->numOfTry, pContext->msgType);
pContext->epSet.inUse++; pContext->epSet.inUse++;
pContext->epSet.inUse = pContext->epSet.inUse % pContext->epSet.numOfEps; pContext->epSet.inUse = pContext->epSet.inUse % pContext->epSet.numOfEps;
rpcSendReqToServer(pRpc, pContext); rpcSendReqToServer(pRpc, pContext);
...@@ -1509,7 +1527,8 @@ static void rpcProcessRetryTimer(void *param, void *tmrId) { ...@@ -1509,7 +1527,8 @@ static void rpcProcessRetryTimer(void *param, void *tmrId) {
pConn->pTimer = taosTmrStart(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl); pConn->pTimer = taosTmrStart(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl);
} else { } else {
// close the connection // close the connection
tDebug("%s, failed to send msg:%s to %s:%hu", pConn->info, taosMsg[pConn->outType], pConn->peerFqdn, pConn->peerPort); tDebug("%s, failed to send msg:%s to %s:%hu", pConn->info, taosMsg[pConn->outType], pConn->peerFqdn,
pConn->peerPort);
if (pConn->pContext) { if (pConn->pContext) {
pConn->pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL; pConn->pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
pConn->pContext->pConn = NULL; pConn->pContext->pConn = NULL;
...@@ -1558,7 +1577,7 @@ static void rpcProcessProgressTimer(void *param, void *tmrId) { ...@@ -1558,7 +1577,7 @@ static void rpcProcessProgressTimer(void *param, void *tmrId) {
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
} }
static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) { static int32_t rpcCompressRpcMsg(char *pCont, int32_t contLen) {
SRpcHead *pHead = rpcHeadFromCont(pCont); SRpcHead *pHead = rpcHeadFromCont(pCont);
int32_t finalLen = 0; int32_t finalLen = 0;
int overhead = sizeof(SRpcComp); int overhead = sizeof(SRpcComp);
...@@ -1567,7 +1586,7 @@ static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) { ...@@ -1567,7 +1586,7 @@ static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
return contLen; return contLen;
} }
char *buf = malloc (contLen + overhead + 8); // 8 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", contLen); tError("failed to allocate memory for rpc msg compression, contLen:%d", contLen);
return contLen; return contLen;
...@@ -1614,7 +1633,7 @@ static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead) { ...@@ -1614,7 +1633,7 @@ static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead) {
if (pNewHead) { if (pNewHead) {
int compLen = rpcContLenFromMsg(pHead->msgLen) - overhead; int compLen = rpcContLenFromMsg(pHead->msgLen) - overhead;
int origLen = LZ4_decompress_safe((char*)(pCont + overhead), (char *)pNewHead->content, compLen, contLen); int origLen = LZ4_decompress_safe((char *)(pCont + overhead), (char *)pNewHead->content, compLen, contLen);
assert(origLen == contLen); assert(origLen == contLen);
memcpy(pNewHead, pHead, sizeof(SRpcHead)); memcpy(pNewHead, pHead, sizeof(SRpcHead));
...@@ -1680,19 +1699,19 @@ static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen) { ...@@ -1680,19 +1699,19 @@ static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen) {
SRpcHead *pHead = (SRpcHead *)msg; SRpcHead *pHead = (SRpcHead *)msg;
int code = 0; int code = 0;
if ((pConn->secured && pHead->spi == 0) || (pHead->spi == 0 && pConn->spi == 0)){ if ((pConn->secured && pHead->spi == 0) || (pHead->spi == 0 && pConn->spi == 0)) {
// secured link, or no authentication // secured link, or no authentication
pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen); pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen);
// tTrace("%s, secured link, no auth is required", pConn->info); // tTrace("%s, secured link, no auth is required", pConn->info);
return 0; return 0;
} }
if ( !rpcIsReq(pHead->msgType) ) { if (!rpcIsReq(pHead->msgType)) {
// for response, if code is auth failure, it shall bypass the auth process // for response, if code is auth failure, it shall bypass the auth process
code = htonl(pHead->code); code = htonl(pHead->code);
if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE || if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE ||
code == TSDB_CODE_RPC_INVALID_VERSION || code == TSDB_CODE_RPC_INVALID_VERSION || code == TSDB_CODE_RPC_AUTH_REQUIRED ||
code == TSDB_CODE_RPC_AUTH_REQUIRED || code == TSDB_CODE_MND_INVALID_USER || code == TSDB_CODE_RPC_NOT_READY) { code == TSDB_CODE_MND_INVALID_USER || code == TSDB_CODE_RPC_NOT_READY) {
pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen); pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen);
// tTrace("%s, dont check authentication since code is:0x%x", pConn->info, code); // tTrace("%s, dont check authentication since code is:0x%x", pConn->info, code);
return 0; return 0;
...@@ -1711,12 +1730,12 @@ static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen) { ...@@ -1711,12 +1730,12 @@ static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen) {
tWarn("%s, time diff:%d is too big, msg discarded", pConn->info, delta); tWarn("%s, time diff:%d is too big, msg discarded", pConn->info, delta);
code = TSDB_CODE_RPC_INVALID_TIME_STAMP; code = TSDB_CODE_RPC_INVALID_TIME_STAMP;
} else { } else {
if (rpcAuthenticateMsg(pHead, msgLen-TSDB_AUTH_LEN, pDigest->auth, pConn->secret) < 0) { if (rpcAuthenticateMsg(pHead, msgLen - TSDB_AUTH_LEN, pDigest->auth, pConn->secret) < 0) {
tDebug("%s, authentication failed, msg discarded", pConn->info); tDebug("%s, authentication failed, msg discarded", pConn->info);
code = TSDB_CODE_RPC_AUTH_FAILURE; code = TSDB_CODE_RPC_AUTH_FAILURE;
} else { } else {
pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen) - sizeof(SRpcDigest); pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen) - sizeof(SRpcDigest);
if ( !rpcIsReq(pHead->msgType) ) pConn->secured = 1; // link is secured for client if (!rpcIsReq(pHead->msgType)) pConn->secured = 1; // link is secured for client
// tTrace("%s, message is authenticated", pConn->info); // tTrace("%s, message is authenticated", pConn->info);
} }
} }
...@@ -1745,13 +1764,9 @@ static void rpcUnlockConn(SRpcConn *pConn) { ...@@ -1745,13 +1764,9 @@ static void rpcUnlockConn(SRpcConn *pConn) {
} }
} }
static void rpcAddRef(SRpcInfo *pRpc) static void rpcAddRef(SRpcInfo *pRpc) { atomic_add_fetch_32(&pRpc->refCount, 1); }
{
atomic_add_fetch_32(&pRpc->refCount, 1);
}
static void rpcDecRef(SRpcInfo *pRpc) static void rpcDecRef(SRpcInfo *pRpc) {
{
if (atomic_sub_fetch_32(&pRpc->refCount, 1) == 0) { if (atomic_sub_fetch_32(&pRpc->refCount, 1) == 0) {
rpcCloseConnCache(pRpc->pCache); rpcCloseConnCache(pRpc->pCache);
taosHashCleanup(pRpc->hash); taosHashCleanup(pRpc->hash);
...@@ -1794,10 +1809,10 @@ bool doRpcSendProbe(SRpcConn *pConn) { ...@@ -1794,10 +1809,10 @@ bool doRpcSendProbe(SRpcConn *pConn) {
} }
// send server syn // send server syn
bool rpcSendProbe(int64_t rpcRid, void* pPrevContext, bool *pReqOver) { bool rpcSendProbe(int64_t rpcRid, void *pPrevContext, bool *pReqOver) {
// return false can kill query // return false can kill query
bool ret = false; bool ret = false;
if(rpcRid < 0) { if (rpcRid < 0) {
tError("PROBE rpcRid=0x%" PRIx64 " less than zero, invalid.", rpcRid); tError("PROBE rpcRid=0x%" PRIx64 " less than zero, invalid.", rpcRid);
return true; return true;
} }
...@@ -1810,26 +1825,27 @@ bool rpcSendProbe(int64_t rpcRid, void* pPrevContext, bool *pReqOver) { ...@@ -1810,26 +1825,27 @@ bool rpcSendProbe(int64_t rpcRid, void* pPrevContext, bool *pReqOver) {
} }
// context same // context same
if(pContext != pPrevContext) { if (pContext != pPrevContext) {
tError("PROBE rpcRid=0x%" PRIx64 " context diff. pContext=%p pPreContent=%p", rpcRid, pContext, pPrevContext); tError("PROBE rpcRid=0x%" PRIx64 " context diff. pContext=%p pPreContent=%p", rpcRid, pContext, pPrevContext);
goto _END; goto _END;
} }
// conn same // conn same
if(pContext->pConn == NULL) { if (pContext->pConn == NULL) {
tInfo("PROBE rpcRid=0x%" PRIx64 " connect obj is NULL. req is response and done.", rpcRid); tInfo("PROBE rpcRid=0x%" PRIx64 " connect obj is NULL. req is response and done.", rpcRid);
if(pReqOver) if (pReqOver) *pReqOver = true;
*pReqOver = true;
ret = true; ret = true;
goto _END; goto _END;
} else if (pContext->pConn != pContext->sendInfo.pConn) { } else if (pContext->pConn != pContext->sendInfo.pConn) {
tInfo("PROBE rpcRid=0x%" PRIx64 " connect obj diff. pContext->pConn=%p pPreConn=%p", rpcRid, pContext->pConn, pContext->sendInfo.pConn); tInfo("PROBE rpcRid=0x%" PRIx64 " connect obj diff. pContext->pConn=%p pPreConn=%p", rpcRid, pContext->pConn,
pContext->sendInfo.pConn);
goto _END; goto _END;
} }
// fdObj same // fdObj same
if (pContext->pConn->chandle != pContext->sendInfo.pFdObj) { if (pContext->pConn->chandle != pContext->sendInfo.pFdObj) {
tInfo("PROBE rpcRid=0x%" PRIx64 " connect fdObj diff. pContext->pConn->chandle=%p pPrevFdObj=%p", rpcRid, pContext->pConn->chandle, pContext->sendInfo.pFdObj); tInfo("PROBE rpcRid=0x%" PRIx64 " connect fdObj diff. pContext->pConn->chandle=%p pPrevFdObj=%p", rpcRid,
pContext->pConn->chandle, pContext->sendInfo.pFdObj);
goto _END; goto _END;
} }
...@@ -1853,8 +1869,8 @@ _END: ...@@ -1853,8 +1869,8 @@ _END:
} }
// after sql request send , save conn info // after sql request send , save conn info
bool rpcSaveSendInfo(int64_t rpcRid, void** ppContext) { bool rpcSaveSendInfo(int64_t rpcRid, void **ppContext) {
if(rpcRid < 0) { if (rpcRid < 0) {
tError("PROBE saveSendInfo rpcRid=0x%" PRIx64 " less than zero, invalid.", rpcRid); tError("PROBE saveSendInfo rpcRid=0x%" PRIx64 " less than zero, invalid.", rpcRid);
return false; return false;
} }
...@@ -1865,8 +1881,7 @@ bool rpcSaveSendInfo(int64_t rpcRid, void** ppContext) { ...@@ -1865,8 +1881,7 @@ bool rpcSaveSendInfo(int64_t rpcRid, void** ppContext) {
return false; return false;
} }
if (ppContext) if (ppContext) *ppContext = pContext;
*ppContext = pContext;
taosReleaseRef(tsRpcRefId, rpcRid); taosReleaseRef(tsRpcRefId, rpcRid);
return true; return true;
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册