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

remove wal seek module

This commit is contained in:
Minglei Jin 2024-07-18 10:16:12 +08:00
parent d2be5e0464
commit f485e4c5e8
3 changed files with 151 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/libs/wal/inc/walInt.h
View File

@ -157,13 +157,7 @@ char* walMetaSerialize(SWal* pWal);
int walMetaDeserialize(SWal* pWal, const char* bytes); int walMetaDeserialize(SWal* pWal, const char* bytes);
// meta section end // meta section end
// seek section
int64_t walChangeWrite(SWal* pWal, int64_t ver);
int walInitWriteFile(SWal* pWal);
// seek section end
int64_t walGetSeq(); int64_t walGetSeq();
int32_t walRollImpl(SWal* pWal);
#ifdef __cplusplus #ifdef __cplusplus
} }

102
source/libs/wal/src/walSeek.c
View File

@ -1,102 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "os.h"
#include "taoserror.h"
#include "tref.h"
#include "walInt.h"
int walInitWriteFile(SWal* pWal) {
TdFilePtr pIdxTFile, pLogTFile;
SWalFileInfo* pRet = taosArrayGetLast(pWal->fileInfoSet);
int64_t fileFirstVer = pRet->firstVer;
char fnameStr[WAL_FILE_LEN];
walBuildIdxName(pWal, fileFirstVer, fnameStr);
pIdxTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pIdxTFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
walBuildLogName(pWal, fileFirstVer, fnameStr);
pLogTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pLogTFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
// switch file
pWal->pIdxFile = pIdxTFile;
pWal->pLogFile = pLogTFile;
pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1;
return 0;
}
int64_t walChangeWrite(SWal* pWal, int64_t ver) {
int code;
TdFilePtr pIdxTFile, pLogTFile;
char fnameStr[WAL_FILE_LEN];
if (pWal->pLogFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pLogFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
code = taosCloseFile(&pWal->pLogFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
}
if (pWal->pIdxFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pIdxFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
code = taosCloseFile(&pWal->pIdxFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
}
SWalFileInfo tmpInfo;
tmpInfo.firstVer = ver;
// bsearch in fileSet
int32_t idx = taosArraySearchIdx(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE);
/*A(idx != -1);*/
SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, idx);
int64_t fileFirstVer = pFileInfo->firstVer;
walBuildIdxName(pWal, fileFirstVer, fnameStr);
pIdxTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pIdxTFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
pWal->pIdxFile = NULL;
return -1;
}
walBuildLogName(pWal, fileFirstVer, fnameStr);
pLogTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pLogTFile == NULL) {
taosCloseFile(&pIdxTFile);
terrno = TAOS_SYSTEM_ERROR(errno);
pWal->pLogFile = NULL;
return -1;
}
pWal->pLogFile = pLogTFile;
pWal->pIdxFile = pIdxTFile;
pWal->writeCur = idx;
return fileFirstVer;
}

220
source/libs/wal/src/walWrite.c
View File

