chore: more code
This commit is contained in:
kailixu
parent
67461d92a7
commit
6ee135cb52
|
|
@ -1,638 +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/>.
|
||||
*/
|
||||
|
||||
#include "sma.h"
|
||||
#include "vnd.h"
|
||||
|
||||
// =================================================================================================
|
||||
|
||||
// static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **output);
|
||||
static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2);
|
||||
|
||||
static FORCE_INLINE int32_t tPutQTaskF(uint8_t *p, SQTaskFile *pFile) {
|
||||
int32_t n = 0;
|
||||
|
||||
n += tPutI8(p ? p + n : p, pFile->level);
|
||||
n += tPutI64v(p ? p + n : p, pFile->size);
|
||||
n += tPutI64v(p ? p + n : p, pFile->suid);
|
||||
n += tPutI64v(p ? p + n : p, pFile->version);
|
||||
n += tPutI64v(p ? p + n : p, pFile->mtime);
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
static int32_t tdRSmaFSToBinary(uint8_t *p, SRSmaFS *pFS) {
|
||||
int32_t n = 0;
|
||||
uint32_t size = taosArrayGetSize(pFS->aQTaskInf);
|
||||
|
||||
// version
|
||||
n += tPutI8(p ? p + n : p, 0);
|
||||
|
||||
// SArray<SQTaskFile>
|
||||
n += tPutU32v(p ? p + n : p, size);
|
||||
for (uint32_t i = 0; i < size; ++i) {
|
||||
n += tPutQTaskF(p ? p + n : p, taosArrayGet(pFS->aQTaskInf, i));
|
||||
}
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
int32_t tdRSmaGetQTaskF(uint8_t *p, SQTaskFile *pFile) {
|
||||
int32_t n = 0;
|
||||
|
||||
n += tGetI8(p + n, &pFile->level);
|
||||
n += tGetI64v(p + n, &pFile->size);
|
||||
n += tGetI64v(p + n, &pFile->suid);
|
||||
n += tGetI64v(p + n, &pFile->version);
|
||||
n += tGetI64v(p + n, &pFile->mtime);
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
static int32_t tsdbBinaryToFS(uint8_t *pData, int64_t nData, SRSmaFS *pFS) {
|
||||
int32_t code = 0;
|
||||
int32_t n = 0;
|
||||
int8_t version = 0;
|
||||
|
||||
// version
|
||||
n += tGetI8(pData + n, &version);
|
||||
|
||||
// SArray<SQTaskFile>
|
||||
taosArrayClear(pFS->aQTaskInf);
|
||||
uint32_t size = 0;
|
||||
n += tGetU32v(pData + n, &size);
|
||||
for (uint32_t i = 0; i < size; ++i) {
|
||||
SQTaskFile qTaskF = {0};
|
||||
|
||||
int32_t nt = tdRSmaGetQTaskF(pData + n, &qTaskF);
|
||||
if (nt < 0) {
|
||||
code = TSDB_CODE_FILE_CORRUPTED;
|
||||
goto _exit;
|
||||
}
|
||||
|
||||
n += nt;
|
||||
if (taosArrayPush(pFS->aQTaskInf, &qTaskF) == NULL) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
goto _exit;
|
||||
}
|
||||
}
|
||||
|
||||
if (ASSERTS(n + sizeof(TSCKSUM) == nData, "n:%d + sizeof(TSCKSUM):%d != nData:%d", n, (int32_t)sizeof(TSCKSUM),
|
||||
nData)) {
|
||||
code = TSDB_CODE_FILE_CORRUPTED;
|
||||
goto _exit;
|
||||
}
|
||||
|
||||
_exit:
|
||||
return code;
|
||||
}
|
||||
|
||||
static int32_t tdRSmaSaveFSToFile(SRSmaFS *pFS, const char *fname) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
|
||||
// encode to binary
|
||||
int32_t size = tdRSmaFSToBinary(NULL, pFS) + sizeof(TSCKSUM);
|
||||
uint8_t *pData = taosMemoryMalloc(size);
|
||||
if (pData == NULL) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
tdRSmaFSToBinary(pData, pFS);
|
||||
taosCalcChecksumAppend(0, pData, size);
|
||||
|
||||
// save to file
|
||||
TdFilePtr pFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
|
||||
if (pFD == NULL) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
int64_t n = taosWriteFile(pFD, pData, size);
|
||||
if (n < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
taosCloseFile(&pFD);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
if (taosFsyncFile(pFD) < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
taosCloseFile(&pFD);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
taosCloseFile(&pFD);
|
||||
|
||||
_exit:
|
||||
if (pData) taosMemoryFree(pData);
|
||||
if (code) {
|
||||
smaError("%s failed at line %d since %s, fname:%s", __func__, lino, tstrerror(code), fname);
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
static int32_t tdRSmaFSCreate(SRSmaFS *pFS, int32_t size) {
|
||||
int32_t code = 0;
|
||||
|
||||
pFS->aQTaskInf = taosArrayInit(size, sizeof(SQTaskFile));
|
||||
if (pFS->aQTaskInf == NULL) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
goto _exit;
|
||||
}
|
||||
|
||||
_exit:
|
||||
return code;
|
||||
}
|
||||
|
||||
static void tdRSmaGetCurrentFName(SSma *pSma, char *current, char *current_t) {
|
||||
SVnode *pVnode = pSma->pVnode;
|
||||
int32_t offset = 0;
|
||||
|
||||
vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, current, TSDB_FILENAME_LEN);
|
||||
offset = strlen(current);
|
||||
snprintf(current + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s%sPRESENT", TD_DIRSEP, VNODE_RSMA_DIR, TD_DIRSEP);
|
||||
|
||||
vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, current_t, TSDB_FILENAME_LEN);
|
||||
offset = strlen(current_t);
|
||||
snprintf(current_t + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s%sPRESENT.t", TD_DIRSEP, VNODE_RSMA_DIR, TD_DIRSEP);
|
||||
}
|
||||
|
||||
static int32_t tdRSmaLoadFSFromFile(const char *fname, SRSmaFS *pFS) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
uint8_t *pData = NULL;
|
||||
|
||||
// load binary
|
||||
TdFilePtr pFD = taosOpenFile(fname, TD_FILE_READ);
|
||||
if (pFD == NULL) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
int64_t size;
|
||||
if (taosFStatFile(pFD, &size, NULL) < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
taosCloseFile(&pFD);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
pData = taosMemoryMalloc(size);
|
||||
if (pData == NULL) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
taosCloseFile(&pFD);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
if (taosReadFile(pFD, pData, size) < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
taosCloseFile(&pFD);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
if (!taosCheckChecksumWhole(pData, size)) {
|
||||
code = TSDB_CODE_FILE_CORRUPTED;
|
||||
taosCloseFile(&pFD);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
taosCloseFile(&pFD);
|
||||
|
||||
// decode binary
|
||||
code = tsdbBinaryToFS(pData, size, pFS);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
_exit:
|
||||
if (pData) taosMemoryFree(pData);
|
||||
if (code) {
|
||||
smaError("%s failed at line %d since %s, fname:%s", __func__, lino, tstrerror(code), fname);
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2) {
|
||||
const SQTaskFile *q1 = (const SQTaskFile *)p1;
|
||||
const SQTaskFile *q2 = (const SQTaskFile *)p2;
|
||||
|
||||
if (q1->suid < q2->suid) {
|
||||
return -1;
|
||||
} else if (q1->suid > q2->suid) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (q1->level < q2->level) {
|
||||
return -1;
|
||||
} else if (q1->level > q2->level) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (q1->version < q2->version) {
|
||||
return -2;
|
||||
} else if (q1->version > q2->version) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tdRSmaFSApplyChange(SSma *pSma, SRSmaFS *pFSNew) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
int32_t nRef = 0;
|
||||
SVnode *pVnode = pSma->pVnode;
|
||||
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
|
||||
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
|
||||
SRSmaFS *pFSOld = RSMA_FS(pStat);
|
||||
int64_t version = pStat->commitAppliedVer;
|
||||
char fname[TSDB_FILENAME_LEN] = {0};
|
||||
|
||||
// SQTaskFile
|
||||
int32_t nNew = taosArrayGetSize(pFSNew->aQTaskInf);
|
||||
int32_t iNew = 0;
|
||||
while (iNew < nNew) {
|
||||
SQTaskFile *pQTaskFNew = TARRAY_GET_ELEM(pFSNew->aQTaskInf, iNew++);
|
||||
|
||||
int32_t idx = taosArraySearchIdx(pFSOld->aQTaskInf, pQTaskFNew, tdQTaskInfCmprFn1, TD_GE);
|
||||
|
||||
if (idx < 0) {
|
||||
idx = taosArrayGetSize(pFSOld->aQTaskInf);
|
||||
pQTaskFNew->nRef = 1;
|
||||
} else {
|
||||
SQTaskFile *pTaskF = TARRAY_GET_ELEM(pFSOld->aQTaskInf, idx);
|
||||
int32_t c1 = tdQTaskInfCmprFn1(pQTaskFNew, pTaskF);
|
||||
if (c1 == 0) {
|
||||
// utilize the item in pFSOld->qQTaskInf, instead of pFSNew
|
||||
continue;
|
||||
} else if (c1 < 0) {
|
||||
// NOTHING TODO
|
||||
} else {
|
||||
code = TSDB_CODE_RSMA_FS_UPDATE;
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
}
|
||||
|
||||
if (taosArrayInsert(pFSOld->aQTaskInf, idx, pQTaskFNew) == NULL) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
// remove previous version
|
||||
while (--idx >= 0) {
|
||||
SQTaskFile *preTaskF = TARRAY_GET_ELEM(pFSOld->aQTaskInf, idx);
|
||||
int32_t c2 = tdQTaskInfCmprFn1(preTaskF, pQTaskFNew);
|
||||
if (c2 == 0) {
|
||||
code = TSDB_CODE_RSMA_FS_UPDATE;
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
} else if (c2 != -2) {
|
||||
break;
|
||||
}
|
||||
|
||||
nRef = atomic_sub_fetch_32(&preTaskF->nRef, 1);
|
||||
if (nRef <= 0) {
|
||||
tdRSmaQTaskInfoGetFullName(pVnode, preTaskF->suid, preTaskF->level, preTaskF->version, pVnode->pTfs, fname);
|
||||
(void)taosRemoveFile(fname);
|
||||
taosArrayRemove(pFSOld->aQTaskInf, idx);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_exit:
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
static int32_t tdRSmaFSScanAndTryFix(SSma *pSma) {
|
||||
int32_t code = 0;
|
||||
#if 0
|
||||
int32_t lino = 0;
|
||||
SVnode *pVnode = pSma->pVnode;
|
||||
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
|
||||
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
|
||||
SRSmaFS *pFS = RSMA_FS(pStat);
|
||||
char fname[TSDB_FILENAME_LEN] = {0};
|
||||
char fnameVer[TSDB_FILENAME_LEN] = {0};
|
||||
|
||||
// SArray<SQTaskFile>
|
||||
int32_t size = taosArrayGetSize(pFS->aQTaskInf);
|
||||
for (int32_t i = 0; i < size; ++i) {
|
||||
SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, i);
|
||||
|
||||
// main.tdb =========
|
||||
tdRSmaQTaskInfoGetFullName(pVnode, pTaskF->suid, pTaskF->level, pTaskF->version,
|
||||
pVnode->pTfs, fnameVer);
|
||||
tdRSmaQTaskInfoGetFullName(pVnode, pTaskF->suid, pTaskF->level, -1, pVnode->pTfs, fname);
|
||||
|
||||
if (taosCheckExistFile(fnameVer)) {
|
||||
if (taosRenameFile(fnameVer, fname) < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
smaDebug("vgId:%d, %s:%d succeed to to rename %s to %s", TD_VID(pVnode), __func__, lino, fnameVer, fname);
|
||||
} else if (taosCheckExistFile(fname)) {
|
||||
if (taosRemoveFile(fname) < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
smaDebug("vgId:%d, %s:%d succeed to to remove %s", TD_VID(pVnode), __func__, lino, fname);
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
// remove those invalid files (todo)
|
||||
// main.tdb-journal.5 // TDB should handle its clear for kill -9
|
||||
}
|
||||
|
||||
_exit:
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
|
||||
}
|
||||
#endif
|
||||
return code;
|
||||
}
|
||||
|
||||
// EXPOSED APIS ====================================================================================
|
||||
|
||||
int32_t tdRSmaFSOpen(SSma *pSma, int64_t version, int8_t rollback) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
SVnode *pVnode = pSma->pVnode;
|
||||
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
|
||||
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
|
||||
|
||||
// open handle
|
||||
code = tdRSmaFSCreate(RSMA_FS(pStat), 0);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
// open impl
|
||||
char current[TSDB_FILENAME_LEN] = {0};
|
||||
char current_t[TSDB_FILENAME_LEN] = {0};
|
||||
tdRSmaGetCurrentFName(pSma, current, current_t);
|
||||
|
||||
if (taosCheckExistFile(current)) {
|
||||
code = tdRSmaLoadFSFromFile(current, RSMA_FS(pStat));
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
if (taosCheckExistFile(current_t)) {
|
||||
if (rollback) {
|
||||
code = tdRSmaFSRollback(pSma);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
} else {
|
||||
code = tdRSmaFSCommit(pSma);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// 1st time open with empty current/qTaskInfoFile
|
||||
code = tdRSmaSaveFSToFile(RSMA_FS(pStat), current);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
// scan and try fix(remove main.db/main.db.xxx and use the one with version)
|
||||
code = tdRSmaFSScanAndTryFix(pSma);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
_exit:
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
void tdRSmaFSClose(SRSmaFS *pFS) { pFS->aQTaskInf = taosArrayDestroy(pFS->aQTaskInf); }
|
||||
|
||||
int32_t tdRSmaFSPrepareCommit(SSma *pSma, SRSmaFS *pFSNew) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
char tfname[TSDB_FILENAME_LEN];
|
||||
|
||||
tdRSmaGetCurrentFName(pSma, NULL, tfname);
|
||||
|
||||
// generate PRESENT.t
|
||||
code = tdRSmaSaveFSToFile(pFSNew, tfname);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
_exit:
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
int32_t tdRSmaFSCommit(SSma *pSma) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
SRSmaFS fs = {0};
|
||||
|
||||
char current[TSDB_FILENAME_LEN] = {0};
|
||||
char current_t[TSDB_FILENAME_LEN] = {0};
|
||||
tdRSmaGetCurrentFName(pSma, current, current_t);
|
||||
|
||||
if (!taosCheckExistFile(current_t)) {
|
||||
goto _exit;
|
||||
}
|
||||
|
||||
// rename the file
|
||||
if (taosRenameFile(current_t, current) < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
|
||||
// load the new FS
|
||||
code = tdRSmaFSCreate(&fs, 1);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
code = tdRSmaLoadFSFromFile(current, &fs);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
// apply file change
|
||||
code = tdRSmaFSApplyChange(pSma, &fs);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
_exit:
|
||||
tdRSmaFSClose(&fs);
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pSma), __func__, lino, tstrerror(code));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
int32_t tdRSmaFSFinishCommit(SSma *pSma) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
|
||||
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);
|
||||
|
||||
taosWLockLatch(RSMA_FS_LOCK(pStat));
|
||||
code = tdRSmaFSCommit(pSma);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
|
||||
_exit:
|
||||
taosWUnLockLatch(RSMA_FS_LOCK(pStat));
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pSma), __func__, lino, tstrerror(code));
|
||||
} else {
|
||||
smaInfo("vgId:%d, rsmaFS finish commit", SMA_VID(pSma));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
int32_t tdRSmaFSRollback(SSma *pSma) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 0;
|
||||
|
||||
char current_t[TSDB_FILENAME_LEN] = {0};
|
||||
tdRSmaGetCurrentFName(pSma, NULL, current_t);
|
||||
(void)taosRemoveFile(current_t);
|
||||
|
||||
_exit:
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pSma), __func__, lino, tstrerror(errno));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
int32_t tdRSmaFSUpsertQTaskFile(SSma *pSma, SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t nSize) {
|
||||
int32_t code = 0;
|
||||
|
||||
for (int32_t i = 0; i < nSize; ++i) {
|
||||
SQTaskFile *qTaskF = qTaskFile + i;
|
||||
|
||||
int32_t idx = taosArraySearchIdx(pFS->aQTaskInf, qTaskF, tdQTaskInfCmprFn1, TD_GE);
|
||||
|
||||
if (idx < 0) {
|
||||
idx = taosArrayGetSize(pFS->aQTaskInf);
|
||||
} else {
|
||||
SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, idx);
|
||||
int32_t c = tdQTaskInfCmprFn1(pTaskF, qTaskF);
|
||||
if (c == 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;
|
||||
}
|
||||
}
|
||||
|
||||
if (!taosArrayInsert(pFS->aQTaskInf, idx, qTaskF)) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
goto _exit;
|
||||
}
|
||||
}
|
||||
|
||||
_exit:
|
||||
return code;
|
||||
}
|
||||
|
||||
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_s(sizeof(SQTaskFile), size);
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
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(pVnode, pTaskF->suid, pTaskF->level, pTaskF->version, 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 lino = 0;
|
||||
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
|
||||
SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
|
||||
|
||||
taosRLockLatch(RSMA_FS_LOCK(pStat));
|
||||
code = tdRSmaFSRef(pSma, pFS);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
_exit:
|
||||
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));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
||||
int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFS) {
|
||||
int32_t code = 0;
|
||||
int32_t lino = 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);
|
||||
|
||||
code = tdRSmaFSCreate(pFS, size);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
taosArrayAddBatch(pFS->aQTaskInf, qFS->aQTaskInf->pData, size);
|
||||
|
||||
_exit:
|
||||
if (code) {
|
||||
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code));
|
||||
}
|
||||
return code;
|
||||
}
|
||||
Loading…
Reference in New Issue