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

Merge branch 'develop' into feature/TD-1413

This commit is contained in:
Hongze Cheng 2020-11-28 03:20:36 +00:00
parent d66d3119a0 6a4481f490
commit 91bc2e9c9d
57 changed files with 614 additions and 537 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/client/inc/tsclient.h
View File

@ -282,7 +282,7 @@ typedef struct {
int8_t dataSourceType; // load data from file or not
int8_t submitSchema; // submit block is built with table schema
STagData tagData;
STagData *pTagData; // NOTE: pTagData->data is used as a variant length array
SHashObj *pTableList; // referred table involved in sql
SArray *pDataBlocks; // SArray<STableDataBlocks*> submit data blocks after parsing sql
} SSqlCmd;

20
src/client/src/tscParseInsert.c
View File

@ -790,9 +790,6 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
sql += index;
tscAllocPayload(pCmd, sizeof(STagData));
STagData *pTag = &pCmd->tagData;
memset(pTag, 0, sizeof(STagData));
//the source super table is moved to the secondary position of the pTableMetaInfo list
if (pQueryInfo->numOfTables < 2) {
@ -805,7 +802,14 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
return code;
}
STagData *pTag = realloc(pCmd->pTagData, offsetof(STagData, data));
if (pTag == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memset(pTag, 0, offsetof(STagData, data));
tstrncpy(pTag->name, pSTableMeterMetaInfo->name, sizeof(pTag->name));
pCmd->pTagData = pTag;
code = tscGetTableMeta(pSql, pSTableMeterMetaInfo);
if (code != TSDB_CODE_SUCCESS) {
return code;
@ -934,7 +938,13 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
tdSortKVRowByColIdx(row);
pTag->dataLen = kvRowLen(row);
pTag = (STagData*)realloc(pCmd->pTagData, offsetof(STagData, data) + kvRowLen(row));
if (pTag == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
pCmd->pTagData = pTag;
pTag->dataLen = htonl(kvRowLen(row));
kvRowCpy(pTag->data, row);
free(row);
@ -945,8 +955,6 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
return tscSQLSyntaxErrMsg(pCmd->payload, ") expected", sToken.z);
}
pTag->dataLen = htonl(pTag->dataLen);
if (tscValidateName(&tableToken) != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(pCmd->payload, "invalid table name", *sqlstr);
}

28
src/client/src/tscServer.c
View File

@ -1565,11 +1565,11 @@ int tscBuildTableMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
char *pMsg = (char *)pInfoMsg + sizeof(STableInfoMsg);
size_t len = htonl(pCmd->tagData.dataLen);
if (pSql->cmd.autoCreated) {
if (pCmd->autoCreated && pCmd->pTagData != NULL) {
int len = htonl(pCmd->pTagData->dataLen);
if (len > 0) {
len += sizeof(pCmd->tagData.name) + sizeof(pCmd->tagData.dataLen);
memcpy(pInfoMsg->tags, &pCmd->tagData, len);
len += sizeof(pCmd->pTagData->name) + sizeof(pCmd->pTagData->dataLen);
memcpy(pInfoMsg->tags, pCmd->pTagData, len);
pMsg += len;
}
}
@ -2239,8 +2239,6 @@ static int32_t getTableMetaFromMgmt(SSqlObj *pSql, STableMetaInfo *pTableMetaInf
pNew->signature = pNew;
pNew->cmd.command = TSDB_SQL_META;
registerSqlObj(pNew);
tscAddSubqueryInfo(&pNew->cmd);
SQueryInfo *pNewQueryInfo = tscGetQueryInfoDetailSafely(&pNew->cmd, 0);
@ -2248,8 +2246,7 @@ static int32_t getTableMetaFromMgmt(SSqlObj *pSql, STableMetaInfo *pTableMetaInf
pNew->cmd.autoCreated = pSql->cmd.autoCreated; // create table if not exists
if (TSDB_CODE_SUCCESS != tscAllocPayload(&pNew->cmd, TSDB_DEFAULT_PAYLOAD_SIZE + pSql->cmd.payloadLen)) {
tscError("%p malloc failed for payload to get table meta", pSql);
free(pNew);
tscFreeSqlObj(pNew);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
@ -2257,12 +2254,25 @@ static int32_t getTableMetaFromMgmt(SSqlObj *pSql, STableMetaInfo *pTableMetaInf
assert(pNew->cmd.numOfClause == 1 && pNewQueryInfo->numOfTables == 1);
tstrncpy(pNewMeterMetaInfo->name, pTableMetaInfo->name, sizeof(pNewMeterMetaInfo->name));
memcpy(&pNew->cmd.tagData, &pSql->cmd.tagData, sizeof(pSql->cmd.tagData));
if (pSql->cmd.pTagData != NULL) {
int size = offsetof(STagData, data) + htonl(pSql->cmd.pTagData->dataLen);
pNew->cmd.pTagData = calloc(1, size);
if (pNew->cmd.pTagData == NULL) {
tscError("%p malloc failed for new tag data to get table meta", pSql);
tscFreeSqlObj(pNew);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memcpy(pNew->cmd.pTagData, pSql->cmd.pTagData, size);
}
tscDebug("%p new pSqlObj:%p to get tableMeta, auto create:%d", pSql, pNew, pNew->cmd.autoCreated);
pNew->fp = tscTableMetaCallBack;
pNew->param = pSql;
registerSqlObj(pNew);
int32_t code = tscProcessSql(pNew);
if (code == TSDB_CODE_SUCCESS) {
code = TSDB_CODE_TSC_ACTION_IN_PROGRESS; // notify upper application that current process need to be terminated

7
src/client/src/tscSub.c
View File

@ -442,7 +442,12 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
size_t size = taosArrayGetSize(pSub->progress) * sizeof(STableIdInfo);
size += sizeof(SQueryTableMsg) + 4096;
tscAllocPayload(&pSql->cmd, (int)size);
int code = tscAllocPayload(&pSql->cmd, (int)size);
if (code != TSDB_CODE_SUCCESS) {
tscError("failed to alloc payload");
return NULL;
}
for (int retry = 0; retry < 3; retry++) {
tscRemoveFromSqlList(pSql);

16
src/client/src/tscUtil.c
View File

@ -408,7 +408,7 @@ void tscResetSqlCmdObj(SSqlCmd* pCmd, bool removeFromCache) {
pCmd->pTableList = NULL;
pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks);
tscFreeQueryInfo(pCmd, removeFromCache);
}
@ -512,6 +512,8 @@ void tscFreeSqlObj(SSqlObj* pSql) {
tscFreeSqlResult(pSql);
tscResetSqlCmdObj(pCmd, false);
tfree(pCmd->pTagData);
memset(pCmd->payload, 0, (size_t)pCmd->allocSize);
tfree(pCmd->payload);
pCmd->allocSize = 0;
@ -1909,7 +1911,17 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm
pCmd->command = cmd;
pCmd->parseFinished = 1;
pCmd->autoCreated = pSql->cmd.autoCreated;
memcpy(&pCmd->tagData, &pSql->cmd.tagData, sizeof(pCmd->tagData));
if (pSql->cmd.pTagData != NULL) {
int size = offsetof(STagData, data) + htonl(pSql->cmd.pTagData->dataLen);
pNew->cmd.pTagData = calloc(1, size);
if (pNew->cmd.pTagData == NULL) {
tscError("%p new subquery failed, unable to malloc tag data, tableIndex:%d", pSql, 0);
free(pNew);
return NULL;
}
memcpy(pNew->cmd.pTagData, pSql->cmd.pTagData, size);
}
if (tscAddSubqueryInfo(pCmd) != TSDB_CODE_SUCCESS) {
tscFreeSqlObj(pNew);

2
src/common/src/tglobal.c
View File

@ -206,7 +206,7 @@ int32_t tsNumOfLogLines = 10000000;
int32_t mDebugFlag = 135;
int32_t sdbDebugFlag = 135;
int32_t dDebugFlag = 135;
int32_t vDebugFlag = 131;
int32_t vDebugFlag = 135;
int32_t cDebugFlag = 131;
int32_t jniDebugFlag = 131;
int32_t odbcDebugFlag = 131;

2
src/connector/go

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

2
src/connector/grafanaplugin

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

4
src/inc/taoserror.h
View File

@ -261,6 +261,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_CONFIG, 0, 0x0900, "Invalid Sy
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_ENABLED, 0, 0x0901, "Sync module not enabled")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_VERSION, 0, 0x0902, "Invalid Sync version")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_CONFIRM_EXPIRED, 0, 0x0903, "Sync confirm expired")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_VND_COMMITING, 0, 0x0904, "Vnode is commiting")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_FILE_CHNAGED, 0, 0x0905, "Vnode file is changed")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_APP_ERROR, 0, 0x1000, "Unexpected generic error in sync")
// wal
TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR, 0, 0x1000, "Unexpected generic error in wal")
@ -367,6 +370,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_HTTP_OP_TAG_VALUE_TOO_LONG, 0, 0x11A4, "tag value
TAOS_DEFINE_ERROR(TSDB_CODE_HTTP_OP_VALUE_NULL, 0, 0x11A5, "value not find")
TAOS_DEFINE_ERROR(TSDB_CODE_HTTP_OP_VALUE_TYPE, 0, 0x11A6, "value type should be boolean, number or string")
// odbc
TAOS_DEFINE_ERROR(TSDB_CODE_ODBC_OOM, 0, 0x2100, "out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_ODBC_CONV_CHAR_NOT_NUM, 0, 0x2101, "convertion not a valid literal input")
TAOS_DEFINE_ERROR(TSDB_CODE_ODBC_CONV_UNDEF, 0, 0x2102, "convertion undefined")

4
src/inc/tsync.h
View File

@ -85,6 +85,9 @@ typedef void (*FNotifyFlowCtrl)(int32_t vgId, int32_t level);
// when data file is synced successfully, notity app
typedef int32_t (*FNotifyFileSynced)(int32_t vgId, uint64_t fversion);
// get file version
typedef int32_t (*FGetFileVersion)(int32_t vgId, uint64_t *fver);
typedef struct {
int32_t vgId; // vgroup ID
uint64_t version; // initial version
@ -97,6 +100,7 @@ typedef struct {
FNotifyRole notifyRole;
FNotifyFlowCtrl notifyFlowCtrl;
FNotifyFileSynced notifyFileSynced;
FGetFileVersion getFileVersion;
} SSyncInfo;
typedef void *tsync_h;

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);
}

2
src/query/inc/qUtil.h
View File

@ -27,7 +27,7 @@
int32_t getOutputInterResultBufSize(SQuery* pQuery);
void clearResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* pRow, int16_t type);
void copyResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* dst, const SResultRow* src);
void copyResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* dst, const SResultRow* src, int16_t type);
SResultRowCellInfo* getResultCell(SQueryRuntimeEnv* pRuntimeEnv, const SResultRow* pRow, int32_t index);
int32_t initWindowResInfo(SWindowResInfo* pWindowResInfo, int32_t size, int32_t threshold, int16_t type);

6
src/query/src/qExecutor.c
View File

@ -5836,9 +5836,9 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
pQueryMsg->interval.interval = htobe64(pQueryMsg->interval.interval);
pQueryMsg->interval.sliding = htobe64(pQueryMsg->interval.sliding);
pQueryMsg->interval.offset = htobe64(pQueryMsg->interval.offset);
pQueryMsg->interval.intervalUnit = pQueryMsg->interval.intervalUnit;
pQueryMsg->interval.slidingUnit = pQueryMsg->interval.slidingUnit;
pQueryMsg->interval.offsetUnit = pQueryMsg->interval.offsetUnit;
// pQueryMsg->interval.intervalUnit = pQueryMsg->interval.intervalUnit;
// pQueryMsg->interval.slidingUnit = pQueryMsg->interval.slidingUnit;
// pQueryMsg->interval.offsetUnit = pQueryMsg->interval.offsetUnit;
pQueryMsg->limit = htobe64(pQueryMsg->limit);
pQueryMsg->offset = htobe64(pQueryMsg->offset);

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);

