Skip to content

TDengine
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
TDengine
TDengine
提交 84eeb2ab 编写于 作者: dengyihao's avatar dengyihao
浏览文件
操作

Merge branch 'rpc' into 3.0

上级 49de494e 305fa254
隐藏空白更改
内联 并排
Showing with 58 addition and 3156 deletion
include/libs/transport/rpcHead.h 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#ifndef TDENGINE_RPCHEAD_H
#define TDENGINE_RPCHEAD_H
#include <tdef.h>
#ifdef __cplusplus
extern "C" {
#endif
#define RPC_CONN_TCP 2
extern int tsRpcOverhead;
typedef struct {
void* msg;
int msgLen;
uint32_t ip;
uint16_t port;
int connType;
void* shandle;
void* thandle;
void* chandle;
} SRecvInfo;
#pragma pack(push, 1)
typedef struct {
char version : 4; // RPC version
char comp : 4; // compression algorithm, 0:no compression 1:lz4
char resflag : 2; // reserved bits
char spi : 3; // security parameter index
char encrypt : 3; // encrypt algorithm, 0: no encryption
uint16_t tranId; // transcation ID
uint32_t linkUid; // for unique connection ID assigned by client
uint64_t ahandle; // ahandle assigned by client
uint32_t sourceId; // source ID, an index for connection list
uint32_t destId; // destination ID, an index for connection list
uint32_t destIp; // destination IP address, for NAT scenario
char user[TSDB_UNI_LEN]; // user ID
uint16_t port; // for UDP only, port may be changed
char empty[1]; // reserved
uint16_t msgType; // message type
int32_t msgLen; // message length including the header iteslf
uint32_t msgVer;
int32_t code; // code in response message
uint8_t content[0]; // message body starts from here
} SRpcHead;
typedef struct {
int32_t reserved;
int32_t contLen;
} SRpcComp;
typedef struct {
uint32_t timeStamp;
uint8_t auth[TSDB_AUTH_LEN];
} SRpcDigest;
#pragma pack(pop)
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_RPCHEAD_H
include/libs/transport/transport.h
浏览文件 @ 84eeb2ab
......@@ -20,9 +20,8 @@
extern "C" {
#endif
#ifdef __cplusplus
}
#endif
#endif /*_TD_TRANSPORT_H_*/
\ No newline at end of file
#endif /*_TD_TRANSPORT_H_*/
source/libs/transport/inc/rpcLog.h 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#ifndef TDENGINE_RPC_LOG_H
#define TDENGINE_RPC_LOG_H
#ifdef __cplusplus
extern "C" {
#endif
#include "tlog.h"
#define tFatal(...) \
{ \
if (rpcDebugFlag & DEBUG_FATAL) { \
taosPrintLog("RPC FATAL ", DEBUG_FATAL, rpcDebugFlag, __VA_ARGS__); \
} \
}
#define tError(...) \
{ \
if (rpcDebugFlag & DEBUG_ERROR) { \
taosPrintLog("RPC ERROR ", DEBUG_ERROR, rpcDebugFlag, __VA_ARGS__); \
} \
}
#define tWarn(...) \
{ \
if (rpcDebugFlag & DEBUG_WARN) { \
taosPrintLog("RPC WARN ", DEBUG_WARN, rpcDebugFlag, __VA_ARGS__); \
} \
}
#define tInfo(...) \
{ \
if (rpcDebugFlag & DEBUG_INFO) { \
taosPrintLog("RPC ", DEBUG_INFO, rpcDebugFlag, __VA_ARGS__); \
} \
}
#define tDebug(...) \
{ \
if (rpcDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("RPC ", DEBUG_DEBUG, rpcDebugFlag, __VA_ARGS__); \
} \
}
#define tTrace(...) \
{ \
if (rpcDebugFlag & DEBUG_TRACE) { \
taosPrintLog("RPC ", DEBUG_TRACE, rpcDebugFlag, __VA_ARGS__); \
} \
}
#define tDump(x, y) \
{ \
if (rpcDebugFlag & DEBUG_DUMP) { \
taosDumpData((unsigned char *)x, y); \
} \
}
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_RPC_LOG_H
source/libs/transport/inc/rpcTcp.h 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#ifndef _rpc_tcp_header_
#define _rpc_tcp_header_
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle);
void taosStopTcpServer(void *param);
void taosCleanUpTcpServer(void *param);
void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *fp, void *shandle);
void taosStopTcpClient(void *chandle);
void taosCleanUpTcpClient(void *chandle);
void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port);
void taosCloseTcpConnection(void *chandle);
int taosSendTcpData(uint32_t ip, uint16_t port, void *data, int len, void *chandle);
#ifdef __cplusplus
}
#endif
#endif
source/libs/transport/inc/rpcUdp.h 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#ifndef _rpc_udp_header_
#define _rpc_udp_header_
#ifdef __cplusplus
extern "C" {
#endif
#include "taosdef.h"
void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int, void *fp, void *shandle);
void taosStopUdpConnection(void *handle);
void taosCleanUpUdpConnection(void *handle);
int taosSendUdpData(uint32_t ip, uint16_t port, void *data, int dataLen, void *chandle);
void *taosOpenUdpConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port);
void taosFreeMsgHdr(void *hdr);
int taosMsgHdrSize(void *hdr);
void taosSendMsgHdr(void *hdr, TdFilePtr pFile);
void taosInitMsgHdr(void **hdr, void *dest, int maxPkts);
void taosSetMsgHdrData(void *hdr, char *data, int dataLen);
#ifdef __cplusplus
}
#endif
#endif
source/libs/transport/inc/transComm.h
浏览文件 @ 84eeb2ab
......@@ -12,7 +12,8 @@
* 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/>.
*/
#ifdef USE_UV
#ifndef _TD_TRANSPORT_COMM_H
#define _TD_TRANSPORT_COMM_H
#ifdef __cplusplus
extern "C" {
......@@ -22,9 +23,6 @@ extern "C" {
#include "lz4.h"
#include "os.h"
#include "osSocket.h"
#include "rpcCache.h"
#include "rpcHead.h"
#include "rpcLog.h"
#include "taoserror.h"
#include "tglobal.h"
#include "thash.h"
......@@ -33,6 +31,7 @@ extern "C" {
#include "tmd5.h"
#include "tmempool.h"
#include "tmsg.h"
#include "transLog.h"
#include "transportInt.h"
#include "tref.h"
#include "trpc.h"
......@@ -193,12 +192,7 @@ typedef enum { ConnNormal, ConnAcquire, ConnRelease, ConnBroken, ConnInPool } Co
#define container_of(ptr, type, member) ((type*)((char*)(ptr)-offsetof(type, member)))
#define RPC_RESERVE_SIZE (sizeof(STranConnCtx))
#define RPC_MSG_OVERHEAD (sizeof(SRpcHead) + sizeof(SRpcDigest))
#define rpcHeadFromCont(cont) ((SRpcHead*)((char*)cont - sizeof(SRpcHead)))
#define rpcContFromHead(msg) (msg + sizeof(SRpcHead))
#define rpcMsgLenFromCont(contLen) (contLen + sizeof(SRpcHead))
#define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHead))
#define rpcIsReq(type) (type & 1U)
#define rpcIsReq(type) (type & 1U)
#define TRANS_RESERVE_SIZE (sizeof(STranConnCtx))
......@@ -209,15 +203,14 @@ typedef enum { ConnNormal, ConnAcquire, ConnRelease, ConnBroken, ConnInPool } Co
#define transContLenFromMsg(msgLen) (msgLen - sizeof(STransMsgHead));
#define transIsReq(type) (type & 1U)
int rpcAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey);
void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen);
SRpcHead* rpcDecompressRpcMsg(SRpcHead* pHead);
int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey);
void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
bool transCompressMsg(char* msg, int32_t len, int32_t* flen);
bool transDecompressMsg(char* msg, int32_t len, int32_t* flen);
// int rpcAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey);
// void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
//// int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen);
//
// int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey);
// void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
// bool transCompressMsg(char* msg, int32_t len, int32_t* flen);
// bool transDecompressMsg(char* msg, int32_t len, int32_t* flen);
void transFreeMsg(void* msg);
......@@ -365,4 +358,4 @@ void transThreadOnce();
}
#endif
#endif
#endif // _TD_TRANSPORT_COMM_H
source/libs/transport/inc/rpcCache.h → source/libs/transport/inc/transLog.h
浏览文件 @ 84eeb2ab
......@@ -13,22 +13,26 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_RPC_CACHE_H
#define TDENGINE_RPC_CACHE_H
#include <stdint.h>
#ifndef _TD_TRANSPORT_LOG_H
#define _TD_TRANSPORT_LOG_H
#ifdef __cplusplus
extern "C" {
#endif
void *rpcOpenConnCache(int maxSessions, void (*cleanFp)(void *), void *tmrCtrl, int64_t keepTimer);
void rpcCloseConnCache(void *handle);
void rpcAddConnIntoCache(void *handle, void *data, char *fqdn, uint16_t port, int8_t connType);
void *rpcGetConnFromCache(void *handle, char *fqdn, uint16_t port, int8_t connType);
// clang-format off
#include "tlog.h"
#define tFatal(...) do {if (rpcDebugFlag & DEBUG_FATAL){ taosPrintLog("RPC FATAL ", DEBUG_FATAL, rpcDebugFlag, __VA_ARGS__); }} while (0)
#define tError(...)do { if (rpcDebugFlag & DEBUG_ERROR){ taosPrintLog("RPC ERROR ", DEBUG_ERROR, rpcDebugFlag, __VA_ARGS__); } } while(0)
#define tWarn(...) do { if (rpcDebugFlag & DEBUG_WARN) { taosPrintLog("RPC WARN ", DEBUG_WARN, rpcDebugFlag, __VA_ARGS__); }} while(0)
#define tInfo(...) do { if (rpcDebugFlag & DEBUG_INFO) { taosPrintLog("RPC ", DEBUG_INFO, rpcDebugFlag, __VA_ARGS__); }} while(0)
#define tDebug(...) do {if (rpcDebugFlag & DEBUG_DEBUG){ taosPrintLog("RPC ", DEBUG_DEBUG, rpcDebugFlag, __VA_ARGS__); }} while(0)
#define tTrace(...) do {if (rpcDebugFlag & DEBUG_TRACE){ taosPrintLog("RPC ", DEBUG_TRACE, rpcDebugFlag, __VA_ARGS__); }} while(0)
#define tDump(x, y) do {if (rpcDebugFlag & DEBUG_DUMP) { taosDumpData((unsigned char *)x, y); } } while(0)
// clang-format on
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_RPC_CACHE_H
#endif // __TRANS_LOG_H
source/libs/transport/inc/transportInt.h
浏览文件 @ 84eeb2ab
......@@ -21,14 +21,12 @@
#endif
#include "lz4.h"
#include "os.h"
#include "rpcCache.h"
#include "rpcHead.h"
#include "rpcLog.h"
#include "taoserror.h"
#include "tglobal.h"
#include "thash.h"
#include "tidpool.h"
#include "tmsg.h"
#include "transLog.h"
#include "tref.h"
#include "trpc.h"
#include "ttimer.h"
......
source/libs/transport/src/rpcCache.c 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#include "rpcCache.h"
#include "os.h"
#include "rpcLog.h"
#include "taosdef.h"
#include "tglobal.h"
#include "tmempool.h"
#include "ttimer.h"
#include "tutil.h"
typedef struct SConnHash {
char fqdn[TSDB_FQDN_LEN];
uint16_t port;
char connType;
struct SConnHash *prev;
struct SConnHash *next;
void * data;
uint64_t time;
} SConnHash;
typedef struct {
SConnHash ** connHashList;
mpool_h connHashMemPool;
int maxSessions;
int total;
int * count;
int64_t keepTimer;
TdThreadMutex mutex;
void (*cleanFp)(void *);
void * tmrCtrl;
void * pTimer;
int64_t *lockedBy;
} SConnCache;
static int rpcHashConn(void *handle, char *fqdn, uint16_t port, int8_t connType);
static void rpcLockCache(int64_t *lockedBy);
static void rpcUnlockCache(int64_t *lockedBy);
static void rpcCleanConnCache(void *handle, void *tmrId);
static void rpcRemoveExpiredNodes(SConnCache *pCache, SConnHash *pNode, int hash, uint64_t time);
void *rpcOpenConnCache(int maxSessions, void (*cleanFp)(void *), void *tmrCtrl, int64_t keepTimer) {
SConnHash **connHashList;
mpool_h connHashMemPool;
SConnCache *pCache;
connHashMemPool = taosMemPoolInit(maxSessions, sizeof(SConnHash));
if (connHashMemPool == 0) return NULL;
connHashList = taosMemoryCalloc(sizeof(SConnHash *), maxSessions);
if (connHashList == 0) {
taosMemPoolCleanUp(connHashMemPool);
return NULL;
}
pCache = taosMemoryMalloc(sizeof(SConnCache));
if (pCache == NULL) {
taosMemPoolCleanUp(connHashMemPool);
taosMemoryFree(connHashList);
return NULL;
}
memset(pCache, 0, sizeof(SConnCache));
pCache->count = taosMemoryCalloc(sizeof(int), maxSessions);
pCache->total = 0;
pCache->keepTimer = keepTimer;
pCache->maxSessions = maxSessions;
pCache->connHashMemPool = connHashMemPool;
pCache->connHashList = connHashList;
pCache->cleanFp = cleanFp;
pCache->tmrCtrl = tmrCtrl;
pCache->lockedBy = taosMemoryCalloc(sizeof(int64_t), maxSessions);
taosTmrReset(rpcCleanConnCache, (int32_t)(pCache->keepTimer * 2), pCache, pCache->tmrCtrl, &pCache->pTimer);
taosThreadMutexInit(&pCache->mutex, NULL);
return pCache;
}
void rpcCloseConnCache(void *handle) {
SConnCache *pCache;
pCache = (SConnCache *)handle;
if (pCache == NULL || pCache->maxSessions == 0) return;
taosThreadMutexLock(&pCache->mutex);
taosTmrStopA(&(pCache->pTimer));
if (pCache->connHashMemPool) taosMemPoolCleanUp(pCache->connHashMemPool);
taosMemoryFreeClear(pCache->connHashList);
taosMemoryFreeClear(pCache->count);
taosMemoryFreeClear(pCache->lockedBy);
taosThreadMutexUnlock(&pCache->mutex);
taosThreadMutexDestroy(&pCache->mutex);
memset(pCache, 0, sizeof(SConnCache));
taosMemoryFree(pCache);
}
void rpcAddConnIntoCache(void *handle, void *data, char *fqdn, uint16_t port, int8_t connType) {
int hash;
SConnHash * pNode;
SConnCache *pCache;
uint64_t time = taosGetTimestampMs();
pCache = (SConnCache *)handle;
assert(pCache);
assert(data);
hash = rpcHashConn(pCache, fqdn, port, connType);
pNode = (SConnHash *)taosMemPoolMalloc(pCache->connHashMemPool);
tstrncpy(pNode->fqdn, fqdn, sizeof(pNode->fqdn));
pNode->port = port;
pNode->connType = connType;
pNode->data = data;
pNode->prev = NULL;
pNode->time = time;
rpcLockCache(pCache->lockedBy + hash);
pNode->next = pCache->connHashList[hash];
if (pCache->connHashList[hash] != NULL) (pCache->connHashList[hash])->prev = pNode;
pCache->connHashList[hash] = pNode;
pCache->count[hash]++;
rpcRemoveExpiredNodes(pCache, pNode->next, hash, time);
rpcUnlockCache(pCache->lockedBy + hash);
pCache->total++;
// tTrace("%p %s:%hu:%d:%d:%p added into cache, connections:%d", data, fqdn, port, connType, hash, pNode,
// pCache->count[hash]);
return;
}
void *rpcGetConnFromCache(void *handle, char *fqdn, uint16_t port, int8_t connType) {
int hash;
SConnHash * pNode;
SConnCache *pCache;
void * pData = NULL;
pCache = (SConnCache *)handle;
assert(pCache);
uint64_t time = taosGetTimestampMs();
hash = rpcHashConn(pCache, fqdn, port, connType);
rpcLockCache(pCache->lockedBy + hash);
pNode = pCache->connHashList[hash];
while (pNode) {
if (time >= pCache->keepTimer + pNode->time) {
rpcRemoveExpiredNodes(pCache, pNode, hash, time);
pNode = NULL;
break;
}
if (strcmp(pNode->fqdn, fqdn) == 0 && pNode->port == port && pNode->connType == connType) break;
pNode = pNode->next;
}
if (pNode) {
rpcRemoveExpiredNodes(pCache, pNode->next, hash, time);
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pCache->connHashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
pData = pNode->data;
taosMemPoolFree(pCache->connHashMemPool, (char *)pNode);
pCache->total--;
pCache->count[hash]--;
}
rpcUnlockCache(pCache->lockedBy + hash);
if (pData) {
// tTrace("%p %s:%hu:%d:%d:%p retrieved from cache, connections:%d", pData, fqdn, port, connType, hash, pNode,
// pCache->count[hash]);
} else {
// tTrace("%s:%hu:%d:%d failed to retrieve conn from cache, connections:%d", fqdn, port, connType, hash,
// pCache->count[hash]);
}
return pData;
}
static void rpcCleanConnCache(void *handle, void *tmrId) {
int hash;
SConnHash * pNode;
SConnCache *pCache;
pCache = (SConnCache *)handle;
if (pCache == NULL || pCache->maxSessions == 0) return;
if (pCache->pTimer != tmrId) return;
taosThreadMutexLock(&pCache->mutex);
uint64_t time = taosGetTimestampMs();
for (hash = 0; hash < pCache->maxSessions; ++hash) {
rpcLockCache(pCache->lockedBy + hash);
pNode = pCache->connHashList[hash];
rpcRemoveExpiredNodes(pCache, pNode, hash, time);
rpcUnlockCache(pCache->lockedBy + hash);
}
// tTrace("timer, total connections in cache:%d", pCache->total);
taosTmrReset(rpcCleanConnCache, (int32_t)(pCache->keepTimer * 2), pCache, pCache->tmrCtrl, &pCache->pTimer);
taosThreadMutexUnlock(&pCache->mutex);
}
static void rpcRemoveExpiredNodes(SConnCache *pCache, SConnHash *pNode, int hash, uint64_t time) {
if (pNode == NULL || (time < pCache->keepTimer + pNode->time)) return;
SConnHash *pPrev = pNode->prev, *pNext;
while (pNode) {
(*pCache->cleanFp)(pNode->data);
pNext = pNode->next;
pCache->total--;
pCache->count[hash]--;
// tTrace("%p %s:%hu:%d:%d:%p removed from cache, connections:%d", pNode->data, pNode->fqdn, pNode->port,
// pNode->connType, hash, pNode,
// pCache->count[hash]);
taosMemPoolFree(pCache->connHashMemPool, (char *)pNode);
pNode = pNext;
}
if (pPrev)
pPrev->next = NULL;
else
pCache->connHashList[hash] = NULL;
}
static int rpcHashConn(void *handle, char *fqdn, uint16_t port, int8_t connType) {
SConnCache *pCache = (SConnCache *)handle;
int hash = 0;
char * temp = fqdn;
while (*temp) {
hash += *temp;
++temp;
}
hash += port;
hash += connType;
hash = hash % pCache->maxSessions;
return hash;
}
static void rpcLockCache(int64_t *lockedBy) {
int64_t tid = taosGetSelfPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(lockedBy, 0, tid) != 0) {
if (++i % 100 == 0) {
sched_yield();
}
}
}
static void rpcUnlockCache(int64_t *lockedBy) {
int64_t tid = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(lockedBy, tid, 0) != tid) {
assert(false);
}
}
source/libs/transport/src/rpcMain.c 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#include "lz4.h"
#include "transportInt.h"
#include "os.h"
#include "rpcCache.h"
#include "rpcHead.h"
#include "rpcLog.h"
#include "rpcTcp.h"
#include "rpcUdp.h"
#include "taoserror.h"
#include "tglobal.h"
#include "thash.h"
#include "tidpool.h"
#include "tmd5.h"
#include "tmempool.h"
#include "tmsg.h"
#include "tref.h"
#include "trpc.h"
#include "ttimer.h"
#include "tutil.h"
static TdThreadOnce tsRpcInitOnce = PTHREAD_ONCE_INIT;
int tsRpcMaxUdpSize = 15000; // bytes
int tsProgressTimer = 100;
// not configurable
int tsRpcMaxRetry;
int tsRpcHeadSize;
int tsRpcOverhead;
SHashObj *tsFqdnHash;
#ifndef USE_UV
typedef struct {
int sessions; // number of sessions allowed
int numOfThreads; // number of threads to process incoming messages
int idleTime; // milliseconds;
uint16_t localPort;
int8_t connType;
int index; // for UDP server only, round robin for multiple threads
char label[TSDB_LABEL_LEN];
char user[TSDB_UNI_LEN]; // meter ID
char spi; // security parameter index
char encrypt; // encrypt algorithm
char secret[TSDB_PASSWORD_LEN]; // secret for the link
char ckey[TSDB_PASSWORD_LEN]; // ciphering key
void (*cfp)(void *parent, SRpcMsg *, SEpSet *);
int (*afp)(void *parent, char *user, char *spi, char *encrypt, char *secret, char *ckey);
int32_t refCount;
void * parent;
void * idPool; // handle to ID pool
void * tmrCtrl; // handle to timer
SHashObj * hash; // handle returned by hash utility
void * tcphandle; // returned handle from TCP initialization
void * udphandle; // returned handle from UDP initialization
void * pCache; // connection cache
TdThreadMutex mutex;
struct SRpcConn *connList; // connection list
} SRpcInfo;
typedef struct {
SRpcInfo * pRpc; // associated SRpcInfo
SEpSet epSet; // ip list provided by app
void * ahandle; // handle provided by app
struct SRpcConn *pConn; // pConn allocated
tmsg_t msgType; // message type
uint8_t * pCont; // content provided by app
int32_t contLen; // content length
int32_t code; // error code
int16_t numOfTry; // number of try for different servers
int8_t oldInUse; // server EP inUse passed by app
int8_t redirect; // flag to indicate redirect
int8_t connType; // connection type
int64_t rid; // refId returned by taosAddRef
SRpcMsg * pRsp; // for synchronous API
tsem_t * pSem; // for synchronous API
SEpSet * pSet; // for synchronous API
char msg[0]; // RpcHead starts from here
} SRpcReqContext;
#define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHead) + sizeof(SRpcDigest))
#define rpcHeadFromCont(cont) ((SRpcHead *)((char *)cont - sizeof(SRpcHead)))
#define rpcContFromHead(msg) (msg + sizeof(SRpcHead))
#define rpcMsgLenFromCont(contLen) (contLen + sizeof(SRpcHead))
#define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHead))
#define rpcIsReq(type) (type & 1U)
typedef struct SRpcConn {
char info[48]; // debug info: label + pConn + ahandle
int sid; // session ID
uint32_t ownId; // own link ID
uint32_t peerId; // peer link ID
char user[TSDB_UNI_LEN]; // user ID for the link
char spi; // security parameter index
char encrypt; // encryption, 0:1
char secret[TSDB_PASSWORD_LEN]; // secret for the link
char ckey[TSDB_PASSWORD_LEN]; // ciphering key
char secured; // if set to 1, no authentication
uint16_t localPort; // for UDP only
uint32_t linkUid; // connection unique ID assigned by client
uint32_t peerIp; // peer IP
uint16_t peerPort; // peer port
char peerFqdn[TSDB_FQDN_LEN]; // peer FQDN or ip string
uint16_t tranId; // outgoing transcation ID, for build message
uint16_t outTranId; // outgoing transcation ID
uint16_t inTranId; // transcation ID for incoming msg
tmsg_t outType; // message type for outgoing request
tmsg_t inType; // message type for incoming request
void * chandle; // handle passed by TCP/UDP connection layer
void * ahandle; // handle provided by upper app layter
int retry; // number of retry for sending request
int tretry; // total retry
void * pTimer; // retry timer to monitor the response
void * pIdleTimer; // idle timer
char * pRspMsg; // response message including header
int rspMsgLen; // response messag length
char * pReqMsg; // request message including header
int reqMsgLen; // request message length
SRpcInfo * pRpc; // the associated SRpcInfo
int8_t connType; // connection type
int64_t lockedBy; // lock for connection
SRpcReqContext *pContext; // request context
} SRpcConn;
static int tsRpcRefId = -1;
static int32_t tsRpcNum = 0;
// static TdThreadOnce tsRpcInit = PTHREAD_ONCE_INIT;
// server:0 client:1 tcp:2 udp:0
#define RPC_CONN_UDPS 0
#define RPC_CONN_UDPC 1
#define RPC_CONN_TCPS 2
#define RPC_CONN_TCPC 3
void *(*taosInitConn[])(uint32_t ip, uint16_t port, char *label, int threads, void *fp, void *shandle) = {
taosInitUdpConnection, taosInitUdpConnection, taosInitTcpServer, taosInitTcpClient};
void (*taosCleanUpConn[])(void *thandle) = {taosCleanUpUdpConnection, taosCleanUpUdpConnection, taosCleanUpTcpServer,
taosCleanUpTcpClient};
void (*taosStopConn[])(void *thandle) = {
taosStopUdpConnection,
taosStopUdpConnection,
taosStopTcpServer,
taosStopTcpClient,
};
int (*taosSendData[])(uint32_t ip, uint16_t port, void *data, int len, void *chandle) = {
taosSendUdpData, taosSendUdpData, taosSendTcpData, taosSendTcpData};
void *(*taosOpenConn[])(void *shandle, void *thandle, uint32_t ip, uint16_t port) = {
taosOpenUdpConnection,
taosOpenUdpConnection,
NULL,
taosOpenTcpClientConnection,
};
void (*taosCloseConn[])(void *chandle) = {NULL, NULL, taosCloseTcpConnection, taosCloseTcpConnection};
static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerFqdn, uint16_t peerPort, int8_t connType);
static void rpcCloseConn(void *thandle);
static SRpcConn *rpcSetupConnToServer(SRpcReqContext *pContext);
static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc);
static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv);
static SRpcConn *rpcGetConnObj(SRpcInfo *pRpc, int sid, SRecvInfo *pRecv);
static void rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext);
static void rpcSendQuickRsp(SRpcConn *pConn, int32_t code);
static void rpcSendErrorMsgToPeer(SRecvInfo *pRecv, int32_t code);
static void rpcSendMsgToPeer(SRpcConn *pConn, void *data, int dataLen);
static void rpcSendReqHead(SRpcConn *pConn);
static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv);
static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqContext *pContext);
static void rpcProcessConnError(void *param, void *id);
static void rpcProcessRetryTimer(void *, void *);
static void rpcProcessIdleTimer(void *param, void *tmrId);
static void rpcProcessProgressTimer(void *param, void *tmrId);
static void rpcFreeMsg(void *msg);
static int32_t rpcCompressRpcMsg(char *pCont, int32_t contLen);
static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead);
static int rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen);
static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen);
static void rpcLockConn(SRpcConn *pConn);
static void rpcUnlockConn(SRpcConn *pConn);
static void rpcAddRef(SRpcInfo *pRpc);
static void rpcDecRef(SRpcInfo *pRpc);
static void rpcFree(void *p) {
tTrace("free mem: %p", p);
taosMemoryFree(p);
}
static void rpcInitImp(void) {
tsProgressTimer = tsRpcTimer / 2;
tsRpcMaxRetry = tsRpcMaxTime * 1000 / tsProgressTimer;
tsRpcHeadSize = RPC_MSG_OVERHEAD;
tsRpcOverhead = sizeof(SRpcReqContext);
tsRpcRefId = taosOpenRef(200, rpcFree);
tsFqdnHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
}
int32_t rpcInit() {
taosThreadOnce(&tsRpcInitOnce, rpcInitImp);
return 0;
}
void rpcCleanup(void) {
taosCloseRef(tsRpcRefId);
taosHashClear(tsFqdnHash);
taosHashCleanup(tsFqdnHash);
tsFqdnHash = NULL;
tsRpcRefId = -1;
}
void *rpcOpen(const SRpcInit *pInit) {
SRpcInfo *pRpc;
// taosThreadOnce(&tsRpcInit, rpcInit);
pRpc = (SRpcInfo *)taosMemoryCalloc(1, sizeof(SRpcInfo));
if (pRpc == NULL) return NULL;
if (pInit->label) tstrncpy(pRpc->label, pInit->label, tListLen(pInit->label));
pRpc->connType = pInit->connType;
if (pRpc->connType == TAOS_CONN_CLIENT) {
pRpc->numOfThreads = pInit->numOfThreads;
if (pRpc->numOfThreads >= 10) {
pRpc->numOfThreads = 10;
}
} else {
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
}
pRpc->idleTime = pInit->idleTime;
pRpc->localPort = pInit->localPort;
pRpc->afp = pInit->afp;
pRpc->sessions = pInit->sessions + 1;
if (pInit->user) tstrncpy(pRpc->user, pInit->user, sizeof(pRpc->user));
if (pInit->secret) memcpy(pRpc->secret, pInit->secret, sizeof(pRpc->secret));
if (pInit->ckey) tstrncpy(pRpc->ckey, pInit->ckey, sizeof(pRpc->ckey));
pRpc->spi = pInit->spi;
pRpc->cfp = pInit->cfp;
pRpc->afp = pInit->afp;
pRpc->parent = pInit->parent;
pRpc->refCount = 1;
atomic_add_fetch_32(&tsRpcNum, 1);
size_t size = sizeof(SRpcConn) * pRpc->sessions;
pRpc->connList = (SRpcConn *)taosMemoryCalloc(1, size);
if (pRpc->connList == NULL) {
tError("%s failed to allocate memory for taos connections, size:%" PRId64, pRpc->label, (int64_t)size);
rpcClose(pRpc);
return NULL;
}
pRpc->idPool = taosInitIdPool(pRpc->sessions - 1);
if (pRpc->idPool == NULL) {
tError("%s failed to init ID pool", pRpc->label);
rpcClose(pRpc);
return NULL;
}
pRpc->tmrCtrl = taosTmrInit(pRpc->sessions * 2 + 1, 50, 10000, pRpc->label);
if (pRpc->tmrCtrl == NULL) {
tError("%s failed to init timers", pRpc->label);
rpcClose(pRpc);
return NULL;
}
if (pRpc->connType == TAOS_CONN_SERVER) {
pRpc->hash = taosHashInit(pRpc->sessions, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, true);
if (pRpc->hash == NULL) {
tError("%s failed to init string hash", pRpc->label);
rpcClose(pRpc);
return NULL;
}
} else {
pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime * 20);
if (pRpc->pCache == NULL) {
tError("%s failed to init connection cache", pRpc->label);
rpcClose(pRpc);
return NULL;
}
}
taosThreadMutexInit(&pRpc->mutex, NULL);
pRpc->tcphandle = (*taosInitConn[pRpc->connType | RPC_CONN_TCP])(0, pRpc->localPort, pRpc->label, pRpc->numOfThreads,
rpcProcessMsgFromPeer, pRpc);
pRpc->udphandle =
(*taosInitConn[pRpc->connType])(0, pRpc->localPort, pRpc->label, pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc);
if (pRpc->tcphandle == NULL || pRpc->udphandle == NULL) {
tError("%s failed to init network, port:%d", pRpc->label, pRpc->localPort);
rpcClose(pRpc);
return NULL;
}
tDebug("%s rpc is opened, threads:%d sessions:%d", pRpc->label, pRpc->numOfThreads, pInit->sessions);
return pRpc;
}
void rpcClose(void *param) {
SRpcInfo *pRpc = (SRpcInfo *)param;
// stop connection to outside first
(*taosStopConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
(*taosStopConn[pRpc->connType])(pRpc->udphandle);
// close all connections
for (int i = 0; i < pRpc->sessions; ++i) {
if (pRpc->connList && pRpc->connList[i].user[0]) {
rpcCloseConn((void *)(pRpc->connList + i));
}
}
// clean up
(*taosCleanUpConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
(*taosCleanUpConn[pRpc->connType])(pRpc->udphandle);
tDebug("%s rpc is closed", pRpc->label);
rpcDecRef(pRpc);
}
void *rpcMallocCont(int contLen) {
int size = contLen + RPC_MSG_OVERHEAD;
char *start = (char *)taosMemoryCalloc(1, (size_t)size);
if (start == NULL) {
tError("failed to malloc msg, size:%d", size);
return NULL;
} else {
tTrace("malloc mem:%p size:%d", start, size);
}
return start + sizeof(SRpcReqContext) + sizeof(SRpcHead);
}
void rpcFreeCont(void *cont) {
if (cont) {
char *temp = ((char *)cont) - sizeof(SRpcHead) - sizeof(SRpcReqContext);
taosMemoryFree(temp);
tTrace("free mem: %p", temp);
}
}
void *rpcReallocCont(void *ptr, int contLen) {
if (ptr == NULL) return rpcMallocCont(contLen);
char *start = ((char *)ptr) - sizeof(SRpcReqContext) - sizeof(SRpcHead);
if (contLen == 0) {
taosMemoryFree(start);
return NULL;
}
int size = contLen + RPC_MSG_OVERHEAD;
start = taosMemoryRealloc(start, size);
if (start == NULL) {
tError("failed to realloc cont, size:%d", size);
return NULL;
}
return start + sizeof(SRpcReqContext) + sizeof(SRpcHead);
}
void rpcSendRequest(void *shandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *pRid) {
SRpcInfo * pRpc = (SRpcInfo *)shandle;
SRpcReqContext *pContext;
int contLen = rpcCompressRpcMsg(pMsg->pCont, pMsg->contLen);
pContext = (SRpcReqContext *)((char *)pMsg->pCont - sizeof(SRpcHead) - sizeof(SRpcReqContext));
pContext->ahandle = pMsg->ahandle;
pContext->pRpc = (SRpcInfo *)shandle;
pContext->epSet = *pEpSet;
pContext->contLen = contLen;
pContext->pCont = pMsg->pCont;
pContext->msgType = pMsg->msgType;
pContext->oldInUse = pEpSet->inUse;
pContext->connType = RPC_CONN_UDPC;
if (contLen > tsRpcMaxUdpSize || tsRpcForceTcp) pContext->connType = RPC_CONN_TCPC;
// connection type is application specific.
// for TDengine, all the query, show commands shall have TCP connection
tmsg_t type = pMsg->msgType;
if (type == TDMT_VND_QUERY || type == TDMT_MND_SHOW_RETRIEVE || type == TDMT_VND_FETCH ||
type == TDMT_MND_VGROUP_LIST || type == TDMT_VND_TABLES_META || type == TDMT_VND_TABLE_META ||
type == TDMT_MND_SHOW || type == TDMT_MND_STATUS || type == TDMT_VND_ALTER_TABLE)
pContext->connType = RPC_CONN_TCPC;
pContext->rid = taosAddRef(tsRpcRefId, pContext);
if (pRid) *pRid = pContext->rid;
rpcSendReqToServer(pRpc, pContext);
}
void rpcSendResponse(const SRpcMsg *pRsp) {
ASSERT(pRsp->handle != NULL);
int msgLen = 0;
SRpcConn *pConn = (SRpcConn *)pRsp->handle;
SRpcMsg rpcMsg = *pRsp;
SRpcMsg * pMsg = &rpcMsg;
SRpcInfo *pRpc = pConn->pRpc;
if (pMsg->pCont == NULL) {
pMsg->pCont = rpcMallocCont(0);
pMsg->contLen = 0;
}
SRpcHead *pHead = rpcHeadFromCont(pMsg->pCont);
char * msg = (char *)pHead;
pMsg->contLen = rpcCompressRpcMsg(pMsg->pCont, pMsg->contLen);
msgLen = rpcMsgLenFromCont(pMsg->contLen);
rpcLockConn(pConn);
if (pConn->inType == 0 || pConn->user[0] == 0) {
tError("%s, connection is already released, rsp wont be sent", pConn->info);
rpcUnlockConn(pConn);
rpcFreeCont(pMsg->pCont);
rpcDecRef(pRpc);
return;
}
// set msg header
pHead->version = 1;
pHead->msgType = pConn->inType + 1;
pHead->spi = pConn->spi;
pHead->encrypt = pConn->encrypt;
pHead->tranId = pConn->inTranId;
pHead->sourceId = pConn->ownId;
pHead->destId = pConn->peerId;
pHead->linkUid = pConn->linkUid;
pHead->port = htons(pConn->localPort);
pHead->code = htonl(pMsg->code);
pHead->ahandle = (uint64_t)pConn->ahandle;
// set pConn parameters
pConn->inType = 0;
// response message is released until new response is sent
rpcFreeMsg(pConn->pRspMsg);
pConn->pRspMsg = msg;
pConn->rspMsgLen = msgLen;
if (pMsg->code == TSDB_CODE_RPC_ACTION_IN_PROGRESS) pConn->inTranId--;
// stop the progress timer
taosTmrStopA(&pConn->pTimer);
// set the idle timer to monitor the activity
taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime * 30, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer);
rpcSendMsgToPeer(pConn, msg, msgLen);
// 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) pConn->secured = 1; // connection shall be secured
if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg);
pConn->pReqMsg = NULL;
pConn->reqMsgLen = 0;
rpcUnlockConn(pConn);
rpcDecRef(pRpc); // decrease the referene count
return;
}
void rpcSendRedirectRsp(void *thandle, const SEpSet *pEpSet) {
SRpcMsg rpcMsg;
memset(&rpcMsg, 0, sizeof(rpcMsg));
rpcMsg.contLen = sizeof(SEpSet);
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
if (rpcMsg.pCont == NULL) return;
memcpy(rpcMsg.pCont, pEpSet, sizeof(SEpSet));
rpcMsg.code = TSDB_CODE_RPC_REDIRECT;
rpcMsg.handle = thandle;
rpcSendResponse(&rpcMsg);
return;
}
int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
SRpcConn *pConn = (SRpcConn *)thandle;
if (pConn->user[0] == 0) return -1;
pInfo->clientIp = pConn->peerIp;
pInfo->clientPort = pConn->peerPort;
// pInfo->serverIp = pConn->destIp;
tstrncpy(pInfo->user, pConn->user, sizeof(pInfo->user));
return 0;
}
void rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
SRpcReqContext *pContext;
pContext = (SRpcReqContext *)((char *)pMsg->pCont - sizeof(SRpcHead) - sizeof(SRpcReqContext));
memset(pRsp, 0, sizeof(SRpcMsg));
tsem_t sem;
tsem_init(&sem, 0, 0);
pContext->pSem = &sem;
pContext->pRsp = pRsp;
pContext->pSet = pEpSet;
rpcSendRequest(shandle, pEpSet, pMsg, NULL);
tsem_wait(&sem);
tsem_destroy(&sem);
return;
}
// this API is used by server app to keep an APP context in case connection is broken
int rpcReportProgress(void *handle, char *pCont, int contLen) {
SRpcConn *pConn = (SRpcConn *)handle;
int code = 0;
rpcLockConn(pConn);
if (pConn->user[0]) {
// pReqMsg and reqMsgLen is re-used to store the context from app server
pConn->pReqMsg = pCont;
pConn->reqMsgLen = contLen;
} else {
tDebug("%s, rpc connection is already released", pConn->info);
rpcFreeCont(pCont);
code = -1;
}
rpcUnlockConn(pConn);
return code;
}
void rpcCancelRequest(int64_t rid) {
SRpcReqContext *pContext = taosAcquireRef(tsRpcRefId, rid);
if (pContext == NULL) return;
rpcCloseConn(pContext->pConn);
taosReleaseRef(tsRpcRefId, rid);
}
static void rpcFreeMsg(void *msg) {
if (msg) {
char *temp = (char *)msg - sizeof(SRpcReqContext);
taosMemoryFree(temp);
tTrace("free mem: %p", temp);
}
}
static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerFqdn, uint16_t peerPort, int8_t connType) {
SRpcConn *pConn;
uint32_t peerIp = 0;
uint32_t *pPeerIp = taosHashGet(tsFqdnHash, peerFqdn, strlen(peerFqdn) + 1);
if (pPeerIp != NULL) {
peerIp = *pPeerIp;
} else {
peerIp = taosGetIpv4FromFqdn(peerFqdn);
if (peerIp != 0xFFFFFFFF) {
taosHashPut(tsFqdnHash, peerFqdn, strlen(peerFqdn) + 1, &peerIp, sizeof(peerIp));
}
}
if (peerIp == 0xFFFFFFFF) {
tError("%s, failed to resolve FQDN:%s", pRpc->label, peerFqdn);
terrno = TSDB_CODE_RPC_FQDN_ERROR;
return NULL;
}
pConn = rpcAllocateClientConn(pRpc);
if (pConn) {
tstrncpy(pConn->peerFqdn, peerFqdn, sizeof(pConn->peerFqdn));
pConn->peerIp = peerIp;
pConn->peerPort = peerPort;
tstrncpy(pConn->user, pRpc->user, sizeof(pConn->user));
pConn->connType = connType;
if (taosOpenConn[connType]) {
void *shandle = (connType & RPC_CONN_TCP) ? pRpc->tcphandle : pRpc->udphandle;
pConn->chandle = (*taosOpenConn[connType])(shandle, pConn, pConn->peerIp, pConn->peerPort);
if (pConn->chandle == NULL) {
tError("failed to connect to:%s:%d", taosIpStr(pConn->peerIp), pConn->peerPort);
terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL;
rpcCloseConn(pConn);
pConn = NULL;
}
}
}
return pConn;
}
static void rpcReleaseConn(SRpcConn *pConn) {
SRpcInfo *pRpc = pConn->pRpc;
if (pConn->user[0] == 0) return;
pConn->user[0] = 0;
if (taosCloseConn[pConn->connType]) (*taosCloseConn[pConn->connType])(pConn->chandle);
taosTmrStopA(&pConn->pTimer);
taosTmrStopA(&pConn->pIdleTimer);
if (pRpc->connType == TAOS_CONN_SERVER) {
char hashstr[40] = {0};
size_t size = snprintf(hashstr, sizeof(hashstr), "%x:%x:%x:%d", pConn->peerIp, pConn->linkUid, pConn->peerId,
pConn->connType);
taosHashRemove(pRpc->hash, hashstr, size);
rpcFreeMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg
pConn->pRspMsg = NULL;
// if server has ever reported progress, free content
if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg); // do not use rpcFreeMsg
} else {
// if there is an outgoing message, free it
if (pConn->outType && pConn->pReqMsg) {
SRpcReqContext *pContext = pConn->pContext;
if (pContext) {
if (pContext->pRsp) {
// for synchronous API, post semaphore to unblock app
pContext->pRsp->code = TSDB_CODE_RPC_APP_ERROR;
pContext->pRsp->pCont = NULL;
pContext->pRsp->contLen = 0;
tsem_post(pContext->pSem);
}
pContext->pConn = NULL;
taosRemoveRef(tsRpcRefId, pContext->rid);
} else {
assert(0);
}
}
}
// memset could not be used, since lockeBy can not be reset
pConn->inType = 0;
pConn->outType = 0;
pConn->inTranId = 0;
pConn->outTranId = 0;
pConn->secured = 0;
pConn->peerId = 0;
pConn->peerIp = 0;
pConn->peerPort = 0;
pConn->pReqMsg = NULL;
pConn->reqMsgLen = 0;
pConn->pContext = NULL;
pConn->chandle = NULL;
taosFreeId(pRpc->idPool, pConn->sid);
tDebug("%s, rpc connection is released", pConn->info);
}
static void rpcCloseConn(void *thandle) {
SRpcConn *pConn = (SRpcConn *)thandle;
if (pConn == NULL) return;
rpcLockConn(pConn);
if (pConn->user[0]) rpcReleaseConn(pConn);
rpcUnlockConn(pConn);
}
static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc) {
SRpcConn *pConn = NULL;
int sid = taosAllocateId(pRpc->idPool);
if (sid <= 0) {
tError("%s maximum number of sessions:%d is reached", pRpc->label, pRpc->sessions);
terrno = TSDB_CODE_RPC_MAX_SESSIONS;
} else {
pConn = pRpc->connList + sid;
pConn->pRpc = pRpc;
pConn->sid = sid;
pConn->tranId = (uint16_t)(taosRand() & 0xFFFF);
pConn->ownId = htonl(pConn->sid);
pConn->linkUid = (uint32_t)((int64_t)pConn + taosGetPId() + (int64_t)pConn->tranId);
pConn->spi = pRpc->spi;
pConn->encrypt = pRpc->encrypt;
if (pConn->spi) memcpy(pConn->secret, pRpc->secret, TSDB_PASSWORD_LEN);
tDebug("%s %p client connection is allocated, uid:0x%x", pRpc->label, pConn, pConn->linkUid);
}
return pConn;
}
static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) {
SRpcConn *pConn = NULL;
char hashstr[40] = {0};
SRpcHead *pHead = (SRpcHead *)pRecv->msg;
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
SRpcConn **ppConn = (SRpcConn **)(taosHashGet(pRpc->hash, hashstr, size));
if (ppConn) pConn = *ppConn;
if (pConn) {
pConn->secured = 0;
return pConn;
}
// if code is not 0, it means it is simple reqhead, just ignore
if (pHead->code != 0) {
terrno = TSDB_CODE_RPC_ALREADY_PROCESSED;
return NULL;
}
int sid = taosAllocateId(pRpc->idPool);
if (sid <= 0) {
tError("%s maximum number of sessions:%d is reached", pRpc->label, pRpc->sessions);
terrno = TSDB_CODE_RPC_MAX_SESSIONS;
} else {
pConn = pRpc->connList + sid;
memcpy(pConn->user, pHead->user, tListLen(pConn->user));
pConn->pRpc = pRpc;
pConn->sid = sid;
pConn->tranId = (uint16_t)(taosRand() & 0xFFFF);
pConn->ownId = htonl(pConn->sid);
pConn->linkUid = pHead->linkUid;
if (pRpc->afp) {
if (pConn->user[0] == 0) {
terrno = TSDB_CODE_RPC_AUTH_REQUIRED;
} else {
terrno = (*pRpc->afp)(pRpc->parent, pConn->user, &pConn->spi, &pConn->encrypt, pConn->secret, pConn->ckey);
}
if (terrno != 0) {
taosFreeId(pRpc->idPool, sid); // sid shall be released
pConn = NULL;
}
}
}
if (pConn) {
if (pRecv->connType == RPC_CONN_UDPS && pRpc->numOfThreads > 1) {
// UDP server, assign to new connection
pRpc->index = (pRpc->index + 1) % pRpc->numOfThreads;
pConn->localPort = (pRpc->localPort + pRpc->index);
}
taosHashPut(pRpc->hash, hashstr, size, (char *)&pConn, POINTER_BYTES);
tDebug("%s %p server connection is allocated, uid:0x%x sid:%d key:%s spi:%d", pRpc->label, pConn, pConn->linkUid,
sid, hashstr, pConn->spi);
}
return pConn;
}
static SRpcConn *rpcGetConnObj(SRpcInfo *pRpc, int sid, SRecvInfo *pRecv) {
SRpcConn *pConn = NULL;
SRpcHead *pHead = (SRpcHead *)pRecv->msg;
if (sid) {
pConn = pRpc->connList + sid;
if (pConn->user[0] == 0) pConn = NULL;
}
if (pConn == NULL) {
if (pRpc->connType == TAOS_CONN_SERVER) {
pConn = rpcAllocateServerConn(pRpc, pRecv);
} else {
terrno = TSDB_CODE_RPC_UNEXPECTED_RESPONSE;
}
}
if (pConn) {
if (pConn->linkUid != pHead->linkUid) {
terrno = TSDB_CODE_RPC_MISMATCHED_LINK_ID;
tDebug("%s %p %p, linkUid:0x%x is not matched with received:0x%x", pRpc->label, pConn, (void *)pHead->ahandle,
pConn->linkUid, pHead->linkUid);
pConn = NULL;
}
}
return pConn;
}
static SRpcConn *rpcSetupConnToServer(SRpcReqContext *pContext) {
SRpcConn *pConn;
SRpcInfo *pRpc = pContext->pRpc;
SEpSet * pEpSet = &pContext->epSet;
pConn = rpcGetConnFromCache(pRpc->pCache, pEpSet->eps[pEpSet->inUse].fqdn, pEpSet->eps[pEpSet->inUse].port,
pContext->connType);
if (pConn == NULL || pConn->user[0] == 0) {
pConn = rpcOpenConn(pRpc, pEpSet->eps[pEpSet->inUse].fqdn, pEpSet->eps[pEpSet->inUse].port, pContext->connType);
}
if (pConn) {
pConn->tretry = 0;
pConn->ahandle = pContext->ahandle;
snprintf(pConn->info, sizeof(pConn->info), "%s %p %p", pRpc->label, pConn, pConn->ahandle);
pConn->tretry = 0;
} else {
tError("%s %p, failed to set up connection(%s)", pRpc->label, pContext->ahandle, tstrerror(terrno));
}
return pConn;
}
static int rpcProcessReqHead(SRpcConn *pConn, SRpcHead *pHead) {
if (pConn->peerId == 0) {
pConn->peerId = pHead->sourceId;
} else {
if (pConn->peerId != pHead->sourceId) {
tDebug("%s, source Id is changed, old:0x%08x new:0x%08x", pConn->info, pConn->peerId, pHead->sourceId);
return TSDB_CODE_RPC_INVALID_VALUE;
}
}
if (pConn->inTranId == pHead->tranId) {
if (pConn->inType == pHead->msgType) {
if (pHead->code == 0) {
tDebug("%s, %s is retransmitted", pConn->info, TMSG_INFO(pHead->msgType));
rpcSendQuickRsp(pConn, TSDB_CODE_RPC_ACTION_IN_PROGRESS);
} else {
// do nothing, it is heart beat from client
}
} else if (pConn->inType == 0) {
tDebug("%s, %s is already processed, tranId:%d", pConn->info, TMSG_INFO(pHead->msgType), pConn->inTranId);
rpcSendMsgToPeer(pConn, pConn->pRspMsg, pConn->rspMsgLen); // resend the response
} else {
tDebug("%s, mismatched message %s and tranId", pConn->info, TMSG_INFO(pHead->msgType));
}
// do not reply any message
return TSDB_CODE_RPC_ALREADY_PROCESSED;
}
if (pConn->inType != 0) {
tDebug("%s, last session is not finished, inTranId:%d tranId:%d", pConn->info, pConn->inTranId, pHead->tranId);
return TSDB_CODE_RPC_LAST_SESSION_NOT_FINISHED;
}
if (rpcContLenFromMsg(pHead->msgLen) <= 0) {
tDebug("%s, message body is empty, ignore", pConn->info);
return TSDB_CODE_RPC_APP_ERROR;
}
pConn->inTranId = pHead->tranId;
pConn->inType = pHead->msgType;
// start the progress timer to monitor the response from server app
if (pConn->connType != RPC_CONN_TCPS)
pConn->pTimer = taosTmrStart(rpcProcessProgressTimer, tsProgressTimer, pConn, pConn->pRpc->tmrCtrl);
return 0;
}
static int rpcProcessRspHead(SRpcConn *pConn, SRpcHead *pHead) {
SRpcInfo *pRpc = pConn->pRpc;
pConn->peerId = pHead->sourceId;
if (pConn->outType == 0 || pConn->pContext == NULL) {
return TSDB_CODE_RPC_UNEXPECTED_RESPONSE;
}
if (pHead->tranId != pConn->outTranId) {
return TSDB_CODE_RPC_INVALID_TRAN_ID;
}
if (pHead->msgType != pConn->outType + 1) {
return TSDB_CODE_RPC_INVALID_RESPONSE_TYPE;
}
taosTmrStopA(&pConn->pTimer);
pConn->retry = 0;
if (pHead->code == TSDB_CODE_RPC_AUTH_REQUIRED && pRpc->spi) {
tDebug("%s, authentication shall be restarted", pConn->info);
pConn->secured = 0;
rpcSendMsgToPeer(pConn, pConn->pReqMsg, pConn->reqMsgLen);
if (pConn->connType != RPC_CONN_TCPC)
pConn->pTimer = taosTmrStart(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl);
return TSDB_CODE_RPC_ALREADY_PROCESSED;
}
if (pHead->code == TSDB_CODE_RPC_MISMATCHED_LINK_ID) {
tDebug("%s, mismatched linkUid, link shall be restarted", pConn->info);
pConn->secured = 0;
((SRpcHead *)pConn->pReqMsg)->destId = 0;
rpcSendMsgToPeer(pConn, pConn->pReqMsg, pConn->reqMsgLen);
if (pConn->connType != RPC_CONN_TCPC)
pConn->pTimer = taosTmrStart(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl);
return TSDB_CODE_RPC_ALREADY_PROCESSED;
}
if (pHead->code == TSDB_CODE_RPC_ACTION_IN_PROGRESS) {
if (pConn->tretry <= tsRpcMaxRetry) {
tDebug("%s, peer is still processing the transaction, retry:%d", pConn->info, pConn->tretry);
pConn->tretry++;
rpcSendReqHead(pConn);
if (pConn->connType != RPC_CONN_TCPC)
pConn->pTimer = taosTmrStart(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl);
return TSDB_CODE_RPC_ALREADY_PROCESSED;
} else {
// peer still in processing, give up
tDebug("%s, server processing takes too long time, give up", pConn->info);
pHead->code = TSDB_CODE_RPC_TOO_SLOW;
}
}
pConn->outType = 0;
pConn->pReqMsg = NULL;
pConn->reqMsgLen = 0;
SRpcReqContext *pContext = pConn->pContext;
if (pHead->code == TSDB_CODE_RPC_REDIRECT) {
if (rpcContLenFromMsg(pHead->msgLen) < sizeof(SEpSet)) {
// if EpSet is not included in the msg, treat it as NOT_READY
pHead->code = TSDB_CODE_RPC_NOT_READY;
} else {
pContext->redirect++;
if (pContext->redirect > TSDB_MAX_REPLICA) {
pHead->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
tWarn("%s, too many redirects, quit", pConn->info);
}
}
}
return TSDB_CODE_SUCCESS;
}
static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqContext **ppContext) {
int32_t sid;
SRpcConn *pConn = NULL;
SRpcHead *pHead = (SRpcHead *)pRecv->msg;
sid = htonl(pHead->destId);
*ppContext = NULL;
if (TMSG_INDEX(pHead->msgType) >= TDMT_MAX || TMSG_INDEX(pHead->msgType) <= 0) {
tDebug("%s sid:%d, invalid message type:%d", pRpc->label, sid, pHead->msgType);
terrno = TSDB_CODE_RPC_INVALID_MSG_TYPE;
return NULL;
}
if (sid < 0 || sid >= pRpc->sessions) {
tDebug("%s sid:%d, sid is out of range, max sid:%d, %s discarded", pRpc->label, sid, pRpc->sessions,
TMSG_INFO(pHead->msgType));
terrno = TSDB_CODE_RPC_INVALID_SESSION_ID;
return NULL;
}
if (rpcIsReq(pHead->msgType) && htonl(pHead->msgVer) != tsVersion >> 8) {
tDebug("%s sid:%d, invalid client version:%x/%x %s", pRpc->label, sid, htonl(pHead->msgVer), tsVersion,
TMSG_INFO(pHead->msgType));
terrno = TSDB_CODE_RPC_INVALID_VERSION;
return NULL;
}
pConn = rpcGetConnObj(pRpc, sid, pRecv);
if (pConn == NULL) {
tDebug("%s %p, failed to get connection obj(%s)", pRpc->label, (void *)pHead->ahandle, tstrerror(terrno));
return NULL;
}
rpcLockConn(pConn);
if (rpcIsReq(pHead->msgType)) {
pConn->ahandle = (void *)pHead->ahandle;
snprintf(pConn->info, sizeof(pConn->info), "%s %p %p", pRpc->label, pConn, pConn->ahandle);
}
sid = pConn->sid;
if (pConn->chandle == NULL) pConn->chandle = pRecv->chandle;
pConn->peerIp = pRecv->ip;
pConn->peerPort = pRecv->port;
if (pHead->port) pConn->peerPort = htons(pHead->port);
terrno = rpcCheckAuthentication(pConn, (char *)pHead, pRecv->msgLen);
// code can be transformed only after authentication
pHead->code = htonl(pHead->code);
if (terrno == 0) {
if (pHead->encrypt) {
// decrypt here
}
if (rpcIsReq(pHead->msgType)) {
pConn->connType = pRecv->connType;
terrno = rpcProcessReqHead(pConn, pHead);
// stop idle timer
taosTmrStopA(&pConn->pIdleTimer);
// client shall send the request within tsRpcTime again for UDP, double it
if (pConn->connType != RPC_CONN_TCPS)
pConn->pIdleTimer = taosTmrStart(rpcProcessIdleTimer, tsRpcTimer * 2, pConn, pRpc->tmrCtrl);
} else {
terrno = rpcProcessRspHead(pConn, pHead);
*ppContext = pConn->pContext;
}
}
rpcUnlockConn(pConn);
return pConn;
}
static void doRpcReportBrokenLinkToServer(void *param, void *id) {
SRpcMsg * pRpcMsg = (SRpcMsg *)(param);
SRpcConn *pConn = (SRpcConn *)(pRpcMsg->handle);
SRpcInfo *pRpc = pConn->pRpc;
(*(pRpc->cfp))(pRpc->parent, pRpcMsg, NULL);
taosMemoryFree(pRpcMsg);
}
static void rpcReportBrokenLinkToServer(SRpcConn *pConn) {
SRpcInfo *pRpc = pConn->pRpc;
if (pConn->pReqMsg == NULL) return;
// if there are pending request, notify the app
rpcAddRef(pRpc);
tDebug("%s, notify the server app, connection is gone", pConn->info);
SRpcMsg *rpcMsg = taosMemoryMalloc(sizeof(SRpcMsg));
rpcMsg->pCont = pConn->pReqMsg; // pReqMsg is re-used to store the APP context from server
rpcMsg->contLen = pConn->reqMsgLen; // reqMsgLen is re-used to store the APP context length
rpcMsg->ahandle = pConn->ahandle;
rpcMsg->handle = pConn;
rpcMsg->msgType = pConn->inType;
rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
pConn->pReqMsg = NULL;
pConn->reqMsgLen = 0;
if (pRpc->cfp) {
taosTmrStart(doRpcReportBrokenLinkToServer, 0, rpcMsg, pRpc->tmrCtrl);
} else {
taosMemoryFree(rpcMsg);
}
}
static void rpcProcessBrokenLink(SRpcConn *pConn) {
if (pConn == NULL) return;
SRpcInfo *pRpc = pConn->pRpc;
tDebug("%s, link is broken", pConn->info);
rpcLockConn(pConn);
if (pConn->outType) {
SRpcReqContext *pContext = pConn->pContext;
pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
pContext->pConn = NULL;
pConn->pReqMsg = NULL;
taosTmrStart(rpcProcessConnError, 0, pContext, pRpc->tmrCtrl);
}
if (pConn->inType) rpcReportBrokenLinkToServer(pConn);
rpcReleaseConn(pConn);
rpcUnlockConn(pConn);
}
static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
SRpcHead *pHead = (SRpcHead *)pRecv->msg;
SRpcInfo *pRpc = (SRpcInfo *)pRecv->shandle;
SRpcConn *pConn = (SRpcConn *)pRecv->thandle;
tDump(pRecv->msg, pRecv->msgLen);
// underlying UDP layer does not know it is server or client
pRecv->connType = pRecv->connType | pRpc->connType;
if (pRecv->msg == NULL) {
rpcProcessBrokenLink(pConn);
return NULL;
}
terrno = 0;
SRpcReqContext *pContext;
pConn = rpcProcessMsgHead(pRpc, pRecv, &pContext);
char ipstr[24] = {0};
taosIpPort2String(pRecv->ip, pRecv->port, ipstr);
if (TMSG_INDEX(pHead->msgType) >= 1 && TMSG_INDEX(pHead->msgType) < TDMT_MAX) {
tDebug("%s %p %p, %s received from %s, parse code:0x%x len:%d sig:0x%08x:0x%08x:%d code:0x%x", pRpc->label, pConn,
(void *)pHead->ahandle, TMSG_INFO(pHead->msgType), ipstr, terrno, pRecv->msgLen, pHead->sourceId,
pHead->destId, pHead->tranId, pHead->code);
} else {
tDebug("%s %p %p, %d received from %s, parse code:0x%x len:%d sig:0x%08x:0x%08x:%d code:0x%x", pRpc->label, pConn,
(void *)pHead->ahandle, pHead->msgType, ipstr, terrno, pRecv->msgLen, pHead->sourceId, pHead->destId,
pHead->tranId, pHead->code);
}
int32_t code = terrno;
if (code != TSDB_CODE_RPC_ALREADY_PROCESSED) {
if (code != 0) { // parsing error
if (rpcIsReq(pHead->msgType)) {
rpcSendErrorMsgToPeer(pRecv, code);
if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE) {
rpcCloseConn(pConn);
}
if (TMSG_INDEX(pHead->msgType) + 1 > 1 && TMSG_INDEX(pHead->msgType) + 1 < TDMT_MAX) {
tDebug("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle,
TMSG_INFO(pHead->msgType + 1), code);
} else {
tError("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle,
TMSG_INFO(pHead->msgType), code);
}
}
} else { // msg is passed to app only parsing is ok
rpcProcessIncomingMsg(pConn, pHead, pContext);
}
}
if (code) rpcFreeMsg(pRecv->msg); // parsing failed, msg shall be freed
return pConn;
}
static void rpcNotifyClient(SRpcReqContext *pContext, SRpcMsg *pMsg) {
SRpcInfo *pRpc = pContext->pRpc;
pContext->pConn = NULL;
if (pContext->pRsp) {
// for synchronous API
memcpy(pContext->pSet, &pContext->epSet, sizeof(SEpSet));
memcpy(pContext->pRsp, pMsg, sizeof(SRpcMsg));
tsem_post(pContext->pSem);
} else {
// for asynchronous API
SEpSet *pEpSet = NULL;
if (pContext->epSet.inUse != pContext->oldInUse || pContext->redirect) pEpSet = &pContext->epSet;
(*pRpc->cfp)(pRpc->parent, pMsg, pEpSet);
}
// free the request message
taosRemoveRef(tsRpcRefId, pContext->rid);
}
static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqContext *pContext) {
SRpcInfo *pRpc = pConn->pRpc;
SRpcMsg rpcMsg;
pHead = rpcDecompressRpcMsg(pHead);
rpcMsg.contLen = rpcContLenFromMsg(pHead->msgLen);
rpcMsg.pCont = pHead->content;
rpcMsg.msgType = pHead->msgType;
rpcMsg.code = pHead->code;
if (rpcIsReq(pHead->msgType)) {
rpcMsg.ahandle = pConn->ahandle;
rpcMsg.handle = pConn;
rpcAddRef(pRpc); // add the refCount for requests
// notify the server app
(*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL);
} else {
// it's a response
rpcMsg.handle = pContext;
rpcMsg.ahandle = pContext->ahandle;
pContext->pConn = NULL;
// 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) {
rpcAddConnIntoCache(pRpc->pCache, pConn, pConn->peerFqdn, pContext->epSet.eps[pContext->epSet.inUse].port,
pConn->connType);
} else {
rpcCloseConn(pConn);
}
if (pHead->code == TSDB_CODE_RPC_REDIRECT) {
pContext->numOfTry = 0;
SEpSet *pEpSet = (SEpSet *)pHead->content;
if (pEpSet->numOfEps > 0) {
memcpy(&pContext->epSet, pHead->content, sizeof(pContext->epSet));
tDebug("%s, redirect is received, numOfEps:%d inUse:%d", pConn->info, pContext->epSet.numOfEps,
pContext->epSet.inUse);
for (int i = 0; i < pContext->epSet.numOfEps; ++i) {
pContext->epSet.eps[i].port = htons(pContext->epSet.eps[i].port);
tDebug("%s, redirect is received, index:%d ep:%s:%u", pConn->info, i, pContext->epSet.eps[i].fqdn,
pContext->epSet.eps[i].port);
}
}
rpcSendReqToServer(pRpc, pContext);
rpcFreeCont(rpcMsg.pCont);
} else if (pHead->code == TSDB_CODE_RPC_NOT_READY || pHead->code == TSDB_CODE_APP_NOT_READY ||
pHead->code == TSDB_CODE_NODE_OFFLINE) {
pContext->code = pHead->code;
rpcProcessConnError(pContext, NULL);
rpcFreeCont(rpcMsg.pCont);
} else {
rpcNotifyClient(pContext, &rpcMsg);
}
}
}
static void rpcSendQuickRsp(SRpcConn *pConn, int32_t code) {
char msg[RPC_MSG_OVERHEAD];
SRpcHead *pHead;
// set msg header
memset(msg, 0, sizeof(SRpcHead));
pHead = (SRpcHead *)msg;
pHead->version = 1;
pHead->msgType = pConn->inType + 1;
pHead->spi = pConn->spi;
pHead->encrypt = 0;
pHead->tranId = pConn->inTranId;
pHead->sourceId = pConn->ownId;
pHead->destId = pConn->peerId;
pHead->linkUid = pConn->linkUid;
pHead->ahandle = (uint64_t)pConn->ahandle;
memcpy(pHead->user, pConn->user, tListLen(pHead->user));
pHead->code = htonl(code);
rpcSendMsgToPeer(pConn, msg, sizeof(SRpcHead));
pConn->secured = 1; // connection shall be secured
}
static void rpcSendReqHead(SRpcConn *pConn) {
char msg[RPC_MSG_OVERHEAD];
SRpcHead *pHead;
// set msg header
memset(msg, 0, sizeof(SRpcHead));
pHead = (SRpcHead *)msg;
pHead->version = 1;
pHead->msgType = pConn->outType;
pHead->msgVer = htonl(tsVersion >> 8);
pHead->spi = pConn->spi;
pHead->encrypt = 0;
pHead->tranId = pConn->outTranId;
pHead->sourceId = pConn->ownId;
pHead->destId = pConn->peerId;
pHead->linkUid = pConn->linkUid;
pHead->ahandle = (uint64_t)pConn->ahandle;
memcpy(pHead->user, pConn->user, tListLen(pHead->user));
pHead->code = 1;
rpcSendMsgToPeer(pConn, msg, sizeof(SRpcHead));
}
static void rpcSendErrorMsgToPeer(SRecvInfo *pRecv, int32_t code) {
SRpcHead *pRecvHead, *pReplyHead;
char msg[sizeof(SRpcHead) + sizeof(SRpcDigest) + sizeof(uint32_t)];
uint32_t timeStamp;
int msgLen;
pRecvHead = (SRpcHead *)pRecv->msg;
pReplyHead = (SRpcHead *)msg;
memset(msg, 0, sizeof(SRpcHead));
pReplyHead->version = pRecvHead->version;
pReplyHead->msgType = (tmsg_t)(pRecvHead->msgType + 1);
pReplyHead->spi = 0;
pReplyHead->encrypt = pRecvHead->encrypt;
pReplyHead->tranId = pRecvHead->tranId;
pReplyHead->sourceId = pRecvHead->destId;
pReplyHead->destId = pRecvHead->sourceId;
pReplyHead->linkUid = pRecvHead->linkUid;
pReplyHead->ahandle = pRecvHead->ahandle;
pReplyHead->code = htonl(code);
msgLen = sizeof(SRpcHead);
if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP) {
// include a time stamp if client's time is not synchronized well
uint8_t *pContent = pReplyHead->content;
timeStamp = htonl(taosGetTimestampSec());
memcpy(pContent, &timeStamp, sizeof(timeStamp));
msgLen += sizeof(timeStamp);
}
pReplyHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
(*taosSendData[pRecv->connType])(pRecv->ip, pRecv->port, msg, msgLen, pRecv->chandle);
return;
}
static void rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) {
SRpcHead *pHead = rpcHeadFromCont(pContext->pCont);
char * msg = (char *)pHead;
int msgLen = rpcMsgLenFromCont(pContext->contLen);
tmsg_t msgType = pContext->msgType;
pContext->numOfTry++;
SRpcConn *pConn = rpcSetupConnToServer(pContext);
if (pConn == NULL) {
pContext->code = terrno;
taosTmrStart(rpcProcessConnError, 1, pContext, pRpc->tmrCtrl);
return;
}
pContext->pConn = pConn;
pConn->ahandle = pContext->ahandle;
rpcLockConn(pConn);
// set the message header
pHead->version = 1;
pHead->msgVer = htonl(tsVersion >> 8);
pHead->msgType = msgType;
pHead->encrypt = 0;
pConn->tranId++;
if (pConn->tranId == 0) pConn->tranId++;
pHead->tranId = pConn->tranId;
pHead->sourceId = pConn->ownId;
pHead->destId = pConn->peerId;
pHead->port = 0;
pHead->linkUid = pConn->linkUid;
pHead->ahandle = (uint64_t)pConn->ahandle;
memcpy(pHead->user, pConn->user, tListLen(pHead->user));
// set the connection parameters
pConn->outType = msgType;
pConn->outTranId = pHead->tranId;
pConn->pReqMsg = msg;
pConn->reqMsgLen = msgLen;
pConn->pContext = pContext;
rpcSendMsgToPeer(pConn, msg, msgLen);
if (pConn->connType != RPC_CONN_TCPC)
taosTmrReset(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl, &pConn->pTimer);
rpcUnlockConn(pConn);
}
static void rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) {
int writtenLen = 0;
SRpcHead *pHead = (SRpcHead *)msg;
msgLen = rpcAddAuthPart(pConn, msg, msgLen);
if (rpcIsReq(pHead->msgType)) {
tDebug("%s, %s is sent to %s:%hu, len:%d sig:0x%08x:0x%08x:%d", pConn->info, TMSG_INFO(pHead->msgType),
pConn->peerFqdn, pConn->peerPort, msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
} else {
if (pHead->code == 0) {
pConn->secured = 1; // for success response, set link as secured
}
char ipport[40] = {0};
taosIpPort2String(pConn->peerIp, pConn->peerPort, ipport);
tDebug("%s, %s is sent to %s, code:0x%x len:%d sig:0x%08x:0x%08x:%d", pConn->info, TMSG_INFO(pHead->msgType),
ipport, htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
}
// tTrace("connection type is: %d", pConn->connType);
writtenLen = (*taosSendData[pConn->connType])(pConn->peerIp, pConn->peerPort, pHead, msgLen, pConn->chandle);
if (writtenLen != msgLen) {
tError("%s, failed to send, msgLen:%d written:%d, reason:%s", pConn->info, msgLen, writtenLen, strerror(errno));
}
tDump(msg, msgLen);
}
static void rpcProcessConnError(void *param, void *id) {
SRpcReqContext *pContext = (SRpcReqContext *)param;
SRpcInfo * pRpc = pContext->pRpc;
SRpcMsg rpcMsg = {0};
if (pRpc == NULL) {
return;
}
tDebug("%s %p, connection error happens", pRpc->label, pContext->ahandle);
if (pContext->numOfTry >= pContext->epSet.numOfEps || pContext->msgType == TDMT_VND_FETCH) {
rpcMsg.msgType = pContext->msgType + 1;
rpcMsg.ahandle = pContext->ahandle;
rpcMsg.code = pContext->code;
rpcMsg.pCont = NULL;
rpcMsg.contLen = 0;
rpcNotifyClient(pContext, &rpcMsg);
} else {
// move to next IP
pContext->epSet.inUse++;
pContext->epSet.inUse = pContext->epSet.inUse % pContext->epSet.numOfEps;
rpcSendReqToServer(pRpc, pContext);
}
}
static void rpcProcessRetryTimer(void *param, void *tmrId) {
SRpcConn *pConn = (SRpcConn *)param;
SRpcInfo *pRpc = pConn->pRpc;
rpcLockConn(pConn);
if (pConn->outType && pConn->user[0]) {
tDebug("%s, expected %s is not received", pConn->info, TMSG_INFO((int)pConn->outType + 1));
pConn->pTimer = NULL;
pConn->retry++;
if (pConn->retry < 4) {
tDebug("%s, re-send msg:%s to %s:%hu", pConn->info, TMSG_INFO(pConn->outType), pConn->peerFqdn, pConn->peerPort);
rpcSendMsgToPeer(pConn, pConn->pReqMsg, pConn->reqMsgLen);
pConn->pTimer = taosTmrStart(rpcProcessRetryTimer, tsRpcTimer, pConn, pRpc->tmrCtrl);
} else {
// close the connection
tDebug("%s, failed to send msg:%s to %s:%hu", pConn->info, TMSG_INFO(pConn->outType), pConn->peerFqdn,
pConn->peerPort);
if (pConn->pContext) {
pConn->pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
pConn->pContext->pConn = NULL;
pConn->pReqMsg = NULL;
taosTmrStart(rpcProcessConnError, 1, pConn->pContext, pRpc->tmrCtrl);
rpcReleaseConn(pConn);
}
}
} else {
tDebug("%s, retry timer not processed", pConn->info);
}
rpcUnlockConn(pConn);
}
static void rpcProcessIdleTimer(void *param, void *tmrId) {
SRpcConn *pConn = (SRpcConn *)param;
rpcLockConn(pConn);
if (pConn->user[0]) {
tDebug("%s, close the connection since no activity", pConn->info);
if (pConn->inType) rpcReportBrokenLinkToServer(pConn);
rpcReleaseConn(pConn);
} else {
tDebug("%s, idle timer:%p not processed", pConn->info, tmrId);
}
rpcUnlockConn(pConn);
}
static void rpcProcessProgressTimer(void *param, void *tmrId) {
SRpcConn *pConn = (SRpcConn *)param;
SRpcInfo *pRpc = pConn->pRpc;
rpcLockConn(pConn);
if (pConn->inType && pConn->user[0]) {
tDebug("%s, progress timer expired, send progress", pConn->info);
rpcSendQuickRsp(pConn, TSDB_CODE_RPC_ACTION_IN_PROGRESS);
pConn->pTimer = taosTmrStart(rpcProcessProgressTimer, tsProgressTimer, pConn, pRpc->tmrCtrl);
} else {
tDebug("%s, progress timer:%p not processed", pConn->info, tmrId);
}
rpcUnlockConn(pConn);
}
static int32_t rpcCompressRpcMsg(char *pCont, int32_t contLen) {
SRpcHead *pHead = rpcHeadFromCont(pCont);
int32_t finalLen = 0;
int overhead = sizeof(SRpcComp);
if (!NEEDTO_COMPRESSS_MSG(contLen)) {
return contLen;
}
char *buf = taosMemoryMalloc(contLen + overhead + 8); // 8 extra bytes
if (buf == NULL) {
tError("failed to allocate memory for rpc msg compression, contLen:%d", contLen);
return contLen;
}
int32_t compLen = LZ4_compress_default(pCont, buf, contLen, contLen + overhead);
tDebug("compress rpc msg, before:%d, after:%d, overhead:%d", contLen, compLen, overhead);
/*
* only the compressed size is less than the value of contLen - overhead, the compression is applied
* The first four bytes is set to 0, the second four bytes are utilized to keep the original length of message
*/
if (compLen > 0 && compLen < contLen - overhead) {
SRpcComp *pComp = (SRpcComp *)pCont;
pComp->reserved = 0;
pComp->contLen = htonl(contLen);
memcpy(pCont + overhead, buf, compLen);
pHead->comp = 1;
tDebug("compress rpc msg, before:%d, after:%d", contLen, compLen);
finalLen = compLen + overhead;
} else {
finalLen = contLen;
}
taosMemoryFree(buf);
return finalLen;
}
static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead) {
int overhead = sizeof(SRpcComp);
SRpcHead *pNewHead = NULL;
uint8_t * pCont = pHead->content;
SRpcComp *pComp = (SRpcComp *)pHead->content;
if (pHead->comp) {
// decompress the content
assert(pComp->reserved == 0);
int contLen = htonl(pComp->contLen);
// prepare the temporary buffer to decompress message
char *temp = (char *)taosMemoryMalloc(contLen + RPC_MSG_OVERHEAD);
pNewHead = (SRpcHead *)(temp + sizeof(SRpcReqContext)); // reserve SRpcReqContext
if (pNewHead) {
int compLen = rpcContLenFromMsg(pHead->msgLen) - overhead;
int origLen = LZ4_decompress_safe((char *)(pCont + overhead), (char *)pNewHead->content, compLen, contLen);
assert(origLen == contLen);
memcpy(pNewHead, pHead, sizeof(SRpcHead));
pNewHead->msgLen = rpcMsgLenFromCont(origLen);
rpcFreeMsg(pHead); // free the compressed message buffer
pHead = pNewHead;
tTrace("decomp malloc mem:%p", temp);
} else {
tError("failed to allocate memory to decompress msg, contLen:%d", contLen);
}
}
return pHead;
}
static int rpcAuthenticateMsg(void *pMsg, int msgLen, void *pAuth, void *pKey) {
T_MD5_CTX context;
int ret = -1;
tMD5Init(&context);
tMD5Update(&context, (uint8_t *)pKey, TSDB_PASSWORD_LEN);
tMD5Update(&context, (uint8_t *)pMsg, msgLen);
tMD5Update(&context, (uint8_t *)pKey, TSDB_PASSWORD_LEN);
tMD5Final(&context);
if (memcmp(context.digest, pAuth, sizeof(context.digest)) == 0) ret = 0;
return ret;
}
static void rpcBuildAuthHead(void *pMsg, int msgLen, void *pAuth, void *pKey) {
T_MD5_CTX context;
tMD5Init(&context);
tMD5Update(&context, (uint8_t *)pKey, TSDB_PASSWORD_LEN);
tMD5Update(&context, (uint8_t *)pMsg, msgLen);
tMD5Update(&context, (uint8_t *)pKey, TSDB_PASSWORD_LEN);
tMD5Final(&context);
memcpy(pAuth, context.digest, sizeof(context.digest));
}
static int rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen) {
SRpcHead *pHead = (SRpcHead *)msg;
if (pConn->spi && pConn->secured == 0) {
// add auth part
pHead->spi = pConn->spi;
SRpcDigest *pDigest = (SRpcDigest *)(msg + msgLen);
pDigest->timeStamp = htonl(taosGetTimestampSec());
msgLen += sizeof(SRpcDigest);
pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
rpcBuildAuthHead(pHead, msgLen - TSDB_AUTH_LEN, pDigest->auth, pConn->secret);
} else {
pHead->spi = 0;
pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
}
return msgLen;
}
static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen) {
SRpcHead *pHead = (SRpcHead *)msg;
int code = 0;
if ((pConn->secured && pHead->spi == 0) || (pHead->spi == 0 && pConn->spi == 0)) {
// secured link, or no authentication
pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen);
// tTrace("%s, secured link, no auth is required", pConn->info);
return 0;
}
if (!rpcIsReq(pHead->msgType)) {
// for response, if code is auth failure, it shall bypass the auth process
code = htonl(pHead->code);
if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE ||
code == TSDB_CODE_RPC_INVALID_VERSION || code == TSDB_CODE_RPC_AUTH_REQUIRED ||
code == TSDB_CODE_MND_USER_NOT_EXIST || code == TSDB_CODE_RPC_NOT_READY) {
pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen);
// tTrace("%s, dont check authentication since code is:0x%x", pConn->info, code);
return 0;
}
}
code = 0;
if (pHead->spi == pConn->spi) {
// authentication
SRpcDigest *pDigest = (SRpcDigest *)((char *)pHead + msgLen - sizeof(SRpcDigest));
int32_t delta;
delta = (int32_t)htonl(pDigest->timeStamp);
delta -= (int32_t)taosGetTimestampSec();
if (abs(delta) > 900) {
tWarn("%s, time diff:%d is too big, msg discarded", pConn->info, delta);
code = TSDB_CODE_RPC_INVALID_TIME_STAMP;
} else {
if (rpcAuthenticateMsg(pHead, msgLen - TSDB_AUTH_LEN, pDigest->auth, pConn->secret) < 0) {
tDebug("%s, authentication failed, msg discarded", pConn->info);
code = TSDB_CODE_RPC_AUTH_FAILURE;
} else {
pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen) - sizeof(SRpcDigest);
if (!rpcIsReq(pHead->msgType)) pConn->secured = 1; // link is secured for client
// tTrace("%s, message is authenticated", pConn->info);
}
}
} else {
tError("%s, auth spi:%d not matched with received:%d %p", pConn->info, pConn->spi, pHead->spi, pConn);
code = pHead->spi ? TSDB_CODE_RPC_AUTH_FAILURE : TSDB_CODE_RPC_AUTH_REQUIRED;
}
return code;
}
static void rpcLockConn(SRpcConn *pConn) {
int64_t tid = taosGetSelfPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(&(pConn->lockedBy), 0, tid) != 0) {
if (++i % 1000 == 0) {
sched_yield();
}
}
}
static void rpcUnlockConn(SRpcConn *pConn) {
int64_t tid = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(&(pConn->lockedBy), tid, 0) != tid) {
assert(false);
}
}
static void rpcAddRef(SRpcInfo *pRpc) { atomic_add_fetch_32(&pRpc->refCount, 1); }
static void rpcDecRef(SRpcInfo *pRpc) {
if (atomic_sub_fetch_32(&pRpc->refCount, 1) == 0) {
rpcCloseConnCache(pRpc->pCache);
taosHashCleanup(pRpc->hash);
taosTmrCleanUp(pRpc->tmrCtrl);
taosIdPoolCleanUp(pRpc->idPool);
taosMemoryFreeClear(pRpc->connList);
taosThreadMutexDestroy(&pRpc->mutex);
tDebug("%s rpc resources are released", pRpc->label);
taosMemoryFreeClear(pRpc);
atomic_sub_fetch_32(&tsRpcNum, 1);
}
}
#endif
source/libs/transport/src/rpcTcp.c 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#include "rpcTcp.h"
#include "os.h"
#include "rpcHead.h"
#include "rpcLog.h"
#include "taosdef.h"
#include "taoserror.h"
#include "tutil.h"
#ifndef USE_UV
typedef struct SFdObj {
void * signature;
TdSocketPtr pSocket; // TCP socket FD
void * thandle; // handle from upper layer, like TAOS
uint32_t ip;
uint16_t port;
int16_t closedByApp; // 1: already closed by App
struct SThreadObj *pThreadObj;
struct SFdObj * prev;
struct SFdObj * next;
} SFdObj;
typedef struct SThreadObj {
TdThread thread;
SFdObj * pHead;
TdThreadMutex mutex;
uint32_t ip;
bool stop;
TdEpollPtr pEpoll;
int numOfFds;
int threadId;
char label[TSDB_LABEL_LEN];
void * shandle; // handle passed by upper layer during server initialization
void *(*processData)(SRecvInfo *pPacket);
} SThreadObj;
typedef struct {
char label[TSDB_LABEL_LEN];
int32_t index;
int numOfThreads;
SThreadObj **pThreadObj;
} SClientObj;
typedef struct {
TdSocketServerPtr pSocketServer;
uint32_t ip;
uint16_t port;
int8_t stop;
int8_t reserve;
char label[TSDB_LABEL_LEN];
int numOfThreads;
void * shandle;
SThreadObj **pThreadObj;
TdThread thread;
} SServerObj;
static void * taosProcessTcpData(void *param);
static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, TdSocketPtr pSocket);
static void taosFreeFdObj(SFdObj *pFdObj);
static void taosReportBrokenLink(SFdObj *pFdObj);
static void * taosAcceptTcpConnection(void *arg);
void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle) {
SServerObj *pServerObj;
SThreadObj *pThreadObj;
pServerObj = (SServerObj *)taosMemoryCalloc(sizeof(SServerObj), 1);
if (pServerObj == NULL) {
tError("TCP:%s no enough memory", label);
terrno = TAOS_SYSTEM_ERROR(errno);
return NULL;
}
pServerObj->pSocketServer = NULL;
taosResetPthread(&pServerObj->thread);
pServerObj->ip = ip;
pServerObj->port = port;
tstrncpy(pServerObj->label, label, sizeof(pServerObj->label));
pServerObj->numOfThreads = numOfThreads;
pServerObj->pThreadObj = (SThreadObj **)taosMemoryCalloc(sizeof(SThreadObj *), numOfThreads);
if (pServerObj->pThreadObj == NULL) {
tError("TCP:%s no enough memory", label);
terrno = TAOS_SYSTEM_ERROR(errno);
taosMemoryFree(pServerObj);
return NULL;
}
int code = 0;
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
taosThreadAttrSetDetachState(&thattr, PTHREAD_CREATE_JOINABLE);
// initialize parameters in case it may encounter error later
for (int i = 0; i < numOfThreads; ++i) {
pThreadObj = (SThreadObj *)taosMemoryCalloc(sizeof(SThreadObj), 1);
if (pThreadObj == NULL) {
tError("TCP:%s no enough memory", label);
terrno = TAOS_SYSTEM_ERROR(errno);
for (int j = 0; j < i; ++j) taosMemoryFree(pServerObj->pThreadObj[j]);
taosMemoryFree(pServerObj->pThreadObj);
taosMemoryFree(pServerObj);
return NULL;
}
pServerObj->pThreadObj[i] = pThreadObj;
pThreadObj->pEpoll = NULL;
taosResetPthread(&pThreadObj->thread);
pThreadObj->processData = fp;
tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
pThreadObj->shandle = shandle;
pThreadObj->stop = false;
}
// initialize mutex, thread, fd which may fail
for (int i = 0; i < numOfThreads; ++i) {
pThreadObj = pServerObj->pThreadObj[i];
code = taosThreadMutexInit(&(pThreadObj->mutex), NULL);
if (code < 0) {
tError("%s failed to init TCP process data mutex(%s)", label, strerror(errno));
break;
}
pThreadObj->pEpoll = taosCreateEpoll(10); // size does not matter
if (pThreadObj->pEpoll == NULL) {
tError("%s failed to create TCP epoll", label);
code = -1;
break;
}
code = taosThreadCreate(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
if (code != 0) {
tError("%s failed to create TCP process data thread(%s)", label, strerror(errno));
break;
}
pThreadObj->threadId = i;
}
pServerObj->pSocketServer = taosOpenTcpServerSocket(pServerObj->ip, pServerObj->port);
if (pServerObj->pSocketServer == NULL) code = -1;
if (code == 0) {
code = taosThreadCreate(&pServerObj->thread, &thattr, taosAcceptTcpConnection, (void *)pServerObj);
if (code != 0) {
tError("%s failed to create TCP accept thread(%s)", label, strerror(code));
}
}
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosCleanUpTcpServer(pServerObj);
pServerObj = NULL;
} else {
tDebug("%s TCP server is initialized, ip:0x%x port:%hu numOfThreads:%d", label, ip, port, numOfThreads);
}
taosThreadAttrDestroy(&thattr);
return (void *)pServerObj;
}
static void taosStopTcpThread(SThreadObj *pThreadObj) {
if (pThreadObj == NULL) {
return;
}
// save thread into local variable and signal thread to stop
TdThread thread = pThreadObj->thread;
if (!taosCheckPthreadValid(thread)) {
return;
}
pThreadObj->stop = true;
if (taosComparePthread(thread, taosThreadSelf())) {
pthread_detach(taosThreadSelf());
return;
}
taosThreadJoin(thread, NULL);
}
void taosStopTcpServer(void *handle) {
SServerObj *pServerObj = handle;
if (pServerObj == NULL) return;
pServerObj->stop = 1;
if (pServerObj->pSocketServer != NULL) {
taosShutDownSocketServerRD(pServerObj->pSocketServer);
}
if (taosCheckPthreadValid(pServerObj->thread)) {
if (taosComparePthread(pServerObj->thread, taosThreadSelf())) {
pthread_detach(taosThreadSelf());
} else {
taosThreadJoin(pServerObj->thread, NULL);
}
}
tDebug("%s TCP server is stopped", pServerObj->label);
}
void taosCleanUpTcpServer(void *handle) {
SServerObj *pServerObj = handle;
SThreadObj *pThreadObj;
if (pServerObj == NULL) return;
for (int i = 0; i < pServerObj->numOfThreads; ++i) {
pThreadObj = pServerObj->pThreadObj[i];
taosStopTcpThread(pThreadObj);
}
tDebug("%s TCP server is cleaned up", pServerObj->label);
taosMemoryFreeClear(pServerObj->pThreadObj);
taosMemoryFreeClear(pServerObj);
}
static void *taosAcceptTcpConnection(void *arg) {
TdSocketPtr pSocket = NULL;
struct sockaddr_in caddr;
int threadId = 0;
SThreadObj * pThreadObj;
SServerObj * pServerObj;
pServerObj = (SServerObj *)arg;
tDebug("%s TCP server is ready, ip:0x%x:%hu", pServerObj->label, pServerObj->ip, pServerObj->port);
setThreadName("acceptTcpConn");
while (1) {
socklen_t addrlen = sizeof(caddr);
pSocket = taosAcceptTcpConnectSocket(pServerObj->pSocketServer, (struct sockaddr *)&caddr, &addrlen);
if (pServerObj->stop) {
tDebug("%s TCP server stop accepting new connections", pServerObj->label);
break;
}
if (pSocket == NULL) {
if (errno == EINVAL) {
tDebug("%s TCP server stop accepting new connections, exiting", pServerObj->label);
break;
}
tError("%s TCP accept failure(%s)", pServerObj->label, strerror(errno));
continue;
}
taosKeepTcpAlive(pSocket);
struct timeval to = {5, 0};
int32_t ret = taosSetSockOpt(pSocket, SOL_SOCKET, SO_RCVTIMEO, &to, sizeof(to));
if (ret != 0) {
taosCloseSocket(&pSocket);
tError("%s failed to set recv timeout fd(%s)for connection from:%s:%hu", pServerObj->label, strerror(errno),
taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port));
continue;
}
// pick up the thread to handle this connection
pThreadObj = pServerObj->pThreadObj[threadId];
SFdObj *pFdObj = taosMallocFdObj(pThreadObj, pSocket);
if (pFdObj) {
pFdObj->ip = caddr.sin_addr.s_addr;
pFdObj->port = htons(caddr.sin_port);
tDebug("%s new TCP connection from %s:%hu, FD:%p numOfFds:%d", pServerObj->label,
taosInetNtoa(caddr.sin_addr), pFdObj->port, pFdObj, pThreadObj->numOfFds);
} else {
taosCloseSocket(&pSocket);
tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno),
taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port));
}
// pick up next thread for next connection
threadId++;
threadId = threadId % pServerObj->numOfThreads;
}
taosCloseSocketServer(&pServerObj->pSocketServer);
return NULL;
}
void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle) {
SClientObj *pClientObj = (SClientObj *)taosMemoryCalloc(1, sizeof(SClientObj));
if (pClientObj == NULL) {
tError("TCP:%s no enough memory", label);
terrno = TAOS_SYSTEM_ERROR(errno);
return NULL;
}
tstrncpy(pClientObj->label, label, sizeof(pClientObj->label));
pClientObj->numOfThreads = numOfThreads;
pClientObj->pThreadObj = (SThreadObj **)taosMemoryCalloc(numOfThreads, sizeof(SThreadObj *));
if (pClientObj->pThreadObj == NULL) {
tError("TCP:%s no enough memory", label);
taosMemoryFreeClear(pClientObj);
terrno = TAOS_SYSTEM_ERROR(errno);
}
int code = 0;
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
taosThreadAttrSetDetachState(&thattr, PTHREAD_CREATE_JOINABLE);
for (int i = 0; i < numOfThreads; ++i) {
SThreadObj *pThreadObj = (SThreadObj *)taosMemoryCalloc(1, sizeof(SThreadObj));
if (pThreadObj == NULL) {
tError("TCP:%s no enough memory", label);
terrno = TAOS_SYSTEM_ERROR(errno);
for (int j = 0; j < i; ++j) taosMemoryFree(pClientObj->pThreadObj[j]);
taosMemoryFree(pClientObj);
taosThreadAttrDestroy(&thattr);
return NULL;
}
pClientObj->pThreadObj[i] = pThreadObj;
taosResetPthread(&pThreadObj->thread);
pThreadObj->ip = ip;
pThreadObj->stop = false;
tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
pThreadObj->shandle = shandle;
pThreadObj->processData = fp;
}
// initialize mutex, thread, fd which may fail
for (int i = 0; i < numOfThreads; ++i) {
SThreadObj *pThreadObj = pClientObj->pThreadObj[i];
code = taosThreadMutexInit(&(pThreadObj->mutex), NULL);
if (code < 0) {
tError("%s failed to init TCP process data mutex(%s)", label, strerror(errno));
break;
}
pThreadObj->pEpoll = taosCreateEpoll(10); // size does not matter
if (pThreadObj->pEpoll == NULL) {
tError("%s failed to create TCP epoll", label);
code = -1;
break;
}
code = taosThreadCreate(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
if (code != 0) {
tError("%s failed to create TCP process data thread(%s)", label, strerror(errno));
break;
}
pThreadObj->threadId = i;
}
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosCleanUpTcpClient(pClientObj);
pClientObj = NULL;
}
return pClientObj;
}
void taosStopTcpClient(void *chandle) {
SClientObj *pClientObj = chandle;
if (pClientObj == NULL) return;
tDebug("%s TCP client is stopped", pClientObj->label);
}
void taosCleanUpTcpClient(void *chandle) {
SClientObj *pClientObj = chandle;
if (pClientObj == NULL) return;
for (int i = 0; i < pClientObj->numOfThreads; ++i) {
SThreadObj *pThreadObj = pClientObj->pThreadObj[i];
taosStopTcpThread(pThreadObj);
}
tDebug("%s TCP client is cleaned up", pClientObj->label);
taosMemoryFreeClear(pClientObj->pThreadObj);
taosMemoryFreeClear(pClientObj);
}
void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) {
SClientObj *pClientObj = shandle;
int32_t index = atomic_load_32(&pClientObj->index) % pClientObj->numOfThreads;
atomic_store_32(&pClientObj->index, index + 1);
SThreadObj *pThreadObj = pClientObj->pThreadObj[index];
TdSocketPtr pSocket = taosOpenTcpClientSocket(ip, port, pThreadObj->ip);
if (pSocket == NULL) return NULL;
struct sockaddr_in sin;
uint16_t localPort = 0;
unsigned int addrlen = sizeof(sin);
if (taosGetSocketName(pSocket, (struct sockaddr *)&sin, &addrlen) == 0 && sin.sin_family == AF_INET && addrlen == sizeof(sin)) {
localPort = (uint16_t)ntohs(sin.sin_port);
}
SFdObj *pFdObj = taosMallocFdObj(pThreadObj, pSocket);
if (pFdObj) {
pFdObj->thandle = thandle;
pFdObj->port = port;
pFdObj->ip = ip;
char ipport[40] = {0};
taosIpPort2String(ip, port, ipport);
tDebug("%s %p TCP connection to %s is created, localPort:%hu FD:%p numOfFds:%d", pThreadObj->label, thandle,
ipport, localPort, pFdObj, pThreadObj->numOfFds);
} else {
tError("%s failed to malloc client FdObj(%s)", pThreadObj->label, strerror(errno));
taosCloseSocket(&pSocket);
}
return pFdObj;
}
void taosCloseTcpConnection(void *chandle) {
SFdObj *pFdObj = chandle;
if (pFdObj == NULL || pFdObj->signature != pFdObj) return;
SThreadObj *pThreadObj = pFdObj->pThreadObj;
tDebug("%s %p TCP connection will be closed, FD:%p", pThreadObj->label, pFdObj->thandle, pFdObj);
// pFdObj->thandle = NULL;
pFdObj->closedByApp = 1;
taosShutDownSocketWR(pFdObj->pSocket);
}
int taosSendTcpData(uint32_t ip, uint16_t port, void *data, int len, void *chandle) {
SFdObj *pFdObj = chandle;
if (pFdObj == NULL || pFdObj->signature != pFdObj) return -1;
SThreadObj *pThreadObj = pFdObj->pThreadObj;
int ret = taosWriteMsg(pFdObj->pSocket, data, len);
tTrace("%s %p TCP data is sent, FD:%p bytes:%d", pThreadObj->label, pFdObj->thandle, pFdObj, ret);
return ret;
}
static void taosReportBrokenLink(SFdObj *pFdObj) {
SThreadObj *pThreadObj = pFdObj->pThreadObj;
// notify the upper layer, so it will clean the associated context
if (pFdObj->closedByApp == 0) {
taosShutDownSocketWR(pFdObj->pSocket);
SRecvInfo recvInfo;
recvInfo.msg = NULL;
recvInfo.msgLen = 0;
recvInfo.ip = 0;
recvInfo.port = 0;
recvInfo.shandle = pThreadObj->shandle;
recvInfo.thandle = pFdObj->thandle;
recvInfo.chandle = NULL;
recvInfo.connType = RPC_CONN_TCP;
(*(pThreadObj->processData))(&recvInfo);
}
taosFreeFdObj(pFdObj);
}
static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
SRpcHead rpcHead;
int32_t msgLen, leftLen, retLen, headLen;
char * buffer, *msg;
SThreadObj *pThreadObj = pFdObj->pThreadObj;
headLen = taosReadMsg(pFdObj->pSocket, &rpcHead, sizeof(SRpcHead));
if (headLen != sizeof(SRpcHead)) {
tDebug("%s %p read error, FD:%p headLen:%d", pThreadObj->label, pFdObj->thandle, pFdObj, headLen);
return -1;
}
msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen);
int32_t size = msgLen + tsRpcOverhead;
buffer = taosMemoryMalloc(size);
if (NULL == buffer) {
tError("%s %p TCP taosMemoryMalloc(size:%d) fail", pThreadObj->label, pFdObj->thandle, msgLen);
return -1;
} else {
tTrace("%s %p read data, FD:%p TCP malloc mem:%p", pThreadObj->label, pFdObj->thandle, pFdObj, buffer);
}
msg = buffer + tsRpcOverhead;
leftLen = msgLen - headLen;
retLen = taosReadMsg(pFdObj->pSocket, msg + headLen, leftLen);
if (leftLen != retLen) {
tError("%s %p read error, leftLen:%d retLen:%d FD:%p", pThreadObj->label, pFdObj->thandle, leftLen, retLen, pFdObj);
taosMemoryFree(buffer);
return -1;
}
memcpy(msg, &rpcHead, sizeof(SRpcHead));
pInfo->msg = msg;
pInfo->msgLen = msgLen;
pInfo->ip = pFdObj->ip;
pInfo->port = pFdObj->port;
pInfo->shandle = pThreadObj->shandle;
pInfo->thandle = pFdObj->thandle;
pInfo->chandle = pFdObj;
pInfo->connType = RPC_CONN_TCP;
if (pFdObj->closedByApp) {
taosMemoryFree(buffer);
return -1;
}
return 0;
}
#define maxEvents 10
static void *taosProcessTcpData(void *param) {
SThreadObj * pThreadObj = param;
SFdObj * pFdObj;
struct epoll_event events[maxEvents];
SRecvInfo recvInfo;
char name[16] = {0};
snprintf(name, tListLen(name), "%s-tcp", pThreadObj->label);
setThreadName(name);
while (1) {
int fdNum = taosWaitEpoll(pThreadObj->pEpoll, events, maxEvents, TAOS_EPOLL_WAIT_TIME);
if (pThreadObj->stop) {
tDebug("%s TCP thread get stop event, exiting...", pThreadObj->label);
break;
}
if (fdNum < 0) continue;
for (int i = 0; i < fdNum; ++i) {
pFdObj = events[i].data.ptr;
if (events[i].events & EPOLLERR) {
tDebug("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj);
taosReportBrokenLink(pFdObj);
continue;
}
if (events[i].events & EPOLLRDHUP) {
tDebug("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
taosReportBrokenLink(pFdObj);
continue;
}
if (events[i].events & EPOLLHUP) {
tDebug("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
taosReportBrokenLink(pFdObj);
continue;
}
if (taosReadTcpData(pFdObj, &recvInfo) < 0) {
taosShutDownSocketWR(pFdObj->pSocket);
continue;
}
pFdObj->thandle = (*(pThreadObj->processData))(&recvInfo);
if (pFdObj->thandle == NULL) taosFreeFdObj(pFdObj);
}
if (pThreadObj->stop) break;
}
if (pThreadObj->pEpoll != NULL) {
taosCloseEpoll(&pThreadObj->pEpoll);
pThreadObj->pEpoll = NULL;
}
while (pThreadObj->pHead) {
pFdObj = pThreadObj->pHead;
pThreadObj->pHead = pFdObj->next;
taosReportBrokenLink(pFdObj);
}
taosThreadMutexDestroy(&(pThreadObj->mutex));
tDebug("%s TCP thread exits ...", pThreadObj->label);
taosMemoryFreeClear(pThreadObj);
return NULL;
}
static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, TdSocketPtr pSocket) {
struct epoll_event event;
SFdObj *pFdObj = (SFdObj *)taosMemoryCalloc(sizeof(SFdObj), 1);
if (pFdObj == NULL) {
return NULL;
}
pFdObj->closedByApp = 0;
pFdObj->pSocket = pSocket;
pFdObj->pThreadObj = pThreadObj;
pFdObj->signature = pFdObj;
event.events = EPOLLIN | EPOLLRDHUP;
event.data.ptr = pFdObj;
if (taosCtlEpoll(pThreadObj->pEpoll, EPOLL_CTL_ADD, pSocket, &event) < 0) {
taosMemoryFreeClear(pFdObj);
terrno = TAOS_SYSTEM_ERROR(errno);
return NULL;
}
// notify the data process, add into the FdObj list
taosThreadMutexLock(&(pThreadObj->mutex));
pFdObj->next = pThreadObj->pHead;
if (pThreadObj->pHead) (pThreadObj->pHead)->prev = pFdObj;
pThreadObj->pHead = pFdObj;
pThreadObj->numOfFds++;
taosThreadMutexUnlock(&(pThreadObj->mutex));
return pFdObj;
}
static void taosFreeFdObj(SFdObj *pFdObj) {
if (pFdObj == NULL) return;
if (pFdObj->signature != pFdObj) return;
SThreadObj *pThreadObj = pFdObj->pThreadObj;
taosThreadMutexLock(&pThreadObj->mutex);
if (pFdObj->signature == NULL) {
taosThreadMutexUnlock(&pThreadObj->mutex);
return;
}
pFdObj->signature = NULL;
taosCtlEpoll(pThreadObj->pEpoll, EPOLL_CTL_DEL, pFdObj->pSocket, NULL);
taosCloseSocket(&pFdObj->pSocket);
pThreadObj->numOfFds--;
if (pThreadObj->numOfFds < 0)
tError("%s %p TCP thread:%d, number of FDs is negative!!!", pThreadObj->label, pFdObj->thandle,
pThreadObj->threadId);
if (pFdObj->prev) {
(pFdObj->prev)->next = pFdObj->next;
} else {
pThreadObj->pHead = pFdObj->next;
}
if (pFdObj->next) {
(pFdObj->next)->prev = pFdObj->prev;
}
taosThreadMutexUnlock(&pThreadObj->mutex);
tDebug("%s %p TCP connection is closed, FD:%p numOfFds:%d", pThreadObj->label, pFdObj->thandle, pFdObj, pThreadObj->numOfFds);
taosMemoryFreeClear(pFdObj);
}
#endif
\ No newline at end of file
source/libs/transport/src/rpcUdp.c 已删除 100644 → 0
浏览文件 @ 49de494e
/*
* 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/>.
*/
#include "rpcUdp.h"
#include "os.h"
#include "rpcHead.h"
#include "rpcLog.h"
#include "taosdef.h"
#include "taoserror.h"
#include "ttimer.h"
#include "tutil.h"
#ifndef USE_UV
#define RPC_MAX_UDP_CONNS 256
#define RPC_MAX_UDP_PKTS 1000
#define RPC_UDP_BUF_TIME 5 // mseconds
#define RPC_MAX_UDP_SIZE 65480
typedef struct {
int index;
TdSocketPtr pSocket;
uint16_t port; // peer port
uint16_t localPort; // local port
char label[TSDB_LABEL_LEN]; // copy from udpConnSet;
TdThread thread;
void *hash;
void *shandle; // handle passed by upper layer during server initialization
void *pSet;
void *(*processData)(SRecvInfo *pRecv);
char *buffer; // buffer to receive data
} SUdpConn;
typedef struct {
int index;
int server;
uint32_t ip; // local IP
uint16_t port; // local Port
void * shandle; // handle passed by upper layer during server initialization
int threads;
char label[TSDB_LABEL_LEN];
void *(*fp)(SRecvInfo *pPacket);
SUdpConn udpConn[];
} SUdpConnSet;
static void *taosRecvUdpData(void *param);
void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int threads, void *fp, void *shandle) {
SUdpConn * pConn;
SUdpConnSet *pSet;
int size = (int)sizeof(SUdpConnSet) + threads * (int)sizeof(SUdpConn);
pSet = (SUdpConnSet *)taosMemoryMalloc((size_t)size);
if (pSet == NULL) {
tError("%s failed to allocate UdpConn", label);
terrno = TAOS_SYSTEM_ERROR(errno);
return NULL;
}
memset(pSet, 0, (size_t)size);
pSet->ip = ip;
pSet->port = port;
pSet->shandle = shandle;
pSet->fp = fp;
pSet->threads = threads;
tstrncpy(pSet->label, label, sizeof(pSet->label));
TdThreadAttr thAttr;
taosThreadAttrInit(&thAttr);
taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
int i;
uint16_t ownPort;
for (i = 0; i < threads; ++i) {
pConn = pSet->udpConn + i;
ownPort = (port ? port + i : 0);
pConn->pSocket = taosOpenUdpSocket(ip, ownPort);
if (pConn->pSocket == NULL) {
tError("%s failed to open UDP socket %x:%hu", label, ip, port);
break;
}
pConn->buffer = taosMemoryMalloc(RPC_MAX_UDP_SIZE);
if (NULL == pConn->buffer) {
tError("%s failed to malloc recv buffer", label);
break;
}
struct sockaddr_in sin;
unsigned int addrlen = sizeof(sin);
if (taosGetSocketName(pConn->pSocket, (struct sockaddr *)&sin, &addrlen) == 0 && sin.sin_family == AF_INET &&
addrlen == sizeof(sin)) {
pConn->localPort = (uint16_t)ntohs(sin.sin_port);
}
tstrncpy(pConn->label, label, sizeof(pConn->label));
pConn->shandle = shandle;
pConn->processData = fp;
pConn->index = i;
pConn->pSet = pSet;
int code = taosThreadCreate(&pConn->thread, &thAttr, taosRecvUdpData, pConn);
if (code != 0) {
tError("%s failed to create thread to process UDP data(%s)", label, strerror(errno));
break;
}
}
taosThreadAttrDestroy(&thAttr);
if (i != threads) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosCleanUpUdpConnection(pSet);
return NULL;
}
tDebug("%s UDP connection is initialized, ip:%x:%hu threads:%d", label, ip, port, threads);
return pSet;
}
void taosStopUdpConnection(void *handle) {
SUdpConnSet *pSet = (SUdpConnSet *)handle;
SUdpConn * pConn;
if (pSet == NULL) return;
for (int i = 0; i < pSet->threads; ++i) {
pConn = pSet->udpConn + i;
if (pConn->pSocket != NULL) taosShutDownSocketRDWR(pConn->pSocket);
if (pConn->pSocket != NULL) taosCloseSocket(&pConn->pSocket);
pConn->pSocket = NULL;
}
for (int i = 0; i < pSet->threads; ++i) {
pConn = pSet->udpConn + i;
if (taosCheckPthreadValid(pConn->thread)) {
taosThreadJoin(pConn->thread, NULL);
}
taosMemoryFreeClear(pConn->buffer);
// tTrace("%s UDP thread is closed, index:%d", pConn->label, i);
}
tDebug("%s UDP is stopped", pSet->label);
}
void taosCleanUpUdpConnection(void *handle) {
SUdpConnSet *pSet = (SUdpConnSet *)handle;
SUdpConn * pConn;
if (pSet == NULL) return;
for (int i = 0; i < pSet->threads; ++i) {
pConn = pSet->udpConn + i;
if (pConn->pSocket != NULL) taosCloseSocket(&pConn->pSocket);
}
tDebug("%s UDP is cleaned up", pSet->label);
taosMemoryFreeClear(pSet);
}
void *taosOpenUdpConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) {
SUdpConnSet *pSet = (SUdpConnSet *)shandle;
pSet->index = (pSet->index + 1) % pSet->threads;
SUdpConn *pConn = pSet->udpConn + pSet->index;
pConn->port = port;
tDebug("%s UDP connection is setup, ip:%x:%hu localPort:%hu", pConn->label, ip, port, pConn->localPort);
return pConn;
}
static void *taosRecvUdpData(void *param) {
SUdpConn * pConn = param;
struct sockaddr_in sourceAdd;
ssize_t dataLen;
unsigned int addLen;
uint16_t port;
SRecvInfo recvInfo;
memset(&sourceAdd, 0, sizeof(sourceAdd));
addLen = sizeof(sourceAdd);
tDebug("%s UDP thread is created, index:%d", pConn->label, pConn->index);
char *msg = pConn->buffer;
setThreadName("recvUdpData");
while (1) {
dataLen = taosReadFromSocket(pConn->pSocket, pConn->buffer, RPC_MAX_UDP_SIZE, 0, (struct sockaddr *)&sourceAdd, &addLen);
if (dataLen <= 0) {
tDebug("%s UDP socket was closed, exiting(%s), dataLen:%d", pConn->label, strerror(errno), (int32_t)dataLen);
// for windows usage, remote shutdown also returns - 1 in windows client
if (pConn->pSocket == NULL) {
break;
} else {
continue;
}
}
port = ntohs(sourceAdd.sin_port);
if (dataLen < sizeof(SRpcHead)) {
tError("%s recvfrom failed(%s)", pConn->label, strerror(errno));
continue;
}
int32_t size = dataLen + tsRpcOverhead;
char * tmsg = taosMemoryMalloc(size);
if (NULL == tmsg) {
tError("%s failed to allocate memory, size:%" PRId64, pConn->label, (int64_t)dataLen);
continue;
} else {
tTrace("UDP malloc mem:%p size:%d", tmsg, size);
}
tmsg += tsRpcOverhead; // overhead for SRpcReqContext
memcpy(tmsg, msg, dataLen);
recvInfo.msg = tmsg;
recvInfo.msgLen = dataLen;
recvInfo.ip = sourceAdd.sin_addr.s_addr;
recvInfo.port = port;
recvInfo.shandle = pConn->shandle;
recvInfo.thandle = NULL;
recvInfo.chandle = pConn;
recvInfo.connType = 0;
(*(pConn->processData))(&recvInfo);
}
return NULL;
}
int taosSendUdpData(uint32_t ip, uint16_t port, void *data, int dataLen, void *chandle) {
SUdpConn *pConn = (SUdpConn *)chandle;
if (pConn == NULL) return -1;
struct sockaddr_in destAdd;
memset(&destAdd, 0, sizeof(destAdd));
destAdd.sin_family = AF_INET;
destAdd.sin_addr.s_addr = ip;
destAdd.sin_port = htons(port);
int ret = taosSendto(pConn->pSocket, data, (size_t)dataLen, 0, (struct sockaddr *)&destAdd, sizeof(destAdd));
return ret;
}
#endif
\ No newline at end of file
source/libs/transport/test/pushServer.c
浏览文件 @ 84eeb2ab
......@@ -15,9 +15,9 @@
//#define _DEFAULT_SOURCE
#include "os.h"
#include "rpcLog.h"
#include "tglobal.h"
#include "tqueue.h"
#include "transLog.h"
#include "trpc.h"
int msgSize = 128;
......
source/libs/transport/test/rclient.c
浏览文件 @ 84eeb2ab
......@@ -14,9 +14,9 @@
*/
#include <tdatablock.h>
#include "os.h"
#include "rpcLog.h"
#include "taoserror.h"
#include "tglobal.h"
#include "transLog.h"
#include "trpc.h"
#include "tutil.h"
......
source/libs/transport/test/rserver.c
浏览文件 @ 84eeb2ab
......@@ -15,9 +15,9 @@
//#define _DEFAULT_SOURCE
#include "os.h"
#include "rpcLog.h"
#include "tglobal.h"
#include "tqueue.h"
#include "transLog.h"
#include "trpc.h"
int msgSize = 128;
......
source/libs/transport/test/syncClient.c
浏览文件 @ 84eeb2ab
......@@ -14,9 +14,9 @@
*/
#include <tdatablock.h>
#include "os.h"
#include "rpcLog.h"
#include "taoserror.h"
#include "tglobal.h"
#include "transLog.h"
#include "trpc.h"
#include "tutil.h"
......
source/libs/transport/test/transUT.cpp
浏览文件 @ 84eeb2ab
......@@ -15,10 +15,10 @@
#include <gtest/gtest.h>
#include <cstdio>
#include <cstring>
#include "rpcLog.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "tlog.h"
#include "transLog.h"
#include "trpc.h"
using namespace std;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或