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

chore: rsma sync and assert

This commit is contained in:
kailixu 2022-12-25 21:54:09 +08:00
parent ed3a0c1dc1
commit 73710da55f
10 changed files with 380 additions and 353 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/util/taoserror.h
View File

@ -708,6 +708,9 @@ int32_t* taosGetErrno();
#define TSDB_CODE_RSMA_REGEX_MATCH TAOS_DEF_ERROR_CODE(0, 0x3158) #define TSDB_CODE_RSMA_REGEX_MATCH TAOS_DEF_ERROR_CODE(0, 0x3158)
#define TSDB_CODE_RSMA_STREAM_STATE_OPEN TAOS_DEF_ERROR_CODE(0, 0x3159) #define TSDB_CODE_RSMA_STREAM_STATE_OPEN TAOS_DEF_ERROR_CODE(0, 0x3159)
#define TSDB_CODE_RSMA_STREAM_STATE_COMMIT TAOS_DEF_ERROR_CODE(0, 0x3160) #define TSDB_CODE_RSMA_STREAM_STATE_COMMIT TAOS_DEF_ERROR_CODE(0, 0x3160)
#define TSDB_CODE_RSMA_FS_REF TAOS_DEF_ERROR_CODE(0, 0x3161)
#define TSDB_CODE_RSMA_FS_SYNC TAOS_DEF_ERROR_CODE(0, 0x3162)
#define TSDB_CODE_RSMA_FS_UPDATE TAOS_DEF_ERROR_CODE(0, 0x3163)
//index //index
#define TSDB_CODE_INDEX_REBUILDING TAOS_DEF_ERROR_CODE(0, 0x3200) #define TSDB_CODE_INDEX_REBUILDING TAOS_DEF_ERROR_CODE(0, 0x3200)

19
source/dnode/vnode/src/inc/sma.h
View File

@ -44,7 +44,6 @@ typedef struct SRSmaInfoItem SRSmaInfoItem;
typedef struct SRSmaFS SRSmaFS; typedef struct SRSmaFS SRSmaFS;
typedef struct SQTaskFile SQTaskFile; typedef struct SQTaskFile SQTaskFile;
typedef struct SQTaskFReader SQTaskFReader; typedef struct SQTaskFReader SQTaskFReader;
typedef struct SQTaskFWriter SQTaskFWriter;
struct SSmaEnv { struct SSmaEnv {
SRWLatch lock; SRWLatch lock;
@ -94,15 +93,11 @@ struct SQTaskFile {
struct SQTaskFReader { struct SQTaskFReader {
SSma *pSma; SSma *pSma;
int8_t level;
int64_t suid;
int64_t version; int64_t version;
TdFilePtr pReadH; TdFilePtr pReadH;
}; };
struct SQTaskFWriter {
SSma *pSma;
int64_t version;
TdFilePtr pWriteH;
char *fname;
};
struct SRSmaFS { struct SRSmaFS {
SArray *aQTaskInf; // array of SQTaskFile SArray *aQTaskInf; // array of SQTaskFile
@ -221,11 +216,11 @@ void tdRSmaFSClose(SRSmaFS *fs);
int32_t tdRSmaFSPrepareCommit(SSma *pSma, SRSmaFS *pFSNew); int32_t tdRSmaFSPrepareCommit(SSma *pSma, SRSmaFS *pFSNew);
int32_t tdRSmaFSCommit(SSma *pSma); int32_t tdRSmaFSCommit(SSma *pSma);
int32_t tdRSmaFSFinishCommit(SSma *pSma); int32_t tdRSmaFSFinishCommit(SSma *pSma);
int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFSOut); int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFS);
int32_t tdRSmaFSTakeSnapshot(SSma *pSma, SRSmaFS *pFSOut); int32_t tdRSmaFSTakeSnapshot(SSma *pSma, SRSmaFS *pFS);
int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t suid, int8_t level, int64_t version); int32_t tdRSmaFSRef(SSma *pSma, SRSmaFS *pFS);
void tdRSmaFSUnRef(SSma *pSma, SRSmaStat *pStat, int64_t suid, int8_t level, int64_t version); void tdRSmaFSUnRef(SSma *pSma, SRSmaFS *pFS);
int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t size); int32_t tdRSmaFSUpsertQTaskFile(SSma *pSma, SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t nSize);
int32_t tdRSmaFSRollback(SSma *pSma); int32_t tdRSmaFSRollback(SSma *pSma);
int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback); int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback);
int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName); int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);

