提交 dfc9531f 编写于 作者: H hjxilinx

Merge branch '2.0' into liaohj_2

......@@ -12,7 +12,6 @@ ADD_SUBDIRECTORY(client)
ADD_SUBDIRECTORY(query)
ADD_SUBDIRECTORY(kit)
ADD_SUBDIRECTORY(plugins)
ADD_SUBDIRECTORY(sdb)
ADD_SUBDIRECTORY(mnode)
ADD_SUBDIRECTORY(vnode)
ADD_SUBDIRECTORY(dnode)
......
......@@ -13,7 +13,7 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
AUX_SOURCE_DIRECTORY(src SRC)
ADD_EXECUTABLE(taosd ${SRC})
TARGET_LINK_LIBRARIES(taosd mnode sdb taos_static monitor http tsdb)
TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http tsdb)
#IF (TD_CLUSTER)
# TARGET_LINK_LIBRARIES(taosd dcluster)
......
......@@ -25,7 +25,6 @@ extern "C" {
#include "taosdef.h"
#include "taosmsg.h"
#include "taoserror.h"
#include "sdb.h"
#include "tglobalcfg.h"
#include "thash.h"
#include "tidpool.h"
......@@ -54,8 +53,8 @@ typedef struct {
int8_t numOfMnodes;
int32_t numOfDnodes;
char mnodeName[TSDB_DNODE_NAME_LEN + 1];
char reserved[7];
char updateEnd[1];
int8_t reserved[15];
int8_t updateEnd[1];
int syncFd;
void *hbTimer;
void *pSync;
......@@ -79,8 +78,8 @@ typedef struct {
float lbScore; // calc in balance function
int32_t customScore; // config by user
char dnodeName[TSDB_DNODE_NAME_LEN + 1];
char reserved[7];
char updateEnd[1];
int8_t reserved[15];
int8_t updateEnd[1];
SVnodeLoad vload[TSDB_MAX_VNODES];
int32_t status;
uint32_t lastReboot; // time stamp for last reboot
......@@ -121,7 +120,7 @@ typedef struct SSuperTableObj {
int32_t sversion;
int32_t numOfColumns;
int32_t numOfTags;
int8_t reserved[5];
int8_t reserved[15];
int8_t updateEnd[1];
int32_t numOfTables;
int16_t nextColId;
......@@ -153,26 +152,29 @@ typedef struct {
int32_t sversion;
int32_t numOfColumns;
int32_t sqlLen;
int8_t reserved[3];
int8_t reserved[7];
int8_t updateEnd[1];
char* sql; //null-terminated string
int16_t nextColId;
SSchema* schema;
} SNormalTableObj;
struct _db_obj;
typedef struct _vg_obj {
uint32_t vgId;
char dbName[TSDB_DB_NAME_LEN + 1];
int64_t createdTime;
SVnodeGid vnodeGid[TSDB_VNODES_SUPPORT];
int32_t numOfVnodes;
int32_t numOfTables;
int32_t lbIp;
int32_t lbTime;
int8_t lbStatus;
int8_t reserved[14];
int8_t updateEnd[1];
struct _vg_obj *prev, *next;
struct _db_obj *pDb;
int32_t numOfTables;
void * idPool;
STableInfo ** tableList;
} SVgObj;
......@@ -182,8 +184,8 @@ typedef struct _db_obj {
int8_t dirty;
int64_t createdTime;
SDbCfg cfg;
char reserved[15];
char updateEnd[1];
int8_t reserved[15];
int8_t updateEnd[1];
struct _db_obj *prev, *next;
int32_t numOfVgroups;
int32_t numOfTables;
......@@ -201,7 +203,7 @@ typedef struct _user_obj {
int64_t createdTime;
int8_t superAuth;
int8_t writeAuth;
int8_t reserved[16];
int8_t reserved[13];
int8_t updateEnd[1];
struct _user_obj *prev, *next;
struct _acctObj * pAcct;
......
/*
* 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_SDB_H
#define TDENGINE_SDB_H
#ifdef __cplusplus
extern "C" {
#endif
#include "taosmsg.h"
#include "taosdef.h"
extern uint16_t tsMgmtMgmtPort;
extern uint16_t tsMgmtSyncPort;
extern int tsMgmtPeerHBTimer; // seconds
extern char * sdbStatusStr[];
extern char * sdbRoleStr[];
extern int sdbMaster;
extern SRpcIpSet *pSdbIpList;
extern SRpcIpSet *pSdbPublicIpList;
extern void (*sdbWorkAsMasterCallback)(); // this function pointer will be set by taosd
enum _keytype {
SDB_KEYTYPE_STRING, SDB_KEYTYPE_UINT32, SDB_KEYTYPE_AUTO, SDB_KEYTYPE_RECYCLE, SDB_KEYTYPE_MAX
};
#define SDB_ROLE_UNAPPROVED 0
#define SDB_ROLE_UNDECIDED 1
#define SDB_ROLE_MASTER 2
#define SDB_ROLE_SLAVE 3
#define SDB_STATUS_OFFLINE 0
#define SDB_STATUS_UNSYNCED 1
#define SDB_STATUS_SYNCING 2
#define SDB_STATUS_SERVING 3
#define SDB_STATUS_DELETED 4
enum _sdbaction {
SDB_TYPE_INSERT,
SDB_TYPE_DELETE,
SDB_TYPE_UPDATE,
SDB_TYPE_DECODE,
SDB_TYPE_ENCODE,
SDB_TYPE_BEFORE_BATCH_UPDATE,
SDB_TYPE_BATCH_UPDATE,
SDB_TYPE_AFTER_BATCH_UPDATE,
SDB_TYPE_RESET,
SDB_TYPE_DESTROY,
SDB_MAX_ACTION_TYPES
};
#define SDB_MAX_PEERS 4
typedef struct {
uint32_t ip;
uint32_t publicIp;
char ipstr[20];
char zone[12];
char role;
int64_t createdTime;
uint64_t dbVersion;
int64_t lostTime;
char status;
char numOfMnodes;
int numOfDnodes;
char updateEnd[1];
// internal
int syncFd;
void *hbTimer;
void *pSync;
} SSdbPeer;
extern SSdbPeer *sdbPeer[];
#define sdbInited (sdbPeer[0])
#define sdbStatus (sdbPeer[0]->status)
void *sdbOpenTable(int maxRows, int32_t maxRowSize, char *name, uint8_t keyType, char *directory,
void *(*appTool)(char, void *, char *, int, int *));
void *sdbGetRow(void *handle, void *key);
int64_t sdbInsertRow(void *handle, void *row, int rowSize);
int sdbDeleteRow(void *handle, void *key);
int sdbUpdateRow(void *handle, void *row, int updateSize, char isUpdated);
void *sdbFetchRow(void *handle, void *pNode, void **ppRow);
int sdbBatchUpdateRow(void *handle, void *row, int rowSize);
int64_t sdbGetId(void *handle);
int64_t sdbGetNumOfRows(void *handle);
void sdbSaveSnapShot(void *handle);
void sdbCloseTable(void *handle);
int sdbRemovePeerByIp(uint32_t ip);
int sdbInitPeers(char *directory);
void sdbCleanUpPeers();
int64_t sdbGetVersion();
int32_t sdbGetRunStatus();
#define TSDB_MAX_NORMAL_TABLES 10000
#define TSDB_MAX_SUPER_TABLES 1000
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_SDB_H
......@@ -309,6 +309,16 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
#define TSDB_SESSIONS_PER_VNODE (300)
#define TSDB_SESSIONS_PER_DNODE (TSDB_SESSIONS_PER_VNODE * TSDB_MAX_VNODES)
#define TSDB_MAX_MNODES 5
#define TSDB_MAX_DNODES 10
#define TSDB_MAX_ACCOUNTS 10
#define TSDB_MAX_USERS 20
#define TSDB_MAX_DBS 100
#define TSDB_MAX_VGROUPS 1000
#define TSDB_MAX_SUPER_TABLES 100
#define TSDB_MAX_NORMAL_TABLES 1000
#define TSDB_MAX_CHILD_TABLES 100000
enum {
TSDB_PRECISION_MILLI,
TSDB_PRECISION_MICRO,
......
......@@ -2,5 +2,5 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
ADD_SUBDIRECTORY(shell)
ADD_SUBDIRECTORY(taosdemo)
ADD_SUBDIRECTORY(taosdump)
#ADD_SUBDIRECTORY(taosdemo)
#ADD_SUBDIRECTORY(taosdump)
......@@ -13,7 +13,7 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(mnode ${SRC})
TARGET_LINK_LIBRARIES(mnode trpc tutil sdb pthread)
TARGET_LINK_LIBRARIES(mnode trpc tutil pthread)
IF (TD_CLUSTER)
TARGET_LINK_LIBRARIES(mnode)
......
......@@ -39,6 +39,7 @@ int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName
int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMetaMsg *pMeta, bool usePublicIp);
void mgmtDropAllChildTables(SDbObj *pDropDb);
void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable);
#ifdef __cplusplus
}
......
......@@ -22,11 +22,11 @@ extern "C" {
#include "mnode.h"
int32_t mgmtAddVgroupIntoDb(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtAddVgroupIntoDbTail(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtRemoveVgroupFromDb(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtMoveVgroupToTail(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtMoveVgroupToHead(SDbObj *pDb, SVgObj *pVgroup);
void mgmtAddVgroupIntoDb(SVgObj *pVgroup);
void mgmtAddVgroupIntoDbTail(SVgObj *pVgroup);
void mgmtRemoveVgroupFromDb(SVgObj *pVgroup);
void mgmtMoveVgroupToTail(SVgObj *pVgroup);
void mgmtMoveVgroupToHead(SVgObj *pVgroup);
int32_t mgmtInitDbs();
void mgmtCleanUpDbs();
......
......@@ -20,9 +20,15 @@
extern "C" {
#endif
bool mgmtCheckRedirect(void *handle);
int32_t mgmtInitMnodes();
void mgmtCleanupMnodes();
bool mgmtInServerStatus();
bool mgmtIsMaster();
void mgmtGetMnodeIpList(SRpcIpSet *ipSet);
bool mgmtCheckRedirect(void *handle);
void mgmtGetMnodePrivateIpList(SRpcIpSet *ipSet);
void mgmtGetMnodePublicIpList(SRpcIpSet *ipSet);
int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp);
int32_t mgmtRemoveMnode(uint32_t privateIp);
......
......@@ -28,6 +28,8 @@ bool mgmtCheckQhandle(uint64_t qhandle);
void mgmtSaveQhandle(void *qhandle);
void mgmtFreeQhandle(void *qhandle);
void mgmtFreeQueuedMsg(SQueuedMsg *pMsg);
#ifdef __cplusplus
}
#endif
......
......@@ -12,47 +12,61 @@
* 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 "sdbint.h"
int32_t (*mpeerInitMnodesFp)(char *directory) = NULL;
void (*mpeerCleanUpMnodesFp)() = NULL;
int32_t (*mpeerForwardRequestFp)(SSdbTable *pTable, char type, void *cont, int32_t contLen) = NULL;
char *sdbStatusStr[] = {
"offline",
"unsynced",
"syncing",
"serving",
"null"
};
char *sdbRoleStr[] = {
"unauthed",
"undecided",
"master",
"slave",
"null"
};
int32_t sdbForwardDbReqToPeer(SSdbTable *pTable, char type, char *data, int32_t dataLen) {
if (mpeerForwardRequestFp) {
return mpeerForwardRequestFp(pTable, type, data, dataLen);
} else {
return 0;
}
}
int32_t sdbInitPeers(char *directory) {
if (mpeerInitMnodesFp) {
return (*mpeerInitMnodesFp)(directory);
} else {
return 0;
}
}
#ifndef TDENGINE_MNODE_SDB_H
#define TDENGINE_MNODE_SDB_H
#ifdef __cplusplus
extern "C" {
#endif
typedef enum {
SDB_KEY_TYPE_STRING,
SDB_KEY_TYPE_AUTO
} ESdbKeyType;
typedef enum {
SDB_OPER_TYPE_GLOBAL,
SDB_OPER_TYPE_LOCAL
} ESdbOperType;
typedef struct {
ESdbOperType type;
void * table;
void * pObj;
int64_t version;
int32_t maxRowSize;
int32_t rowSize;
void * rowData;
} SSdbOperDesc;
void sdbCleanUpPeers() {
if (mpeerCleanUpMnodesFp) {
(*mpeerCleanUpMnodesFp)();
}
typedef struct {
char *tableName;
int32_t hashSessions;
int32_t maxRowSize;
ESdbKeyType keyType;
int32_t (*insertFp)(SSdbOperDesc *pOper);
int32_t (*deleteFp)(SSdbOperDesc *pOper);
int32_t (*updateFp)(SSdbOperDesc *pOper);
int32_t (*encodeFp)(SSdbOperDesc *pOper);
int32_t (*decodeFp)(SSdbOperDesc *pDesc);
int32_t (*destroyFp)(SSdbOperDesc *pDesc);
} SSdbTableDesc;
void *sdbOpenTable(SSdbTableDesc *desc);
void sdbCloseTable(void *handle);
int32_t sdbInsertRow(SSdbOperDesc *pOper);
int32_t sdbDeleteRow(SSdbOperDesc *pOper);
int32_t sdbUpdateRow(SSdbOperDesc *pOper);
void *sdbGetRow(void *handle, void *key);
void *sdbFetchRow(void *handle, void *pNode, void **ppRow);
int64_t sdbGetNumOfRows(void *handle);
uint64_t sdbGetVersion();
#ifdef __cplusplus
}
#endif
#endif
\ No newline at end of file
......@@ -28,13 +28,12 @@ int32_t mgmtInitVgroups();
void mgmtCleanUpVgroups();
SVgObj *mgmtGetVgroup(int32_t vgId);
SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode);
void mgmtDropAllVgroups(SDbObj *pDropDb);
void mgmtCreateVgroup(SQueuedMsg *pMsg);
void mgmtDropVgroup(SVgObj *pVgroup, void *ahandle);
void mgmtUpdateVgroup(SVgObj *pVgroup);
void mgmtUpdateVgroupIp(SDnodeObj *pDnode);
void mgmtSetVgroupIdPool();
void mgmtAlterVgroup(SVgObj *pVgroup, void *ahandle);
SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb);
void mgmtAddTableIntoVgroup(SVgObj *pVgroup, STableInfo *pTable);
......
......@@ -85,8 +85,10 @@ int32_t mgmtRemoveUserFromAcct(SAcctObj *pAcct, SUserObj *pUser) {
pUser->prev->next = pUser->next;
}
if (pUser->next) pUser->next->prev = pUser->prev;
if (pUser->next) {
pUser->next->prev = pUser->prev;
}
if (pUser->prev == NULL) {
pAcct->pUser = pUser->next;
}
......
......@@ -25,79 +25,47 @@
#include "mgmtAcct.h"
#include "mgmtChildTable.h"
#include "mgmtDb.h"
#include "mgmtDClient.h"
#include "mgmtGrant.h"
#include "mgmtMnode.h"
#include "mgmtProfile.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtDClient.h"
#include "mgmtSuperTable.h"
#include "mgmtTable.h"
#include "mgmtVgroup.h"
void *tsChildTableSdb;
int32_t tsChildTableUpdateSize;
void *(*mgmtChildTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtChildTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtChildTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtChildTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtChildTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtChildTableActionReset(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtChildTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
static void mgmtDestroyChildTable(SChildTableObj *pTable) {
free(pTable);
}
static void mgmtChildTableActionInit() {
mgmtChildTableActionFp[SDB_TYPE_INSERT] = mgmtChildTableActionInsert;
mgmtChildTableActionFp[SDB_TYPE_DELETE] = mgmtChildTableActionDelete;
mgmtChildTableActionFp[SDB_TYPE_UPDATE] = mgmtChildTableActionUpdate;
mgmtChildTableActionFp[SDB_TYPE_ENCODE] = mgmtChildTableActionEncode;
mgmtChildTableActionFp[SDB_TYPE_DECODE] = mgmtChildTableActionDecode;
mgmtChildTableActionFp[SDB_TYPE_RESET] = mgmtChildTableActionReset;
mgmtChildTableActionFp[SDB_TYPE_DESTROY] = mgmtChildTableActionDestroy;
}
void *mgmtChildTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
SChildTableObj *pTable = (SChildTableObj *) row;
memcpy(pTable, str, tsChildTableUpdateSize);
return NULL;
tfree(pTable);
}
void *mgmtChildTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
SChildTableObj *pTable = (SChildTableObj *)row;
mgmtDestroyChildTable(pTable);
return NULL;
static int32_t mgmtChildTableActionDestroy(SSdbOperDesc *pOper) {
mgmtDestroyChildTable(pOper->pObj);
return TSDB_CODE_SUCCESS;
}
void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
SChildTableObj *pTable = (SChildTableObj *) row;
static int32_t mgmtChildTableActionInsert(SSdbOperDesc *pOper) {
SChildTableObj *pTable = pOper->pObj;
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {
mError("ctable:%s, not in vgroup:%d", pTable->tableId, pTable->vgId);
return NULL;
return TSDB_CODE_INVALID_VGROUP_ID;
}
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) {
mError("ctable:%s, vgroup:%d not in db:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
return NULL;
return TSDB_CODE_INVALID_DB;
}
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
if (pAcct == NULL) {
mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
return NULL;
}
if (!sdbMaster) {
int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid != pTable->sid) {
mError("ctable:%s, sid:%d is not matched from the master:%d", pTable->tableId, sid, pTable->sid);
return NULL;
}
return TSDB_CODE_INVALID_ACCT;
}
pTable->superTable = mgmtGetSuperTable(pTable->superTableId);
......@@ -107,83 +75,65 @@ void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ss
mgmtAddTableIntoDb(pDb);
mgmtAddTableIntoVgroup(pVgroup, (STableInfo *) pTable);
if (pVgroup->numOfTables >= pDb->cfg.maxSessions - 1 && pDb->numOfVgroups > 1) {
mgmtMoveVgroupToTail(pDb, pVgroup);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtChildTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
SChildTableObj *pTable = (SChildTableObj *) row;
static int32_t mgmtChildTableActionDelete(SSdbOperDesc *pOper) {
SChildTableObj *pTable = pOper->pObj;
if (pTable->vgId == 0) {
return NULL;
return TSDB_CODE_INVALID_VGROUP_ID;
}
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {
return NULL;
return TSDB_CODE_INVALID_VGROUP_ID;
}
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) {
mError("ctable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
return NULL;
return TSDB_CODE_INVALID_DB;
}
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
if (pAcct == NULL) {
mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
return NULL;
return TSDB_CODE_INVALID_ACCT;
}
mgmtRestoreTimeSeries(pAcct, pTable->superTable->numOfColumns - 1);
mgmtRemoveTableFromDb(pDb);
mgmtRemoveTableFromVgroup(pVgroup, (STableInfo *) pTable);
mgmtRemoveTableFromSuperTable(pTable->superTable);
if (pVgroup->numOfTables > 0) {
mgmtMoveVgroupToHead(pDb, pVgroup);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtChildTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
return mgmtChildTableActionReset(row, str, size, NULL);
static int32_t mgmtChildTableActionUpdate(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
void *mgmtChildTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
SChildTableObj *pTable = (SChildTableObj *) row;
assert(row != NULL && str != NULL);
static int32_t mgmtChildTableActionEncode(SSdbOperDesc *pOper) {
SChildTableObj *pTable = pOper->pObj;
assert(pTable != NULL && pOper->rowData != NULL);
memcpy(str, pTable, tsChildTableUpdateSize);
*ssize = tsChildTableUpdateSize;
memcpy(pOper->rowData, pTable, tsChildTableUpdateSize);
pOper->rowSize = tsChildTableUpdateSize;
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtChildTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
assert(str != NULL);
SChildTableObj *pTable = (SChildTableObj *)calloc(sizeof(SChildTableObj), 1);
if (pTable == NULL) return NULL;
static int32_t mgmtChildTableActionDecode(SSdbOperDesc *pOper) {
assert(pOper->rowData != NULL);
if (size < tsChildTableUpdateSize) {
mgmtDestroyChildTable(pTable);
return NULL;
pOper->pObj = calloc(1, sizeof(SChildTableObj));
if (pOper->pObj == NULL) {
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
memcpy(pTable, str, tsChildTableUpdateSize);
return (void *)pTable;
}
memcpy(pOper->pObj, pOper->rowData, tsChildTableUpdateSize);
void *mgmtChildTableAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
if (mgmtChildTableActionFp[(uint8_t)action] != NULL) {
return (*(mgmtChildTableActionFp[(uint8_t)action]))(row, str, size, ssize);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
int32_t mgmtInitChildTables() {
......@@ -191,12 +141,23 @@ int32_t mgmtInitChildTables() {
void *pLastNode = NULL;
SChildTableObj *pTable = NULL;
mgmtChildTableActionInit();
SChildTableObj tObj;
tsChildTableUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = {
.tableName = "ctables",
.hashSessions = tsMaxTables,
.maxRowSize = tsChildTableUpdateSize,
.keyType = SDB_KEY_TYPE_STRING,
.insertFp = mgmtChildTableActionInsert,
.deleteFp = mgmtChildTableActionDelete,
.updateFp = mgmtChildTableActionUpdate,
.encodeFp = mgmtChildTableActionEncode,
.decodeFp = mgmtChildTableActionDecode,
.destroyFp = mgmtChildTableActionDestroy,
};
tsChildTableSdb = sdbOpenTable(tsMaxTables, tsChildTableUpdateSize,
"ctables", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtChildTableAction);
tsChildTableSdb = sdbOpenTable(&tableDesc);
if (tsChildTableSdb == NULL) {
mError("failed to init child table data");
return -1;
......@@ -212,7 +173,11 @@ int32_t mgmtInitChildTables() {
SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
if (pDb == NULL) {
mError("ctable:%s, failed to get db, discard it", pTable->tableId);
sdbDeleteRow(tsChildTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
......@@ -221,7 +186,11 @@ int32_t mgmtInitChildTables() {
if (pVgroup == NULL) {
mError("ctable:%s, failed to get vgroup:%d sid:%d, discard it", pTable->tableId, pTable->vgId, pTable->sid);
pTable->vgId = 0;
sdbDeleteRow(tsChildTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
......@@ -230,7 +199,11 @@ int32_t mgmtInitChildTables() {
mError("ctable:%s, db:%s not match with vgroup:%d db:%s sid:%d, discard it",
pTable->tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->sid);
pTable->vgId = 0;
sdbDeleteRow(tsChildTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
......@@ -238,28 +211,27 @@ int32_t mgmtInitChildTables() {
if (pVgroup->tableList == NULL) {
mError("ctable:%s, vgroup:%d tableList is null", pTable->tableId, pTable->vgId);
pTable->vgId = 0;
sdbDeleteRow(tsChildTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
pVgroup->tableList[pTable->sid] = (STableInfo*)pTable;
taosIdPoolMarkStatus(pVgroup->idPool, pTable->sid, 1);
SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTable->superTableId);
if (pSuperTable == NULL) {
mError("ctable:%s, stable:%s not exist", pTable->tableId, pTable->superTableId);
pTable->vgId = 0;
sdbDeleteRow(tsChildTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
pTable->superTable = pSuperTable;
mgmtAddTableIntoSuperTable(pSuperTable);
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
mgmtAddTimeSeries(pAcct, pTable->superTable->numOfColumns - 1);
}
mTrace("child table is initialized");
......@@ -271,8 +243,13 @@ void mgmtCleanUpChildTables() {
}
void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableObj *pTable) {
char *pTagData = pMsg->schema + TSDB_TABLE_ID_LEN + 1;
int32_t tagDataLen = htonl(pMsg->contLen) - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
char * pTagData = NULL;
int32_t tagDataLen = 0;
if (pMsg != NULL) {
pTagData = pMsg->schema + TSDB_TABLE_ID_LEN + 1;
tagDataLen = htonl(pMsg->contLen) - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
}
int32_t totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags;
int32_t contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen;
......@@ -305,18 +282,14 @@ void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableObj *pTab
pSchema++;
}
memcpy(pCreate->data + totalCols * sizeof(SSchema), pTagData, tagDataLen);
if (pMsg != NULL) {
memcpy(pCreate->data + totalCols * sizeof(SSchema), pTagData, tagDataLen);
}
return pCreate;
}
void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t tid) {
int32_t numOfTables = sdbGetNumOfRows(tsChildTableSdb);
if (numOfTables >= tsMaxTables) {
mError("ctable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
terrno = TSDB_CODE_TOO_MANY_TABLES;
return NULL;
}
char *pTagData = (char *) pCreate->schema; // it is a tag key
SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTagData);
if (pSuperTable == NULL) {
......@@ -337,12 +310,18 @@ void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
pTable->type = TSDB_CHILD_TABLE;
pTable->createdTime = taosGetTimestampMs();
pTable->uid = (((uint64_t) pTable->vgId) << 40) + ((((uint64_t) pTable->sid) & ((1ul << 24) - 1ul)) << 16) +
((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
(sdbGetVersion() & ((1ul << 16) - 1ul));
pTable->sid = tid;
pTable->vgId = pVgroup->vgId;
pTable->superTable = pSuperTable;
if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) {
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_GLOBAL;
desc.pObj = pTable;
desc.table = tsChildTableSdb;
sdbInsertRow(&desc);
if (sdbInsertRow(&desc) < 0) {
free(pTable);
mError("ctable:%s, update sdb error", pCreate->tableId);
terrno = TSDB_CODE_SDB_ERROR;
......@@ -485,12 +464,45 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
}
if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
sdbDeleteRow(tsChildTableSdb, pTable);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_LOCAL,
.table = tsChildTableSdb,
.pObj = pTable,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
continue;
}
}
mTrace("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
void *pNode = NULL;
void *pLastNode = NULL;
int32_t numOfTables = 0;
SChildTableObj *pTable = NULL;
while (1) {
pNode = sdbFetchRow(tsChildTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (pTable->superTable == pStable) {
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_LOCAL,
.table = tsChildTableSdb,
.pObj = pTable,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables ++;
numOfTables++;
continue;
}
}
mTrace("db:%s, all child tables:%d is dropped", pDropDb->name, numOfTables);
mTrace("stable:%s, all child tables:%d is dropped from sdb", pStable->tableId, numOfTables);
}
\ No newline at end of file
......@@ -42,7 +42,7 @@ int32_t mgmtInitDClient() {
rpcInit.label = "MND-DC";
rpcInit.numOfThreads = 1;
rpcInit.cfp = mgmtProcessRspFromDnode;
rpcInit.sessions = tsMaxDnodes * 5;
rpcInit.sessions = 100;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.user = "mgmtDClient";
......
......@@ -45,7 +45,7 @@ int32_t mgmtInitDServer() {
rpcInit.label = "MND-DS";
rpcInit.numOfThreads = 1;
rpcInit.cfp = mgmtProcessMsgFromDnode;
rpcInit.sessions = tsMaxDnodes * 5;
rpcInit.sessions = 100;
rpcInit.connType = TAOS_CONN_SERVER;
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.afp = mgmtDServerRetrieveAuth;
......
此差异已折叠。
......@@ -547,9 +547,6 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
return ;
}
uint32_t lastPrivateIp = pDnode->privateIp;
uint32_t lastPublicIp = pDnode->publicIp;
pDnode->privateIp = htonl(pStatus->privateIp);
pDnode->publicIp = htonl(pStatus->publicIp);
pDnode->lastReboot = htonl(pStatus->lastReboot);
......@@ -566,11 +563,6 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
mgmtSetDnodeMaxVnodes(pDnode);
}
if (lastPrivateIp != pDnode->privateIp || lastPublicIp != pDnode->publicIp) {
mgmtUpdateVgroupIp(pDnode);
//mgmtUpdateMnodeIp();
}
int32_t openVnodes = htons(pStatus->openVnodes);
for (int32_t j = 0; j < openVnodes; ++j) {
pDnode->vload[j].vgId = htonl(pStatus->load[j].vgId);
......@@ -599,7 +591,7 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
return;
}
mgmtGetMnodeIpList(&pRsp->ipList);
mgmtGetMnodePrivateIpList(&pRsp->ipList);
pRsp->dnodeState.dnodeId = htonl(pDnode->dnodeId);
pRsp->dnodeState.moduleStatus = htonl(pDnode->moduleStatus);
......
......@@ -25,6 +25,8 @@
#include "mgmtDClient.h"
#include "mgmtDnode.h"
#include "mgmtDServer.h"
#include "mgmtMnode.h"
#include "mgmtSdb.h"
#include "mgmtVgroup.h"
#include "mgmtUser.h"
#include "mgmtTable.h"
......@@ -65,7 +67,7 @@ int32_t mgmtStartSystem() {
return 0;
}
tsMgmtTmr = taosTmrInit((tsMaxDnodes + tsMaxShellConns) * 3, 200, 3600000, "MND");
tsMgmtTmr = taosTmrInit((tsMaxShellConns) * 3, 200, 3600000, "MND");
if (tsMgmtTmr == NULL) {
mError("failed to init timer");
return -1;
......@@ -109,8 +111,8 @@ int32_t mgmtStartSystem() {
return -1;
}
if (sdbInitPeers(tsMnodeDir) < 0) {
mError("failed to init peers");
if (mgmtInitMnodes() < 0) {
mError("failed to init mnodes");
return -1;
}
......@@ -125,7 +127,7 @@ int32_t mgmtStartSystem() {
void mgmtStopSystem() {
if (sdbMaster) {
if (mgmtIsMaster()) {
mTrace("it is a master mgmt node, it could not be stopped");
return;
}
......@@ -136,7 +138,7 @@ void mgmtStopSystem() {
void mgmtCleanUpSystem() {
mPrint("starting to clean up mgmt");
sdbCleanUpPeers();
mgmtCleanupMnodes();
mgmtCleanupBalance();
mgmtCleanUpShell();
mgmtCleanupDClient();
......
......@@ -17,48 +17,93 @@
#include "os.h"
#include "trpc.h"
#include "mgmtMnode.h"
#include "mgmtSdb.h"
#include "mgmtUser.h"
int32_t (*mgmtAddMnodeFp)(uint32_t privateIp, uint32_t publicIp) = NULL;
int32_t (*mgmtRemoveMnodeFp)(uint32_t privateIp) = NULL;
int32_t (*mgmtGetMnodesNumFp)() = NULL;
void * (*mgmtGetNextMnodeFp)(SShowObj *pShow, SMnodeObj **pMnode) = NULL;
int32_t (*mpeerAddMnodeFp)(uint32_t privateIp, uint32_t publicIp) = NULL;
int32_t (*mpeerRemoveMnodeFp)(uint32_t privateIp) = NULL;
int32_t (*mpeerGetMnodesNumFp)() = NULL;
void * (*mpeerGetNextMnodeFp)(SShowObj *pShow, SMnodeObj **pMnode) = NULL;
int32_t (*mpeerInitMnodesFp)() = NULL;
void (*mpeerCleanUpMnodesFp)() = NULL;
static SMnodeObj tsMnodeObj = {0};
static bool tsMnodeIsMaster = false;
static bool tsMnodeIsServing = false;
static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static char *mgmtMnodeStatusStr[] = {
"offline",
"unsynced",
"syncing",
"serving",
"null"
};
static char *mgmtMnodeRoleStr[] = {
"unauthed",
"undecided",
"master",
"slave",
"null"
};
int32_t mgmtInitMnodes() {
if (mpeerInitMnodesFp) {
return (*mpeerInitMnodesFp)();
} else {
tsMnodeIsServing = true;
tsMnodeIsMaster = true;
return 0;
}
}
void mgmtCleanupMnodes() {
if (mpeerCleanUpMnodesFp) {
(*mpeerCleanUpMnodesFp)();
}
}
bool mgmtInServerStatus() {
return tsMnodeIsServing;
}
bool mgmtIsMaster() {
return tsMnodeIsMaster;
}
bool mgmtCheckRedirect(void *handle) {
return false;
}
int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp) {
if (mgmtAddMnodeFp) {
return (*mgmtAddMnodeFp)(privateIp, publicIp);
if (mpeerAddMnodeFp) {
return (*mpeerAddMnodeFp)(privateIp, publicIp);
} else {
return 0;
}
}
int32_t mgmtRemoveMnode(uint32_t privateIp) {
if (mgmtRemoveMnodeFp) {
return (*mgmtRemoveMnodeFp)(privateIp);
if (mpeerRemoveMnodeFp) {
return (*mpeerRemoveMnodeFp)(privateIp);
} else {
return 0;
}
}
static int32_t mgmtGetMnodesNum() {
if (mgmtGetMnodesNumFp) {
return (*mgmtGetMnodesNumFp)();
if (mpeerGetMnodesNumFp) {
return (*mpeerGetMnodesNumFp)();
} else {
return 1;
}
}
static void *mgmtGetNextMnode(SShowObj *pShow, SMnodeObj **pMnode) {
if (mgmtGetNextMnodeFp) {
return (*mgmtGetNextMnodeFp)(pShow, pMnode);
if (mpeerGetNextMnodeFp) {
return (*mpeerGetNextMnodeFp)(pShow, pMnode);
} else {
if (*pMnode == NULL) {
*pMnode = &tsMnodeObj;
......@@ -148,11 +193,11 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, sdbStatusStr[(uint8_t)pMnode->status]);
strcpy(pWrite, mgmtMnodeStatusStr[pMnode->status]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, sdbRoleStr[(uint8_t)pMnode->role]);
strcpy(pWrite, mgmtMnodeRoleStr[pMnode->role]);
cols++;
tinet_ntoa(ipstr, pMnode->publicIp);
......@@ -167,7 +212,14 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
return numOfRows;
}
void mgmtGetMnodeIpList(SRpcIpSet *ipSet) {
void mgmtGetMnodePrivateIpList(SRpcIpSet *ipSet) {
ipSet->inUse = 0;
ipSet->port = htons(tsMnodeDnodePort);
ipSet->numOfIps = 1;
ipSet->ip[0] = htonl(inet_addr(tsMasterIp));
}
void mgmtGetMnodePublicIpList(SRpcIpSet *ipSet) {
ipSet->inUse = 0;
ipSet->port = htons(tsMnodeDnodePort);
ipSet->numOfIps = 1;
......
......@@ -26,191 +26,132 @@
#include "mgmtDb.h"
#include "mgmtDClient.h"
#include "mgmtGrant.h"
#include "mgmtMnode.h"
#include "mgmtNormalTable.h"
#include "mgmtSdb.h"
#include "mgmtSuperTable.h"
#include "mgmtTable.h"
#include "mgmtVgroup.h"
void *tsNormalTableSdb;
void *tsNormalTableSdb;
int32_t tsNormalTableUpdateSize;
void *(*mgmtNormalTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtNormalTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtNormalTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtNormalTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtNormalTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtNormalTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtNormalTableActionReset(void *row, char *str, int32_t size, int32_t *ssize);
void *mgmtNormalTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
static void mgmtDestroyNormalTable(SNormalTableObj *pTable) {
free(pTable->schema);
free(pTable->sql);
free(pTable);
}
static void mgmtNormalTableActionInit() {
mgmtNormalTableActionFp[SDB_TYPE_INSERT] = mgmtNormalTableActionInsert;
mgmtNormalTableActionFp[SDB_TYPE_DELETE] = mgmtNormalTableActionDelete;
mgmtNormalTableActionFp[SDB_TYPE_UPDATE] = mgmtNormalTableActionUpdate;
mgmtNormalTableActionFp[SDB_TYPE_ENCODE] = mgmtNormalTableActionEncode;
mgmtNormalTableActionFp[SDB_TYPE_DECODE] = mgmtNormalTableActionDecode;
mgmtNormalTableActionFp[SDB_TYPE_RESET] = mgmtNormalTableActionReset;
mgmtNormalTableActionFp[SDB_TYPE_DESTROY] = mgmtNormalTableActionDestroy;
}
void *mgmtNormalTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
SNormalTableObj *pTable = (SNormalTableObj *) row;
memcpy(pTable, str, tsNormalTableUpdateSize);
int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns) + pTable->sqlLen;
pTable->schema = realloc(pTable->schema, schemaSize);
pTable->sql = (char*)pTable->schema + sizeof(SSchema) * (pTable->numOfColumns);
memcpy(pTable->schema, str + tsNormalTableUpdateSize, schemaSize);
return NULL;
tfree(pTable->schema);
tfree(pTable->sql);
tfree(pTable);
}
void *mgmtNormalTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
SNormalTableObj *pTable = (SNormalTableObj *)row;
mgmtDestroyNormalTable(pTable);
return NULL;
static int32_t mgmtNormalTableActionDestroy(SSdbOperDesc *pOper) {
mgmtDestroyNormalTable(pOper->pObj);
return TSDB_CODE_SUCCESS;
}
void *mgmtNormalTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
SNormalTableObj *pTable = (SNormalTableObj *) row;
static int32_t mgmtNormalTableActionInsert(SSdbOperDesc *pOper) {
SNormalTableObj *pTable = pOper->pObj;
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {
mError("id:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
return NULL;
mError("ntable:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
return TSDB_CODE_INVALID_VGROUP_ID;
}
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) {
mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
return NULL;
mError("ntable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
return TSDB_CODE_INVALID_DB;
}
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
if (pAcct == NULL) {
mError("account not exists");
return NULL;
}
if (!sdbMaster) {
int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid != pTable->sid) {
mError("sid:%d is not matched from the master:%d", sid, pTable->sid);
return NULL;
}
mError("ntable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
return TSDB_CODE_INVALID_ACCT;
}
mgmtAddTimeSeries(pAcct, pTable->numOfColumns - 1);
mgmtAddTableIntoDb(pDb);
mgmtAddTableIntoVgroup(pVgroup, (STableInfo *) pTable);
if (pVgroup->numOfTables >= pDb->cfg.maxSessions - 1 && pDb->numOfVgroups > 1) {
mgmtMoveVgroupToTail(pDb, pVgroup);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtNormalTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
SNormalTableObj *pTable = (SNormalTableObj *) row;
static int32_t mgmtNormalTableActionDelete(SSdbOperDesc *pOper) {
SNormalTableObj *pTable = pOper->pObj;
if (pTable->vgId == 0) {
return NULL;
return TSDB_CODE_INVALID_VGROUP_ID;
}
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {
return NULL;
return TSDB_CODE_INVALID_VGROUP_ID;
}
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) {
mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
return NULL;
mError("ntable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
return TSDB_CODE_INVALID_DB;
}
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
if (pAcct == NULL) {
mError("account not exists");
return NULL;
return TSDB_CODE_INVALID_ACCT;
}
mgmtRestoreTimeSeries(pAcct, pTable->numOfColumns - 1);
mgmtRemoveTableFromDb(pDb);
mgmtRemoveTableFromVgroup(pVgroup, (STableInfo *) pTable);
if (pVgroup->numOfTables > 0) {
mgmtMoveVgroupToHead(pDb, pVgroup);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtNormalTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
return mgmtNormalTableActionReset(row, str, size, NULL);
static int32_t mgmtNormalTableActionUpdate(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
void *mgmtNormalTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
SNormalTableObj *pTable = (SNormalTableObj *) row;
assert(row != NULL && str != NULL);
static int32_t mgmtNormalTableActionEncode(SSdbOperDesc *pOper) {
SNormalTableObj *pTable = pOper->pObj;
assert(pOper->pObj != NULL && pOper->rowData != NULL);
int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema);
if (size < tsNormalTableUpdateSize + schemaSize + 1) {
*ssize = -1;
return NULL;
if (pOper->maxRowSize < tsNormalTableUpdateSize + schemaSize) {
return TSDB_CODE_INVALID_MSG_LEN;
}
memcpy(str, pTable, tsNormalTableUpdateSize);
memcpy(str + tsNormalTableUpdateSize, pTable->schema, schemaSize);
memcpy(str + tsNormalTableUpdateSize + schemaSize, pTable->sql, pTable->sqlLen);
*ssize = tsNormalTableUpdateSize + schemaSize + pTable->sqlLen;
memcpy(pOper->rowData, pTable, tsNormalTableUpdateSize);
memcpy(pOper->rowData + tsNormalTableUpdateSize, pTable->schema, schemaSize);
memcpy(pOper->rowData + tsNormalTableUpdateSize + schemaSize, pTable->sql, pTable->sqlLen);
return NULL;
pOper->rowSize = tsNormalTableUpdateSize + schemaSize + pTable->sqlLen;
return TSDB_CODE_SUCCESS;
}
void *mgmtNormalTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
assert(str != NULL);
static int32_t mgmtNormalTableActionDecode(SSdbOperDesc *pOper) {
assert(pOper->rowData != NULL);
SNormalTableObj *pTable = (SNormalTableObj *)malloc(sizeof(SNormalTableObj));
if (pTable == NULL) {
return NULL;
}
memset(pTable, 0, sizeof(SNormalTableObj));
SNormalTableObj *pTable = (SNormalTableObj *)calloc(1, sizeof(SNormalTableObj));
if (pTable == NULL) TSDB_CODE_SERV_OUT_OF_MEMORY;
if (size < tsNormalTableUpdateSize) {
mgmtDestroyNormalTable(pTable);
return NULL;
}
memcpy(pTable, str, tsNormalTableUpdateSize);
memcpy(pTable, pOper->rowData, tsNormalTableUpdateSize);
int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema);
pTable->schema = (SSchema *)malloc(schemaSize);
if (pTable->schema == NULL) {
mgmtDestroyNormalTable(pTable);
return NULL;
return -1;
}
memcpy(pTable->schema, str + tsNormalTableUpdateSize, schemaSize);
memcpy(pTable->schema, pOper->rowData + tsNormalTableUpdateSize, schemaSize);
pTable->sql = (char *)malloc(pTable->sqlLen);
if (pTable->sql == NULL) {
mgmtDestroyNormalTable(pTable);
return NULL;
}
memcpy(pTable->sql, str + tsNormalTableUpdateSize + schemaSize, pTable->sqlLen);
return (void *)pTable;
}
void *mgmtNormalTableAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
if (mgmtNormalTableActionFp[(uint8_t)action] != NULL) {
return (*(mgmtNormalTableActionFp[(uint8_t)action]))(row, str, size, ssize);
return -1;
}
return NULL;
memcpy(pTable->sql, pOper->rowData + tsNormalTableUpdateSize + schemaSize, pTable->sqlLen);
pOper->pObj = pTable;
return TSDB_CODE_SUCCESS;
}
int32_t mgmtInitNormalTables() {
......@@ -218,12 +159,23 @@ int32_t mgmtInitNormalTables() {
void *pLastNode = NULL;
SNormalTableObj *pTable = NULL;
mgmtNormalTableActionInit();
SNormalTableObj tObj;
tsNormalTableUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
tsNormalTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = {
.tableName = "ntables",
.hashSessions = TSDB_MAX_NORMAL_TABLES,
.maxRowSize = sizeof(SNormalTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS,
.keyType = SDB_KEY_TYPE_STRING,
.insertFp = mgmtNormalTableActionInsert,
.deleteFp = mgmtNormalTableActionDelete,
.updateFp = mgmtNormalTableActionUpdate,
.encodeFp = mgmtNormalTableActionEncode,
.decodeFp = mgmtNormalTableActionDecode,
.destroyFp = mgmtNormalTableActionDestroy,
};
tsNormalTableSdb = sdbOpenTable(tsMaxTables, sizeof(SNormalTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS,
"ntables", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtNormalTableAction);
tsNormalTableSdb = sdbOpenTable(&tableDesc);
if (tsNormalTableSdb == NULL) {
mError("failed to init ntables data");
return -1;
......@@ -237,7 +189,11 @@ int32_t mgmtInitNormalTables() {
SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
if (pDb == NULL) {
mError("ntable:%s, failed to get db, discard it", pTable->tableId);
sdbDeleteRow(tsNormalTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsNormalTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
......@@ -246,7 +202,11 @@ int32_t mgmtInitNormalTables() {
if (pVgroup == NULL) {
mError("ntable:%s, failed to get vgroup:%d sid:%d, discard it", pTable->tableId, pTable->vgId, pTable->sid);
pTable->vgId = 0;
sdbDeleteRow(tsNormalTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsNormalTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
......@@ -255,7 +215,12 @@ int32_t mgmtInitNormalTables() {
mError("ntable:%s, db:%s not match with vgroup:%d db:%s sid:%d, discard it",
pTable->tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->sid);
pTable->vgId = 0;
sdbDeleteRow(tsNormalTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsNormalTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
......@@ -263,19 +228,14 @@ int32_t mgmtInitNormalTables() {
if (pVgroup->tableList == NULL) {
mError("ntable:%s, vgroup:%d tableList is null", pTable->tableId, pTable->vgId);
pTable->vgId = 0;
sdbDeleteRow(tsNormalTableSdb, pTable);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_LOCAL;
desc.pObj = pTable;
desc.table = tsNormalTableSdb;
sdbDeleteRow(&desc);
pNode = pLastNode;
continue;
}
mgmtAddTableIntoVgroup(pVgroup, (STableInfo *)pTable);
//pVgroup->tableList[pTable->sid] = (STableInfo*)pTable;
taosIdPoolMarkStatus(pVgroup->idPool, pTable->sid, 1);
pTable->sql = (char *)pTable->schema + sizeof(SSchema) * pTable->numOfColumns;
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
mgmtAddTimeSeries(pAcct, pTable->numOfColumns - 1);
}
mTrace("ntables is initialized");
......@@ -323,13 +283,6 @@ void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable) {
}
void *mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t sid) {
int32_t numOfTables = sdbGetNumOfRows(tsNormalTableSdb);
if (numOfTables >= TSDB_MAX_NORMAL_TABLES) {
mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_NORMAL_TABLES);
terrno = TSDB_CODE_TOO_MANY_TABLES;
return NULL;
}
SNormalTableObj *pTable = (SNormalTableObj *) calloc(sizeof(SNormalTableObj), 1);
if (pTable == NULL) {
mError("table:%s, failed to alloc memory", pCreate->tableId);
......@@ -341,7 +294,7 @@ void *mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
pTable->type = TSDB_NORMAL_TABLE;
pTable->vgId = pVgroup->vgId;
pTable->createdTime = taosGetTimestampMs();
pTable->uid = (((uint64_t) pTable->createdTime) << 16) + ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
pTable->uid = (((uint64_t) pTable->createdTime) << 16) + (sdbGetVersion() & ((1ul << 16) - 1ul));
pTable->sid = sid;
pTable->sversion = 0;
pTable->numOfColumns = htons(pCreate->numOfColumns);
......@@ -377,7 +330,11 @@ void *mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
mTrace("table:%s, stream sql len:%d sql:%s", pTable->tableId, pTable->sqlLen, pTable->sql);
}
if (sdbInsertRow(tsNormalTableSdb, pTable, 0) < 0) {
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_GLOBAL;
desc.pObj = pTable;
desc.table = tsNormalTableSdb;
if (sdbInsertRow(&desc) < 0) {
mError("table:%s, update sdb error", pTable->tableId);
free(pTable);
terrno = TSDB_CODE_SDB_ERROR;
......@@ -474,7 +431,14 @@ int32_t mgmtAddNormalTableColumn(SNormalTableObj *pTable, SSchema schema[], int3
pTable->sversion++;
pAcct->acctInfo.numOfTimeSeries += ncols;
sdbUpdateRow(tsNormalTableSdb, pTable, 0, 1);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_GLOBAL;
desc.pObj = pTable;
desc.table = tsNormalTableSdb;
desc.rowData = pTable;
desc.rowSize = tsNormalTableUpdateSize;
sdbUpdateRow(&desc);
return TSDB_CODE_SUCCESS;
}
......@@ -503,7 +467,14 @@ int32_t mgmtDropNormalTableColumnByName(SNormalTableObj *pTable, char *colName)
pTable->sversion++;
pAcct->acctInfo.numOfTimeSeries--;
sdbUpdateRow(tsNormalTableSdb, pTable, 0, 1);
SSdbOperDesc desc = {0};
desc.type = SDB_OPER_TYPE_GLOBAL;
desc.pObj = pTable;
desc.table = tsNormalTableSdb;
desc.rowData = pTable;
desc.rowSize = tsNormalTableUpdateSize;
sdbUpdateRow(&desc);
return TSDB_CODE_SUCCESS;
}
......@@ -564,12 +535,17 @@ void mgmtDropAllNormalTables(SDbObj *pDropDb) {
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
sdbDeleteRow(tsNormalTableSdb, pTable);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_LOCAL,
.table = tsNormalTableSdb,
.pObj = pTable,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables ++;
numOfTables++;
continue;
}
}
mTrace("db:%s, all normal tables:%d is dropped", pDropDb->name, numOfTables);
mTrace("db:%s, all normal tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
......@@ -762,3 +762,13 @@ int32_t mgmtInitProfile() {
void mgmtCleanUpProfile() {
}
void mgmtFreeQueuedMsg(SQueuedMsg *pMsg) {
if (pMsg != NULL) {
if (pMsg->pCont != NULL) {
rpcFreeCont(pMsg->pCont);
pMsg->pCont = NULL;
}
free(pMsg);
}
}
\ No newline at end of file
此差异已折叠。
......@@ -32,6 +32,7 @@
#include "mgmtMnode.h"
#include "mgmtNormalTable.h"
#include "mgmtProfile.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtSuperTable.h"
#include "mgmtTable.h"
......@@ -63,7 +64,7 @@ int32_t mgmtInitShell() {
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_HEARTBEAT, mgmtProcessHeartBeatMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CONNECT, mgmtProcessConnectMsg);
tsMgmtTranQhandle = taosInitScheduler(tsMaxDnodes + tsMaxShellConns, 1, "mnodeT");
tsMgmtTranQhandle = taosInitScheduler(tsMaxShellConns, 1, "mnodeT");
int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore / 4.0;
if (numOfThreads < 1) {
......@@ -131,7 +132,7 @@ void mgmtAddToShellQueue(SQueuedMsg *queuedMsg) {
}
static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
if (sdbGetRunStatus() != SDB_STATUS_SERVING) {
if (!mgmtInServerStatus()) {
mgmtProcessMsgWhileNotReady(rpcMsg);
rpcFreeCont(rpcMsg->pCont);
return;
......@@ -309,20 +310,10 @@ static void mgmtProcessHeartBeatMsg(SQueuedMsg *pMsg) {
return;
}
pHBRsp->ipList.inUse = 0;
pHBRsp->ipList.port = htons(tsMnodeShellPort);
pHBRsp->ipList.numOfIps = 0;
if (pSdbPublicIpList != NULL && pSdbIpList != NULL) {
pHBRsp->ipList.numOfIps = htons(pSdbPublicIpList->numOfIps);
if (connInfo.serverIp == tsPublicIpInt) {
for (int i = 0; i < pSdbPublicIpList->numOfIps; ++i) {
pHBRsp->ipList.ip[i] = htonl(pSdbPublicIpList->ip[i]);
}
} else {
for (int i = 0; i < pSdbIpList->numOfIps; ++i) {
pHBRsp->ipList.ip[i] = htonl(pSdbIpList->ip[i]);
}
}
if (connInfo.serverIp == tsPublicIpInt) {
mgmtGetMnodePublicIpList(&pHBRsp->ipList);
} else {
mgmtGetMnodePrivateIpList(&pHBRsp->ipList);
}
/*
......@@ -411,20 +402,11 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
strcpy(pConnectRsp->serverVersion, version);
pConnectRsp->writeAuth = pUser->writeAuth;
pConnectRsp->superAuth = pUser->superAuth;
pConnectRsp->ipList.inUse = 0;
pConnectRsp->ipList.port = htons(tsMnodeShellPort);
pConnectRsp->ipList.numOfIps = 0;
if (pSdbPublicIpList != NULL && pSdbIpList != NULL) {
pConnectRsp->ipList.numOfIps = htons(pSdbPublicIpList->numOfIps);
if (connInfo.serverIp == tsPublicIpInt) {
for (int i = 0; i < pSdbPublicIpList->numOfIps; ++i) {
pConnectRsp->ipList.ip[i] = htonl(pSdbPublicIpList->ip[i]);
}
} else {
for (int i = 0; i < pSdbIpList->numOfIps; ++i) {
pConnectRsp->ipList.ip[i] = htonl(pSdbIpList->ip[i]);
}
}
if (connInfo.serverIp == tsPublicIpInt) {
mgmtGetMnodePublicIpList(&pConnectRsp->ipList);
} else {
mgmtGetMnodePrivateIpList(&pConnectRsp->ipList);
}
connect_over:
......
......@@ -23,6 +23,7 @@
#include "mgmtGrant.h"
#include "mgmtShell.h"
#include "mgmtSuperTable.h"
#include "mgmtSdb.h"
#include "mgmtTable.h"
#include "mgmtUser.h"
#include "mgmtVgroup.h"
......@@ -31,160 +32,106 @@
#include "name.h"
#include "tsqlfunction.h"
static void *tsSuperTableSdb;
static void *tsSuperTableSdb;
static int32_t tsSuperTableUpdateSize;
static void *(*mgmtSuperTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtSuperTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtSuperTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtSuperTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtSuperTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtSuperTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtSuperTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
static int32_t mgmtRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static int32_t mgmtGetShowSuperTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static void mgmtDestroySuperTable(SSuperTableObj *pTable) {
free(pTable->schema);
free(pTable);
static void mgmtDestroySuperTable(SSuperTableObj *pStable) {
tfree(pStable->schema);
tfree(pStable);
}
static void mgmtSuperTableActionInit() {
SSuperTableObj tObj;
tsSuperTableUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
mgmtSuperTableActionFp[SDB_TYPE_INSERT] = mgmtSuperTableActionInsert;
mgmtSuperTableActionFp[SDB_TYPE_DELETE] = mgmtSuperTableActionDelete;
mgmtSuperTableActionFp[SDB_TYPE_UPDATE] = mgmtSuperTableActionUpdate;
mgmtSuperTableActionFp[SDB_TYPE_ENCODE] = mgmtSuperTableActionEncode;
mgmtSuperTableActionFp[SDB_TYPE_DECODE] = mgmtSuperTableActionDecode;
mgmtSuperTableActionFp[SDB_TYPE_RESET] = mgmtSuperTableActionReset;
mgmtSuperTableActionFp[SDB_TYPE_DESTROY] = mgmtSuperTableActionDestroy;
}
void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
SSuperTableObj *pTable = (SSuperTableObj *) row;
memcpy(pTable, str, tsSuperTableUpdateSize);
int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns + pTable->numOfTags);
pTable->schema = realloc(pTable->schema, schemaSize);
memcpy(pTable->schema, str + tsSuperTableUpdateSize, schemaSize);
return NULL;
}
void *mgmtSuperTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
SSuperTableObj *pTable = (SSuperTableObj *) row;
mgmtDestroySuperTable(pTable);
return NULL;
static int32_t mgmtSuperTableActionDestroy(SSdbOperDesc *pOper) {
mgmtDestroySuperTable(pOper->pObj);
return TSDB_CODE_SUCCESS;
}
void *mgmtSuperTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
STableInfo *pTable = (STableInfo *) row;
SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
if (pDb) {
static int32_t mgmtSuperTableActionInsert(SSdbOperDesc *pOper) {
STableInfo *pStable = pOper->pObj;
SDbObj *pDb = mgmtGetDbByTableId(pStable->tableId);
if (pDb != NULL) {
mgmtAddSuperTableIntoDb(pDb);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtSuperTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
STableInfo *pTable = (STableInfo *) row;
SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
if (pDb) {
static int32_t mgmtSuperTableActionDelete(SSdbOperDesc *pOper) {
STableInfo *pStable = pOper->pObj;
SDbObj *pDb = mgmtGetDbByTableId(pStable->tableId);
if (pDb != NULL) {
mgmtRemoveSuperTableFromDb(pDb);
mgmtDropAllChildTablesInStable((SSuperTableObj *)pStable);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtSuperTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
return mgmtSuperTableActionReset(row, str, size, NULL);
static int32_t mgmtSuperTableActionUpdate(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
void *mgmtSuperTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
SSuperTableObj *pTable = (SSuperTableObj *) row;
assert(row != NULL && str != NULL);
static int32_t mgmtSuperTableActionEncode(SSdbOperDesc *pOper) {
SSuperTableObj *pStable = pOper->pObj;
assert(pOper->pObj != NULL && pOper->rowData != NULL);
int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns + pTable->numOfTags);
int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags);
if (size < tsSuperTableUpdateSize + schemaSize + 1) {
*ssize = -1;
return NULL;
if (pOper->maxRowSize < tsSuperTableUpdateSize + schemaSize) {
return TSDB_CODE_INVALID_MSG_LEN;
}
memcpy(str, pTable, tsSuperTableUpdateSize);
memcpy(str + tsSuperTableUpdateSize, pTable->schema, schemaSize);
*ssize = tsSuperTableUpdateSize + schemaSize;
memcpy(pOper->rowData, pStable, tsSuperTableUpdateSize);
memcpy(pOper->rowData + tsSuperTableUpdateSize, pStable->schema, schemaSize);
pOper->rowSize = tsSuperTableUpdateSize + schemaSize;
return NULL;
return TSDB_CODE_SUCCESS;
}
void *mgmtSuperTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
assert(str != NULL);
static int32_t mgmtSuperTableActionDecode(SSdbOperDesc *pOper) {
assert(pOper->rowData != NULL);
SSuperTableObj *pTable = (SSuperTableObj *) malloc(sizeof(SSuperTableObj));
if (pTable == NULL) {
return NULL;
}
memset(pTable, 0, sizeof(SSuperTableObj));
SSuperTableObj *pStable = (SSuperTableObj *) calloc(1, sizeof(SSuperTableObj));
if (pStable == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
if (size < tsSuperTableUpdateSize) {
mgmtDestroySuperTable(pTable);
return NULL;
}
memcpy(pTable, str, tsSuperTableUpdateSize);
memcpy(pStable, pOper->rowData, tsSuperTableUpdateSize);
int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns + pTable->numOfTags);
pTable->schema = malloc(schemaSize);
if (pTable->schema == NULL) {
mgmtDestroySuperTable(pTable);
return NULL;
int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags);
pStable->schema = malloc(schemaSize);
if (pStable->schema == NULL) {
mgmtDestroySuperTable(pStable);
return -1;
}
memcpy(pTable->schema, str + tsSuperTableUpdateSize, schemaSize);
return (void *) pTable;
}
memcpy(pStable->schema, pOper->rowData + tsSuperTableUpdateSize, schemaSize);
pOper->pObj = pStable;
void *mgmtSuperTableAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
if (mgmtSuperTableActionFp[(uint8_t) action] != NULL) {
return (*(mgmtSuperTableActionFp[(uint8_t) action]))(row, str, size, ssize);
}
return NULL;
return TSDB_CODE_SUCCESS;
}
int32_t mgmtInitSuperTables() {
void *pNode = NULL;
void *pLastNode = NULL;
SSuperTableObj *pTable = NULL;
mgmtSuperTableActionInit();
tsSuperTableSdb = sdbOpenTable(tsMaxTables, tsSuperTableUpdateSize + sizeof(SSchema) * TSDB_MAX_COLUMNS,
"stables", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtSuperTableAction);
SSuperTableObj tObj;
tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = {
.tableName = "stables",
.hashSessions = TSDB_MAX_SUPER_TABLES,
.maxRowSize = tsSuperTableUpdateSize + sizeof(SSchema) * TSDB_MAX_COLUMNS,
.keyType = SDB_KEY_TYPE_STRING,
.insertFp = mgmtSuperTableActionInsert,
.deleteFp = mgmtSuperTableActionDelete,
.updateFp = mgmtSuperTableActionUpdate,
.encodeFp = mgmtSuperTableActionEncode,
.decodeFp = mgmtSuperTableActionDecode,
.destroyFp = mgmtSuperTableActionDestroy,
};
tsSuperTableSdb = sdbOpenTable(&tableDesc);
if (tsSuperTableSdb == NULL) {
mError("failed to init stables data");
return -1;
}
pNode = NULL;
while (1) {
pNode = sdbFetchRow(tsSuperTableSdb, pNode, (void **) &pTable);
if (pTable == NULL) {
break;
}
SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
if (pDb == NULL) {
mError("super table:%s, failed to get db, discard it", pTable->tableId);
sdbDeleteRow(tsSuperTableSdb, pTable);
pNode = pLastNode;
continue;
}
mgmtAddSuperTableIntoDb(pDb);
}
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_METRIC, mgmtGetShowSuperTableMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_METRIC, mgmtRetrieveShowSuperTables);
......@@ -197,13 +144,7 @@ void mgmtCleanUpSuperTables() {
}
int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate) {
int32_t numOfTables = sdbGetNumOfRows(tsSuperTableSdb);
if (numOfTables >= TSDB_MAX_SUPER_TABLES) {
mError("stable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_SUPER_TABLES);
return TSDB_CODE_TOO_MANY_TABLES;
}
SSuperTableObj *pStable = (SSuperTableObj *)calloc(sizeof(SSuperTableObj), 1);
SSuperTableObj *pStable = (SSuperTableObj *)calloc(1, sizeof(SSuperTableObj));
if (pStable == NULL) {
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
......@@ -213,7 +154,7 @@ int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate) {
pStable->createdTime = taosGetTimestampMs();
pStable->vgId = 0;
pStable->sid = 0;
pStable->uid = (((uint64_t) pStable->createdTime) << 16) + ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
pStable->uid = (((uint64_t) pStable->createdTime) << 16) + (sdbGetVersion() & ((1ul << 16) - 1ul));
pStable->sversion = 0;
pStable->numOfColumns = htons(pCreate->numOfColumns);
pStable->numOfTags = htons(pCreate->numOfTags);
......@@ -235,13 +176,21 @@ int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate) {
tschema[col].bytes = htons(tschema[col].bytes);
}
if (sdbInsertRow(tsSuperTableSdb, pStable, 0) < 0) {
mError("stable:%s, update sdb error", pStable->tableId);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsSuperTableSdb,
.pObj = pStable,
.rowSize = sizeof(SSuperTableObj) + schemaSize
};
int32_t code = sdbInsertRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
mgmtDestroySuperTable(pStable);
return TSDB_CODE_SDB_ERROR;
} else {
mLPrint("stable:%s, is created, tags:%d cols:%d", pStable->tableId, pStable->numOfTags, pStable->numOfColumns);
return TSDB_CODE_SUCCESS;
}
mPrint("stable:%s, is created, tags:%d cols:%d", pStable->tableId, pStable->numOfTags, pStable->numOfColumns);
return TSDB_CODE_SUCCESS;
}
int32_t mgmtDropSuperTable(SQueuedMsg *newMsg, SDbObj *pDb, SSuperTableObj *pStable) {
......@@ -249,10 +198,14 @@ int32_t mgmtDropSuperTable(SQueuedMsg *newMsg, SDbObj *pDb, SSuperTableObj *pSta
mError("stable:%s, numOfTables:%d not 0", pStable->tableId, pStable->numOfTables);
return TSDB_CODE_OTHERS;
} else {
//TODO: drop child tables
mError("stable:%s, is dropped from sdb", pStable->tableId);
mgmtRemoveSuperTableFromDb(pDb);
return TSDB_CODE_OTHERS;
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsSuperTableSdb,
.pObj = pStable
};
int32_t code = sdbDeleteRow(&oper);
mLPrint("stable:%s, is dropped from sdb, result:%s", pStable->tableId, tstrerror(code));
return code;
}
}
......@@ -261,10 +214,9 @@ void* mgmtGetSuperTable(char *tableId) {
}
void *mgmtGetSuperTableVgroup(SSuperTableObj *pStable) {
//TODO get vgroup of dnodes
SCMSuperTableInfoRsp *rsp = rpcMallocCont(sizeof(SCMSuperTableInfoRsp) + sizeof(uint32_t) * mgmtGetDnodesNum());
rsp->numOfDnodes = 1;
rsp->dnodeIps[0] = 0;
rsp->numOfDnodes = htonl(1);
rsp->dnodeIps[0] = htonl(inet_addr(tsPrivateIp));
return rsp;
}
......@@ -321,7 +273,7 @@ int32_t mgmtAddSuperTableTag(SSuperTableObj *pStable, SSchema schema[], int32_t
pStable->sversion++;
pAcct->acctInfo.numOfTimeSeries += (ntags * pStable->numOfTables);
sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
// sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
mTrace("Succeed to add tag column %s to table %s", schema[0].name, pStable->tableId);
return TSDB_CODE_SUCCESS;
......@@ -354,7 +306,7 @@ int32_t mgmtDropSuperTableTag(SSuperTableObj *pStable, char *tagName) {
int32_t schemaSize = sizeof(SSchema) * (pStable->numOfTags + pStable->numOfColumns);
pStable->schema = realloc(pStable->schema, schemaSize);
sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
// sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
return TSDB_CODE_SUCCESS;
}
......@@ -367,8 +319,8 @@ int32_t mgmtModifySuperTableTagNameByName(SSuperTableObj *pStable, char *oldTagN
return TSDB_CODE_INVALID_MSG_TYPE;
}
int32_t rowSize = 0;
uint32_t len = strlen(newTagName);
int32_t rowSize = 0;
uint32_t len = strlen(newTagName);
if (col >= pStable->numOfTags || len >= TSDB_COL_NAME_LEN || mgmtFindSuperTableTagIndex(pStable, newTagName) >= 0) {
return TSDB_CODE_APP_ERROR;
......@@ -384,9 +336,10 @@ int32_t mgmtModifySuperTableTagNameByName(SSuperTableObj *pStable, char *oldTagN
if (msg == NULL) return TSDB_CODE_APP_ERROR;
memset(msg, 0, size);
mgmtSuperTableActionEncode(pStable, msg, size, &rowSize);
// mgmtSuperTableActionEncode(pStable, msg, size, &rowSize);
int32_t ret = sdbUpdateRow(tsSuperTableSdb, msg, rowSize, 1);
int32_t ret = 0;
// int32_t ret = sdbUpdateRow(tsSuperTableSdb, msg, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
tfree(msg);
if (ret < 0) {
......@@ -448,7 +401,7 @@ int32_t mgmtAddSuperTableColumn(SSuperTableObj *pStable, SSchema schema[], int32
pStable->sversion++;
pAcct->acctInfo.numOfTimeSeries += (ncols * pStable->numOfTables);
sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
// sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
return TSDB_CODE_SUCCESS;
}
......@@ -481,7 +434,7 @@ int32_t mgmtDropSuperTableColumnByName(SSuperTableObj *pStable, char *colName) {
pStable->schema = realloc(pStable->schema, schemaSize);
pAcct->acctInfo.numOfTimeSeries -= (pStable->numOfTables);
sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
// sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
return TSDB_CODE_SUCCESS;
}
......@@ -620,14 +573,19 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
}
if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
sdbDeleteRow(tsSuperTableSdb, pTable);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_LOCAL,
.table = tsSuperTableSdb,
.pObj = pTable,
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables ++;
continue;
}
}
mTrace("db:%s, all super tables:%d is dropped", pDropDb->name, numOfTables);
mTrace("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable) {
......
......@@ -22,10 +22,12 @@
#include "mgmtDClient.h"
#include "mgmtDb.h"
#include "mgmtDnode.h"
#include "mgmtDServer.h"
#include "mgmtGrant.h"
#include "mgmtMnode.h"
#include "mgmtNormalTable.h"
#include "mgmtProfile.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtSuperTable.h"
#include "mgmtUser.h"
......@@ -60,6 +62,7 @@ static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg);
static int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
static void mgmtProcessTableCfgMsg(SRpcMsg *rpcMsg);
int32_t mgmtInitTables() {
int32_t code = mgmtInitSuperTables();
......@@ -77,8 +80,6 @@ int32_t mgmtInitTables() {
return code;
}
mgmtSetVgroupIdPool();
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CREATE_TABLE, mgmtProcessCreateTableMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_DROP_TABLE, mgmtProcessDropTableMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_ALTER_TABLE, mgmtProcessAlterTableMsg);
......@@ -91,6 +92,7 @@ int32_t mgmtInitTables() {
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_TABLE_RSP, mgmtProcessDropTableRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP, mgmtProcessAlterTableRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_STABLE_RSP, mgmtProcessDropStableRsp);
mgmtAddDServerMsgHandle(TSDB_MSG_TYPE_DM_CONFIG_TABLE, mgmtProcessTableCfgMsg);
return TSDB_CODE_SUCCESS;
}
......@@ -201,13 +203,13 @@ int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn)
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "table_name");
strcpy(pSchema[cols].name, "table name");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "created_time");
strcpy(pSchema[cols].name, "create time");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
......@@ -219,7 +221,7 @@ int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn)
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "stable_name");
strcpy(pSchema[cols].name, "stable name");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
......@@ -237,9 +239,6 @@ int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn)
return 0;
}
/*
* remove the hole in result set
*/
static void mgmtVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capacity, SShowObj *pShow) {
if (rows < capacity) {
for (int32_t i = 0; i < numOfCols; ++i) {
......@@ -433,11 +432,13 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
pTable = mgmtCreateChildTable(pCreate, pVgroup, sid);
if (pTable == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
mgmtFreeQueuedMsg(newMsg);
return;
}
pMDCreate = mgmtBuildCreateChildTableMsg(pCreate, (SChildTableObj *) pTable);
if (pMDCreate == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
mgmtFreeQueuedMsg(newMsg);
return;
}
} else {
......@@ -445,11 +446,13 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
pTable = mgmtCreateNormalTable(pCreate, pVgroup, sid);
if (pTable == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
mgmtFreeQueuedMsg(newMsg);
return;
}
pMDCreate = mgmtBuildCreateNormalTableMsg((SNormalTableObj *) pTable);
if (pMDCreate == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
mgmtFreeQueuedMsg(newMsg);
return;
}
}
......@@ -760,9 +763,19 @@ static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
if (rpcMsg->code != TSDB_CODE_SUCCESS) {
if (pTable->type == TSDB_CHILD_TABLE) {
sdbDeleteRow(tsChildTableSdb, pTable);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsChildTableSdb,
.pObj = pTable
};
sdbDeleteRow(&oper);
} else if (pTable->type == TSDB_NORMAL_TABLE){
sdbDeleteRow(tsNormalTableSdb, pTable);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsNormalTableSdb,
.pObj = pTable
};
sdbDeleteRow(&oper);
} else {}
mError("table:%s, failed to create in dnode, reason:%s", pTable->tableId, tstrerror(rpcMsg->code));
mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);
......@@ -816,14 +829,26 @@ static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
}
if (pTable->type == TSDB_CHILD_TABLE) {
if (sdbDeleteRow(tsChildTableSdb, pTable) < 0) {
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsChildTableSdb,
.pObj = pTable
};
int32_t code = sdbDeleteRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
mError("table:%s, update ctables sdb error", pTable->tableId);
mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SDB_ERROR);
free(queueMsg);
return;
}
} else if (pTable->type == TSDB_NORMAL_TABLE){
if (sdbDeleteRow(tsNormalTableSdb, pTable) < 0) {
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsNormalTableSdb,
.pObj = pTable
};
int32_t code = sdbDeleteRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
mError("table:%s, update ntables sdb error", pTable->tableId);
mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SDB_ERROR);
free(queueMsg);
......@@ -844,32 +869,48 @@ static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg) {
mTrace("drop stable rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
}
//
//
//static void mgmtProcessTableCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle) {
// SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) pCont;
// pCfg->dnode = htonl(pCfg->dnode);
// pCfg->vnode = htonl(pCfg->vnode);
// pCfg->sid = htonl(pCfg->sid);
// mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
//
// if (!sdbMaster) {
// mError("dnode:%s, vnode:%d, sid:%d, not master, redirect it", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
// return;
// }
//
// STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
// if (pTable == NULL) {
// mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_INVALID_TABLE, NULL, 0);
// return;
// }
//
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
//
// //TODO
// SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
// mgmtSendCreateTableMsg(NULL, &ipSet, NULL);
//}
//
\ No newline at end of file
static void mgmtProcessTableCfgMsg(SRpcMsg *rpcMsg) {
if (mgmtCheckRedirect(rpcMsg->handle)) return;
SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) rpcMsg->pCont;
pCfg->dnode = htonl(pCfg->dnode);
pCfg->vnode = htonl(pCfg->vnode);
pCfg->sid = htonl(pCfg->sid);
mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
if (pTable == NULL) {
mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_NOT_ACTIVE_TABLE);
return;
}
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_SUCCESS);
SMDCreateTableMsg *pMDCreate = NULL;
if (pTable->type == TSDB_CHILD_TABLE) {
mTrace("table:%s, is a child table, vgroup:%d sid:%d", pTable->tableId, pCfg->vnode, pCfg->sid);
pMDCreate = mgmtBuildCreateChildTableMsg(NULL, (SChildTableObj *) pTable);
if (pMDCreate == NULL) {
return;
}
} else if (pTable->type == TSDB_NORMAL_TABLE) {
mTrace("table:%s, is a normal table, vgroup:%d sid:%d", pTable->tableId, pCfg->vnode, pCfg->sid);
pMDCreate = mgmtBuildCreateNormalTableMsg((SNormalTableObj *) pTable);
if (pMDCreate == NULL) {
return;
}
} else {
mError("table:%s, invalid msg type, vgroup:%d sid:%d", pTable->tableId, pCfg->vnode, pCfg->sid);
}
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
SRpcMsg rpcRsp = {
.handle = NULL,
.pCont = pMDCreate,
.contLen = htonl(pMDCreate->contLen),
.code = 0,
.msgType = TSDB_MSG_TYPE_MD_CREATE_TABLE
};
mgmtSendMsgToDnode(&ipSet, &rpcRsp);
}
......@@ -17,9 +17,11 @@
#include "os.h"
#include "trpc.h"
#include "ttime.h"
#include "tutil.h"
#include "mgmtAcct.h"
#include "mgmtGrant.h"
#include "mgmtMnode.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtUser.h"
......@@ -36,48 +38,84 @@ static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg);
static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg);
static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg);
static void *(*mgmtUserActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionReset(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserAction(char action, void *row, char *str, int32_t size, int32_t *ssize);
static void mgmtUserActionInit();
static int32_t mgmtUserActionDestroy(SSdbOperDesc *pOper) {
tfree(pOper->pObj);
return TSDB_CODE_SUCCESS;
}
int32_t mgmtInitUsers() {
void *pNode = NULL;
SUserObj *pUser = NULL;
SAcctObj *pAcct = NULL;
int32_t numOfUsers = 0;
static int32_t mgmtUserActionInsert(SSdbOperDesc *pOper) {
SUserObj *pUser = pOper->pObj;
SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
mgmtUserActionInit();
if (pAcct != NULL) {
mgmtAddUserIntoAcct(pAcct, pUser);
}
else {
mError("user:%s, acct:%s info not exist in sdb", pUser->user, pUser->acct);
return TSDB_CODE_INVALID_ACCT;
}
SUserObj tObj;
tsUserUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
return TSDB_CODE_SUCCESS;
}
tsUserSdb = sdbOpenTable(tsMaxUsers, tsUserUpdateSize, "users", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtUserAction);
if (tsUserSdb == NULL) {
mError("failed to init user data");
static int32_t mgmtUserActionDelete(SSdbOperDesc *pOper) {
SUserObj *pUser = pOper->pObj;
SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
mgmtRemoveUserFromAcct(pAcct, pUser);
return TSDB_CODE_SUCCESS;
}
static int32_t mgmtUserActionUpdate(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
static int32_t mgmtUserActionEncode(SSdbOperDesc *pOper) {
SUserObj *pUser = pOper->pObj;
if (pOper->maxRowSize < tsUserUpdateSize) {
return -1;
} else {
memcpy(pOper->rowData, pUser, tsUserUpdateSize);
pOper->rowSize = tsUserUpdateSize;
return TSDB_CODE_SUCCESS;
}
}
while (1) {
pNode = sdbFetchRow(tsUserSdb, pNode, (void **)&pUser);
if (pUser == NULL) break;
pUser->prev = NULL;
pUser->next = NULL;
static int32_t mgmtUserActionDecode(SSdbOperDesc *pOper) {
SUserObj *pUser = (SUserObj *) calloc(1, sizeof(SUserObj));
if (pUser == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
pAcct = mgmtGetAcct(pUser->acct);
mgmtAddUserIntoAcct(pAcct, pUser);
memcpy(pUser, pOper->rowData, tsUserUpdateSize);
pOper->pObj = pUser;
return TSDB_CODE_SUCCESS;
}
numOfUsers++;
int32_t mgmtInitUsers() {
SUserObj tObj;
tsUserUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = {
.tableName = "users",
.hashSessions = TSDB_MAX_USERS,
.maxRowSize = tsUserUpdateSize,
.keyType = SDB_KEY_TYPE_STRING,
.insertFp = mgmtUserActionInsert,
.deleteFp = mgmtUserActionDelete,
.updateFp = mgmtUserActionUpdate,
.encodeFp = mgmtUserActionEncode,
.decodeFp = mgmtUserActionDecode,
.destroyFp = mgmtUserActionDestroy,
};
tsUserSdb = sdbOpenTable(&tableDesc);
if (tsUserSdb == NULL) {
mError("failed to init user data");
return -1;
}
pAcct = mgmtGetAcct("root");
SAcctObj *pAcct = mgmtGetAcct("root");
mgmtCreateUser(pAcct, "root", "taosdata");
mgmtCreateUser(pAcct, "monitor", tsInternalPass);
mgmtCreateUser(pAcct, "_root", tsInternalPass);
......@@ -101,16 +139,23 @@ SUserObj *mgmtGetUser(char *name) {
}
static int32_t mgmtUpdateUser(SUserObj *pUser) {
return sdbUpdateRow(tsUserSdb, pUser, 0, 1);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsUserSdb,
.pObj = pUser,
.rowSize = tsUserUpdateSize
};
int32_t code = sdbUpdateRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
tfree(pUser);
code = TSDB_CODE_SDB_ERROR;
}
return code;
}
static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
int32_t numOfUsers = sdbGetNumOfRows(tsUserSdb);
if (numOfUsers >= tsMaxUsers) {
mWarn("numOfUsers:%d, exceed tsMaxUsers:%d", numOfUsers, tsMaxUsers);
return TSDB_CODE_TOO_MANY_USERS;
}
int32_t code = mgmtCheckUserLimit(pAcct);
if (code != 0) {
return code;
......@@ -131,8 +176,7 @@ static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
return code;
}
pUser = malloc(sizeof(SUserObj));
memset(pUser, 0, sizeof(SUserObj));
pUser = calloc(1, sizeof(SUserObj));
strcpy(pUser->user, name);
taosEncryptPass((uint8_t*) pass, strlen(pass), pUser->pass);
strcpy(pUser->acct, pAcct->user);
......@@ -143,8 +187,15 @@ static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
pUser->superAuth = 1;
}
code = TSDB_CODE_SUCCESS;
if (sdbInsertRow(tsUserSdb, pUser, 0) < 0) {
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsUserSdb,
.pObj = pUser,
.rowSize = sizeof(SUserObj)
};
code = sdbInsertRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
tfree(pUser);
code = TSDB_CODE_SDB_ERROR;
}
......@@ -165,9 +216,18 @@ static int32_t mgmtDropUser(SAcctObj *pAcct, char *name) {
return TSDB_CODE_NO_RIGHTS;
}
sdbDeleteRow(tsUserSdb, pUser);
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsUserSdb,
.pObj = pUser
};
return 0;
int32_t code = sdbDeleteRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
code = TSDB_CODE_SDB_ERROR;
}
return code;
}
static int32_t mgmtGetUserMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
......@@ -250,83 +310,6 @@ static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void
return numOfRows;
}
static void mgmtUserActionInit() {
mgmtUserActionFp[SDB_TYPE_INSERT] = mgmtUserActionInsert;
mgmtUserActionFp[SDB_TYPE_DELETE] = mgmtUserActionDelete;
mgmtUserActionFp[SDB_TYPE_UPDATE] = mgmtUserActionUpdate;
mgmtUserActionFp[SDB_TYPE_ENCODE] = mgmtUserActionEncode;
mgmtUserActionFp[SDB_TYPE_DECODE] = mgmtUserActionDecode;
mgmtUserActionFp[SDB_TYPE_RESET] = mgmtUserActionReset;
mgmtUserActionFp[SDB_TYPE_DESTROY] = mgmtUserActionDestroy;
}
static void *mgmtUserAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
if (mgmtUserActionFp[(uint8_t) action] != NULL) {
return (*(mgmtUserActionFp[(uint8_t) action]))(row, str, size, ssize);
}
return NULL;
}
static void *mgmtUserActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
SUserObj *pUser = (SUserObj *) row;
SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
pUser->pAcct = pAcct;
mgmtAddUserIntoAcct(pAcct, pUser);
return NULL;
}
static void *mgmtUserActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
SUserObj *pUser = (SUserObj *) row;
SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
mgmtRemoveUserFromAcct(pAcct, pUser);
return NULL;
}
static void *mgmtUserActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
return mgmtUserActionReset(row, str, size, ssize);
}
static void *mgmtUserActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
SUserObj *pUser = (SUserObj *) row;
if (size < tsUserUpdateSize) {
*ssize = -1;
} else {
memcpy(str, pUser, tsUserUpdateSize);
*ssize = tsUserUpdateSize;
}
return NULL;
}
static void *mgmtUserActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
SUserObj *pUser = (SUserObj *) malloc(sizeof(SUserObj));
if (pUser == NULL) return NULL;
memset(pUser, 0, sizeof(SUserObj));
memcpy(pUser, str, tsUserUpdateSize);
return (void *)pUser;
}
static void *mgmtUserActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
SUserObj *pUser = (SUserObj *)row;
memcpy(pUser, str, tsUserUpdateSize);
return NULL;
}
static void *mgmtUserActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
tfree(row);
return NULL;
}
SUserObj *mgmtGetUserFromConn(void *pConn) {
SRpcConnInfo connInfo;
if (rpcGetConnInfo(pConn, &connInfo) == 0) {
......@@ -393,7 +376,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
memset(pUser->pass, 0, sizeof(pUser->pass));
taosEncryptPass((uint8_t*)pAlter->pass, strlen(pAlter->pass), pUser->pass);
code = mgmtUpdateUser(pUser);
mLPrint("user:%s password is altered by %s, code:%d", pAlter->user, pUser->user, code);
mLPrint("user:%s password is altered by %s, result:%d", pUser->user, pOperUser->user, tstrerror(code));
} else {
code = TSDB_CODE_NO_RIGHTS;
}
......@@ -428,10 +411,6 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
}
if (hasRight) {
//if (pAlter->privilege == 1) { // super
// pUser->superAuth = 1;
// pUser->writeAuth = 1;
//}
if (pAlter->privilege == 2) { // read
pUser->superAuth = 0;
pUser->writeAuth = 0;
......@@ -442,7 +421,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
}
code = mgmtUpdateUser(pUser);
mLPrint("user:%s privilege is altered by %s, code:%d", pAlter->user, pUser->user, code);
mLPrint("user:%s privilege is altered by %s, result:%d", pUser->user, pOperUser->user, tstrerror(code));
} else {
code = TSDB_CODE_NO_RIGHTS;
}
......@@ -492,7 +471,7 @@ static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
if (hasRight) {
code = mgmtDropUser(pUser->pAcct, pDrop->user);
if (code == TSDB_CODE_SUCCESS) {
mLPrint("user:%s is dropped by %s", pDrop->user, pUser->user);
mLPrint("user:%s is dropped by %s, result:%d", pUser->user, pOperUser->user, tstrerror(code));
}
} else {
code = TSDB_CODE_NO_RIGHTS;
......
此差异已折叠。
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(inc)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(sdb ${SRC})
TARGET_LINK_LIBRARIES(sdb trpc)
IF (TD_CLUSTER)
TARGET_LINK_LIBRARIES(sdb)
ENDIF()
ENDIF ()
/*
* 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 _sdbint_header_
#define _sdbint_header_
#include <errno.h>
#include <pthread.h>
#include <semaphore.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "hashint.h"
#include "hashstr.h"
#include "sdb.h"
#include "tchecksum.h"
#include "tlog.h"
#include "trpc.h"
#include "tutil.h"
#define sdbError(...) \
if (sdbDebugFlag & DEBUG_ERROR) { \
tprintf("ERROR MND-SDB ", 255, __VA_ARGS__); \
}
#define sdbWarn(...) \
if (sdbDebugFlag & DEBUG_WARN) { \
tprintf("WARN MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
}
#define sdbTrace(...) \
if (sdbDebugFlag & DEBUG_TRACE) { \
tprintf("MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
}
#define sdbPrint(...) \
{ tprintf("MND-SDB ", 255, __VA_ARGS__); }
#define mpeerError(...) \
if (sdbDebugFlag & DEBUG_ERROR) { \
tprintf("ERROR MND-MPEER ", 255, __VA_ARGS__); \
}
#define mpeerWarn(...) \
if (sdbDebugFlag & DEBUG_WARN) { \
tprintf("WARN MND-MPEER ", sdbDebugFlag, __VA_ARGS__); \
}
#define mpeerTrace(...) \
if (sdbDebugFlag & DEBUG_TRACE) { \
tprintf("MND-MPEER ", sdbDebugFlag, __VA_ARGS__); \
}
#define mpeerPrint(...) \
{ tprintf("MND-MPEER ", 255, __VA_ARGS__); }
#define sdbLError(...) taosLogError(__VA_ARGS__) sdbError(__VA_ARGS__)
#define sdbLWarn(...) taosLogWarn(__VA_ARGS__) sdbWarn(__VA_ARGS__)
#define sdbLPrint(...) taosLogPrint(__VA_ARGS__) sdbPrint(__VA_ARGS__)
#define SDB_MAX_PEERS 4
#define SDB_DELIMITER 0xFFF00F00
#define SDB_ENDCOMMIT 0xAFFFAAAF
typedef struct {
uint64_t swVersion;
int16_t sdbFileVersion;
char reserved[6];
TSCKSUM checkSum;
} SSdbHeader;
typedef struct {
char type;
// short rowSize;
char *row;
} SSdbUpdate;
typedef struct _SSdbTable {
SSdbHeader header;
int maxRows;
int dbId;
int32_t maxRowSize;
char name[TSDB_DB_NAME_LEN];
char fn[128];
int keyType;
uint32_t autoIndex;
int64_t numOfRows;
int64_t id;
int64_t size;
void * iHandle;
int fd;
void *(*appTool)(char, void *, char *, int, int *);
pthread_mutex_t mutex;
SSdbUpdate * update;
int numOfUpdates;
int updatePos;
} SSdbTable;
typedef struct {
int64_t id;
int64_t offset;
int rowSize;
void * row;
} SRowMeta;
typedef struct {
int32_t delimiter;
int32_t rowSize;
int64_t id;
char data[];
} SRowHead;
typedef struct {
uint8_t dbId;
char type;
uint64_t version;
short dataLen;
char data[];
} SForwardMsg;
extern SSdbTable *tableList[];
extern int sdbMaxPeers;
extern int sdbNumOfTables;
extern int64_t sdbVersion;
int sdbForwardDbReqToPeer(SSdbTable *pTable, char type, char *data, int dataLen);
int mpeerRetrieveRows(int fd, SSdbTable *pTable, uint64_t version);
void sdbResetTable(SSdbTable *pTable);
extern const int16_t sdbFileVersion;
#endif
此差异已折叠。
......@@ -103,13 +103,7 @@ extern int tsReplications;
extern int tsNumOfMPeers;
extern int tsMaxShellConns;
extern int tsMaxAccounts;
extern int tsMaxUsers;
extern int tsMaxDbs;
extern int tsMaxTables;
extern int tsMaxDnodes;
extern int tsMaxVGroups;
extern char tsMgmtZone[];
extern char tsLocalIp[];
extern char tsDefaultDB[];
......
......@@ -34,11 +34,7 @@ void taosIdPoolCleanUp(void *handle);
int taosIdPoolNumOfUsed(void *handle);
void taosIdPoolReinit(void *handle);
void taosIdPoolMarkStatus(void *handle, int id, int status);
void taosIdPoolSetFreeList(void *handle);
void taosIdPoolMarkStatus(void *handle, int id);
#ifdef __cplusplus
}
......
......@@ -239,6 +239,25 @@ extern uint32_t cdebugFlag;
#define monitorLWarn(...) taosLogWarn(__VA_ARGS__) monitorWarn(__VA_ARGS__)
#define monitorLPrint(...) taosLogPrint(__VA_ARGS__) monitorPrint(__VA_ARGS__)
#define sdbError(...) \
if (sdbDebugFlag & DEBUG_ERROR) { \
tprintf("ERROR MND-SDB ", 255, __VA_ARGS__); \
}
#define sdbWarn(...) \
if (sdbDebugFlag & DEBUG_WARN) { \
tprintf("WARN MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
}
#define sdbTrace(...) \
if (sdbDebugFlag & DEBUG_TRACE) { \
tprintf("MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
}
#define sdbPrint(...) \
{ tprintf("MND-SDB ", 255, __VA_ARGS__); }
#define sdbLError(...) taosLogError(__VA_ARGS__) sdbError(__VA_ARGS__)
#define sdbLWarn(...) taosLogWarn(__VA_ARGS__) sdbWarn(__VA_ARGS__)
#define sdbLPrint(...) taosLogPrint(__VA_ARGS__) sdbPrint(__VA_ARGS__)
#ifdef __cplusplus
}
#endif
......
......@@ -112,13 +112,7 @@ int tsReplications = TSDB_REPLICA_MIN_NUM;
int tsNumOfMPeers = 3;
int tsMaxShellConns = 2000;
int tsMaxAccounts = 100;
int tsMaxUsers = 1000;
int tsMaxDbs = 1000;
int tsMaxTables = 650000;
int tsMaxDnodes = 1000;
int tsMaxVGroups = 1000;
char tsMgmtZone[16] = "rzone";
int tsMaxTables = 100000;
char tsLocalIp[TSDB_IPv4ADDR_LEN] = {0};
char tsDefaultDB[TSDB_DB_NAME_LEN] = {0};
......@@ -612,28 +606,10 @@ static void doInitGlobalConfig() {
1, 8640000, 0, TSDB_CFG_UTYPE_SECOND);
// mgmt configs
tsInitConfigOption(cfg++, "mgmtZone", tsMgmtZone, TSDB_CFG_VTYPE_STRING,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLUSTER,
0, 0, 16, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "maxAccounts", &tsMaxAccounts, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLUSTER,
1, 1000, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "maxUsers", &tsMaxUsers, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
1, 1000, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "maxDbs", &tsMaxDbs, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
1, 10000, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "maxTables", &tsMaxTables, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
1, 100000000, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "maxDnodes", &tsMaxDnodes, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLUSTER,
1, 1000, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "maxVGroups", &tsMaxVGroups, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
1, 1000000, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "minSlidingTime", &tsMinSlidingTime, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
10, 1000000, 0, TSDB_CFG_UTYPE_MS);
......
......@@ -15,88 +15,81 @@
#include "os.h"
#include "tlog.h"
#include <stdbool.h>
typedef struct {
int maxId;
int numOfFree;
int freeSlot;
int * freeList;
bool * freeList;
pthread_mutex_t mutex;
} id_pool_t;
void *taosInitIdPool(int maxId) {
id_pool_t *pIdPool;
int * idList, i;
if (maxId < 3) maxId = 3;
pIdPool = (id_pool_t *)malloc(sizeof(id_pool_t));
id_pool_t *pIdPool = calloc(1, sizeof(id_pool_t));
if (pIdPool == NULL) return NULL;
idList = (int *)malloc(sizeof(int) * (size_t)maxId);
if (idList == NULL) {
pIdPool->freeList = calloc(maxId, sizeof(bool));
if (pIdPool->freeList == NULL) {
free(pIdPool);
return NULL;
}
memset(pIdPool, 0, sizeof(id_pool_t));
pIdPool->maxId = maxId;
pIdPool->numOfFree = maxId - 1;
pIdPool->numOfFree = maxId;
pIdPool->freeSlot = 0;
pIdPool->freeList = idList;
pthread_mutex_init(&pIdPool->mutex, NULL);
for (i = 1; i < maxId; ++i) idList[i - 1] = i;
pTrace("pool:%p is setup, maxId:%d", pIdPool, pIdPool->maxId);
return (void *)pIdPool;
return pIdPool;
}
int taosAllocateId(void *handle) {
id_pool_t *pIdPool;
int id = -1;
if (handle == NULL) return id;
pIdPool = (id_pool_t *)handle;
if (pIdPool->maxId < 3) pError("pool:%p is messed up, maxId:%d", pIdPool, pIdPool->maxId);
id_pool_t *pIdPool = handle;
if (handle == NULL) {
return -1;
}
if (pthread_mutex_lock(&pIdPool->mutex) != 0) perror("lock pIdPool Mutex");
int slot = -1;
pthread_mutex_lock(&pIdPool->mutex);
if (pIdPool->numOfFree > 0) {
id = pIdPool->freeList[pIdPool->freeSlot];
pIdPool->freeSlot = (pIdPool->freeSlot + 1) % pIdPool->maxId;
pIdPool->numOfFree--;
for (int i = 0; i < pIdPool->maxId; ++i) {
slot = (i + pIdPool->freeSlot) % pIdPool->maxId;
if (!pIdPool->freeList[slot]) {
pIdPool->freeList[slot] = true;
pIdPool->freeSlot = slot + 1;
pIdPool->numOfFree--;
break;
}
}
}
if (pthread_mutex_unlock(&pIdPool->mutex) != 0) perror("unlock pIdPool Mutex");
return id;
pthread_mutex_unlock(&pIdPool->mutex);
return slot + 1;
}
void taosFreeId(void *handle, int id) {
id_pool_t *pIdPool;
int slot;
id_pool_t *pIdPool = handle;
if (handle == NULL) return;
pIdPool = (id_pool_t *)handle;
if (pIdPool->freeList == NULL || pIdPool->maxId == 0) return;
if (id <= 0 || id >= pIdPool->maxId) return;
if (pthread_mutex_lock(&pIdPool->mutex) != 0) perror("lock pIdPool Mutex");
pthread_mutex_lock(&pIdPool->mutex);
slot = (pIdPool->freeSlot + pIdPool->numOfFree) % pIdPool->maxId;
pIdPool->freeList[slot] = id;
pIdPool->numOfFree++;
int slot = (id - 1) % pIdPool->maxId;
if (pIdPool->freeList[slot]) {
pIdPool->freeList[slot] = false;
pIdPool->numOfFree++;
}
if (pthread_mutex_unlock(&pIdPool->mutex) != 0) perror("unlock pIdPool Mutex");
pthread_mutex_unlock(&pIdPool->mutex);
}
void taosIdPoolCleanUp(void *handle) {
id_pool_t *pIdPool;
id_pool_t *pIdPool = handle;
if (handle == NULL) return;
pIdPool = (id_pool_t *)handle;
if (pIdPool == NULL) return;
pTrace("pool:%p is cleaned", pIdPool);
......@@ -110,42 +103,21 @@ void taosIdPoolCleanUp(void *handle) {
}
int taosIdPoolNumOfUsed(void *handle) {
id_pool_t *pIdPool = (id_pool_t *)handle;
return pIdPool->maxId - pIdPool->numOfFree - 1;
}
void taosIdPoolReinit(void *handle) {
id_pool_t *pIdPool;
pIdPool = (id_pool_t *)handle;
pIdPool->numOfFree = 0;
pIdPool->freeSlot = 0;
for (int i = 0; i < pIdPool->maxId; ++i) pIdPool->freeList[i] = 0;
id_pool_t *pIdPool = handle;
return pIdPool->maxId - pIdPool->numOfFree;
}
void taosIdPoolMarkStatus(void *handle, int id, int status) {
id_pool_t *pIdPool = (id_pool_t *)handle;
pIdPool->freeList[id] = status;
}
void taosIdPoolMarkStatus(void *handle, int id) {
id_pool_t *pIdPool = handle;
pthread_mutex_lock(&pIdPool->mutex);
void taosIdPoolSetFreeList(void *handle) {
id_pool_t *pIdPool;
int pos = 0;
pIdPool = (id_pool_t *)handle;
pIdPool->numOfFree = 0;
pIdPool->freeSlot = 0;
for (int i = 1; i < pIdPool->maxId; ++i) {
if (pIdPool->freeList[i] == 0) {
pIdPool->freeList[pos] = i;
pIdPool->numOfFree++;
pos++;
}
int slot = (id - 1) % pIdPool->maxId;
if (!pIdPool->freeList[slot]) {
pIdPool->freeList[slot] = true;
pIdPool->numOfFree--;
}
pthread_mutex_unlock(&pIdPool->mutex);
}
int taosUpdateIdPool(id_pool_t *handle, int maxId) {
......@@ -154,18 +126,14 @@ int taosUpdateIdPool(id_pool_t *handle, int maxId) {
return -1;
}
int *idList, i;
idList = (int *)malloc(sizeof(int) * (size_t)maxId);
int *idList = calloc(maxId, sizeof(bool));
if (idList == NULL) {
return -1;
}
for (i = 1; i < maxId; ++i) {
idList[i - 1] = i;
}
if (pthread_mutex_lock(&pIdPool->mutex) != 0) perror("lock pIdPool Mutex");
pthread_mutex_lock(&pIdPool->mutex);
memcpy(idList, pIdPool->freeList, sizeof(int) * (size_t)pIdPool->maxId);
memcpy(idList, pIdPool->freeList, sizeof(bool) * pIdPool->maxId);
pIdPool->numOfFree += (maxId - pIdPool->maxId);
pIdPool->maxId = maxId;
......@@ -173,7 +141,7 @@ int taosUpdateIdPool(id_pool_t *handle, int maxId) {
pIdPool->freeList = idList;
free(oldIdList);
if (pthread_mutex_unlock(&pIdPool->mutex) != 0) perror("unlock pIdPool Mutex");
pthread_mutex_unlock(&pIdPool->mutex);
return 0;
}
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册