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

feat: add vgroup epset for tsma

This commit is contained in:
Cary Xu 2022-06-06 13:47:32 +08:00
parent e75f14b98b
commit 79769dd2ad
4 changed files with 142 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tmsg.h
View File

@ -2319,7 +2319,7 @@ typedef struct {
int64_t sliding;
char* expr; // sma expression
char* tagsFilter;
SVgEpSet vgEpSet[];
SVgEpSet* pVgEpSet;
} STSma; // Time-range-wise SMA
typedef STSma SVCreateTSmaReq;

32
source/common/src/tmsg.c
View File

@ -694,7 +694,6 @@ void tFreeSMAltertbReq(SMAlterStbReq *pReq) {
pReq->pFields = NULL;
}
int32_t tSerializeSEpSet(void *buf, int32_t bufLen, const SEpSet *pEpset) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
@ -3674,12 +3673,12 @@ int32_t tEncodeTSma(SEncoder *pCoder, const STSma *pSma) {
if (tEncodeCStr(pCoder, pSma->tagsFilter) < 0) return -1;
}
for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
if (tEncodeI32(pCoder, pSma->vgEpSet[v].vgId) < 0) return -1;
if (tEncodeI8(pCoder, pSma->vgEpSet[v].epSet.inUse) < 0) return -1;
int8_t numOfEps = pSma->vgEpSet[v].epSet.numOfEps;
if (tEncodeI32(pCoder, pSma->pVgEpSet[v].vgId) < 0) return -1;
if (tEncodeI8(pCoder, pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
if (tEncodeI8(pCoder, numOfEps) < 0) return -1;
for (int32_t n = 0; n < numOfEps; ++n) {
const SEp *pEp = &pSma->vgEpSet[v].epSet.eps[n];
const SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
if (tEncodeCStr(pCoder, pEp->fqdn) < 0) return -1;
if (tEncodeU16(pCoder, pEp->port) < 0) return -1;
}
@ -3712,17 +3711,27 @@ int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
} else {
pSma->tagsFilter = NULL;
}
if (pSma->numOfVgroups > 0) {
pSma->pVgEpSet = (SVgEpSet *)tDecoderMalloc(pCoder, pSma->numOfVgroups * sizeof(SVgEpSet));
if (!pSma->pVgEpSet) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
memset(pSma->pVgEpSet, 0, pSma->numOfVgroups * sizeof(SVgEpSet));
for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
if (tDecodeI32(pCoder, &pSma->vgEpSet[v].vgId) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->vgEpSet[v].epSet.inUse) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->vgEpSet[v].epSet.numOfEps) < 0) return -1;
int8_t numOfEps = pSma->vgEpSet[v].epSet.numOfEps;
if (tDecodeI32(pCoder, &pSma->pVgEpSet[v].vgId) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.numOfEps) < 0) return -1;
int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
for (int32_t n = 0; n < numOfEps; ++n) {
SEp *pEp = &pSma->vgEpSet[v].epSet.eps[n];
SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
if (tDecodeCStrTo(pCoder, pEp->fqdn) < 0) return -1;
if (tDecodeU16(pCoder, &pEp->port) < 0) return -1;
}
}
}
return 0;
}
@ -4569,6 +4578,3 @@ void tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp) {
taosMemoryFree(pRsp->pMeta);
}
}

2
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -316,10 +316,12 @@ typedef struct {
int32_t tagsFilterLen;
int32_t sqlLen;
int32_t astLen;
int32_t numOfVgroups;
char* expr;
char* tagsFilter;
char* sql;
char* ast;
SVgEpSet* pVgEpSet;
} SSmaObj;
typedef struct {

61
source/dnode/mnode/impl/src/mndSma.c
View File

@ -36,6 +36,7 @@ static SSdbRow *mndSmaActionDecode(SSdbRaw *pRaw);
static int32_t mndSmaActionInsert(SSdb *pSdb, SSmaObj *pSma);
static int32_t mndSmaActionDelete(SSdb *pSdb, SSmaObj *pSpSmatb);
static int32_t mndSmaActionUpdate(SSdb *pSdb, SSmaObj *pOld, SSmaObj *pNew);
static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups);
static int32_t mndProcessMCreateSmaReq(SRpcMsg *pReq);
static int32_t mndProcessMDropSmaReq(SRpcMsg *pReq);
static int32_t mndProcessGetSmaReq(SRpcMsg *pReq);
@ -262,6 +263,8 @@ static void *mndBuildVCreateSmaReq(SMnode *pMnode, SVgObj *pVgroup, SSmaObj *pSm
req.sliding = pSma->sliding;
req.expr = pSma->expr;
req.tagsFilter = pSma->tagsFilter;
req.numOfVgroups = pSma->numOfVgroups;
req.pVgEpSet = pSma->pVgEpSet;
// get length
int32_t ret = 0;
@ -420,6 +423,15 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
mndReleaseDnode(pMnode, pDnode);
// todo add sma info here
SVgEpSet *pVgEpSet = NULL;
int32_t numOfVgroups = 0;
if (mndSmaGetVgEpSet(pMnode, pDb, &pVgEpSet, &numOfVgroups) != 0) {
return -1;
}
pSma->pVgEpSet = pVgEpSet;
pSma->numOfVgroups = numOfVgroups;
int32_t smaContLen = 0;
void *pSmaReq = mndBuildVCreateSmaReq(pMnode, pVgroup, pSma, &smaContLen);
if (pSmaReq == NULL) return -1;
@ -964,3 +976,52 @@ static void mndCancelGetNextSma(SMnode *pMnode, void *pIter) {
SSdb *pSdb = pMnode->pSdb;
sdbCancelFetch(pSdb, pIter);
}
static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups) {
SSdb *pSdb = pMnode->pSdb;
SVgObj *pVgroup = NULL;
void *pIter = NULL;
SVgEpSet *pVgEpSet = NULL;
int32_t nAllocVgs = 16;
int32_t nVgs = 0;
pVgEpSet = taosMemoryCalloc(nAllocVgs, sizeof(SVgEpSet));
if (!pVgEpSet) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
if (pVgroup->dbUid != pDb->uid) {
sdbRelease(pSdb, pVgroup);
continue;
}
if (nVgs >= nAllocVgs) {
void *p = taosMemoryRealloc(pVgEpSet, nAllocVgs * 2 * sizeof(SVgEpSet));
if (!p) {
taosMemoryFree(pVgEpSet);
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pVgroup);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pVgEpSet = (SVgEpSet *)p;
nAllocVgs *= 2;
}
(pVgEpSet + nVgs)->vgId = pVgroup->vgId;
(pVgEpSet + nVgs)->epSet = mndGetVgroupEpset(pMnode, pVgroup);
++nVgs;
sdbRelease(pSdb, pVgroup);
}
*ppVgEpSet = pVgEpSet;
*numOfVgroups = nVgs;
return 0;
}