From 6b69b286494e68ed46e1fa810d3615a77bec9451 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 2 Jun 2022 16:36:48 +0800 Subject: [PATCH 1/6] enh: refactor tag filter code --- include/common/ttypes.h | 86 +++++++++++++------------ source/dnode/vnode/src/meta/metaQuery.c | 30 +++++---- source/dnode/vnode/src/meta/metaTable.c | 69 ++++++++++---------- source/libs/index/src/indexFilter.c | 79 ++++++++++++++--------- source/libs/transport/src/transCli.c | 1 - 5 files changed, 149 insertions(+), 116 deletions(-) diff --git a/include/common/ttypes.h b/include/common/ttypes.h index 31cdb28690..3602b045a9 100644 --- a/include/common/ttypes.h +++ b/include/common/ttypes.h @@ -30,7 +30,7 @@ typedef uint64_t TDRowVerT; typedef int16_t col_id_t; typedef int8_t col_type_t; typedef int32_t col_bytes_t; -typedef int32_t schema_ver_t; +typedef int32_t schema_ver_t; typedef int32_t func_id_t; #pragma pack(push, 1) @@ -49,8 +49,9 @@ typedef struct { #define varDataCopy(dst, v) memcpy((dst), (void *)(v), varDataTLen(v)) #define varDataLenByData(v) (*(VarDataLenT *)(((char *)(v)) - VARSTR_HEADER_SIZE)) #define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT)(_len)) -#define IS_VAR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON)) -#define IS_STR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR)) +#define IS_VAR_DATA_TYPE(t) \ + (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON)) +#define IS_STR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR)) #define varDataNetLen(v) (htons(((VarDataLenT *)(v))[0])) #define varDataNetTLen(v) (sizeof(VarDataLenT) + varDataNetLen(v)) @@ -142,44 +143,44 @@ typedef struct { } \ } while (0) -#define NUM_TO_STRING(_inputType, _input, _outputBytes, _output) \ - do { \ - switch (_inputType) { \ - case TSDB_DATA_TYPE_TINYINT: \ - snprintf(_output, (int32_t)(_outputBytes), "%d", *(int8_t *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_UTINYINT: \ - snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint8_t *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_SMALLINT: \ - snprintf(_output, (int32_t)(_outputBytes), "%d", *(int16_t *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_USMALLINT: \ - snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint16_t *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_TIMESTAMP: \ - case TSDB_DATA_TYPE_BIGINT: \ - snprintf(_output, (int32_t)(_outputBytes), "%" PRId64, *(int64_t *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_UBIGINT: \ - snprintf(_output, (int32_t)(_outputBytes), "%" PRIu64, *(uint64_t *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_FLOAT: \ - snprintf(_output, (int32_t)(_outputBytes), "%f", *(float *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_DOUBLE: \ - snprintf(_output, (int32_t)(_outputBytes), "%f", *(double *)(_input)); \ - break; \ - case TSDB_DATA_TYPE_UINT: \ - snprintf(_output, (int32_t)(_outputBytes), "%u", *(uint32_t *)(_input)); \ - break; \ - default: \ - snprintf(_output, (int32_t)(_outputBytes), "%d", *(int32_t *)(_input)); \ - break; \ - } \ +#define NUM_TO_STRING(_inputType, _input, _outputBytes, _output) \ + do { \ + switch (_inputType) { \ + case TSDB_DATA_TYPE_TINYINT: \ + snprintf(_output, (int32_t)(_outputBytes), "%d", *(int8_t *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_UTINYINT: \ + snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint8_t *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_SMALLINT: \ + snprintf(_output, (int32_t)(_outputBytes), "%d", *(int16_t *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_USMALLINT: \ + snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint16_t *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_TIMESTAMP: \ + case TSDB_DATA_TYPE_BIGINT: \ + snprintf(_output, (int32_t)(_outputBytes), "%" PRId64, *(int64_t *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_UBIGINT: \ + snprintf(_output, (int32_t)(_outputBytes), "%" PRIu64, *(uint64_t *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_FLOAT: \ + snprintf(_output, (int32_t)(_outputBytes), "%f", *(float *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_DOUBLE: \ + snprintf(_output, (int32_t)(_outputBytes), "%f", *(double *)(_input)); \ + break; \ + case TSDB_DATA_TYPE_UINT: \ + snprintf(_output, (int32_t)(_outputBytes), "%u", *(uint32_t *)(_input)); \ + break; \ + default: \ + snprintf(_output, (int32_t)(_outputBytes), "%d", *(int32_t *)(_input)); \ + break; \ + } \ } while (0) -//TODO: use varchar(0) to represent NULL type +// TODO: use varchar(0) to represent NULL type #define IS_NULL_TYPE(_t) ((_t) == TSDB_DATA_TYPE_NULL) #define IS_SIGNED_NUMERIC_TYPE(_t) ((_t) >= TSDB_DATA_TYPE_TINYINT && (_t) <= TSDB_DATA_TYPE_BIGINT) #define IS_UNSIGNED_NUMERIC_TYPE(_t) ((_t) >= TSDB_DATA_TYPE_UTINYINT && (_t) <= TSDB_DATA_TYPE_UBIGINT) @@ -187,7 +188,8 @@ typedef struct { #define IS_INTEGER_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t))) #define IS_NUMERIC_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)) || (IS_FLOAT_TYPE(_t))) -#define IS_MATHABLE_TYPE(_t) (IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP)) +#define IS_MATHABLE_TYPE(_t) \ + (IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP)) #define IS_VALID_TINYINT(_t) ((_t) >= INT8_MIN && (_t) <= INT8_MAX) #define IS_VALID_SMALLINT(_t) ((_t) >= INT16_MIN && (_t) <= INT16_MAX) @@ -244,7 +246,7 @@ typedef struct tDataTypeDescriptor { int16_t type; int16_t nameLen; int32_t bytes; - char *name; + char * name; int64_t minValue; int64_t maxValue; int32_t (*compFunc)(const char *const input, int32_t inputSize, const int32_t nelements, char *const output, @@ -277,4 +279,4 @@ void *getDataMax(int32_t type); } #endif -#endif /*_TD_COMMON_TTYPE_H_*/ +#endif /*_TD_COMMON_TTYPE_H_*/ diff --git a/source/dnode/vnode/src/meta/metaQuery.c b/source/dnode/vnode/src/meta/metaQuery.c index a57eb4e899..7928f715e2 100644 --- a/source/dnode/vnode/src/meta/metaQuery.c +++ b/source/dnode/vnode/src/meta/metaQuery.c @@ -576,15 +576,15 @@ SArray *metaGetSmaTbUids(SMeta *pMeta) { const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t type, STagVal *val) { ASSERT(pEntry->type == TSDB_CHILD_TABLE); STag *tag = (STag *)pEntry->ctbEntry.pTags; - if (type == TSDB_DATA_TYPE_JSON){ - if(tag->nTag == 0){ + if (type == TSDB_DATA_TYPE_JSON) { + if (tag->nTag == 0) { return NULL; } return tag; } bool find = tTagGet(tag, val); - if(!find){ + if (!find) { return NULL; } return val; @@ -605,8 +605,6 @@ typedef struct { int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) { SIdxCursor *pCursor = NULL; - char *tagData = param->val; - int32_t ret = 0, valid = 0; pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor)); pCursor->pMeta = pMeta; @@ -623,12 +621,16 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) { int32_t nKey = 0; int32_t nTagData = 0; - if(IS_VAR_DATA_TYPE(param->type)){ - nTagData = strlen(param->val); - }else{ + void * tagData = NULL; + + if (IS_VAR_DATA_TYPE(param->type)) { + tagData = varDataVal(param->val); + nTagData = varDataLen(param->val); + } else { + tagData = param->val; nTagData = tDataTypes[param->type].bytes; } - ret = metaCreateTagIdxKey(pCursor->suid, pCursor->cid, param->val, nTagData, pCursor->type, + ret = metaCreateTagIdxKey(pCursor->suid, pCursor->cid, tagData, nTagData, pCursor->type, param->reverse ? INT64_MAX : INT64_MIN, &pKey, &nKey); if (ret != 0) { goto END; @@ -637,6 +639,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) { if (tdbTbcMoveTo(pCursor->pCur, pKey, nKey, &cmp) < 0) { goto END; } + void * entryKey = NULL, *entryVal = NULL; int32_t nEntryKey, nEntryVal; while (1) { @@ -649,7 +652,12 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) { int32_t cmp = (*param->filterFunc)(p->data, pKey->data, pKey->type); if (cmp == 0) { // match - tb_uid_t tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes); + tb_uid_t tuid = 0; + if (IS_VAR_DATA_TYPE(pKey->type)) { + tuid = *(tb_uid_t *)(p->data + varDataTLen(p->data)); + } else { + tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes); + } taosArrayPush(pUids, &tuid); } else if (cmp == 1) { // not match but should continue to iter @@ -670,4 +678,4 @@ END: taosMemoryFree(pCursor); return ret; -} \ No newline at end of file +} diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index a300741942..4e0b4b8b44 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -31,9 +31,9 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { int vLen = 0; const void *pKey = NULL; const void *pVal = NULL; - void *pBuf = NULL; + void * pBuf = NULL; int32_t szBuf = 0; - void *p = NULL; + void * p = NULL; SMetaReader mr = {0}; // validate req @@ -87,7 +87,7 @@ int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) { } // drop all child tables - TBC *pCtbIdxc = NULL; + TBC * pCtbIdxc = NULL; SArray *pArray = taosArrayInit(8, sizeof(tb_uid_t)); tdbTbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn); @@ -142,8 +142,8 @@ _exit: int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { SMetaEntry oStbEntry = {0}; SMetaEntry nStbEntry = {0}; - TBC *pUidIdxc = NULL; - TBC *pTbDbc = NULL; + TBC * pUidIdxc = NULL; + TBC * pTbDbc = NULL; const void *pData; int nData; int64_t oversion; @@ -262,7 +262,7 @@ _err: } int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) { - void *pData = NULL; + void * pData = NULL; int nData = 0; int rc = 0; tb_uid_t uid; @@ -288,7 +288,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi } static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { - void *pData = NULL; + void * pData = NULL; int nData = 0; int rc = 0; int64_t version; @@ -324,14 +324,14 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { } static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) { - void *pVal = NULL; + void * pVal = NULL; int nVal = 0; - const void *pData = NULL; + const void * pData = NULL; int nData = 0; int ret = 0; tb_uid_t uid; int64_t oversion; - SSchema *pColumn = NULL; + SSchema * pColumn = NULL; SMetaEntry entry = {0}; SSchemaWrapper *pSchema; int c; @@ -479,7 +479,7 @@ _err: static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) { SMetaEntry ctbEntry = {0}; SMetaEntry stbEntry = {0}; - void *pVal = NULL; + void * pVal = NULL; int nVal = 0; int ret; int c; @@ -510,7 +510,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA oversion = *(int64_t *)pData; // search table.db - TBC *pTbDbc = NULL; + TBC * pTbDbc = NULL; SDecoder dc1 = {0}; SDecoder dc2 = {0}; @@ -534,7 +534,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA metaDecodeEntry(&dc2, &stbEntry); SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag; - SSchema *pColumn = NULL; + SSchema * pColumn = NULL; int32_t iCol = 0; for (;;) { pColumn = NULL; @@ -564,8 +564,8 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA memcpy((void *)ctbEntry.ctbEntry.pTags, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal); } else { const STag *pOldTag = (const STag *)ctbEntry.ctbEntry.pTags; - STag *pNewTag = NULL; - SArray *pTagArray = taosArrayInit(pTagSchema->nCols, sizeof(STagVal)); + STag * pNewTag = NULL; + SArray * pTagArray = taosArrayInit(pTagSchema->nCols, sizeof(STagVal)); if (!pTagArray) { terrno = TSDB_CODE_OUT_OF_MEMORY; goto _err; @@ -649,8 +649,8 @@ int metaAlterTable(SMeta *pMeta, int64_t version, SVAlterTbReq *pReq) { static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) { STbDbKey tbDbKey; - void *pKey = NULL; - void *pVal = NULL; + void * pKey = NULL; + void * pVal = NULL; int kLen = 0; int vLen = 0; SEncoder coder = {0}; @@ -733,16 +733,11 @@ static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME) { int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_t nTagData, int8_t type, tb_uid_t uid, STagIdxKey **ppTagIdxKey, int32_t *nTagIdxKey) { - // int32_t nTagData = 0; - - // if (pTagData) { - // if (IS_VAR_DATA_TYPE(type)) { - // nTagData = varDataTLen(pTagData); - // } else { - // nTagData = tDataTypes[type].bytes; - // } - // } - *nTagIdxKey = sizeof(STagIdxKey) + nTagData + sizeof(tb_uid_t); + if (IS_VAR_DATA_TYPE(type)) { + *nTagIdxKey = sizeof(STagIdxKey) + nTagData + VARSTR_HEADER_SIZE + sizeof(tb_uid_t); + } else { + *nTagIdxKey = sizeof(STagIdxKey) + nTagData + sizeof(tb_uid_t); + } *ppTagIdxKey = (STagIdxKey *)taosMemoryMalloc(*nTagIdxKey); if (*ppTagIdxKey == NULL) { @@ -754,8 +749,16 @@ int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_ (*ppTagIdxKey)->cid = cid; (*ppTagIdxKey)->isNull = (pTagData == NULL) ? 1 : 0; (*ppTagIdxKey)->type = type; - if (nTagData) memcpy((*ppTagIdxKey)->data, pTagData, nTagData); - *(tb_uid_t *)((*ppTagIdxKey)->data + nTagData) = uid; + + // refactor + if (IS_VAR_DATA_TYPE(type)) { + memcpy((*ppTagIdxKey)->data, (uint16_t *)&nTagData, VARSTR_HEADER_SIZE); + memcpy((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE, pTagData, nTagData); + *(tb_uid_t *)((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE + nTagData) = uid; + } else { + memcpy((*ppTagIdxKey)->data, pTagData, nTagData); + *(tb_uid_t *)((*ppTagIdxKey)->data + nTagData) = uid; + } return 0; } @@ -765,14 +768,14 @@ static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey) { } static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) { - void *pData = NULL; + void * pData = NULL; int nData = 0; STbDbKey tbDbKey = {0}; SMetaEntry stbEntry = {0}; - STagIdxKey *pTagIdxKey = NULL; + STagIdxKey * pTagIdxKey = NULL; int32_t nTagIdxKey; const SSchema *pTagColumn; // = &stbEntry.stbEntry.schema.pSchema[0]; - const void *pTagData = NULL; // + const void * pTagData = NULL; // int32_t nTagData = 0; SDecoder dc = {0}; @@ -829,7 +832,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) { static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) { SEncoder coder = {0}; - void *pVal = NULL; + void * pVal = NULL; int vLen = 0; int rcode = 0; SSkmDbKey skmDbKey = {0}; diff --git a/source/libs/index/src/indexFilter.c b/source/libs/index/src/indexFilter.c index 766746dd2a..b41006b6dd 100644 --- a/source/libs/index/src/indexFilter.c +++ b/source/libs/index/src/indexFilter.c @@ -265,23 +265,24 @@ static int32_t sifExecFunction(SFunctionNode *node, SIFCtx *ctx, SIFParam *outpu typedef int (*Filter)(void *a, void *b, int16_t dtype); int sifGreaterThan(void *a, void *b, int16_t dtype) { - __compar_fn_t func = indexGetCompar(dtype); + __compar_fn_t func = getComparFunc(dtype, 0); return tDoCompare(func, QUERY_GREATER_THAN, a, b); } int sifGreaterEqual(void *a, void *b, int16_t dtype) { - __compar_fn_t func = indexGetCompar(dtype); + __compar_fn_t func = getComparFunc(dtype, 0); return tDoCompare(func, QUERY_GREATER_EQUAL, a, b); } int sifLessEqual(void *a, void *b, int16_t dtype) { - __compar_fn_t func = indexGetCompar(dtype); + __compar_fn_t func = getComparFunc(dtype, 0); return tDoCompare(func, QUERY_LESS_EQUAL, a, b); } int sifLessThan(void *a, void *b, int16_t dtype) { - __compar_fn_t func = indexGetCompar(dtype); + __compar_fn_t func = getComparFunc(dtype, 0); return (int)tDoCompare(func, QUERY_LESS_THAN, a, b); } int sifEqual(void *a, void *b, int16_t dtype) { - __compar_fn_t func = indexGetCompar(dtype); + __compar_fn_t func = getComparFunc(dtype, 0); + //__compar_fn_t func = indexGetCompar(dtype); return (int)tDoCompare(func, QUERY_TERM, a, b); } static Filter sifGetFilterFunc(EIndexQueryType type, bool *reverse) { @@ -397,38 +398,62 @@ static int32_t sifDefaultFunc(SIFParam *left, SIFParam *right, SIFParam *output) return TSDB_CODE_QRY_INVALID_INPUT; } -static sif_func_t sifGetOperFn(int32_t funcId) { +static int32_t sifGetOperFn(int32_t funcId, sif_func_t *func, SIdxFltStatus *status) { // impl later + *status = SFLT_ACCURATE_INDEX; switch (funcId) { case OP_TYPE_GREATER_THAN: - return sifGreaterThanFunc; + *func = sifGreaterThanFunc; + return 0; case OP_TYPE_GREATER_EQUAL: - return sifGreaterEqualFunc; + *func = sifGreaterEqualFunc; + return 0; case OP_TYPE_LOWER_THAN: - return sifLessThanFunc; + *func = sifLessThanFunc; + return 0; case OP_TYPE_LOWER_EQUAL: - return sifLessEqualFunc; + *func = sifLessEqualFunc; + return 0; case OP_TYPE_EQUAL: - return sifEqualFunc; + *func = sifEqualFunc; + return 0; case OP_TYPE_NOT_EQUAL: - return sifNotEqualFunc; + *status = SFLT_NOT_INDEX; + *func = sifNotEqualFunc; + return 0; case OP_TYPE_IN: - return sifInFunc; + *status = SFLT_NOT_INDEX; + *func = sifInFunc; + return 0; case OP_TYPE_NOT_IN: - return sifNotInFunc; + *status = SFLT_NOT_INDEX; + *func = sifNotInFunc; + return 0; case OP_TYPE_LIKE: - return sifLikeFunc; + *status = SFLT_NOT_INDEX; + *func = sifLikeFunc; + return 0; case OP_TYPE_NOT_LIKE: - return sifNotLikeFunc; + *status = SFLT_NOT_INDEX; + *func = sifNotLikeFunc; + return 0; case OP_TYPE_MATCH: - return sifMatchFunc; + *status = SFLT_NOT_INDEX; + *func = sifMatchFunc; + return 0; case OP_TYPE_NMATCH: - return sifNotMatchFunc; + *status = SFLT_NOT_INDEX; + *func = sifNotMatchFunc; + return 0; default: - return sifNullFunc; + *status = SFLT_NOT_INDEX; + *func = sifNullFunc; + return 0; } - return sifNullFunc; + return 0; } +// typedef struct filterFuncDict { + static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) { int32_t code = 0; int32_t nParam = sifGetOperParamNum(node->opType); @@ -441,18 +466,14 @@ static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) { // ugly code, refactor later output->arg = ctx->arg; - sif_func_t operFn = sifGetOperFn(node->opType); - if (ctx->noExec && operFn == NULL) { - output->status = SFLT_NOT_INDEX; - } else { - output->status = SFLT_ACCURATE_INDEX; - } + sif_func_t operFn = sifNullFunc; + code = sifGetOperFn(node->opType, &operFn, &output->status); if (ctx->noExec) { SIF_RET(code); + } else { + return operFn(¶ms[0], nParam > 1 ? ¶ms[1] : NULL, output); } - - return operFn(¶ms[0], nParam > 1 ? ¶ms[1] : NULL, output); _return: taosMemoryFree(params); SIF_RET(code); @@ -477,7 +498,7 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou } else if (node->condType == LOGIC_COND_TYPE_OR) { taosArrayAddAll(output->result, params[m].result); } else if (node->condType == LOGIC_COND_TYPE_NOT) { - taosArrayAddAll(output->result, params[m].result); + // taosArrayAddAll(output->result, params[m].result); } } } else { diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index d82b1dc540..3b6fc74a53 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -1151,7 +1151,6 @@ void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) { SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i]; tDebug("update epset at thread:%d, threadID:%" PRId64 "", i, thrd->thread); - tsem_t* pSem = pCtx->pSem; transSendAsync(thrd->asyncPool, &(cliMsg->q)); } } From a682999c1d38a3dd70bc890ca33f4391621780e2 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 2 Jun 2022 17:43:03 +0800 Subject: [PATCH 2/6] refactor: adjust vnode queue --- source/dnode/mgmt/mgmt_vnode/inc/vmInt.h | 14 +- source/dnode/mgmt/mgmt_vnode/src/vmFile.c | 10 +- source/dnode/mgmt/mgmt_vnode/src/vmHandle.c | 98 +++++----- source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 14 +- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 182 +++++++----------- source/dnode/mgmt/node_mgmt/src/dmTransport.c | 1 + 6 files changed, 141 insertions(+), 178 deletions(-) diff --git a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h index 5ec33fe810..f74fd119c5 100644 --- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h +++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h @@ -108,13 +108,13 @@ int32_t vmPutRpcMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutRpcMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype); -int32_t vmPutNodeMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutNodeMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutNodeMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutNodeMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutNodeMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutNodeMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutNodeMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); #ifdef __cplusplus } diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c index cf5a7ad885..71f11e0d97 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c @@ -16,6 +16,8 @@ #define _DEFAULT_SOURCE #include "vmInt.h" +#define MAX_CONTENT_LEN 1024 * 1024 + SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) { taosThreadRwlockRdlock(&pMgmt->lock); @@ -47,7 +49,7 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) { int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes) { int32_t code = TSDB_CODE_INVALID_JSON_FORMAT; int32_t len = 0; - int32_t maxLen = 1024 * 1024; + int32_t maxLen = MAX_CONTENT_LEN; char *content = taosMemoryCalloc(1, maxLen + 1); cJSON *root = NULL; FILE *fp = NULL; @@ -128,7 +130,7 @@ int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes = vnodesNum; code = 0; - dDebug("succcessed to read file %s", file); + dDebug("succcessed to read file %s, numOfVnodes:%d", file, vnodesNum); _OVER: if (content != NULL) taosMemoryFree(content); @@ -156,7 +158,7 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) { SVnodeObj **pVnodes = vmGetVnodeListFromHash(pMgmt, &numOfVnodes); int32_t len = 0; - int32_t maxLen = 1024 * 1024; + int32_t maxLen = MAX_CONTENT_LEN; char *content = taosMemoryCalloc(1, maxLen + 1); if (content == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; @@ -195,6 +197,6 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) { taosMemoryFree(pVnodes); } - dDebug("successed to write %s", realfile); + dDebug("successed to write %s, numOfVnodes:%d", realfile, numOfVnodes); return taosRenameFile(file, realfile); } \ No newline at end of file diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index 86087a0165..f43f00e02a 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -297,59 +297,59 @@ SArray *vmGetMsgHandles() { SArray *pArray = taosArrayInit(32, sizeof(SMgmtHandle)); if (pArray == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_MON_VM_INFO, vmPutNodeMsgToMonitorQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_MON_VM_LOAD, vmPutNodeMsgToMonitorQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_MON_VM_INFO, vmPutMsgToMonitorQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_MON_VM_LOAD, vmPutMsgToMonitorQueue, 0) == NULL) goto _OVER; // Requests handled by VNODE - if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_CONTINUE, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_TABLE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONSUME, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_QUERY, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONNECT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_DISCONNECT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - // if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_SET_CUR, vmPutNodeMsgToWriteQueue, 0)== NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_TASK, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_STB, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_STB, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_STB, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_TABLE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TABLE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_SMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_SMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DEPLOY, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RUN, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DISPATCH, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RECOVER, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_CONTINUE, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONSUME, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONNECT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_DISCONNECT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + // if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_SET_CUR, vmPutMsgToWriteQueue, 0)== NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RUN, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DISPATCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RECOVER, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_TIMEOUT, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_TIMEOUT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER; code = 0; diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c index 0c8d492ef4..5d3479f74b 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c @@ -18,21 +18,17 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) { SVnodeObj *pVnode = NULL; - int32_t refCount = 0; taosThreadRwlockRdlock(&pMgmt->lock); taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode); if (pVnode == NULL) { terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; } else { - refCount = atomic_add_fetch_32(&pVnode->refCount, 1); + int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1); + dTrace("vgId:%d, acquire vnode, ref:%d", pVnode->vgId, refCount); } taosThreadRwlockUnlock(&pMgmt->lock); - if (pVnode != NULL) { - dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount); - } - return pVnode; } @@ -41,8 +37,8 @@ void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { taosThreadRwlockRdlock(&pMgmt->lock); int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1); + dTrace("vgId:%d, release vnode, ref:%d", pVnode->vgId, refCount); taosThreadRwlockUnlock(&pMgmt->lock); - dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount); } int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) { @@ -138,7 +134,7 @@ static void *vmOpenVnodeInThread(void *param) { } } - dDebug("thread:%d, total vnodes:%d, opened:%d failed:%d", pThread->threadIndex, pThread->vnodeNum, pThread->opened, + dDebug("thread:%d, numOfVnodes:%d, opened:%d failed:%d", pThread->threadIndex, pThread->vnodeNum, pThread->opened, pThread->failed); return NULL; } @@ -160,7 +156,7 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) { pMgmt->state.totalVnodes = numOfVnodes; - int32_t threadNum = 1; // tsNumOfCores; + int32_t threadNum = tsNumOfCores; int32_t vnodesPerThread = numOfVnodes / threadNum + 1; SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread)); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index 03db74abd7..e5953b8383 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -29,11 +29,11 @@ static inline void vmSendRsp(SRpcMsg *pMsg, int32_t code) { tmsgSendRsp(&rsp); } -static void vmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { +static void vmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SVnodeMgmt *pMgmt = pInfo->ahandle; int32_t code = -1; - dTrace("msg:%p, get from vnode queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("msg:%p, get from vnode-mgmt queue", pMsg); switch (pMsg->msgType) { case TDMT_MON_VM_INFO: code = vmProcessGetMonitorInfoReq(pMgmt, pMsg); @@ -49,11 +49,14 @@ static void vmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { break; default: terrno = TSDB_CODE_MSG_NOT_PROCESSED; - dError("msg:%p, not processed in vnode queue", pMsg); + dError("msg:%p, not processed in vnode-mgmt queue", pMsg); } if (IsReq(pMsg)) { - if (code != 0 && terrno != 0) code = terrno; + if (code != 0 && terrno != 0) { + dError("msg:%p failed to process since %s", pMsg, terrstr()); + code = terrno; + } vmSendRsp(pMsg, code); } @@ -65,13 +68,15 @@ static void vmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SVnodeObj *pVnode = pInfo->ahandle; - dTrace("msg:%p, get from vnode-query queue", pMsg); + dTrace("vgId:%d, msg:%p get from vnode-query queue", pVnode->vgId, pMsg); int32_t code = vnodeProcessQueryMsg(pVnode->pImpl, pMsg); if (code != 0) { if (terrno != 0) code = terrno; + dError("vgId:%d, msg:%p failed to query since %s", pVnode->vgId, pMsg, terrstr()); vmSendRsp(pMsg, code); } - dTrace("msg:%p, is freed, code:0x%x", pMsg, code); + + dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } @@ -79,13 +84,15 @@ static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { static void vmProcessFetchQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SVnodeObj *pVnode = pInfo->ahandle; - dTrace("msg:%p, get from vnode-fetch queue", pMsg); + dTrace("vgId:%d, msg:%p get from vnode-fetch queue", pVnode->vgId, pMsg); int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo); if (code != 0) { if (terrno != 0) code = terrno; + dError("vgId:%d, msg:%p failed to fetch since %s", pVnode->vgId, pMsg, terrstr()); vmSendRsp(pMsg, code); } - dTrace("msg:%p, is freed, code:0x%x", pMsg, code); + + dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } @@ -99,7 +106,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO for (int32_t m = 0; m < numOfMsgs; m++) { if (taosGetQitem(qall, (void **)&pMsg) == 0) continue; - dTrace("vgId:%d, get msg:%p from vnode-write queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p get from vnode-write queue", pVnode->vgId, pMsg); if (taosArrayPush(pArray, &pMsg) == NULL) { dError("vgId:%d, failed to push msg:%p to vnode-write array", pVnode->vgId, pMsg); @@ -113,7 +120,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO if (code == TSDB_CODE_ACTION_IN_PROGRESS) continue; if (code != 0) { - dError("vgId:%d, msg:%p failed to process since %s", pVnode->vgId, pMsg, tstrerror(code)); + dError("vgId:%d, msg:%p failed to write since %s", pVnode->vgId, pMsg, tstrerror(code)); vmSendRsp(pMsg, code); continue; } @@ -129,14 +136,14 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info}; tmsgSendRedirectRsp(&rsp, &newEpSet); } else { - dError("vgId:%d, msg:%p failed to process since %s", pVnode->vgId, pMsg, tstrerror(code)); + dError("vgId:%d, msg:%p failed to write since %s", pVnode->vgId, pMsg, tstrerror(code)); vmSendRsp(pMsg, code); } } for (int32_t i = 0; i < numOfMsgs; i++) { pMsg = *(SRpcMsg **)taosArrayGet(pArray, i); - dTrace("vgId:%d, msg:%p, is freed", pVnode->vgId, pMsg); + dTrace("vgId:%d, msg:%p is freed", pVnode->vgId, pMsg); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } @@ -146,10 +153,11 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { SVnodeObj *pVnode = pInfo->ahandle; + SRpcMsg *pMsg = NULL; for (int32_t i = 0; i < numOfMsgs; ++i) { - SRpcMsg *pMsg = NULL; - taosGetQitem(qall, (void **)&pMsg); + if (taosGetQitem(qall, (void **)&pMsg) == 0) continue; + dTrace("vgId:%d, msg:%p get from vnode-apply queue", pVnode->vgId, pMsg); // init response rpc msg SRpcMsg rsp = {0}; @@ -164,7 +172,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO // apply data into tsdb if (vnodeProcessWriteReq(pVnode->pImpl, &originalRpcMsg, pSyncApplyMsg->fsmMeta.index, &rsp) < 0) { rsp.code = terrno; - dTrace("msg:%p, process write error since %s", pMsg, terrstr()); + dError("vgId:%d, msg:%p failed to apply since %s", pVnode->vgId, pMsg, terrstr()); } syncApplyMsgDestroy(pSyncApplyMsg); @@ -176,6 +184,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO tmsgSendRsp(&rsp); } + dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, rsp.code); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } @@ -183,23 +192,22 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { SVnodeObj *pVnode = pInfo->ahandle; + SRpcMsg *pMsg = NULL; for (int32_t i = 0; i < numOfMsgs; ++i) { - SRpcMsg *pMsg = NULL; - taosGetQitem(qall, (void **)&pMsg); + if (taosGetQitem(qall, (void **)&pMsg) == 0) continue; + dTrace("vgId:%d, msg:%p get from vnode-sync queue", pVnode->vgId, pMsg); int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL); if (code != 0) { + dError("vgId:%d, msg:%p failed to sync since %s", pVnode->vgId, pMsg, terrstr()); if (pMsg->info.handle != NULL) { - SRpcMsg rsp = { - .code = (terrno < 0) ? terrno : code, - .info = pMsg->info, - }; - dTrace("msg:%p, failed to process sync queue since %s", pMsg, terrstr()); - tmsgSendRsp(&rsp); + if (terrno != 0) code = terrno; + vmSendRsp(pMsg, code); } } + dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } @@ -207,24 +215,26 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { SVnodeObj *pVnode = pInfo->ahandle; + SRpcMsg *pMsg = NULL; for (int32_t i = 0; i < numOfMsgs; ++i) { - SRpcMsg *pMsg = NULL; - taosGetQitem(qall, (void **)&pMsg); + if (taosGetQitem(qall, (void **)&pMsg) == 0) continue; + dTrace("vgId:%d, msg:%p get from vnode-merge queue", pVnode->vgId, pMsg); - dTrace("msg:%p, get from vnode-merge queue", pMsg); int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo); if (code != 0) { + dError("vgId:%d, msg:%p failed to merge since %s", pVnode->vgId, pMsg, terrstr()); if (terrno != 0) code = terrno; vmSendRsp(pMsg, code); } + dTrace("msg:%p, is freed, code:0x%x", pMsg, code); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } } -static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) { +static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) { SMsgHead *pHead = pMsg->pCont; int32_t code = 0; @@ -233,31 +243,35 @@ static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); if (pVnode == NULL) { - dError("vgId:%d, failed to put msg:%p into vnode-queue since %s", pHead->vgId, pMsg, terrstr()); + dError("vgId:%d, failed to put msg:%p into vnode queue since %s", pHead->vgId, pMsg, terrstr()); return terrno != 0 ? terrno : -1; } switch (qtype) { case QUERY_QUEUE: - dTrace("msg:%p, put into vnode-query worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-query queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pQueryQ, pMsg); break; case FETCH_QUEUE: - dTrace("msg:%p, put into vnode-fetch worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-fetch queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pFetchQ, pMsg); break; case WRITE_QUEUE: - dTrace("msg:%p, put into vnode-write worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-write queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pWriteQ, pMsg); break; case SYNC_QUEUE: - dTrace("msg:%p, put into vnode-sync worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-sync queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pSyncQ, pMsg); break; case MERGE_QUEUE: - dTrace("msg:%p, put into vnode-merge worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-merge queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pMergeQ, pMsg); break; + case APPLY_QUEUE: + dTrace("vgId:%d, msg:%p put into vnode-apply queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + taosWriteQitem(pVnode->pApplyQ, pMsg); + break; default: code = -1; terrno = TSDB_CODE_INVALID_PARA; @@ -268,88 +282,44 @@ static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType return code; } -int32_t vmPutNodeMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return vmPutNodeMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); -} +int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); } -int32_t vmPutNodeMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return vmPutNodeMsgToQueue(pMgmt, pMsg, WRITE_QUEUE); -} +int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, WRITE_QUEUE); } -int32_t vmPutNodeMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return vmPutNodeMsgToQueue(pMgmt, pMsg, QUERY_QUEUE); -} +int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, QUERY_QUEUE); } -int32_t vmPutNodeMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return vmPutNodeMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); -} +int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); } -int32_t vmPutNodeMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return vmPutNodeMsgToQueue(pMgmt, pMsg, MERGE_QUEUE); -} +int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, MERGE_QUEUE); } -int32_t vmPutNodeMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - SSingleWorker *pWorker = &pMgmt->mgmtWorker; - dTrace("msg:%p, put into vnode-mgmt worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); - taosWriteQitem(pWorker->queue, pMsg); +int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { + dTrace("msg:%p, put into vnode-mgmt queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + taosWriteQitem(pMgmt->mgmtWorker.queue, pMsg); return 0; } -int32_t vmPutNodeMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - SSingleWorker *pWorker = &pMgmt->monitorWorker; - dTrace("msg:%p, put into vnode-monitor worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); - taosWriteQitem(pWorker->queue, pMsg); +int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { + dTrace("msg:%p, put into vnode-monitor queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + taosWriteQitem(pMgmt->monitorWorker.queue, pMsg); return 0; } static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType qtype) { - SMsgHead *pHead = pRpc->pCont; - SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); - if (pVnode == NULL) return -1; - SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); - int32_t code = 0; - if (pMsg == NULL) { rpcFreeCont(pRpc->pCont); pRpc->pCont = NULL; - code = -1; - } else { - memcpy(pMsg, pRpc, sizeof(SRpcMsg)); - switch (qtype) { - case WRITE_QUEUE: - dTrace("msg:%p, create and put into vnode-write worker, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); - taosWriteQitem(pVnode->pWriteQ, pMsg); - break; - case QUERY_QUEUE: - dTrace("msg:%p, create and put into vnode-query queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); - taosWriteQitem(pVnode->pQueryQ, pMsg); - break; - case FETCH_QUEUE: - dTrace("msg:%p, create and put into vnode-fetch queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); - taosWriteQitem(pVnode->pFetchQ, pMsg); - break; - case APPLY_QUEUE: - dTrace("msg:%p, create and put into vnode-apply queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); - taosWriteQitem(pVnode->pApplyQ, pMsg); - break; - case MERGE_QUEUE: - dTrace("msg:%p, create and put into vnode-merge queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); - taosWriteQitem(pVnode->pMergeQ, pMsg); - break; - case SYNC_QUEUE: - dTrace("msg:%p, create and put into vnode-sync queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); - taosWriteQitem(pVnode->pSyncQ, pMsg); - break; - default: - code = -1; - terrno = TSDB_CODE_INVALID_PARA; - break; - } + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; } - vmReleaseVnode(pMgmt, pVnode); - return code; + SMsgHead *pHead = pRpc->pCont; + pHead->contLen = htonl(pHead->contLen); + pHead->vgId = htonl(pHead->vgId); + memcpy(pMsg, pRpc, sizeof(SRpcMsg)); + + dTrace("msg:%p, is created and will put into vnode queue", pMsg); + return vmPutMsgToQueue(pMgmt, pMsg, qtype); } int32_t vmPutRpcMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { @@ -467,29 +437,23 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { pMPool->max = tsNumOfVnodeMergeThreads; if (tWWorkerInit(pMPool) != 0) return -1; - SSingleWorkerCfg cfg = { + SSingleWorkerCfg mgmtCfg = { .min = 1, .max = 1, .name = "vnode-mgmt", - .fp = (FItem)vmProcessQueue, + .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt, }; - if (tSingleWorkerInit(&pMgmt->mgmtWorker, &cfg) != 0) { - dError("failed to start vnode-mgmt worker since %s", terrstr()); - return -1; - } + if (tSingleWorkerInit(&pMgmt->mgmtWorker, &mgmtCfg) != 0) return -1; - SSingleWorkerCfg mCfg = { + SSingleWorkerCfg monitorCfg = { .min = 1, .max = 1, .name = "vnode-monitor", - .fp = (FItem)vmProcessQueue, + .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt, }; - if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) { - dError("failed to start vnode-monitor worker since %s", terrstr()); - return -1; - } + if (tSingleWorkerInit(&pMgmt->monitorWorker, &monitorCfg) != 0) return -1; dDebug("vnode workers are initialized"); return 0; diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index e5893fd947..955b94dd08 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -117,6 +117,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { if (pMsg == NULL) { goto _OVER; } + dTrace("msg:%p, is created", pMsg); if (dmBuildNodeMsg(pMsg, pRpc) != 0) { goto _OVER; From de167cd46df90418b404d2bb2f48a833ee24e5ea Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 2 Jun 2022 10:17:07 +0000 Subject: [PATCH 3/6] fix(query): concat/concat_ws function output incorrect when all param are constant and include NULL value. TD-16232 TD-16233 --- include/common/ttypes.h | 2 ++ source/client/src/clientImpl.c | 2 +- source/libs/function/src/builtins.c | 8 +++++++- source/libs/scalar/src/sclfunc.c | 14 +++++++------- 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/include/common/ttypes.h b/include/common/ttypes.h index 31cdb28690..3416b2fa41 100644 --- a/include/common/ttypes.h +++ b/include/common/ttypes.h @@ -180,7 +180,9 @@ typedef struct { } while (0) //TODO: use varchar(0) to represent NULL type +#define IS_VAR_NULL_TYPE(_t, _b) ((_t) == TSDB_DATA_TYPE_VARCHAR && (_b) == 0) #define IS_NULL_TYPE(_t) ((_t) == TSDB_DATA_TYPE_NULL) + #define IS_SIGNED_NUMERIC_TYPE(_t) ((_t) >= TSDB_DATA_TYPE_TINYINT && (_t) <= TSDB_DATA_TYPE_BIGINT) #define IS_UNSIGNED_NUMERIC_TYPE(_t) ((_t) >= TSDB_DATA_TYPE_UTINYINT && (_t) <= TSDB_DATA_TYPE_UBIGINT) #define IS_FLOAT_TYPE(_t) ((_t) == TSDB_DATA_TYPE_FLOAT || (_t) == TSDB_DATA_TYPE_DOUBLE) diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 375e1c0da9..d0ff480acc 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -898,7 +898,7 @@ void doSetOneRowPtr(SReqResultInfo* pResultInfo) { int32_t bytes = pResultInfo->fields[i].bytes; if (IS_VAR_DATA_TYPE(type)) { - if (pCol->offset[pResultInfo->current] != -1) { + if (!IS_VAR_NULL_TYPE(type, bytes) && pCol->offset[pResultInfo->current] != -1) { char* pStart = pResultInfo->pCol[i].offset[pResultInfo->current] + pResultInfo->pCol[i].pData; pResultInfo->length[i] = varDataLen(pStart); diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index 5560c9c1d5..74219b3919 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -851,7 +851,7 @@ static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t for (int32_t i = 0; i < numOfParams; ++i) { SNode* pPara = nodesListGetNode(pFunc->pParameterList, i); uint8_t paraType = ((SExprNode*)pPara)->resType.type; - if (!IS_VAR_DATA_TYPE(paraType) && TSDB_DATA_TYPE_NULL != paraType) { + if (!IS_VAR_DATA_TYPE(paraType) && !IS_NULL_TYPE(paraType)) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); } if (TSDB_DATA_TYPE_NCHAR == paraType) { @@ -864,6 +864,12 @@ static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t uint8_t paraType = ((SExprNode*)pPara)->resType.type; int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes; int32_t factor = 1; + if (IS_NULL_TYPE(paraType)) { + resultType = TSDB_DATA_TYPE_VARCHAR; + resultBytes = 0; + sepBytes = 0; + break; + } if (TSDB_DATA_TYPE_NCHAR == resultType && TSDB_DATA_TYPE_VARCHAR == paraType) { factor *= TSDB_NCHAR_SIZE; } diff --git a/source/libs/scalar/src/sclfunc.c b/source/libs/scalar/src/sclfunc.c index 370ea8fa2e..587f9b5515 100644 --- a/source/libs/scalar/src/sclfunc.c +++ b/source/libs/scalar/src/sclfunc.c @@ -449,7 +449,7 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu bool hasNull = false; for (int32_t i = 0; i < inputNum; ++i) { if (colDataIsNull_s(pInputData[i], k) || - GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_NULL) { + IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[i]))) { colDataAppendNULL(pOutputData, k); hasNull = true; break; @@ -526,7 +526,7 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p for (int32_t k = 0; k < numOfRows; ++k) { if (colDataIsNull_s(pInputData[0], k) || - GET_PARAM_TYPE(&pInput[0]) == TSDB_DATA_TYPE_NULL) { + IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[0]))) { colDataAppendNULL(pOutputData, k); continue; } @@ -535,7 +535,7 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p bool hasNull = false; for (int32_t i = 1; i < inputNum; ++i) { if (colDataIsNull_s(pInputData[i], k) || - GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_NULL) { + IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[i]))) { hasNull = true; break; } @@ -1221,8 +1221,8 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p if (IS_VAR_DATA_TYPE(type)) { /* datetime format strings */ int32_t ret = convertStringToTimestamp(type, input[k], TSDB_TIME_PRECISION_NANO, &timeVal[k]); if (ret != TSDB_CODE_SUCCESS) { - colDataAppendNULL(pOutput->columnData, i); - continue; + hasNull = true; + break; } } else if (type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_TIMESTAMP) { /* unix timestamp or ts column*/ GET_TYPED_DATA(timeVal[k], int64_t, type, input[k]); @@ -1247,8 +1247,8 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) { timeVal[k] = timeVal[k]; } else { - colDataAppendNULL(pOutput->columnData, i); - continue; + hasNull = true; + break; } } } From a9b32dc3276755c266932ae8bd4d2d5d32fb7bd5 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 2 Jun 2022 18:26:29 +0800 Subject: [PATCH 4/6] refactor: adjust vnode queue --- include/common/tmsgcb.h | 4 +- source/common/src/tmsgcb.c | 36 +++----------- source/dnode/mgmt/mgmt_mnode/inc/mmInt.h | 5 +- source/dnode/mgmt/mgmt_mnode/src/mmInt.c | 5 +- source/dnode/mgmt/mgmt_mnode/src/mmWorker.c | 54 ++++++++++----------- source/dnode/mgmt/mgmt_qnode/inc/qmInt.h | 3 +- source/dnode/mgmt/mgmt_qnode/src/qmInt.c | 3 +- source/dnode/mgmt/mgmt_qnode/src/qmWorker.c | 26 +++++----- source/dnode/mgmt/mgmt_vnode/inc/vmInt.h | 41 +++++++--------- source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 9 +--- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 31 +----------- source/libs/qworker/test/qworkerTests.cpp | 33 ++++++------- 12 files changed, 92 insertions(+), 158 deletions(-) diff --git a/include/common/tmsgcb.h b/include/common/tmsgcb.h index 9fa657a2a6..e99377f9b4 100644 --- a/include/common/tmsgcb.h +++ b/include/common/tmsgcb.h @@ -38,7 +38,7 @@ typedef enum { QUEUE_MAX, } EQueueType; -typedef int32_t (*PutToQueueFp)(void* pMgmt, SRpcMsg* pMsg); +typedef int32_t (*PutToQueueFp)(void* pMgmt, EQueueType qtype, SRpcMsg* pMsg); typedef int32_t (*GetQueueSizeFp)(void* pMgmt, int32_t vgId, EQueueType qtype); typedef int32_t (*SendReqFp)(const SEpSet* pEpSet, SRpcMsg* pMsg); typedef void (*SendRspFp)(SRpcMsg* pMsg); @@ -50,7 +50,7 @@ typedef void (*ReportStartup)(const char* name, const char* desc); typedef struct { void* mgmt; void* clientRpc; - PutToQueueFp queueFps[QUEUE_MAX]; + PutToQueueFp putToQueueFp; GetQueueSizeFp qsizeFp; SendReqFp sendReqFp; SendRspFp sendRspFp; diff --git a/source/common/src/tmsgcb.c b/source/common/src/tmsgcb.c index 126a4c023a..b8eec655b1 100644 --- a/source/common/src/tmsgcb.c +++ b/source/common/src/tmsgcb.c @@ -22,41 +22,21 @@ static SMsgCb defaultMsgCb; void tmsgSetDefault(const SMsgCb* msgcb) { defaultMsgCb = *msgcb; } int32_t tmsgPutToQueue(const SMsgCb* msgcb, EQueueType qtype, SRpcMsg* pMsg) { - PutToQueueFp fp = msgcb->queueFps[qtype]; - return (*fp)(msgcb->mgmt, pMsg); + return (*msgcb->putToQueueFp)(msgcb->mgmt, qtype, pMsg); } int32_t tmsgGetQueueSize(const SMsgCb* msgcb, int32_t vgId, EQueueType qtype) { - GetQueueSizeFp fp = msgcb->qsizeFp; - return (*fp)(msgcb->mgmt, vgId, qtype); + return (*msgcb->qsizeFp)(msgcb->mgmt, vgId, qtype); } -int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) { - SendReqFp fp = defaultMsgCb.sendReqFp; - return (*fp)(epSet, pMsg); -} +int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) { return (*defaultMsgCb.sendReqFp)(epSet, pMsg); } -void tmsgSendRsp(SRpcMsg* pMsg) { - SendRspFp fp = defaultMsgCb.sendRspFp; - return (*fp)(pMsg); -} +void tmsgSendRsp(SRpcMsg* pMsg) { return (*defaultMsgCb.sendRspFp)(pMsg); } -void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) { - SendRedirectRspFp fp = defaultMsgCb.sendRedirectRspFp; - (*fp)(pMsg, pNewEpSet); -} +void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) { (*defaultMsgCb.sendRedirectRspFp)(pMsg, pNewEpSet); } -void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) { - RegisterBrokenLinkArgFp fp = defaultMsgCb.registerBrokenLinkArgFp; - (*fp)(pMsg); -} +void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) { (*defaultMsgCb.registerBrokenLinkArgFp)(pMsg); } -void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) { - ReleaseHandleFp fp = defaultMsgCb.releaseHandleFp; - (*fp)(pHandle, type); -} +void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) { (*defaultMsgCb.releaseHandleFp)(pHandle, type); } -void tmsgReportStartup(const char* name, const char* desc) { - ReportStartup fp = defaultMsgCb.reportStartupFp; - (*fp)(name, desc); -} \ No newline at end of file +void tmsgReportStartup(const char* name, const char* desc) { (*defaultMsgCb.reportStartupFp)(name, desc); } \ No newline at end of file diff --git a/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h b/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h index bd034fe7d6..9a0cfdfc93 100644 --- a/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h +++ b/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h @@ -65,10 +65,7 @@ int32_t mmPutNodeMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutNodeMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutNodeMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc); #ifdef __cplusplus } diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmInt.c b/source/dnode/mgmt/mgmt_mnode/src/mmInt.c index 1b973f3045..0f3c06cb3a 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmInt.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmInt.c @@ -105,10 +105,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->path = pInput->path; pMgmt->name = pInput->name; pMgmt->msgCb = pInput->msgCb; - pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)mmPutRpcMsgToQueryQueue; - pMgmt->msgCb.queueFps[READ_QUEUE] = (PutToQueueFp)mmPutRpcMsgToReadQueue; - pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)mmPutRpcMsgToWriteQueue; - pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)mmPutRpcMsgToSyncQueue; + pMgmt->msgCb.putToQueueFp = (PutToQueueFp)mmPutRpcMsgToQueue; pMgmt->msgCb.mgmt = pMgmt; taosThreadRwlockInit(&pMgmt->lock, NULL); diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c index 1de9875d06..460f2242f2 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c @@ -96,40 +96,38 @@ int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { return mmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg); } -static inline int32_t mmPutRpcMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pRpc) { +int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); if (pMsg == NULL) return -1; - - dTrace("msg:%p, create and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType)); memcpy(pMsg, pRpc, sizeof(SRpcMsg)); - taosWriteQitem(pWorker->queue, pMsg); - return 0; -} -int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return mmPutRpcMsgToWorker(&pMgmt->queryWorker, pMsg); -} + switch (qtype) { + case WRITE_QUEUE: + dTrace("msg:%p, is created and will put into vnode-write queue", pMsg); + taosWriteQitem(pMgmt->writeWorker.queue, pMsg); + return 0; + case QUERY_QUEUE: + dTrace("msg:%p, is created and will put into vnode-query queue", pMsg); + taosWriteQitem(pMgmt->queryWorker.queue, pMsg); + return 0; -int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return mmPutRpcMsgToWorker(&pMgmt->writeWorker, pMsg); -} - -int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - return mmPutRpcMsgToWorker(&pMgmt->readWorker, pMsg); -} - -int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { - int32_t code = -1; - if (mmAcquire(pMgmt) == 0) { - code = mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg); - mmRelease(pMgmt); + case READ_QUEUE: + dTrace("msg:%p, is created and will put into vnode-read queue", pMsg); + taosWriteQitem(pMgmt->readWorker.queue, pMsg); + return 0; + case SYNC_QUEUE: + if (mmAcquire(pMgmt) == 0) { + dTrace("msg:%p, is created and will put into vnode-sync queue", pMsg); + taosWriteQitem(pMgmt->syncWorker.queue, pMsg); + mmRelease(pMgmt); + return 0; + } else { + return -1; + } + default: + terrno = TSDB_CODE_INVALID_PARA; + return -1; } - - if (code != 0) { - rpcFreeCont(pMsg->pCont); - pMsg->pCont = NULL; - } - return code; } int32_t mmStartWorker(SMnodeMgmt *pMgmt) { diff --git a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h index 9738fb0c45..54e9da24a4 100644 --- a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h +++ b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h @@ -42,8 +42,7 @@ int32_t qmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg); int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); // qmWorker.c -int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pMsg); int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype); int32_t qmStartWorker(SQnodeMgmt *pMgmt); diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmInt.c b/source/dnode/mgmt/mgmt_qnode/src/qmInt.c index 06c18ab288..3b425a0b49 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmInt.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmInt.c @@ -43,8 +43,7 @@ static int32_t qmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->path = pInput->path; pMgmt->name = pInput->name; pMgmt->msgCb = pInput->msgCb; - pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qmPutRpcMsgToQueryQueue; - pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)qmPutRpcMsgToFetchQueue; + pMgmt->msgCb.putToQueueFp = (PutToQueueFp)qmPutRpcMsgToQueue; pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)qmGetQueueSize; pMgmt->msgCb.mgmt = pMgmt; diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c index e7fc261b67..e36efa83db 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c @@ -68,22 +68,24 @@ int32_t qmPutNodeMsgToMonitorQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) { return qmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg); } -static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) { +int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); if (pMsg == NULL) return -1; - - dTrace("msg:%p, create and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType)); memcpy(pMsg, pRpc, sizeof(SRpcMsg)); - taosWriteQitem(pWorker->queue, pMsg); - return 0; -} -int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return qmPutRpcMsgToWorker(pMgmt, &pMgmt->queryWorker, pRpc); -} - -int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return qmPutRpcMsgToWorker(pMgmt, &pMgmt->fetchWorker, pRpc); + switch (qtype) { + case QUERY_QUEUE: + dTrace("msg:%p, is created and will put into qnode-query queue", pMsg); + taosWriteQitem(pMgmt->queryWorker.queue, pMsg); + return 0; + case READ_QUEUE: + dTrace("msg:%p, is created and will put into qnode-fetch queue", pMsg); + taosWriteQitem(pMgmt->fetchWorker.queue, pMsg); + return 0; + default: + terrno = TSDB_CODE_INVALID_PARA; + return -1; + } } int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { diff --git a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h index f74fd119c5..adc18fdced 100644 --- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h +++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h @@ -44,26 +44,26 @@ typedef struct SVnodeMgmt { } SVnodeMgmt; typedef struct { - int32_t vgId; - int32_t vgVersion; - int8_t dropped; - char path[PATH_MAX + 20]; + int32_t vgId; + int32_t vgVersion; + int8_t dropped; + char path[PATH_MAX + 20]; } SWrapperCfg; typedef struct { - int32_t vgId; - int32_t refCount; - int32_t vgVersion; - int8_t dropped; - int8_t accessState; - char *path; - SVnode *pImpl; - STaosQueue *pWriteQ; - STaosQueue *pSyncQ; - STaosQueue *pApplyQ; - STaosQueue *pQueryQ; - STaosQueue *pFetchQ; - STaosQueue *pMergeQ; + int32_t vgId; + int32_t refCount; + int32_t vgVersion; + int8_t dropped; + int8_t accessState; + char *path; + SVnode *pImpl; + STaosQueue *pWriteQ; + STaosQueue *pSyncQ; + STaosQueue *pApplyQ; + STaosQueue *pQueryQ; + STaosQueue *pFetchQ; + STaosQueue *pMergeQ; } SVnodeObj; typedef struct { @@ -100,13 +100,8 @@ void vmStopWorker(SVnodeMgmt *pMgmt); int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode); void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode); -int32_t vmPutRpcMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToApplyQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); -int32_t vmPutRpcMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype); +int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc); int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c index 5d3479f74b..7e76ee3af2 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c @@ -156,7 +156,7 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) { pMgmt->state.totalVnodes = numOfVnodes; - int32_t threadNum = tsNumOfCores; + int32_t threadNum = 1; int32_t vnodesPerThread = numOfVnodes / threadNum + 1; SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread)); @@ -249,12 +249,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->path = pInput->path; pMgmt->name = pInput->name; pMgmt->msgCb = pInput->msgCb; - pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToWriteQueue; - pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)vmPutRpcMsgToSyncQueue; - pMgmt->msgCb.queueFps[APPLY_QUEUE] = (PutToQueueFp)vmPutRpcMsgToApplyQueue; - pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)vmPutRpcMsgToQueryQueue; - pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)vmPutRpcMsgToFetchQueue; - pMgmt->msgCb.queueFps[MERGE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToMergeQueue; + pMgmt->msgCb.putToQueueFp = (PutToQueueFp)vmPutRpcMsgToQueue; pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)vmGetQueueSize; pMgmt->msgCb.mgmt = pMgmt; taosThreadRwlockInit(&pMgmt->lock, NULL); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index e5953b8383..dbd653a44e 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -304,14 +304,9 @@ int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return 0; } -static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType qtype) { +int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); - if (pMsg == NULL) { - rpcFreeCont(pRpc->pCont); - pRpc->pCont = NULL; - terrno = TSDB_CODE_OUT_OF_MEMORY; - return -1; - } + if (pMsg == NULL) return -1; SMsgHead *pHead = pRpc->pCont; pHead->contLen = htonl(pHead->contLen); @@ -322,28 +317,6 @@ static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType q return vmPutMsgToQueue(pMgmt, pMsg, qtype); } -int32_t vmPutRpcMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, WRITE_QUEUE); -} - -int32_t vmPutRpcMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { return vmPutRpcMsgToQueue(pMgmt, pRpc, SYNC_QUEUE); } - -int32_t vmPutRpcMsgToApplyQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, APPLY_QUEUE); -} - -int32_t vmPutRpcMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, QUERY_QUEUE); -} - -int32_t vmPutRpcMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, FETCH_QUEUE); -} - -int32_t vmPutRpcMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { - return vmPutRpcMsgToQueue(pMgmt, pRpc, MERGE_QUEUE); -} - int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { int32_t size = -1; SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId); diff --git a/source/libs/qworker/test/qworkerTests.cpp b/source/libs/qworker/test/qworkerTests.cpp index 16dcd7b6e0..5bb6acee60 100644 --- a/source/libs/qworker/test/qworkerTests.cpp +++ b/source/libs/qworker/test/qworkerTests.cpp @@ -172,8 +172,7 @@ int32_t qwtPutReqToFetchQueue(void *node, struct SRpcMsg *pMsg) { return 0; } - -int32_t qwtPutReqToQueue(void *node, struct SRpcMsg *pMsg) { +int32_t qwtPutReqToQueue(void *node, EQueueType qtype, struct SRpcMsg *pMsg) { taosWLockLatch(&qwtTestQueryQueueLock); struct SRpcMsg *newMsg = (struct SRpcMsg *)taosMemoryCalloc(1, sizeof(struct SRpcMsg)); memcpy(newMsg, pMsg, sizeof(struct SRpcMsg)); @@ -181,7 +180,7 @@ int32_t qwtPutReqToQueue(void *node, struct SRpcMsg *pMsg) { if (qwtTestQueryQueueWIdx >= qwtTestQueryQueueSize) { qwtTestQueryQueueWIdx = 0; } - + qwtTestQueryQueueNum++; if (qwtTestQueryQueueWIdx == qwtTestQueryQueueRIdx) { @@ -702,9 +701,9 @@ void *qwtclientThread(void *param) { while (!qwtTestStop) { qwtTestCaseFinished = false; - + qwtBuildQueryReqMsg(&queryRpc); - qwtPutReqToQueue((void *)0x1, &queryRpc); + qwtPutReqToQueue((void *)0x1, QUERY_QUEUE, &queryRpc); while (!qwtTestCaseFinished) { taosUsleep(1); @@ -874,7 +873,7 @@ TEST(seqTest, normalCase) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -907,10 +906,10 @@ TEST(seqTest, cancelFirst) { stubSetStringToPlan(); stubSetRpcSendResponse(); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -944,10 +943,10 @@ TEST(seqTest, randCase) { stubSetCreateExecTask(); taosSeedRand(taosGetTimestampSec()); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1015,10 +1014,10 @@ TEST(seqTest, multithreadRand) { stubSetGetDataBlock(); taosSeedRand(taosGetTimestampSec()); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1081,7 +1080,7 @@ TEST(rcTest, shortExecshortDelay) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1165,7 +1164,7 @@ TEST(rcTest, longExecshortDelay) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1251,7 +1250,7 @@ TEST(rcTest, shortExeclongDelay) { SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); @@ -1332,10 +1331,10 @@ TEST(rcTest, dropTest) { stubSetGetDataBlock(); taosSeedRand(taosGetTimestampSec()); - + SMsgCb msgCb = {0}; msgCb.mgmt = (void *)mockPointer; - msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue; + msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue; code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb); ASSERT_EQ(code, 0); From 465ec5003bd455c5f485c744979dfa2375079460 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 2 Jun 2022 19:00:14 +0800 Subject: [PATCH 5/6] refactor: adjust msg type --- include/common/tmsgdef.h | 172 ++++++++++---------- source/dnode/mgmt/mgmt_mnode/src/mmHandle.c | 4 +- source/dnode/mgmt/test/mnode/dmnode.cpp | 8 +- source/dnode/mnode/impl/src/mndMnode.c | 8 +- tests/script/tsim/db/alter_replica_13.sim | 1 - 5 files changed, 94 insertions(+), 99 deletions(-) diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index ba4a221a9f..94cd1518ed 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -70,106 +70,102 @@ enum { // Requests handled by DNODE TD_NEW_MSG_SEG(TDMT_DND_MSG) TD_DEF_MSG_TYPE(TDMT_DND_CREATE_MNODE, "dnode-create-mnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_ALTER_MNODE, "dnode-alter-mnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_DROP_MNODE, "dnode-drop-mnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_CREATE_QNODE, "dnode-create-qnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_ALTER_QNODE, "dnode-alter-qnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_DROP_QNODE, "dnode-drop-qnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_CREATE_SNODE, "dnode-create-snode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_ALTER_SNODE, "dnode-alter-snode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_DROP_SNODE, "dnode-drop-snode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_CREATE_BNODE, "dnode-create-bnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_ALTER_BNODE, "dnode-alter-bnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_DROP_BNODE, "dnode-drop-bnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_CREATE_VNODE, "dnode-create-vnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "dnode-drop-vnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "dnode-config-dnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_SERVER_STATUS, "dnode-server-status", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_DND_NET_TEST, "dnode-net-test", NULL, NULL) - + TD_DEF_MSG_TYPE(TDMT_DND_CREATE_VNODE, "create-vnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "drop-vnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_DND_SERVER_STATUS, "server-status", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_DND_NET_TEST, "net-test", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "config-dnode", NULL, NULL) + // Requests handled by MNODE TD_NEW_MSG_SEG(TDMT_MND_MSG) - TD_DEF_MSG_TYPE(TDMT_MND_CONNECT, "mnode-connect", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_ACCT, "mnode-create-acct", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_ACCT, "mnode-alter-acct", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_ACCT, "mnode-drop-acct", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_USER, "mnode-create-user", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_USER, "mnode-alter-user", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_USER, "mnode-drop-user", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_GET_USER_AUTH, "mnode-get-user-auth", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DNODE, "mnode-create-dnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CONFIG_DNODE, "mnode-config-dnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DNODE, "mnode-alter-dnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_DNODE, "mnode-drop-dnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_MNODE, "mnode-create-mnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_MNODE, "mnode-alter-mnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_MNODE, "mnode-drop-mnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_QNODE, "mnode-create-qnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_QNODE, "mnode-alter-qnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_QNODE, "mnode-drop-qnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SNODE, "mnode-create-snode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_SNODE, "mnode-alter-snode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_SNODE, "mnode-drop-snode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_BNODE, "mnode-create-bnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_BNODE, "mnode-alter-bnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_BNODE, "mnode-drop-bnode", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DB, "mnode-create-db", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_DB, "mnode-drop-db", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_USE_DB, "mnode-use-db", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DB, "mnode-alter-db", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_SYNC_DB, "mnode-sync-db", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_COMPACT_DB, "mnode-compact-db", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_FUNC, "mnode-create-func", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_RETRIEVE_FUNC, "mnode-retrieve-func", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_FUNC, "mnode-drop-func", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STB, "mnode-create-stb", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STB, "mnode-alter-stb", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_STB, "mnode-drop-stb", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SMA, "mnode-create-sma", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_SMA, "mnode-drop-sma", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_TABLE_META, "mnode-table-meta", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_VGROUP_LIST, "mnode-vgroup-list", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_QNODE_LIST, "mnode-qnode-list", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "mnode-kill-query", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "mnode-kill-conn", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "mnode-heartbeat", SClientHbBatchReq, SClientHbBatchRsp) - TD_DEF_MSG_TYPE(TDMT_MND_SHOW, "mnode-show", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_SYSTABLE_RETRIEVE, "mnode-systable-retrieve", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_STATUS, "mnode-status", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "mnode-trans-tmr", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "mnode-kill-trans", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "mnode-telem-tmr", SMTimerReq, SMTimerReq) - TD_DEF_MSG_TYPE(TDMT_MND_GRANT, "mnode-grant", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "mnode-auth", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "mnode-create-topic", SMCreateTopicReq, SMCreateTopicRsp) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp) - TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mnode-mq-ask-ep", SMqAskEpReq, SMqAskEpRsp) - TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMqConsumerLostMsg, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_RECOVER, "mnode-mq-consumer-recover", SMqConsumerRecoverMsg, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_MQ_DROP_CGROUP, "mnode-mq-drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp) - TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mnode-mq-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STREAM, "mnode-create-stream", SCMCreateStreamReq, SCMCreateStreamRsp) - TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STREAM, "mnode-alter-stream", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_STREAM, "mnode-drop-stream", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_CREATE_INDEX, "mnode-create-index", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_DROP_INDEX, "mnode-drop-index", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_GET_DB_CFG, "mnode-get-db-cfg", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_GET_INDEX, "mnode-get-index", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_APPLY_MSG, "mnode-apply-msg", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CONNECT, "connect", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_ACCT, "create-acct", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_ACCT, "alter-acct", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_ACCT, "drop-acct", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_USER, "create-user", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_USER, "alter-user", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_USER, "drop-user", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_GET_USER_AUTH, "get-user-auth", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DNODE, "create-dnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CONFIG_DNODE, "config-dnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DNODE, "alter-dnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_DNODE, "drop-dnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_MNODE, "create-mnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_MNODE, "alter-mnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_MNODE, "drop-mnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_QNODE, "create-qnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_QNODE, "alter-qnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_QNODE, "drop-qnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SNODE, "create-snode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_SNODE, "alter-snode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_SNODE, "drop-snode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_BNODE, "create-bnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_BNODE, "alter-bnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_BNODE, "drop-bnode", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DB, "create-db", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_DB, "drop-db", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_USE_DB, "use-db", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DB, "alter-db", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_SYNC_DB, "sync-db", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_COMPACT_DB, "compact-db", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_FUNC, "create-func", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_RETRIEVE_FUNC, "retrieve-func", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_FUNC, "drop-func", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STB, "create-stb", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STB, "alter-stb", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_STB, "drop-stb", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SMA, "create-sma", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_SMA, "drop-sma", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_TABLE_META, "table-meta", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_VGROUP_LIST, "vgroup-list", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_QNODE_LIST, "qnode-list", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "kill-query", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "kill-conn", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "heartbeat", SClientHbBatchReq, SClientHbBatchRsp) + TD_DEF_MSG_TYPE(TDMT_MND_SHOW, "show", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_SYSTABLE_RETRIEVE, "systable-retrieve", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_STATUS, "status", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "trans-tmr", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "kill-trans", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "telem-tmr", SMTimerReq, SMTimerReq) + TD_DEF_MSG_TYPE(TDMT_MND_GRANT, "grant", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "auth", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "create-topic", SMCreateTopicReq, SMCreateTopicRsp) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "alter-topic", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "drop-topic", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "subscribe", SCMSubscribeReq, SCMSubscribeRsp) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mq-ask-ep", SMqAskEpReq, SMqAskEpRsp) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mq-tmr", SMTimerReq, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mq-consumer-lost", SMqConsumerLostMsg, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_RECOVER, "mq-consumer-recover", SMqConsumerRecoverMsg, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mq-do-rebalance", SMqDoRebalanceMsg, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_DROP_CGROUP, "mq-drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mq-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STREAM, "create-stream", SCMCreateStreamReq, SCMCreateStreamRsp) + TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STREAM, "alter-stream", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_STREAM, "drop-stream", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_CREATE_INDEX, "create-index", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_DROP_INDEX, "drop-index", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_GET_DB_CFG, "get-db-cfg", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_GET_INDEX, "get-index", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_APPLY_MSG, "apply-msg", NULL, NULL) // Requests handled by VNODE TD_NEW_MSG_SEG(TDMT_VND_MSG) - TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT, "vnode-submit", SSubmitReq, SSubmitRsp) - TD_DEF_MSG_TYPE(TDMT_VND_QUERY, "vnode-query", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_VND_FETCH, "vnode-fetch", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_VND_CREATE_TABLE, "vnode-create-table", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_VND_ALTER_TABLE, "vnode-alter-table", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_VND_DROP_TABLE, "vnode-drop-table", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_VND_UPDATE_TAG_VAL, "vnode-update-tag-val", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT, "submit", SSubmitReq, SSubmitRsp) + TD_DEF_MSG_TYPE(TDMT_VND_QUERY, "query", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_VND_FETCH, "fetch", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_VND_CREATE_TABLE, "create-table", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_VND_ALTER_TABLE, "alter-table", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_VND_DROP_TABLE, "drop-table", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_VND_UPDATE_TAG_VAL, "update-tag-val", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_TABLE_META, "vnode-table-meta", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_TABLES_META, "vnode-tables-meta", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_CREATE_STB, "vnode-create-stb", SVCreateStbReq, NULL) diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c index 5c5316e3a3..8d0d503d8f 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c @@ -134,7 +134,7 @@ SArray *mmGetMsgHandles() { // Requests handled by DNODE if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_DND_ALTER_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_QNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_QNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; @@ -190,7 +190,7 @@ SArray *mmGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_MND_KILL_TRANS, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_AUTH, mmPutNodeMsgToReadQueue, 0) == NULL) goto _OVER; - if (dmSetMgmtHandle(pArray, TDMT_DND_ALTER_MNODE, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_MNODE, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_TOPIC, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_TOPIC, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_TOPIC, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; diff --git a/source/dnode/mgmt/test/mnode/dmnode.cpp b/source/dnode/mgmt/test/mnode/dmnode.cpp index 8c945b50ac..857f58befc 100644 --- a/source/dnode/mgmt/test/mnode/dmnode.cpp +++ b/source/dnode/mgmt/test/mnode/dmnode.cpp @@ -94,7 +94,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) { void* pReq = rpcMallocCont(contLen); tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq); - SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen); + SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen); ASSERT_NE(pRsp, nullptr); ASSERT_EQ(pRsp->code, TSDB_CODE_INVALID_OPTION); } @@ -111,7 +111,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) { void* pReq = rpcMallocCont(contLen); tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq); - SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen); + SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen); ASSERT_NE(pRsp, nullptr); ASSERT_EQ(pRsp->code, TSDB_CODE_INVALID_OPTION); } @@ -128,7 +128,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) { void* pReq = rpcMallocCont(contLen); tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq); - SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen); + SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen); ASSERT_NE(pRsp, nullptr); ASSERT_EQ(pRsp->code, 0); } @@ -186,7 +186,7 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) { void* pReq = rpcMallocCont(contLen); tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq); - SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen); + SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen); ASSERT_NE(pRsp, nullptr); ASSERT_EQ(pRsp->code, TSDB_CODE_NODE_NOT_DEPLOYED); } diff --git a/source/dnode/mnode/impl/src/mndMnode.c b/source/dnode/mnode/impl/src/mndMnode.c index 160fc8d428..013a991e98 100644 --- a/source/dnode/mnode/impl/src/mndMnode.c +++ b/source/dnode/mnode/impl/src/mndMnode.c @@ -53,10 +53,10 @@ int32_t mndInitMnode(SMnode *pMnode) { }; mndSetMsgHandle(pMnode, TDMT_MND_CREATE_MNODE, mndProcessCreateMnodeReq); - mndSetMsgHandle(pMnode, TDMT_DND_ALTER_MNODE, mndProcessAlterMnodeReq); + mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE, mndProcessAlterMnodeReq); mndSetMsgHandle(pMnode, TDMT_MND_DROP_MNODE, mndProcessDropMnodeReq); mndSetMsgHandle(pMnode, TDMT_DND_CREATE_MNODE_RSP, mndProcessCreateMnodeRsp); - mndSetMsgHandle(pMnode, TDMT_DND_ALTER_MNODE_RSP, mndProcessAlterMnodeRsp); + mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE_RSP, mndProcessAlterMnodeRsp); mndSetMsgHandle(pMnode, TDMT_DND_DROP_MNODE_RSP, mndProcessDropMnodeRsp); mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndRetrieveMnodes); @@ -338,7 +338,7 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno .epSet = alterEpset, .pCont = pReq, .contLen = contLen, - .msgType = TDMT_DND_ALTER_MNODE, + .msgType = TDMT_MND_ALTER_MNODE, .acceptableCode = 0, }; @@ -506,7 +506,7 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode .epSet = alterEpset, .pCont = pReq, .contLen = contLen, - .msgType = TDMT_DND_ALTER_MNODE, + .msgType = TDMT_MND_ALTER_MNODE, .acceptableCode = 0, }; diff --git a/tests/script/tsim/db/alter_replica_13.sim b/tests/script/tsim/db/alter_replica_13.sim index e20bf6937d..4d45b92967 100644 --- a/tests/script/tsim/db/alter_replica_13.sim +++ b/tests/script/tsim/db/alter_replica_13.sim @@ -128,7 +128,6 @@ if $data[0][8] != follower then endi print ============= step5: stop dnode 2 -return sql select * from db.stb if $rows != 1 then From ecaea66954b0bf5a3d9b2dd8289418ed97c5d015 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 2 Jun 2022 19:12:52 +0800 Subject: [PATCH 6/6] refactor: adjust some logs --- source/dnode/mgmt/mgmt_vnode/src/vmFile.c | 2 +- source/dnode/mgmt/mgmt_vnode/src/vmHandle.c | 4 ++-- source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 4 ++-- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 23 ++++++++++--------- source/dnode/mgmt/node_mgmt/src/dmMgmt.c | 6 ++--- source/dnode/mgmt/node_mgmt/src/dmTransport.c | 2 +- 6 files changed, 21 insertions(+), 20 deletions(-) diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c index 71f11e0d97..613f3fb994 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c @@ -31,7 +31,7 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) { SVnodeObj *pVnode = *ppVnode; if (pVnode && num < size) { int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1); - dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount); + // dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount); pVnodes[num] = (*ppVnode); num++; pIter = taosHashIterate(pMgmt->hash, pIter); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index f43f00e02a..f1668304b7 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -192,7 +192,7 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return -1; } - dDebug("vgId:%d, create vnode req is received, tsma:%d standby:%d", createReq.vgId, createReq.isTsma, + dDebug("vgId:%d, start to create vnode, tsma:%d standby:%d", createReq.vgId, createReq.isTsma, createReq.standby); vmGenerateVnodeCfg(&createReq, &vnodeCfg); vmGenerateWrapperCfg(pMgmt, &createReq, &wrapperCfg); @@ -270,7 +270,7 @@ int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { } int32_t vgId = dropReq.vgId; - dDebug("vgId:%d, drop vnode req is received", vgId); + dDebug("vgId:%d, start to drop vnode", vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId); if (pVnode == NULL) { diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c index 7e76ee3af2..23927255bb 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c @@ -25,7 +25,7 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) { terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; } else { int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1); - dTrace("vgId:%d, acquire vnode, ref:%d", pVnode->vgId, refCount); + // dTrace("vgId:%d, acquire vnode, ref:%d", pVnode->vgId, refCount); } taosThreadRwlockUnlock(&pMgmt->lock); @@ -37,7 +37,7 @@ void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { taosThreadRwlockRdlock(&pMgmt->lock); int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1); - dTrace("vgId:%d, release vnode, ref:%d", pVnode->vgId, refCount); + // dTrace("vgId:%d, release vnode, ref:%d", pVnode->vgId, refCount); taosThreadRwlockUnlock(&pMgmt->lock); } diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index dbd653a44e..bedf473379 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -243,33 +243,34 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); if (pVnode == NULL) { - dError("vgId:%d, failed to put msg:%p into vnode queue since %s", pHead->vgId, pMsg, terrstr()); + dError("vgId:%d, failed to put msg:%p into vnode queue since %s, type:%s", pHead->vgId, pMsg, terrstr(), + TMSG_INFO(pMsg->msgType)); return terrno != 0 ? terrno : -1; } switch (qtype) { case QUERY_QUEUE: - dTrace("vgId:%d, msg:%p put into vnode-query queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pQueryQ, pMsg); break; case FETCH_QUEUE: - dTrace("vgId:%d, msg:%p put into vnode-fetch queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-fetch queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pFetchQ, pMsg); break; case WRITE_QUEUE: - dTrace("vgId:%d, msg:%p put into vnode-write queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pWriteQ, pMsg); break; case SYNC_QUEUE: - dTrace("vgId:%d, msg:%p put into vnode-sync queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pSyncQ, pMsg); break; case MERGE_QUEUE: - dTrace("vgId:%d, msg:%p put into vnode-merge queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-merge queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pMergeQ, pMsg); break; case APPLY_QUEUE: - dTrace("vgId:%d, msg:%p put into vnode-apply queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pApplyQ, pMsg); break; default: @@ -293,13 +294,13 @@ int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsg int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, MERGE_QUEUE); } int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - dTrace("msg:%p, put into vnode-mgmt queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("msg:%p, put into vnode-mgmt queue", pMsg); taosWriteQitem(pMgmt->mgmtWorker.queue, pMsg); return 0; } int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { - dTrace("msg:%p, put into vnode-monitor queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); + dTrace("msg:%p, put into vnode-monitor queue", pMsg); taosWriteQitem(pMgmt->monitorWorker.queue, pMsg); return 0; } @@ -309,11 +310,11 @@ int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { if (pMsg == NULL) return -1; SMsgHead *pHead = pRpc->pCont; + dTrace("vgId:%d, msg:%p is created, type:%s", pHead->vgId, pMsg, TMSG_INFO(pMsg->msgType)); + pHead->contLen = htonl(pHead->contLen); pHead->vgId = htonl(pHead->vgId); memcpy(pMsg, pRpc, sizeof(SRpcMsg)); - - dTrace("msg:%p, is created and will put into vnode queue", pMsg); return vmPutMsgToQueue(pMgmt, pMsg, qtype); } diff --git a/source/dnode/mgmt/node_mgmt/src/dmMgmt.c b/source/dnode/mgmt/node_mgmt/src/dmMgmt.c index 787f5e5019..ee27f27f06 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmMgmt.c +++ b/source/dnode/mgmt/node_mgmt/src/dmMgmt.c @@ -229,7 +229,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) { taosThreadRwlockRdlock(&pWrapper->lock); if (pWrapper->deployed) { int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1); - dTrace("node:%s, is acquired, ref:%d", pWrapper->name, refCount); + // dTrace("node:%s, is acquired, ref:%d", pWrapper->name, refCount); } else { terrno = TSDB_CODE_NODE_NOT_DEPLOYED; pRetWrapper = NULL; @@ -245,7 +245,7 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) { taosThreadRwlockRdlock(&pWrapper->lock); if (pWrapper->deployed || (InParentProc(pWrapper) && pWrapper->required)) { int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1); - dTrace("node:%s, is marked, ref:%d", pWrapper->name, refCount); + // dTrace("node:%s, is marked, ref:%d", pWrapper->name, refCount); } else { terrno = TSDB_CODE_NODE_NOT_DEPLOYED; code = -1; @@ -261,7 +261,7 @@ void dmReleaseWrapper(SMgmtWrapper *pWrapper) { taosThreadRwlockRdlock(&pWrapper->lock); int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1); taosThreadRwlockUnlock(&pWrapper->lock); - dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount); + // dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount); } static void dmGetServerStartupStatus(SDnode *pDnode, SServerStatusRsp *pStatus) { diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 955b94dd08..8b939d15ce 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -117,7 +117,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { if (pMsg == NULL) { goto _OVER; } - dTrace("msg:%p, is created", pMsg); + dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); if (dmBuildNodeMsg(pMsg, pRpc) != 0) { goto _OVER;