Merge pull request #16791 from taosdata/fix/ctgMemleak
fix(ctg): fix memleak
This commit is contained in:
Shengliang Guan
GitHub
commit
abdd28f355
|
|
@ -13,72 +13,23 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "trpc.h"
|
||||
#include "query.h"
|
||||
#include "tname.h"
|
||||
#include "catalogInt.h"
|
||||
#include "query.h"
|
||||
#include "systable.h"
|
||||
#include "tname.h"
|
||||
#include "trpc.h"
|
||||
|
||||
SCtgOperation gCtgCacheOperation[CTG_OP_MAX] = {
|
||||
{
|
||||
CTG_OP_UPDATE_VGROUP,
|
||||
"update vgInfo",
|
||||
ctgOpUpdateVgroup
|
||||
},
|
||||
{
|
||||
CTG_OP_UPDATE_TB_META,
|
||||
"update tbMeta",
|
||||
ctgOpUpdateTbMeta
|
||||
},
|
||||
{
|
||||
CTG_OP_DROP_DB_CACHE,
|
||||
"drop DB",
|
||||
ctgOpDropDbCache
|
||||
},
|
||||
{
|
||||
CTG_OP_DROP_DB_VGROUP,
|
||||
"drop DBVgroup",
|
||||
ctgOpDropDbVgroup
|
||||
},
|
||||
{
|
||||
CTG_OP_DROP_STB_META,
|
||||
"drop stbMeta",
|
||||
ctgOpDropStbMeta
|
||||
},
|
||||
{
|
||||
CTG_OP_DROP_TB_META,
|
||||
"drop tbMeta",
|
||||
ctgOpDropTbMeta
|
||||
},
|
||||
{
|
||||
CTG_OP_UPDATE_USER,
|
||||
"update user",
|
||||
ctgOpUpdateUser
|
||||
},
|
||||
{
|
||||
CTG_OP_UPDATE_VG_EPSET,
|
||||
"update epset",
|
||||
ctgOpUpdateEpset
|
||||
},
|
||||
{
|
||||
CTG_OP_UPDATE_TB_INDEX,
|
||||
"update tbIndex",
|
||||
ctgOpUpdateTbIndex
|
||||
},
|
||||
{
|
||||
CTG_OP_DROP_TB_INDEX,
|
||||
"drop tbIndex",
|
||||
ctgOpDropTbIndex
|
||||
},
|
||||
{
|
||||
CTG_OP_CLEAR_CACHE,
|
||||
"clear cache",
|
||||
ctgOpClearCache
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
|
||||
SCtgOperation gCtgCacheOperation[CTG_OP_MAX] = {{CTG_OP_UPDATE_VGROUP, "update vgInfo", ctgOpUpdateVgroup},
|
||||
{CTG_OP_UPDATE_TB_META, "update tbMeta", ctgOpUpdateTbMeta},
|
||||
{CTG_OP_DROP_DB_CACHE, "drop DB", ctgOpDropDbCache},
|
||||
{CTG_OP_DROP_DB_VGROUP, "drop DBVgroup", ctgOpDropDbVgroup},
|
||||
{CTG_OP_DROP_STB_META, "drop stbMeta", ctgOpDropStbMeta},
|
||||
{CTG_OP_DROP_TB_META, "drop tbMeta", ctgOpDropTbMeta},
|
||||
{CTG_OP_UPDATE_USER, "update user", ctgOpUpdateUser},
|
||||
{CTG_OP_UPDATE_VG_EPSET, "update epset", ctgOpUpdateEpset},
|
||||
{CTG_OP_UPDATE_TB_INDEX, "update tbIndex", ctgOpUpdateTbIndex},
|
||||
{CTG_OP_DROP_TB_INDEX, "drop tbIndex", ctgOpDropTbIndex},
|
||||
{CTG_OP_CLEAR_CACHE, "clear cache", ctgOpClearCache}};
|
||||
|
||||
int32_t ctgRLockVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) {
|
||||
CTG_LOCK(CTG_READ, &dbCache->vgCache.vgLock);
|
||||
|
|
@ -92,7 +43,6 @@ int32_t ctgRLockVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
if (NULL == dbCache->vgCache.vgInfo) {
|
||||
CTG_UNLOCK(CTG_READ, &dbCache->vgCache.vgLock);
|
||||
|
||||
|
|
@ -118,17 +68,11 @@ int32_t ctgWLockVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
void ctgRUnlockVgInfo(SCtgDBCache *dbCache) {
|
||||
CTG_UNLOCK(CTG_READ, &dbCache->vgCache.vgLock);
|
||||
}
|
||||
void ctgRUnlockVgInfo(SCtgDBCache *dbCache) { CTG_UNLOCK(CTG_READ, &dbCache->vgCache.vgLock); }
|
||||
|
||||
void ctgWUnlockVgInfo(SCtgDBCache *dbCache) {
|
||||
CTG_UNLOCK(CTG_WRITE, &dbCache->vgCache.vgLock);
|
||||
}
|
||||
void ctgWUnlockVgInfo(SCtgDBCache *dbCache) { CTG_UNLOCK(CTG_WRITE, &dbCache->vgCache.vgLock); }
|
||||
|
||||
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache) {
|
||||
CTG_UNLOCK(CTG_READ, &dbCache->dbLock);
|
||||
}
|
||||
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache) { CTG_UNLOCK(CTG_READ, &dbCache->dbLock); }
|
||||
|
||||
int32_t ctgAcquireDBCacheImpl(SCatalog *pCtg, const char *dbFName, SCtgDBCache **pCache, bool acquire) {
|
||||
char *p = strchr(dbFName, '.');
|
||||
|
|
@ -321,7 +265,6 @@ _return:
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgAcquireTbIndexFromCache(SCatalog *pCtg, char *dbFName, char *tbName, SCtgDBCache **pDb, SCtgTbCache **pTb) {
|
||||
SCtgDBCache *dbCache = NULL;
|
||||
SCtgTbCache *pCache = NULL;
|
||||
|
|
@ -362,7 +305,6 @@ _return:
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgTbMetaExistInCache(SCatalog *pCtg, char *dbFName, char *tbName, int32_t *exist) {
|
||||
SCtgDBCache *dbCache = NULL;
|
||||
SCtgTbCache *tbCache = NULL;
|
||||
|
|
@ -431,8 +373,8 @@ int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta**
|
|||
memcpy(*pTableMeta, tbMeta, metaSize);
|
||||
|
||||
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
|
||||
ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s",
|
||||
ctx->pName->tname, ctx->tbInfo.tbType, dbFName);
|
||||
ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s", ctx->pName->tname,
|
||||
ctx->tbInfo.tbType, dbFName);
|
||||
|
||||
ctgAcquireStbMetaFromCache(pCtg, dbFName, ctx->tbInfo.suid, &dbCache, &tbCache);
|
||||
if (NULL == tbCache) {
|
||||
|
|
@ -472,8 +414,8 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid,
|
||||
char *stbName) {
|
||||
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType,
|
||||
uint64_t *suid, char *stbName) {
|
||||
*sver = -1;
|
||||
*tver = -1;
|
||||
|
||||
|
|
@ -496,8 +438,8 @@ int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver,
|
|||
*sver = tbMeta->sversion;
|
||||
*tver = tbMeta->tversion;
|
||||
|
||||
ctgDebug("Got tb %s ver from cache, dbFName:%s, tbType:%d, sver:%d, tver:%d, suid:0x%" PRIx64,
|
||||
pTableName->tname, dbFName, *tbType, *sver, *tver, *suid);
|
||||
ctgDebug("Got tb %s ver from cache, dbFName:%s, tbType:%d, sver:%d, tver:%d, suid:0x%" PRIx64, pTableName->tname,
|
||||
dbFName, *tbType, *sver, *tver, *suid);
|
||||
|
||||
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
|
@ -533,12 +475,12 @@ int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver,
|
|||
|
||||
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
|
||||
|
||||
ctgDebug("Got tb %s sver %d tver %d from cache, type:%d, dbFName:%s", pTableName->tname, *sver, *tver, *tbType, dbFName);
|
||||
ctgDebug("Got tb %s sver %d tver %d from cache, type:%d, dbFName:%s", pTableName->tname, *sver, *tver, *tbType,
|
||||
dbFName);
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgReadTbTypeFromCache(SCatalog *pCtg, char *dbFName, char *tbName, int32_t *tbType) {
|
||||
SCtgDBCache *dbCache = NULL;
|
||||
SCtgTbCache *tbCache = NULL;
|
||||
|
|
@ -651,7 +593,6 @@ void ctgDequeue(SCtgCacheOperation **op) {
|
|||
*op = node->op;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgEnqueue(SCatalog *pCtg, SCtgCacheOperation *operation) {
|
||||
SCtgQNode *node = taosMemoryCalloc(1, sizeof(SCtgQNode));
|
||||
if (NULL == node) {
|
||||
|
|
@ -699,7 +640,6 @@ int32_t ctgEnqueue(SCatalog* pCtg, SCtgCacheOperation *operation) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgDropDbCacheEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId) {
|
||||
int32_t code = 0;
|
||||
SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
|
||||
|
|
@ -764,9 +704,8 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
|
||||
int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncOp) {
|
||||
int32_t ctgDropStbMetaEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid,
|
||||
bool syncOp) {
|
||||
int32_t code = 0;
|
||||
SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
|
||||
op->opId = CTG_OP_DROP_STB_META;
|
||||
|
|
@ -796,8 +735,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
|
||||
int32_t ctgDropTbMetaEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncOp) {
|
||||
int32_t code = 0;
|
||||
SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
|
||||
|
|
@ -931,8 +868,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
|
||||
int32_t ctgUpdateUserEnqueue(SCatalog *pCtg, SGetUserAuthRsp *pAuth, bool syncOp) {
|
||||
int32_t code = 0;
|
||||
SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
|
||||
|
|
@ -1022,7 +957,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgClearCacheEnqueue(SCatalog *pCtg, bool freeCtg, bool stopQueue, bool syncOp) {
|
||||
int32_t code = 0;
|
||||
SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
|
||||
|
|
@ -1052,7 +986,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
|
||||
mgmt->slotRIdx = 0;
|
||||
mgmt->slotNum = rentSec / CTG_RENT_SLOT_SECOND;
|
||||
|
|
@ -1071,7 +1004,6 @@ int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size) {
|
||||
int16_t widx = abs((int)(id % mgmt->slotNum));
|
||||
|
||||
|
|
@ -1082,7 +1014,8 @@ int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size)
|
|||
if (NULL == slot->meta) {
|
||||
slot->meta = taosArrayInit(CTG_DEFAULT_RENT_SLOT_SIZE, size);
|
||||
if (NULL == slot->meta) {
|
||||
qError("taosArrayInit %d failed, id:0x%"PRIx64", slot idx:%d, type:%d", CTG_DEFAULT_RENT_SLOT_SIZE, id, widx, mgmt->type);
|
||||
qError("taosArrayInit %d failed, id:0x%" PRIx64 ", slot idx:%d, type:%d", CTG_DEFAULT_RENT_SLOT_SIZE, id, widx,
|
||||
mgmt->type);
|
||||
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
|
|
@ -1102,7 +1035,8 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
|
||||
int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size, __compar_fn_t sortCompare,
|
||||
__compar_fn_t searchCompare) {
|
||||
int16_t widx = abs((int)(id % mgmt->slotNum));
|
||||
|
||||
SCtgRentSlot *slot = &mgmt->slots[widx];
|
||||
|
|
@ -1115,7 +1049,8 @@ int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t si
|
|||
}
|
||||
|
||||
if (slot->needSort) {
|
||||
qDebug("meta slot before sorte, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
|
||||
qDebug("meta slot before sorte, slot idx:%d, type:%d, size:%d", widx, mgmt->type,
|
||||
(int32_t)taosArrayGetSize(slot->meta));
|
||||
taosArraySort(slot->meta, sortCompare);
|
||||
slot->needSort = false;
|
||||
qDebug("meta slot sorted, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
|
||||
|
|
@ -1123,7 +1058,8 @@ int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t si
|
|||
|
||||
void *orig = taosArraySearch(slot->meta, &id, searchCompare, TD_EQ);
|
||||
if (NULL == orig) {
|
||||
qDebug("meta not found in slot, id:0x%"PRIx64", slot idx:%d, type:%d, size:%d", id, widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
|
||||
qDebug("meta not found in slot, id:0x%" PRIx64 ", slot idx:%d, type:%d, size:%d", id, widx, mgmt->type,
|
||||
(int32_t)taosArrayGetSize(slot->meta));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
|
|
@ -1136,7 +1072,8 @@ _return:
|
|||
CTG_UNLOCK(CTG_WRITE, &slot->lock);
|
||||
|
||||
if (code) {
|
||||
qDebug("meta in rent update failed, will try to add it, code:%x, id:0x%"PRIx64", slot idx:%d, type:%d", code, id, widx, mgmt->type);
|
||||
qDebug("meta in rent update failed, will try to add it, code:%x, id:0x%" PRIx64 ", slot idx:%d, type:%d", code, id,
|
||||
widx, mgmt->type);
|
||||
CTG_RET(ctgMetaRentAdd(mgmt, meta, id, size));
|
||||
}
|
||||
|
||||
|
|
@ -1178,7 +1115,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgMetaRentGetImpl(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
|
||||
int16_t ridx = atomic_add_fetch_16(&mgmt->slotRIdx, 1);
|
||||
if (ridx >= mgmt->slotNum) {
|
||||
|
|
@ -1254,13 +1190,15 @@ int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
|
|||
SCtgDBCache newDBCache = {0};
|
||||
newDBCache.dbId = dbId;
|
||||
|
||||
newDBCache.tbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
|
||||
newDBCache.tbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY),
|
||||
true, HASH_ENTRY_LOCK);
|
||||
if (NULL == newDBCache.tbCache) {
|
||||
ctgError("taosHashInit %d metaCache failed", gCtgMgmt.cfg.maxTblCacheNum);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
newDBCache.stbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
|
||||
newDBCache.stbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT),
|
||||
true, HASH_ENTRY_LOCK);
|
||||
if (NULL == newDBCache.stbCache) {
|
||||
ctgError("taosHashInit %d stbCache failed", gCtgMgmt.cfg.maxTblCacheNum);
|
||||
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
|
|
@ -1297,7 +1235,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
void ctgRemoveStbRent(SCatalog *pCtg, SCtgDBCache *dbCache) {
|
||||
if (NULL == dbCache->stbCache) {
|
||||
return;
|
||||
|
|
@ -1308,7 +1245,8 @@ void ctgRemoveStbRent(SCatalog* pCtg, SCtgDBCache *dbCache) {
|
|||
uint64_t *suid = NULL;
|
||||
suid = taosHashGetKey(pIter, NULL);
|
||||
|
||||
if (TSDB_CODE_SUCCESS == ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare)) {
|
||||
if (TSDB_CODE_SUCCESS ==
|
||||
ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare)) {
|
||||
ctgDebug("stb removed from rent, suid:0x%" PRIx64, *suid);
|
||||
}
|
||||
|
||||
|
|
@ -1316,7 +1254,6 @@ void ctgRemoveStbRent(SCatalog* pCtg, SCtgDBCache *dbCache) {
|
|||
}
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgRemoveDBFromCache(SCatalog *pCtg, SCtgDBCache *dbCache, const char *dbFName) {
|
||||
uint64_t dbId = dbCache->dbId;
|
||||
|
||||
|
|
@ -1344,7 +1281,6 @@ int32_t ctgRemoveDBFromCache(SCatalog* pCtg, SCtgDBCache *dbCache, const char* d
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgGetAddDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId, SCtgDBCache **pCache) {
|
||||
int32_t code = 0;
|
||||
SCtgDBCache *dbCache = NULL;
|
||||
|
|
@ -1386,7 +1322,8 @@ int32_t ctgGetAddDBCache(SCatalog* pCtg, const char *dbFName, uint64_t dbId, SCt
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t ctgUpdateRentStbVersion(SCatalog *pCtg, char* dbFName, char* tbName, uint64_t dbId, uint64_t suid, SCtgTbCache* pCache) {
|
||||
int32_t ctgUpdateRentStbVersion(SCatalog *pCtg, char *dbFName, char *tbName, uint64_t dbId, uint64_t suid,
|
||||
SCtgTbCache *pCache) {
|
||||
SSTableVersion metaRent = {.dbId = dbId, .suid = suid};
|
||||
if (pCache->pMeta) {
|
||||
metaRent.sversion = pCache->pMeta->sversion;
|
||||
|
|
@ -1400,16 +1337,17 @@ int32_t ctgUpdateRentStbVersion(SCatalog *pCtg, char* dbFName, char* tbName, uin
|
|||
strcpy(metaRent.dbFName, dbFName);
|
||||
strcpy(metaRent.stbName, tbName);
|
||||
|
||||
CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->stbRent, &metaRent, metaRent.suid, sizeof(SSTableVersion), ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
|
||||
CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->stbRent, &metaRent, metaRent.suid, sizeof(SSTableVersion),
|
||||
ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
|
||||
|
||||
ctgDebug("db %s,0x%" PRIx64 " stb %s,0x%" PRIx64 " sver %d tver %d smaVer %d updated to stbRent",
|
||||
dbFName, dbId, tbName, suid, metaRent.sversion, metaRent.tversion, metaRent.smaVer);
|
||||
ctgDebug("db %s,0x%" PRIx64 " stb %s,0x%" PRIx64 " sver %d tver %d smaVer %d updated to stbRent", dbFName, dbId,
|
||||
tbName, suid, metaRent.sversion, metaRent.tversion, metaRent.smaVer);
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFName, uint64_t dbId, char *tbName, STableMeta *meta, int32_t metaSize) {
|
||||
int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFName, uint64_t dbId, char *tbName,
|
||||
STableMeta *meta, int32_t metaSize) {
|
||||
if (NULL == dbCache->tbCache || NULL == dbCache->stbCache) {
|
||||
taosMemoryFree(meta);
|
||||
ctgError("db is dropping, dbId:0x%" PRIx64, dbCache->dbId);
|
||||
|
|
@ -1425,7 +1363,8 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
|
|||
if (orig) {
|
||||
origType = orig->tableType;
|
||||
|
||||
if (origType == meta->tableType && orig->uid == meta->uid && (origType == TSDB_CHILD_TABLE || (orig->sversion >= meta->sversion && orig->tversion >= meta->tversion))) {
|
||||
if (origType == meta->tableType && orig->uid == meta->uid &&
|
||||
(origType == TSDB_CHILD_TABLE || (orig->sversion >= meta->sversion && orig->tversion >= meta->tversion))) {
|
||||
taosMemoryFree(meta);
|
||||
ctgDebug("ignore table %s meta update", tbName);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
|
@ -1469,14 +1408,16 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
if (origSuid != meta->suid && taosHashPut(dbCache->stbCache, &meta->suid, sizeof(meta->suid), tbName, strlen(tbName) + 1) != 0) {
|
||||
if (origSuid != meta->suid &&
|
||||
taosHashPut(dbCache->stbCache, &meta->suid, sizeof(meta->suid), tbName, strlen(tbName) + 1) != 0) {
|
||||
ctgError("taosHashPut to stable cache failed, suid:0x%" PRIx64, meta->suid);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_CACHE_STAT_INC(numOfStb, 1);
|
||||
|
||||
ctgDebug("stb 0x%" PRIx64 " updated to cache, dbFName:%s, tbName:%s, tbType:%d", meta->suid, dbFName, tbName, meta->tableType);
|
||||
ctgDebug("stb 0x%" PRIx64 " updated to cache, dbFName:%s, tbName:%s, tbType:%d", meta->suid, dbFName, tbName,
|
||||
meta->tableType);
|
||||
|
||||
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbId, meta->suid, pCache));
|
||||
|
||||
|
|
@ -1506,7 +1447,8 @@ int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char* dbFNa
|
|||
}
|
||||
|
||||
*index = NULL;
|
||||
ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version, (int32_t)taosArrayGetSize(pIndex->pIndex));
|
||||
ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version,
|
||||
(int32_t)taosArrayGetSize(pIndex->pIndex));
|
||||
|
||||
if (suid) {
|
||||
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, pIndex->suid, &cache));
|
||||
|
|
@ -1526,7 +1468,8 @@ int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char* dbFNa
|
|||
pCache->pIndex = pIndex;
|
||||
*index = NULL;
|
||||
|
||||
ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version, (int32_t)taosArrayGetSize(pIndex->pIndex));
|
||||
ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version,
|
||||
(int32_t)taosArrayGetSize(pIndex->pIndex));
|
||||
|
||||
if (suid) {
|
||||
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, suid, pCache));
|
||||
|
|
@ -1582,7 +1525,6 @@ void ctgFreeAllInstance(void) {
|
|||
taosHashClear(gCtgMgmt.pCluster);
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgOpUpdateVgroup(SCtgCacheOperation *operation) {
|
||||
int32_t code = 0;
|
||||
SCtgUpdateVgMsg *msg = operation->data;
|
||||
|
|
@ -1595,8 +1537,8 @@ int32_t ctgOpUpdateVgroup(SCtgCacheOperation *operation) {
|
|||
}
|
||||
|
||||
if (dbInfo->vgVersion < 0 || taosHashGetSize(dbInfo->vgHash) <= 0) {
|
||||
ctgError("invalid db vgInfo, dbFName:%s, vgHash:%p, vgVersion:%d, vgHashSize:%d",
|
||||
dbFName, dbInfo->vgHash, dbInfo->vgVersion, taosHashGetSize(dbInfo->vgHash));
|
||||
ctgError("invalid db vgInfo, dbFName:%s, vgHash:%p, vgVersion:%d, vgHashSize:%d", dbFName, dbInfo->vgHash,
|
||||
dbInfo->vgVersion, taosHashGetSize(dbInfo->vgHash));
|
||||
CTG_ERR_JRET(TSDB_CODE_APP_ERROR);
|
||||
}
|
||||
|
||||
|
|
@ -1624,7 +1566,8 @@ int32_t ctgOpUpdateVgroup(SCtgCacheOperation *operation) {
|
|||
}
|
||||
|
||||
if (dbInfo->vgVersion == vgInfo->vgVersion && dbInfo->numOfTable == vgInfo->numOfTable) {
|
||||
ctgDebug("no new db vgVer or numOfTable, dbFName:%s, vgVer:%d, numOfTable:%d", dbFName, dbInfo->vgVersion, dbInfo->numOfTable);
|
||||
ctgDebug("no new db vgVer or numOfTable, dbFName:%s, vgVer:%d, numOfTable:%d", dbFName, dbInfo->vgVersion,
|
||||
dbInfo->numOfTable);
|
||||
ctgWUnlockVgInfo(dbCache);
|
||||
|
||||
goto _return;
|
||||
|
|
@ -1643,7 +1586,8 @@ int32_t ctgOpUpdateVgroup(SCtgCacheOperation *operation) {
|
|||
dbCache = NULL;
|
||||
|
||||
strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
|
||||
CTG_ERR_RET(ctgMetaRentUpdate(&msg->pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion), ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
|
||||
CTG_ERR_JRET(ctgMetaRentUpdate(&msg->pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion),
|
||||
ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
|
||||
|
||||
_return:
|
||||
|
||||
|
|
@ -1665,7 +1609,8 @@ int32_t ctgOpDropDbCache(SCtgCacheOperation *operation) {
|
|||
}
|
||||
|
||||
if (dbCache->dbId != msg->dbId) {
|
||||
ctgInfo("dbId already updated, dbFName:%s, dbId:0x%"PRIx64 ", targetId:0x%"PRIx64, msg->dbFName, dbCache->dbId, msg->dbId);
|
||||
ctgInfo("dbId already updated, dbFName:%s, dbId:0x%" PRIx64 ", targetId:0x%" PRIx64, msg->dbFName, dbCache->dbId,
|
||||
msg->dbId);
|
||||
goto _return;
|
||||
}
|
||||
|
||||
|
|
@ -1689,7 +1634,7 @@ int32_t ctgOpDropDbVgroup(SCtgCacheOperation *operation) {
|
|||
goto _return;
|
||||
}
|
||||
|
||||
CTG_ERR_RET(ctgWLockVgInfo(pCtg, dbCache));
|
||||
CTG_ERR_JRET(ctgWLockVgInfo(pCtg, dbCache));
|
||||
|
||||
ctgFreeVgInfo(dbCache->vgCache.vgInfo);
|
||||
dbCache->vgCache.vgInfo = NULL;
|
||||
|
|
@ -1705,7 +1650,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) {
|
||||
int32_t code = 0;
|
||||
SCtgUpdateTbMetaMsg *msg = operation->data;
|
||||
|
|
@ -1731,7 +1675,8 @@ int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) {
|
|||
|
||||
if (CTG_IS_META_TABLE(pMeta->metaType) || CTG_IS_META_BOTH(pMeta->metaType)) {
|
||||
int32_t metaSize = CTG_META_SIZE(pMeta->tbMeta);
|
||||
CTG_ERR_JRET(ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->tbName, pMeta->tbMeta, metaSize));
|
||||
CTG_ERR_JRET(
|
||||
ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->tbName, pMeta->tbMeta, metaSize));
|
||||
pMeta->tbMeta = NULL;
|
||||
}
|
||||
|
||||
|
|
@ -1741,7 +1686,8 @@ int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) {
|
|||
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
memcpy(ctbMeta, &pMeta->ctbMeta, sizeof(SCTableMeta));
|
||||
CTG_ERR_JRET(ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->ctbName, (STableMeta *)ctbMeta, sizeof(SCTableMeta)));
|
||||
CTG_ERR_JRET(ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->ctbName,
|
||||
(STableMeta *)ctbMeta, sizeof(SCTableMeta)));
|
||||
}
|
||||
|
||||
_return:
|
||||
|
|
@ -1756,7 +1702,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgOpDropStbMeta(SCtgCacheOperation *operation) {
|
||||
int32_t code = 0;
|
||||
SCtgDropStbMetaMsg *msg = operation->data;
|
||||
|
|
@ -1765,17 +1710,18 @@ int32_t ctgOpDropStbMeta(SCtgCacheOperation *operation) {
|
|||
SCtgDBCache *dbCache = NULL;
|
||||
ctgGetDBCache(pCtg, msg->dbFName, &dbCache);
|
||||
if (NULL == dbCache) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
goto _return;
|
||||
}
|
||||
|
||||
if (msg->dbId && (dbCache->dbId != msg->dbId)) {
|
||||
ctgDebug("dbId already modified, dbFName:%s, current:0x%" PRIx64 ", dbId:0x%" PRIx64 ", stb:%s, suid:0x%" PRIx64,
|
||||
msg->dbFName, dbCache->dbId, msg->dbId, msg->stbName, msg->suid);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
goto _return;
|
||||
}
|
||||
|
||||
if (taosHashRemove(dbCache->stbCache, &msg->suid, sizeof(msg->suid))) {
|
||||
ctgDebug("stb not exist in stbCache, may be removed, dbFName:%s, stb:%s, suid:0x%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
|
||||
ctgDebug("stb not exist in stbCache, may be removed, dbFName:%s, stb:%s, suid:0x%" PRIx64, msg->dbFName,
|
||||
msg->stbName, msg->suid);
|
||||
} else {
|
||||
CTG_CACHE_STAT_DEC(numOfStb, 1);
|
||||
}
|
||||
|
|
@ -1821,7 +1767,8 @@ int32_t ctgOpDropTbMeta(SCtgCacheOperation *operation) {
|
|||
}
|
||||
|
||||
if (dbCache->dbId != msg->dbId) {
|
||||
ctgDebug("dbId 0x%" PRIx64 " not match with curId 0x%"PRIx64", dbFName:%s, tbName:%s", msg->dbId, dbCache->dbId, msg->dbFName, msg->tbName);
|
||||
ctgDebug("dbId 0x%" PRIx64 " not match with curId 0x%" PRIx64 ", dbFName:%s, tbName:%s", msg->dbId, dbCache->dbId,
|
||||
msg->dbFName, msg->tbName);
|
||||
goto _return;
|
||||
}
|
||||
|
||||
|
|
@ -1933,9 +1880,9 @@ int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation) {
|
|||
|
||||
SEp *pOrigEp = &pInfo->epSet.eps[pInfo->epSet.inUse];
|
||||
SEp *pNewEp = &msg->epSet.eps[msg->epSet.inUse];
|
||||
ctgDebug("vgroup %d epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d, dbFName:%s in ctg",
|
||||
pInfo->vgId, pInfo->epSet.inUse, pInfo->epSet.numOfEps, pOrigEp->fqdn, pOrigEp->port,
|
||||
msg->epSet.inUse, msg->epSet.numOfEps, pNewEp->fqdn, pNewEp->port, msg->dbFName);
|
||||
ctgDebug("vgroup %d epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d, dbFName:%s in ctg", pInfo->vgId,
|
||||
pInfo->epSet.inUse, pInfo->epSet.numOfEps, pOrigEp->fqdn, pOrigEp->port, msg->epSet.inUse,
|
||||
msg->epSet.numOfEps, pNewEp->fqdn, pNewEp->port, msg->dbFName);
|
||||
|
||||
pInfo->epSet = msg->epSet;
|
||||
|
||||
|
|
@ -2006,7 +1953,6 @@ _return:
|
|||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgOpClearCache(SCtgCacheOperation *operation) {
|
||||
int32_t code = 0;
|
||||
SCtgClearCacheMsg *msg = operation->data;
|
||||
|
|
@ -2125,7 +2071,6 @@ void* ctgUpdateThreadFunc(void* param) {
|
|||
return NULL;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgStartUpdateThread() {
|
||||
TdThreadAttr thAttr;
|
||||
taosThreadAttrInit(&thAttr);
|
||||
|
|
@ -2140,7 +2085,6 @@ int32_t ctgStartUpdateThread() {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgGetTbMetaFromCache(SCatalog *pCtg, SRequestConnInfo *pConn, SCtgTbMetaCtx *ctx, STableMeta **pTableMeta) {
|
||||
if (IS_SYS_DBNAME(ctx->pName->dbname)) {
|
||||
CTG_FLAG_SET_SYS_DB(ctx->flag);
|
||||
|
|
@ -2221,7 +2165,8 @@ int32_t ctgGetTbMetaBFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMe
|
|||
}
|
||||
#endif
|
||||
|
||||
int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetasCtx* ctx, int32_t dbIdx, int32_t *fetchIdx, int32_t baseResIdx, SArray* pList) {
|
||||
int32_t ctgGetTbMetasFromCache(SCatalog *pCtg, SRequestConnInfo *pConn, SCtgTbMetasCtx *ctx, int32_t dbIdx,
|
||||
int32_t *fetchIdx, int32_t baseResIdx, SArray *pList) {
|
||||
int32_t tbNum = taosArrayGetSize(pList);
|
||||
SName *pName = taosArrayGet(pList, 0);
|
||||
char dbFName[TSDB_DB_FNAME_LEN] = {0};
|
||||
|
|
@ -2332,8 +2277,8 @@ int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMe
|
|||
CTG_UNLOCK(CTG_READ, &pCache->metaLock);
|
||||
taosHashRelease(dbCache->tbCache, pCache);
|
||||
|
||||
ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s",
|
||||
pName->tname, nctx.tbInfo.tbType, dbFName);
|
||||
ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s", pName->tname,
|
||||
nctx.tbInfo.tbType, dbFName);
|
||||
|
||||
char *stName = taosHashAcquire(dbCache->stbCache, &pTableMeta->suid, sizeof(pTableMeta->suid));
|
||||
if (NULL == stName) {
|
||||
|
|
@ -2378,7 +2323,8 @@ int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMe
|
|||
CTG_UNLOCK(CTG_READ, &pCache->metaLock);
|
||||
taosHashRelease(dbCache->tbCache, pCache);
|
||||
|
||||
ctgError("stb suid 0x%" PRIx64 " in stbCache mis-match, expected suid 0x%"PRIx64 , stbMeta->suid, nctx.tbInfo.suid);
|
||||
ctgError("stb suid 0x%" PRIx64 " in stbCache mis-match, expected suid 0x%" PRIx64, stbMeta->suid,
|
||||
nctx.tbInfo.suid);
|
||||
|
||||
ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
|
||||
taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
|
||||
|
|
@ -2412,7 +2358,6 @@ int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMe
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
int32_t ctgRemoveTbMetaFromCache(SCatalog *pCtg, SName *pTableName, bool syncReq) {
|
||||
int32_t code = 0;
|
||||
STableMeta *tblMeta = NULL;
|
||||
|
|
@ -2476,5 +2421,3 @@ _return:
|
|||
|
||||
CTG_RET(code);
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue