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 16:06:06 +08:00
parent 97e41cdf8d
commit 98ec34b40d
10 changed files with 234 additions and 238 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -57,7 +57,7 @@ typedef struct SSWriteMsg {
void * rowData; void * rowData;
int32_t (*fpReq)(SMnodeMsg *pMsg); int32_t (*fpReq)(SMnodeMsg *pMsg);
int32_t (*fpWrite)(SMnodeMsg *pMsg, int32_t code); int32_t (*fpWrite)(SMnodeMsg *pMsg, int32_t code);
void * pObj; void * pRow;
SMnodeMsg *pMsg; SMnodeMsg *pMsg;
struct SSdbTable *pTable; struct SSdbTable *pTable;
} SSWriteMsg; } SSWriteMsg;
@ -75,7 +75,7 @@ typedef struct {
int32_t (*fpEncode)(SSWriteMsg *pWrite); int32_t (*fpEncode)(SSWriteMsg *pWrite);
int32_t (*fpDecode)(SSWriteMsg *pWrite); int32_t (*fpDecode)(SSWriteMsg *pWrite);
int32_t (*fpDestroy)(SSWriteMsg *pWrite); int32_t (*fpDestroy)(SSWriteMsg *pWrite);
int32_t (*fpDestored)(); int32_t (*fpRestored)();
} SSdbTableDesc; } SSdbTableDesc;
int32_t sdbInit(); int32_t sdbInit();

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