2
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -271,6 +271,7 @@ int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData);
// SRSmaSnapWriter ======================================== // SRSmaSnapWriter ========================================
int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter); int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter);
int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData); int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
int32_t rsmaSnapWriterPrepareClose(SRSmaSnapWriter* pWriter);
int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback); int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback);
typedef struct { typedef struct {
@ -414,6 +415,7 @@ enum {
struct SSnapDataHdr { struct SSnapDataHdr {
int8_t type; int8_t type;
int8_t flag;
int64_t index; int64_t index;
int64_t size; int64_t size;
uint8_t data[]; uint8_t data[];

98
source/dnode/vnode/src/sma/smaFS.c
View File

@ -588,10 +588,10 @@ _exit:
return code; return code;
} }
int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t size) { int32_t tdRSmaFSUpsertQTaskFile(SSma *pSma, SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t nSize) {
int32_t code = 0; int32_t code = 0;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < nSize; ++i) {
SQTaskFile *qTaskF = qTaskFile + i; SQTaskFile *qTaskF = qTaskFile + i;
int32_t idx = taosArraySearchIdx(pFS->aQTaskInf, qTaskF, tdQTaskInfCmprFn1, TD_GE); int32_t idx = taosArraySearchIdx(pFS->aQTaskInf, qTaskF, tdQTaskInfCmprFn1, TD_GE);
@ -602,7 +602,14 @@ int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t siz
SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, idx); SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, idx);
int32_t c = tdQTaskInfCmprFn1(pTaskF, qTaskF); int32_t c = tdQTaskInfCmprFn1(pTaskF, qTaskF);
if (c == 0) { if (c == 0) {
ASSERT(0); if (pTaskF->size != qTaskF->size) {
code = TSDB_CODE_RSMA_FS_UPDATE;
smaError("vgId:%d, %s failed at line %d since %s, level:%" PRIi8 ", suid:%" PRIi64 ", version:%" PRIi64
", size:%" PRIi64 " != %" PRIi64,
SMA_VID(pSma), __func__, __LINE__, tstrerror(code), pTaskF->level, pTaskF->suid, pTaskF->version,
pTaskF->size, qTaskF->size);
goto _exit;
}
continue; continue;
} }
} }
@ -616,7 +623,7 @@ int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t siz
_exit: _exit:
return code; return code;
} }
#if 0
int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t suid, int8_t level, int64_t version) { int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t suid, int8_t level, int64_t version) {
SArray *aQTaskInf = RSMA_FS(pStat)->aQTaskInf; SArray *aQTaskInf = RSMA_FS(pStat)->aQTaskInf;
SQTaskFile qTaskF = {.level = level, .suid = suid, .version = version}; SQTaskFile qTaskF = {.level = level, .suid = suid, .version = version};
@ -696,38 +703,101 @@ void tdRSmaFSUnRef(SSma *pSma, SRSmaStat *pStat, int64_t suid, int8_t level, int
taosWUnLockLatch(RSMA_FS_LOCK(pStat)); taosWUnLockLatch(RSMA_FS_LOCK(pStat));
} }
#endif
int32_t tdRSmaFSTakeSnapshot(SSma *pSma, SRSmaFS *pFSOut) { int32_t tdRSmaFSRef(SSma *pSma, SRSmaFS *pFS) {
int32_t code = 0;
int32_t lino = 0;
int32_t nRef = 0;
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
SRSmaFS *qFS = RSMA_FS(pStat);
int32_t size = taosArrayGetSize(qFS->aQTaskInf);
pFS->aQTaskInf = taosArrayInit(size, sizeof(SQTaskFile));
if (pFS->aQTaskInf == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
for (int32_t i = 0; i < size; ++i) {
SQTaskFile *qTaskF = (SQTaskFile *)taosArrayGet(qFS->aQTaskInf, i);
nRef = atomic_fetch_add_32(&qTaskF->nRef, 1);
if(nRef <= 0) {
code = TSDB_CODE_RSMA_FS_REF;
TSDB_CHECK_CODE(code, lino, _exit);
}
}
taosArraySetSize(pFS->aQTaskInf, size);
memcpy(pFS->aQTaskInf->pData, qFS->aQTaskInf->pData, size * sizeof(SQTaskFile));
_exit:
if (code) {
smaError("vgId:%d, %s failed at line %d since %s, nRef %d", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code),
nRef);
}
return code;
}
void tdRSmaFSUnRef(SSma *pSma, SRSmaFS *pFS) {
int32_t nRef = 0;
char fname[TSDB_FILENAME_LEN];
SVnode *pVnode = pSma->pVnode;
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
int32_t size = taosArrayGetSize(pFS->aQTaskInf);
for (int32_t i = 0; i < size; ++i) {
SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, i);
nRef = atomic_sub_fetch_32(&pTaskF->nRef, 1);
if (nRef == 0) {
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, pTaskF->version,
tfsGetPrimaryPath(pVnode->pTfs), fname);
if (taosRemoveFile(fname) < 0) {
smaWarn("vgId:%d, failed to remove %s since %s", TD_VID(pVnode), fname, tstrerror(TAOS_SYSTEM_ERROR(errno)));
} else {
smaDebug("vgId:%d, success to remove %s", TD_VID(pVnode), fname);
}
} else if (nRef < 0) {
smaWarn("vgId:%d, abnormal unref %s since %s", TD_VID(pVnode), fname, tstrerror(TSDB_CODE_RSMA_FS_REF));
}
}
taosArrayDestroy(pFS->aQTaskInf);
}
int32_t tdRSmaFSTakeSnapshot(SSma *pSma, SRSmaFS *pFS) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma); SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv); SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
taosRLockLatch(RSMA_FS_LOCK(pStat)); taosRLockLatch(RSMA_FS_LOCK(pStat));
code = tdRSmaFSCopy(pSma, pFSOut); code = tdRSmaFSRef(pSma, pFS);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
taosRUnLockLatch(RSMA_FS_LOCK(pStat));
_exit: _exit:
if (code) {
taosRUnLockLatch(RSMA_FS_LOCK(pStat)); taosRUnLockLatch(RSMA_FS_LOCK(pStat));
if (code) {
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code)); smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code));
} }
return code; return code;
} }
int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFSOut) { int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFS) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma); SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv); SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
SRSmaFS *pFS = RSMA_FS(pStat); SRSmaFS *qFS = RSMA_FS(pStat);
int32_t size = taosArrayGetSize(pFS->aQTaskInf); int32_t size = taosArrayGetSize(qFS->aQTaskInf);
code = tdRSmaFSCreate(pFSOut, size); code = tdRSmaFSCreate(pFS, size);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
taosArraySetSize(pFSOut->aQTaskInf, size); taosArraySetSize(pFS->aQTaskInf, size);
memcpy(pFSOut->aQTaskInf->pData, pFS->aQTaskInf->pData, size * sizeof(SQTaskFile)); memcpy(pFS->aQTaskInf->pData, qFS->aQTaskInf->pData, size * sizeof(SQTaskFile));
_exit: _exit:
if (code) { if (code) {

13
source/dnode/vnode/src/sma/smaOpen.c
View File

@ -121,8 +121,6 @@ int smaSetKeepCfg(SVnode *pVnode, STsdbKeepCfg *pKeepCfg, STsdbCfg *pCfg, int ty
int32_t smaOpen(SVnode *pVnode, int8_t rollback) { int32_t smaOpen(SVnode *pVnode, int8_t rollback) {
STsdbCfg *pCfg = &pVnode->config.tsdbCfg; STsdbCfg *pCfg = &pVnode->config.tsdbCfg;
ASSERT(!pVnode->pSma);
SSma *pSma = taosMemoryCalloc(1, sizeof(SSma)); SSma *pSma = taosMemoryCalloc(1, sizeof(SSma));
if (!pSma) { if (!pSma) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -137,7 +135,7 @@ int32_t smaOpen(SVnode *pVnode, int8_t rollback) {
if (VND_IS_RSMA(pVnode)) { if (VND_IS_RSMA(pVnode)) {
STsdbKeepCfg keepCfg = {0}; STsdbKeepCfg keepCfg = {0};
for (int i = 0; i < TSDB_RETENTION_MAX; ++i) { for (int32_t i = 0; i < TSDB_RETENTION_MAX; ++i) {
if (i == TSDB_RETENTION_L0) { if (i == TSDB_RETENTION_L0) {
SMA_OPEN_RSMA_IMPL(pVnode, 0); SMA_OPEN_RSMA_IMPL(pVnode, 0);
} else if (i == TSDB_RETENTION_L1) { } else if (i == TSDB_RETENTION_L1) {
@ -145,7 +143,9 @@ int32_t smaOpen(SVnode *pVnode, int8_t rollback) {
} else if (i == TSDB_RETENTION_L2) { } else if (i == TSDB_RETENTION_L2) {
SMA_OPEN_RSMA_IMPL(pVnode, 2); SMA_OPEN_RSMA_IMPL(pVnode, 2);
} else { } else {
ASSERT(0); terrno = TSDB_CODE_APP_ERROR;
smaError("vgId:%d, sma open failed since %s, level:%d", TD_VID(pVnode), terrstr(), i);
goto _err;
} }
} }
@ -182,7 +182,10 @@ int32_t smaClose(SSma *pSma) {
* @return int32_t * @return int32_t
*/ */
int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback) { int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback) {
ASSERT(VND_IS_RSMA(pSma->pVnode)); if (!VND_IS_RSMA(pSma->pVnode)) {
terrno = TSDB_CODE_RSMA_INVALID_ENV;
return TSDB_CODE_FAILED;
}
return tdRSmaProcessRestoreImpl(pSma, type, committedVer, rollback); return tdRSmaProcessRestoreImpl(pSma, type, committedVer, rollback);
} }

4
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -1272,10 +1272,10 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
} }
// prepare // prepare
code = tdRSmaFSTakeSnapshot(pSma, &fs); code = tdRSmaFSCopy(pSma, &fs);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tdRSmaFSUpsertQTaskFile(&fs, qTaskFArray->pData, taosArrayGetSize(qTaskFArray)); code = tdRSmaFSUpsertQTaskFile(pSma, &fs, qTaskFArray->pData, taosArrayGetSize(qTaskFArray));
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tdRSmaFSPrepareCommit(pSma, &fs); code = tdRSmaFSPrepareCommit(pSma, &fs);

433
source/dnode/vnode/src/sma/smaSnapshot.c
View File

@ -17,14 +17,13 @@
static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppData); static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppData);
static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData); static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
static int32_t rsmaQTaskInfSnapReaderOpen(SRSmaSnapReader* pReader, int64_t version);
static int32_t rsmaQTaskInfSnapReaderClose(SQTaskFReader** ppReader);
// SRSmaSnapReader ======================================== // SRSmaSnapReader ========================================
struct SRSmaSnapReader { struct SRSmaSnapReader {
SSma* pSma; SSma* pSma;
int64_t sver; int64_t sver;
int64_t ever; int64_t ever;
SRSmaFS fs;
// for data file // for data file
int8_t rsmaDataDone[TSDB_RETENTION_L2]; int8_t rsmaDataDone[TSDB_RETENTION_L2];
@ -32,19 +31,23 @@ struct SRSmaSnapReader {
// for qtaskinfo file // for qtaskinfo file
int8_t qTaskDone; int8_t qTaskDone;
int32_t fsIter;
SQTaskFReader* pQTaskFReader; SQTaskFReader* pQTaskFReader;
}; };
int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapReader** ppReader) { int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapReader** ppReader) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0;
SVnode* pVnode = pSma->pVnode; SVnode* pVnode = pSma->pVnode;
SRSmaSnapReader* pReader = NULL; SRSmaSnapReader* pReader = NULL;
SSmaEnv* pEnv = SMA_RSMA_ENV(pSma);
SRSmaStat* pStat = (SRSmaStat*)SMA_ENV_STAT(pEnv);
// alloc // alloc
pReader = (SRSmaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader)); pReader = (SRSmaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));
if (pReader == NULL) { if (pReader == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
pReader->pSma = pSma; pReader->pSma = pSma;
pReader->sver = sver; pReader->sver = sver;
@ -55,174 +58,144 @@ int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapRead
if (pSma->pRSmaTsdb[i]) { if (pSma->pRSmaTsdb[i]) {
code = tsdbSnapReaderOpen(pSma->pRSmaTsdb[i], sver, ever, i == 0 ? SNAP_DATA_RSMA1 : SNAP_DATA_RSMA2, code = tsdbSnapReaderOpen(pSma->pRSmaTsdb[i], sver, ever, i == 0 ? SNAP_DATA_RSMA1 : SNAP_DATA_RSMA2,
&pReader->pDataReader[i]); &pReader->pDataReader[i]);
if (code < 0) { TSDB_CHECK_CODE(code, lino, _exit);
goto _err;
}
} }
} }
// open qtaskinfo // open qtaskinfo
if ((code = rsmaQTaskInfSnapReaderOpen(pReader, ever)) < 0) { taosRLockLatch(RSMA_FS_LOCK(pStat));
goto _err; code = tdRSmaFSRef(pSma, &pReader->fs);
} taosRUnLockLatch(RSMA_FS_LOCK(pStat));
TSDB_CHECK_CODE(code, lino, _exit);
*ppReader = pReader;
return TSDB_CODE_SUCCESS;
_err:
if (pReader) rsmaSnapReaderClose(&pReader);
*ppReader = NULL;
smaError("vgId:%d, vnode snapshot rsma reader open failed since %s", TD_VID(pVnode), tstrerror(code));
return TSDB_CODE_FAILED;
}
static int32_t rsmaQTaskInfSnapReaderOpen(SRSmaSnapReader* pReader, int64_t version) {
#if 0
int32_t code = 0;
SSma* pSma = pReader->pSma;
SVnode* pVnode = pSma->pVnode;
SSmaEnv* pEnv = NULL;
SRSmaStat* pStat = NULL;
if (!(pEnv = SMA_RSMA_ENV(pSma))) {
smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as env is NULL",
TD_VID(pVnode), version);
return TSDB_CODE_SUCCESS;
}
pStat = (SRSmaStat*)SMA_ENV_STAT(pEnv);
int32_t ref = tdRSmaFSRef(pReader->pSma, pStat, version);
if (ref < 1) {
smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as ref is %d",
TD_VID(pVnode), version, ref);
return TSDB_CODE_SUCCESS;
}
char qTaskInfoFullName[TSDB_FILENAME_LEN];
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), version, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
if (!taosCheckExistFile(qTaskInfoFullName)) {
tdRSmaFSUnRef(pSma, pStat, version);
smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as %s not exist",
TD_VID(pVnode), version, qTaskInfoFullName);
return TSDB_CODE_SUCCESS;
}
if (taosArrayGetSize(pReader->fs.aQTaskInf) > 0) {
pReader->pQTaskFReader = taosMemoryCalloc(1, sizeof(SQTaskFReader)); pReader->pQTaskFReader = taosMemoryCalloc(1, sizeof(SQTaskFReader));
if (!pReader->pQTaskFReader) { if (!pReader->pQTaskFReader) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _end; TSDB_CHECK_CODE(code, lino, _exit);
} }
TdFilePtr fp = taosOpenFile(qTaskInfoFullName, TD_FILE_READ);
if (!fp) {
code = TAOS_SYSTEM_ERROR(errno);
taosMemoryFreeClear(pReader->pQTaskFReader);
goto _end;
}
pReader->pQTaskFReader->pReadH = fp;
pReader->pQTaskFReader->pSma = pSma; pReader->pQTaskFReader->pSma = pSma;
pReader->pQTaskFReader->version = pReader->ever; pReader->pQTaskFReader->version = pReader->ever;
_end:
if (code < 0) {
tdRSmaFSUnRef(pSma, pStat, version);
smaError("vgId:%d, vnode snapshot rsma reader open %s succeed", TD_VID(pVnode), qTaskInfoFullName);
return TSDB_CODE_FAILED;
} }
smaInfo("vgId:%d, vnode snapshot rsma reader open %s succeed", TD_VID(pVnode), qTaskInfoFullName); *ppReader = pReader;
#endif _exit:
return TSDB_CODE_SUCCESS; if (code) {
if (pReader) rsmaSnapReaderClose(&pReader);
*ppReader = NULL;
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
} }
return code;
static int32_t rsmaQTaskInfSnapReaderClose(SQTaskFReader** ppReader) {
#if 0
if (!(*ppReader)) {
return TSDB_CODE_SUCCESS;
}
SSma* pSma = (*ppReader)->pSma;
SRSmaStat* pStat = SMA_RSMA_STAT(pSma);
int64_t version = (*ppReader)->version;
taosCloseFile(&(*ppReader)->pReadH);
tdRSmaFSUnRef(pSma, pStat, version);
taosMemoryFreeClear(*ppReader);
smaInfo("vgId:%d, vnode snapshot rsma reader closed for qTaskInfo version %" PRIi64, SMA_VID(pSma), version);
#endif
return TSDB_CODE_SUCCESS;
} }
static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppBuf) { static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppBuf) {
int32_t code = 0; int32_t code = 0;
SSma* pSma = pReader->pSma; int32_t lino = 0;
SVnode* pVnode = pReader->pSma->pVnode;
SQTaskFReader* qReader = pReader->pQTaskFReader;
SRSmaFS* pFS = &pReader->fs;
int64_t n = 0; int64_t n = 0;
uint8_t* pBuf = NULL; uint8_t* pBuf = NULL;
SQTaskFReader* qReader = pReader->pQTaskFReader; int64_t version = pReader->ever;
char fname[TSDB_FILENAME_LEN];
if (!qReader) { if (!qReader) {
*ppBuf = NULL; *ppBuf = NULL;
smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, qTaskReader is NULL", SMA_VID(pSma)); smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, qTaskReader is NULL", TD_VID(pVnode));
return 0; return 0;
} }
if (pReader->fsIter >= taosArrayGetSize(pFS->aQTaskInf)) {
*ppBuf = NULL;
smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, fsIter reach end", TD_VID(pVnode));
return 0;
}
while (pReader->fsIter < taosArrayGetSize(pFS->aQTaskInf)) {
SQTaskFile* qTaskF = taosArrayGet(pFS->aQTaskInf, pReader->fsIter);
if (qTaskF->version != version) {
continue;
}
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), qTaskF->suid, qTaskF->level, version, tfsGetPrimaryPath(pVnode->pTfs),
fname);
if (!taosCheckExistFile(fname)) {
smaWarn("vgId:%d, vnode snapshot rsma reader for qtaskinfo, table %" PRIi64 ", level %" PRIi8 ", version %" PRIi64
" is not needed as %s not exist",
TD_VID(pVnode), qTaskF->suid, qTaskF->level, version, fname);
continue;
}
TdFilePtr fp = taosOpenFile(fname, TD_FILE_READ);
if (!fp) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
}
qReader->pReadH = fp;
qReader->level = qTaskF->level;
qReader->suid = qTaskF->suid;
}
if (!qReader->pReadH) { if (!qReader->pReadH) {
*ppBuf = NULL; *ppBuf = NULL;
smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, readh is NULL", SMA_VID(pSma)); smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, not needed since readh is NULL", TD_VID(pVnode));
return 0; return 0;
} }
int64_t size = 0; int64_t size = 0;
if (taosFStatFile(qReader->pReadH, &size, NULL) < 0) { if (taosFStatFile(qReader->pReadH, &size, NULL) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
// seek // seek
if (taosLSeekFile(qReader->pReadH, 0, SEEK_SET) < 0) { if (taosLSeekFile(qReader->pReadH, 0, SEEK_SET) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
ASSERT(!(*ppBuf)); if (*ppBuf) {
// alloc *ppBuf = taosMemoryRealloc(*ppBuf, sizeof(SSnapDataHdr) + size);
*ppBuf = taosMemoryCalloc(1, sizeof(SSnapDataHdr) + size); } else {
*ppBuf = taosMemoryMalloc(sizeof(SSnapDataHdr) + size);
}
if (!(*ppBuf)) { if (!(*ppBuf)) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
// read // read
n = taosReadFile(qReader->pReadH, POINTER_SHIFT(*ppBuf, sizeof(SSnapDataHdr)), size); n = taosReadFile(qReader->pReadH, POINTER_SHIFT(*ppBuf, sizeof(SSnapDataHdr)), size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} else if (n != size) { } else if (n != size) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
smaInfo("vgId:%d, vnode snapshot rsma read qtaskinfo, size:%" PRIi64, SMA_VID(pSma), size); smaInfo("vgId:%d, vnode snapshot rsma read qtaskinfo, version:%" PRIi64 ", size:%" PRIi64, TD_VID(pVnode), version,
size);
SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppBuf); SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppBuf);
pHdr->type = SNAP_DATA_QTASK; pHdr->type = SNAP_DATA_QTASK;
pHdr->flag = qReader->level;
pHdr->index = qReader->suid;
pHdr->size = size; pHdr->size = size;
_exit: _exit:
smaInfo("vgId:%d, vnode snapshot rsma read qtaskinfo succeed", SMA_VID(pSma)); if (code) {
return code;
_err:
*ppBuf = NULL; *ppBuf = NULL;
smaError("vgId:%d, vnode snapshot rsma read qtaskinfo failed since %s", SMA_VID(pSma), tstrerror(code)); smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
} else {
smaInfo("vgId:%d, vnode snapshot rsma read qtaskinfo succeed", TD_VID(pVnode));
}
return code; return code;
} }
int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData) { int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0;
*ppData = NULL; *ppData = NULL;
@ -236,15 +209,12 @@ int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData) {
if (!pReader->rsmaDataDone[i]) { if (!pReader->rsmaDataDone[i]) {
smaInfo("vgId:%d, vnode snapshot rsma read level %d not done", SMA_VID(pReader->pSma), i); smaInfo("vgId:%d, vnode snapshot rsma read level %d not done", SMA_VID(pReader->pSma), i);
code = tsdbSnapRead(pTsdbSnapReader, ppData); code = tsdbSnapRead(pTsdbSnapReader, ppData);
if (code) { TSDB_CHECK_CODE(code, lino, _exit);
goto _err;
} else {
if (*ppData) { if (*ppData) {
goto _exit; goto _exit;
} else { } else {
pReader->rsmaDataDone[i] = 1; pReader->rsmaDataDone[i] = 1;
} }
}
} else { } else {
smaInfo("vgId:%d, vnode snapshot rsma read level %d is done", SMA_VID(pReader->pSma), i); smaInfo("vgId:%d, vnode snapshot rsma read level %d is done", SMA_VID(pReader->pSma), i);
} }
@ -254,22 +224,20 @@ int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData) {
if (!pReader->qTaskDone) { if (!pReader->qTaskDone) {
smaInfo("vgId:%d, vnode snapshot rsma qtaskinfo not done", SMA_VID(pReader->pSma)); smaInfo("vgId:%d, vnode snapshot rsma qtaskinfo not done", SMA_VID(pReader->pSma));
code = rsmaSnapReadQTaskInfo(pReader, ppData); code = rsmaSnapReadQTaskInfo(pReader, ppData);
if (code) { TSDB_CHECK_CODE(code, lino, _exit);
goto _err;
} else {
pReader->qTaskDone = 1;
if (*ppData) { if (*ppData) {
goto _exit; goto _exit;
} } else {
pReader->qTaskDone = 1;
} }
} }
_exit: _exit:
smaInfo("vgId:%d, vnode snapshot rsma read succeed", SMA_VID(pReader->pSma)); if (code) {
return code;
_err:
smaError("vgId:%d, vnode snapshot rsma read failed since %s", SMA_VID(pReader->pSma), tstrerror(code)); smaError("vgId:%d, vnode snapshot rsma read failed since %s", SMA_VID(pReader->pSma), tstrerror(code));
} else {
smaInfo("vgId:%d, vnode snapshot rsma read succeed", SMA_VID(pReader->pSma));
}
return code; return code;
} }
@ -277,14 +245,18 @@ int32_t rsmaSnapReaderClose(SRSmaSnapReader** ppReader) {
int32_t code = 0; int32_t code = 0;
SRSmaSnapReader* pReader = *ppReader; SRSmaSnapReader* pReader = *ppReader;
tdRSmaFSUnRef(pReader->pSma, &pReader->fs);
if (pReader->pQTaskFReader) {
taosCloseFile(&pReader->pQTaskFReader->pReadH);
taosMemoryFree(pReader->pQTaskFReader);
}
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) { for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pReader->pDataReader[i]) { if (pReader->pDataReader[i]) {
tsdbSnapReaderClose(&pReader->pDataReader[i]); tsdbSnapReaderClose(&pReader->pDataReader[i]);
} }
} }
rsmaQTaskInfSnapReaderClose(&pReader->pQTaskFReader);
smaInfo("vgId:%d, vnode snapshot rsma reader closed", SMA_VID(pReader->pSma)); smaInfo("vgId:%d, vnode snapshot rsma reader closed", SMA_VID(pReader->pSma));
taosMemoryFreeClear(*ppReader); taosMemoryFreeClear(*ppReader);
@ -296,29 +268,23 @@ struct SRSmaSnapWriter {
SSma* pSma; SSma* pSma;
int64_t sver; int64_t sver;
int64_t ever; int64_t ever;
SRSmaFS fs;
// config
int64_t commitID;
// for data file // for data file
STsdbSnapWriter* pDataWriter[TSDB_RETENTION_L2]; STsdbSnapWriter* pDataWriter[TSDB_RETENTION_L2];
// for qtaskinfo file
SQTaskFReader* pQTaskFReader;
SQTaskFWriter* pQTaskFWriter;
}; };
int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter) { int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter) {
int32_t code = 0; int32_t code = 0;
#if 0 int32_t lino = 0;
SRSmaSnapWriter* pWriter = NULL;
SVnode* pVnode = pSma->pVnode; SVnode* pVnode = pSma->pVnode;
SRSmaSnapWriter* pWriter = NULL;
// alloc // alloc
pWriter = (SRSmaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter)); pWriter = (SRSmaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
if (pWriter == NULL) { if (!pWriter) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
pWriter->pSma = pSma; pWriter->pSma = pSma;
pWriter->sver = sver; pWriter->sver = sver;
@ -328,104 +294,103 @@ int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWrit
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) { for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pSma->pRSmaTsdb[i]) { if (pSma->pRSmaTsdb[i]) {
code = tsdbSnapWriterOpen(pSma->pRSmaTsdb[i], sver, ever, &pWriter->pDataWriter[i]); code = tsdbSnapWriterOpen(pSma->pRSmaTsdb[i], sver, ever, &pWriter->pDataWriter[i]);
if (code < 0) { TSDB_CHECK_CODE(code, lino, _exit);
goto _err;
}
} }
} }
// qtaskinfo // qtaskinfo
SQTaskFWriter* qWriter = (SQTaskFWriter*)taosMemoryCalloc(1, sizeof(SQTaskFWriter)); code = tdRSmaFSCopy(pSma, &pWriter->fs);
qWriter->pSma = pSma; TSDB_CHECK_CODE(code, lino, _exit);
char qTaskInfoFullName[TSDB_FILENAME_LEN];
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), 0, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
TdFilePtr qTaskF = taosCreateFile(qTaskInfoFullName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (!qTaskF) {
taosMemoryFree(qWriter);
code = TAOS_SYSTEM_ERROR(errno);
smaError("vgId:%d, rsma snapshot writer open %s failed since %s", TD_VID(pSma->pVnode), qTaskInfoFullName,
tstrerror(code));
goto _err;
}
qWriter->pWriteH = qTaskF;
int32_t fnameLen = strlen(qTaskInfoFullName) + 1;
qWriter->fname = taosMemoryCalloc(1, fnameLen);
strncpy(qWriter->fname, qTaskInfoFullName, fnameLen);
pWriter->pQTaskFWriter = qWriter;
smaDebug("vgId:%d, rsma snapshot writer open succeed for %s", TD_VID(pSma->pVnode), qTaskInfoFullName);
// snapWriter // snapWriter
*ppWriter = pWriter; *ppWriter = pWriter;
_exit:
smaInfo("vgId:%d, rsma snapshot writer open succeed", TD_VID(pSma->pVnode)); if (code) {
return code;
_err:
smaError("vgId:%d, rsma snapshot writer open failed since %s", TD_VID(pSma->pVnode), tstrerror(code));
if (pWriter) rsmaSnapWriterClose(&pWriter, 0); if (pWriter) rsmaSnapWriterClose(&pWriter, 0);
*ppWriter = NULL; *ppWriter = NULL;
#endif smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
} else {
smaInfo("vgId:%d, rsma snapshot writer open succeed", TD_VID(pSma->pVnode));
}
return code;
}
int32_t rsmaSnapWriterPrepareClose(SRSmaSnapWriter* pWriter) {
int32_t code = 0;
int32_t lino = 0;
// rsmaSnapWriterClose
code = tdRSmaFSPrepareCommit(pWriter->pSma, &pWriter->fs);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pWriter->pSma), __func__, lino, tstrerror(code));
}
return code; return code;
} }
int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback) { int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback) {
int32_t code = 0; int32_t code = 0;
#if 0 int32_t lino = 0;
SSma* pSma = NULL;
SSmaEnv* pEnv = NULL;
SRSmaStat* pStat = NULL;
SRSmaSnapWriter* pWriter = *ppWriter; SRSmaSnapWriter* pWriter = *ppWriter;
SVnode* pVnode = pWriter->pSma->pVnode;
if (rollback) { if (!pWriter) {
// TODO: rsma1/rsma2 goto _exit;
// qtaskinfo
if (pWriter->pQTaskFWriter) {
if (taosRemoveFile(pWriter->pQTaskFWriter->fname) != 0) {
smaWarn("vgId:%d, vnode snapshot rsma writer failed to remove %s since %s", SMA_VID(pWriter->pSma),
pWriter->pQTaskFWriter->fname ? pWriter->pQTaskFWriter->fname : "NULL",
tstrerror(TAOS_SYSTEM_ERROR(errno)));
} }
}
} else { pSma = pWriter->pSma;
pEnv = SMA_RSMA_ENV(pSma);
pStat = (SRSmaStat*)SMA_ENV_STAT(pEnv);
// rsma1/rsma2 // rsma1/rsma2
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) { for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pWriter->pDataWriter[i]) { if (pWriter->pDataWriter[i]) {
code = tsdbSnapWriterClose(&pWriter->pDataWriter[i], rollback); code = tsdbSnapWriterClose(&pWriter->pDataWriter[i], rollback);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
} }
// qtaskinfo // qtaskinfo
if (pWriter->pQTaskFWriter) { if (rollback) {
char qTaskInfoFullName[TSDB_FILENAME_LEN]; tdRSmaFSRollback(pSma);
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pWriter->ever, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName); // remove qTaskFiles
if (taosRenameFile(pWriter->pQTaskFWriter->fname, qTaskInfoFullName) < 0) { } else {
code = TAOS_SYSTEM_ERROR(errno); // lock
goto _err; taosWLockLatch(RSMA_FS_LOCK(pStat));
code = tdRSmaFSCommit(pSma);
if (code) {
taosWUnLockLatch(RSMA_FS_LOCK(pStat));
goto _exit;
}
// unlock
taosWUnLockLatch(RSMA_FS_LOCK(pStat));
} }
smaInfo("vgId:%d, vnode snapshot rsma writer rename %s to %s", SMA_VID(pWriter->pSma),
pWriter->pQTaskFWriter->fname, qTaskInfoFullName);
// rsma restore // rsma restore
int8_t rollback = 0; code = tdRSmaRestore(pWriter->pSma, RSMA_RESTORE_SYNC, pWriter->ever, rollback);
if ((code = tdRSmaRestore(pWriter->pSma, RSMA_RESTORE_SYNC, pWriter->ever, rollback)) < 0) { TSDB_CHECK_CODE(code, lino, _exit);
goto _err; smaInfo("vgId:%d, vnode snapshot rsma writer restore from sync succeed", SMA_VID(pSma));
}
smaInfo("vgId:%d, vnode snapshot rsma writer restore from %s succeed", SMA_VID(pWriter->pSma), qTaskInfoFullName);
}
}
smaInfo("vgId:%d, vnode snapshot rsma writer close succeed", SMA_VID(pWriter->pSma)); _exit:
taosMemoryFree(pWriter); if (pWriter) taosMemoryFree(pWriter);
*ppWriter = NULL; *ppWriter = NULL;
return code; if (code) {
smaError("vgId:%d, vnode snapshot rsma writer close failed since %s", SMA_VID(pSma), tstrerror(code));
} else {
smaInfo("vgId:%d, vnode snapshot rsma writer close succeed", SMA_VID(pSma));
}
_err:
smaError("vgId:%d, vnode snapshot rsma writer close failed since %s", SMA_VID(pWriter->pSma), tstrerror(code));
#endif
return code; return code;
} }
int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0;
SSnapDataHdr* pHdr = (SSnapDataHdr*)pData; SSnapDataHdr* pHdr = (SSnapDataHdr*)pData;
// rsma1/rsma2 // rsma1/rsma2
@ -438,42 +403,78 @@ int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData)
} else if (pHdr->type == SNAP_DATA_QTASK) { } else if (pHdr->type == SNAP_DATA_QTASK) {
code = rsmaSnapWriteQTaskInfo(pWriter, pData, nData); code = rsmaSnapWriteQTaskInfo(pWriter, pData, nData);
} else { } else {
ASSERT(0); code = TSDB_CODE_RSMA_FS_SYNC;
} }
if (code < 0) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
_exit: _exit:
if (code) {
smaError("vgId:%d, %s failed at line %d since %s, data type %" PRIi8, SMA_VID(pWriter->pSma), __func__, lino,
tstrerror(code), pHdr->type);
} else {
smaInfo("vgId:%d, rsma snapshot write for data type %" PRIi8 " succeed", SMA_VID(pWriter->pSma), pHdr->type); smaInfo("vgId:%d, rsma snapshot write for data type %" PRIi8 " succeed", SMA_VID(pWriter->pSma), pHdr->type);
return code; }
_err:
smaError("vgId:%d, rsma snapshot write for data type %" PRIi8 " failed since %s", SMA_VID(pWriter->pSma), pHdr->type,
tstrerror(code));
return code; return code;
} }
static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
int32_t code = 0; int32_t code = 0;
SQTaskFWriter* qWriter = pWriter->pQTaskFWriter; int32_t lino = 0;
SSma* pSma = pWriter->pSma;
if (qWriter && qWriter->pWriteH) { SVnode* pVnode = pSma->pVnode;
char fname[TSDB_FILENAME_LEN];
TdFilePtr fp = NULL;
SSnapDataHdr* pHdr = (SSnapDataHdr*)pData; SSnapDataHdr* pHdr = (SSnapDataHdr*)pData;
int64_t size = pHdr->size;
ASSERT(size == (nData - sizeof(SSnapDataHdr))); fname[0] = '\0';
int64_t contLen = taosWriteFile(qWriter->pWriteH, pHdr->data, size);
if (contLen != size) { if (pHdr->size != (nData - sizeof(SSnapDataHdr))) {
code = TSDB_CODE_RSMA_FS_SYNC;
TSDB_CHECK_CODE(code, lino, _exit);
}
SQTaskFile qTaskFile = {
.nRef = 1, .level = pHdr->flag, .suid = pHdr->index, .version = pWriter->ever, .size = pHdr->size};
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pHdr->index, pHdr->flag, qTaskFile.version,
tfsGetPrimaryPath(pVnode->pTfs), fname);
fp = taosCreateFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (!fp) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
smaInfo("vgId:%d, vnode snapshot rsma write qtaskinfo %s succeed", SMA_VID(pWriter->pSma), qWriter->fname);
} else { int64_t contLen = taosWriteFile(fp, pHdr->data, pHdr->size);
smaInfo("vgId:%d, vnode snapshot rsma write qtaskinfo is not needed", SMA_VID(pWriter->pSma)); if (contLen != pHdr->size) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
} }
uint32_t mtime = 0;
if (taosFStatFile(fp, NULL, &mtime) == 0) {
qTaskFile.mtime = mtime;
}
if (taosFsyncFile(fp) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
}
taosCloseFile(&fp);
code = tdRSmaFSUpsertQTaskFile(pSma, &pWriter->fs, &qTaskFile, 1);
TSDB_CHECK_CODE(code, lino, _exit);
_exit: _exit:
return code; if (code) {
if (fp) {
(void)taosRemoveFile(fname);
}
smaError("vgId:%d, %s failed at line %d since %s, file:%s", TD_VID(pVnode), __func__, lino, tstrerror(code), fname);
} else {
smaInfo("vgId:%d, vnode snapshot rsma write qtaskinfo %s succeed", TD_VID(pVnode), fname);
}
_err:
smaError("vgId:%d, vnode snapshot rsma write qtaskinfo failed since %s", SMA_VID(pWriter->pSma), tstrerror(code));
return code; return code;
} }

