Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
6da0b736
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1181
Star
22014
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看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
6da0b736
编写于
11月 13, 2022
作者:
S
Shengliang Guan
提交者:
GitHub
11月 13, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #18095 from taosdata/fix/TD-20052
refact: remove sync pre snapshot
上级
1c21f2d0
7bcda46c
变更
8
隐藏空白更改
内联
并排
Showing
8 changed file
with
422 addition
and
391 deletion
+422
-391
include/common/tmsgdef.h
include/common/tmsgdef.h
+2
-2
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
+0
-1
source/libs/sync/inc/syncMessage.h
source/libs/sync/inc/syncMessage.h
+2
-44
source/libs/sync/src/syncMessage.c
source/libs/sync/src/syncMessage.c
+24
-287
source/libs/sync/src/syncSnapshot.c
source/libs/sync/src/syncSnapshot.c
+0
-57
source/libs/sync/test/sync_test_lib/inc/syncTest.h
source/libs/sync/test/sync_test_lib/inc/syncTest.h
+40
-0
source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c
source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c
+297
-0
source/libs/sync/test/sync_test_lib/src/syncSnapshotDebug.c
source/libs/sync/test/sync_test_lib/src/syncSnapshotDebug.c
+57
-0
未找到文件。
include/common/tmsgdef.h
浏览文件 @
6da0b736
...
...
@@ -274,8 +274,8 @@ enum {
TD_DEF_MSG_TYPE
(
TDMT_SYNC_HEARTBEAT
,
"sync-heartbeat"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_SYNC_HEARTBEAT_REPLY
,
"sync-heartbeat-reply"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_SYNC_LOCAL_CMD
,
"sync-local-cmd"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_SYNC_PRE_SNAPSHOT
,
"sync-pre-snapshot"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_SYNC_PRE_SNAPSHOT_REPLY
,
"sync-pre-snapshot-reply"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_SYNC_PRE_SNAPSHOT
,
"sync-pre-snapshot"
,
NULL
,
NULL
)
// no longer used
TD_DEF_MSG_TYPE
(
TDMT_SYNC_PRE_SNAPSHOT_REPLY
,
"sync-pre-snapshot-reply"
,
NULL
,
NULL
)
// no longer used
TD_DEF_MSG_TYPE
(
TDMT_SYNC_MAX_MSG
,
"sync-max"
,
NULL
,
NULL
)
TD_NEW_MSG_SEG
(
TDMT_VND_STREAM_MSG
)
...
...
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
浏览文件 @
6da0b736
...
...
@@ -195,7 +195,6 @@ SArray *mmGetMsgHandles() {
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_SYNC_SNAPSHOT_RSP
,
mmPutMsgToSyncQueue
,
1
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_SYNC_PRE_SNAPSHOT
,
mmPutMsgToSyncQueue
,
1
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_SYNC_PRE_SNAPSHOT_REPLY
,
mmPutMsgToSyncQueue
,
1
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_SYNC_HEARTBEAT
,
mmPutMsgToSyncCtrlQueue
,
1
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_SYNC_HEARTBEAT_REPLY
,
mmPutMsgToSyncCtrlQueue
,
1
)
==
NULL
)
goto
_OVER
;
...
...
source/libs/sync/inc/syncMessage.h
浏览文件 @
6da0b736
...
...
@@ -112,7 +112,6 @@ typedef struct SyncAppendEntriesReply {
int64_t
startTime
;
}
SyncAppendEntriesReply
;
// ---------------------------------------------
typedef
struct
SyncHeartbeat
{
uint32_t
bytes
;
int32_t
vgId
;
...
...
@@ -149,28 +148,8 @@ typedef struct SyncPreSnapshot {
// private data
SyncTerm
term
;
}
SyncPreSnapshot
;
SyncPreSnapshot
*
syncPreSnapshotBuild
(
int32_t
vgId
);
void
syncPreSnapshotDestroy
(
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotSerialize
(
const
SyncPreSnapshot
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
);
void
syncPreSnapshotDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshot
*
pMsg
);
char
*
syncPreSnapshotSerialize2
(
const
SyncPreSnapshot
*
pMsg
,
uint32_t
*
len
);
SyncPreSnapshot
*
syncPreSnapshotDeserialize2
(
const
char
*
buf
,
uint32_t
len
);
void
syncPreSnapshot2RpcMsg
(
const
SyncPreSnapshot
*
pMsg
,
SRpcMsg
*
pRpcMsg
);
void
syncPreSnapshotFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshot
*
pMsg
);
SyncPreSnapshot
*
syncPreSnapshotFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
);
cJSON
*
syncPreSnapshot2Json
(
const
SyncPreSnapshot
*
pMsg
);
char
*
syncPreSnapshot2Str
(
const
SyncPreSnapshot
*
pMsg
);
// for debug ----------------------
void
syncPreSnapshotPrint
(
const
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotPrint2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotLog
(
const
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotLog2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
);
// ---------------------------------------------
typedef
struct
SyncPreSnapshotReply
{
uint32_t
bytes
;
int32_t
vgId
;
...
...
@@ -181,28 +160,8 @@ typedef struct SyncPreSnapshotReply {
// private data
SyncTerm
term
;
SyncIndex
snapStart
;
}
SyncPreSnapshotReply
;
SyncPreSnapshotReply
*
syncPreSnapshotReplyBuild
(
int32_t
vgId
);
void
syncPreSnapshotReplyDestroy
(
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplySerialize
(
const
SyncPreSnapshotReply
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
);
void
syncPreSnapshotReplyDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshotReply
*
pMsg
);
char
*
syncPreSnapshotReplySerialize2
(
const
SyncPreSnapshotReply
*
pMsg
,
uint32_t
*
len
);
SyncPreSnapshotReply
*
syncPreSnapshotReplyDeserialize2
(
const
char
*
buf
,
uint32_t
len
);
void
syncPreSnapshotReply2RpcMsg
(
const
SyncPreSnapshotReply
*
pMsg
,
SRpcMsg
*
pRpcMsg
);
void
syncPreSnapshotReplyFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshotReply
*
pMsg
);
SyncPreSnapshotReply
*
syncPreSnapshotReplyFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
);
cJSON
*
syncPreSnapshotReply2Json
(
const
SyncPreSnapshotReply
*
pMsg
);
char
*
syncPreSnapshotReply2Str
(
const
SyncPreSnapshotReply
*
pMsg
);
// for debug ----------------------
void
syncPreSnapshotReplyPrint
(
const
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplyPrint2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplyLog
(
const
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplyLog2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
);
// ---------------------------------------------
typedef
struct
SyncApplyMsg
{
uint32_t
bytes
;
int32_t
vgId
;
...
...
@@ -375,9 +334,6 @@ int32_t syncNodeOnRequestVoteReply(SSyncNode* pNode, const SRpcMsg* pMsg);
int32_t
syncNodeOnAppendEntries
(
SSyncNode
*
pNode
,
const
SRpcMsg
*
pMsg
);
int32_t
syncNodeOnAppendEntriesReply
(
SSyncNode
*
ths
,
const
SRpcMsg
*
pMsg
);
int32_t
syncNodeOnPreSnapshot
(
SSyncNode
*
ths
,
SyncPreSnapshot
*
pMsg
);
int32_t
syncNodeOnPreSnapshotReply
(
SSyncNode
*
ths
,
SyncPreSnapshotReply
*
pMsg
);
int32_t
syncNodeOnSnapshot
(
SSyncNode
*
ths
,
SyncSnapshotSend
*
pMsg
);
int32_t
syncNodeOnSnapshotReply
(
SSyncNode
*
ths
,
SyncSnapshotRsp
*
pMsg
);
...
...
@@ -404,6 +360,8 @@ int32_t syncBuildAppendEntries(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId);
int32_t
syncBuildAppendEntriesReply
(
SRpcMsg
*
pMsg
,
int32_t
vgId
);
int32_t
syncBuildHeartbeat
(
SRpcMsg
*
pMsg
,
int32_t
vgId
);
int32_t
syncBuildHeartbeatReply
(
SRpcMsg
*
pMsg
,
int32_t
vgId
);
int32_t
syncBuildPreSnapshot
(
SRpcMsg
*
pMsg
,
int32_t
vgId
);
int32_t
syncBuildPreSnapshotReply
(
SRpcMsg
*
pMsg
,
int32_t
vgId
);
#ifdef __cplusplus
}
...
...
source/libs/sync/src/syncMessage.c
浏览文件 @
6da0b736
...
...
@@ -189,301 +189,38 @@ int32_t syncBuildHeartbeatReply(SRpcMsg* pMsg, int32_t vgId) {
return
0
;
}
// ---- message process SyncPreSnapshot----
SyncPreSnapshot
*
syncPreSnapshotBuild
(
int32_t
vgId
)
{
uint32_t
bytes
=
sizeof
(
SyncPreSnapshot
);
SyncPreSnapshot
*
pMsg
=
taosMemoryMalloc
(
bytes
);
memset
(
pMsg
,
0
,
bytes
);
pMsg
->
bytes
=
bytes
;
pMsg
->
vgId
=
vgId
;
int32_t
syncBuildPreSnapshot
(
SRpcMsg
*
pMsg
,
int32_t
vgId
)
{
int32_t
bytes
=
sizeof
(
SyncPreSnapshot
);
pMsg
->
pCont
=
rpcMallocCont
(
bytes
);
pMsg
->
msgType
=
TDMT_SYNC_PRE_SNAPSHOT
;
return
pMsg
;
}
void
syncPreSnapshotDestroy
(
SyncPreSnapshot
*
pMsg
)
{
if
(
pMsg
!=
NULL
)
{
taosMemoryFree
(
pMsg
);
}
}
void
syncPreSnapshotSerialize
(
const
SyncPreSnapshot
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
)
{
ASSERT
(
pMsg
->
bytes
<=
bufLen
);
memcpy
(
buf
,
pMsg
,
pMsg
->
bytes
);
}
void
syncPreSnapshotDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshot
*
pMsg
)
{
memcpy
(
pMsg
,
buf
,
len
);
ASSERT
(
len
==
pMsg
->
bytes
);
}
char
*
syncPreSnapshotSerialize2
(
const
SyncPreSnapshot
*
pMsg
,
uint32_t
*
len
)
{
char
*
buf
=
taosMemoryMalloc
(
pMsg
->
bytes
);
ASSERT
(
buf
!=
NULL
);
syncPreSnapshotSerialize
(
pMsg
,
buf
,
pMsg
->
bytes
);
if
(
len
!=
NULL
)
{
*
len
=
pMsg
->
bytes
;
}
return
buf
;
}
SyncPreSnapshot
*
syncPreSnapshotDeserialize2
(
const
char
*
buf
,
uint32_t
len
)
{
uint32_t
bytes
=
*
((
uint32_t
*
)
buf
);
SyncPreSnapshot
*
pMsg
=
taosMemoryMalloc
(
bytes
);
ASSERT
(
pMsg
!=
NULL
);
syncPreSnapshotDeserialize
(
buf
,
len
,
pMsg
);
ASSERT
(
len
==
pMsg
->
bytes
);
return
pMsg
;
}
void
syncPreSnapshot2RpcMsg
(
const
SyncPreSnapshot
*
pMsg
,
SRpcMsg
*
pRpcMsg
)
{
memset
(
pRpcMsg
,
0
,
sizeof
(
*
pRpcMsg
));
pRpcMsg
->
msgType
=
pMsg
->
msgType
;
pRpcMsg
->
contLen
=
pMsg
->
bytes
;
pRpcMsg
->
pCont
=
rpcMallocCont
(
pRpcMsg
->
contLen
);
syncPreSnapshotSerialize
(
pMsg
,
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
}
void
syncPreSnapshotFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshot
*
pMsg
)
{
syncPreSnapshotDeserialize
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
,
pMsg
);
}
SyncPreSnapshot
*
syncPreSnapshotFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
)
{
SyncPreSnapshot
*
pMsg
=
syncPreSnapshotDeserialize2
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
ASSERT
(
pMsg
!=
NULL
);
return
pMsg
;
}
cJSON
*
syncPreSnapshot2Json
(
const
SyncPreSnapshot
*
pMsg
)
{
char
u64buf
[
128
]
=
{
0
};
cJSON
*
pRoot
=
cJSON_CreateObject
();
if
(
pMsg
!=
NULL
)
{
cJSON_AddNumberToObject
(
pRoot
,
"bytes"
,
pMsg
->
bytes
);
cJSON_AddNumberToObject
(
pRoot
,
"vgId"
,
pMsg
->
vgId
);
cJSON_AddNumberToObject
(
pRoot
,
"msgType"
,
pMsg
->
msgType
);
cJSON
*
pSrcId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
srcId
.
addr
);
cJSON_AddStringToObject
(
pSrcId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
srcId
.
addr
;
cJSON
*
pTmp
=
pSrcId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pSrcId
,
"vgId"
,
pMsg
->
srcId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"srcId"
,
pSrcId
);
cJSON
*
pDestId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
destId
.
addr
);
cJSON_AddStringToObject
(
pDestId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
destId
.
addr
;
cJSON
*
pTmp
=
pDestId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pDestId
,
"vgId"
,
pMsg
->
destId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"destId"
,
pDestId
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
pMsg
->
contLen
=
bytes
;
if
(
pMsg
->
pCont
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
cJSON
*
pJson
=
cJSON_CreateObject
();
cJSON_AddItemToObject
(
pJson
,
"SyncPreSnapshot"
,
pRoot
);
return
pJson
;
}
char
*
syncPreSnapshot2Str
(
const
SyncPreSnapshot
*
pMsg
)
{
cJSON
*
pJson
=
syncPreSnapshot2Json
(
pMsg
);
char
*
serialized
=
cJSON_Print
(
pJson
);
cJSON_Delete
(
pJson
);
return
serialized
;
}
void
syncPreSnapshotPrint
(
const
SyncPreSnapshot
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
printf
(
"syncPreSnapshotPrint | len:%d | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotPrint2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
printf
(
"syncPreSnapshotPrint2 | len:%d | %s | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotLog
(
const
SyncPreSnapshot
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotLog | len:%d | %s"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotLog2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
)
{
if
(
gRaftDetailLog
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotLog2 | len:%d | %s | %s"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
taosMemoryFree
(
serialized
);
}
SyncPreSnapshot
*
pPreSnapshot
=
pMsg
->
pCont
;
pPreSnapshot
->
bytes
=
bytes
;
pPreSnapshot
->
msgType
=
TDMT_SYNC_PRE_SNAPSHOT
;
pPreSnapshot
->
vgId
=
vgId
;
return
0
;
}
// ---- message process SyncPreSnapshotReply----
SyncPreSnapshotReply
*
syncPreSnapshotReplyBuild
(
int32_t
vgId
)
{
uint32_t
bytes
=
sizeof
(
SyncPreSnapshotReply
);
SyncPreSnapshotReply
*
pMsg
=
taosMemoryMalloc
(
bytes
);
memset
(
pMsg
,
0
,
bytes
);
pMsg
->
bytes
=
bytes
;
pMsg
->
vgId
=
vgId
;
int32_t
syncBuildPreSnapshotReply
(
SRpcMsg
*
pMsg
,
int32_t
vgId
)
{
int32_t
bytes
=
sizeof
(
SyncPreSnapshotReply
);
pMsg
->
pCont
=
rpcMallocCont
(
bytes
);
pMsg
->
msgType
=
TDMT_SYNC_PRE_SNAPSHOT_REPLY
;
return
pMsg
;
}
void
syncPreSnapshotReplyDestroy
(
SyncPreSnapshotReply
*
pMsg
)
{
if
(
pMsg
!=
NULL
)
{
taosMemoryFree
(
pMsg
);
}
}
void
syncPreSnapshotReplySerialize
(
const
SyncPreSnapshotReply
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
)
{
ASSERT
(
pMsg
->
bytes
<=
bufLen
);
memcpy
(
buf
,
pMsg
,
pMsg
->
bytes
);
}
void
syncPreSnapshotReplyDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshotReply
*
pMsg
)
{
memcpy
(
pMsg
,
buf
,
len
);
ASSERT
(
len
==
pMsg
->
bytes
);
}
char
*
syncPreSnapshotReplySerialize2
(
const
SyncPreSnapshotReply
*
pMsg
,
uint32_t
*
len
)
{
char
*
buf
=
taosMemoryMalloc
(
pMsg
->
bytes
);
ASSERT
(
buf
!=
NULL
);
syncPreSnapshotReplySerialize
(
pMsg
,
buf
,
pMsg
->
bytes
);
if
(
len
!=
NULL
)
{
*
len
=
pMsg
->
bytes
;
}
return
buf
;
}
SyncPreSnapshotReply
*
syncPreSnapshotReplyDeserialize2
(
const
char
*
buf
,
uint32_t
len
)
{
uint32_t
bytes
=
*
((
uint32_t
*
)
buf
);
SyncPreSnapshotReply
*
pMsg
=
taosMemoryMalloc
(
bytes
);
ASSERT
(
pMsg
!=
NULL
);
syncPreSnapshotReplyDeserialize
(
buf
,
len
,
pMsg
);
ASSERT
(
len
==
pMsg
->
bytes
);
return
pMsg
;
}
void
syncPreSnapshotReply2RpcMsg
(
const
SyncPreSnapshotReply
*
pMsg
,
SRpcMsg
*
pRpcMsg
)
{
memset
(
pRpcMsg
,
0
,
sizeof
(
*
pRpcMsg
));
pRpcMsg
->
msgType
=
pMsg
->
msgType
;
pRpcMsg
->
contLen
=
pMsg
->
bytes
;
pRpcMsg
->
pCont
=
rpcMallocCont
(
pRpcMsg
->
contLen
);
syncPreSnapshotReplySerialize
(
pMsg
,
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
}
void
syncPreSnapshotReplyFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshotReply
*
pMsg
)
{
syncPreSnapshotReplyDeserialize
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
,
pMsg
);
}
SyncPreSnapshotReply
*
syncPreSnapshotReplyFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
)
{
SyncPreSnapshotReply
*
pMsg
=
syncPreSnapshotReplyDeserialize2
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
ASSERT
(
pMsg
!=
NULL
);
return
pMsg
;
}
cJSON
*
syncPreSnapshotReply2Json
(
const
SyncPreSnapshotReply
*
pMsg
)
{
char
u64buf
[
128
]
=
{
0
};
cJSON
*
pRoot
=
cJSON_CreateObject
();
if
(
pMsg
!=
NULL
)
{
cJSON_AddNumberToObject
(
pRoot
,
"bytes"
,
pMsg
->
bytes
);
cJSON_AddNumberToObject
(
pRoot
,
"vgId"
,
pMsg
->
vgId
);
cJSON_AddNumberToObject
(
pRoot
,
"msgType"
,
pMsg
->
msgType
);
cJSON
*
pSrcId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
srcId
.
addr
);
cJSON_AddStringToObject
(
pSrcId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
srcId
.
addr
;
cJSON
*
pTmp
=
pSrcId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pSrcId
,
"vgId"
,
pMsg
->
srcId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"srcId"
,
pSrcId
);
cJSON
*
pDestId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
destId
.
addr
);
cJSON_AddStringToObject
(
pDestId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
destId
.
addr
;
cJSON
*
pTmp
=
pDestId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pDestId
,
"vgId"
,
pMsg
->
destId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"destId"
,
pDestId
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRId64
,
pMsg
->
snapStart
);
cJSON_AddStringToObject
(
pRoot
,
"snap-start"
,
u64buf
);
pMsg
->
contLen
=
bytes
;
if
(
pMsg
->
pCont
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
cJSON
*
pJson
=
cJSON_CreateObject
();
cJSON_AddItemToObject
(
pJson
,
"SyncPreSnapshotReply"
,
pRoot
);
return
pJson
;
}
char
*
syncPreSnapshotReply2Str
(
const
SyncPreSnapshotReply
*
pMsg
)
{
cJSON
*
pJson
=
syncPreSnapshotReply2Json
(
pMsg
);
char
*
serialized
=
cJSON_Print
(
pJson
);
cJSON_Delete
(
pJson
);
return
serialized
;
}
void
syncPreSnapshotReplyPrint
(
const
SyncPreSnapshotReply
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
printf
(
"syncPreSnapshotReplyPrint | len:%d | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotReplyPrint2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
printf
(
"syncPreSnapshotReplyPrint2 | len:%d | %s | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotReplyLog
(
const
SyncPreSnapshotReply
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotReplyLog | len:%d | %s"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotReplyLog2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
)
{
if
(
gRaftDetailLog
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotReplyLog2 | len:%d | %s | %s"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
taosMemoryFree
(
serialized
);
}
SyncPreSnapshotReply
*
pPreSnapshotReply
=
pMsg
->
pCont
;
pPreSnapshotReply
->
bytes
=
bytes
;
pPreSnapshotReply
->
msgType
=
TDMT_SYNC_PRE_SNAPSHOT_REPLY
;
pPreSnapshotReply
->
vgId
=
vgId
;
return
0
;
}
// ---- message process SyncApplyMsg----
...
...
source/libs/sync/src/syncSnapshot.c
浏览文件 @
6da0b736
...
...
@@ -903,60 +903,3 @@ int32_t syncNodeOnSnapshotReply(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
return
0
;
}
int32_t
syncNodeOnPreSnapshot
(
SSyncNode
*
ths
,
SyncPreSnapshot
*
pMsg
)
{
syncLogRecvSyncPreSnapshot
(
ths
,
pMsg
,
""
);
SyncPreSnapshotReply
*
pMsgReply
=
syncPreSnapshotReplyBuild
(
ths
->
vgId
);
pMsgReply
->
srcId
=
ths
->
myRaftId
;
pMsgReply
->
destId
=
pMsg
->
srcId
;
pMsgReply
->
term
=
ths
->
pRaftStore
->
currentTerm
;
SSyncLogStoreData
*
pData
=
ths
->
pLogStore
->
data
;
SWal
*
pWal
=
pData
->
pWal
;
if
(
syncNodeIsMnode
(
ths
))
{
pMsgReply
->
snapStart
=
SYNC_INDEX_BEGIN
;
}
else
{
bool
isEmpty
=
ths
->
pLogStore
->
syncLogIsEmpty
(
ths
->
pLogStore
);
int64_t
walCommitVer
=
walGetCommittedVer
(
pWal
);
if
(
!
isEmpty
&&
ths
->
commitIndex
!=
walCommitVer
)
{
sNError
(
ths
,
"commit not same, wal-commit:%"
PRId64
", commit:%"
PRId64
", ignore"
,
walCommitVer
,
ths
->
commitIndex
);
goto
_IGNORE
;
}
pMsgReply
->
snapStart
=
ths
->
commitIndex
+
1
;
// make local log clean
int32_t
code
=
ths
->
pLogStore
->
syncLogTruncate
(
ths
->
pLogStore
,
pMsgReply
->
snapStart
);
if
(
code
!=
0
)
{
sNError
(
ths
,
"truncate wal error"
);
goto
_IGNORE
;
}
}
// can not write behind _RESPONSE
SRpcMsg
rpcMsg
;
_RESPONSE:
syncPreSnapshotReply2RpcMsg
(
pMsgReply
,
&
rpcMsg
);
syncNodeSendMsgById
(
&
pMsgReply
->
destId
,
ths
,
&
rpcMsg
);
syncPreSnapshotReplyDestroy
(
pMsgReply
);
return
0
;
_IGNORE:
syncPreSnapshotReplyDestroy
(
pMsgReply
);
return
0
;
}
int32_t
syncNodeOnPreSnapshotReply
(
SSyncNode
*
ths
,
SyncPreSnapshotReply
*
pMsg
)
{
syncLogRecvSyncPreSnapshotReply
(
ths
,
pMsg
,
""
);
// start snapshot
return
0
;
}
\ No newline at end of file
source/libs/sync/test/sync_test_lib/inc/syncTest.h
浏览文件 @
6da0b736
...
...
@@ -349,6 +349,46 @@ void syncHeartbeatReplyPrint2(char* s, const SyncHeartbeatReply* pMsg);
void
syncHeartbeatReplyLog
(
const
SyncHeartbeatReply
*
pMsg
);
void
syncHeartbeatReplyLog2
(
char
*
s
,
const
SyncHeartbeatReply
*
pMsg
);
SyncPreSnapshot
*
syncPreSnapshotBuild
(
int32_t
vgId
);
void
syncPreSnapshotDestroy
(
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotSerialize
(
const
SyncPreSnapshot
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
);
void
syncPreSnapshotDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshot
*
pMsg
);
char
*
syncPreSnapshotSerialize2
(
const
SyncPreSnapshot
*
pMsg
,
uint32_t
*
len
);
SyncPreSnapshot
*
syncPreSnapshotDeserialize2
(
const
char
*
buf
,
uint32_t
len
);
void
syncPreSnapshot2RpcMsg
(
const
SyncPreSnapshot
*
pMsg
,
SRpcMsg
*
pRpcMsg
);
void
syncPreSnapshotFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshot
*
pMsg
);
SyncPreSnapshot
*
syncPreSnapshotFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
);
cJSON
*
syncPreSnapshot2Json
(
const
SyncPreSnapshot
*
pMsg
);
char
*
syncPreSnapshot2Str
(
const
SyncPreSnapshot
*
pMsg
);
// for debug ----------------------
void
syncPreSnapshotPrint
(
const
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotPrint2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotLog
(
const
SyncPreSnapshot
*
pMsg
);
void
syncPreSnapshotLog2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
);
SyncPreSnapshotReply
*
syncPreSnapshotReplyBuild
(
int32_t
vgId
);
void
syncPreSnapshotReplyDestroy
(
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplySerialize
(
const
SyncPreSnapshotReply
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
);
void
syncPreSnapshotReplyDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshotReply
*
pMsg
);
char
*
syncPreSnapshotReplySerialize2
(
const
SyncPreSnapshotReply
*
pMsg
,
uint32_t
*
len
);
SyncPreSnapshotReply
*
syncPreSnapshotReplyDeserialize2
(
const
char
*
buf
,
uint32_t
len
);
void
syncPreSnapshotReply2RpcMsg
(
const
SyncPreSnapshotReply
*
pMsg
,
SRpcMsg
*
pRpcMsg
);
void
syncPreSnapshotReplyFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshotReply
*
pMsg
);
SyncPreSnapshotReply
*
syncPreSnapshotReplyFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
);
cJSON
*
syncPreSnapshotReply2Json
(
const
SyncPreSnapshotReply
*
pMsg
);
char
*
syncPreSnapshotReply2Str
(
const
SyncPreSnapshotReply
*
pMsg
);
// for debug ----------------------
void
syncPreSnapshotReplyPrint
(
const
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplyPrint2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplyLog
(
const
SyncPreSnapshotReply
*
pMsg
);
void
syncPreSnapshotReplyLog2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
);
// ---------------------------------------------
int32_t
syncNodeOnPreSnapshot
(
SSyncNode
*
ths
,
SyncPreSnapshot
*
pMsg
);
int32_t
syncNodeOnPreSnapshotReply
(
SSyncNode
*
ths
,
SyncPreSnapshotReply
*
pMsg
);
#ifdef __cplusplus
}
#endif
...
...
source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c
浏览文件 @
6da0b736
...
...
@@ -1942,3 +1942,300 @@ void syncHeartbeatReplyLog2(char* s, const SyncHeartbeatReply* pMsg) {
taosMemoryFree
(
serialized
);
}
}
// ---- message process SyncPreSnapshot----
SyncPreSnapshot
*
syncPreSnapshotBuild
(
int32_t
vgId
)
{
uint32_t
bytes
=
sizeof
(
SyncPreSnapshot
);
SyncPreSnapshot
*
pMsg
=
taosMemoryMalloc
(
bytes
);
memset
(
pMsg
,
0
,
bytes
);
pMsg
->
bytes
=
bytes
;
pMsg
->
vgId
=
vgId
;
pMsg
->
msgType
=
TDMT_SYNC_PRE_SNAPSHOT
;
return
pMsg
;
}
void
syncPreSnapshotDestroy
(
SyncPreSnapshot
*
pMsg
)
{
if
(
pMsg
!=
NULL
)
{
taosMemoryFree
(
pMsg
);
}
}
void
syncPreSnapshotSerialize
(
const
SyncPreSnapshot
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
)
{
ASSERT
(
pMsg
->
bytes
<=
bufLen
);
memcpy
(
buf
,
pMsg
,
pMsg
->
bytes
);
}
void
syncPreSnapshotDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshot
*
pMsg
)
{
memcpy
(
pMsg
,
buf
,
len
);
ASSERT
(
len
==
pMsg
->
bytes
);
}
char
*
syncPreSnapshotSerialize2
(
const
SyncPreSnapshot
*
pMsg
,
uint32_t
*
len
)
{
char
*
buf
=
taosMemoryMalloc
(
pMsg
->
bytes
);
ASSERT
(
buf
!=
NULL
);
syncPreSnapshotSerialize
(
pMsg
,
buf
,
pMsg
->
bytes
);
if
(
len
!=
NULL
)
{
*
len
=
pMsg
->
bytes
;
}
return
buf
;
}
SyncPreSnapshot
*
syncPreSnapshotDeserialize2
(
const
char
*
buf
,
uint32_t
len
)
{
uint32_t
bytes
=
*
((
uint32_t
*
)
buf
);
SyncPreSnapshot
*
pMsg
=
taosMemoryMalloc
(
bytes
);
ASSERT
(
pMsg
!=
NULL
);
syncPreSnapshotDeserialize
(
buf
,
len
,
pMsg
);
ASSERT
(
len
==
pMsg
->
bytes
);
return
pMsg
;
}
void
syncPreSnapshot2RpcMsg
(
const
SyncPreSnapshot
*
pMsg
,
SRpcMsg
*
pRpcMsg
)
{
memset
(
pRpcMsg
,
0
,
sizeof
(
*
pRpcMsg
));
pRpcMsg
->
msgType
=
pMsg
->
msgType
;
pRpcMsg
->
contLen
=
pMsg
->
bytes
;
pRpcMsg
->
pCont
=
rpcMallocCont
(
pRpcMsg
->
contLen
);
syncPreSnapshotSerialize
(
pMsg
,
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
}
void
syncPreSnapshotFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshot
*
pMsg
)
{
syncPreSnapshotDeserialize
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
,
pMsg
);
}
SyncPreSnapshot
*
syncPreSnapshotFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
)
{
SyncPreSnapshot
*
pMsg
=
syncPreSnapshotDeserialize2
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
ASSERT
(
pMsg
!=
NULL
);
return
pMsg
;
}
cJSON
*
syncPreSnapshot2Json
(
const
SyncPreSnapshot
*
pMsg
)
{
char
u64buf
[
128
]
=
{
0
};
cJSON
*
pRoot
=
cJSON_CreateObject
();
if
(
pMsg
!=
NULL
)
{
cJSON_AddNumberToObject
(
pRoot
,
"bytes"
,
pMsg
->
bytes
);
cJSON_AddNumberToObject
(
pRoot
,
"vgId"
,
pMsg
->
vgId
);
cJSON_AddNumberToObject
(
pRoot
,
"msgType"
,
pMsg
->
msgType
);
cJSON
*
pSrcId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
srcId
.
addr
);
cJSON_AddStringToObject
(
pSrcId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
srcId
.
addr
;
cJSON
*
pTmp
=
pSrcId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pSrcId
,
"vgId"
,
pMsg
->
srcId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"srcId"
,
pSrcId
);
cJSON
*
pDestId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
destId
.
addr
);
cJSON_AddStringToObject
(
pDestId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
destId
.
addr
;
cJSON
*
pTmp
=
pDestId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pDestId
,
"vgId"
,
pMsg
->
destId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"destId"
,
pDestId
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
}
cJSON
*
pJson
=
cJSON_CreateObject
();
cJSON_AddItemToObject
(
pJson
,
"SyncPreSnapshot"
,
pRoot
);
return
pJson
;
}
char
*
syncPreSnapshot2Str
(
const
SyncPreSnapshot
*
pMsg
)
{
cJSON
*
pJson
=
syncPreSnapshot2Json
(
pMsg
);
char
*
serialized
=
cJSON_Print
(
pJson
);
cJSON_Delete
(
pJson
);
return
serialized
;
}
void
syncPreSnapshotPrint
(
const
SyncPreSnapshot
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
printf
(
"syncPreSnapshotPrint | len:%d | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotPrint2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
printf
(
"syncPreSnapshotPrint2 | len:%d | %s | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotLog
(
const
SyncPreSnapshot
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotLog | len:%d | %s"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotLog2
(
char
*
s
,
const
SyncPreSnapshot
*
pMsg
)
{
if
(
gRaftDetailLog
)
{
char
*
serialized
=
syncPreSnapshot2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotLog2 | len:%d | %s | %s"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
taosMemoryFree
(
serialized
);
}
}
// ---- message process SyncPreSnapshotReply----
SyncPreSnapshotReply
*
syncPreSnapshotReplyBuild
(
int32_t
vgId
)
{
uint32_t
bytes
=
sizeof
(
SyncPreSnapshotReply
);
SyncPreSnapshotReply
*
pMsg
=
taosMemoryMalloc
(
bytes
);
memset
(
pMsg
,
0
,
bytes
);
pMsg
->
bytes
=
bytes
;
pMsg
->
vgId
=
vgId
;
pMsg
->
msgType
=
TDMT_SYNC_PRE_SNAPSHOT_REPLY
;
return
pMsg
;
}
void
syncPreSnapshotReplyDestroy
(
SyncPreSnapshotReply
*
pMsg
)
{
if
(
pMsg
!=
NULL
)
{
taosMemoryFree
(
pMsg
);
}
}
void
syncPreSnapshotReplySerialize
(
const
SyncPreSnapshotReply
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
)
{
ASSERT
(
pMsg
->
bytes
<=
bufLen
);
memcpy
(
buf
,
pMsg
,
pMsg
->
bytes
);
}
void
syncPreSnapshotReplyDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncPreSnapshotReply
*
pMsg
)
{
memcpy
(
pMsg
,
buf
,
len
);
ASSERT
(
len
==
pMsg
->
bytes
);
}
char
*
syncPreSnapshotReplySerialize2
(
const
SyncPreSnapshotReply
*
pMsg
,
uint32_t
*
len
)
{
char
*
buf
=
taosMemoryMalloc
(
pMsg
->
bytes
);
ASSERT
(
buf
!=
NULL
);
syncPreSnapshotReplySerialize
(
pMsg
,
buf
,
pMsg
->
bytes
);
if
(
len
!=
NULL
)
{
*
len
=
pMsg
->
bytes
;
}
return
buf
;
}
SyncPreSnapshotReply
*
syncPreSnapshotReplyDeserialize2
(
const
char
*
buf
,
uint32_t
len
)
{
uint32_t
bytes
=
*
((
uint32_t
*
)
buf
);
SyncPreSnapshotReply
*
pMsg
=
taosMemoryMalloc
(
bytes
);
ASSERT
(
pMsg
!=
NULL
);
syncPreSnapshotReplyDeserialize
(
buf
,
len
,
pMsg
);
ASSERT
(
len
==
pMsg
->
bytes
);
return
pMsg
;
}
void
syncPreSnapshotReply2RpcMsg
(
const
SyncPreSnapshotReply
*
pMsg
,
SRpcMsg
*
pRpcMsg
)
{
memset
(
pRpcMsg
,
0
,
sizeof
(
*
pRpcMsg
));
pRpcMsg
->
msgType
=
pMsg
->
msgType
;
pRpcMsg
->
contLen
=
pMsg
->
bytes
;
pRpcMsg
->
pCont
=
rpcMallocCont
(
pRpcMsg
->
contLen
);
syncPreSnapshotReplySerialize
(
pMsg
,
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
}
void
syncPreSnapshotReplyFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncPreSnapshotReply
*
pMsg
)
{
syncPreSnapshotReplyDeserialize
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
,
pMsg
);
}
SyncPreSnapshotReply
*
syncPreSnapshotReplyFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
)
{
SyncPreSnapshotReply
*
pMsg
=
syncPreSnapshotReplyDeserialize2
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
ASSERT
(
pMsg
!=
NULL
);
return
pMsg
;
}
cJSON
*
syncPreSnapshotReply2Json
(
const
SyncPreSnapshotReply
*
pMsg
)
{
char
u64buf
[
128
]
=
{
0
};
cJSON
*
pRoot
=
cJSON_CreateObject
();
if
(
pMsg
!=
NULL
)
{
cJSON_AddNumberToObject
(
pRoot
,
"bytes"
,
pMsg
->
bytes
);
cJSON_AddNumberToObject
(
pRoot
,
"vgId"
,
pMsg
->
vgId
);
cJSON_AddNumberToObject
(
pRoot
,
"msgType"
,
pMsg
->
msgType
);
cJSON
*
pSrcId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
srcId
.
addr
);
cJSON_AddStringToObject
(
pSrcId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
srcId
.
addr
;
cJSON
*
pTmp
=
pSrcId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pSrcId
,
"vgId"
,
pMsg
->
srcId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"srcId"
,
pSrcId
);
cJSON
*
pDestId
=
cJSON_CreateObject
();
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
destId
.
addr
);
cJSON_AddStringToObject
(
pDestId
,
"addr"
,
u64buf
);
{
uint64_t
u64
=
pMsg
->
destId
.
addr
;
cJSON
*
pTmp
=
pDestId
;
char
host
[
128
]
=
{
0
};
uint16_t
port
;
syncUtilU642Addr
(
u64
,
host
,
sizeof
(
host
),
&
port
);
cJSON_AddStringToObject
(
pTmp
,
"addr_host"
,
host
);
cJSON_AddNumberToObject
(
pTmp
,
"addr_port"
,
port
);
}
cJSON_AddNumberToObject
(
pDestId
,
"vgId"
,
pMsg
->
destId
.
vgId
);
cJSON_AddItemToObject
(
pRoot
,
"destId"
,
pDestId
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRId64
,
pMsg
->
snapStart
);
cJSON_AddStringToObject
(
pRoot
,
"snap-start"
,
u64buf
);
}
cJSON
*
pJson
=
cJSON_CreateObject
();
cJSON_AddItemToObject
(
pJson
,
"SyncPreSnapshotReply"
,
pRoot
);
return
pJson
;
}
char
*
syncPreSnapshotReply2Str
(
const
SyncPreSnapshotReply
*
pMsg
)
{
cJSON
*
pJson
=
syncPreSnapshotReply2Json
(
pMsg
);
char
*
serialized
=
cJSON_Print
(
pJson
);
cJSON_Delete
(
pJson
);
return
serialized
;
}
void
syncPreSnapshotReplyPrint
(
const
SyncPreSnapshotReply
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
printf
(
"syncPreSnapshotReplyPrint | len:%d | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotReplyPrint2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
printf
(
"syncPreSnapshotReplyPrint2 | len:%d | %s | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotReplyLog
(
const
SyncPreSnapshotReply
*
pMsg
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotReplyLog | len:%d | %s"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
taosMemoryFree
(
serialized
);
}
void
syncPreSnapshotReplyLog2
(
char
*
s
,
const
SyncPreSnapshotReply
*
pMsg
)
{
if
(
gRaftDetailLog
)
{
char
*
serialized
=
syncPreSnapshotReply2Str
(
pMsg
);
sTrace
(
"syncPreSnapshotReplyLog2 | len:%d | %s | %s"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
taosMemoryFree
(
serialized
);
}
}
source/libs/sync/test/sync_test_lib/src/syncSnapshotDebug.c
浏览文件 @
6da0b736
...
...
@@ -130,3 +130,60 @@ char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) {
cJSON_Delete
(
pJson
);
return
serialized
;
}
int32_t
syncNodeOnPreSnapshot
(
SSyncNode
*
ths
,
SyncPreSnapshot
*
pMsg
)
{
syncLogRecvSyncPreSnapshot
(
ths
,
pMsg
,
""
);
SyncPreSnapshotReply
*
pMsgReply
=
syncPreSnapshotReplyBuild
(
ths
->
vgId
);
pMsgReply
->
srcId
=
ths
->
myRaftId
;
pMsgReply
->
destId
=
pMsg
->
srcId
;
pMsgReply
->
term
=
ths
->
pRaftStore
->
currentTerm
;
SSyncLogStoreData
*
pData
=
ths
->
pLogStore
->
data
;
SWal
*
pWal
=
pData
->
pWal
;
if
(
syncNodeIsMnode
(
ths
))
{
pMsgReply
->
snapStart
=
SYNC_INDEX_BEGIN
;
}
else
{
bool
isEmpty
=
ths
->
pLogStore
->
syncLogIsEmpty
(
ths
->
pLogStore
);
int64_t
walCommitVer
=
walGetCommittedVer
(
pWal
);
if
(
!
isEmpty
&&
ths
->
commitIndex
!=
walCommitVer
)
{
sNError
(
ths
,
"commit not same, wal-commit:%"
PRId64
", commit:%"
PRId64
", ignore"
,
walCommitVer
,
ths
->
commitIndex
);
goto
_IGNORE
;
}
pMsgReply
->
snapStart
=
ths
->
commitIndex
+
1
;
// make local log clean
int32_t
code
=
ths
->
pLogStore
->
syncLogTruncate
(
ths
->
pLogStore
,
pMsgReply
->
snapStart
);
if
(
code
!=
0
)
{
sNError
(
ths
,
"truncate wal error"
);
goto
_IGNORE
;
}
}
// can not write behind _RESPONSE
SRpcMsg
rpcMsg
;
_RESPONSE:
syncPreSnapshotReply2RpcMsg
(
pMsgReply
,
&
rpcMsg
);
syncNodeSendMsgById
(
&
pMsgReply
->
destId
,
ths
,
&
rpcMsg
);
syncPreSnapshotReplyDestroy
(
pMsgReply
);
return
0
;
_IGNORE:
syncPreSnapshotReplyDestroy
(
pMsgReply
);
return
0
;
}
int32_t
syncNodeOnPreSnapshotReply
(
SSyncNode
*
ths
,
SyncPreSnapshotReply
*
pMsg
)
{
syncLogRecvSyncPreSnapshotReply
(
ths
,
pMsg
,
""
);
// start snapshot
return
0
;
}
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录