提交 31f3bed3 编写于 作者: D dapan1121

support index cache

上级 041002a5
...@@ -2513,6 +2513,7 @@ typedef struct { ...@@ -2513,6 +2513,7 @@ typedef struct {
} STableIndexInfo; } STableIndexInfo;
typedef struct { typedef struct {
int32_t version;
SArray* pIndex; SArray* pIndex;
} STableIndexRsp; } STableIndexRsp;
......
...@@ -2437,6 +2437,7 @@ int32_t tSerializeSTableIndexRsp(void *buf, int32_t bufLen, const STableIndexRsp ...@@ -2437,6 +2437,7 @@ int32_t tSerializeSTableIndexRsp(void *buf, int32_t bufLen, const STableIndexRsp
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1; if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->version) < 0) return -1;
int32_t num = taosArrayGetSize(pRsp->pIndex); int32_t num = taosArrayGetSize(pRsp->pIndex);
if (tEncodeI32(&encoder, num) < 0) return -1; if (tEncodeI32(&encoder, num) < 0) return -1;
if (num > 0) { if (num > 0) {
...@@ -2471,6 +2472,7 @@ int32_t tDeserializeSTableIndexRsp(void *buf, int32_t bufLen, STableIndexRsp *pR ...@@ -2471,6 +2472,7 @@ int32_t tDeserializeSTableIndexRsp(void *buf, int32_t bufLen, STableIndexRsp *pR
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1; if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->version) < 0) return -1;
int32_t num = 0; int32_t num = 0;
if (tDecodeI32(&decoder, &num) < 0) return -1; if (tDecodeI32(&decoder, &num) < 0) return -1;
if (num > 0) { if (num > 0) {
......
...@@ -57,6 +57,8 @@ enum { ...@@ -57,6 +57,8 @@ enum {
CTG_OP_DROP_TB_META, CTG_OP_DROP_TB_META,
CTG_OP_UPDATE_USER, CTG_OP_UPDATE_USER,
CTG_OP_UPDATE_VG_EPSET, CTG_OP_UPDATE_VG_EPSET,
CTG_OP_UPDATE_TB_INDEX,
CTG_OP_DROP_TB_INDEX,
CTG_OP_MAX CTG_OP_MAX
}; };
...@@ -128,19 +130,27 @@ typedef struct SCtgUserCtx { ...@@ -128,19 +130,27 @@ typedef struct SCtgUserCtx {
SUserAuthInfo user; SUserAuthInfo user;
} SCtgUserCtx; } SCtgUserCtx;
typedef struct SCtgTbMetaCache { typedef STableIndexRsp STableIndex;
SRWLatch stbLock;
SRWLatch metaLock; // RC between cache destroy and all other operations
SHashObj *metaCache; //key:tbname, value:STableMeta
SHashObj *stbCache; //key:suid, value:STableMeta*
} SCtgTbMetaCache;
typedef struct SCtgDBCache { typedef struct SCtgTbCache {
SRWLatch metaLock;
STableMeta *pMeta;
SRWLatch indexLock;
STableIndex *pIndex;
} SCtgTbCache;
typedef struct SCtgVgCache {
SRWLatch vgLock; SRWLatch vgLock;
SDBVgInfo *vgInfo;
} SCtgVgCache;
typedef struct SCtgDBCache {
SRWLatch dbLock; // RC between destroy tbCache/stbCache and all reads
uint64_t dbId; uint64_t dbId;
int8_t deleted; int8_t deleted;
SDBVgInfo *vgInfo; SCtgVgCache vgCache;
SCtgTbMetaCache tbCache; SHashObj *tbCache; // key:tbname, value:SCtgTbCache
SHashObj *stbCache; // key:suid, value:STableMeta*
} SCtgDBCache; } SCtgDBCache;
typedef struct SCtgRentSlot { typedef struct SCtgRentSlot {
...@@ -245,8 +255,10 @@ typedef struct SCtgCacheStat { ...@@ -245,8 +255,10 @@ typedef struct SCtgCacheStat {
uint64_t userNum; uint64_t userNum;
uint64_t vgHitNum; uint64_t vgHitNum;
uint64_t vgMissNum; uint64_t vgMissNum;
uint64_t tblHitNum; uint64_t tbMetaHitNum;
uint64_t tblMissNum; uint64_t tbMetaMissNum;
uint64_t tbIndexHitNum;
uint64_t tbIndexMissNum;
uint64_t userHitNum; uint64_t userHitNum;
uint64_t userMissNum; uint64_t userMissNum;
} SCtgCacheStat; } SCtgCacheStat;
...@@ -268,10 +280,10 @@ typedef struct SCtgUpdateVgMsg { ...@@ -268,10 +280,10 @@ typedef struct SCtgUpdateVgMsg {
SDBVgInfo* dbInfo; SDBVgInfo* dbInfo;
} SCtgUpdateVgMsg; } SCtgUpdateVgMsg;
typedef struct SCtgUpdateTblMsg { typedef struct SCtgUpdateTbMetaMsg {
SCatalog* pCtg; SCatalog* pCtg;
STableMetaOutput* output; STableMetaOutput* pMeta;
} SCtgUpdateTblMsg; } SCtgUpdateTbMetaMsg;
typedef struct SCtgDropDBMsg { typedef struct SCtgDropDBMsg {
SCatalog* pCtg; SCatalog* pCtg;
...@@ -305,6 +317,19 @@ typedef struct SCtgUpdateUserMsg { ...@@ -305,6 +317,19 @@ typedef struct SCtgUpdateUserMsg {
SGetUserAuthRsp userAuth; SGetUserAuthRsp userAuth;
} SCtgUpdateUserMsg; } SCtgUpdateUserMsg;
typedef struct SCtgUpdateTbIndexMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
char tbName[TSDB_TABLE_NAME_LEN];
STableIndex* pIndex;
} SCtgUpdateTbIndexMsg;
typedef struct SCtgDropTbIndexMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
char tbName[TSDB_TABLE_NAME_LEN];
} SCtgDropTbIndexMsg;
typedef struct SCtgUpdateEpsetMsg { typedef struct SCtgUpdateEpsetMsg {
SCatalog* pCtg; SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN]; char dbFName[TSDB_DB_FNAME_LEN];
...@@ -465,8 +490,7 @@ int32_t ctgOpUpdateUser(SCtgCacheOperation *action); ...@@ -465,8 +490,7 @@ int32_t ctgOpUpdateUser(SCtgCacheOperation *action);
int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation); int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation);
int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache); int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache); void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache);
void ctgReleaseVgInfo(SCtgDBCache *dbCache); void ctgRUnlockVgInfo(SCtgDBCache *dbCache);
int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist); int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist);
int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta); int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName); int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName);
...@@ -479,6 +503,7 @@ int32_t ctgUpdateVgroupEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId ...@@ -479,6 +503,7 @@ int32_t ctgUpdateVgroupEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId
int32_t ctgUpdateTbMetaEnqueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq); int32_t ctgUpdateTbMetaEnqueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq);
int32_t ctgUpdateUserEnqueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq); int32_t ctgUpdateUserEnqueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq);
int32_t ctgUpdateVgEpsetEnqueue(SCatalog* pCtg, char *dbFName, int32_t vgId, SEpSet* pEpSet); int32_t ctgUpdateVgEpsetEnqueue(SCatalog* pCtg, char *dbFName, int32_t vgId, SEpSet* pEpSet);
int32_t ctgUpdateTbIndexEnqueue(SCatalog* pCtg, SName* pName, STableIndex *pIndex, bool syncOp);
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type); int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type);
int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size); int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size);
int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size); int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size);
...@@ -521,6 +546,7 @@ void ctgFreeSTableMetaOutput(STableMetaOutput* pOutput); ...@@ -521,6 +546,7 @@ void ctgFreeSTableMetaOutput(STableMetaOutput* pOutput);
int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target); int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target);
char *ctgTaskTypeStr(CTG_TASK_TYPE type); char *ctgTaskTypeStr(CTG_TASK_TYPE type);
int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask); int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask);
int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes);
extern SCatalogMgmt gCtgMgmt; extern SCatalogMgmt gCtgMgmt;
......
...@@ -97,8 +97,7 @@ int32_t ctgRefreshDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* ...@@ -97,8 +97,7 @@ int32_t ctgRefreshDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char*
if (NULL != dbCache) { if (NULL != dbCache) {
input.dbId = dbCache->dbId; input.dbId = dbCache->dbId;
ctgReleaseVgInfo(dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
ctgReleaseDBCache(pCtg, dbCache);
} }
input.vgVersion = CTG_DEFAULT_INVALID_VERSION; input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
...@@ -383,6 +382,18 @@ _return: ...@@ -383,6 +382,18 @@ _return:
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgGetTbIndex(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SArray** pRes) {
CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pTableName, pRes));
if (*pRes) {
return TSDB_CODE_SUCCESS;
}
CTG_ERR_RET(ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pRes, NULL));
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, SArray** pVgList) { int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, SArray** pVgList) {
STableMeta *tbMeta = NULL; STableMeta *tbMeta = NULL;
int32_t code = 0; int32_t code = 0;
...@@ -443,7 +454,7 @@ int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTabl ...@@ -443,7 +454,7 @@ int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTabl
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgRUnlockVgInfo(dbCache);
ctgReleaseDBCache(pCtg, dbCache); ctgReleaseDBCache(pCtg, dbCache);
} }
...@@ -633,8 +644,7 @@ int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* vers ...@@ -633,8 +644,7 @@ int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* vers
*dbId = dbCache->dbId; *dbId = dbCache->dbId;
*tableNum = dbCache->vgInfo->numOfTable; *tableNum = dbCache->vgInfo->numOfTable;
ctgReleaseVgInfo(dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
ctgReleaseDBCache(pCtg, dbCache);
ctgDebug("Got db vgVersion from cache, dbFName:%s, vgVersion:%d", dbFName, *version); ctgDebug("Got db vgVersion from cache, dbFName:%s, vgVersion:%d", dbFName, *version);
...@@ -659,7 +669,7 @@ int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* ...@@ -659,7 +669,7 @@ int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char*
SDBVgInfo *vgInfo = NULL; SDBVgInfo *vgInfo = NULL;
CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, dbFName, &dbCache, &vgInfo)); CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, dbFName, &dbCache, &vgInfo));
if (dbCache) { if (dbCache) {
vgHash = dbCache->vgInfo->vgHash; vgHash = dbCache->vgCache.vgInfo->vgHash;
} else { } else {
vgHash = vgInfo->vgHash; vgHash = vgInfo->vgHash;
} }
...@@ -672,7 +682,7 @@ int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* ...@@ -672,7 +682,7 @@ int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char*
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgRUnlockVgInfo(dbCache);
ctgReleaseDBCache(pCtg, dbCache); ctgReleaseDBCache(pCtg, dbCache);
} }
...@@ -940,7 +950,7 @@ int32_t catalogGetTableHashVgroup(SCatalog *pCtg, SRequestConnInfo *pConn, const ...@@ -940,7 +950,7 @@ int32_t catalogGetTableHashVgroup(SCatalog *pCtg, SRequestConnInfo *pConn, const
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgRUnlockVgInfo(dbCache);
ctgReleaseDBCache(pCtg, dbCache); ctgReleaseDBCache(pCtg, dbCache);
} }
...@@ -1143,7 +1153,17 @@ int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo *pConn, const SNam ...@@ -1143,7 +1153,17 @@ int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo *pConn, const SNam
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT); CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
} }
CTG_API_LEAVE(ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pRes, NULL)); int32_t code = 0;
CTG_ERR_JRET(ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pRes, NULL));
SArray* pInfo = NULL;
CTG_ERR_JRET(ctgCloneTableIndex(*pRes, &pInfo));
CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pCtg, pTableName, pInfo, false));
_return:
CTG_API_LEAVE(code);
} }
......
...@@ -344,6 +344,11 @@ int32_t ctgHandleForceUpdate(SCatalog* pCtg, SCtgJob *pJob, const SCatalogReq* p ...@@ -344,6 +344,11 @@ int32_t ctgHandleForceUpdate(SCatalog* pCtg, SCtgJob *pJob, const SCatalogReq* p
} }
} }
for (int32_t i = 0; i < pJob->tbIndexNum; ++i) {
SName* name = taosArrayGet(pReq->pTableIndex, i);
ctgDropTbIndexEnqueue(pCtg, name, true);
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -524,7 +529,7 @@ int32_t ctgDumpTbIndexRes(SCtgTask* pTask) { ...@@ -524,7 +529,7 @@ int32_t ctgDumpTbIndexRes(SCtgTask* pTask) {
} }
SMetaRes res = {.code = pTask->code, .pRes = pTask->res}; SMetaRes res = {.code = pTask->code, .pRes = pTask->res};
taosArrayPush(pJob->jobRes.pTableHash, &res); taosArrayPush(pJob->jobRes.pTableIndex, &res);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -687,8 +692,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf * ...@@ -687,8 +692,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
ctx->vgId = vgInfo.vgId; ctx->vgId = vgInfo.vgId;
CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, ctx->pName, &vgInfo, NULL, pTask)); CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, ctx->pName, &vgInfo, NULL, pTask));
ctgReleaseVgInfo(dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
ctgReleaseDBCache(pCtg, dbCache);
} else { } else {
SBuildUseDBInput input = {0}; SBuildUseDBInput input = {0};
...@@ -786,7 +790,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf * ...@@ -786,7 +790,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgRUnlockVgInfo(dbCache);
ctgReleaseDBCache(pCtg, dbCache); ctgReleaseDBCache(pCtg, dbCache);
} }
...@@ -866,9 +870,16 @@ _return: ...@@ -866,9 +870,16 @@ _return:
int32_t ctgHandleGetTbIndexRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pMsg, int32_t rspCode) { int32_t ctgHandleGetTbIndexRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pMsg, int32_t rspCode) {
int32_t code = 0; int32_t code = 0;
CTG_ERR_JRET(ctgProcessRspMsg(pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target)); CTG_ERR_JRET(ctgProcessRspMsg(&pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target));
TSWAP(pTask->res, pTask->msgCtx.out); STableIndex* pOut = (STableIndex*)pTask->msgCtx.out;
SArray* pInfo = NULL;
CTG_ERR_JRET(ctgCloneTableIndex(pOut->pIndex, &pInfo));
pTask->res = pInfo;
pTask->msgCtx.out = NULL;
SCtgTbIndexCtx* ctx = pTask->taskCtx;
CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pTask->pJob->pCtg, ctx->pName, pOut, false));
_return: _return:
...@@ -1024,8 +1035,7 @@ int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) { ...@@ -1024,8 +1035,7 @@ int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) {
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
ctgReleaseDBCache(pCtg, dbCache);
} }
CTG_RET(code); CTG_RET(code);
...@@ -1070,8 +1080,7 @@ int32_t ctgLaunchGetDbVgTask(SCtgTask *pTask) { ...@@ -1070,8 +1080,7 @@ int32_t ctgLaunchGetDbVgTask(SCtgTask *pTask) {
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
ctgReleaseDBCache(pCtg, dbCache);
} }
CTG_RET(code); CTG_RET(code);
...@@ -1105,8 +1114,7 @@ int32_t ctgLaunchGetTbHashTask(SCtgTask *pTask) { ...@@ -1105,8 +1114,7 @@ int32_t ctgLaunchGetTbHashTask(SCtgTask *pTask) {
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
ctgReleaseDBCache(pCtg, dbCache);
} }
CTG_RET(code); CTG_RET(code);
...@@ -1117,6 +1125,15 @@ int32_t ctgLaunchGetTbIndexTask(SCtgTask *pTask) { ...@@ -1117,6 +1125,15 @@ int32_t ctgLaunchGetTbIndexTask(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
SRequestConnInfo* pConn = &pTask->pJob->conn; SRequestConnInfo* pConn = &pTask->pJob->conn;
SCtgTbIndexCtx* pCtx = (SCtgTbIndexCtx*)pTask->taskCtx; SCtgTbIndexCtx* pCtx = (SCtgTbIndexCtx*)pTask->taskCtx;
SArray* pRes = NULL;
CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pCtx->pName, &pRes));
if (pRes) {
pTask->res = pRes;
CTG_ERR_RET(ctgHandleTaskEnd(pTask, 0));
return TSDB_CODE_SUCCESS;
}
CTG_ERR_RET(ctgGetTbIndexFromMnode(pCtg, pConn, pCtx->pName, NULL, pTask)); CTG_ERR_RET(ctgGetTbIndexFromMnode(pCtg, pConn, pCtx->pName, NULL, pTask));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
...@@ -1167,8 +1184,7 @@ int32_t ctgLaunchGetDbInfoTask(SCtgTask *pTask) { ...@@ -1167,8 +1184,7 @@ int32_t ctgLaunchGetDbInfoTask(SCtgTask *pTask) {
_return: _return:
if (dbCache) { if (dbCache) {
ctgReleaseVgInfo(dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
ctgReleaseDBCache(pCtg, dbCache);
} }
CTG_RET(code); CTG_RET(code);
......
此差异已折叠。
...@@ -266,11 +266,11 @@ int32_t ctgdGetStatNum(char *option, void *res) { ...@@ -266,11 +266,11 @@ int32_t ctgdGetStatNum(char *option, void *res) {
} }
int32_t ctgdGetTbMetaNum(SCtgDBCache *dbCache) { int32_t ctgdGetTbMetaNum(SCtgDBCache *dbCache) {
return dbCache->tbCache.metaCache ? (int32_t)taosHashGetSize(dbCache->tbCache.metaCache) : 0; return dbCache->table.tbCache ? (int32_t)taosHashGetSize(dbCache->table.tbCache) : 0;
} }
int32_t ctgdGetStbNum(SCtgDBCache *dbCache) { int32_t ctgdGetStbNum(SCtgDBCache *dbCache) {
return dbCache->tbCache.stbCache ? (int32_t)taosHashGetSize(dbCache->tbCache.stbCache) : 0; return dbCache->table.stbCache ? (int32_t)taosHashGetSize(dbCache->table.stbCache) : 0;
} }
int32_t ctgdGetRentNum(SCtgRentMgmt *rent) { int32_t ctgdGetRentNum(SCtgRentMgmt *rent) {
...@@ -363,8 +363,8 @@ void ctgdShowDBCache(SCatalog* pCtg, SHashObj *dbHash) { ...@@ -363,8 +363,8 @@ void ctgdShowDBCache(SCatalog* pCtg, SHashObj *dbHash) {
dbFName = taosHashGetKey(pIter, &len); dbFName = taosHashGetKey(pIter, &len);
int32_t metaNum = dbCache->tbCache.metaCache ? taosHashGetSize(dbCache->tbCache.metaCache) : 0; int32_t metaNum = dbCache->table.tbCache ? taosHashGetSize(dbCache->table.tbCache) : 0;
int32_t stbNum = dbCache->tbCache.stbCache ? taosHashGetSize(dbCache->tbCache.stbCache) : 0; int32_t stbNum = dbCache->table.stbCache ? taosHashGetSize(dbCache->table.stbCache) : 0;
int32_t vgVersion = CTG_DEFAULT_INVALID_VERSION; int32_t vgVersion = CTG_DEFAULT_INVALID_VERSION;
int32_t hashMethod = -1; int32_t hashMethod = -1;
int32_t vgNum = 0; int32_t vgNum = 0;
......
...@@ -448,10 +448,7 @@ int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *n ...@@ -448,10 +448,7 @@ int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *n
} }
if (pTask) { if (pTask) {
void* pOut = taosMemoryCalloc(1, POINTER_BYTES); void* pOut = NULL;
if (NULL == pOut) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, (char*)tbFName)); CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, (char*)tbFName));
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask, reqType, msg, msgLen)); CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask, reqType, msg, msgLen));
......
...@@ -110,25 +110,39 @@ void ctgFreeMetaRent(SCtgRentMgmt *mgmt) { ...@@ -110,25 +110,39 @@ void ctgFreeMetaRent(SCtgRentMgmt *mgmt) {
taosMemoryFreeClear(mgmt->slots); taosMemoryFreeClear(mgmt->slots);
} }
void ctgFreeStbMetaCache(SCtgDBCache *dbCache) {
if (NULL == dbCache->stbCache) {
return;
}
int32_t stblNum = taosHashGetSize(dbCache->stbCache);
taosHashCleanup(dbCache->stbCache);
dbCache->stbCache = NULL;
CTG_CACHE_STAT_DEC(stblNum, stblNum);
}
void ctgFreeTbCacheImpl(SCtgTbCache *pCache) {
taosMemoryFreeClear(pCache->pMeta);
if (pCache->pIndex) {
taosArrayDestroyEx(pCache->pIndex->pIndex, tFreeSTableIndexInfo);
taosMemoryFreeClear(pCache->pIndex);
}
}
void ctgFreeTbMetaCache(SCtgTbMetaCache *cache) { void ctgFreeTbCache(SCtgDBCache *dbCache) {
CTG_LOCK(CTG_WRITE, &cache->stbLock); if (NULL == dbCache->tbCache) {
if (cache->stbCache) { return;
int32_t stblNum = taosHashGetSize(cache->stbCache);
taosHashCleanup(cache->stbCache);
cache->stbCache = NULL;
CTG_CACHE_STAT_DEC(stblNum, stblNum);
} }
CTG_UNLOCK(CTG_WRITE, &cache->stbLock);
CTG_LOCK(CTG_WRITE, &cache->metaLock); int32_t tblNum = taosHashGetSize(dbCache->tbCache);
if (cache->metaCache) { SCtgTbCache *pCache = taosHashIterate(dbCache->tbCache, NULL);
int32_t tblNum = taosHashGetSize(cache->metaCache); while (NULL != pCache) {
taosHashCleanup(cache->metaCache); ctgFreeTbCacheImpl(pCache);
cache->metaCache = NULL; pCache = taosHashIterate(dbCache->tbCache, pCache);
CTG_CACHE_STAT_DEC(tblNum, tblNum);
} }
CTG_UNLOCK(CTG_WRITE, &cache->metaLock); taosHashCleanup(dbCache->tbCache);
dbCache->tbCache = NULL;
CTG_CACHE_STAT_DEC(tblNum, tblNum);
} }
void ctgFreeVgInfo(SDBVgInfo *vgInfo) { void ctgFreeVgInfo(SDBVgInfo *vgInfo) {
...@@ -144,16 +158,18 @@ void ctgFreeVgInfo(SDBVgInfo *vgInfo) { ...@@ -144,16 +158,18 @@ void ctgFreeVgInfo(SDBVgInfo *vgInfo) {
taosMemoryFreeClear(vgInfo); taosMemoryFreeClear(vgInfo);
} }
void ctgFreeVgInfoCache(SCtgDBCache *dbCache) {
ctgFreeVgInfo(dbCache->vgCache.vgInfo);
}
void ctgFreeDbCache(SCtgDBCache *dbCache) { void ctgFreeDbCache(SCtgDBCache *dbCache) {
if (NULL == dbCache) { if (NULL == dbCache) {
return; return;
} }
CTG_LOCK(CTG_WRITE, &dbCache->vgLock); ctgFreeVgInfoCache(dbCache);
ctgFreeVgInfo (dbCache->vgInfo); ctgFreeStbMetaCache(dbCache);
CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock); ctgFreeTbCache(dbCache);
ctgFreeTbMetaCache(&dbCache->tbCache);
} }
...@@ -167,16 +183,13 @@ void ctgFreeHandle(SCatalog* pCtg) { ...@@ -167,16 +183,13 @@ void ctgFreeHandle(SCatalog* pCtg) {
void *pIter = taosHashIterate(pCtg->dbCache, NULL); void *pIter = taosHashIterate(pCtg->dbCache, NULL);
while (pIter) { while (pIter) {
SCtgDBCache *dbCache = pIter; SCtgDBCache *dbCache = pIter;
atomic_store_8(&dbCache->deleted, 1); atomic_store_8(&dbCache->deleted, 1);
ctgFreeDbCache(dbCache); ctgFreeDbCache(dbCache);
pIter = taosHashIterate(pCtg->dbCache, pIter); pIter = taosHashIterate(pCtg->dbCache, pIter);
} }
taosHashCleanup(pCtg->dbCache); taosHashCleanup(pCtg->dbCache);
CTG_CACHE_STAT_DEC(dbNum, dbNum); CTG_CACHE_STAT_DEC(dbNum, dbNum);
} }
...@@ -186,14 +199,12 @@ void ctgFreeHandle(SCatalog* pCtg) { ...@@ -186,14 +199,12 @@ void ctgFreeHandle(SCatalog* pCtg) {
void *pIter = taosHashIterate(pCtg->userCache, NULL); void *pIter = taosHashIterate(pCtg->userCache, NULL);
while (pIter) { while (pIter) {
SCtgUserAuth *userCache = pIter; SCtgUserAuth *userCache = pIter;
ctgFreeSCtgUserAuth(userCache); ctgFreeSCtgUserAuth(userCache);
pIter = taosHashIterate(pCtg->userCache, pIter); pIter = taosHashIterate(pCtg->userCache, pIter);
} }
taosHashCleanup(pCtg->userCache); taosHashCleanup(pCtg->userCache);
CTG_CACHE_STAT_DEC(userNum, userNum); CTG_CACHE_STAT_DEC(userNum, userNum);
} }
...@@ -252,9 +263,9 @@ void ctgFreeMsgCtx(SCtgMsgCtx* pCtx) { ...@@ -252,9 +263,9 @@ void ctgFreeMsgCtx(SCtgMsgCtx* pCtx) {
break; break;
} }
case TDMT_MND_GET_TABLE_INDEX: { case TDMT_MND_GET_TABLE_INDEX: {
SArray** pOut = (SArray**)pCtx->out; STableIndex* pOut = (STableIndex*)pCtx->out;
if (pOut) { if (pOut) {
taosArrayDestroyEx(*pOut, tFreeSTableIndexInfo); taosArrayDestroyEx(pOut->pIndex, tFreeSTableIndexInfo);
taosMemoryFreeClear(pCtx->out); taosMemoryFreeClear(pCtx->out);
} }
break; break;
...@@ -640,6 +651,27 @@ int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput) ...@@ -640,6 +651,27 @@ int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput)
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes) {
if (NULL == pIndex) {
*pRes = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t num = taosArrayGetSize(pIndex);
*pRes = taosArrayInit(num, sizeof(STableIndexInfo));
if (NULL == *pRes) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
for (int32_t i = 0; i < num; ++i) {
STableIndexInfo *pInfo = taosArrayGet(pIndex, i);
taosArrayPush(*pRes, pInfo);
}
return TSDB_CODE_SUCCESS;
}
int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask) { int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask) {
if (msgType == TDMT_VND_TABLE_META) { if (msgType == TDMT_VND_TABLE_META) {
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx; SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
......
...@@ -895,9 +895,9 @@ void *ctgTestSetCtableMetaThread(void *param) { ...@@ -895,9 +895,9 @@ void *ctgTestSetCtableMetaThread(void *param) {
output = (STableMetaOutput *)taosMemoryMalloc(sizeof(STableMetaOutput)); output = (STableMetaOutput *)taosMemoryMalloc(sizeof(STableMetaOutput));
ctgTestBuildCTableMetaOutput(output); ctgTestBuildCTableMetaOutput(output);
SCtgUpdateTblMsg *msg = (SCtgUpdateTblMsg *)taosMemoryMalloc(sizeof(SCtgUpdateTblMsg)); SCtgUpdateTbMetaMsg *msg = (SCtgUpdateTbMetaMsg *)taosMemoryMalloc(sizeof(SCtgUpdateTbMetaMsg));
msg->pCtg = pCtg; msg->pCtg = pCtg;
msg->output = output; msg->pMeta = output;
operation.data = msg; operation.data = msg;
code = ctgOpUpdateTbMeta(&operation); code = ctgOpUpdateTbMeta(&operation);
......
...@@ -484,13 +484,17 @@ int32_t queryProcessGetTbIndexRsp(void *output, char *msg, int32_t msgSize) { ...@@ -484,13 +484,17 @@ int32_t queryProcessGetTbIndexRsp(void *output, char *msg, int32_t msgSize) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
} }
STableIndexRsp out = {0}; STableIndexRsp *out = taosMemoryCalloc(1, sizeof(STableIndexRsp));
if (tDeserializeSTableIndexRsp(msg, msgSize, &out) != 0) { if (NULL == out) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (tDeserializeSTableIndexRsp(msg, msgSize, out) != 0) {
qError("tDeserializeSTableIndexRsp failed, msgSize:%d", msgSize); qError("tDeserializeSTableIndexRsp failed, msgSize:%d", msgSize);
return TSDB_CODE_INVALID_MSG; return TSDB_CODE_INVALID_MSG;
} }
*(void **)output = out.pIndex; *(void **)output = out;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
......
...@@ -837,7 +837,6 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { ...@@ -837,7 +837,6 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
SQWorker *mgmt = qwAcquire(refId); SQWorker *mgmt = qwAcquire(refId);
if (NULL == mgmt) { if (NULL == mgmt) {
QW_DLOG("qwAcquire %" PRIx64 "failed", refId); QW_DLOG("qwAcquire %" PRIx64 "failed", refId);
taosMemoryFree(param);
return; return;
} }
......
...@@ -1476,6 +1476,11 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options) ...@@ -1476,6 +1476,11 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options)
for (uint32_t i = 0; i < info->fields[FLD_TYPE_VALUE].num; ++i) { for (uint32_t i = 0; i < info->fields[FLD_TYPE_VALUE].num; ++i) {
SFilterField *field = &info->fields[FLD_TYPE_VALUE].fields[i]; SFilterField *field = &info->fields[FLD_TYPE_VALUE].fields[i];
if (field->desc) { if (field->desc) {
if (QUERY_NODE_VALUE != nodeType(field->desc)) {
qDebug("VAL%d => [type:not value node][val:NIL]", i); //TODO
continue;
}
SValueNode *var = (SValueNode *)field->desc; SValueNode *var = (SValueNode *)field->desc;
SDataType *dType = &var->node.resType; SDataType *dType = &var->node.resType;
if (dType->type == TSDB_DATA_TYPE_VALUE_ARRAY) { if (dType->type == TSDB_DATA_TYPE_VALUE_ARRAY) {
......
...@@ -458,6 +458,9 @@ static void taosArrayInsertSort(SArray* pArray, __ext_compar_fn_t fn, const void ...@@ -458,6 +458,9 @@ static void taosArrayInsertSort(SArray* pArray, __ext_compar_fn_t fn, const void
} }
SArray* taosArrayDeepCopy(const SArray* pSrc, FCopy deepCopy) { SArray* taosArrayDeepCopy(const SArray* pSrc, FCopy deepCopy) {
if (NULL == pSrc) {
return NULL;
}
ASSERT(pSrc->elemSize == sizeof(void*)); ASSERT(pSrc->elemSize == sizeof(void*));
SArray* pArray = taosArrayInit(pSrc->size, sizeof(void*)); SArray* pArray = taosArrayInit(pSrc->size, sizeof(void*));
for (int32_t i = 0; i < pSrc->size; i++) { for (int32_t i = 0; i < pSrc->size; i++) {
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册