diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 1c5510ba0d..bef581a31b 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -145,6 +145,7 @@ extern bool tsUseAdapter; extern int32_t tsMetaCacheMaxSize; extern int32_t tsSlowLogThreshold; extern int32_t tsSlowLogScope; +extern int32_t tsTimeSeriesThreshold; // client extern int32_t tsMinSlidingTime; diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 1217f3f6b4..d1d38ca2f8 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1464,6 +1464,11 @@ typedef struct { int32_t learnerProgress; // use one reservered } SVnodeLoad; +typedef struct { + int32_t vgId; + int64_t nTimeSeries; +} SVnodeLoadLite; + typedef struct { int8_t syncState; int64_t syncTerm; @@ -1512,15 +1517,9 @@ int32_t tDeserializeSStatusReq(void* buf, int32_t bufLen, SStatusReq* pReq); void tFreeSStatusReq(SStatusReq* pReq); typedef struct { - int32_t vgId; - int64_t nTimeSeries; -} SDndNotifyInfo; - -int32_t dmProcessNotifyReq(SDndNotifyInfo* pInfo); - -typedef struct { - int32_t nVgroup; - SDndNotifyInfo* payload; + int32_t dnodeId; + int64_t clusterId; + SArray* pVloads; } SNotifyReq; int32_t tSerializeSNotifyReq(void* buf, int32_t bufLen, SNotifyReq* pReq); diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index 39a36a248b..7e030bef3f 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -180,9 +180,6 @@ enum { // WARN: new msg should be appended to segment tail TD_DEF_MSG_TYPE(TDMT_MND_RETRIEVE_IP_WHITE, "retrieve-ip-white", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_GET_USER_WHITELIST, "get-user-whitelist", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_NOTIFY, "notify", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_GRANT_NOTIFY, "grant-notify", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_MAX_MSG, "mnd-max", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_MND_BALANCE_VGROUP_LEADER, "balance-vgroup-leader", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_RESTORE_DNODE, "restore-dnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_PAUSE_STREAM, "pause-stream", NULL, NULL) @@ -191,6 +188,8 @@ enum { // WARN: new msg should be appended to segment tail TD_DEF_MSG_TYPE(TDMT_MND_STREAM_BEGIN_CHECKPOINT, "stream-begin-checkpoint", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_STREAM_NODECHANGE_CHECK, "stream-nodechange-check", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_TRIM_DB_TIMER, "trim-db-tmr", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_GRANT_NOTIFY, "grant-notify", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_MAX_MSG, "mnd-max", NULL, NULL) TD_NEW_MSG_SEG(TDMT_VND_MSG) TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT, "submit", SSubmitReq, SSubmitRsp) diff --git a/include/libs/executor/storageapi.h b/include/libs/executor/storageapi.h index 0a240dd8f5..d21fb8c459 100644 --- a/include/libs/executor/storageapi.h +++ b/include/libs/executor/storageapi.h @@ -142,6 +142,7 @@ typedef struct SSnapContext { typedef struct { int64_t uid; int64_t ctbNum; + int32_t colNum; } SMetaStbStats; // void tqReaderSetColIdList(STqReader *pReader, SArray *pColIdList); @@ -285,8 +286,8 @@ typedef struct SStoreMeta { // db name, vgId, numOfTables, numOfSTables int32_t (*getNumOfChildTables)( - void* pVnode, int64_t uid, - int64_t* numOfTables); // int32_t metaGetStbStats(SMeta *pMeta, int64_t uid, SMetaStbStats *pInfo); + void* pVnode, int64_t uid, int64_t* numOfTables, + int32_t* numOfCols); // int32_t metaGetStbStats(SMeta *pMeta, int64_t uid, SMetaStbStats *pInfo); void (*getBasicInfo)(void* pVnode, const char** dbname, int32_t* vgId, int64_t* numOfTables, int64_t* numOfNormalTables); // vnodeGetInfo(void *pVnode, const char **dbname, int32_t *vgId) & // metaGetTbNum(SMeta *pMeta) & metaGetNtbNum(SMeta *pMeta); diff --git a/include/libs/monitor/monitor.h b/include/libs/monitor/monitor.h index 607179c446..708953f45e 100644 --- a/include/libs/monitor/monitor.h +++ b/include/libs/monitor/monitor.h @@ -191,7 +191,7 @@ typedef struct { } SMonBmInfo; typedef struct { - SArray *pVloads; // SVnodeLoad + SArray *pVloads; // SVnodeLoad/SVnodeLoadLite } SMonVloadInfo; typedef struct { diff --git a/include/os/osThread.h b/include/os/osThread.h index 3fdafe2c34..f0b79ac2c9 100644 --- a/include/os/osThread.h +++ b/include/os/osThread.h @@ -46,7 +46,7 @@ typedef HANDLE TdThreadMutexAttr; // windows api typedef struct { SRWLOCK lock; int8_t excl; -} TdThreadRwlock; // pthread api +} TdThreadRwlock; // windows api typedef pthread_attr_t TdThreadAttr; // pthread api typedef pthread_once_t TdThreadOnce; // pthread api typedef HANDLE TdThreadRwlockAttr; // windows api diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 03285edcf7..251919568f 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -144,6 +144,7 @@ bool tsUseAdapter = false; int32_t tsMetaCacheMaxSize = -1; // MB int32_t tsSlowLogThreshold = 3; // seconds int32_t tsSlowLogScope = SLOW_LOG_TYPE_ALL; +int32_t tsTimeSeriesThreshold = 50; /* * denote if the server needs to compress response message at the application layer to client, including query rsp, @@ -630,6 +631,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "trimVDbIntervalSec", tsTrimVDbIntervalSec, 1, 100000, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "uptimeInterval", tsUptimeInterval, 1, 100000, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "queryRsmaTolerance", tsQueryRsmaTolerance, 0, 900000, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddInt32(pCfg, "timeseriesThreshold", tsTimeSeriesThreshold, 0, 2000, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt64(pCfg, "walFsyncDataSizeLimit", tsWalFsyncDataSizeLimit, 100 * 1024 * 1024, INT64_MAX, CFG_SCOPE_SERVER) != 0) @@ -1036,6 +1038,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsTrimVDbIntervalSec = cfgGetItem(pCfg, "trimVDbIntervalSec")->i32; tsUptimeInterval = cfgGetItem(pCfg, "uptimeInterval")->i32; tsQueryRsmaTolerance = cfgGetItem(pCfg, "queryRsmaTolerance")->i32; + tsTimeSeriesThreshold = cfgGetItem(pCfg, "timeseriesThreshold")->i32; tsWalFsyncDataSizeLimit = cfgGetItem(pCfg, "walFsyncDataSizeLimit")->i64; @@ -1448,6 +1451,8 @@ int32_t taosApplyLocalCfg(SConfig *pCfg, char *name) { tqDebugFlag = cfgGetItem(pCfg, "tqDebugFlag")->i32; } else if (strcasecmp("ttlFlushThreshold", name) == 0) { tsTtlFlushThreshold = cfgGetItem(pCfg, "ttlFlushThreshold")->i32; + } else if (strcasecmp("timeseriesThreshold", name) == 0) { + tsTimeSeriesThreshold = cfgGetItem(pCfg, "timeseriesThreshold")->i32; } break; } diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index c805053a5c..e1ea76f463 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -1041,10 +1041,15 @@ int32_t tSerializeSNotifyReq(void *buf, int32_t bufLen, SNotifyReq *pReq) { if (tStartEncode(&encoder) < 0) return -1; - if (tEncodeI32(&encoder, pReq->nVgroup) < 0) return -1; - for (int32_t i = 0; i < pReq->nVgroup; ++i) { - if (tEncodeI32(&encoder, (pReq->payload + i)->vgId) < 0) return -1; - if (tEncodeI64(&encoder, (pReq->payload + i)->nTimeSeries) < 0) return -1; + if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1; + if (tEncodeI64(&encoder, pReq->clusterId) < 0) return -1; + + int32_t nVgroup = taosArrayGetSize(pReq->pVloads); + if (tEncodeI32(&encoder, nVgroup) < 0) return -1; + for (int32_t i = 0; i < nVgroup; ++i) { + SVnodeLoadLite *vload = TARRAY_GET_ELEM(pReq->pVloads, i); + if (tEncodeI32(&encoder, vload->vgId) < 0) return -1; + if (tEncodeI64(&encoder, vload->nTimeSeries) < 0) return -1; } tEndEncode(&encoder); @@ -1055,29 +1060,39 @@ int32_t tSerializeSNotifyReq(void *buf, int32_t bufLen, SNotifyReq *pReq) { } int32_t tDeserializeSNotifyReq(void *buf, int32_t bufLen, SNotifyReq *pReq) { + int32_t code = TSDB_CODE_INVALID_MSG; SDecoder decoder = {0}; tDecoderInit(&decoder, buf, bufLen); - if (tStartDecode(&decoder) < 0) return -1; - - if (tDecodeI32(&decoder, &pReq->nVgroup) < 0) return -1; - - pReq->payload = taosMemoryMalloc(pReq->nVgroup * (sizeof(SDndNotifyInfo))); - if (!pReq->payload) return -1; - - for (int32_t i = 0; i < pReq->nVgroup; ++i) { - if (tDecodeI32(&decoder, &((pReq->payload + i)->vgId)) < 0) return -1; - if (tDecodeI64(&decoder, &((pReq->payload + i)->nTimeSeries)) < 0) return -1; + if (tStartDecode(&decoder) < 0) goto _exit; + if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) goto _exit; + if (tDecodeI64(&decoder, &pReq->clusterId) < 0) goto _exit; + int32_t nVgroup = 0; + if (tDecodeI32(&decoder, &nVgroup) < 0) goto _exit; + if (nVgroup > 0) { + pReq->pVloads = taosArrayInit(nVgroup, sizeof(SVnodeLoadLite)); + if (!pReq->pVloads) { + code = TSDB_CODE_OUT_OF_MEMORY; + goto _exit; + } + for (int32_t i = 0; i < nVgroup; ++i) { + SVnodeLoadLite *vload = TARRAY_GET_ELEM(pReq->pVloads, i); + if (tDecodeI32(&decoder, &(vload->vgId)) < 0) goto _exit; + if (tDecodeI64(&decoder, &(vload->nTimeSeries)) < 0) goto _exit; + } } + code = 0; + +_exit: tEndDecode(&decoder); tDecoderClear(&decoder); - return 0; + return code; } void tFreeSNotifyReq(SNotifyReq *pReq) { if (pReq) { - taosMemoryFreeClear(pReq->payload); + taosArrayDestroy(pReq->pVloads); } } diff --git a/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h b/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h index 0cf2c4f7b8..93e2513f4a 100644 --- a/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h +++ b/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h @@ -28,6 +28,7 @@ typedef struct SDnodeMgmt { const char *path; const char *name; TdThread statusThread; + TdThread notifyThread; TdThread monitorThread; TdThread crashReportThread; SSingleWorker mgmtWorker; @@ -36,6 +37,7 @@ typedef struct SDnodeMgmt { ProcessDropNodeFp processDropNodeFp; SendMonitorReportFp sendMonitorReportFp; GetVnodeLoadsFp getVnodeLoadsFp; + GetVnodeLoadsFp getVnodeLoadsLiteFp; GetMnodeLoadsFp getMnodeLoadsFp; GetQnodeLoadsFp getQnodeLoadsFp; int32_t statusSeq; @@ -44,6 +46,7 @@ typedef struct SDnodeMgmt { // dmHandle.c SArray *dmGetMsgHandles(); void dmSendStatusReq(SDnodeMgmt *pMgmt); +void dmSendNotifyReq(SDnodeMgmt *pMgmt); int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SRpcMsg *pMsg); @@ -51,11 +54,11 @@ int32_t dmProcessServerRunStatus(SDnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t dmProcessGrantReq(void *pInfo, SRpcMsg *pMsg); int32_t dmProcessGrantNotify(void *pInfo, SRpcMsg *pMsg); -int32_t dmStartNotify(SDnodeMgmt *pMgmt); // dmWorker.c int32_t dmPutNodeMsgToMgmtQueue(SDnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t dmStartStatusThread(SDnodeMgmt *pMgmt); +int32_t dmStartNotifyThread(SDnodeMgmt *pMgmt); void dmStopStatusThread(SDnodeMgmt *pMgmt); int32_t dmStartMonitorThread(SDnodeMgmt *pMgmt); void dmStopMonitorThread(SDnodeMgmt *pMgmt); diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c b/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c index bc53827222..d6a4b76964 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c @@ -170,6 +170,36 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) { dmProcessStatusRsp(pMgmt, &rpcRsp); } +void dmSendNotifyReq(SDnodeMgmt *pMgmt) { + SNotifyReq req = {0}; + + // taosThreadRwlockRdlock(&pMgmt->pData->lock); + // req.dnodeId = pMgmt->pData->dnodeId; + // taosThreadRwlockUnlock(&pMgmt->pData->lock); + + // req.clusterId = pMgmt->pData->clusterId; + + SMonVloadInfo vinfo = {0}; + (*pMgmt->getVnodeLoadsLiteFp)(&vinfo); + req.pVloads = vinfo.pVloads; + + int32_t contLen = tSerializeSNotifyReq(NULL, 0, &req); + void *pHead = rpcMallocCont(contLen); + tSerializeSNotifyReq(pHead, contLen, &req); + tFreeSNotifyReq(&req); + + SRpcMsg rpcMsg = {.pCont = pHead, + .contLen = contLen, + .msgType = TDMT_MND_NOTIFY, + .info.ahandle = (void *)0x9527, + .info.refId = 0, + .info.noResp = 1}; + + SEpSet epSet = {0}; + dmGetMnodeEpSet(pMgmt->pData, &epSet); + rpcSendRequest(pMgmt->msgCb.clientRpc, &epSet, &rpcMsg, NULL); +} + int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) { dError("auth rsp is received, but not supported yet"); return 0; @@ -408,99 +438,3 @@ _OVER: return pArray; } } - -#ifndef TD_ENTERPRISE -int32_t dmStartNotify(SDnodeMgmt *pMgmt) { return 0; } -int32_t dmProcessNotifyReq(SDndNotifyInfo *pInfo) { return 0; } -#else -static SHashObj *tsDndNotifyInfo = NULL; -static int64_t tsTimeSeries = 0; -static int8_t tsDndNotifyInitLock = 0; -static int8_t tsDndNotifyLock = 0; -static SDnodeMgmt *pSDnodeMgmt = NULL; - -int32_t dmStartNotify(SDnodeMgmt *pMgmt) { - pSDnodeMgmt = pMgmt; - return 0; -} - -int32_t dmProcessNotifyReq(SDndNotifyInfo *pInfo) { - if (atomic_load_8(&tsDndNotifyInitLock) != 2) { - if (atomic_val_compare_exchange_8(&tsDndNotifyInitLock, 0, 1) == 0) { - tsDndNotifyInfo = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK); - - // fetch the latest tsTimeSeries of vgroups in current dnode - atomic_store_8(&tsDndNotifyInitLock, 2); - } else { - int32_t nLoops = 0; - while (atomic_load_8(&tsDndNotifyInitLock) != 2) { - if (++nLoops > 1000) { - sched_yield(); - nLoops = 0; - } - } - } - } - - if (atomic_val_compare_exchange_8(&tsDndNotifyLock, 0, 1) != 0) { - return 0; - } - - int64_t lastTimeSeries = atomic_load_64(&tsTimeSeries); - - int64_t *val = NULL; - if ((val = taosHashGet(tsDndNotifyInfo, &pInfo->vgId, sizeof(pInfo->vgId)))) { - if (*val != pInfo->nTimeSeries) { - assert(*val > 0); - atomic_add_fetch_64(&tsTimeSeries, pInfo->nTimeSeries - *val); - taosHashPut(tsDndNotifyInfo, &pInfo->vgId, sizeof(pInfo->vgId), &pInfo->nTimeSeries, sizeof(pInfo->nTimeSeries)); - } - } else { - atomic_add_fetch_64(&tsTimeSeries, pInfo->nTimeSeries); - taosHashPut(tsDndNotifyInfo, &pInfo->vgId, sizeof(pInfo->vgId), &pInfo->nTimeSeries, sizeof(pInfo->nTimeSeries)); - } - - if (atomic_load_64(&tsTimeSeries) - lastTimeSeries > 100) { - int32_t hashSize = taosHashGetSize(tsDndNotifyInfo); - SNotifyReq req = {.nVgroup = hashSize}; - req.payload = taosMemoryMalloc(sizeof(SDndNotifyInfo) * hashSize); - - int64_t *nVal = NULL; - int32_t iter = 0; - - int64_t nnn = 0; - size_t kLen = sizeof(int32_t); - while ((nVal = taosHashIterate(tsDndNotifyInfo, nVal)) && iter < hashSize) { - int32_t *nVgId = taosHashGetKey(nVal, &kLen); - (req.payload + iter)->vgId = *nVgId; - (req.payload + iter)->nTimeSeries = *nVal; - ++iter; - } - // assert(nnn < 1000000); - - int32_t contLen = tSerializeSNotifyReq(NULL, 0, &req); - if (contLen <= 0) return -1; - void *pHead = rpcMallocCont(contLen); - if (!pHead) { - tFreeSNotifyReq(&req); - atomic_store_8(&tsDndNotifyLock, 0); - return -1; - } - tSerializeSNotifyReq(pHead, contLen, &req); - tFreeSNotifyReq(&req); - - SRpcMsg rpcMsg = {.pCont = pHead, - .contLen = contLen, - .msgType = TDMT_MND_NOTIFY, - .info.ahandle = NULL, - .info.refId = 0, - .info.noResp = 1}; - - SEpSet epSet = {0}; - dmGetMnodeEpSet(pSDnodeMgmt->pData, &epSet); - rpcSendRequest(pSDnodeMgmt->msgCb.clientRpc, &epSet, &rpcMsg, NULL); - } - atomic_store_8(&tsDndNotifyLock, 0); - return 0; -} -#endif diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmInt.c b/source/dnode/mgmt/mgmt_dnode/src/dmInt.c index 2f232a995f..1e8ad2f7c2 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmInt.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmInt.c @@ -20,15 +20,17 @@ static int32_t dmStartMgmt(SDnodeMgmt *pMgmt) { if (dmStartStatusThread(pMgmt) != 0) { return -1; } +#ifdef TD_ENTERPRISE + if (dmStartNotifyThread(pMgmt) != 0) { + return -1; + } +#endif if (dmStartMonitorThread(pMgmt) != 0) { return -1; } if (dmStartCrashReportThread(pMgmt) != 0) { return -1; } - if(dmStartNotify(pMgmt) != 0) { - return -1; - } return 0; } @@ -55,6 +57,7 @@ static int32_t dmOpenMgmt(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->processDropNodeFp = pInput->processDropNodeFp; pMgmt->sendMonitorReportFp = pInput->sendMonitorReportFp; pMgmt->getVnodeLoadsFp = pInput->getVnodeLoadsFp; + pMgmt->getVnodeLoadsLiteFp = pInput->getVnodeLoadsLiteFp; pMgmt->getMnodeLoadsFp = pInput->getMnodeLoadsFp; pMgmt->getQnodeLoadsFp = pInput->getQnodeLoadsFp; diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c index 9ddc64358a..4cd7a9f951 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c @@ -17,7 +17,6 @@ #include "dmInt.h" #include "thttp.h" -int8_t tsNeedUpdStatus = 0; static void *dmStatusThreadFp(void *param) { SDnodeMgmt *pMgmt = param; int64_t lastTime = taosGetTimestampMs(); @@ -29,13 +28,13 @@ static void *dmStatusThreadFp(void *param) { int64_t upTime = 0; while (1) { - taosMsleep(100); + taosMsleep(200); if (pMgmt->pData->dropped || pMgmt->pData->stopped) break; int64_t curTime = taosGetTimestampMs(); if (curTime < lastTime) lastTime = curTime; float interval = (curTime - lastTime) / 1000.0f; - if (atomic_val_compare_exchange_8(&tsNeedUpdStatus, 1, 0) == 1 || interval >= tsStatusInterval) { + if (interval >= tsStatusInterval) { dmSendStatusReq(pMgmt); lastTime = curTime; @@ -54,6 +53,26 @@ static void *dmStatusThreadFp(void *param) { return NULL; } +tsem_t dmNotifySem; +static void *dmNotifyThreadFp(void *param) { + SDnodeMgmt *pMgmt = param; + int64_t lastTime = taosGetTimestampMs(); + setThreadName("dnode-notify"); + + if (tsem_init(&dmNotifySem, 0, 0) != 0) { + return NULL; + } + + while (1) { + if (pMgmt->pData->dropped || pMgmt->pData->stopped) break; + + tsem_wait(&dmNotifySem); + dmSendNotifyReq(pMgmt); + } + + return NULL; +} + static void *dmMonitorThreadFp(void *param) { SDnodeMgmt *pMgmt = param; int64_t lastTime = taosGetTimestampMs(); @@ -154,6 +173,29 @@ void dmStopStatusThread(SDnodeMgmt *pMgmt) { } } +int32_t dmStartNotifyThread(SDnodeMgmt *pMgmt) { + TdThreadAttr thAttr; + taosThreadAttrInit(&thAttr); + taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE); + if (taosThreadCreate(&pMgmt->notifyThread, &thAttr, dmNotifyThreadFp, pMgmt) != 0) { + dError("failed to create notify thread since %s", strerror(errno)); + return -1; + } + + taosThreadAttrDestroy(&thAttr); + tmsgReportStartup("dnode-notify", "initialized"); + return 0; +} + +void dmStopNotifyThread(SDnodeMgmt *pMgmt) { + if (taosCheckPthreadValid(pMgmt->notifyThread)) { + tsem_post(&dmNotifySem); + taosThreadJoin(pMgmt->notifyThread, NULL); + taosThreadClear(&pMgmt->notifyThread); + } + tsem_destroy(&dmNotifySem); +} + int32_t dmStartMonitorThread(SDnodeMgmt *pMgmt) { TdThreadAttr thAttr; taosThreadAttrInit(&thAttr); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index 0251b9b636..a13c8fcc57 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -40,6 +40,28 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo, bool isReset) { taosThreadRwlockUnlock(&pMgmt->lock); } +void vmGetVnodeLoadsLite(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) { + pInfo->pVloads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoadLite)); + if (!pInfo->pVloads) return; + + taosThreadRwlockRdlock(&pMgmt->lock); + + void *pIter = taosHashIterate(pMgmt->hash, NULL); + while (pIter) { + SVnodeObj **ppVnode = pIter; + if (ppVnode == NULL || *ppVnode == NULL) continue; + + SVnodeObj *pVnode = *ppVnode; + SVnodeLoadLite vload = {0}; + if (vnodeGetLoadLite(pVnode->pImpl, &vload) == 0) { + taosArrayPush(pInfo->pVloads, &vload); + } + pIter = taosHashIterate(pMgmt->hash, pIter); + } + + taosThreadRwlockUnlock(&pMgmt->lock); +} + void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) { SMonVloadInfo vloads = {0}; vmGetVnodeLoads(pMgmt, &vloads, true); diff --git a/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h b/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h index 98489433b9..1c6625ba1a 100644 --- a/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h +++ b/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h @@ -119,6 +119,7 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); // dmMonitor.c void dmSendMonitorReport(); void dmGetVnodeLoads(SMonVloadInfo *pInfo); +void dmGetVnodeLoadsLite(SMonVloadInfo *pInfo); void dmGetMnodeLoads(SMonMloadInfo *pInfo); void dmGetQnodeLoads(SQnodeLoad *pInfo); diff --git a/source/dnode/mgmt/node_mgmt/inc/dmNodes.h b/source/dnode/mgmt/node_mgmt/inc/dmNodes.h index 5f36309038..7d635c6bdc 100644 --- a/source/dnode/mgmt/node_mgmt/inc/dmNodes.h +++ b/source/dnode/mgmt/node_mgmt/inc/dmNodes.h @@ -35,6 +35,7 @@ void smGetMonitorInfo(void *pMgmt, SMonSmInfo *pInfo); void bmGetMonitorInfo(void *pMgmt, SMonBmInfo *pInfo); void vmGetVnodeLoads(void *pMgmt, SMonVloadInfo *pInfo, bool isReset); +void vmGetVnodeLoadsLite(void *pMgmt, SMonVloadInfo *pInfo); void mmGetMnodeLoads(void *pMgmt, SMonMloadInfo *pInfo); void qmGetQnodeLoads(void *pMgmt, SQnodeLoad *pInfo); diff --git a/source/dnode/mgmt/node_mgmt/src/dmEnv.c b/source/dnode/mgmt/node_mgmt/src/dmEnv.c index 320c9db37d..8815647047 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmEnv.c +++ b/source/dnode/mgmt/node_mgmt/src/dmEnv.c @@ -419,6 +419,7 @@ SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper) { .processDropNodeFp = dmProcessDropNodeReq, .sendMonitorReportFp = dmSendMonitorReport, .getVnodeLoadsFp = dmGetVnodeLoads, + .getVnodeLoadsLiteFp = dmGetVnodeLoadsLite, .getMnodeLoadsFp = dmGetMnodeLoads, .getQnodeLoadsFp = dmGetQnodeLoads, }; diff --git a/source/dnode/mgmt/node_mgmt/src/dmMonitor.c b/source/dnode/mgmt/node_mgmt/src/dmMonitor.c index 4ab56ed682..b3db7c3058 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmMonitor.c +++ b/source/dnode/mgmt/node_mgmt/src/dmMonitor.c @@ -119,6 +119,17 @@ void dmGetVnodeLoads(SMonVloadInfo *pInfo) { } } +void dmGetVnodeLoadsLite(SMonVloadInfo *pInfo) { + SDnode *pDnode = dmInstance(); + SMgmtWrapper *pWrapper = &pDnode->wrappers[VNODE]; + if (dmMarkWrapper(pWrapper) == 0) { + if (pWrapper->pMgmt != NULL) { + vmGetVnodeLoadsLite(pWrapper->pMgmt, pInfo); + } + dmReleaseWrapper(pWrapper); + } +} + void dmGetMnodeLoads(SMonMloadInfo *pInfo) { SDnode *pDnode = dmInstance(); SMgmtWrapper *pWrapper = &pDnode->wrappers[MNODE]; diff --git a/source/dnode/mgmt/node_util/inc/dmUtil.h b/source/dnode/mgmt/node_util/inc/dmUtil.h index 8d506563ec..9d97e6ae9f 100644 --- a/source/dnode/mgmt/node_util/inc/dmUtil.h +++ b/source/dnode/mgmt/node_util/inc/dmUtil.h @@ -92,8 +92,6 @@ typedef void (*GetMnodeLoadsFp)(SMonMloadInfo *pInfo); typedef void (*GetQnodeLoadsFp)(SQnodeLoad *pInfo); typedef int32_t (*ProcessAlterNodeTypeFp)(EDndNodeType ntype, SRpcMsg *pMsg); -int32_t dmProcessNotifyReq(SDndNotifyInfo *pInfo); - typedef struct { int32_t dnodeId; int64_t clusterId; @@ -123,6 +121,7 @@ typedef struct { ProcessDropNodeFp processDropNodeFp; SendMonitorReportFp sendMonitorReportFp; GetVnodeLoadsFp getVnodeLoadsFp; + GetVnodeLoadsFp getVnodeLoadsLiteFp; GetMnodeLoadsFp getMnodeLoadsFp; GetQnodeLoadsFp getQnodeLoadsFp; } SMgmtInputOpt; diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c index 8f9d37187b..300bd6fbb9 100644 --- a/source/dnode/mnode/impl/src/mndDnode.c +++ b/source/dnode/mnode/impl/src/mndDnode.c @@ -70,6 +70,7 @@ static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq); static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq); static int32_t mndProcessConfigDnodeRsp(SRpcMsg *pRsp); static int32_t mndProcessStatusReq(SRpcMsg *pReq); +static int32_t mndProcessNotifyReq(SRpcMsg *pReq); static int32_t mndProcessRestoreDnodeReq(SRpcMsg *pReq); static int32_t mndRetrieveConfigs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows); @@ -80,11 +81,9 @@ static void mndCancelGetNextDnode(SMnode *pMnode, void *pIter); static int32_t mndMCfgGetValInt32(SMCfgDnodeReq *pInMCfgReq, int32_t opLen, int32_t *pOutValue); #ifndef TD_ENTERPRISE -static int32_t mndUpdateClusterInfo(SRpcMsg *pReq) { return 0; } -static int32_t mndProcessNotifyReq(SRpcMsg *pReq) { return 0; } +static int32_t mndUpdClusterInfo(SRpcMsg *pReq) { return 0; } #else -int32_t mndUpdateClusterInfo(SRpcMsg *pReq); -int32_t mndProcessNotifyReq(SRpcMsg *pReq); +int32_t mndUpdClusterInfo(SRpcMsg *pReq); #endif int32_t mndInitDnode(SMnode *pMnode) { @@ -540,7 +539,6 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) { int64_t dnodeVer = sdbGetTableVer(pMnode->pSdb, SDB_DNODE) + sdbGetTableVer(pMnode->pSdb, SDB_MNODE); int64_t curMs = taosGetTimestampMs(); - int64_t nDiffTimeSeries = 0; bool online = mndIsDnodeOnline(pDnode, curMs); bool dnodeChanged = (statusReq.dnodeVer == 0) || (statusReq.dnodeVer != dnodeVer); bool reboot = (pDnode->rebootTime != statusReq.rebootTime); @@ -557,7 +555,6 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) { SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVload->vgId); if (pVgroup != NULL) { - nDiffTimeSeries = pVload->numOfTimeSeries - pVgroup->numOfTimeSeries; if (pVload->syncState == TAOS_SYNC_STATE_LEADER) { pVgroup->cacheUsage = pVload->cacheUsage; pVgroup->numOfCachedTables = pVload->numOfCachedTables; @@ -686,9 +683,41 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) { _OVER: mndReleaseDnode(pMnode, pDnode); taosArrayDestroy(statusReq.pVloads); - if (nDiffTimeSeries > 0) { - mndUpdateClusterInfo(pReq); + mndUpdClusterInfo(pReq); + return code; +} + +static int32_t mndProcessNotifyReq(SRpcMsg *pReq) { + SMnode *pMnode = pReq->info.node; + SNotifyReq notifyReq = {0}; + int32_t code = 0; + + if ((code = tDeserializeSNotifyReq(pReq->pCont, pReq->contLen, ¬ifyReq)) != 0) { + terrno = code; + goto _OVER; } + + // int64_t clusterid = mndGetClusterId(pMnode); + // if (notifyReq.clusterId != 0 && notifyReq.clusterId != clusterid) { + // code = TSDB_CODE_MND_DNODE_DIFF_CLUSTER; + // mWarn("dnode:%d, its clusterid:%" PRId64 " differ from current cluster:%" PRId64 " since %s", notifyReq.dnodeId, + // notifyReq.clusterId, clusterid, tstrerror(code)); + // goto _OVER; + // } + + int32_t nVgroup = taosArrayGetSize(notifyReq.pVloads); + for (int32_t v = 0; v < nVgroup; ++v) { + SVnodeLoadLite *pVload = taosArrayGet(notifyReq.pVloads, v); + + SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVload->vgId); + if (pVgroup != NULL) { + pVgroup->numOfTimeSeries = pVload->nTimeSeries; + mndReleaseVgroup(pMnode, pVgroup); + } + } +_OVER: + mndUpdClusterInfo(pReq); + tFreeSNotifyReq(¬ifyReq); return code; } diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 9d422d8068..727d790b91 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -86,11 +86,13 @@ void *vnodeGetIdx(void *pVnode); void *vnodeGetIvtIdx(void *pVnode); int32_t vnodeGetCtbNum(SVnode *pVnode, int64_t suid, int64_t *num); +int32_t vnodeGetStbColumnNum(SVnode *pVnode, tb_uid_t suid, int *num); int32_t vnodeGetTimeSeriesNum(SVnode *pVnode, int64_t *num); int32_t vnodeGetAllCtbNum(SVnode *pVnode, int64_t *num); void vnodeResetLoad(SVnode *pVnode, SVnodeLoad *pLoad); int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad); +int32_t vnodeGetLoadLite(SVnode *pVnode, SVnodeLoadLite *pLoad); int32_t vnodeValidateTableHash(SVnode *pVnode, char *tableFName); int32_t vnodePreProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg); @@ -134,7 +136,7 @@ bool metaTbInFilterCache(void *pVnode, tb_uid_t suid, int8_t type); int32_t metaPutTbToFilterCache(void *pVnode, tb_uid_t suid, int8_t type); int32_t metaSizeOfTbFilterCache(void *pVnode, int8_t type); -int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables); +int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables, int32_t *numOfCols); // tsdb typedef struct STsdbReader STsdbReader; @@ -288,10 +290,10 @@ typedef struct { int64_t numOfSTables; int64_t numOfCTables; int64_t numOfNTables; - int64_t numOfCmprTimeSeries; + int64_t numOfReportedTimeSeries; int64_t numOfNTimeSeries; int64_t numOfTimeSeries; - int64_t itvTimeSeries; + // int64_t itvTimeSeries; int64_t pointsWritten; int64_t totalStorage; int64_t compStorage; diff --git a/source/dnode/vnode/src/inc/meta.h b/source/dnode/vnode/src/inc/meta.h index 4f25bf31a2..b57e046ad8 100644 --- a/source/dnode/vnode/src/inc/meta.h +++ b/source/dnode/vnode/src/inc/meta.h @@ -71,7 +71,7 @@ int32_t metaCacheDrop(SMeta* pMeta, int64_t uid); int32_t metaStatsCacheUpsert(SMeta* pMeta, SMetaStbStats* pInfo); int32_t metaStatsCacheDrop(SMeta* pMeta, int64_t uid); int32_t metaStatsCacheGet(SMeta* pMeta, int64_t uid, SMetaStbStats* pInfo); -void metaUpdateStbStats(SMeta* pMeta, int64_t uid, int64_t delta); +void metaUpdateStbStats(SMeta* pMeta, int64_t uid, int64_t deltaCtb, int32_t totalCols); int32_t metaUidFilterCacheGet(SMeta* pMeta, uint64_t suid, const void* pKey, int32_t keyLen, LRUHandle** pHandle); struct SMeta { diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 88f0bf890a..2952c863cb 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -171,6 +171,7 @@ int metaAddIndexToSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq); int metaDropIndexFromSTable(SMeta* pMeta, int64_t version, SDropIndexReq* pReq); int64_t metaGetTimeSeriesNum(SMeta* pMeta, int type); +void metaUpdTimeSeriesNum(SMeta* pMeta); SMCtbCursor* metaOpenCtbCursor(void* pVnode, tb_uid_t uid, int lock); int32_t metaResumeCtbCursor(SMCtbCursor* pCtbCur, int8_t first); void metaPauseCtbCursor(SMCtbCursor* pCtbCur); diff --git a/source/dnode/vnode/src/meta/metaQuery.c b/source/dnode/vnode/src/meta/metaQuery.c index c7b7ce9686..1a9001ce2b 100644 --- a/source/dnode/vnode/src/meta/metaQuery.c +++ b/source/dnode/vnode/src/meta/metaQuery.c @@ -696,19 +696,31 @@ int64_t metaGetTbNum(SMeta *pMeta) { return pMeta->pVnode->config.vndStats.numOfCTables + pMeta->pVnode->config.vndStats.numOfNTables; } -// N.B. Called by statusReq per second -int64_t metaGetTimeSeriesNum(SMeta *pMeta, int type) { - // sum of (number of columns of stable - 1) * number of ctables (excluding timestamp column) - if (type || pMeta->pVnode->config.vndStats.numOfTimeSeries <= 0 || - ++pMeta->pVnode->config.vndStats.itvTimeSeries % (60 * 5) == 0) { - int64_t num = 0; - vnodeGetTimeSeriesNum(pMeta->pVnode, &num); - pMeta->pVnode->config.vndStats.numOfTimeSeries = num; +void metaUpdTimeSeriesNum(SMeta *pMeta) { + SVnodeStats *pStats = &pMeta->pVnode->config.vndStats; + int64_t nCtbTimeSeries = 0; - pMeta->pVnode->config.vndStats.itvTimeSeries = (TD_VID(pMeta->pVnode) % 100) * 2; + vnodeGetTimeSeriesNum(pMeta->pVnode, &nCtbTimeSeries); + atomic_store_64(&pStats->numOfTimeSeries, nCtbTimeSeries); +} + +static int64_t metaGetTimeSeriesNumImpl(SMeta *pMeta, bool forceUpd) { + // sum of (number of columns of stable - 1) * number of ctables (excluding timestamp column) + SVnodeStats *pStats = &pMeta->pVnode->config.vndStats; + if (forceUpd || pStats->numOfTimeSeries < 0) { + metaUpdTimeSeriesNum(pMeta); } - return pMeta->pVnode->config.vndStats.numOfTimeSeries + pMeta->pVnode->config.vndStats.numOfNTimeSeries; + return pStats->numOfTimeSeries + pStats->numOfNTimeSeries; +} + +// type: 1 report timeseries +int64_t metaGetTimeSeriesNum(SMeta *pMeta, int type) { + int64_t nTimeSeries = metaGetTimeSeriesNumImpl(pMeta, false); + if (type == 1) { + atomic_store_64(&pMeta->pVnode->config.vndStats.numOfReportedTimeSeries, nTimeSeries); + } + return nTimeSeries; } typedef struct { @@ -1506,9 +1518,10 @@ _exit: return code; } -int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables) { +int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables, int32_t *numOfCols) { int32_t code = 0; - *numOfTables = 0; + + if (!numOfTables && !numOfCols) goto _exit; SVnode *pVnodeObj = pVnode; metaRLock(pVnodeObj->pMeta); @@ -1517,7 +1530,8 @@ int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables) { SMetaStbStats state = {0}; if (metaStatsCacheGet(pVnodeObj->pMeta, uid, &state) == TSDB_CODE_SUCCESS) { metaULock(pVnodeObj->pMeta); - *numOfTables = state.ctbNum; + if (numOfTables) *numOfTables = state.ctbNum; + if (numOfCols) *numOfCols = state.colNum; goto _exit; } @@ -1526,10 +1540,15 @@ int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables) { vnodeGetCtbNum(pVnode, uid, &ctbNum); metaULock(pVnodeObj->pMeta); - *numOfTables = ctbNum; + if (numOfTables) *numOfTables = ctbNum; + + int32_t colNum = 0; + vnodeGetStbColumnNum(pVnode, uid, &colNum); + if (numOfCols) *numOfCols = colNum; state.uid = uid; state.ctbNum = ctbNum; + state.colNum = colNum; // upsert the cache metaWLock(pVnodeObj->pMeta); @@ -1540,11 +1559,12 @@ _exit: return code; } -void metaUpdateStbStats(SMeta *pMeta, int64_t uid, int64_t delta) { +void metaUpdateStbStats(SMeta *pMeta, int64_t uid, int64_t deltaCtb, int32_t totalCols) { SMetaStbStats stats = {0}; if (metaStatsCacheGet(pMeta, uid, &stats) == TSDB_CODE_SUCCESS) { - stats.ctbNum += delta; + stats.ctbNum += deltaCtb; + if (totalCols > 0) stats.colNum = totalCols; metaStatsCacheUpsert(pMeta, &stats); } diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index 4102fa4312..c10811083f 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -15,7 +15,7 @@ #include "meta.h" -extern int8_t tsNeedUpdStatus; +extern tsem_t dmNotifySem; static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema); static int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema); @@ -194,6 +194,16 @@ int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSche return 0; } +static inline void metaTimeSeriesNotifyCheck(SMeta *pMeta) { +#ifdef TD_ENTERPRISE + int64_t nTimeSeries = metaGetTimeSeriesNum(pMeta, 0); + int64_t deltaTS = nTimeSeries - pMeta->pVnode->config.vndStats.numOfReportedTimeSeries; + if (abs(deltaTS) > tsTimeSeriesThreshold) { + tsem_post(&dmNotifySem); + } +#endif +} + int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { SMetaEntry me = {0}; int kLen = 0; @@ -393,6 +403,8 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { // metaStatsCacheDrop(pMeta, nStbEntry.uid); + metaUpdateStbStats(pMeta, pReq->suid, 0, pReq->schemaRow.nCols); + metaULock(pMeta); if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf); @@ -773,10 +785,12 @@ int metaCreateTable(SMeta *pMeta, int64_t ver, SVCreateTbReq *pReq, STableMetaRs ++pStats->numOfCTables; - pStats->numOfTimeSeries += 2; // 2 cols for test. + int32_t nCols = 0; + metaGetStbStats(pMeta->pVnode, me.ctbEntry.suid, 0, &nCols); + pStats->numOfTimeSeries = pStats->numOfCTables * (nCols - 1); metaWLock(pMeta); - metaUpdateStbStats(pMeta, me.ctbEntry.suid, 1); + metaUpdateStbStats(pMeta, me.ctbEntry.suid, 1, 0); metaUidCacheClear(pMeta, me.ctbEntry.suid); metaTbGroupCacheClear(pMeta, me.ctbEntry.suid); metaULock(pMeta); @@ -793,15 +807,8 @@ int metaCreateTable(SMeta *pMeta, int64_t ver, SVCreateTbReq *pReq, STableMetaRs } if (metaHandleEntry(pMeta, &me) < 0) goto _err; - // assert(pStats->numOfTimeSeries + pStats->numOfNTimeSeries < 200000); - // if (pStats->numOfTimeSeries + pStats->numOfNTimeSeries - pStats->numOfCmprTimeSeries > 100) { - // pStats->numOfCmprTimeSeries = pStats->numOfTimeSeries + pStats->numOfNTimeSeries; - // SDndNotifyInfo dNotifyInfo = {.vgId = pMeta->pVnode->config.vgId, - // .nTimeSeries = pStats->numOfTimeSeries + pStats->numOfNTimeSeries}; - // dmProcessNotifyReq(&dNotifyInfo); - // } - atomic_val_compare_exchange_8(&tsNeedUpdStatus, 0, 1); + metaTimeSeriesNotifyCheck(pMeta); if (pMetaRsp) { *pMetaRsp = taosMemoryCalloc(1, sizeof(STableMetaRsp)); @@ -1086,19 +1093,23 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { if (e.type != TSDB_SUPER_TABLE) metaDeleteTtl(pMeta, &e); + SVnodeStats *pStats = &pMeta->pVnode->config.vndStats; if (e.type == TSDB_CHILD_TABLE) { tdbTbDelete(pMeta->pCtbIdx, &(SCtbIdxKey){.suid = e.ctbEntry.suid, .uid = uid}, sizeof(SCtbIdxKey), pMeta->txn); - --pMeta->pVnode->config.vndStats.numOfCTables; + --pStats->numOfCTables; + int32_t nCols = 0; + metaGetStbStats(pMeta->pVnode, e.ctbEntry.suid, 0, &nCols); + pStats->numOfTimeSeries = pStats->numOfCTables * (nCols - 1); - metaUpdateStbStats(pMeta, e.ctbEntry.suid, -1); + metaUpdateStbStats(pMeta, e.ctbEntry.suid, -1, 0); metaUidCacheClear(pMeta, e.ctbEntry.suid); metaTbGroupCacheClear(pMeta, e.ctbEntry.suid); } else if (e.type == TSDB_NORMAL_TABLE) { // drop schema.db (todo) - --pMeta->pVnode->config.vndStats.numOfNTables; - pMeta->pVnode->config.vndStats.numOfNTimeSeries -= e.ntbEntry.schemaRow.nCols - 1; + --pStats->numOfNTables; + pStats->numOfNTimeSeries -= e.ntbEntry.schemaRow.nCols - 1; } else if (e.type == TSDB_SUPER_TABLE) { tdbTbDelete(pMeta->pSuidIdx, &e.uid, sizeof(tb_uid_t), pMeta->txn); // drop schema.db (todo) @@ -1106,13 +1117,14 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { metaStatsCacheDrop(pMeta, uid); metaUidCacheClear(pMeta, uid); metaTbGroupCacheClear(pMeta, uid); - --pMeta->pVnode->config.vndStats.numOfSTables; + metaUpdTimeSeriesNum(pMeta); + --pStats->numOfSTables; } - atomic_val_compare_exchange_8(&tsNeedUpdStatus, 0, 1); - metaCacheDrop(pMeta, uid); + metaTimeSeriesNotifyCheck(pMeta); + tDecoderClear(&dc); tdbFree(pData); @@ -1272,6 +1284,7 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl strcpy(pSchema->pSchema[entry.ntbEntry.schemaRow.nCols - 1].name, pAlterTbReq->colName); ++pMeta->pVnode->config.vndStats.numOfNTimeSeries; + metaTimeSeriesNotifyCheck(pMeta); break; case TSDB_ALTER_TABLE_DROP_COLUMN: if (pColumn == NULL) { @@ -1294,6 +1307,7 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl pSchema->nCols--; --pMeta->pVnode->config.vndStats.numOfNTimeSeries; + metaTimeSeriesNotifyCheck(pMeta); break; case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: if (pColumn == NULL) { diff --git a/source/dnode/vnode/src/vnd/vnodeQuery.c b/source/dnode/vnode/src/vnd/vnodeQuery.c index be4941d444..c8b2a797fa 100644 --- a/source/dnode/vnode/src/vnd/vnodeQuery.c +++ b/source/dnode/vnode/src/vnd/vnodeQuery.c @@ -400,6 +400,15 @@ int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) { return 0; } +int32_t vnodeGetLoadLite(SVnode *pVnode, SVnodeLoadLite *pLoad) { + SSyncState syncState = syncGetState(pVnode->sync); + if (syncState.state == TAOS_SYNC_STATE_LEADER) { + pLoad->vgId = TD_VID(pVnode); + pLoad->nTimeSeries = metaGetTimeSeriesNum(pVnode->pMeta, 1); + return 0; + } + return -1; +} /** * @brief Reset the statistics value by monitor interval * @@ -544,7 +553,7 @@ int32_t vnodeGetCtbNum(SVnode *pVnode, int64_t suid, int64_t *num) { return TSDB_CODE_SUCCESS; } -static int32_t vnodeGetStbColumnNum(SVnode *pVnode, tb_uid_t suid, int *num) { +int32_t vnodeGetStbColumnNum(SVnode *pVnode, tb_uid_t suid, int *num) { SSchemaWrapper *pSW = metaGetTableSchema(pVnode->pMeta, suid, -1, 1); if (pSW) { *num = pSW->nCols; @@ -634,10 +643,8 @@ int32_t vnodeGetTimeSeriesNum(SVnode *pVnode, int64_t *num) { tb_uid_t suid = *(tb_uid_t *)taosArrayGet(suidList, i); int64_t ctbNum = 0; - metaGetStbStats(pVnode, suid, &ctbNum); - - int numOfCols = 0; - vnodeGetStbColumnNum(pVnode, suid, &numOfCols); + int32_t numOfCols = 0; + metaGetStbStats(pVnode, suid, &ctbNum, &numOfCols); *num += ctbNum * (numOfCols - 1); } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 474128007a..b95a948937 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -3928,7 +3928,7 @@ static void buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanO pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid); int64_t numOfChildTables = 0; - pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables); + pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL); fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes); } else { @@ -3979,7 +3979,7 @@ static void buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, S pRes->info.id.groupId = groupId; int64_t ctbNum = 0; - int32_t code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum); + int32_t code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL); fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes); }