/* * Copyright (c) 2019 TAOS Data, Inc. * * 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 . */ #ifndef _TD_WAL_H_ #define _TD_WAL_H_ #include "os.h" #include "tarray.h" #include "tdef.h" #include "tlog.h" #include "tmsg.h" #ifdef __cplusplus extern "C" { #endif #define wFatal(...) \ { \ if (wDebugFlag & DEBUG_FATAL) { \ taosPrintLog("WAL FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); \ } \ } #define wError(...) \ { \ if (wDebugFlag & DEBUG_ERROR) { \ taosPrintLog("WAL ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); \ } \ } #define wWarn(...) \ { \ if (wDebugFlag & DEBUG_WARN) { \ taosPrintLog("WAL WARN ", DEBUG_WARN, 255, __VA_ARGS__); \ } \ } #define wInfo(...) \ { \ if (wDebugFlag & DEBUG_INFO) { \ taosPrintLog("WAL ", DEBUG_INFO, 255, __VA_ARGS__); \ } \ } #define wDebug(...) \ { \ if (wDebugFlag & DEBUG_DEBUG) { \ taosPrintLog("WAL ", DEBUG_DEBUG, wDebugFlag, __VA_ARGS__); \ } \ } #define wTrace(...) \ { \ if (wDebugFlag & DEBUG_TRACE) { \ taosPrintLog("WAL ", DEBUG_TRACE, wDebugFlag, __VA_ARGS__); \ } \ } #define WAL_HEAD_VER 0 #define WAL_NOSUFFIX_LEN 20 #define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN + 1) #define WAL_LOG_SUFFIX "log" #define WAL_INDEX_SUFFIX "idx" #define WAL_REFRESH_MS 1000 #define WAL_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead)) #define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12) #define WAL_FILE_LEN (WAL_PATH_LEN + 32) #define WAL_MAGIC 0xFAFBFCFDULL #pragma pack(push, 1) typedef enum { TAOS_WAL_NOLOG = 0, TAOS_WAL_WRITE = 1, TAOS_WAL_FSYNC = 2, } EWalType; // used by sync module typedef struct { int8_t isWeek; uint64_t seqNum; uint64_t term; } SSyncLogMeta; typedef struct SWalReadHead { int8_t headVer; int8_t reserved; int16_t msgType; int32_t bodyLen; int64_t ingestTs; // not implemented int64_t version; // sync meta SSyncLogMeta syncMeta; char body[]; } SWalReadHead; typedef struct { int32_t vgId; int32_t fsyncPeriod; // millisecond int32_t retentionPeriod; // secs int32_t rollPeriod; // secs int64_t retentionSize; int64_t segSize; EWalType level; // wal level } SWalCfg; typedef struct { uint64_t magic; uint32_t cksumHead; uint32_t cksumBody; SWalReadHead head; } SWalHead; typedef struct SWalVer { int64_t firstVer; int64_t verInSnapshotting; int64_t snapshotVer; int64_t commitVer; int64_t lastVer; } SWalVer; typedef struct SWal { // cfg SWalCfg cfg; int32_t fsyncSeq; // meta SWalVer vers; TdFilePtr pWriteLogTFile; TdFilePtr pWriteIdxTFile; int32_t writeCur; SArray *fileInfoSet; // status int64_t totSize; int64_t lastRollSeq; // ctl int64_t refId; TdThreadMutex mutex; // ref SHashObj *pRefHash; // ref -> SWalRef // path char path[WAL_PATH_LEN]; // reusable write head SWalHead writeHead; } SWal; // WAL HANDLE typedef struct SWalReadHandle { SWal *pWal; TdFilePtr pReadLogTFile; TdFilePtr pReadIdxTFile; int64_t curFileFirstVer; int64_t curVersion; int64_t capacity; int64_t status; // if cursor valid TdThreadMutex mutex; SWalHead *pHead; } SWalReadHandle; #pragma pack(pop) // typedef int32_t (*FWalWrite)(void *ahandle, void *pHead); // module initialization int32_t walInit(); void walCleanUp(); // handle open and ctl SWal *walOpen(const char *path, SWalCfg *pCfg); int32_t walAlter(SWal *, SWalCfg *pCfg); void walClose(SWal *); // write int64_t walWriteWithSyncInfo(SWal *, int64_t index, tmsg_t msgType, SSyncLogMeta syncMeta, const void *body, int32_t bodyLen); int64_t walWrite(SWal *, int64_t index, tmsg_t msgType, const void *body, int32_t bodyLen); void walFsync(SWal *, bool force); // apis for lifecycle management int32_t walCommit(SWal *, int64_t ver); // truncate after int32_t walRollback(SWal *, int64_t ver); // notify that previous logs can be pruned safely int32_t walBeginSnapshot(SWal *, int64_t ver); int32_t walEndSnapshot(SWal *); int32_t walRestoreFromSnapshot(SWal *, int64_t ver); // int32_t walDataCorrupted(SWal*); // read SWalReadHandle *walOpenReadHandle(SWal *); void walCloseReadHandle(SWalReadHandle *); int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver); // only for tq usage // int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead); void walSetReaderCapacity(SWalReadHandle *pRead, int32_t capacity); int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead); int32_t walFetchBody(SWalReadHandle *pRead, SWalHead **ppHead); int32_t walSkipFetchBody(SWalReadHandle *pRead, const SWalHead *pHead); typedef struct { int64_t refId; int64_t ver; } SWalRef; SWalRef *walOpenRef(SWal *); void walCloseRef(SWalRef *); int32_t walRefVer(SWalRef *, int64_t ver); int32_t walUnrefVer(SWal *); // deprecated #if 0 int32_t walRead(SWal *, SWalHead **, int64_t ver); int32_t walReadWithFp(SWal *, FWalWrite writeFp, int64_t verStart, int32_t readNum); #endif // lifecycle check int64_t walGetFirstVer(SWal *); int64_t walGetSnapshotVer(SWal *); int64_t walGetLastVer(SWal *); int64_t walGetCommittedVer(SWal *); #ifdef __cplusplus } #endif #endif // _TD_WAL_H_