@ -99,6 +99,63 @@ int32_t walCommit(SWal *pWal, int64_t ver) {
return 0; return 0;
} }
static int64_t walChangeWrite(SWal *pWal, int64_t ver) {
int code;
TdFilePtr pIdxTFile, pLogTFile;
char fnameStr[WAL_FILE_LEN];
if (pWal->pLogFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pLogFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
code = taosCloseFile(&pWal->pLogFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
}
if (pWal->pIdxFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pIdxFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
code = taosCloseFile(&pWal->pIdxFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
}
SWalFileInfo tmpInfo;
tmpInfo.firstVer = ver;
// bsearch in fileSet
int32_t idx = taosArraySearchIdx(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE);
/*A(idx != -1);*/
SWalFileInfo *pFileInfo = taosArrayGet(pWal->fileInfoSet, idx);
int64_t fileFirstVer = pFileInfo->firstVer;
walBuildIdxName(pWal, fileFirstVer, fnameStr);
pIdxTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pIdxTFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
pWal->pIdxFile = NULL;
return -1;
}
walBuildLogName(pWal, fileFirstVer, fnameStr);
pLogTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pLogTFile == NULL) {
taosCloseFile(&pIdxTFile);
terrno = TAOS_SYSTEM_ERROR(errno);
pWal->pLogFile = NULL;
return -1;
}
pWal->pLogFile = pLogTFile;
pWal->pIdxFile = pIdxTFile;
pWal->writeCur = idx;
return fileFirstVer;
}
int32_t walRollback(SWal *pWal, int64_t ver) { int32_t walRollback(SWal *pWal, int64_t ver) {
taosThreadMutexLock(&pWal->mutex); taosThreadMutexLock(&pWal->mutex);
wInfo("vgId:%d, wal rollback for version %" PRId64, pWal->cfg.vgId, ver); wInfo("vgId:%d, wal rollback for version %" PRId64, pWal->cfg.vgId, ver);
@ -223,6 +280,75 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
return 0; return 0;
} }
static int32_t walRollImpl(SWal *pWal) {
int32_t code = 0;
if (pWal->pIdxFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pIdxFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
code = taosCloseFile(&pWal->pIdxFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
}
if (pWal->pLogFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pLogFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
code = taosCloseFile(&pWal->pLogFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
}
TdFilePtr pIdxFile, pLogFile;
// create new file
int64_t newFileFirstVer = pWal->vers.lastVer + 1;
char fnameStr[WAL_FILE_LEN];
walBuildIdxName(pWal, newFileFirstVer, fnameStr);
pIdxFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pIdxFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
code = -1;
goto END;
}
walBuildLogName(pWal, newFileFirstVer, fnameStr);
pLogFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
wDebug("vgId:%d, wal create new file for write:%s", pWal->cfg.vgId, fnameStr);
if (pLogFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
code = -1;
goto END;
}
// error code was set inner
code = walRollFileInfo(pWal);
if (code != 0) {
goto END;
}
// switch file
pWal->pIdxFile = pIdxFile;
pWal->pLogFile = pLogFile;
pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1;
pWal->lastRollSeq = walGetSeq();
code = walSaveMeta(pWal);
if (code < 0) {
wError("vgId:%d, failed to save meta since %s", pWal->cfg.vgId, terrstr());
goto END;
}
END:
return code;
}
static FORCE_INLINE int32_t walCheckAndRoll(SWal *pWal) { static FORCE_INLINE int32_t walCheckAndRoll(SWal *pWal) {
if (taosArrayGetSize(pWal->fileInfoSet) == 0) { if (taosArrayGetSize(pWal->fileInfoSet) == 0) {
if (walRollImpl(pWal) < 0) { if (walRollImpl(pWal) < 0) {
@ -392,75 +518,6 @@ END:
return code; return code;
} }
int32_t walRollImpl(SWal *pWal) {
int32_t code = 0;
if (pWal->pIdxFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pIdxFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
code = taosCloseFile(&pWal->pIdxFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
}
if (pWal->pLogFile != NULL) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pLogFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
code = taosCloseFile(&pWal->pLogFile);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
}
TdFilePtr pIdxFile, pLogFile;
// create new file
int64_t newFileFirstVer = pWal->vers.lastVer + 1;
char fnameStr[WAL_FILE_LEN];
walBuildIdxName(pWal, newFileFirstVer, fnameStr);
pIdxFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pIdxFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
code = -1;
goto END;
}
walBuildLogName(pWal, newFileFirstVer, fnameStr);
pLogFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
wDebug("vgId:%d, wal create new file for write:%s", pWal->cfg.vgId, fnameStr);
if (pLogFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
code = -1;
goto END;
}
// error code was set inner
code = walRollFileInfo(pWal);
if (code != 0) {
goto END;
}
// switch file
pWal->pIdxFile = pIdxFile;
pWal->pLogFile = pLogFile;
pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1;
pWal->lastRollSeq = walGetSeq();
code = walSaveMeta(pWal);
if (code < 0) {
wError("vgId:%d, failed to save meta since %s", pWal->cfg.vgId, terrstr());
goto END;
}
END:
return code;
}
static int32_t walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) { static int32_t walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
SWalIdxEntry entry = {.ver = ver, .offset = offset}; SWalIdxEntry entry = {.ver = ver, .offset = offset};
SWalFileInfo *pFileInfo = walGetCurFileInfo(pWal); SWalFileInfo *pFileInfo = walGetCurFileInfo(pWal);
@ -617,6 +674,31 @@ END:
return -1; return -1;
} }
static int walInitWriteFile(SWal *pWal) {
TdFilePtr pIdxTFile, pLogTFile;
SWalFileInfo *pRet = taosArrayGetLast(pWal->fileInfoSet);
int64_t fileFirstVer = pRet->firstVer;
char fnameStr[WAL_FILE_LEN];
walBuildIdxName(pWal, fileFirstVer, fnameStr);
pIdxTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pIdxTFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
walBuildLogName(pWal, fileFirstVer, fnameStr);
pLogTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pLogTFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
// switch file
pWal->pIdxFile = pIdxTFile;
pWal->pLogFile = pLogTFile;
pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1;
return 0;
}
int32_t walAppendLog(SWal *pWal, int64_t index, tmsg_t msgType, SWalSyncInfo syncMeta, const void *body, int32_t walAppendLog(SWal *pWal, int64_t index, tmsg_t msgType, SWalSyncInfo syncMeta, const void *body,
int32_t bodyLen) { int32_t bodyLen) {
int32_t code = 0, lino = 0; int32_t code = 0, lino = 0;