Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
6437d20c
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看板
提交
6437d20c
编写于
7月 04, 2022
作者:
M
Minghao Li
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor(sync): add resp syncRespCleanByTTL
上级
eae44cd9
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
95 addition
and
6 deletion
+95
-6
source/libs/sync/src/syncRespMgr.c
source/libs/sync/src/syncRespMgr.c
+57
-1
source/libs/sync/test/syncIndexTest.cpp
source/libs/sync/test/syncIndexTest.cpp
+7
-1
source/libs/sync/test/syncRespMgrTest.cpp
source/libs/sync/test/syncRespMgrTest.cpp
+31
-4
未找到文件。
source/libs/sync/src/syncRespMgr.c
浏览文件 @
6437d20c
...
...
@@ -14,6 +14,7 @@
*/
#include "syncRespMgr.h"
#include "syncRaftEntry.h"
#include "syncRaftStore.h"
SSyncRespMgr
*
syncRespMgrCreate
(
void
*
data
,
int64_t
ttl
)
{
...
...
@@ -116,4 +117,59 @@ void syncRespClean(SSyncRespMgr *pObj) {
taosThreadMutexUnlock
(
&
(
pObj
->
mutex
));
}
void
syncRespCleanByTTL
(
SSyncRespMgr
*
pObj
,
int64_t
ttl
)
{}
void
syncRespCleanByTTL
(
SSyncRespMgr
*
pObj
,
int64_t
ttl
)
{
SRespStub
*
pStub
=
(
SRespStub
*
)
taosHashIterate
(
pObj
->
pRespHash
,
NULL
);
int
cnt
=
0
;
SSyncNode
*
pSyncNode
=
pObj
->
data
;
SArray
*
delIndexArray
=
taosArrayInit
(
0
,
sizeof
(
SyncIndex
));
ASSERT
(
delIndexArray
!=
NULL
);
while
(
pStub
)
{
size_t
len
;
void
*
key
=
taosHashGetKey
(
pStub
,
&
len
);
SyncIndex
*
pIndex
=
(
SyncIndex
*
)
key
;
int64_t
nowMS
=
taosGetTimestampMs
();
if
(
nowMS
-
pStub
->
createTime
>
ttl
)
{
taosArrayPush
(
delIndexArray
,
pIndex
);
cnt
++
;
SSyncRaftEntry
*
pEntry
=
NULL
;
int32_t
code
=
0
;
if
(
pSyncNode
->
pLogStore
!=
NULL
)
{
code
=
pSyncNode
->
pLogStore
->
syncLogGetEntry
(
pSyncNode
->
pLogStore
,
*
pIndex
,
&
pEntry
);
if
(
code
==
0
&&
pEntry
!=
NULL
)
{
SFsmCbMeta
cbMeta
=
{
0
};
cbMeta
.
index
=
pEntry
->
index
;
cbMeta
.
lastConfigIndex
=
syncNodeGetSnapshotConfigIndex
(
pSyncNode
,
cbMeta
.
index
);
cbMeta
.
isWeak
=
pEntry
->
isWeak
;
cbMeta
.
code
=
TSDB_CODE_SYN_TIMEOUT
;
cbMeta
.
state
=
pSyncNode
->
state
;
cbMeta
.
seqNum
=
pEntry
->
seqNum
;
cbMeta
.
term
=
pEntry
->
term
;
cbMeta
.
currentTerm
=
pSyncNode
->
pRaftStore
->
currentTerm
;
cbMeta
.
flag
=
0
;
SRpcMsg
rpcMsg
=
pStub
->
rpcMsg
;
rpcMsg
.
pCont
=
rpcMallocCont
(
pEntry
->
dataLen
);
memcpy
(
rpcMsg
.
pCont
,
pEntry
->
data
,
pEntry
->
dataLen
);
pSyncNode
->
pFsm
->
FpCommitCb
(
pSyncNode
->
pFsm
,
&
rpcMsg
,
cbMeta
);
syncEntryDestory
(
pEntry
);
}
}
}
pStub
=
(
SRespStub
*
)
taosHashIterate
(
pObj
->
pRespHash
,
pStub
);
}
int32_t
arraySize
=
taosArrayGetSize
(
delIndexArray
);
sDebug
(
"vgId:%d, resp clean by ttl, cnt:%d, array-size:%d"
,
pSyncNode
->
vgId
,
cnt
,
arraySize
);
for
(
int32_t
i
=
0
;
i
<
arraySize
;
++
i
)
{
SyncIndex
*
pIndex
=
taosArrayGet
(
delIndexArray
,
i
);
taosHashRemove
(
pObj
->
pRespHash
,
pIndex
,
sizeof
(
SyncIndex
));
}
taosArrayDestroy
(
delIndexArray
);
}
source/libs/sync/test/syncIndexTest.cpp
浏览文件 @
6437d20c
...
...
@@ -8,7 +8,13 @@ void print(SHashObj *pNextIndex) {
printf
(
"----------------
\n
"
);
uint64_t
*
p
=
(
uint64_t
*
)
taosHashIterate
(
pNextIndex
,
NULL
);
while
(
p
)
{
printf
(
"%lu
\n
"
,
*
p
);
size_t
len
;
void
*
key
=
taosHashGetKey
(
p
,
&
len
);
SRaftId
*
pRaftId
=
(
SRaftId
*
)
key
;
printf
(
"key:<%lu, %d>, value:%lu
\n
"
,
pRaftId
->
addr
,
pRaftId
->
vgId
,
*
p
);
p
=
(
uint64_t
*
)
taosHashIterate
(
pNextIndex
,
p
);
}
}
...
...
source/libs/sync/test/syncRespMgrTest.cpp
浏览文件 @
6437d20c
...
...
@@ -73,9 +73,15 @@ void syncRespMgrGetAndDelTest(uint64_t i) {
}
}
SSyncNode
*
createSyncNode
()
{
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
taosMemoryMalloc
(
sizeof
(
SSyncNode
));
memset
(
pSyncNode
,
0
,
sizeof
(
SSyncNode
));
return
pSyncNode
;
}
void
test1
()
{
printf
(
"------- test1 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
NULL
,
0
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
()
,
0
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
10
);
...
...
@@ -100,7 +106,7 @@ void test1() {
void
test2
()
{
printf
(
"------- test2 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
NULL
,
0
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
()
,
0
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
10
);
...
...
@@ -117,7 +123,7 @@ void test2() {
void
test3
()
{
printf
(
"------- test3 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
NULL
,
0
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
()
,
0
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
10
);
...
...
@@ -132,13 +138,34 @@ void test3() {
syncRespMgrDestroy
(
pMgr
);
}
void
test4
()
{
printf
(
"------- test4 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
(),
2
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
5
);
syncRespMgrPrint
();
taosMsleep
(
3000
);
syncRespMgrInsert
(
3
);
syncRespMgrPrint
();
printf
(
"====== after clean ttl
\n
"
);
syncRespClean
(
pMgr
);
syncRespMgrPrint
();
syncRespMgrDestroy
(
pMgr
);
}
int
main
()
{
tsAsyncLog
=
0
;
sDebugFlag
=
DEBUG_TRACE
+
DEBUG_SCREEN
+
DEBUG_FILE
;
sDebugFlag
=
DEBUG_
DEBUG
+
DEBUG_
TRACE
+
DEBUG_SCREEN
+
DEBUG_FILE
;
logTest
();
test1
();
test2
();
test3
();
test4
();
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录