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

Merge pull request #14214 from taosdata/feat/rpcRefactor1 

feat: rpc refactor
This commit is contained in:
Yihao Deng 2022-06-25 14:47:37 +08:00 committed by GitHub
parent 930613db8b 25a12d96b0
commit 6e4e10ad41
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 469 additions and 396 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
include/libs/qcom/query.h
View File

@ -16,6 +16,7 @@
#ifndef _TD_QUERY_H_ #ifndef _TD_QUERY_H_
#define _TD_QUERY_H_ #define _TD_QUERY_H_
// clang-foramt off
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
@ -71,7 +72,7 @@ typedef struct SIndexMeta {
} SIndexMeta; } SIndexMeta;
typedef struct STbVerInfo { typedef struct STbVerInfo {
char tbFName[TSDB_TABLE_FNAME_LEN]; char tbFName[TSDB_TABLE_FNAME_LEN];
int32_t sversion; int32_t sversion;
int32_t tversion; int32_t tversion;
} STbVerInfo; } STbVerInfo;
@ -141,7 +142,7 @@ typedef struct SDataBuf {
typedef struct STargetInfo { typedef struct STargetInfo {
ETargetType type; ETargetType type;
char* dbFName; // used to update db's vgroup epset char* dbFName; // used to update db's vgroup epset
int32_t vgId; int32_t vgId;
} STargetInfo; } STargetInfo;
@ -149,15 +150,15 @@ typedef int32_t (*__async_send_cb_fn_t)(void* param, const SDataBuf* pMsg, int32
typedef int32_t (*__async_exec_fn_t)(void* param); typedef int32_t (*__async_exec_fn_t)(void* param);
typedef struct SRequestConnInfo { typedef struct SRequestConnInfo {
void* pTrans; void* pTrans;
uint64_t requestId; uint64_t requestId;
int64_t requestObjRefId; int64_t requestObjRefId;
SEpSet mgmtEps; SEpSet mgmtEps;
} SRequestConnInfo; } SRequestConnInfo;
typedef struct SMsgSendInfo { typedef struct SMsgSendInfo {
__async_send_cb_fn_t fp; // async callback function __async_send_cb_fn_t fp; // async callback function
STargetInfo target; // for update epset STargetInfo target; // for update epset
void* param; void* param;
uint64_t requestId; uint64_t requestId;
uint64_t requestObjRefId; uint64_t requestObjRefId;
@ -206,13 +207,15 @@ int32_t queryCreateTableMetaFromMsg(STableMetaRsp* msg, bool isSuperTable, STabl
char* jobTaskStatusStr(int32_t status); char* jobTaskStatusStr(int32_t status);
SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name); SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name);
void destroyQueryExecRes(SQueryExecRes* pRes);
int32_t dataConverToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *len); void destroyQueryExecRes(SQueryExecRes* pRes);
char* parseTagDatatoJson(void* p); int32_t dataConverToStr(char* str, int type, void* buf, int32_t bufSize, int32_t* len);
char* parseTagDatatoJson(void* p);
int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst); int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst);
int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst); int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst);
extern int32_t (*queryBuildMsg[TDMT_MAX])(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallocFp)(int32_t)); extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char** msg, int32_t msgSize, int32_t* msgLen,
void* (*mallocFp)(int32_t));
extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t msgSize); extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t msgSize);
#define SET_META_TYPE_NULL(t) (t) = META_TYPE_NULL_TABLE #define SET_META_TYPE_NULL(t) (t) = META_TYPE_NULL_TABLE
@ -223,7 +226,7 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
#define NEED_CLIENT_RM_TBLMETA_ERROR(_code) \ #define NEED_CLIENT_RM_TBLMETA_ERROR(_code) \
((_code) == TSDB_CODE_PAR_TABLE_NOT_EXIST || (_code) == TSDB_CODE_VND_TB_NOT_EXIST || \ ((_code) == TSDB_CODE_PAR_TABLE_NOT_EXIST || (_code) == TSDB_CODE_VND_TB_NOT_EXIST || \
(_code) == TSDB_CODE_PAR_INVALID_COLUMNS_NUM || (_code) == TSDB_CODE_PAR_INVALID_COLUMN || \ (_code) == TSDB_CODE_PAR_INVALID_COLUMNS_NUM || (_code) == TSDB_CODE_PAR_INVALID_COLUMN || \
(_code) == TSDB_CODE_PAR_TAGS_NOT_MATCHED || (_code) == TSDB_CODE_PAR_VALUE_TOO_LONG || \ (_code) == TSDB_CODE_PAR_TAGS_NOT_MATCHED || (_code) == TSDB_CODE_PAR_VALUE_TOO_LONG || \
(_code) == TSDB_CODE_PAR_INVALID_DROP_COL || ((_code) == TSDB_CODE_TDB_INVALID_TABLE_ID)) (_code) == TSDB_CODE_PAR_INVALID_DROP_COL || ((_code) == TSDB_CODE_TDB_INVALID_TABLE_ID))
#define NEED_CLIENT_REFRESH_VG_ERROR(_code) \ #define NEED_CLIENT_REFRESH_VG_ERROR(_code) \
((_code) == TSDB_CODE_VND_HASH_MISMATCH || (_code) == TSDB_CODE_VND_INVALID_VGROUP_ID) ((_code) == TSDB_CODE_VND_HASH_MISMATCH || (_code) == TSDB_CODE_VND_INVALID_VGROUP_ID)
@ -231,11 +234,13 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
#define NEED_CLIENT_HANDLE_ERROR(_code) \ #define NEED_CLIENT_HANDLE_ERROR(_code) \
(NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \ (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \
NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code)) NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))
#define NEED_CLIENT_RM_TBLMETA_REQ(_type) ((_type) == TDMT_VND_CREATE_TABLE || (_type) == TDMT_VND_CREATE_STB \ #define NEED_CLIENT_RM_TBLMETA_REQ(_type) \
|| (_type) == TDMT_VND_DROP_TABLE || (_type) == TDMT_VND_DROP_STB) ((_type) == TDMT_VND_CREATE_TABLE || (_type) == TDMT_VND_CREATE_STB || (_type) == TDMT_VND_DROP_TABLE || \
(_type) == TDMT_VND_DROP_STB)
#define NEED_SCHEDULER_RETRY_ERROR(_code) \ #define NEED_SCHEDULER_RETRY_ERROR(_code) \
((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || (_code) == TSDB_CODE_SCH_TIMEOUT_ERROR) ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || \
(_code) == TSDB_CODE_SCH_TIMEOUT_ERROR)
#define REQUEST_TOTAL_EXEC_TIMES 2 #define REQUEST_TOTAL_EXEC_TIMES 2
@ -312,3 +317,4 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
#endif #endif
#endif /*_TD_QUERY_H_*/ #endif /*_TD_QUERY_H_*/
// clang-foramt on

5
include/os/osSocket.h
View File

@ -157,7 +157,10 @@ int32_t taosNonblockwrite(TdSocketPtr pSocket, char *ptr, int32_t nbytes);
int64_t taosCopyFds(TdSocketPtr pSrcSocket, TdSocketPtr pDestSocket, int64_t len); int64_t taosCopyFds(TdSocketPtr pSrcSocket, TdSocketPtr pDestSocket, int64_t len);
void taosWinSocketInit(); void taosWinSocketInit();
int taosCreateSocketWithTimeOutOpt(uint32_t conn_timeout_sec); /*
* set timeout(ms)
*/
int32_t taosCreateSocketWithTimeout(uint32_t timeout);
TdSocketPtr taosOpenUdpSocket(uint32_t localIp, uint16_t localPort); TdSocketPtr taosOpenUdpSocket(uint32_t localIp, uint16_t localPort);
TdSocketPtr taosOpenTcpClientSocket(uint32_t ip, uint16_t port, uint32_t localIp); TdSocketPtr taosOpenTcpClientSocket(uint32_t ip, uint16_t port, uint32_t localIp);

8
include/util/ttrace.h
View File

@ -45,9 +45,11 @@ typedef struct STraceId {
#define TRACE_GET_MSGID(traceId) (traceId)->msgId #define TRACE_GET_MSGID(traceId) (traceId)->msgId
#define TRACE_TO_STR(traceId, buf) \ #define TRACE_TO_STR(traceId, buf) \
do { \ do { \
sprintf(buf, "0x%" PRIx64 ":0x%" PRIx64 "", traceId->rootId, traceId->msgId); \ int64_t rootId = (traceId) != NULL ? (traceId)->rootId : 0; \
int64_t msgId = (traceId) != NULL ? (traceId)->msgId : 0; \
sprintf(buf, "0x%" PRIx64 ":0x%" PRIx64 "", rootId, msgId); \
} while (0) } while (0)
#ifdef __cplusplus #ifdef __cplusplus

34
source/client/src/clientEnv.c
View File

@ -13,11 +13,11 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "os.h"
#include "catalog.h" #include "catalog.h"
#include "functionMgt.h"
#include "clientInt.h" #include "clientInt.h"
#include "clientLog.h" #include "clientLog.h"
#include "functionMgt.h"
#include "os.h"
#include "query.h" #include "query.h"
#include "scheduler.h" #include "scheduler.h"
#include "tcache.h" #include "tcache.h"
@ -38,7 +38,7 @@ static TdThreadOnce tscinit = PTHREAD_ONCE_INIT;
volatile int32_t tscInitRes = 0; volatile int32_t tscInitRes = 0;
static void registerRequest(SRequestObj *pRequest) { static void registerRequest(SRequestObj *pRequest) {
STscObj *pTscObj = acquireTscObj(*(int64_t*)pRequest->pTscObj->id); STscObj *pTscObj = acquireTscObj(*(int64_t *)pRequest->pTscObj->id);
assert(pTscObj != NULL); assert(pTscObj != NULL);
@ -54,14 +54,14 @@ static void registerRequest(SRequestObj *pRequest) {
int32_t currentInst = atomic_add_fetch_64((int64_t *)&pSummary->currentRequests, 1); int32_t currentInst = atomic_add_fetch_64((int64_t *)&pSummary->currentRequests, 1);
tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64 tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64
", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64, ", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64,
pRequest->self, *(int64_t*)pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId); pRequest->self, *(int64_t *)pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId);
} }
} }
static void deregisterRequest(SRequestObj *pRequest) { static void deregisterRequest(SRequestObj *pRequest) {
assert(pRequest != NULL); assert(pRequest != NULL);
STscObj *pTscObj = pRequest->pTscObj; STscObj * pTscObj = pRequest->pTscObj;
SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary; SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1); int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1);
@ -70,8 +70,8 @@ static void deregisterRequest(SRequestObj *pRequest) {
int64_t duration = taosGetTimestampUs() - pRequest->metric.start; int64_t duration = taosGetTimestampUs() - pRequest->metric.start;
tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%" PRIu64 tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%" PRIu64
" ms, current:%d, app current:%d", " ms, current:%d, app current:%d",
pRequest->self, *(int64_t*)pTscObj->id, pRequest->requestId, duration / 1000, num, currentInst); pRequest->self, *(int64_t *)pTscObj->id, pRequest->requestId, duration / 1000, num, currentInst);
releaseTscObj(*(int64_t*)pTscObj->id); releaseTscObj(*(int64_t *)pTscObj->id);
} }
// todo close the transporter properly // todo close the transporter properly
@ -80,12 +80,13 @@ void closeTransporter(STscObj *pTscObj) {
return; return;
} }
tscDebug("free transporter:%p in connObj: 0x%" PRIx64, pTscObj->pAppInfo->pTransporter, *(int64_t*)pTscObj->id); tscDebug("free transporter:%p in connObj: 0x%" PRIx64, pTscObj->pAppInfo->pTransporter, *(int64_t *)pTscObj->id);
rpcClose(pTscObj->pAppInfo->pTransporter); rpcClose(pTscObj->pAppInfo->pTransporter);
} }
static bool clientRpcRfp(int32_t code) { static bool clientRpcRfp(int32_t code) {
if (code == TSDB_CODE_RPC_REDIRECT) { if (code == TSDB_CODE_RPC_REDIRECT || code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_NODE_NOT_DEPLOYED ||
code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_APP_NOT_READY) {
return true; return true;
} else { } else {
return false; return false;
@ -128,16 +129,17 @@ void closeAllRequests(SHashObj *pRequests) {
void destroyTscObj(void *pObj) { void destroyTscObj(void *pObj) {
STscObj *pTscObj = pObj; STscObj *pTscObj = pObj;
SClientHbKey connKey = {.tscRid = *(int64_t*)pTscObj->id, .connType = pTscObj->connType}; SClientHbKey connKey = {.tscRid = *(int64_t *)pTscObj->id, .connType = pTscObj->connType};
hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey); hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey);
int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1); int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
closeAllRequests(pTscObj->pRequests); closeAllRequests(pTscObj->pRequests);
schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter); schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter);
if (0 == connNum) { if (0 == connNum) {
// TODO // TODO
//closeTransporter(pTscObj); // closeTransporter(pTscObj);
} }
tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, *(int64_t*)pTscObj->id, pTscObj->pAppInfo->numOfConns); tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, *(int64_t *)pTscObj->id,
pTscObj->pAppInfo->numOfConns);
taosThreadMutexDestroy(&pTscObj->mutex); taosThreadMutexDestroy(&pTscObj->mutex);
taosMemoryFreeClear(pTscObj); taosMemoryFreeClear(pTscObj);
} }
@ -167,10 +169,10 @@ void *createTscObj(const char *user, const char *auth, const char *db, int32_t c
taosThreadMutexInit(&pObj->mutex, NULL); taosThreadMutexInit(&pObj->mutex, NULL);
pObj->id = taosMemoryMalloc(sizeof(int64_t)); pObj->id = taosMemoryMalloc(sizeof(int64_t));
*(int64_t*)pObj->id = taosAddRef(clientConnRefPool, pObj); *(int64_t *)pObj->id = taosAddRef(clientConnRefPool, pObj);
pObj->schemalessType = 1; pObj->schemalessType = 1;
tscDebug("connObj created, 0x%" PRIx64, *(int64_t*)pObj->id); tscDebug("connObj created, 0x%" PRIx64, *(int64_t *)pObj->id);
return pObj; return pObj;
} }
@ -325,7 +327,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
return 0; return 0;
} }
SConfig *pCfg = taosGetCfg(); SConfig * pCfg = taosGetCfg();
SConfigItem *pItem = NULL; SConfigItem *pItem = NULL;
switch (option) { switch (option) {

26
source/client/src/clientImpl.c
View File

@ -617,12 +617,12 @@ int32_t scheduleAsyncQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNod
.requestId = pRequest->requestId, .requestId = pRequest->requestId,
.requestObjRefId = pRequest->self}; .requestObjRefId = pRequest->self};
SSchedulerReq req = {.pConn = &conn, SSchedulerReq req = {.pConn = &conn,
.pNodeList = pNodeList, .pNodeList = pNodeList,
.pDag = pDag, .pDag = pDag,
.sql = pRequest->sqlstr, .sql = pRequest->sqlstr,
.startTs = pRequest->metric.start, .startTs = pRequest->metric.start,
.fp = schdExecCallback, .fp = schdExecCallback,
.cbParam = &res}; .cbParam = &res};
int32_t code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob); int32_t code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
@ -669,13 +669,13 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
.requestId = pRequest->requestId, .requestId = pRequest->requestId,
.requestObjRefId = pRequest->self}; .requestObjRefId = pRequest->self};
SSchedulerReq req = {.pConn = &conn, SSchedulerReq req = {.pConn = &conn,
.pNodeList = pNodeList, .pNodeList = pNodeList,
.pDag = pDag, .pDag = pDag,
.sql = pRequest->sqlstr, .sql = pRequest->sqlstr,
.startTs = pRequest->metric.start, .startTs = pRequest->metric.start,
.fp = NULL, .fp = NULL,
.cbParam = NULL, .cbParam = NULL,
.reqKilled = &pRequest->killed}; .reqKilled = &pRequest->killed};
int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob, &res); int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob, &res);
pRequest->body.resInfo.execRes = res.res; pRequest->body.resInfo.execRes = res.res;

