Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
未验证 提交 5e7f8430 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #1907 from taosdata/feature/hash 

Feature/hash
上级 20d372a4 174f0354
隐藏空白更改
内联 并排
Showing with 303 addition and 1665 deletion
src/client/src/tscSub.c
浏览文件 @ 5e7f8430
......@@ -14,8 +14,6 @@
*/
#include "os.h"
#include "shash.h"
#include "taos.h"
#include "trpc.h"
#include "tsclient.h"
......
src/dnode/src/dnodeMgmt.c
浏览文件 @ 5e7f8430
......@@ -16,7 +16,6 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "cJSON.h"
#include "ihash.h"
#include "taoserror.h"
#include "taosmsg.h"
#include "ttime.h"
......
src/mnode/inc/mgmtAcct.h
浏览文件 @ 5e7f8430
......@@ -25,7 +25,7 @@ extern "C" {
int32_t mgmtInitAccts();
void mgmtCleanUpAccts();
void * mgmtGetAcct(char *acctName);
void * mgmtGetNextAcct(void *pNode, SAcctObj **pAcct);
void * mgmtGetNextAcct(void *pIter, SAcctObj **pAcct);
void mgmtIncAcctRef(SAcctObj *pAcct);
void mgmtDecAcctRef(SAcctObj *pAcct);
void mgmtAddDbToAcct(SAcctObj *pAcct, SDbObj *pDb);
......
src/mnode/inc/mgmtDb.h
浏览文件 @ 5e7f8430
......@@ -32,7 +32,7 @@ int32_t mgmtInitDbs();
void mgmtCleanUpDbs();
SDbObj *mgmtGetDb(char *db);
SDbObj *mgmtGetDbByTableId(char *db);
void * mgmtGetNextDb(void *pNode, SDbObj **pDb);
void * mgmtGetNextDb(void *pIter, SDbObj **pDb);
void mgmtIncDbRef(SDbObj *pDb);
void mgmtDecDbRef(SDbObj *pDb);
bool mgmtCheckIsMonitorDB(char *db, char *monitordb);
......
src/mnode/inc/mgmtDef.h
浏览文件 @ 5e7f8430
......@@ -223,7 +223,7 @@ typedef struct SAcctObj {
typedef struct {
int8_t type;
char db[TSDB_DB_NAME_LEN + 1];
void * pNode;
void * pIter;
int16_t numOfColumns;
int32_t rowSize;
int32_t numOfRows;
......
src/mnode/inc/mgmtDnode.h
浏览文件 @ 5e7f8430
......@@ -34,7 +34,7 @@ char* mgmtGetDnodeStatusStr(int32_t dnodeStatus);
void mgmtMonitorDnodeModule();
int32_t mgmtGetDnodesNum();
void * mgmtGetNextDnode(void *pNode, SDnodeObj **pDnode);
void * mgmtGetNextDnode(void *pIter, SDnodeObj **pDnode);
void mgmtIncDnodeRef(SDnodeObj *pDnode);
void mgmtDecDnodeRef(SDnodeObj *pDnode);
void * mgmtGetDnode(int32_t dnodeId);
......
src/mnode/inc/mgmtMnode.h
浏览文件 @ 5e7f8430
......@@ -37,7 +37,7 @@ void mgmtDropMnodeLocal(int32_t dnodeId);
void * mgmtGetMnode(int32_t mnodeId);
int32_t mgmtGetMnodesNum();
void * mgmtGetNextMnode(void *pNode, struct SMnodeObj **pMnode);
void * mgmtGetNextMnode(void *pIter, struct SMnodeObj **pMnode);
void mgmtIncMnodeRef(struct SMnodeObj *pMnode);
void mgmtDecMnodeRef(struct SMnodeObj *pMnode);
......
src/mnode/inc/mgmtSdb.h
浏览文件 @ 5e7f8430
......@@ -80,7 +80,8 @@ int32_t sdbDeleteRow(SSdbOper *pOper);
int32_t sdbUpdateRow(SSdbOper *pOper);
void *sdbGetRow(void *handle, void *key);
void *sdbFetchRow(void *handle, void *pNode, void **ppRow);
void *sdbFetchRow(void *handle, void *pIter, void **ppRow);
void sdbFreeIter(void *pIter);
void sdbIncRef(void *thandle, void *pRow);
void sdbDecRef(void *thandle, void *pRow);
int64_t sdbGetNumOfRows(void *handle);
......
src/mnode/inc/mgmtTable.h
浏览文件 @ 5e7f8430
......@@ -27,8 +27,8 @@ void mgmtCleanUpTables();
void * mgmtGetTable(char *tableId);
void mgmtIncTableRef(void *pTable);
void mgmtDecTableRef(void *pTable);
void * mgmtGetNextChildTable(void *pNode, SChildTableObj **pTable);
void * mgmtGetNextSuperTable(void *pNode, SSuperTableObj **pTable);
void * mgmtGetNextChildTable(void *pIter, SChildTableObj **pTable);
void * mgmtGetNextSuperTable(void *pIter, SSuperTableObj **pTable);
void mgmtDropAllChildTables(SDbObj *pDropDb);
void mgmtDropAllSuperTables(SDbObj *pDropDb);
......
src/mnode/inc/mgmtUser.h
浏览文件 @ 5e7f8430
......@@ -24,7 +24,7 @@ extern "C" {
int32_t mgmtInitUsers();
void mgmtCleanUpUsers();
SUserObj *mgmtGetUser(char *name);
void * mgmtGetNextUser(void *pNode, SUserObj **pUser);
void * mgmtGetNextUser(void *pIter, SUserObj **pUser);
void mgmtIncUserRef(SUserObj *pUser);
void mgmtDecUserRef(SUserObj *pUser);
SUserObj *mgmtGetUserFromConn(void *pConn);
......
src/mnode/inc/mgmtVgroup.h
浏览文件 @ 5e7f8430
......@@ -35,7 +35,7 @@ void mgmtDecVgroupRef(SVgObj *pVgroup);
void mgmtDropAllDbVgroups(SDbObj *pDropDb, bool sendMsg);
void mgmtDropAllDnodeVgroups(SDnodeObj *pDropDnode);
void * mgmtGetNextVgroup(void *pNode, SVgObj **pVgroup);
void * mgmtGetNextVgroup(void *pIter, SVgObj **pVgroup);
void mgmtUpdateVgroup(SVgObj *pVgroup);
void mgmtUpdateVgroupStatus(SVgObj *pVgroup, SDnodeObj *dnodeId, SVnodeLoad *pVload);
......
src/mnode/src/mgmtAcct.c
浏览文件 @ 5e7f8430
......@@ -126,8 +126,8 @@ void *mgmtGetAcct(char *name) {
return sdbGetRow(tsAcctSdb, name);
}
void *mgmtGetNextAcct(void *pNode, SAcctObj **pAcct) {
return sdbFetchRow(tsAcctSdb, pNode, (void **)pAcct);
void *mgmtGetNextAcct(void *pIter, SAcctObj **pAcct) {
return sdbFetchRow(tsAcctSdb, pIter, (void **)pAcct);
}
void mgmtIncAcctRef(SAcctObj *pAcct) {
......
src/mnode/src/mgmtBalance.c
浏览文件 @ 5e7f8430
......@@ -22,6 +22,7 @@
#include "mgmtInt.h"
#include "mgmtMnode.h"
#include "mgmtDnode.h"
#include "mgmtSdb.h"
#include "mgmtVgroup.h"
#ifndef _SYNC
......@@ -33,13 +34,13 @@ void balanceUpdateMgmt() {}
void balanceReset() {}
int32_t balanceAllocVnodes(SVgObj *pVgroup) {
void * pNode = NULL;
void * pIter = NULL;
SDnodeObj *pDnode = NULL;
SDnodeObj *pSelDnode = NULL;
float vnodeUsage = 1000.0;
while (1) {
pNode = mgmtGetNextDnode(pNode, &pDnode);
pIter = mgmtGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
if (pDnode->totalVnodes > 0 && pDnode->openVnodes < pDnode->totalVnodes) {
......@@ -55,6 +56,8 @@ int32_t balanceAllocVnodes(SVgObj *pVgroup) {
mgmtDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
if (pSelDnode == NULL) {
mError("failed to alloc vnode to vgroup");
return TSDB_CODE_NO_ENOUGH_DNODES;
......
src/mnode/src/mgmtDb.c
浏览文件 @ 5e7f8430
......@@ -156,8 +156,8 @@ int32_t mgmtInitDbs() {
return 0;
}
void *mgmtGetNextDb(void *pNode, SDbObj **pDb) {
return sdbFetchRow(tsDbSdb, pNode, (void **)pDb);
void *mgmtGetNextDb(void *pIter, SDbObj **pDb) {
return sdbFetchRow(tsDbSdb, pIter, (void **)pDb);
}
SDbObj *mgmtGetDb(char *db) {
......@@ -583,7 +583,7 @@ static int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *
if (pUser == NULL) return 0;
while (numOfRows < rows) {
pShow->pNode = mgmtGetNextDb(pShow->pNode, &pDb);
pShow->pIter = mgmtGetNextDb(pShow->pIter, &pDb);
if (pDb == NULL) break;
cols = 0;
......@@ -865,14 +865,15 @@ static int32_t mgmtAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
}
}
void *pNode = NULL;
void *pIter = NULL;
while (1) {
SVgObj *pVgroup = NULL;
pNode = mgmtGetNextVgroup(pNode, &pVgroup);
pIter = mgmtGetNextVgroup(pIter, &pVgroup);
if (pVgroup == NULL) break;
mgmtSendCreateVgroupMsg(pVgroup, NULL);
mgmtDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
if (oldReplica != pDb->cfg.replications) {
balanceNotify();
......@@ -983,12 +984,12 @@ static void mgmtProcessDropDbMsg(SQueuedMsg *pMsg) {
void mgmtDropAllDbs(SAcctObj *pAcct) {
int32_t numOfDbs = 0;
SDbObj *pDb = NULL;
void * pNode = NULL;
void * pIter = NULL;
mPrint("acct:%s, all dbs will be dropped from sdb", pAcct->user);
while (1) {
pNode = mgmtGetNextDb(pNode, &pDb);
pIter = mgmtGetNextDb(pIter, &pDb);
if (pDb == NULL) break;
if (pDb->pAcct == pAcct) {
......@@ -1005,5 +1006,7 @@ void mgmtDropAllDbs(SAcctObj *pAcct) {
mgmtDecDbRef(pDb);
}
sdbFreeIter(pIter);
mPrint("acct:%s, all dbs:%d is dropped from sdb", pAcct->user, numOfDbs);
}
src/mnode/src/mgmtDnode.c
浏览文件 @ 5e7f8430
......@@ -170,8 +170,8 @@ void mgmtCleanupDnodes() {
sdbCloseTable(tsDnodeSdb);
}
void *mgmtGetNextDnode(void *pNode, SDnodeObj **pDnode) {
return sdbFetchRow(tsDnodeSdb, pNode, (void **)pDnode);
void *mgmtGetNextDnode(void *pIter, SDnodeObj **pDnode) {
return sdbFetchRow(tsDnodeSdb, pIter, (void **)pDnode);
}
int32_t mgmtGetDnodesNum() {
......@@ -184,17 +184,20 @@ void *mgmtGetDnode(int32_t dnodeId) {
void *mgmtGetDnodeByEp(char *ep) {
SDnodeObj *pDnode = NULL;
void * pNode = NULL;
void * pIter = NULL;
while (1) {
pNode = mgmtGetNextDnode(pNode, &pDnode);
pIter = mgmtGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
if (strcmp(ep, pDnode->dnodeEp) == 0) {
sdbFreeIter(pIter);
return pDnode;
}
mgmtDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
return NULL;
}
......@@ -530,7 +533,7 @@ static int32_t mgmtGetDnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
pShow->numOfRows = mgmtGetDnodesNum();
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
pShow->pNode = NULL;
pShow->pIter = NULL;
mgmtDecUserRef(pUser);
......@@ -544,7 +547,7 @@ static int32_t mgmtRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, voi
char *pWrite;
while (numOfRows < rows) {
pShow->pNode = mgmtGetNextDnode(pShow->pNode, &pDnode);
pShow->pIter = mgmtGetNextDnode(pShow->pIter, &pDnode);
if (pDnode == NULL) break;
cols = 0;
......@@ -636,7 +639,7 @@ static int32_t mgmtGetModuleMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pShow->numOfRows = mgmtGetDnodesNum() * TSDB_MOD_MAX;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
pShow->pNode = NULL;
pShow->pIter = NULL;
mgmtDecUserRef(pUser);
return 0;
......@@ -648,7 +651,7 @@ int32_t mgmtRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pCo
while (numOfRows < rows) {
SDnodeObj *pDnode = NULL;
pShow->pNode = mgmtGetNextDnode(pShow->pNode, (SDnodeObj **)&pDnode);
pShow->pIter = mgmtGetNextDnode(pShow->pIter, (SDnodeObj **)&pDnode);
if (pDnode == NULL) break;
for (int32_t moduleType = 0; moduleType < TSDB_MOD_MAX; ++moduleType) {
......@@ -738,7 +741,7 @@ static int32_t mgmtGetConfigMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
}
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
pShow->pNode = NULL;
pShow->pIter = NULL;
mgmtDecUserRef(pUser);
return 0;
......@@ -821,7 +824,8 @@ static int32_t mgmtGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
if (pShow->payloadLen > 0 ) {
pDnode = mgmtGetDnodeByEp(pShow->payload);
} else {
mgmtGetNextDnode(NULL, (SDnodeObj **)&pDnode);
void *pIter = mgmtGetNextDnode(NULL, (SDnodeObj **)&pDnode);
sdbFreeIter(pIter);
}
if (pDnode != NULL) {
......@@ -830,7 +834,7 @@ static int32_t mgmtGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
}
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
pShow->pNode = pDnode;
pShow->pIter = pDnode;
mgmtDecUserRef(pUser);
return 0;
......@@ -844,12 +848,12 @@ static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, voi
if (0 == rows) return 0;
pDnode = (SDnodeObj *)(pShow->pNode);
pDnode = (SDnodeObj *)(pShow->pIter);
if (pDnode != NULL) {
void *pNode = NULL;
void *pIter = NULL;
SVgObj *pVgroup;
while (1) {
pNode = mgmtGetNextVgroup(pNode, &pVgroup);
pIter = mgmtGetNextVgroup(pIter, &pVgroup);
if (pVgroup == NULL) break;
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
......@@ -869,6 +873,7 @@ static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, voi
mgmtDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
} else {
numOfRows = 0;
}
......
src/mnode/src/mgmtMnode.c
浏览文件 @ 5e7f8430
......@@ -95,11 +95,12 @@ static int32_t mgmtMnodeActionDecode(SSdbOper *pOper) {
static int32_t mgmtMnodeActionRestored() {
if (mgmtGetMnodesNum() == 1) {
SMnodeObj *pMnode = NULL;
mgmtGetNextMnode(NULL, &pMnode);
void *pIter = mgmtGetNextMnode(NULL, &pMnode);
if (pMnode != NULL) {
pMnode->role = TAOS_SYNC_ROLE_MASTER;
mgmtDecMnodeRef(pMnode);
}
sdbFreeIter(pIter);
}
return TSDB_CODE_SUCCESS;
}
......@@ -157,8 +158,8 @@ void mgmtDecMnodeRef(SMnodeObj *pMnode) {
sdbDecRef(tsMnodeSdb, pMnode);
}
void *mgmtGetNextMnode(void *pNode, SMnodeObj **pMnode) {
return sdbFetchRow(tsMnodeSdb, pNode, (void **)pMnode);
void *mgmtGetNextMnode(void *pIter, SMnodeObj **pMnode) {
return sdbFetchRow(tsMnodeSdb, pIter, (void **)pMnode);
}
char *mgmtGetMnodeRoleStr(int32_t role) {
......@@ -177,10 +178,10 @@ char *mgmtGetMnodeRoleStr(int32_t role) {
}
void mgmtGetMnodeIpSet(SRpcIpSet *ipSet) {
void *pNode = NULL;
void *pIter = NULL;
while (1) {
SMnodeObj *pMnode = NULL;
pNode = mgmtGetNextMnode(pNode, &pMnode);
pIter = mgmtGetNextMnode(pIter, &pMnode);
if (pMnode == NULL) break;
strcpy(ipSet->fqdn[ipSet->numOfIps], pMnode->pDnode->dnodeFqdn);
......@@ -194,6 +195,7 @@ void mgmtGetMnodeIpSet(SRpcIpSet *ipSet) {
mgmtDecMnodeRef(pMnode);
}
sdbFreeIter(pIter);
}
void mgmtGetMnodeInfos(void *param) {
......@@ -201,10 +203,10 @@ void mgmtGetMnodeInfos(void *param) {
mnodes->inUse = 0;
int32_t index = 0;
void *pNode = NULL;
void *pIter = NULL;
while (1) {
SMnodeObj *pMnode = NULL;
pNode = mgmtGetNextMnode(pNode, &pMnode);
pIter = mgmtGetNextMnode(pIter, &pMnode);
if (pMnode == NULL) break;
mnodes->nodeInfos[index].nodeId = htonl(pMnode->mnodeId);
......@@ -216,6 +218,7 @@ void mgmtGetMnodeInfos(void *param) {
index++;
mgmtDecMnodeRef(pMnode);
}
sdbFreeIter(pIter);
mnodes->nodeNum = index;
}
......@@ -317,7 +320,7 @@ static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
pShow->numOfRows = mgmtGetMnodesNum();
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
pShow->pNode = NULL;
pShow->pIter = NULL;
mgmtDecUserRef(pUser);
return 0;
......@@ -330,7 +333,7 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
char *pWrite;
while (numOfRows < rows) {
pShow->pNode = mgmtGetNextMnode(pShow->pNode, &pMnode);
pShow->pIter = mgmtGetNextMnode(pShow->pIter, &pMnode);
if (pMnode == NULL) break;
cols = 0;
......
src/mnode/src/mgmtProfile.c
浏览文件 @ 5e7f8430
......@@ -140,7 +140,7 @@ int32_t mgmtGetQueries(SShowObj *pShow, void *pConn) {
//
// // sorting based on useconds
//
// pShow->pNode = pQueryShow;
// pShow->pIter = pQueryShow;
return 0;
}
......@@ -187,7 +187,7 @@ int32_t mgmtGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
for (int32_t i = 1; i < cols; ++i) pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
pShow->numOfRows = 1000000;
pShow->pNode = NULL;
pShow->pIter = NULL;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
mgmtGetQueries(pShow, pConn);
......@@ -252,7 +252,7 @@ int32_t mgmtRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, void *pCo
char *pWrite;
int32_t cols = 0;
SQueryShow *pQueryShow = (SQueryShow *)pShow->pNode;
SQueryShow *pQueryShow = (SQueryShow *)pShow->pIter;
if (rows > pQueryShow->numOfQueries - pQueryShow->index) rows = pQueryShow->numOfQueries - pQueryShow->index;
......@@ -339,7 +339,7 @@ int32_t mgmtGetStreams(SShowObj *pShow, void *pConn) {
//
// // sorting based on useconds
//
// pShow->pNode = pStreamShow;
// pShow->pIter = pStreamShow;
return 0;
}
......@@ -397,7 +397,7 @@ int32_t mgmtGetStreamMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
for (int32_t i = 1; i < cols; ++i) pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
pShow->numOfRows = 1000000;
pShow->pNode = NULL;
pShow->pIter = NULL;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
mgmtGetStreams(pShow, pConn);
......@@ -409,7 +409,7 @@ int32_t mgmtRetrieveStreams(SShowObj *pShow, char *data, int32_t rows, void *pCo
char *pWrite;
int32_t cols = 0;
SStreamShow *pStreamShow = (SStreamShow *)pShow->pNode;
SStreamShow *pStreamShow = (SStreamShow *)pShow->pIter;
if (rows > pStreamShow->numOfStreams - pStreamShow->index) rows = pStreamShow->numOfStreams - pStreamShow->index;
......@@ -592,7 +592,7 @@ int mgmtGetConns(SShowObj *pShow, void *pConn) {
//
// // sorting based on useconds
//
// pShow->pNode = pConnShow;
// pShow->pIter = pConnShow;
return 0;
}
......@@ -627,7 +627,7 @@ int32_t mgmtGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
for (int i = 1; i < cols; ++i) pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
pShow->numOfRows = 1000000;
pShow->pNode = NULL;
pShow->pIter = NULL;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
mgmtGetConns(pShow, pConn);
......@@ -639,7 +639,7 @@ int32_t mgmtRetrieveConns(SShowObj *pShow, char *data, int32_t rows, void *pConn
char *pWrite;
int32_t cols = 0;
SConnShow *pConnShow = (SConnShow *)pShow->pNode;
SConnShow *pConnShow = (SConnShow *)pShow->pIter;
if (rows > pConnShow->numOfConns - pConnShow->index) rows = pConnShow->numOfConns - pConnShow->index;
......
src/mnode/src/mgmtSdb.c
浏览文件 @ 5e7f8430
......@@ -16,6 +16,7 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "taoserror.h"
#include "hash.h"
#include "trpc.h"
#include "tutil.h"
#include "tbalance.h"
......@@ -23,8 +24,6 @@
#include "twal.h"
#include "tsync.h"
#include "tglobal.h"
#include "hashint.h"
#include "hashstr.h"
#include "dnode.h"
#include "mgmtDef.h"
#include "mgmtInt.h"
......@@ -83,12 +82,6 @@ typedef struct {
} SSdbRow;
static SSdbObject tsSdbObj = {0};
static void *(*sdbInitIndexFp[])(int32_t maxRows, int32_t dataSize) = {sdbOpenStrHash, sdbOpenIntHash, sdbOpenIntHash};
static void *(*sdbAddIndexFp[])(void *handle, void *key, void *data) = {sdbAddStrHash, sdbAddIntHash, sdbAddIntHash};
static void (*sdbDeleteIndexFp[])(void *handle, void *key) = {sdbDeleteStrHash, sdbDeleteIntHash, sdbDeleteIntHash};
static void *(*sdbGetIndexFp[])(void *handle, void *key) = {sdbGetStrHashData, sdbGetIntHashData, sdbGetIntHashData};
static void (*sdbCleanUpIndexFp[])(void *handle) = {sdbCloseStrHash, sdbCloseIntHash, sdbCloseIntHash};
static void *(*sdbFetchRowFp[])(void *handle, void *ptr, void **ppRow) = {sdbFetchStrHashData, sdbFetchIntHashData, sdbFetchIntHashData};
static int sdbWrite(void *param, void *data, int type);
int32_t sdbGetId(void *handle) {
......@@ -244,10 +237,10 @@ void sdbUpdateSync() {
}
if (index == 0) {
void *pNode = NULL;
void *pIter = NULL;
while (1) {
SMnodeObj *pMnode = NULL;
pNode = mgmtGetNextMnode(pNode, &pMnode);
pIter = mgmtGetNextMnode(pIter, &pMnode);
if (pMnode == NULL) break;
syncCfg.nodeInfo[index].nodeId = pMnode->mnodeId;
......@@ -257,6 +250,7 @@ void sdbUpdateSync() {
mgmtDecMnodeRef(pMnode);
}
sdbFreeIter(pIter);
}
syncCfg.replica = index;
......@@ -375,7 +369,11 @@ static SSdbRow *sdbGetRowMeta(void *handle, void *key) {
if (handle == NULL) return NULL;
pMeta = (*sdbGetIndexFp[pTable->keyType])(pTable->iHandle, key);
int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) {
keySize = strlen((char *)key);
}
pMeta = taosHashGet(pTable->iHandle, key, keySize);
return pMeta;
}
......@@ -387,13 +385,17 @@ void *sdbGetRow(void *handle, void *key) {
if (handle == NULL) return NULL;
pthread_mutex_lock(&pTable->mutex);
pMeta = (*sdbGetIndexFp[pTable->keyType])(pTable->iHandle, key);
int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) {
keySize = strlen((char *)key);
}
pMeta = taosHashGet(pTable->iHandle, key, keySize);
if (pMeta) sdbIncRef(pTable, pMeta->row);
pthread_mutex_unlock(&pTable->mutex);
if (pMeta == NULL) {
return NULL;
}
if (pMeta == NULL) return NULL;
return pMeta->row;
}
......@@ -404,7 +406,13 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
rowMeta.row = pOper->pObj;
pthread_mutex_lock(&pTable->mutex);
(*sdbAddIndexFp[pTable->keyType])(pTable->iHandle, pOper->pObj, &rowMeta);
int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) {
keySize = strlen((char *)pOper->pObj);
}
taosHashPut(pTable->iHandle, pOper->pObj, keySize, &rowMeta, sizeof(SSdbRow));
sdbIncRef(pTable, pOper->pObj);
pTable->numOfRows++;
......@@ -427,7 +435,13 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
(*pTable->deleteFp)(pOper);
pthread_mutex_lock(&pTable->mutex);
(*sdbDeleteIndexFp[pTable->keyType])(pTable->iHandle, pOper->pObj);
int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) {
keySize = strlen((char *)pOper->pObj);
}
taosHashRemove(pTable->iHandle, pOper->pObj, keySize);
pTable->numOfRows--;
pthread_mutex_unlock(&pTable->mutex);
......@@ -637,18 +651,35 @@ int32_t sdbUpdateRow(SSdbOper *pOper) {
void *sdbFetchRow(void *handle, void *pNode, void **ppRow) {
SSdbTable *pTable = (SSdbTable *)handle;
SSdbRow * pMeta;
*ppRow = NULL;
if (pTable == NULL) return NULL;
pNode = (*sdbFetchRowFp[pTable->keyType])(pTable->iHandle, pNode, (void **)&pMeta);
if (pMeta == NULL) return NULL;
SHashMutableIterator *pIter = pNode;
if (pIter == NULL) {
pIter = taosHashCreateIter(pTable->iHandle);
}
if (!taosHashIterNext(pIter)) {
taosHashDestroyIter(pIter);
return NULL;
}
SSdbRow *pMeta = taosHashIterGet(pIter);
if (pMeta == NULL) {
taosHashDestroyIter(pIter);
return NULL;
}
*ppRow = pMeta->row;
sdbIncRef(handle, pMeta->row);
return pNode;
return pIter;
}
void sdbFreeIter(void *pIter) {
if (pIter != NULL) {
taosHashDestroyIter(pIter);
}
}
void *sdbOpenTable(SSdbTableDesc *pDesc) {
......@@ -669,10 +700,12 @@ void *sdbOpenTable(SSdbTableDesc *pDesc) {
pTable->decodeFp = pDesc->decodeFp;
pTable->destroyFp = pDesc->destroyFp;
pTable->restoredFp = pDesc->restoredFp;
if (sdbInitIndexFp[pTable->keyType] != NULL) {
pTable->iHandle = (*sdbInitIndexFp[pTable->keyType])(pTable->maxRowSize, sizeof(SSdbRow));
_hash_fn_t hashFp = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
if (pTable->keyType == SDB_KEY_STRING) {
hashFp = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
}
pTable->iHandle = taosHashInit(pTable->hashSessions, hashFp, true);
pthread_mutex_init(&pTable->mutex, NULL);
......@@ -688,11 +721,10 @@ void sdbCloseTable(void *handle) {
tsSdbObj.numOfTables--;
tsSdbObj.tableList[pTable->tableId] = NULL;
void *pNode = NULL;
while (1) {
SSdbRow *pMeta;
pNode = (*sdbFetchRowFp[pTable->keyType])(pTable->iHandle, pNode, (void **)&pMeta);
if (pMeta == NULL) break;
SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle);
while (taosHashIterNext(pIter)) {
SSdbRow *pMeta = taosHashIterGet(pIter);
if (pMeta == NULL) continue;
SSdbOper oper = {
.pObj = pMeta->row,
......@@ -702,9 +734,8 @@ void sdbCloseTable(void *handle) {
(*pTable->destroyFp)(&oper);
}
if (sdbCleanUpIndexFp[pTable->keyType]) {
(*sdbCleanUpIndexFp[pTable->keyType])(pTable->iHandle);
}
taosHashDestroyIter(pIter);
taosHashCleanup(pTable->iHandle);
pthread_mutex_destroy(&pTable->mutex);
......
src/mnode/src/mgmtShell.c
浏览文件 @ 5e7f8430
......@@ -48,6 +48,7 @@ static void mgmtProcessRetrieveMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessHeartBeatMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessConnectMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessUseMsg(SQueuedMsg *queuedMsg);
static void mgmtFreeShowObj(void *data);
void *tsMgmtTmr;
static void *tsMgmtTranQhandle = NULL;
......@@ -65,7 +66,7 @@ int32_t mgmtInitShell() {
tsMgmtTmr = taosTmrInit((tsMaxShellConns) * 3, 200, 3600000, "MND");
tsMgmtTranQhandle = taosInitScheduler(tsMaxShellConns, 1, "mnodeT");
tsQhandleCache = taosCacheInit(tsMgmtTmr, 10);
tsQhandleCache = taosCacheInitWithCb(tsMgmtTmr, 10, mgmtFreeShowObj);
return 0;
}
......@@ -476,10 +477,10 @@ void mgmtSendSimpleResp(void *thandle, int32_t code) {
bool mgmtCheckQhandle(uint64_t qhandle) {
void *pSaved = taosCacheAcquireByData(tsQhandleCache, (void *)qhandle);
if (pSaved == (void *)qhandle) {
mTrace("qhandle:%p is retrived", qhandle);
mTrace("show:%p, is retrieved", qhandle);
return true;
} else {
mTrace("qhandle:%p is already freed", qhandle);
mTrace("show:%p, is already released", qhandle);
return false;
}
}
......@@ -491,15 +492,21 @@ void* mgmtSaveQhandle(void *qhandle, int32_t size) {
void *newQhandle = taosCachePut(tsQhandleCache, key, qhandle, size, 60);
free(qhandle);
mTrace("qhandle:%p is saved", newQhandle);
mTrace("show:%p, is saved", newQhandle);
return newQhandle;
}
return NULL;
}
static void mgmtFreeShowObj(void *data) {
SShowObj *pShow = data;
sdbFreeIter(pShow->pIter);
mTrace("show:%p, is destroyed", pShow);
}
void mgmtFreeQhandle(void *qhandle, bool forceRemove) {
mTrace("qhandle:%p is freed", qhandle);
mTrace("show:%p, is released, force:%s", qhandle, forceRemove ? "true" : "false");
taosCacheRelease(tsQhandleCache, &qhandle, forceRemove);
}
......
src/mnode/src/mgmtTable.c
浏览文件 @ 5e7f8430
......@@ -247,25 +247,19 @@ static int32_t mgmtChildTableActionDecode(SSdbOper *pOper) {
}
static int32_t mgmtChildTableActionRestored() {
void *pNode = NULL;
void *pLastNode = NULL;
void *pIter = NULL;
SChildTableObj *pTable = NULL;
while (1) {
pLastNode = pNode;
mgmtDecTableRef(pTable);
pNode = mgmtGetNextChildTable(pNode, &pTable);
pIter = mgmtGetNextChildTable(pIter, &pTable);
if (pTable == NULL) break;
SDbObj *pDb = mgmtGetDbByTableId(pTable->info.tableId);
if (pDb == NULL) {
mError("ctable:%s, failed to get db, discard it", pTable->info.tableId);
SSdbOper desc = {0};
desc.type = SDB_OPER_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
SSdbOper desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .table = tsChildTableSdb};
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
mgmtDecDbRef(pDb);
......@@ -274,12 +268,8 @@ static int32_t mgmtChildTableActionRestored() {
if (pVgroup == NULL) {
mError("ctable:%s, failed to get vgId:%d sid:%d, discard it", pTable->info.tableId, pTable->vgId, pTable->sid);
pTable->vgId = 0;
SSdbOper desc = {0};
desc.type = SDB_OPER_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
SSdbOper desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .table = tsChildTableSdb};
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
mgmtDecVgroupRef(pVgroup);
......@@ -288,24 +278,16 @@ static int32_t mgmtChildTableActionRestored() {
mError("ctable:%s, db:%s not match with vgId:%d db:%s sid:%d, discard it",
pTable->info.tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->sid);
pTable->vgId = 0;
SSdbOper desc = {0};
desc.type = SDB_OPER_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
SSdbOper desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .table = tsChildTableSdb};
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
if (pVgroup->tableList == NULL) {
mError("ctable:%s, vgId:%d tableList is null", pTable->info.tableId, pTable->vgId);
pTable->vgId = 0;
SSdbOper desc = {0};
desc.type = SDB_OPER_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
SSdbOper desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .table = tsChildTableSdb};
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
......@@ -314,18 +296,16 @@ static int32_t mgmtChildTableActionRestored() {
if (pSuperTable == NULL) {
mError("ctable:%s, stable:%" PRIu64 " not exist", pTable->info.tableId, pTable->suid);
pTable->vgId = 0;
SSdbOper desc = {0};
desc.type = SDB_OPER_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
SSdbOper desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .table = tsChildTableSdb};
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
mgmtDecTableRef(pSuperTable);
}
}
sdbFreeIter(pIter);
return 0;
}
......@@ -560,10 +540,10 @@ static void *mgmtGetSuperTable(char *tableId) {
static void *mgmtGetSuperTableByUid(uint64_t uid) {
SSuperTableObj *pStable = NULL;
void * pNode = NULL;
void *pIter = NULL;
while (1) {
pNode = mgmtGetNextSuperTable(pNode, &pStable);
pIter = mgmtGetNextSuperTable(pIter, &pStable);
if (pStable == NULL) break;
if (pStable->uid == uid) {
return pStable;
......@@ -571,6 +551,8 @@ static void *mgmtGetSuperTableByUid(uint64_t uid) {
mgmtDecTableRef(pStable);
}
sdbFreeIter(pIter);
return NULL;
}
......@@ -588,12 +570,12 @@ void *mgmtGetTable(char *tableId) {
return NULL;
}
void *mgmtGetNextChildTable(void *pNode, SChildTableObj **pTable) {
return sdbFetchRow(tsChildTableSdb, pNode, (void **)pTable);
void *mgmtGetNextChildTable(void *pIter, SChildTableObj **pTable) {
return sdbFetchRow(tsChildTableSdb, pIter, (void **)pTable);
}
void *mgmtGetNextSuperTable(void *pNode, SSuperTableObj **pTable) {
return sdbFetchRow(tsSuperTableSdb, pNode, (void **)pTable);
void *mgmtGetNextSuperTable(void *pIter, SSuperTableObj **pTable) {
return sdbFetchRow(tsSuperTableSdb, pIter, (void **)pTable);
}
void mgmtIncTableRef(void *p1) {
......@@ -812,6 +794,7 @@ static void mgmtProcessDropSuperTableMsg(SQueuedMsg *pMsg) {
dnodeSendMsgToDnode(&ipSet, &rpcMsg);
mgmtDecVgroupRef(pVgroup);
}
taosHashDestroyIter(pIter);
mgmtDropAllChildTablesInStable(pStable);
}
......@@ -1121,7 +1104,7 @@ int32_t mgmtRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows, v
while (numOfRows < rows) {
mgmtDecTableRef(pTable);
pShow->pNode = mgmtGetNextSuperTable(pShow->pNode, &pTable);
pShow->pIter = mgmtGetNextSuperTable(pShow->pIter, &pTable);
if (pTable == NULL) break;
if (strncmp(pTable->info.tableId, prefix, prefixLen)) {
continue;
......@@ -1171,8 +1154,7 @@ int32_t mgmtRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows, v
}
void mgmtDropAllSuperTables(SDbObj *pDropDb) {
void *pNode = NULL;
void *pLastNode = NULL;
void * pIter= NULL;
int32_t numOfTables = 0;
int32_t dbNameLen = strlen(pDropDb->name);
SSuperTableObj *pTable = NULL;
......@@ -1180,8 +1162,7 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
mPrint("db:%s, all super tables will be dropped from sdb", pDropDb->name);
while (1) {
pLastNode = pNode;
pNode = mgmtGetNextSuperTable(pNode, &pTable);
pIter = mgmtGetNextSuperTable(pIter, &pTable);
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
......@@ -1191,13 +1172,14 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
.pObj = pTable,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables ++;
}
mgmtDecTableRef(pTable);
}
sdbFreeIter(pIter);
mPrint("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
......@@ -1290,6 +1272,8 @@ static void mgmtProcessSuperTableVgroupMsg(SQueuedMsg *pMsg) {
mgmtDecVgroupRef(pVgroup);
}
taosHashDestroyIter(pIter);
pVgroupInfo->numOfVgroups = htonl(vgSize);
// one table is done, try the next table
......@@ -1751,8 +1735,7 @@ static void mgmtGetChildTableMeta(SQueuedMsg *pMsg) {
}
void mgmtDropAllChildTables(SDbObj *pDropDb) {
void *pNode = NULL;
void *pLastNode = NULL;
void * pIter = NULL;
int32_t numOfTables = 0;
int32_t dbNameLen = strlen(pDropDb->name);
SChildTableObj *pTable = NULL;
......@@ -1760,8 +1743,7 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
mPrint("db:%s, all child tables will be dropped from sdb", pDropDb->name);
while (1) {
pLastNode = pNode;
pNode = mgmtGetNextChildTable(pNode, &pTable);
pIter = mgmtGetNextChildTable(pIter, &pTable);
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
......@@ -1771,26 +1753,25 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
.pObj = pTable,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
}
mgmtDecTableRef(pTable);
}
sdbFreeIter(pIter);
mPrint("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
void *pNode = NULL;
void *pLastNode = NULL;
void * pIter = NULL;
int32_t numOfTables = 0;
SChildTableObj *pTable = NULL;
mPrint("stable:%s, all child tables will dropped from sdb", pStable->info.tableId, numOfTables);
while (1) {
pLastNode = pNode;
pNode = mgmtGetNextChildTable(pNode, &pTable);
pIter = mgmtGetNextChildTable(pIter, &pTable);
if (pTable == NULL) break;
if (pTable->superTable == pStable) {
......@@ -1800,13 +1781,14 @@ static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
.pObj = pTable,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
}
mgmtDecTableRef(pTable);
}
sdbFreeIter(pIter);
mPrint("stable:%s, all child tables:%d is dropped from sdb", pStable->info.tableId, numOfTables);
}
......@@ -2077,7 +2059,7 @@ static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows,
while (numOfRows < rows) {
mgmtDecTableRef(pTable);
pShow->pNode = mgmtGetNextChildTable(pShow->pNode, &pTable);
pShow->pIter = mgmtGetNextChildTable(pShow->pIter, &pTable);
if (pTable == NULL) break;
// not belong to current db
......
src/mnode/src/mgmtUser.c
浏览文件 @ 5e7f8430
......@@ -155,8 +155,8 @@ SUserObj *mgmtGetUser(char *name) {
return (SUserObj *)sdbGetRow(tsUserSdb, name);
}
void *mgmtGetNextUser(void *pNode, SUserObj **pUser) {
return sdbFetchRow(tsUserSdb, pNode, (void **)pUser);
void *mgmtGetNextUser(void *pIter, SUserObj **pUser) {
return sdbFetchRow(tsUserSdb, pIter, (void **)pUser);
}
void mgmtIncUserRef(SUserObj *pUser) {
......@@ -300,7 +300,7 @@ static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void
char *pWrite;
while (numOfRows < rows) {
pShow->pNode = mgmtGetNextUser(pShow->pNode, &pUser);
pShow->pIter = mgmtGetNextUser(pShow->pIter, &pUser);
if (pUser == NULL) break;
cols = 0;
......@@ -504,15 +504,13 @@ static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
}
void mgmtDropAllUsers(SAcctObj *pAcct) {
void * pNode = NULL;
void * pLastNode = NULL;
void * pIter = NULL;
int32_t numOfUsers = 0;
int32_t acctNameLen = strlen(pAcct->user);
SUserObj *pUser = NULL;
while (1) {
pLastNode = pNode;
pNode = mgmtGetNextUser(pNode, &pUser);
pIter = mgmtGetNextUser(pIter, &pUser);
if (pUser == NULL) break;
if (strncmp(pUser->acct, pAcct->user, acctNameLen) == 0) {
......@@ -522,13 +520,14 @@ void mgmtDropAllUsers(SAcctObj *pAcct) {
.pObj = pUser,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfUsers++;
}
mgmtDecUserRef(pUser);
}
sdbFreeIter(pIter);
mTrace("acct:%s, all users:%d is dropped from sdb", pAcct->user, numOfUsers);
}
......
src/mnode/src/mgmtVgroup.c
浏览文件 @ 5e7f8430
......@@ -288,8 +288,8 @@ SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb) {
return pDb->pHead;
}
void *mgmtGetNextVgroup(void *pNode, SVgObj **pVgroup) {
return sdbFetchRow(tsVgroupSdb, pNode, (void **)pVgroup);
void *mgmtGetNextVgroup(void *pIter, SVgObj **pVgroup) {
return sdbFetchRow(tsVgroupSdb, pIter, (void **)pVgroup);
}
void mgmtCreateVgroup(SQueuedMsg *pMsg, SDbObj *pDb) {
......@@ -429,10 +429,10 @@ int32_t mgmtGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
if (NULL == pTable) {
pShow->numOfRows = pDb->numOfVgroups;
pShow->pNode = pDb->pHead;
pShow->pIter = pDb->pHead;
} else {
pShow->numOfRows = 1;
pShow->pNode = pVgroup;
pShow->pIter = pVgroup;
}
mgmtDecDbRef(pDb);
......@@ -457,9 +457,9 @@ int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pCo
}
while (numOfRows < rows) {
pVgroup = (SVgObj *) pShow->pNode;
pVgroup = (SVgObj *) pShow->pIter;
if (pVgroup == NULL) break;
pShow->pNode = (void *) pVgroup->next;
pShow->pIter = (void *) pVgroup->next;
cols = 0;
......@@ -749,14 +749,12 @@ static void mgmtProcessVnodeCfgMsg(SRpcMsg *rpcMsg) {
}
void mgmtDropAllDnodeVgroups(SDnodeObj *pDropDnode) {
void * pNode = NULL;
void * pLastNode = NULL;
void * pIter = NULL;
SVgObj *pVgroup = NULL;
int32_t numOfVgroups = 0;
while (1) {
pLastNode = pNode;
pNode = mgmtGetNextVgroup(pNode, &pVgroup);
pIter = mgmtGetNextVgroup(pIter, &pVgroup);
if (pVgroup == NULL) break;
if (pVgroup->vnodeGid[0].dnodeId == pDropDnode->dnodeId) {
......@@ -766,24 +764,23 @@ void mgmtDropAllDnodeVgroups(SDnodeObj *pDropDnode) {
.pObj = pVgroup,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfVgroups++;
continue;
}
mgmtDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
}
void mgmtDropAllDbVgroups(SDbObj *pDropDb, bool sendMsg) {
void *pNode = NULL;
void *pLastNode = NULL;
void * pIter = NULL;
int32_t numOfVgroups = 0;
SVgObj *pVgroup = NULL;
mPrint("db:%s, all vgroups will be dropped from sdb", pDropDb->name);
while (1) {
pLastNode = pNode;
pNode = mgmtGetNextVgroup(pNode, &pVgroup);
pIter = mgmtGetNextVgroup(pIter, &pVgroup);
if (pVgroup == NULL) break;
if (pVgroup->pDb == pDropDb) {
......@@ -793,7 +790,6 @@ void mgmtDropAllDbVgroups(SDbObj *pDropDb, bool sendMsg) {
.pObj = pVgroup,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfVgroups++;
if (sendMsg) {
......@@ -804,5 +800,7 @@ void mgmtDropAllDbVgroups(SDbObj *pDropDb, bool sendMsg) {
mgmtDecVgroupRef(pVgroup);
}
sdbFreeIter(pIter);
mPrint("db:%s, all vgroups:%d is dropped from sdb", pDropDb->name, numOfVgroups);
}
src/plugins/http/src/httpHandle.c
浏览文件 @ 5e7f8430
......@@ -19,7 +19,6 @@
#include "tglobal.h"
#include "tsocket.h"
#include "ttimer.h"
#include "shash.h"
#include "http.h"
#include "httpLog.h"
#include "httpCode.h"
......
src/plugins/http/src/httpSession.c
浏览文件 @ 5e7f8430
......@@ -15,7 +15,7 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "shash.h"
#include "hash.h"
#include "taos.h"
#include "ttime.h"
#include "ttimer.h"
......@@ -25,7 +25,6 @@
#include "httpHandle.h"
#include "httpResp.h"
void httpAccessSession(HttpContext *pContext) {
HttpServer *server = pContext->pThread->pServer;
pthread_mutex_lock(&server->serverMutex);
......@@ -50,8 +49,11 @@ void httpCreateSession(HttpContext *pContext, void *taos) {
session.taos = taos;
session.expire = (int)taosGetTimestampSec() + server->sessionExpire;
session.access = 1;
snprintf(session.id, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass);
pContext->session = (HttpSession *)taosAddStrHash(server->pSessionHash, session.id, (char *)(&session));
int sessionIdLen = snprintf(session.id, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass);
taosHashPut(server->pSessionHash, session.id, sessionIdLen, (char *)(&session), sizeof(HttpSession));
pContext->session = taosHashGet(server->pSessionHash, session.id, sessionIdLen);
if (pContext->session == NULL) {
httpError("context:%p, fd:%d, ip:%s, user:%s, error:%s", pContext, pContext->fd, pContext->ipstr, pContext->user,
httpMsg[HTTP_SESSION_FULL]);
......@@ -71,9 +73,9 @@ void httpFetchSessionImp(HttpContext *pContext) {
pthread_mutex_lock(&server->serverMutex);
char sessionId[HTTP_SESSION_ID_LEN];
snprintf(sessionId, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass);
int sessonIdLen = snprintf(sessionId, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass);
pContext->session = (HttpSession *)taosGetStrHashData(server->pSessionHash, sessionId);
pContext->session = taosHashGet(server->pSessionHash, sessionId, sessonIdLen);
if (pContext->session != NULL && pContext->session == pContext->session->signature) {
pContext->session->access++;
httpTrace("context:%p, fd:%d, ip:%s, user:%s, find an exist session:%p:%p, access:%d, expire:%d",
......@@ -120,8 +122,7 @@ void httpRestoreSession(HttpContext *pContext) {
pthread_mutex_unlock(&server->serverMutex);
}
void httpResetSession(char *session) {
HttpSession *pSession = (HttpSession *)session;
void httpResetSession(HttpSession *pSession) {
httpTrace("close session:%p:%p", pSession, pSession->taos);
if (pSession->taos != NULL) {
taos_close(pSession->taos);
......@@ -131,15 +132,20 @@ void httpResetSession(char *session) {
}
void httpRemoveAllSessions(HttpServer *pServer) {
if (pServer->pSessionHash != NULL) {
taosCleanUpStrHashWithFp(pServer->pSessionHash, httpResetSession);
pServer->pSessionHash = NULL;
SHashMutableIterator *pIter = taosHashCreateIter(pServer->pSessionHash);
while (taosHashIterNext(pIter)) {
HttpSession *pSession = taosHashIterGet(pIter);
if (pSession == NULL) continue;
httpResetSession(pSession);
}
taosHashDestroyIter(pIter);
}
bool httpInitAllSessions(HttpServer *pServer) {
if (pServer->pSessionHash == NULL) {
pServer->pSessionHash = taosInitStrHash(100, sizeof(HttpSession), taosHashStringStep1);
pServer->pSessionHash = taosHashInit(10, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true);
}
if (pServer->pSessionHash == NULL) {
httpError("http init session pool failed");
......@@ -152,46 +158,41 @@ bool httpInitAllSessions(HttpServer *pServer) {
return true;
}
int httpSessionExpired(char *session) {
HttpSession *pSession = (HttpSession *)session;
time_t cur = taosGetTimestampSec();
bool httpSessionExpired(HttpSession *pSession) {
time_t cur = taosGetTimestampSec();
if (pSession->taos != NULL) {
if (pSession->expire > cur) {
return 0; // un-expired, so return false
return false; // un-expired, so return false
}
if (pSession->access > 0) {
httpTrace("session:%p:%p is expired, but still access:%d", pSession, pSession->taos,
pSession->access);
return 0; // still used, so return false
return false; // still used, so return false
}
httpTrace("need close session:%p:%p for it expired, cur:%d, expire:%d, invertal:%d",
pSession, pSession->taos, cur, pSession->expire, cur - pSession->expire);
}
return 1;
return true;
}
void httpRemoveExpireSessions(HttpServer *pServer) {
int expiredNum = 0;
do {
pthread_mutex_lock(&pServer->serverMutex);
void httpRemoveExpireSessions(HttpServer *pServer) {
SHashMutableIterator *pIter = taosHashCreateIter(pServer->pSessionHash);
HttpSession *pSession = (HttpSession *)taosVisitStrHashWithFp(pServer->pSessionHash, httpSessionExpired);
if (pSession == NULL) {
pthread_mutex_unlock(&pServer->serverMutex);
break;
}
httpResetSession((char *)pSession);
taosDeleteStrHashNode(pServer->pSessionHash, pSession->id, pSession);
while (taosHashIterNext(pIter)) {
HttpSession *pSession = taosHashIterGet(pIter);
if (pSession == NULL) continue;
pthread_mutex_lock(&pServer->serverMutex);
if (httpSessionExpired(pSession)) {
httpResetSession(pSession);
taosHashRemove(pServer->pSessionHash, pSession->id, strlen(pSession->id));
}
pthread_mutex_unlock(&pServer->serverMutex);
}
if (++expiredNum > 10) {
break;
}
} while (true);
taosHashDestroyIter(pIter);
}
void httpProcessSessionExpire(void *handle, void *tmrId) {
......
src/plugins/http/src/httpSystem.c
浏览文件 @ 5e7f8430
......@@ -15,7 +15,6 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "shash.h"
#include "taos.h"
#include "tglobal.h"
#include "tsocket.h"
......
src/plugins/http/src/httpUtil.c
浏览文件 @ 5e7f8430
......@@ -16,7 +16,6 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "tmd5.h"
#include "shash.h"
#include "taos.h"
#include "http.h"
#include "httpLog.h"
......
src/plugins/http/src/tgHandle.c
浏览文件 @ 5e7f8430
......@@ -16,7 +16,6 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "tglobal.h"
#include "shash.h"
#include "taosdef.h"
#include "taosmsg.h"
#include "tgHandle.h"
......
src/query/src/tvariant.c
浏览文件 @ 5e7f8430
......@@ -17,7 +17,7 @@
#include "hash.h"
#include "hashfunc.h"
#include "os.h"
#include "shash.h"
#include "hash.h"
#include "taos.h"
#include "taosdef.h"
#include "tstoken.h"
......
src/rpc/src/rpcUdp.c
浏览文件 @ 5e7f8430
......@@ -18,7 +18,6 @@
#include "tsystem.h"
#include "ttimer.h"
#include "tutil.h"
#include "thash.h"
#include "rpcLog.h"
#include "rpcHaship.h"
#include "rpcUdp.h"
......
src/util/inc/hashint.h 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#ifndef _sdb_int_hash_header_
#define _sdb_int_hash_header_
void *sdbOpenIntHash(int maxSessions, int dataSize);
void sdbCloseIntHash(void *handle);
void *sdbAddIntHash(void *handle, void *key, void *pData);
void sdbDeleteIntHash(void *handle, void *key);
void *sdbGetIntHashData(void *handle, void *key);
void *sdbFetchIntHashData(void *handle, void *ptr, void **ppMeta);
#endif
src/util/inc/hashstr.h 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#ifndef _sdb_str_hash_header_
#define _sdb_str_hash_header_
void *sdbOpenStrHash(int maxSessions, int dataSize);
void sdbCloseStrHash(void *handle);
void *sdbAddStrHash(void *handle, void *key, void *pData);
void sdbDeleteStrHash(void *handle, void *key);
void *sdbGetStrHashData(void *handle, void *key);
void *sdbFetchStrHashData(void *handle, void *ptr, void **ppMeta);
#endif
src/util/inc/ihash.h 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#ifndef TDENGINE_IHASH_H
#define TDENGINE_IHASH_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
void *taosInitIntHash(int32_t maxSessions, int32_t dataSize, int32_t (*fp)(void *, uint64_t));
void taosCleanUpIntHash(void *handle);
char *taosGetIntHashData(void *handle, uint64_t key);
void taosDeleteIntHash(void *handle, uint64_t key);
char *taosAddIntHash(void *handle, uint64_t key, char *pData);
int32_t taosHashInt(void *handle, uint64_t key);
void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *));
void taosVisitIntHashWithFp(void *handle, void (*fp)(char *, void *), void *param);
int32_t taosGetIntHashSize(void *handle);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_IHASH_H
src/util/inc/shash.h 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#ifndef TDENGINE_TSHASH_H
#define TDENGINE_TSHASH_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
void *taosInitStrHash(uint32_t maxSessions, uint32_t dataSize, uint32_t (*fp)(void *, char *));
void taosCleanUpStrHash(void *handle);
void *taosGetStrHashData(void *handle, char *string);
void taosDeleteStrHash(void *handle, char *string);
void taosDeleteStrHashNode(void *handle, char *string, void *pDeleteNode);
void *taosAddStrHash(void *handle, char *string, char *pData);
void *taosAddStrHashWithSize(void *handle, char *string, char *pData, int dataSize);
uint32_t taosHashString(void *handle, char *string);
uint32_t taosHashStringStep1(void *handle, char *string);
char *taosVisitStrHashWithFp(void *handle, int (*fp)(char *));
void taosCleanUpStrHashWithFp(void *handle, void (*fp)(char *));
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TSHASH_H
src/util/inc/tcache.h
浏览文件 @ 5e7f8430
......@@ -67,6 +67,7 @@ typedef struct {
void * pTimer;
SCacheStatis statistics;
SHashObj * pHashTable;
_hash_free_fn_t freeFp;
int numOfElemsInTrash; // number of element in trash
int16_t deleting; // set the deleting flag to stop refreshing ASAP.
T_REF_DECLARE()
......@@ -88,6 +89,7 @@ typedef struct {
* @return
*/
SCacheObj *taosCacheInit(void *tmrCtrl, int64_t refreshTimeInSeconds);
SCacheObj *taosCacheInitWithCb(void *tmrCtrl, int64_t refreshTimeInSeconds, void (*freeCb)(void *data));
/**
* add data into cache
......
src/util/inc/thash.h 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#ifndef TDENGINE_THASH_H
#define TDENGINE_THASH_H
#ifdef __cplusplus
extern "C" {
#endif
void *taosOpenHash(int maxSessions, int (*fp)(void *, uint64_t));
void taosCloseHash(void *handle);
int taosAddHash(void *handle, uint64_t, uint32_t id);
void taosDeleteHash(void *handle, uint64_t);
int32_t taosGetIdFromHash(void *handle, uint64_t);
int taosHashLong(void *, uint64_t ip);
uint64_t taosHashUInt64(uint64_t handle);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_THASH_H
src/util/src/hashint.c 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#include <pthread.h>
#include <semaphore.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "tmempool.h"
#include "taosdef.h"
typedef struct _long_hash_t {
uint32_t key;
int hash;
struct _long_hash_t *prev;
struct _long_hash_t *next;
char data[];
} SLongHash;
typedef struct {
SLongHash **longHashList;
mpool_h longHashMemPool;
int maxSessions;
int dataSize;
} SHashObj;
int sdbHashLong(void *handle, uint32_t ip) {
SHashObj *pObj = (SHashObj *)handle;
int hash = 0;
hash = ip >> 16;
hash += (ip & 0xFFFF);
hash = hash % pObj->maxSessions;
return hash;
}
void *sdbAddIntHash(void *handle, void *pKey, void *data) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
uint32_t key = *((uint32_t *)pKey);
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashLong(pObj, key);
pNode = (SLongHash *)taosMemPoolMalloc(pObj->longHashMemPool);
pNode->key = key;
memcpy(pNode->data, data, pObj->dataSize);
pNode->prev = 0;
pNode->next = pObj->longHashList[hash];
pNode->hash = hash;
if (pObj->longHashList[hash] != 0) (pObj->longHashList[hash])->prev = pNode;
pObj->longHashList[hash] = pNode;
return pObj;
}
void sdbDeleteIntHash(void *handle, void *pKey) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
uint32_t key = *((uint32_t *)pKey);
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
hash = sdbHashLong(pObj, key);
pNode = pObj->longHashList[hash];
while (pNode) {
if (pNode->key == key) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->longHashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
taosMemPoolFree(pObj->longHashMemPool, (char *)pNode);
}
}
void *sdbGetIntHashData(void *handle, void *pKey) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
uint32_t key = *((uint32_t *)pKey);
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashLong(pObj, key);
pNode = pObj->longHashList[hash];
while (pNode) {
if (pNode->key == key) {
break;
}
pNode = pNode->next;
}
if (pNode) return pNode->data;
return NULL;
}
void *sdbOpenIntHash(int maxSessions, int dataSize) {
SLongHash **longHashList;
mpool_h longHashMemPool;
SHashObj * pObj;
longHashMemPool = taosMemPoolInit(maxSessions, sizeof(SLongHash) + dataSize);
if (longHashMemPool == 0) return NULL;
longHashList = calloc(sizeof(SLongHash *), maxSessions);
if (longHashList == 0) {
taosMemPoolCleanUp(longHashMemPool);
return NULL;
}
pObj = malloc(sizeof(SHashObj));
if (pObj == NULL) {
taosMemPoolCleanUp(longHashMemPool);
free(longHashList);
return NULL;
}
pObj->maxSessions = maxSessions;
pObj->longHashMemPool = longHashMemPool;
pObj->longHashList = longHashList;
pObj->dataSize = dataSize;
return pObj;
}
void sdbCloseIntHash(void *handle) {
SHashObj *pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
if (pObj->longHashMemPool) taosMemPoolCleanUp(pObj->longHashMemPool);
if (pObj->longHashList) free(pObj->longHashList);
memset(pObj, 0, sizeof(SHashObj));
free(pObj);
}
void *sdbFetchIntHashData(void *handle, void *ptr, void **ppMeta) {
SHashObj * pObj = (SHashObj *)handle;
SLongHash *pNode = (SLongHash *)ptr;
int hash = 0;
*ppMeta = NULL;
if (pObj == NULL || pObj->maxSessions <= 0) return NULL;
if (pObj->longHashList == NULL) return NULL;
if (pNode) {
hash = pNode->hash + 1;
pNode = pNode->next;
}
if (pNode == NULL) {
for (int i = hash; i < pObj->maxSessions; ++i) {
pNode = pObj->longHashList[i];
if (pNode) break;
}
}
if (pNode) *ppMeta = pNode->data;
return pNode;
}
src/util/src/hashstr.c 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#include "os.h"
#include "taosdef.h"
#define MAX_STR_LEN 40
typedef struct _str_node_t {
char string[TSDB_TABLE_ID_LEN];
int hash;
struct _str_node_t *prev;
struct _str_node_t *next;
char data[];
} SHashNode;
typedef struct {
SHashNode **hashList;
int maxSessions;
int dataSize;
} SHashObj;
int sdbHashString(void *handle, char *string) {
SHashObj * pObj = (SHashObj *)handle;
unsigned int hash = 0, hashv;
char * c;
int len = strlen(string);
c = string;
while (len >= 4) {
hash += *((int *)c);
c += 4;
len -= 4;
}
while (len > 0) {
hash += *c;
c++;
len--;
}
hashv = hash / pObj->maxSessions;
hash = (hashv + hash % pObj->maxSessions) % pObj->maxSessions;
return hash;
}
void *sdbAddStrHash(void *handle, void *key, void *pData) {
int hash;
SHashNode *pNode;
SHashObj * pObj;
char * string = (char *)key;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashString(pObj, string);
int size = sizeof(SHashNode) + pObj->dataSize;
pNode = (SHashNode *)malloc(size);
memset(pNode, 0, size);
strcpy(pNode->string, string);
memcpy(pNode->data, pData, pObj->dataSize);
pNode->prev = 0;
pNode->next = pObj->hashList[hash];
pNode->hash = hash;
if (pObj->hashList[hash] != 0) (pObj->hashList[hash])->prev = pNode;
pObj->hashList[hash] = pNode;
return pNode->data;
}
void sdbDeleteStrHash(void *handle, void *key) {
int hash;
SHashNode *pNode;
SHashObj * pObj;
char * string = (char *)key;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
hash = sdbHashString(pObj, string);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
memset(pNode, 0, sizeof(SHashNode));
free(pNode);
}
}
void *sdbGetStrHashData(void *handle, void *key) {
int hash;
SHashNode *pNode;
SHashObj * pObj;
char * string = (char *)key;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashString(pObj, string);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) {
break;
}
pNode = pNode->next;
}
if (pNode) return pNode->data;
return NULL;
}
void *sdbOpenStrHash(int maxSessions, int dataSize) {
SHashObj *pObj;
pObj = (SHashObj *)malloc(sizeof(SHashObj));
if (pObj == NULL) {
return NULL;
}
memset(pObj, 0, sizeof(SHashObj));
pObj->maxSessions = maxSessions;
pObj->dataSize = dataSize;
pObj->hashList = (SHashNode **)malloc(sizeof(SHashNode *) * maxSessions);
if (pObj->hashList == NULL) {
free(pObj);
return NULL;
}
memset(pObj->hashList, 0, sizeof(SHashNode *) * maxSessions);
return (void *)pObj;
}
void sdbCloseStrHash(void *handle) {
SHashObj *pObj;
SHashNode *pNode, *pNext;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
free(pNode);
pNode = pNext;
}
}
free(pObj->hashList);
}
memset(pObj, 0, sizeof(SHashObj));
free(pObj);
}
void *sdbFetchStrHashData(void *handle, void *ptr, void **ppMeta) {
SHashObj *pObj = (SHashObj *)handle;
SHashNode *pNode = (SHashNode *)ptr;
int hash = 0;
*ppMeta = NULL;
if (pObj == NULL || pObj->maxSessions <= 0) return NULL;
if (pObj->hashList == NULL) return NULL;
if (pNode) {
hash = pNode->hash + 1;
pNode = pNode->next;
}
if (pNode == NULL) {
for (int i = hash; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
if (pNode) break;
}
}
if (pNode) *ppMeta = pNode->data;
return pNode;
}
src/util/src/ihash.c 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#include "os.h"
typedef struct _str_node_t {
uint64_t key;
struct _str_node_t *prev;
struct _str_node_t *next;
char data[];
} IHashNode;
typedef struct {
IHashNode **hashList;
int64_t *lockedBy;
int32_t maxSessions;
int32_t dataSize;
int32_t (*hashFp)(void *, uint64_t key);
} IHashObj;
int32_t taosHashInt(void *handle, uint64_t key) {
IHashObj *pObj = (IHashObj *)handle;
int32_t hash = key % pObj->maxSessions;
return hash;
}
static void taosLockIntHash(IHashObj *pObj, int hash);
static void taosUnlockIntHash(IHashObj *pObj, int hash);
char *taosAddIntHash(void *handle, uint64_t key, char *pData) {
int32_t hash;
IHashNode *pNode;
IHashObj * pObj;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = (*pObj->hashFp)(pObj, key);
pNode = (IHashNode *)malloc(sizeof(IHashNode) + (size_t)pObj->dataSize);
if (pNode == NULL)
return NULL;
taosLockIntHash(pObj, hash);
pNode->key = key;
if (pData != NULL) {
memcpy(pNode->data, pData, (size_t)pObj->dataSize);
}
pNode->prev = 0;
pNode->next = pObj->hashList[hash];
if (pObj->hashList[hash] != 0) (pObj->hashList[hash])->prev = pNode;
pObj->hashList[hash] = pNode;
taosUnlockIntHash(pObj, hash);
return (char *)pNode->data;
}
void taosDeleteIntHash(void *handle, uint64_t key) {
int32_t hash;
IHashNode *pNode;
IHashObj * pObj;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
hash = (*(pObj->hashFp))(pObj, key);
taosLockIntHash(pObj, hash);
pNode = pObj->hashList[hash];
while (pNode) {
if (pNode->key == key) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
free(pNode);
}
taosUnlockIntHash(pObj, hash);
}
char *taosGetIntHashData(void *handle, uint64_t key) {
int32_t hash;
IHashNode *pNode;
IHashObj * pObj;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = (*pObj->hashFp)(pObj, key);
taosLockIntHash(pObj, hash);
pNode = pObj->hashList[hash];
while (pNode) {
if (pNode->key == key) {
break;
}
pNode = pNode->next;
}
taosUnlockIntHash(pObj, hash);
if (pNode) return pNode->data;
return NULL;
}
void *taosInitIntHash(int32_t maxSessions, int32_t dataSize, int32_t (*fp)(void *, uint64_t)) {
IHashObj *pObj;
pObj = (IHashObj *)malloc(sizeof(IHashObj));
if (pObj == NULL) {
return NULL;
}
memset(pObj, 0, sizeof(IHashObj));
pObj->maxSessions = maxSessions;
pObj->dataSize = dataSize;
pObj->hashFp = fp;
pObj->hashList = (IHashNode **)malloc(sizeof(IHashNode *) * (size_t)maxSessions);
if (pObj->hashList == NULL) {
free(pObj);
return NULL;
}
memset(pObj->hashList, 0, sizeof(IHashNode *) * (size_t)maxSessions);
pObj->lockedBy = (int64_t *)calloc(sizeof(int64_t), maxSessions);
if (pObj->lockedBy == NULL) {
free(pObj);
free(pObj->hashList);
pObj = NULL;
}
return pObj;
}
void taosCleanUpIntHash(void *handle) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int32_t i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
free(pNode);
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
free(pObj->hashList);
}
free(pObj->lockedBy);
free(pObj);
}
void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
if (fp != NULL) (*fp)(pNode->data);
free(pNode);
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
free(pObj->hashList);
}
memset(pObj, 0, sizeof(IHashObj));
free(pObj);
}
void taosVisitIntHashWithFp(void *handle, int (*fp)(char *, void *), void *param) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
(*fp)(pNode->data, param);
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
}
}
int32_t taosGetIntHashSize(void *handle) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
int32_t num = 0;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return 0;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
num++;
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
}
return num;
}
static void taosLockIntHash(IHashObj *pObj, int hash) {
int64_t tid = taosGetPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(&(pObj->lockedBy[hash]), 0, tid) != 0) {
if (++i % 1000 == 0) {
sched_yield();
}
}
}
static void taosUnlockIntHash(IHashObj *pObj, int hash) {
int64_t tid = taosGetPthreadId();
if (atomic_val_compare_exchange_64(&(pObj->lockedBy[hash]), tid, 0) != tid) {
assert(false);
}
}
src/util/src/shash.c 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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 "shash.h"
#include "tulog.h"
typedef struct _str_node_t {
char * string;
struct _str_node_t *prev;
struct _str_node_t *next;
char data[];
} SHashNode;
typedef struct {
SHashNode **hashList;
uint32_t maxSessions;
uint32_t dataSize;
uint32_t (*hashFp)(void *, char *string);
pthread_mutex_t mutex;
} SHashObj;
uint32_t taosHashString(void *handle, char *string) {
SHashObj *pObj = (SHashObj *)handle;
uint32_t hash = 0, hashv;
char * c;
c = string;
while (*c) {
hash += *((int *)c);
c += 4;
}
hashv = hash / pObj->maxSessions;
hash = (hashv + hash % pObj->maxSessions) % pObj->maxSessions;
return hash;
}
uint32_t taosHashStringStep1(void *handle, char *string) {
SHashObj *pObj = (SHashObj *)handle;
uint32_t hash = 0, hashv;
char * c;
c = string;
while (*c) {
hash += *c;
c++;
}
hashv = hash / pObj->maxSessions;
hash = (hashv + hash % pObj->maxSessions) % pObj->maxSessions;
return hash;
}
void *taosAddStrHashWithSize(void *handle, char *string, char *pData, int dataSize) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
if (string == NULL || string[0] == 0) return NULL;
hash = (*pObj->hashFp)(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = (SHashNode *)malloc(sizeof(SHashNode) + (size_t)dataSize + strlen(string) + 1);
memcpy(pNode->data, pData, (size_t)dataSize);
pNode->prev = 0;
pNode->next = pObj->hashList[hash];
pNode->string = pNode->data + dataSize;
strcpy(pNode->string, string);
if (pObj->hashList[hash] != 0) (pObj->hashList[hash])->prev = pNode;
pObj->hashList[hash] = pNode;
pthread_mutex_unlock(&pObj->mutex);
uTrace("hash:%d:%s is added", hash, string);
return pNode->data;
}
void *taosAddStrHash(void *handle, char *string, char *pData) {
if (string == NULL || string[0] == 0) return NULL;
SHashObj *pObj = (SHashObj *)handle;
return taosAddStrHashWithSize(handle, string, pData, pObj->dataSize);
}
void taosDeleteStrHashNode(void *handle, char *string, void *pDeleteNode) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
bool find = false;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
if (string == NULL || string[0] == 0) return;
hash = (*(pObj->hashFp))(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) != 0) continue;
if (pNode->data == pDeleteNode) {
find = true;
break;
}
pNode = pNode->next;
}
if (find) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
uTrace("hash:%d:%s:%p is removed", hash, string, pNode);
free(pNode);
}
pthread_mutex_unlock(&pObj->mutex);
}
void taosDeleteStrHash(void *handle, char *string) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
if (string == NULL || string[0] == 0) return;
hash = (*(pObj->hashFp))(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
uTrace("hash:%d:%s:%p is removed", hash, string, pNode);
free(pNode);
}
pthread_mutex_unlock(&pObj->mutex);
}
void *taosGetStrHashData(void *handle, char *string) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
if (string == NULL || string[0] == 0) return NULL;
hash = (*pObj->hashFp)(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) {
uTrace("hash:%d:%s is retrieved", hash, string);
break;
}
pNode = pNode->next;
}
pthread_mutex_unlock(&pObj->mutex);
if (pNode) return pNode->data;
return NULL;
}
void *taosInitStrHash(uint32_t maxSessions, uint32_t dataSize, uint32_t (*fp)(void *, char *)) {
SHashObj *pObj;
pObj = (SHashObj *)malloc(sizeof(SHashObj));
if (pObj == NULL) {
return NULL;
}
memset(pObj, 0, sizeof(SHashObj));
pObj->maxSessions = maxSessions;
pObj->dataSize = dataSize;
pObj->hashFp = fp;
pObj->hashList = (SHashNode **)malloc(sizeof(SHashNode *) * (size_t)maxSessions);
if (pObj->hashList == NULL) {
free(pObj);
return NULL;
}
memset(pObj->hashList, 0, sizeof(SHashNode *) * (size_t)maxSessions);
pthread_mutex_init(&pObj->mutex, NULL);
return pObj;
}
void taosCleanUpStrHashWithFp(void *handle, void (*fp)(char *)) {
SHashObj * pObj;
SHashNode *pNode, *pNext;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
pthread_mutex_lock(&pObj->mutex);
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
if (fp != NULL) fp(pNode->data);
free(pNode);
pNode = pNext;
}
}
free(pObj->hashList);
}
pthread_mutex_unlock(&pObj->mutex);
pthread_mutex_destroy(&pObj->mutex);
memset(pObj, 0, sizeof(SHashObj));
free(pObj);
}
void taosCleanUpStrHash(void *handle) { taosCleanUpStrHashWithFp(handle, NULL); }
char *taosVisitStrHashWithFp(void *handle, int (*fp)(char *)) {
SHashObj * pObj;
SHashNode *pNode, *pNext;
char * pData = NULL;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return NULL;
pthread_mutex_lock(&pObj->mutex);
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
int flag = fp(pNode->data);
if (flag) {
pData = pNode->data;
goto VisitEnd;
}
pNode = pNext;
}
}
}
VisitEnd:
pthread_mutex_unlock(&pObj->mutex);
return pData;
}
src/util/src/tcache.c
浏览文件 @ 5e7f8430
......@@ -149,6 +149,7 @@ static void taosRemoveFromTrash(SCacheObj *pCacheObj, STrashElem *pElem) {
}
pElem->pData->signature = 0;
if (pCacheObj->freeFp) pCacheObj->freeFp(pElem->pData->data);
free(pElem->pData);
free(pElem);
}
......@@ -210,7 +211,8 @@ static FORCE_INLINE void taosCacheReleaseNode(SCacheObj *pCacheObj, SCacheDataNo
int32_t size = pNode->size;
taosHashRemove(pCacheObj->pHashTable, pNode->key, pNode->keySize);
uTrace("key:%s is removed from cache,total:%d,size:%ldbytes", pNode->key, pCacheObj->totalSize, size);
uTrace("key:%s is removed from cache,total:%d,size:%ldbytes", pNode->key, pCacheObj->totalSize, size);
if (pCacheObj->freeFp) pCacheObj->freeFp(pNode->data);
free(pNode);
}
......@@ -380,7 +382,7 @@ static void taosCacheRefresh(void *handle, void *tmrId) {
}
}
SCacheObj *taosCacheInit(void *tmrCtrl, int64_t refreshTime) {
SCacheObj *taosCacheInitWithCb(void *tmrCtrl, int64_t refreshTime, void (*freeCb)(void *data)) {
if (tmrCtrl == NULL || refreshTime <= 0) {
return NULL;
}
......@@ -401,6 +403,7 @@ SCacheObj *taosCacheInit(void *tmrCtrl, int64_t refreshTime) {
// set free cache node callback function for hash table
taosHashSetFreecb(pCacheObj->pHashTable, taosFreeNode);
pCacheObj->freeFp = freeCb;
pCacheObj->refreshTime = refreshTime * 1000;
pCacheObj->tmrCtrl = tmrCtrl;
......@@ -419,6 +422,10 @@ SCacheObj *taosCacheInit(void *tmrCtrl, int64_t refreshTime) {
return pCacheObj;
}
SCacheObj *taosCacheInit(void *tmrCtrl, int64_t refreshTime) {
return taosCacheInitWithCb(tmrCtrl, refreshTime, NULL);
}
void *taosCachePut(SCacheObj *pCacheObj, const char *key, const void *pData, size_t dataSize, int duration) {
SCacheDataNode *pNode;
......
src/util/src/thash.c 已删除 100644 → 0
浏览文件 @ 20d372a4
/*
* 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/>.
*/
#include "os.h"
#include "tmempool.h"
typedef struct _long_hash_t {
unsigned int id;
struct _long_hash_t *prev;
struct _long_hash_t *next;
uint64_t cont;
} SLongHash;
typedef struct {
SLongHash **longHashList;
mpool_h longHashMemPool;
int (*hashFp)(void *, uint64_t);
int maxSessions;
pthread_mutex_t mutex;
} SHashObj;
uint64_t taosHashUInt64(uint64_t handle) {
uint64_t hash = handle >> 16;
hash += handle & 0xFFFF;
return hash;
}
int taosHashLong(void *handle, uint64_t ip) {
SHashObj *pObj = (SHashObj *)handle;
int hash = 0;
hash = (int)(ip >> 16);
hash += (int)(ip & 0xFFFF);
hash = hash % pObj->maxSessions;
return hash;
}
int taosAddHash(void *handle, uint64_t cont, unsigned int id) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return -1;
pthread_mutex_lock(&pObj->mutex);
hash = (*pObj->hashFp)(pObj, cont);
pNode = (SLongHash *)taosMemPoolMalloc(pObj->longHashMemPool);
pNode->cont = cont;
pNode->id = id;
pNode->prev = 0;
pNode->next = pObj->longHashList[hash];
if (pObj->longHashList[hash] != 0) (pObj->longHashList[hash])->prev = pNode;
pObj->longHashList[hash] = pNode;
pthread_mutex_unlock(&pObj->mutex);
return 0;
}
void taosDeleteHash(void *handle, uint64_t cont) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
hash = (*pObj->hashFp)(pObj, cont);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->longHashList[hash];
while (pNode) {
if (pNode->cont == cont) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->longHashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
taosMemPoolFree(pObj->longHashMemPool, (char *)pNode);
}
pthread_mutex_unlock(&pObj->mutex);
}
int32_t taosGetIdFromHash(void *handle, uint64_t cont) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return -1;
hash = (*pObj->hashFp)(pObj, cont);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->longHashList[hash];
while (pNode) {
if (pNode->cont == cont) {
break;
}
pNode = pNode->next;
}
pthread_mutex_unlock(&pObj->mutex);
if (pNode) return (int32_t)pNode->id;
return -1;
}
void *taosOpenHash(int maxSessions, int (*fp)(void *, uint64_t)) {
SLongHash **longHashList;
mpool_h longHashMemPool;
SHashObj * pObj;
longHashMemPool = taosMemPoolInit(maxSessions, sizeof(SLongHash));
if (longHashMemPool == 0) return NULL;
longHashList = calloc(sizeof(SLongHash *), (size_t)maxSessions);
if (longHashList == 0) {
taosMemPoolCleanUp(longHashMemPool);
return NULL;
}
pObj = malloc(sizeof(SHashObj));
if (pObj == NULL) {
taosMemPoolCleanUp(longHashMemPool);
free(longHashList);
return NULL;
}
pObj->maxSessions = maxSessions;
pObj->longHashMemPool = longHashMemPool;
pObj->longHashList = longHashList;
pObj->hashFp = fp;
pthread_mutex_init(&pObj->mutex, NULL);
return pObj;
}
void taosCloseHash(void *handle) {
SHashObj *pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
pthread_mutex_lock(&pObj->mutex);
if (pObj->longHashMemPool) taosMemPoolCleanUp(pObj->longHashMemPool);
if (pObj->longHashList) free(pObj->longHashList);
pthread_mutex_unlock(&pObj->mutex);
pthread_mutex_destroy(&pObj->mutex);
memset(pObj, 0, sizeof(SHashObj));
free(pObj);
}
src/vnode/src/vnodeMain.c
浏览文件 @ 5e7f8430
......@@ -15,7 +15,7 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "ihash.h"
#include "hash.h"
#include "taoserror.h"
#include "taosmsg.h"
#include "tutil.h"
......@@ -32,7 +32,6 @@
static int32_t tsOpennedVnodes;
static void *tsDnodeVnodesHash;
static void vnodeCleanUp(SVnodeObj *pVnode);
static void vnodeBuildVloadMsg(char *pNode, void * param);
static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg);
static int32_t vnodeReadCfg(SVnodeObj *pVnode);
static int32_t vnodeSaveVersion(SVnodeObj *pVnode);
......@@ -58,7 +57,7 @@ static void vnodeInit() {
vnodeInitWriteFp();
vnodeInitReadFp();
tsDnodeVnodesHash = taosInitIntHash(TSDB_MAX_VNODES, sizeof(SVnodeObj *), taosHashInt);
tsDnodeVnodesHash = taosHashInit(TSDB_MAX_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true);
if (tsDnodeVnodesHash == NULL) {
vError("failed to init vnode list");
}
......@@ -68,7 +67,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
int32_t code;
pthread_once(&vnodeModuleInit, vnodeInit);
SVnodeObj *pTemp = (SVnodeObj *)taosGetIntHashData(tsDnodeVnodesHash, pVnodeCfg->cfg.vgId);
SVnodeObj *pTemp = (SVnodeObj *)taosHashGet(tsDnodeVnodesHash, (const char *)&pVnodeCfg->cfg.vgId, sizeof(int32_t));
if (pTemp != NULL) {
vPrint("vgId:%d, vnode already exist, pVnode:%p", pVnodeCfg->cfg.vgId, pTemp);
return TSDB_CODE_SUCCESS;
......@@ -120,7 +119,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
}
int32_t vnodeDrop(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId);
SVnodeObj **ppVnode = (SVnodeObj **)taosHashGet(tsDnodeVnodesHash, (const char *)&vgId, sizeof(int32_t));
if (ppVnode == NULL || *ppVnode == NULL) {
vTrace("vgId:%d, failed to drop, vgId not exist", vgId);
return TSDB_CODE_INVALID_VGROUP_ID;
......@@ -147,7 +146,7 @@ int32_t vnodeAlter(void *param, SMDCreateVnodeMsg *pVnodeCfg) {
code = vnodeReadCfg(pVnode);
if (code != TSDB_CODE_SUCCESS) {
vError("vgId:%d, failed to read cfg file", pVnode->vgId);
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
return code;
}
......@@ -182,12 +181,12 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->version = 0;
pVnode->tsdbCfg.tsdbId = pVnode->vgId;
pVnode->rootDir = strdup(rootDir);
taosAddIntHash(tsDnodeVnodesHash, pVnode->vgId, (char *)(&pVnode));
taosHashPut(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *));
int32_t code = vnodeReadCfg(pVnode);
if (code != TSDB_CODE_SUCCESS) {
vError("vgId:%d, failed to read cfg file", pVnode->vgId);
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
return code;
}
......@@ -212,7 +211,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->tsdb = tsdbOpenRepo(temp, &appH);
if (pVnode->tsdb == NULL) {
vError("vgId:%d, failed to open tsdb at %s(%s)", pVnode->vgId, temp, tstrerror(terrno));
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
return terrno;
}
......@@ -248,7 +247,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
}
int32_t vnodeClose(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId);
SVnodeObj **ppVnode = (SVnodeObj **)taosHashGet(tsDnodeVnodesHash, (const char *)&vgId, sizeof(int32_t));
if (ppVnode == NULL || *ppVnode == NULL) return 0;
SVnodeObj *pVnode = *ppVnode;
......@@ -292,14 +291,16 @@ void vnodeRelease(void *pVnodeRaw) {
vTrace("vgId:%d, vnode is released, vnodes:%d", vgId, count);
if (count <= 0) {
taosCleanUpIntHash(tsDnodeVnodesHash);
taosHashCleanup(tsDnodeVnodesHash);
vnodeModuleInit = PTHREAD_ONCE_INIT;
tsDnodeVnodesHash = NULL;
}
}
void *vnodeGetVnode(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId);
if (tsDnodeVnodesHash == NULL) return NULL;
SVnodeObj **ppVnode = (SVnodeObj **)taosHashGet(tsDnodeVnodesHash, (const char *)&vgId, sizeof(int32_t));
if (ppVnode == NULL || *ppVnode == NULL) {
terrno = TSDB_CODE_INVALID_VGROUP_ID;
vPrint("vgId:%d, not exist", vgId);
......@@ -333,16 +334,8 @@ void *vnodeGetWal(void *pVnode) {
return ((SVnodeObj *)pVnode)->wal;
}
void vnodeBuildStatusMsg(void *param) {
SDMStatusMsg *pStatus = param;
taosVisitIntHashWithFp(tsDnodeVnodesHash, vnodeBuildVloadMsg, pStatus);
}
static void vnodeBuildVloadMsg(char *pNode, void * param) {
SVnodeObj *pVnode = *(SVnodeObj **) pNode;
static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SDMStatusMsg *pStatus) {
if (pVnode->status == TAOS_VN_STATUS_DELETING) return;
SDMStatusMsg *pStatus = param;
if (pStatus->openVnodes >= TSDB_MAX_VNODES) return;
SVnodeLoad *pLoad = &pStatus->load[pStatus->openVnodes++];
......@@ -356,8 +349,24 @@ static void vnodeBuildVloadMsg(char *pNode, void * param) {
pLoad->replica = pVnode->syncCfg.replica;
}
void vnodeBuildStatusMsg(void *param) {
SDMStatusMsg *pStatus = param;
SHashMutableIterator *pIter = taosHashCreateIter(tsDnodeVnodesHash);
while (taosHashIterNext(pIter)) {
SVnodeObj **pVnode = taosHashIterGet(pIter);
if (pVnode == NULL) continue;
if (*pVnode == NULL) continue;
vnodeBuildVloadMsg(*pVnode, pStatus);
pStatus++;
}
taosHashDestroyIter(pIter);
}
static void vnodeCleanUp(SVnodeObj *pVnode) {
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
if (pVnode->sync) {
syncStop(pVnode->sync);
......
tests/script/general/parser/auto_create_tb.sim
浏览文件 @ 5e7f8430
......@@ -149,16 +149,17 @@ sql drop table tb1
#### auto create multiple tables
sql insert into tb1 using $stb tags(1) values ( $ts0 , 1, 1, 1, 1, 'bin1', 1, 1, 1, '涛思数据1') tb2 using $stb tags(2) values ( $ts0 , 2, 2, 2, 2, 'bin2', 2, 2, 2, '涛思数据2') tb3 using $stb tags(3) values ( $ts0 , 3, 3, 3, 3, 'bin3', 3, 3, 3, '涛思数据3')
sql show tables
print $rows $data00 $data10 $data20
if $rows != 3 then
return -1
endi
if $data00 != tb1 then
if $data00 != tb3 then
return -1
endi
if $data10 != tb2 then
return -1
endi
if $data20 != tb3 then
if $data20 != tb1 then
return -1
endi
......@@ -220,13 +221,13 @@ sql show tables
if $rows != 3 then
return -1
endi
if $data00 != tb1 then
if $data00 != tb3 then
return -1
endi
if $data10 != tb2 then
return -1
endi
if $data20 != tb3 then
if $data20 != tb1 then
return -1
endi
......
tests/script/jenkins/basic.txt
浏览文件 @ 5e7f8430
......@@ -15,7 +15,9 @@ cd ../../../debug; make
./test.sh -f general/cache/restart_metrics.sim
./test.sh -f general/cache/restart_table.sim
#hongze ./test.sh -f general/column/commit.sim
./test.sh -f general/connection/connections.sim
./test.sh -f general/column/commit.sim
./test.sh -f general/column/metrics.sim
./test.sh -f general/column/table.sim
......@@ -32,7 +34,7 @@ cd ../../../debug; make
./test.sh -f general/compute/first.sim
# liao ./test.sh -f general/compute/interval.sim
./test.sh -f general/compute/last.sim
# liao ./test.sh -f general/compute/leastsquare.sim
./test.sh -f general/compute/leastsquare.sim
./test.sh -f general/compute/max.sim
./test.sh -f general/compute/min.sim
./test.sh -f general/compute/null.sim
......@@ -59,10 +61,10 @@ cd ../../../debug; make
./test.sh -f general/db/tables.sim
./test.sh -f general/field/2.sim
#liao ./test.sh -f general/field/3.sim
#liao? ./test.sh -f general/field/4.sim
#liao? ./test.sh -f general/field/5.sim
#liao? ./test.sh -f general/field/6.sim
./test.sh -f general/field/3.sim
./test.sh -f general/field/4.sim
./test.sh -f general/field/5.sim
./test.sh -f general/field/6.sim
./test.sh -f general/field/bigint.sim
./test.sh -f general/field/binary.sim
./test.sh -f general/field/bool.sim
......@@ -98,7 +100,7 @@ cd ../../../debug; make
#unsupport ./test.sh -f general/parser/alter1.sim
#unsupport ./test.sh -f general/parser/alter_stable.sim
./test.sh -f general/parser/auto_create_tb.sim
#slguan ./test.sh -f general/parser/auto_create_tb_drop_tb.sim
./test.sh -f general/parser/auto_create_tb_drop_tb.sim
./test.sh -f general/parser/col_arithmetic_operation.sim
./test.sh -f general/parser/columnValue.sim
./test.sh -f general/parser/commit.sim
......@@ -148,7 +150,7 @@ cd ../../../debug; make
./test.sh -f general/stable/disk.sim
./test.sh -f general/stable/metrics.sim
#liao? ./test.sh -f general/stable/values.sim
./test.sh -f general/stable/values.sim
./test.sh -f general/stable/vnode3.sim
./test.sh -f general/table/autocreate.sim
......@@ -179,10 +181,10 @@ cd ../../../debug; make
./test.sh -f general/table/tinyint.sim
./test.sh -f general/table/vgroup.sim
#liao ./test.sh -f general/tag/3.sim
#liao? ./test.sh -f general/tag/4.sim
#liao? ./test.sh -f general/tag/5.sim
#liao? ./test.sh -f general/tag/6.sim
./test.sh -f general/tag/3.sim
./test.sh -f general/tag/4.sim
./test.sh -f general/tag/5.sim
./test.sh -f general/tag/6.sim
#unsupport ./test.sh -f general/tag/add.sim
./test.sh -f general/tag/bigint.sim
./test.sh -f general/tag/binary_binary.sim
......@@ -195,16 +197,17 @@ cd ../../../debug; make
#unsupport ./test.sh -f general/tag/commit.sim
./test.sh -f general/tag/create.sim
#unsupport ./test.sh -f general/tag/delete.sim
#./test.sh -f general/tag/double.sim
./test.sh -f general/tag/double.sim
./test.sh -f general/tag/filter.sim
#liao? ./test.sh -f general/tag/float.sim
#liao? ./test.sh -f general/tag/int_binary.sim
./test.sh -f general/tag/float.sim
./test.sh -f general/tag/int_binary.sim
./test.sh -f general/tag/int_float.sim
./test.sh -f general/tag/int.sim
#unsupport ./test.sh -f general/tag/set.sim
./test.sh -f general/tag/smallint.sim
./test.sh -f general/tag/tinyint.sim
./test.sh -f general/user/authority.sim
./test.sh -f general/user/basic1.sim
./test.sh -f general/user/monitor.sim
./test.sh -f general/user/pass_alter.sim
......
tests/script/unique/mnode/mgmt22.sim
浏览文件 @ 5e7f8430
......@@ -8,7 +8,7 @@ system sh/cfg.sh -n dnode2 -c numOfMPeers -v 2
system sh/cfg.sh -n dnode3 -c numOfMPeers -v 2
print ============== step1
system sh/exec_up.sh -n dnode1 -s start -t
system sh/exec_up.sh -n dnode1 -s start
sleep 3000
sql connect
......@@ -20,7 +20,7 @@ if $data2_1 != master then
endi
print ============== step2
system sh/exec_up.sh -n dnode2 -s start -t
system sh/exec_up.sh -n dnode2 -s start
sql create dnode $hostname2
$x = 0
......@@ -41,17 +41,6 @@ if $data2_2 != slave then
goto show2
endi
system sh/exec_up.sh -n dnode1 -s stop -x SIGINT
system sh/exec_up.sh -n dnode2 -s stop -x SIGINT
system sh/exec_up.sh -n dnode3 -s stop -x SIGINT
system sh/exec_up.sh -n dnode4 -s stop -x SIGINT
system sh/exec_up.sh -n dnode5 -s stop -x SIGINT
system sh/exec_up.sh -n dnode6 -s stop -x SIGINT
system sh/exec_up.sh -n dnode7 -s stop -x SIGINT
system sh/exec_up.sh -n dnode8 -s stop -x SIGINT
return
print ============== step3
sql_error drop dnode $hostname1 -x error1
print should not drop master
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或