Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
XKZJ_WCL
TDengine
提交
966a5e18
T
TDengine
项目概览
XKZJ_WCL
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
966a5e18
编写于
12月 15, 2020
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
rename some files
上级
7c1c99d3
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
50 addition
and
53 deletion
+50
-53
src/sync/CMakeLists.txt
src/sync/CMakeLists.txt
+3
-3
src/sync/inc/syncTcp.h
src/sync/inc/syncTcp.h
+6
-9
src/sync/src/syncArbitrator.c
src/sync/src/syncArbitrator.c
+11
-11
src/sync/src/syncMain.c
src/sync/src/syncMain.c
+12
-12
src/sync/src/syncTcp.c
src/sync/src/syncTcp.c
+18
-18
未找到文件。
src/sync/CMakeLists.txt
浏览文件 @
966a5e18
...
...
@@ -5,12 +5,12 @@ INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY
(
src SRC
)
IF
(
TD_LINUX
)
LIST
(
REMOVE_ITEM SRC src/
ta
rbitrator.c
)
LIST
(
REMOVE_ITEM SRC src/
syncA
rbitrator.c
)
ADD_LIBRARY
(
sync
${
SRC
}
)
TARGET_LINK_LIBRARIES
(
sync tutil pthread common
)
LIST
(
APPEND BIN_SRC src/
ta
rbitrator.c
)
LIST
(
APPEND BIN_SRC src/
taosTcpPool
.c
)
LIST
(
APPEND BIN_SRC src/
syncA
rbitrator.c
)
LIST
(
APPEND BIN_SRC src/
syncTcp
.c
)
ADD_EXECUTABLE
(
tarbitrator
${
BIN_SRC
}
)
TARGET_LINK_LIBRARIES
(
tarbitrator sync common osdetail tutil
)
...
...
src/sync/inc/
taosTcpPool
.h
→
src/sync/inc/
syncTcp
.h
浏览文件 @
966a5e18
...
...
@@ -13,16 +13,13 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_TCP_POOL_H
#define TDENGINE_TCP_POOL_H
#ifndef TDENGINE_
SYNC_
TCP_POOL_H
#define TDENGINE_
SYNC_
TCP_POOL_H
#ifdef __cplusplus
extern
"C"
{
#endif
typedef
void
*
ttpool_h
;
typedef
void
*
tthread_h
;
typedef
struct
{
int32_t
numOfThreads
;
uint32_t
serverIp
;
...
...
@@ -33,10 +30,10 @@ typedef struct {
void
(
*
processIncomingConn
)(
int32_t
fd
,
uint32_t
ip
);
}
SPoolInfo
;
ttpool_h
taos
OpenTcpThreadPool
(
SPoolInfo
*
pInfo
);
void
taosCloseTcpThreadPool
(
ttpool_h
);
void
*
taos
AllocateTcpConn
(
void
*
,
void
*
ahandle
,
int32_t
connFd
);
void
taos
FreeTcpConn
(
void
*
);
void
*
sync
OpenTcpThreadPool
(
SPoolInfo
*
pInfo
);
void
syncCloseTcpThreadPool
(
void
*
);
void
*
sync
AllocateTcpConn
(
void
*
,
void
*
ahandle
,
int32_t
connFd
);
void
sync
FreeTcpConn
(
void
*
);
#ifdef __cplusplus
}
...
...
src/sync/src/
ta
rbitrator.c
→
src/sync/src/
syncA
rbitrator.c
浏览文件 @
966a5e18
...
...
@@ -22,17 +22,17 @@
#include "tsocket.h"
#include "tglobal.h"
#include "taoserror.h"
#include "taosTcpPool.h"
#include "twal.h"
#include "tsync.h"
#include "syncInt.h"
#include "syncTcp.h"
static
void
arbSignalHandler
(
int32_t
signum
,
siginfo_t
*
sigInfo
,
void
*
context
);
static
void
arbProcessIncommingConnection
(
int32_t
connFd
,
uint32_t
sourceIp
);
static
void
arbProcessBrokenLink
(
void
*
param
);
static
int32_t
arbProcessPeerMsg
(
void
*
param
,
void
*
buffer
);
static
tsem_t
tsArbSem
;
static
ttpool_h
tsArbTcpPool
;
static
void
arbSignalHandler
(
int32_t
signum
,
siginfo_t
*
sigInfo
,
void
*
context
);
static
void
arbProcessIncommingConnection
(
int32_t
connFd
,
uint32_t
sourceIp
);
static
void
arbProcessBrokenLink
(
void
*
param
);
static
int32_t
arbProcessPeerMsg
(
void
*
param
,
void
*
buffer
);
static
tsem_t
tsArbSem
;
static
void
*
tsArbTcpPool
;
typedef
struct
{
char
id
[
TSDB_EP_LEN
+
24
];
...
...
@@ -90,7 +90,7 @@ int32_t main(int32_t argc, char *argv[]) {
info
.
processBrokenLink
=
arbProcessBrokenLink
;
info
.
processIncomingMsg
=
arbProcessPeerMsg
;
info
.
processIncomingConn
=
arbProcessIncommingConnection
;
tsArbTcpPool
=
taos
OpenTcpThreadPool
(
&
info
);
tsArbTcpPool
=
sync
OpenTcpThreadPool
(
&
info
);
if
(
tsArbTcpPool
==
NULL
)
{
sDebug
(
"failed to open TCP thread pool, exit..."
);
...
...
@@ -101,8 +101,8 @@ int32_t main(int32_t argc, char *argv[]) {
tsem_wait
(
&
tsArbSem
);
taos
CloseTcpThreadPool
(
tsArbTcpPool
);
sInfo
(
"TAOS arbitrator is shut down
\n
"
);
sync
CloseTcpThreadPool
(
tsArbTcpPool
);
sInfo
(
"TAOS arbitrator is shut down"
);
closelog
();
return
0
;
...
...
@@ -138,7 +138,7 @@ static void arbProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
sDebug
(
"%s, arbitrator request is accepted"
,
pNode
->
id
);
pNode
->
nodeFd
=
connFd
;
pNode
->
pConn
=
taos
AllocateTcpConn
(
tsArbTcpPool
,
pNode
,
connFd
);
pNode
->
pConn
=
sync
AllocateTcpConn
(
tsArbTcpPool
,
pNode
,
connFd
);
return
;
}
...
...
src/sync/src/syncMain.c
浏览文件 @
966a5e18
...
...
@@ -23,10 +23,10 @@
#include "tsocket.h"
#include "tglobal.h"
#include "taoserror.h"
#include "taosTcpPool.h"
#include "tqueue.h"
#include "twal.h"
#include "tsync.h"
#include "syncTcp.h"
#include "syncInt.h"
// global configurable
...
...
@@ -39,10 +39,10 @@ int32_t tsSyncTimer = 1;
int32_t
tsSyncNum
;
// number of sync in process in whole system
char
tsNodeFqdn
[
TSDB_FQDN_LEN
];
static
ttpool_h
tsTcpPool
;
static
void
*
tsSyncTmrCtrl
=
NULL
;
static
void
*
tsVgIdHash
;
static
int32_t
tsSyncRefId
=
-
1
;
static
void
*
tsTcpPool
;
static
void
*
tsSyncTmrCtrl
=
NULL
;
static
void
*
tsVgIdHash
;
static
int32_t
tsSyncRefId
=
-
1
;
// local functions
static
void
syncProcessSyncRequest
(
char
*
pMsg
,
SSyncPeer
*
pPeer
);
...
...
@@ -117,7 +117,7 @@ int32_t syncInit() {
info
.
processIncomingMsg
=
syncProcessPeerMsg
;
info
.
processIncomingConn
=
syncProcessIncommingConnection
;
tsTcpPool
=
taos
OpenTcpThreadPool
(
&
info
);
tsTcpPool
=
sync
OpenTcpThreadPool
(
&
info
);
if
(
tsTcpPool
==
NULL
)
{
sError
(
"failed to init tcpPool"
);
return
-
1
;
...
...
@@ -126,7 +126,7 @@ int32_t syncInit() {
tsSyncTmrCtrl
=
taosTmrInit
(
1000
,
50
,
10000
,
"SYNC"
);
if
(
tsSyncTmrCtrl
==
NULL
)
{
sError
(
"failed to init tmrCtrl"
);
taos
CloseTcpThreadPool
(
tsTcpPool
);
sync
CloseTcpThreadPool
(
tsTcpPool
);
tsTcpPool
=
NULL
;
return
-
1
;
}
...
...
@@ -135,7 +135,7 @@ int32_t syncInit() {
if
(
tsVgIdHash
==
NULL
)
{
sError
(
"failed to init tsVgIdHash"
);
taosTmrCleanUp
(
tsSyncTmrCtrl
);
taos
CloseTcpThreadPool
(
tsTcpPool
);
sync
CloseTcpThreadPool
(
tsTcpPool
);
tsTcpPool
=
NULL
;
tsSyncTmrCtrl
=
NULL
;
return
-
1
;
...
...
@@ -155,7 +155,7 @@ int32_t syncInit() {
void
syncCleanUp
()
{
if
(
tsTcpPool
)
{
taos
CloseTcpThreadPool
(
tsTcpPool
);
sync
CloseTcpThreadPool
(
tsTcpPool
);
tsTcpPool
=
NULL
;
}
...
...
@@ -509,7 +509,7 @@ static void syncClosePeerConn(SSyncPeer *pPeer) {
taosClose
(
pPeer
->
syncFd
);
if
(
pPeer
->
peerFd
>=
0
)
{
pPeer
->
peerFd
=
-
1
;
taos
FreeTcpConn
(
pPeer
->
pConn
);
sync
FreeTcpConn
(
pPeer
->
pConn
);
}
}
...
...
@@ -1065,7 +1065,7 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) {
sDebug
(
"%s, connection to peer server is setup, pfd:%d sfd:%d tranId:%u"
,
pPeer
->
id
,
connFd
,
pPeer
->
syncFd
,
firstPkt
.
tranId
);
pPeer
->
peerFd
=
connFd
;
pPeer
->
role
=
TAOS_SYNC_ROLE_UNSYNCED
;
pPeer
->
pConn
=
taos
AllocateTcpConn
(
tsTcpPool
,
pPeer
,
connFd
);
pPeer
->
pConn
=
sync
AllocateTcpConn
(
tsTcpPool
,
pPeer
,
connFd
);
syncAddPeerRef
(
pPeer
);
}
else
{
sDebug
(
"%s, failed to setup peer connection to server since %s, try later"
,
pPeer
->
id
,
strerror
(
errno
));
...
...
@@ -1159,7 +1159,7 @@ static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
sDebug
(
"%s, TCP connection is up, pfd:%d sfd:%d, old pfd:%d"
,
pPeer
->
id
,
connFd
,
pPeer
->
syncFd
,
pPeer
->
peerFd
);
syncClosePeerConn
(
pPeer
);
pPeer
->
peerFd
=
connFd
;
pPeer
->
pConn
=
taos
AllocateTcpConn
(
tsTcpPool
,
pPeer
,
connFd
);
pPeer
->
pConn
=
sync
AllocateTcpConn
(
tsTcpPool
,
pPeer
,
connFd
);
syncAddPeerRef
(
pPeer
);
sDebug
(
"%s, ready to exchange data"
,
pPeer
->
id
);
syncSendPeersStatusMsgToPeer
(
pPeer
,
1
,
SYNC_STATUS_EXCHANGE_DATA
,
syncGenTranId
());
...
...
src/sync/src/
taosTcpPool
.c
→
src/sync/src/
syncTcp
.c
浏览文件 @
966a5e18
...
...
@@ -19,10 +19,10 @@
#include "tutil.h"
#include "tsocket.h"
#include "taoserror.h"
#include "taosTcpPool.h"
#include "twal.h"
#include "tsync.h"
#include "syncInt.h"
#include "syncTcp.h"
typedef
struct
SThreadObj
{
pthread_t
thread
;
...
...
@@ -47,12 +47,12 @@ typedef struct {
int32_t
closedByApp
;
}
SConnObj
;
static
void
*
taos
AcceptPeerTcpConnection
(
void
*
argv
);
static
void
*
taos
ProcessTcpData
(
void
*
param
);
static
void
taos
StopPoolThread
(
SThreadObj
*
pThread
);
static
SThreadObj
*
taos
GetTcpThread
(
SPoolObj
*
pPool
);
static
void
*
sync
AcceptPeerTcpConnection
(
void
*
argv
);
static
void
*
sync
ProcessTcpData
(
void
*
param
);
static
void
sync
StopPoolThread
(
SThreadObj
*
pThread
);
static
SThreadObj
*
sync
GetTcpThread
(
SPoolObj
*
pPool
);
void
*
taos
OpenTcpThreadPool
(
SPoolInfo
*
pInfo
)
{
void
*
sync
OpenTcpThreadPool
(
SPoolInfo
*
pInfo
)
{
pthread_attr_t
thattr
;
SPoolObj
*
pPool
=
calloc
(
sizeof
(
SPoolObj
),
1
);
...
...
@@ -80,7 +80,7 @@ void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
pthread_attr_init
(
&
thattr
);
pthread_attr_setdetachstate
(
&
thattr
,
PTHREAD_CREATE_JOINABLE
);
if
(
pthread_create
(
&
(
pPool
->
thread
),
&
thattr
,
(
void
*
)
taos
AcceptPeerTcpConnection
,
pPool
)
!=
0
)
{
if
(
pthread_create
(
&
(
pPool
->
thread
),
&
thattr
,
(
void
*
)
sync
AcceptPeerTcpConnection
,
pPool
)
!=
0
)
{
sError
(
"failed to create accept thread for TCP server since %s"
,
strerror
(
errno
));
close
(
pPool
->
acceptFd
);
tfree
(
pPool
->
pThread
);
...
...
@@ -94,7 +94,7 @@ void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
return
pPool
;
}
void
taos
CloseTcpThreadPool
(
void
*
param
)
{
void
sync
CloseTcpThreadPool
(
void
*
param
)
{
SPoolObj
*
pPool
=
param
;
SThreadObj
*
pThread
;
...
...
@@ -103,7 +103,7 @@ void taosCloseTcpThreadPool(void *param) {
for
(
int32_t
i
=
0
;
i
<
pPool
->
info
.
numOfThreads
;
++
i
)
{
pThread
=
pPool
->
pThread
[
i
];
if
(
pThread
)
taos
StopPoolThread
(
pThread
);
if
(
pThread
)
sync
StopPoolThread
(
pThread
);
}
sDebug
(
"%p TCP pool is closed"
,
pPool
);
...
...
@@ -112,7 +112,7 @@ void taosCloseTcpThreadPool(void *param) {
tfree
(
pPool
);
}
void
*
taos
AllocateTcpConn
(
void
*
param
,
void
*
pPeer
,
int32_t
connFd
)
{
void
*
sync
AllocateTcpConn
(
void
*
param
,
void
*
pPeer
,
int32_t
connFd
)
{
struct
epoll_event
event
;
SPoolObj
*
pPool
=
param
;
...
...
@@ -122,7 +122,7 @@ void *taosAllocateTcpConn(void *param, void *pPeer, int32_t connFd) {
return
NULL
;
}
SThreadObj
*
pThread
=
taos
GetTcpThread
(
pPool
);
SThreadObj
*
pThread
=
sync
GetTcpThread
(
pPool
);
if
(
pThread
==
NULL
)
{
tfree
(
pConn
);
return
NULL
;
...
...
@@ -149,7 +149,7 @@ void *taosAllocateTcpConn(void *param, void *pPeer, int32_t connFd) {
return
pConn
;
}
void
taos
FreeTcpConn
(
void
*
param
)
{
void
sync
FreeTcpConn
(
void
*
param
)
{
SConnObj
*
pConn
=
param
;
SThreadObj
*
pThread
=
pConn
->
pThread
;
...
...
@@ -175,7 +175,7 @@ static void taosProcessBrokenLink(SConnObj *pConn) {
#define maxEvents 10
static
void
*
taos
ProcessTcpData
(
void
*
param
)
{
static
void
*
sync
ProcessTcpData
(
void
*
param
)
{
SThreadObj
*
pThread
=
(
SThreadObj
*
)
param
;
SPoolObj
*
pPool
=
pThread
->
pPool
;
SPoolInfo
*
pInfo
=
&
pPool
->
info
;
...
...
@@ -222,7 +222,7 @@ static void *taosProcessTcpData(void *param) {
if
(
pConn
->
closedByApp
==
0
)
{
if
((
*
pInfo
->
processIncomingMsg
)(
pConn
->
ahandle
,
buffer
)
<
0
)
{
taos
FreeTcpConn
(
pConn
);
sync
FreeTcpConn
(
pConn
);
continue
;
}
}
...
...
@@ -239,7 +239,7 @@ static void *taosProcessTcpData(void *param) {
return
NULL
;
}
static
void
*
taos
AcceptPeerTcpConnection
(
void
*
argv
)
{
static
void
*
sync
AcceptPeerTcpConnection
(
void
*
argv
)
{
SPoolObj
*
pPool
=
(
SPoolObj
*
)
argv
;
SPoolInfo
*
pInfo
=
&
pPool
->
info
;
...
...
@@ -268,7 +268,7 @@ static void *taosAcceptPeerTcpConnection(void *argv) {
return
NULL
;
}
static
SThreadObj
*
taos
GetTcpThread
(
SPoolObj
*
pPool
)
{
static
SThreadObj
*
sync
GetTcpThread
(
SPoolObj
*
pPool
)
{
SThreadObj
*
pThread
=
pPool
->
pThread
[
pPool
->
nextId
];
if
(
pThread
)
return
pThread
;
...
...
@@ -286,7 +286,7 @@ static SThreadObj *taosGetTcpThread(SPoolObj *pPool) {
pthread_attr_t
thattr
;
pthread_attr_init
(
&
thattr
);
pthread_attr_setdetachstate
(
&
thattr
,
PTHREAD_CREATE_JOINABLE
);
int32_t
ret
=
pthread_create
(
&
(
pThread
->
thread
),
&
thattr
,
(
void
*
)
taos
ProcessTcpData
,
pThread
);
int32_t
ret
=
pthread_create
(
&
(
pThread
->
thread
),
&
thattr
,
(
void
*
)
sync
ProcessTcpData
,
pThread
);
pthread_attr_destroy
(
&
thattr
);
if
(
ret
!=
0
)
{
...
...
@@ -303,7 +303,7 @@ static SThreadObj *taosGetTcpThread(SPoolObj *pPool) {
return
pThread
;
}
static
void
taos
StopPoolThread
(
SThreadObj
*
pThread
)
{
static
void
sync
StopPoolThread
(
SThreadObj
*
pThread
)
{
pthread_t
thread
=
pThread
->
thread
;
if
(
!
taosCheckPthreadValid
(
thread
))
{
return
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录