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

Merge pull request #4368 from taosdata/patch/TD-1983 

Patch/td 1983
This commit is contained in:
Shengliang Guan 2020-11-27 16:16:04 +08:00 committed by GitHub
parent 07bf2b3bf0 15691bfb08
commit de7bbc0d96
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
32 changed files with 430 additions and 382 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
src/balance/src/balance.c
View File

@ -162,7 +162,6 @@ int32_t balanceAllocVnodes(SVgObj *pVgroup) {
pDnode->openVnodes, pDnode->diskAvailable, pDnode->alternativeRole);
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
if (mnodeGetOnlineDnodesNum() == 0) {
return TSDB_CODE_MND_NOT_READY;
@ -377,15 +376,13 @@ static bool balanceMonitorBalance() {
srcScore, pDestDnode->score, destScore);
balanceAddVnode(pVgroup, pSrcDnode, pDestDnode);
mnodeDecVgroupRef(pVgroup);
sdbFreeIter(pIter);
mnodeCancelGetNextVgroup(pIter);
return true;
}
}
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
}
return false;
@ -413,8 +410,6 @@ void balanceReset() {
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
tsAccessSquence = 0;
}
@ -443,12 +438,11 @@ static int32_t balanceMonitorVgroups() {
mnodeDecVgroupRef(pVgroup);
if (code == TSDB_CODE_SUCCESS) {
mnodeCancelGetNextVgroup(pIter);
break;
}
}
sdbFreeIter(pIter);
return hasUpdatingVgroup;
}
@ -465,11 +459,12 @@ static bool balanceMonitorDnodeDropping(SDnodeObj *pDnode) {
hasThisDnode = balanceCheckDnodeInVgroup(pDnode, pVgroup);
mnodeDecVgroupRef(pVgroup);
if (hasThisDnode) break;
if (hasThisDnode) {
mnodeCancelGetNextVgroup(pIter);
break;
}
}
sdbFreeIter(pIter);
if (!hasThisDnode) {
mInfo("dnode:%d, dropped for all vnodes are moving to other dnodes", pDnode->dnodeId);
mnodeDropDnode(pDnode, NULL);
@ -499,20 +494,18 @@ static bool balanceMontiorDropping() {
pDnode->status = TAOS_DN_STATUS_DROPPING;
mnodeUpdateDnode(pDnode);
mnodeDecDnodeRef(pDnode);
sdbFreeIter(pIter);
mnodeCancelGetNextDnode(pIter);
return true;
}
if (pDnode->status == TAOS_DN_STATUS_DROPPING) {
bool ret = balanceMonitorDnodeDropping(pDnode);
mnodeDecDnodeRef(pDnode);
sdbFreeIter(pIter);
mnodeCancelGetNextDnode(pIter);
return ret;
}
}
sdbFreeIter(pIter);
return false;
}
@ -556,8 +549,6 @@ static void balanceSetVgroupOffline(SDnodeObj* pDnode) {
}
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
}
static void balanceCheckDnodeAccess() {
@ -578,8 +569,6 @@ static void balanceCheckDnodeAccess() {
}
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
}
static void balanceProcessBalanceTimer(void *handle, void *tmrId) {
@ -630,6 +619,7 @@ void balanceAsyncNotify() {
int32_t balanceInit() {
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_SCORES, balanceGetScoresMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_SCORES, balanceRetrieveScores);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_SCORES, mnodeCancelGetNextDnode);
pthread_mutex_init(&tsBalanceMutex, NULL);
balanceInitDnodeList();
@ -667,8 +657,6 @@ int32_t balanceDropDnode(SDnodeObj *pDnode) {
mnodeDecDnodeRef(pTempDnode);
}
sdbFreeIter(pIter);
if (pDnode->openVnodes > totalFreeVnodes) {
mError("dnode:%d, openVnodes:%d totalFreeVnodes:%d no enough dnodes", pDnode->dnodeId, pDnode->openVnodes, totalFreeVnodes);
return TSDB_CODE_MND_NO_ENOUGH_DNODES;
@ -780,8 +768,12 @@ void balanceAccquireDnodeList() {
SDnodeObj *pDnode = NULL;
int32_t dnodeIndex = 0;
while (1) {
if (dnodeIndex >= dnodesNum) break;
while (1) {
if (dnodeIndex >= dnodesNum) {
mnodeCancelGetNextDnode(pIter);
break;
}
pIter = mnodeGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
if (pDnode->status == TAOS_DN_STATUS_OFFLINE) {
@ -802,8 +794,6 @@ void balanceAccquireDnodeList() {
dnodeIndex++;
}
sdbFreeIter(pIter);
tsBalanceDnodeListSize = dnodeIndex;
}

2
src/connector/go

@ -1 +1 @@
Subproject commit 050667e5b4d0eafa5387e4283e713559b421203f
Subproject commit 8c58c512b6acda8bcdfa48fdc7140227b5221766

2
src/connector/grafanaplugin

@ -1 +1 @@
Subproject commit ec77d9049a719dabfd1a7c1122a209e201861944
Subproject commit d598db167eb256fe67409b7bb3d0eb7fffc3ff8c

1
src/mnode/inc/mnodeAcct.h
View File

@ -27,6 +27,7 @@ void mnodeCleanupAccts();
void mnodeGetStatOfAllAcct(SAcctInfo* pAcctInfo);
void * mnodeGetAcct(char *acctName);
void * mnodeGetNextAcct(void *pIter, SAcctObj **pAcct);
void mnodeCancelGetNextAcct(void *pIter);
void mnodeIncAcctRef(SAcctObj *pAcct);
void mnodeDecAcctRef(SAcctObj *pAcct);
void mnodeAddDbToAcct(SAcctObj *pAcct, SDbObj *pDb);

1
src/mnode/inc/mnodeDb.h
View File

@ -34,6 +34,7 @@ int64_t mnodeGetDbNum();
SDbObj *mnodeGetDb(char *db);
SDbObj *mnodeGetDbByTableId(char *db);
void * mnodeGetNextDb(void *pIter, SDbObj **pDb);
void mnodeCancelGetNextDb(void *pIter);
void mnodeIncDbRef(SDbObj *pDb);
void mnodeDecDbRef(SDbObj *pDb);
bool mnodeCheckIsMonitorDB(char *db, char *monitordb);

1
src/mnode/inc/mnodeDnode.h
View File

@ -65,6 +65,7 @@ int32_t mnodeGetDnodesNum();
int32_t mnodeGetOnlinDnodesCpuCoreNum();
int32_t mnodeGetOnlineDnodesNum();
void * mnodeGetNextDnode(void *pIter, SDnodeObj **pDnode);
void mnodeCancelGetNextDnode(void *pIter);
void mnodeIncDnodeRef(SDnodeObj *pDnode);
void mnodeDecDnodeRef(SDnodeObj *pDnode);
void * mnodeGetDnode(int32_t dnodeId);

1
src/mnode/inc/mnodeMnode.h
View File

@ -38,6 +38,7 @@ void mnodeDropMnodeLocal(int32_t dnodeId);
void * mnodeGetMnode(int32_t mnodeId);
int32_t mnodeGetMnodesNum();
void * mnodeGetNextMnode(void *pIter, struct SMnodeObj **pMnode);
void mnodeCancelGetNextMnode(void *pIter);
void mnodeIncMnodeRef(struct SMnodeObj *pMnode);
void mnodeDecMnodeRef(struct SMnodeObj *pMnode);

6
src/mnode/inc/mnodeSdb.h
View File

@ -92,9 +92,9 @@ int32_t sdbDeleteRow(SSdbRow *pRow);
int32_t sdbUpdateRow(SSdbRow *pRow);
int32_t sdbInsertRowToQueue(SSdbRow *pRow);
void *sdbGetRow(void *pTable, void *key);
void *sdbFetchRow(void *pTable, void *pIter, void **ppRow);
void sdbFreeIter(void *pIter);
void * sdbGetRow(void *pTable, void *key);
void * sdbFetchRow(void *pTable, void *pIter, void **ppRow);
void sdbFreeIter(void *pTable, void *pIter);
void sdbIncRef(void *pTable, void *pRow);
void sdbDecRef(void *pTable, void *pRow);
int64_t sdbGetNumOfRows(void *pTable);

2
src/mnode/inc/mnodeShow.h
View File

@ -26,8 +26,10 @@ void mnodeCleanUpShow();
typedef int32_t (*SShowMetaFp)(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
typedef int32_t (*SShowRetrieveFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
typedef void (*SShowFreeIterFp)(void *pIter);
void mnodeAddShowMetaHandle(uint8_t showType, SShowMetaFp fp);
void mnodeAddShowRetrieveHandle(uint8_t showType, SShowRetrieveFp fp);
void mnodeAddShowFreeIterHandle(uint8_t msgType, SShowFreeIterFp fp);
void mnodeVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capacity, SShowObj *pShow);
#ifdef __cplusplus

2
src/mnode/inc/mnodeTable.h
View File

@ -31,6 +31,8 @@ void mnodeIncTableRef(void *pTable);
void mnodeDecTableRef(void *pTable);
void * mnodeGetNextChildTable(void *pIter, SCTableObj **pTable);
void * mnodeGetNextSuperTable(void *pIter, SSTableObj **pTable);
void mnodeCancelGetNextChildTable(void *pIter);
void mnodeCancelGetNextSuperTable(void *pIter);
void mnodeDropAllChildTables(SDbObj *pDropDb);
void mnodeDropAllSuperTables(SDbObj *pDropDb);
void mnodeDropAllChildTablesInVgroups(SVgObj *pVgroup);

1
src/mnode/inc/mnodeUser.h
View File

@ -25,6 +25,7 @@ int32_t mnodeInitUsers();
void mnodeCleanupUsers();
SUserObj *mnodeGetUser(char *name);
void * mnodeGetNextUser(void *pIter, SUserObj **pUser);
void mnodeCancelGetNextUser(void *pIter);
void mnodeIncUserRef(SUserObj *pUser);
void mnodeDecUserRef(SUserObj *pUser);
SUserObj *mnodeGetUserFromConn(void *pConn);

1
src/mnode/inc/mnodeVgroup.h
View File

@ -34,6 +34,7 @@ void mnodeDropAllDnodeVgroups(SDnodeObj *pDropDnode);
//void mnodeUpdateAllDbVgroups(SDbObj *pAlterDb);
void * mnodeGetNextVgroup(void *pIter, SVgObj **pVgroup);
void mnodeCancelGetNextVgroup(void *pIter);
void mnodeUpdateVgroup(SVgObj *pVgroup);
void mnodeUpdateVgroupStatus(SVgObj *pVgroup, SDnodeObj *pDnode, SVnodeLoad *pVload);
void mnodeCheckUnCreatedVgroup(SDnodeObj *pDnode, SVnodeLoad *pVloads, int32_t openVnodes);

6
src/mnode/src/mnodeAcct.c
View File

@ -144,7 +144,6 @@ void mnodeGetStatOfAllAcct(SAcctInfo* pAcctInfo) {
pAcctInfo->numOfTimeSeries += pAcct->acctInfo.numOfTimeSeries;
mnodeDecAcctRef(pAcct);
}
sdbFreeIter(pIter);
SVgObj *pVgroup = NULL;
pIter = NULL;
@ -158,7 +157,6 @@ void mnodeGetStatOfAllAcct(SAcctInfo* pAcctInfo) {
pAcctInfo->totalPoints += pVgroup->pointsWritten;
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
}
void *mnodeGetAcct(char *name) {
@ -169,6 +167,10 @@ void *mnodeGetNextAcct(void *pIter, SAcctObj **pAcct) {
return sdbFetchRow(tsAcctSdb, pIter, (void **)pAcct);
}
void mnodeCancelGetNextAcct(void *pIter) {
sdbFreeIter(tsAcctSdb, pIter);
}
void mnodeIncAcctRef(SAcctObj *pAcct) {
sdbIncRef(tsAcctSdb, pAcct);
}

8
src/mnode/src/mnodeCluster.c
View File

@ -31,6 +31,7 @@ static int32_t mnodeCreateCluster();
static int32_t mnodeGetClusterMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mnodeCancelGetNextCluster(void *pIter);
static int32_t mnodeClusterActionDestroy(SSdbRow *pRow) {
tfree(pRow->pObj);
@ -108,6 +109,7 @@ int32_t mnodeInitCluster() {
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeGetClusterMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeRetrieveClusters);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeCancelGetNextCluster);
mDebug("table:%s, hash is created", desc.name);
return TSDB_CODE_SUCCESS;
@ -122,6 +124,10 @@ void *mnodeGetNextCluster(void *pIter, SClusterObj **pCluster) {
return sdbFetchRow(tsClusterSdb, pIter, (void **)pCluster);
}
void mnodeCancelGetNextCluster(void *pIter) {
sdbFreeIter(tsClusterSdb, pIter);
}
void mnodeIncClusterRef(SClusterObj *pCluster) {
sdbIncRef(tsClusterSdb, pCluster);
}
@ -167,7 +173,7 @@ void mnodeUpdateClusterId() {
}
mnodeDecClusterRef(pCluster);
sdbFreeIter(pIter);
mnodeCancelGetNextCluster(pIter);
}
static int32_t mnodeGetClusterMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {

12
src/mnode/src/mnodeDb.c
View File

@ -171,6 +171,7 @@ int32_t mnodeInitDbs() {
mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_DROP_DB, mnodeProcessDropDbMsg);
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_DB, mnodeGetDbMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_DB, mnodeRetrieveDbs);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_DB, mnodeCancelGetNextDb);
mDebug("table:dbs table is created");
return 0;
@ -180,6 +181,10 @@ void *mnodeGetNextDb(void *pIter, SDbObj **pDb) {
return sdbFetchRow(tsDbSdb, pIter, (void **)pDb);
}
void mnodeCancelGetNextDb(void *pIter) {
sdbFreeIter(tsDbSdb, pIter);
}
SDbObj *mnodeGetDb(char *db) {
return (SDbObj *)sdbGetRow(tsDbSdb, db);
}
@ -986,8 +991,8 @@ static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) {
SDbObj *pDb = pMsg->pDb;
void *pIter = NULL;
while (1) {
SVgObj *pVgroup = NULL;
SVgObj *pVgroup = NULL;
while (1) {
pIter = mnodeGetNextVgroup(pIter, &pVgroup);
if (pVgroup == NULL) break;
if (pVgroup->pDb == pDb) {
@ -995,7 +1000,6 @@ static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) {
}
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
mDebug("db:%s, all vgroups is altered", pDb->name);
mLInfo("db:%s, is alterd by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
@ -1146,7 +1150,5 @@ void mnodeDropAllDbs(SAcctObj *pAcct) {
mnodeDecDbRef(pDb);
}
sdbFreeIter(pIter);
mInfo("acct:%s, all dbs:%d is dropped from sdb", pAcct->user, numOfDbs);
}

23
src/mnode/src/mnodeDnode.c
View File

@ -206,6 +206,7 @@ int32_t mnodeInitDnodes() {
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_VNODES, mnodeRetrieveVnodes);
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_DNODE, mnodeGetDnodeMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_DNODE, mnodeRetrieveDnodes);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_DNODE, mnodeCancelGetNextDnode);
mDebug("table:dnodes table is created");
return 0;
@ -223,6 +224,10 @@ void *mnodeGetNextDnode(void *pIter, SDnodeObj **pDnode) {
return sdbFetchRow(tsDnodeSdb, pIter, (void **)pDnode);
}
void mnodeCancelGetNextDnode(void *pIter) {
sdbFreeIter(tsDnodeSdb, pIter);
}
int32_t mnodeGetDnodesNum() {
return sdbGetNumOfRows(tsDnodeSdb);
}
@ -241,8 +246,6 @@ int32_t mnodeGetOnlinDnodesCpuCoreNum() {
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
if (cpuCores < 2) cpuCores = 2;
return cpuCores;
}
@ -259,8 +262,6 @@ int32_t mnodeGetOnlineDnodesNum() {
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
return onlineDnodes;
}
@ -276,13 +277,12 @@ void *mnodeGetDnodeByEp(char *ep) {
pIter = mnodeGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
if (strcmp(ep, pDnode->dnodeEp) == 0) {
sdbFreeIter(pIter);
mnodeCancelGetNextDnode(pIter);
return pDnode;
}
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
return NULL;
}
@ -464,7 +464,10 @@ static void mnodeUpdateDnodeEps() {
while (1) {
pIter = mnodeGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
if (dnodesNum >= totalDnodes) break;
if (dnodesNum >= totalDnodes) {
mnodeCancelGetNextDnode(pIter);
break;
}
SDnodeEp *pEp = &tsDnodeEps->dnodeEps[dnodesNum];
dnodesNum++;
@ -474,7 +477,6 @@ static void mnodeUpdateDnodeEps() {
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
pthread_mutex_unlock(&tsDnodeEpsMutex);
}
@ -1100,7 +1102,7 @@ static int32_t mnodeGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pDnode = mnodeGetDnodeByEp(pShow->payload);
} else {
void *pIter = mnodeGetNextDnode(NULL, (SDnodeObj **)&pDnode);
sdbFreeIter(pIter);
mnodeCancelGetNextDnode(pIter);
}
if (pDnode != NULL) {
@ -1148,7 +1150,6 @@ static int32_t mnodeRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, vo
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
} else {
numOfRows = 0;
}
@ -1217,8 +1218,6 @@ int32_t balanceAllocVnodes(SVgObj *pVgroup) {
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
if (pSelDnode == NULL) {
mError("failed to alloc vnode to vgroup");
return TSDB_CODE_MND_NO_ENOUGH_DNODES;

9
src/mnode/src/mnodeMnode.c
View File

@ -123,7 +123,7 @@ static int32_t mnodeMnodeActionRestored() {
pMnode->role = TAOS_SYNC_ROLE_MASTER;
mnodeDecMnodeRef(pMnode);
}
sdbFreeIter(pIter);
mnodeCancelGetNextMnode(pIter);
}
mnodeUpdateMnodeEpSet();
@ -161,6 +161,7 @@ int32_t mnodeInitMnodes() {
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_MNODE, mnodeGetMnodeMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_MNODE, mnodeRetrieveMnodes);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_MNODE, mnodeCancelGetNextMnode);
mDebug("table:mnodes table is created");
return TSDB_CODE_SUCCESS;
@ -192,6 +193,10 @@ void *mnodeGetNextMnode(void *pIter, SMnodeObj **pMnode) {
return sdbFetchRow(tsMnodeSdb, pIter, (void **)pMnode);
}
void mnodeCancelGetNextMnode(void *pIter) {
sdbFreeIter(tsMnodeSdb, pIter);
}
void mnodeUpdateMnodeEpSet() {
mInfo("update mnodes epSet, numOfEps:%d ", mnodeGetMnodesNum());
@ -239,8 +244,6 @@ void mnodeUpdateMnodeEpSet() {
tsMnodeEpSetForShell.numOfEps = index;
tsMnodeEpSetForPeer.numOfEps = index;
sdbFreeIter(pIter);
mnodeMnodeUnLock();
}

24
src/mnode/src/mnodeProfile.c
View File

@ -42,6 +42,7 @@ static int32_t mnodeGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
static int32_t mnodeRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static int32_t mnodeGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mnodeCancelGetNextConn(void *pIter);
static int32_t mnodeGetStreamMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mnodeRetrieveStreams(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mnodeFreeConn(void *data);
@ -52,10 +53,13 @@ static int32_t mnodeProcessKillConnectionMsg(SMnodeMsg *pMsg);
int32_t mnodeInitProfile() {
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_QUERIES, mnodeGetQueryMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_QUERIES, mnodeRetrieveQueries);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_QUERIES, mnodeCancelGetNextConn);
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_CONNS, mnodeGetConnsMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_CONNS, mnodeRetrieveConns);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_CONNS, mnodeCancelGetNextConn);
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_STREAMS, mnodeGetStreamMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_STREAMS, mnodeRetrieveStreams);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_STREAMS, mnodeCancelGetNextConn);
mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_KILL_QUERY, mnodeProcessKillQueryMsg);
mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_KILL_STREAM, mnodeProcessKillStreamMsg);
@ -137,21 +141,15 @@ static void mnodeFreeConn(void *data) {
mDebug("connId:%d, is destroyed", pConn->connId);
}
static void *mnodeGetNextConn(SHashMutableIterator *pIter, SConnObj **pConn) {
static void *mnodeGetNextConn(void *pIter, SConnObj **pConn) {
*pConn = NULL;
if (pIter == NULL) {
pIter = taosHashCreateIter(tsMnodeConnCache->pHashTable);
}
pIter = taosHashIterate(tsMnodeConnCache->pHashTable, pIter);
if (pIter == NULL) return NULL;
if (!taosHashIterNext(pIter)) {
taosHashDestroyIter(pIter);
return NULL;
}
SCacheDataNode **pNode = taosHashIterGet(pIter);
SCacheDataNode **pNode = pIter;
if (pNode == NULL || *pNode == NULL) {
taosHashDestroyIter(pIter);
taosHashCancelIterate(tsMnodeConnCache->pHashTable, pIter);
return NULL;
}
@ -159,6 +157,10 @@ static void *mnodeGetNextConn(SHashMutableIterator *pIter, SConnObj **pConn) {
return pIter;
}
static void mnodeCancelGetNextConn(void *pIter) {
taosHashCancelIterate(tsMnodeConnCache->pHashTable, pIter);
}
static int32_t mnodeGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
SUserObj *pUser = mnodeGetUserFromConn(pConn);
if (pUser == NULL) return 0;

36
src/mnode/src/mnodeSdb.c
View File

@ -325,7 +325,6 @@ void sdbUpdateSync(void *pMnodes) {
mnodeDecDnodeRef(pDnode);
mnodeDecMnodeRef(pMnode);
}
sdbFreeIter(pIter);
syncCfg.replica = index;
mDebug("vgId:1, mnodes info not input, use infos in sdb, numOfMnodes:%d", syncCfg.replica);
} else {
@ -775,24 +774,17 @@ int32_t sdbUpdateRow(SSdbRow *pRow) {
}
}
void *sdbFetchRow(void *tparam, void *pNode, void **ppRow) {
void *sdbFetchRow(void *tparam, void *pIter, void **ppRow) {
SSdbTable *pTable = tparam;
*ppRow = NULL;
if (pTable == NULL) return NULL;
SHashMutableIterator *pIter = pNode;
if (pIter == NULL) {
pIter = taosHashCreateIter(pTable->iHandle);
}
pIter = taosHashIterate(pTable->iHandle, pIter);
if (pIter == NULL) return NULL;
if (!taosHashIterNext(pIter)) {
taosHashDestroyIter(pIter);
return NULL;
}
void **ppMetaRow = taosHashIterGet(pIter);
void **ppMetaRow = pIter;
if (ppMetaRow == NULL) {
taosHashDestroyIter(pIter);
taosHashCancelIterate(pTable->iHandle, pIter);
return NULL;
}
@ -802,10 +794,11 @@ void *sdbFetchRow(void *tparam, void *pNode, void **ppRow) {
return pIter;
}
void sdbFreeIter(void *pIter) {
if (pIter != NULL) {
taosHashDestroyIter(pIter);
}
void sdbFreeIter(void *tparam, void *pIter) {
SSdbTable *pTable = tparam;
if (pTable == NULL || pIter == NULL) return;
taosHashCancelIterate(pTable->iHandle, pIter);
}
void *sdbOpenTable(SSdbTableDesc *pDesc) {
@ -846,9 +839,10 @@ void sdbCloseTable(void *handle) {
tsSdbMgmt.numOfTables--;
tsSdbMgmt.tableList[pTable->id] = NULL;
SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle);
while (taosHashIterNext(pIter)) {
void **ppRow = taosHashIterGet(pIter);
void *pIter = taosHashIterate(pTable->iHandle, NULL);
while (pIter) {
void **ppRow = pIter;
pIter = taosHashIterate(pTable->iHandle, pIter);
if (ppRow == NULL) continue;
SSdbRow row = {
@ -859,7 +853,7 @@ void sdbCloseTable(void *handle) {
(*pTable->fpDestroy)(&row);
}
taosHashDestroyIter(pIter);
taosHashCancelIterate(pTable->iHandle, pIter);
taosHashCleanup(pTable->iHandle);
pthread_mutex_destroy(&pTable->mutex);

9
src/mnode/src/mnodeShow.c
View File

@ -57,6 +57,7 @@ static void *tsMnodeShowCache = NULL;
static int32_t tsShowObjIndex = 0;
static SShowMetaFp tsMnodeShowMetaFp[TSDB_MGMT_TABLE_MAX] = {0};
static SShowRetrieveFp tsMnodeShowRetrieveFp[TSDB_MGMT_TABLE_MAX] = {0};
static SShowFreeIterFp tsMnodeShowFreeIterFp[TSDB_MGMT_TABLE_MAX] = {0};
int32_t mnodeInitShow() {
mnodeAddReadMsgHandle(TSDB_MSG_TYPE_CM_SHOW, mnodeProcessShowMsg);
@ -85,6 +86,10 @@ void mnodeAddShowRetrieveHandle(uint8_t msgType, SShowRetrieveFp fp) {
tsMnodeShowRetrieveFp[msgType] = fp;
}
void mnodeAddShowFreeIterHandle(uint8_t msgType, SShowFreeIterFp fp) {
tsMnodeShowFreeIterFp[msgType] = fp;
}
static char *mnodeGetShowType(int32_t showType) {
switch (showType) {
case TSDB_MGMT_TABLE_ACCT: return "show accounts";
@ -412,7 +417,9 @@ static void* mnodePutShowObj(SShowObj *pShow) {
static void mnodeFreeShowObj(void *data) {
SShowObj *pShow = *(SShowObj **)data;
sdbFreeIter(pShow->pIter);
if (tsMnodeShowFreeIterFp[pShow->type] != NULL && pShow->pIter != NULL) {
(*tsMnodeShowFreeIterFp[pShow->type])(pShow->pIter);
}
mDebug("%p, show is destroyed, data:%p index:%d", pShow, data, pShow->index);
tfree(pShow);

47
src/mnode/src/mnodeTable.c
View File

@ -342,8 +342,7 @@ static int32_t mnodeChildTableActionRestored() {
mnodeDecTableRef(pTable);
}
sdbFreeIter(pIter);
mnodeCancelGetNextChildTable(pIter);
return 0;
}
@ -602,10 +601,13 @@ int32_t mnodeInitTables() {
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_TABLE, mnodeGetShowTableMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mnodeRetrieveShowTables);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_TABLE, mnodeCancelGetNextChildTable);
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_METRIC, mnodeGetShowSuperTableMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_METRIC, mnodeRetrieveShowSuperTables);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_METRIC, mnodeCancelGetNextSuperTable);
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_STREAMTABLES, mnodeGetStreamTableMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_STREAMTABLES, mnodeRetrieveStreamTables);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_STREAMTABLES, mnodeCancelGetNextChildTable);
return TSDB_CODE_SUCCESS;
}
@ -626,14 +628,12 @@ static void *mnodeGetSuperTableByUid(uint64_t uid) {
pIter = mnodeGetNextSuperTable(pIter, &pStable);
if (pStable == NULL) break;
if (pStable->uid == uid) {
sdbFreeIter(pIter);
mnodeCancelGetNextSuperTable(pIter);
return pStable;
}
mnodeDecTableRef(pStable);
}
sdbFreeIter(pIter);
return NULL;
}
@ -655,10 +655,18 @@ void *mnodeGetNextChildTable(void *pIter, SCTableObj **pTable) {
return sdbFetchRow(tsChildTableSdb, pIter, (void **)pTable);
}
void mnodeCancelGetNextChildTable(void *pIter) {
sdbFreeIter(tsChildTableSdb, pIter);
}
void *mnodeGetNextSuperTable(void *pIter, SSTableObj **pTable) {
return sdbFetchRow(tsSuperTableSdb, pIter, (void **)pTable);
}
void mnodeCancelGetNextSuperTable(void *pIter) {
sdbFreeIter(tsSuperTableSdb, pIter);
}
void mnodeIncTableRef(void *p1) {
STableObj *pTable = (STableObj *)p1;
if (pTable->type == TSDB_SUPER_TABLE) {
@ -914,10 +922,10 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
SSTableObj *pStable = (SSTableObj *)pMsg->pTable;
if (pStable->vgHash != NULL /*pStable->numOfTables != 0*/) {
SHashMutableIterator *pIter = taosHashCreateIter(pStable->vgHash);
while (taosHashIterNext(pIter)) {
int32_t *pVgId = taosHashIterGet(pIter);
int32_t *pVgId = taosHashIterate(pStable->vgHash, NULL);
while (pVgId) {
SVgObj *pVgroup = mnodeGetVgroup(*pVgId);
pVgId = taosHashIterate(pStable->vgHash, pVgId);
if (pVgroup == NULL) break;
SDropSTableMsg *pDrop = rpcMallocCont(sizeof(SDropSTableMsg));
@ -933,7 +941,8 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
dnodeSendMsgToDnode(&epSet, &rpcMsg);
mnodeDecVgroupRef(pVgroup);
}
taosHashDestroyIter(pIter);
taosHashCancelIterate(pStable->vgHash, pVgId);
mnodeDropAllChildTablesInStable(pStable);
}
@ -1430,8 +1439,6 @@ void mnodeDropAllSuperTables(SDbObj *pDropDb) {
mnodeDecTableRef(pTable);
}
sdbFreeIter(pIter);
mInfo("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
@ -1523,11 +1530,11 @@ static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
} else {
SVgroupsMsg *pVgroupMsg = (SVgroupsMsg *)msg;
SHashMutableIterator *pIter = taosHashCreateIter(pTable->vgHash);
int32_t vgSize = 0;
while (taosHashIterNext(pIter)) {
int32_t *pVgId = taosHashIterGet(pIter);
SVgObj * pVgroup = mnodeGetVgroup(*pVgId);
int32_t *pVgId = taosHashIterate(pTable->vgHash, NULL);
int32_t vgSize = 0;
while (pVgId) {
SVgObj *pVgroup = mnodeGetVgroup(*pVgId);
pVgId = taosHashIterate(pTable->vgHash, pVgId);
if (pVgroup == NULL) continue;
pVgroupMsg->vgroups[vgSize].vgId = htonl(pVgroup->vgId);
@ -1547,7 +1554,7 @@ static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
mnodeDecVgroupRef(pVgroup);
}
taosHashDestroyIter(pIter);
taosHashCancelIterate(pTable->vgHash, pVgId);
mnodeDecTableRef(pTable);
pVgroupMsg->numOfVgroups = htonl(vgSize);
@ -2230,8 +2237,6 @@ void mnodeDropAllChildTablesInVgroups(SVgObj *pVgroup) {
mnodeDecTableRef(pTable);
}
sdbFreeIter(pIter);
mInfo("vgId:%d, all child tables is dropped from sdb", pVgroup->vgId);
}
@ -2263,8 +2268,6 @@ void mnodeDropAllChildTables(SDbObj *pDropDb) {
mnodeDecTableRef(pTable);
}
sdbFreeIter(pIter);
mInfo("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
@ -2293,8 +2296,6 @@ static void mnodeDropAllChildTablesInStable(SSTableObj *pStable) {
mnodeDecTableRef(pTable);
}
sdbFreeIter(pIter);
mInfo("stable:%s, all child tables:%d is dropped from sdb", pStable->info.tableId, numOfTables);
}

9
src/mnode/src/mnodeUser.c
View File

@ -123,7 +123,6 @@ static void mnodePrintUserAuth() {
}
fflush(fp);
sdbFreeIter(pIter);
fclose(fp);
}
@ -177,6 +176,8 @@ int32_t mnodeInitUsers() {
mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_DROP_USER, mnodeProcessDropUserMsg);
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_USER, mnodeGetUserMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_USER, mnodeRetrieveUsers);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_USER, mnodeCancelGetNextUser);
mnodeAddPeerMsgHandle(TSDB_MSG_TYPE_DM_AUTH, mnodeProcessAuthMsg);
mDebug("table:%s, hash is created", desc.name);
@ -196,6 +197,10 @@ void *mnodeGetNextUser(void *pIter, SUserObj **pUser) {
return sdbFetchRow(tsUserSdb, pIter, (void **)pUser);
}
void mnodeCancelGetNextUser(void *pIter) {
sdbFreeIter(tsUserSdb, pIter);
}
void mnodeIncUserRef(SUserObj *pUser) {
return sdbIncRef(tsUserSdb, pUser);
}
@ -574,8 +579,6 @@ void mnodeDropAllUsers(SAcctObj *pAcct) {
mnodeDecUserRef(pUser);
}
sdbFreeIter(pIter);
mDebug("acct:%s, all users:%d is dropped from sdb", pAcct->user, numOfUsers);
}

15
src/mnode/src/mnodeVgroup.c
View File

@ -230,6 +230,7 @@ int32_t mnodeInitVgroups() {
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_VGROUP, mnodeGetVgroupMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_VGROUP, mnodeRetrieveVgroups);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_VGROUP, mnodeCancelGetNextVgroup);
mnodeAddPeerRspHandle(TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP, mnodeProcessCreateVnodeRsp);
mnodeAddPeerRspHandle(TSDB_MSG_TYPE_MD_ALTER_VNODE_RSP, mnodeProcessAlterVnodeRsp);
mnodeAddPeerRspHandle(TSDB_MSG_TYPE_MD_DROP_VNODE_RSP, mnodeProcessDropVnodeRsp);
@ -304,7 +305,7 @@ void mnodeCheckUnCreatedVgroup(SDnodeObj *pDnode, SVnodeLoad *pVloads, int32_t o
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
mnodeCancelGetNextVgroup(pIter);
}
void mnodeUpdateVgroupStatus(SVgObj *pVgroup, SDnodeObj *pDnode, SVnodeLoad *pVload) {
@ -491,6 +492,10 @@ void *mnodeGetNextVgroup(void *pIter, SVgObj **pVgroup) {
return sdbFetchRow(tsVgroupSdb, pIter, (void **)pVgroup);
}
void mnodeCancelGetNextVgroup(void *pIter) {
sdbFreeIter(tsVgroupSdb, pIter);
}
static int32_t mnodeCreateVgroupFp(SMnodeMsg *pMsg) {
SVgObj *pVgroup = pMsg->pVgroup;
SDbObj *pDb = pMsg->pDb;
@ -1095,8 +1100,6 @@ void mnodeDropAllDnodeVgroups(SDnodeObj *pDropDnode) {
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
mInfo("dnode:%d, all vgroups:%d is dropped from sdb", pDropDnode->dnodeId, numOfVgroups);
}
@ -1118,8 +1121,6 @@ void mnodeUpdateAllDbVgroups(SDbObj *pAlterDb) {
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
mInfo("db:%s, all vgroups is updated in sdb", pAlterDb->name);
}
#endif
@ -1147,8 +1148,6 @@ void mnodeDropAllDbVgroups(SDbObj *pDropDb) {
mnodeDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
mInfo("db:%s, all vgroups:%d is dropped from sdb", pDropDb->name, numOfVgroups);
}
@ -1170,7 +1169,5 @@ void mnodeSendDropAllDbVgroupsMsg(SDbObj *pDropDb) {
numOfVgroups++;
}
sdbFreeIter(pIter);
mInfo("db:%s, all vgroups:%d drop msg is sent to dnode", pDropDb->name, numOfVgroups);
}

8
src/query/src/qResultbuf.c
View File

@ -423,9 +423,8 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf) {
unlink(pResultBuf->path);
tfree(pResultBuf->path);
SHashMutableIterator* iter = taosHashCreateIter(pResultBuf->groupSet);
while(taosHashIterNext(iter)) {
SArray** p = (SArray**) taosHashIterGet(iter);
SArray** p = taosHashIterate(pResultBuf->groupSet, NULL);
while(p) {
size_t n = taosArrayGetSize(*p);
for(int32_t i = 0; i < n; ++i) {
SPageInfo* pi = taosArrayGetP(*p, i);
@ -434,10 +433,9 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf) {
}
taosArrayDestroy(*p);
p = taosHashIterate(pResultBuf->groupSet, p);
}
taosHashDestroyIter(iter);
tdListFree(pResultBuf->lruList);
taosArrayDestroy(pResultBuf->emptyDummyIdList);
taosHashCleanup(pResultBuf->groupSet);

15
src/rpc/src/rpcTcp.c
View File

@ -371,10 +371,13 @@ void taosCloseTcpConnection(void *chandle) {
int taosSendTcpData(uint32_t ip, uint16_t port, void *data, int len, void *chandle) {
SFdObj *pFdObj = chandle;
if (pFdObj == NULL || pFdObj->signature != pFdObj) return -1;
SThreadObj *pThreadObj = pFdObj->pThreadObj;
return taosWriteMsg(pFdObj->fd, data, len);
int ret = taosWriteMsg(pFdObj->fd, data, len);
tTrace("%s %p TCP data is sent, FD:%p fd:%d bytes:%d", pThreadObj->label, pFdObj->thandle, pFdObj, pFdObj->fd, ret);
return ret;
}
static void taosReportBrokenLink(SFdObj *pFdObj) {
@ -409,7 +412,7 @@ static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
headLen = taosReadMsg(pFdObj->fd, &rpcHead, sizeof(SRpcHead));
if (headLen != sizeof(SRpcHead)) {
tDebug("%s %p read error, headLen:%d", pThreadObj->label, pFdObj->thandle, headLen);
tDebug("%s %p read error, FD:%p headLen:%d", pThreadObj->label, pFdObj->thandle, pFdObj, headLen);
return -1;
}
@ -420,7 +423,7 @@ static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
tError("%s %p TCP malloc(size:%d) fail", pThreadObj->label, pFdObj->thandle, msgLen);
return -1;
} else {
tTrace("TCP malloc mem:%p size:%d", buffer, size);
tTrace("%s %p read data, FD:%p fd:%d TCP malloc mem:%p", pThreadObj->label, pFdObj->thandle, pFdObj, pFdObj->fd, buffer);
}
msg = buffer + tsRpcOverhead;
@ -583,8 +586,8 @@ static void taosFreeFdObj(SFdObj *pFdObj) {
pthread_mutex_unlock(&pThreadObj->mutex);
tDebug("%s %p TCP connection is closed, FD:%p numOfFds:%d",
pThreadObj->label, pFdObj->thandle, pFdObj, pThreadObj->numOfFds);
tDebug("%s %p TCP connection is closed, FD:%p fd:%d numOfFds:%d",
pThreadObj->label, pFdObj->thandle, pFdObj, pFdObj->fd, pThreadObj->numOfFds);
tfree(pFdObj);
}

54
src/util/inc/hash.h
View File

@ -31,16 +31,18 @@ extern "C" {
typedef void (*_hash_free_fn_t)(void *param);
typedef struct SHashNode {
// char *key;
struct SHashNode *next;
uint32_t hashVal; // the hash value of key
uint32_t keyLen; // length of the key
// char *data;
size_t dataLen; // length of data
int8_t count; // reference count
int8_t removed; // flag to indicate removed
char data[];
} SHashNode;
#define GET_HASH_NODE_KEY(_n) ((char*)(_n) + sizeof(SHashNode))
#define GET_HASH_NODE_DATA(_n) ((char*)(_n) + sizeof(SHashNode) + (_n)->keyLen)
#define GET_HASH_NODE_KEY(_n) ((char*)(_n) + sizeof(SHashNode) + (_n)->dataLen)
#define GET_HASH_NODE_DATA(_n) ((char*)(_n) + sizeof(SHashNode))
#define GET_HASH_PNODE(_n) ((char*)(_n) - sizeof(SHashNode));
typedef enum SHashLockTypeE {
HASH_NO_LOCK = 0,
HASH_ENTRY_LOCK = 1,
@ -65,15 +67,6 @@ typedef struct SHashObj {
SArray *pMemBlock; // memory block allocated for SHashEntry
} SHashObj;
typedef struct SHashMutableIterator {
SHashObj *pHashObj;
int32_t entryIndex;
SHashNode *pCur;
SHashNode *pNext; // current node can be deleted for mutable iterator, so keep the next one before return current
size_t numOfChecked; // already check number of elements in hash table
size_t numOfEntries; // number of entries while the iterator is created
} SHashMutableIterator;
/**
* init the hash table
*
@ -142,33 +135,9 @@ int32_t taosHashCondTraverse(SHashObj *pHashObj, bool (*fp)(void *, void *), voi
*/
void taosHashCleanup(SHashObj *pHashObj);
/**
*
* @param pHashObj
* @return
*/
SHashMutableIterator* taosHashCreateIter(SHashObj *pHashObj);
/**
*
* @param iter
* @return
*/
bool taosHashIterNext(SHashMutableIterator *iter);
/**
*
* @param iter
* @return
*/
void *taosHashIterGet(SHashMutableIterator *iter);
/**
*
* @param iter
* @return
*/
void* taosHashDestroyIter(SHashMutableIterator* iter);
/*
void *SHashMutableIterator* taosHashCreateIter(SHashObj *pHashObj, void *);
*/
/**
*
@ -179,6 +148,9 @@ int32_t taosHashGetMaxOverflowLinkLength(const SHashObj *pHashObj);
size_t taosHashGetMemSize(const SHashObj *pHashObj);
void *taosHashIterate(SHashObj *pHashObj, void *p);
void taosHashCancelIterate(SHashObj *pHashObj, void *p);
#ifdef __cplusplus
}
#endif

347
src/util/src/hash.c
View File

@ -76,7 +76,7 @@ static FORCE_INLINE int32_t taosHashCapacity(int32_t length) {
static FORCE_INLINE SHashNode *doSearchInEntryList(SHashEntry *pe, const void *key, size_t keyLen, uint32_t hashVal) {
SHashNode *pNode = pe->next;
while (pNode) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0) && pNode->removed == 0) {
assert(pNode->hashVal == hashVal);
break;
}
@ -114,15 +114,25 @@ static SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *p
* @param dsize size of actual data
* @return hash node
*/
static FORCE_INLINE SHashNode *doUpdateHashNode(SHashEntry* pe, SHashNode* prev, SHashNode *pNode, SHashNode *pNewNode) {
static FORCE_INLINE SHashNode *doUpdateHashNode(SHashObj *pHashObj, SHashEntry* pe, SHashNode* prev, SHashNode *pNode, SHashNode *pNewNode) {
assert(pNode->keyLen == pNewNode->keyLen);
pNode->count--;
if (prev != NULL) {
prev->next = pNewNode;
} else {
pe->next = pNewNode;
}
pNewNode->next = pNode->next;
if (pNode->count <= 0) {
pNewNode->next = pNode->next;
DO_FREE_HASH_NODE(pNode);
} else {
pNewNode->next = pNode;
pe->num++;
atomic_add_fetch_64(&pHashObj->size, 1);
}
return pNewNode;
}
@ -139,7 +149,6 @@ static void pushfrontNodeInEntryList(SHashEntry *pEntry, SHashNode *pNode);
* @param pIter
* @return
*/
static SHashNode *getNextHashNode(SHashMutableIterator *pIter);
SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTypeE type) {
if (capacity == 0 || fn == NULL) {
@ -213,7 +222,7 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
SHashNode* prev = NULL;
while (pNode) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0) && pNode->removed == 0) {
assert(pNode->hashVal == hashVal);
break;
}
@ -244,8 +253,7 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
} else {
// not support the update operation, return error
if (pHashObj->enableUpdate) {
doUpdateHashNode(pe, prev, pNode, pNewNode);
DO_FREE_HASH_NODE(pNode);
doUpdateHashNode(pHashObj, pe, prev, pNode, pNewNode);
} else {
DO_FREE_HASH_NODE(pNewNode);
}
@ -335,22 +343,10 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
int32_t slot = HASH_INDEX(hashVal, pHashObj->capacity);
SHashEntry *pe = pHashObj->hashList[slot];
// no data, return directly
if (pe->num == 0) {
__rd_unlock(&pHashObj->lock, pHashObj->type);
return -1;
}
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWLockLatch(&pe->latch);
}
if (pe->num == 0) {
assert(pe->next == NULL);
} else {
assert(pe->next != NULL);
}
// double check after locked
if (pe->num == 0) {
assert(pe->next == NULL);
@ -360,37 +356,37 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
return -1;
}
int code = -1;
SHashNode *pNode = pe->next;
SHashNode *pRes = NULL;
SHashNode *prevNode = NULL;
// remove it
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) {
pe->num -= 1;
pRes = pNode;
pe->next = pNode->next;
} else {
while (pNode->next != NULL) {
if (((pNode->next)->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY((pNode->next)), key, keyLen) == 0)) {
assert((pNode->next)->hashVal == hashVal);
break;
while (pNode) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0) && pNode->removed == 0)
break;
prevNode = pNode;
pNode = pNode->next;
}
if (pNode) {
code = 0; // it is found
pNode->count--;
pNode->removed = 1;
if (pNode->count <= 0) {
if (prevNode) {
prevNode->next = pNode->next;
} else {
pe->next = pNode->next;
}
if (data) memcpy(data, GET_HASH_NODE_DATA(pNode), dsize);
pNode = pNode->next;
pe->num--;
atomic_sub_fetch_64(&pHashObj->size, 1);
FREE_HASH_NODE(pHashObj, pNode);
}
if (pNode->next != NULL) {
pe->num -= 1;
pRes = pNode->next;
pNode->next = pNode->next->next;
}
}
if (pe->num == 0) {
assert(pe->next == NULL);
} else {
assert(pe->next != NULL);
}
}
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWUnLockLatch(&pe->latch);
@ -398,17 +394,7 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
__rd_unlock(&pHashObj->lock, pHashObj->type);
if (data != NULL && pRes != NULL) {
memcpy(data, GET_HASH_NODE_DATA(pRes), dsize);
}
if (pRes != NULL) {
atomic_sub_fetch_64(&pHashObj->size, 1);
FREE_HASH_NODE(pHashObj, pRes);
return 0;
} else {
return -1;
}
return code;
}
int32_t taosHashCondTraverse(SHashObj *pHashObj, bool (*fp)(void *, void *), void *param) {
@ -531,98 +517,6 @@ void taosHashCleanup(SHashObj *pHashObj) {
free(pHashObj);
}
SHashMutableIterator *taosHashCreateIter(SHashObj *pHashObj) {
SHashMutableIterator *pIter = calloc(1, sizeof(SHashMutableIterator));
if (pIter == NULL) {
return NULL;
}
pIter->pHashObj = pHashObj;
// keep it in local variable, in case the resize operation expand the size
pIter->numOfEntries = pHashObj->capacity;
return pIter;
}
bool taosHashIterNext(SHashMutableIterator *pIter) {
if (pIter == NULL) {
return false;
}
size_t size = taosHashGetSize(pIter->pHashObj);
if (size == 0) {
return false;
}
// check the first one
if (pIter->numOfChecked == 0) {
assert(pIter->pCur == NULL && pIter->pNext == NULL);
while (1) {
SHashEntry *pEntry = pIter->pHashObj->hashList[pIter->entryIndex];
if (pEntry->num == 0) {
assert(pEntry->next == NULL);
pIter->entryIndex++;
continue;
}
if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
taosRLockLatch(&pEntry->latch);
}
pIter->pCur = pEntry->next;
if (pIter->pCur->next) {
pIter->pNext = pIter->pCur->next;
if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
taosRUnLockLatch(&pEntry->latch);
}
} else {
if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
taosRUnLockLatch(&pEntry->latch);
}
pIter->pNext = getNextHashNode(pIter);
}
break;
}
pIter->numOfChecked++;
return true;
} else {
assert(pIter->pCur != NULL);
if (pIter->pNext) {
pIter->pCur = pIter->pNext;
} else { // no more data in the hash list
return false;
}
pIter->numOfChecked++;
if (pIter->pCur->next) {
pIter->pNext = pIter->pCur->next;
} else {
pIter->pNext = getNextHashNode(pIter);
}
return true;
}
}
void *taosHashIterGet(SHashMutableIterator *iter) { return (iter == NULL) ? NULL : GET_HASH_NODE_DATA(iter->pCur); }
void *taosHashDestroyIter(SHashMutableIterator *iter) {
if (iter == NULL) {
return NULL;
}
free(iter);
return NULL;
}
// for profile only
int32_t taosHashGetMaxOverflowLinkLength(const SHashObj *pHashObj) {
if (pHashObj == NULL || pHashObj->size == 0) {
@ -759,6 +653,8 @@ SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *pData, s
pNewNode->keyLen = (uint32_t)keyLen;
pNewNode->hashVal = hashVal;
pNewNode->dataLen = dsize;
pNewNode->count = 1;
memcpy(GET_HASH_NODE_DATA(pNewNode), pData, dsize);
memcpy(GET_HASH_NODE_KEY(pNewNode), key, keyLen);
@ -775,35 +671,6 @@ void pushfrontNodeInEntryList(SHashEntry *pEntry, SHashNode *pNode) {
pEntry->num += 1;
}
SHashNode *getNextHashNode(SHashMutableIterator *pIter) {
assert(pIter != NULL);
pIter->entryIndex++;
SHashNode *p = NULL;
while (pIter->entryIndex < pIter->numOfEntries) {
SHashEntry *pEntry = pIter->pHashObj->hashList[pIter->entryIndex];
if (pEntry->num == 0) {
pIter->entryIndex++;
continue;
}
if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
taosRLockLatch(&pEntry->latch);
}
p = pEntry->next;
if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
taosRUnLockLatch(&pEntry->latch);
}
return p;
}
return NULL;
}
size_t taosHashGetMemSize(const SHashObj *pHashObj) {
if (pHashObj == NULL) {
return 0;
@ -811,3 +678,129 @@ size_t taosHashGetMemSize(const SHashObj *pHashObj) {
return (pHashObj->capacity * (sizeof(SHashEntry) + POINTER_BYTES)) + sizeof(SHashNode) * taosHashGetSize(pHashObj) + sizeof(SHashObj);
}
// release the pNode, return next pNode, and lock the current entry
static void *taosHashReleaseNode(SHashObj *pHashObj, void *p, int *slot) {
SHashNode *pOld = (SHashNode *)GET_HASH_PNODE(p);
SHashNode *prevNode = NULL;
*slot = HASH_INDEX(pOld->hashVal, pHashObj->capacity);
SHashEntry *pe = pHashObj->hashList[*slot];
// lock entry
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWLockLatch(&pe->latch);
}
SHashNode *pNode = pe->next;
while (pNode) {
if (pNode == pOld)
break;
prevNode = pNode;
pNode = pNode->next;
}
if (pNode) {
pNode = pNode->next;
while (pNode) {
if (pNode->removed == 0) break;
pNode = pNode->next;
}
pOld->count--;
if (pOld->count <=0) {
if (prevNode) {
prevNode->next = pOld->next;
} else {
pe->next = pOld->next;
}
pe->num--;
atomic_sub_fetch_64(&pHashObj->size, 1);
FREE_HASH_NODE(pHashObj, pOld);
}
} else {
uError("pNode:%p data:%p is not there!!!", pNode, p);
}
return pNode;
}
void *taosHashIterate(SHashObj *pHashObj, void *p) {
if (pHashObj == NULL) return NULL;
int slot = 0;
char *data = NULL;
// only add the read lock to disable the resize process
__rd_lock(&pHashObj->lock, pHashObj->type);
SHashNode *pNode = NULL;
if (p) {
pNode = taosHashReleaseNode(pHashObj, p, &slot);
if (pNode == NULL) {
SHashEntry *pe = pHashObj->hashList[slot];
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWUnLockLatch(&pe->latch);
}
slot = slot + 1;
}
}
if (pNode == NULL) {
for (; slot < pHashObj->capacity; ++slot) {
SHashEntry *pe = pHashObj->hashList[slot];
// lock entry
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWLockLatch(&pe->latch);
}
pNode = pe->next;
while (pNode) {
if (pNode->removed == 0) break;
pNode = pNode->next;
}
if (pNode) break;
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWUnLockLatch(&pe->latch);
}
}
}
if (pNode) {
SHashEntry *pe = pHashObj->hashList[slot];
pNode->count++;
data = GET_HASH_NODE_DATA(pNode);
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWUnLockLatch(&pe->latch);
}
}
__rd_unlock(&pHashObj->lock, pHashObj->type);
return data;
}
void taosHashCancelIterate(SHashObj *pHashObj, void *p) {
if (pHashObj == NULL || p == NULL) return;
// only add the read lock to disable the resize process
__rd_lock(&pHashObj->lock, pHashObj->type);
int slot;
taosHashReleaseNode(pHashObj, p, &slot);
SHashEntry *pe = pHashObj->hashList[slot];
if (pHashObj->type == HASH_ENTRY_LOCK) {
taosWUnLockLatch(&pe->latch);
}
__rd_unlock(&pHashObj->lock, pHashObj->type);
}