12
src/query/src/qUtil.c
View File

@ -133,7 +133,7 @@ void clearFirstNWindowRes(SQueryRuntimeEnv *pRuntimeEnv, int32_t num) {
// clear all the closed windows from the window list
for (int32_t k = 0; k < remain; ++k) {
copyResultRow(pRuntimeEnv, pWindowResInfo->pResult[k], pWindowResInfo->pResult[num + k]);
copyResultRow(pRuntimeEnv, pWindowResInfo->pResult[k], pWindowResInfo->pResult[num + k], type);
}
// move the unclosed window in the front of the window list
@ -272,9 +272,15 @@ void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pWindowRes, int16
* since the attribute of "Pos" is bound to each window result when the window result is created in the
* disk-based result buffer.
*/
void copyResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *dst, const SResultRow *src) {
void copyResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *dst, const SResultRow *src, int16_t type) {
dst->numOfRows = src->numOfRows;
dst->win = src->win;
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
dst->key = realloc(dst->key, varDataTLen(src->key));
varDataCopy(dst->key, src->key);
} else {
dst->win = src->win;
}
dst->closed = src->closed;
int32_t nOutputCols = pRuntimeEnv->pQuery->numOfOutput;

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);
}

2
src/sync/inc/syncInt.h
View File

@ -139,6 +139,7 @@ typedef struct SsyncPeer {
char id[TSDB_EP_LEN + 32]; // peer vgId + end point
uint64_t version;
uint64_t sversion; // track the peer version in retrieve process
uint64_t lastVer; // track the file version while retrieve
int32_t syncFd;
int32_t peerFd; // forward FD
int32_t numOfRetrieves; // number of retrieves tried
@ -172,6 +173,7 @@ typedef struct SSyncNode {
FNotifyRole notifyRole;
FNotifyFlowCtrl notifyFlowCtrl;
FNotifyFileSynced notifyFileSynced;
FGetFileVersion getFileVersion;
pthread_mutex_t mutex;
} SSyncNode;

6
src/sync/src/syncMain.c
View File

@ -196,6 +196,7 @@ int64_t syncStart(const SSyncInfo *pInfo) {
pNode->confirmForward = pInfo->confirmForward;
pNode->notifyFlowCtrl = pInfo->notifyFlowCtrl;
pNode->notifyFileSynced = pInfo->notifyFileSynced;
pNode->getFileVersion = pInfo->getFileVersion;
pNode->selfIndex = -1;
pNode->vgId = pInfo->vgId;
@ -540,7 +541,7 @@ static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo) {
pPeer->ip = ip;
pPeer->port = pInfo->nodePort;
pPeer->fqdn[sizeof(pPeer->fqdn) - 1] = 0;
snprintf(pPeer->id, sizeof(pPeer->id), "vgId:%d, peer:%s:%u", pNode->vgId, pPeer->fqdn, pPeer->port);
snprintf(pPeer->id, sizeof(pPeer->id), "vgId:%d, nodeId:%d", pNode->vgId, pPeer->nodeId);
pPeer->peerFd = -1;
pPeer->syncFd = -1;
@ -1143,8 +1144,7 @@ static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
pPeer->syncFd = connFd;
syncCreateRestoreDataThread(pPeer);
} else {
sDebug("%s, TCP connection is already up(pfd:%d), close one, new pfd:%d sfd:%d", pPeer->id, pPeer->peerFd, connFd,
pPeer->syncFd);
sDebug("%s, TCP connection is up, pfd:%d sfd:%d, old pfd:%d", pPeer->id, connFd, pPeer->syncFd, pPeer->peerFd);
syncClosePeerConn(pPeer);
pPeer->peerFd = connFd;
pPeer->pConn = taosAllocateTcpConn(tsTcpPool, pPeer, connFd);

12
src/sync/src/syncRestore.c
View File

@ -52,12 +52,12 @@ static void syncRemoveExtraFile(SSyncPeer *pPeer, int32_t sindex, int32_t eindex
static int32_t syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
SSyncNode *pNode = pPeer->pSyncNode;
SFileInfo minfo; memset(&minfo, 0, sizeof(minfo)); /* = {0}; */ // master file info
SFileInfo sinfo; memset(&sinfo, 0, sizeof(sinfo)); /* = {0}; */ // slave file info
SFileAck fileAck;
SFileInfo minfo; memset(&minfo, 0, sizeof(SFileInfo)); /* = {0}; */
SFileInfo sinfo; memset(&sinfo, 0, sizeof(SFileInfo)); /* = {0}; */
SFileAck fileAck = {0};
int32_t code = -1;
char name[TSDB_FILENAME_LEN * 2] = {0};
uint32_t pindex = 0; // index in last restore
uint32_t pindex = 0; // index in last restore
bool fileChanged = false;
*fversion = 0;
@ -134,7 +134,7 @@ static int32_t syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
// data file is changed, code shall be set to 1
*fversion = minfo.fversion;
code = 1;
sDebug("%s, file changed while restore file", pPeer->id);
sDebug("%s, file changed after restore file, fver:%" PRIu64, pPeer->id, *fversion);
}
if (code < 0) {
@ -160,7 +160,7 @@ static int32_t syncRestoreWal(SSyncPeer *pPeer) {
}
if (pHead->len == 0) {
sDebug("%s, wal is synced over", pPeer->id);
sDebug("%s, wal is synced over, last wver:%" PRIu64, pPeer->id, lastVer);
code = 0;
break;
} // wal sync over

135
src/sync/src/syncRetrieve.c
View File

@ -16,6 +16,7 @@
#define _DEFAULT_SOURCE
#include <sys/inotify.h>
#include "os.h"
#include "taoserror.h"
#include "tlog.h"
#include "tutil.h"
#include "tglobal.h"
@ -25,85 +26,35 @@
#include "tsync.h"
#include "syncInt.h"
static int32_t syncAddIntoWatchList(SSyncPeer *pPeer, char *name) {
sDebug("%s, start to monitor:%s", pPeer->id, name);
static int32_t syncAreFilesModified(SSyncNode *pNode, SSyncPeer *pPeer) {
if (pNode->getFileVersion == NULL) return TSDB_CODE_SUCCESS;
if (pPeer->notifyFd <= 0) {
pPeer->watchNum = 0;
pPeer->notifyFd = inotify_init1(IN_NONBLOCK);
if (pPeer->notifyFd < 0) {
sError("%s, failed to init inotify since %s", pPeer->id, strerror(errno));
return -1;
}
if (pPeer->watchFd == NULL) pPeer->watchFd = malloc(sizeof(int32_t) * tsMaxWatchFiles);
if (pPeer->watchFd == NULL) {
sError("%s, failed to allocate watchFd", pPeer->id);
return -1;
}
memset(pPeer->watchFd, -1, sizeof(int32_t) * tsMaxWatchFiles);
uint64_t fver = 0;
int32_t code = (*pNode->getFileVersion)(pNode->vgId, &fver);
if (code != 0) {
sInfo("%s, vnode is commiting while retrieve, last fver:%" PRIu64, pPeer->id, pPeer->lastVer);
pPeer->fileChanged = 1;
return TSDB_CODE_SYN_VND_COMMITING;
}
int32_t *wd = pPeer->watchFd + pPeer->watchNum;
if (*wd >= 0) {
if (inotify_rm_watch(pPeer->notifyFd, *wd) < 0) {
sError("%s, failed to remove wd:%d since %s", pPeer->id, *wd, strerror(errno));
return -1;
}
if (fver != pPeer->lastVer) {
sInfo("%s, files are modified while retrieve, fver:%" PRIu64 ", last fver:%" PRIu64, pPeer->id, fver, pPeer->lastVer);
pPeer->fileChanged = 1;
return TSDB_CODE_SYN_FILE_CHNAGED;
}
*wd = inotify_add_watch(pPeer->notifyFd, name, IN_MODIFY | IN_DELETE);
if (*wd == -1) {
sError("%s, failed to add %s since %s", pPeer->id, name, strerror(errno));
return -1;
} else {
sDebug("%s, monitor %s, wd:%d watchNum:%d", pPeer->id, name, *wd, pPeer->watchNum);
}
pPeer->watchNum = (pPeer->watchNum + 1) % tsMaxWatchFiles;
return 0;
}
static int32_t syncAreFilesModified(SSyncPeer *pPeer) {
if (pPeer->notifyFd <= 0) return 0;
char buf[2048];
int32_t len = read(pPeer->notifyFd, buf, sizeof(buf));
if (len < 0 && errno != EAGAIN) {
sError("%s, failed to read notify FD since %s", pPeer->id, strerror(errno));
return -1;
}
int32_t code = 0;
if (len > 0) {
const struct inotify_event *event;
char *ptr;
for (ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
event = (const struct inotify_event *)ptr;
if ((event->mask & IN_MODIFY) || (event->mask & IN_DELETE)) {
sDebug("%s, processed file is changed", pPeer->id);
pPeer->fileChanged = 1;
code = 1;
break;
}
}
}
return code;
pPeer->fileChanged = 0;
return TSDB_CODE_SUCCESS;
}
static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
SSyncNode *pNode = pPeer->pSyncNode;
SFileInfo fileInfo;
SFileAck fileAck;
int32_t code = -1;
SFileInfo fileInfo; memset(&fileInfo, 0, sizeof(SFileInfo));
SFileAck fileAck = {0};
int32_t code = TSDB_CODE_SYN_APP_ERROR;
char name[TSDB_FILENAME_LEN * 2] = {0};
memset(&fileInfo, 0, sizeof(fileInfo));
memset(&fileAck, 0, sizeof(fileAck));
if (pNode->getFileVersion) (*pNode->getFileVersion)(pNode->vgId, &pPeer->lastVer);
while (1) {
// retrieve file info
@ -111,24 +62,27 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
fileInfo.magic = (*pNode->getFileInfo)(pNode->vgId, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX,
&fileInfo.size, &fileInfo.fversion);
// fileInfo.size = htonl(size);
sDebug("%s, file:%s info is sent, size:%" PRId64, pPeer->id, fileInfo.name, fileInfo.size);
// send the file info
int32_t ret = taosWriteMsg(pPeer->syncFd, &(fileInfo), sizeof(fileInfo));
if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno);
sError("%s, failed to write file:%s info while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
// if no file anymore, break
if (fileInfo.magic == 0 || fileInfo.name[0] == 0) {
code = TSDB_CODE_SUCCESS;
sDebug("%s, no more files to sync", pPeer->id);
code = 0;
break;
}
// wait for the ack from peer
ret = taosReadMsg(pPeer->syncFd, &fileAck, sizeof(fileAck));
if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno);
sError("%s, failed to read file:%s ack while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
@ -136,15 +90,6 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
// set the peer sync version
pPeer->sversion = fileInfo.fversion;
// get the full path to file
snprintf(name, sizeof(name), "%s/%s", pNode->path, fileInfo.name);
// add the file into watch list
if (syncAddIntoWatchList(pPeer, name) < 0) {
sError("%s, failed to watch file:%s while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
// if sync is not required, continue
if (fileAck.sync == 0) {
fileInfo.index++;
@ -152,9 +97,13 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
continue;
}
// get the full path to file
snprintf(name, sizeof(name), "%s/%s", pNode->path, fileInfo.name);
// send the file to peer
int32_t sfd = open(name, O_RDONLY);
if (sfd < 0) {
code = TAOS_SYSTEM_ERROR(errno);
sError("%s, failed to open file:%s while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
@ -162,22 +111,21 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
ret = taosSendFile(pPeer->syncFd, sfd, NULL, fileInfo.size);
close(sfd);
if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno);
sError("%s, failed to send file:%s while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
sDebug("%s, %s is sent, size:%" PRId64, pPeer->id, name, fileInfo.size);
sDebug("%s, file:%s is sent, size:%" PRId64, pPeer->id, fileInfo.name, fileInfo.size);
fileInfo.index++;
// check if processed files are modified
if (syncAreFilesModified(pPeer) != 0) {
sInfo("%s, file:%s are modified while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
code = syncAreFilesModified(pNode, pPeer);
if (code != TSDB_CODE_SUCCESS) break;
}
if (code < 0) {
sError("%s, failed to retrieve file", pPeer->id);
if (code != TSDB_CODE_SUCCESS) {
sError("%s, failed to retrieve file since %s", pPeer->id, tstrerror(code));
}
return code;
@ -308,9 +256,9 @@ static int32_t syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversi
static int32_t syncProcessLastWal(SSyncPeer *pPeer, char *wname, int64_t index) {
SSyncNode *pNode = pPeer->pSyncNode;
int32_t code = -1;
char fname[TSDB_FILENAME_LEN * 2]; // full path to wal file
char fname[TSDB_FILENAME_LEN * 2] = {0}; // full path to wal file
if (syncAreFilesModified(pPeer) != 0) return -1;
if (syncAreFilesModified(pNode, pPeer) != 0) return -1;
while (1) {
int32_t once = 0; // last WAL has once ever been processed
@ -429,9 +377,7 @@ static int32_t syncRetrieveWal(SSyncPeer *pPeer) {
close(sfd);
if (code < 0) break;
index++;
if (syncAreFilesModified(pPeer) != 0) break;
if (syncAreFilesModified(pNode, pPeer) != 0) break;
}
if (code == 0) {
@ -481,18 +427,18 @@ static int32_t syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
pPeer->sversion = 0;
pPeer->sstatus = TAOS_SYNC_STATUS_FILE;
sInfo("%s, start to retrieve file, set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
sInfo("%s, start to retrieve files, set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
if (syncRetrieveFile(pPeer) < 0) {
sError("%s, failed to retrieve file", pPeer->id);
sError("%s, failed to retrieve files", pPeer->id);
return -1;
}
// if no files are synced, there must be wal to sync, sversion must be larger than one
if (pPeer->sversion == 0) pPeer->sversion = 1;
sInfo("%s, start to retrieve wal", pPeer->id);
sInfo("%s, start to retrieve wals", pPeer->id);
if (syncRetrieveWal(pPeer) < 0) {
sError("%s, failed to retrieve wal", pPeer->id);
sError("%s, failed to retrieve wals", pPeer->id);
return -1;
}
@ -506,7 +452,6 @@ void *syncRetrieveData(void *param) {
if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->vgId, pPeer->numOfRetrieves);
pPeer->fileChanged = 0;
pPeer->syncFd = taosOpenTcpClientSocket(pPeer->ip, pPeer->port, 0);
if (pPeer->syncFd < 0) {
sError("%s, failed to open socket to sync", pPeer->id);

4
src/tsdb/src/tsdbMemTable.c
View File

@ -874,10 +874,10 @@ static void tsdbFreeRows(STsdbRepo *pRepo, void **rows, int rowCounter) {
listNEles(pRepo->mem->bufBlockList), pBufBlock->offset, pBufBlock->remain);
if (pBufBlock->offset == 0) { // return the block to buffer pool
tsdbLockRepo(pRepo);
if (tsdbLockRepo(pRepo) < 0) return;
SListNode *pNode = tdListPopTail(pRepo->mem->bufBlockList);
tdListPrependNode(pBufPool->bufBlockList, pNode);
tsdbUnlockRepo(pRepo);
if (tsdbUnlockRepo(pRepo) < 0) return;
}
} else {
ASSERT(listNEles(pRepo->mem->extraBuffList) > 0);

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;
}

3
src/util/src/tqueue.c
View File

@ -79,15 +79,12 @@ void taosCloseQueue(taos_queue param) {
if (queue->qset) taosRemoveFromQset(qset, queue);
pthread_mutex_lock(&queue->mutex);
while (pNode) {
pTemp = pNode;
pNode = pNode->next;
free (pTemp);
}
pthread_mutex_unlock(&queue->mutex);
pthread_mutex_destroy(&queue->mutex);
free(queue);

5
src/util/src/tskiplist.c
View File

@ -291,7 +291,10 @@ bool tSkipListIterNext(SSkipListIterator *iter) {
iter->next = SL_NODE_GET_FORWARD_POINTER(iter->cur, 0);
iter->step++;
} else {
if (iter->cur == pSkipList->pHead) return false;
if (iter->cur == pSkipList->pHead) {
tSkipListUnlock(pSkipList);
return false;
}
iter->cur = SL_NODE_GET_BACKWARD_POINTER(iter->cur, 0);
// a new node is inserted into between iter->cur and iter->next, ignore it

1
src/vnode/inc/vnodeInt.h
View File

@ -44,6 +44,7 @@ typedef struct {
int8_t role;
int8_t accessState;
int8_t isFull;
int8_t isCommiting;
uint64_t version; // current version
uint64_t fversion; // version on saved data file
void *wqueue;

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

@ -41,6 +41,7 @@ static void vnodeCtrlFlow(int32_t vgId, int32_t level);
static int32_t vnodeNotifyFileSynced(int32_t vgId, uint64_t fversion);
static void vnodeConfirmForard(int32_t vgId, void *wparam, int32_t code);
static int32_t vnodeWriteToCache(int32_t vgId, void *wparam, int32_t qtype, void *rparam);
static int32_t vnodeGetFileVersion(int32_t vgId, uint64_t *fver);
#ifndef _SYNC
int64_t syncStart(const SSyncInfo *info) { return NULL; }
@ -344,6 +345,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
syncInfo.notifyRole = vnodeNotifyRole;
syncInfo.notifyFlowCtrl = vnodeCtrlFlow;
syncInfo.notifyFileSynced = vnodeNotifyFileSynced;
syncInfo.getFileVersion = vnodeGetFileVersion;
pVnode->sync = syncStart(&syncInfo);
#ifndef _SYNC
@ -512,11 +514,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) {
@ -525,25 +526,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) {
@ -589,18 +590,19 @@ static void vnodeCleanUp(SVnodeObj *pVnode) {
vnodeRelease(pVnode);
}
// TODO: this is a simple implement
static int32_t vnodeProcessTsdbStatus(void *arg, int32_t status, int32_t eno) {
SVnodeObj *pVnode = arg;
if (eno != TSDB_CODE_SUCCESS) {
vError("vgId:%d, failed to commit since %s, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, tstrerror(eno),
pVnode->fversion, pVnode->version);
pVnode->isCommiting = 0;
pVnode->isFull = 1;
return 0;
}
if (status == TSDB_STATUS_COMMIT_START) {
pVnode->isCommiting = 1;
pVnode->fversion = pVnode->version;
vDebug("vgId:%d, start commit, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
if (pVnode->status != TAOS_VN_STATUS_INIT) {
@ -611,6 +613,7 @@ static int32_t vnodeProcessTsdbStatus(void *arg, int32_t status, int32_t eno) {
if (status == TSDB_STATUS_COMMIT_OVER) {
vDebug("vgId:%d, commit over, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
pVnode->isCommiting = 0;
pVnode->isFull = 0;
if (pVnode->status != TAOS_VN_STATUS_INIT) {
walRemoveOneOldFile(pVnode->wal);
@ -676,8 +679,10 @@ static void vnodeCtrlFlow(int32_t vgId, int32_t level) {
return;
}
pVnode->flowctrlLevel = level;
vDebug("vgId:%d, set flowctrl level:%d", pVnode->vgId, level);
if (pVnode->flowctrlLevel != level) {
vDebug("vgId:%d, set flowctrl level from %d to %d", pVnode->vgId, pVnode->flowctrlLevel, level);
pVnode->flowctrlLevel = level;
}
vnodeRelease(pVnode);
}
@ -757,3 +762,22 @@ static int32_t vnodeWriteToCache(int32_t vgId, void *wparam, int32_t qtype, void
vnodeRelease(pVnode);
return code;
}
static int32_t vnodeGetFileVersion(int32_t vgId, uint64_t *fver) {
SVnodeObj *pVnode = vnodeAcquire(vgId);
if (pVnode == NULL) {
vError("vgId:%d, vnode not found while write to cache", vgId);
return -1;
}
int32_t code = 0;
if (pVnode->isCommiting) {
vDebug("vgId:%d, vnode is commiting while get file version", vgId);
code = -1;
} else {
*fver = pVnode->fversion;
}
vnodeRelease(pVnode);
return code;
}

8
src/vnode/src/vnodeWrite.c
View File

@ -282,13 +282,15 @@ static void vnodeFlowCtrlMsgToWQueue(void *param, void *tmrId) {
pWrite->processedCount++;
if (pWrite->processedCount > 100) {
vError("vgId:%d, msg:%p, failed to process since %s", pVnode->vgId, pWrite, tstrerror(code));
vError("vgId:%d, msg:%p, failed to process since %s, retry:%d", pVnode->vgId, pWrite, tstrerror(code),
pWrite->processedCount);
pWrite->processedCount = 1;
dnodeSendRpcVWriteRsp(pWrite->pVnode, pWrite, code);
} else {
code = vnodePerformFlowCtrl(pWrite);
if (code == 0) {
vTrace("vgId:%d, write into vwqueue after flowctrl", pVnode->vgId);
vDebug("vgId:%d, msg:%p, write into vwqueue after flowctrl, retry:%d", pVnode->vgId, pWrite,
pWrite->processedCount);
pWrite->processedCount = 0;
taosWriteQitem(pVnode->wqueue, pWrite->qtype, pWrite);
}
@ -310,7 +312,7 @@ static int32_t vnodePerformFlowCtrl(SVWriteMsg *pWrite) {
void *unUsed = NULL;
taosTmrReset(vnodeFlowCtrlMsgToWQueue, 100, pWrite, tsDnodeTmr, &unUsed);
vTrace("vgId:%d, msg:%p, app:%p, perform flowctrl, count:%d", pVnode->vgId, pWrite, pWrite->rpcMsg.ahandle,
vTrace("vgId:%d, msg:%p, app:%p, perform flowctrl, retry:%d", pVnode->vgId, pWrite, pWrite->rpcMsg.ahandle,
pWrite->processedCount);
return TSDB_CODE_VND_ACTION_IN_PROGRESS;
}

4
src/wal/src/walWrite.c
View File

@ -211,7 +211,7 @@ int32_t walGetWalFile(void *handle, char *fileName, int64_t *fileId) {
code = (*fileId == pWal->fileId) ? 0 : 1;
}
wTrace("vgId:%d, get wal file, code:%d curId:%" PRId64 " outId:%" PRId64, pWal->vgId, code, pWal->fileId, *fileId);
wDebug("vgId:%d, get wal file, code:%d curId:%" PRId64 " outId:%" PRId64, pWal->vgId, code, pWal->fileId, *fileId);
pthread_mutex_unlock(&(pWal->mutex));
return code;
@ -325,7 +325,7 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch
offset = offset + sizeof(SWalHead) + pHead->len;
wTrace("vgId:%d, restore wal, fileId:%" PRId64 " hver:%" PRIu64 " wver:%" PRIu64 " len:%d", pWal->vgId,
wDebug("vgId:%d, restore wal, fileId:%" PRId64 " hver:%" PRIu64 " wver:%" PRIu64 " len:%d", pWal->vgId,
fileId, pHead->version, pWal->version, pHead->len);
pWal->version = pHead->version;

12
tests/examples/c/apitest.c
View File

@ -79,11 +79,11 @@ static int print_result(TAOS_RES* res, int blockFetch) {
if (blockFetch) {
int rows = 0;
while ((rows = taos_fetch_block(res, &row))) {
for (int i = 0; i < rows; i++) {
char temp[256];
taos_print_row(temp, row + i, fields, num_fields);
puts(temp);
}
//for (int i = 0; i < rows; i++) {
// char temp[256];
// taos_print_row(temp, row + i, fields, num_fields);
// puts(temp);
//}
nRows += rows;
}
} else {
@ -498,4 +498,4 @@ int main(int argc, char *argv[]) {
taos_close(taos);
taos_cleanup();
}
}

8
tests/examples/c/subscribe.c
View File

@ -19,10 +19,10 @@ void print_result(TAOS_RES* res, int blockFetch) {
if (blockFetch) {
nRows = taos_fetch_block(res, &row);
for (int i = 0; i < nRows; i++) {
taos_print_row(buf, row + i, fields, num_fields);
puts(buf);
}
//for (int i = 0; i < nRows; i++) {
// taos_print_row(buf, row + i, fields, num_fields);
// puts(buf);
//}
} else {
while ((row = taos_fetch_row(res))) {
taos_print_row(buf, row, fields, num_fields);

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

1
tests/script/general/wal/sync.sim
View File

@ -82,6 +82,7 @@ restful d1 table_rest 1591772800 30000
restful d1 table_rest 1591872800 30000
restful d1 table_rest 1591972800 30000
sleep 1000
sql select * from table_rest;
print rows: $rows
if $rows != 300000 then

4
tests/script/tmp/mnodes.sim
View File

@ -20,6 +20,10 @@ system sh/cfg.sh -n dnode1 -c maxTablesPerVnode -v 20000
system sh/cfg.sh -n dnode2 -c maxTablesPerVnode -v 20000
system sh/cfg.sh -n dnode3 -c maxTablesPerVnode -v 20000
system sh/cfg.sh -n dnode1 -c minTablesPerVnode -v 1000
system sh/cfg.sh -n dnode2 -c minTablesPerVnode -v 1000
system sh/cfg.sh -n dnode3 -c minTablesPerVnode -v 1000
system sh/cfg.sh -n dnode1 -c maxVgroupsPerDb -v 20
system sh/cfg.sh -n dnode2 -c maxVgroupsPerDb -v 20
system sh/cfg.sh -n dnode3 -c maxVgroupsPerDb -v 20

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;
}