Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
398161eb
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看板
提交
398161eb
编写于
1月 09, 2023
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'main' into merge/mainto3.0_0108
上级
054c3133
f2d6ed86
变更
37
隐藏空白更改
内联
并排
Showing
37 changed file
with
408 addition
and
168 deletion
+408
-168
cmake/taostools_CMakeLists.txt.in
cmake/taostools_CMakeLists.txt.in
+1
-1
include/common/tglobal.h
include/common/tglobal.h
+3
-0
include/libs/sync/sync.h
include/libs/sync/sync.h
+3
-1
source/common/src/tglobal.c
source/common/src/tglobal.c
+7
-0
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+2
-0
source/dnode/mnode/impl/src/mndMnode.c
source/dnode/mnode/impl/src/mndMnode.c
+2
-2
source/dnode/mnode/sdb/src/sdbFile.c
source/dnode/mnode/sdb/src/sdbFile.c
+1
-1
source/dnode/snode/src/snode.c
source/dnode/snode/src/snode.c
+1
-0
source/dnode/vnode/inc/vnode.h
source/dnode/vnode/inc/vnode.h
+1
-0
source/dnode/vnode/src/inc/vnd.h
source/dnode/vnode/src/inc/vnd.h
+2
-0
source/dnode/vnode/src/inc/vnodeInt.h
source/dnode/vnode/src/inc/vnodeInt.h
+6
-0
source/dnode/vnode/src/meta/metaOpen.c
source/dnode/vnode/src/meta/metaOpen.c
+1
-0
source/dnode/vnode/src/tq/tqCommit.c
source/dnode/vnode/src/tq/tqCommit.c
+8
-1
source/dnode/vnode/src/tsdb/tsdbCache.c
source/dnode/vnode/src/tsdb/tsdbCache.c
+2
-0
source/dnode/vnode/src/tsdb/tsdbRead.c
source/dnode/vnode/src/tsdb/tsdbRead.c
+41
-16
source/dnode/vnode/src/vnd/vnodeCommit.c
source/dnode/vnode/src/vnd/vnodeCommit.c
+22
-1
source/dnode/vnode/src/vnd/vnodeOpen.c
source/dnode/vnode/src/vnd/vnodeOpen.c
+5
-1
source/dnode/vnode/src/vnd/vnodeSvr.c
source/dnode/vnode/src/vnd/vnodeSvr.c
+4
-4
source/dnode/vnode/src/vnd/vnodeSync.c
source/dnode/vnode/src/vnd/vnodeSync.c
+62
-28
source/libs/stream/src/streamMeta.c
source/libs/stream/src/streamMeta.c
+2
-2
source/libs/sync/inc/syncMessage.h
source/libs/sync/inc/syncMessage.h
+2
-2
source/libs/sync/inc/syncPipeline.h
source/libs/sync/inc/syncPipeline.h
+1
-0
source/libs/sync/src/syncAppendEntries.c
source/libs/sync/src/syncAppendEntries.c
+3
-1
source/libs/sync/src/syncCommit.c
source/libs/sync/src/syncCommit.c
+1
-0
source/libs/sync/src/syncEnv.c
source/libs/sync/src/syncEnv.c
+1
-1
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+27
-14
source/libs/sync/src/syncPipeline.c
source/libs/sync/src/syncPipeline.c
+53
-18
source/libs/sync/src/syncRaftLog.c
source/libs/sync/src/syncRaftLog.c
+4
-23
source/libs/sync/src/syncUtil.c
source/libs/sync/src/syncUtil.c
+59
-33
source/libs/tdb/inc/tdb.h
source/libs/tdb/inc/tdb.h
+6
-1
source/libs/tdb/src/db/tdbTxn.c
source/libs/tdb/src/db/tdbTxn.c
+6
-1
source/libs/transport/src/transCli.c
source/libs/transport/src/transCli.c
+1
-1
source/libs/wal/src/walMeta.c
source/libs/wal/src/walMeta.c
+4
-13
source/libs/wal/src/walWrite.c
source/libs/wal/src/walWrite.c
+1
-1
tests/script/tsim/parser/regressiontest.sim
tests/script/tsim/parser/regressiontest.sim
+61
-0
tests/script/tsim/sma/rsmaCreateInsertQuery.sim
tests/script/tsim/sma/rsmaCreateInsertQuery.sim
+1
-1
tests/script/tsim/sma/rsmaPersistenceRecovery.sim
tests/script/tsim/sma/rsmaPersistenceRecovery.sim
+1
-0
未找到文件。
cmake/taostools_CMakeLists.txt.in
浏览文件 @
398161eb
...
...
@@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG
4efbc10
GIT_TAG
94d6895
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
...
...
include/common/tglobal.h
浏览文件 @
398161eb
...
...
@@ -69,6 +69,9 @@ extern int32_t tsElectInterval;
extern
int32_t
tsHeartbeatInterval
;
extern
int32_t
tsHeartbeatTimeout
;
// vnode
extern
int64_t
tsVndCommitMaxIntervalMs
;
// monitor
extern
bool
tsEnableMonitor
;
extern
int32_t
tsMonitorInterval
;
...
...
include/libs/sync/sync.h
浏览文件 @
398161eb
...
...
@@ -49,10 +49,12 @@ extern "C" {
#define SYNC_HEARTBEAT_REPLY_SLOW_MS 1500
#define SYNC_SNAP_RESEND_MS 1000 * 60
#define SYNC_VND_COMMIT_MIN_MS 1000
#define SYNC_MAX_BATCH_SIZE 1
#define SYNC_INDEX_BEGIN 0
#define SYNC_INDEX_INVALID -1
#define SYNC_TERM_INVALID -1
// 0xFFFFFFFFFFFFFFFF
#define SYNC_TERM_INVALID -1
typedef
enum
{
SYNC_STRATEGY_NO_SNAPSHOT
=
0
,
...
...
source/common/src/tglobal.c
浏览文件 @
398161eb
...
...
@@ -60,6 +60,9 @@ int32_t tsElectInterval = 25 * 1000;
int32_t
tsHeartbeatInterval
=
1000
;
int32_t
tsHeartbeatTimeout
=
20
*
1000
;
// vnode
int64_t
tsVndCommitMaxIntervalMs
=
60
*
1000
;
// monitor
bool
tsEnableMonitor
=
true
;
int32_t
tsMonitorInterval
=
30
;
...
...
@@ -435,6 +438,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if
(
cfgAddInt32
(
pCfg
,
"syncHeartbeatInterval"
,
tsHeartbeatInterval
,
10
,
1000
*
60
*
24
*
2
,
0
)
!=
0
)
return
-
1
;
if
(
cfgAddInt32
(
pCfg
,
"syncHeartbeatTimeout"
,
tsHeartbeatTimeout
,
10
,
1000
*
60
*
24
*
2
,
0
)
!=
0
)
return
-
1
;
if
(
cfgAddInt64
(
pCfg
,
"vndCommitMaxInterval"
,
tsVndCommitMaxIntervalMs
,
1000
,
1000
*
60
*
60
,
0
)
!=
0
)
return
-
1
;
if
(
cfgAddBool
(
pCfg
,
"monitor"
,
tsEnableMonitor
,
0
)
!=
0
)
return
-
1
;
if
(
cfgAddInt32
(
pCfg
,
"monitorInterval"
,
tsMonitorInterval
,
1
,
200000
,
0
)
!=
0
)
return
-
1
;
if
(
cfgAddString
(
pCfg
,
"monitorFqdn"
,
tsMonitorFqdn
,
0
)
!=
0
)
return
-
1
;
...
...
@@ -754,6 +759,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsHeartbeatInterval
=
cfgGetItem
(
pCfg
,
"syncHeartbeatInterval"
)
->
i32
;
tsHeartbeatTimeout
=
cfgGetItem
(
pCfg
,
"syncHeartbeatTimeout"
)
->
i32
;
tsVndCommitMaxIntervalMs
=
cfgGetItem
(
pCfg
,
"vndCommitMaxInterval"
)
->
i64
;
tsStartUdfd
=
cfgGetItem
(
pCfg
,
"udf"
)
->
bval
;
tstrncpy
(
tsUdfdResFuncs
,
cfgGetItem
(
pCfg
,
"udfdResFuncs"
)
->
str
,
sizeof
(
tsUdfdResFuncs
));
tstrncpy
(
tsUdfdLdLibPath
,
cfgGetItem
(
pCfg
,
"udfdLdLibPath"
)
->
str
,
sizeof
(
tsUdfdLdLibPath
));
...
...
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
浏览文件 @
398161eb
...
...
@@ -79,6 +79,8 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
void
vmCloseVnode
(
SVnodeMgmt
*
pMgmt
,
SVnodeObj
*
pVnode
)
{
char
path
[
TSDB_FILENAME_LEN
]
=
{
0
};
vnodeProposeCommitOnNeed
(
pVnode
->
pImpl
);
taosThreadRwlockWrlock
(
&
pMgmt
->
lock
);
taosHashRemove
(
pMgmt
->
hash
,
&
pVnode
->
vgId
,
sizeof
(
int32_t
));
taosThreadRwlockUnlock
(
&
pMgmt
->
lock
);
...
...
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @
398161eb
...
...
@@ -785,9 +785,9 @@ static void mndReloadSyncConfig(SMnode *pMnode) {
int32_t
code
=
syncReconfig
(
pMnode
->
syncMgmt
.
sync
,
&
cfg
);
if
(
code
!=
0
)
{
mError
(
"vgId:1,
failed to reconfig mnode sync
since %s"
,
terrstr
());
mError
(
"vgId:1,
mnode sync reconfig failed
since %s"
,
terrstr
());
}
else
{
mInfo
(
"vgId:1,
reconfig mnode sync
success"
);
mInfo
(
"vgId:1,
mnode sync reconfig
success"
);
}
}
}
source/dnode/mnode/sdb/src/sdbFile.c
浏览文件 @
398161eb
...
...
@@ -243,7 +243,7 @@ static int32_t sdbReadFileImp(SSdb *pSdb) {
if
(
pFile
==
NULL
)
{
taosMemoryFree
(
pRaw
);
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
m
Debug
(
"failed to read sdb file:%s
since %s"
,
file
,
terrstr
());
m
Info
(
"read sdb file:%s finished
since %s"
,
file
,
terrstr
());
return
0
;
}
...
...
source/dnode/snode/src/snode.c
浏览文件 @
398161eb
...
...
@@ -124,6 +124,7 @@ FAIL:
}
void
sndClose
(
SSnode
*
pSnode
)
{
streamMetaCommit
(
pSnode
->
pMeta
);
streamMetaClose
(
pSnode
->
pMeta
);
taosMemoryFree
(
pSnode
->
path
);
taosMemoryFree
(
pSnode
);
...
...
source/dnode/vnode/inc/vnode.h
浏览文件 @
398161eb
...
...
@@ -89,6 +89,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
int32_t
vnodeProcessFetchMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
,
SQueueInfo
*
pInfo
);
void
vnodeProposeWriteMsg
(
SQueueInfo
*
pInfo
,
STaosQall
*
qall
,
int32_t
numOfMsgs
);
void
vnodeApplyWriteMsg
(
SQueueInfo
*
pInfo
,
STaosQall
*
qall
,
int32_t
numOfMsgs
);
void
vnodeProposeCommitOnNeed
(
SVnode
*
pVnode
);
// meta
typedef
struct
SMeta
SMeta
;
// todo: remove
...
...
source/dnode/vnode/src/inc/vnd.h
浏览文件 @
398161eb
...
...
@@ -86,6 +86,7 @@ int32_t vnodeGetBatchMeta(SVnode* pVnode, SRpcMsg* pMsg);
// vnodeCommit.c
int32_t
vnodeBegin
(
SVnode
*
pVnode
);
int32_t
vnodeShouldCommit
(
SVnode
*
pVnode
);
void
vnodeUpdCommitSched
(
SVnode
*
pVnode
);
void
vnodeRollback
(
SVnode
*
pVnode
);
int32_t
vnodeSaveInfo
(
const
char
*
dir
,
const
SVnodeInfo
*
pCfg
);
int32_t
vnodeCommitInfo
(
const
char
*
dir
,
const
SVnodeInfo
*
pInfo
);
...
...
@@ -103,6 +104,7 @@ void vnodeSyncClose(SVnode* pVnode);
void
vnodeRedirectRpcMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
,
int32_t
code
);
bool
vnodeIsLeader
(
SVnode
*
pVnode
);
bool
vnodeIsRoleLeader
(
SVnode
*
pVnode
);
int
vnodeShouldCommit
(
SVnode
*
pVnode
);
#ifdef __cplusplus
}
...
...
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @
398161eb
...
...
@@ -325,6 +325,11 @@ struct STsdbKeepCfg {
int32_t
keep2
;
};
typedef
struct
SVCommitSched
{
int64_t
commitMs
;
int64_t
maxWaitMs
;
}
SVCommitSched
;
struct
SVnode
{
char
*
path
;
SVnodeCfg
config
;
...
...
@@ -343,6 +348,7 @@ struct SVnode {
STQ
*
pTq
;
SSink
*
pSink
;
tsem_t
canCommit
;
SVCommitSched
commitSched
;
int64_t
sync
;
TdThreadMutex
lock
;
bool
blocked
;
...
...
source/dnode/vnode/src/meta/metaOpen.c
浏览文件 @
398161eb
...
...
@@ -203,6 +203,7 @@ _err:
int
metaClose
(
SMeta
*
pMeta
)
{
if
(
pMeta
)
{
if
(
pMeta
->
pEnv
)
tdbAbort
(
pMeta
->
pEnv
,
pMeta
->
txn
);
if
(
pMeta
->
pCache
)
metaCacheClose
(
pMeta
);
if
(
pMeta
->
pIdx
)
metaCloseIdx
(
pMeta
);
if
(
pMeta
->
pStreamDb
)
tdbTbClose
(
pMeta
->
pStreamDb
);
...
...
source/dnode/vnode/src/tq/tqCommit.c
浏览文件 @
398161eb
...
...
@@ -15,4 +15,11 @@
#include "tq.h"
int
tqCommit
(
STQ
*
pTq
)
{
return
tqOffsetCommitFile
(
pTq
->
pOffsetStore
);
}
int
tqCommit
(
STQ
*
pTq
)
{
if
(
streamMetaCommit
(
pTq
->
pStreamMeta
)
<
0
)
{
tqError
(
"vgId:%d, failed to commit stream meta since %s"
,
TD_VID
(
pTq
->
pVnode
),
terrstr
());
return
-
1
;
}
return
tqOffsetCommitFile
(
pTq
->
pOffsetStore
);
}
source/dnode/vnode/src/tsdb/tsdbCache.c
浏览文件 @
398161eb
...
...
@@ -1153,6 +1153,8 @@ static int32_t nextRowIterGet(CacheNextRowIter *pIter, TSDBROW **ppRow) {
iMax
[
nMax
]
=
i
;
max
[
nMax
++
]
=
pIter
->
input
[
i
].
pRow
;
}
else
{
pIter
->
input
[
i
].
next
=
false
;
}
}
}
...
...
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @
398161eb
...
...
@@ -2421,6 +2421,19 @@ static int32_t loadNeighborIfOverlap(SFileDataBlockInfo* pBlockInfo, STableBlock
return
code
;
}
static
void
updateComposedBlockInfo
(
STsdbReader
*
pReader
,
double
el
,
STableBlockScanInfo
*
pBlockScanInfo
)
{
SSDataBlock
*
pResBlock
=
pReader
->
pResBlock
;
pResBlock
->
info
.
id
.
uid
=
(
pBlockScanInfo
!=
NULL
)
?
pBlockScanInfo
->
uid
:
0
;
pResBlock
->
info
.
dataLoad
=
1
;
blockDataUpdateTsWindow
(
pResBlock
,
pReader
->
suppInfo
.
slotId
[
0
]);
setComposedBlockFlag
(
pReader
,
true
);
pReader
->
cost
.
composedBlocks
+=
1
;
pReader
->
cost
.
buildComposedBlockTime
+=
el
;
}
static
int32_t
buildComposedDataBlock
(
STsdbReader
*
pReader
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
...
...
@@ -2432,6 +2445,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
bool
asc
=
ASCENDING_TRAVERSE
(
pReader
->
order
);
int64_t
st
=
taosGetTimestampUs
();
int32_t
step
=
asc
?
1
:
-
1
;
double
el
=
0
;
STableBlockScanInfo
*
pBlockScanInfo
=
NULL
;
if
(
pBlockInfo
!=
NULL
)
{
...
...
@@ -2494,10 +2508,8 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
}
}
bool
hasBlockLData
=
hasDataInLastBlock
(
pLastBlockReader
);
// no data in last block and block, no need to proceed.
if
(
(
hasBlockData
==
false
)
&&
(
hasBlockLData
==
false
)
)
{
if
(
hasBlockData
==
false
)
{
break
;
}
...
...
@@ -2516,15 +2528,8 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
}
_end:
pResBlock
->
info
.
id
.
uid
=
(
pBlockScanInfo
!=
NULL
)
?
pBlockScanInfo
->
uid
:
0
;
pResBlock
->
info
.
dataLoad
=
1
;
blockDataUpdateTsWindow
(
pResBlock
,
pReader
->
suppInfo
.
slotId
[
0
]);
setComposedBlockFlag
(
pReader
,
true
);
double
el
=
(
taosGetTimestampUs
()
-
st
)
/
1000
.
0
;
pReader
->
cost
.
composedBlocks
+=
1
;
pReader
->
cost
.
buildComposedBlockTime
+=
el
;
el
=
(
taosGetTimestampUs
()
-
st
)
/
1000
.
0
;
updateComposedBlockInfo
(
pReader
,
el
,
pBlockScanInfo
);
if
(
pResBlock
->
info
.
rows
>
0
)
{
tsdbDebug
(
"%p uid:%"
PRIu64
", composed data block created, brange:%"
PRIu64
"-%"
PRIu64
...
...
@@ -2768,6 +2773,8 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
return
code
;
}
SSDataBlock
*
pResBlock
=
pReader
->
pResBlock
;
while
(
1
)
{
// load the last data block of current table
STableBlockScanInfo
*
pScanInfo
=
*
(
STableBlockScanInfo
**
)
pStatus
->
pTableIter
;
...
...
@@ -2778,15 +2785,33 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
if
(
!
hasNexTable
)
{
return
TSDB_CODE_SUCCESS
;
}
continue
;
}
code
=
doBuildDataBlock
(
pReader
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
int64_t
st
=
taosGetTimestampUs
();
while
(
1
)
{
bool
hasBlockLData
=
hasDataInLastBlock
(
pLastBlockReader
);
// no data in last block and block, no need to proceed.
if
(
hasBlockLData
==
false
)
{
break
;
}
buildComposedDataBlockImpl
(
pReader
,
pScanInfo
,
&
pReader
->
status
.
fileBlockData
,
pLastBlockReader
);
if
(
pResBlock
->
info
.
rows
>=
pReader
->
capacity
)
{
break
;
}
}
if
(
pReader
->
pResBlock
->
info
.
rows
>
0
)
{
double
el
=
(
taosGetTimestampUs
()
-
st
)
/
1000
.
0
;
updateComposedBlockInfo
(
pReader
,
el
,
pScanInfo
);
if
(
pResBlock
->
info
.
rows
>
0
)
{
tsdbDebug
(
"%p uid:%"
PRIu64
", composed data block created, brange:%"
PRIu64
"-%"
PRIu64
" rows:%d, elapsed time:%.2f ms %s"
,
pReader
,
pResBlock
->
info
.
id
.
uid
,
pResBlock
->
info
.
window
.
skey
,
pResBlock
->
info
.
window
.
ekey
,
pResBlock
->
info
.
rows
,
el
,
pReader
->
idStr
);
return
TSDB_CODE_SUCCESS
;
}
...
...
source/dnode/vnode/src/vnd/vnodeCommit.c
浏览文件 @
398161eb
...
...
@@ -58,7 +58,25 @@ int vnodeBegin(SVnode *pVnode) {
return
0
;
}
void
vnodeUpdCommitSched
(
SVnode
*
pVnode
)
{
int64_t
randNum
=
taosRand
();
pVnode
->
commitSched
.
commitMs
=
taosGetMonoTimestampMs
();
pVnode
->
commitSched
.
maxWaitMs
=
tsVndCommitMaxIntervalMs
+
(
randNum
%
tsVndCommitMaxIntervalMs
);
}
int
vnodeShouldCommit
(
SVnode
*
pVnode
)
{
if
(
!
pVnode
->
inUse
||
!
osDataSpaceAvailable
())
{
return
false
;
}
SVCommitSched
*
pSched
=
&
pVnode
->
commitSched
;
int64_t
nowMs
=
taosGetMonoTimestampMs
();
return
(((
pVnode
->
inUse
->
size
>
pVnode
->
inUse
->
node
.
size
)
&&
(
pSched
->
commitMs
+
SYNC_VND_COMMIT_MIN_MS
<
nowMs
))
||
(
pVnode
->
inUse
->
size
>
0
&&
pSched
->
commitMs
+
pSched
->
maxWaitMs
<
nowMs
));
}
int
vnodeShouldCommitOld
(
SVnode
*
pVnode
)
{
if
(
pVnode
->
inUse
)
{
return
osDataSpaceAvailable
()
&&
(
pVnode
->
inUse
->
size
>
pVnode
->
inUse
->
node
.
size
);
}
...
...
@@ -249,6 +267,7 @@ _exit:
taosMemoryFree
(
pInfo
);
return
code
;
}
int
vnodeAsyncCommit
(
SVnode
*
pVnode
)
{
int32_t
code
=
0
;
...
...
@@ -297,7 +316,9 @@ static int vnodeCommitImpl(SCommitInfo *pInfo) {
SVnode
*
pVnode
=
pInfo
->
pVnode
;
vInfo
(
"vgId:%d, start to commit, commitId:%"
PRId64
" version:%"
PRId64
" term: %"
PRId64
,
TD_VID
(
pVnode
),
pVnode
->
state
.
commitID
,
pVnode
->
state
.
applied
,
pVnode
->
state
.
applyTerm
);
pInfo
->
info
.
state
.
commitID
,
pInfo
->
info
.
state
.
committed
,
pInfo
->
info
.
state
.
commitTerm
);
vnodeUpdCommitSched
(
pVnode
);
// persist wal before starting
if
(
walPersist
(
pVnode
->
pWal
)
<
0
)
{
...
...
source/dnode/vnode/src/vnd/vnodeOpen.c
浏览文件 @
398161eb
...
...
@@ -160,6 +160,8 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
taosThreadMutexInit
(
&
pVnode
->
mutex
,
NULL
);
taosThreadCondInit
(
&
pVnode
->
poolNotEmpty
,
NULL
);
vnodeUpdCommitSched
(
pVnode
);
int8_t
rollback
=
vnodeShouldRollback
(
pVnode
);
// open buffer pool
...
...
@@ -254,7 +256,7 @@ void vnodePostClose(SVnode *pVnode) { vnodeSyncPostClose(pVnode); }
void
vnodeClose
(
SVnode
*
pVnode
)
{
if
(
pVnode
)
{
vnodeSyncCommit
(
pVnode
);
tsem_wait
(
&
pVnode
->
canCommit
);
vnodeSyncClose
(
pVnode
);
vnodeQueryClose
(
pVnode
);
walClose
(
pVnode
->
pWal
);
...
...
@@ -263,6 +265,8 @@ void vnodeClose(SVnode *pVnode) {
smaClose
(
pVnode
->
pSma
);
metaClose
(
pVnode
->
pMeta
);
vnodeCloseBufPool
(
pVnode
);
tsem_post
(
&
pVnode
->
canCommit
);
// destroy handle
tsem_destroy
(
&
(
pVnode
->
canCommit
));
tsem_destroy
(
&
pVnode
->
syncSem
);
...
...
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @
398161eb
...
...
@@ -223,6 +223,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
// skip header
pReq
=
POINTER_SHIFT
(
pMsg
->
pCont
,
sizeof
(
SMsgHead
));
len
=
pMsg
->
contLen
-
sizeof
(
SMsgHead
);
bool
needCommit
=
false
;
switch
(
pMsg
->
msgType
)
{
/* META */
...
...
@@ -319,9 +320,8 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
vnodeProcessAlterConfigReq
(
pVnode
,
version
,
pReq
,
len
,
pRsp
);
break
;
case
TDMT_VND_COMMIT
:
vnodeSyncCommit
(
pVnode
);
vnodeBegin
(
pVnode
);
goto
_exit
;
needCommit
=
true
;
break
;
default:
vError
(
"vgId:%d, unprocessed msg, %d"
,
TD_VID
(
pVnode
),
pMsg
->
msgType
);
return
-
1
;
...
...
@@ -338,7 +338,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
}
// commit if need
if
(
vnodeShouldCommit
(
pVnode
)
)
{
if
(
needCommit
)
{
vInfo
(
"vgId:%d, commit at version %"
PRId64
,
TD_VID
(
pVnode
),
version
);
if
(
vnodeAsyncCommit
(
pVnode
)
<
0
)
{
vError
(
"vgId:%d, failed to vnode async commit since %s."
,
TD_VID
(
pVnode
),
tstrerror
(
terrno
));
...
...
source/dnode/vnode/src/vnd/vnodeSync.c
浏览文件 @
398161eb
...
...
@@ -101,6 +101,64 @@ static void vnodeHandleProposeError(SVnode *pVnode, SRpcMsg *pMsg, int32_t code)
}
}
static
int32_t
inline
vnodeProposeMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
,
bool
isWeak
)
{
int64_t
seq
=
0
;
taosThreadMutexLock
(
&
pVnode
->
lock
);
int32_t
code
=
syncPropose
(
pVnode
->
sync
,
pMsg
,
isWeak
,
&
seq
);
bool
wait
=
(
code
==
0
&&
vnodeIsMsgBlock
(
pMsg
->
msgType
));
if
(
wait
)
{
ASSERT
(
!
pVnode
->
blocked
);
pVnode
->
blocked
=
true
;
pVnode
->
blockSec
=
taosGetTimestampSec
();
pVnode
->
blockSeq
=
seq
;
#if 0
pVnode->blockInfo = pMsg->info;
#endif
}
taosThreadMutexUnlock
(
&
pVnode
->
lock
);
if
(
code
>
0
)
{
vnodeHandleWriteMsg
(
pVnode
,
pMsg
);
}
else
if
(
code
<
0
)
{
if
(
terrno
!=
0
)
code
=
terrno
;
vnodeHandleProposeError
(
pVnode
,
pMsg
,
code
);
}
if
(
wait
)
vnodeWaitBlockMsg
(
pVnode
,
pMsg
);
return
code
;
}
void
vnodeProposeCommitOnNeed
(
SVnode
*
pVnode
)
{
if
(
!
vnodeShouldCommit
(
pVnode
))
{
return
;
}
int32_t
contLen
=
sizeof
(
SMsgHead
);
SMsgHead
*
pHead
=
rpcMallocCont
(
contLen
);
pHead
->
contLen
=
contLen
;
pHead
->
vgId
=
pVnode
->
config
.
vgId
;
SRpcMsg
rpcMsg
=
{
0
};
rpcMsg
.
msgType
=
TDMT_VND_COMMIT
;
rpcMsg
.
contLen
=
contLen
;
rpcMsg
.
pCont
=
pHead
;
rpcMsg
.
info
.
noResp
=
1
;
bool
isWeak
=
false
;
if
(
vnodeProposeMsg
(
pVnode
,
&
rpcMsg
,
isWeak
)
<
0
)
{
vTrace
(
"vgId:%d, failed to propose vnode commit since %s"
,
pVnode
->
config
.
vgId
,
terrstr
());
goto
_out
;
}
vInfo
(
"vgId:%d, proposed vnode commit"
,
pVnode
->
config
.
vgId
);
_out:
vnodeUpdCommitSched
(
pVnode
);
rpcFreeCont
(
rpcMsg
.
pCont
);
rpcMsg
.
pCont
=
NULL
;
}
#if BATCH_ENABLE
static
void
inline
vnodeProposeBatchMsg
(
SVnode
*
pVnode
,
SRpcMsg
**
pMsgArr
,
bool
*
pIsWeakArr
,
int32_t
*
arrSize
)
{
...
...
@@ -178,6 +236,8 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
continue
;
}
vnodeProposeCommitOnNeed
(
pVnode
);
code
=
vnodePreProcessWriteMsg
(
pVnode
,
pMsg
);
if
(
code
!=
0
)
{
vGError
(
"vgId:%d, msg:%p failed to pre-process since %s"
,
vgId
,
pMsg
,
terrstr
());
...
...
@@ -205,34 +265,6 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
#else
static
int32_t
inline
vnodeProposeMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
,
bool
isWeak
)
{
int64_t
seq
=
0
;
taosThreadMutexLock
(
&
pVnode
->
lock
);
int32_t
code
=
syncPropose
(
pVnode
->
sync
,
pMsg
,
isWeak
,
&
seq
);
bool
wait
=
(
code
==
0
&&
vnodeIsMsgBlock
(
pMsg
->
msgType
));
if
(
wait
)
{
ASSERT
(
!
pVnode
->
blocked
);
pVnode
->
blocked
=
true
;
pVnode
->
blockSec
=
taosGetTimestampSec
();
pVnode
->
blockSeq
=
seq
;
#if 0
pVnode->blockInfo = pMsg->info;
#endif
}
taosThreadMutexUnlock
(
&
pVnode
->
lock
);
if
(
code
>
0
)
{
vnodeHandleWriteMsg
(
pVnode
,
pMsg
);
}
else
if
(
code
<
0
)
{
if
(
terrno
!=
0
)
code
=
terrno
;
vnodeHandleProposeError
(
pVnode
,
pMsg
,
code
);
}
if
(
wait
)
vnodeWaitBlockMsg
(
pVnode
,
pMsg
);
return
code
;
}
void
vnodeProposeWriteMsg
(
SQueueInfo
*
pInfo
,
STaosQall
*
qall
,
int32_t
numOfMsgs
)
{
SVnode
*
pVnode
=
pInfo
->
ahandle
;
int32_t
vgId
=
pVnode
->
config
.
vgId
;
...
...
@@ -256,6 +288,8 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
continue
;
}
vnodeProposeCommitOnNeed
(
pVnode
);
code
=
vnodePreProcessWriteMsg
(
pVnode
,
pMsg
);
if
(
code
!=
0
)
{
vGError
(
"vgId:%d, msg:%p failed to pre-process since %s"
,
vgId
,
pMsg
,
terrstr
());
...
...
source/libs/stream/src/streamMeta.c
浏览文件 @
398161eb
...
...
@@ -69,8 +69,7 @@ _err:
}
void
streamMetaClose
(
SStreamMeta
*
pMeta
)
{
tdbCommit
(
pMeta
->
db
,
pMeta
->
txn
);
tdbPostCommit
(
pMeta
->
db
,
pMeta
->
txn
);
tdbAbort
(
pMeta
->
db
,
pMeta
->
txn
);
tdbTbClose
(
pMeta
->
pTaskDb
);
tdbTbClose
(
pMeta
->
pCheckpointDb
);
tdbClose
(
pMeta
->
db
);
...
...
@@ -88,6 +87,7 @@ void streamMetaClose(SStreamMeta* pMeta) {
/*streamMetaReleaseTask(pMeta, pTask);*/
}
taosHashCleanup
(
pMeta
->
pTasks
);
taosHashCleanup
(
pMeta
->
pRecoverStatus
);
taosMemoryFree
(
pMeta
->
path
);
taosMemoryFree
(
pMeta
);
}
...
...
source/libs/sync/inc/syncMessage.h
浏览文件 @
398161eb
...
...
@@ -247,8 +247,8 @@ typedef struct SyncLocalCmd {
SRaftId
destId
;
int32_t
cmd
;
SyncTerm
sdNew
Term
;
// step down new term
SyncIndex
fcIndex
;
// follower commit index
SyncTerm
current
Term
;
// step down new term
SyncIndex
commitIndex
;
// follower commit index
}
SyncLocalCmd
;
int32_t
syncBuildTimeout
(
SRpcMsg
*
pMsg
,
ESyncTimeoutType
ttype
,
uint64_t
logicClock
,
int32_t
ms
,
SSyncNode
*
pNode
);
...
...
source/libs/sync/inc/syncPipeline.h
浏览文件 @
398161eb
...
...
@@ -98,6 +98,7 @@ int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode);
// access
int64_t
syncLogBufferGetEndIndex
(
SSyncLogBuffer
*
pBuf
);
SyncTerm
syncLogBufferGetLastMatchTerm
(
SSyncLogBuffer
*
pBuf
);
int32_t
syncLogBufferAppend
(
SSyncLogBuffer
*
pBuf
,
SSyncNode
*
pNode
,
SSyncRaftEntry
*
pEntry
);
int32_t
syncLogBufferAccept
(
SSyncLogBuffer
*
pBuf
,
SSyncNode
*
pNode
,
SSyncRaftEntry
*
pEntry
,
SyncTerm
prevTerm
);
int64_t
syncLogBufferProceed
(
SSyncLogBuffer
*
pBuf
,
SSyncNode
*
pNode
,
SyncTerm
*
pMatchTerm
);
...
...
source/libs/sync/src/syncAppendEntries.c
浏览文件 @
398161eb
...
...
@@ -90,6 +90,7 @@
//
int32_t
syncNodeFollowerCommit
(
SSyncNode
*
ths
,
SyncIndex
newCommitIndex
)
{
ASSERT
(
false
&&
"deprecated"
);
if
(
ths
->
state
!=
TAOS_SYNC_STATE_FOLLOWER
)
{
sNTrace
(
ths
,
"can not do follower commit"
);
return
-
1
;
...
...
@@ -206,12 +207,13 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
accepted
=
true
;
_SEND_RESPONSE:
pEntry
=
NULL
;
pReply
->
matchIndex
=
syncLogBufferProceed
(
ths
->
pLogBuf
,
ths
,
&
pReply
->
lastMatchTerm
);
bool
matched
=
(
pReply
->
matchIndex
>=
pReply
->
lastSendIndex
);
if
(
accepted
&&
matched
)
{
pReply
->
success
=
true
;
// update commit index only after matching
(
void
)
syncNodeUpdateCommitIndex
(
ths
,
pMsg
->
commitIndex
);
(
void
)
syncNodeUpdateCommitIndex
(
ths
,
TMIN
(
pMsg
->
commitIndex
,
pReply
->
lastSendIndex
)
);
}
// ack, i.e. send response
...
...
source/libs/sync/src/syncCommit.c
浏览文件 @
398161eb
...
...
@@ -44,6 +44,7 @@
// /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
//
void
syncOneReplicaAdvance
(
SSyncNode
*
pSyncNode
)
{
ASSERT
(
false
&&
"deprecated"
);
if
(
pSyncNode
==
NULL
)
{
sError
(
"pSyncNode is NULL"
);
return
;
...
...
source/libs/sync/src/syncEnv.c
浏览文件 @
398161eb
...
...
@@ -114,7 +114,7 @@ void syncHbTimerDataRemove(int64_t rid) { taosRemoveRef(gHbDataRefId, rid); }
SSyncHbTimerData
*
syncHbTimerDataAcquire
(
int64_t
rid
)
{
SSyncHbTimerData
*
pData
=
taosAcquireRef
(
gHbDataRefId
,
rid
);
if
(
pData
==
NULL
)
{
if
(
pData
==
NULL
&&
rid
>
0
)
{
sInfo
(
"failed to acquire hb-timer-data from refId:%"
PRId64
,
rid
);
terrno
=
TSDB_CODE_SYN_INTERNAL_ERROR
;
}
...
...
source/libs/sync/src/syncMain.c
浏览文件 @
398161eb
...
...
@@ -1030,6 +1030,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
}
}
pSyncNode
->
commitIndex
=
commitIndex
;
sInfo
(
"vgId:%d, sync node commitIndex initialized as %"
PRId64
,
pSyncNode
->
vgId
,
pSyncNode
->
commitIndex
);
if
(
syncNodeLogStoreRestoreOnNeed
(
pSyncNode
)
<
0
)
{
goto
_error
;
...
...
@@ -1170,9 +1171,10 @@ int32_t syncNodeRestore(SSyncNode* pSyncNode) {
}
ASSERT
(
endIndex
==
lastVer
+
1
);
commitIndex
=
TMAX
(
pSyncNode
->
commitIndex
,
commitIndex
);
pSyncNode
->
commitIndex
=
TMAX
(
pSyncNode
->
commitIndex
,
commitIndex
);
sInfo
(
"vgId:%d, restore sync until commitIndex:%"
PRId64
,
pSyncNode
->
vgId
,
pSyncNode
->
commitIndex
);
if
(
syncLogBufferCommit
(
pSyncNode
->
pLogBuf
,
pSyncNode
,
commitIndex
)
<
0
)
{
if
(
syncLogBufferCommit
(
pSyncNode
->
pLogBuf
,
pSyncNode
,
pSyncNode
->
commitIndex
)
<
0
)
{
return
-
1
;
}
...
...
@@ -1480,16 +1482,21 @@ int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pNode, SRpcMsg
}
}
int32_t
code
=
-
1
;
if
(
pNode
->
syncSendMSg
!=
NULL
&&
epSet
!=
NULL
)
{
syncUtilMsgHtoN
(
pMsg
->
pCont
);
pMsg
->
info
.
noResp
=
1
;
return
pNode
->
syncSendMSg
(
epSet
,
pMsg
);
}
else
{
sError
(
"vgId:%d, sync send msg by id error, fp:%p epset:%p"
,
pNode
->
vgId
,
pNode
->
syncSendMSg
,
epSet
);
code
=
pNode
->
syncSendMSg
(
epSet
,
pMsg
);
}
if
(
code
<
0
)
{
sError
(
"vgId:%d, sync send msg by id error, epset:%p dnode:%d addr:%"
PRId64
" err:0x%x"
,
pNode
->
vgId
,
epSet
,
DID
(
destRaftId
),
destRaftId
->
addr
,
terrno
);
rpcFreeCont
(
pMsg
->
pCont
);
terrno
=
TSDB_CODE_SYN_INTERNAL_ERROR
;
return
-
1
;
}
return
code
;
}
inline
bool
syncNodeInConfig
(
SSyncNode
*
pNode
,
const
SSyncCfg
*
pCfg
)
{
...
...
@@ -2537,8 +2544,9 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncLocalCmd
*
pSyncMsg
=
rpcMsgLocalCmd
.
pCont
;
pSyncMsg
->
cmd
=
SYNC_LOCAL_CMD_FOLLOWER_CMT
;
pSyncMsg
->
fcIndex
=
pMsg
->
commitIndex
;
SyncIndex
fcIndex
=
pSyncMsg
->
fcIndex
;
pSyncMsg
->
commitIndex
=
pMsg
->
commitIndex
;
pSyncMsg
->
currentTerm
=
pMsg
->
term
;
SyncIndex
fcIndex
=
pSyncMsg
->
commitIndex
;
if
(
ths
->
syncEqMsg
!=
NULL
&&
ths
->
msgcb
!=
NULL
)
{
int32_t
code
=
ths
->
syncEqMsg
(
ths
->
msgcb
,
&
rpcMsgLocalCmd
);
...
...
@@ -2559,7 +2567,8 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncLocalCmd
*
pSyncMsg
=
rpcMsgLocalCmd
.
pCont
;
pSyncMsg
->
cmd
=
SYNC_LOCAL_CMD_STEP_DOWN
;
pSyncMsg
->
sdNewTerm
=
pMsg
->
term
;
pSyncMsg
->
currentTerm
=
pMsg
->
term
;
pSyncMsg
->
commitIndex
=
pMsg
->
commitIndex
;
if
(
ths
->
syncEqMsg
!=
NULL
&&
ths
->
msgcb
!=
NULL
)
{
int32_t
code
=
ths
->
syncEqMsg
(
ths
->
msgcb
,
&
rpcMsgLocalCmd
);
...
...
@@ -2567,7 +2576,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
sError
(
"vgId:%d, sync enqueue step-down msg error, code:%d"
,
ths
->
vgId
,
code
);
rpcFreeCont
(
rpcMsgLocalCmd
.
pCont
);
}
else
{
sTrace
(
"vgId:%d, sync enqueue step-down msg, new-term: %"
PRId64
,
ths
->
vgId
,
pSyncMsg
->
sdNew
Term
);
sTrace
(
"vgId:%d, sync enqueue step-down msg, new-term: %"
PRId64
,
ths
->
vgId
,
pSyncMsg
->
current
Term
);
}
}
}
...
...
@@ -2625,10 +2634,13 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
syncLogRecvLocalCmd
(
ths
,
pMsg
,
""
);
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_STEP_DOWN
)
{
syncNodeStepDown
(
ths
,
pMsg
->
sdNew
Term
);
syncNodeStepDown
(
ths
,
pMsg
->
current
Term
);
}
else
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_FOLLOWER_CMT
)
{
(
void
)
syncNodeUpdateCommitIndex
(
ths
,
pMsg
->
fcIndex
);
SyncTerm
matchTerm
=
syncLogBufferGetLastMatchTerm
(
ths
->
pLogBuf
);
if
(
pMsg
->
currentTerm
==
matchTerm
)
{
(
void
)
syncNodeUpdateCommitIndex
(
ths
,
pMsg
->
commitIndex
);
}
if
(
syncLogBufferCommit
(
ths
->
pLogBuf
,
ths
,
ths
->
commitIndex
)
<
0
)
{
sError
(
"vgId:%d, failed to commit raft log since %s. commit index: %"
PRId64
""
,
ths
->
vgId
,
terrstr
(),
ths
->
commitIndex
);
...
...
@@ -2641,14 +2653,15 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
}
int32_t
syncNodeOnLocalCmdOld
(
SSyncNode
*
ths
,
const
SRpcMsg
*
pRpcMsg
)
{
ASSERT
(
false
&&
"deprecated"
);
SyncLocalCmd
*
pMsg
=
pRpcMsg
->
pCont
;
syncLogRecvLocalCmd
(
ths
,
pMsg
,
""
);
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_STEP_DOWN
)
{
syncNodeStepDown
(
ths
,
pMsg
->
sdNew
Term
);
syncNodeStepDown
(
ths
,
pMsg
->
current
Term
);
}
else
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_FOLLOWER_CMT
)
{
syncNodeFollowerCommit
(
ths
,
pMsg
->
fc
Index
);
syncNodeFollowerCommit
(
ths
,
pMsg
->
commit
Index
);
}
else
{
sError
(
"error local cmd"
);
...
...
source/libs/sync/src/syncPipeline.c
浏览文件 @
398161eb
...
...
@@ -31,6 +31,10 @@ static bool syncIsMsgBlock(tmsg_t type) {
(
type
==
TDMT_VND_UPDATE_TAG_VAL
)
||
(
type
==
TDMT_VND_ALTER_CONFIRM
);
}
FORCE_INLINE
static
int64_t
syncGetRetryMaxWaitMs
()
{
return
SYNC_LOG_REPL_RETRY_WAIT_MS
*
(
1
<<
SYNC_MAX_RETRY_BACKOFF
);
}
int64_t
syncLogBufferGetEndIndex
(
SSyncLogBuffer
*
pBuf
)
{
taosThreadMutexLock
(
&
pBuf
->
mutex
);
int64_t
index
=
pBuf
->
endIndex
;
...
...
@@ -264,20 +268,27 @@ int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode) {
return
ret
;
}
FORCE_INLINE
SyncTerm
syncLogBufferGetLastMatchTerm
(
SSyncLogBuffer
*
pBuf
)
{
FORCE_INLINE
SyncTerm
syncLogBufferGetLastMatchTerm
WithoutLock
(
SSyncLogBuffer
*
pBuf
)
{
SyncIndex
index
=
pBuf
->
matchIndex
;
SSyncRaftEntry
*
pEntry
=
pBuf
->
entries
[(
index
+
pBuf
->
size
)
%
pBuf
->
size
].
pItem
;
ASSERT
(
pEntry
!=
NULL
);
return
pEntry
->
term
;
}
SyncTerm
syncLogBufferGetLastMatchTerm
(
SSyncLogBuffer
*
pBuf
)
{
taosThreadMutexLock
(
&
pBuf
->
mutex
);
SyncTerm
term
=
syncLogBufferGetLastMatchTermWithoutLock
(
pBuf
);
taosThreadMutexUnlock
(
&
pBuf
->
mutex
);
return
term
;
}
int32_t
syncLogBufferAccept
(
SSyncLogBuffer
*
pBuf
,
SSyncNode
*
pNode
,
SSyncRaftEntry
*
pEntry
,
SyncTerm
prevTerm
)
{
taosThreadMutexLock
(
&
pBuf
->
mutex
);
syncLogBufferValidate
(
pBuf
);
int32_t
ret
=
-
1
;
SyncIndex
index
=
pEntry
->
index
;
SyncIndex
prevIndex
=
pEntry
->
index
-
1
;
SyncTerm
lastMatchTerm
=
syncLogBufferGetLastMatchTerm
(
pBuf
);
SyncTerm
lastMatchTerm
=
syncLogBufferGetLastMatchTermWithoutLock
(
pBuf
);
SSyncRaftEntry
*
pExist
=
NULL
;
bool
inBuf
=
true
;
...
...
@@ -329,6 +340,8 @@ int32_t syncLogBufferAccept(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt
}
// update
ASSERT
(
pBuf
->
startIndex
<
index
);
ASSERT
(
index
-
pBuf
->
startIndex
<
pBuf
->
size
);
ASSERT
(
pBuf
->
entries
[
index
%
pBuf
->
size
].
pItem
==
NULL
);
SSyncLogBufEntry
tmp
=
{.
pItem
=
pEntry
,
.
prevLogIndex
=
prevIndex
,
.
prevLogTerm
=
prevTerm
};
pEntry
=
NULL
;
...
...
@@ -456,6 +469,11 @@ int32_t syncLogFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, Syn
pNode
->
vgId
,
pEntry
->
index
,
pEntry
->
term
,
TMSG_INFO
(
pEntry
->
originalRpcType
),
applyCode
);
}
if
(
pEntry
->
originalRpcType
==
TDMT_VND_COMMIT
)
{
sInfo
(
"vgId:%d, fsm execute vnode commit. index: %"
PRId64
", term: %"
PRId64
""
,
pNode
->
vgId
,
pEntry
->
index
,
pEntry
->
term
);
}
SRpcMsg
rpcMsg
=
{.
code
=
applyCode
};
syncEntry2OriginalRpc
(
pEntry
,
&
rpcMsg
);
...
...
@@ -552,7 +570,8 @@ int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t comm
ret
=
0
;
_out:
// mark as restored if needed
if
(
!
pNode
->
restoreFinish
&&
pBuf
->
commitIndex
>=
pNode
->
commitIndex
)
{
if
(
!
pNode
->
restoreFinish
&&
pBuf
->
commitIndex
>=
pNode
->
commitIndex
&&
pEntry
!=
NULL
&&
pNode
->
pRaftStore
->
currentTerm
<=
pEntry
->
term
)
{
pNode
->
pFsm
->
FpRestoreFinishCb
(
pNode
->
pFsm
);
pNode
->
restoreFinish
=
true
;
sInfo
(
"vgId:%d, restore finished. log buffer: [%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pNode
->
vgId
,
...
...
@@ -613,6 +632,12 @@ int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
}
if
(
pMgr
->
states
[
pos
].
acked
)
{
if
(
pMgr
->
matchIndex
<
index
&&
pMgr
->
states
[
pos
].
timeMs
+
(
syncGetRetryMaxWaitMs
()
<<
3
)
<
nowMs
)
{
syncLogReplMgrReset
(
pMgr
);
sWarn
(
"vgId:%d, reset sync log repl mgr since stagnation. index: %"
PRId64
", peer: %"
PRIx64
,
pNode
->
vgId
,
index
,
pDestId
->
addr
);
goto
_out
;
}
continue
;
}
...
...
@@ -639,10 +664,12 @@ int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
_out:
if
(
retried
)
{
pMgr
->
retryBackoff
=
syncLogGetNextRetryBackoff
(
pMgr
);
sInfo
(
"vgId:%d, resent %d sync log entries. dest: %"
PRIx64
", indexes: %"
PRId64
" ..., terms: ... %"
PRId64
", retryWaitMs: %"
PRId64
", repl mgr: [%"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
SSyncLogBuffer
*
pBuf
=
pNode
->
pLogBuf
;
sInfo
(
"vgId:%d, resend %d sync log entries. dest: %"
PRIx64
", indexes: %"
PRId64
" ..., terms: ... %"
PRId64
", retryWaitMs: %"
PRId64
", mgr: [%"
PRId64
" %"
PRId64
", %"
PRId64
"), buffer: [%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pNode
->
vgId
,
count
,
pDestId
->
addr
,
firstIndex
,
term
,
retryWaitMs
,
pMgr
->
startIndex
,
pMgr
->
matchIndex
,
pMgr
->
endIndex
);
pMgr
->
endIndex
,
pBuf
->
startIndex
,
pBuf
->
commitIndex
,
pBuf
->
matchIndex
,
pBuf
->
endIndex
);
}
return
ret
;
}
...
...
@@ -771,7 +798,7 @@ int32_t syncLogReplMgrReplicateOnce(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
int32_t
syncLogReplMgrReplicateProbeOnce
(
SSyncLogReplMgr
*
pMgr
,
SSyncNode
*
pNode
,
SyncIndex
index
)
{
ASSERT
(
!
pMgr
->
restored
);
ASSERT
(
pMgr
->
startIndex
>=
0
);
int64_t
retryMaxWaitMs
=
SYNC_LOG_REPL_RETRY_WAIT_MS
*
(
1
<<
SYNC_MAX_RETRY_BACKOFF
);
int64_t
retryMaxWaitMs
=
syncGetRetryMaxWaitMs
(
);
int64_t
nowMs
=
taosGetMonoTimestampMs
();
if
(
pMgr
->
endIndex
>
pMgr
->
startIndex
&&
...
...
@@ -799,11 +826,10 @@ int32_t syncLogReplMgrReplicateProbeOnce(SSyncLogReplMgr* pMgr, SSyncNode* pNode
pMgr
->
endIndex
=
index
+
1
;
SSyncLogBuffer
*
pBuf
=
pNode
->
pLogBuf
;
sTrace
(
"vgId:%d, attempted to probe the %d'th peer with msg of index:%"
PRId64
" term: %"
PRId64
". pMgr(rs:%d): [%"
PRId64
" %"
PRId64
", %"
PRId64
"), pBuf: [%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pNode
->
vgId
,
pMgr
->
peerId
,
index
,
term
,
pMgr
->
restored
,
pMgr
->
startIndex
,
pMgr
->
matchIndex
,
pMgr
->
endIndex
,
pBuf
->
startIndex
,
pBuf
->
commitIndex
,
pBuf
->
matchIndex
,
pBuf
->
endIndex
);
sInfo
(
"vgId:%d, probe peer:%"
PRIx64
" with msg of index:%"
PRId64
" term: %"
PRId64
". mgr (rs:%d): [%"
PRId64
" %"
PRId64
", %"
PRId64
"), buffer: [%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pNode
->
vgId
,
pDestId
->
addr
,
index
,
term
,
pMgr
->
restored
,
pMgr
->
startIndex
,
pMgr
->
matchIndex
,
pMgr
->
endIndex
,
pBuf
->
startIndex
,
pBuf
->
commitIndex
,
pBuf
->
matchIndex
,
pBuf
->
endIndex
);
return
0
;
}
...
...
@@ -815,9 +841,11 @@ int32_t syncLogReplMgrReplicateAttemptedOnce(SSyncLogReplMgr* pMgr, SSyncNode* p
int32_t
count
=
0
;
int64_t
nowMs
=
taosGetMonoTimestampMs
();
int64_t
limit
=
pMgr
->
size
>>
1
;
SyncTerm
term
=
-
1
;
SyncIndex
firstIndex
=
-
1
;
for
(
SyncIndex
index
=
pMgr
->
endIndex
;
index
<=
pNode
->
pLogBuf
->
matchIndex
;
index
++
)
{
if
(
batchSize
<
count
++
||
limit
<=
index
-
pMgr
->
startIndex
)
{
if
(
batchSize
<
count
||
limit
<=
index
-
pMgr
->
startIndex
)
{
break
;
}
if
(
pMgr
->
startIndex
+
1
<
index
&&
pMgr
->
states
[(
index
-
1
)
%
pMgr
->
size
].
barrier
)
{
...
...
@@ -826,7 +854,6 @@ int32_t syncLogReplMgrReplicateAttemptedOnce(SSyncLogReplMgr* pMgr, SSyncNode* p
int64_t
pos
=
index
%
pMgr
->
size
;
SRaftId
*
pDestId
=
&
pNode
->
replicasId
[
pMgr
->
peerId
];
bool
barrier
=
false
;
SyncTerm
term
=
-
1
;
if
(
syncLogBufferReplicateOneTo
(
pMgr
,
pNode
,
index
,
&
term
,
pDestId
,
&
barrier
)
<
0
)
{
sError
(
"vgId:%d, failed to replicate log entry since %s. index: %"
PRId64
", dest: 0x%016"
PRIx64
""
,
pNode
->
vgId
,
terrstr
(),
index
,
pDestId
->
addr
);
...
...
@@ -837,6 +864,9 @@ int32_t syncLogReplMgrReplicateAttemptedOnce(SSyncLogReplMgr* pMgr, SSyncNode* p
pMgr
->
states
[
pos
].
term
=
term
;
pMgr
->
states
[
pos
].
acked
=
false
;
if
(
firstIndex
==
-
1
)
firstIndex
=
index
;
count
++
;
pMgr
->
endIndex
=
index
+
1
;
if
(
barrier
)
{
sInfo
(
"vgId:%d, replicated sync barrier to dest: %"
PRIx64
". index: %"
PRId64
", term: %"
PRId64
...
...
@@ -850,10 +880,11 @@ int32_t syncLogReplMgrReplicateAttemptedOnce(SSyncLogReplMgr* pMgr, SSyncNode* p
syncLogReplMgrRetryOnNeed
(
pMgr
,
pNode
);
SSyncLogBuffer
*
pBuf
=
pNode
->
pLogBuf
;
sTrace
(
"vgId:%d, attempted to replicate %d msgs to the %d'th peer. pMgr(rs:%d): [%"
PRId64
" %"
PRId64
", %"
PRId64
"), pBuf: [%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pNode
->
vgId
,
count
,
pMgr
->
peerId
,
pMgr
->
restored
,
pMgr
->
startIndex
,
pMgr
->
matchIndex
,
pMgr
->
endIndex
,
pBuf
->
startIndex
,
pBuf
->
commitIndex
,
pBuf
->
matchIndex
,
pBuf
->
endIndex
);
sTrace
(
"vgId:%d, replicated %d msgs to peer: %"
PRIx64
". indexes: %"
PRId64
"..., terms: ...%"
PRId64
", mgr: (rs:%d) [%"
PRId64
" %"
PRId64
", %"
PRId64
"), buffer: [%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pNode
->
vgId
,
count
,
pDestId
->
addr
,
firstIndex
,
term
,
pMgr
->
restored
,
pMgr
->
startIndex
,
pMgr
->
matchIndex
,
pMgr
->
endIndex
,
pBuf
->
startIndex
,
pBuf
->
commitIndex
,
pBuf
->
matchIndex
,
pBuf
->
endIndex
);
return
0
;
}
...
...
@@ -985,6 +1016,10 @@ void syncLogBufferDestroy(SSyncLogBuffer* pBuf) {
int32_t
syncLogBufferRollback
(
SSyncLogBuffer
*
pBuf
,
SSyncNode
*
pNode
,
SyncIndex
toIndex
)
{
ASSERT
(
pBuf
->
commitIndex
<
toIndex
&&
toIndex
<=
pBuf
->
endIndex
);
if
(
toIndex
==
pBuf
->
endIndex
)
{
return
0
;
}
sInfo
(
"vgId:%d, rollback sync log buffer. toindex: %"
PRId64
", buffer: [%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pNode
->
vgId
,
toIndex
,
pBuf
->
startIndex
,
pBuf
->
commitIndex
,
pBuf
->
matchIndex
,
pBuf
->
endIndex
);
...
...
source/libs/sync/src/syncRaftLog.c
浏览文件 @
398161eb
...
...
@@ -219,6 +219,10 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
ASSERT
(
pEntry
->
index
==
index
);
if
(
pEntry
->
originalRpcType
==
TDMT_VND_COMMIT
)
{
walFsync
(
pWal
,
true
);
}
sNTrace
(
pData
->
pSyncNode
,
"write index:%"
PRId64
", type:%s, origin type:%s, elapsed:%"
PRId64
,
pEntry
->
index
,
TMSG_INFO
(
pEntry
->
msgType
),
TMSG_INFO
(
pEntry
->
originalRpcType
),
tsElapsed
);
return
0
;
...
...
@@ -312,29 +316,6 @@ static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIn
SSyncLogStoreData
*
pData
=
pLogStore
->
data
;
SWal
*
pWal
=
pData
->
pWal
;
// need not truncate
SyncIndex
wallastVer
=
walGetLastVer
(
pWal
);
if
(
fromIndex
>
wallastVer
)
{
return
0
;
}
// need not truncate
SyncIndex
walCommitVer
=
walGetCommittedVer
(
pWal
);
if
(
fromIndex
<=
walCommitVer
)
{
return
0
;
}
// delete from cache
for
(
SyncIndex
index
=
fromIndex
;
index
<=
wallastVer
;
++
index
)
{
SLRUCache
*
pCache
=
pData
->
pSyncNode
->
pLogStore
->
pCache
;
LRUHandle
*
h
=
taosLRUCacheLookup
(
pCache
,
&
index
,
sizeof
(
index
));
if
(
h
)
{
sNTrace
(
pData
->
pSyncNode
,
"cache delete index:%"
PRId64
,
index
);
taosLRUCacheRelease
(
pData
->
pSyncNode
->
pLogStore
->
pCache
,
h
,
true
);
}
}
int32_t
code
=
walRollback
(
pWal
,
fromIndex
);
if
(
code
!=
0
)
{
int32_t
err
=
terrno
;
...
...
source/libs/sync/src/syncUtil.c
浏览文件 @
398161eb
...
...
@@ -17,20 +17,20 @@
#include "syncUtil.h"
#include "syncIndexMgr.h"
#include "syncMessage.h"
#include "syncPipeline.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
#include "syncSnapshot.h"
void
syncCfg2SimpleStr
(
const
SSyncCfg
*
pCfg
,
char
*
buf
,
int32_t
bufLen
)
{
int32_t
len
=
snprintf
(
buf
,
bufLen
,
"{r-num:%d, my:%d, "
,
pCfg
->
replicaNum
,
pCfg
->
myIndex
);
int32_t
len
=
snprintf
(
buf
,
bufLen
,
"{num:%d, as:%d, ["
,
pCfg
->
replicaNum
,
pCfg
->
myIndex
);
for
(
int32_t
i
=
0
;
i
<
pCfg
->
replicaNum
;
++
i
)
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s:%d"
,
pCfg
->
nodeInfo
[
i
].
nodeFqdn
,
pCfg
->
nodeInfo
[
i
].
nodePort
);
if
(
i
<
pCfg
->
replicaNum
-
1
)
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s:%d, "
,
pCfg
->
nodeInfo
[
i
].
nodeFqdn
,
pCfg
->
nodeInfo
[
i
].
nodePort
);
}
else
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s:%d}"
,
pCfg
->
nodeInfo
[
i
].
nodeFqdn
,
pCfg
->
nodeInfo
[
i
].
nodePort
);
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
", "
);
}
}
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
"]}"
);
}
void
syncUtilNodeInfo2EpSet
(
const
SNodeInfo
*
pInfo
,
SEpSet
*
pEpSet
)
{
...
...
@@ -89,32 +89,55 @@ bool syncUtilUserRollback(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP &&
// for leader
static
void
syncHearbeatReplyTime2Str
(
SSyncNode
*
pSyncNode
,
char
*
buf
,
int32_t
bufLen
)
{
int32_t
len
=
5
;
int32_t
len
=
0
;
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
"{"
);
for
(
int32_t
i
=
0
;
i
<
pSyncNode
->
replicaNum
;
++
i
)
{
int64_t
tsMs
=
syncIndexMgrGetRecvTime
(
pSyncNode
->
pMatchIndex
,
&
(
pSyncNode
->
replicasId
[
i
]));
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%d:%"
PRId64
,
i
,
tsMs
);
if
(
i
<
pSyncNode
->
replicaNum
-
1
)
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%d:%"
PRId64
","
,
i
,
tsMs
);
}
else
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%d:%"
PRId64
"}"
,
i
,
tsMs
);
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
","
);
}
}
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
"}"
);
}
// for follower
static
void
syncHearbeatTime2Str
(
SSyncNode
*
pSyncNode
,
char
*
buf
,
int32_t
bufLen
)
{
int32_t
len
=
4
;
int32_t
len
=
0
;
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
"{"
);
for
(
int32_t
i
=
0
;
i
<
pSyncNode
->
replicaNum
;
++
i
)
{
int64_t
tsMs
=
syncIndexMgrGetRecvTime
(
pSyncNode
->
pNextIndex
,
&
(
pSyncNode
->
replicasId
[
i
]));
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%d:%"
PRId64
,
i
,
tsMs
);
if
(
i
<
pSyncNode
->
replicaNum
-
1
)
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%d:%"
PRId64
","
,
i
,
tsMs
);
}
else
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%d:%"
PRId64
"}"
,
i
,
tsMs
);
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
","
);
}
}
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
"}"
);
}
static
void
syncLogBufferStates2Str
(
SSyncNode
*
pSyncNode
,
char
*
buf
,
int32_t
bufLen
)
{
SSyncLogBuffer
*
pBuf
=
pSyncNode
->
pLogBuf
;
if
(
pBuf
==
NULL
)
{
return
;
}
int
len
=
0
;
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"[%"
PRId64
" %"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
pBuf
->
startIndex
,
pBuf
->
commitIndex
,
pBuf
->
matchIndex
,
pBuf
->
endIndex
);
}
static
void
syncLogReplMgrStates2Str
(
SSyncNode
*
pSyncNode
,
char
*
buf
,
int32_t
bufLen
)
{
int
len
=
0
;
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
"{"
);
for
(
int32_t
i
=
0
;
i
<
pSyncNode
->
replicaNum
;
i
++
)
{
SSyncLogReplMgr
*
pMgr
=
pSyncNode
->
logReplMgrs
[
i
];
if
(
pMgr
==
NULL
)
break
;
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%d:%d [%"
PRId64
" %"
PRId64
", %"
PRId64
")"
,
i
,
pMgr
->
restored
,
pMgr
->
startIndex
,
pMgr
->
matchIndex
,
pMgr
->
endIndex
);
if
(
i
+
1
<
pSyncNode
->
replicaNum
)
{
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
", "
);
}
}
len
+=
snprintf
(
buf
+
len
,
bufLen
-
len
,
"%s"
,
"}"
);
}
static
void
syncPeerState2Str
(
SSyncNode
*
pSyncNode
,
char
*
buf
,
int32_t
bufLen
)
{
...
...
@@ -156,16 +179,19 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
int32_t
cacheHit
=
pNode
->
pLogStore
->
cacheHit
;
int32_t
cacheMiss
=
pNode
->
pLogStore
->
cacheMiss
;
char
cfgStr
[
1024
];
char
cfgStr
[
1024
]
=
""
;
syncCfg2SimpleStr
(
&
pNode
->
raftCfg
.
cfg
,
cfgStr
,
sizeof
(
cfgStr
));
char
peerStr
[
1024
]
=
"{
"
;
sync
PeerState2Str
(
pNode
,
peerStr
,
sizeof
(
peer
Str
));
char
replMgrStatesStr
[
1024
]
=
"
"
;
sync
LogReplMgrStates2Str
(
pNode
,
replMgrStatesStr
,
sizeof
(
replMgrStates
Str
));
char
hbrTimeStr
[
256
]
=
"hbr:{"
;
char
bufferStatesStr
[
256
]
=
""
;
syncLogBufferStates2Str
(
pNode
,
bufferStatesStr
,
sizeof
(
bufferStatesStr
));
char
hbrTimeStr
[
256
]
=
""
;
syncHearbeatReplyTime2Str
(
pNode
,
hbrTimeStr
,
sizeof
(
hbrTimeStr
));
char
hbTimeStr
[
256
]
=
"
hb:{
"
;
char
hbTimeStr
[
256
]
=
""
;
syncHearbeatTime2Str
(
pNode
,
hbTimeStr
,
sizeof
(
hbTimeStr
));
char
eventLog
[
512
];
// {0};
...
...
@@ -181,21 +207,21 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
// restore error code
terrno
=
errCode
;
if
(
pNode
!=
NULL
)
{
taosPrintLog
(
flags
,
level
,
dflag
,
"vgId:%d, %s, sync:%s, term:%"
PRIu64
", commit-index:%"
PRId64
", first-ver:%"
PRId64
", last-ver:%"
PRId64
", min:%"
PRId64
", snap:%"
PRId64
", snap-term:%"
PRIu64
", elect-times:%d, as-leader-times:%d, cfg-ch-times:%d, h
it:%d, mis:%d, h
b-slow:%d, hbr-slow:%d, "
", elect-times:%d, as-leader-times:%d, cfg-ch-times:%d, hb-slow:%d, hbr-slow:%d, "
"aq-items:%d, snaping:%"
PRId64
", replicas:%d, last-cfg:%"
PRId64
", chging:%d, restore:%d, quorum:%d, elect-lc-timer:%"
PRId64
", hb:%"
PRId64
", %s, %s, %s, %s"
,
", chging:%d, restore:%d, quorum:%d, elect-lc-timer:%"
PRId64
", hb:%"
PRId64
", buffer:%s, repl-mgrs:%s, members:%s, hb:%s, hb-reply:%s"
,
pNode
->
vgId
,
eventLog
,
syncStr
(
pNode
->
state
),
currentTerm
,
pNode
->
commitIndex
,
logBeginIndex
,
logLastIndex
,
pNode
->
minMatchIndex
,
snapshot
.
lastApplyIndex
,
snapshot
.
lastApplyTerm
,
pNode
->
electNum
,
pNode
->
becomeLeaderNum
,
pNode
->
configChangeNum
,
cacheHit
,
cacheMiss
,
pNode
->
hbSlowNum
,
pNode
->
hbrSlowNum
,
aqItems
,
pNode
->
snapshottingIndex
,
pNode
->
replicaNum
,
pNode
->
raftCfg
.
lastConfigIndex
,
pNode
->
changing
,
pNode
->
restoreFinish
,
syncNodeDynamicQuorum
(
pNode
),
pNode
->
electTimerLogicClock
,
pNode
->
heartbeatTimerLogicClockUser
,
peerStr
,
cfgStr
,
hbTimeStr
,
hbrTimeStr
);
pNode
->
becomeLeaderNum
,
pNode
->
configChangeNum
,
pNode
->
hbSlowNum
,
pNode
->
hbrSlowNum
,
aqItems
,
pNode
->
snapshottingIndex
,
pNode
->
replicaNum
,
pNode
->
raftCfg
.
lastConfigIndex
,
pNode
->
changing
,
pNode
->
restoreFinish
,
syncNodeDynamicQuorum
(
pNode
),
pNode
->
electTimerLogicClock
,
pNode
->
heartbeatTimerLogicClockUser
,
bufferStatesStr
,
replMgrStatesStr
,
cfgStr
,
hbTimeStr
,
hbrTimeStr
);
}
}
...
...
@@ -216,7 +242,7 @@ void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dfla
logBeginIndex
=
pNode
->
pLogStore
->
syncLogBeginIndex
(
pNode
->
pLogStore
);
}
char
cfgStr
[
1024
];
char
cfgStr
[
1024
]
=
""
;
syncCfg2SimpleStr
(
&
pNode
->
raftCfg
.
cfg
,
cfgStr
,
sizeof
(
cfgStr
));
char
peerStr
[
1024
]
=
"{"
;
...
...
@@ -262,7 +288,7 @@ void syncPrintSnapshotReceiverLog(const char* flags, ELogLevel level, int32_t df
logBeginIndex
=
pNode
->
pLogStore
->
syncLogBeginIndex
(
pNode
->
pLogStore
);
}
char
cfgStr
[
1024
];
char
cfgStr
[
1024
]
=
""
;
syncCfg2SimpleStr
(
&
pNode
->
raftCfg
.
cfg
,
cfgStr
,
sizeof
(
cfgStr
));
char
peerStr
[
1024
]
=
"{"
;
...
...
@@ -304,7 +330,7 @@ void syncLogRecvTimer(SSyncNode* pSyncNode, const SyncTimeout* pMsg, const char*
void
syncLogRecvLocalCmd
(
SSyncNode
*
pSyncNode
,
const
SyncLocalCmd
*
pMsg
,
const
char
*
s
)
{
sNTrace
(
pSyncNode
,
"recv sync-local-cmd {cmd:%d-%s, sd-new-term:%"
PRId64
", fc-index:%"
PRId64
"}, %s"
,
pMsg
->
cmd
,
syncLocalCmdGetStr
(
pMsg
->
cmd
),
pMsg
->
sdNewTerm
,
pMsg
->
fc
Index
,
s
);
syncLocalCmdGetStr
(
pMsg
->
cmd
),
pMsg
->
currentTerm
,
pMsg
->
commit
Index
,
s
);
}
void
syncLogSendAppendEntriesReply
(
SSyncNode
*
pSyncNode
,
const
SyncAppendEntriesReply
*
pMsg
,
const
char
*
s
)
{
...
...
source/libs/tdb/inc/tdb.h
浏览文件 @
398161eb
...
...
@@ -74,7 +74,12 @@ int32_t tdbTbcUpsert(TBC *pTbc, const void *pKey, int nKey, const void *pData, i
int32_t
tdbTxnOpen
(
TXN
*
pTxn
,
int64_t
txnid
,
void
*
(
*
xMalloc
)(
void
*
,
size_t
),
void
(
*
xFree
)(
void
*
,
void
*
),
void
*
xArg
,
int
flags
);
int32_t
tdbTxnClose
(
TXN
*
pTxn
);
int32_t
tdbTxnCloseImpl
(
TXN
*
pTxn
);
#define tdbTxnClose(pTxn) \
do { \
tdbTxnCloseImpl(pTxn); \
(pTxn) = NULL; \
} while (0)
// other
void
tdbFree
(
void
*
);
...
...
source/libs/tdb/src/db/tdbTxn.c
浏览文件 @
398161eb
...
...
@@ -31,13 +31,18 @@ int tdbTxnOpen(TXN *pTxn, int64_t txnid, void *(*xMalloc)(void *, size_t), void
return
0
;
}
int
tdbTxnClose
(
TXN
*
pTxn
)
{
int
tdbTxnClose
Impl
(
TXN
*
pTxn
)
{
if
(
pTxn
)
{
if
(
pTxn
->
jPageSet
)
{
hashset_destroy
(
pTxn
->
jPageSet
);
pTxn
->
jPageSet
=
NULL
;
}
if
(
pTxn
->
jfd
)
{
tdbOsClose
(
pTxn
->
jfd
);
ASSERT
(
pTxn
->
jfd
==
NULL
);
}
tdbOsFree
(
pTxn
);
}
...
...
source/libs/transport/src/transCli.c
浏览文件 @
398161eb
...
...
@@ -1126,7 +1126,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
int
ret
=
uv_tcp_connect
(
&
conn
->
connReq
,
(
uv_tcp_t
*
)(
conn
->
stream
),
(
const
struct
sockaddr
*
)
&
addr
,
cliConnCb
);
if
(
ret
!=
0
)
{
tG
Trace
(
"%s conn %p failed to connect to %s:%d, reason:%s"
,
pTransInst
->
label
,
conn
,
conn
->
ip
,
conn
->
port
,
tG
Error
(
"%s conn %p failed to connect to %s:%d, reason:%s"
,
pTransInst
->
label
,
conn
,
conn
->
ip
,
conn
->
port
,
uv_err_name
(
ret
));
uv_timer_stop
(
conn
->
timer
);
...
...
source/libs/wal/src/walMeta.c
浏览文件 @
398161eb
...
...
@@ -289,19 +289,10 @@ void walAlignVersions(SWal* pWal) {
}
pWal
->
vers
.
lastVer
=
pWal
->
vers
.
snapshotVer
;
}
if
(
pWal
->
vers
.
commitVer
<
pWal
->
vers
.
snapshotVer
)
{
wWarn
(
"vgId:%d, commitVer:%"
PRId64
" is less than snapshotVer:%"
PRId64
". align with it."
,
pWal
->
cfg
.
vgId
,
pWal
->
vers
.
commitVer
,
pWal
->
vers
.
snapshotVer
);
pWal
->
vers
.
commitVer
=
pWal
->
vers
.
snapshotVer
;
}
if
(
pWal
->
vers
.
appliedVer
<
pWal
->
vers
.
snapshotVer
)
{
wWarn
(
"vgId:%d, appliedVer:%"
PRId64
" is less than snapshotVer:%"
PRId64
". align with it."
,
pWal
->
cfg
.
vgId
,
pWal
->
vers
.
appliedVer
,
pWal
->
vers
.
snapshotVer
);
pWal
->
vers
.
appliedVer
=
pWal
->
vers
.
snapshotVer
;
}
pWal
->
vers
.
commitVer
=
TMIN
(
pWal
->
vers
.
lastVer
,
pWal
->
vers
.
commitVer
);
pWal
->
vers
.
appliedVer
=
TMIN
(
pWal
->
vers
.
commitVer
,
pWal
->
vers
.
appliedVer
);
// reset commitVer and appliedVer
pWal
->
vers
.
commitVer
=
pWal
->
vers
.
snapshotVer
;
pWal
->
vers
.
appliedVer
=
pWal
->
vers
.
snapshotVer
;
wInfo
(
"vgId:%d, reset commitVer to %"
PRId64
,
pWal
->
cfg
.
vgId
,
pWal
->
vers
.
commitVer
);
}
bool
walLogEntriesComplete
(
const
SWal
*
pWal
)
{
...
...
source/libs/wal/src/walWrite.c
浏览文件 @
398161eb
...
...
@@ -103,7 +103,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
wInfo
(
"vgId:%d, wal rollback for version %"
PRId64
,
pWal
->
cfg
.
vgId
,
ver
);
int64_t
code
;
char
fnameStr
[
WAL_FILE_LEN
];
if
(
ver
>
pWal
->
vers
.
lastVer
||
ver
<
pWal
->
vers
.
commitVer
||
ver
<=
pWal
->
vers
.
snapshotVer
)
{
if
(
ver
>
pWal
->
vers
.
lastVer
||
ver
<
=
pWal
->
vers
.
commitVer
||
ver
<=
pWal
->
vers
.
snapshotVer
)
{
terrno
=
TSDB_CODE_WAL_INVALID_VER
;
taosThreadMutexUnlock
(
&
pWal
->
mutex
);
return
-
1
;
...
...
tests/script/tsim/parser/regressiontest.sim
0 → 100644
浏览文件 @
398161eb
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sql connect
$dbPrefix = reg_db
$tb = tb
$rowNum = 8200
$ts0 = 1537146000000
$delta = 100
print ========== reg.sim
$i = 0
$db = $dbPrefix . $i
sql drop database if exists $db -x step1
step1:
sql create database $db vgroups 1;
sql use $db
sql create table $tb (ts timestamp, c1 int)
$i = 0
$ts = $ts0
$x = 0
while $x < $rowNum
$xs = $x * $delta
$ts = $ts0 + $xs
sql insert into $tb values ( $ts , $x )
$x = $x + 1
endw
print ================== restart server to commit data into disk
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode1 -s start
print ================== server restart completed
sql connect
sql use $db
sql delete from $tb where ts=1537146000000
sql delete from $tb where ts=1537146409500
print =========================> TS-2410
sql select * from $tb limit 20 offset 4090
print $data00
print $data10
print $data20
print $data30
print $data40
print $data50
print $data60
print $data70
print $data80
print $data90
if $data40 != @18-09-17 09:06:49.600@ then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
tests/script/tsim/sma/rsmaCreateInsertQuery.sim
浏览文件 @
398161eb
...
...
@@ -82,8 +82,8 @@ endi
#===================================================================
#==================== reboot to trigger commit data to file
sql flush database d0;
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode1 -s start
...
...
tests/script/tsim/sma/rsmaPersistenceRecovery.sim
浏览文件 @
398161eb
...
...
@@ -85,6 +85,7 @@ endi
#==================== reboot to trigger commit data to file
sql flush database d0;
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode1 -s start
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录