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

TD-2046

This commit is contained in:
Shengliang Guan 2020-11-18 17:55:57 +08:00
parent c1e1dd6427
commit cf5bcc4b58
10 changed files with 95 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -63,11 +63,11 @@ typedef struct SSWriteMsg {
} SSWriteMsg; } SSWriteMsg;
typedef struct { typedef struct {
char * tableName; char * name;
int32_t hashSessions; int32_t hashSessions;
int32_t maxRowSize; int32_t maxRowSize;
int32_t refCountPos; int32_t refCountPos;
ESdbTable tableId; ESdbTable id;
ESdbKey keyType; ESdbKey keyType;
int32_t (*fpInsert)(SSWriteMsg *pWrite); int32_t (*fpInsert)(SSWriteMsg *pWrite);
int32_t (*fpDelete)(SSWriteMsg *pWrite); int32_t (*fpDelete)(SSWriteMsg *pWrite);

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

@ -99,9 +99,9 @@ int32_t mnodeInitAccts() {
SAcctObj tObj; SAcctObj tObj;
tsAcctUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsAcctUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = { SSdbTableDesc desc = {
.tableId = SDB_TABLE_ACCOUNT, .id = SDB_TABLE_ACCOUNT,
.tableName = "accounts", .name = "accounts",
.hashSessions = TSDB_DEFAULT_ACCOUNTS_HASH_SIZE, .hashSessions = TSDB_DEFAULT_ACCOUNTS_HASH_SIZE,
.maxRowSize = tsAcctUpdateSize, .maxRowSize = tsAcctUpdateSize,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -115,13 +115,13 @@ int32_t mnodeInitAccts() {
.fpRestored = mnodeAcctActionRestored .fpRestored = mnodeAcctActionRestored
}; };
tsAcctSdb = sdbOpenTable(&tableDesc); tsAcctSdb = sdbOpenTable(&desc);
if (tsAcctSdb == NULL) { if (tsAcctSdb == NULL) {
mError("table:%s, failed to create hash", tableDesc.tableName); mError("table:%s, failed to create hash", desc.name);
return -1; return -1;
} }
mDebug("table:%s, hash is created", tableDesc.tableName); mDebug("table:%s, hash is created", desc.name);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

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

@ -84,9 +84,9 @@ int32_t mnodeInitCluster() {
SClusterObj tObj; SClusterObj tObj;
tsClusterUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsClusterUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = { SSdbTableDesc desc = {
.tableId = SDB_TABLE_CLUSTER, .id = SDB_TABLE_CLUSTER,
.tableName = "cluster", .name = "cluster",
.hashSessions = TSDB_DEFAULT_CLUSTER_HASH_SIZE, .hashSessions = TSDB_DEFAULT_CLUSTER_HASH_SIZE,
.maxRowSize = tsClusterUpdateSize, .maxRowSize = tsClusterUpdateSize,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -100,16 +100,16 @@ int32_t mnodeInitCluster() {
.fpRestored = mnodeClusterActionRestored .fpRestored = mnodeClusterActionRestored
}; };
tsClusterSdb = sdbOpenTable(&tableDesc); tsClusterSdb = sdbOpenTable(&desc);
if (tsClusterSdb == NULL) { if (tsClusterSdb == NULL) {
mError("table:%s, failed to create hash", tableDesc.tableName); mError("table:%s, failed to create hash", desc.name);
return -1; return -1;
} }
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeGetClusterMeta); mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeGetClusterMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeRetrieveClusters); mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeRetrieveClusters);
mDebug("table:%s, hash is created", tableDesc.tableName); mDebug("table:%s, hash is created", desc.name);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

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

@ -144,9 +144,9 @@ int32_t mnodeInitDbs() {
SDbObj tObj; SDbObj tObj;
tsDbUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsDbUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = { SSdbTableDesc desc = {
.tableId = SDB_TABLE_DB, .id = SDB_TABLE_DB,
.tableName = "dbs", .name = "dbs",
.hashSessions = TSDB_DEFAULT_DBS_HASH_SIZE, .hashSessions = TSDB_DEFAULT_DBS_HASH_SIZE,
.maxRowSize = tsDbUpdateSize, .maxRowSize = tsDbUpdateSize,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -160,7 +160,7 @@ int32_t mnodeInitDbs() {
.fpRestored = mnodeDbActionRestored .fpRestored = mnodeDbActionRestored
}; };
tsDbSdb = sdbOpenTable(&tableDesc); tsDbSdb = sdbOpenTable(&desc);
if (tsDbSdb == NULL) { if (tsDbSdb == NULL) {
mError("failed to init db data"); mError("failed to init db data");
return -1; return -1;

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

@ -171,9 +171,9 @@ int32_t mnodeInitDnodes() {
tsDnodeUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsDnodeUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
pthread_mutex_init(&tsDnodeEpsMutex, NULL); pthread_mutex_init(&tsDnodeEpsMutex, NULL);
SSdbTableDesc tableDesc = { SSdbTableDesc desc = {
.tableId = SDB_TABLE_DNODE, .id = SDB_TABLE_DNODE,
.tableName = "dnodes", .name = "dnodes",
.hashSessions = TSDB_DEFAULT_DNODES_HASH_SIZE, .hashSessions = TSDB_DEFAULT_DNODES_HASH_SIZE,
.maxRowSize = tsDnodeUpdateSize, .maxRowSize = tsDnodeUpdateSize,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -187,7 +187,7 @@ int32_t mnodeInitDnodes() {
.fpRestored = mnodeDnodeActionRestored .fpRestored = mnodeDnodeActionRestored
}; };
tsDnodeSdb = sdbOpenTable(&tableDesc); tsDnodeSdb = sdbOpenTable(&desc);
if (tsDnodeSdb == NULL) { if (tsDnodeSdb == NULL) {
mError("failed to init dnodes data"); mError("failed to init dnodes data");
return -1; return -1;

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

@ -137,9 +137,9 @@ int32_t mnodeInitMnodes() {
SMnodeObj tObj; SMnodeObj tObj;
tsMnodeUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsMnodeUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = { SSdbTableDesc desc = {
.tableId = SDB_TABLE_MNODE, .id = SDB_TABLE_MNODE,
.tableName = "mnodes", .name = "mnodes",
.hashSessions = TSDB_DEFAULT_MNODES_HASH_SIZE, .hashSessions = TSDB_DEFAULT_MNODES_HASH_SIZE,
.maxRowSize = tsMnodeUpdateSize, .maxRowSize = tsMnodeUpdateSize,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -153,7 +153,7 @@ int32_t mnodeInitMnodes() {
.fpRestored = mnodeMnodeActionRestored .fpRestored = mnodeMnodeActionRestored
}; };
tsMnodeSdb = sdbOpenTable(&tableDesc); tsMnodeSdb = sdbOpenTable(&desc);
if (tsMnodeSdb == NULL) { if (tsMnodeSdb == NULL) {
mError("failed to init mnodes data"); mError("failed to init mnodes data");
return -1; return -1;

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

@ -56,8 +56,8 @@ char *actStr[] = {
}; };
typedef struct SSdbTable { typedef struct SSdbTable {
char tableName[SDB_TABLE_LEN]; char name[SDB_TABLE_LEN];
ESdbTable tableId; ESdbTable id;
ESdbKey keyType; ESdbKey keyType;
int32_t hashSessions; int32_t hashSessions;
int32_t maxRowSize; int32_t maxRowSize;
@ -201,7 +201,7 @@ static void sdbRestoreTables() {
totalRows += pTable->numOfRows; totalRows += pTable->numOfRows;
numOfTables++; numOfTables++;
sdbDebug("vgId:1, sdb:%s is restored, rows:%" PRId64, pTable->tableName, pTable->numOfRows); sdbDebug("vgId:1, sdb:%s is restored, rows:%" PRId64, pTable->name, pTable->numOfRows);
} }
sdbInfo("vgId:1, sdb is restored, mver:%" PRIu64 " rows:%d tables:%d", tsSdbMgmt.version, totalRows, numOfTables); sdbInfo("vgId:1, sdb is restored, mver:%" PRIu64 " rows:%d tables:%d", tsSdbMgmt.version, totalRows, numOfTables);
@ -248,6 +248,21 @@ static void sdbNotifyRole(void *ahandle, int8_t role) {
sdbUpdateMnodeRoles(); sdbUpdateMnodeRoles();
} }
// failed to forward, need revert insert
static void sdbHandleFailedConfirm(SSWriteMsg *pWrite) {
SWalHead *pHead = (SWalHead *)((char *)pWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK);
int32_t action = pHead->msgType % 10;
sdbError("vgId:1, row:%p:%s hver:%" PRIu64 " action:%s, failed to foward since %s", pWrite->pRow,
sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version, actStr[action], tstrerror(pWrite->code));
// It's better to create a table in two stages, create it first and then set it success
if (action == SDB_ACTION_INSERT) {
SSWriteMsg wmsg = {.type = SDB_OPER_GLOBAL, .pTable = pWrite->pTable, .pRow = pWrite->pRow};
sdbDeleteRow(&wmsg);
}
}
FORCE_INLINE FORCE_INLINE
static void sdbConfirmForward(void *ahandle, void *wparam, int32_t code) { static void sdbConfirmForward(void *ahandle, void *wparam, int32_t code) {
if (wparam == NULL) return; if (wparam == NULL) return;
@ -263,26 +278,12 @@ static void sdbConfirmForward(void *ahandle, void *wparam, int32_t code) {
if (pMsg != NULL) sdbTrace("vgId:1, msg:%p is confirmed, code:%x", pMsg, code); if (pMsg != NULL) sdbTrace("vgId:1, msg:%p is confirmed, code:%x", pMsg, code);
} }
// failed to forward, need revert insert if (pWrite->code != TSDB_CODE_SUCCESS) sdbHandleFailedConfirm(pWrite);
if (pWrite->code != TSDB_CODE_SUCCESS) {
SWalHead *pHead = (SWalHead *)((char *)pWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK);
int32_t action = pHead->msgType % 10;
sdbError("vgId:1, row:%p:%s hver:%" PRIu64 " action:%s, failed to foward since %s", pWrite->pRow,
sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version, actStr[action], tstrerror(pWrite->code));
if (action == SDB_ACTION_INSERT) {
// It's better to create a table in two stages, create it first and then set it success
SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL,
.pTable = pWrite->pTable,
.pRow = pWrite->pRow
};
sdbDeleteRow(&wmsg);
}
}
if (pWrite->fpRsp != NULL) { if (pWrite->fpRsp != NULL) {
pWrite->code = (*pWrite->fpRsp)(pMsg, pWrite->code); pWrite->code = (*pWrite->fpRsp)(pMsg, pWrite->code);
} }
dnodeSendRpcMWriteRsp(pMsg, pWrite->code); dnodeSendRpcMWriteRsp(pMsg, pWrite->code);
// if ahandle, means this func is called by sdb write // if ahandle, means this func is called by sdb write
@ -439,7 +440,7 @@ void sdbIncRef(void *tparam, void *pRow) {
SSdbTable *pTable = tparam; SSdbTable *pTable = tparam;
int32_t * pRefCount = (int32_t *)(pRow + pTable->refCountPos); int32_t * pRefCount = (int32_t *)(pRow + pTable->refCountPos);
int32_t refCount = atomic_add_fetch_32(pRefCount, 1); int32_t refCount = atomic_add_fetch_32(pRefCount, 1);
sdbTrace("vgId:1, sdb:%s, inc ref to row:%p:%s:%d", pTable->tableName, pRow, sdbGetRowStr(pTable, pRow), refCount); sdbTrace("vgId:1, sdb:%s, inc ref to row:%p:%s:%d", pTable->name, pRow, sdbGetRowStr(pTable, pRow), refCount);
} }
void sdbDecRef(void *tparam, void *pRow) { void sdbDecRef(void *tparam, void *pRow) {
@ -448,11 +449,11 @@ void sdbDecRef(void *tparam, void *pRow) {
SSdbTable *pTable = tparam; SSdbTable *pTable = tparam;
int32_t * pRefCount = (int32_t *)(pRow + pTable->refCountPos); int32_t * pRefCount = (int32_t *)(pRow + pTable->refCountPos);
int32_t refCount = atomic_sub_fetch_32(pRefCount, 1); int32_t refCount = atomic_sub_fetch_32(pRefCount, 1);
sdbTrace("vgId:1, sdb:%s, dec ref to row:%p:%s:%d", pTable->tableName, pRow, sdbGetRowStr(pTable, pRow), refCount); sdbTrace("vgId:1, sdb:%s, dec ref to row:%p:%s:%d", pTable->name, pRow, sdbGetRowStr(pTable, pRow), refCount);
int32_t *updateEnd = pRow + pTable->refCountPos - 4; int32_t *updateEnd = pRow + pTable->refCountPos - 4;
if (refCount <= 0 && *updateEnd) { if (refCount <= 0 && *updateEnd) {
sdbTrace("vgId:1, sdb:%s, row:%p:%s:%d destroyed", pTable->tableName, pRow, sdbGetRowStr(pTable, pRow), refCount); sdbTrace("vgId:1, sdb:%s, row:%p:%s:%d destroyed", pTable->name, pRow, sdbGetRowStr(pTable, pRow), refCount);
SSWriteMsg wmsg = {.pRow = pRow}; SSWriteMsg wmsg = {.pRow = pRow};
(*pTable->fpDestroy)(&wmsg); (*pTable->fpDestroy)(&wmsg);
} }
@ -512,12 +513,12 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
atomic_add_fetch_32(&pTable->autoIndex, 1); atomic_add_fetch_32(&pTable->autoIndex, 1);
} }
sdbDebug("vgId:1, sdb:%s, insert key:%s to hash, rowSize:%d rows:%" PRId64 ", msg:%p", pTable->tableName, sdbDebug("vgId:1, sdb:%s, insert key:%s to hash, rowSize:%d rows:%" PRId64 ", msg:%p", pTable->name,
sdbGetRowStr(pTable, pWrite->pRow), pWrite->rowSize, pTable->numOfRows, pWrite->pMsg); sdbGetRowStr(pTable, pWrite->pRow), pWrite->rowSize, pTable->numOfRows, pWrite->pMsg);
int32_t code = (*pTable->fpInsert)(pWrite); int32_t code = (*pTable->fpInsert)(pWrite);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
sdbError("vgId:1, sdb:%s, failed to insert key:%s to hash, remove it", pTable->tableName, sdbError("vgId:1, sdb:%s, failed to insert key:%s to hash, remove it", pTable->name,
sdbGetRowStr(pTable, pWrite->pRow)); sdbGetRowStr(pTable, pWrite->pRow));
sdbDeleteHash(pTable, pWrite); sdbDeleteHash(pTable, pWrite);
} }
@ -529,7 +530,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
int32_t *updateEnd = pWrite->pRow + pTable->refCountPos - 4; int32_t *updateEnd = pWrite->pRow + pTable->refCountPos - 4;
bool set = atomic_val_compare_exchange_32(updateEnd, 0, 1) == 0; bool set = atomic_val_compare_exchange_32(updateEnd, 0, 1) == 0;
if (!set) { if (!set) {
sdbError("vgId:1, sdb:%s, failed to delete key:%s from hash, for it already removed", pTable->tableName, sdbError("vgId:1, sdb:%s, failed to delete key:%s from hash, for it already removed", pTable->name,
sdbGetRowStr(pTable, pWrite->pRow)); sdbGetRowStr(pTable, pWrite->pRow));
return TSDB_CODE_MND_SDB_OBJ_NOT_THERE; return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
} }
@ -548,7 +549,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
atomic_sub_fetch_32(&pTable->numOfRows, 1); atomic_sub_fetch_32(&pTable->numOfRows, 1);
sdbDebug("vgId:1, sdb:%s, delete key:%s from hash, numOfRows:%" PRId64 ", msg:%p", pTable->tableName, sdbDebug("vgId:1, sdb:%s, delete key:%s from hash, numOfRows:%" PRId64 ", msg:%p", pTable->name,
sdbGetRowStr(pTable, pWrite->pRow), pTable->numOfRows, pWrite->pMsg); sdbGetRowStr(pTable, pWrite->pRow), pTable->numOfRows, pWrite->pMsg);
sdbDecRef(pTable, pWrite->pRow); sdbDecRef(pTable, pWrite->pRow);
@ -557,7 +558,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
} }
static int32_t sdbUpdateHash(SSdbTable *pTable, SSWriteMsg *pWrite) { static int32_t sdbUpdateHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
sdbDebug("vgId:1, sdb:%s, update key:%s in hash, numOfRows:%" PRId64 ", msg:%p", pTable->tableName, sdbDebug("vgId:1, sdb:%s, update key:%s in hash, numOfRows:%" PRId64 ", msg:%p", pTable->name,
sdbGetRowStr(pTable, pWrite->pRow), pTable->numOfRows, pWrite->pMsg); sdbGetRowStr(pTable, pWrite->pRow), pTable->numOfRows, pWrite->pMsg);
(*pTable->fpUpdate)(pWrite); (*pTable->fpUpdate)(pWrite);
@ -584,12 +585,12 @@ static int sdbWrite(void *wparam, void *hparam, int32_t qtype, void *unused) {
if (pHead->version <= tsSdbMgmt.version) { if (pHead->version <= tsSdbMgmt.version) {
pthread_mutex_unlock(&tsSdbMgmt.mutex); pthread_mutex_unlock(&tsSdbMgmt.mutex);
sdbDebug("vgId:1, sdb:%s, failed to restore %s key:%s from source(%d), hver:%" PRIu64 " too large, mver:%" PRIu64, sdbDebug("vgId:1, sdb:%s, failed to restore %s key:%s from source(%d), hver:%" PRIu64 " too large, mver:%" PRIu64,
pTable->tableName, actStr[action], sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version); pTable->name, actStr[action], sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (pHead->version != tsSdbMgmt.version + 1) { } else if (pHead->version != tsSdbMgmt.version + 1) {
pthread_mutex_unlock(&tsSdbMgmt.mutex); pthread_mutex_unlock(&tsSdbMgmt.mutex);
sdbError("vgId:1, sdb:%s, failed to restore %s key:%s from source(%d), hver:%" PRIu64 " too large, mver:%" PRIu64, sdbError("vgId:1, sdb:%s, failed to restore %s key:%s from source(%d), hver:%" PRIu64 " too large, mver:%" PRIu64,
pTable->tableName, actStr[action], sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version); pTable->name, actStr[action], sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version);
return TSDB_CODE_SYN_INVALID_VERSION; return TSDB_CODE_SYN_INVALID_VERSION;
} else { } else {
tsSdbMgmt.version = pHead->version; tsSdbMgmt.version = pHead->version;
@ -612,19 +613,19 @@ static int sdbWrite(void *wparam, void *hparam, int32_t qtype, void *unused) {
if (syncCode <= 0) pWrite->processedCount = 1; if (syncCode <= 0) pWrite->processedCount = 1;
if (syncCode < 0) { if (syncCode < 0) {
sdbError("vgId:1, sdb:%s, failed to forward req since %s action:%s key:%s hver:%" PRIu64 ", msg:%p", pTable->tableName, sdbError("vgId:1, sdb:%s, failed to forward req since %s action:%s key:%s hver:%" PRIu64 ", msg:%p", pTable->name,
tstrerror(syncCode), actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version, pWrite->pMsg); tstrerror(syncCode), actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version, pWrite->pMsg);
} else if (syncCode > 0) { } else if (syncCode > 0) {
sdbDebug("vgId:1, sdb:%s, forward req is sent, action:%s key:%s hver:%" PRIu64 ", msg:%p", pTable->tableName, sdbDebug("vgId:1, sdb:%s, forward req is sent, action:%s key:%s hver:%" PRIu64 ", msg:%p", pTable->name,
actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version, pWrite->pMsg); actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version, pWrite->pMsg);
} else { } else {
sdbTrace("vgId:1, sdb:%s, no need to send fwd req, action:%s key:%s hver:%" PRIu64 ", msg:%p", pTable->tableName, sdbTrace("vgId:1, sdb:%s, no need to send fwd req, action:%s key:%s hver:%" PRIu64 ", msg:%p", pTable->name,
actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version, pWrite->pMsg); actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version, pWrite->pMsg);
} }
return syncCode; return syncCode;
} }
sdbDebug("vgId:1, sdb:%s, record from wal/fwd is disposed, action:%s key:%s hver:%" PRIu64, pTable->tableName, sdbDebug("vgId:1, sdb:%s, record from wal/fwd is disposed, action:%s key:%s hver:%" PRIu64, pTable->name,
actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version); actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version);
// even it is WAL/FWD, it shall be called to update version in sync // even it is WAL/FWD, it shall be called to update version in sync
@ -638,7 +639,7 @@ static int sdbWrite(void *wparam, void *hparam, int32_t qtype, void *unused) {
} else if (action == SDB_ACTION_DELETE) { } else if (action == SDB_ACTION_DELETE) {
void *pRow = sdbGetRowMeta(pTable, pHead->cont); void *pRow = sdbGetRowMeta(pTable, pHead->cont);
if (pRow == NULL) { if (pRow == NULL) {
sdbDebug("vgId:1, sdb:%s, object:%s not exist in hash, ignore delete action", pTable->tableName, sdbDebug("vgId:1, sdb:%s, object:%s not exist in hash, ignore delete action", pTable->name,
sdbGetKeyStr(pTable, pHead->cont)); sdbGetKeyStr(pTable, pHead->cont));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -647,7 +648,7 @@ static int sdbWrite(void *wparam, void *hparam, int32_t qtype, void *unused) {
} else if (action == SDB_ACTION_UPDATE) { } else if (action == SDB_ACTION_UPDATE) {
void *pRow = sdbGetRowMeta(pTable, pHead->cont); void *pRow = sdbGetRowMeta(pTable, pHead->cont);
if (pRow == NULL) { if (pRow == NULL) {
sdbDebug("vgId:1, sdb:%s, object:%s not exist in hash, ignore update action", pTable->tableName, sdbDebug("vgId:1, sdb:%s, object:%s not exist in hash, ignore update action", pTable->name,
sdbGetKeyStr(pTable, pHead->cont)); sdbGetKeyStr(pTable, pHead->cont));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -664,7 +665,7 @@ int32_t sdbInsertRow(SSWriteMsg *pWrite) {
if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE; if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
if (sdbGetRowFromObj(pTable, pWrite->pRow)) { if (sdbGetRowFromObj(pTable, pWrite->pRow)) {
sdbError("vgId:1, sdb:%s, failed to insert key:%s, already exist", pTable->tableName, sdbError("vgId:1, sdb:%s, failed to insert key:%s since it already exist", pTable->name,
sdbGetRowStr(pTable, pWrite->pRow)); sdbGetRowStr(pTable, pWrite->pRow));
sdbDecRef(pTable, pWrite->pRow); sdbDecRef(pTable, pWrite->pRow);
return TSDB_CODE_MND_SDB_OBJ_ALREADY_THERE; return TSDB_CODE_MND_SDB_OBJ_ALREADY_THERE;
@ -674,14 +675,14 @@ int32_t sdbInsertRow(SSWriteMsg *pWrite) {
*((uint32_t *)pWrite->pRow) = atomic_add_fetch_32(&pTable->autoIndex, 1); *((uint32_t *)pWrite->pRow) = atomic_add_fetch_32(&pTable->autoIndex, 1);
// let vgId increase from 2 // let vgId increase from 2
if (pTable->autoIndex == 1 && strcmp(pTable->tableName, "vgroups") == 0) { if (pTable->autoIndex == 1 && strcmp(pTable->name, "vgroups") == 0) {
*((uint32_t *)pWrite->pRow) = atomic_add_fetch_32(&pTable->autoIndex, 1); *((uint32_t *)pWrite->pRow) = atomic_add_fetch_32(&pTable->autoIndex, 1);
} }
} }
int32_t code = sdbInsertHash(pTable, pWrite); int32_t code = sdbInsertHash(pTable, pWrite);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
sdbError("vgId:1, sdb:%s, failed to insert into hash", pTable->tableName); sdbError("vgId:1, sdb:%s, failed to insert into hash", pTable->name);
return code; return code;
} }
@ -707,7 +708,7 @@ int32_t sdbInsertRowImp(SSWriteMsg *pWrite) {
SWalHead *pHead = (SWalHead *)((char *)pNewWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK); SWalHead *pHead = (SWalHead *)((char *)pNewWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK);
pHead->version = 0; pHead->version = 0;
pHead->len = pWrite->rowSize; pHead->len = pWrite->rowSize;
pHead->msgType = pTable->tableId * 10 + SDB_ACTION_INSERT; pHead->msgType = pTable->id * 10 + SDB_ACTION_INSERT;
pWrite->rowData = pHead->cont; pWrite->rowData = pHead->cont;
(*pTable->fpEncode)(pWrite); (*pTable->fpEncode)(pWrite);
@ -717,7 +718,7 @@ int32_t sdbInsertRowImp(SSWriteMsg *pWrite) {
if (pNewWrite->pMsg != NULL) { if (pNewWrite->pMsg != NULL) {
sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s, insert action is add to sdb queue", pNewWrite->pMsg->rpcMsg.ahandle, sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s, insert action is add to sdb queue", pNewWrite->pMsg->rpcMsg.ahandle,
pNewWrite->pMsg, pTable->tableName, pWrite->pRow, sdbGetRowStr(pTable, pWrite->pRow)); pNewWrite->pMsg, pTable->name, pWrite->pRow, sdbGetRowStr(pTable, pWrite->pRow));
} }
sdbIncRef(pNewWrite->pTable, pNewWrite->pRow); sdbIncRef(pNewWrite->pTable, pNewWrite->pRow);
@ -740,7 +741,7 @@ int32_t sdbDeleteRow(SSWriteMsg *pWrite) {
void *pRow = sdbGetRowMetaFromObj(pTable, pWrite->pRow); void *pRow = sdbGetRowMetaFromObj(pTable, pWrite->pRow);
if (pRow == NULL) { if (pRow == NULL) {
sdbDebug("vgId:1, sdb:%s, record is not there, delete failed", pTable->tableName); sdbDebug("vgId:1, sdb:%s, record is not there, delete failed", pTable->name);
return TSDB_CODE_MND_SDB_OBJ_NOT_THERE; return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
} }
@ -748,7 +749,7 @@ int32_t sdbDeleteRow(SSWriteMsg *pWrite) {
int32_t code = sdbDeleteHash(pTable, pWrite); int32_t code = sdbDeleteHash(pTable, pWrite);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
sdbError("vgId:1, sdb:%s, failed to delete from hash", pTable->tableName); sdbError("vgId:1, sdb:%s, failed to delete from hash", pTable->name);
sdbDecRef(pTable, pWrite->pRow); sdbDecRef(pTable, pWrite->pRow);
return code; return code;
} }
@ -775,7 +776,7 @@ int32_t sdbDeleteRowImp(SSWriteMsg *pWrite) {
SWalHead *pHead = (SWalHead *)((void *)pNewWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK); SWalHead *pHead = (SWalHead *)((void *)pNewWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK);
pHead->version = 0; pHead->version = 0;
pHead->msgType = pTable->tableId * 10 + SDB_ACTION_DELETE; pHead->msgType = pTable->id * 10 + SDB_ACTION_DELETE;
pWrite->rowData = pHead->cont; pWrite->rowData = pHead->cont;
(*pTable->fpEncode)(pWrite); (*pTable->fpEncode)(pWrite);
@ -785,7 +786,7 @@ int32_t sdbDeleteRowImp(SSWriteMsg *pWrite) {
if (pNewWrite->pMsg != NULL) { if (pNewWrite->pMsg != NULL) {
sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s, delete action is add to sdb queue", pNewWrite->pMsg->rpcMsg.ahandle, sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s, delete action is add to sdb queue", pNewWrite->pMsg->rpcMsg.ahandle,
pNewWrite->pMsg, pTable->tableName, pWrite->pRow, sdbGetRowStr(pTable, pWrite->pRow)); pNewWrite->pMsg, pTable->name, pWrite->pRow, sdbGetRowStr(pTable, pWrite->pRow));
} }
taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewWrite); taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewWrite);
@ -799,13 +800,13 @@ int32_t sdbUpdateRow(SSWriteMsg *pWrite) {
void *pRow = sdbGetRowMetaFromObj(pTable, pWrite->pRow); void *pRow = sdbGetRowMetaFromObj(pTable, pWrite->pRow);
if (pRow == NULL) { if (pRow == NULL) {
sdbDebug("vgId:1, sdb:%s, record is not there, update failed", pTable->tableName); sdbDebug("vgId:1, sdb:%s, record is not there, update failed", pTable->name);
return TSDB_CODE_MND_SDB_OBJ_NOT_THERE; return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
} }
int32_t code = sdbUpdateHash(pTable, pWrite); int32_t code = sdbUpdateHash(pTable, pWrite);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
sdbError("vgId:1, sdb:%s, failed to update hash", pTable->tableName); sdbError("vgId:1, sdb:%s, failed to update hash", pTable->name);
return code; return code;
} }
@ -830,7 +831,7 @@ int32_t sdbUpdateRowImp(SSWriteMsg *pWrite) {
SWalHead *pHead = (SWalHead *)((void *)pNewWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK); SWalHead *pHead = (SWalHead *)((void *)pNewWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK);
pHead->version = 0; pHead->version = 0;
pHead->msgType = pTable->tableId * 10 + SDB_ACTION_UPDATE; pHead->msgType = pTable->id * 10 + SDB_ACTION_UPDATE;
pWrite->rowData = pHead->cont; pWrite->rowData = pHead->cont;
(*pTable->fpEncode)(pWrite); (*pTable->fpEncode)(pWrite);
@ -840,7 +841,7 @@ int32_t sdbUpdateRowImp(SSWriteMsg *pWrite) {
if (pNewWrite->pMsg != NULL) { if (pNewWrite->pMsg != NULL) {
sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s, update action is add to sdb queue", pNewWrite->pMsg->rpcMsg.ahandle, sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s, update action is add to sdb queue", pNewWrite->pMsg->rpcMsg.ahandle,
pNewWrite->pMsg, pTable->tableName, pWrite->pRow, sdbGetRowStr(pTable, pWrite->pRow)); pNewWrite->pMsg, pTable->name, pWrite->pRow, sdbGetRowStr(pTable, pWrite->pRow));
} }
sdbIncRef(pNewWrite->pTable, pNewWrite->pRow); sdbIncRef(pNewWrite->pTable, pNewWrite->pRow);
@ -888,9 +889,9 @@ void *sdbOpenTable(SSdbTableDesc *pDesc) {
if (pTable == NULL) return NULL; if (pTable == NULL) return NULL;
pthread_mutex_init(&pTable->mutex, NULL); pthread_mutex_init(&pTable->mutex, NULL);
tstrncpy(pTable->tableName, pDesc->tableName, SDB_TABLE_LEN); tstrncpy(pTable->name, pDesc->name, SDB_TABLE_LEN);
pTable->keyType = pDesc->keyType; pTable->keyType = pDesc->keyType;
pTable->tableId = pDesc->tableId; pTable->id = pDesc->id;
pTable->hashSessions = pDesc->hashSessions; pTable->hashSessions = pDesc->hashSessions;
pTable->maxRowSize = pDesc->maxRowSize; pTable->maxRowSize = pDesc->maxRowSize;
pTable->refCountPos = pDesc->refCountPos; pTable->refCountPos = pDesc->refCountPos;
@ -909,7 +910,7 @@ void *sdbOpenTable(SSdbTableDesc *pDesc) {
pTable->iHandle = taosHashInit(pTable->hashSessions, hashFp, true, true); pTable->iHandle = taosHashInit(pTable->hashSessions, hashFp, true, true);
tsSdbMgmt.numOfTables++; tsSdbMgmt.numOfTables++;
tsSdbMgmt.tableList[pTable->tableId] = pTable; tsSdbMgmt.tableList[pTable->id] = pTable;
return pTable; return pTable;
} }
@ -918,7 +919,7 @@ void sdbCloseTable(void *handle) {
if (pTable == NULL) return; if (pTable == NULL) return;
tsSdbMgmt.numOfTables--; tsSdbMgmt.numOfTables--;
tsSdbMgmt.tableList[pTable->tableId] = NULL; tsSdbMgmt.tableList[pTable->id] = NULL;
SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle); SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle);
while (taosHashIterNext(pIter)) { while (taosHashIterNext(pIter)) {
@ -937,7 +938,7 @@ void sdbCloseTable(void *handle) {
taosHashCleanup(pTable->iHandle); taosHashCleanup(pTable->iHandle);
pthread_mutex_destroy(&pTable->mutex); pthread_mutex_destroy(&pTable->mutex);
sdbDebug("vgId:1, sdb:%s, is closed, numOfTables:%d", pTable->tableName, tsSdbMgmt.numOfTables); sdbDebug("vgId:1, sdb:%s, is closed, numOfTables:%d", pTable->name, tsSdbMgmt.numOfTables);
free(pTable); free(pTable);
} }
@ -1062,7 +1063,7 @@ static void *sdbWorkerFp(void *pWorker) {
pHead = (void *)pWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK; pHead = (void *)pWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK;
if (pWrite->pMsg != NULL) { if (pWrite->pMsg != NULL) {
sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s hver:%" PRIu64 ", will be processed in sdb queue", sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s row:%p:%s hver:%" PRIu64 ", will be processed in sdb queue",
pWrite->pMsg->rpcMsg.ahandle, pWrite->pMsg, pWrite->pTable->tableName, pWrite->pRow, pWrite->pMsg->rpcMsg.ahandle, pWrite->pMsg, pWrite->pTable->name, pWrite->pRow,
sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version); sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version);
} }
} else { } else {

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

@ -352,8 +352,8 @@ static int32_t mnodeInitChildTables() {
tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type; tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type;
SSdbTableDesc tableDesc = { SSdbTableDesc tableDesc = {
.tableId = SDB_TABLE_CTABLE, .id = SDB_TABLE_CTABLE,
.tableName = "ctables", .name = "ctables",
.hashSessions = TSDB_DEFAULT_CTABLES_HASH_SIZE, .hashSessions = TSDB_DEFAULT_CTABLES_HASH_SIZE,
.maxRowSize = sizeof(SCTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16) + TSDB_TABLE_FNAME_LEN + TSDB_CQ_SQL_SIZE, .maxRowSize = sizeof(SCTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16) + TSDB_TABLE_FNAME_LEN + TSDB_CQ_SQL_SIZE,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -545,8 +545,8 @@ static int32_t mnodeInitSuperTables() {
tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type; tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type;
SSdbTableDesc tableDesc = { SSdbTableDesc tableDesc = {
.tableId = SDB_TABLE_STABLE, .id = SDB_TABLE_STABLE,
.tableName = "stables", .name = "stables",
.hashSessions = TSDB_DEFAULT_STABLES_HASH_SIZE, .hashSessions = TSDB_DEFAULT_STABLES_HASH_SIZE,
.maxRowSize = sizeof(SSTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16) + TSDB_TABLE_FNAME_LEN, .maxRowSize = sizeof(SSTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16) + TSDB_TABLE_FNAME_LEN,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,

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

@ -150,9 +150,9 @@ int32_t mnodeInitUsers() {
SUserObj tObj; SUserObj tObj;
tsUserUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsUserUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = { SSdbTableDesc desc = {
.tableId = SDB_TABLE_USER, .id = SDB_TABLE_USER,
.tableName = "users", .name = "users",
.hashSessions = TSDB_DEFAULT_USERS_HASH_SIZE, .hashSessions = TSDB_DEFAULT_USERS_HASH_SIZE,
.maxRowSize = tsUserUpdateSize, .maxRowSize = tsUserUpdateSize,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -166,9 +166,9 @@ int32_t mnodeInitUsers() {
.fpRestored = mnodeUserActionRestored .fpRestored = mnodeUserActionRestored
}; };
tsUserSdb = sdbOpenTable(&tableDesc); tsUserSdb = sdbOpenTable(&desc);
if (tsUserSdb == NULL) { if (tsUserSdb == NULL) {
mError("table:%s, failed to create hash", tableDesc.tableName); mError("table:%s, failed to create hash", desc.name);
return -1; return -1;
} }
@ -179,7 +179,7 @@ int32_t mnodeInitUsers() {
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_USER, mnodeRetrieveUsers); mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_USER, mnodeRetrieveUsers);
mnodeAddPeerMsgHandle(TSDB_MSG_TYPE_DM_AUTH, mnodeProcessAuthMsg); mnodeAddPeerMsgHandle(TSDB_MSG_TYPE_DM_AUTH, mnodeProcessAuthMsg);
mDebug("table:%s, hash is created", tableDesc.tableName); mDebug("table:%s, hash is created", desc.name);
return 0; return 0;
} }

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

@ -206,9 +206,9 @@ int32_t mnodeInitVgroups() {
SVgObj tObj; SVgObj tObj;
tsVgUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsVgUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
SSdbTableDesc tableDesc = { SSdbTableDesc desc = {
.tableId = SDB_TABLE_VGROUP, .id = SDB_TABLE_VGROUP,
.tableName = "vgroups", .name = "vgroups",
.hashSessions = TSDB_DEFAULT_VGROUPS_HASH_SIZE, .hashSessions = TSDB_DEFAULT_VGROUPS_HASH_SIZE,
.maxRowSize = tsVgUpdateSize, .maxRowSize = tsVgUpdateSize,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
@ -222,7 +222,7 @@ int32_t mnodeInitVgroups() {
.fpRestored = mnodeVgroupActionRestored, .fpRestored = mnodeVgroupActionRestored,
}; };
tsVgroupSdb = sdbOpenTable(&tableDesc); tsVgroupSdb = sdbOpenTable(&desc);
if (tsVgroupSdb == NULL) { if (tsVgroupSdb == NULL) {
mError("failed to init vgroups data"); mError("failed to init vgroups data");
return -1; return -1;