Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
30f602fa
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
30f602fa
编写于
3月 12, 2022
作者:
U
ubuntu
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
update UT test
上级
baf0ac2d
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
99 addition
and
98 deletion
+99
-98
source/libs/transport/src/transCli.c
source/libs/transport/src/transCli.c
+99
-98
未找到文件。
source/libs/transport/src/transCli.c
浏览文件 @
30f602fa
...
...
@@ -63,12 +63,12 @@ typedef struct SCliThrdObj {
bool
quit
;
}
SCliThrdObj
;
typedef
struct
SCli
ent
Obj
{
typedef
struct
SCliObj
{
char
label
[
TSDB_LABEL_LEN
];
int32_t
index
;
int
numOfThreads
;
SCliThrdObj
**
pThreadObj
;
}
SCli
ent
Obj
;
}
SCliObj
;
typedef
struct
SConnList
{
queue
conn
;
...
...
@@ -82,32 +82,32 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port);
static
void
addConnToPool
(
void
*
pool
,
char
*
ip
,
uint32_t
port
,
SCliConn
*
conn
);
// register timer in each thread to clear expire conn
static
void
cli
ent
TimeoutCb
(
uv_timer_t
*
handle
);
// alloc buf for re
ad
static
void
cli
ent
AllocBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
);
// callback after read nbytes from socket
static
void
cli
ent
RecvCb
(
uv_stream_t
*
cli
,
ssize_t
nread
,
const
uv_buf_t
*
buf
);
static
void
cliTimeoutCb
(
uv_timer_t
*
handle
);
// alloc buf for re
cv
static
void
cliAllocBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
);
// callback after
read nbytes from socket
static
void
cliRecvCb
(
uv_stream_t
*
cli
,
ssize_t
nread
,
const
uv_buf_t
*
buf
);
// callback after write data to socket
static
void
cli
entSendData
Cb
(
uv_write_t
*
req
,
int
status
);
static
void
cli
Send
Cb
(
uv_write_t
*
req
,
int
status
);
// callback after conn to server
static
void
cli
ent
ConnCb
(
uv_connect_t
*
req
,
int
status
);
static
void
cli
ent
AsyncCb
(
uv_async_t
*
handle
);
static
void
cliConnCb
(
uv_connect_t
*
req
,
int
status
);
static
void
cliAsyncCb
(
uv_async_t
*
handle
);
static
SCliConn
*
cli
entConnCreate
(
SCliThrdObj
*
thrd
);
static
void
cli
entConnDestroy
(
SCliConn
*
pConn
,
bool
clear
/*clear tcp handle or not*/
);
static
void
cli
ent
Destroy
(
uv_handle_t
*
handle
);
static
SCliConn
*
cli
CreateConn
(
SCliThrdObj
*
thrd
);
static
void
cli
DestroyConn
(
SCliConn
*
pConn
,
bool
clear
/*clear tcp handle or not*/
);
static
void
cliDestroy
(
uv_handle_t
*
handle
);
// process data read from server, add decompress etc later
static
void
cli
ent
HandleResp
(
SCliConn
*
conn
);
static
void
cliHandleResp
(
SCliConn
*
conn
);
// handle except about conn
static
void
cli
ent
HandleExcept
(
SCliConn
*
conn
);
static
void
cliHandleExcept
(
SCliConn
*
conn
);
// handle req from app
static
void
cli
ent
HandleReq
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
);
static
void
cli
ent
HandleQuit
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
);
static
void
cli
ent
SendQuit
(
SCliThrdObj
*
thrd
);
static
void
cliHandleReq
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
);
static
void
cliHandleQuit
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
);
static
void
cliSendQuit
(
SCliThrdObj
*
thrd
);
static
void
destroyUserdata
(
SRpcMsg
*
userdata
);
static
int
cli
ent
RBChoseIdx
(
SRpcInfo
*
pTransInst
);
static
int
cliRBChoseIdx
(
SRpcInfo
*
pTransInst
);
static
void
destroyCmsg
(
SCliMsg
*
cmsg
);
static
void
transDestroyConnCtx
(
STransConnCtx
*
ctx
);
...
...
@@ -122,7 +122,7 @@ static void destroyThrdObj(SCliThrdObj* pThrd);
#define CONN_HANDLE_THREAD_QUIT(conn, thrd) \
do { \
if (thrd->quit) { \
cli
ent
HandleExcept(conn); \
cliHandleExcept(conn); \
goto _RETURE; \
} \
} while (0)
...
...
@@ -130,15 +130,15 @@ static void destroyThrdObj(SCliThrdObj* pThrd);
#define CONN_HANDLE_BROKEN(conn) \
do { \
if (conn->broken) { \
cli
ent
HandleExcept(conn); \
cliHandleExcept(conn); \
goto _RETURE; \
} \
} while (0);
static
void
*
cli
ent
Thread
(
void
*
arg
);
static
void
*
cli
Work
Thread
(
void
*
arg
);
static
void
*
cli
ent
NotifyApp
()
{}
static
void
cli
ent
HandleResp
(
SCliConn
*
conn
)
{
static
void
*
cliNotifyApp
()
{}
static
void
cliHandleResp
(
SCliConn
*
conn
)
{
SCliMsg
*
pMsg
=
conn
->
data
;
STransConnCtx
*
pCtx
=
pMsg
->
ctx
;
...
...
@@ -164,25 +164,25 @@ static void clientHandleResp(SCliConn* conn) {
transRefCliHandle
(
conn
);
conn
->
persist
=
1
;
tDebug
(
"cli
ent
conn %p persist by app"
,
conn
);
tDebug
(
"cli conn %p persist by app"
,
conn
);
}
tDebug
(
"%s cli
ent
conn %p %s received from %s:%d, local info: %s:%d, msg size: %d"
,
pTransInst
->
label
,
conn
,
tDebug
(
"%s cli conn %p %s received from %s:%d, local info: %s:%d, msg size: %d"
,
pTransInst
->
label
,
conn
,
TMSG_INFO
(
pHead
->
msgType
),
inet_ntoa
(
conn
->
addr
.
sin_addr
),
ntohs
(
conn
->
addr
.
sin_port
),
inet_ntoa
(
conn
->
locaddr
.
sin_addr
),
ntohs
(
conn
->
locaddr
.
sin_port
),
rpcMsg
.
contLen
);
conn
->
secured
=
pHead
->
secured
;
if
(
pCtx
->
pSem
==
NULL
)
{
tTrace
(
"%s cli
ent
conn %p handle resp"
,
pTransInst
->
label
,
conn
);
tTrace
(
"%s cli conn %p handle resp"
,
pTransInst
->
label
,
conn
);
(
pTransInst
->
cfp
)(
pTransInst
->
parent
,
&
rpcMsg
,
NULL
);
}
else
{
tTrace
(
"%s cli
ent
conn(sync) %p handle resp"
,
pTransInst
->
label
,
conn
);
tTrace
(
"%s cli conn(sync) %p handle resp"
,
pTransInst
->
label
,
conn
);
memcpy
((
char
*
)
pCtx
->
pRsp
,
(
char
*
)
&
rpcMsg
,
sizeof
(
rpcMsg
));
tsem_post
(
pCtx
->
pSem
);
}
uv_read_start
((
uv_stream_t
*
)
conn
->
stream
,
cli
entAllocBufferCb
,
client
RecvCb
);
uv_read_start
((
uv_stream_t
*
)
conn
->
stream
,
cli
AllocBufferCb
,
cli
RecvCb
);
// user owns conn->persist = 1
if
(
conn
->
persist
==
0
)
{
...
...
@@ -193,10 +193,10 @@ static void clientHandleResp(SCliConn* conn) {
// start thread's timer of conn pool if not active
if
(
!
uv_is_active
((
uv_handle_t
*
)
&
pThrd
->
timer
)
&&
pTransInst
->
idleTime
>
0
)
{
// uv_timer_start((uv_timer_t*)&pThrd->timer, cli
ent
TimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
// uv_timer_start((uv_timer_t*)&pThrd->timer, cliTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
}
}
static
void
cli
ent
HandleExcept
(
SCliConn
*
pConn
)
{
static
void
cliHandleExcept
(
SCliConn
*
pConn
)
{
if
(
pConn
->
data
==
NULL
)
{
// handle conn except in conn pool
transUnrefCliHandle
(
pConn
);
...
...
@@ -214,25 +214,25 @@ static void clientHandleExcept(SCliConn* pConn) {
rpcMsg
.
msgType
=
pMsg
->
msg
.
msgType
+
1
;
if
(
pCtx
->
pSem
==
NULL
)
{
tTrace
(
"%s cli
ent
conn %p handle resp"
,
pTransInst
->
label
,
pConn
);
tTrace
(
"%s cli conn %p handle resp"
,
pTransInst
->
label
,
pConn
);
(
pTransInst
->
cfp
)(
pTransInst
->
parent
,
&
rpcMsg
,
NULL
);
}
else
{
tTrace
(
"%s cli
ent
conn(sync) %p handle resp"
,
pTransInst
->
label
,
pConn
);
tTrace
(
"%s cli conn(sync) %p handle resp"
,
pTransInst
->
label
,
pConn
);
memcpy
((
char
*
)(
pCtx
->
pRsp
),
(
char
*
)(
&
rpcMsg
),
sizeof
(
rpcMsg
));
tsem_post
(
pCtx
->
pSem
);
}
destroyCmsg
(
pConn
->
data
);
pConn
->
data
=
NULL
;
tTrace
(
"%s cli
ent
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
);
transUnrefCliHandle
(
pConn
);
}
static
void
cli
ent
TimeoutCb
(
uv_timer_t
*
handle
)
{
static
void
cliTimeoutCb
(
uv_timer_t
*
handle
)
{
SCliThrdObj
*
pThrd
=
handle
->
data
;
SRpcInfo
*
pRpc
=
pThrd
->
pTransInst
;
int64_t
currentTime
=
pThrd
->
nextTimeout
;
tTrace
(
"%s, cli
ent
conn timeout, try to remove expire conn from conn pool"
,
pRpc
->
label
);
tTrace
(
"%s, cli conn timeout, try to remove expire conn from conn pool"
,
pRpc
->
label
);
SConnList
*
p
=
taosHashIterate
((
SHashObj
*
)
pThrd
->
pool
,
NULL
);
while
(
p
!=
NULL
)
{
...
...
@@ -250,7 +250,7 @@ static void clientTimeoutCb(uv_timer_t* handle) {
}
pThrd
->
nextTimeout
=
taosGetTimestampMs
()
+
CONN_PERSIST_TIME
(
pRpc
->
idleTime
);
uv_timer_start
(
handle
,
cli
ent
TimeoutCb
,
CONN_PERSIST_TIME
(
pRpc
->
idleTime
)
/
2
,
0
);
uv_timer_start
(
handle
,
cliTimeoutCb
,
CONN_PERSIST_TIME
(
pRpc
->
idleTime
)
/
2
,
0
);
}
static
void
*
createConnPool
(
int
size
)
{
// thread local, no lock
...
...
@@ -263,7 +263,7 @@ static void* destroyConnPool(void* pool) {
queue
*
h
=
QUEUE_HEAD
(
&
connList
->
conn
);
QUEUE_REMOVE
(
h
);
SCliConn
*
c
=
QUEUE_DATA
(
h
,
SCliConn
,
conn
);
cli
entConnDestroy
(
c
,
true
);
cli
DestroyConn
(
c
,
true
);
}
connList
=
taosHashIterate
((
SHashObj
*
)
pool
,
connList
);
}
...
...
@@ -299,7 +299,7 @@ static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) {
tstrncpy
(
key
,
ip
,
strlen
(
ip
));
tstrncpy
(
key
+
strlen
(
key
),
(
char
*
)(
&
port
),
sizeof
(
port
));
tTrace
(
"cli
ent
conn %p added to conn pool, read buf cap: %d"
,
conn
,
conn
->
readBuf
.
cap
);
tTrace
(
"cli conn %p added to conn pool, read buf cap: %d"
,
conn
,
conn
->
readBuf
.
cap
);
SRpcInfo
*
pRpc
=
((
SCliThrdObj
*
)
conn
->
hostThrd
)
->
pTransInst
;
...
...
@@ -309,12 +309,12 @@ static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) {
assert
(
plist
!=
NULL
);
QUEUE_PUSH
(
&
plist
->
conn
,
&
conn
->
conn
);
}
static
void
cli
ent
AllocBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
)
{
static
void
cliAllocBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
)
{
SCliConn
*
conn
=
handle
->
data
;
SConnBuffer
*
pBuf
=
&
conn
->
readBuf
;
transAllocBuffer
(
pBuf
,
buf
);
}
static
void
cli
ent
RecvCb
(
uv_stream_t
*
handle
,
ssize_t
nread
,
const
uv_buf_t
*
buf
)
{
static
void
cliRecvCb
(
uv_stream_t
*
handle
,
ssize_t
nread
,
const
uv_buf_t
*
buf
)
{
// impl later
if
(
handle
->
data
==
NULL
)
{
return
;
...
...
@@ -324,10 +324,10 @@ static void clientRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
if
(
nread
>
0
)
{
pBuf
->
len
+=
nread
;
if
(
transReadComplete
(
pBuf
))
{
tTrace
(
"%s cli
ent
conn %p read complete"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
cli
ent
HandleResp
(
conn
);
tTrace
(
"%s cli conn %p read complete"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
cliHandleResp
(
conn
);
}
else
{
tTrace
(
"%s cli
ent
conn %p read partial packet, continue to read"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
tTrace
(
"%s cli conn %p read partial packet, continue to read"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
}
return
;
}
...
...
@@ -340,13 +340,13 @@ static void clientRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
return
;
}
if
(
nread
<
0
)
{
tError
(
"%s cli
ent
conn %p read error: %s"
,
CONN_GET_INST_LABEL
(
conn
),
conn
,
uv_err_name
(
nread
));
tError
(
"%s cli conn %p read error: %s"
,
CONN_GET_INST_LABEL
(
conn
),
conn
,
uv_err_name
(
nread
));
conn
->
broken
=
true
;
cli
ent
HandleExcept
(
conn
);
cliHandleExcept
(
conn
);
}
}
static
SCliConn
*
cli
entConnCreate
(
SCliThrdObj
*
pThrd
)
{
static
SCliConn
*
cli
CreateConn
(
SCliThrdObj
*
pThrd
)
{
SCliConn
*
conn
=
calloc
(
1
,
sizeof
(
SCliConn
));
// read/write stream handle
conn
->
stream
=
(
uv_stream_t
*
)
malloc
(
sizeof
(
uv_tcp_t
));
...
...
@@ -362,40 +362,40 @@ static SCliConn* clientConnCreate(SCliThrdObj* pThrd) {
transRefCliHandle
(
conn
);
return
conn
;
}
static
void
cli
entConnDestroy
(
SCliConn
*
conn
,
bool
clear
)
{
tTrace
(
"%s cli
ent
conn %p remove from conn pool"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
static
void
cli
DestroyConn
(
SCliConn
*
conn
,
bool
clear
)
{
tTrace
(
"%s cli conn %p remove from conn pool"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
QUEUE_REMOVE
(
&
conn
->
conn
);
if
(
clear
)
{
uv_close
((
uv_handle_t
*
)
conn
->
stream
,
cli
ent
Destroy
);
uv_close
((
uv_handle_t
*
)
conn
->
stream
,
cliDestroy
);
}
}
static
void
cli
ent
Destroy
(
uv_handle_t
*
handle
)
{
static
void
cliDestroy
(
uv_handle_t
*
handle
)
{
SCliConn
*
conn
=
handle
->
data
;
free
(
conn
->
stream
);
tTrace
(
"%s cli
ent
conn %p destroy successfully"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
tTrace
(
"%s cli conn %p destroy successfully"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
free
(
conn
);
}
static
void
cli
entSendData
Cb
(
uv_write_t
*
req
,
int
status
)
{
static
void
cli
Send
Cb
(
uv_write_t
*
req
,
int
status
)
{
SCliConn
*
pConn
=
req
->
data
;
if
(
status
==
0
)
{
tTrace
(
"%s cli
ent
conn %p data already was written out"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
);
tTrace
(
"%s cli conn %p data already was written out"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
);
SCliMsg
*
pMsg
=
pConn
->
data
;
if
(
pMsg
==
NULL
)
{
return
;
}
destroyUserdata
(
&
pMsg
->
msg
);
}
else
{
tError
(
"%s cli
ent
conn %p failed to write: %s"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
,
uv_err_name
(
status
));
cli
ent
HandleExcept
(
pConn
);
tError
(
"%s cli conn %p failed to write: %s"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
,
uv_err_name
(
status
));
cliHandleExcept
(
pConn
);
return
;
}
uv_read_start
((
uv_stream_t
*
)
pConn
->
stream
,
cli
entAllocBufferCb
,
client
RecvCb
);
uv_read_start
((
uv_stream_t
*
)
pConn
->
stream
,
cli
AllocBufferCb
,
cli
RecvCb
);
}
static
void
cli
entSendData
(
SCliConn
*
pConn
)
{
static
void
cli
Send
(
SCliConn
*
pConn
)
{
CONN_HANDLE_BROKEN
(
pConn
);
SCliMsg
*
pCliMsg
=
pConn
->
data
;
...
...
@@ -432,22 +432,22 @@ static void clientSendData(SCliConn* pConn) {
pHead
->
msgLen
=
(
int32_t
)
htonl
((
uint32_t
)
msgLen
);
uv_buf_t
wb
=
uv_buf_init
((
char
*
)
pHead
,
msgLen
);
tDebug
(
"%s cli
ent
conn %p %s is send to %s:%d, local info %s:%d"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
,
tDebug
(
"%s cli conn %p %s is send to %s:%d, local info %s:%d"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
,
TMSG_INFO
(
pHead
->
msgType
),
inet_ntoa
(
pConn
->
addr
.
sin_addr
),
ntohs
(
pConn
->
addr
.
sin_port
),
inet_ntoa
(
pConn
->
locaddr
.
sin_addr
),
ntohs
(
pConn
->
locaddr
.
sin_port
));
uv_write
(
&
pConn
->
writeReq
,
(
uv_stream_t
*
)
pConn
->
stream
,
&
wb
,
1
,
cli
entSendData
Cb
);
uv_write
(
&
pConn
->
writeReq
,
(
uv_stream_t
*
)
pConn
->
stream
,
&
wb
,
1
,
cli
Send
Cb
);
return
;
_RETURE:
return
;
}
static
void
cli
ent
ConnCb
(
uv_connect_t
*
req
,
int
status
)
{
static
void
cliConnCb
(
uv_connect_t
*
req
,
int
status
)
{
// impl later
SCliConn
*
pConn
=
req
->
data
;
if
(
status
!=
0
)
{
tError
(
"%s cli
ent
conn %p failed to connect server: %s"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
,
uv_strerror
(
status
));
cli
ent
HandleExcept
(
pConn
);
tError
(
"%s cli conn %p failed to connect server: %s"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
,
uv_strerror
(
status
));
cliHandleExcept
(
pConn
);
return
;
}
int
addrlen
=
sizeof
(
pConn
->
addr
);
...
...
@@ -456,14 +456,14 @@ static void clientConnCb(uv_connect_t* req, int status) {
addrlen
=
sizeof
(
pConn
->
locaddr
);
uv_tcp_getsockname
((
uv_tcp_t
*
)
pConn
->
stream
,
(
struct
sockaddr
*
)
&
pConn
->
locaddr
,
&
addrlen
);
tTrace
(
"%s cli
ent
conn %p connect to server successfully"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
);
tTrace
(
"%s cli conn %p connect to server successfully"
,
CONN_GET_INST_LABEL
(
pConn
),
pConn
);
assert
(
pConn
->
stream
==
req
->
handle
);
cli
entSendData
(
pConn
);
cli
Send
(
pConn
);
}
static
void
cli
ent
HandleQuit
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
)
{
tDebug
(
"cli
ent
work thread %p start to quit"
,
pThrd
);
static
void
cliHandleQuit
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
)
{
tDebug
(
"cli work thread %p start to quit"
,
pThrd
);
destroyCmsg
(
pMsg
);
destroyConnPool
(
pThrd
->
pool
);
...
...
@@ -472,57 +472,57 @@ static void clientHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) {
pThrd
->
quit
=
true
;
uv_stop
(
pThrd
->
loop
);
}
static
SCliConn
*
cli
ent
GetConn
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
)
{
static
SCliConn
*
cliGetConn
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
)
{
SCliConn
*
conn
=
NULL
;
if
(
pMsg
->
msg
.
handle
!=
NULL
)
{
conn
=
(
SCliConn
*
)(
pMsg
->
msg
.
handle
);
transUnrefCliHandle
(
conn
);
if
(
conn
!=
NULL
)
{
tTrace
(
"%s cli
ent
conn %p reused"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
tTrace
(
"%s cli conn %p reused"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
}
}
else
{
STransConnCtx
*
pCtx
=
pMsg
->
ctx
;
conn
=
getConnFromPool
(
pThrd
->
pool
,
pCtx
->
ip
,
pCtx
->
port
);
if
(
conn
!=
NULL
)
tTrace
(
"%s cli
ent
conn %p get from conn pool"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
if
(
conn
!=
NULL
)
tTrace
(
"%s cli conn %p get from conn pool"
,
CONN_GET_INST_LABEL
(
conn
),
conn
);
}
return
conn
;
}
static
void
cli
ent
HandleReq
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
)
{
static
void
cliHandleReq
(
SCliMsg
*
pMsg
,
SCliThrdObj
*
pThrd
)
{
uint64_t
et
=
taosGetTimestampUs
();
uint64_t
el
=
et
-
pMsg
->
st
;
tTrace
(
"%s cli
ent
msg tran time cost: %"
PRIu64
"us"
,
((
SRpcInfo
*
)
pThrd
->
pTransInst
)
->
label
,
el
);
tTrace
(
"%s cli msg tran time cost: %"
PRIu64
"us"
,
((
SRpcInfo
*
)
pThrd
->
pTransInst
)
->
label
,
el
);
STransConnCtx
*
pCtx
=
pMsg
->
ctx
;
SRpcInfo
*
pTransInst
=
pThrd
->
pTransInst
;
SCliConn
*
conn
=
cli
ent
GetConn
(
pMsg
,
pThrd
);
SCliConn
*
conn
=
cliGetConn
(
pMsg
,
pThrd
);
if
(
conn
!=
NULL
)
{
conn
->
data
=
pMsg
;
transDestroyBuffer
(
&
conn
->
readBuf
);
cli
entSendData
(
conn
);
cli
Send
(
conn
);
}
else
{
conn
=
cli
entConnCreate
(
pThrd
);
conn
=
cli
CreateConn
(
pThrd
);
conn
->
data
=
pMsg
;
int
ret
=
transSetConnOption
((
uv_tcp_t
*
)
conn
->
stream
);
if
(
ret
)
{
tError
(
"%s cli
ent
conn %p failed to set conn option, errmsg %s"
,
pTransInst
->
label
,
conn
,
uv_err_name
(
ret
));
tError
(
"%s cli conn %p failed to set conn option, errmsg %s"
,
pTransInst
->
label
,
conn
,
uv_err_name
(
ret
));
}
struct
sockaddr_in
addr
;
uv_ip4_addr
(
pMsg
->
ctx
->
ip
,
pMsg
->
ctx
->
port
,
&
addr
);
// handle error in callback if fail to connect
tTrace
(
"%s cli
ent
conn %p try to connect to %s:%d"
,
pTransInst
->
label
,
conn
,
pMsg
->
ctx
->
ip
,
pMsg
->
ctx
->
port
);
uv_tcp_connect
(
&
conn
->
connReq
,
(
uv_tcp_t
*
)(
conn
->
stream
),
(
const
struct
sockaddr
*
)
&
addr
,
cli
ent
ConnCb
);
tTrace
(
"%s cli conn %p try to connect to %s:%d"
,
pTransInst
->
label
,
conn
,
pMsg
->
ctx
->
ip
,
pMsg
->
ctx
->
port
);
uv_tcp_connect
(
&
conn
->
connReq
,
(
uv_tcp_t
*
)(
conn
->
stream
),
(
const
struct
sockaddr
*
)
&
addr
,
cliConnCb
);
}
conn
->
hThrdIdx
=
pCtx
->
hThrdIdx
;
}
static
void
cli
ent
AsyncCb
(
uv_async_t
*
handle
)
{
static
void
cliAsyncCb
(
uv_async_t
*
handle
)
{
SAsyncItem
*
item
=
handle
->
data
;
SCliThrdObj
*
pThrd
=
item
->
pThrd
;
SCliMsg
*
pMsg
=
NULL
;
queue
wq
;
// batch process to avoid to lock/unlock frequently
queue
wq
;
pthread_mutex_lock
(
&
item
->
mtx
);
QUEUE_MOVE
(
&
item
->
qmsg
,
&
wq
);
pthread_mutex_unlock
(
&
item
->
mtx
);
...
...
@@ -534,25 +534,25 @@ static void clientAsyncCb(uv_async_t* handle) {
SCliMsg
*
pMsg
=
QUEUE_DATA
(
h
,
SCliMsg
,
q
);
if
(
pMsg
->
ctx
==
NULL
)
{
cli
ent
HandleQuit
(
pMsg
,
pThrd
);
cliHandleQuit
(
pMsg
,
pThrd
);
}
else
{
cli
ent
HandleReq
(
pMsg
,
pThrd
);
cliHandleReq
(
pMsg
,
pThrd
);
}
count
++
;
}
if
(
count
>=
2
)
{
tTrace
(
"cli
ent
process batch size: %d"
,
count
);
tTrace
(
"cli process batch size: %d"
,
count
);
}
}
static
void
*
cli
ent
Thread
(
void
*
arg
)
{
static
void
*
cli
Work
Thread
(
void
*
arg
)
{
SCliThrdObj
*
pThrd
=
(
SCliThrdObj
*
)
arg
;
setThreadName
(
"trans-cli
ent
-work"
);
setThreadName
(
"trans-cli-work"
);
uv_run
(
pThrd
->
loop
,
UV_RUN_DEFAULT
);
}
void
*
taosInitClient
(
uint32_t
ip
,
uint32_t
port
,
char
*
label
,
int
numOfThreads
,
void
*
fp
,
void
*
shandle
)
{
SCli
entObj
*
cli
=
calloc
(
1
,
sizeof
(
SClient
Obj
));
SCli
Obj
*
cli
=
calloc
(
1
,
sizeof
(
SCli
Obj
));
SRpcInfo
*
pRpc
=
shandle
;
memcpy
(
cli
->
label
,
label
,
strlen
(
label
));
...
...
@@ -564,9 +564,9 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
pThrd
->
nextTimeout
=
taosGetTimestampMs
()
+
CONN_PERSIST_TIME
(
pRpc
->
idleTime
);
pThrd
->
pTransInst
=
shandle
;
int
err
=
pthread_create
(
&
pThrd
->
thread
,
NULL
,
cli
ent
Thread
,
(
void
*
)(
pThrd
));
int
err
=
pthread_create
(
&
pThrd
->
thread
,
NULL
,
cli
Work
Thread
,
(
void
*
)(
pThrd
));
if
(
err
==
0
)
{
tDebug
(
"success to create tranport-cli
ent
thread %d"
,
i
);
tDebug
(
"success to create tranport-cli thread %d"
,
i
);
}
cli
->
pThreadObj
[
i
]
=
pThrd
;
}
...
...
@@ -591,13 +591,14 @@ static void destroyCmsg(SCliMsg* pMsg) {
static
SCliThrdObj
*
createThrdObj
()
{
SCliThrdObj
*
pThrd
=
(
SCliThrdObj
*
)
calloc
(
1
,
sizeof
(
SCliThrdObj
));
QUEUE_INIT
(
&
pThrd
->
msg
);
pthread_mutex_init
(
&
pThrd
->
msgMtx
,
NULL
);
pThrd
->
loop
=
(
uv_loop_t
*
)
malloc
(
sizeof
(
uv_loop_t
));
uv_loop_init
(
pThrd
->
loop
);
pThrd
->
asyncPool
=
transCreateAsyncPool
(
pThrd
->
loop
,
5
,
pThrd
,
cli
ent
AsyncCb
);
pThrd
->
asyncPool
=
transCreateAsyncPool
(
pThrd
->
loop
,
5
,
pThrd
,
cliAsyncCb
);
uv_timer_init
(
pThrd
->
loop
,
&
pThrd
->
timer
);
pThrd
->
timer
.
data
=
pThrd
;
...
...
@@ -628,21 +629,21 @@ static void transDestroyConnCtx(STransConnCtx* ctx) {
free
(
ctx
);
}
//
static
void
cli
ent
SendQuit
(
SCliThrdObj
*
thrd
)
{
static
void
cliSendQuit
(
SCliThrdObj
*
thrd
)
{
// cli can stop gracefully
SCliMsg
*
msg
=
calloc
(
1
,
sizeof
(
SCliMsg
));
transSendAsync
(
thrd
->
asyncPool
,
&
msg
->
q
);
}
void
taosCloseClient
(
void
*
arg
)
{
SCli
ent
Obj
*
cli
=
arg
;
SCliObj
*
cli
=
arg
;
for
(
int
i
=
0
;
i
<
cli
->
numOfThreads
;
i
++
)
{
cli
ent
SendQuit
(
cli
->
pThreadObj
[
i
]);
cliSendQuit
(
cli
->
pThreadObj
[
i
]);
destroyThrdObj
(
cli
->
pThreadObj
[
i
]);
}
free
(
cli
->
pThreadObj
);
free
(
cli
);
}
static
int
cli
ent
RBChoseIdx
(
SRpcInfo
*
pTransInst
)
{
static
int
cliRBChoseIdx
(
SRpcInfo
*
pTransInst
)
{
int64_t
index
=
pTransInst
->
index
;
if
(
pTransInst
->
index
++
>=
pTransInst
->
numOfThreads
)
{
pTransInst
->
index
=
0
;
...
...
@@ -662,7 +663,7 @@ void transUnrefCliHandle(void* handle) {
}
int
ref
=
T_REF_DEC
((
SCliConn
*
)
handle
);
if
(
ref
==
0
)
{
cli
entConnDestroy
((
SCliConn
*
)
handle
,
true
);
cli
DestroyConn
((
SCliConn
*
)
handle
,
true
);
}
// unref cli handle
...
...
@@ -676,7 +677,7 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t*
int
index
=
CONN_HOST_THREAD_INDEX
(
pMsg
->
handle
);
if
(
index
==
-
1
)
{
index
=
cli
ent
RBChoseIdx
(
pTransInst
);
index
=
cliRBChoseIdx
(
pTransInst
);
}
int32_t
flen
=
0
;
if
(
transCompressMsg
(
pMsg
->
pCont
,
pMsg
->
contLen
,
&
flen
))
{
...
...
@@ -697,7 +698,7 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t*
cliMsg
->
msg
=
*
pMsg
;
cliMsg
->
st
=
taosGetTimestampUs
();
SCliThrdObj
*
thrd
=
((
SCli
ent
Obj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
SCliThrdObj
*
thrd
=
((
SCliObj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
transSendAsync
(
thrd
->
asyncPool
,
&
(
cliMsg
->
q
));
}
...
...
@@ -709,7 +710,7 @@ void rpcSendRecv(void* shandle, SEpSet* pEpSet, SRpcMsg* pReq, SRpcMsg* pRsp) {
int
index
=
CONN_HOST_THREAD_INDEX
(
pReq
->
handle
);
if
(
index
==
-
1
)
{
index
=
cli
ent
RBChoseIdx
(
pTransInst
);
index
=
cliRBChoseIdx
(
pTransInst
);
}
STransConnCtx
*
pCtx
=
calloc
(
1
,
sizeof
(
STransConnCtx
));
...
...
@@ -727,7 +728,7 @@ void rpcSendRecv(void* shandle, SEpSet* pEpSet, SRpcMsg* pReq, SRpcMsg* pRsp) {
cliMsg
->
msg
=
*
pReq
;
cliMsg
->
st
=
taosGetTimestampUs
();
SCliThrdObj
*
thrd
=
((
SCli
ent
Obj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
SCliThrdObj
*
thrd
=
((
SCliObj
*
)
pTransInst
->
tcphandle
)
->
pThreadObj
[
index
];
transSendAsync
(
thrd
->
asyncPool
,
&
(
cliMsg
->
q
));
tsem_t
*
pSem
=
pCtx
->
pSem
;
tsem_wait
(
pSem
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录