34
source/dnode/vnode/src/sma/smaTimeRange.c
View File

@ -111,35 +111,48 @@ _err:
* @return int32_t * @return int32_t
*/ */
static int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) { static int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
int32_t code = 0;
int32_t lino = 0;
SSmaCfg *pCfg = (SSmaCfg *)pMsg; SSmaCfg *pCfg = (SSmaCfg *)pMsg;
SVCreateStbReq pReq = {0};
if (TD_VID(pSma->pVnode) == pCfg->dstVgId) { if (TD_VID(pSma->pVnode) == pCfg->dstVgId) {
// create tsma meta in dstVgId // create tsma meta in dstVgId
if (metaCreateTSma(SMA_META(pSma), version, pCfg) < 0) { if (metaCreateTSma(SMA_META(pSma), version, pCfg) < 0) {
return -1; code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
} }
// create stable to save tsma result in dstVgId // create stable to save tsma result in dstVgId
SName stbFullName = {0}; SName stbFullName = {0};
tNameFromString(&stbFullName, pCfg->dstTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE); tNameFromString(&stbFullName, pCfg->dstTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
SVCreateStbReq pReq = {0};
pReq.name = (char *)tNameGetTableName(&stbFullName); pReq.name = (char *)tNameGetTableName(&stbFullName);
pReq.suid = pCfg->dstTbUid; pReq.suid = pCfg->dstTbUid;
pReq.schemaRow = pCfg->schemaRow; pReq.schemaRow = pCfg->schemaRow;
pReq.schemaTag = pCfg->schemaTag; pReq.schemaTag = pCfg->schemaTag;
if (metaCreateSTable(SMA_META(pSma), version, &pReq) < 0) { if (metaCreateSTable(SMA_META(pSma), version, &pReq) < 0) {
return -1; code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
}
} else {
code = terrno = TSDB_CODE_TSMA_INVALID_STAT;
TSDB_CHECK_CODE(code, lino, _exit);
} }
_exit:
if (code) {
smaError("vgId:%d, failed at line %d to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64
" dstTb:%s dstVg:%d",
SMA_VID(pSma), lino, pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name,
pCfg->dstVgId);
} else {
smaDebug("vgId:%d, success to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64 smaDebug("vgId:%d, success to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64
" dstTb:%s dstVg:%d", " dstTb:%s dstVg:%d",
SMA_VID(pSma), pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name, pCfg->dstVgId); SMA_VID(pSma), pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name, pCfg->dstVgId);
} else {
ASSERT(0);
} }
return 0; return code;
} }
/** /**
@ -174,7 +187,7 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
STSmaStat *pTsmaStat = NULL; STSmaStat *pTsmaStat = NULL;
if (!pEnv || !(pStat = SMA_ENV_STAT(pEnv))) { if (!pEnv || !(pStat = SMA_ENV_STAT(pEnv))) {
terrno = TSDB_CODE_TSMA_INVALID_STAT; terrno = TSDB_CODE_TSMA_INVALID_ENV;
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
@ -218,7 +231,12 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
} }
#if 0 #if 0
ASSERT(!strncasecmp("td.tsma.rst.tb", pTsmaStat->pTSma->dstTbName, 14)); if (!strncasecmp("td.tsma.rst.tb", pTsmaStat->pTSma->dstTbName, 14)) {
terrno = TSDB_CODE_APP_ERROR;
smaError("vgId:%d, tsma insert for smaIndex %" PRIi64 " failed since %s, %s", SMA_VID(pSma), indexUid,
pTsmaStat->pTSma->indexUid, tstrerror(terrno), pTsmaStat->pTSma->dstTbName);
goto _err;
}
#endif #endif
SRpcMsg submitReqMsg = { SRpcMsg submitReqMsg = {

68
source/dnode/vnode/src/sma/smaUtil.c
View File

@ -98,74 +98,6 @@ void tdRSmaGetDirName(int32_t vgId, const char *pdname, const char *dname, bool
} }
} }
#if 0
int32_t tdInitTFile(STFile *pTFile, const char *dname, const char *fname) {
TD_TFILE_SET_STATE(pTFile, TD_FILE_STATE_OK);
TD_TFILE_SET_CLOSED(pTFile);
memset(&(pTFile->info), 0, sizeof(pTFile->info));
pTFile->info.magic = TD_FILE_INIT_MAGIC;
char tmpName[TSDB_FILENAME_LEN * 2 + 32] = {0};
snprintf(tmpName, TSDB_FILENAME_LEN * 2 + 32, "%s%s%s", dname, TD_DIRSEP, fname);
int32_t tmpNameLen = strlen(tmpName) + 1;
pTFile->fname = taosMemoryMalloc(tmpNameLen);
if (!pTFile->fname) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tstrncpy(pTFile->fname, tmpName, tmpNameLen);
return 0;
}
int32_t tdCreateTFile(STFile *pTFile, bool updateHeader, int8_t fType) {
ASSERT(pTFile->info.fsize == 0 && pTFile->info.magic == TD_FILE_INIT_MAGIC);
pTFile->pFile = taosOpenFile(TD_TFILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pTFile->pFile == NULL) {
if (errno == ENOENT) {
// Try to create directory recursively
char *s = strdup(TD_TFILE_FULL_NAME(pTFile));
if (taosMulMkDir(taosDirName(s)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosMemoryFree(s);
return -1;
}
taosMemoryFree(s);
pTFile->pFile = taosOpenFile(TD_TFILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pTFile->pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
}
}
if (!updateHeader) {
return 0;
}
pTFile->info.fsize += TD_FILE_HEAD_SIZE;
pTFile->info.fver = 0;
if (tdUpdateTFileHeader(pTFile) < 0) {
tdCloseTFile(pTFile);
tdRemoveTFile(pTFile);
return -1;
}
return 0;
}
int32_t tdRemoveTFile(STFile *pTFile) {
if (taosRemoveFile(TD_TFILE_FULL_NAME(pTFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
};
return 0;
}
#endif
// smaXXXUtil ================ // smaXXXUtil ================
void *tdAcquireSmaRef(int32_t rsetId, int64_t refId) { void *tdAcquireSmaRef(int32_t rsetId, int64_t refId) {
void *pResult = taosAcquireRef(rsetId, refId); void *pResult = taosAcquireRef(rsetId, refId);

5
source/util/src/terror.c
View File

@ -585,7 +585,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_NO_INDEX_IN_CACHE, "No tsma index in ca
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_ENV, "Invalid rsma env") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_ENV, "Invalid rsma env")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_STAT, "Invalid rsma state") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_STAT, "Invalid rsma state")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_QTASKINFO_CREATE, "Rsma qtaskinfo creation error") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_QTASKINFO_CREATE, "Rsma qtaskinfo creation error")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FS_COMMIT, "Invalid rsma fs commit") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FS_COMMIT, "Rsma fs commit error")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_REMOVE_EXISTS, "Rsma remove exists") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_REMOVE_EXISTS, "Rsma remove exists")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP, "Rsma fetch msg is messed up") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP, "Rsma fetch msg is messed up")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_EMPTY_INFO, "Rsma info is empty") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_EMPTY_INFO, "Rsma info is empty")
@ -593,6 +593,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_SCHEMA, "Rsma invalid schema
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_REGEX_MATCH, "Rsma regex match") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_REGEX_MATCH, "Rsma regex match")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_STREAM_STATE_OPEN, "Rsma stream state open") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_STREAM_STATE_OPEN, "Rsma stream state open")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_STREAM_STATE_COMMIT, "Rsma stream state commit") TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_STREAM_STATE_COMMIT, "Rsma stream state commit")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FS_REF, "Rsma fs ref error")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FS_SYNC, "Rsma fs sync error")
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FS_UPDATE, "Rsma fs update error")
//index //index
TAOS_DEFINE_ERROR(TSDB_CODE_INDEX_REBUILDING, "Index is rebuilding") TAOS_DEFINE_ERROR(TSDB_CODE_INDEX_REBUILDING, "Index is rebuilding")