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

enh: dnode notify logic

This commit is contained in:
kailixu 2023-10-05 21:35:57 +08:00
parent b1a0912a61
commit 086213a6c2
4 changed files with 52 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/libs/monitor/monitor.h
View File

@ -206,6 +206,11 @@ typedef struct {
bool comp;
} SMonCfg;
typedef struct {
int8_t state;
tsem_t sem;
} SDmNotifyHandle;
int32_t monInit(const SMonCfg *pCfg);
void monCleanup();
void monRecordLog(int64_t ts, ELogLevel level, const char *content);

19
source/dnode/mgmt/mgmt_dnode/src/dmWorker.c
View File

@ -53,21 +53,28 @@ static void *dmStatusThreadFp(void *param) {
return NULL;
}
tsem_t dmNotifySem;
static void *dmNotifyThreadFp(void *param) {
SDmNotifyHandle dmNotifyHdl = {.state = 0};
static void *dmNotifyThreadFp(void *param) {
SDnodeMgmt *pMgmt = param;
int64_t lastTime = taosGetTimestampMs();
setThreadName("dnode-notify");
if (tsem_init(&dmNotifySem, 0, 0) != 0) {
if (tsem_init(&dmNotifyHdl.sem, 0, 0) != 0) {
return NULL;
}
while (1) {
if (pMgmt->pData->dropped || pMgmt->pData->stopped) break;
tsem_wait(&dmNotifySem);
_wait:
tsem_wait(&dmNotifyHdl.sem);
_send:
atomic_store_8(&dmNotifyHdl.state, 1);
dmSendNotifyReq(pMgmt);
if (1 == atomic_val_compare_exchange_8(&dmNotifyHdl.state, 1, 0)) {
goto _wait;
}
goto _send;
}
return NULL;
@ -189,11 +196,11 @@ int32_t dmStartNotifyThread(SDnodeMgmt *pMgmt) {
void dmStopNotifyThread(SDnodeMgmt *pMgmt) {
if (taosCheckPthreadValid(pMgmt->notifyThread)) {
tsem_post(&dmNotifySem);
tsem_post(&dmNotifyHdl.sem);
taosThreadJoin(pMgmt->notifyThread, NULL);
taosThreadClear(&pMgmt->notifyThread);
}
tsem_destroy(&dmNotifySem);
tsem_destroy(&dmNotifyHdl.sem);
}
int32_t dmStartMonitorThread(SDnodeMgmt *pMgmt) {

10
source/dnode/vnode/src/meta/metaTable.c
View File

@ -15,7 +15,7 @@
#include "meta.h"
extern tsem_t dmNotifySem;
extern SDmNotifyHandle dmNotifyHdl;
static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema);
static int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema);
@ -198,7 +198,11 @@ static inline void metaTimeSeriesNotifyCheck(SMeta *pMeta) {
#if defined(TD_ENTERPRISE) && !defined(_TD_DARWIN_64)
int64_t nTimeSeries = metaGetTimeSeriesNum(pMeta, 0);
int64_t deltaTS = nTimeSeries - pMeta->pVnode->config.vndStats.numOfReportedTimeSeries;
if (deltaTS > tsTimeSeriesThreshold) tsem_post(&dmNotifySem);
if (deltaTS > tsTimeSeriesThreshold) {
if (1 != atomic_val_compare_exchange_8(&dmNotifyHdl.state, 1, 2)) {
tsem_post(&dmNotifyHdl.sem);
}
}
#endif
}
@ -1066,7 +1070,7 @@ static int metaDeleteTtl(SMeta *pMeta, const SMetaEntry *pME) {
return ttlMgrDeleteTtl(pMeta->pTtlMgr, &ctx);
}
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *pSuid, char* stbName) {
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *pSuid) { //}, char* stbName) {
void *pData = NULL;
int nData = 0;
int rc = 0;

39
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -566,7 +566,8 @@ int32_t vnodeGetStbColumnNum(SVnode *pVnode, tb_uid_t suid, int *num) {
}
#ifdef TD_ENTERPRISE
#define TK_LOG_STB_NUM 20
#define TK_LOG_STB_NUM 19
#define TK_AUDIT_STB_NUM 1
static const char *tkLogStb[TK_LOG_STB_NUM] = {"cluster_info",
"data_dir",
"dnodes_info",
@ -585,24 +586,38 @@ static const char *tkLogStb[TK_LOG_STB_NUM] = {"cluster_info",
"taosadapter_system_mem_percent",
"temp_dir",
"vgroups_info",
"vnodes_role",
"operations"};
"vnodes_role"};
static const char *tkAuditStb[TK_AUDIT_STB_NUM] = {"operations"};
// exclude stbs of taoskeeper log
static int32_t vnodeGetTimeSeriesBlackList(SVnode *pVnode) {
char *dbName = strchr(pVnode->config.dbname, '.');
if (!dbName || 0 != strncmp(++dbName, "log", TSDB_DB_NAME_LEN)) {
if (!dbName) {
return 0;
}
int32_t tbSize = metaSizeOfTbFilterCache(pVnode, 0);
if (tbSize < TK_LOG_STB_NUM) {
for (int32_t i = 0; i < TK_LOG_STB_NUM; ++i) {
tb_uid_t suid = metaGetTableEntryUidByName(pVnode->pMeta, tkLogStb[i]);
if (suid != 0) {
metaPutTbToFilterCache(pVnode, &suid, 0);
}
}
int32_t tbSize = 0;
if (0 == strncmp(++dbName, "log", TSDB_DB_NAME_LEN)) {
tbSize = metaSizeOfTbFilterCache(pVnode, 0);
if (tbSize < TK_LOG_STB_NUM) {
for (int32_t i = 0; i < TK_LOG_STB_NUM; ++i) {
tb_uid_t suid = metaGetTableEntryUidByName(pVnode->pMeta, tkLogStb[i]);
if (suid != 0) {
metaPutTbToFilterCache(pVnode, &suid, 0);
}
}
tbSize = metaSizeOfTbFilterCache(pVnode, 0);
}
} else if (0 == strncmp(dbName, "audit", TSDB_DB_NAME_LEN)) {
tbSize = metaSizeOfTbFilterCache(pVnode, 0);
if (tbSize < TK_AUDIT_STB_NUM) {
for (int32_t i = 0; i < TK_AUDIT_STB_NUM; ++i) {
tb_uid_t suid = metaGetTableEntryUidByName(pVnode->pMeta, tkAuditStb[i]);
if (suid != 0) {
metaPutTbToFilterCache(pVnode, &suid, 0);
}
}
tbSize = metaSizeOfTbFilterCache(pVnode, 0);
}
}
return tbSize;