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

Merge pull request #11529 from taosdata/feature/vnode_refact 

Feature/vnode refact
This commit is contained in:
Hongze Cheng 2022-04-15 17:36:52 +08:00 committed by GitHub
parent 980f7b707b 911c81d230
commit 3fedb0042e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 294 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -91,9 +91,9 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg->keep = pCreate->daysToKeep0; pCfg->keep = pCreate->daysToKeep0;
pCfg->streamMode = pCreate->streamMode; pCfg->streamMode = pCreate->streamMode;
pCfg->isWeak = true; pCfg->isWeak = true;
pCfg->tsdbCfg.keep = pCreate->daysToKeep0;
pCfg->tsdbCfg.keep1 = pCreate->daysToKeep2;
pCfg->tsdbCfg.keep2 = pCreate->daysToKeep0; pCfg->tsdbCfg.keep2 = pCreate->daysToKeep0;
pCfg->tsdbCfg.keep0 = pCreate->daysToKeep2;
pCfg->tsdbCfg.keep1 = pCreate->daysToKeep0;
pCfg->tsdbCfg.lruCacheSize = pCreate->cacheBlockSize; pCfg->tsdbCfg.lruCacheSize = pCreate->cacheBlockSize;
pCfg->tsdbCfg.retentions = pCreate->pRetensions; pCfg->tsdbCfg.retentions = pCreate->pRetensions;
pCfg->metaCfg.lruSize = pCreate->cacheBlockSize; pCfg->metaCfg.lruSize = pCreate->cacheBlockSize;
@ -121,6 +121,8 @@ static void vmGenerateWrapperCfg(SVnodesMgmt *pMgmt, SCreateVnodeReq *pCreate, S
int32_t vmProcessCreateVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { int32_t vmProcessCreateVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SRpcMsg *pReq = &pMsg->rpcMsg; SRpcMsg *pReq = &pMsg->rpcMsg;
SCreateVnodeReq createReq = {0}; SCreateVnodeReq createReq = {0};
char path[TSDB_FILENAME_LEN];
if (tDeserializeSCreateVnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) { if (tDeserializeSCreateVnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
return -1; return -1;
@ -143,6 +145,14 @@ int32_t vmProcessCreateVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
return -1; return -1;
} }
// create vnode
snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
tFreeSCreateVnodeReq(&createReq);
dError("vgId:%d, failed to create vnode since %s", createReq.vgId, terrstr());
return -1;
}
SMsgCb msgCb = pMgmt->pDnode->data.msgCb; SMsgCb msgCb = pMgmt->pDnode->data.msgCb;
msgCb.pWrapper = pMgmt->pWrapper; msgCb.pWrapper = pMgmt->pWrapper;
msgCb.queueFps[QUERY_QUEUE] = vmPutMsgToQueryQueue; msgCb.queueFps[QUERY_QUEUE] = vmPutMsgToQueryQueue;

10
source/dnode/vnode/inc/vnode.h
View File

@ -124,12 +124,12 @@ struct STsdbCfg {
int8_t precision; int8_t precision;
int8_t update; int8_t update;
int8_t compression; int8_t compression;
int32_t daysPerFile; int32_t days;
int32_t minRowsPerFileBlock; int32_t minRows;
int32_t maxRowsPerFileBlock; int32_t maxRows;
int32_t keep;
int32_t keep1;
int32_t keep2; int32_t keep2;
int32_t keep0;
int32_t keep1;
uint64_t lruCacheSize; uint64_t lruCacheSize;
SArray *retentions; SArray *retentions;
}; };

1
source/dnode/vnode/src/inc/vnd.h
View File

@ -43,6 +43,7 @@ int vnodeGetTableMeta(SVnode* pVnode, SRpcMsg* pMsg);
// vnodeCommit ==================== // vnodeCommit ====================
int vnodeSaveInfo(const char* dir, const SVnodeInfo* pCfg); int vnodeSaveInfo(const char* dir, const SVnodeInfo* pCfg);
int vnodeCommitInfo(const char* dir, const SVnodeInfo* pInfo); int vnodeCommitInfo(const char* dir, const SVnodeInfo* pInfo);
int vnodeLoadInfo(const char* dir, SVnodeInfo* pInfo);
#if 1 #if 1
// SVBufPool // SVBufPool

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

@ -27,6 +27,7 @@
#include "tdbInt.h" #include "tdbInt.h"
#include "tfs.h" #include "tfs.h"
#include "tglobal.h" #include "tglobal.h"
#include "tjson.h"
#include "tlist.h" #include "tlist.h"
#include "tlockfree.h" #include "tlockfree.h"
#include "tlosertree.h" #include "tlosertree.h"

