refactor(sync): delete old functions
This commit is contained in:
Minghao Li
parent
aded474d5b
commit
4bd650609d
|
|
@ -19,7 +19,6 @@
|
|||
|
||||
// refactor, log[0 .. n] ==> log[m .. n]
|
||||
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
|
||||
// static int32_t raftLogSetBeginIndex(struct SSyncLogStore* pLogStore, SyncIndex beginIndex);
|
||||
static SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore);
|
||||
static SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore);
|
||||
static SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore);
|
||||
|
|
@ -44,28 +43,27 @@ static int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex from
|
|||
static int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
|
||||
static SyncIndex logStoreGetCommitIndex(SSyncLogStore* pLogStore);
|
||||
|
||||
// refactor, log[0 .. n] ==> log[m .. n]
|
||||
/*
|
||||
static int32_t raftLogSetBeginIndex(struct SSyncLogStore* pLogStore, SyncIndex beginIndex) {
|
||||
// if beginIndex == 0, donot need call this funciton
|
||||
ASSERT(beginIndex > 0);
|
||||
|
||||
SSyncLogStoreData* pData = pLogStore->data;
|
||||
SWal* pWal = pData->pWal;
|
||||
sTrace("vgId:%d, reset wal begin index:%ld", pData->pSyncNode->vgId, beginIndex);
|
||||
|
||||
pData->beginIndex = beginIndex;
|
||||
walRestoreFromSnapshot(pWal, beginIndex - 1);
|
||||
return 0;
|
||||
}
|
||||
*/
|
||||
|
||||
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {
|
||||
ASSERT(snapshotIndex >= 0);
|
||||
|
||||
SSyncLogStoreData* pData = pLogStore->data;
|
||||
SWal* pWal = pData->pWal;
|
||||
walRestoreFromSnapshot(pWal, snapshotIndex);
|
||||
int32_t code = walRestoreFromSnapshot(pWal, snapshotIndex);
|
||||
if (code != 0) {
|
||||
int32_t err = terrno;
|
||||
const char* errStr = tstrerror(err);
|
||||
int32_t sysErr = errno;
|
||||
const char* sysErrStr = strerror(errno);
|
||||
|
||||
char logBuf[128];
|
||||
snprintf(logBuf, sizeof(logBuf),
|
||||
"wal restore from snapshot error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", snapshotIndex, err,
|
||||
err, errStr, sysErr, sysErrStr);
|
||||
syncNodeErrorLog(pData->pSyncNode, logBuf);
|
||||
|
||||
ASSERT(0);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -91,18 +89,6 @@ static int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) {
|
|||
return count > 0 ? count : 0;
|
||||
}
|
||||
|
||||
#if 0
|
||||
static bool raftLogInRange(struct SSyncLogStore* pLogStore, SyncIndex index) {
|
||||
SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
|
||||
SyncIndex endIndex = raftLogEndIndex(pLogStore);
|
||||
if (index >= beginIndex && index <= endIndex) {
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) {
|
||||
SyncIndex lastIndex;
|
||||
SSyncLogStoreData* pData = pLogStore->data;
|
||||
|
|
@ -138,24 +124,6 @@ static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {
|
|||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {
|
||||
SyncTerm lastTerm = 0;
|
||||
if (raftLogEntryCount(pLogStore) == 0) {
|
||||
lastTerm = 0;
|
||||
} else {
|
||||
SSyncRaftEntry* pLastEntry;
|
||||
int32_t code = raftLogGetLastEntry(pLogStore, &pLastEntry);
|
||||
ASSERT(code == 0);
|
||||
if (pLastEntry != NULL) {
|
||||
lastTerm = pLastEntry->term;
|
||||
taosMemoryFree(pLastEntry);
|
||||
}
|
||||
}
|
||||
return lastTerm;
|
||||
}
|
||||
*/
|
||||
|
||||
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
|
||||
SSyncLogStoreData* pData = pLogStore->data;
|
||||
SWal* pWal = pData->pWal;
|
||||
|
|
@ -197,53 +165,6 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
|
|||
return code;
|
||||
}
|
||||
|
||||
#if 0
|
||||
static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
|
||||
SSyncLogStoreData* pData = pLogStore->data;
|
||||
SWal* pWal = pData->pWal;
|
||||
int32_t code;
|
||||
|
||||
*ppEntry = NULL;
|
||||
if (raftLogInRange(pLogStore, index)) {
|
||||
SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
|
||||
ASSERT(pWalHandle != NULL);
|
||||
|
||||
code = walReadWithHandle(pWalHandle, index);
|
||||
if (code != 0) {
|
||||
int32_t err = terrno;
|
||||
const char* errStr = tstrerror(err);
|
||||
int32_t linuxErr = errno;
|
||||
const char* linuxErrMsg = strerror(errno);
|
||||
sError("raftLogGetEntry error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
|
||||
linuxErrMsg);
|
||||
ASSERT(0);
|
||||
walCloseReadHandle(pWalHandle);
|
||||
return code;
|
||||
}
|
||||
|
||||
*ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
|
||||
ASSERT(*ppEntry != NULL);
|
||||
(*ppEntry)->msgType = TDMT_SYNC_CLIENT_REQUEST;
|
||||
(*ppEntry)->originalRpcType = pWalHandle->pHead->head.msgType;
|
||||
(*ppEntry)->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;
|
||||
(*ppEntry)->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;
|
||||
(*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;
|
||||
(*ppEntry)->index = index;
|
||||
ASSERT((*ppEntry)->dataLen == pWalHandle->pHead->head.bodyLen);
|
||||
memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
|
||||
|
||||
// need to hold, do not new every time!!
|
||||
walCloseReadHandle(pWalHandle);
|
||||
|
||||
} else {
|
||||
// index not in range
|
||||
code = 0;
|
||||
}
|
||||
|
||||
return code;
|
||||
}
|
||||
#endif
|
||||
|
||||
static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
|
||||
SSyncLogStoreData* pData = pLogStore->data;
|
||||
SWal* pWal = pData->pWal;
|
||||
|
|
@ -343,18 +264,6 @@ static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** pp
|
|||
return -1;
|
||||
}
|
||||
|
||||
/*
|
||||
static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry) {
|
||||
*ppLastEntry = NULL;
|
||||
if (raftLogEntryCount(pLogStore) == 0) {
|
||||
return 0;
|
||||
}
|
||||
SyncIndex lastIndex = raftLogLastIndex(pLogStore);
|
||||
int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);
|
||||
return code;
|
||||
}
|
||||
*/
|
||||
|
||||
//-------------------------------
|
||||
SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
|
||||
SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore));
|
||||
|
|
@ -372,18 +281,6 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
|
|||
pData->pWalHandle = walOpenReadHandle(pData->pWal);
|
||||
ASSERT(pData->pWalHandle != NULL);
|
||||
|
||||
/*
|
||||
SyncIndex firstVer = walGetFirstVer(pData->pWal);
|
||||
SyncIndex lastVer = walGetLastVer(pData->pWal);
|
||||
if (firstVer >= 0) {
|
||||
pData->beginIndex = firstVer;
|
||||
} else if (firstVer == -1) {
|
||||
pData->beginIndex = lastVer + 1;
|
||||
} else {
|
||||
ASSERT(0);
|
||||
}
|
||||
*/
|
||||
|
||||
pLogStore->appendEntry = logStoreAppendEntry;
|
||||
pLogStore->getEntry = logStoreGetEntry;
|
||||
pLogStore->truncate = logStoreTruncate;
|
||||
|
|
@ -392,7 +289,6 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
|
|||
pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
|
||||
pLogStore->getCommitIndex = logStoreGetCommitIndex;
|
||||
|
||||
// pLogStore->syncLogSetBeginIndex = raftLogSetBeginIndex;
|
||||
pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;
|
||||
pLogStore->syncLogBeginIndex = raftLogBeginIndex;
|
||||
pLogStore->syncLogEndIndex = raftLogEndIndex;
|
||||
|
|
@ -405,8 +301,6 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
|
|||
pLogStore->syncLogTruncate = raftLogTruncate;
|
||||
pLogStore->syncLogWriteIndex = raftLogWriteIndex;
|
||||
|
||||
// pLogStore->syncLogInRange = raftLogInRange;
|
||||
|
||||
return pLogStore;
|
||||
}
|
||||
|
||||
|
|
@ -593,55 +487,6 @@ SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) {
|
|||
return pEntry;
|
||||
}
|
||||
|
||||
/*
|
||||
cJSON* logStore2Json(SSyncLogStore* pLogStore) {
|
||||
char u64buf[128] = {0};
|
||||
SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
|
||||
cJSON* pRoot = cJSON_CreateObject();
|
||||
|
||||
if (pData != NULL && pData->pWal != NULL) {
|
||||
snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
|
||||
cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
|
||||
snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
|
||||
cJSON_AddStringToObject(pRoot, "pWal", u64buf);
|
||||
|
||||
snprintf(u64buf, sizeof(u64buf), "%ld", pData->beginIndex);
|
||||
cJSON_AddStringToObject(pRoot, "beginIndex", u64buf);
|
||||
|
||||
SyncIndex endIndex = raftLogEndIndex(pLogStore);
|
||||
snprintf(u64buf, sizeof(u64buf), "%ld", endIndex);
|
||||
cJSON_AddStringToObject(pRoot, "endIndex", u64buf);
|
||||
|
||||
int32_t count = raftLogEntryCount(pLogStore);
|
||||
cJSON_AddNumberToObject(pRoot, "entryCount", count);
|
||||
|
||||
snprintf(u64buf, sizeof(u64buf), "%ld", raftLogWriteIndex(pLogStore));
|
||||
cJSON_AddStringToObject(pRoot, "WriteIndex", u64buf);
|
||||
|
||||
snprintf(u64buf, sizeof(u64buf), "%d", raftLogIsEmpty(pLogStore));
|
||||
cJSON_AddStringToObject(pRoot, "IsEmpty", u64buf);
|
||||
|
||||
snprintf(u64buf, sizeof(u64buf), "%ld", raftLogLastIndex(pLogStore));
|
||||
cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
|
||||
snprintf(u64buf, sizeof(u64buf), "%lu", raftLogLastTerm(pLogStore));
|
||||
cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
|
||||
|
||||
cJSON* pEntries = cJSON_CreateArray();
|
||||
cJSON_AddItemToObject(pRoot, "pEntries", pEntries);
|
||||
|
||||
for (SyncIndex i = pData->beginIndex; i <= endIndex; ++i) {
|
||||
SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
|
||||
cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
|
||||
syncEntryDestory(pEntry);
|
||||
}
|
||||
}
|
||||
|
||||
cJSON* pJson = cJSON_CreateObject();
|
||||
cJSON_AddItemToObject(pJson, "SSyncLogStore", pRoot);
|
||||
return pJson;
|
||||
}
|
||||
*/
|
||||
|
||||
cJSON* logStore2Json(SSyncLogStore* pLogStore) {
|
||||
char u64buf[128] = {0};
|
||||
SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
|
||||
|
|
|
|||
|
|
@ -772,6 +772,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
|
|||
snapshotReSend(pSender);
|
||||
|
||||
} else {
|
||||
// error log
|
||||
do {
|
||||
char logBuf[96];
|
||||
snprintf(logBuf, sizeof(logBuf), "snapshot sender recv error ack:%d, my seq:%d", pMsg->ack, pSender->seq);
|
||||
|
|
|
|||
Loading…
Reference in New Issue