Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
ca4d7329
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
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看板
提交
ca4d7329
编写于
6月 23, 2022
作者:
M
Minghao Li
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor(sync): modify wal error log
上级
81eae2a9
变更
4
显示空白变更内容
内联
并排
Showing
4 changed file
with
64 addition
and
56 deletion
+64
-56
source/libs/sync/src/syncAppendEntries.c
source/libs/sync/src/syncAppendEntries.c
+7
-20
source/libs/sync/src/syncAppendEntriesReply.c
source/libs/sync/src/syncAppendEntriesReply.c
+19
-23
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+3
-2
source/libs/sync/src/syncRaftLog.c
source/libs/sync/src/syncRaftLog.c
+35
-11
未找到文件。
source/libs/sync/src/syncAppendEntries.c
浏览文件 @
ca4d7329
...
...
@@ -466,14 +466,11 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
int32_t
code
=
0
;
// print log
char
logBuf
[
128
]
=
{
0
};
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"recv SyncAppendEntries, vgId:%d, term:%lu"
,
ths
->
vgId
,
ths
->
pRaftStore
->
currentTerm
);
syncAppendEntriesLog2
(
logBuf
,
pMsg
);
syncAppendEntriesLog2
(
"==syncNodeOnAppendEntriesSnapshotCb=="
,
pMsg
);
// if already drop replica, do not process
if
(
!
syncNodeInRaftGroup
(
ths
,
&
(
pMsg
->
srcId
))
&&
!
ths
->
pRaftCfg
->
isStandBy
)
{
s
Info
(
"recv SyncAppendEntries
maybe replica already dropped"
);
s
yncNodeEventLog
(
ths
,
"recv sync-append-entries,
maybe replica already dropped"
);
return
ret
;
}
...
...
@@ -497,7 +494,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
do
{
bool
condition
=
pMsg
->
term
==
ths
->
pRaftStore
->
currentTerm
&&
ths
->
state
==
TAOS_SYNC_STATE_CANDIDATE
;
if
(
condition
)
{
s
Trace
(
"recv SyncAppendE
ntries, candidate to follower"
);
s
yncNodeEventLog
(
ths
,
"recv sync-append-e
ntries, candidate to follower"
);
syncNodeBecomeFollower
(
ths
,
"from candidate by append entries"
);
// do not reply?
...
...
@@ -538,11 +535,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
bool
condition
=
condition1
||
condition2
||
condition3
||
condition4
;
if
(
condition
)
{
sTrace
(
"recv SyncAppendEntries, fake match, myLastIndex:%ld, syncLogBeginIndex:%ld, syncLogEndIndex:%ld, "
"condition1:%d, condition2:%d, condition3:%d, condition4:%d"
,
myLastIndex
,
ths
->
pLogStore
->
syncLogBeginIndex
(
ths
->
pLogStore
),
ths
->
pLogStore
->
syncLogEndIndex
(
ths
->
pLogStore
),
condition1
,
condition2
,
condition3
,
condition4
);
syncNodeEventLog
(
ths
,
"recv sync-append-entries, fake match"
);
// prepare response msg
SyncAppendEntriesReply
*
pReply
=
syncAppendEntriesReplyBuild
(
ths
->
vgId
);
...
...
@@ -576,8 +569,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
bool
condition
=
(
pMsg
->
term
==
ths
->
pRaftStore
->
currentTerm
)
&&
(
ths
->
state
==
TAOS_SYNC_STATE_FOLLOWER
)
&&
(
pMsg
->
prevLogIndex
<=
ths
->
commitIndex
);
if
(
condition
)
{
sTrace
(
"recv SyncAppendEntries, fake match2, msg-prevLogIndex:%ld, my-commitIndex:%ld"
,
pMsg
->
prevLogIndex
,
ths
->
commitIndex
);
syncNodeEventLog
(
ths
,
"recv sync-append-entries, fake match2"
);
SyncIndex
matchIndex
=
ths
->
commitIndex
;
bool
hasAppendEntries
=
pMsg
->
dataLen
>
0
;
...
...
@@ -650,11 +642,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
bool
condition
=
condition1
||
condition2
;
if
(
condition
)
{
sTrace
(
"recv SyncAppendEntries, not match, syncLogBeginIndex:%ld, syncLogEndIndex:%ld, condition1:%d, "
"condition2:%d, logOK:%d"
,
ths
->
pLogStore
->
syncLogBeginIndex
(
ths
->
pLogStore
),
ths
->
pLogStore
->
syncLogEndIndex
(
ths
->
pLogStore
),
condition1
,
condition2
,
logOK
);
syncNodeEventLog
(
ths
,
"recv sync-append-entries, not match"
);
// prepare response msg
SyncAppendEntriesReply
*
pReply
=
syncAppendEntriesReplyBuild
(
ths
->
vgId
);
...
...
@@ -693,8 +681,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
// has entries in SyncAppendEntries msg
bool
hasAppendEntries
=
pMsg
->
dataLen
>
0
;
sTrace
(
"recv SyncAppendEntries, match, myLastIndex:%ld, hasExtraEntries:%d, hasAppendEntries:%d"
,
myLastIndex
,
hasExtraEntries
,
hasAppendEntries
);
syncNodeEventLog
(
ths
,
"recv sync-append-entries, match"
);
if
(
hasExtraEntries
)
{
// make log same, rollback deleted entries
...
...
source/libs/sync/src/syncAppendEntriesReply.c
浏览文件 @
ca4d7329
...
...
@@ -101,32 +101,27 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
int32_t
ret
=
0
;
// print log
char
logBuf
[
128
]
=
{
0
};
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"recv SyncAppendEntriesReply, vgId:%d, term:%lu"
,
ths
->
vgId
,
ths
->
pRaftStore
->
currentTerm
);
syncAppendEntriesReplyLog2
(
logBuf
,
pMsg
);
syncAppendEntriesReplyLog2
(
"==syncNodeOnAppendEntriesReplySnapshotCb=="
,
pMsg
);
// if already drop replica, do not process
if
(
!
syncNodeInRaftGroup
(
ths
,
&
(
pMsg
->
srcId
))
&&
!
ths
->
pRaftCfg
->
isStandBy
)
{
s
Info
(
"recv SyncAppendEntriesR
eply, maybe replica already dropped"
);
return
ret
;
s
yncNodeEventLog
(
ths
,
"recv sync-append-entries-r
eply, maybe replica already dropped"
);
return
0
;
}
// drop stale response
if
(
pMsg
->
term
<
ths
->
pRaftStore
->
currentTerm
)
{
sTrace
(
"recv SyncAppendEntriesReply, drop stale response, receive_term:%lu current_term:%lu"
,
pMsg
->
term
,
ths
->
pRaftStore
->
currentTerm
);
return
ret
;
char
logBuf
[
128
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"recv sync-append-entries-reply, recv-term:%lu, drop stale response"
,
pMsg
->
term
);
syncNodeEventLog
(
ths
,
logBuf
);
return
0
;
}
syncIndexMgrLog2
(
"recv SyncAppendEntriesReply, before pNextIndex:"
,
ths
->
pNextIndex
);
syncIndexMgrLog2
(
"recv SyncAppendEntriesReply, before pMatchIndex:"
,
ths
->
pMatchIndex
);
if
(
gRaftDetailLog
)
{
SSnapshot
snapshot
;
ths
->
pFsm
->
FpGetSnapshotInfo
(
ths
->
pFsm
,
&
snapshot
);
sTrace
(
"recv SyncAppendEntriesReply, before snapshot.lastApplyIndex:%ld, snapshot.lastApplyTerm:%lu"
,
snapshot
.
lastApplyIndex
,
snapshot
.
lastApplyTerm
);
syncNodeEventLog
(
ths
,
"recv sync-append-entries-reply, before"
);
}
syncIndexMgrLog2
(
"recv sync-append-entries-reply, before pNextIndex:"
,
ths
->
pNextIndex
);
syncIndexMgrLog2
(
"recv sync-append-entries-reply, before pMatchIndex:"
,
ths
->
pMatchIndex
);
// no need this code, because if I receive reply.term, then I must have sent for that term.
// if (pMsg->term > ths->pRaftStore->currentTerm) {
...
...
@@ -134,12 +129,10 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
// }
if
(
pMsg
->
term
>
ths
->
pRaftStore
->
currentTerm
)
{
char
logBuf
[
128
]
=
{
0
};
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"recv SyncAppendEntriesReply, error term, receive_term:%lu current_term:%lu"
,
pMsg
->
term
,
ths
->
pRaftStore
->
currentTerm
);
syncNodeLog2
(
logBuf
,
ths
);
sError
(
"%s"
,
logBuf
);
return
ret
;
char
logBuf
[
128
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"recv sync-append-entries-reply, error term, recv-term:%lu"
,
pMsg
->
term
);
syncNodeErrorLog
(
ths
,
logBuf
);
return
-
1
;
}
ASSERT
(
pMsg
->
term
==
ths
->
pRaftStore
->
currentTerm
);
...
...
@@ -228,8 +221,11 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
}
}
syncIndexMgrLog2
(
"recv SyncAppendEntriesReply, after pNextIndex:"
,
ths
->
pNextIndex
);
syncIndexMgrLog2
(
"recv SyncAppendEntriesReply, after pMatchIndex:"
,
ths
->
pMatchIndex
);
if
(
gRaftDetailLog
)
{
syncNodeEventLog
(
ths
,
"recv sync-append-entries-reply, after"
);
}
syncIndexMgrLog2
(
"recv sync-append-entries-reply, after pNextIndex:"
,
ths
->
pNextIndex
);
syncIndexMgrLog2
(
"recv sync-append-entries-reply, after pMatchIndex:"
,
ths
->
pMatchIndex
);
return
ret
;
}
\ No newline at end of file
source/libs/sync/src/syncMain.c
浏览文件 @
ca4d7329
...
...
@@ -416,7 +416,7 @@ int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct
}
}
sMeta
->
lastConfigIndex
=
lastIndex
;
sTrace
(
"vgId:%d, get snapshot meta by index:%"
PRId64
" l
astConfigI
ndex:%"
PRId64
,
pSyncNode
->
vgId
,
snapshotIndex
,
sTrace
(
"vgId:%d, get snapshot meta by index:%"
PRId64
" l
ci
ndex:%"
PRId64
,
pSyncNode
->
vgId
,
snapshotIndex
,
sMeta
->
lastConfigIndex
);
taosReleaseRef
(
tsNodeRefId
,
pSyncNode
->
rid
);
...
...
@@ -433,8 +433,9 @@ SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapsho
lastIndex
=
(
pSyncNode
->
pRaftCfg
->
configIndexArr
)[
i
];
}
}
sTrace
(
"vgId:%d, sync get snapshot last config index, index:%ld lcindex:%ld"
,
pSyncNode
->
vgId
,
snapshotLastApplyIndex
,
lastIndex
);
sTrace
(
"sync syncNodeGetSnapshotConfigIndex index:%ld lastConfigIndex:%ld"
,
snapshotLastApplyIndex
,
lastIndex
);
return
lastIndex
;
}
...
...
source/libs/sync/src/syncRaftLog.c
浏览文件 @
ca4d7329
...
...
@@ -46,13 +46,13 @@ static SyncIndex logStoreGetCommitIndex(SSyncLogStore* pLogStore);
// refactor, log[0 .. n] ==> log[m .. n]
static
int32_t
raftLogSetBeginIndex
(
struct
SSyncLogStore
*
pLogStore
,
SyncIndex
beginIndex
)
{
sTrace
(
"raftLogSetBeginIndex beginIndex:%ld"
,
beginIndex
);
// if beginIndex == 0, donot need call this funciton
ASSERT
(
beginIndex
>
0
);
SSyncLogStoreData
*
pData
=
pLogStore
->
data
;
SWal
*
pWal
=
pData
->
pWal
;
sTrace
(
"vgId:%d, reset wal begin index:%ld"
,
pData
->
pSyncNode
->
vgId
,
beginIndex
);
pData
->
beginIndex
=
beginIndex
;
walRestoreFromSnapshot
(
pWal
,
beginIndex
-
1
);
return
0
;
...
...
@@ -160,8 +160,12 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
sysErr
=
errno
;
const
char
*
sysErrStr
=
strerror
(
errno
);
sError
(
"vgId:%d wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
pData
->
pSyncNode
->
vgId
,
char
logBuf
[
128
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
pEntry
->
index
,
err
,
err
,
errStr
,
sysErr
,
sysErrStr
);
syncNodeErrorLog
(
pData
->
pSyncNode
,
logBuf
);
ASSERT
(
0
);
}
...
...
@@ -240,8 +244,17 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
sysErr
=
errno
;
const
char
*
sysErrStr
=
strerror
(
errno
);
sError
(
"vgId:%d wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
pData
->
pSyncNode
->
vgId
,
index
,
err
,
err
,
errStr
,
sysErr
,
sysErrStr
);
do
{
char
logBuf
[
128
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
index
,
err
,
err
,
errStr
,
sysErr
,
sysErrStr
);
if
(
terrno
==
TSDB_CODE_WAL_LOG_NOT_EXIST
)
{
syncNodeEventLog
(
pData
->
pSyncNode
,
logBuf
);
}
else
{
syncNodeErrorLog
(
pData
->
pSyncNode
,
logBuf
);
}
}
while
(
0
);
int32_t
saveErr
=
terrno
;
walCloseReadHandle
(
pWalHandle
);
...
...
@@ -368,8 +381,11 @@ int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
sysErr
=
errno
;
const
char
*
sysErrStr
=
strerror
(
errno
);
sError
(
"vgId:%d wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
pData
->
pSyncNode
->
vgId
,
char
logBuf
[
128
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
pEntry
->
index
,
err
,
err
,
errStr
,
sysErr
,
sysErrStr
);
syncNodeErrorLog
(
pData
->
pSyncNode
,
logBuf
);
ASSERT
(
0
);
}
...
...
@@ -398,12 +414,20 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
sysErr
=
errno
;
const
char
*
sysErrStr
=
strerror
(
errno
);
sError
(
"vgId:%d wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
pData
->
pSyncNode
->
vgId
,
index
,
err
,
err
,
errStr
,
sysErr
,
sysErrStr
);
do
{
char
logBuf
[
128
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s"
,
index
,
err
,
err
,
errStr
,
sysErr
,
sysErrStr
);
if
(
terrno
==
TSDB_CODE_WAL_LOG_NOT_EXIST
)
{
syncNodeEventLog
(
pData
->
pSyncNode
,
logBuf
);
}
else
{
syncNodeErrorLog
(
pData
->
pSyncNode
,
logBuf
);
}
}
while
(
0
);
ASSERT
(
0
);
}
// ASSERT(walReadWithHandle(pWalHandle, index) == 0);
SSyncRaftEntry
*
pEntry
=
syncEntryBuild
(
pWalHandle
->
pHead
->
head
.
bodyLen
);
ASSERT
(
pEntry
!=
NULL
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录