37
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -55,7 +55,7 @@ typedef struct {
#define TSDB_COMMIT_BUF(ch) TSDB_READ_BUF(&((ch)->readh)) #define TSDB_COMMIT_BUF(ch) TSDB_READ_BUF(&((ch)->readh))
#define TSDB_COMMIT_COMP_BUF(ch) TSDB_READ_COMP_BUF(&((ch)->readh)) #define TSDB_COMMIT_COMP_BUF(ch) TSDB_READ_COMP_BUF(&((ch)->readh))
#define TSDB_COMMIT_EXBUF(ch) TSDB_READ_EXBUF(&((ch)->readh)) #define TSDB_COMMIT_EXBUF(ch) TSDB_READ_EXBUF(&((ch)->readh))
#define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->config.maxRowsPerFileBlock) #define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->config.maxRows)
#define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch))) #define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch)))
static void tsdbStartCommit(STsdb *pRepo); static void tsdbStartCommit(STsdb *pRepo);
@ -217,14 +217,14 @@ void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
TSKEY minKey, midKey, maxKey, now; TSKEY minKey, midKey, maxKey, now;
now = taosGetTimestamp(pCfg->precision); now = taosGetTimestamp(pCfg->precision);
minKey = now - pCfg->keep * tsTickPerDay[pCfg->precision]; minKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
midKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision]; midKey = now - pCfg->keep1 * tsTickPerDay[pCfg->precision];
maxKey = now - pCfg->keep1 * tsTickPerDay[pCfg->precision]; maxKey = now - pCfg->keep0 * tsTickPerDay[pCfg->precision];
pRtn->minKey = minKey; pRtn->minKey = minKey;
pRtn->minFid = (int)(TSDB_KEY_FID(minKey, pCfg->daysPerFile, pCfg->precision)); pRtn->minFid = (int)(TSDB_KEY_FID(minKey, pCfg->days, pCfg->precision));
pRtn->midFid = (int)(TSDB_KEY_FID(midKey, pCfg->daysPerFile, pCfg->precision)); pRtn->midFid = (int)(TSDB_KEY_FID(midKey, pCfg->days, pCfg->precision));
pRtn->maxFid = (int)(TSDB_KEY_FID(maxKey, pCfg->daysPerFile, pCfg->precision)); pRtn->maxFid = (int)(TSDB_KEY_FID(maxKey, pCfg->days, pCfg->precision));
tsdbDebug("vgId:%d now:%" PRId64 " minKey:%" PRId64 " minFid:%d, midFid:%d, maxFid:%d", REPO_ID(pRepo), now, minKey, tsdbDebug("vgId:%d now:%" PRId64 " minKey:%" PRId64 " minFid:%d, midFid:%d, maxFid:%d", REPO_ID(pRepo), now, minKey,
pRtn->minFid, pRtn->midFid, pRtn->maxFid); pRtn->minFid, pRtn->midFid, pRtn->maxFid);
} }
@ -286,7 +286,7 @@ static int tsdbInitCommitH(SCommitH *pCommith, STsdb *pRepo) {
return -1; return -1;
} }
pCommith->pDataCols = tdNewDataCols(0, pCfg->maxRowsPerFileBlock); pCommith->pDataCols = tdNewDataCols(0, pCfg->maxRows);
if (pCommith->pDataCols == NULL) { if (pCommith->pDataCols == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tsdbDestroyCommitH(pCommith); tsdbDestroyCommitH(pCommith);
@ -319,7 +319,7 @@ static int tsdbNextCommitFid(SCommitH *pCommith) {
if (nextKey == TSDB_DATA_TIMESTAMP_NULL) { if (nextKey == TSDB_DATA_TIMESTAMP_NULL) {
continue; continue;
} else { } else {
int tfid = (int)(TSDB_KEY_FID(nextKey, pCfg->daysPerFile, pCfg->precision)); int tfid = (int)(TSDB_KEY_FID(nextKey, pCfg->days, pCfg->precision));
if (fid == TSDB_IVLD_FID || fid > tfid) { if (fid == TSDB_IVLD_FID || fid > tfid) {
fid = tfid; fid = tfid;
} }
@ -346,7 +346,7 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
ASSERT(pSet == NULL || pSet->fid == fid); ASSERT(pSet == NULL || pSet->fid == fid);
tsdbResetCommitFile(pCommith); tsdbResetCommitFile(pCommith);
tsdbGetFidKeyRange(pCfg->daysPerFile, pCfg->precision, fid, &(pCommith->minKey), &(pCommith->maxKey)); tsdbGetFidKeyRange(pCfg->days, pCfg->precision, fid, &(pCommith->minKey), &(pCommith->maxKey));
// Set and open files // Set and open files
if (tsdbSetAndOpenCommitFile(pCommith, pSet, fid) < 0) { if (tsdbSetAndOpenCommitFile(pCommith, pSet, fid) < 0) {
@ -1210,8 +1210,8 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDF
int64_t offset = 0, offsetAggr = 0; int64_t offset = 0, offsetAggr = 0;
int rowsToWrite = pDataCols->numOfRows; int rowsToWrite = pDataCols->numOfRows;
ASSERT(rowsToWrite > 0 && rowsToWrite <= pCfg->maxRowsPerFileBlock); ASSERT(rowsToWrite > 0 && rowsToWrite <= pCfg->maxRows);
ASSERT((!isLast) || rowsToWrite < pCfg->minRowsPerFileBlock); ASSERT((!isLast) || rowsToWrite < pCfg->minRows);
// Make buffer space // Make buffer space
if (tsdbMakeRoom(ppBuf, tsdbBlockStatisSize(pDataCols->numOfCols, SBlockVerLatest)) < 0) { if (tsdbMakeRoom(ppBuf, tsdbBlockStatisSize(pDataCols->numOfCols, SBlockVerLatest)) < 0) {
@ -1460,7 +1460,7 @@ static int tsdbCommitMemData(SCommitH *pCommith, SCommitIter *pIter, TSKEY keyLi
if (pCommith->pDataCols->numOfRows <= 0) break; if (pCommith->pDataCols->numOfRows <= 0) break;
if (toData || pCommith->pDataCols->numOfRows >= pCfg->minRowsPerFileBlock) { if (toData || pCommith->pDataCols->numOfRows >= pCfg->minRows) {
pDFile = TSDB_COMMIT_DATA_FILE(pCommith); pDFile = TSDB_COMMIT_DATA_FILE(pCommith);
isLast = false; isLast = false;
} else { } else {
@ -1619,7 +1619,7 @@ static int tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols
if (pCommith->pDataCols->numOfRows == 0) break; if (pCommith->pDataCols->numOfRows == 0) break;
if (isLastOneBlock) { if (isLastOneBlock) {
if (pCommith->pDataCols->numOfRows < pCfg->minRowsPerFileBlock) { if (pCommith->pDataCols->numOfRows < pCfg->minRows) {
pDFile = TSDB_COMMIT_LAST_FILE(pCommith); pDFile = TSDB_COMMIT_LAST_FILE(pCommith);
isLast = true; isLast = true;
} else { } else {
@ -1667,7 +1667,8 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) { if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) {
TASSERT(0); TASSERT(0);
} }
tdAppendValToDataCol(pTarget->cols + i, sVal.valType, sVal.val, pTarget->numOfRows, pTarget->maxPoints, pTarget->bitmapMode); tdAppendValToDataCol(pTarget->cols + i, sVal.valType, sVal.val, pTarget->numOfRows, pTarget->maxPoints,
pTarget->bitmapMode);
} }
++pTarget->numOfRows; ++pTarget->numOfRows;
@ -1774,11 +1775,11 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
ASSERT(mergeRows > 0); ASSERT(mergeRows > 0);
if (pBlock->numOfSubBlocks < TSDB_MAX_SUBBLOCKS && pInfo->nOperations <= pCfg->maxRowsPerFileBlock) { if (pBlock->numOfSubBlocks < TSDB_MAX_SUBBLOCKS && pInfo->nOperations <= pCfg->maxRows) {
if (pBlock->last) { if (pBlock->last) {
if (pCommith->isLFileSame && mergeRows < pCfg->minRowsPerFileBlock) return true; if (pCommith->isLFileSame && mergeRows < pCfg->minRows) return true;
} else { } else {
if (pCommith->isDFileSame && mergeRows <= pCfg->maxRowsPerFileBlock) return true; if (pCommith->isDFileSame && mergeRows <= pCfg->maxRows) return true;
} }
} }

4
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

@ -190,8 +190,8 @@ static int tsdbAddDFileSetToStatus(SFSStatus *pStatus, const SDFileSet *pSet) {
// ================== STsdbFS // ================== STsdbFS
STsdbFS *tsdbNewFS(const STsdbCfg *pCfg) { STsdbFS *tsdbNewFS(const STsdbCfg *pCfg) {
int keep = pCfg->keep; int keep = pCfg->keep2;
int days = pCfg->daysPerFile; int days = pCfg->days;
int maxFSet = TSDB_MAX_FSETS(keep, days); int maxFSet = TSDB_MAX_FSETS(keep, days);
STsdbFS *pfs; STsdbFS *pfs;

4
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -263,8 +263,8 @@ static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
SSubmitBlkIter blkIter = {0}; SSubmitBlkIter blkIter = {0};
STSRow *row = NULL; STSRow *row = NULL;
TSKEY now = taosGetTimestamp(pTsdb->config.precision); TSKEY now = taosGetTimestamp(pTsdb->config.precision);
TSKEY minKey = now - tsTickPerDay[pTsdb->config.precision] * pTsdb->config.keep; TSKEY minKey = now - tsTickPerDay[pTsdb->config.precision] * pTsdb->config.keep2;
TSKEY maxKey = now + tsTickPerDay[pTsdb->config.precision] * pTsdb->config.daysPerFile; TSKEY maxKey = now + tsTickPerDay[pTsdb->config.precision] * pTsdb->config.days;
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
pMsg->length = htonl(pMsg->length); pMsg->length = htonl(pMsg->length);

10
source/dnode/vnode/src/tsdb/tsdbOptions.c
View File

@ -17,12 +17,12 @@
const STsdbCfg defautlTsdbOptions = {.precision = 0, const STsdbCfg defautlTsdbOptions = {.precision = 0,
.lruCacheSize = 0, .lruCacheSize = 0,
.daysPerFile = 10, .days = 10,
.minRowsPerFileBlock = 100, .minRows = 100,
.maxRowsPerFileBlock = 4096, .maxRows = 4096,
.keep = 3650,
.keep1 = 3650,
.keep2 = 3650, .keep2 = 3650,
.keep0 = 3650,
.keep1 = 3650,
.update = 0, .update = 0,
.compression = TWO_STAGE_COMP}; .compression = TWO_STAGE_COMP};

12
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -314,7 +314,7 @@ static int64_t getEarliestValidTimestamp(STsdb* pTsdb) {
STsdbCfg* pCfg = &pTsdb->config; STsdbCfg* pCfg = &pTsdb->config;
int64_t now = taosGetTimestamp(pCfg->precision); int64_t now = taosGetTimestamp(pCfg->precision);
return now - (tsTickPerDay[pCfg->precision] * pCfg->keep) + 1; // needs to add one tick return now - (tsTickPerDay[pCfg->precision] * pCfg->keep2) + 1; // needs to add one tick
} }
static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, STsdbQueryCond* pCond) { static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, STsdbQueryCond* pCond) {
@ -404,7 +404,7 @@ static STsdbReadHandle* tsdbQueryTablesImpl(STsdb* tsdb, STsdbQueryCond* pCond,
pReadHandle->defaultLoadColumn = getDefaultLoadColumns(pReadHandle, true); pReadHandle->defaultLoadColumn = getDefaultLoadColumns(pReadHandle, true);
} }
pReadHandle->pDataCols = tdNewDataCols(1000, pReadHandle->pTsdb->config.maxRowsPerFileBlock); pReadHandle->pDataCols = tdNewDataCols(1000, pReadHandle->pTsdb->config.maxRows);
if (pReadHandle->pDataCols == NULL) { if (pReadHandle->pDataCols == NULL) {
tsdbError("%p failed to malloc buf for pDataCols, %s", pReadHandle, pReadHandle->idStr); tsdbError("%p failed to malloc buf for pDataCols, %s", pReadHandle, pReadHandle->idStr);
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
@ -2199,7 +2199,7 @@ static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exi
break; break;
} }
tsdbGetFidKeyRange(pCfg->daysPerFile, pCfg->precision, pTsdbReadHandle->pFileGroup->fid, &win.skey, &win.ekey); tsdbGetFidKeyRange(pCfg->days, pCfg->precision, pTsdbReadHandle->pFileGroup->fid, &win.skey, &win.ekey);
// current file are not overlapped with query time window, ignore remain files // current file are not overlapped with query time window, ignore remain files
if ((ASCENDING_TRAVERSE(pTsdbReadHandle->order) && win.skey > pTsdbReadHandle->window.ekey) || if ((ASCENDING_TRAVERSE(pTsdbReadHandle->order) && win.skey > pTsdbReadHandle->window.ekey) ||
@ -2295,7 +2295,7 @@ int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDistInfo*
// find the start data block in file // find the start data block in file
pTsdbReadHandle->locateStart = true; pTsdbReadHandle->locateStart = true;
STsdbCfg* pCfg = &pTsdbReadHandle->pTsdb->config; STsdbCfg* pCfg = &pTsdbReadHandle->pTsdb->config;
int32_t fid = getFileIdFromKey(pTsdbReadHandle->window.skey, pCfg->daysPerFile, pCfg->precision); int32_t fid = getFileIdFromKey(pTsdbReadHandle->window.skey, pCfg->days, pCfg->precision);
tsdbRLockFS(pFileHandle); tsdbRLockFS(pFileHandle);
tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order); tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
@ -2321,7 +2321,7 @@ int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDistInfo*
break; break;
} }
tsdbGetFidKeyRange(pCfg->daysPerFile, pCfg->precision, pTsdbReadHandle->pFileGroup->fid, &win.skey, &win.ekey); tsdbGetFidKeyRange(pCfg->days, pCfg->precision, pTsdbReadHandle->pFileGroup->fid, &win.skey, &win.ekey);
// current file are not overlapped with query time window, ignore remain files // current file are not overlapped with query time window, ignore remain files
if ((ascTraverse && win.skey > pTsdbReadHandle->window.ekey) || if ((ascTraverse && win.skey > pTsdbReadHandle->window.ekey) ||
@ -2396,7 +2396,7 @@ static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exis
if (!pTsdbReadHandle->locateStart) { if (!pTsdbReadHandle->locateStart) {
pTsdbReadHandle->locateStart = true; pTsdbReadHandle->locateStart = true;
STsdbCfg* pCfg = &pTsdbReadHandle->pTsdb->config; STsdbCfg* pCfg = &pTsdbReadHandle->pTsdb->config;
int32_t fid = getFileIdFromKey(pTsdbReadHandle->window.skey, pCfg->daysPerFile, pCfg->precision); int32_t fid = getFileIdFromKey(pTsdbReadHandle->window.skey, pCfg->days, pCfg->precision);
tsdbRLockFS(pFileHandle); tsdbRLockFS(pFileHandle);
tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order); tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);

10
source/dnode/vnode/src/tsdb/tsdbReadImpl.c
View File

@ -43,14 +43,14 @@ int tsdbInitReadH(SReadH *pReadh, STsdb *pRepo) {
return -1; return -1;
} }
pReadh->pDCols[0] = tdNewDataCols(0, pCfg->maxRowsPerFileBlock); pReadh->pDCols[0] = tdNewDataCols(0, pCfg->maxRows);
if (pReadh->pDCols[0] == NULL) { if (pReadh->pDCols[0] == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tsdbDestroyReadH(pReadh); tsdbDestroyReadH(pReadh);
return -1; return -1;
} }
pReadh->pDCols[1] = tdNewDataCols(0, pCfg->maxRowsPerFileBlock); pReadh->pDCols[1] = tdNewDataCols(0, pCfg->maxRows);
if (pReadh->pDCols[1] == NULL) { if (pReadh->pDCols[1] == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tsdbDestroyReadH(pReadh); tsdbDestroyReadH(pReadh);
@ -276,8 +276,8 @@ int tsdbLoadBlockData(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo) {
return 0; return 0;
} }
int tsdbLoadBlockDataCols(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo, const int16_t *colIds, int tsdbLoadBlockDataCols(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo, const int16_t *colIds, int numOfColsIds,
int numOfColsIds, bool mergeBitmap) { bool mergeBitmap) {
ASSERT(pBlock->numOfSubBlocks > 0); ASSERT(pBlock->numOfSubBlocks > 0);
int8_t update = pReadh->pRepo->config.update; int8_t update = pReadh->pRepo->config.update;
@ -836,7 +836,7 @@ static int tsdbLoadColData(SReadH *pReadh, SDFile *pDFile, SBlock *pBlock, SBloc
} }
if (tsdbCheckAndDecodeColumnData(pDataCol, pReadh->pBuf, pBlockCol->len, pBlockCol->blen, pBlock->algorithm, if (tsdbCheckAndDecodeColumnData(pDataCol, pReadh->pBuf, pBlockCol->len, pBlockCol->blen, pBlock->algorithm,
pBlock->numOfRows, tLenBitmap, pCfg->maxRowsPerFileBlock, pReadh->pCBuf, pBlock->numOfRows, tLenBitmap, pCfg->maxRows, pReadh->pCBuf,
(int32_t)taosTSizeof(pReadh->pCBuf)) < 0) { (int32_t)taosTSizeof(pReadh->pCBuf)) < 0) {
tsdbError("vgId:%d file %s is broken at column %d offset %" PRId64, REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), tsdbError("vgId:%d file %s is broken at column %d offset %" PRId64, REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile),
pBlockCol->colId, offset); pBlockCol->colId, offset);

8
source/dnode/vnode/src/tsdb/tsdbSma.c
View File

@ -106,7 +106,8 @@ struct SSmaStat {
// expired window // expired window
static int32_t tsdbUpdateExpiredWindowImpl(STsdb *pTsdb, SSubmitReq *pMsg, int64_t version); static int32_t tsdbUpdateExpiredWindowImpl(STsdb *pTsdb, SSubmitReq *pMsg, int64_t version);
static int32_t tsdbSetExpiredWindow(STsdb *pTsdb, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey, int64_t version); static int32_t tsdbSetExpiredWindow(STsdb *pTsdb, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey,
int64_t version);
static int32_t tsdbInitSmaStat(SSmaStat **pSmaStat); static int32_t tsdbInitSmaStat(SSmaStat **pSmaStat);
static void *tsdbFreeSmaStatItem(SSmaStatItem *pSmaStatItem); static void *tsdbFreeSmaStatItem(SSmaStatItem *pSmaStatItem);
static int32_t tsdbDestroySmaState(SSmaStat *pSmaStat); static int32_t tsdbDestroySmaState(SSmaStat *pSmaStat);
@ -544,7 +545,8 @@ static int32_t tsdbCheckAndInitSmaEnv(STsdb *pTsdb, int8_t smaType) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
}; };
static int32_t tsdbSetExpiredWindow(STsdb *pTsdb, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey, int64_t version) { static int32_t tsdbSetExpiredWindow(STsdb *pTsdb, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey,
int64_t version) {
SSmaStatItem *pItem = taosHashGet(pItemsHash, &indexUid, sizeof(indexUid)); SSmaStatItem *pItem = taosHashGet(pItemsHash, &indexUid, sizeof(indexUid));
if (pItem == NULL) { if (pItem == NULL) {
// TODO: use TSDB_SMA_STAT_EXPIRED and update by stream computing later // TODO: use TSDB_SMA_STAT_EXPIRED and update by stream computing later
@ -946,7 +948,7 @@ static int32_t tsdbSetTSmaDataFile(STSmaWriteH *pSmaH, int64_t indexUid, int32_t
*/ */
static int32_t tsdbGetTSmaDays(STsdb *pTsdb, int64_t interval, int32_t storageLevel) { static int32_t tsdbGetTSmaDays(STsdb *pTsdb, int64_t interval, int32_t storageLevel) {
STsdbCfg *pCfg = REPO_CFG(pTsdb); STsdbCfg *pCfg = REPO_CFG(pTsdb);
int32_t daysPerFile = pCfg->daysPerFile; int32_t daysPerFile = pCfg->days;
if (storageLevel == SMA_STORAGE_LEVEL_TSDB) { if (storageLevel == SMA_STORAGE_LEVEL_TSDB) {
int32_t days = SMA_STORAGE_TSDB_TIMES * (interval / tsTickPerDay[pCfg->precision]); int32_t days = SMA_STORAGE_TSDB_TIMES * (interval / tsTickPerDay[pCfg->precision]);

187
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -18,8 +18,8 @@
#define VND_INFO_FNAME "vnode.json" #define VND_INFO_FNAME "vnode.json"
#define VND_INFO_FNAME_TMP "vnode_tmp.json" #define VND_INFO_FNAME_TMP "vnode_tmp.json"
static int vnodeEncodeInfo(const SVnodeInfo *pInfo, uint8_t **ppData, int *len); static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData);
static int vnodeDecodeInfo(uint8_t *pData, int len, SVnodeInfo *pInfo); static int vnodeDecodeInfo(uint8_t *pData, SVnodeInfo *pInfo);
static int vnodeStartCommit(SVnode *pVnode); static int vnodeStartCommit(SVnode *pVnode);
static int vnodeEndCommit(SVnode *pVnode); static int vnodeEndCommit(SVnode *pVnode);
static int vnodeCommit(void *arg); static int vnodeCommit(void *arg);
@ -28,16 +28,14 @@ static void vnodeWaitCommit(SVnode *pVnode);
int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) { int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
char fname[TSDB_FILENAME_LEN]; char fname[TSDB_FILENAME_LEN];
TdFilePtr pFile; TdFilePtr pFile;
uint8_t *data; char *data;
int len;
snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP); snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);
// encode info // encode info
data = NULL; data = NULL;
len = 0;
if (vnodeEncodeInfo(pInfo, &data, &len) < 0) { if (vnodeEncodeInfo(pInfo, &data) < 0) {
return -1; return -1;
} }
@ -48,7 +46,7 @@ int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
return -1; return -1;
} }
if (taosWriteFile(pFile, data, len) < 0) { if (taosWriteFile(pFile, data, strlen(data)) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
@ -90,9 +88,53 @@ int vnodeCommitInfo(const char *dir, const SVnodeInfo *pInfo) {
return 0; return 0;
} }
int vnodeLoadInfo(const char *dir) { int vnodeLoadInfo(const char *dir, SVnodeInfo *pInfo) {
// TODO char fname[TSDB_FILENAME_LEN];
TdFilePtr pFile = NULL;
char *pData = NULL;
int64_t size;
snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);
// read info
pFile = taosOpenFile(fname, TD_FILE_READ);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (taosFStatFile(pFile, &size, NULL) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
pData = taosMemoryMalloc(size);
if (pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
if (taosReadFile(pFile, pData, size) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pFile);
// decode info
if (vnodeDecodeInfo(pData, pInfo) < 0) {
taosMemoryFree(pData);
return -1;
}
taosMemoryFree(pData);
return 0; return 0;
_err:
taosCloseFile(&pFile);
taosMemoryFree(pData);
return -1;
} }
int vnodeAsyncCommit(SVnode *pVnode) { int vnodeAsyncCommit(SVnode *pVnode) {
@ -137,12 +179,131 @@ static int vnodeEndCommit(SVnode *pVnode) {
static FORCE_INLINE void vnodeWaitCommit(SVnode *pVnode) { tsem_wait(&pVnode->canCommit); } static FORCE_INLINE void vnodeWaitCommit(SVnode *pVnode) { tsem_wait(&pVnode->canCommit); }
static int vnodeEncodeInfo(const SVnodeInfo *pInfo, uint8_t **ppData, int *len) { static int vnodeEncodeConfig(const void *pObj, SJson *pJson) {
// TODO const SVnodeCfg *pCfg = (SVnodeCfg *)pObj;
if (tjsonAddIntegerToObject(pJson, "vgId", pCfg->vgId) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "dbId", pCfg->dbId) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "wsize", pCfg->wsize) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "ssize", pCfg->ssize) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "lsize", pCfg->lsize) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "isHeap", pCfg->isHeapAllocator) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "ttl", pCfg->ttl) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "keep", pCfg->keep) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "streamMode", pCfg->streamMode) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "isWeak", pCfg->isWeak) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "precision", pCfg->tsdbCfg.precision) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "update", pCfg->tsdbCfg.update) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "compression", pCfg->tsdbCfg.compression) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "daysPerFile", pCfg->tsdbCfg.days) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "minRows", pCfg->tsdbCfg.minRows) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "maxRows", pCfg->tsdbCfg.maxRows) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "keep0", pCfg->tsdbCfg.keep0) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "keep1", pCfg->tsdbCfg.keep1) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "keep2", pCfg->tsdbCfg.keep2) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "lruCacheSize", pCfg->tsdbCfg.lruCacheSize) < 0) return -1;
return 0; return 0;
} }
static int vnodeDecodeInfo(uint8_t *pData, int len, SVnodeInfo *pInfo) { static int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
// TODO SVnodeCfg *pCfg = (SVnodeCfg *)pObj;
if (tjsonGetNumberValue(pJson, "vgId", pCfg->vgId) < 0) return -1;
if (tjsonGetNumberValue(pJson, "dbId", pCfg->dbId) < 0) return -1;
if (tjsonGetNumberValue(pJson, "wsize", pCfg->wsize) < 0) return -1;
if (tjsonGetNumberValue(pJson, "ssize", pCfg->ssize) < 0) return -1;
if (tjsonGetNumberValue(pJson, "lsize", pCfg->lsize) < 0) return -1;
if (tjsonGetNumberValue(pJson, "isHeap", pCfg->isHeapAllocator) < 0) return -1;
if (tjsonGetNumberValue(pJson, "ttl", pCfg->ttl) < 0) return -1;
if (tjsonGetNumberValue(pJson, "keep", pCfg->keep) < 0) return -1;
if (tjsonGetNumberValue(pJson, "streamMode", pCfg->streamMode) < 0) return -1;
if (tjsonGetNumberValue(pJson, "isWeak", pCfg->isWeak) < 0) return -1;
if (tjsonGetNumberValue(pJson, "precision", pCfg->tsdbCfg.precision) < 0) return -1;
if (tjsonGetNumberValue(pJson, "update", pCfg->tsdbCfg.update) < 0) return -1;
if (tjsonGetNumberValue(pJson, "compression", pCfg->tsdbCfg.compression) < 0) return -1;
if (tjsonGetNumberValue(pJson, "daysPerFile", pCfg->tsdbCfg.days) < 0) return -1;
if (tjsonGetNumberValue(pJson, "minRows", pCfg->tsdbCfg.minRows) < 0) return -1;
if (tjsonGetNumberValue(pJson, "maxRows", pCfg->tsdbCfg.maxRows) < 0) return -1;
if (tjsonGetNumberValue(pJson, "keep0", pCfg->tsdbCfg.keep0) < 0) return -1;
if (tjsonGetNumberValue(pJson, "keep1", pCfg->tsdbCfg.keep1) < 0) return -1;
if (tjsonGetNumberValue(pJson, "keep2", pCfg->tsdbCfg.keep2) < 0) return -1;
if (tjsonGetNumberValue(pJson, "lruCacheSize", pCfg->tsdbCfg.lruCacheSize) < 0) return -1;
return 0; return 0;
} }
static int vnodeEncodeState(const void *pObj, SJson *pJson) {
const SVState *pState = (SVState *)pObj;
if (tjsonAddIntegerToObject(pJson, "commit version", pState->committed) < 0) return -1;
return 0;
}
static int vnodeDecodeState(const SJson *pJson, void *pObj) {
SVState *pState = (SVState *)pObj;
if (tjsonGetNumberValue(pJson, "commit version", pState->committed) < 0) return -1;
return 0;
}
static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData) {
SJson *pJson;
char *pData;
*ppData = NULL;
pJson = tjsonCreateObject();
if (pJson == NULL) {
return -1;
}
if (tjsonAddObject(pJson, "config", vnodeEncodeConfig, (void *)&pInfo->config) < 0) {
goto _err;
}
if (tjsonAddObject(pJson, "state", vnodeEncodeState, (void *)&pInfo->state) < 0) {
goto _err;
}
pData = tjsonToString(pJson);
if (pData == NULL) {
goto _err;
}
tjsonDelete(pJson);
*ppData = pData;
return 0;
_err:
tjsonDelete(pJson);
return -1;
}
static int vnodeDecodeInfo(uint8_t *pData, SVnodeInfo *pInfo) {
SJson *pJson = NULL;
pJson = tjsonCreateObject();
if (pJson == NULL) {
return -1;
}
if (tjsonToObject(pJson, "config", vnodeDecodeConfig, (void *)&pInfo->config) < 0) {
goto _err;
}
if (tjsonToObject(pJson, "state", vnodeDecodeState, (void *)&pInfo->state) < 0) {
goto _err;
}
tjsonDelete(pJson);
return 0;
_err:
tjsonDelete(pJson);
return -1;
}

2
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -46,6 +46,8 @@ int vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
return -1; return -1;
} }
vInfo("vgId: %d vnode is created", pCfg->vgId);
return 0; return 0;
} }

