xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #15568 from taosdata/feature/3.0_mhli 

refactor(sync): speed up sync point
This commit is contained in:
Li Minghao 2022-07-29 22:06:41 +08:00 committed by GitHub
parent 29c94d17b6 e0e23fc2cb
commit 93aa8f9add
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 139 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -213,6 +213,11 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie
if (nextIndex > SYNC_INDEX_BEGIN) {
--nextIndex;
// speed up
if (nextIndex > pMsg->matchIndex + 1) {
nextIndex = pMsg->matchIndex + 1;
}
bool needStartSnapshot = false;
if (nextIndex >= SYNC_INDEX_BEGIN && !ths->pLogStore->syncLogExist(ths->pLogStore, nextIndex)) {
needStartSnapshot = true;

10
source/libs/sync/src/syncMain.c
View File

@ -2222,13 +2222,18 @@ SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
SyncIndex preIndex = index - 1;
SSyncRaftEntry* pPreEntry = NULL;
int32_t code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry);
SSnapshot snapshot = {.data = NULL,
.lastApplyIndex = SYNC_INDEX_INVALID,
.lastApplyTerm = SYNC_TERM_INVALID,
.lastConfigIndex = SYNC_INDEX_INVALID};
if (code == 0) {
ASSERT(pPreEntry != NULL);
preTerm = pPreEntry->term;
taosMemoryFree(pPreEntry);
return preTerm;
} else {
SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
if (snapshot.lastApplyIndex == preIndex) {
@ -2239,7 +2244,8 @@ SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "sync node get pre term error, index:%" PRId64, index);
snprintf(logBuf, sizeof(logBuf), "sync node get pre term error, index:%ld, snap-index:%ld, snap-term:%lu", index,
snapshot.lastApplyIndex, snapshot.lastApplyTerm);
syncNodeErrorLog(pSyncNode, logBuf);
} while (0);

6
source/libs/sync/src/syncReplication.c
View File

@ -132,7 +132,8 @@ int32_t syncNodeAppendEntriesPeersSnapshot2(SSyncNode* pSyncNode) {
SyncIndex preLogIndex = syncNodeGetPreIndex(pSyncNode, nextIndex);
SyncTerm preLogTerm = syncNodeGetPreTerm(pSyncNode, nextIndex);
if (preLogTerm == SYNC_TERM_INVALID) {
SyncIndex newNextIndex = syncNodeGetLastIndex(pSyncNode) + 1;
// SyncIndex newNextIndex = syncNodeGetLastIndex(pSyncNode) + 1;
SyncIndex newNextIndex = nextIndex + 1;
syncIndexMgrSetIndex(pSyncNode->pNextIndex, pDestId, newNextIndex);
syncIndexMgrSetIndex(pSyncNode->pMatchIndex, pDestId, SYNC_INDEX_INVALID);
sError("vgId:%d sync get pre term error, nextIndex:%" PRId64 ", update next-index:%" PRId64
@ -222,7 +223,8 @@ int32_t syncNodeAppendEntriesPeersSnapshot(SSyncNode* pSyncNode) {
SyncIndex preLogIndex = syncNodeGetPreIndex(pSyncNode, nextIndex);
SyncTerm preLogTerm = syncNodeGetPreTerm(pSyncNode, nextIndex);
if (preLogTerm == SYNC_TERM_INVALID) {
SyncIndex newNextIndex = syncNodeGetLastIndex(pSyncNode) + 1;
// SyncIndex newNextIndex = syncNodeGetLastIndex(pSyncNode) + 1;
SyncIndex newNextIndex = nextIndex + 1;
syncIndexMgrSetIndex(pSyncNode->pNextIndex, pDestId, newNextIndex);
syncIndexMgrSetIndex(pSyncNode->pMatchIndex, pDestId, SYNC_INDEX_INVALID);
sError("vgId:%d sync get pre term error, nextIndex:%" PRId64 ", update next-index:%" PRId64

45
source/libs/sync/test/sh/auto_bench.sh Normal file
View File

@ -0,0 +1,45 @@
#!/bin/bash
if [ $# != 5 ] ; then
echo "Uasge: $0 instances vgroups replica ctables rows"
echo ""
exit 1
fi
instances=$1
vgroups=$2
replica=$3
ctables=$4
rows=$5
echo "params: instances:${instances}, vgroups:${vgroups}, replica:${replica}, ctables:${ctables}, rows:${rows}"
dt=`date "+%Y-%m-%d-%H-%M-%S"`
casedir=instances_${instances}_vgroups_${vgroups}_replica_${replica}_ctables_${ctables}_rows_${rows}_${dt}
mkdir ${casedir}
cp ./insert.tpl.json ${casedir}
cd ${casedir}
for i in `seq 1 ${instances}`;do
#echo ===$i===
cfg_file=bench_${i}.json
cp ./insert.tpl.json ${cfg_file}
rstfile=result_${i}
sed -i 's/tpl_vgroups_tpl/'${vgroups}'/g' ${cfg_file}
sed -i 's/tpl_replica_tpl/'${replica}'/g' ${cfg_file}
sed -i 's/tpl_ctables_tpl/'${ctables}'/g' ${cfg_file}
sed -i 's/tpl_stid_tpl/'${i}'/g' ${cfg_file}
sed -i 's/tpl_rows_tpl/'${rows}'/g' ${cfg_file}
sed -i 's/tpl_insert_result_tpl/'${rstfile}'/g' ${cfg_file}
done
for conf_file in `ls ./bench_*.json`;do
echo "nohup taosBenchmark -f ${conf_file} &"
nohup taosBenchmark -f ${conf_file} &
done
cd -
exit 0

77
source/libs/sync/test/sh/insert.tpl.json Normal file
View File

@ -0,0 +1,77 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos/",
"host": "v3cluster-0001",
"port": 7100,
"user": "root",
"password": "taosdata",
"thread_count": 8,
"thread_count_create_tbl": 8,
"result_file": "./tpl_insert_result_tpl",
"confirm_parameter_prompt": "no",
"insert_interval": 0,
"interlace_rows": 0,
"num_of_records_per_req": 100000,
"databases": [
{
"dbinfo": {
"name": "db1",
"drop": "yes",
"vgroups": tpl_vgroups_tpl,
"replica": tpl_replica_tpl
},
"super_tables": [
{
"name": "stb_tpl_stid_tpl",
"child_table_exists": "no",
"childtable_count": tpl_ctables_tpl,
"childtable_prefix": "stb_tpl_stid_tpl_",
"auto_create_table": "no",
"batch_create_tbl_num": 50000,
"data_source": "rand",
"insert_mode": "taosc",
"insert_rows": tpl_rows_tpl,
"interlace_rows": 0,
"insert_interval": 0,
"max_sql_len": 10000000,
"disorder_ratio": 0,
"disorder_range": 1000,
"timestamp_step": 10,
"sample_format": "csv",
"use_sample_ts": "no",
"tags_file": "",
"columns": [
{
"type": "INT"
},
{
"type": "DOUBLE",
"count": 1
},
{
"type": "BINARY",
"len": 40,
"count": 1
},
{
"type": "nchar",
"len": 20,
"count": 1
}
],
"tags": [
{
"type": "TINYINT",
"count": 1
},
{
"type": "BINARY",
"len": 16,
"count": 1
}
]
}
]
}
]
}