Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
adminstrot
TDengine
提交
5621f48f
TDengine
项目概览
adminstrot
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
1
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
5621f48f
编写于
10月 05, 2020
作者:
S
Shengliang Guan
提交者:
GitHub
10月 05, 2020
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #3744 from taosdata/patch/TD-1645
remove a race condition in TCP socket
上级
c522e902
c608fba8
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
52 addition
and
22 deletion
+52
-22
src/connector/go
src/connector/go
+1
-1
src/rpc/src/rpcTcp.c
src/rpc/src/rpcTcp.c
+47
-20
src/sync/src/taosTcpPool.c
src/sync/src/taosTcpPool.c
+4
-1
未找到文件。
go
@
8c58c512
比较
567b7b12
...
8c58c512
Subproject commit
567b7b12f3fd2775c718d284beffc8c38dd6c219
Subproject commit
8c58c512b6acda8bcdfa48fdc7140227b5221766
src/rpc/src/rpcTcp.c
浏览文件 @
5621f48f
...
...
@@ -62,7 +62,7 @@ typedef struct {
char
label
[
TSDB_LABEL_LEN
];
int
numOfThreads
;
void
*
shandle
;
SThreadObj
*
pThreadObj
;
SThreadObj
*
*
pThreadObj
;
pthread_t
thread
;
}
SServerObj
;
...
...
@@ -90,7 +90,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
tstrncpy
(
pServerObj
->
label
,
label
,
sizeof
(
pServerObj
->
label
));
pServerObj
->
numOfThreads
=
numOfThreads
;
pServerObj
->
pThreadObj
=
(
SThreadObj
*
)
calloc
(
sizeof
(
SThreadObj
),
numOfThreads
);
pServerObj
->
pThreadObj
=
(
SThreadObj
*
*
)
calloc
(
sizeof
(
SThreadObj
*
),
numOfThreads
);
if
(
pServerObj
->
pThreadObj
==
NULL
)
{
tError
(
"TCP:%s no enough memory"
,
label
);
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
...
...
@@ -104,19 +104,28 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
pthread_attr_setdetachstate
(
&
thattr
,
PTHREAD_CREATE_JOINABLE
);
// initialize parameters in case it may encounter error later
pThreadObj
=
pServerObj
->
pThreadObj
;
for
(
int
i
=
0
;
i
<
numOfThreads
;
++
i
)
{
pThreadObj
=
(
SThreadObj
*
)
calloc
(
sizeof
(
SThreadObj
),
1
);
if
(
pThreadObj
==
NULL
)
{
tError
(
"TCP:%s no enough memory"
,
label
);
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
for
(
int
j
=
0
;
j
<
i
;
++
j
)
free
(
pServerObj
->
pThreadObj
[
j
]);
free
(
pServerObj
->
pThreadObj
);
free
(
pServerObj
);
return
NULL
;
}
pServerObj
->
pThreadObj
[
i
]
=
pThreadObj
;
pThreadObj
->
pollFd
=
-
1
;
taosResetPthread
(
&
pThreadObj
->
thread
);
pThreadObj
->
processData
=
fp
;
tstrncpy
(
pThreadObj
->
label
,
label
,
sizeof
(
pThreadObj
->
label
));
pThreadObj
->
shandle
=
shandle
;
pThreadObj
++
;
}
// initialize mutex, thread, fd which may fail
pThreadObj
=
pServerObj
->
pThreadObj
;
for
(
int
i
=
0
;
i
<
numOfThreads
;
++
i
)
{
pThreadObj
=
pServerObj
->
pThreadObj
[
i
];
code
=
pthread_mutex_init
(
&
(
pThreadObj
->
mutex
),
NULL
);
if
(
code
<
0
)
{
tError
(
"%s failed to init TCP process data mutex(%s)"
,
label
,
strerror
(
errno
));
...
...
@@ -137,7 +146,6 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
}
pThreadObj
->
threadId
=
i
;
pThreadObj
++
;
}
pServerObj
->
fd
=
taosOpenTcpServerSocket
(
pServerObj
->
ip
,
pServerObj
->
port
);
...
...
@@ -166,6 +174,11 @@ static void taosStopTcpThread(SThreadObj* pThreadObj) {
pThreadObj
->
stop
=
true
;
eventfd_t
fd
=
-
1
;
if
(
taosComparePthread
(
pThreadObj
->
thread
,
pthread_self
()))
{
pthread_detach
(
pthread_self
());
return
;
}
if
(
taosCheckPthreadValid
(
pThreadObj
->
thread
)
&&
pThreadObj
->
pollFd
>=
0
)
{
// signal the thread to stop, try graceful method first,
// and use pthread_cancel when failed
...
...
@@ -183,15 +196,11 @@ static void taosStopTcpThread(SThreadObj* pThreadObj) {
}
}
if
(
taosCheckPthreadValid
(
pThreadObj
->
thread
))
pthread_join
(
pThreadObj
->
thread
,
NULL
);
if
(
pThreadObj
->
pollFd
>=
0
)
taosCloseSocket
(
pThreadObj
->
pollFd
);
if
(
fd
!=
-
1
)
taosCloseSocket
(
fd
);
while
(
pThreadObj
->
pHead
)
{
SFdObj
*
pFdObj
=
pThreadObj
->
pHead
;
pThreadObj
->
pHead
=
pFdObj
->
next
;
taosFreeFdObj
(
pFdObj
);
if
(
taosCheckPthreadValid
(
pThreadObj
->
thread
)
&&
pThreadObj
->
pollFd
>=
0
)
{
pthread_join
(
pThreadObj
->
thread
,
NULL
);
}
if
(
fd
!=
-
1
)
taosCloseSocket
(
fd
);
}
void
taosStopTcpServer
(
void
*
handle
)
{
...
...
@@ -199,7 +208,14 @@ void taosStopTcpServer(void *handle) {
if
(
pServerObj
==
NULL
)
return
;
if
(
pServerObj
->
fd
>=
0
)
shutdown
(
pServerObj
->
fd
,
SHUT_RD
);
if
(
taosCheckPthreadValid
(
pServerObj
->
thread
))
pthread_join
(
pServerObj
->
thread
,
NULL
);
if
(
taosCheckPthreadValid
(
pServerObj
->
thread
))
{
if
(
taosComparePthread
(
pServerObj
->
thread
,
pthread_self
()))
{
pthread_detach
(
pthread_self
());
}
else
{
pthread_join
(
pServerObj
->
thread
,
NULL
);
}
}
tDebug
(
"%s TCP server is stopped"
,
pServerObj
->
label
);
}
...
...
@@ -210,9 +226,8 @@ void taosCleanUpTcpServer(void *handle) {
if
(
pServerObj
==
NULL
)
return
;
for
(
int
i
=
0
;
i
<
pServerObj
->
numOfThreads
;
++
i
)
{
pThreadObj
=
pServerObj
->
pThreadObj
+
i
;
pThreadObj
=
pServerObj
->
pThreadObj
[
i
]
;
taosStopTcpThread
(
pThreadObj
);
pthread_mutex_destroy
(
&
(
pThreadObj
->
mutex
));
}
tDebug
(
"%s TCP server is cleaned up"
,
pServerObj
->
label
);
...
...
@@ -249,7 +264,7 @@ static void *taosAcceptTcpConnection(void *arg) {
taosSetSockOpt
(
connFd
,
SOL_SOCKET
,
SO_RCVTIMEO
,
&
to
,
sizeof
(
to
));
// pick up the thread to handle this connection
pThreadObj
=
pServerObj
->
pThreadObj
+
threadId
;
pThreadObj
=
pServerObj
->
pThreadObj
[
threadId
]
;
SFdObj
*
pFdObj
=
taosMallocFdObj
(
pThreadObj
,
connFd
);
if
(
pFdObj
)
{
...
...
@@ -329,8 +344,6 @@ void taosCleanUpTcpClient(void *chandle) {
taosStopTcpThread
(
pThreadObj
);
tDebug
(
"%s TCP client is cleaned up"
,
pThreadObj
->
label
);
taosTFree
(
pThreadObj
);
}
void
*
taosOpenTcpClientConnection
(
void
*
shandle
,
void
*
thandle
,
uint32_t
ip
,
uint16_t
port
)
{
...
...
@@ -503,8 +516,22 @@ static void *taosProcessTcpData(void *param) {
pFdObj
->
thandle
=
(
*
(
pThreadObj
->
processData
))(
&
recvInfo
);
if
(
pFdObj
->
thandle
==
NULL
)
taosFreeFdObj
(
pFdObj
);
}
if
(
pThreadObj
->
stop
)
break
;
}
if
(
pThreadObj
->
pollFd
>=
0
)
taosCloseSocket
(
pThreadObj
->
pollFd
);
while
(
pThreadObj
->
pHead
)
{
SFdObj
*
pFdObj
=
pThreadObj
->
pHead
;
pThreadObj
->
pHead
=
pFdObj
->
next
;
taosFreeFdObj
(
pFdObj
);
}
pthread_mutex_destroy
(
&
(
pThreadObj
->
mutex
));
tDebug
(
"%s TCP thread exits ..."
,
pThreadObj
->
label
);
taosTFree
(
pThreadObj
);
return
NULL
;
}
...
...
src/sync/src/taosTcpPool.c
浏览文件 @
5621f48f
...
...
@@ -219,7 +219,10 @@ static void *taosProcessTcpData(void *param) {
continue
;
}
}
}
if
(
pThread
->
stop
)
break
;
}
uDebug
(
"%p TCP epoll thread exits"
,
pThread
);
...
...
@@ -321,5 +324,5 @@ static void taosStopPoolThread(SThreadObj *pThread) {
}
pthread_join
(
thread
,
NULL
);
taosClose
(
fd
);
if
(
fd
>=
0
)
taosClose
(
fd
);
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录