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

memory leak

This commit is contained in:
dmchen 2023-07-19 17:33:55 +08:00
parent 8062d603b7
commit 154bc177a7
1 changed files with 29 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
source/libs/sync/src/syncPipeline.c
View File

@ -638,39 +638,40 @@ int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t comm
pEntry->index, pEntry->term, role, currentTerm); pEntry->index, pEntry->term, role, currentTerm);
pNextEntry = syncLogBufferGetOneEntry(pBuf, pNode, index + 1, &nextInBuf); pNextEntry = syncLogBufferGetOneEntry(pBuf, pNode, index + 1, &nextInBuf);
if (pNextEntry != NULL && pNextEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE) { if (pNextEntry != NULL) {
sInfo("vgId:%d, to change config at Commit. " if(pNextEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE){
"current entry, index:%" PRId64 ", term:%" PRId64", " sInfo("vgId:%d, to change config at Commit. "
"node, role:%d, current term:%" PRId64 ", restore:%d, " "current entry, index:%" PRId64 ", term:%" PRId64", "
"cond, next entry index:%" PRId64 ", msgType:%s", "node, role:%d, current term:%" PRId64 ", restore:%d, "
vgId, "cond, next entry index:%" PRId64 ", msgType:%s",
pEntry->index, pEntry->term, vgId,
role, currentTerm, pNode->restoreFinish, pEntry->index, pEntry->term,
pNextEntry->index, TMSG_INFO(pNextEntry->originalRpcType)); role, currentTerm, pNode->restoreFinish,
pNextEntry->index, TMSG_INFO(pNextEntry->originalRpcType));
if(syncNodeChangeConfig(pNode, pNextEntry, "Commit") != 0){ if(syncNodeChangeConfig(pNode, pNextEntry, "Commit") != 0){
sError("vgId:%d, failed to change config from Commit. index:%" PRId64 ", term:%" PRId64 sError("vgId:%d, failed to change config from Commit. index:%" PRId64 ", term:%" PRId64
", role:%d, current term:%" PRId64, ", role:%d, current term:%" PRId64,
vgId, pNextEntry->index, pNextEntry->term, role, currentTerm); vgId, pNextEntry->index, pNextEntry->term, role, currentTerm);
goto _out; goto _out;
}
//for 2->1, need to apply config change entry in sync thread,
if(pNode->replicaNum == 1){
if (syncFsmExecute(pNode, pFsm, role, currentTerm, pNextEntry, 0, true) != 0) {
sError("vgId:%d, failed to execute sync log entry. index:%" PRId64 ", term:%" PRId64
", role:%d, current term:%" PRId64,
vgId, pNextEntry->index, pNextEntry->term, role, currentTerm);
goto _out;
} }
index++; //for 2->1, need to apply config change entry in sync thread,
pBuf->commitIndex = index; if(pNode->replicaNum == 1){
if (syncFsmExecute(pNode, pFsm, role, currentTerm, pNextEntry, 0, true) != 0) {
sError("vgId:%d, failed to execute sync log entry. index:%" PRId64 ", term:%" PRId64
", role:%d, current term:%" PRId64,
vgId, pNextEntry->index, pNextEntry->term, role, currentTerm);
goto _out;
}
sTrace("vgId:%d, committed index:%" PRId64 ", term:%" PRId64 ", role:%d, current term:%" PRId64 "", pNode->vgId, index++;
pNextEntry->index, pNextEntry->term, role, currentTerm); pBuf->commitIndex = index;
sTrace("vgId:%d, committed index:%" PRId64 ", term:%" PRId64 ", role:%d, current term:%" PRId64 "", pNode->vgId,
pNextEntry->index, pNextEntry->term, role, currentTerm);
}
} }
if (!nextInBuf) { if (!nextInBuf) {
syncEntryDestroy(pNextEntry); syncEntryDestroy(pNextEntry);
pNextEntry = NULL; pNextEntry = NULL;