19
src/util/src/tkvstore.c
View File

@ -529,7 +529,7 @@ static int tdRestoreKVStore(SKVStore *pStore) {
void * buf = NULL;
int64_t maxBufSize = 0;
SKVRecord rInfo = {0};
SHashMutableIterator *pIter = NULL;
SKVRecord *pRecord = NULL;
ASSERT(TD_KVSTORE_HEADER_SIZE == lseek(pStore->fd, 0, SEEK_CUR));
ASSERT(pStore->info.size == TD_KVSTORE_HEADER_SIZE);
@ -582,16 +582,8 @@ static int tdRestoreKVStore(SKVStore *pStore) {
goto _err;
}
pIter = taosHashCreateIter(pStore->map);
if (pIter == NULL) {
uError("failed to create hash iter while opening KV store %s", pStore->fname);
terrno = TSDB_CODE_COM_OUT_OF_MEMORY;
goto _err;
}
while (taosHashIterNext(pIter)) {
SKVRecord *pRecord = taosHashIterGet(pIter);
pRecord = taosHashIterate(pStore->map, NULL);
while (pRecord) {
if (lseek(pStore->fd, (off_t)(pRecord->offset + sizeof(SKVRecord)), SEEK_SET) < 0) {
uError("failed to lseek file %s since %s, offset %" PRId64, pStore->fname, strerror(errno), pRecord->offset);
terrno = TAOS_SYSTEM_ERROR(errno);
@ -613,16 +605,17 @@ static int tdRestoreKVStore(SKVStore *pStore) {
goto _err;
}
}
pRecord = taosHashIterate(pStore->map, pRecord);
}
if (pStore->aFunc) (*pStore->aFunc)(pStore->appH);
taosHashDestroyIter(pIter);
tfree(buf);
return 0;
_err:
taosHashDestroyIter(pIter);
taosHashCancelIterate(pStore->map, pRecord);
tfree(buf);
return -1;
}

31
src/vnode/src/vnodeMain.c
View File

@ -520,11 +520,10 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) {
}
int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash);
while (taosHashIterNext(pIter)) {
SVnodeObj **pVnode = taosHashIterGet(pIter);
if (pVnode == NULL) continue;
if (*pVnode == NULL) continue;
void *pIter = taosHashIterate(tsVnodesHash, NULL);
while (pIter) {
SVnodeObj **pVnode = pIter;
if (*pVnode) {
(*numOfVnodes)++;
if (*numOfVnodes >= TSDB_MAX_VNODES) {
@ -533,25 +532,25 @@ int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
} else {
vnodeList[*numOfVnodes - 1] = (*pVnode)->vgId;
}
}
taosHashDestroyIter(pIter);
}
pIter = taosHashIterate(tsVnodesHash, pIter);
}
return TSDB_CODE_SUCCESS;
}
void vnodeBuildStatusMsg(void *param) {
SStatusMsg *pStatus = param;
SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash);
while (taosHashIterNext(pIter)) {
SVnodeObj **pVnode = taosHashIterGet(pIter);
if (pVnode == NULL) continue;
if (*pVnode == NULL) continue;
vnodeBuildVloadMsg(*pVnode, pStatus);
void *pIter = taosHashIterate(tsVnodesHash, NULL);
while (pIter) {
SVnodeObj **pVnode = pIter;
if (*pVnode) {
vnodeBuildVloadMsg(*pVnode, pStatus);
}
pIter = taosHashIterate(tsVnodesHash, pIter);
}
taosHashDestroyIter(pIter);
}
void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes) {

1
tests/script/basicSuite.sim
View File

@ -14,7 +14,6 @@ run general/table/vgroup.sim
run general/user/authority.sim
run general/vector/metrics_mix.sim
run general/vector/table_field.sim
run general/user/authority.sim
run general/tag/set.sim
run general/table/delete_writing.sim
run general/stable/disk.sim

7
tests/test/c/CMakeLists.txt
View File

@ -31,8 +31,8 @@ IF (TD_LINUX)
#add_executable(createTablePerformance createTablePerformance.c)
#target_link_libraries(createTablePerformance taos_static tutil common pthread)
#add_executable(createNormalTable createNormalTable.c)
#target_link_libraries(createNormalTable taos_static tutil common pthread)
add_executable(createNormalTable createNormalTable.c)
target_link_libraries(createNormalTable taos_static tutil common pthread)
#add_executable(queryPerformance queryPerformance.c)
#target_link_libraries(queryPerformance taos_static tutil common pthread)
@ -45,5 +45,8 @@ IF (TD_LINUX)
#add_executable(invalidTableId invalidTableId.c)
#target_link_libraries(invalidTableId taos_static tutil common pthread)
add_executable(hashIterator hashIterator.c)
target_link_libraries(hashIterator taos_static tutil common pthread)
ENDIF()

72
tests/test/c/hashIterator.c Normal file
View File

@ -0,0 +1,72 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "os.h"
#include "taos.h"
#include "tulog.h"
#include "tutil.h"
#include "hash.h"
typedef struct HashTestRow {
int32_t keySize;
char key[100];
} HashTestRow;
int main(int argc, char *argv[]) {
_hash_fn_t hashFp = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
void * hashHandle = taosHashInit(100, hashFp, true, HASH_ENTRY_LOCK);
pPrint("insert 3 rows to hash");
for (int32_t t = 0; t < 3; ++t) {
HashTestRow row = {0};
row.keySize = sprintf(row.key, "0.db.st%d", t);
taosHashPut(hashHandle, row.key, row.keySize, &row, sizeof(HashTestRow));
}
pPrint("start iterator");
HashTestRow *row = taosHashIterate(hashHandle, NULL);
while (row) {
pPrint("drop key:%s", row->key);
taosHashRemove(hashHandle, row->key, row->keySize);
pPrint("get rows from hash");
for (int32_t t = 0; t < 3; ++t) {
HashTestRow r = {0};
r.keySize = sprintf(r.key, "0.db.st%d", t);
void *result = taosHashGet(hashHandle, r.key, r.keySize);
pPrint("get key:%s result:%p", r.key, result);
}
//Before getting the next iterator, the object just deleted can be obtained
row = taosHashIterate(hashHandle, row);
}
pPrint("stop iterator");
taosHashCancelIterate(hashHandle, row);
pPrint("get rows from hash");
for (int32_t t = 0; t < 3; ++t) {
HashTestRow r = {0};
r.keySize = sprintf(r.key, "0.db.st%d", t);
void *result = taosHashGet(hashHandle, r.key, r.keySize);
pPrint("get key:%s result:%p", r.key, result);
}
return 0;
}