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

rename msg

This commit is contained in:
Shengliang Guan 2022-03-15 20:53:29 +08:00
parent bde9b43933
commit 332da70335
31 changed files with 383 additions and 394 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
include/dnode/mnode/mnode.h
View File

@ -31,18 +31,6 @@ typedef int32_t (*PutReqToMWriteQFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef int32_t (*PutReqToMReadQFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef struct {
char user[TSDB_USER_LEN];
char db[TSDB_DB_FNAME_LEN];
int32_t acctId;
SMnode* pMnode;
int64_t createdTime;
SRpcMsg rpcMsg;
int32_t contLen;
void* pCont;
} SMndMsg;
typedef struct {
int32_t dnodeId;
int64_t clusterId;
@ -121,7 +109,7 @@ int32_t mndRetriveAuth(SMnode *pMnode, char *user, char *spi, char *encrypt, cha
* @param pMsg The request msg.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t mndProcessMsg(SMndMsg *pMsg);
int32_t mndProcessMsg(SNodeMsg *pMsg);
#ifdef __cplusplus
}

2
source/dnode/mgmt/bnode/inc/bmWorker.h
View File

@ -28,7 +28,7 @@ void bmInitMsgFp(SMnodeMgmt *pMgmt);
void bmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
int32_t bmPutMsgToWriteQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t bmPutMsgToReadQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
void bmConsumeChildQueue(SDnode *pDnode, SMndMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void bmConsumeChildQueue(SDnode *pDnode, SNodeMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void bmConsumeParentQueue(SDnode *pDnode, SRpcMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);

2
source/dnode/mgmt/mnode/src/mmWorker.c
View File

@ -29,7 +29,7 @@ static void mmProcessQueue(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
if (pMnode != NULL) {
pMsg->pNode = pMnode;
code = mndProcessMsg((SMndMsg *)pMsg);
code = mndProcessMsg((SNodeMsg *)pMsg);
mmRelease(pMgmt, pMnode);
}

2
source/dnode/mgmt/qnode/inc/qmWorker.h
View File

@ -28,7 +28,7 @@ void qmInitMsgFp(SMnodeMgmt *pMgmt);
void qmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
int32_t qmPutMsgToWriteQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t qmPutMsgToReadQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
void qmConsumeChildQueue(SDnode *pDnode, SMndMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void qmConsumeChildQueue(SDnode *pDnode, SNodeMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void qmConsumeParentQueue(SDnode *pDnode, SRpcMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void qmProcessWriteMsg(SDnode *pDnode, SMgmtWrapper *pWrapper, SNodeMsg *pMsg);

2
source/dnode/mgmt/snode/inc/smWorker.h
View File

@ -28,7 +28,7 @@ void smInitMsgFp(SMnodeMgmt *pMgmt);
void smProcessRpcMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
int32_t smPutMsgToWriteQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t smPutMsgToReadQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
void smConsumeChildQueue(SDnode *pDnode, SMndMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void smConsumeChildQueue(SDnode *pDnode, SNodeMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void smConsumeParentQueue(SDnode *pDnode, SRpcMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void smProcessWriteMsg(SDnode *pDnode, SMgmtWrapper *pWrapper, SNodeMsg *pMsg);

2
source/dnode/mgmt/vnode/inc/vmWorker.h
View File

@ -28,7 +28,7 @@ void vmInitMsgFp(SMnodeMgmt *pMgmt);
void vmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
int32_t vmPutMsgToWriteQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t vmPutMsgToReadQueue(SDnode *pDnode, SRpcMsg *pRpcMsg);
void vmConsumeChildQueue(SDnode *pDnode, SMndMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void vmConsumeChildQueue(SDnode *pDnode, SNodeMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
void vmConsumeParentQueue(SDnode *pDnode, SRpcMsg *pMsg, int32_t msgLen, void *pCont, int32_t contLen);
int32_t vmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);

6
source/dnode/mnode/impl/inc/mndInt.h
View File

@ -38,11 +38,11 @@ extern "C" {
#define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }}
#define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }}
typedef int32_t (*MndMsgFp)(SMndMsg *pMsg);
typedef int32_t (*MndMsgFp)(SNodeMsg *pMsg);
typedef int32_t (*MndInitFp)(SMnode *pMnode);
typedef void (*MndCleanupFp)(SMnode *pMnode);
typedef int32_t (*ShowMetaFp)(SMndMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
typedef int32_t (*ShowRetrieveFp)(SMndMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
typedef int32_t (*ShowMetaFp)(SNodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
typedef int32_t (*ShowRetrieveFp)(SNodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
typedef void (*ShowFreeIterFp)(SMnode *pMnode, void *pIter);
typedef struct SMnodeLoad {

2
source/dnode/mnode/impl/inc/mndTrans.h
View File

@ -47,7 +47,7 @@ void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen);
void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb);
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
void mndTransProcessRsp(SMndMsg *pRsp);
void mndTransProcessRsp(SNodeMsg *pRsp);
void mndTransPullup(SMnode *pMnode);
#ifdef __cplusplus

12
source/dnode/mnode/impl/src/mndAcct.c
View File

@ -26,9 +26,9 @@ static SSdbRow *mndAcctActionDecode(SSdbRaw *pRaw);
static int32_t mndAcctActionInsert(SSdb *pSdb, SAcctObj *pAcct);
static int32_t mndAcctActionDelete(SSdb *pSdb, SAcctObj *pAcct);
static int32_t mndAcctActionUpdate(SSdb *pSdb, SAcctObj *pOld, SAcctObj *pNew);
static int32_t mndProcessCreateAcctReq(SMndMsg *pReq);
static int32_t mndProcessAlterAcctReq(SMndMsg *pReq);
static int32_t mndProcessDropAcctReq(SMndMsg *pReq);
static int32_t mndProcessCreateAcctReq(SNodeMsg *pReq);
static int32_t mndProcessAlterAcctReq(SNodeMsg *pReq);
static int32_t mndProcessDropAcctReq(SNodeMsg *pReq);
int32_t mndInitAcct(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_ACCT,
@ -185,19 +185,19 @@ static int32_t mndAcctActionUpdate(SSdb *pSdb, SAcctObj *pOld, SAcctObj *pNew) {
return 0;
}
static int32_t mndProcessCreateAcctReq(SMndMsg *pReq) {
static int32_t mndProcessCreateAcctReq(SNodeMsg *pReq) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process create acct request since %s", terrstr());
return -1;
}
static int32_t mndProcessAlterAcctReq(SMndMsg *pReq) {
static int32_t mndProcessAlterAcctReq(SNodeMsg *pReq) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process create acct request since %s", terrstr());
return -1;
}
static int32_t mndProcessDropAcctReq(SMndMsg *pReq) {
static int32_t mndProcessDropAcctReq(SNodeMsg *pReq) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process create acct request since %s", terrstr());
return -1;

10
source/dnode/mnode/impl/src/mndAuth.c
View File

@ -17,7 +17,7 @@
#include "mndAuth.h"
#include "mndUser.h"
static int32_t mndProcessAuthReq(SMndMsg *pReq);
static int32_t mndProcessAuthReq(SNodeMsg *pReq);
int32_t mndInitAuth(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_MND_AUTH, mndProcessAuthReq);
@ -45,7 +45,7 @@ int32_t mndRetriveAuth(SMnode *pMnode, char *user, char *spi, char *encrypt, cha
return 0;
}
static int32_t mndProcessAuthReq(SMndMsg *pReq) {
static int32_t mndProcessAuthReq(SNodeMsg *pReq) {
SAuthReq authReq = {0};
if (tDeserializeSAuthReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &authReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
@ -56,15 +56,15 @@ static int32_t mndProcessAuthReq(SMndMsg *pReq) {
memcpy(authRsp.user, authReq.user, TSDB_USER_LEN);
int32_t code =
mndRetriveAuth(pReq->pMnode, authRsp.user, &authRsp.spi, &authRsp.encrypt, authRsp.secret, authRsp.ckey);
mndRetriveAuth(pReq->pNode, authRsp.user, &authRsp.spi, &authRsp.encrypt, authRsp.secret, authRsp.ckey);
mTrace("user:%s, auth req received, spi:%d encrypt:%d ruser:%s", pReq->user, authRsp.spi, authRsp.encrypt,
authRsp.user);
int32_t contLen = tSerializeSAuthReq(NULL, 0, &authRsp);
void *pRsp = rpcMallocCont(contLen);
tSerializeSAuthReq(pRsp, contLen, &authRsp);
pReq->pCont = pRsp;
pReq->contLen = contLen;
pReq->pRsp = pRsp;
pReq->rspLen = contLen;
return code;
}

36
source/dnode/mnode/impl/src/mndBnode.c
View File

@ -29,12 +29,12 @@ static SSdbRow *mndBnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndBnodeActionInsert(SSdb *pSdb, SBnodeObj *pObj);
static int32_t mndBnodeActionDelete(SSdb *pSdb, SBnodeObj *pObj);
static int32_t mndBnodeActionUpdate(SSdb *pSdb, SBnodeObj *pOld, SBnodeObj *pNew);
static int32_t mndProcessCreateBnodeReq(SMndMsg *pReq);
static int32_t mndProcessDropBnodeReq(SMndMsg *pReq);
static int32_t mndProcessCreateBnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessDropBnodeRsp(SMndMsg *pRsp);
static int32_t mndGetBnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveBnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessCreateBnodeReq(SNodeMsg *pReq);
static int32_t mndProcessDropBnodeReq(SNodeMsg *pReq);
static int32_t mndProcessCreateBnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessDropBnodeRsp(SNodeMsg *pRsp);
static int32_t mndGetBnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveBnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextBnode(SMnode *pMnode, void *pIter);
int32_t mndInitBnode(SMnode *pMnode) {
@ -240,7 +240,7 @@ static int32_t mndSetCreateBnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, S
return 0;
}
static int32_t mndCreateBnode(SMnode *pMnode, SMndMsg *pReq, SDnodeObj *pDnode, SMCreateBnodeReq *pCreate) {
static int32_t mndCreateBnode(SMnode *pMnode, SNodeMsg *pReq, SDnodeObj *pDnode, SMCreateBnodeReq *pCreate) {
int32_t code = -1;
SBnodeObj bnodeObj = {0};
@ -266,8 +266,8 @@ CREATE_BNODE_OVER:
return code;
}
static int32_t mndProcessCreateBnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateBnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SBnodeObj *pObj = NULL;
SDnodeObj *pDnode = NULL;
@ -363,7 +363,7 @@ static int32_t mndSetDropBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SBn
return 0;
}
static int32_t mndDropBnode(SMnode *pMnode, SMndMsg *pReq, SBnodeObj *pObj) {
static int32_t mndDropBnode(SMnode *pMnode, SNodeMsg *pReq, SBnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_BNODE, &pReq->rpcMsg);
@ -382,8 +382,8 @@ DROP_BNODE_OVER:
return code;
}
static int32_t mndProcessDropBnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropBnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SBnodeObj *pObj = NULL;
@ -430,18 +430,18 @@ DROP_BNODE_OVER:
return code;
}
static int32_t mndProcessCreateBnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessCreateBnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessDropBnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessDropBnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndGetBnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetBnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -480,8 +480,8 @@ static int32_t mndGetBnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveBnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveBnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
int32_t cols = 0;

10
source/dnode/mnode/impl/src/mndCluster.c
View File

@ -26,8 +26,8 @@ static int32_t mndClusterActionInsert(SSdb *pSdb, SClusterObj *pCluster);
static int32_t mndClusterActionDelete(SSdb *pSdb, SClusterObj *pCluster);
static int32_t mndClusterActionUpdate(SSdb *pSdb, SClusterObj *pOldCluster, SClusterObj *pNewCluster);
static int32_t mndCreateDefaultCluster(SMnode *pMnode);
static int32_t mndGetClusterMeta(SMndMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveClusters(SMndMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetClusterMeta(SNodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveClusters(SNodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextCluster(SMnode *pMnode, void *pIter);
int32_t mndInitCluster(SMnode *pMnode) {
@ -163,7 +163,7 @@ static int32_t mndCreateDefaultCluster(SMnode *pMnode) {
return sdbWrite(pMnode->pSdb, pRaw);
}
static int32_t mndGetClusterMeta(SMndMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta) {
static int32_t mndGetClusterMeta(SNodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta) {
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
@ -201,8 +201,8 @@ static int32_t mndGetClusterMeta(SMndMsg *pMsg, SShowObj *pShow, STableMetaRsp *
return 0;
}
static int32_t mndRetrieveClusters(SMndMsg *pMsg, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndRetrieveClusters(SNodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pMsg->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
int32_t cols = 0;

6
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -34,9 +34,9 @@
static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer);
static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer);
static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pConsumer, SMqConsumerObj *pNewConsumer);
static int32_t mndProcessConsumerMetaMsg(SMndMsg *pMsg);
static int32_t mndGetConsumerMeta(SMndMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConsumer(SMndMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessConsumerMetaMsg(SNodeMsg *pMsg);
static int32_t mndGetConsumerMeta(SNodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConsumer(SNodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter);
int32_t mndInitConsumer(SMnode *pMnode) {

58
source/dnode/mnode/impl/src/mndDb.c
View File

@ -31,14 +31,14 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw);
static int32_t mndDbActionInsert(SSdb *pSdb, SDbObj *pDb);
static int32_t mndDbActionDelete(SSdb *pSdb, SDbObj *pDb);
static int32_t mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew);
static int32_t mndProcessCreateDbReq(SMndMsg *pReq);
static int32_t mndProcessAlterDbReq(SMndMsg *pReq);
static int32_t mndProcessDropDbReq(SMndMsg *pReq);
static int32_t mndProcessUseDbReq(SMndMsg *pReq);
static int32_t mndProcessSyncDbReq(SMndMsg *pReq);
static int32_t mndProcessCompactDbReq(SMndMsg *pReq);
static int32_t mndGetDbMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveDbs(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessCreateDbReq(SNodeMsg *pReq);
static int32_t mndProcessAlterDbReq(SNodeMsg *pReq);
static int32_t mndProcessDropDbReq(SNodeMsg *pReq);
static int32_t mndProcessUseDbReq(SNodeMsg *pReq);
static int32_t mndProcessSyncDbReq(SNodeMsg *pReq);
static int32_t mndProcessCompactDbReq(SNodeMsg *pReq);
static int32_t mndGetDbMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveDbs(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextDb(SMnode *pMnode, void *pIter);
int32_t mndInitDb(SMnode *pMnode) {
@ -384,7 +384,7 @@ static int32_t mndSetCreateDbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
return 0;
}
static int32_t mndCreateDb(SMnode *pMnode, SMndMsg *pReq, SCreateDbReq *pCreate, SUserObj *pUser) {
static int32_t mndCreateDb(SMnode *pMnode, SNodeMsg *pReq, SCreateDbReq *pCreate, SUserObj *pUser) {
SDbObj dbObj = {0};
memcpy(dbObj.name, pCreate->db, TSDB_DB_FNAME_LEN);
memcpy(dbObj.acct, pUser->acct, TSDB_USER_LEN);
@ -458,8 +458,8 @@ CREATE_DB_OVER:
return code;
}
static int32_t mndProcessCreateDbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateDbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SDbObj *pDb = NULL;
SUserObj *pUser = NULL;
@ -622,7 +622,7 @@ static int32_t mndSetUpdateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
return 0;
}
static int32_t mndUpdateDb(SMnode *pMnode, SMndMsg *pReq, SDbObj *pOld, SDbObj *pNew) {
static int32_t mndUpdateDb(SMnode *pMnode, SNodeMsg *pReq, SDbObj *pOld, SDbObj *pNew) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_ALTER_DB, &pReq->rpcMsg);
if (pTrans == NULL) goto UPDATE_DB_OVER;
@ -642,8 +642,8 @@ UPDATE_DB_OVER:
return code;
}
static int32_t mndProcessAlterDbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessAlterDbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SDbObj *pDb = NULL;
SUserObj *pUser = NULL;
@ -802,7 +802,7 @@ static int32_t mndSetDropDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *p
return 0;
}
static int32_t mndDropDb(SMnode *pMnode, SMndMsg *pReq, SDbObj *pDb) {
static int32_t mndDropDb(SMnode *pMnode, SNodeMsg *pReq, SDbObj *pDb) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_DB, &pReq->rpcMsg);
if (pTrans == NULL) goto DROP_DB_OVER;
@ -837,8 +837,8 @@ DROP_DB_OVER:
return code;
}
static int32_t mndProcessDropDbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropDbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SDbObj *pDb = NULL;
SUserObj *pUser = NULL;
@ -948,8 +948,8 @@ static void mndBuildDBVgroupInfo(SDbObj *pDb, SMnode *pMnode, SArray *pVgList) {
sdbCancelFetch(pSdb, pIter);
}
static int32_t mndProcessUseDbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessUseDbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SDbObj *pDb = NULL;
SUserObj *pUser = NULL;
@ -1017,8 +1017,8 @@ static int32_t mndProcessUseDbReq(SMndMsg *pReq) {
tSerializeSUseDbRsp(pRsp, contLen, &usedbRsp);
pReq->pCont = pRsp;
pReq->contLen = contLen;
pReq->pRsp = pRsp;
pReq->rspLen = contLen;
USE_DB_OVER:
if (code != 0) {
@ -1101,8 +1101,8 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbVgVersion *pDbs, int32_t numOfDbs,
return 0;
}
static int32_t mndProcessSyncDbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessSyncDbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SDbObj *pDb = NULL;
SUserObj *pUser = NULL;
@ -1142,8 +1142,8 @@ SYNC_DB_OVER:
return code;
}
static int32_t mndProcessCompactDbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCompactDbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SDbObj *pDb = NULL;
SUserObj *pUser = NULL;
@ -1183,8 +1183,8 @@ SYNC_DB_OVER:
return code;
}
static int32_t mndGetDbMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetDbMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -1324,8 +1324,8 @@ char *mnGetDbStr(char *src) {
return pos;
}
static int32_t mndRetrieveDbs(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveDbs(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SDbObj *pDb = NULL;

58
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -49,17 +49,17 @@ static int32_t mndDnodeActionInsert(SSdb *pSdb, SDnodeObj *pDnode);
static int32_t mndDnodeActionDelete(SSdb *pSdb, SDnodeObj *pDnode);
static int32_t mndDnodeActionUpdate(SSdb *pSdb, SDnodeObj *pOld, SDnodeObj *pNew);
static int32_t mndProcessCreateDnodeReq(SMndMsg *pReq);
static int32_t mndProcessDropDnodeReq(SMndMsg *pReq);
static int32_t mndProcessConfigDnodeReq(SMndMsg *pReq);
static int32_t mndProcessConfigDnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessStatusReq(SMndMsg *pReq);
static int32_t mndProcessCreateDnodeReq(SNodeMsg *pReq);
static int32_t mndProcessDropDnodeReq(SNodeMsg *pReq);
static int32_t mndProcessConfigDnodeReq(SNodeMsg *pReq);
static int32_t mndProcessConfigDnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessStatusReq(SNodeMsg *pReq);
static int32_t mndGetConfigMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConfigs(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetConfigMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConfigs(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextConfig(SMnode *pMnode, void *pIter);
static int32_t mndGetDnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveDnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetDnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveDnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextDnode(SMnode *pMnode, void *pIter);
int32_t mndInitDnode(SMnode *pMnode) {
@ -296,8 +296,8 @@ static int32_t mndCheckClusterCfgPara(SMnode *pMnode, const SClusterCfg *pCfg) {
return 0;
}
static int32_t mndProcessStatusReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessStatusReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SStatusReq statusReq = {0};
SDnodeObj *pDnode = NULL;
int32_t code = -1;
@ -424,8 +424,8 @@ static int32_t mndProcessStatusReq(SMndMsg *pReq) {
tSerializeSStatusRsp(pHead, contLen, &statusRsp);
taosArrayDestroy(statusRsp.pDnodeEps);
pReq->contLen = contLen;
pReq->pCont = pHead;
pReq->rspLen = contLen;
pReq->pRsp = pHead;
}
pDnode->lastAccessTime = curMs;
@ -437,7 +437,7 @@ PROCESS_STATUS_MSG_OVER:
return code;
}
static int32_t mndCreateDnode(SMnode *pMnode, SMndMsg *pReq, SCreateDnodeReq *pCreate) {
static int32_t mndCreateDnode(SMnode *pMnode, SNodeMsg *pReq, SCreateDnodeReq *pCreate) {
SDnodeObj dnodeObj = {0};
dnodeObj.id = sdbGetMaxId(pMnode->pSdb, SDB_DNODE);
dnodeObj.createdTime = taosGetTimestampMs();
@ -471,8 +471,8 @@ static int32_t mndCreateDnode(SMnode *pMnode, SMndMsg *pReq, SCreateDnodeReq *pC
return 0;
}
static int32_t mndProcessCreateDnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateDnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SDnodeObj *pDnode = NULL;
@ -521,7 +521,7 @@ CREATE_DNODE_OVER:
return code;
}
static int32_t mndDropDnode(SMnode *pMnode, SMndMsg *pReq, SDnodeObj *pDnode) {
static int32_t mndDropDnode(SMnode *pMnode, SNodeMsg *pReq, SDnodeObj *pDnode) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_DNODE, &pReq->rpcMsg);
if (pTrans == NULL) {
mError("dnode:%d, failed to drop since %s", pDnode->id, terrstr());
@ -547,8 +547,8 @@ static int32_t mndDropDnode(SMnode *pMnode, SMndMsg *pReq, SDnodeObj *pDnode) {
return 0;
}
static int32_t mndProcessDropDnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropDnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SDnodeObj *pDnode = NULL;
@ -596,8 +596,8 @@ DROP_DNODE_OVER:
return code;
}
static int32_t mndProcessConfigDnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessConfigDnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SMCfgDnodeReq cfgReq = {0};
if (tDeserializeSMCfgDnodeReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &cfgReq) != 0) {
@ -628,11 +628,11 @@ static int32_t mndProcessConfigDnodeReq(SMndMsg *pReq) {
return 0;
}
static int32_t mndProcessConfigDnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessConfigDnodeRsp(SNodeMsg *pRsp) {
mInfo("app:%p config rsp from dnode", pRsp->rpcMsg.ahandle);
}
static int32_t mndGetConfigMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
static int32_t mndGetConfigMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
@ -663,8 +663,8 @@ static int32_t mndGetConfigMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *p
return 0;
}
static int32_t mndRetrieveConfigs(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveConfigs(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
int32_t totalRows = 0;
int32_t numOfRows = 0;
char *cfgOpts[TSDB_CONFIG_NUMBER] = {0};
@ -709,8 +709,8 @@ static int32_t mndRetrieveConfigs(SMndMsg *pReq, SShowObj *pShow, char *data, in
static void mndCancelGetNextConfig(SMnode *pMnode, void *pIter) {}
static int32_t mndGetDnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetDnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -773,8 +773,8 @@ static int32_t mndGetDnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveDnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveDnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
int32_t cols = 0;

42
source/dnode/mnode/impl/src/mndFunc.c
View File

@ -29,13 +29,13 @@ static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw);
static int32_t mndFuncActionInsert(SSdb *pSdb, SFuncObj *pFunc);
static int32_t mndFuncActionDelete(SSdb *pSdb, SFuncObj *pFunc);
static int32_t mndFuncActionUpdate(SSdb *pSdb, SFuncObj *pOld, SFuncObj *pNew);
static int32_t mndCreateFunc(SMnode *pMnode, SMndMsg *pReq, SCreateFuncReq *pCreate);
static int32_t mndDropFunc(SMnode *pMnode, SMndMsg *pReq, SFuncObj *pFunc);
static int32_t mndProcessCreateFuncReq(SMndMsg *pReq);
static int32_t mndProcessDropFuncReq(SMndMsg *pReq);
static int32_t mndProcessRetrieveFuncReq(SMndMsg *pReq);
static int32_t mndGetFuncMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveFuncs(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndCreateFunc(SMnode *pMnode, SNodeMsg *pReq, SCreateFuncReq *pCreate);
static int32_t mndDropFunc(SMnode *pMnode, SNodeMsg *pReq, SFuncObj *pFunc);
static int32_t mndProcessCreateFuncReq(SNodeMsg *pReq);
static int32_t mndProcessDropFuncReq(SNodeMsg *pReq);
static int32_t mndProcessRetrieveFuncReq(SNodeMsg *pReq);
static int32_t mndGetFuncMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveFuncs(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextFunc(SMnode *pMnode, void *pIter);
int32_t mndInitFunc(SMnode *pMnode) {
@ -181,7 +181,7 @@ static void mndReleaseFunc(SMnode *pMnode, SFuncObj *pFunc) {
sdbRelease(pSdb, pFunc);
}
static int32_t mndCreateFunc(SMnode *pMnode, SMndMsg *pReq, SCreateFuncReq *pCreate) {
static int32_t mndCreateFunc(SMnode *pMnode, SNodeMsg *pReq, SCreateFuncReq *pCreate) {
int32_t code = -1;
STrans *pTrans = NULL;
@ -234,7 +234,7 @@ CREATE_FUNC_OVER:
return code;
}
static int32_t mndDropFunc(SMnode *pMnode, SMndMsg *pReq, SFuncObj *pFunc) {
static int32_t mndDropFunc(SMnode *pMnode, SNodeMsg *pReq, SFuncObj *pFunc) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_FUNC, &pReq->rpcMsg);
if (pTrans == NULL) goto DROP_FUNC_OVER;
@ -262,8 +262,8 @@ DROP_FUNC_OVER:
return code;
}
static int32_t mndProcessCreateFuncReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateFuncReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SFuncObj *pFunc = NULL;
@ -339,8 +339,8 @@ CREATE_FUNC_OVER:
return code;
}
static int32_t mndProcessDropFuncReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropFuncReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SFuncObj *pFunc = NULL;
@ -394,8 +394,8 @@ DROP_FUNC_OVER:
return code;
}
static int32_t mndProcessRetrieveFuncReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessRetrieveFuncReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SRetrieveFuncReq retrieveReq = {0};
SRetrieveFuncRsp retrieveRsp = {0};
@ -451,8 +451,8 @@ static int32_t mndProcessRetrieveFuncReq(SMndMsg *pReq) {
tSerializeSRetrieveFuncRsp(pRsp, contLen, &retrieveRsp);
pReq->pCont = pRsp;
pReq->contLen = contLen;
pReq->pRsp = pRsp;
pReq->rspLen = contLen;
code = 0;
@ -463,8 +463,8 @@ RETRIEVE_FUNC_OVER:
return code;
}
static int32_t mndGetFuncMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetFuncMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -545,8 +545,8 @@ static void *mnodeGenTypeStr(char *buf, int32_t buflen, uint8_t type, int16_t le
return tDataTypes[type].name;
}
static int32_t mndRetrieveFuncs(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveFuncs(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SFuncObj *pFunc = NULL;

40
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -30,13 +30,13 @@ static SSdbRow *mndMnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj);
static int32_t mndMnodeActionDelete(SSdb *pSdb, SMnodeObj *pObj);
static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOld, SMnodeObj *pNew);
static int32_t mndProcessCreateMnodeReq(SMndMsg *pReq);
static int32_t mndProcessDropMnodeReq(SMndMsg *pReq);
static int32_t mndProcessCreateMnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessAlterMnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessDropMnodeRsp(SMndMsg *pRsp);
static int32_t mndGetMnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveMnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessCreateMnodeReq(SNodeMsg *pReq);
static int32_t mndProcessDropMnodeReq(SNodeMsg *pReq);
static int32_t mndProcessCreateMnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessAlterMnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessDropMnodeRsp(SNodeMsg *pRsp);
static int32_t mndGetMnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveMnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextMnode(SMnode *pMnode, void *pIter);
int32_t mndInitMnode(SMnode *pMnode) {
@ -355,7 +355,7 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
return 0;
}
static int32_t mndCreateMnode(SMnode *pMnode, SMndMsg *pReq, SDnodeObj *pDnode, SMCreateMnodeReq *pCreate) {
static int32_t mndCreateMnode(SMnode *pMnode, SNodeMsg *pReq, SDnodeObj *pDnode, SMCreateMnodeReq *pCreate) {
int32_t code = -1;
SMnodeObj mnodeObj = {0};
@ -380,8 +380,8 @@ CREATE_MNODE_OVER:
return code;
}
static int32_t mndProcessCreateMnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateMnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SMnodeObj *pObj = NULL;
SDnodeObj *pDnode = NULL;
@ -527,7 +527,7 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
return 0;
}
static int32_t mndDropMnode(SMnode *pMnode, SMndMsg *pReq, SMnodeObj *pObj) {
static int32_t mndDropMnode(SMnode *pMnode, SNodeMsg *pReq, SMnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_MNODE, &pReq->rpcMsg);
@ -547,8 +547,8 @@ DROP_MNODE_OVER:
return code;
}
static int32_t mndProcessDropMnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropMnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SMnodeObj *pObj = NULL;
@ -595,23 +595,23 @@ DROP_MNODE_OVER:
return code;
}
static int32_t mndProcessCreateMnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessCreateMnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessAlterMnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessAlterMnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessDropMnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessDropMnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndGetMnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetMnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -663,8 +663,8 @@ static int32_t mndGetMnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveMnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveMnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
int32_t cols = 0;

6
source/dnode/mnode/impl/src/mndOffset.c
View File

@ -32,7 +32,7 @@
static int32_t mndOffsetActionInsert(SSdb *pSdb, SMqOffsetObj *pOffset);
static int32_t mndOffsetActionDelete(SSdb *pSdb, SMqOffsetObj *pOffset);
static int32_t mndOffsetActionUpdate(SSdb *pSdb, SMqOffsetObj *pOffset, SMqOffsetObj *pNewOffset);
static int32_t mndProcessCommitOffsetReq(SMndMsg *pReq);
static int32_t mndProcessCommitOffsetReq(SNodeMsg *pReq);
int32_t mndInitOffset(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_OFFSET,
@ -152,10 +152,10 @@ int32_t mndCreateOffset(STrans *pTrans, const char *cgroup, const char *topicNam
return 0;
}
static int32_t mndProcessCommitOffsetReq(SMndMsg *pMsg) {
static int32_t mndProcessCommitOffsetReq(SNodeMsg *pMsg) {
char key[TSDB_PARTITION_KEY_LEN];
SMnode *pMnode = pMsg->pMnode;
SMnode *pMnode = pMsg->pNode;
char *msgStr = pMsg->rpcMsg.pCont;
SMqCMCommitOffsetReq commitOffsetReq;
SCoder decoder;

65
source/dnode/mnode/impl/src/mndProfile.c
View File

@ -50,14 +50,14 @@ static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId);
static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn);
static void *mndGetNextConn(SMnode *pMnode, SCacheIter *pIter);
static void mndCancelGetNextConn(SMnode *pMnode, void *pIter);
static int32_t mndProcessHeartBeatReq(SMndMsg *pReq);
static int32_t mndProcessConnectReq(SMndMsg *pReq);
static int32_t mndProcessKillQueryReq(SMndMsg *pReq);
static int32_t mndProcessKillConnReq(SMndMsg *pReq);
static int32_t mndGetConnsMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConns(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetQueryMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveQueries(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq);
static int32_t mndProcessConnectReq(SNodeMsg *pReq);
static int32_t mndProcessKillQueryReq(SNodeMsg *pReq);
static int32_t mndProcessKillConnReq(SNodeMsg *pReq);
static int32_t mndGetConnsMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetQueryMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter);
int32_t mndInitProfile(SMnode *pMnode) {
@ -177,8 +177,8 @@ static void mndCancelGetNextConn(SMnode *pMnode, void *pIter) {
}
}
static int32_t mndProcessConnectReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessConnectReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SUserObj *pUser = NULL;
SDbObj *pDb = NULL;
SConnObj *pConn = NULL;
@ -206,8 +206,9 @@ static int32_t mndProcessConnectReq(SMndMsg *pReq) {
}
if (connReq.db[0]) {
snprintf(pReq->db, TSDB_DB_FNAME_LEN, "%d%s%s", pUser->acctId, TS_PATH_DELIMITER, connReq.db);
pDb = mndAcquireDb(pMnode, pReq->db);
char db[TSDB_DB_FNAME_LEN];
snprintf(db, TSDB_DB_FNAME_LEN, "%d%s%s", pUser->acctId, TS_PATH_DELIMITER, connReq.db);
pDb = mndAcquireDb(pMnode, db);
if (pDb == NULL) {
terrno = TSDB_CODE_MND_INVALID_DB;
mError("user:%s, failed to login from %s while use db:%s since %s", pReq->user, ip, connReq.db, terrstr());
@ -237,8 +238,8 @@ static int32_t mndProcessConnectReq(SMndMsg *pReq) {
if (pRsp == NULL) goto CONN_OVER;
tSerializeSConnectRsp(pRsp, contLen, &connectRsp);
pReq->contLen = contLen;
pReq->pCont = pRsp;
pReq->rspLen = contLen;
pReq->pRsp = pRsp;
mDebug("user:%s, login from %s, conn:%d, app:%s", info.user, ip, pConn->id, connReq.app);
@ -338,8 +339,8 @@ static SClientHbRsp *mndMqHbBuildRsp(SMnode *pMnode, SClientHbReq *pReq) {
return NULL;
}
static int32_t mndProcessHeartBeatReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SClientHbBatchReq batchReq = {0};
if (tDeserializeSClientHbBatchReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &batchReq) != 0) {
@ -423,12 +424,12 @@ static int32_t mndProcessHeartBeatReq(SMndMsg *pReq) {
}
taosArrayDestroy(batchRsp.rsps);
pReq->contLen = tlen;
pReq->pCont = buf;
pReq->rspLen = tlen;
pReq->pRsp = buf;
return 0;
#if 0
SMnode *pMnode = pReq->pMnode;
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SHeartBeatReq *pHeartbeat = pReq->rpcMsg.pCont;
@ -495,13 +496,13 @@ static int32_t mndProcessHeartBeatReq(SMndMsg *pReq) {
mndReleaseConn(pMnode, pConn);
pReq->contLen = sizeof(SConnectRsp);
pReq->pCont = pRsp;
pReq->pRsp = pRsp;
return 0;
#endif
}
static int32_t mndProcessKillQueryReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessKillQueryReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pReq->user);
@ -534,8 +535,8 @@ static int32_t mndProcessKillQueryReq(SMndMsg *pReq) {
}
}
static int32_t mndProcessKillConnReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessKillConnReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pReq->user);
@ -566,8 +567,8 @@ static int32_t mndProcessKillConnReq(SMndMsg *pReq) {
}
}
static int32_t mndGetConnsMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetConnsMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pReq->user);
@ -641,8 +642,8 @@ static int32_t mndGetConnsMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveConns(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
int32_t numOfRows = 0;
SConnObj *pConn = NULL;
int32_t cols = 0;
@ -700,8 +701,8 @@ static int32_t mndRetrieveConns(SMndMsg *pReq, SShowObj *pShow, char *data, int3
return numOfRows;
}
static int32_t mndGetQueryMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetQueryMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pReq->user);
@ -815,8 +816,8 @@ static int32_t mndGetQueryMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveQueries(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
int32_t numOfRows = 0;
SConnObj *pConn = NULL;
int32_t cols = 0;

46
source/dnode/mnode/impl/src/mndQnode.c
View File

@ -29,14 +29,14 @@ static SSdbRow *mndQnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndQnodeActionInsert(SSdb *pSdb, SQnodeObj *pObj);
static int32_t mndQnodeActionDelete(SSdb *pSdb, SQnodeObj *pObj);
static int32_t mndQnodeActionUpdate(SSdb *pSdb, SQnodeObj *pOld, SQnodeObj *pNew);
static int32_t mndProcessCreateQnodeReq(SMndMsg *pReq);
static int32_t mndProcessDropQnodeReq(SMndMsg *pReq);
static int32_t mndProcessCreateQnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessDropQnodeRsp(SMndMsg *pRsp);
static int32_t mndGetQnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveQnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessCreateQnodeReq(SNodeMsg *pReq);
static int32_t mndProcessDropQnodeReq(SNodeMsg *pReq);
static int32_t mndProcessCreateQnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessDropQnodeRsp(SNodeMsg *pRsp);
static int32_t mndGetQnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveQnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextQnode(SMnode *pMnode, void *pIter);
static int32_t mndProcessQnodeListReq(SMndMsg *pReq);
static int32_t mndProcessQnodeListReq(SNodeMsg *pReq);
int32_t mndInitQnode(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_QNODE,
@ -242,7 +242,7 @@ static int32_t mndSetCreateQnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, S
return 0;
}
static int32_t mndCreateQnode(SMnode *pMnode, SMndMsg *pReq, SDnodeObj *pDnode, SMCreateQnodeReq *pCreate) {
static int32_t mndCreateQnode(SMnode *pMnode, SNodeMsg *pReq, SDnodeObj *pDnode, SMCreateQnodeReq *pCreate) {
int32_t code = -1;
SQnodeObj qnodeObj = {0};
@ -268,8 +268,8 @@ CREATE_QNODE_OVER:
return code;
}
static int32_t mndProcessCreateQnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateQnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SQnodeObj *pObj = NULL;
SDnodeObj *pDnode = NULL;
@ -365,7 +365,7 @@ static int32_t mndSetDropQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQn
return 0;
}
static int32_t mndDropQnode(SMnode *pMnode, SMndMsg *pReq, SQnodeObj *pObj) {
static int32_t mndDropQnode(SMnode *pMnode, SNodeMsg *pReq, SQnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_QNODE, &pReq->rpcMsg);
@ -384,8 +384,8 @@ DROP_QNODE_OVER:
return code;
}
static int32_t mndProcessDropQnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropQnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SQnodeObj *pObj = NULL;
@ -432,11 +432,11 @@ DROP_QNODE_OVER:
return code;
}
static int32_t mndProcessQnodeListReq(SMndMsg *pReq) {
static int32_t mndProcessQnodeListReq(SNodeMsg *pReq) {
int32_t code = -1;
SQnodeListReq qlistReq = {0};
int32_t numOfRows = 0;
SMnode *pMnode = pReq->pMnode;
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
SQnodeObj *pObj = NULL;
SQnodeListRsp qlistRsp = {0};
@ -482,8 +482,8 @@ static int32_t mndProcessQnodeListReq(SMndMsg *pReq) {
tSerializeSQnodeListRsp(pRsp, rspLen, &qlistRsp);
pReq->contLen = rspLen;
pReq->pCont = pRsp;
pReq->rspLen = rspLen;
pReq->pRsp = pRsp;
code = 0;
QNODE_LIST_OVER:
@ -493,18 +493,18 @@ QNODE_LIST_OVER:
return code;
}
static int32_t mndProcessCreateQnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessCreateQnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessDropQnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessDropQnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndGetQnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetQnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -543,8 +543,8 @@ static int32_t mndGetQnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveQnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveQnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
int32_t cols = 0;

30
source/dnode/mnode/impl/src/mndShow.c
View File

@ -22,10 +22,10 @@ static SShowObj *mndCreateShowObj(SMnode *pMnode, SShowReq *pReq);
static void mndFreeShowObj(SShowObj *pShow);
static SShowObj *mndAcquireShowObj(SMnode *pMnode, int64_t showId);
static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove);
static int32_t mndProcessShowReq(SMndMsg *pReq);
static int32_t mndProcessRetrieveReq(SMndMsg *pReq);
static int32_t mndProcessShowReq(SNodeMsg *pReq);
static int32_t mndProcessRetrieveReq(SNodeMsg *pReq);
static bool mndCheckRetrieveFinished(SShowObj *pShow);
static int32_t mndProcessRetrieveSysTableReq(SMndMsg *pReq);
static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq);
int32_t mndInitShow(SMnode *pMnode) {
SShowMgmt *pMgmt = &pMnode->showMgmt;
@ -117,8 +117,8 @@ static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove) {
taosCacheRelease(pMgmt->cache, (void **)(&pShow), forceRemove);
}
static int32_t mndProcessShowReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessShowReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SShowMgmt *pMgmt = &pMnode->showMgmt;
int32_t code = -1;
SShowReq showReq = {0};
@ -161,8 +161,8 @@ static int32_t mndProcessShowReq(SMndMsg *pReq) {
int32_t bufLen = tSerializeSShowRsp(NULL, 0, &showRsp);
void *pBuf = rpcMallocCont(bufLen);
tSerializeSShowRsp(pBuf, bufLen, &showRsp);
pReq->contLen = bufLen;
pReq->pCont = pBuf;
pReq->rspLen = bufLen;
pReq->pRsp = pBuf;
mndReleaseShowObj(pShow, false);
} else {
mndReleaseShowObj(pShow, true);
@ -178,8 +178,8 @@ SHOW_OVER:
return code;
}
static int32_t mndProcessRetrieveReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessRetrieveReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SShowMgmt *pMgmt = &pMnode->showMgmt;
int32_t rowsToRead = 0;
int32_t size = 0;
@ -248,8 +248,8 @@ static int32_t mndProcessRetrieveReq(SMndMsg *pReq) {
pRsp->numOfRows = htonl(rowsRead);
pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision
pReq->pCont = pRsp;
pReq->contLen = size;
pReq->pRsp = pRsp;
pReq->rspLen = size;
if (rowsRead == 0 || rowsToRead == 0 || (rowsRead == rowsToRead && pShow->numOfRows == pShow->numOfReads)) {
pRsp->completed = 1;
@ -263,8 +263,8 @@ static int32_t mndProcessRetrieveReq(SMndMsg *pReq) {
return TSDB_CODE_SUCCESS;
}
static int32_t mndProcessRetrieveSysTableReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SShowMgmt *pMgmt = &pMnode->showMgmt;
int32_t rowsToRead = 0;
int32_t size = 0;
@ -348,8 +348,8 @@ static int32_t mndProcessRetrieveSysTableReq(SMndMsg *pReq) {
pRsp->numOfRows = htonl(rowsRead);
pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision
pReq->pCont = pRsp;
pReq->contLen = size;
pReq->pRsp = pRsp;
pReq->rspLen = size;
if (rowsRead == 0 || rowsToRead == 0 || (rowsRead == rowsToRead && pShow->numOfRows == pShow->numOfReads)) {
pRsp->completed = 1;

36
source/dnode/mnode/impl/src/mndSnode.c
View File

@ -29,12 +29,12 @@ static SSdbRow *mndSnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndSnodeActionInsert(SSdb *pSdb, SSnodeObj *pObj);
static int32_t mndSnodeActionDelete(SSdb *pSdb, SSnodeObj *pObj);
static int32_t mndSnodeActionUpdate(SSdb *pSdb, SSnodeObj *pOld, SSnodeObj *pNew);
static int32_t mndProcessCreateSnodeReq(SMndMsg *pReq);
static int32_t mndProcessDropSnodeReq(SMndMsg *pReq);
static int32_t mndProcessCreateSnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessDropSnodeRsp(SMndMsg *pRsp);
static int32_t mndGetSnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveSnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessCreateSnodeReq(SNodeMsg *pReq);
static int32_t mndProcessDropSnodeReq(SNodeMsg *pReq);
static int32_t mndProcessCreateSnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessDropSnodeRsp(SNodeMsg *pRsp);
static int32_t mndGetSnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveSnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextSnode(SMnode *pMnode, void *pIter);
int32_t mndInitSnode(SMnode *pMnode) {
@ -240,7 +240,7 @@ static int32_t mndSetCreateSnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, S
return 0;
}
static int32_t mndCreateSnode(SMnode *pMnode, SMndMsg *pReq, SDnodeObj *pDnode, SMCreateSnodeReq *pCreate) {
static int32_t mndCreateSnode(SMnode *pMnode, SNodeMsg *pReq, SDnodeObj *pDnode, SMCreateSnodeReq *pCreate) {
int32_t code = -1;
SSnodeObj snodeObj = {0};
@ -267,8 +267,8 @@ CREATE_SNODE_OVER:
return code;
}
static int32_t mndProcessCreateSnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateSnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SSnodeObj *pObj = NULL;
SDnodeObj *pDnode = NULL;
@ -365,7 +365,7 @@ static int32_t mndSetDropSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SSn
return 0;
}
static int32_t mndDropSnode(SMnode *pMnode, SMndMsg *pReq, SSnodeObj *pObj) {
static int32_t mndDropSnode(SMnode *pMnode, SNodeMsg *pReq, SSnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_SNODE, &pReq->rpcMsg);
@ -385,8 +385,8 @@ DROP_SNODE_OVER:
return code;
}
static int32_t mndProcessDropSnodeReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropSnodeReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SSnodeObj *pObj = NULL;
@ -433,18 +433,18 @@ DROP_SNODE_OVER:
return code;
}
static int32_t mndProcessCreateSnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessCreateSnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessDropSnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessDropSnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndGetSnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetSnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -483,8 +483,8 @@ static int32_t mndGetSnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveSnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveSnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
int32_t cols = 0;

58
source/dnode/mnode/impl/src/mndStb.c
View File

@ -33,15 +33,15 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw);
static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb);
static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb);
static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew);
static int32_t mndProcessMCreateStbReq(SMndMsg *pReq);
static int32_t mndProcessMAlterStbReq(SMndMsg *pReq);
static int32_t mndProcessMDropStbReq(SMndMsg *pReq);
static int32_t mndProcessVCreateStbRsp(SMndMsg *pRsp);
static int32_t mndProcessVAlterStbRsp(SMndMsg *pRsp);
static int32_t mndProcessVDropStbRsp(SMndMsg *pRsp);
static int32_t mndProcessTableMetaReq(SMndMsg *pReq);
static int32_t mndGetStbMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveStb(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessMCreateStbReq(SNodeMsg *pReq);
static int32_t mndProcessMAlterStbReq(SNodeMsg *pReq);
static int32_t mndProcessMDropStbReq(SNodeMsg *pReq);
static int32_t mndProcessVCreateStbRsp(SNodeMsg *pRsp);
static int32_t mndProcessVAlterStbRsp(SNodeMsg *pRsp);
static int32_t mndProcessVDropStbRsp(SNodeMsg *pRsp);
static int32_t mndProcessTableMetaReq(SNodeMsg *pReq);
static int32_t mndGetStbMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveStb(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter);
int32_t mndInitStb(SMnode *pMnode) {
@ -490,7 +490,7 @@ static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
return 0;
}
static int32_t mndCreateStb(SMnode *pMnode, SMndMsg *pReq, SMCreateStbReq *pCreate, SDbObj *pDb) {
static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCreate, SDbObj *pDb) {
SStbObj stbObj = {0};
memcpy(stbObj.name, pCreate->name, TSDB_TABLE_FNAME_LEN);
memcpy(stbObj.db, pDb->name, TSDB_DB_FNAME_LEN);
@ -551,8 +551,8 @@ CREATE_STB_OVER:
return code;
}
static int32_t mndProcessMCreateStbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessMCreateStbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SStbObj *pTopicStb = NULL;
SStbObj *pStb = NULL;
@ -623,7 +623,7 @@ CREATE_STB_OVER:
return code;
}
static int32_t mndProcessVCreateStbRsp(SMndMsg *pRsp) {
static int32_t mndProcessVCreateStbRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
@ -980,7 +980,7 @@ static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
return 0;
}
static int32_t mndAlterStb(SMnode *pMnode, SMndMsg *pReq, const SMAltertbReq *pAlter, SDbObj *pDb, SStbObj *pOld) {
static int32_t mndAlterStb(SMnode *pMnode, SNodeMsg *pReq, const SMAltertbReq *pAlter, SDbObj *pDb, SStbObj *pOld) {
SStbObj stbObj = {0};
taosRLockLatch(&pOld->lock);
memcpy(&stbObj, pOld, sizeof(SStbObj));
@ -1043,8 +1043,8 @@ ALTER_STB_OVER:
return code;
}
static int32_t mndProcessMAlterStbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessMAlterStbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SDbObj *pDb = NULL;
SStbObj *pStb = NULL;
@ -1096,7 +1096,7 @@ ALTER_STB_OVER:
return code;
}
static int32_t mndProcessVAlterStbRsp(SMndMsg *pRsp) {
static int32_t mndProcessVAlterStbRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
@ -1160,7 +1160,7 @@ static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
return 0;
}
static int32_t mndDropStb(SMnode *pMnode, SMndMsg *pReq, SDbObj *pDb, SStbObj *pStb) {
static int32_t mndDropStb(SMnode *pMnode, SNodeMsg *pReq, SDbObj *pDb, SStbObj *pStb) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK,TRN_TYPE_DROP_STB, &pReq->rpcMsg);
if (pTrans == NULL) goto DROP_STB_OVER;
@ -1180,8 +1180,8 @@ DROP_STB_OVER:
return code;
}
static int32_t mndProcessMDropStbReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessMDropStbReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SDbObj *pDb = NULL;
@ -1237,7 +1237,7 @@ DROP_STB_OVER:
return code;
}
static int32_t mndProcessVDropStbRsp(SMndMsg *pRsp) {
static int32_t mndProcessVDropStbRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
@ -1311,8 +1311,8 @@ static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char
return code;
}
static int32_t mndProcessTableMetaReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessTableMetaReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
STableInfoReq infoReq = {0};
STableMetaRsp metaRsp = {0};
@ -1347,8 +1347,8 @@ static int32_t mndProcessTableMetaReq(SMndMsg *pReq) {
}
tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp);
pReq->pCont = pRsp;
pReq->contLen = rspLen;
pReq->pRsp = pRsp;
pReq->rspLen = rspLen;
code = 0;
mDebug("stb:%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName);
@ -1436,8 +1436,8 @@ static int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs
return 0;
}
static int32_t mndGetStbMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetStbMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
if (mndGetNumOfStbs(pMnode, pShow->db, &pShow->numOfRows) != 0) {
@ -1499,8 +1499,8 @@ static void mndExtractTableName(char *tableId, char *name) {
}
}
static int32_t mndRetrieveStb(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveStb(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SStbObj *pStb = NULL;

26
source/dnode/mnode/impl/src/mndStream.c
View File

@ -31,12 +31,12 @@
static int32_t mndStreamActionInsert(SSdb *pSdb, SStreamObj *pStream);
static int32_t mndStreamActionDelete(SSdb *pSdb, SStreamObj *pStream);
static int32_t mndStreamActionUpdate(SSdb *pSdb, SStreamObj *pStream, SStreamObj *pNewStream);
static int32_t mndProcessCreateStreamReq(SMndMsg *pReq);
/*static int32_t mndProcessDropStreamReq(SMndMsg *pReq);*/
/*static int32_t mndProcessDropStreamInRsp(SMndMsg *pRsp);*/
static int32_t mndProcessStreamMetaReq(SMndMsg *pReq);
static int32_t mndGetStreamMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveStream(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessCreateStreamReq(SNodeMsg *pReq);
/*static int32_t mndProcessDropStreamReq(SNodeMsg *pReq);*/
/*static int32_t mndProcessDropStreamInRsp(SNodeMsg *pRsp);*/
static int32_t mndProcessStreamMetaReq(SNodeMsg *pReq);
static int32_t mndGetStreamMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveStream(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextStream(SMnode *pMnode, void *pIter);
int32_t mndInitStream(SMnode *pMnode) {
@ -208,7 +208,7 @@ static int32_t mndCheckCreateStreamReq(SCMCreateStreamReq *pCreate) {
return 0;
}
static int32_t mndCreateStream(SMnode *pMnode, SMndMsg *pReq, SCMCreateStreamReq *pCreate, SDbObj *pDb) {
static int32_t mndCreateStream(SMnode *pMnode, SNodeMsg *pReq, SCMCreateStreamReq *pCreate, SDbObj *pDb) {
mDebug("stream:%s to create", pCreate->name);
SStreamObj streamObj = {0};
tstrncpy(streamObj.name, pCreate->name, TSDB_STREAM_FNAME_LEN);
@ -247,8 +247,8 @@ static int32_t mndCreateStream(SMnode *pMnode, SMndMsg *pReq, SCMCreateStreamReq
return 0;
}
static int32_t mndProcessCreateStreamReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateStreamReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SStreamObj *pStream = NULL;
SDbObj *pDb = NULL;
@ -339,8 +339,8 @@ static int32_t mndGetNumOfStreams(SMnode *pMnode, char *dbName, int32_t *pNumOfS
return 0;
}
static int32_t mndGetStreamMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetStreamMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
if (mndGetNumOfStreams(pMnode, pShow->db, &pShow->numOfRows) != 0) {
@ -383,8 +383,8 @@ static int32_t mndGetStreamMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *p
return 0;
}
static int32_t mndRetrieveStream(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveStream(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SStreamObj *pStream = NULL;

44
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -48,14 +48,14 @@ static int32_t mndSubActionInsert(SSdb *pSdb, SMqSubscribeObj *);
static int32_t mndSubActionDelete(SSdb *pSdb, SMqSubscribeObj *);
static int32_t mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubscribeObj *pNewSub);
static int32_t mndProcessSubscribeReq(SMndMsg *pMsg);
static int32_t mndProcessSubscribeRsp(SMndMsg *pMsg);
static int32_t mndProcessSubscribeInternalReq(SMndMsg *pMsg);
static int32_t mndProcessSubscribeInternalRsp(SMndMsg *pMsg);
static int32_t mndProcessMqTimerMsg(SMndMsg *pMsg);
static int32_t mndProcessGetSubEpReq(SMndMsg *pMsg);
static int32_t mndProcessDoRebalanceMsg(SMndMsg *pMsg);
static int32_t mndProcessResetOffsetReq(SMndMsg *pMsg);
static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg);
static int32_t mndProcessSubscribeRsp(SNodeMsg *pMsg);
static int32_t mndProcessSubscribeInternalReq(SNodeMsg *pMsg);
static int32_t mndProcessSubscribeInternalRsp(SNodeMsg *pMsg);
static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg);
static int32_t mndProcessGetSubEpReq(SNodeMsg *pMsg);
static int32_t mndProcessDoRebalanceMsg(SNodeMsg *pMsg);
static int32_t mndProcessResetOffsetReq(SNodeMsg *pMsg);
static int32_t mndPersistMqSetConnReq(SMnode *pMnode, STrans *pTrans, const SMqTopicObj *pTopic, const char *cgroup,
const SMqConsumerEp *pConsumerEp);
@ -211,8 +211,8 @@ static int32_t mndPersistCancelConnReq(SMnode *pMnode, STrans *pTrans, const SMq
}
#if 0
static int32_t mndProcessResetOffsetReq(SMndMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndProcessResetOffsetReq(SNodeMsg *pMsg) {
SMnode *pMnode = pMsg->pNode;
uint8_t *str = pMsg->rpcMsg.pCont;
SMqCMResetOffsetReq req;
@ -249,14 +249,14 @@ static int32_t mndProcessResetOffsetReq(SMndMsg *pMsg) {
}
#endif
static int32_t mndProcessGetSubEpReq(SMndMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndProcessGetSubEpReq(SNodeMsg *pMsg) {
SMnode *pMnode = pMsg->pNode;
SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->rpcMsg.pCont;
SMqCMGetSubEpRsp rsp = {0};
int64_t consumerId = be64toh(pReq->consumerId);
int32_t epoch = ntohl(pReq->epoch);
SMqConsumerObj *pConsumer = mndAcquireConsumer(pMsg->pMnode, consumerId);
SMqConsumerObj *pConsumer = mndAcquireConsumer(pMsg->pNode, consumerId);
if (pConsumer == NULL) {
terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
return -1;
@ -327,8 +327,8 @@ static int32_t mndProcessGetSubEpReq(SMndMsg *pMsg) {
tEncodeSMqCMGetSubEpRsp(&abuf, &rsp);
tDeleteSMqCMGetSubEpRsp(&rsp);
mndReleaseConsumer(pMnode, pConsumer);
pMsg->pCont = buf;
pMsg->contLen = tlen;
pMsg->pRsp = buf;
pMsg->rspLen = tlen;
return 0;
}
@ -356,8 +356,8 @@ static SMqRebSubscribe *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
return pRebSub;
}
static int32_t mndProcessMqTimerMsg(SMndMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg) {
SMnode *pMnode = pMsg->pNode;
SSdb *pSdb = pMnode->pSdb;
SMqConsumerObj *pConsumer;
void *pIter = NULL;
@ -428,8 +428,8 @@ static int32_t mndProcessMqTimerMsg(SMndMsg *pMsg) {
return 0;
}
static int32_t mndProcessDoRebalanceMsg(SMndMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndProcessDoRebalanceMsg(SNodeMsg *pMsg) {
SMnode *pMnode = pMsg->pNode;
SMqDoRebalanceMsg *pReq = pMsg->rpcMsg.pCont;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_REBALANCE, &pMsg->rpcMsg);
void *pIter = NULL;
@ -994,8 +994,8 @@ void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub) {
sdbRelease(pSdb, pSub);
}
static int32_t mndProcessSubscribeReq(SMndMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) {
SMnode *pMnode = pMsg->pNode;
char *msgStr = pMsg->rpcMsg.pCont;
SCMSubscribeReq subscribe;
tDeserializeSCMSubscribeReq(msgStr, &subscribe);
@ -1156,7 +1156,7 @@ static int32_t mndProcessSubscribeReq(SMndMsg *pMsg) {
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}
static int32_t mndProcessSubscribeInternalRsp(SMndMsg *pRsp) {
static int32_t mndProcessSubscribeInternalRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}

4
source/dnode/mnode/impl/src/mndTelem.c
View File

@ -79,8 +79,8 @@ static char* mndBuildTelemetryReport(SMnode* pMnode) {
return pCont;
}
static int32_t mndProcessTelemTimer(SMndMsg* pReq) {
SMnode* pMnode = pReq->pMnode;
static int32_t mndProcessTelemTimer(SNodeMsg* pReq) {
SMnode* pMnode = pReq->pNode;
STelemMgmt* pMgmt = &pMnode->telemMgmt;
if (!pMgmt->enable) return 0;

34
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -31,12 +31,12 @@
static int32_t mndTopicActionInsert(SSdb *pSdb, SMqTopicObj *pTopic);
static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic);
static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pTopic, SMqTopicObj *pNewTopic);
static int32_t mndProcessCreateTopicReq(SMndMsg *pReq);
static int32_t mndProcessDropTopicReq(SMndMsg *pReq);
static int32_t mndProcessDropTopicInRsp(SMndMsg *pRsp);
static int32_t mndProcessTopicMetaReq(SMndMsg *pReq);
static int32_t mndGetTopicMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveTopic(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndProcessCreateTopicReq(SNodeMsg *pReq);
static int32_t mndProcessDropTopicReq(SNodeMsg *pReq);
static int32_t mndProcessDropTopicInRsp(SNodeMsg *pRsp);
static int32_t mndProcessTopicMetaReq(SNodeMsg *pReq);
static int32_t mndGetTopicMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter);
int32_t mndInitTopic(SMnode *pMnode) {
@ -236,7 +236,7 @@ static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
return 0;
}
static int32_t mndCreateTopic(SMnode *pMnode, SMndMsg *pReq, SCMCreateTopicReq *pCreate, SDbObj *pDb) {
static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq *pCreate, SDbObj *pDb) {
mDebug("topic:%s to create", pCreate->name);
SMqTopicObj topicObj = {0};
tstrncpy(topicObj.name, pCreate->name, TSDB_TOPIC_FNAME_LEN);
@ -276,8 +276,8 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMndMsg *pReq, SCMCreateTopicReq *
return 0;
}
static int32_t mndProcessCreateTopicReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateTopicReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SMqTopicObj *pTopic = NULL;
SDbObj *pDb = NULL;
@ -341,7 +341,7 @@ CREATE_TOPIC_OVER:
return code;
}
static int32_t mndDropTopic(SMnode *pMnode, SMndMsg *pReq, SMqTopicObj *pTopic) {
static int32_t mndDropTopic(SMnode *pMnode, SNodeMsg *pReq, SMqTopicObj *pTopic) {
// TODO: cannot drop when subscribed
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_TOPIC, &pReq->rpcMsg);
if (pTrans == NULL) {
@ -368,8 +368,8 @@ static int32_t mndDropTopic(SMnode *pMnode, SMndMsg *pReq, SMqTopicObj *pTopic)
return 0;
}
static int32_t mndProcessDropTopicReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropTopicReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SMDropTopicReq dropReq = {0};
if (tDeserializeSMDropTopicReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &dropReq) != 0) {
@ -403,7 +403,7 @@ static int32_t mndProcessDropTopicReq(SMndMsg *pReq) {
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}
static int32_t mndProcessDropTopicInRsp(SMndMsg *pRsp) {
static int32_t mndProcessDropTopicInRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
@ -435,8 +435,8 @@ static int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTo
return 0;
}
static int32_t mndGetTopicMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetTopicMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
if (mndGetNumOfTopics(pMnode, pShow->db, &pShow->numOfRows) != 0) {
@ -479,8 +479,8 @@ static int32_t mndGetTopicMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveTopic(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SMqTopicObj *pTopic = NULL;

28
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -55,11 +55,11 @@ static bool mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans);
static void mndTransExecute(SMnode *pMnode, STrans *pTrans);
static void mndTransSendRpcRsp(STrans *pTrans);
static int32_t mndProcessTransReq(SMndMsg *pReq);
static int32_t mndProcessKillTransReq(SMndMsg *pReq);
static int32_t mndProcessTransReq(SNodeMsg *pReq);
static int32_t mndProcessKillTransReq(SNodeMsg *pReq);
static int32_t mndGetTransMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveTrans(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetTransMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter);
int32_t mndInitTrans(SMnode *pMnode) {
@ -774,8 +774,8 @@ static void mndTransSendRpcRsp(STrans *pTrans) {
}
}
void mndTransProcessRsp(SMndMsg *pRsp) {
SMnode *pMnode = pRsp->pMnode;
void mndTransProcessRsp(SNodeMsg *pRsp) {
SMnode *pMnode = pRsp->pNode;
int64_t signature = (int64_t)(pRsp->rpcMsg.ahandle);
int32_t transId = (int32_t)(signature >> 32);
int32_t action = (int32_t)((signature << 32) >> 32);
@ -1157,8 +1157,8 @@ static void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
mndTransSendRpcRsp(pTrans);
}
static int32_t mndProcessTransReq(SMndMsg *pReq) {
mndTransPullup(pReq->pMnode);
static int32_t mndProcessTransReq(SNodeMsg *pReq) {
mndTransPullup(pReq->pNode);
return 0;
}
@ -1199,8 +1199,8 @@ static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
return 0;
}
static int32_t mndProcessKillTransReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessKillTransReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SKillTransReq killReq = {0};
int32_t code = -1;
SUserObj *pUser = NULL;
@ -1257,8 +1257,8 @@ void mndTransPullup(SMnode *pMnode) {
sdbWriteFile(pMnode->pSdb);
}
static int32_t mndGetTransMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetTransMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -1320,8 +1320,8 @@ static int32_t mndGetTransMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveTrans(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
STrans *pTrans = NULL;

48
source/dnode/mnode/impl/src/mndUser.c
View File

@ -30,13 +30,13 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw);
static int32_t mndUserActionInsert(SSdb *pSdb, SUserObj *pUser);
static int32_t mndUserActionDelete(SSdb *pSdb, SUserObj *pUser);
static int32_t mndUserActionUpdate(SSdb *pSdb, SUserObj *pOld, SUserObj *pNew);
static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate, SMndMsg *pReq);
static int32_t mndProcessCreateUserReq(SMndMsg *pReq);
static int32_t mndProcessAlterUserReq(SMndMsg *pReq);
static int32_t mndProcessDropUserReq(SMndMsg *pReq);
static int32_t mndProcessGetUserAuthReq(SMndMsg *pReq);
static int32_t mndGetUserMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveUsers(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate, SNodeMsg *pReq);
static int32_t mndProcessCreateUserReq(SNodeMsg *pReq);
static int32_t mndProcessAlterUserReq(SNodeMsg *pReq);
static int32_t mndProcessDropUserReq(SNodeMsg *pReq);
static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq);
static int32_t mndGetUserMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveUsers(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextUser(SMnode *pMnode, void *pIter);
int32_t mndInitUser(SMnode *pMnode) {
@ -261,7 +261,7 @@ void mndReleaseUser(SMnode *pMnode, SUserObj *pUser) {
sdbRelease(pSdb, pUser);
}
static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate, SMndMsg *pReq) {
static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate, SNodeMsg *pReq) {
SUserObj userObj = {0};
taosEncryptPass_c((uint8_t *)pCreate->pass, strlen(pCreate->pass), userObj.pass);
tstrncpy(userObj.user, pCreate->user, TSDB_USER_LEN);
@ -295,8 +295,8 @@ static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate
return 0;
}
static int32_t mndProcessCreateUserReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessCreateUserReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SUserObj *pOperUser = NULL;
@ -349,7 +349,7 @@ CREATE_USER_OVER:
return code;
}
static int32_t mndUpdateUser(SMnode *pMnode, SUserObj *pOld, SUserObj *pNew, SMndMsg *pReq) {
static int32_t mndUpdateUser(SMnode *pMnode, SUserObj *pOld, SUserObj *pNew, SNodeMsg *pReq) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_ALTER_USER,&pReq->rpcMsg);
if (pTrans == NULL) {
mError("user:%s, failed to update since %s", pOld->user, terrstr());
@ -397,8 +397,8 @@ static SHashObj *mndDupDbHash(SHashObj *pOld) {
return pNew;
}
static int32_t mndProcessAlterUserReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SUserObj *pOperUser = NULL;
@ -510,7 +510,7 @@ ALTER_USER_OVER:
return code;
}
static int32_t mndDropUser(SMnode *pMnode, SMndMsg *pReq, SUserObj *pUser) {
static int32_t mndDropUser(SMnode *pMnode, SNodeMsg *pReq, SUserObj *pUser) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK,TRN_TYPE_DROP_USER, &pReq->rpcMsg);
if (pTrans == NULL) {
mError("user:%s, failed to drop since %s", pUser->user, terrstr());
@ -536,8 +536,8 @@ static int32_t mndDropUser(SMnode *pMnode, SMndMsg *pReq, SUserObj *pUser) {
return 0;
}
static int32_t mndProcessDropUserReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessDropUserReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SUserObj *pOperUser = NULL;
@ -585,8 +585,8 @@ DROP_USER_OVER:
return code;
}
static int32_t mndProcessGetUserAuthReq(SMndMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
int32_t code = -1;
SUserObj *pUser = NULL;
SGetUserAuthReq authReq = {0};
@ -635,8 +635,8 @@ static int32_t mndProcessGetUserAuthReq(SMndMsg *pReq) {
tSerializeSGetUserAuthRsp(pRsp, contLen, &authRsp);
pReq->pCont = pRsp;
pReq->contLen = contLen;
pReq->pRsp = pRsp;
pReq->rspLen = contLen;
code = 0;
GET_AUTH_OVER:
@ -647,8 +647,8 @@ GET_AUTH_OVER:
return code;
}
static int32_t mndGetUserMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetUserMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -693,8 +693,8 @@ static int32_t mndGetUserMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMe
return 0;
}
static int32_t mndRetrieveUsers(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveUsers(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SUserObj *pUser = NULL;

44
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -29,17 +29,17 @@ static int32_t mndVgroupActionInsert(SSdb *pSdb, SVgObj *pVgroup);
static int32_t mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup);
static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOld, SVgObj *pNew);
static int32_t mndProcessCreateVnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessAlterVnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessDropVnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessSyncVnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessCompactVnodeRsp(SMndMsg *pRsp);
static int32_t mndProcessCreateVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessAlterVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessDropVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessSyncVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessCompactVnodeRsp(SNodeMsg *pRsp);
static int32_t mndGetVgroupMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveVgroups(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetVgroupMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveVgroups(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextVgroup(SMnode *pMnode, void *pIter);
static int32_t mndGetVnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveVnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetVnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveVnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextVnode(SMnode *pMnode, void *pIter);
int32_t mndInitVgroup(SMnode *pMnode) {
@ -452,24 +452,24 @@ SEpSet mndGetVgroupEpset(SMnode *pMnode, SVgObj *pVgroup) {
return epset;
}
static int32_t mndProcessCreateVnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessCreateVnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessAlterVnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessAlterVnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessDropVnodeRsp(SMndMsg *pRsp) {
static int32_t mndProcessDropVnodeRsp(SNodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessSyncVnodeRsp(SMndMsg *pRsp) { return 0; }
static int32_t mndProcessSyncVnodeRsp(SNodeMsg *pRsp) { return 0; }
static int32_t mndProcessCompactVnodeRsp(SMndMsg *pRsp) { return 0; }
static int32_t mndProcessCompactVnodeRsp(SNodeMsg *pRsp) { return 0; }
static bool mndGetVgroupMaxReplicaFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
SVgObj *pVgroup = pObj;
@ -500,8 +500,8 @@ static int32_t mndGetVgroupMaxReplica(SMnode *pMnode, char *dbName, int8_t *pRep
return 0;
}
static int32_t mndGetVgroupMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetVgroupMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
if (mndGetVgroupMaxReplica(pMnode, pShow->db, &pShow->replica, &pShow->numOfRows) != 0) {
@ -551,8 +551,8 @@ static int32_t mndGetVgroupMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *p
return 0;
}
static int32_t mndRetrieveVgroups(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveVgroups(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SVgObj *pVgroup = NULL;
@ -624,8 +624,8 @@ int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId) {
return numOfVnodes;
}
static int32_t mndGetVnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndGetVnodeMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
@ -664,8 +664,8 @@ static int32_t mndGetVnodeMeta(SMndMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
return 0;
}
static int32_t mndRetrieveVnodes(SMndMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
static int32_t mndRetrieveVnodes(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SVgObj *pVgroup = NULL;

4
source/dnode/mnode/impl/src/mnode.c
View File

@ -386,8 +386,8 @@ void mndDestroy(const char *path) {
mDebug("mnode is destroyed");
}
int32_t mndProcessMsg(SMndMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
int32_t mndProcessMsg(SNodeMsg *pMsg) {
SMnode *pMnode = pMsg->pNode;
SRpcMsg *pRpc = &pMsg->rpcMsg;
tmsg_t msgType = pMsg->rpcMsg.msgType;
void *ahandle = pMsg->rpcMsg.ahandle;