20
source/libs/scheduler/src/scheduler.c
View File

@ -14,12 +14,12 @@
*/ */
#include "catalog.h" #include "catalog.h"
#include "command.h"
#include "query.h" #include "query.h"
#include "schedulerInt.h" #include "schedulerInt.h"
#include "tmsg.h" #include "tmsg.h"
#include "tref.h" #include "tref.h"
#include "trpc.h" #include "trpc.h"
#include "command.h"
SSchedulerMgmt schMgmt = {0}; SSchedulerMgmt schMgmt = {0};
@ -141,7 +141,6 @@ _return:
SCH_RET(code); SCH_RET(code);
} }
void schFreeRpcCtx(SRpcCtx *pCtx) { void schFreeRpcCtx(SRpcCtx *pCtx) {
if (NULL == pCtx) { if (NULL == pCtx) {
return; return;
@ -1254,7 +1253,8 @@ int32_t schGetTaskFromTaskList(SHashObj *pTaskList, uint64_t taskId, SSchTask **
} }
int32_t schUpdateTaskExecNodeHandle(SSchTask *pTask, void *handle, int32_t rspCode) { int32_t schUpdateTaskExecNodeHandle(SSchTask *pTask, void *handle, int32_t rspCode) {
if (rspCode || NULL == pTask->execNodes || taosArrayGetSize(pTask->execNodes) > 1 || taosArrayGetSize(pTask->execNodes) <= 0) { if (rspCode || NULL == pTask->execNodes || taosArrayGetSize(pTask->execNodes) > 1 ||
taosArrayGetSize(pTask->execNodes) <= 0) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1264,7 +1264,6 @@ int32_t schUpdateTaskExecNodeHandle(SSchTask *pTask, void *handle, int32_t rspCo
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t msgType, int32_t rspCode) { int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t msgType, int32_t rspCode) {
int32_t code = 0; int32_t code = 0;
SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param; SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
@ -1282,13 +1281,15 @@ int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t msgType, in
if (TDMT_VND_EXPLAIN_RSP == msgType) { if (TDMT_VND_EXPLAIN_RSP == msgType) {
schGetTaskFromTaskList(pJob->succTasks, pParam->taskId, &pTask); schGetTaskFromTaskList(pJob->succTasks, pParam->taskId, &pTask);
} else { } else {
SCH_JOB_ELOG("task not found in execTask list, refId:%" PRIx64 ", taskId:%" PRIx64, pParam->refId, pParam->taskId); SCH_JOB_ELOG("task not found in execTask list, refId:%" PRIx64 ", taskId:%" PRIx64, pParam->refId,
pParam->taskId);
SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR); SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
} }
} }
if (NULL == pTask) { if (NULL == pTask) {
SCH_JOB_ELOG("task not found in execList & succList, refId:%" PRIx64 ", taskId:%" PRIx64, pParam->refId, pParam->taskId); SCH_JOB_ELOG("task not found in execList & succList, refId:%" PRIx64 ", taskId:%" PRIx64, pParam->refId,
pParam->taskId);
SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR); SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
} }
@ -2682,7 +2683,6 @@ int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t scheduleCancelJob(int64_t job) { int32_t scheduleCancelJob(int64_t job) {
SSchJob *pJob = schAcquireJob(job); SSchJob *pJob = schAcquireJob(job);
if (NULL == pJob) { if (NULL == pJob) {
@ -2734,15 +2734,17 @@ void schedulerFreeTaskList(SArray *taskList) {
void schedulerDestroy(void) { void schedulerDestroy(void) {
if (schMgmt.jobRef) { if (schMgmt.jobRef) {
SSchJob *pJob = taosIterateRef(schMgmt.jobRef, 0); SSchJob *pJob = taosIterateRef(schMgmt.jobRef, 0);
int64_t refId = 0;
while (pJob) { while (pJob) {
refId = pJob->refId;
taosRemoveRef(schMgmt.jobRef, pJob->refId); taosRemoveRef(schMgmt.jobRef, pJob->refId);
pJob = taosIterateRef(schMgmt.jobRef, pJob->refId); pJob = taosIterateRef(schMgmt.jobRef, refId);
} }
taosCloseRef(schMgmt.jobRef); taosCloseRef(schMgmt.jobRef);
schMgmt.jobRef = 0; schMgmt.jobRef = 0;
} }
} }