17
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -70,9 +70,9 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
} }
static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
SDnodeTrans *pTrans = &pDnode->trans; SDnodeTrans * pTrans = &pDnode->trans;
int32_t code = -1; int32_t code = -1;
SRpcMsg *pMsg = NULL; SRpcMsg * pMsg = NULL;
SMgmtWrapper *pWrapper = NULL; SMgmtWrapper *pWrapper = NULL;
SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)]; SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)];
@ -194,11 +194,11 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) { for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype]; SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
SArray *pArray = (*pWrapper->func.getHandlesFp)(); SArray * pArray = (*pWrapper->func.getHandlesFp)();
if (pArray == NULL) return -1; if (pArray == NULL) return -1;
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) { for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
SMgmtHandle *pMgmt = taosArrayGet(pArray, i); SMgmtHandle * pMgmt = taosArrayGet(pArray, i);
SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)]; SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)];
if (pMgmt->needCheckVgId) { if (pMgmt->needCheckVgId) {
pHandle->needCheckVgId = pMgmt->needCheckVgId; pHandle->needCheckVgId = pMgmt->needCheckVgId;
@ -248,7 +248,14 @@ static inline void dmReleaseHandle(SRpcHandleInfo *pHandle, int8_t type) {
} }
} }
static bool rpcRfp(int32_t code) { return code == TSDB_CODE_RPC_REDIRECT; } static bool rpcRfp(int32_t code) {
if (code == TSDB_CODE_RPC_REDIRECT || code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_NODE_NOT_DEPLOYED ||
code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_APP_NOT_READY) {
return true;
} else {
return false;
}
}
int32_t dmInitClient(SDnode *pDnode) { int32_t dmInitClient(SDnode *pDnode) {
SDnodeTrans *pTrans = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;

40
source/libs/function/src/udfd.c
View File

@ -12,6 +12,8 @@
* You should have received a copy of the GNU Affero General Public License * 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/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
// clang-format off
#include "uv.h" #include "uv.h"
#include "os.h" #include "os.h"
#include "fnLog.h" #include "fnLog.h"
@ -25,6 +27,7 @@
#include "tglobal.h" #include "tglobal.h"
#include "tmsg.h" #include "tmsg.h"
#include "trpc.h" #include "trpc.h"
// clang-foramt on
typedef struct SUdfdContext { typedef struct SUdfdContext {
uv_loop_t * loop; uv_loop_t * loop;
@ -103,12 +106,12 @@ typedef struct SUdfdRpcSendRecvInfo {
uv_sem_t resultSem; uv_sem_t resultSem;
} SUdfdRpcSendRecvInfo; } SUdfdRpcSendRecvInfo;
static void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet); static void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet);
static int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf); static int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf);
static int32_t udfdConnectToMnode(); static int32_t udfdConnectToMnode();
static int32_t udfdLoadUdf(char *udfName, SUdf *udf); static int32_t udfdLoadUdf(char *udfName, SUdf *udf);
static bool udfdRpcRfp(int32_t code); static bool udfdRpcRfp(int32_t code);
static int initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet); static int initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet);
static int32_t udfdOpenClientRpc(); static int32_t udfdOpenClientRpc();
static int32_t udfdCloseClientRpc(); static int32_t udfdCloseClientRpc();
@ -126,19 +129,19 @@ static void udfdUvHandleError(SUdfdUvConn *conn) { uv_close((uv_handle_t *)conn-
static void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf); static void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf);
static void udfdOnNewConnection(uv_stream_t *server, int status); static void udfdOnNewConnection(uv_stream_t *server, int status);
static void udfdIntrSignalHandler(uv_signal_t *handle, int signum); static void udfdIntrSignalHandler(uv_signal_t *handle, int signum);
static int32_t removeListeningPipe(); static int32_t removeListeningPipe();
static void udfdPrintVersion(); static void udfdPrintVersion();
static int32_t udfdParseArgs(int32_t argc, char *argv[]); static int32_t udfdParseArgs(int32_t argc, char *argv[]);
static int32_t udfdInitLog(); static int32_t udfdInitLog();
static void udfdCtrlAllocBufCb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf); static void udfdCtrlAllocBufCb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf);
static void udfdCtrlReadCb(uv_stream_t *q, ssize_t nread, const uv_buf_t *buf); static void udfdCtrlReadCb(uv_stream_t *q, ssize_t nread, const uv_buf_t *buf);
static int32_t udfdUvInit(); static int32_t udfdUvInit();
static void udfdCloseWalkCb(uv_handle_t *handle, void *arg); static void udfdCloseWalkCb(uv_handle_t *handle, void *arg);
static int32_t udfdRun(); static int32_t udfdRun();
static void udfdConnectMnodeThreadFunc(void* args); static void udfdConnectMnodeThreadFunc(void *args);
void udfdProcessRequest(uv_work_t *req) { void udfdProcessRequest(uv_work_t *req) {
SUvUdfWork *uvUdf = (SUvUdfWork *)(req->data); SUvUdfWork *uvUdf = (SUvUdfWork *)(req->data);
@ -401,11 +404,11 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
udf->bufSize = pFuncInfo->bufSize; udf->bufSize = pFuncInfo->bufSize;
char path[PATH_MAX] = {0}; char path[PATH_MAX] = {0};
#ifdef WINDOWS #ifdef WINDOWS
snprintf(path, sizeof(path), "%s%s.dll", TD_TMP_DIR_PATH, pFuncInfo->name); snprintf(path, sizeof(path), "%s%s.dll", TD_TMP_DIR_PATH, pFuncInfo->name);
#else #else
snprintf(path, sizeof(path), "%s/lib%s.so", TD_TMP_DIR_PATH, pFuncInfo->name); snprintf(path, sizeof(path), "%s/lib%s.so", TD_TMP_DIR_PATH, pFuncInfo->name);
#endif #endif
TdFilePtr file = TdFilePtr file =
taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL); taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
if (file == NULL) { if (file == NULL) {
@ -544,7 +547,8 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
return 0; return 0;
} }
static bool udfdRpcRfp(int32_t code) { static bool udfdRpcRfp(int32_t code) {
if (code == TSDB_CODE_RPC_REDIRECT) { if (code == TSDB_CODE_RPC_REDIRECT || code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_NODE_NOT_DEPLOYED ||
code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_APP_NOT_READY) {
return true; return true;
} else { } else {
return false; return false;
@ -652,8 +656,7 @@ void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
buf->base = ctx->inputBuf; buf->base = ctx->inputBuf;
buf->len = ctx->inputCap; buf->len = ctx->inputCap;
} else { } else {
fnError("udfd can not allocate enough memory") fnError("udfd can not allocate enough memory") buf->base = NULL;
buf->base = NULL;
buf->len = 0; buf->len = 0;
} }
} else { } else {
@ -664,8 +667,7 @@ void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
buf->base = ctx->inputBuf + ctx->inputLen; buf->base = ctx->inputBuf + ctx->inputLen;
buf->len = ctx->inputCap - ctx->inputLen; buf->len = ctx->inputCap - ctx->inputLen;
} else { } else {
fnError("udfd can not allocate enough memory") fnError("udfd can not allocate enough memory") buf->base = NULL;
buf->base = NULL;
buf->len = 0; buf->len = 0;
} }
} }
@ -881,7 +883,7 @@ static int32_t udfdRun() {
return 0; return 0;
} }
void udfdConnectMnodeThreadFunc(void* args) { void udfdConnectMnodeThreadFunc(void *args) {
int32_t retryMnodeTimes = 0; int32_t retryMnodeTimes = 0;
int32_t code = 0; int32_t code = 0;
while (retryMnodeTimes++ <= TSDB_MAX_REPLICA) { while (retryMnodeTimes++ <= TSDB_MAX_REPLICA) {
@ -939,7 +941,7 @@ int main(int argc, char *argv[]) {
uv_thread_create(&mnodeConnectThread, udfdConnectMnodeThreadFunc, NULL); uv_thread_create(&mnodeConnectThread, udfdConnectMnodeThreadFunc, NULL);
udfdRun(); udfdRun();
removeListeningPipe(); removeListeningPipe();
udfdCloseClientRpc(); udfdCloseClientRpc();

44
source/libs/qcom/src/queryUtil.c
View File

@ -19,6 +19,7 @@
#include "tmsg.h" #include "tmsg.h"
#include "trpc.h" #include "trpc.h"
#include "tsched.h" #include "tsched.h"
// clang-format off
#include "cJSON.h" #include "cJSON.h"
#define VALIDNUMOFCOLS(x) ((x) >= TSDB_MIN_COLUMNS && (x) <= TSDB_MAX_COLUMNS) #define VALIDNUMOFCOLS(x) ((x) >= TSDB_MIN_COLUMNS && (x) <= TSDB_MAX_COLUMNS)
@ -147,13 +148,15 @@ int32_t asyncSendMsgToServerExt(void* pTransporter, SEpSet* epSet, int64_t* pTra
} }
memcpy(pMsg, pInfo->msgInfo.pData, pInfo->msgInfo.len); memcpy(pMsg, pInfo->msgInfo.pData, pInfo->msgInfo.len);
SRpcMsg rpcMsg = {.msgType = pInfo->msgType, SRpcMsg rpcMsg = {
.pCont = pMsg, .msgType = pInfo->msgType,
.contLen = pInfo->msgInfo.len, .pCont = pMsg,
.info.ahandle = (void*)pInfo, .contLen = pInfo->msgInfo.len,
.info.handle = pInfo->msgInfo.handle, .info.ahandle = (void*)pInfo,
.info.persistHandle = persistHandle, .info.handle = pInfo->msgInfo.handle,
.code = 0}; .info.persistHandle = persistHandle,
.code = 0
};
assert(pInfo->fp != NULL); assert(pInfo->fp != NULL);
TRACE_SET_ROOTID(&rpcMsg.info.traceId, pInfo->requestId); TRACE_SET_ROOTID(&rpcMsg.info.traceId, pInfo->requestId);
rpcSendRequestWithCtx(pTransporter, epSet, &rpcMsg, pTransporterId, rpcCtx); rpcSendRequestWithCtx(pTransporter, epSet, &rpcMsg, pTransporterId, rpcCtx);
@ -221,8 +224,9 @@ void destroyQueryExecRes(SQueryExecRes* pRes) {
qError("invalid exec result for request type %d", pRes->msgType); qError("invalid exec result for request type %d", pRes->msgType);
} }
} }
// clang-format on
int32_t dataConverToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *len) { int32_t dataConverToStr(char* str, int type, void* buf, int32_t bufSize, int32_t* len) {
int32_t n = 0; int32_t n = 0;
switch (type) { switch (type) {
@ -262,7 +266,7 @@ int32_t dataConverToStr(char *str, int type, void *buf, int32_t bufSize, int32_t
case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR: case TSDB_DATA_TYPE_NCHAR:
if (bufSize < 0) { if (bufSize < 0) {
// tscError("invalid buf size"); // tscError("invalid buf size");
return TSDB_CODE_TSC_INVALID_VALUE; return TSDB_CODE_TSC_INVALID_VALUE;
} }
@ -289,7 +293,7 @@ int32_t dataConverToStr(char *str, int type, void *buf, int32_t bufSize, int32_t
break; break;
default: default:
// tscError("unsupported type:%d", type); // tscError("unsupported type:%d", type);
return TSDB_CODE_TSC_INVALID_VALUE; return TSDB_CODE_TSC_INVALID_VALUE;
} }
@ -332,7 +336,7 @@ char* parseTagDatatoJson(void* p) {
int32_t length = taosUcs4ToMbs((TdUcs4*)pTagVal->pData, pTagVal->nData, tagJsonValue); int32_t length = taosUcs4ToMbs((TdUcs4*)pTagVal->pData, pTagVal->nData, tagJsonValue);
if (length < 0) { if (length < 0) {
qError("charset:%s to %s. val:%s convert json value failed.", DEFAULT_UNICODE_ENCODEC, tsCharset, qError("charset:%s to %s. val:%s convert json value failed.", DEFAULT_UNICODE_ENCODEC, tsCharset,
pTagVal->pData); pTagVal->pData);
taosMemoryFree(tagJsonValue); taosMemoryFree(tagJsonValue);
goto end; goto end;
} }
@ -372,7 +376,6 @@ end:
return string; return string;
} }
int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) { int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) {
if (NULL == pSrc) { if (NULL == pSrc) {
*pDst = NULL; *pDst = NULL;
@ -393,37 +396,36 @@ int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst) {
*pDst = NULL; *pDst = NULL;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
*pDst = taosMemoryMalloc(sizeof(*pSrc)); *pDst = taosMemoryMalloc(sizeof(*pSrc));
if (NULL == *pDst) { if (NULL == *pDst) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
memcpy(*pDst, pSrc, sizeof(*pSrc)); memcpy(*pDst, pSrc, sizeof(*pSrc));
if (pSrc->vgHash) { if (pSrc->vgHash) {
(*pDst)->vgHash = taosHashInit(taosHashGetSize(pSrc->vgHash), taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK); (*pDst)->vgHash = taosHashInit(taosHashGetSize(pSrc->vgHash), taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true,
HASH_ENTRY_LOCK);
if (NULL == (*pDst)->vgHash) { if (NULL == (*pDst)->vgHash) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
SVgroupInfo* vgInfo = NULL; SVgroupInfo* vgInfo = NULL;
void *pIter = taosHashIterate(pSrc->vgHash, NULL); void* pIter = taosHashIterate(pSrc->vgHash, NULL);
while (pIter) { while (pIter) {
vgInfo = pIter; vgInfo = pIter;
int32_t* vgId = taosHashGetKey(pIter, NULL); int32_t* vgId = taosHashGetKey(pIter, NULL);
if (0 != taosHashPut((*pDst)->vgHash, vgId, sizeof(*vgId), vgInfo, sizeof(*vgInfo))) { if (0 != taosHashPut((*pDst)->vgHash, vgId, sizeof(*vgId), vgInfo, sizeof(*vgInfo))) {
qError("taosHashPut failed, vgId:%d", vgInfo->vgId); qError("taosHashPut failed, vgId:%d", vgInfo->vgId);
taosHashCancelIterate(pSrc->vgHash, pIter); taosHashCancelIterate(pSrc->vgHash, pIter);
taosHashCleanup((*pDst)->vgHash); taosHashCleanup((*pDst)->vgHash);
taosMemoryFreeClear(*pDst); taosMemoryFreeClear(*pDst);
return TSDB_CODE_CTG_MEM_ERROR; return TSDB_CODE_CTG_MEM_ERROR;
} }
pIter = taosHashIterate(pSrc->vgHash, pIter); pIter = taosHashIterate(pSrc->vgHash, pIter);
} }
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

58
source/libs/transport/inc/transComm.h
View File

@ -105,6 +105,13 @@ typedef SRpcCtxVal STransCtxVal;
typedef SRpcInfo STrans; typedef SRpcInfo STrans;
typedef SRpcConnInfo STransHandleInfo; typedef SRpcConnInfo STransHandleInfo;
// ref mgt
// handle
typedef struct SExHandle {
void* handle;
int64_t refId;
void* pThrd;
} SExHandle;
/*convet from fqdn to ip */ /*convet from fqdn to ip */
typedef struct SCvtAddr { typedef struct SCvtAddr {
char ip[TSDB_FQDN_LEN]; char ip[TSDB_FQDN_LEN];
@ -113,14 +120,15 @@ typedef struct SCvtAddr {
} SCvtAddr; } SCvtAddr;
typedef struct { typedef struct {
SEpSet epSet; // ip list provided by app SEpSet epSet; // ip list provided by app
SEpSet origEpSet; SEpSet origEpSet;
void* ahandle; // handle provided by app void* ahandle; // handle provided by app
tmsg_t msgType; // message type tmsg_t msgType; // message type
int8_t connType; // connection type cli/srv int8_t connType; // connection type cli/srv
int64_t rid; // refId returned by taosAddRef
int8_t retryCount; int8_t retryCnt;
int8_t retryLimit;
// bool setMaxRetry;
STransCtx appCtx; // STransCtx appCtx; //
STransMsg* pRsp; // for synchronous API STransMsg* pRsp; // for synchronous API
tsem_t* pSem; // for synchronous API tsem_t* pSem; // for synchronous API
@ -239,6 +247,32 @@ int transSendAsync(SAsyncPool* pool, queue* mq);
} \ } \
} \ } \
} while (0) } while (0)
#define ASYNC_CHECK_HANDLE(exh1, id) \
do { \
if (id > 0) { \
tTrace("handle step1"); \
SExHandle* exh2 = transAcquireExHandle(refMgt, id); \
if (exh2 == NULL || id != exh2->refId) { \
tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
exh2 ? exh2->refId : 0, id); \
goto _return1; \
} \
} else if (id == 0) { \
tTrace("handle step2"); \
SExHandle* exh2 = transAcquireExHandle(refMgt, id); \
if (exh2 == NULL || id == exh2->refId) { \
tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, id, \
exh2 ? exh2->refId : 0); \
goto _return1; \
} else { \
id = exh1->refId; \
} \
} else if (id < 0) { \
tTrace("handle step3"); \
goto _return2; \
} \
} while (0)
int transInitBuffer(SConnBuffer* buf); int transInitBuffer(SConnBuffer* buf);
int transClearBuffer(SConnBuffer* buf); int transClearBuffer(SConnBuffer* buf);
int transDestroyBuffer(SConnBuffer* buf); int transDestroyBuffer(SConnBuffer* buf);
@ -349,21 +383,13 @@ void transDQDestroy(SDelayQueue* queue);
int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs); int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs);
void transPrintEpSet(SEpSet* pEpSet); // void transPrintEpSet(SEpSet* pEpSet);
bool transEpSetIsEqual(SEpSet* a, SEpSet* b); bool transEpSetIsEqual(SEpSet* a, SEpSet* b);
/* /*
* init global func * init global func
*/ */
void transThreadOnce(); void transThreadOnce();
// ref mgt
// handle
typedef struct SExHandle {
void* handle;
int64_t refId;
void* pThrd;
} SExHandle;
void transInitEnv(); void transInitEnv();
int32_t transOpenExHandleMgt(int size); int32_t transOpenExHandleMgt(int size);
void transCloseExHandleMgt(int32_t mgt); void transCloseExHandleMgt(int32_t mgt);

1
source/libs/transport/src/trans.c
View File

@ -79,6 +79,7 @@ void* rpcOpen(const SRpcInit* pInit) {
return pRpc; return pRpc;
} }
void rpcClose(void* arg) { void rpcClose(void* arg) {
tInfo("start to close rpc");
SRpcInfo* pRpc = (SRpcInfo*)arg; SRpcInfo* pRpc = (SRpcInfo*)arg;
(*taosCloseHandle[pRpc->connType])(pRpc->tcphandle); (*taosCloseHandle[pRpc->connType])(pRpc->tcphandle);
transCloseExHandleMgt(pRpc->refMgt); transCloseExHandleMgt(pRpc->refMgt);

466
source/libs/transport/src/transCli.c
View File

@ -1,4 +1,4 @@
/* * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com> /** Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
* *
* This program is free software: you can use, redistribute, and/or modify * 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 * it under the terms of the GNU Affero General Public License, version 3
@ -25,7 +25,6 @@ typedef struct SCliConn {
uv_write_t writeReq; uv_write_t writeReq;
void* hostThrd; void* hostThrd;
int hThrdIdx;
SConnBuffer readBuf; SConnBuffer readBuf;
STransQueue cliMsgs; STransQueue cliMsgs;
@ -36,6 +35,7 @@ typedef struct SCliConn {
bool broken; // link broken or not bool broken; // link broken or not
ConnStatus status; // ConnStatus status; //
int64_t refId;
char* ip; char* ip;
uint32_t port; uint32_t port;
@ -54,7 +54,7 @@ typedef struct SCliMsg {
int sent; //(0: no send, 1: alread sent) int sent; //(0: no send, 1: alread sent)
} SCliMsg; } SCliMsg;
typedef struct SCliThrdObj { typedef struct SCliThrd {
TdThread thread; // tid TdThread thread; // tid
int64_t pid; // pid int64_t pid; // pid
uv_loop_t* loop; uv_loop_t* loop;
@ -72,13 +72,13 @@ typedef struct SCliThrdObj {
SCvtAddr cvtAddr; SCvtAddr cvtAddr;
bool quit; bool quit;
} SCliThrdObj; } SCliThrd;
typedef struct SCliObj { typedef struct SCliObj {
char label[TSDB_LABEL_LEN]; char label[TSDB_LABEL_LEN];
int32_t index; int32_t index;
int numOfThreads; int numOfThreads;
SCliThrdObj** pThreadObj; SCliThrd** pThreadObj;
} SCliObj; } SCliObj;
typedef struct SConnList { typedef struct SConnList {
@ -106,11 +106,18 @@ static void cliAsyncCb(uv_async_t* handle);
static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg); static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg);
static SCliConn* cliCreateConn(SCliThrdObj* thrd); static SCliConn* cliCreateConn(SCliThrd* thrd);
static void cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/); static void cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
static void cliDestroy(uv_handle_t* handle); static void cliDestroy(uv_handle_t* handle);
static void cliSend(SCliConn* pConn); static void cliSend(SCliConn* pConn);
static bool cliIsEpsetUpdated(int32_t code, STransConnCtx* pCtx) {
if (code != 0) return false;
if (pCtx->retryCnt == 0) return false;
if (transEpSetIsEqual(&pCtx->epSet, &pCtx->origEpSet)) return false;
return true;
}
void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr); void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr);
/* /*
* set TCP connection timeout per-socket level * set TCP connection timeout per-socket level
@ -122,14 +129,14 @@ static void cliHandleResp(SCliConn* conn);
static void cliHandleExcept(SCliConn* conn); static void cliHandleExcept(SCliConn* conn);
// handle req from app // handle req from app
static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd);
static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd);
static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd);
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleUpdate(SCliMsg* pMsg, SCliThrd* pThrd);
static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrdObj* pThrd) = {cliHandleReq, cliHandleQuit, cliHandleRelease, static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrd* pThrd) = {cliHandleReq, cliHandleQuit, cliHandleRelease, NULL,
NULL, cliHandleUpdate}; cliHandleUpdate};
static void cliSendQuit(SCliThrdObj* thrd); static void cliSendQuit(SCliThrd* thrd);
static void destroyUserdata(STransMsg* userdata); static void destroyUserdata(STransMsg* userdata);
static int cliRBChoseIdx(STrans* pTransInst); static int cliRBChoseIdx(STrans* pTransInst);
@ -137,8 +144,8 @@ static int cliRBChoseIdx(STrans* pTransInst);
static void destroyCmsg(SCliMsg* cmsg); static void destroyCmsg(SCliMsg* cmsg);
static void transDestroyConnCtx(STransConnCtx* ctx); static void transDestroyConnCtx(STransConnCtx* ctx);
// thread obj // thread obj
static SCliThrdObj* createThrdObj(); static SCliThrd* createThrdObj();
static void destroyThrdObj(SCliThrdObj* pThrd); static void destroyThrdObj(SCliThrd* pThrd);
static void cliWalkCb(uv_handle_t* handle, void* arg); static void cliWalkCb(uv_handle_t* handle, void* arg);
@ -154,7 +161,6 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
destroyCmsg(pMsg); destroyCmsg(pMsg);
} }
} }
#define CLI_RELEASE_UV(loop) \ #define CLI_RELEASE_UV(loop) \
do { \ do { \
uv_walk(loop, cliWalkCb, NULL); \ uv_walk(loop, cliWalkCb, NULL); \
@ -168,17 +174,23 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
snprintf(key, sizeof(key), "%s:%d", ip, (int)port); \ snprintf(key, sizeof(key), "%s:%d", ip, (int)port); \
} while (0) } while (0)
#define CONN_HOST_THREAD_IDX(conn) (conn ? ((SCliConn*)conn)->hThrdIdx : -1) #define CONN_HOST_THREAD_IDX1(idx, exh, refId, pThrd) \
do { \
if (exh == NULL) { \
idx = -1; \
} else { \
ASYNC_CHECK_HANDLE((exh), refId); \
pThrd = (SCliThrd*)(exh)->pThrd; \
} \
} while (0)
#define CONN_PERSIST_TIME(para) (para * 1000 * 10) #define CONN_PERSIST_TIME(para) (para * 1000 * 10)
#define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL) #define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL)
#define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrdObj*)(conn)->hostThrd)->pTransInst))->label) #define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label)
#define CONN_SHOULD_RELEASE(conn, head) \ #define CONN_SHOULD_RELEASE(conn, head) \
do { \ do { \
if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) { \ if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) { \
int connStatus = conn->status; \
uint64_t ahandle = head->ahandle; \ uint64_t ahandle = head->ahandle; \
CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); \ CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); \
conn->status = ConnRelease; \
transClearBuffer(&conn->readBuf); \ transClearBuffer(&conn->readBuf); \
transFreeMsg(transContFromHead((char*)head)); \ transFreeMsg(transContFromHead((char*)head)); \
tDebug("%s conn %p receive release request, ref: %d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn)); \ tDebug("%s conn %p receive release request, ref: %d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn)); \
@ -187,9 +199,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
} \ } \
destroyCmsg(pMsg); \ destroyCmsg(pMsg); \
cliReleaseUnfinishedMsg(conn); \ cliReleaseUnfinishedMsg(conn); \
if (connStatus != ConnInPool) { \ addConnToPool(((SCliThrd*)conn->hostThrd)->pool, conn); \
addConnToPool(((SCliThrdObj*)conn->hostThrd)->pool, conn); \
} \
return; \ return; \
} \ } \
} while (0) } while (0)
@ -255,8 +265,25 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
#define REQUEST_PERSIS_HANDLE(msg) ((msg)->info.persistHandle == 1) #define REQUEST_PERSIS_HANDLE(msg) ((msg)->info.persistHandle == 1)
#define REQUEST_RELEASE_HANDLE(cmsg) ((cmsg)->type == Release) #define REQUEST_RELEASE_HANDLE(cmsg) ((cmsg)->type == Release)
#define EPSET_GET_SIZE(epSet) (epSet)->numOfEps
#define EPSET_GET_INUSE_IP(epSet) ((epSet)->eps[(epSet)->inUse].fqdn) #define EPSET_GET_INUSE_IP(epSet) ((epSet)->eps[(epSet)->inUse].fqdn)
#define EPSET_GET_INUSE_PORT(epSet) ((epSet)->eps[(epSet)->inUse].port) #define EPSET_GET_INUSE_PORT(epSet) ((epSet)->eps[(epSet)->inUse].port)
#define EPSET_FORWARD_INUSE(epSet) \
do { \
(epSet)->inUse = (++((epSet)->inUse)) % ((epSet)->numOfEps); \
} while (0)
#define EPSET_DEBUG_STR(epSet, tbuf) \
do { \
int len = snprintf(tbuf, sizeof(tbuf), "epset:{"); \
for (int i = 0; i < (epSet)->numOfEps; i++) { \
if (i == (epSet)->numOfEps - 1) { \
len += snprintf(tbuf + len, sizeof(tbuf) - len, "%d. %s:%d", i, (epSet)->eps[i].fqdn, (epSet)->eps[i].port); \
} else { \
len += snprintf(tbuf + len, sizeof(tbuf) - len, "%d. %s:%d, ", i, (epSet)->eps[i].fqdn, (epSet)->eps[i].port); \
} \
} \
len += snprintf(tbuf + len, sizeof(tbuf) - len, "}, inUse:%d", (epSet)->inUse); \
} while (0);
static void* cliWorkThread(void* arg); static void* cliWorkThread(void* arg);
@ -271,8 +298,8 @@ _RETURN:
return false; return false;
} }
void cliHandleResp(SCliConn* conn) { void cliHandleResp(SCliConn* conn) {
SCliThrdObj* pThrd = conn->hostThrd; SCliThrd* pThrd = conn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf); STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
pHead->code = htonl(pHead->code); pHead->code = htonl(pHead->code);
@ -292,17 +319,9 @@ void cliHandleResp(SCliConn* conn) {
if (CONN_NO_PERSIST_BY_APP(conn)) { if (CONN_NO_PERSIST_BY_APP(conn)) {
pMsg = transQueuePop(&conn->cliMsgs); pMsg = transQueuePop(&conn->cliMsgs);
pCtx = pMsg ? pMsg->ctx : NULL; pCtx = pMsg->ctx;
if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(conn)) { transMsg.info.ahandle = pCtx->ahandle;
transMsg.info.ahandle = transCtxDumpVal(&conn->ctx, transMsg.msgType); tDebug("%s conn %p get ahandle %p, persist: 0", CONN_GET_INST_LABEL(conn), conn, transMsg.info.ahandle);
if (transMsg.info.ahandle == NULL) {
transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&conn->ctx, (int32_t*)&(transMsg.msgType));
}
tDebug("%s conn %p construct ahandle %p, persist: 0", CONN_GET_INST_LABEL(conn), conn, transMsg.info.ahandle);
} else {
transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL;
tDebug("%s conn %p get ahandle %p, persist: 0", CONN_GET_INST_LABEL(conn), conn, transMsg.info.ahandle);
}
} else { } else {
uint64_t ahandle = (uint64_t)pHead->ahandle; uint64_t ahandle = (uint64_t)pHead->ahandle;
CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle);
@ -324,26 +343,22 @@ void cliHandleResp(SCliConn* conn) {
} }
// buf's mem alread translated to transMsg.pCont // buf's mem alread translated to transMsg.pCont
transClearBuffer(&conn->readBuf); transClearBuffer(&conn->readBuf);
if (!CONN_NO_PERSIST_BY_APP(conn)) { if (!CONN_NO_PERSIST_BY_APP(conn)) {
transMsg.info.handle = conn; transMsg.info.handle = (void*)conn->refId;
tDebug("%s conn %p ref by app", CONN_GET_INST_LABEL(conn), conn); tDebug("%s conn %p ref by app", CONN_GET_INST_LABEL(conn), conn);
} }
// char buf[64] = {0};
// TRACE_TO_STR(&transMsg.info.traceId, buf);
STraceId* trace = &transMsg.info.traceId; STraceId* trace = &transMsg.info.traceId;
tGTrace("conn %p %s received from %s:%d, local info: %s:%d, msg size: %d, code: %d", conn, TMSG_INFO(pHead->msgType), tGTrace("%s conn %p %s received from %s:%d, local info: %s:%d, msg size: %d, code: %d", CONN_GET_INST_LABEL(conn),
taosInetNtoa(conn->addr.sin_addr), ntohs(conn->addr.sin_port), taosInetNtoa(conn->localAddr.sin_addr), conn, TMSG_INFO(pHead->msgType), taosInetNtoa(conn->addr.sin_addr), ntohs(conn->addr.sin_port),
ntohs(conn->localAddr.sin_port), transMsg.contLen, transMsg.code); taosInetNtoa(conn->localAddr.sin_addr), ntohs(conn->localAddr.sin_port), transMsg.contLen, transMsg.code);
if (pCtx == NULL && CONN_NO_PERSIST_BY_APP(conn)) { if (pCtx == NULL && CONN_NO_PERSIST_BY_APP(conn)) {
tDebug("%s except, server continue send while cli ignore it", CONN_GET_INST_LABEL(conn)); tDebug("%s except, conn %p read while cli ignore it", CONN_GET_INST_LABEL(conn), conn);
// transUnrefCliHandle(conn);
return; return;
} }
if (CONN_RELEASE_BY_SERVER(conn) && transMsg.info.ahandle == NULL) { if (CONN_RELEASE_BY_SERVER(conn) && transMsg.info.ahandle == NULL) {
tDebug("%s except, server continue send while cli ignore it", CONN_GET_INST_LABEL(conn)); tDebug("%s except, conn %p read while cli ignore it", CONN_GET_INST_LABEL(conn), conn);
// transUnrefCliHandle(conn);
return; return;
} }
@ -375,9 +390,9 @@ void cliHandleExcept(SCliConn* pConn) {
return; return;
} }
} }
SCliThrdObj* pThrd = pConn->hostThrd; SCliThrd* pThrd = pConn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
bool once = false; bool once = false;
do { do {
SCliMsg* pMsg = transQueuePop(&pConn->cliMsgs); SCliMsg* pMsg = transQueuePop(&pConn->cliMsgs);
if (pMsg == NULL && once) { if (pMsg == NULL && once) {
@ -389,7 +404,6 @@ void cliHandleExcept(SCliConn* pConn) {
transMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL; transMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0; transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0;
transMsg.info.ahandle = NULL; transMsg.info.ahandle = NULL;
transMsg.info.handle = pConn;
if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(pConn)) { if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(pConn)) {
transMsg.info.ahandle = transCtxDumpVal(&pConn->ctx, transMsg.msgType); transMsg.info.ahandle = transCtxDumpVal(&pConn->ctx, transMsg.msgType);
@ -414,15 +428,15 @@ void cliHandleExcept(SCliConn* pConn) {
return; return;
} }
destroyCmsg(pMsg); destroyCmsg(pMsg);
tTrace("%s conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn); tTrace("%s conn %p start to destroy, ref:%d", CONN_GET_INST_LABEL(pConn), pConn, T_REF_VAL_GET(pConn));
} while (!transQueueEmpty(&pConn->cliMsgs)); } while (!transQueueEmpty(&pConn->cliMsgs));
transUnrefCliHandle(pConn); transUnrefCliHandle(pConn);
} }
void cliTimeoutCb(uv_timer_t* handle) { void cliTimeoutCb(uv_timer_t* handle) {
SCliThrdObj* pThrd = handle->data; SCliThrd* pThrd = handle->data;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
int64_t currentTime = pThrd->nextTimeout; int64_t currentTime = pThrd->nextTimeout;
tTrace("%s conn timeout, try to remove expire conn from conn pool", pTransInst->label); tTrace("%s conn timeout, try to remove expire conn from conn pool", pTransInst->label);
SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL); SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL);
@ -487,10 +501,26 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
assert(h == &conn->conn); assert(h == &conn->conn);
return conn; return conn;
} }
static void allocConnRef(SCliConn* conn, bool update) {
if (update) {
transRemoveExHandle(refMgt, conn->refId);
}
SExHandle* exh = taosMemoryCalloc(1, sizeof(SExHandle));
exh->handle = conn;
exh->pThrd = conn->hostThrd;
exh->refId = transAddExHandle(refMgt, exh);
conn->refId = exh->refId;
}
static void addConnToPool(void* pool, SCliConn* conn) { static void addConnToPool(void* pool, SCliConn* conn) {
SCliThrdObj* thrd = conn->hostThrd; if (conn->status == ConnInPool) {
// assert(0);
return;
}
SCliThrd* thrd = conn->hostThrd;
CONN_HANDLE_THREAD_QUIT(thrd); CONN_HANDLE_THREAD_QUIT(thrd);
allocConnRef(conn, true);
STrans* pTransInst = thrd->pTransInst; STrans* pTransInst = thrd->pTransInst;
conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime); conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
transQueueClear(&conn->cliMsgs); transQueueClear(&conn->cliMsgs);
@ -499,7 +529,7 @@ static void addConnToPool(void* pool, SCliConn* conn) {
char key[128] = {0}; char key[128] = {0};
CONN_CONSTRUCT_HASH_KEY(key, conn->ip, conn->port); CONN_CONSTRUCT_HASH_KEY(key, conn->ip, conn->port);
tTrace("%s conn %p added to conn pool, read buf cap: %d", CONN_GET_INST_LABEL(conn), conn, conn->readBuf.cap); tTrace("%s conn %p added to conn pool, read buf cap:%d", CONN_GET_INST_LABEL(conn), conn, conn->readBuf.cap);
SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key)); SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
// list already create before // list already create before
@ -540,13 +570,14 @@ static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
return; return;
} }
if (nread < 0) { if (nread < 0) {
tError("%s conn %p read error: %s", CONN_GET_INST_LABEL(conn), conn, uv_err_name(nread)); tError("%s conn %p read error: %s, ref: %d", CONN_GET_INST_LABEL(conn), conn, uv_err_name(nread),
T_REF_VAL_GET(conn));
conn->broken = true; conn->broken = true;
cliHandleExcept(conn); cliHandleExcept(conn);
} }
} }
static SCliConn* cliCreateConn(SCliThrdObj* pThrd) { static SCliConn* cliCreateConn(SCliThrd* pThrd) {
SCliConn* conn = taosMemoryCalloc(1, sizeof(SCliConn)); SCliConn* conn = taosMemoryCalloc(1, sizeof(SCliConn));
// read/write stream handle // read/write stream handle
conn->stream = (uv_stream_t*)taosMemoryMalloc(sizeof(uv_tcp_t)); conn->stream = (uv_stream_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
@ -562,11 +593,16 @@ static SCliConn* cliCreateConn(SCliThrdObj* pThrd) {
conn->status = ConnNormal; conn->status = ConnNormal;
conn->broken = 0; conn->broken = 0;
transRefCliHandle(conn); transRefCliHandle(conn);
allocConnRef(conn, false);
return conn; return conn;
} }
static void cliDestroyConn(SCliConn* conn, bool clear) { static void cliDestroyConn(SCliConn* conn, bool clear) {
tTrace("%s conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn); tTrace("%s conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn);
QUEUE_REMOVE(&conn->conn); QUEUE_REMOVE(&conn->conn);
QUEUE_INIT(&conn->conn);
transRemoveExHandle(refMgt, conn->refId);
if (clear) { if (clear) {
uv_close((uv_handle_t*)conn->stream, cliDestroy); uv_close((uv_handle_t*)conn->stream, cliDestroy);
} }
@ -593,7 +629,7 @@ static bool cliHandleNoResp(SCliConn* conn) {
} }
if (res == true) { if (res == true) {
if (cliMaySendCachedMsg(conn) == false) { if (cliMaySendCachedMsg(conn) == false) {
SCliThrdObj* thrd = conn->hostThrd; SCliThrd* thrd = conn->hostThrd;
addConnToPool(thrd->pool, conn); addConnToPool(thrd->pool, conn);
} }
} }
@ -629,8 +665,8 @@ void cliSend(SCliConn* pConn) {
STransConnCtx* pCtx = pCliMsg->ctx; STransConnCtx* pCtx = pCliMsg->ctx;
SCliThrdObj* pThrd = pConn->hostThrd; SCliThrd* pThrd = pConn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
STransMsg* pMsg = (STransMsg*)(&pCliMsg->msg); STransMsg* pMsg = (STransMsg*)(&pCliMsg->msg);
if (pMsg->pCont == 0) { if (pMsg->pCont == 0) {
@ -651,12 +687,10 @@ void cliSend(SCliConn* pConn) {
uv_buf_t wb = uv_buf_init((char*)pHead, msgLen); uv_buf_t wb = uv_buf_init((char*)pHead, msgLen);
// char buf[64] = {0};
// TRACE_TO_STR(&pMsg->info.traceId, buf);
STraceId* trace = &pMsg->info.traceId; STraceId* trace = &pMsg->info.traceId;
tGTrace("conn %p %s is sent to %s:%d, local info %s:%d", pConn, TMSG_INFO(pHead->msgType), tGTrace("%s conn %p %s is sent to %s:%d, local info %s:%d", CONN_GET_INST_LABEL(pConn), pConn,
taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->localAddr.sin_addr), TMSG_INFO(pHead->msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
ntohs(pConn->localAddr.sin_port)); taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port));
if (pHead->persist == 1) { if (pHead->persist == 1) {
CONN_SET_PERSIST_BY_APP(pConn); CONN_SET_PERSIST_BY_APP(pConn);
@ -664,7 +698,6 @@ void cliSend(SCliConn* pConn) {
pConn->writeReq.data = pConn; pConn->writeReq.data = pConn;
uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb); uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb);
return; return;
_RETURN: _RETURN:
return; return;
@ -690,19 +723,24 @@ void cliConnCb(uv_connect_t* req, int status) {
cliSend(pConn); cliSend(pConn);
} }
static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) { static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd) {
pThrd->quit = true;
tDebug("cli work thread %p start to quit", pThrd); tDebug("cli work thread %p start to quit", pThrd);
destroyCmsg(pMsg); destroyCmsg(pMsg);
destroyConnPool(pThrd->pool); destroyConnPool(pThrd->pool);
uv_timer_stop(&pThrd->timer); uv_timer_stop(&pThrd->timer);
uv_walk(pThrd->loop, cliWalkCb, NULL); uv_walk(pThrd->loop, cliWalkCb, NULL);
pThrd->quit = true;
// uv_stop(pThrd->loop); // uv_stop(pThrd->loop);
} }
static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) { static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
SCliConn* conn = pMsg->msg.info.handle; int64_t refId = (int64_t)(pMsg->msg.info.handle);
SExHandle* exh = transAcquireExHandle(refMgt, refId);
if (exh == NULL) {
tDebug("%" PRId64 " already release", refId);
}
SCliConn* conn = exh->handle;
tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn); tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn);
if (T_REF_VAL_GET(conn) == 2) { if (T_REF_VAL_GET(conn) == 2) {
@ -711,33 +749,37 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) {
return; return;
} }
cliSend(conn); cliSend(conn);
} else {
// conn already broken down
transUnrefCliHandle(conn);
} }
} }
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd) { static void cliHandleUpdate(SCliMsg* pMsg, SCliThrd* pThrd) {
STransConnCtx* pCtx = pMsg->ctx; STransConnCtx* pCtx = pMsg->ctx;
pThrd->cvtAddr = pCtx->cvtAddr; pThrd->cvtAddr = pCtx->cvtAddr;
destroyCmsg(pMsg); destroyCmsg(pMsg);
} }
SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrd* pThrd, bool* ignore) {
SCliConn* conn = NULL; SCliConn* conn = NULL;
if (pMsg->msg.info.handle != NULL) { int64_t refId = (int64_t)(pMsg->msg.info.handle);
conn = (SCliConn*)(pMsg->msg.info.handle); if (refId != 0) {
if (conn != NULL) { SExHandle* exh = transAcquireExHandle(refMgt, refId);
tTrace("%s conn %p reused", CONN_GET_INST_LABEL(conn), conn); if (exh == NULL) {
} *ignore = true;
} else { destroyCmsg(pMsg);
STransConnCtx* pCtx = pMsg->ctx; return NULL;
conn = getConnFromPool(pThrd->pool, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet)); // assert(0);
if (conn != NULL) {
tTrace("%s conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn);
} else { } else {
tTrace("%s not found conn in conn pool %p", ((STrans*)pThrd->pTransInst)->label, pThrd->pool); conn = exh->handle;
transReleaseExHandle(refMgt, refId);
} }
return conn;
};
STransConnCtx* pCtx = pMsg->ctx;
conn = getConnFromPool(pThrd->pool, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet));
if (conn != NULL) {
tTrace("%s conn %p get from conn pool:%p", CONN_GET_INST_LABEL(conn), conn, pThrd->pool);
} else {
tTrace("%s not found conn in conn pool:%p", ((STrans*)pThrd->pTransInst)->label, pThrd->pool);
} }
return conn; return conn;
} }
@ -752,22 +794,19 @@ void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr) {
} }
} }
} }
void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
uint64_t et = taosGetTimestampUs();
uint64_t el = et - pMsg->st;
// tTrace("%s cli msg tran time cost: %" PRIu64 "us", ((STrans*)pThrd->pTransInst)->label, el);
STransConnCtx* pCtx = pMsg->ctx; STransConnCtx* pCtx = pMsg->ctx;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr); cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr);
transPrintEpSet(&pCtx->epSet); // transPrintEpSet(&pCtx->epSet);
bool ignore = false;
SCliConn* conn = cliGetConn(pMsg, pThrd); SCliConn* conn = cliGetConn(pMsg, pThrd, &ignore);
if (ignore == true) {
return;
}
if (conn != NULL) { if (conn != NULL) {
conn->hThrdIdx = pCtx->hThrdIdx;
transCtxMerge(&conn->ctx, &pCtx->appCtx); transCtxMerge(&conn->ctx, &pCtx->appCtx);
transQueuePush(&conn->cliMsgs, pMsg); transQueuePush(&conn->cliMsgs, pMsg);
cliSend(conn); cliSend(conn);
@ -776,7 +815,6 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
transCtxMerge(&conn->ctx, &pCtx->appCtx); transCtxMerge(&conn->ctx, &pCtx->appCtx);
transQueuePush(&conn->cliMsgs, pMsg); transQueuePush(&conn->cliMsgs, pMsg);
conn->hThrdIdx = pCtx->hThrdIdx;
conn->ip = strdup(EPSET_GET_INUSE_IP(&pCtx->epSet)); conn->ip = strdup(EPSET_GET_INUSE_IP(&pCtx->epSet));
conn->port = EPSET_GET_INUSE_PORT(&pCtx->epSet); conn->port = EPSET_GET_INUSE_PORT(&pCtx->epSet);
@ -784,7 +822,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
if (ret) { if (ret) {
tError("%s conn %p failed to set conn option, errmsg %s", transLabel(pTransInst), conn, uv_err_name(ret)); tError("%s conn %p failed to set conn option, errmsg %s", transLabel(pTransInst), conn, uv_err_name(ret));
} }
int fd = taosCreateSocketWithTimeOutOpt(TRANS_CONN_TIMEOUT); int32_t fd = taosCreateSocketWithTimeout(TRANS_CONN_TIMEOUT);
if (fd == -1) { if (fd == -1) {
tTrace("%s conn %p failed to create socket", transLabel(pTransInst), conn); tTrace("%s conn %p failed to create socket", transLabel(pTransInst), conn);
cliHandleExcept(conn); cliHandleExcept(conn);
@ -807,9 +845,9 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
} }
} }
static void cliAsyncCb(uv_async_t* handle) { static void cliAsyncCb(uv_async_t* handle) {
SAsyncItem* item = handle->data; SAsyncItem* item = handle->data;
SCliThrdObj* pThrd = item->pThrd; SCliThrd* pThrd = item->pThrd;
SCliMsg* pMsg = NULL; SCliMsg* pMsg = NULL;
// batch process to avoid to lock/unlock frequently // batch process to avoid to lock/unlock frequently
queue wq; queue wq;
@ -835,7 +873,7 @@ static void cliAsyncCb(uv_async_t* handle) {
} }
static void* cliWorkThread(void* arg) { static void* cliWorkThread(void* arg) {
SCliThrdObj* pThrd = (SCliThrdObj*)arg; SCliThrd* pThrd = (SCliThrd*)arg;
pThrd->pid = taosGetSelfPthreadId(); pThrd->pid = taosGetSelfPthreadId();
setThreadName("trans-cli-work"); setThreadName("trans-cli-work");
uv_run(pThrd->loop, UV_RUN_DEFAULT); uv_run(pThrd->loop, UV_RUN_DEFAULT);
@ -848,10 +886,10 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
STrans* pTransInst = shandle; STrans* pTransInst = shandle;
memcpy(cli->label, label, strlen(label)); memcpy(cli->label, label, strlen(label));
cli->numOfThreads = numOfThreads; cli->numOfThreads = numOfThreads;
cli->pThreadObj = (SCliThrdObj**)taosMemoryCalloc(cli->numOfThreads, sizeof(SCliThrdObj*)); cli->pThreadObj = (SCliThrd**)taosMemoryCalloc(cli->numOfThreads, sizeof(SCliThrd*));
for (int i = 0; i < cli->numOfThreads; i++) { for (int i = 0; i < cli->numOfThreads; i++) {
SCliThrdObj* pThrd = createThrdObj(); SCliThrd* pThrd = createThrdObj();
pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime); pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
pThrd->pTransInst = shandle; pThrd->pTransInst = shandle;
@ -885,8 +923,8 @@ static void destroyCmsg(SCliMsg* pMsg) {
taosMemoryFree(pMsg); taosMemoryFree(pMsg);
} }
static SCliThrdObj* createThrdObj() { static SCliThrd* createThrdObj() {
SCliThrdObj* pThrd = (SCliThrdObj*)taosMemoryCalloc(1, sizeof(SCliThrdObj)); SCliThrd* pThrd = (SCliThrd*)taosMemoryCalloc(1, sizeof(SCliThrd));
QUEUE_INIT(&pThrd->msg); QUEUE_INIT(&pThrd->msg);
taosThreadMutexInit(&pThrd->msgMtx, NULL); taosThreadMutexInit(&pThrd->msgMtx, NULL);
@ -904,7 +942,7 @@ static SCliThrdObj* createThrdObj() {
pThrd->quit = false; pThrd->quit = false;
return pThrd; return pThrd;
} }
static void destroyThrdObj(SCliThrdObj* pThrd) { static void destroyThrdObj(SCliThrd* pThrd) {
if (pThrd == NULL) { if (pThrd == NULL) {
return; return;
} }
@ -925,7 +963,7 @@ static void transDestroyConnCtx(STransConnCtx* ctx) {
taosMemoryFree(ctx); taosMemoryFree(ctx);
} }
void cliSendQuit(SCliThrdObj* thrd) { void cliSendQuit(SCliThrd* thrd) {
// cli can stop gracefully // cli can stop gracefully
SCliMsg* msg = taosMemoryCalloc(1, sizeof(SCliMsg)); SCliMsg* msg = taosMemoryCalloc(1, sizeof(SCliMsg));
msg->type = Quit; msg->type = Quit;
@ -938,7 +976,10 @@ void cliWalkCb(uv_handle_t* handle, void* arg) {
} }
int cliRBChoseIdx(STrans* pTransInst) { int cliRBChoseIdx(STrans* pTransInst) {
int64_t index = pTransInst->index; int8_t index = pTransInst->index;
if (pTransInst->numOfThreads == 0) {
return -1;
}
if (pTransInst->index++ >= pTransInst->numOfThreads) { if (pTransInst->index++ >= pTransInst->numOfThreads) {
pTransInst->index = 0; pTransInst->index = 0;
} }
@ -946,82 +987,71 @@ int cliRBChoseIdx(STrans* pTransInst) {
} }
static void doDelayTask(void* param) { static void doDelayTask(void* param) {
STaskArg* arg = param; STaskArg* arg = param;
SCliMsg* pMsg = arg->param1;
SCliMsg* pMsg = arg->param1; SCliThrd* pThrd = arg->param2;
SCliThrdObj* pThrd = arg->param2;
cliHandleReq(pMsg, pThrd);
taosMemoryFree(arg); taosMemoryFree(arg);
cliHandleReq(pMsg, pThrd);
}
static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
STransConnCtx* pCtx = pMsg->ctx;
STraceId* trace = &pMsg->msg.info.traceId;
char tbuf[256] = {0};
EPSET_DEBUG_STR(&pCtx->epSet, tbuf);
tGTrace("%s retry on next node, use %s, retryCnt:%d, limit:%d", transLabel(pThrd->pTransInst), tbuf,
pCtx->retryCnt + 1, pCtx->retryLimit);
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg;
arg->param2 = pThrd;
transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
}
void cliCompareAndSwap(int8_t* val, int8_t exp, int8_t newVal) {
if (*val != exp) {
*val = newVal;
}
} }
int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
SCliThrdObj* pThrd = pConn->hostThrd; SCliThrd* pThrd = pConn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
if (pMsg == NULL || pMsg->ctx == NULL) { if (pMsg == NULL || pMsg->ctx == NULL) {
tTrace("%s conn %p handle resp", pTransInst->label, pConn); tTrace("%s conn %p handle resp", pTransInst->label, pConn);
pTransInst->cfp(pTransInst->parent, pResp, NULL); pTransInst->cfp(pTransInst->parent, pResp, NULL);
return 0; return 0;
} }
STransConnCtx* pCtx = pMsg->ctx;
SEpSet* pEpSet = &pCtx->epSet;
if (pCtx->retryCount == 0) {
pCtx->origEpSet = pCtx->epSet;
}
/* /*
* upper layer handle retry if code equal TSDB_CODE_RPC_NETWORK_UNAVAIL * no retry
* 1. query conn
* 2. rpc thread already receive quit msg
*/ */
tmsg_t msgType = pCtx->msgType; STransConnCtx* pCtx = pMsg->ctx;
if ((pTransInst->retry != NULL && pEpSet->numOfEps > 1 && (pTransInst->retry(pResp->code))) || int32_t code = pResp->code;
(pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pResp->code == TSDB_CODE_APP_NOT_READY || if (pTransInst->retry != NULL && pTransInst->retry(code)) {
pResp->code == TSDB_CODE_NODE_NOT_DEPLOYED || pResp->code == TSDB_CODE_SYN_NOT_LEADER)) {
pMsg->sent = 0; pMsg->sent = 0;
tTrace("try to send req to next node"); pCtx->retryCnt += 1;
pMsg->st = taosGetTimestampUs(); if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
pCtx->retryCount += 1; cliCompareAndSwap(&pCtx->retryLimit, TRANS_RETRY_COUNT_LIMIT, EPSET_GET_SIZE(&pCtx->epSet) * 3);
if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL && pCtx->setMaxRetry == false) { if (pCtx->retryCnt < pCtx->retryLimit) {
if (pCtx->retryCount < pEpSet->numOfEps * 3) { transUnrefCliHandle(pConn);
pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps; EPSET_FORWARD_INUSE(&pCtx->epSet);
if (pThrd->quit == false) { cliSchedMsgToNextNode(pMsg, pThrd);
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg)); return -1;
arg->param1 = pMsg;
arg->param2 = pThrd;
transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
transPrintEpSet(pEpSet);
tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, pEpSet->numOfEps * 3);
transUnrefCliHandle(pConn);
return -1;
}
} }
} else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) { } else {
pCtx->setMaxRetry = true; cliCompareAndSwap(&pCtx->retryLimit, TRANS_RETRY_COUNT_LIMIT, TRANS_RETRY_COUNT_LIMIT);
if (pResp->contLen == 0) { if (pCtx->retryCnt < pCtx->retryLimit) {
pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps; addConnToPool(pThrd->pool, pConn);
transPrintEpSet(&pCtx->epSet); if (pResp->contLen == 0) {
tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse, EPSET_FORWARD_INUSE(&pCtx->epSet);
pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
} else {
SEpSet epSet = {0};
tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet);
pCtx->epSet = epSet;
transPrintEpSet(&pCtx->epSet);
tTrace("%s use remote epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
}
if (pThrd->quit == false) {
if (pResp->code != TSDB_CODE_RPC_NETWORK_UNAVAIL) {
if (pConn->status != ConnInPool) addConnToPool(pThrd->pool, pConn);
} else { } else {
transUnrefCliHandle(pConn); tDeserializeSEpSet(pResp->pCont, pResp->contLen, &pCtx->epSet);
} }
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg)); transFreeMsg(pResp->pCont);
arg->param1 = pMsg; cliSchedMsgToNextNode(pMsg, pThrd);
arg->param2 = pThrd;
transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
return -1; return -1;
} }
} }
@ -1029,20 +1059,20 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
STraceId* trace = &pResp->info.traceId; STraceId* trace = &pResp->info.traceId;
if (pCtx->pSem != NULL) { if (pCtx->pSem != NULL) {
tGTrace("conn %p(sync) handle resp", pConn); tGTrace("%s conn %p(sync) handle resp", CONN_GET_INST_LABEL(pConn), pConn);
if (pCtx->pRsp == NULL) { if (pCtx->pRsp == NULL) {
tGTrace("conn %p(sync) failed to resp, ignore", pConn); tGTrace("%s conn %p(sync) failed to resp, ignore", CONN_GET_INST_LABEL(pConn), pConn);
} else { } else {
memcpy((char*)pCtx->pRsp, (char*)pResp, sizeof(*pResp)); memcpy((char*)pCtx->pRsp, (char*)pResp, sizeof(*pResp));
} }
tsem_post(pCtx->pSem); tsem_post(pCtx->pSem);
pCtx->pRsp = NULL; pCtx->pRsp = NULL;
} else { } else {
tGTrace("conn %p handle resp", pConn); tGTrace("%s conn %p handle resp", CONN_GET_INST_LABEL(pConn), pConn);
if (pResp->code != 0 || pCtx->retryCount == 0 || transEpSetIsEqual(&pCtx->epSet, &pCtx->origEpSet)) { if (!cliIsEpsetUpdated(code, pCtx)) {
pTransInst->cfp(pTransInst->parent, pResp, NULL); pTransInst->cfp(pTransInst->parent, pResp, NULL);
} else { } else {
pTransInst->cfp(pTransInst->parent, pResp, pEpSet); pTransInst->cfp(pTransInst->parent, pResp, &pCtx->epSet);
} }
} }
return 0; return 0;
@ -1074,38 +1104,58 @@ void transUnrefCliHandle(void* handle) {
return; return;
} }
int ref = T_REF_DEC((SCliConn*)handle); int ref = T_REF_DEC((SCliConn*)handle);
tTrace("%s conn %p ref %d", CONN_GET_INST_LABEL((SCliConn*)handle), handle, ref); tTrace("%s conn %p ref:%d", CONN_GET_INST_LABEL((SCliConn*)handle), handle, ref);
if (ref == 0) { if (ref == 0) {
cliDestroyConn((SCliConn*)handle, true); cliDestroyConn((SCliConn*)handle, true);
} }
} }
SCliThrd* transGetWorkThrdFromHandle(int64_t handle) {
SCliThrd* pThrd = NULL;
SExHandle* exh = transAcquireExHandle(refMgt, handle);
if (exh == NULL) {
return NULL;
}
pThrd = exh->pThrd;
transReleaseExHandle(refMgt, handle);
return pThrd;
}
SCliThrd* transGetWorkThrd(STrans* trans, int64_t handle) {
if (handle == 0) {
int idx = cliRBChoseIdx(trans);
if (idx < 0) return NULL;
return ((SCliObj*)trans->tcphandle)->pThreadObj[idx];
}
return transGetWorkThrdFromHandle(handle);
}
void transReleaseCliHandle(void* handle) { void transReleaseCliHandle(void* handle) {
SCliThrdObj* thrd = CONN_GET_HOST_THREAD(handle); int idx = -1;
if (thrd == NULL) { SCliThrd* pThrd = transGetWorkThrdFromHandle((int64_t)handle);
if (pThrd == NULL) {
return; return;
} }
STransMsg tmsg = {.info.handle = handle}; STransMsg tmsg = {.info.handle = handle};
SCliMsg* cmsg = taosMemoryCalloc(1, sizeof(SCliMsg)); SCliMsg* cmsg = taosMemoryCalloc(1, sizeof(SCliMsg));
cmsg->msg = tmsg; cmsg->msg = tmsg;
cmsg->type = Release; cmsg->type = Release;
transSendAsync(thrd->asyncPool, &cmsg->q); transSendAsync(pThrd->asyncPool, &cmsg->q);
return;
} }
void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) { void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) {
STrans* pTransInst = (STrans*)shandle; STrans* pTransInst = (STrans*)shandle;
int idx = CONN_HOST_THREAD_IDX((SCliConn*)pReq->info.handle); SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle);
if (idx == -1) { if (pThrd == NULL) {
idx = cliRBChoseIdx(pTransInst); transFreeMsg(pReq->pCont);
return;
} }
TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64()); TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64());
STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx)); STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
pCtx->epSet = *pEpSet; pCtx->epSet = *pEpSet;
pCtx->ahandle = pReq->info.ahandle; pCtx->ahandle = pReq->info.ahandle;
pCtx->msgType = pReq->msgType; pCtx->msgType = pReq->msgType;
pCtx->hThrdIdx = idx;
if (ctx != NULL) { if (ctx != NULL) {
pCtx->appCtx = *ctx; pCtx->appCtx = *ctx;
@ -1118,19 +1168,19 @@ void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
cliMsg->st = taosGetTimestampUs(); cliMsg->st = taosGetTimestampUs();
cliMsg->type = Normal; cliMsg->type = Normal;
SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[idx];
STraceId* trace = &pReq->info.traceId; STraceId* trace = &pReq->info.traceId;
tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), thrd->pid, tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), pThrd->pid,
EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle); EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle);
ASSERT(transSendAsync(thrd->asyncPool, &(cliMsg->q)) == 0); ASSERT(transSendAsync(pThrd->asyncPool, &(cliMsg->q)) == 0);
return;
} }
void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMsg* pRsp) { void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMsg* pRsp) {
STrans* pTransInst = (STrans*)shandle; STrans* pTransInst = (STrans*)shandle;
int idx = CONN_HOST_THREAD_IDX(pReq->info.handle); SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle);
if (idx == -1) { if (pThrd == NULL) {
idx = cliRBChoseIdx(pTransInst); transFreeMsg(pReq->pCont);
return;
} }
tsem_t* sem = taosMemoryCalloc(1, sizeof(tsem_t)); tsem_t* sem = taosMemoryCalloc(1, sizeof(tsem_t));
tsem_init(sem, 0, 0); tsem_init(sem, 0, 0);
@ -1139,9 +1189,9 @@ void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransM
STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx)); STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
pCtx->epSet = *pEpSet; pCtx->epSet = *pEpSet;
pCtx->origEpSet = *pEpSet;
pCtx->ahandle = pReq->info.ahandle; pCtx->ahandle = pReq->info.ahandle;
pCtx->msgType = pReq->msgType; pCtx->msgType = pReq->msgType;
pCtx->hThrdIdx = idx;
pCtx->pSem = sem; pCtx->pSem = sem;
pCtx->pRsp = pRsp; pCtx->pRsp = pRsp;
@ -1151,22 +1201,21 @@ void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransM
cliMsg->st = taosGetTimestampUs(); cliMsg->st = taosGetTimestampUs();
cliMsg->type = Normal; cliMsg->type = Normal;
SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[idx];
STraceId* trace = &pReq->info.traceId; STraceId* trace = &pReq->info.traceId;
tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), thrd->pid, tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), pThrd->pid,
EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle); EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle);
transSendAsync(thrd->asyncPool, &(cliMsg->q)); transSendAsync(pThrd->asyncPool, &(cliMsg->q));
tsem_wait(sem); tsem_wait(sem);
tsem_destroy(sem); tsem_destroy(sem);
taosMemoryFree(sem); taosMemoryFree(sem);
return;
} }
/* /*
* *
**/ **/
void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) { void transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
STrans* pTransInst = ahandle; STrans* pTransInst = shandle;
SCvtAddr cvtAddr = {0}; SCvtAddr cvtAddr = {0};
if (ip != NULL && fqdn != NULL) { if (ip != NULL && fqdn != NULL) {
@ -1176,14 +1225,13 @@ void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) {
} }
for (int i = 0; i < pTransInst->numOfThreads; i++) { for (int i = 0; i < pTransInst->numOfThreads; i++) {
STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx)); STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
pCtx->hThrdIdx = i;
pCtx->cvtAddr = cvtAddr; pCtx->cvtAddr = cvtAddr;
SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg)); SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
cliMsg->ctx = pCtx; cliMsg->ctx = pCtx;
cliMsg->type = Update; cliMsg->type = Update;
SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i]; SCliThrd* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i];
tDebug("%s update epset at thread:%08" PRId64 "", pTransInst->label, thrd->pid); tDebug("%s update epset at thread:%08" PRId64 "", pTransInst->label, thrd->pid);
transSendAsync(thrd->asyncPool, &(cliMsg->q)); transSendAsync(thrd->asyncPool, &(cliMsg->q));

