Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
a558b445
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
a558b445
编写于
4月 27, 2022
作者:
dengyihao
提交者:
GitHub
4月 27, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #11906 from taosdata/feature/support_retry
Feature/support retry
上级
a196de93
731dc4ff
变更
7
隐藏空白更改
内联
并排
Showing
7 changed file
with
108 addition
and
84 deletion
+108
-84
include/libs/transport/trpc.h
include/libs/transport/trpc.h
+6
-2
source/libs/transport/inc/transComm.h
source/libs/transport/inc/transComm.h
+7
-8
source/libs/transport/inc/transportInt.h
source/libs/transport/inc/transportInt.h
+1
-2
source/libs/transport/src/trans.c
source/libs/transport/src/trans.c
+3
-10
source/libs/transport/src/transCli.c
source/libs/transport/src/transCli.c
+87
-50
source/libs/transport/src/transComm.c
source/libs/transport/src/transComm.c
+0
-11
source/libs/transport/src/transSrv.c
source/libs/transport/src/transSrv.c
+4
-1
未找到文件。
include/libs/transport/trpc.h
浏览文件 @
a558b445
...
@@ -59,9 +59,13 @@ typedef struct {
...
@@ -59,9 +59,13 @@ typedef struct {
void
*
pNode
;
void
*
pNode
;
}
SNodeMsg
;
}
SNodeMsg
;
typedef
void
(
*
RpcCfp
)(
void
*
parent
,
SRpcMsg
*
,
SEpSet
*
);
typedef
void
(
*
RpcCfp
)(
void
*
parent
,
SRpcMsg
*
,
SEpSet
*
rf
);
typedef
int
(
*
RpcAfp
)(
void
*
parent
,
char
*
tableId
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
typedef
int
(
*
RpcAfp
)(
void
*
parent
,
char
*
tableId
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
typedef
int
(
*
RpcRfp
)(
void
*
parent
,
SRpcMsg
*
,
SEpSet
*
);
///
// // SRpcMsg code
// REDIERE,
// NOT READY, EpSet
typedef
bool
(
*
RpcRfp
)(
int32_t
code
);
typedef
struct
SRpcInit
{
typedef
struct
SRpcInit
{
uint16_t
localPort
;
// local port
uint16_t
localPort
;
// local port
...
...
source/libs/transport/inc/transComm.h
浏览文件 @
a558b445
...
@@ -103,6 +103,9 @@ typedef void* queue[2];
...
@@ -103,6 +103,9 @@ typedef void* queue[2];
/* Return the structure holding the given element. */
/* Return the structure holding the given element. */
#define QUEUE_DATA(e, type, field) ((type*)((void*)((char*)(e)-offsetof(type, field))))
#define QUEUE_DATA(e, type, field) ((type*)((void*)((char*)(e)-offsetof(type, field))))
#define TRANS_RETRY_COUNT_LIMIT 10 // retry count limit
#define TRANS_RETRY_INTERVAL 5 // ms retry interval
typedef
struct
{
typedef
struct
{
SRpcInfo
*
pRpc
;
// associated SRpcInfo
SRpcInfo
*
pRpc
;
// associated SRpcInfo
SEpSet
epSet
;
// ip list provided by app
SEpSet
epSet
;
// ip list provided by app
...
@@ -137,14 +140,12 @@ typedef struct {
...
@@ -137,14 +140,12 @@ typedef struct {
int8_t
connType
;
// connection type cli/srv
int8_t
connType
;
// connection type cli/srv
int64_t
rid
;
// refId returned by taosAddRef
int64_t
rid
;
// refId returned by taosAddRef
int8_t
retryCount
;
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
int
hThrdIdx
;
int
hThrdIdx
;
char
*
ip
;
uint32_t
port
;
// SEpSet* pSet; // for synchronous API
}
STransConnCtx
;
}
STransConnCtx
;
#pragma pack(push, 1)
#pragma pack(push, 1)
...
@@ -215,8 +216,6 @@ void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
...
@@ -215,8 +216,6 @@ void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
bool
transCompressMsg
(
char
*
msg
,
int32_t
len
,
int32_t
*
flen
);
bool
transCompressMsg
(
char
*
msg
,
int32_t
len
,
int32_t
*
flen
);
bool
transDecompressMsg
(
char
*
msg
,
int32_t
len
,
int32_t
*
flen
);
bool
transDecompressMsg
(
char
*
msg
,
int32_t
len
,
int32_t
*
flen
);
void
transConnCtxDestroy
(
STransConnCtx
*
ctx
);
void
transFreeMsg
(
void
*
msg
);
void
transFreeMsg
(
void
*
msg
);
//
//
...
@@ -262,8 +261,8 @@ void transUnrefCliHandle(void* handle);
...
@@ -262,8 +261,8 @@ void transUnrefCliHandle(void* handle);
void
transReleaseCliHandle
(
void
*
handle
);
void
transReleaseCliHandle
(
void
*
handle
);
void
transReleaseSrvHandle
(
void
*
handle
);
void
transReleaseSrvHandle
(
void
*
handle
);
void
transSendRequest
(
void
*
shandle
,
const
char
*
ip
,
uint32_t
por
t
,
STransMsg
*
pMsg
,
STransCtx
*
pCtx
);
void
transSendRequest
(
void
*
shandle
,
const
SEpSet
*
pEpSe
t
,
STransMsg
*
pMsg
,
STransCtx
*
pCtx
);
void
transSendRecv
(
void
*
shandle
,
const
char
*
ip
,
uint32_t
por
t
,
STransMsg
*
pMsg
,
STransMsg
*
pRsp
);
void
transSendRecv
(
void
*
shandle
,
const
SEpSet
*
pEpSe
t
,
STransMsg
*
pMsg
,
STransMsg
*
pRsp
);
void
transSendResponse
(
const
STransMsg
*
msg
);
void
transSendResponse
(
const
STransMsg
*
msg
);
void
transRegisterMsg
(
const
STransMsg
*
msg
);
void
transRegisterMsg
(
const
STransMsg
*
msg
);
int
transGetConnInfo
(
void
*
thandle
,
STransHandleInfo
*
pInfo
);
int
transGetConnInfo
(
void
*
thandle
,
STransHandleInfo
*
pInfo
);
...
...
source/libs/transport/inc/transportInt.h
浏览文件 @
a558b445
...
@@ -62,8 +62,7 @@ typedef struct {
...
@@ -62,8 +62,7 @@ typedef struct {
char
ckey
[
TSDB_PASSWORD_LEN
];
// ciphering key
char
ckey
[
TSDB_PASSWORD_LEN
];
// ciphering key
void
(
*
cfp
)(
void
*
parent
,
SRpcMsg
*
,
SEpSet
*
);
void
(
*
cfp
)(
void
*
parent
,
SRpcMsg
*
,
SEpSet
*
);
int
(
*
afp
)(
void
*
parent
,
char
*
user
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
bool
(
*
retry
)(
int32_t
code
);
int
(
*
retry
)(
void
*
parent
,
SRpcMsg
*
,
SEpSet
*
);
int32_t
refCount
;
int32_t
refCount
;
void
*
parent
;
void
*
parent
;
...
...
source/libs/transport/src/trans.c
浏览文件 @
a558b445
...
@@ -38,7 +38,6 @@ void* rpcOpen(const SRpcInit* pInit) {
...
@@ -38,7 +38,6 @@ void* rpcOpen(const SRpcInit* pInit) {
// register callback handle
// register callback handle
pRpc
->
cfp
=
pInit
->
cfp
;
pRpc
->
cfp
=
pInit
->
cfp
;
pRpc
->
afp
=
pInit
->
afp
;
pRpc
->
retry
=
pInit
->
rfp
;
pRpc
->
retry
=
pInit
->
rfp
;
if
(
pInit
->
connType
==
TAOS_CONN_SERVER
)
{
if
(
pInit
->
connType
==
TAOS_CONN_SERVER
)
{
...
@@ -116,19 +115,13 @@ int rpcReportProgress(void* pConn, char* pCont, int contLen) { return -1; }
...
@@ -116,19 +115,13 @@ int rpcReportProgress(void* pConn, char* pCont, int contLen) { return -1; }
void
rpcCancelRequest
(
int64_t
rid
)
{
return
;
}
void
rpcCancelRequest
(
int64_t
rid
)
{
return
;
}
void
rpcSendRequest
(
void
*
shandle
,
const
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
,
int64_t
*
pRid
)
{
void
rpcSendRequest
(
void
*
shandle
,
const
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
,
int64_t
*
pRid
)
{
char
*
ip
=
(
char
*
)(
pEpSet
->
eps
[
pEpSet
->
inUse
].
fqdn
);
transSendRequest
(
shandle
,
pEpSet
,
pMsg
,
NULL
);
uint32_t
port
=
pEpSet
->
eps
[
pEpSet
->
inUse
].
port
;
transSendRequest
(
shandle
,
ip
,
port
,
pMsg
,
NULL
);
}
}
void
rpcSendRequestWithCtx
(
void
*
shandle
,
const
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
,
int64_t
*
pRid
,
SRpcCtx
*
pCtx
)
{
void
rpcSendRequestWithCtx
(
void
*
shandle
,
const
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
,
int64_t
*
pRid
,
SRpcCtx
*
pCtx
)
{
char
*
ip
=
(
char
*
)(
pEpSet
->
eps
[
pEpSet
->
inUse
].
fqdn
);
transSendRequest
(
shandle
,
pEpSet
,
pMsg
,
pCtx
);
uint32_t
port
=
pEpSet
->
eps
[
pEpSet
->
inUse
].
port
;
transSendRequest
(
shandle
,
ip
,
port
,
pMsg
,
pCtx
);
}
}
void
rpcSendRecv
(
void
*
shandle
,
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
,
SRpcMsg
*
pRsp
)
{
void
rpcSendRecv
(
void
*
shandle
,
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
,
SRpcMsg
*
pRsp
)
{
char
*
ip
=
(
char
*
)(
pEpSet
->
eps
[
pEpSet
->
inUse
].
fqdn
);
transSendRecv
(
shandle
,
pEpSet
,
pMsg
,
pRsp
);
uint32_t
port
=
pEpSet
->
eps
[
pEpSet
->
inUse
].
port
;
transSendRecv
(
shandle
,
ip
,
port
,
pMsg
,
pRsp
);
}
}
void
rpcSendResponse
(
const
SRpcMsg
*
pMsg
)
{
transSendResponse
(
pMsg
);
}
void
rpcSendResponse
(
const
SRpcMsg
*
pMsg
)
{
transSendResponse
(
pMsg
);
}
...
...
source/libs/transport/src/transCli.c
浏览文件 @
a558b445
/*
/* * 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
...
@@ -97,7 +96,7 @@ static void cliSendCb(uv_write_t* req, int status);
...
@@ -97,7 +96,7 @@ static void cliSendCb(uv_write_t* req, int status);
static
void
cliConnCb
(
uv_connect_t
*
req
,
int
status
);
static
void
cliConnCb
(
uv_connect_t
*
req
,
int
status
);
static
void
cliAsyncCb
(
uv_async_t
*
handle
);
static
void
cliAsyncCb
(
uv_async_t
*
handle
);
static
void
cliAppCb
(
SCliConn
*
pConn
,
STrans
Msg
*
pMsg
);
static
int
cliAppCb
(
SCliConn
*
pConn
,
STransMsg
*
pResp
,
SCli
Msg
*
pMsg
);
static
SCliConn
*
cliCreateConn
(
SCliThrdObj
*
thrd
);
static
SCliConn
*
cliCreateConn
(
SCliThrdObj
*
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*/
);
...
@@ -227,6 +226,9 @@ static void cliWalkCb(uv_handle_t* handle, void* arg);
...
@@ -227,6 +226,9 @@ static void cliWalkCb(uv_handle_t* handle, void* arg);
#define REQUEST_PERSIS_HANDLE(msg) ((msg)->persistHandle == 1)
#define REQUEST_PERSIS_HANDLE(msg) ((msg)->persistHandle == 1)
#define REQUEST_RELEASE_HANDLE(cmsg) ((cmsg)->type == Release)
#define REQUEST_RELEASE_HANDLE(cmsg) ((cmsg)->type == Release)
#define EPSET_GET_INUSE_IP(epSet) ((epSet)->eps[(epSet)->inUse].fqdn)
#define EPSET_GET_INUSE_PORT(epSet) ((epSet)->eps[(epSet)->inUse].port)
static
void
*
cliWorkThread
(
void
*
arg
);
static
void
*
cliWorkThread
(
void
*
arg
);
bool
cliMaySendCachedMsg
(
SCliConn
*
conn
)
{
bool
cliMaySendCachedMsg
(
SCliConn
*
conn
)
{
...
@@ -311,14 +313,10 @@ void cliHandleResp(SCliConn* conn) {
...
@@ -311,14 +313,10 @@ void cliHandleResp(SCliConn* conn) {
return
;
return
;
}
}
if
(
pCtx
==
NULL
||
pCtx
->
pSem
==
NULL
)
{
int
ret
=
cliAppCb
(
conn
,
&
transMsg
,
pMsg
);
tTrace
(
"%s cli conn %p handle resp"
,
pTransInst
->
label
,
conn
);
if
(
ret
!=
0
)
{
cliAppCb
(
conn
,
&
transMsg
);
tTrace
(
"try to send req to next node"
);
//(pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
return
;
}
else
{
tTrace
(
"%s cli conn(sync) %p handle resp"
,
pTransInst
->
label
,
conn
);
memcpy
((
char
*
)
pCtx
->
pRsp
,
(
char
*
)
&
transMsg
,
sizeof
(
transMsg
));
tsem_post
(
pCtx
->
pSem
);
}
}
destroyCmsg
(
pMsg
);
destroyCmsg
(
pMsg
);
...
@@ -375,17 +373,15 @@ void cliHandleExcept(SCliConn* pConn) {
...
@@ -375,17 +373,15 @@ void cliHandleExcept(SCliConn* pConn) {
}
}
if
(
pCtx
==
NULL
||
pCtx
->
pSem
==
NULL
)
{
if
(
pCtx
==
NULL
||
pCtx
->
pSem
==
NULL
)
{
tTrace
(
"%s cli conn %p handle except"
,
pTransInst
->
label
,
pConn
);
if
(
transMsg
.
ahandle
==
NULL
)
{
if
(
transMsg
.
ahandle
==
NULL
)
{
once
=
true
;
once
=
true
;
continue
;
continue
;
}
}
cliAppCb
(
pConn
,
&
transMsg
);
}
//(pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
int
ret
=
cliAppCb
(
pConn
,
&
transMsg
,
pMsg
);
}
else
{
if
(
ret
!=
0
)
{
tTrace
(
"%s cli conn(sync) %p handle except"
,
pTransInst
->
label
,
pConn
);
tTrace
(
"try to send req to next node"
);
memcpy
((
char
*
)(
pCtx
->
pRsp
),
(
char
*
)(
&
transMsg
),
sizeof
(
transMsg
));
return
;
tsem_post
(
pCtx
->
pSem
);
}
}
destroyCmsg
(
pMsg
);
destroyCmsg
(
pMsg
);
tTrace
(
"%s cli conn %p start to destroy"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
);
tTrace
(
"%s cli conn %p start to destroy"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
);
...
@@ -695,7 +691,7 @@ SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) {
...
@@ -695,7 +691,7 @@ SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) {
}
}
}
else
{
}
else
{
STransConnCtx
*
pCtx
=
pMsg
->
ctx
;
STransConnCtx
*
pCtx
=
pMsg
->
ctx
;
conn
=
getConnFromPool
(
pThrd
->
pool
,
pCtx
->
ip
,
pCtx
->
port
);
conn
=
getConnFromPool
(
pThrd
->
pool
,
EPSET_GET_INUSE_IP
(
&
pCtx
->
epSet
),
EPSET_GET_INUSE_PORT
(
&
pCtx
->
epSet
)
);
if
(
conn
!=
NULL
)
{
if
(
conn
!=
NULL
)
{
tTrace
(
"%s cli conn %p get from conn pool"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
tTrace
(
"%s cli conn %p get from conn pool"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
}
else
{
}
else
{
...
@@ -719,10 +715,6 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
...
@@ -719,10 +715,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
);
// tTrace("%s cli conn %p queue msg size %d", ((STrans*)pThrd->pTransInst)->label, conn, 2);
// return;
//}
// transDestroyBuffer(&conn->readBuf);
cliSend
(
conn
);
cliSend
(
conn
);
}
else
{
}
else
{
conn
=
cliCreateConn
(
pThrd
);
conn
=
cliCreateConn
(
pThrd
);
...
@@ -730,8 +722,8 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
...
@@ -730,8 +722,8 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
transQueuePush
(
&
conn
->
cliMsgs
,
pMsg
);
transQueuePush
(
&
conn
->
cliMsgs
,
pMsg
);
conn
->
hThrdIdx
=
pCtx
->
hThrdIdx
;
conn
->
hThrdIdx
=
pCtx
->
hThrdIdx
;
conn
->
ip
=
strdup
(
pMsg
->
ctx
->
ip
);
conn
->
ip
=
strdup
(
EPSET_GET_INUSE_IP
(
&
pCtx
->
epSet
)
);
conn
->
port
=
pMsg
->
ctx
->
port
;
conn
->
port
=
EPSET_GET_INUSE_PORT
(
&
pCtx
->
epSet
)
;
int
ret
=
transSetConnOption
((
uv_tcp_t
*
)
conn
->
stream
);
int
ret
=
transSetConnOption
((
uv_tcp_t
*
)
conn
->
stream
);
if
(
ret
)
{
if
(
ret
)
{
...
@@ -743,10 +735,14 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
...
@@ -743,10 +735,14 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
addr
.
sin_family
=
AF_INET
;
addr
.
sin_family
=
AF_INET
;
addr
.
sin_addr
.
s_addr
=
taosGetIpv4FromFqdn
(
conn
->
ip
);
addr
.
sin_addr
.
s_addr
=
taosGetIpv4FromFqdn
(
conn
->
ip
);
addr
.
sin_port
=
(
uint16_t
)
htons
((
uint16_t
)
conn
->
port
);
addr
.
sin_port
=
(
uint16_t
)
htons
((
uint16_t
)
conn
->
port
);
// uv_ip4_addr(pMsg->ctx->ip, pMsg->ctx->port, &addr);
tTrace
(
"%s cli conn %p try to connect to %s:%d"
,
pTransInst
->
label
,
conn
,
conn
->
ip
,
conn
->
port
);
// handle error in callback if fail to connect
ret
=
uv_tcp_connect
(
&
conn
->
connReq
,
(
uv_tcp_t
*
)(
conn
->
stream
),
(
const
struct
sockaddr
*
)
&
addr
,
cliConnCb
);
tTrace
(
"%s cli conn %p try to connect to %s:%d"
,
pTransInst
->
label
,
conn
,
pMsg
->
ctx
->
ip
,
pMsg
->
ctx
->
port
);
if
(
ret
!=
0
)
{
uv_tcp_connect
(
&
conn
->
connReq
,
(
uv_tcp_t
*
)(
conn
->
stream
),
(
const
struct
sockaddr
*
)
&
addr
,
cliConnCb
);
tTrace
(
"%s cli conn %p failed to connect to %s:%d, reason: %s"
,
pTransInst
->
label
,
conn
,
conn
->
ip
,
conn
->
port
,
uv_err_name
(
ret
));
cliHandleExcept
(
conn
);
return
;
}
}
}
}
}
static
void
cliAsyncCb
(
uv_async_t
*
handle
)
{
static
void
cliAsyncCb
(
uv_async_t
*
handle
)
{
...
@@ -856,12 +852,10 @@ static void destroyThrdObj(SCliThrdObj* pThrd) {
...
@@ -856,12 +852,10 @@ static void destroyThrdObj(SCliThrdObj* pThrd) {
}
}
static
void
transDestroyConnCtx
(
STransConnCtx
*
ctx
)
{
static
void
transDestroyConnCtx
(
STransConnCtx
*
ctx
)
{
if
(
ctx
!=
NULL
)
{
//
taosMemoryFree
(
ctx
->
ip
);
}
taosMemoryFree
(
ctx
);
taosMemoryFree
(
ctx
);
}
}
//
void
cliSendQuit
(
SCliThrdObj
*
thrd
)
{
void
cliSendQuit
(
SCliThrdObj
*
thrd
)
{
// cli can stop gracefully
// cli can stop gracefully
SCliMsg
*
msg
=
taosMemoryCalloc
(
1
,
sizeof
(
SCliMsg
));
SCliMsg
*
msg
=
taosMemoryCalloc
(
1
,
sizeof
(
SCliMsg
));
...
@@ -881,17 +875,58 @@ int cliRBChoseIdx(STrans* pTransInst) {
...
@@ -881,17 +875,58 @@ int cliRBChoseIdx(STrans* pTransInst) {
}
}
return
index
%
pTransInst
->
numOfThreads
;
return
index
%
pTransInst
->
numOfThreads
;
}
}
void
cliAppCb
(
SCliConn
*
pConn
,
STransMsg
*
trans
Msg
)
{
int
cliAppCb
(
SCliConn
*
pConn
,
STransMsg
*
pResp
,
SCliMsg
*
p
Msg
)
{
SCliThrdObj
*
pThrd
=
pConn
->
hostThrd
;
SCliThrdObj
*
pThrd
=
pConn
->
hostThrd
;
STrans
*
pTransInst
=
pThrd
->
pTransInst
;
STrans
*
pTransInst
=
pThrd
->
pTransInst
;
if
(
transMsg
->
code
==
TSDB_CODE_RPC_REDIRECT
&&
pTransInst
->
retry
!=
NULL
)
{
if
(
pMsg
==
NULL
||
pMsg
->
ctx
==
NULL
)
{
SMEpSet
emsg
=
{
0
};
tTrace
(
"%s cli conn %p handle resp"
,
pTransInst
->
label
,
pConn
);
tDeserializeSMEpSet
(
transMsg
->
pCont
,
transMsg
->
contLen
,
&
emsg
);
pTransInst
->
cfp
(
pTransInst
->
parent
,
pResp
,
NULL
);
pTransInst
->
retry
(
pTransInst
,
transMsg
,
&
(
emsg
.
epSet
));
return
0
;
}
STransConnCtx
*
pCtx
=
pMsg
->
ctx
;
SEpSet
*
pEpSet
=
&
pCtx
->
epSet
;
/*
* upper layer handle retry if code equal TSDB_CODE_RPC_NETWORK_UNAVAIL
*/
tmsg_t
msgType
=
pCtx
->
msgType
;
if
((
pTransInst
->
retry
!=
NULL
&&
(
pTransInst
->
retry
(
pResp
->
code
)))
||
((
pResp
->
code
==
TSDB_CODE_RPC_NETWORK_UNAVAIL
)
&&
msgType
==
TDMT_MND_CONNECT
))
{
pCtx
->
retryCount
+=
1
;
pMsg
->
st
=
taosGetTimestampUs
();
if
(
msgType
==
TDMT_MND_CONNECT
&&
pResp
->
code
==
TSDB_CODE_RPC_NETWORK_UNAVAIL
)
{
if
(
pCtx
->
retryCount
<
pEpSet
->
numOfEps
)
{
pEpSet
->
inUse
=
(
++
pEpSet
->
inUse
)
%
pEpSet
->
numOfEps
;
cliHandleReq
(
pMsg
,
pThrd
);
cliDestroy
((
uv_handle_t
*
)
pConn
->
stream
);
return
-
1
;
}
}
else
if
(
pCtx
->
retryCount
<
TRANS_RETRY_COUNT_LIMIT
)
{
if
(
pResp
->
contLen
==
0
)
{
pEpSet
->
inUse
=
(
pEpSet
->
inUse
++
)
%
pEpSet
->
numOfEps
;
}
else
{
SMEpSet
emsg
=
{
0
};
tDeserializeSMEpSet
(
pResp
->
pCont
,
pResp
->
contLen
,
&
emsg
);
pCtx
->
epSet
=
emsg
.
epSet
;
}
cliHandleReq
(
pMsg
,
pThrd
);
// release pConn
addConnToPool
(
pThrd
,
pConn
);
return
-
1
;
}
}
if
(
pCtx
->
pSem
!=
NULL
)
{
tTrace
(
"%s cli conn %p handle resp"
,
pTransInst
->
label
,
pConn
);
memcpy
((
char
*
)
pCtx
->
pRsp
,
(
char
*
)
pResp
,
sizeof
(
*
pResp
));
tsem_post
(
pCtx
->
pSem
);
}
else
{
}
else
{
pTransInst
->
cfp
(
pTransInst
->
parent
,
transMsg
,
NULL
);
tTrace
(
"%s cli conn %p handle resp"
,
pTransInst
->
label
,
pConn
);
pTransInst
->
cfp
(
pTransInst
->
parent
,
pResp
,
pEpSet
);
}
}
return
0
;
}
}
void
transCloseClient
(
void
*
arg
)
{
void
transCloseClient
(
void
*
arg
)
{
...
@@ -934,18 +969,17 @@ void transReleaseCliHandle(void* handle) {
...
@@ -934,18 +969,17 @@ void transReleaseCliHandle(void* handle) {
transSendAsync
(
thrd
->
asyncPool
,
&
cmsg
->
q
);
transSendAsync
(
thrd
->
asyncPool
,
&
cmsg
->
q
);
}
}
void
transSendRequest
(
void
*
shandle
,
const
char
*
ip
,
uint32_t
port
,
STransMsg
*
pMsg
,
STransCtx
*
ctx
)
{
void
transSendRequest
(
void
*
shandle
,
const
SEpSet
*
pEpSet
,
STransMsg
*
pReq
,
STransCtx
*
ctx
)
{
STrans
*
pTransInst
=
(
STrans
*
)
shandle
;
STrans
*
pTransInst
=
(
STrans
*
)
shandle
;
int
index
=
CONN_HOST_THREAD_INDEX
((
SCliConn
*
)
p
Msg
->
handle
);
int
index
=
CONN_HOST_THREAD_INDEX
((
SCliConn
*
)
p
Req
->
handle
);
if
(
index
==
-
1
)
{
if
(
index
==
-
1
)
{
index
=
cliRBChoseIdx
(
pTransInst
);
index
=
cliRBChoseIdx
(
pTransInst
);
}
}
STransConnCtx
*
pCtx
=
taosMemoryCalloc
(
1
,
sizeof
(
STransConnCtx
));
STransConnCtx
*
pCtx
=
taosMemoryCalloc
(
1
,
sizeof
(
STransConnCtx
));
pCtx
->
ahandle
=
pMsg
->
ahandle
;
pCtx
->
epSet
=
*
pEpSet
;
pCtx
->
msgType
=
pMsg
->
msgType
;
pCtx
->
ahandle
=
pReq
->
ahandle
;
pCtx
->
ip
=
strdup
(
ip
);
pCtx
->
msgType
=
pReq
->
msgType
;
pCtx
->
port
=
port
;
pCtx
->
hThrdIdx
=
index
;
pCtx
->
hThrdIdx
=
index
;
if
(
ctx
!=
NULL
)
{
if
(
ctx
!=
NULL
)
{
...
@@ -955,17 +989,18 @@ void transSendRequest(void* shandle, const char* ip, uint32_t port, STransMsg* p
...
@@ -955,17 +989,18 @@ void transSendRequest(void* shandle, const char* ip, uint32_t port, STransMsg* p
SCliMsg
*
cliMsg
=
taosMemoryCalloc
(
1
,
sizeof
(
SCliMsg
));
SCliMsg
*
cliMsg
=
taosMemoryCalloc
(
1
,
sizeof
(
SCliMsg
));
cliMsg
->
ctx
=
pCtx
;
cliMsg
->
ctx
=
pCtx
;
cliMsg
->
msg
=
*
p
Msg
;
cliMsg
->
msg
=
*
p
Req
;
cliMsg
->
st
=
taosGetTimestampUs
();
cliMsg
->
st
=
taosGetTimestampUs
();
cliMsg
->
type
=
Normal
;
cliMsg
->
type
=
Normal
;
SCliThrdObj
*
thrd
=
((
SCliObj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
SCliThrdObj
*
thrd
=
((
SCliObj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
tDebug
(
"send request at thread:%d %p, dst: %s:%d, app:%p"
,
index
,
pMsg
,
ip
,
port
,
pMsg
->
ahandle
);
tDebug
(
"send request at thread:%d %p, dst: %s:%d, app:%p"
,
index
,
pReq
,
EPSET_GET_INUSE_IP
(
&
pCtx
->
epSet
),
EPSET_GET_INUSE_PORT
(
&
pCtx
->
epSet
),
pReq
->
ahandle
);
ASSERT
(
transSendAsync
(
thrd
->
asyncPool
,
&
(
cliMsg
->
q
))
==
0
);
ASSERT
(
transSendAsync
(
thrd
->
asyncPool
,
&
(
cliMsg
->
q
))
==
0
);
}
}
void
transSendRecv
(
void
*
shandle
,
const
char
*
ip
,
uint32_t
por
t
,
STransMsg
*
pReq
,
STransMsg
*
pRsp
)
{
void
transSendRecv
(
void
*
shandle
,
const
SEpSet
*
pEpSe
t
,
STransMsg
*
pReq
,
STransMsg
*
pRsp
)
{
STrans
*
pTransInst
=
(
STrans
*
)
shandle
;
STrans
*
pTransInst
=
(
STrans
*
)
shandle
;
int
index
=
CONN_HOST_THREAD_INDEX
(
pReq
->
handle
);
int
index
=
CONN_HOST_THREAD_INDEX
(
pReq
->
handle
);
if
(
index
==
-
1
)
{
if
(
index
==
-
1
)
{
...
@@ -973,10 +1008,9 @@ void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pReq
...
@@ -973,10 +1008,9 @@ void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pReq
}
}
STransConnCtx
*
pCtx
=
taosMemoryCalloc
(
1
,
sizeof
(
STransConnCtx
));
STransConnCtx
*
pCtx
=
taosMemoryCalloc
(
1
,
sizeof
(
STransConnCtx
));
pCtx
->
epSet
=
*
pEpSet
;
pCtx
->
ahandle
=
pReq
->
ahandle
;
pCtx
->
ahandle
=
pReq
->
ahandle
;
pCtx
->
msgType
=
pReq
->
msgType
;
pCtx
->
msgType
=
pReq
->
msgType
;
pCtx
->
ip
=
strdup
(
ip
);
pCtx
->
port
=
port
;
pCtx
->
hThrdIdx
=
index
;
pCtx
->
hThrdIdx
=
index
;
pCtx
->
pSem
=
taosMemoryCalloc
(
1
,
sizeof
(
tsem_t
));
pCtx
->
pSem
=
taosMemoryCalloc
(
1
,
sizeof
(
tsem_t
));
pCtx
->
pRsp
=
pRsp
;
pCtx
->
pRsp
=
pRsp
;
...
@@ -989,6 +1023,9 @@ void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pReq
...
@@ -989,6 +1023,9 @@ void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pReq
cliMsg
->
type
=
Normal
;
cliMsg
->
type
=
Normal
;
SCliThrdObj
*
thrd
=
((
SCliObj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
SCliThrdObj
*
thrd
=
((
SCliObj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
tDebug
(
"send request at thread:%d %p, dst: %s:%d, app:%p"
,
index
,
pReq
,
EPSET_GET_INUSE_IP
(
&
pCtx
->
epSet
),
EPSET_GET_INUSE_PORT
(
&
pCtx
->
epSet
),
pReq
->
ahandle
);
transSendAsync
(
thrd
->
asyncPool
,
&
(
cliMsg
->
q
));
transSendAsync
(
thrd
->
asyncPool
,
&
(
cliMsg
->
q
));
tsem_t
*
pSem
=
pCtx
->
pSem
;
tsem_t
*
pSem
=
pCtx
->
pSem
;
tsem_wait
(
pSem
);
tsem_wait
(
pSem
);
...
...
source/libs/transport/src/transComm.c
浏览文件 @
a558b445
...
@@ -93,11 +93,6 @@ bool transDecompressMsg(char* msg, int32_t len, int32_t* flen) {
...
@@ -93,11 +93,6 @@ bool transDecompressMsg(char* msg, int32_t len, int32_t* flen) {
return
false
;
return
false
;
}
}
void
transConnCtxDestroy
(
STransConnCtx
*
ctx
)
{
taosMemoryFree
(
ctx
->
ip
);
taosMemoryFree
(
ctx
);
}
void
transFreeMsg
(
void
*
msg
)
{
void
transFreeMsg
(
void
*
msg
)
{
if
(
msg
==
NULL
)
{
if
(
msg
==
NULL
)
{
return
;
return
;
...
@@ -363,10 +358,4 @@ void transQueueDestroy(STransQueue* queue) {
...
@@ -363,10 +358,4 @@ void transQueueDestroy(STransQueue* queue) {
transQueueClear
(
queue
);
transQueueClear
(
queue
);
taosArrayDestroy
(
queue
->
q
);
taosArrayDestroy
(
queue
->
q
);
}
}
// int32_t transGetExHandle() {
// static
//}
// void transThreadOnce() {
// taosThreadOnce(&transModuleInit, );
//}
#endif
#endif
source/libs/transport/src/transSrv.c
浏览文件 @
a558b445
...
@@ -802,7 +802,6 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
...
@@ -802,7 +802,6 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
taosThreadOnce
(
&
transModuleInit
,
uvInitExHandleMgt
);
taosThreadOnce
(
&
transModuleInit
,
uvInitExHandleMgt
);
transSrvInst
++
;
transSrvInst
++
;
// uvOpenExHandleMgt(10000);
for
(
int
i
=
0
;
i
<
srv
->
numOfThreads
;
i
++
)
{
for
(
int
i
=
0
;
i
<
srv
->
numOfThreads
;
i
++
)
{
SWorkThrdObj
*
thrd
=
(
SWorkThrdObj
*
)
taosMemoryCalloc
(
1
,
sizeof
(
SWorkThrdObj
));
SWorkThrdObj
*
thrd
=
(
SWorkThrdObj
*
)
taosMemoryCalloc
(
1
,
sizeof
(
SWorkThrdObj
));
...
@@ -831,6 +830,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
...
@@ -831,6 +830,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
}
else
{
}
else
{
// TODO: clear all other resource later
// TODO: clear all other resource later
tError
(
"failed to create worker-thread %d"
,
i
);
tError
(
"failed to create worker-thread %d"
,
i
);
goto
End
;
}
}
}
}
if
(
false
==
addHandleToAcceptloop
(
srv
))
{
if
(
false
==
addHandleToAcceptloop
(
srv
))
{
...
@@ -840,6 +840,8 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
...
@@ -840,6 +840,8 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
if
(
err
==
0
)
{
if
(
err
==
0
)
{
tDebug
(
"success to create accept-thread"
);
tDebug
(
"success to create accept-thread"
);
}
else
{
}
else
{
tError
(
"failed to create accept-thread"
);
goto
End
;
// clear all resource later
// clear all resource later
}
}
...
@@ -1078,6 +1080,7 @@ void transRegisterMsg(const STransMsg* msg) {
...
@@ -1078,6 +1080,7 @@ void transRegisterMsg(const STransMsg* msg) {
transSendAsync
(
pThrd
->
asyncPool
,
&
srvMsg
->
q
);
transSendAsync
(
pThrd
->
asyncPool
,
&
srvMsg
->
q
);
uvReleaseExHandle
(
refId
);
uvReleaseExHandle
(
refId
);
return
;
return
;
_return1:
_return1:
tTrace
(
"server handle %p failed to send to register brokenlink"
,
exh
);
tTrace
(
"server handle %p failed to send to register brokenlink"
,
exh
);
rpcFreeCont
(
msg
->
pCont
);
rpcFreeCont
(
msg
->
pCont
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录