Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
lzh_me
TDengine
提交
a5fdb787
TDengine
项目概览
lzh_me
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
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,发现更多精彩内容 >>
提交
a5fdb787
编写于
11月 10, 2020
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
change int to int32_t
上级
42693517
变更
6
展开全部
隐藏空白更改
内联
并排
Showing
6 changed file
with
208 addition
and
209 deletion
+208
-209
src/inc/tsync.h
src/inc/tsync.h
+28
-29
src/sync/inc/syncInt.h
src/sync/inc/syncInt.h
+23
-23
src/sync/src/syncMain.c
src/sync/src/syncMain.c
+103
-103
src/sync/src/syncRestore.c
src/sync/src/syncRestore.c
+13
-13
src/sync/src/syncRetrieve.c
src/sync/src/syncRetrieve.c
+30
-30
src/sync/src/tarbitrator.c
src/sync/src/tarbitrator.c
+11
-11
未找到文件。
src/inc/tsync.h
浏览文件 @
a5fdb787
...
...
@@ -51,9 +51,9 @@ typedef struct {
}
SSyncCfg
;
typedef
struct
{
int
selfIndex
;
uint32_t
nodeId
[
TAOS_SYNC_MAX_REPLICA
];
int
role
[
TAOS_SYNC_MAX_REPLICA
];
int
32_t
selfIndex
;
uint32_t
nodeId
[
TAOS_SYNC_MAX_REPLICA
];
int
32_t
role
[
TAOS_SYNC_MAX_REPLICA
];
}
SNodesRole
;
/*
...
...
@@ -83,25 +83,24 @@ typedef void (*FNotifyRole)(void *ahandle, int8_t role);
typedef
void
(
*
FNotifyFlowCtrl
)(
void
*
ahandle
,
int32_t
mseconds
);
// when data file is synced successfully, notity app
typedef
int
(
*
FNotifyFileSynced
)(
void
*
ahandle
,
uint64_t
fversion
);
typedef
int
32_t
(
*
FNotifyFileSynced
)(
void
*
ahandle
,
uint64_t
fversion
);
typedef
struct
{
int32_t
vgId
;
// vgroup ID
uint64_t
version
;
// initial version
SSyncCfg
syncCfg
;
// configuration from mgmt
char
path
[
128
];
// path to the file
void
*
ahandle
;
// handle provided by APP
FGetFileInfo
getFileInfo
;
FGetWalInfo
getWalInfo
;
FWriteToCache
writeToCache
;
FConfirmForward
confirmForward
;
FNotifyRole
notifyRole
;
FNotifyFlowCtrl
notifyFlowCtrl
;
int32_t
vgId
;
// vgroup ID
uint64_t
version
;
// initial version
SSyncCfg
syncCfg
;
// configuration from mgmt
char
path
[
128
];
// path to the file
void
*
ahandle
;
// handle provided by APP
FGetFileInfo
getFileInfo
;
FGetWalInfo
getWalInfo
;
FWriteToCache
writeToCache
;
FConfirmForward
confirmForward
;
FNotifyRole
notifyRole
;
FNotifyFlowCtrl
notifyFlowCtrl
;
FNotifyFileSynced
notifyFileSynced
;
}
SSyncInfo
;
typedef
void
*
tsync_h
;
typedef
void
*
tsync_h
;
int32_t
syncInit
();
void
syncCleanUp
();
...
...
@@ -109,22 +108,22 @@ void syncCleanUp();
int64_t
syncStart
(
const
SSyncInfo
*
);
void
syncStop
(
int64_t
rid
);
int32_t
syncReconfig
(
int64_t
rid
,
const
SSyncCfg
*
);
int32_t
syncForwardToPeer
(
int64_t
rid
,
void
*
pHead
,
void
*
mhandle
,
int
qtype
);
int32_t
syncForwardToPeer
(
int64_t
rid
,
void
*
pHead
,
void
*
mhandle
,
int
32_t
qtype
);
void
syncConfirmForward
(
int64_t
rid
,
uint64_t
version
,
int32_t
code
);
void
syncRecover
(
int64_t
rid
);
// recover from other nodes:
int
syncGetNodesRole
(
int64_t
rid
,
SNodesRole
*
);
void
syncRecover
(
int64_t
rid
);
// recover from other nodes:
int
32_t
syncGetNodesRole
(
int64_t
rid
,
SNodesRole
*
);
extern
char
*
syncRole
[];
extern
char
*
syncRole
[];
//global configurable parameters
extern
int
tsMaxSyncNum
;
extern
int
tsSyncTcpThreads
;
extern
int
tsMaxWatchFiles
;
extern
int
tsSyncTimer
;
extern
int
tsMaxFwdInfo
;
extern
int
sDebugFlag
;
extern
char
tsArbitrator
[];
extern
uint16_t
tsSyncPort
;
extern
int32_t
tsMaxSyncNum
;
extern
int32_t
tsSyncTcpThreads
;
extern
int32_t
tsMaxWatchFiles
;
extern
int32_t
tsSyncTimer
;
extern
int32_t
tsMaxFwdInfo
;
extern
int32_t
sDebugFlag
;
extern
char
tsArbitrator
[];
extern
uint16_t
tsSyncPort
;
#ifdef __cplusplus
}
...
...
src/sync/inc/syncInt.h
浏览文件 @
a5fdb787
...
...
@@ -89,11 +89,11 @@ typedef struct {
#pragma pack(pop)
typedef
struct
{
char
*
buffer
;
int
bufferSize
;
char
*
offset
;
int
forwards
;
int
code
;
char
*
buffer
;
int
32_t
bufferSize
;
char
*
offset
;
int
32_t
forwards
;
int
32_t
code
;
}
SRecvBuffer
;
typedef
struct
{
...
...
@@ -107,10 +107,10 @@ typedef struct {
}
SFwdInfo
;
typedef
struct
{
int
first
;
int
last
;
int
fwds
;
// number of forwards
SFwdInfo
fwdInfo
[];
int
32_t
first
;
int
32_t
last
;
int
32_t
fwds
;
// number of forwards
SFwdInfo
fwdInfo
[];
}
SSyncFwds
;
typedef
struct
SsyncPeer
{
...
...
@@ -123,15 +123,15 @@ typedef struct SsyncPeer {
int8_t
sstatus
;
// sync status
uint64_t
version
;
uint64_t
sversion
;
// track the peer version in retrieve process
int
syncFd
;
int
peerFd
;
// forward FD
int
numOfRetrieves
;
// number of retrieves tried
int
fileChanged
;
// a flag to indicate file is changed during retrieving process
int
32_t
syncFd
;
int
32_t
peerFd
;
// forward FD
int
32_t
numOfRetrieves
;
// number of retrieves tried
int
32_t
fileChanged
;
// a flag to indicate file is changed during retrieving process
void
*
timer
;
void
*
pConn
;
int
notifyFd
;
int
watchNum
;
int
*
watchFd
;
int
32_t
notifyFd
;
int
32_t
watchNum
;
int
32_t
*
watchFd
;
int8_t
refCount
;
// reference count
struct
SSyncNode
*
pSyncNode
;
}
SSyncPeer
;
...
...
@@ -161,16 +161,16 @@ typedef struct SSyncNode {
}
SSyncNode
;
// sync module global
extern
int
tsSyncNum
;
extern
char
tsNodeFqdn
[
TSDB_FQDN_LEN
];
extern
int
32_t
tsSyncNum
;
extern
char
tsNodeFqdn
[
TSDB_FQDN_LEN
];
void
*
syncRetrieveData
(
void
*
param
);
void
*
syncRestoreData
(
void
*
param
);
int
syncSaveIntoBuffer
(
SSyncPeer
*
pPeer
,
SWalHead
*
pHead
);
void
syncRestartConnection
(
SSyncPeer
*
pPeer
);
void
syncBroadcastStatus
(
SSyncNode
*
pNode
);
void
syncAddPeerRef
(
SSyncPeer
*
pPeer
);
int
syncDecPeerRef
(
SSyncPeer
*
pPeer
);
int
32_t
syncSaveIntoBuffer
(
SSyncPeer
*
pPeer
,
SWalHead
*
pHead
);
void
syncRestartConnection
(
SSyncPeer
*
pPeer
);
void
syncBroadcastStatus
(
SSyncNode
*
pNode
);
void
syncAddPeerRef
(
SSyncPeer
*
pPeer
);
int
32_t
syncDecPeerRef
(
SSyncPeer
*
pPeer
);
#ifdef __cplusplus
}
...
...
src/sync/src/syncMain.c
浏览文件 @
a5fdb787
此差异已折叠。
点击以展开。
src/sync/src/syncRestore.c
浏览文件 @
a5fdb787
...
...
@@ -48,12 +48,12 @@ static void syncRemoveExtraFile(SSyncPeer *pPeer, int32_t sindex, int32_t eindex
}
}
static
int
syncRestoreFile
(
SSyncPeer
*
pPeer
,
uint64_t
*
fversion
)
{
static
int
32_t
syncRestoreFile
(
SSyncPeer
*
pPeer
,
uint64_t
*
fversion
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
SFileInfo
minfo
;
memset
(
&
minfo
,
0
,
sizeof
(
minfo
));
/* = {0}; */
// master file info
SFileInfo
sinfo
;
memset
(
&
sinfo
,
0
,
sizeof
(
sinfo
));
/* = {0}; */
// slave file info
SFileAck
fileAck
;
int
code
=
-
1
;
int
32_t
code
=
-
1
;
char
name
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
uint32_t
pindex
=
0
;
// index in last restore
bool
fileChanged
=
false
;
...
...
@@ -62,7 +62,7 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
sinfo
.
index
=
0
;
while
(
1
)
{
// read file info
int
ret
=
taosReadMsg
(
pPeer
->
syncFd
,
&
(
minfo
),
sizeof
(
minfo
));
int
32_t
ret
=
taosReadMsg
(
pPeer
->
syncFd
,
&
(
minfo
),
sizeof
(
minfo
));
if
(
ret
<
0
)
break
;
// if no more file from master, break;
...
...
@@ -104,7 +104,7 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
minfo
.
name
[
sizeof
(
minfo
.
name
)
-
1
]
=
0
;
snprintf
(
name
,
sizeof
(
name
),
"%s/%s"
,
pNode
->
path
,
minfo
.
name
);
int
dfd
=
open
(
name
,
O_WRONLY
|
O_CREAT
|
O_TRUNC
,
S_IRWXU
|
S_IRWXG
|
S_IRWXO
);
int
32_t
dfd
=
open
(
name
,
O_WRONLY
|
O_CREAT
|
O_TRUNC
,
S_IRWXU
|
S_IRWXG
|
S_IRWXO
);
if
(
dfd
<
0
)
{
sError
(
"%s, failed to open file:%s"
,
pPeer
->
id
,
name
);
break
;
...
...
@@ -132,9 +132,9 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
return
code
;
}
static
int
syncRestoreWal
(
SSyncPeer
*
pPeer
)
{
static
int
32_t
syncRestoreWal
(
SSyncPeer
*
pPeer
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
int
ret
,
code
=
-
1
;
int
32_t
ret
,
code
=
-
1
;
void
*
buffer
=
calloc
(
1024000
,
1
);
// size for one record
if
(
buffer
==
NULL
)
return
-
1
;
...
...
@@ -175,10 +175,10 @@ static char *syncProcessOneBufferedFwd(SSyncPeer *pPeer, char *offset) {
return
offset
;
}
static
int
syncProcessBufferedFwd
(
SSyncPeer
*
pPeer
)
{
static
int
32_t
syncProcessBufferedFwd
(
SSyncPeer
*
pPeer
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
SRecvBuffer
*
pRecv
=
pNode
->
pRecv
;
int
forwards
=
0
;
int
32_t
forwards
=
0
;
sDebug
(
"%s, number of buffered forwards:%d"
,
pPeer
->
id
,
pRecv
->
forwards
);
...
...
@@ -203,12 +203,12 @@ static int syncProcessBufferedFwd(SSyncPeer *pPeer) {
return
pRecv
->
code
;
}
int
syncSaveIntoBuffer
(
SSyncPeer
*
pPeer
,
SWalHead
*
pHead
)
{
int
32_t
syncSaveIntoBuffer
(
SSyncPeer
*
pPeer
,
SWalHead
*
pHead
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
SRecvBuffer
*
pRecv
=
pNode
->
pRecv
;
if
(
pRecv
==
NULL
)
return
-
1
;
int
len
=
pHead
->
len
+
sizeof
(
SWalHead
);
int
32_t
len
=
pHead
->
len
+
sizeof
(
SWalHead
);
if
(
pRecv
->
bufferSize
-
(
pRecv
->
offset
-
pRecv
->
buffer
)
>=
len
)
{
memcpy
(
pRecv
->
offset
,
pHead
,
len
);
...
...
@@ -231,7 +231,7 @@ static void syncCloseRecvBuffer(SSyncNode *pNode) {
tfree
(
pNode
->
pRecv
);
}
static
int
syncOpenRecvBuffer
(
SSyncNode
*
pNode
)
{
static
int
32_t
syncOpenRecvBuffer
(
SSyncNode
*
pNode
)
{
syncCloseRecvBuffer
(
pNode
);
SRecvBuffer
*
pRecv
=
calloc
(
sizeof
(
SRecvBuffer
),
1
);
...
...
@@ -252,13 +252,13 @@ static int syncOpenRecvBuffer(SSyncNode *pNode) {
return
0
;
}
static
int
syncRestoreDataStepByStep
(
SSyncPeer
*
pPeer
)
{
static
int
32_t
syncRestoreDataStepByStep
(
SSyncPeer
*
pPeer
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
nodeSStatus
=
TAOS_SYNC_STATUS_FILE
;
uint64_t
fversion
=
0
;
sDebug
(
"%s, start to restore file"
,
pPeer
->
id
);
int
code
=
syncRestoreFile
(
pPeer
,
&
fversion
);
int
32_t
code
=
syncRestoreFile
(
pPeer
,
&
fversion
);
if
(
code
<
0
)
{
sError
(
"%s, failed to restore file"
,
pPeer
->
id
);
return
-
1
;
...
...
src/sync/src/syncRetrieve.c
浏览文件 @
a5fdb787
...
...
@@ -27,7 +27,7 @@
#include "tsync.h"
#include "syncInt.h"
static
int
syncAddIntoWatchList
(
SSyncPeer
*
pPeer
,
char
*
name
)
{
static
int
32_t
syncAddIntoWatchList
(
SSyncPeer
*
pPeer
,
char
*
name
)
{
sDebug
(
"%s, start to monitor:%s"
,
pPeer
->
id
,
name
);
if
(
pPeer
->
notifyFd
<=
0
)
{
...
...
@@ -38,16 +38,16 @@ static int syncAddIntoWatchList(SSyncPeer *pPeer, char *name) {
return
-
1
;
}
if
(
pPeer
->
watchFd
==
NULL
)
pPeer
->
watchFd
=
malloc
(
sizeof
(
int
)
*
tsMaxWatchFiles
);
if
(
pPeer
->
watchFd
==
NULL
)
pPeer
->
watchFd
=
malloc
(
sizeof
(
int
32_t
)
*
tsMaxWatchFiles
);
if
(
pPeer
->
watchFd
==
NULL
)
{
sError
(
"%s, failed to allocate watchFd"
,
pPeer
->
id
);
return
-
1
;
}
memset
(
pPeer
->
watchFd
,
-
1
,
sizeof
(
int
)
*
tsMaxWatchFiles
);
memset
(
pPeer
->
watchFd
,
-
1
,
sizeof
(
int
32_t
)
*
tsMaxWatchFiles
);
}
int
*
wd
=
pPeer
->
watchFd
+
pPeer
->
watchNum
;
int
32_t
*
wd
=
pPeer
->
watchFd
+
pPeer
->
watchNum
;
if
(
*
wd
>=
0
)
{
if
(
inotify_rm_watch
(
pPeer
->
notifyFd
,
*
wd
)
<
0
)
{
...
...
@@ -69,17 +69,17 @@ static int syncAddIntoWatchList(SSyncPeer *pPeer, char *name) {
return
0
;
}
static
int
syncAreFilesModified
(
SSyncPeer
*
pPeer
)
{
static
int
32_t
syncAreFilesModified
(
SSyncPeer
*
pPeer
)
{
if
(
pPeer
->
notifyFd
<=
0
)
return
0
;
char
buf
[
2048
];
int
len
=
read
(
pPeer
->
notifyFd
,
buf
,
sizeof
(
buf
));
char
buf
[
2048
];
int
32_t
len
=
read
(
pPeer
->
notifyFd
,
buf
,
sizeof
(
buf
));
if
(
len
<
0
&&
errno
!=
EAGAIN
)
{
sError
(
"%s, failed to read notify FD(%s)"
,
pPeer
->
id
,
strerror
(
errno
));
return
-
1
;
}
int
code
=
0
;
int
32_t
code
=
0
;
if
(
len
>
0
)
{
const
struct
inotify_event
*
event
;
char
*
ptr
;
...
...
@@ -97,11 +97,11 @@ static int syncAreFilesModified(SSyncPeer *pPeer) {
return
code
;
}
static
int
syncRetrieveFile
(
SSyncPeer
*
pPeer
)
{
static
int
32_t
syncRetrieveFile
(
SSyncPeer
*
pPeer
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
SFileInfo
fileInfo
;
SFileAck
fileAck
;
int
code
=
-
1
;
int
32_t
code
=
-
1
;
char
name
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
memset
(
&
fileInfo
,
0
,
sizeof
(
fileInfo
));
...
...
@@ -146,7 +146,7 @@ static int syncRetrieveFile(SSyncPeer *pPeer) {
}
// send the file to peer
int
sfd
=
open
(
name
,
O_RDONLY
);
int
32_t
sfd
=
open
(
name
,
O_RDONLY
);
if
(
sfd
<
0
)
break
;
ret
=
taosSendFile
(
pPeer
->
syncFd
,
sfd
,
NULL
,
fileInfo
.
size
);
...
...
@@ -169,8 +169,8 @@ static int syncRetrieveFile(SSyncPeer *pPeer) {
/* if only a partial record is read out, set the IN_MODIFY flag in event,
so upper layer will reload the file to get a complete record */
static
int
syncReadOneWalRecord
(
in
t
sfd
,
SWalHead
*
pHead
,
uint32_t
*
pEvent
)
{
int
ret
;
static
int
32_t
syncReadOneWalRecord
(
int32_
t
sfd
,
SWalHead
*
pHead
,
uint32_t
*
pEvent
)
{
int
32_t
ret
;
ret
=
read
(
sfd
,
pHead
,
sizeof
(
SWalHead
));
if
(
ret
<
0
)
return
-
1
;
...
...
@@ -194,7 +194,7 @@ static int syncReadOneWalRecord(int sfd, SWalHead *pHead, uint32_t *pEvent) {
return
sizeof
(
SWalHead
)
+
pHead
->
len
;
}
static
int
syncMonitorLastWal
(
SSyncPeer
*
pPeer
,
char
*
name
)
{
static
int
32_t
syncMonitorLastWal
(
SSyncPeer
*
pPeer
,
char
*
name
)
{
pPeer
->
watchNum
=
0
;
taosClose
(
pPeer
->
notifyFd
);
pPeer
->
notifyFd
=
inotify_init1
(
IN_NONBLOCK
);
...
...
@@ -203,14 +203,14 @@ static int syncMonitorLastWal(SSyncPeer *pPeer, char *name) {
return
-
1
;
}
if
(
pPeer
->
watchFd
==
NULL
)
pPeer
->
watchFd
=
malloc
(
sizeof
(
int
)
*
tsMaxWatchFiles
);
if
(
pPeer
->
watchFd
==
NULL
)
pPeer
->
watchFd
=
malloc
(
sizeof
(
int
32_t
)
*
tsMaxWatchFiles
);
if
(
pPeer
->
watchFd
==
NULL
)
{
sError
(
"%s, failed to allocate watchFd"
,
pPeer
->
id
);
return
-
1
;
}
memset
(
pPeer
->
watchFd
,
-
1
,
sizeof
(
int
)
*
tsMaxWatchFiles
);
int
*
wd
=
pPeer
->
watchFd
;
memset
(
pPeer
->
watchFd
,
-
1
,
sizeof
(
int
32_t
)
*
tsMaxWatchFiles
);
int
32_t
*
wd
=
pPeer
->
watchFd
;
*
wd
=
inotify_add_watch
(
pPeer
->
notifyFd
,
name
,
IN_MODIFY
|
IN_CLOSE_WRITE
);
if
(
*
wd
==
-
1
)
{
...
...
@@ -222,8 +222,8 @@ static int syncMonitorLastWal(SSyncPeer *pPeer, char *name) {
}
static
int32_t
syncCheckLastWalChanges
(
SSyncPeer
*
pPeer
,
uint32_t
*
pEvent
)
{
char
buf
[
2048
];
int
len
=
read
(
pPeer
->
notifyFd
,
buf
,
sizeof
(
buf
));
char
buf
[
2048
];
int
32_t
len
=
read
(
pPeer
->
notifyFd
,
buf
,
sizeof
(
buf
));
if
(
len
<
0
&&
errno
!=
EAGAIN
)
{
sError
(
"%s, failed to read notify FD(%s)"
,
pPeer
->
id
,
strerror
(
errno
));
return
-
1
;
...
...
@@ -243,11 +243,11 @@ static int32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
return
0
;
}
static
int
syncRetrieveLastWal
(
SSyncPeer
*
pPeer
,
char
*
name
,
uint64_t
fversion
,
int64_t
offset
,
uint32_t
*
pEvent
)
{
static
int
32_t
syncRetrieveLastWal
(
SSyncPeer
*
pPeer
,
char
*
name
,
uint64_t
fversion
,
int64_t
offset
,
uint32_t
*
pEvent
)
{
SWalHead
*
pHead
=
malloc
(
640000
);
int
code
=
-
1
;
int
32_t
code
=
-
1
;
int32_t
bytes
=
0
;
int
sfd
;
int
32_t
sfd
;
sfd
=
open
(
name
,
O_RDONLY
);
if
(
sfd
<
0
)
{
...
...
@@ -259,7 +259,7 @@ static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion,
sDebug
(
"%s, retrieve last wal, offset:%"
PRId64
" fver:%"
PRIu64
,
pPeer
->
id
,
offset
,
fversion
);
while
(
1
)
{
int
wsize
=
syncReadOneWalRecord
(
sfd
,
pHead
,
pEvent
);
int
32_t
wsize
=
syncReadOneWalRecord
(
sfd
,
pHead
,
pEvent
);
if
(
wsize
<
0
)
break
;
if
(
wsize
==
0
)
{
code
=
0
;
...
...
@@ -267,7 +267,7 @@ static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion,
}
sDebug
(
"%s, last wal is forwarded, ver:%"
PRIu64
,
pPeer
->
id
,
pHead
->
version
);
int
ret
=
taosWriteMsg
(
pPeer
->
syncFd
,
pHead
,
wsize
);
int
32_t
ret
=
taosWriteMsg
(
pPeer
->
syncFd
,
pHead
,
wsize
);
if
(
ret
!=
wsize
)
break
;
pPeer
->
sversion
=
pHead
->
version
;
...
...
@@ -287,9 +287,9 @@ static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion,
return
-
1
;
}
static
int
syncProcessLastWal
(
SSyncPeer
*
pPeer
,
char
*
wname
,
int64_t
index
)
{
static
int
32_t
syncProcessLastWal
(
SSyncPeer
*
pPeer
,
char
*
wname
,
int64_t
index
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
int
code
=
-
1
;
int
32_t
code
=
-
1
;
char
fname
[
TSDB_FILENAME_LEN
*
2
];
// full path to wal file
if
(
syncAreFilesModified
(
pPeer
)
!=
0
)
return
-
1
;
...
...
@@ -370,13 +370,13 @@ static int syncProcessLastWal(SSyncPeer *pPeer, char *wname, int64_t index) {
return
code
;
}
static
int
syncRetrieveWal
(
SSyncPeer
*
pPeer
)
{
static
int
32_t
syncRetrieveWal
(
SSyncPeer
*
pPeer
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
char
fname
[
TSDB_FILENAME_LEN
*
3
];
char
wname
[
TSDB_FILENAME_LEN
*
2
];
int32_t
size
;
struct
stat
fstat
;
int
code
=
-
1
;
int
32_t
code
=
-
1
;
int64_t
index
=
0
;
while
(
1
)
{
...
...
@@ -403,7 +403,7 @@ static int syncRetrieveWal(SSyncPeer *pPeer) {
size
=
fstat
.
st_size
;
sDebug
(
"%s, retrieve wal:%s size:%d"
,
pPeer
->
id
,
fname
,
size
);
int
sfd
=
open
(
fname
,
O_RDONLY
);
int
32_t
sfd
=
open
(
fname
,
O_RDONLY
);
if
(
sfd
<
0
)
break
;
code
=
taosSendFile
(
pPeer
->
syncFd
,
sfd
,
NULL
,
size
);
...
...
@@ -428,7 +428,7 @@ static int syncRetrieveWal(SSyncPeer *pPeer) {
return
code
;
}
static
int
syncRetrieveDataStepByStep
(
SSyncPeer
*
pPeer
)
{
static
int
32_t
syncRetrieveDataStepByStep
(
SSyncPeer
*
pPeer
)
{
SSyncNode
*
pNode
=
pPeer
->
pSyncNode
;
SFirstPkt
firstPkt
;
...
...
src/sync/src/tarbitrator.c
浏览文件 @
a5fdb787
...
...
@@ -28,22 +28,22 @@
#include "syncInt.h"
static
void
arbSignalHandler
(
int32_t
signum
,
siginfo_t
*
sigInfo
,
void
*
context
);
static
void
arbProcessIncommingConnection
(
int
connFd
,
uint32_t
sourceIp
);
static
void
arbProcessIncommingConnection
(
int
32_t
connFd
,
uint32_t
sourceIp
);
static
void
arbProcessBrokenLink
(
void
*
param
);
static
int
arbProcessPeerMsg
(
void
*
param
,
void
*
buffer
);
static
int
32_t
arbProcessPeerMsg
(
void
*
param
,
void
*
buffer
);
static
tsem_t
tsArbSem
;
static
ttpool_h
tsArbTcpPool
;
typedef
struct
{
char
id
[
TSDB_EP_LEN
+
24
];
int
nodeFd
;
void
*
pConn
;
char
id
[
TSDB_EP_LEN
+
24
];
int
32_t
nodeFd
;
void
*
pConn
;
}
SNodeConn
;
int
main
(
in
t
argc
,
char
*
argv
[])
{
int
32_t
main
(
int32_
t
argc
,
char
*
argv
[])
{
char
arbLogPath
[
TSDB_FILENAME_LEN
+
16
]
=
{
0
};
for
(
int
i
=
1
;
i
<
argc
;
++
i
)
{
for
(
int
32_t
i
=
1
;
i
<
argc
;
++
i
)
{
if
(
strcmp
(
argv
[
i
],
"-p"
)
==
0
&&
i
<
argc
-
1
)
{
tsArbitratorPort
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-d"
)
==
0
&&
i
<
argc
-
1
)
{
...
...
@@ -108,7 +108,7 @@ int main(int argc, char *argv[]) {
return
0
;
}
static
void
arbProcessIncommingConnection
(
int
connFd
,
uint32_t
sourceIp
)
{
static
void
arbProcessIncommingConnection
(
int
32_t
connFd
,
uint32_t
sourceIp
)
{
char
ipstr
[
24
];
tinet_ntoa
(
ipstr
,
sourceIp
);
sDebug
(
"peer TCP connection from ip:%s"
,
ipstr
);
...
...
@@ -150,13 +150,13 @@ static void arbProcessBrokenLink(void *param) {
tfree
(
pNode
);
}
static
int
arbProcessPeerMsg
(
void
*
param
,
void
*
buffer
)
{
static
int
32_t
arbProcessPeerMsg
(
void
*
param
,
void
*
buffer
)
{
SNodeConn
*
pNode
=
param
;
SSyncHead
head
;
int
bytes
=
0
;
int
32_t
bytes
=
0
;
char
*
cont
=
(
char
*
)
buffer
;
int
hlen
=
taosReadMsg
(
pNode
->
nodeFd
,
&
head
,
sizeof
(
head
));
int
32_t
hlen
=
taosReadMsg
(
pNode
->
nodeFd
,
&
head
,
sizeof
(
head
));
if
(
hlen
!=
sizeof
(
head
))
{
sDebug
(
"%s, failed to read msg, hlen:%d"
,
pNode
->
id
,
hlen
);
return
-
1
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录