提交 9b4c8811 编写于 作者: K kailixu

chore: rsma logic optimization

上级 58477d3b
...@@ -225,7 +225,6 @@ int32_t tdProcessRSmaSubmit(SSma* pSma, void* pMsg, int32_t inputType); ...@@ -225,7 +225,6 @@ int32_t tdProcessRSmaSubmit(SSma* pSma, void* pMsg, int32_t inputType);
int32_t tdProcessRSmaDrop(SSma* pSma, SVDropStbReq* pReq); int32_t tdProcessRSmaDrop(SSma* pSma, SVDropStbReq* pReq);
int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid); int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid);
int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd); int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd);
void tdUidStoreDestory(STbUidStore* pStore);
void* tdUidStoreFree(STbUidStore* pStore); void* tdUidStoreFree(STbUidStore* pStore);
// SMetaSnapReader ======================================== // SMetaSnapReader ========================================
......
...@@ -29,9 +29,9 @@ SSmaMgmt smaMgmt = { ...@@ -29,9 +29,9 @@ SSmaMgmt smaMgmt = {
#define TD_QTASKINFO_FNAME_PREFIX "qinf.v" #define TD_QTASKINFO_FNAME_PREFIX "qinf.v"
typedef struct SRSmaQTaskInfoItem SRSmaQTaskInfoItem; typedef struct SRSmaQTaskInfoItem SRSmaQTaskInfoItem;
typedef struct SRSmaQTaskInfoIter SRSmaQTaskInfoIter;
static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid); static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid);
static void tdUidStoreDestory(STbUidStore *pStore);
static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids, bool isAdd); static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids, bool isAdd);
static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo, static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo,
int8_t idx); int8_t idx);
...@@ -57,18 +57,6 @@ struct SRSmaQTaskInfoItem { ...@@ -57,18 +57,6 @@ struct SRSmaQTaskInfoItem {
void *qTaskInfo; void *qTaskInfo;
}; };
struct SRSmaQTaskInfoIter {
STFile *pTFile;
int64_t offset;
int64_t fsize;
int32_t nBytes;
int32_t nAlloc;
char *pBuf;
// ------------
char *qBuf; // for iterator
int32_t nBufPos;
};
void tdRSmaQTaskInfoGetFileName(int32_t vgId, int64_t version, char *outputName) { void tdRSmaQTaskInfoGetFileName(int32_t vgId, int64_t version, char *outputName) {
tdGetVndFileName(vgId, NULL, VNODE_RSMA_DIR, TD_QTASKINFO_FNAME_PREFIX, version, outputName); tdGetVndFileName(vgId, NULL, VNODE_RSMA_DIR, TD_QTASKINFO_FNAME_PREFIX, version, outputName);
} }
...@@ -449,8 +437,8 @@ int32_t tdProcessRSmaCreate(SSma *pSma, SVCreateStbReq *pReq) { ...@@ -449,8 +437,8 @@ int32_t tdProcessRSmaCreate(SSma *pSma, SVCreateStbReq *pReq) {
} }
if (!VND_IS_RSMA(pVnode)) { if (!VND_IS_RSMA(pVnode)) {
smaTrace("vgId:%d, not create rsma for stable %s %" PRIi64 " since vnd is not rsma", TD_VID(pVnode), pReq->name, smaWarn("vgId:%d, not create rsma for stable %s %" PRIi64 " since vnd is not rsma", TD_VID(pVnode), pReq->name,
pReq->suid); pReq->suid);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -494,9 +482,8 @@ int32_t tdProcessRSmaDrop(SSma *pSma, SVDropStbReq *pReq) { ...@@ -494,9 +482,8 @@ int32_t tdProcessRSmaDrop(SSma *pSma, SVDropStbReq *pReq) {
tdReleaseRSmaInfo(pSma, pRSmaInfo); tdReleaseRSmaInfo(pSma, pRSmaInfo);
// save to file // no need to save to file as triggered by dropping stable
// TODO smaDebug("vgId:%d, drop rsma for stable %" PRIi64 " succeed", TD_VID(pVnode), pReq->suid);
smaDebug("vgId:%d, drop rsma for table %" PRIi64 " succeed", TD_VID(pVnode), pReq->suid);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -561,7 +548,7 @@ static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid) ...@@ -561,7 +548,7 @@ static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid)
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void tdUidStoreDestory(STbUidStore *pStore) { static void tdUidStoreDestory(STbUidStore *pStore) {
if (pStore) { if (pStore) {
if (pStore->uidHash) { if (pStore->uidHash) {
if (pStore->tbUids) { if (pStore->tbUids) {
...@@ -602,7 +589,7 @@ static int32_t tdProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) { ...@@ -602,7 +589,7 @@ static int32_t tdProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
} }
SSubmitReq *pSubmitReq = (SSubmitReq *)pReq; SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;
// TODO: spin lock for race conditiond // TODO: spin lock for race condition
if (tsdbInsertData(pTsdb, version, pSubmitReq, NULL) < 0) { if (tsdbInsertData(pTsdb, version, pSubmitReq, NULL) < 0) {
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
...@@ -871,6 +858,12 @@ static int32_t tdCloneQTaskInfo(SSma *pSma, qTaskInfo_t dstTaskInfo, qTaskInfo_t ...@@ -871,6 +858,12 @@ static int32_t tdCloneQTaskInfo(SSma *pSma, qTaskInfo_t dstTaskInfo, qTaskInfo_t
char *pOutput = NULL; char *pOutput = NULL;
int32_t len = 0; int32_t len = 0;
if (!srcTaskInfo) {
terrno = TSDB_CODE_INVALID_PTR;
smaWarn("vgId:%d, rsma clone, table %" PRIi64 ", no need since srcTaskInfo is NULL", TD_VID(pVnode), suid);
return TSDB_CODE_FAILED;
}
if ((terrno = qSerializeTaskStatus(srcTaskInfo, &pOutput, &len)) < 0) { if ((terrno = qSerializeTaskStatus(srcTaskInfo, &pOutput, &len)) < 0) {
smaError("vgId:%d, rsma clone, table %" PRIi64 " serialize qTaskInfo failed since %s", TD_VID(pVnode), suid, smaError("vgId:%d, rsma clone, table %" PRIi64 " serialize qTaskInfo failed since %s", TD_VID(pVnode), suid,
terrstr()); terrstr());
...@@ -1051,12 +1044,8 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, void *pMsg, int32_t inputType) { ...@@ -1051,12 +1044,8 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, void *pMsg, int32_t inputType) {
// only applicable when rsma env exists // only applicable when rsma env exists
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
STbUidStore uidStore = {0}; STbUidStore uidStore = {0};
SRetention *pRetention = SMA_RETENTION(pSma);
if (!RETENTION_VALID(pRetention + 1)) {
// return directly if retention level 1 is invalid
return TSDB_CODE_SUCCESS;
}
if (inputType == STREAM_INPUT__DATA_SUBMIT) { if (inputType == STREAM_INPUT__DATA_SUBMIT) {
if (tdFetchSubmitReqSuids(pMsg, &uidStore) < 0) { if (tdFetchSubmitReqSuids(pMsg, &uidStore) < 0) {
...@@ -1191,10 +1180,12 @@ _err: ...@@ -1191,10 +1180,12 @@ _err:
* @param pSma * @param pSma
* @return int32_t * @return int32_t
*/ */
#if 0
static int32_t tdRSmaRestoreTSDataReload(SSma *pSma) { static int32_t tdRSmaRestoreTSDataReload(SSma *pSma) {
// NOTHING TODO: the data would be restored from the unified WAL replay procedure // NOTHING TODO: the data would be restored from the unified WAL replay procedure
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
#endif
int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer) { int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer) {
// step 1: iterate all stables to restore the rsma env // step 1: iterate all stables to restore the rsma env
...@@ -1208,9 +1199,11 @@ int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer) ...@@ -1208,9 +1199,11 @@ int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer)
} }
// step 2: reload ts data from checkpoint // step 2: reload ts data from checkpoint
#if 0
if (tdRSmaRestoreTSDataReload(pSma) < 0) { if (tdRSmaRestoreTSDataReload(pSma) < 0) {
goto _err; goto _err;
} }
#endif
// step 3: open SRSmaFS for qTaskFiles // step 3: open SRSmaFS for qTaskFiles
if (tdRSmaFSOpen(pSma, qtaskFileVer) < 0) { if (tdRSmaFSOpen(pSma, qtaskFileVer) < 0) {
......
...@@ -3041,15 +3041,15 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* ret ...@@ -3041,15 +3041,15 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* ret
if (level == TSDB_RETENTION_L0) { if (level == TSDB_RETENTION_L0) {
*pLevel = TSDB_RETENTION_L0; *pLevel = TSDB_RETENTION_L0;
tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L0, str); tsdbInfo("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L0, str);
return VND_RSMA0(pVnode); return VND_RSMA0(pVnode);
} else if (level == TSDB_RETENTION_L1) { } else if (level == TSDB_RETENTION_L1) {
*pLevel = TSDB_RETENTION_L1; *pLevel = TSDB_RETENTION_L1;
tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L1, str); tsdbInfo("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L1, str);
return VND_RSMA1(pVnode); return VND_RSMA1(pVnode);
} else { } else {
*pLevel = TSDB_RETENTION_L2; *pLevel = TSDB_RETENTION_L2;
tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L2, str); tsdbInfo("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L2, str);
return VND_RSMA2(pVnode); return VND_RSMA2(pVnode);
} }
} }
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册