6
source/libs/transport/src/transComm.c
View File

@ -455,16 +455,16 @@ void transPrintEpSet(SEpSet* pEpSet) {
return; return;
} }
char buf[512] = {0}; char buf[512] = {0};
int len = snprintf(buf, sizeof(buf), "epset { "); int len = snprintf(buf, sizeof(buf), "epset:{");
for (int i = 0; i < pEpSet->numOfEps; i++) { for (int i = 0; i < pEpSet->numOfEps; i++) {
if (i == pEpSet->numOfEps - 1) { if (i == pEpSet->numOfEps - 1) {
len += snprintf(buf + len, sizeof(buf) - len, "%d. %s:%d ", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); len += snprintf(buf + len, sizeof(buf) - len, "%d. %s:%d", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
} else { } else {
len += snprintf(buf + len, sizeof(buf) - len, "%d. %s:%d, ", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); len += snprintf(buf + len, sizeof(buf) - len, "%d. %s:%d, ", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
} }
} }
len += snprintf(buf + len, sizeof(buf) - len, "}"); len += snprintf(buf + len, sizeof(buf) - len, "}");
tTrace("%s, inUse: %d", buf, pEpSet->inUse); tTrace("%s, inUse:%d", buf, pEpSet->inUse);
} }
bool transEpSetIsEqual(SEpSet* a, SEpSet* b) { bool transEpSetIsEqual(SEpSet* a, SEpSet* b) {
if (a->numOfEps != b->numOfEps || a->inUse != b->inUse) { if (a->numOfEps != b->numOfEps || a->inUse != b->inUse) {

112
source/libs/transport/src/transSvr.c
View File

@ -65,7 +65,7 @@ typedef struct SSvrMsg {
STransMsgType type; STransMsgType type;
} SSvrMsg; } SSvrMsg;
typedef struct SWorkThrdObj { typedef struct SWorkThrd {
TdThread thread; TdThread thread;
uv_connect_t connect_req; uv_connect_t connect_req;
uv_pipe_t* pipe; uv_pipe_t* pipe;
@ -78,7 +78,7 @@ typedef struct SWorkThrdObj {
queue conn; queue conn;
void* pTransInst; void* pTransInst;
bool quit; bool quit;
} SWorkThrdObj; } SWorkThrd;
typedef struct SServerObj { typedef struct SServerObj {
TdThread thread; TdThread thread;
@ -86,10 +86,10 @@ typedef struct SServerObj {
uv_loop_t* loop; uv_loop_t* loop;
// work thread info // work thread info
int workerIdx; int workerIdx;
int numOfThreads; int numOfThreads;
int numOfWorkerReady; int numOfWorkerReady;
SWorkThrdObj** pThreadObj; SWorkThrd** pThreadObj;
uv_pipe_t pipeListen; uv_pipe_t pipeListen;
uv_pipe_t** pipe; uv_pipe_t** pipe;
@ -133,14 +133,14 @@ static SSvrConn* createConn(void* hThrd);
static void destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/); static void destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/);
static void destroyConnRegArg(SSvrConn* conn); static void destroyConnRegArg(SSvrConn* conn);
static int reallocConnRefHandle(SSvrConn* conn); static int reallocConnRef(SSvrConn* conn);
static void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleQuit(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleRelease(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleRelease(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleResp(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleResp(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd);
static void (*transAsyncHandle[])(SSvrMsg* msg, SWorkThrdObj* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease, static void (*transAsyncHandle[])(SSvrMsg* msg, SWorkThrd* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease,
uvHandleRegister, NULL}; uvHandleRegister, NULL};
static int32_t exHandlesMgt; static int32_t exHandlesMgt;
@ -160,7 +160,7 @@ static void* transWorkerThread(void* arg);
static void* transAcceptThread(void* arg); static void* transAcceptThread(void* arg);
// add handle loop // add handle loop
static bool addHandleToWorkloop(SWorkThrdObj* pThrd, char* pipeName); static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName);
static bool addHandleToAcceptloop(void* arg); static bool addHandleToAcceptloop(void* arg);
#define CONN_SHOULD_RELEASE(conn, head) \ #define CONN_SHOULD_RELEASE(conn, head) \
@ -176,7 +176,7 @@ static bool addHandleToAcceptloop(void* arg);
srvMsg->msg = tmsg; \ srvMsg->msg = tmsg; \
srvMsg->type = Release; \ srvMsg->type = Release; \
srvMsg->pConn = conn; \ srvMsg->pConn = conn; \
reallocConnRefHandle(conn); \ reallocConnRef(conn); \
if (!transQueuePush(&conn->srvMsgs, srvMsg)) { \ if (!transQueuePush(&conn->srvMsgs, srvMsg)) { \
return; \ return; \
} \ } \
@ -206,32 +206,6 @@ static bool addHandleToAcceptloop(void* arg);
} \ } \
} while (0) } while (0)
#define ASYNC_CHECK_HANDLE(exh1, refId) \
do { \
if (refId > 0) { \
tTrace("handle step1"); \
SExHandle* exh2 = transAcquireExHandle(refMgt, refId); \
if (exh2 == NULL || refId != exh2->refId) { \
tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
exh2 ? exh2->refId : 0, refId); \
goto _return1; \
} \
} else if (refId == 0) { \
tTrace("handle step2"); \
SExHandle* exh2 = transAcquireExHandle(refMgt, refId); \
if (exh2 == NULL || refId != exh2->refId) { \
tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, refId, \
exh2 ? exh2->refId : 0); \
goto _return1; \
} else { \
refId = exh1->refId; \
} \
} else if (refId < 0) { \
tTrace("handle step3"); \
goto _return2; \
} \
} while (0)
void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
SSvrConn* conn = handle->data; SSvrConn* conn = handle->data;
SConnBuffer* pBuf = &conn->readBuf; SConnBuffer* pBuf = &conn->readBuf;
@ -259,7 +233,7 @@ static void uvHandleReq(SSvrConn* pConn) {
// wreq->data = pConn; // wreq->data = pConn;
// uv_read_stop((uv_stream_t*)pConn->pTcp); // uv_read_stop((uv_stream_t*)pConn->pTcp);
// transRefSrvHandle(pConn); // transRefSrvHandle(pConn);
// uv_queue_work(((SWorkThrdObj*)pConn->hostThrd)->loop, wreq, uvWorkDoTask, uvWorkAfterTask); // uv_queue_work(((SWorkThrd*)pConn->hostThrd)->loop, wreq, uvWorkDoTask, uvWorkAfterTask);
CONN_SHOULD_RELEASE(pConn, pHead); CONN_SHOULD_RELEASE(pConn, pHead);
@ -379,7 +353,7 @@ void uvOnSendCb(uv_write_t* req, int status) {
// if (msg->type == Release && conn->status != ConnNormal) { // if (msg->type == Release && conn->status != ConnNormal) {
// conn->status = ConnNormal; // conn->status = ConnNormal;
// transUnrefSrvHandle(conn); // transUnrefSrvHandle(conn);
// reallocConnRefHandle(conn); // reallocConnRef(conn);
// destroySmsg(msg); // destroySmsg(msg);
// transQueueClear(&conn->srvMsgs); // transQueueClear(&conn->srvMsgs);
// return; // return;
@ -448,6 +422,8 @@ static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
transUnrefSrvHandle(pConn); transUnrefSrvHandle(pConn);
} else { } else {
pHead->msgType = pMsg->msgType; pHead->msgType = pMsg->msgType;
if (pHead->msgType == 0 && transMsgLenFromCont(pMsg->contLen) == sizeof(STransMsgHead))
pHead->msgType = pConn->inType + 1;
} }
} }
@ -504,7 +480,7 @@ static void destroySmsg(SSvrMsg* smsg) {
transFreeMsg(smsg->msg.pCont); transFreeMsg(smsg->msg.pCont);
taosMemoryFree(smsg); taosMemoryFree(smsg);
} }
static void destroyAllConn(SWorkThrdObj* pThrd) { static void destroyAllConn(SWorkThrd* pThrd) {
tTrace("thread %p destroy all conn ", pThrd); tTrace("thread %p destroy all conn ", pThrd);
while (!QUEUE_IS_EMPTY(&pThrd->conn)) { while (!QUEUE_IS_EMPTY(&pThrd->conn)) {
queue* h = QUEUE_HEAD(&pThrd->conn); queue* h = QUEUE_HEAD(&pThrd->conn);
@ -519,10 +495,10 @@ static void destroyAllConn(SWorkThrdObj* pThrd) {
} }
} }
void uvWorkerAsyncCb(uv_async_t* handle) { void uvWorkerAsyncCb(uv_async_t* handle) {
SAsyncItem* item = handle->data; SAsyncItem* item = handle->data;
SWorkThrdObj* pThrd = item->pThrd; SWorkThrd* pThrd = item->pThrd;
SSvrConn* conn = NULL; SSvrConn* conn = NULL;
queue wq; queue wq;
// batch process to avoid to lock/unlock frequently // batch process to avoid to lock/unlock frequently
taosThreadMutexLock(&item->mtx); taosThreadMutexLock(&item->mtx);
@ -650,7 +626,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
assert(buf->base[0] == notify[0]); assert(buf->base[0] == notify[0]);
taosMemoryFree(buf->base); taosMemoryFree(buf->base);
SWorkThrdObj* pThrd = q->data; SWorkThrd* pThrd = q->data;
uv_pipe_t* pipe = (uv_pipe_t*)q; uv_pipe_t* pipe = (uv_pipe_t*)q;
if (!uv_pipe_pending_count(pipe)) { if (!uv_pipe_pending_count(pipe)) {
@ -718,10 +694,10 @@ void uvOnPipeConnectionCb(uv_connect_t* connect, int status) {
if (status != 0) { if (status != 0) {
return; return;
} }
SWorkThrdObj* pThrd = container_of(connect, SWorkThrdObj, connect_req); SWorkThrd* pThrd = container_of(connect, SWorkThrd, connect_req);
uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb); uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
} }
static bool addHandleToWorkloop(SWorkThrdObj* pThrd, char* pipeName) { static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t)); pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
if (0 != uv_loop_init(pThrd->loop)) { if (0 != uv_loop_init(pThrd->loop)) {
return false; return false;
@ -774,14 +750,14 @@ static bool addHandleToAcceptloop(void* arg) {
} }
void* transWorkerThread(void* arg) { void* transWorkerThread(void* arg) {
setThreadName("trans-worker"); setThreadName("trans-worker");
SWorkThrdObj* pThrd = (SWorkThrdObj*)arg; SWorkThrd* pThrd = (SWorkThrd*)arg;
uv_run(pThrd->loop, UV_RUN_DEFAULT); uv_run(pThrd->loop, UV_RUN_DEFAULT);
return NULL; return NULL;
} }
static SSvrConn* createConn(void* hThrd) { static SSvrConn* createConn(void* hThrd) {
SWorkThrdObj* pThrd = hThrd; SWorkThrd* pThrd = hThrd;
SSvrConn* pConn = (SSvrConn*)taosMemoryCalloc(1, sizeof(SSvrConn)); SSvrConn* pConn = (SSvrConn*)taosMemoryCalloc(1, sizeof(SSvrConn));
QUEUE_INIT(&pConn->queue); QUEUE_INIT(&pConn->queue);
@ -826,7 +802,7 @@ static void destroyConnRegArg(SSvrConn* conn) {
conn->regArg.init = 0; conn->regArg.init = 0;
} }
} }
static int reallocConnRefHandle(SSvrConn* conn) { static int reallocConnRef(SSvrConn* conn) {
transReleaseExHandle(refMgt, conn->refId); transReleaseExHandle(refMgt, conn->refId);
transRemoveExHandle(refMgt, conn->refId); transRemoveExHandle(refMgt, conn->refId);
// avoid app continue to send msg on invalid handle // avoid app continue to send msg on invalid handle
@ -844,7 +820,7 @@ static void uvDestroyConn(uv_handle_t* handle) {
if (conn == NULL) { if (conn == NULL) {
return; return;
} }
SWorkThrdObj* thrd = conn->hostThrd; SWorkThrd* thrd = conn->hostThrd;
transReleaseExHandle(refMgt, conn->refId); transReleaseExHandle(refMgt, conn->refId);
transRemoveExHandle(refMgt, conn->refId); transRemoveExHandle(refMgt, conn->refId);
@ -889,7 +865,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
srv->numOfThreads = numOfThreads; srv->numOfThreads = numOfThreads;
srv->workerIdx = 0; srv->workerIdx = 0;
srv->numOfWorkerReady = 0; srv->numOfWorkerReady = 0;
srv->pThreadObj = (SWorkThrdObj**)taosMemoryCalloc(srv->numOfThreads, sizeof(SWorkThrdObj*)); srv->pThreadObj = (SWorkThrd**)taosMemoryCalloc(srv->numOfThreads, sizeof(SWorkThrd*));
srv->pipe = (uv_pipe_t**)taosMemoryCalloc(srv->numOfThreads, sizeof(uv_pipe_t*)); srv->pipe = (uv_pipe_t**)taosMemoryCalloc(srv->numOfThreads, sizeof(uv_pipe_t*));
srv->ip = ip; srv->ip = ip;
srv->port = port; srv->port = port;
@ -914,7 +890,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
assert(0 == uv_listen((uv_stream_t*)&srv->pipeListen, SOMAXCONN, uvPipeListenCb)); assert(0 == uv_listen((uv_stream_t*)&srv->pipeListen, SOMAXCONN, uvPipeListenCb));
for (int i = 0; i < srv->numOfThreads; i++) { for (int i = 0; i < srv->numOfThreads; i++) {
SWorkThrdObj* thrd = (SWorkThrdObj*)taosMemoryCalloc(1, sizeof(SWorkThrdObj)); SWorkThrd* thrd = (SWorkThrd*)taosMemoryCalloc(1, sizeof(SWorkThrd));
thrd->pTransInst = shandle; thrd->pTransInst = shandle;
thrd->quit = false; thrd->quit = false;
srv->pThreadObj[i] = thrd; srv->pThreadObj[i] = thrd;
@ -959,7 +935,7 @@ End:
return NULL; return NULL;
} }
void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleQuit(SSvrMsg* msg, SWorkThrd* thrd) {
thrd->quit = true; thrd->quit = true;
if (QUEUE_IS_EMPTY(&thrd->conn)) { if (QUEUE_IS_EMPTY(&thrd->conn)) {
uv_walk(thrd->loop, uvWalkCb, NULL); uv_walk(thrd->loop, uvWalkCb, NULL);
@ -968,10 +944,10 @@ void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd) {
} }
taosMemoryFree(msg); taosMemoryFree(msg);
} }
void uvHandleRelease(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleRelease(SSvrMsg* msg, SWorkThrd* thrd) {
SSvrConn* conn = msg->pConn; SSvrConn* conn = msg->pConn;
if (conn->status == ConnAcquire) { if (conn->status == ConnAcquire) {
reallocConnRefHandle(conn); reallocConnRef(conn);
if (!transQueuePush(&conn->srvMsgs, msg)) { if (!transQueuePush(&conn->srvMsgs, msg)) {
return; return;
} }
@ -982,12 +958,12 @@ void uvHandleRelease(SSvrMsg* msg, SWorkThrdObj* thrd) {
} }
destroySmsg(msg); destroySmsg(msg);
} }
void uvHandleResp(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleResp(SSvrMsg* msg, SWorkThrd* thrd) {
// send msg to client // send msg to client
tDebug("%s conn %p start to send resp (2/2)", transLabel(thrd->pTransInst), msg->pConn); tDebug("%s conn %p start to send resp (2/2)", transLabel(thrd->pTransInst), msg->pConn);
uvStartSendResp(msg); uvStartSendResp(msg);
} }
void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
SSvrConn* conn = msg->pConn; SSvrConn* conn = msg->pConn;
tDebug("%s conn %p register brokenlink callback", transLabel(thrd->pTransInst), conn); tDebug("%s conn %p register brokenlink callback", transLabel(thrd->pTransInst), conn);
if (conn->status == ConnAcquire) { if (conn->status == ConnAcquire) {
@ -1008,7 +984,7 @@ void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd) {
taosMemoryFree(msg); taosMemoryFree(msg);
} }
} }
void destroyWorkThrd(SWorkThrdObj* pThrd) { void destroyWorkThrd(SWorkThrd* pThrd) {
if (pThrd == NULL) { if (pThrd == NULL) {
return; return;
} }
@ -1019,7 +995,7 @@ void destroyWorkThrd(SWorkThrdObj* pThrd) {
taosMemoryFree(pThrd->loop); taosMemoryFree(pThrd->loop);
taosMemoryFree(pThrd); taosMemoryFree(pThrd);
} }
void sendQuitToWorkThrd(SWorkThrdObj* pThrd) { void sendQuitToWorkThrd(SWorkThrd* pThrd) {
SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg)); SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg));
msg->type = Quit; msg->type = Quit;
tDebug("server send quit msg to work thread"); tDebug("server send quit msg to work thread");
@ -1055,8 +1031,6 @@ void transCloseServer(void* arg) {
int ref = atomic_sub_fetch_32(&tranSSvrInst, 1); int ref = atomic_sub_fetch_32(&tranSSvrInst, 1);
if (ref == 0) { if (ref == 0) {
// TdThreadOnce tmpInit = PTHREAD_ONCE_INIT;
// memcpy(&transModuleInit, &tmpInit, sizeof(TdThreadOnce));
transCloseExHandleMgt(refMgt); transCloseExHandleMgt(refMgt);
} }
} }
@ -1086,7 +1060,7 @@ void transReleaseSrvHandle(void* handle) {
ASYNC_CHECK_HANDLE(exh, refId); ASYNC_CHECK_HANDLE(exh, refId);
SWorkThrdObj* pThrd = exh->pThrd; SWorkThrd* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd); ASYNC_ERR_JRET(pThrd);
STransMsg tmsg = {.code = 0, .info.handle = exh, .info.ahandle = NULL, .info.refId = refId}; STransMsg tmsg = {.code = 0, .info.handle = exh, .info.ahandle = NULL, .info.refId = refId};
@ -1116,7 +1090,7 @@ void transSendResponse(const STransMsg* msg) {
STransMsg tmsg = *msg; STransMsg tmsg = *msg;
tmsg.info.refId = refId; tmsg.info.refId = refId;
SWorkThrdObj* pThrd = exh->pThrd; SWorkThrd* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd); ASYNC_ERR_JRET(pThrd);
SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg)); SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
@ -1146,7 +1120,7 @@ void transRegisterMsg(const STransMsg* msg) {
STransMsg tmsg = *msg; STransMsg tmsg = *msg;
tmsg.info.refId = refId; tmsg.info.refId = refId;
SWorkThrdObj* pThrd = exh->pThrd; SWorkThrd* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd); ASYNC_ERR_JRET(pThrd);
SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg)); SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));