@ -32,14 +32,14 @@ static int32_t tsAcctUpdateSize;
static int32_t mnodeCreateRootAcct(); static int32_t mnodeCreateRootAcct();
static int32_t mnodeAcctActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeAcctActionDestroy(SSWriteMsg *pWMsg) {
SAcctObj *pAcct = pWMsg->pObj; SAcctObj *pAcct = pWMsg->pRow;
pthread_mutex_destroy(&pAcct->mutex); pthread_mutex_destroy(&pAcct->mutex);
tfree(pWMsg->pObj); tfree(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeAcctActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeAcctActionInsert(SSWriteMsg *pWMsg) {
SAcctObj *pAcct = pWMsg->pObj; SAcctObj *pAcct = pWMsg->pRow;
memset(&pAcct->acctInfo, 0, sizeof(SAcctInfo)); memset(&pAcct->acctInfo, 0, sizeof(SAcctInfo));
pAcct->acctInfo.accessState = TSDB_VN_ALL_ACCCESS; pAcct->acctInfo.accessState = TSDB_VN_ALL_ACCCESS;
pthread_mutex_init(&pAcct->mutex, NULL); pthread_mutex_init(&pAcct->mutex, NULL);
@ -47,14 +47,14 @@ static int32_t mnodeAcctActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeAcctActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeAcctActionDelete(SSWriteMsg *pWMsg) {
SAcctObj *pAcct = pWMsg->pObj; SAcctObj *pAcct = pWMsg->pRow;
mnodeDropAllUsers(pAcct); mnodeDropAllUsers(pAcct);
mnodeDropAllDbs(pAcct); mnodeDropAllDbs(pAcct);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeAcctActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeAcctActionUpdate(SSWriteMsg *pWMsg) {
SAcctObj *pAcct = pWMsg->pObj; SAcctObj *pAcct = pWMsg->pRow;
SAcctObj *pSaved = mnodeGetAcct(pAcct->user); SAcctObj *pSaved = mnodeGetAcct(pAcct->user);
if (pAcct != pSaved) { if (pAcct != pSaved) {
memcpy(pSaved, pAcct, tsAcctUpdateSize); memcpy(pSaved, pAcct, tsAcctUpdateSize);
@ -65,7 +65,7 @@ static int32_t mnodeAcctActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeAcctActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeAcctActionEncode(SSWriteMsg *pWMsg) {
SAcctObj *pAcct = pWMsg->pObj; SAcctObj *pAcct = pWMsg->pRow;
memcpy(pWMsg->rowData, pAcct, tsAcctUpdateSize); memcpy(pWMsg->rowData, pAcct, tsAcctUpdateSize);
pWMsg->rowSize = tsAcctUpdateSize; pWMsg->rowSize = tsAcctUpdateSize;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -76,7 +76,7 @@ static int32_t mnodeAcctActionDecode(SSWriteMsg *pWMsg) {
if (pAcct == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY; if (pAcct == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
memcpy(pAcct, pWMsg->rowData, tsAcctUpdateSize); memcpy(pAcct, pWMsg->rowData, tsAcctUpdateSize);
pWMsg->pObj = pAcct; pWMsg->pRow = pAcct;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -112,7 +112,7 @@ int32_t mnodeInitAccts() {
.fpEncode = mnodeAcctActionEncode, .fpEncode = mnodeAcctActionEncode,
.fpDecode = mnodeAcctActionDecode, .fpDecode = mnodeAcctActionDecode,
.fpDestroy = mnodeAcctActionDestroy, .fpDestroy = mnodeAcctActionDestroy,
.fpDestored = mnodeAcctActionRestored .fpRestored = mnodeAcctActionRestored
}; };
tsAcctSdb = sdbOpenTable(&tableDesc); tsAcctSdb = sdbOpenTable(&tableDesc);
@ -229,7 +229,7 @@ static int32_t mnodeCreateRootAcct() {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsAcctSdb, .pTable = tsAcctSdb,
.pObj = pAcct, .pRow = pAcct,
}; };
return sdbInsertRow(&wmsg); return sdbInsertRow(&wmsg);

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

@ -33,7 +33,7 @@ static int32_t mnodeGetClusterMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *
static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static int32_t mnodeClusterActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeClusterActionDestroy(SSWriteMsg *pWMsg) {
tfree(pWMsg->pObj); tfree(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -50,7 +50,7 @@ static int32_t mnodeClusterActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeClusterActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeClusterActionEncode(SSWriteMsg *pWMsg) {
SClusterObj *pCluster = pWMsg->pObj; SClusterObj *pCluster = pWMsg->pRow;
memcpy(pWMsg->rowData, pCluster, tsClusterUpdateSize); memcpy(pWMsg->rowData, pCluster, tsClusterUpdateSize);
pWMsg->rowSize = tsClusterUpdateSize; pWMsg->rowSize = tsClusterUpdateSize;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -61,7 +61,7 @@ static int32_t mnodeClusterActionDecode(SSWriteMsg *pWMsg) {
if (pCluster == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY; if (pCluster == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
memcpy(pCluster, pWMsg->rowData, tsClusterUpdateSize); memcpy(pCluster, pWMsg->rowData, tsClusterUpdateSize);
pWMsg->pObj = pCluster; pWMsg->pRow = pCluster;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -97,7 +97,7 @@ int32_t mnodeInitCluster() {
.fpEncode = mnodeClusterActionEncode, .fpEncode = mnodeClusterActionEncode,
.fpDecode = mnodeClusterActionDecode, .fpDecode = mnodeClusterActionDecode,
.fpDestroy = mnodeClusterActionDestroy, .fpDestroy = mnodeClusterActionDestroy,
.fpDestored = mnodeClusterActionRestored .fpRestored = mnodeClusterActionRestored
}; };
tsClusterSdb = sdbOpenTable(&tableDesc); tsClusterSdb = sdbOpenTable(&tableDesc);
@ -148,7 +148,7 @@ static int32_t mnodeCreateCluster() {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsClusterSdb, .pTable = tsClusterSdb,
.pObj = pCluster, .pRow = pCluster,
}; };
return sdbInsertRow(&wmsg); return sdbInsertRow(&wmsg);

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

@ -57,7 +57,7 @@ static void mnodeDestroyDb(SDbObj *pDb) {
} }
static int32_t mnodeDbActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeDbActionDestroy(SSWriteMsg *pWMsg) {
mnodeDestroyDb(pWMsg->pObj); mnodeDestroyDb(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -66,7 +66,7 @@ int64_t mnodeGetDbNum() {
} }
static int32_t mnodeDbActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeDbActionInsert(SSWriteMsg *pWMsg) {
SDbObj *pDb = pWMsg->pObj; SDbObj *pDb = pWMsg->pRow;
SAcctObj *pAcct = mnodeGetAcct(pDb->acct); SAcctObj *pAcct = mnodeGetAcct(pDb->acct);
pthread_mutex_init(&pDb->mutex, NULL); pthread_mutex_init(&pDb->mutex, NULL);
@ -92,7 +92,7 @@ static int32_t mnodeDbActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeDbActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeDbActionDelete(SSWriteMsg *pWMsg) {
SDbObj *pDb = pWMsg->pObj; SDbObj *pDb = pWMsg->pRow;
SAcctObj *pAcct = mnodeGetAcct(pDb->acct); SAcctObj *pAcct = mnodeGetAcct(pDb->acct);
mnodeDropAllChildTables(pDb); mnodeDropAllChildTables(pDb);
@ -108,7 +108,7 @@ static int32_t mnodeDbActionDelete(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeDbActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeDbActionUpdate(SSWriteMsg *pWMsg) {
SDbObj *pNew = pWMsg->pObj; SDbObj *pNew = pWMsg->pRow;
SDbObj *pDb = mnodeGetDb(pNew->name); SDbObj *pDb = mnodeGetDb(pNew->name);
if (pDb != NULL && pNew != pDb) { if (pDb != NULL && pNew != pDb) {
memcpy(pDb, pNew, pWMsg->rowSize); memcpy(pDb, pNew, pWMsg->rowSize);
@ -121,7 +121,7 @@ static int32_t mnodeDbActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeDbActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeDbActionEncode(SSWriteMsg *pWMsg) {
SDbObj *pDb = pWMsg->pObj; SDbObj *pDb = pWMsg->pRow;
memcpy(pWMsg->rowData, pDb, tsDbUpdateSize); memcpy(pWMsg->rowData, pDb, tsDbUpdateSize);
pWMsg->rowSize = tsDbUpdateSize; pWMsg->rowSize = tsDbUpdateSize;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -132,7 +132,7 @@ static int32_t mnodeDbActionDecode(SSWriteMsg *pWMsg) {
if (pDb == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY; if (pDb == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
memcpy(pDb, pWMsg->rowData, tsDbUpdateSize); memcpy(pDb, pWMsg->rowData, tsDbUpdateSize);
pWMsg->pObj = pDb; pWMsg->pRow = pDb;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -157,7 +157,7 @@ int32_t mnodeInitDbs() {
.fpEncode = mnodeDbActionEncode, .fpEncode = mnodeDbActionEncode,
.fpDecode = mnodeDbActionDecode, .fpDecode = mnodeDbActionDecode,
.fpDestroy = mnodeDbActionDestroy, .fpDestroy = mnodeDbActionDestroy,
.fpDestored = mnodeDbActionRestored .fpRestored = mnodeDbActionRestored
}; };
tsDbSdb = sdbOpenTable(&tableDesc); tsDbSdb = sdbOpenTable(&tableDesc);
@ -415,7 +415,7 @@ static int32_t mnodeCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate, SMnodeMsg *
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsDbSdb, .pTable = tsDbSdb,
.pObj = pDb, .pRow = pDb,
.rowSize = sizeof(SDbObj), .rowSize = sizeof(SDbObj),
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeCreateDbCb .fpWrite = mnodeCreateDbCb
@ -810,7 +810,7 @@ static int32_t mnodeSetDbDropping(SDbObj *pDb) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsDbSdb, .pTable = tsDbSdb,
.pObj = pDb .pRow = pDb
}; };
int32_t code = sdbUpdateRow(&wmsg); int32_t code = sdbUpdateRow(&wmsg);
@ -1022,7 +1022,7 @@ static int32_t mnodeAlterDb(SDbObj *pDb, SAlterDbMsg *pAlter, void *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsDbSdb, .pTable = tsDbSdb,
.pObj = pDb, .pRow = pDb,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeAlterDbCb .fpWrite = mnodeAlterDbCb
}; };
@ -1074,7 +1074,7 @@ static int32_t mnodeDropDb(SMnodeMsg *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsDbSdb, .pTable = tsDbSdb,
.pObj = pDb, .pRow = pDb,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeDropDbCb .fpWrite = mnodeDropDbCb
}; };
@ -1137,7 +1137,7 @@ void mnodeDropAllDbs(SAcctObj *pAcct) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsDbSdb, .pTable = tsDbSdb,
.pObj = pDb .pRow = pDb
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);

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

@ -88,12 +88,12 @@ static char* offlineReason[] = {
}; };
static int32_t mnodeDnodeActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeDnodeActionDestroy(SSWriteMsg *pWMsg) {
tfree(pWMsg->pObj); tfree(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeDnodeActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeDnodeActionInsert(SSWriteMsg *pWMsg) {
SDnodeObj *pDnode = pWMsg->pObj; SDnodeObj *pDnode = pWMsg->pRow;
if (pDnode->status != TAOS_DN_STATUS_DROPPING) { if (pDnode->status != TAOS_DN_STATUS_DROPPING) {
pDnode->status = TAOS_DN_STATUS_OFFLINE; pDnode->status = TAOS_DN_STATUS_OFFLINE;
pDnode->lastAccess = tsAccessSquence; pDnode->lastAccess = tsAccessSquence;
@ -108,7 +108,7 @@ static int32_t mnodeDnodeActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeDnodeActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeDnodeActionDelete(SSWriteMsg *pWMsg) {
SDnodeObj *pDnode = pWMsg->pObj; SDnodeObj *pDnode = pWMsg->pRow;
#ifndef _SYNC #ifndef _SYNC
mnodeDropAllDnodeVgroups(pDnode); mnodeDropAllDnodeVgroups(pDnode);
@ -122,7 +122,7 @@ static int32_t mnodeDnodeActionDelete(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeDnodeActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeDnodeActionUpdate(SSWriteMsg *pWMsg) {
SDnodeObj *pNew = pWMsg->pObj; SDnodeObj *pNew = pWMsg->pRow;
SDnodeObj *pDnode = mnodeGetDnode(pNew->dnodeId); SDnodeObj *pDnode = mnodeGetDnode(pNew->dnodeId);
if (pDnode != NULL && pNew != pDnode) { if (pDnode != NULL && pNew != pDnode) {
memcpy(pDnode, pNew, pWMsg->rowSize); memcpy(pDnode, pNew, pWMsg->rowSize);
@ -135,7 +135,7 @@ static int32_t mnodeDnodeActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeDnodeActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeDnodeActionEncode(SSWriteMsg *pWMsg) {
SDnodeObj *pDnode = pWMsg->pObj; SDnodeObj *pDnode = pWMsg->pRow;
memcpy(pWMsg->rowData, pDnode, tsDnodeUpdateSize); memcpy(pWMsg->rowData, pDnode, tsDnodeUpdateSize);
pWMsg->rowSize = tsDnodeUpdateSize; pWMsg->rowSize = tsDnodeUpdateSize;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -146,7 +146,7 @@ static int32_t mnodeDnodeActionDecode(SSWriteMsg *pWMsg) {
if (pDnode == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY; if (pDnode == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
memcpy(pDnode, pWMsg->rowData, tsDnodeUpdateSize); memcpy(pDnode, pWMsg->rowData, tsDnodeUpdateSize);
pWMsg->pObj = pDnode; pWMsg->pRow = pDnode;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -184,7 +184,7 @@ int32_t mnodeInitDnodes() {
.fpEncode = mnodeDnodeActionEncode, .fpEncode = mnodeDnodeActionEncode,
.fpDecode = mnodeDnodeActionDecode, .fpDecode = mnodeDnodeActionDecode,
.fpDestroy = mnodeDnodeActionDestroy, .fpDestroy = mnodeDnodeActionDestroy,
.fpDestored = mnodeDnodeActionRestored .fpRestored = mnodeDnodeActionRestored
}; };
tsDnodeSdb = sdbOpenTable(&tableDesc); tsDnodeSdb = sdbOpenTable(&tableDesc);
@ -299,7 +299,7 @@ void mnodeUpdateDnode(SDnodeObj *pDnode) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsDnodeSdb, .pTable = tsDnodeSdb,
.pObj = pDnode .pRow = pDnode
}; };
int32_t code = sdbUpdateRow(&wmsg); int32_t code = sdbUpdateRow(&wmsg);
@ -647,7 +647,7 @@ static int32_t mnodeCreateDnode(char *ep, SMnodeMsg *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsDnodeSdb, .pTable = tsDnodeSdb,
.pObj = pDnode, .pRow = pDnode,
.rowSize = sizeof(SDnodeObj), .rowSize = sizeof(SDnodeObj),
.pMsg = pMsg .pMsg = pMsg
}; };
@ -668,7 +668,7 @@ int32_t mnodeDropDnode(SDnodeObj *pDnode, void *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsDnodeSdb, .pTable = tsDnodeSdb,
.pObj = pDnode, .pRow = pDnode,
.pMsg = pMsg .pMsg = pMsg
}; };
@ -1141,7 +1141,7 @@ static int32_t mnodeRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, vo
cols++; cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, mnodeGetMnodeRoleStr(pVgid->role)); strcpy(pWrite, syncRole[pVgid->role]);
cols++; cols++;
} }
} }

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

@ -59,12 +59,12 @@ static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, vo
#endif #endif
static int32_t mnodeMnodeActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeMnodeActionDestroy(SSWriteMsg *pWMsg) {
tfree(pWMsg->pObj); tfree(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeMnodeActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeMnodeActionInsert(SSWriteMsg *pWMsg) {
SMnodeObj *pMnode = pWMsg->pObj; SMnodeObj *pMnode = pWMsg->pRow;
SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId); SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId);
if (pDnode == NULL) return TSDB_CODE_MND_DNODE_NOT_EXIST; if (pDnode == NULL) return TSDB_CODE_MND_DNODE_NOT_EXIST;
@ -77,7 +77,7 @@ static int32_t mnodeMnodeActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeMnodeActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeMnodeActionDelete(SSWriteMsg *pWMsg) {
SMnodeObj *pMnode = pWMsg->pObj; SMnodeObj *pMnode = pWMsg->pRow;
SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId); SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId);
if (pDnode == NULL) return TSDB_CODE_MND_DNODE_NOT_EXIST; if (pDnode == NULL) return TSDB_CODE_MND_DNODE_NOT_EXIST;
@ -89,7 +89,7 @@ static int32_t mnodeMnodeActionDelete(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeMnodeActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeMnodeActionUpdate(SSWriteMsg *pWMsg) {
SMnodeObj *pMnode = pWMsg->pObj; SMnodeObj *pMnode = pWMsg->pRow;
SMnodeObj *pSaved = mnodeGetMnode(pMnode->mnodeId); SMnodeObj *pSaved = mnodeGetMnode(pMnode->mnodeId);
if (pMnode != pSaved) { if (pMnode != pSaved) {
memcpy(pSaved, pMnode, pWMsg->rowSize); memcpy(pSaved, pMnode, pWMsg->rowSize);
@ -100,7 +100,7 @@ static int32_t mnodeMnodeActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeMnodeActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeMnodeActionEncode(SSWriteMsg *pWMsg) {
SMnodeObj *pMnode = pWMsg->pObj; SMnodeObj *pMnode = pWMsg->pRow;
memcpy(pWMsg->rowData, pMnode, tsMnodeUpdateSize); memcpy(pWMsg->rowData, pMnode, tsMnodeUpdateSize);
pWMsg->rowSize = tsMnodeUpdateSize; pWMsg->rowSize = tsMnodeUpdateSize;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -111,7 +111,7 @@ static int32_t mnodeMnodeActionDecode(SSWriteMsg *pWMsg) {
if (pMnode == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY; if (pMnode == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
memcpy(pMnode, pWMsg->rowData, tsMnodeUpdateSize); memcpy(pMnode, pWMsg->rowData, tsMnodeUpdateSize);
pWMsg->pObj = pMnode; pWMsg->pRow = pMnode;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -150,7 +150,7 @@ int32_t mnodeInitMnodes() {
.fpEncode = mnodeMnodeActionEncode, .fpEncode = mnodeMnodeActionEncode,
.fpDecode = mnodeMnodeActionDecode, .fpDecode = mnodeMnodeActionDecode,
.fpDestroy = mnodeMnodeActionDestroy, .fpDestroy = mnodeMnodeActionDestroy,
.fpDestored = mnodeMnodeActionRestored .fpRestored = mnodeMnodeActionRestored
}; };
tsMnodeSdb = sdbOpenTable(&tableDesc); tsMnodeSdb = sdbOpenTable(&tableDesc);
@ -192,10 +192,6 @@ void *mnodeGetNextMnode(void *pIter, SMnodeObj **pMnode) {
return sdbFetchRow(tsMnodeSdb, pIter, (void **)pMnode); return sdbFetchRow(tsMnodeSdb, pIter, (void **)pMnode);
} }
char *mnodeGetMnodeRoleStr(int32_t role) {
return syncRole[role];
}
void mnodeUpdateMnodeEpSet() { void mnodeUpdateMnodeEpSet() {
mInfo("update mnodes epSet, numOfEps:%d ", mnodeGetMnodesNum()); mInfo("update mnodes epSet, numOfEps:%d ", mnodeGetMnodesNum());
@ -332,7 +328,7 @@ void mnodeCreateMnode(int32_t dnodeId, char *dnodeEp, bool needConfirm) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsMnodeSdb, .pTable = tsMnodeSdb,
.pObj = pMnode, .pRow = pMnode,
.fpWrite = mnodeCreateMnodeCb .fpWrite = mnodeCreateMnodeCb
}; };
@ -356,7 +352,7 @@ void mnodeCreateMnode(int32_t dnodeId, char *dnodeEp, bool needConfirm) {
void mnodeDropMnodeLocal(int32_t dnodeId) { void mnodeDropMnodeLocal(int32_t dnodeId) {
SMnodeObj *pMnode = mnodeGetMnode(dnodeId); SMnodeObj *pMnode = mnodeGetMnode(dnodeId);
if (pMnode != NULL) { if (pMnode != NULL) {
SSWriteMsg wmsg = {.type = SDB_OPER_LOCAL, .pTable = tsMnodeSdb, .pObj = pMnode}; SSWriteMsg wmsg = {.type = SDB_OPER_LOCAL, .pTable = tsMnodeSdb, .pRow = pMnode};
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
mnodeDecMnodeRef(pMnode); mnodeDecMnodeRef(pMnode);
} }
@ -374,7 +370,7 @@ int32_t mnodeDropMnode(int32_t dnodeId) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsMnodeSdb, .pTable = tsMnodeSdb,
.pObj = pMnode .pRow = pMnode
}; };
int32_t code = sdbDeleteRow(&wmsg); int32_t code = sdbDeleteRow(&wmsg);
@ -469,7 +465,7 @@ static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, vo
cols++; cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
char* roles = mnodeGetMnodeRoleStr(pMnode->role); char* roles = syncRole[pMnode->role];
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, roles, pShow->bytes[cols]); STR_WITH_MAXSIZE_TO_VARSTR(pWrite, roles, pShow->bytes[cols]);
cols++; cols++;

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

@ -37,15 +37,15 @@
#define SDB_SYNC_HACK 16 #define SDB_SYNC_HACK 16
typedef enum { typedef enum {
SDB_ACTION_INSERT, SDB_ACTION_INSERT = 0,
SDB_ACTION_DELETE, SDB_ACTION_DELETE = 1,
SDB_ACTION_UPDATE SDB_ACTION_UPDATE = 2
} ESdbAction; } ESdbAction;
typedef enum { typedef enum {
SDB_STATUS_OFFLINE, SDB_STATUS_OFFLINE = 0,
SDB_STATUS_SERVING, SDB_STATUS_SERVING = 1,
SDB_STATUS_CLOSING SDB_STATUS_CLOSING = 2
} ESdbStatus; } ESdbStatus;
typedef struct SSdbTable { typedef struct SSdbTable {
@ -64,7 +64,7 @@ typedef struct SSdbTable {
int32_t (*fpDecode)(SSWriteMsg *pWrite); int32_t (*fpDecode)(SSWriteMsg *pWrite);
int32_t (*fpEncode)(SSWriteMsg *pWrite); int32_t (*fpEncode)(SSWriteMsg *pWrite);
int32_t (*fpDestroy)(SSWriteMsg *pWrite); int32_t (*fpDestroy)(SSWriteMsg *pWrite);
int32_t (*fpDestored)(); int32_t (*fpRestored)();
pthread_mutex_t mutex; pthread_mutex_t mutex;
} SSdbTable; } SSdbTable;
@ -78,7 +78,7 @@ typedef struct {
int32_t numOfTables; int32_t numOfTables;
SSdbTable *tableList[SDB_TABLE_MAX]; SSdbTable *tableList[SDB_TABLE_MAX];
pthread_mutex_t mutex; pthread_mutex_t mutex;
} SSdbObject; } SSdbMgmt;
typedef struct { typedef struct {
pthread_t thread; pthread_t thread;
@ -92,7 +92,7 @@ typedef struct {
extern void * tsMnodeTmr; extern void * tsMnodeTmr;
static void * tsSdbTmr; static void * tsSdbTmr;
static SSdbObject tsSdbObj = {0}; static SSdbMgmt tsSdbMgmt = {0};
static taos_qset tsSdbWQset; static taos_qset tsSdbWQset;
static taos_qall tsSdbWQall; static taos_qall tsSdbWQall;
static taos_queue tsSdbWQueue; static taos_queue tsSdbWQueue;
@ -121,15 +121,15 @@ int64_t sdbGetNumOfRows(void *pTable) {
} }
uint64_t sdbGetVersion() { uint64_t sdbGetVersion() {
return tsSdbObj.version; return tsSdbMgmt.version;
} }
bool sdbIsMaster() { bool sdbIsMaster() {
return tsSdbObj.role == TAOS_SYNC_ROLE_MASTER; return tsSdbMgmt.role == TAOS_SYNC_ROLE_MASTER;
} }
bool sdbIsServing() { bool sdbIsServing() {
return tsSdbObj.status == SDB_STATUS_SERVING; return tsSdbMgmt.status == SDB_STATUS_SERVING;
} }
static void *sdbGetObjKey(SSdbTable *pTable, void *key) { static void *sdbGetObjKey(SSdbTable *pTable, void *key) {
@ -172,21 +172,21 @@ static char *sdbGetObjStr(SSdbTable *pTable, void *key) {
} }
static void *sdbGetTableFromId(int32_t tableId) { static void *sdbGetTableFromId(int32_t tableId) {
return tsSdbObj.tableList[tableId]; return tsSdbMgmt.tableList[tableId];
} }
static int32_t sdbInitWal() { static int32_t sdbInitWal() {
SWalCfg walCfg = {.vgId = 1, .walLevel = TAOS_WAL_FSYNC, .keep = TAOS_WAL_KEEP, .fsyncPeriod = 0}; SWalCfg walCfg = {.vgId = 1, .walLevel = TAOS_WAL_FSYNC, .keep = TAOS_WAL_KEEP, .fsyncPeriod = 0};
char temp[TSDB_FILENAME_LEN] = {0}; char temp[TSDB_FILENAME_LEN] = {0};
sprintf(temp, "%s/wal", tsMnodeDir); sprintf(temp, "%s/wal", tsMnodeDir);
tsSdbObj.wal = walOpen(temp, &walCfg); tsSdbMgmt.wal = walOpen(temp, &walCfg);
if (tsSdbObj.wal == NULL) { if (tsSdbMgmt.wal == NULL) {
sdbError("vgId:1, failed to open wal in %s", tsMnodeDir); sdbError("vgId:1, failed to open wal in %s", tsMnodeDir);
return -1; return -1;
} }
sdbInfo("vgId:1, open wal for restore"); sdbInfo("vgId:1, open wal for restore");
int code = walRestore(tsSdbObj.wal, NULL, sdbWrite); int code = walRestore(tsSdbMgmt.wal, NULL, sdbWrite);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
sdbError("vgId:1, failed to open wal for restore since %s", tstrerror(code)); sdbError("vgId:1, failed to open wal for restore since %s", tstrerror(code));
return -1; return -1;
@ -200,8 +200,8 @@ static void sdbRestoreTables() {
for (int32_t tableId = 0; tableId < SDB_TABLE_MAX; ++tableId) { for (int32_t tableId = 0; tableId < SDB_TABLE_MAX; ++tableId) {
SSdbTable *pTable = sdbGetTableFromId(tableId); SSdbTable *pTable = sdbGetTableFromId(tableId);
if (pTable == NULL) continue; if (pTable == NULL) continue;
if (pTable->fpDestored) { if (pTable->fpRestored) {
(*pTable->fpDestored)(); (*pTable->fpRestored)();
} }
totalRows += pTable->numOfRows; totalRows += pTable->numOfRows;
@ -209,22 +209,22 @@ static void sdbRestoreTables() {
sdbDebug("vgId:1, sdb:%s is restored, rows:%" PRId64, pTable->tableName, pTable->numOfRows); sdbDebug("vgId:1, sdb:%s is restored, rows:%" PRId64, pTable->tableName, pTable->numOfRows);
} }
sdbInfo("vgId:1, sdb is restored, mver:%" PRIu64 " rows:%d tables:%d", tsSdbObj.version, totalRows, numOfTables); sdbInfo("vgId:1, sdb is restored, mver:%" PRIu64 " rows:%d tables:%d", tsSdbMgmt.version, totalRows, numOfTables);
} }
void sdbUpdateMnodeRoles() { void sdbUpdateMnodeRoles() {
if (tsSdbObj.sync <= 0) return; if (tsSdbMgmt.sync <= 0) return;
SNodesRole roles = {0}; SNodesRole roles = {0};
syncGetNodesRole(tsSdbObj.sync, &roles); syncGetNodesRole(tsSdbMgmt.sync, &roles);
sdbInfo("vgId:1, update mnodes roles, replica:%d", tsSdbObj.cfg.replica); sdbInfo("vgId:1, update mnodes role, replica:%d", tsSdbMgmt.cfg.replica);
for (int32_t i = 0; i < tsSdbObj.cfg.replica; ++i) { for (int32_t i = 0; i < tsSdbMgmt.cfg.replica; ++i) {
SMnodeObj *pMnode = mnodeGetMnode(roles.nodeId[i]); SMnodeObj *pMnode = mnodeGetMnode(roles.nodeId[i]);
if (pMnode != NULL) { if (pMnode != NULL) {
pMnode->role = roles.role[i]; pMnode->role = roles.role[i];
sdbInfo("vgId:1, mnode:%d, role:%s", pMnode->mnodeId, mnodeGetMnodeRoleStr(pMnode->role)); sdbInfo("vgId:1, mnode:%d, role:%s", pMnode->mnodeId, syncRole[pMnode->role]);
if (pMnode->mnodeId == dnodeGetDnodeId()) tsSdbObj.role = pMnode->role; if (pMnode->mnodeId == dnodeGetDnodeId()) tsSdbMgmt.role = pMnode->role;
mnodeDecMnodeRef(pMnode); mnodeDecMnodeRef(pMnode);
} }
} }
@ -239,16 +239,16 @@ static uint32_t sdbGetFileInfo(void *ahandle, char *name, uint32_t *index, uint3
} }
static int32_t sdbGetWalInfo(void *ahandle, char *fileName, int64_t *fileId) { static int32_t sdbGetWalInfo(void *ahandle, char *fileName, int64_t *fileId) {
return walGetWalFile(tsSdbObj.wal, fileName, fileId); return walGetWalFile(tsSdbMgmt.wal, fileName, fileId);
} }
static void sdbNotifyRole(void *ahandle, int8_t role) { static void sdbNotifyRole(void *ahandle, int8_t role) {
sdbInfo("vgId:1, mnode role changed from %s to %s", mnodeGetMnodeRoleStr(tsSdbObj.role), mnodeGetMnodeRoleStr(role)); sdbInfo("vgId:1, mnode role changed from %s to %s", syncRole[tsSdbMgmt.role], syncRole[role]);
if (role == TAOS_SYNC_ROLE_MASTER && tsSdbObj.role != TAOS_SYNC_ROLE_MASTER) { if (role == TAOS_SYNC_ROLE_MASTER && tsSdbMgmt.role != TAOS_SYNC_ROLE_MASTER) {
balanceReset(); balanceReset();
} }
tsSdbObj.role = role; tsSdbMgmt.role = role;
sdbUpdateMnodeRoles(); sdbUpdateMnodeRoles();
} }
@ -276,7 +276,7 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) {
if (pWrite->retCode != TSDB_CODE_SUCCESS) { if (pWrite->retCode != TSDB_CODE_SUCCESS) {
SWalHead *pHead = (void *)pWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK; SWalHead *pHead = (void *)pWrite + sizeof(SSWriteMsg) + SDB_SYNC_HACK;
int32_t action = pHead->msgType % 10; int32_t action = pHead->msgType % 10;
sdbError("vgId:1, key:%p:%s hver:%" PRIu64 " action:%d, failed to foward since %s", pWrite->pObj, sdbError("vgId:1, key:%p:%s hver:%" PRIu64 " action:%d, failed to foward since %s", pWrite->pRow,
sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version, action, tstrerror(pWrite->retCode)); sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version, action, tstrerror(pWrite->retCode));
if (action == SDB_ACTION_INSERT) { if (action == SDB_ACTION_INSERT) {
// It's better to create a table in two stages, create it first and then set it success // It's better to create a table in two stages, create it first and then set it success
@ -284,7 +284,7 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = pWrite->pTable, .pTable = pWrite->pTable,
.pObj = pWrite->pObj .pRow = pWrite->pRow
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
} }
@ -297,7 +297,7 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) {
// if ahandle, means this func is called by sdb write // if ahandle, means this func is called by sdb write
if (ahandle == NULL) { if (ahandle == NULL) {
sdbDecRef(pWrite->pTable, pWrite->pObj); sdbDecRef(pWrite->pTable, pWrite->pRow);
} }
taosFreeQitem(pWrite); taosFreeQitem(pWrite);
@ -369,7 +369,7 @@ void sdbUpdateSync(void *pMnodes) {
return; return;
} }
if (memcmp(&syncCfg, &tsSdbObj.cfg, sizeof(SSyncCfg)) == 0) { if (memcmp(&syncCfg, &tsSdbMgmt.cfg, sizeof(SSyncCfg)) == 0) {
sdbDebug("vgId:1, update sync config, info not changed"); sdbDebug("vgId:1, update sync config, info not changed");
return; return;
} }
@ -391,18 +391,18 @@ void sdbUpdateSync(void *pMnodes) {
syncInfo.writeToCache = sdbWriteToQueue; syncInfo.writeToCache = sdbWriteToQueue;
syncInfo.confirmForward = sdbConfirmForward; syncInfo.confirmForward = sdbConfirmForward;
syncInfo.notifyRole = sdbNotifyRole; syncInfo.notifyRole = sdbNotifyRole;
tsSdbObj.cfg = syncCfg; tsSdbMgmt.cfg = syncCfg;
if (tsSdbObj.sync) { if (tsSdbMgmt.sync) {
syncReconfig(tsSdbObj.sync, &syncCfg); syncReconfig(tsSdbMgmt.sync, &syncCfg);
} else { } else {
tsSdbObj.sync = syncStart(&syncInfo); tsSdbMgmt.sync = syncStart(&syncInfo);
} }
sdbUpdateMnodeRoles(); sdbUpdateMnodeRoles();
} }
int32_t sdbInit() { int32_t sdbInit() {
pthread_mutex_init(&tsSdbObj.mutex, NULL); pthread_mutex_init(&tsSdbMgmt.mutex, NULL);
if (sdbInitWorker() != 0) { if (sdbInitWorker() != 0) {
return -1; return -1;
@ -415,55 +415,55 @@ int32_t sdbInit() {
sdbRestoreTables(); sdbRestoreTables();
if (mnodeGetMnodesNum() == 1) { if (mnodeGetMnodesNum() == 1) {
tsSdbObj.role = TAOS_SYNC_ROLE_MASTER; tsSdbMgmt.role = TAOS_SYNC_ROLE_MASTER;
} }
tsSdbObj.status = SDB_STATUS_SERVING; tsSdbMgmt.status = SDB_STATUS_SERVING;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void sdbCleanUp() { void sdbCleanUp() {
if (tsSdbObj.status != SDB_STATUS_SERVING) return; if (tsSdbMgmt.status != SDB_STATUS_SERVING) return;
tsSdbObj.status = SDB_STATUS_CLOSING; tsSdbMgmt.status = SDB_STATUS_CLOSING;
sdbCleanupWorker(); sdbCleanupWorker();
sdbDebug("vgId:1, sdb will be closed, mver:%" PRIu64, tsSdbObj.version); sdbDebug("vgId:1, sdb will be closed, mver:%" PRIu64, tsSdbMgmt.version);
if (tsSdbObj.sync) { if (tsSdbMgmt.sync) {
syncStop(tsSdbObj.sync); syncStop(tsSdbMgmt.sync);
tsSdbObj.sync = -1; tsSdbMgmt.sync = -1;
} }
if (tsSdbObj.wal) { if (tsSdbMgmt.wal) {
walClose(tsSdbObj.wal); walClose(tsSdbMgmt.wal);
tsSdbObj.wal = NULL; tsSdbMgmt.wal = NULL;
} }
pthread_mutex_destroy(&tsSdbObj.mutex); pthread_mutex_destroy(&tsSdbMgmt.mutex);
} }
void sdbIncRef(void *tparam, void *pObj) { void sdbIncRef(void *tparam, void *pRow) {
if (pObj == NULL || tparam == NULL) return; if (pRow == NULL || tparam == NULL) return;
SSdbTable *pTable = tparam; SSdbTable *pTable = tparam;
int32_t * pRefCount = (int32_t *)(pObj + 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 key:%p:%s:%d", pTable->tableName, pObj, sdbGetObjStr(pTable, pObj), refCount); sdbTrace("vgId:1, sdb:%s, inc ref to key:%p:%s:%d", pTable->tableName, pRow, sdbGetObjStr(pTable, pRow), refCount);
} }
void sdbDecRef(void *tparam, void *pObj) { void sdbDecRef(void *tparam, void *pRow) {
if (pObj == NULL || tparam == NULL) return; if (pRow == NULL || tparam == NULL) return;
SSdbTable *pTable = tparam; SSdbTable *pTable = tparam;
int32_t * pRefCount = (int32_t *)(pObj + 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 key:%p:%s:%d", pTable->tableName, pObj, sdbGetObjStr(pTable, pObj), refCount); sdbTrace("vgId:1, sdb:%s, dec ref to key:%p:%s:%d", pTable->tableName, pRow, sdbGetObjStr(pTable, pRow), refCount);
int32_t *updateEnd = pObj + pTable->refCountPos - 4; int32_t *updateEnd = pRow + pTable->refCountPos - 4;
if (refCount <= 0 && *updateEnd) { if (refCount <= 0 && *updateEnd) {
sdbTrace("vgId:1, sdb:%s, key:%p:%s:%d destroyed", pTable->tableName, pObj, sdbGetObjStr(pTable, pObj), refCount); sdbTrace("vgId:1, sdb:%s, key:%p:%s:%d destroyed", pTable->tableName, pRow, sdbGetObjStr(pTable, pRow), refCount);
SSWriteMsg wmsg = {.pObj = pObj}; SSWriteMsg wmsg = {.pRow = pRow};
(*pTable->fpDestroy)(&wmsg); (*pTable->fpDestroy)(&wmsg);
} }
} }
@ -502,7 +502,7 @@ static void *sdbGetRowFromObj(SSdbTable *pTable, void *key) {
} }
static int32_t sdbInsertHash(SSdbTable *pTable, SSWriteMsg *pWrite) { static int32_t sdbInsertHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
void * key = sdbGetObjKey(pTable, pWrite->pObj); void * key = sdbGetObjKey(pTable, pWrite->pRow);
int32_t keySize = sizeof(int32_t); int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) { if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
@ -510,25 +510,25 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
} }
pthread_mutex_lock(&pTable->mutex); pthread_mutex_lock(&pTable->mutex);
taosHashPut(pTable->iHandle, key, keySize, &pWrite->pObj, sizeof(int64_t)); taosHashPut(pTable->iHandle, key, keySize, &pWrite->pRow, sizeof(int64_t));
pthread_mutex_unlock(&pTable->mutex); pthread_mutex_unlock(&pTable->mutex);
sdbIncRef(pTable, pWrite->pObj); sdbIncRef(pTable, pWrite->pRow);
atomic_add_fetch_32(&pTable->numOfRows, 1); atomic_add_fetch_32(&pTable->numOfRows, 1);
if (pTable->keyType == SDB_KEY_AUTO) { if (pTable->keyType == SDB_KEY_AUTO) {
pTable->autoIndex = MAX(pTable->autoIndex, *((uint32_t *)pWrite->pObj)); pTable->autoIndex = MAX(pTable->autoIndex, *((uint32_t *)pWrite->pRow));
} else { } else {
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->tableName,
sdbGetObjStr(pTable, pWrite->pObj), pWrite->rowSize, pTable->numOfRows, pWrite->pMsg); sdbGetObjStr(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->tableName,
sdbGetObjStr(pTable, pWrite->pObj)); sdbGetObjStr(pTable, pWrite->pRow));
sdbDeleteHash(pTable, pWrite); sdbDeleteHash(pTable, pWrite);
} }
@ -536,17 +536,17 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
} }
static int32_t sdbDeleteHash(SSdbTable *pTable, SSWriteMsg *pWrite) { static int32_t sdbDeleteHash(SSdbTable *pTable, SSWriteMsg *pWrite) {
int32_t *updateEnd = pWrite->pObj + 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->tableName,
sdbGetObjStr(pTable, pWrite->pObj)); sdbGetObjStr(pTable, pWrite->pRow));
return TSDB_CODE_MND_SDB_OBJ_NOT_THERE; return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
} }
(*pTable->fpDelete)(pWrite); (*pTable->fpDelete)(pWrite);
void * key = sdbGetObjKey(pTable, pWrite->pObj); void * key = sdbGetObjKey(pTable, pWrite->pRow);
int32_t keySize = sizeof(int32_t); int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) { if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
keySize = strlen((char *)key); keySize = strlen((char *)key);
@ -559,16 +559,16 @@ 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->tableName,
sdbGetObjStr(pTable, pWrite->pObj), pTable->numOfRows, pWrite->pMsg); sdbGetObjStr(pTable, pWrite->pRow), pTable->numOfRows, pWrite->pMsg);
sdbDecRef(pTable, pWrite->pObj); sdbDecRef(pTable, pWrite->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
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->tableName,
sdbGetObjStr(pTable, pWrite->pObj), pTable->numOfRows, pWrite->pMsg); sdbGetObjStr(pTable, pWrite->pRow), pTable->numOfRows, pWrite->pMsg);
(*pTable->fpUpdate)(pWrite); (*pTable->fpUpdate)(pWrite);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -583,42 +583,42 @@ static int sdbWrite(void *wparam, void *hparam, int32_t qtype, void *unused) {
SSdbTable *pTable = sdbGetTableFromId(tableId); SSdbTable *pTable = sdbGetTableFromId(tableId);
assert(pTable != NULL); assert(pTable != NULL);
pthread_mutex_lock(&tsSdbObj.mutex); pthread_mutex_lock(&tsSdbMgmt.mutex);
if (pHead->version == 0) { if (pHead->version == 0) {
// assign version // assign version
tsSdbObj.version++; tsSdbMgmt.version++;
pHead->version = tsSdbObj.version; pHead->version = tsSdbMgmt.version;
} else { } else {
// for data from WAL or forward, version may be smaller // for data from WAL or forward, version may be smaller
if (pHead->version <= tsSdbObj.version) { if (pHead->version <= tsSdbMgmt.version) {
pthread_mutex_unlock(&tsSdbObj.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, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbObj.version); pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (pHead->version != tsSdbObj.version + 1) { } else if (pHead->version != tsSdbMgmt.version + 1) {
pthread_mutex_unlock(&tsSdbObj.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, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbObj.version); pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version);
return TSDB_CODE_SYN_INVALID_VERSION; return TSDB_CODE_SYN_INVALID_VERSION;
} else { } else {
tsSdbObj.version = pHead->version; tsSdbMgmt.version = pHead->version;
} }
} }
int32_t code = walWrite(tsSdbObj.wal, pHead); int32_t code = walWrite(tsSdbMgmt.wal, pHead);
if (code < 0) { if (code < 0) {
pthread_mutex_unlock(&tsSdbObj.mutex); pthread_mutex_unlock(&tsSdbMgmt.mutex);
return code; return code;
} }
pthread_mutex_unlock(&tsSdbObj.mutex); pthread_mutex_unlock(&tsSdbMgmt.mutex);
// from app, wmsg is created // from app, wmsg is created
if (pWrite != NULL) { if (pWrite != NULL) {
// forward to peers // forward to peers
pWrite->processedCount = 0; pWrite->processedCount = 0;
int32_t syncCode = syncForwardToPeer(tsSdbObj.sync, pHead, pWrite, TAOS_QTYPE_RPC); int32_t syncCode = syncForwardToPeer(tsSdbMgmt.sync, pHead, pWrite, TAOS_QTYPE_RPC);
if (syncCode <= 0) pWrite->processedCount = 1; if (syncCode <= 0) pWrite->processedCount = 1;
if (syncCode < 0) { if (syncCode < 0) {
@ -638,7 +638,7 @@ static int sdbWrite(void *wparam, void *hparam, int32_t qtype, void *unused) {
sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version); sdbGetActionStr(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
syncForwardToPeer(tsSdbObj.sync, pHead, pWrite, TAOS_QTYPE_RPC); syncForwardToPeer(tsSdbMgmt.sync, pHead, pWrite, TAOS_QTYPE_RPC);
// from wal or forward msg, wmsg not created, should add into hash // from wal or forward msg, wmsg not created, should add into hash
if (action == SDB_ACTION_INSERT) { if (action == SDB_ACTION_INSERT) {
@ -652,7 +652,7 @@ static int sdbWrite(void *wparam, void *hparam, int32_t qtype, void *unused) {
sdbGetKeyStr(pTable, pHead->cont)); sdbGetKeyStr(pTable, pHead->cont));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SSWriteMsg wmsg = {.pTable = pTable, .pObj = pRow}; SSWriteMsg wmsg = {.pTable = pTable, .pRow = pRow};
return sdbDeleteHash(pTable, &wmsg); return sdbDeleteHash(pTable, &wmsg);
} else if (action == SDB_ACTION_UPDATE) { } else if (action == SDB_ACTION_UPDATE) {
void *pRow = sdbGetRowMeta(pTable, pHead->cont); void *pRow = sdbGetRowMeta(pTable, pHead->cont);
@ -673,19 +673,19 @@ int32_t sdbInsertRow(SSWriteMsg *pWrite) {
SSdbTable *pTable = pWrite->pTable; SSdbTable *pTable = pWrite->pTable;
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->pObj)) { 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, already exist", pTable->tableName,
sdbGetObjStr(pTable, pWrite->pObj)); sdbGetObjStr(pTable, pWrite->pRow));
sdbDecRef(pTable, pWrite->pObj); sdbDecRef(pTable, pWrite->pRow);
return TSDB_CODE_MND_SDB_OBJ_ALREADY_THERE; return TSDB_CODE_MND_SDB_OBJ_ALREADY_THERE;
} }
if (pTable->keyType == SDB_KEY_AUTO) { if (pTable->keyType == SDB_KEY_AUTO) {
*((uint32_t *)pWrite->pObj) = 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->tableName, "vgroups") == 0) {
*((uint32_t *)pWrite->pObj) = atomic_add_fetch_32(&pTable->autoIndex, 1); *((uint32_t *)pWrite->pRow) = atomic_add_fetch_32(&pTable->autoIndex, 1);
} }
} }
@ -727,10 +727,10 @@ int32_t sdbInsertRowImp(SSWriteMsg *pWrite) {
if (pNewOper->pMsg != NULL) { if (pNewOper->pMsg != NULL) {
sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s key:%p:%s, insert action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle, sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s key:%p:%s, insert action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle,
pNewOper->pMsg, pTable->tableName, pWrite->pObj, sdbGetObjStr(pTable, pWrite->pObj)); pNewOper->pMsg, pTable->tableName, pWrite->pRow, sdbGetObjStr(pTable, pWrite->pRow));
} }
sdbIncRef(pNewOper->pTable, pNewOper->pObj); sdbIncRef(pNewOper->pTable, pNewOper->pRow);
taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewOper); taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewOper);
return TSDB_CODE_MND_ACTION_IN_PROGRESS; return TSDB_CODE_MND_ACTION_IN_PROGRESS;
@ -748,24 +748,24 @@ int32_t sdbDeleteRow(SSWriteMsg *pWrite) {
SSdbTable *pTable = pWrite->pTable; SSdbTable *pTable = pWrite->pTable;
if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE; if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
void *pRow = sdbGetRowMetaFromObj(pTable, pWrite->pObj); 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->tableName);
return TSDB_CODE_MND_SDB_OBJ_NOT_THERE; return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
} }
sdbIncRef(pTable, pWrite->pObj); sdbIncRef(pTable, pWrite->pRow);
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->tableName);
sdbDecRef(pTable, pWrite->pObj); sdbDecRef(pTable, pWrite->pRow);
return code; return code;
} }
// just delete data from memory // just delete data from memory
if (pWrite->type != SDB_OPER_GLOBAL) { if (pWrite->type != SDB_OPER_GLOBAL) {
sdbDecRef(pTable, pWrite->pObj); sdbDecRef(pTable, pWrite->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -795,7 +795,7 @@ int32_t sdbDeleteRowImp(SSWriteMsg *pWrite) {
if (pNewOper->pMsg != NULL) { if (pNewOper->pMsg != NULL) {
sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s key:%p:%s, delete action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle, sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s key:%p:%s, delete action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle,
pNewOper->pMsg, pTable->tableName, pWrite->pObj, sdbGetObjStr(pTable, pWrite->pObj)); pNewOper->pMsg, pTable->tableName, pWrite->pRow, sdbGetObjStr(pTable, pWrite->pRow));
} }
taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewOper); taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewOper);
@ -807,7 +807,7 @@ int32_t sdbUpdateRow(SSWriteMsg *pWrite) {
SSdbTable *pTable = pWrite->pTable; SSdbTable *pTable = pWrite->pTable;
if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE; if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
void *pRow = sdbGetRowMetaFromObj(pTable, pWrite->pObj); 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->tableName);
return TSDB_CODE_MND_SDB_OBJ_NOT_THERE; return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
@ -850,10 +850,10 @@ int32_t sdbUpdateRowImp(SSWriteMsg *pWrite) {
if (pNewOper->pMsg != NULL) { if (pNewOper->pMsg != NULL) {
sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s key:%p:%s, update action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle, sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s key:%p:%s, update action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle,
pNewOper->pMsg, pTable->tableName, pWrite->pObj, sdbGetObjStr(pTable, pWrite->pObj)); pNewOper->pMsg, pTable->tableName, pWrite->pRow, sdbGetObjStr(pTable, pWrite->pRow));
} }
sdbIncRef(pNewOper->pTable, pNewOper->pObj); sdbIncRef(pNewOper->pTable, pNewOper->pRow);
taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewOper); taosWriteQitem(tsSdbWQueue, TAOS_QTYPE_RPC, pNewOper);
return TSDB_CODE_MND_ACTION_IN_PROGRESS; return TSDB_CODE_MND_ACTION_IN_PROGRESS;
@ -910,7 +910,7 @@ void *sdbOpenTable(SSdbTableDesc *pDesc) {
pTable->fpEncode = pDesc->fpEncode; pTable->fpEncode = pDesc->fpEncode;
pTable->fpDecode = pDesc->fpDecode; pTable->fpDecode = pDesc->fpDecode;
pTable->fpDestroy = pDesc->fpDestroy; pTable->fpDestroy = pDesc->fpDestroy;
pTable->fpDestored = pDesc->fpDestored; pTable->fpRestored = pDesc->fpRestored;
_hash_fn_t hashFp = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT); _hash_fn_t hashFp = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) { if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
@ -918,8 +918,8 @@ void *sdbOpenTable(SSdbTableDesc *pDesc) {
} }
pTable->iHandle = taosHashInit(pTable->hashSessions, hashFp, true, true); pTable->iHandle = taosHashInit(pTable->hashSessions, hashFp, true, true);
tsSdbObj.numOfTables++; tsSdbMgmt.numOfTables++;
tsSdbObj.tableList[pTable->tableId] = pTable; tsSdbMgmt.tableList[pTable->tableId] = pTable;
return pTable; return pTable;
} }
@ -927,8 +927,8 @@ void sdbCloseTable(void *handle) {
SSdbTable *pTable = (SSdbTable *)handle; SSdbTable *pTable = (SSdbTable *)handle;
if (pTable == NULL) return; if (pTable == NULL) return;
tsSdbObj.numOfTables--; tsSdbMgmt.numOfTables--;
tsSdbObj.tableList[pTable->tableId] = NULL; tsSdbMgmt.tableList[pTable->tableId] = NULL;
SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle); SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle);
while (taosHashIterNext(pIter)) { while (taosHashIterNext(pIter)) {
@ -936,7 +936,7 @@ void sdbCloseTable(void *handle) {
if (ppRow == NULL) continue; if (ppRow == NULL) continue;
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.pObj = *ppRow, .pRow = *ppRow,
.pTable = pTable, .pTable = pTable,
}; };
@ -947,7 +947,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, tsSdbObj.numOfTables); sdbDebug("vgId:1, sdb:%s, is closed, numOfTables:%d", pTable->tableName, tsSdbMgmt.numOfTables);
free(pTable); free(pTable);
} }
@ -1072,7 +1072,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 key:%p:%s hver:%" PRIu64 ", will be processed in sdb queue", sdbDebug("vgId:1, ahandle:%p msg:%p, sdb:%s key:%p:%s hver:%" PRIu64 ", will be processed in sdb queue",
pWrite->pMsg->rpcMsg.ahandle, pWrite->pMsg, pWrite->pTable->tableName, pWrite->pObj, pWrite->pMsg->rpcMsg.ahandle, pWrite->pMsg, pWrite->pTable->tableName, pWrite->pRow,
sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version); sdbGetKeyStr(pWrite->pTable, pHead->cont), pHead->version);
} }
} else { } else {
@ -1089,7 +1089,7 @@ static void *sdbWorkerFp(void *pWorker) {
} }
} }
walFsync(tsSdbObj.wal, true); walFsync(tsSdbMgmt.wal, true);
// browse all items, and process them one by one // browse all items, and process them one by one
taosResetQitems(tsSdbWQall); taosResetQitems(tsSdbWQall);
@ -1101,7 +1101,7 @@ static void *sdbWorkerFp(void *pWorker) {
sdbConfirmForward(NULL, pWrite, pWrite->retCode); sdbConfirmForward(NULL, pWrite, pWrite->retCode);
} else if (qtype == TAOS_QTYPE_FWD) { } else if (qtype == TAOS_QTYPE_FWD) {
pHead = (SWalHead *)item; pHead = (SWalHead *)item;
syncConfirmForward(tsSdbObj.sync, pHead->version, pHead->len); syncConfirmForward(tsSdbMgmt.sync, pHead->version, pHead->len);
taosFreeQitem(item); taosFreeQitem(item);
} else { } else {
taosFreeQitem(item); taosFreeQitem(item);

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

@ -100,12 +100,12 @@ static void mnodeDestroyChildTable(SCTableObj *pTable) {
} }
static int32_t mnodeChildTableActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeChildTableActionDestroy(SSWriteMsg *pWMsg) {
mnodeDestroyChildTable(pWMsg->pObj); mnodeDestroyChildTable(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeChildTableActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeChildTableActionInsert(SSWriteMsg *pWMsg) {
SCTableObj *pTable = pWMsg->pObj; SCTableObj *pTable = pWMsg->pRow;
SVgObj *pVgroup = mnodeGetVgroup(pTable->vgId); SVgObj *pVgroup = mnodeGetVgroup(pTable->vgId);
if (pVgroup == NULL) { if (pVgroup == NULL) {
@ -154,7 +154,7 @@ static int32_t mnodeChildTableActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeChildTableActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeChildTableActionDelete(SSWriteMsg *pWMsg) {
SCTableObj *pTable = pWMsg->pObj; SCTableObj *pTable = pWMsg->pRow;
if (pTable->vgId == 0) { if (pTable->vgId == 0) {
return TSDB_CODE_MND_VGROUP_NOT_EXIST; return TSDB_CODE_MND_VGROUP_NOT_EXIST;
} }
@ -190,7 +190,7 @@ static int32_t mnodeChildTableActionDelete(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeChildTableActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeChildTableActionUpdate(SSWriteMsg *pWMsg) {
SCTableObj *pNew = pWMsg->pObj; SCTableObj *pNew = pWMsg->pRow;
SCTableObj *pTable = mnodeGetChildTable(pNew->info.tableId); SCTableObj *pTable = mnodeGetChildTable(pNew->info.tableId);
if (pTable != pNew) { if (pTable != pNew) {
void *oldTableId = pTable->info.tableId; void *oldTableId = pTable->info.tableId;
@ -217,7 +217,7 @@ static int32_t mnodeChildTableActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeChildTableActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeChildTableActionEncode(SSWriteMsg *pWMsg) {
SCTableObj *pTable = pWMsg->pObj; SCTableObj *pTable = pWMsg->pRow;
assert(pTable != NULL && pWMsg->rowData != NULL); assert(pTable != NULL && pWMsg->rowData != NULL);
int32_t len = strlen(pTable->info.tableId); int32_t len = strlen(pTable->info.tableId);
@ -282,7 +282,7 @@ static int32_t mnodeChildTableActionDecode(SSWriteMsg *pWMsg) {
} }
} }
pWMsg->pObj = pTable; pWMsg->pRow = pTable;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -297,7 +297,7 @@ static int32_t mnodeChildTableActionRestored() {
SDbObj *pDb = mnodeGetDbByTableId(pTable->info.tableId); SDbObj *pDb = mnodeGetDbByTableId(pTable->info.tableId);
if (pDb == NULL || pDb->status != TSDB_DB_STATUS_READY) { if (pDb == NULL || pDb->status != TSDB_DB_STATUS_READY) {
mError("ctable:%s, failed to get db or db in dropping, discard it", pTable->info.tableId); mError("ctable:%s, failed to get db or db in dropping, discard it", pTable->info.tableId);
SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .pTable = tsChildTableSdb}; SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pRow = pTable, .pTable = tsChildTableSdb};
sdbDeleteRow(&desc); sdbDeleteRow(&desc);
mnodeDecTableRef(pTable); mnodeDecTableRef(pTable);
mnodeDecDbRef(pDb); mnodeDecDbRef(pDb);
@ -309,7 +309,7 @@ static int32_t mnodeChildTableActionRestored() {
if (pVgroup == NULL) { if (pVgroup == NULL) {
mError("ctable:%s, failed to get vgId:%d tid:%d, discard it", pTable->info.tableId, pTable->vgId, pTable->tid); mError("ctable:%s, failed to get vgId:%d tid:%d, discard it", pTable->info.tableId, pTable->vgId, pTable->tid);
pTable->vgId = 0; pTable->vgId = 0;
SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .pTable = tsChildTableSdb}; SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pRow = pTable, .pTable = tsChildTableSdb};
sdbDeleteRow(&desc); sdbDeleteRow(&desc);
mnodeDecTableRef(pTable); mnodeDecTableRef(pTable);
continue; continue;
@ -320,7 +320,7 @@ static int32_t mnodeChildTableActionRestored() {
mError("ctable:%s, db:%s not match with vgId:%d db:%s sid:%d, discard it", mError("ctable:%s, db:%s not match with vgId:%d db:%s sid:%d, discard it",
pTable->info.tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->tid); pTable->info.tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->tid);
pTable->vgId = 0; pTable->vgId = 0;
SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .pTable = tsChildTableSdb}; SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pRow = pTable, .pTable = tsChildTableSdb};
sdbDeleteRow(&desc); sdbDeleteRow(&desc);
mnodeDecTableRef(pTable); mnodeDecTableRef(pTable);
continue; continue;
@ -331,7 +331,7 @@ static int32_t mnodeChildTableActionRestored() {
if (pSuperTable == NULL) { if (pSuperTable == NULL) {
mError("ctable:%s, stable:%" PRIu64 " not exist", pTable->info.tableId, pTable->suid); mError("ctable:%s, stable:%" PRIu64 " not exist", pTable->info.tableId, pTable->suid);
pTable->vgId = 0; pTable->vgId = 0;
SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .pTable = tsChildTableSdb}; SSWriteMsg desc = {.type = SDB_OPER_LOCAL, .pRow = pTable, .pTable = tsChildTableSdb};
sdbDeleteRow(&desc); sdbDeleteRow(&desc);
mnodeDecTableRef(pTable); mnodeDecTableRef(pTable);
continue; continue;
@ -364,7 +364,7 @@ static int32_t mnodeInitChildTables() {
.fpEncode = mnodeChildTableActionEncode, .fpEncode = mnodeChildTableActionEncode,
.fpDecode = mnodeChildTableActionDecode, .fpDecode = mnodeChildTableActionDecode,
.fpDestroy = mnodeChildTableActionDestroy, .fpDestroy = mnodeChildTableActionDestroy,
.fpDestored = mnodeChildTableActionRestored .fpRestored = mnodeChildTableActionRestored
}; };
tsChildTableSdb = sdbOpenTable(&tableDesc); tsChildTableSdb = sdbOpenTable(&tableDesc);
@ -431,12 +431,12 @@ static void mnodeDestroySuperTable(SSTableObj *pStable) {
} }
static int32_t mnodeSuperTableActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeSuperTableActionDestroy(SSWriteMsg *pWMsg) {
mnodeDestroySuperTable(pWMsg->pObj); mnodeDestroySuperTable(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeSuperTableActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeSuperTableActionInsert(SSWriteMsg *pWMsg) {
SSTableObj *pStable = pWMsg->pObj; SSTableObj *pStable = pWMsg->pRow;
SDbObj *pDb = mnodeGetDbByTableId(pStable->info.tableId); SDbObj *pDb = mnodeGetDbByTableId(pStable->info.tableId);
if (pDb != NULL && pDb->status == TSDB_DB_STATUS_READY) { if (pDb != NULL && pDb->status == TSDB_DB_STATUS_READY) {
mnodeAddSuperTableIntoDb(pDb); mnodeAddSuperTableIntoDb(pDb);
@ -447,7 +447,7 @@ static int32_t mnodeSuperTableActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeSuperTableActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeSuperTableActionDelete(SSWriteMsg *pWMsg) {
SSTableObj *pStable = pWMsg->pObj; SSTableObj *pStable = pWMsg->pRow;
SDbObj *pDb = mnodeGetDbByTableId(pStable->info.tableId); SDbObj *pDb = mnodeGetDbByTableId(pStable->info.tableId);
if (pDb != NULL) { if (pDb != NULL) {
mnodeRemoveSuperTableFromDb(pDb); mnodeRemoveSuperTableFromDb(pDb);
@ -459,7 +459,7 @@ static int32_t mnodeSuperTableActionDelete(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeSuperTableActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeSuperTableActionUpdate(SSWriteMsg *pWMsg) {
SSTableObj *pNew = pWMsg->pObj; SSTableObj *pNew = pWMsg->pRow;
SSTableObj *pTable = mnodeGetSuperTable(pNew->info.tableId); SSTableObj *pTable = mnodeGetSuperTable(pNew->info.tableId);
if (pTable != NULL && pTable != pNew) { if (pTable != NULL && pTable != pNew) {
void *oldTableId = pTable->info.tableId; void *oldTableId = pTable->info.tableId;
@ -484,8 +484,8 @@ static int32_t mnodeSuperTableActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeSuperTableActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeSuperTableActionEncode(SSWriteMsg *pWMsg) {
SSTableObj *pStable = pWMsg->pObj; SSTableObj *pStable = pWMsg->pRow;
assert(pWMsg->pObj != NULL && pWMsg->rowData != NULL); assert(pWMsg->pRow != NULL && pWMsg->rowData != NULL);
int32_t len = strlen(pStable->info.tableId); int32_t len = strlen(pStable->info.tableId);
if (len >= TSDB_TABLE_FNAME_LEN) len = TSDB_CODE_MND_INVALID_TABLE_ID; if (len >= TSDB_TABLE_FNAME_LEN) len = TSDB_CODE_MND_INVALID_TABLE_ID;
@ -531,7 +531,7 @@ static int32_t mnodeSuperTableActionDecode(SSWriteMsg *pWMsg) {
memcpy(pStable->schema, pWMsg->rowData + len, schemaSize); memcpy(pStable->schema, pWMsg->rowData + len, schemaSize);
pWMsg->pObj = pStable; pWMsg->pRow = pStable;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -557,7 +557,7 @@ static int32_t mnodeInitSuperTables() {
.fpEncode = mnodeSuperTableActionEncode, .fpEncode = mnodeSuperTableActionEncode,
.fpDecode = mnodeSuperTableActionDecode, .fpDecode = mnodeSuperTableActionDecode,
.fpDestroy = mnodeSuperTableActionDestroy, .fpDestroy = mnodeSuperTableActionDestroy,
.fpDestored = mnodeSuperTableActionRestored .fpRestored = mnodeSuperTableActionRestored
}; };
tsSuperTableSdb = sdbOpenTable(&tableDesc); tsSuperTableSdb = sdbOpenTable(&tableDesc);
@ -828,7 +828,7 @@ static int32_t mnodeCreateSuperTableCb(SMnodeMsg *pMsg, int32_t code) {
} else { } else {
mError("app:%p:%p, stable:%s, failed to create in sdb, reason:%s", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId, mError("app:%p:%p, stable:%s, failed to create in sdb, reason:%s", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId,
tstrerror(code)); tstrerror(code));
SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pObj = pTable, .pTable = tsSuperTableSdb}; SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pRow = pTable, .pTable = tsSuperTableSdb};
sdbDeleteRow(&desc); sdbDeleteRow(&desc);
} }
@ -881,7 +881,7 @@ static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.rowSize = sizeof(SSTableObj) + schemaSize, .rowSize = sizeof(SSTableObj) + schemaSize,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeCreateSuperTableCb .fpWrite = mnodeCreateSuperTableCb
@ -940,7 +940,7 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeDropSuperTableCb .fpWrite = mnodeDropSuperTableCb
}; };
@ -1013,7 +1013,7 @@ static int32_t mnodeAddSuperTableTag(SMnodeMsg *pMsg, SSchema schema[], int32_t
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeAddSuperTableTagCb .fpWrite = mnodeAddSuperTableTagCb
}; };
@ -1047,7 +1047,7 @@ static int32_t mnodeDropSuperTableTag(SMnodeMsg *pMsg, char *tagName) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeDropSuperTableTagCb .fpWrite = mnodeDropSuperTableTagCb
}; };
@ -1091,7 +1091,7 @@ static int32_t mnodeModifySuperTableTagName(SMnodeMsg *pMsg, char *oldTagName, c
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeModifySuperTableTagNameCb .fpWrite = mnodeModifySuperTableTagNameCb
}; };
@ -1165,7 +1165,7 @@ static int32_t mnodeAddSuperTableColumn(SMnodeMsg *pMsg, SSchema schema[], int32
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeAddSuperTableColumnCb .fpWrite = mnodeAddSuperTableColumnCb
}; };
@ -1210,7 +1210,7 @@ static int32_t mnodeDropSuperTableColumn(SMnodeMsg *pMsg, char *colName) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeDropSuperTableColumnCb .fpWrite = mnodeDropSuperTableColumnCb
}; };
@ -1254,7 +1254,7 @@ static int32_t mnodeChangeSuperTableColumn(SMnodeMsg *pMsg, char *oldName, char
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pStable, .pRow = pStable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeChangeSuperTableColumnCb .fpWrite = mnodeChangeSuperTableColumnCb
}; };
@ -1420,7 +1420,7 @@ void mnodeDropAllSuperTables(SDbObj *pDropDb) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsSuperTableSdb, .pTable = tsSuperTableSdb,
.pObj = pTable, .pRow = pTable,
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
numOfTables ++; numOfTables ++;
@ -1694,7 +1694,7 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) {
} else { } else {
mError("app:%p:%p, table:%s, failed to create table sid:%d, uid:%" PRIu64 ", reason:%s", pMsg->rpcMsg.ahandle, pMsg, mError("app:%p:%p, table:%s, failed to create table sid:%d, uid:%" PRIu64 ", reason:%s", pMsg->rpcMsg.ahandle, pMsg,
pTable->info.tableId, pTable->tid, pTable->uid, tstrerror(code)); pTable->info.tableId, pTable->tid, pTable->uid, tstrerror(code));
SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pObj = pTable, .pTable = tsChildTableSdb}; SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pRow = pTable, .pTable = tsChildTableSdb};
sdbDeleteRow(&desc); sdbDeleteRow(&desc);
return code; return code;
} }
@ -1782,7 +1782,7 @@ static int32_t mnodeDoCreateChildTable(SMnodeMsg *pMsg, int32_t tid) {
SSWriteMsg desc = { SSWriteMsg desc = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pObj = pTable, .pRow = pTable,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pMsg = pMsg, .pMsg = pMsg,
.fpReq = mnodeDoCreateChildTableFp .fpReq = mnodeDoCreateChildTableFp
@ -1904,7 +1904,7 @@ static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pObj = pTable, .pRow = pTable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeDropChildTableCb .fpWrite = mnodeDropChildTableCb
}; };
@ -2008,7 +2008,7 @@ static int32_t mnodeAddNormalTableColumn(SMnodeMsg *pMsg, SSchema schema[], int3
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pObj = pTable, .pRow = pTable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeAlterNormalTableColumnCb .fpWrite = mnodeAlterNormalTableColumnCb
}; };
@ -2041,7 +2041,7 @@ static int32_t mnodeDropNormalTableColumn(SMnodeMsg *pMsg, char *colName) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pObj = pTable, .pRow = pTable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeAlterNormalTableColumnCb .fpWrite = mnodeAlterNormalTableColumnCb
}; };
@ -2078,7 +2078,7 @@ static int32_t mnodeChangeNormalTableColumn(SMnodeMsg *pMsg, char *oldName, char
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pObj = pTable, .pRow = pTable,
.pMsg = pMsg, .pMsg = pMsg,
.fpWrite = mnodeAlterNormalTableColumnCb .fpWrite = mnodeAlterNormalTableColumnCb
}; };
@ -2221,7 +2221,7 @@ void mnodeDropAllChildTablesInVgroups(SVgObj *pVgroup) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pObj = pTable, .pRow = pTable,
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
numOfTables++; numOfTables++;
@ -2254,7 +2254,7 @@ void mnodeDropAllChildTables(SDbObj *pDropDb) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pObj = pTable, .pRow = pTable,
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
numOfTables++; numOfTables++;
@ -2283,7 +2283,7 @@ static void mnodeDropAllChildTablesInStable(SSTableObj *pStable) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pObj = pTable, .pRow = pTable,
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
numOfTables++; numOfTables++;
@ -2412,7 +2412,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
if (rpcMsg->code == TSDB_CODE_SUCCESS || rpcMsg->code == TSDB_CODE_TDB_TABLE_ALREADY_EXIST) { if (rpcMsg->code == TSDB_CODE_SUCCESS || rpcMsg->code == TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
SSWriteMsg desc = { SSWriteMsg desc = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pObj = pTable, .pRow = pTable,
.pTable = tsChildTableSdb, .pTable = tsChildTableSdb,
.pMsg = mnodeMsg, .pMsg = mnodeMsg,
.fpWrite = mnodeDoCreateChildTableCb .fpWrite = mnodeDoCreateChildTableCb
@ -2440,7 +2440,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid, mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid,
tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle, mnodeMsg->incomingTs, sec, mnodeMsg->retry); tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle, mnodeMsg->incomingTs, sec, mnodeMsg->retry);
SSWriteMsg wmsg = {.type = SDB_OPER_GLOBAL, .pTable = tsChildTableSdb, .pObj = pTable}; SSWriteMsg wmsg = {.type = SDB_OPER_GLOBAL, .pTable = tsChildTableSdb, .pRow = pTable};
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
if (rpcMsg->code == TSDB_CODE_APP_NOT_READY) { if (rpcMsg->code == TSDB_CODE_APP_NOT_READY) {

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

@ -43,12 +43,12 @@ static int32_t mnodeProcessDropUserMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessAuthMsg(SMnodeMsg *pMsg); static int32_t mnodeProcessAuthMsg(SMnodeMsg *pMsg);
static int32_t mnodeUserActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeUserActionDestroy(SSWriteMsg *pWMsg) {
tfree(pWMsg->pObj); tfree(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeUserActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeUserActionInsert(SSWriteMsg *pWMsg) {
SUserObj *pUser = pWMsg->pObj; SUserObj *pUser = pWMsg->pRow;
SAcctObj *pAcct = mnodeGetAcct(pUser->acct); SAcctObj *pAcct = mnodeGetAcct(pUser->acct);
if (pAcct != NULL) { if (pAcct != NULL) {
@ -63,7 +63,7 @@ static int32_t mnodeUserActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeUserActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeUserActionDelete(SSWriteMsg *pWMsg) {
SUserObj *pUser = pWMsg->pObj; SUserObj *pUser = pWMsg->pRow;
SAcctObj *pAcct = mnodeGetAcct(pUser->acct); SAcctObj *pAcct = mnodeGetAcct(pUser->acct);
if (pAcct != NULL) { if (pAcct != NULL) {
@ -75,7 +75,7 @@ static int32_t mnodeUserActionDelete(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeUserActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeUserActionUpdate(SSWriteMsg *pWMsg) {
SUserObj *pUser = pWMsg->pObj; SUserObj *pUser = pWMsg->pRow;
SUserObj *pSaved = mnodeGetUser(pUser->user); SUserObj *pSaved = mnodeGetUser(pUser->user);
if (pUser != pSaved) { if (pUser != pSaved) {
memcpy(pSaved, pUser, tsUserUpdateSize); memcpy(pSaved, pUser, tsUserUpdateSize);
@ -86,7 +86,7 @@ static int32_t mnodeUserActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeUserActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeUserActionEncode(SSWriteMsg *pWMsg) {
SUserObj *pUser = pWMsg->pObj; SUserObj *pUser = pWMsg->pRow;
memcpy(pWMsg->rowData, pUser, tsUserUpdateSize); memcpy(pWMsg->rowData, pUser, tsUserUpdateSize);
pWMsg->rowSize = tsUserUpdateSize; pWMsg->rowSize = tsUserUpdateSize;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -97,7 +97,7 @@ static int32_t mnodeUserActionDecode(SSWriteMsg *pWMsg) {
if (pUser == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY; if (pUser == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
memcpy(pUser, pWMsg->rowData, tsUserUpdateSize); memcpy(pUser, pWMsg->rowData, tsUserUpdateSize);
pWMsg->pObj = pUser; pWMsg->pRow = pUser;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -163,7 +163,7 @@ int32_t mnodeInitUsers() {
.fpEncode = mnodeUserActionEncode, .fpEncode = mnodeUserActionEncode,
.fpDecode = mnodeUserActionDecode, .fpDecode = mnodeUserActionDecode,
.fpDestroy = mnodeUserActionDestroy, .fpDestroy = mnodeUserActionDestroy,
.fpDestored = mnodeUserActionRestored .fpRestored = mnodeUserActionRestored
}; };
tsUserSdb = sdbOpenTable(&tableDesc); tsUserSdb = sdbOpenTable(&tableDesc);
@ -208,7 +208,7 @@ static int32_t mnodeUpdateUser(SUserObj *pUser, void *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsUserSdb, .pTable = tsUserSdb,
.pObj = pUser, .pRow = pUser,
.pMsg = pMsg .pMsg = pMsg
}; };
@ -262,7 +262,7 @@ int32_t mnodeCreateUser(SAcctObj *pAcct, char *name, char *pass, void *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsUserSdb, .pTable = tsUserSdb,
.pObj = pUser, .pRow = pUser,
.rowSize = sizeof(SUserObj), .rowSize = sizeof(SUserObj),
.pMsg = pMsg .pMsg = pMsg
}; };
@ -282,7 +282,7 @@ static int32_t mnodeDropUser(SUserObj *pUser, void *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsUserSdb, .pTable = tsUserSdb,
.pObj = pUser, .pRow = pUser,
.pMsg = pMsg .pMsg = pMsg
}; };
@ -565,7 +565,7 @@ void mnodeDropAllUsers(SAcctObj *pAcct) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsUserSdb, .pTable = tsUserSdb,
.pObj = pUser, .pRow = pUser,
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
numOfUsers++; numOfUsers++;

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

@ -73,12 +73,12 @@ static void mnodeDestroyVgroup(SVgObj *pVgroup) {
} }
static int32_t mnodeVgroupActionDestroy(SSWriteMsg *pWMsg) { static int32_t mnodeVgroupActionDestroy(SSWriteMsg *pWMsg) {
mnodeDestroyVgroup(pWMsg->pObj); mnodeDestroyVgroup(pWMsg->pRow);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mnodeVgroupActionInsert(SSWriteMsg *pWMsg) { static int32_t mnodeVgroupActionInsert(SSWriteMsg *pWMsg) {
SVgObj *pVgroup = pWMsg->pObj; SVgObj *pVgroup = pWMsg->pRow;
// refer to db // refer to db
SDbObj *pDb = mnodeGetDb(pVgroup->dbName); SDbObj *pDb = mnodeGetDb(pVgroup->dbName);
@ -116,7 +116,7 @@ static int32_t mnodeVgroupActionInsert(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeVgroupActionDelete(SSWriteMsg *pWMsg) { static int32_t mnodeVgroupActionDelete(SSWriteMsg *pWMsg) {
SVgObj *pVgroup = pWMsg->pObj; SVgObj *pVgroup = pWMsg->pRow;
if (pVgroup->pDb == NULL) { if (pVgroup->pDb == NULL) {
mError("vgId:%d, db:%s is not exist while insert into hash", pVgroup->vgId, pVgroup->dbName); mError("vgId:%d, db:%s is not exist while insert into hash", pVgroup->vgId, pVgroup->dbName);
@ -138,7 +138,7 @@ static int32_t mnodeVgroupActionDelete(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeVgroupActionUpdate(SSWriteMsg *pWMsg) { static int32_t mnodeVgroupActionUpdate(SSWriteMsg *pWMsg) {
SVgObj *pNew = pWMsg->pObj; SVgObj *pNew = pWMsg->pRow;
SVgObj *pVgroup = mnodeGetVgroup(pNew->vgId); SVgObj *pVgroup = mnodeGetVgroup(pNew->vgId);
if (pVgroup != pNew) { if (pVgroup != pNew) {
@ -177,7 +177,7 @@ static int32_t mnodeVgroupActionUpdate(SSWriteMsg *pWMsg) {
} }
static int32_t mnodeVgroupActionEncode(SSWriteMsg *pWMsg) { static int32_t mnodeVgroupActionEncode(SSWriteMsg *pWMsg) {
SVgObj *pVgroup = pWMsg->pObj; SVgObj *pVgroup = pWMsg->pRow;
memcpy(pWMsg->rowData, pVgroup, tsVgUpdateSize); memcpy(pWMsg->rowData, pVgroup, tsVgUpdateSize);
SVgObj *pTmpVgroup = pWMsg->rowData; SVgObj *pTmpVgroup = pWMsg->rowData;
for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) { for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
@ -194,7 +194,7 @@ static int32_t mnodeVgroupActionDecode(SSWriteMsg *pWMsg) {
if (pVgroup == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY; if (pVgroup == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
memcpy(pVgroup, pWMsg->rowData, tsVgUpdateSize); memcpy(pVgroup, pWMsg->rowData, tsVgUpdateSize);
pWMsg->pObj = pVgroup; pWMsg->pRow = pVgroup;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -219,7 +219,7 @@ int32_t mnodeInitVgroups() {
.fpEncode = mnodeVgroupActionEncode, .fpEncode = mnodeVgroupActionEncode,
.fpDecode = mnodeVgroupActionDecode, .fpDecode = mnodeVgroupActionDecode,
.fpDestroy = mnodeVgroupActionDestroy, .fpDestroy = mnodeVgroupActionDestroy,
.fpDestored = mnodeVgroupActionRestored, .fpRestored = mnodeVgroupActionRestored,
}; };
tsVgroupSdb = sdbOpenTable(&tableDesc); tsVgroupSdb = sdbOpenTable(&tableDesc);
@ -256,7 +256,7 @@ void mnodeUpdateVgroup(SVgObj *pVgroup) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup .pRow = pVgroup
}; };
int32_t code = sdbUpdateRow(&wmsg); int32_t code = sdbUpdateRow(&wmsg);
@ -519,14 +519,14 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
mError("app:%p:%p, vgId:%d, failed to create in sdb, reason:%s", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId, mError("app:%p:%p, vgId:%d, failed to create in sdb, reason:%s", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId,
tstrerror(code)); tstrerror(code));
SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .pTable = tsVgroupSdb}; SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pRow = pVgroup, .pTable = tsVgroupSdb};
sdbDeleteRow(&desc); sdbDeleteRow(&desc);
return code; return code;
} else { } else {
mInfo("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId, mInfo("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId,
pDb->name, pVgroup->numOfVnodes); pDb->name, pVgroup->numOfVnodes);
pVgroup->status = TAOS_VG_STATUS_READY; pVgroup->status = TAOS_VG_STATUS_READY;
SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .pTable = tsVgroupSdb}; SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pRow = pVgroup, .pTable = tsVgroupSdb};
(void)sdbUpdateRow(&desc); (void)sdbUpdateRow(&desc);
dnodeReprocessMWriteMsg(pMsg); dnodeReprocessMWriteMsg(pMsg);
@ -535,7 +535,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
// mInfo("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId, // mInfo("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId,
// pDb->name, pVgroup->numOfVnodes); // pDb->name, pVgroup->numOfVnodes);
// pVgroup->status = TAOS_VG_STATUS_READY; // pVgroup->status = TAOS_VG_STATUS_READY;
// SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .pTable = tsVgroupSdb}; // SSWriteMsg desc = {.type = SDB_OPER_GLOBAL, .pRow = pVgroup, .pTable = tsVgroupSdb};
// (void)sdbUpdateRow(&desc); // (void)sdbUpdateRow(&desc);
// dnodeReprocessMWriteMsg(pMsg); // dnodeReprocessMWriteMsg(pMsg);
// return TSDB_CODE_MND_ACTION_IN_PROGRESS; // return TSDB_CODE_MND_ACTION_IN_PROGRESS;
@ -574,7 +574,7 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup, .pRow = pVgroup,
.rowSize = sizeof(SVgObj), .rowSize = sizeof(SVgObj),
.pMsg = pMsg, .pMsg = pMsg,
.fpReq = mnodeCreateVgroupFp .fpReq = mnodeCreateVgroupFp
@ -598,7 +598,7 @@ void mnodeDropVgroup(SVgObj *pVgroup, void *ahandle) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup .pRow = pVgroup
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
} }
@ -770,7 +770,7 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
SDnodeObj * pDnode = pVgroup->vnodeGid[i].pDnode; SDnodeObj * pDnode = pVgroup->vnodeGid[i].pDnode;
const char *role = "NULL"; const char *role = "NULL";
if (pDnode != NULL) { if (pDnode != NULL) {
role = mnodeGetMnodeRoleStr(pVgroup->vnodeGid[i].role); role = syncRole[pVgroup->vnodeGid[i].role];
} }
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
@ -960,7 +960,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup, .pRow = pVgroup,
.rowSize = sizeof(SVgObj), .rowSize = sizeof(SVgObj),
.pMsg = mnodeMsg, .pMsg = mnodeMsg,
.fpWrite = mnodeCreateVgroupCb .fpWrite = mnodeCreateVgroupCb
@ -976,7 +976,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup .pRow = pVgroup
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
dnodeSendRpcMWriteRsp(mnodeMsg, mnodeMsg->code); dnodeSendRpcMWriteRsp(mnodeMsg, mnodeMsg->code);
@ -1034,7 +1034,7 @@ static void mnodeProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup .pRow = pVgroup
}; };
int32_t code = sdbDeleteRow(&wmsg); int32_t code = sdbDeleteRow(&wmsg);
if (code != 0) { if (code != 0) {
@ -1087,7 +1087,7 @@ void mnodeDropAllDnodeVgroups(SDnodeObj *pDropDnode) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup, .pRow = pVgroup,
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
numOfVgroups++; numOfVgroups++;
@ -1138,7 +1138,7 @@ void mnodeDropAllDbVgroups(SDbObj *pDropDb) {
SSWriteMsg wmsg = { SSWriteMsg wmsg = {
.type = SDB_OPER_LOCAL, .type = SDB_OPER_LOCAL,
.pTable = tsVgroupSdb, .pTable = tsVgroupSdb,
.pObj = pVgroup, .pRow = pVgroup,
}; };
sdbDeleteRow(&wmsg); sdbDeleteRow(&wmsg);
numOfVgroups++; numOfVgroups++;