8
source/os/src/osSocket.c
View File

@ -946,7 +946,7 @@ int32_t taosGetFqdn(char *fqdn) {
#endif // __APPLE__ #endif // __APPLE__
int32_t ret = getaddrinfo(hostname, NULL, &hints, &result); int32_t ret = getaddrinfo(hostname, NULL, &hints, &result);
if (!result) { if (!result) {
fprintf(stderr,"failed to get fqdn, code:%d, reason:%s", ret, gai_strerror(ret)); fprintf(stderr, "failed to get fqdn, code:%d, reason:%s", ret, gai_strerror(ret));
return -1; return -1;
} }
@ -1073,7 +1073,7 @@ int32_t taosCloseEpoll(TdEpollPtr *ppEpoll) {
* Set TCP connection timeout per-socket level. * Set TCP connection timeout per-socket level.
* ref [https://github.com/libuv/help/issues/54] * ref [https://github.com/libuv/help/issues/54]
*/ */
int taosCreateSocketWithTimeOutOpt(uint32_t conn_timeout_sec) { int32_t taosCreateSocketWithTimeout(uint32_t timeout) {
#if defined(WINDOWS) #if defined(WINDOWS)
SOCKET fd; SOCKET fd;
#else #else
@ -1083,11 +1083,11 @@ int taosCreateSocketWithTimeOutOpt(uint32_t conn_timeout_sec) {
return -1; return -1;
} }
#if defined(WINDOWS) #if defined(WINDOWS)
if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&conn_timeout_sec, sizeof(conn_timeout_sec))) { if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&timeout, sizeof(timeout))) {
return -1; return -1;
} }
#else // Linux like systems #else // Linux like systems
uint32_t conn_timeout_ms = conn_timeout_sec * 1000; uint32_t conn_timeout_ms = timeout * 1000;
if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) { if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
return -1; return -1;
} }