Skip to content

T
TDengine

taosdata / TDengine
11 个月 前同步成功

通知 1179
Star 22014
Fork 4786
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 5e0a6d96 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #10285 from taosdata/feature/privilege 

serialize msg
上级 27cc205a 88ee9d94
展开全部 隐藏空白更改
内联 并排
Showing with 1168 addition and 731 deletion
include/common/tmsg.h
浏览文件 @ 5e0a6d96
......@@ -201,18 +201,6 @@ typedef struct SSubmitBlk {
char data[];
} SSubmitBlk;
typedef struct {
/* data */
} SSubmitReq;
typedef struct {
/* data */
} SSubmitRsp;
typedef struct {
/* data */
} SSubmitReqReader;
// Submit message for this TSDB
typedef struct {
SMsgHead header;
......@@ -220,7 +208,7 @@ typedef struct {
int32_t length;
int32_t numOfBlocks;
char blocks[];
} SSubmitMsg;
} SSubmitReq;
typedef struct {
int32_t totalLen;
......@@ -234,7 +222,7 @@ typedef struct {
void* pMsg;
} SSubmitMsgIter;
int32_t tInitSubmitMsgIter(SSubmitMsg* pMsg, SSubmitMsgIter* pIter);
int32_t tInitSubmitMsgIter(SSubmitReq* pMsg, SSubmitMsgIter* pIter);
int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
int32_t tInitSubmitBlkIter(SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
......@@ -244,16 +232,16 @@ typedef struct {
int32_t vnode; // vnode index of failed block
int32_t sid; // table index of failed block
int32_t code; // errorcode while write data to vnode, such as not created, dropped, no space, invalid table
} SShellSubmitRspBlock;
} SSubmitRspBlock;
typedef struct {
int32_t code; // 0-success, > 0 error code
int32_t numOfRows; // number of records the client is trying to write
int32_t affectedRows; // number of records actually written
int32_t failedRows; // number of failed records (exclude duplicate records)
int32_t numOfFailedBlocks;
SShellSubmitRspBlock failedBlocks[];
} SShellSubmitRsp;
int32_t code; // 0-success, > 0 error code
int32_t numOfRows; // number of records the client is trying to write
int32_t affectedRows; // number of records actually written
int32_t failedRows; // number of failed records (exclude duplicate records)
int32_t numOfFailedBlocks;
SSubmitRspBlock failedBlocks[];
} SSubmitRsp;
typedef struct SSchema {
int8_t type;
......@@ -696,8 +684,8 @@ typedef struct {
SArray* pVloads; // array of SVnodeLoad
} SStatusReq;
int32_t tSerializeSStatusReq(void** buf, SStatusReq* pReq);
void* tDeserializeSStatusReq(void* buf, SStatusReq* pReq);
int32_t tSerializeSStatusReq(void* buf, int32_t bufLen, SStatusReq* pReq);
int32_t tDeserializeSStatusReq(void* buf, int32_t bufLen, SStatusReq* pReq);
typedef struct {
int32_t dnodeId;
......@@ -716,12 +704,15 @@ typedef struct {
SArray* pDnodeEps; // Array of SDnodeEp
} SStatusRsp;
int32_t tSerializeSStatusRsp(void** buf, SStatusRsp* pRsp);
void* tDeserializeSStatusRsp(void* buf, SStatusRsp* pRsp);
int32_t tSerializeSStatusRsp(void* buf, int32_t bufLen, SStatusRsp* pRsp);
int32_t tDeserializeSStatusRsp(void* buf, int32_t bufLen, SStatusRsp* pRsp);
typedef struct {
int32_t reserve;
} STransReq;
int32_t reserved;
} SMTimerReq;
int32_t tSerializeSMTimerMsg(void* buf, int32_t bufLen, SMTimerReq* pReq);
int32_t tDeserializeSMTimerMsg(void* buf, int32_t bufLen, SMTimerReq* pReq);
typedef struct {
int32_t id;
......@@ -756,6 +747,9 @@ typedef struct {
SReplica replicas[TSDB_MAX_REPLICA];
} SCreateVnodeReq, SAlterVnodeReq;
int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
int32_t tDeserializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
typedef struct {
int32_t vgId;
int32_t dnodeId;
......@@ -763,10 +757,8 @@ typedef struct {
char db[TSDB_DB_FNAME_LEN];
} SDropVnodeReq, SSyncVnodeReq, SCompactVnodeReq;
typedef struct {
int32_t vgId;
int8_t accessState;
} SAuthVnodeReq;
int32_t tSerializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq);
int32_t tDeserializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq);
typedef struct {
SMsgHead header;
......@@ -884,7 +876,6 @@ typedef struct {
int8_t precision;
int8_t compressed;
int32_t compLen;
int32_t numOfRows;
char data[];
} SRetrieveTableRsp;
......@@ -926,6 +917,9 @@ typedef struct {
SReplica replicas[TSDB_MAX_REPLICA];
} SDCreateMnodeReq, SDAlterMnodeReq;
int32_t tSerializeSDCreateMnodeReq(void* buf, int32_t bufLen, SDCreateMnodeReq* pReq);
int32_t tDeserializeSDCreateMnodeReq(void* buf, int32_t bufLen, SDCreateMnodeReq* pReq);
typedef struct {
int32_t dnodeId;
} SMCreateQnodeReq, SMDropQnodeReq, SDCreateQnodeReq, SDDropQnodeReq, SMCreateSnodeReq, SMDropSnodeReq,
......@@ -973,10 +967,16 @@ typedef struct {
int32_t queryId;
} SKillQueryReq;
int32_t tSerializeSKillQueryReq(void* buf, int32_t bufLen, SKillQueryReq* pReq);
int32_t tDeserializeSKillQueryReq(void* buf, int32_t bufLen, SKillQueryReq* pReq);
typedef struct {
int32_t connId;
} SKillConnReq;
int32_t tSerializeSKillConnReq(void* buf, int32_t bufLen, SKillConnReq* pReq);
int32_t tDeserializeSKillConnReq(void* buf, int32_t bufLen, SKillConnReq* pReq);
typedef struct {
char user[TSDB_USER_LEN];
char spi;
......@@ -985,9 +985,11 @@ typedef struct {
char ckey[TSDB_PASSWORD_LEN];
} SAuthReq, SAuthRsp;
int32_t tSerializeSAuthReq(void* buf, int32_t bufLen, SAuthReq* pReq);
int32_t tDeserializeSAuthReq(void* buf, int32_t bufLen, SAuthReq* pReq);
typedef struct {
int8_t finished;
int8_t align[7];
char name[TSDB_STEP_NAME_LEN];
char desc[TSDB_STEP_DESC_LEN];
} SStartupReq;
......@@ -1195,10 +1197,6 @@ static FORCE_INLINE void* tDeserializeSMVSubscribeReq(void* buf, SMVSubscribeReq
return buf;
}
typedef struct {
int32_t reserved;
} SMqTmrMsg;
typedef struct {
const char* key;
SArray* lostConsumers; // SArray<int64_t>
......@@ -1362,7 +1360,6 @@ typedef struct {
int8_t precision;
int8_t compressed;
int32_t compLen;
int32_t numOfRows;
char data[];
} SVShowTablesFetchRsp;
......
include/common/tmsgdef.h
浏览文件 @ 5e0a6d96
......@@ -84,7 +84,6 @@ enum {
TD_DEF_MSG_TYPE(TDMT_DND_CREATE_VNODE, "dnode-create-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_ALTER_VNODE, "dnode-alter-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "dnode-drop-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_AUTH_VNODE, "dnode-auth-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_SYNC_VNODE, "dnode-sync-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_COMPACT_VNODE, "dnode-compact-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "dnode-config-dnode", NULL, NULL)
......@@ -144,7 +143,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
TD_DEF_MSG_TYPE(TDMT_MND_GET_SUB_EP, "mnode-get-sub-ep", SMqCMGetSubEpReq, SMqCMGetSubEpRsp)
TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-timer", SMqTmrMsg, SMqTmrMsg)
TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-timer", SMTimerReq, SMTimerReq)
TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, SMqDoRebalanceMsg)
// Requests handled by VNODE
......
include/libs/parser/parsenodes.h
浏览文件 @ 5e0a6d96
......@@ -135,7 +135,7 @@ typedef struct SVgDataBlocks {
SVgroupInfo vg;
int32_t numOfTables; // number of tables in current submit block
uint32_t size;
char *pData; // SMsgDesc + SSubmitMsg + SSubmitBlk + ...
char *pData; // SMsgDesc + SSubmitReq + SSubmitBlk + ...
} SVgDataBlocks;
typedef struct SVnodeModifOpStmtInfo {
......
source/client/src/clientHb.c
浏览文件 @ 5e0a6d96
......@@ -315,7 +315,7 @@ void hbFreeReq(void *req) {
SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
SClientHbBatchReq* pBatchReq = malloc(sizeof(SClientHbBatchReq));
SClientHbBatchReq* pBatchReq = calloc(1, sizeof(SClientHbBatchReq));
if (pBatchReq == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
return NULL;
......
source/common/src/tmsg.c
浏览文件 @ 5e0a6d96
此差异已折叠。
source/dnode/mgmt/impl/src/dndBnode.c
浏览文件 @ 5e0a6d96
......@@ -258,11 +258,14 @@ static int32_t dndDropBnode(SDnode *pDnode) {
return 0;
}
int32_t dndProcessCreateBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDCreateBnodeReq *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
int32_t dndProcessCreateBnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDCreateBnodeReq createReq = {0};
if (tDeserializeSMCreateDropQSBNodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
if (createReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_BNODE_INVALID_OPTION;
dError("failed to create bnode since %s", terrstr());
return -1;
......@@ -271,11 +274,14 @@ int32_t dndProcessCreateBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
}
int32_t dndProcessDropBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDDropBnodeReq *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
int32_t dndProcessDropBnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDDropBnodeReq dropReq = {0};
if (tDeserializeSMCreateDropQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
if (dropReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_BNODE_INVALID_OPTION;
dError("failed to drop bnode since %s", terrstr());
return -1;
......
source/dnode/mgmt/impl/src/dndMgmt.c
浏览文件 @ 5e0a6d96
......@@ -379,10 +379,9 @@ void dndSendStatusReq(SDnode *pDnode) {
req.pVloads = taosArrayInit(TSDB_MAX_VNODES, sizeof(SVnodeLoad));
dndGetVnodeLoads(pDnode, req.pVloads);
int32_t contLen = tSerializeSStatusReq(NULL, &req);
int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
void *pHead = rpcMallocCont(contLen);
void *pBuf = pHead;
tSerializeSStatusReq(&pBuf, &req);
tSerializeSStatusReq(pHead, contLen, &req);
taosArrayDestroy(req.pVloads);
SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)9527};
......@@ -395,7 +394,7 @@ void dndSendStatusReq(SDnode *pDnode) {
static void dndUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
if (pMgmt->dnodeId == 0) {
dInfo("set dnodeId:%d clusterId:0x%" PRId64, pCfg->dnodeId, pCfg->clusterId);
dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
taosWLockLatch(&pMgmt->latch);
pMgmt->dnodeId = pCfg->dnodeId;
pMgmt->clusterId = pCfg->clusterId;
......@@ -437,7 +436,8 @@ static void dndProcessStatusRsp(SDnode *pDnode, SRpcMsg *pRsp) {
}
} else {
SStatusRsp statusRsp = {0};
if (pRsp->pCont != NULL && pRsp->contLen != 0 && tDeserializeSStatusRsp(pRsp->pCont, &statusRsp) != NULL) {
if (pRsp->pCont != NULL && pRsp->contLen != 0 &&
tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
pMgmt->dver = statusRsp.dver;
dndUpdateDnodeCfg(pDnode, &statusRsp.dnodeCfg);
dndUpdateDnodeEps(pDnode, statusRsp.pDnodeEps);
......@@ -652,9 +652,6 @@ static void dndProcessMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg) {
case TDMT_DND_DROP_VNODE:
code = dndProcessDropVnodeReq(pDnode, pMsg);
break;
case TDMT_DND_AUTH_VNODE:
code = dndProcessAuthVnodeReq(pDnode, pMsg);
break;
case TDMT_DND_SYNC_VNODE:
code = dndProcessSyncVnodeReq(pDnode, pMsg);
break;
......
source/dnode/mgmt/impl/src/dndMnode.c
浏览文件 @ 5e0a6d96
......@@ -424,28 +424,22 @@ static int32_t dndDropMnode(SDnode *pDnode) {
return 0;
}
static SDCreateMnodeReq *dndParseCreateMnodeReq(SRpcMsg *pReq) {
SDCreateMnodeReq *pCreate = pReq->pCont;
pCreate->dnodeId = htonl(pCreate->dnodeId);
for (int32_t i = 0; i < pCreate->replica; ++i) {
pCreate->replicas[i].id = htonl(pCreate->replicas[i].id);
pCreate->replicas[i].port = htons(pCreate->replicas[i].port);
}
return pCreate;
}
int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDCreateMnodeReq *pCreate = dndParseCreateMnodeReq(pReq);
SDCreateMnodeReq createReq = {0};
if (tDeserializeSDCreateMnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pCreate->replica <= 1 || pCreate->dnodeId != dndGetDnodeId(pDnode)) {
if (createReq.replica <= 1 || createReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to create mnode since %s", terrstr());
return -1;
}
SMnodeOpt option = {0};
if (dndBuildMnodeOptionFromReq(pDnode, &option, pCreate) != 0) {
if (dndBuildMnodeOptionFromReq(pDnode, &option, &createReq) != 0) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to create mnode since %s", terrstr());
return -1;
......@@ -464,16 +458,20 @@ int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
}
int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDAlterMnodeReq *pAlter = dndParseCreateMnodeReq(pReq);
SDAlterMnodeReq alterReq = {0};
if (tDeserializeSDCreateMnodeReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pAlter->dnodeId != dndGetDnodeId(pDnode)) {
if (alterReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to alter mnode since %s", terrstr());
return -1;
}
SMnodeOpt option = {0};
if (dndBuildMnodeOptionFromReq(pDnode, &option, pAlter) != 0) {
if (dndBuildMnodeOptionFromReq(pDnode, &option, &alterReq) != 0) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to alter mnode since %s", terrstr());
return -1;
......@@ -494,10 +492,13 @@ int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
}
int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDDropMnodeReq *pDrop = pReq->pCont;
pDrop->dnodeId = htonl(pDrop->dnodeId);
SDDropMnodeReq dropReq = {0};
if (tDeserializeSMCreateDropMnodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pDrop->dnodeId != dndGetDnodeId(pDnode)) {
if (dropReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to drop mnode since %s", terrstr());
return -1;
......
source/dnode/mgmt/impl/src/dndQnode.c
浏览文件 @ 5e0a6d96
......@@ -264,11 +264,14 @@ static int32_t dndDropQnode(SDnode *pDnode) {
return 0;
}
int32_t dndProcessCreateQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDCreateQnodeReq *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
int32_t dndProcessCreateQnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDCreateQnodeReq createReq = {0};
if (tDeserializeSMCreateDropQSBNodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
if (createReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_QNODE_INVALID_OPTION;
dError("failed to create qnode since %s", terrstr());
return -1;
......@@ -277,11 +280,14 @@ int32_t dndProcessCreateQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
}
int32_t dndProcessDropQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDDropQnodeReq *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
int32_t dndProcessDropQnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDDropQnodeReq dropReq = {0};
if (tDeserializeSMCreateDropQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
if (dropReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_QNODE_INVALID_OPTION;
dError("failed to drop qnode since %s", terrstr());
return -1;
......
source/dnode/mgmt/impl/src/dndSnode.c
浏览文件 @ 5e0a6d96
......@@ -258,11 +258,14 @@ static int32_t dndDropSnode(SDnode *pDnode) {
return 0;
}
int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDCreateSnodeReq *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDCreateSnodeReq createReq = {0};
if (tDeserializeSMCreateDropQSBNodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
if (createReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_SNODE_INVALID_OPTION;
dError("failed to create snode since %s", terrstr());
return -1;
......@@ -271,11 +274,14 @@ int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
}
int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDDropSnodeReq *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDDropSnodeReq dropReq = {0};
if (tDeserializeSMCreateDropQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
if (dropReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_SNODE_INVALID_OPTION;
dError("failed to drop snode since %s", terrstr());
return -1;
......
source/dnode/mgmt/impl/src/dndTransport.c
浏览文件 @ 5e0a6d96
......@@ -57,8 +57,6 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_VNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_SYNC_VNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_SYNC_VNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_AUTH_VNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_AUTH_VNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_COMPACT_VNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_COMPACT_VNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CONFIG_DNODE)] = dndProcessMgmtMsg;
......@@ -310,24 +308,29 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
return -1;
}
SAuthReq *pReq = rpcMallocCont(sizeof(SAuthReq));
tstrncpy(pReq->user, user, TSDB_USER_LEN);
SAuthReq authReq = {0};
tstrncpy(authReq.user, user, TSDB_USER_LEN);
int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
void *pReq = rpcMallocCont(contLen);
tSerializeSAuthReq(pReq, contLen, &authReq);
SRpcMsg rpcMsg = {.pCont = pReq, .contLen = sizeof(SAuthReq), .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};
SRpcMsg rpcMsg = {.pCont = pReq, .contLen = contLen, .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};
SRpcMsg rpcRsp = {0};
dTrace("user:%s, send user auth req to other mnodes, spi:%d encrypt:%d", user, pReq->spi, pReq->encrypt);
dTrace("user:%s, send user auth req to other mnodes, spi:%d encrypt:%d", user, authReq.spi, authReq.encrypt);
dndSendMsgToMnodeRecv(pDnode, &rpcMsg, &rpcRsp);
if (rpcRsp.code != 0) {
terrno = rpcRsp.code;
dError("user:%s, failed to get user auth from other mnodes since %s", user, terrstr());
} else {
SAuthRsp *pRsp = rpcRsp.pCont;
memcpy(secret, pRsp->secret, TSDB_PASSWORD_LEN);
memcpy(ckey, pRsp->ckey, TSDB_PASSWORD_LEN);
*spi = pRsp->spi;
*encrypt = pRsp->encrypt;
dTrace("user:%s, success to get user auth from other mnodes, spi:%d encrypt:%d", user, pRsp->spi, pRsp->encrypt);
SAuthRsp authRsp = {0};
tDeserializeSAuthReq(rpcRsp.pCont, rpcRsp.contLen, &authRsp);
memcpy(secret, authRsp.secret, TSDB_PASSWORD_LEN);
memcpy(ckey, authRsp.ckey, TSDB_PASSWORD_LEN);
*spi = authRsp.spi;
*encrypt = authRsp.encrypt;
dTrace("user:%s, success to get user auth from other mnodes, spi:%d encrypt:%d", user, authRsp.spi,
authRsp.encrypt);
}
rpcFreeCont(rpcRsp.pCont);
......
source/dnode/mgmt/impl/src/dndVnodes.c
浏览文件 @ 5e0a6d96
......@@ -15,8 +15,8 @@
#define _DEFAULT_SOURCE
#include "dndVnodes.h"
#include "dndTransport.h"
#include "dndMgmt.h"
#include "dndTransport.h"
typedef struct {
int32_t vgId;
......@@ -34,9 +34,9 @@ typedef struct {
int8_t dropped;
int8_t accessState;
uint64_t dbUid;
char * db;
char * path;
SVnode * pImpl;
char *db;
char *path;
SVnode *pImpl;
STaosQueue *pWriteQ;
STaosQueue *pSyncQ;
STaosQueue *pApplyQ;
......@@ -50,7 +50,7 @@ typedef struct {
int32_t failed;
int32_t threadIndex;
pthread_t thread;
SDnode * pDnode;
SDnode *pDnode;
SWrapperCfg *pCfgs;
} SVnodeThread;
......@@ -68,7 +68,7 @@ void dndProcessVnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pE
void dndProcessVnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
static int32_t dndPutMsgIntoVnodeApplyQueue(SDnode *pDnode, int32_t vgId, SRpcMsg *pMsg);
static SVnodeObj * dndAcquireVnode(SDnode *pDnode, int32_t vgId);
static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId);
static void dndReleaseVnode(SDnode *pDnode, SVnodeObj *pVnode);
static int32_t dndOpenVnode(SDnode *pDnode, SWrapperCfg *pCfg, SVnode *pImpl);
static void dndCloseVnode(SDnode *pDnode, SVnodeObj *pVnode);
......@@ -81,7 +81,7 @@ static void dndCloseVnodes(SDnode *pDnode);
static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId) {
SVnodesMgmt *pMgmt = &pDnode->vmgmt;
SVnodeObj * pVnode = NULL;
SVnodeObj *pVnode = NULL;
int32_t refCount = 0;
taosRLockLatch(&pMgmt->latch);
......@@ -112,7 +112,7 @@ static void dndReleaseVnode(SDnode *pDnode, SVnodeObj *pVnode) {
static int32_t dndOpenVnode(SDnode *pDnode, SWrapperCfg *pCfg, SVnode *pImpl) {
SVnodesMgmt *pMgmt = &pDnode->vmgmt;
SVnodeObj * pVnode = calloc(1, sizeof(SVnodeObj));
SVnodeObj *pVnode = calloc(1, sizeof(SVnodeObj));
if (pVnode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
......@@ -189,7 +189,7 @@ static SVnodeObj **dndGetVnodesFromHash(SDnode *pDnode, int32_t *numOfVnodes) {
void *pIter = taosHashIterate(pMgmt->hash, NULL);
while (pIter) {
SVnodeObj **ppVnode = pIter;
SVnodeObj * pVnode = *ppVnode;
SVnodeObj *pVnode = *ppVnode;
if (pVnode && num < size) {
int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
......@@ -211,9 +211,9 @@ static int32_t dndGetVnodesFromFile(SDnode *pDnode, SWrapperCfg **ppCfgs, int32_
int32_t code = TSDB_CODE_DND_VNODE_READ_FILE_ERROR;
int32_t len = 0;
int32_t maxLen = 30000;
char * content = calloc(1, maxLen + 1);
cJSON * root = NULL;
FILE * fp = NULL;
char *content = calloc(1, maxLen + 1);
cJSON *root = NULL;
FILE *fp = NULL;
char file[PATH_MAX + 20] = {0};
SWrapperCfg *pCfgs = NULL;
......@@ -254,7 +254,7 @@ static int32_t dndGetVnodesFromFile(SDnode *pDnode, SWrapperCfg **ppCfgs, int32_
}
for (int32_t i = 0; i < vnodesNum; ++i) {
cJSON * vnode = cJSON_GetArrayItem(vnodes, i);
cJSON *vnode = cJSON_GetArrayItem(vnodes, i);
SWrapperCfg *pCfg = &pCfgs[i];
cJSON *vgId = cJSON_GetObjectItem(vnode, "vgId");
......@@ -326,7 +326,7 @@ static int32_t dndWriteVnodesToFile(SDnode *pDnode) {
int32_t len = 0;
int32_t maxLen = 65536;
char * content = calloc(1, maxLen + 1);
char *content = calloc(1, maxLen + 1);
len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"vnodes\": [\n");
......@@ -368,8 +368,8 @@ static int32_t dndWriteVnodesToFile(SDnode *pDnode) {
static void *dnodeOpenVnodeFunc(void *param) {
SVnodeThread *pThread = param;
SDnode * pDnode = pThread->pDnode;
SVnodesMgmt * pMgmt = &pDnode->vmgmt;
SDnode *pDnode = pThread->pDnode;
SVnodesMgmt *pMgmt = &pDnode->vmgmt;
dDebug("thread:%d, start to open %d vnodes", pThread->threadIndex, pThread->vnodeNum);
setThreadName("open-vnodes");
......@@ -383,7 +383,7 @@ static void *dnodeOpenVnodeFunc(void *param) {
dndReportStartup(pDnode, "open-vnodes", stepDesc);
SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId, .dbId = pCfg->dbUid};
SVnode * pImpl = vnodeOpen(pCfg->path, &cfg);
SVnode *pImpl = vnodeOpen(pCfg->path, &cfg);
if (pImpl == NULL) {
dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
pThread->failed++;
......@@ -499,31 +499,6 @@ static void dndCloseVnodes(SDnode *pDnode) {
dInfo("total vnodes:%d are all closed", numOfVnodes);
}
static SCreateVnodeReq *dndParseCreateVnodeReq(SRpcMsg *pReq) {
SCreateVnodeReq *pCreate = pReq->pCont;
pCreate->vgId = htonl(pCreate->vgId);
pCreate->dnodeId = htonl(pCreate->dnodeId);
pCreate->dbUid = htobe64(pCreate->dbUid);
pCreate->vgVersion = htonl(pCreate->vgVersion);
pCreate->cacheBlockSize = htonl(pCreate->cacheBlockSize);
pCreate->totalBlocks = htonl(pCreate->totalBlocks);
pCreate->daysPerFile = htonl(pCreate->daysPerFile);
pCreate->daysToKeep0 = htonl(pCreate->daysToKeep0);
pCreate->daysToKeep1 = htonl(pCreate->daysToKeep1);
pCreate->daysToKeep2 = htonl(pCreate->daysToKeep2);
pCreate->minRows = htonl(pCreate->minRows);
pCreate->maxRows = htonl(pCreate->maxRows);
pCreate->commitTime = htonl(pCreate->commitTime);
pCreate->fsyncPeriod = htonl(pCreate->fsyncPeriod);
for (int r = 0; r < pCreate->replica; ++r) {
SReplica *pReplica = &pCreate->replicas[r];
pReplica->id = htonl(pReplica->id);
pReplica->port = htons(pReplica->port);
}
return pCreate;
}
static void dndGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg->vgId = pCreate->vgId;
pCfg->wsize = pCreate->cacheBlockSize;
......@@ -556,37 +531,30 @@ static void dndGenerateWrapperCfg(SDnode *pDnode, SCreateVnodeReq *pCreate, SWra
pCfg->vgVersion = pCreate->vgVersion;
}
static SDropVnodeReq *dndParseDropVnodeReq(SRpcMsg *pReq) {
SDropVnodeReq *pDrop = pReq->pCont;
pDrop->vgId = htonl(pDrop->vgId);
return pDrop;
}
static SAuthVnodeReq *dndParseAuthVnodeReq(SRpcMsg *pReq) {
SAuthVnodeReq *pAuth = pReq->pCont;
pAuth->vgId = htonl(pAuth->vgId);
return pAuth;
}
int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SCreateVnodeReq *pCreate = dndParseCreateVnodeReq(pReq);
dDebug("vgId:%d, create vnode req is received", pCreate->vgId);
SCreateVnodeReq createReq = {0};
if (tDeserializeSCreateVnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
dDebug("vgId:%d, create vnode req is received", createReq.vgId);
SVnodeCfg vnodeCfg = {0};
dndGenerateVnodeCfg(pCreate, &vnodeCfg);
dndGenerateVnodeCfg(&createReq, &vnodeCfg);
SWrapperCfg wrapperCfg = {0};
dndGenerateWrapperCfg(pDnode, pCreate, &wrapperCfg);
dndGenerateWrapperCfg(pDnode, &createReq, &wrapperCfg);
if (pCreate->dnodeId != dndGetDnodeId(pDnode)) {
if (createReq.dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_VNODE_INVALID_OPTION;
dDebug("vgId:%d, failed to create vnode since %s", pCreate->vgId, terrstr());
dDebug("vgId:%d, failed to create vnode since %s", createReq.vgId, terrstr());
return -1;
}
SVnodeObj *pVnode = dndAcquireVnode(pDnode, pCreate->vgId);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, createReq.vgId);
if (pVnode != NULL) {
dDebug("vgId:%d, already exist", pCreate->vgId);
dDebug("vgId:%d, already exist", createReq.vgId);
dndReleaseVnode(pDnode, pVnode);
terrno = TSDB_CODE_DND_VNODE_ALREADY_DEPLOYED;
return -1;
......@@ -597,13 +565,13 @@ int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
vnodeCfg.dbId = wrapperCfg.dbUid;
SVnode *pImpl = vnodeOpen(wrapperCfg.path, &vnodeCfg);
if (pImpl == NULL) {
dError("vgId:%d, failed to create vnode since %s", pCreate->vgId, terrstr());
dError("vgId:%d, failed to create vnode since %s", createReq.vgId, terrstr());
return -1;
}
int32_t code = dndOpenVnode(pDnode, &wrapperCfg, pImpl);
if (code != 0) {
dError("vgId:%d, failed to open vnode since %s", pCreate->vgId, terrstr());
dError("vgId:%d, failed to open vnode since %s", createReq.vgId, terrstr());
vnodeClose(pImpl);
vnodeDestroy(wrapperCfg.path);
terrno = code;
......@@ -622,32 +590,37 @@ int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
}
int32_t dndProcessAlterVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SAlterVnodeReq *pAlter = (SAlterVnodeReq *)dndParseCreateVnodeReq(pReq);
dDebug("vgId:%d, alter vnode req is received", pAlter->vgId);
SAlterVnodeReq alterReq = {0};
if (tDeserializeSCreateVnodeReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
dDebug("vgId:%d, alter vnode req is received", alterReq.vgId);
SVnodeCfg vnodeCfg = {0};
dndGenerateVnodeCfg(pAlter, &vnodeCfg);
dndGenerateVnodeCfg(&alterReq, &vnodeCfg);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, pAlter->vgId);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, alterReq.vgId);
if (pVnode == NULL) {
dDebug("vgId:%d, failed to alter vnode since %s", pAlter->vgId, terrstr());
dDebug("vgId:%d, failed to alter vnode since %s", alterReq.vgId, terrstr());
return -1;
}
if (pAlter->vgVersion == pVnode->vgVersion) {
if (alterReq.vgVersion == pVnode->vgVersion) {
dndReleaseVnode(pDnode, pVnode);
dDebug("vgId:%d, no need to alter vnode cfg for version unchanged ", pAlter->vgId);
dDebug("vgId:%d, no need to alter vnode cfg for version unchanged ", alterReq.vgId);
return 0;
}
if (vnodeAlter(pVnode->pImpl, &vnodeCfg) != 0) {
dError("vgId:%d, failed to alter vnode since %s", pAlter->vgId, terrstr());
dError("vgId:%d, failed to alter vnode since %s", alterReq.vgId, terrstr());
dndReleaseVnode(pDnode, pVnode);
return -1;
}
int32_t oldVersion = pVnode->vgVersion;
pVnode->vgVersion = pAlter->vgVersion;
pVnode->vgVersion = alterReq.vgVersion;
int32_t code = dndWriteVnodesToFile(pDnode);
if (code != 0) {
pVnode->vgVersion = oldVersion;
......@@ -658,9 +631,13 @@ int32_t dndProcessAlterVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
}
int32_t dndProcessDropVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDropVnodeReq *pDrop = dndParseDropVnodeReq(pReq);
SDropVnodeReq dropReq = {0};
if (tDeserializeSDropVnodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
int32_t vgId = pDrop->vgId;
int32_t vgId = dropReq.vgId;
dDebug("vgId:%d, drop vnode req is received", vgId);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, vgId);
......@@ -683,27 +660,11 @@ int32_t dndProcessDropVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
return 0;
}
int32_t dndProcessAuthVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SAuthVnodeReq *pAuth = (SAuthVnodeReq *)dndParseAuthVnodeReq(pReq);
int32_t vgId = pAuth->vgId;
dDebug("vgId:%d, auth vnode req is received", vgId);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, vgId);
if (pVnode == NULL) {
dDebug("vgId:%d, failed to auth since %s", vgId, terrstr());
return -1;
}
pVnode->accessState = pAuth->accessState;
dndReleaseVnode(pDnode, pVnode);
return 0;
}
int32_t dndProcessSyncVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SSyncVnodeReq *pSync = (SSyncVnodeReq *)dndParseDropVnodeReq(pReq);
SSyncVnodeReq syncReq = {0};
tDeserializeSDropVnodeReq(pReq->pCont, pReq->contLen, &syncReq);
int32_t vgId = pSync->vgId;
int32_t vgId = syncReq.vgId;
dDebug("vgId:%d, sync vnode req is received", vgId);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, vgId);
......@@ -723,9 +684,10 @@ int32_t dndProcessSyncVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
}
int32_t dndProcessCompactVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SCompactVnodeReq *pCompact = (SCompactVnodeReq *)dndParseDropVnodeReq(pReq);
SCompactVnodeReq compatcReq = {0};
tDeserializeSDropVnodeReq(pReq->pCont, pReq->contLen, &compatcReq);
int32_t vgId = pCompact->vgId;
int32_t vgId = compatcReq.vgId;
dDebug("vgId:%d, compact vnode req is received", vgId);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, vgId);
......
source/dnode/mgmt/impl/test/bnode/dbnode.cpp
浏览文件 @ 5e0a6d96
......@@ -27,10 +27,12 @@ Testbase DndTestBnode::test;
TEST_F(DndTestBnode, 01_Create_Bnode) {
{
int32_t contLen = sizeof(SDCreateBnodeReq);
SDCreateBnodeReq createReq = {0};
createReq.dnodeId = 2;
SDCreateBnodeReq* pReq = (SDCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -38,10 +40,12 @@ TEST_F(DndTestBnode, 01_Create_Bnode) {
}
{
int32_t contLen = sizeof(SDCreateBnodeReq);
SDCreateBnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateBnodeReq* pReq = (SDCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -49,10 +53,12 @@ TEST_F(DndTestBnode, 01_Create_Bnode) {
}
{
int32_t contLen = sizeof(SDCreateBnodeReq);
SDCreateBnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateBnodeReq* pReq = (SDCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -62,11 +68,12 @@ TEST_F(DndTestBnode, 01_Create_Bnode) {
test.Restart();
{
int32_t contLen = sizeof(SDCreateBnodeReq);
SDCreateBnodeReq* pReq = (SDCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
SDCreateBnodeReq createReq = {0};
createReq.dnodeId = 1;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_BNODE_ALREADY_DEPLOYED);
......@@ -75,10 +82,12 @@ TEST_F(DndTestBnode, 01_Create_Bnode) {
TEST_F(DndTestBnode, 01_Drop_Bnode) {
{
int32_t contLen = sizeof(SDDropBnodeReq);
SDDropBnodeReq dropReq = {0};
dropReq.dnodeId = 2;
SDDropBnodeReq* pReq = (SDDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -86,10 +95,12 @@ TEST_F(DndTestBnode, 01_Drop_Bnode) {
}
{
int32_t contLen = sizeof(SDDropBnodeReq);
SDDropBnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropBnodeReq* pReq = (SDDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -97,10 +108,12 @@ TEST_F(DndTestBnode, 01_Drop_Bnode) {
}
{
int32_t contLen = sizeof(SDDropBnodeReq);
SDDropBnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropBnodeReq* pReq = (SDDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -110,10 +123,12 @@ TEST_F(DndTestBnode, 01_Drop_Bnode) {
test.Restart();
{
int32_t contLen = sizeof(SDDropBnodeReq);
SDDropBnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropBnodeReq* pReq = (SDDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -121,10 +136,12 @@ TEST_F(DndTestBnode, 01_Drop_Bnode) {
}
{
int32_t contLen = sizeof(SDCreateBnodeReq);
SDCreateBnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateBnodeReq* pReq = (SDCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......
source/dnode/mgmt/impl/test/mnode/dmnode.cpp
浏览文件 @ 5e0a6d96
......@@ -16,25 +16,27 @@ class DndTestMnode : public ::testing::Test {
static void SetUpTestSuite() { test.Init("/tmp/dnode_test_mnode", 9114); }
static void TearDownTestSuite() { test.Cleanup(); }
static Testbase test;
static Testbase test;
public:
void SetUp() override {}
void TearDown() override {}
};
Testbase DndTestMnode::test;
Testbase DndTestMnode::test;
TEST_F(DndTestMnode, 01_Create_Mnode) {
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq createReq = {0};
createReq.dnodeId = 2;
createReq.replica = 1;
createReq.replicas[0].id = 1;
createReq.replicas[0].port = 9113;
strcpy(createReq.replicas[0].fqdn, "localhost");
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -42,14 +44,16 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
}
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq createReq = {0};
createReq.dnodeId = 1;
createReq.replica = 1;
createReq.replicas[0].id = 2;
createReq.replicas[0].port = 9113;
strcpy(createReq.replicas[0].fqdn, "localhost");
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(2);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -57,17 +61,19 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
}
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 2;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
pReq->replicas[1].id = htonl(1);
pReq->replicas[1].port = htonl(9114);
strcpy(pReq->replicas[1].fqdn, "localhost");
SDCreateMnodeReq createReq = {0};
createReq.dnodeId = 1;
createReq.replica = 2;
createReq.replicas[0].id = 1;
createReq.replicas[0].port = 9113;
strcpy(createReq.replicas[0].fqdn, "localhost");
createReq.replicas[1].id = 1;
createReq.replicas[1].port = 9114;
strcpy(createReq.replicas[1].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -76,15 +82,17 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
}
TEST_F(DndTestMnode, 02_Alter_Mnode) {
{
int32_t contLen = sizeof(SDAlterMnodeReq);
{
SDAlterMnodeReq alterReq = {0};
alterReq.dnodeId = 2;
alterReq.replica = 1;
alterReq.replicas[0].id = 1;
alterReq.replicas[0].port = 9113;
strcpy(alterReq.replicas[0].fqdn, "localhost");
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &alterReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -92,14 +100,16 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
}
{
int32_t contLen = sizeof(SDAlterMnodeReq);
SDAlterMnodeReq alterReq = {0};
alterReq.dnodeId = 1;
alterReq.replica = 1;
alterReq.replicas[0].id = 2;
alterReq.replicas[0].port = 9113;
strcpy(alterReq.replicas[0].fqdn, "localhost");
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(2);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &alterReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -107,14 +117,16 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
}
{
int32_t contLen = sizeof(SDAlterMnodeReq);
SDAlterMnodeReq alterReq = {0};
alterReq.dnodeId = 1;
alterReq.replica = 1;
alterReq.replicas[0].id = 1;
alterReq.replicas[0].port = 9113;
strcpy(alterReq.replicas[0].fqdn, "localhost");
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &alterReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -124,10 +136,12 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
TEST_F(DndTestMnode, 03_Drop_Mnode) {
{
int32_t contLen = sizeof(SDDropMnodeReq);
SDDropMnodeReq dropReq = {0};
dropReq.dnodeId = 2;
SDDropMnodeReq* pReq = (SDDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
int32_t contLen = tSerializeSMCreateDropMnodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropMnodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -135,10 +149,12 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
}
{
int32_t contLen = sizeof(SDDropMnodeReq);
SDDropMnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropMnodeReq* pReq = (SDDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropMnodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropMnodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -146,10 +162,12 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
}
{
int32_t contLen = sizeof(SDDropMnodeReq);
SDDropMnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropMnodeReq* pReq = (SDDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropMnodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropMnodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -157,30 +175,33 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
}
{
int32_t contLen = sizeof(SDAlterMnodeReq);
SDAlterMnodeReq alterReq = {0};
alterReq.dnodeId = 1;
alterReq.replica = 1;
alterReq.replicas[0].id = 1;
alterReq.replicas[0].port = 9113;
strcpy(alterReq.replicas[0].fqdn, "localhost");
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &alterReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_NOT_DEPLOYED);
}
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 2;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SDCreateMnodeReq createReq = {0};
createReq.dnodeId = 1;
createReq.replica = 2;
createReq.replicas[0].id = 1;
createReq.replicas[0].port = 9113;
strcpy(createReq.replicas[0].fqdn, "localhost");
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......
source/dnode/mgmt/impl/test/qnode/dqnode.cpp
浏览文件 @ 5e0a6d96
......@@ -27,10 +27,12 @@ Testbase DndTestQnode::test;
TEST_F(DndTestQnode, 01_Create_Qnode) {
{
int32_t contLen = sizeof(SDCreateQnodeReq);
SDCreateQnodeReq createReq = {0};
createReq.dnodeId = 2;
SDCreateQnodeReq* pReq = (SDCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -38,10 +40,12 @@ TEST_F(DndTestQnode, 01_Create_Qnode) {
}
{
int32_t contLen = sizeof(SDCreateQnodeReq);
SDCreateQnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateQnodeReq* pReq = (SDCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -49,10 +53,12 @@ TEST_F(DndTestQnode, 01_Create_Qnode) {
}
{
int32_t contLen = sizeof(SDCreateQnodeReq);
SDCreateQnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateQnodeReq* pReq = (SDCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -62,10 +68,12 @@ TEST_F(DndTestQnode, 01_Create_Qnode) {
test.Restart();
{
int32_t contLen = sizeof(SDCreateQnodeReq);
SDCreateQnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateQnodeReq* pReq = (SDCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -75,10 +83,12 @@ TEST_F(DndTestQnode, 01_Create_Qnode) {
TEST_F(DndTestQnode, 02_Drop_Qnode) {
{
int32_t contLen = sizeof(SDDropQnodeReq);
SDDropQnodeReq dropReq = {0};
dropReq.dnodeId = 2;
SDDropQnodeReq* pReq = (SDDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -86,10 +96,12 @@ TEST_F(DndTestQnode, 02_Drop_Qnode) {
}
{
int32_t contLen = sizeof(SDDropQnodeReq);
SDDropQnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropQnodeReq* pReq = (SDDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -97,10 +109,12 @@ TEST_F(DndTestQnode, 02_Drop_Qnode) {
}
{
int32_t contLen = sizeof(SDDropQnodeReq);
SDDropQnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropQnodeReq* pReq = (SDDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -110,10 +124,12 @@ TEST_F(DndTestQnode, 02_Drop_Qnode) {
test.Restart();
{
int32_t contLen = sizeof(SDDropQnodeReq);
SDDropQnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropQnodeReq* pReq = (SDDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -121,10 +137,12 @@ TEST_F(DndTestQnode, 02_Drop_Qnode) {
}
{
int32_t contLen = sizeof(SDCreateQnodeReq);
SDCreateQnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateQnodeReq* pReq = (SDCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......
source/dnode/mgmt/impl/test/snode/dsnode.cpp
浏览文件 @ 5e0a6d96
......@@ -27,10 +27,12 @@ Testbase DndTestSnode::test;
TEST_F(DndTestSnode, 01_Create_Snode) {
{
int32_t contLen = sizeof(SDCreateSnodeReq);
SDCreateSnodeReq createReq = {0};
createReq.dnodeId = 2;
SDCreateSnodeReq* pReq = (SDCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -38,10 +40,12 @@ TEST_F(DndTestSnode, 01_Create_Snode) {
}
{
int32_t contLen = sizeof(SDCreateSnodeReq);
SDCreateSnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateSnodeReq* pReq = (SDCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -49,10 +53,12 @@ TEST_F(DndTestSnode, 01_Create_Snode) {
}
{
int32_t contLen = sizeof(SDCreateSnodeReq);
SDCreateSnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateSnodeReq* pReq = (SDCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -62,10 +68,12 @@ TEST_F(DndTestSnode, 01_Create_Snode) {
test.Restart();
{
int32_t contLen = sizeof(SDCreateSnodeReq);
SDCreateSnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateSnodeReq* pReq = (SDCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -75,10 +83,12 @@ TEST_F(DndTestSnode, 01_Create_Snode) {
TEST_F(DndTestSnode, 01_Drop_Snode) {
{
int32_t contLen = sizeof(SDDropSnodeReq);
SDDropSnodeReq dropReq = {0};
dropReq.dnodeId = 2;
SDDropSnodeReq* pReq = (SDDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -86,10 +96,12 @@ TEST_F(DndTestSnode, 01_Drop_Snode) {
}
{
int32_t contLen = sizeof(SDDropSnodeReq);
SDDropSnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropSnodeReq* pReq = (SDDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -97,10 +109,12 @@ TEST_F(DndTestSnode, 01_Drop_Snode) {
}
{
int32_t contLen = sizeof(SDDropSnodeReq);
SDDropSnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropSnodeReq* pReq = (SDDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -110,10 +124,12 @@ TEST_F(DndTestSnode, 01_Drop_Snode) {
test.Restart();
{
int32_t contLen = sizeof(SDDropSnodeReq);
SDDropSnodeReq dropReq = {0};
dropReq.dnodeId = 1;
SDDropSnodeReq* pReq = (SDDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -121,10 +137,12 @@ TEST_F(DndTestSnode, 01_Drop_Snode) {
}
{
int32_t contLen = sizeof(SDCreateSnodeReq);
SDCreateSnodeReq createReq = {0};
createReq.dnodeId = 1;
SDCreateSnodeReq* pReq = (SDCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......
source/dnode/mgmt/impl/test/vnode/vnode.cpp
浏览文件 @ 5e0a6d96
......@@ -27,38 +27,40 @@ Testbase DndTestVnode::test;
TEST_F(DndTestVnode, 01_Create_Vnode) {
for (int i = 0; i < 3; ++i) {
int32_t contLen = sizeof(SCreateVnodeReq);
SCreateVnodeReq* pReq = (SCreateVnodeReq*)rpcMallocCont(contLen);
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(1);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
pReq->vgVersion = htonl(1);
pReq->cacheBlockSize = htonl(16);
pReq->totalBlocks = htonl(10);
pReq->daysPerFile = htonl(10);
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRows = htonl(100);
pReq->minRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
pReq->precision = 0;
pReq->compression = 2;
pReq->replica = 1;
pReq->quorum = 1;
pReq->update = 0;
pReq->cacheLastRow = 0;
pReq->selfIndex = 0;
for (int r = 0; r < pReq->replica; ++r) {
SReplica* pReplica = &pReq->replicas[r];
pReplica->id = htonl(1);
pReplica->port = htons(9527);
SCreateVnodeReq createReq = {0};
createReq.vgId = 2;
createReq.dnodeId = 1;
strcpy(createReq.db, "1.d1");
createReq.dbUid = 9527;
createReq.vgVersion = 1;
createReq.cacheBlockSize = 16;
createReq.totalBlocks = 10;
createReq.daysPerFile = 10;
createReq.daysToKeep0 = 3650;
createReq.daysToKeep1 = 3650;
createReq.daysToKeep2 = 3650;
createReq.minRows = 100;
createReq.minRows = 4096;
createReq.commitTime = 3600;
createReq.fsyncPeriod = 3000;
createReq.walLevel = 1;
createReq.precision = 0;
createReq.compression = 2;
createReq.replica = 1;
createReq.quorum = 1;
createReq.update = 0;
createReq.cacheLastRow = 0;
createReq.selfIndex = 0;
for (int r = 0; r < createReq.replica; ++r) {
SReplica* pReplica = &createReq.replicas[r];
pReplica->id = 1;
pReplica->port = 9527;
}
int32_t contLen = tSerializeSCreateVnodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSCreateVnodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_VNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
if (i == 0) {
......@@ -70,38 +72,40 @@ TEST_F(DndTestVnode, 01_Create_Vnode) {
}
{
int32_t contLen = sizeof(SCreateVnodeReq);
SCreateVnodeReq* pReq = (SCreateVnodeReq*)rpcMallocCont(contLen);
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(3);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
pReq->vgVersion = htonl(1);
pReq->cacheBlockSize = htonl(16);
pReq->totalBlocks = htonl(10);
pReq->daysPerFile = htonl(10);
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRows = htonl(100);
pReq->minRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
pReq->precision = 0;
pReq->compression = 2;
pReq->replica = 1;
pReq->quorum = 1;
pReq->update = 0;
pReq->cacheLastRow = 0;
pReq->selfIndex = 0;
for (int r = 0; r < pReq->replica; ++r) {
SReplica* pReplica = &pReq->replicas[r];
pReplica->id = htonl(1);
pReplica->port = htons(9527);
SCreateVnodeReq createReq = {0};
createReq.vgId = 2;
createReq.dnodeId = 3;
strcpy(createReq.db, "1.d1");
createReq.dbUid = 9527;
createReq.vgVersion = 1;
createReq.cacheBlockSize = 16;
createReq.totalBlocks = 10;
createReq.daysPerFile = 10;
createReq.daysToKeep0 = 3650;
createReq.daysToKeep1 = 3650;
createReq.daysToKeep2 = 3650;
createReq.minRows = 100;
createReq.minRows = 4096;
createReq.commitTime = 3600;
createReq.fsyncPeriod = 3000;
createReq.walLevel = 1;
createReq.precision = 0;
createReq.compression = 2;
createReq.replica = 1;
createReq.quorum = 1;
createReq.update = 0;
createReq.cacheLastRow = 0;
createReq.selfIndex = 0;
for (int r = 0; r < createReq.replica; ++r) {
SReplica* pReplica = &createReq.replicas[r];
pReplica->id = 1;
pReplica->port = 9527;
}
int32_t contLen = tSerializeSCreateVnodeReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSCreateVnodeReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_VNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_VNODE_INVALID_OPTION);
......@@ -110,38 +114,40 @@ TEST_F(DndTestVnode, 01_Create_Vnode) {
TEST_F(DndTestVnode, 02_Alter_Vnode) {
for (int i = 0; i < 3; ++i) {
int32_t contLen = sizeof(SAlterVnodeReq);
SAlterVnodeReq* pReq = (SAlterVnodeReq*)rpcMallocCont(contLen);
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(1);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
pReq->vgVersion = htonl(2);
pReq->cacheBlockSize = htonl(16);
pReq->totalBlocks = htonl(10);
pReq->daysPerFile = htonl(10);
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRows = htonl(100);
pReq->minRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
pReq->precision = 0;
pReq->compression = 2;
pReq->replica = 1;
pReq->quorum = 1;
pReq->update = 0;
pReq->cacheLastRow = 0;
pReq->selfIndex = 0;
for (int r = 0; r < pReq->replica; ++r) {
SReplica* pReplica = &pReq->replicas[r];
pReplica->id = htonl(1);
pReplica->port = htons(9527);
SAlterVnodeReq alterReq = {0};
alterReq.vgId = 2;
alterReq.dnodeId = 1;
strcpy(alterReq.db, "1.d1");
alterReq.dbUid = 9527;
alterReq.vgVersion = 2;
alterReq.cacheBlockSize = 16;
alterReq.totalBlocks = 10;
alterReq.daysPerFile = 10;
alterReq.daysToKeep0 = 3650;
alterReq.daysToKeep1 = 3650;
alterReq.daysToKeep2 = 3650;
alterReq.minRows = 100;
alterReq.minRows = 4096;
alterReq.commitTime = 3600;
alterReq.fsyncPeriod = 3000;
alterReq.walLevel = 1;
alterReq.precision = 0;
alterReq.compression = 2;
alterReq.replica = 1;
alterReq.quorum = 1;
alterReq.update = 0;
alterReq.cacheLastRow = 0;
alterReq.selfIndex = 0;
for (int r = 0; r < alterReq.replica; ++r) {
SReplica* pReplica = &alterReq.replicas[r];
pReplica->id = 1;
pReplica->port = 9527;
}
int32_t contLen = tSerializeSCreateVnodeReq(NULL, 0, &alterReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSCreateVnodeReq(pReq, contLen, &alterReq);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_VNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
......@@ -312,17 +318,19 @@ TEST_F(DndTestVnode, 05_DROP_Stb) {
TEST_F(DndTestVnode, 06_Drop_Vnode) {
for (int i = 0; i < 3; ++i) {
int32_t contLen = sizeof(SDropVnodeReq);
SDropVnodeReq dropReq = {0};
dropReq.vgId = 2;
dropReq.dnodeId = 1;
strcpy(dropReq.db, "1.d1");
dropReq.dbUid = 9527;
SDropVnodeReq* pReq = (SDropVnodeReq*)rpcMallocCont(contLen);
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(1);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
int32_t contLen = tSerializeSDropVnodeReq(NULL, 0, &dropReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSDropVnodeReq(pReq, contLen, &dropReq);
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SDropVnodeReq);
rpcMsg.contLen = contLen;
rpcMsg.msgType = TDMT_DND_DROP_VNODE;
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_VNODE, pReq, contLen);
......
source/dnode/mnode/impl/inc/mndVgroup.h
浏览文件 @ 5e0a6d96
......@@ -31,8 +31,8 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
SEpSet mndGetVgroupEpset(SMnode *pMnode, SVgObj *pVgroup);
int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId);
SCreateVnodeReq *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);
SDropVnodeReq *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);
void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
#ifdef __cplusplus
}
......
source/dnode/mnode/impl/src/mndAuth.c
浏览文件 @ 5e0a6d96
......@@ -46,15 +46,25 @@ int32_t mndRetriveAuth(SMnode *pMnode, char *user, char *spi, char *encrypt, cha
}
static int32_t mndProcessAuthReq(SMnodeMsg *pReq) {
SAuthReq *pAuth = pReq->rpcMsg.pCont;
SAuthReq authReq = {0};
if (tDeserializeSAuthReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &authReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
SAuthReq authRsp = {0};
memcpy(authRsp.user, authReq.user, TSDB_USER_LEN);
int32_t contLen = sizeof(SAuthRsp);
SAuthRsp *pRsp = rpcMallocCont(contLen);
int32_t code =
mndRetriveAuth(pReq->pMnode, 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;
int32_t code = mndRetriveAuth(pReq->pMnode, pAuth->user, &pRsp->spi, &pRsp->encrypt, pRsp->secret, pRsp->ckey);
mTrace("user:%s, auth req received, spi:%d encrypt:%d ruser:%s", pReq->user, pAuth->spi, pAuth->encrypt, pAuth->user);
return code;
}
......
source/dnode/mnode/impl/src/mndBnode.c
浏览文件 @ 5e0a6d96
......@@ -187,17 +187,21 @@ static int32_t mndSetCreateBnodeCommitLogs(STrans *pTrans, SBnodeObj *pObj) {
}
static int32_t mndSetCreateBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SBnodeObj *pObj) {
SDCreateBnodeReq *pReq = malloc(sizeof(SDCreateBnodeReq));
SDCreateBnodeReq createReq = {0};
createReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDCreateBnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_CREATE_BNODE;
action.acceptableCode = TSDB_CODE_DND_BNODE_ALREADY_DEPLOYED;
......@@ -210,17 +214,21 @@ static int32_t mndSetCreateBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
}
static int32_t mndSetCreateBnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, SBnodeObj *pObj) {
SDDropBnodeReq *pReq = malloc(sizeof(SDDropBnodeReq));
SDDropBnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDDropBnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_BNODE;
action.acceptableCode = TSDB_CODE_DND_BNODE_NOT_DEPLOYED;
......@@ -329,17 +337,21 @@ static int32_t mndSetDropBnodeCommitLogs(STrans *pTrans, SBnodeObj *pObj) {
}
static int32_t mndSetDropBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SBnodeObj *pObj) {
SDDropBnodeReq *pReq = malloc(sizeof(SDDropBnodeReq));
SDDropBnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDDropBnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_BNODE;
action.acceptableCode = TSDB_CODE_DND_BNODE_NOT_DEPLOYED;
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ 5e0a6d96
......@@ -335,11 +335,12 @@ static int32_t mndSetCreateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
SCreateVnodeReq *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup);
int32_t contLen = 0;
void *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = sizeof(SCreateVnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_CREATE_VNODE;
action.acceptableCode = TSDB_CODE_DND_VNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
......@@ -365,11 +366,12 @@ static int32_t mndSetCreateDbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
SDropVnodeReq *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup);
int32_t contLen = 0;
void *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = sizeof(SDropVnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_VNODE;
action.acceptableCode = TSDB_CODE_DND_VNODE_NOT_DEPLOYED;
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
......@@ -577,11 +579,12 @@ static int32_t mndBuildUpdateVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
SAlterVnodeReq *pReq = (SAlterVnodeReq *)mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup);
int32_t contLen = 0;
void *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = sizeof(SAlterVnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_ALTER_VNODE;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pReq);
......@@ -754,11 +757,12 @@ static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
SDropVnodeReq *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup);
int32_t contLen = 0;
void *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = sizeof(SDropVnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_VNODE;
action.acceptableCode = TSDB_CODE_DND_VNODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
......
source/dnode/mnode/impl/src/mndDnode.c
浏览文件 @ 5e0a6d96
......@@ -302,7 +302,10 @@ static int32_t mndProcessStatusReq(SMnodeMsg *pReq) {
SDnodeObj *pDnode = NULL;
int32_t code = -1;
if (tDeserializeSStatusReq(pReq->rpcMsg.pCont, &statusReq) == NULL) goto PROCESS_STATUS_MSG_OVER;
if (tDeserializeSStatusReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &statusReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto PROCESS_STATUS_MSG_OVER;
}
if (statusReq.dnodeId == 0) {
pDnode = mndAcquireDnodeByEp(pMnode, statusReq.dnodeEp);
......@@ -410,10 +413,9 @@ static int32_t mndProcessStatusReq(SMnodeMsg *pReq) {
mndGetDnodeData(pMnode, statusRsp.pDnodeEps);
int32_t contLen = tSerializeSStatusRsp(NULL, &statusRsp);
int32_t contLen = tSerializeSStatusRsp(NULL, 0, &statusRsp);
void *pHead = rpcMallocCont(contLen);
void *pBuf = pHead;
tSerializeSStatusRsp(&pBuf, &statusRsp);
tSerializeSStatusRsp(pHead, contLen, &statusRsp);
taosArrayDestroy(statusRsp.pDnodeEps);
pReq->contLen = contLen;
......@@ -607,14 +609,12 @@ static int32_t mndProcessConfigDnodeReq(SMnodeMsg *pReq) {
SEpSet epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
SDCfgDnodeReq *pCfgDnode = rpcMallocCont(sizeof(SDCfgDnodeReq));
pCfgDnode->dnodeId = htonl(cfgReq.dnodeId);
memcpy(pCfgDnode->config, cfgReq.config, TSDB_DNODE_CONFIG_LEN);
int32_t bufLen = tSerializeSMCfgDnodeReq(NULL, 0, &cfgReq);
void *pBuf = rpcMallocCont(bufLen);
tSerializeSMCfgDnodeReq(pBuf, bufLen, &cfgReq);
SRpcMsg rpcMsg = {.msgType = TDMT_DND_CONFIG_DNODE,
.pCont = pCfgDnode,
.contLen = sizeof(SDCfgDnodeReq),
.ahandle = pReq->rpcMsg.ahandle};
SRpcMsg rpcMsg = {
.msgType = TDMT_DND_CONFIG_DNODE, .pCont = pBuf, .contLen = bufLen, .ahandle = pReq->rpcMsg.ahandle};
mInfo("dnode:%d, app:%p config:%s req send to dnode", cfgReq.dnodeId, rpcMsg.ahandle, cfgReq.config);
mndSendReqToDnode(pMnode, &epSet, &rpcMsg);
......
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @ 5e0a6d96
......@@ -284,8 +284,8 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
if (pIter == NULL) break;
SReplica *pReplica = &createReq.replicas[numOfReplicas];
pReplica->id = htonl(pMObj->id);
pReplica->port = htons(pMObj->pDnode->port);
pReplica->id = pMObj->id;
pReplica->port = pMObj->pDnode->port;
memcpy(pReplica->fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
numOfReplicas++;
......@@ -293,8 +293,8 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
}
SReplica *pReplica = &createReq.replicas[numOfReplicas];
pReplica->id = htonl(pDnode->id);
pReplica->port = htons(pDnode->port);
pReplica->id = pDnode->id;
pReplica->port = pDnode->port;
memcpy(pReplica->fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
numOfReplicas++;
......@@ -307,18 +307,14 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
STransAction action = {0};
SDAlterMnodeReq *pReq = malloc(sizeof(SDAlterMnodeReq));
if (pReq == NULL) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pMObj);
return -1;
}
memcpy(pReq, &createReq, sizeof(SDAlterMnodeReq));
createReq.dnodeId = pMObj->id;
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &createReq);
void *pReq = malloc(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &createReq);
pReq->dnodeId = htonl(pMObj->id);
action.epSet = mndGetDnodeEpset(pMObj->pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDAlterMnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_ALTER_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
......@@ -336,14 +332,14 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
SDCreateMnodeReq *pReq = malloc(sizeof(SDCreateMnodeReq));
if (pReq == NULL) return -1;
memcpy(pReq, &createReq, sizeof(SDAlterMnodeReq));
pReq->dnodeId = htonl(pObj->id);
createReq.dnodeId = pObj->id;
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &createReq);
void *pReq = malloc(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &createReq);
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDCreateMnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_CREATE_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
......@@ -463,8 +459,8 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
if (pMObj->id != pObj->id) {
SReplica *pReplica = &alterReq.replicas[numOfReplicas];
pReplica->id = htonl(pMObj->id);
pReplica->port = htons(pMObj->pDnode->port);
pReplica->id = pMObj->id;
pReplica->port = pMObj->pDnode->port;
memcpy(pReplica->fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
numOfReplicas++;
}
......@@ -481,18 +477,14 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
if (pMObj->id != pObj->id) {
STransAction action = {0};
SDAlterMnodeReq *pReq = malloc(sizeof(SDAlterMnodeReq));
if (pReq == NULL) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pMObj);
return -1;
}
memcpy(pReq, &alterReq, sizeof(SDAlterMnodeReq));
alterReq.dnodeId = pMObj->id;
int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, &alterReq);
void *pReq = malloc(contLen);
tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
pReq->dnodeId = htonl(pMObj->id);
action.epSet = mndGetDnodeEpset(pMObj->pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDAlterMnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_ALTER_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
......@@ -511,16 +503,15 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
SDDropMnodeReq *pReq = malloc(sizeof(SDDropMnodeReq));
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pObj->id);
SDDropMnodeReq dropReq = {0};
dropReq.dnodeId = pObj->id;
int32_t contLen = tSerializeSMCreateDropMnodeReq(NULL, 0, &dropReq);
void *pReq = malloc(contLen);
tSerializeSMCreateDropMnodeReq(pReq, contLen, &dropReq);
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDDropMnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
......
source/dnode/mnode/impl/src/mndProfile.c
浏览文件 @ 5e0a6d96
......@@ -14,16 +14,13 @@
*/
#define _DEFAULT_SOURCE
#include "tglobal.h"
#include "mndProfile.h"
//#include "mndConsumer.h"
#include "mndDb.h"
#include "mndStb.h"
#include "mndMnode.h"
#include "mndShow.h"
//#include "mndTopic.h"
#include "mndStb.h"
#include "mndUser.h"
//#include "mndVgroup.h"
#include "tglobal.h"
#define QUERY_ID_SIZE 20
#define QUERY_OBJ_ID_SIZE 18
......@@ -279,7 +276,7 @@ static int32_t mndSaveQueryStreamList(SConnObj *pConn, SHeartBeatReq *pReq) {
return TSDB_CODE_SUCCESS;
}
static SClientHbRsp* mndMqHbBuildRsp(SMnode* pMnode, SClientHbReq* pReq) {
static SClientHbRsp *mndMqHbBuildRsp(SMnode *pMnode, SClientHbReq *pReq) {
#if 0
SClientHbRsp* pRsp = malloc(sizeof(SClientHbRsp));
if (pRsp == NULL) {
......@@ -353,14 +350,13 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
return -1;
}
SArray *pArray = batchReq.reqs;
int32_t sz = taosArrayGetSize(pArray);
SClientHbBatchRsp batchRsp = {0};
batchRsp.rsps = taosArrayInit(0, sizeof(SClientHbRsp));
int32_t sz = taosArrayGetSize(batchReq.reqs);
for (int i = 0; i < sz; i++) {
SClientHbReq *pHbReq = taosArrayGet(pArray, i);
SClientHbReq *pHbReq = taosArrayGet(batchReq.reqs, i);
if (pHbReq->connKey.hbType == HEARTBEAT_TYPE_QUERY) {
int32_t kvNum = taosHashGetSize(pHbReq->info);
if (NULL == pHbReq->info || kvNum <= 0) {
......@@ -412,7 +408,7 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
}
}
}
taosArrayDestroyEx(pArray, tFreeClientHbReq);
taosArrayDestroyEx(batchReq.reqs, tFreeClientHbReq);
int32_t tlen = tSerializeSClientHbBatchRsp(NULL, 0, &batchRsp);
void *buf = rpcMallocCont(tlen);
......@@ -520,19 +516,22 @@ static int32_t mndProcessKillQueryReq(SMnodeMsg *pReq) {
}
mndReleaseUser(pMnode, pUser);
SKillQueryReq *pKill = pReq->rpcMsg.pCont;
int32_t connId = htonl(pKill->connId);
int32_t queryId = htonl(pKill->queryId);
mInfo("kill query msg is received, queryId:%d", pKill->queryId);
SKillQueryReq killReq = {0};
if (tDeserializeSKillQueryReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
mInfo("kill query msg is received, queryId:%d", killReq.queryId);
SConnObj *pConn = taosCacheAcquireByKey(pMgmt->cache, &connId, sizeof(int32_t));
SConnObj *pConn = taosCacheAcquireByKey(pMgmt->cache, &killReq.connId, sizeof(int32_t));
if (pConn == NULL) {
mError("connId:%d, failed to kill queryId:%d, conn not exist", connId, queryId);
mError("connId:%d, failed to kill queryId:%d, conn not exist", killReq.connId, killReq.queryId);
terrno = TSDB_CODE_MND_INVALID_CONN_ID;
return -1;
} else {
mInfo("connId:%d, queryId:%d is killed by user:%s", connId, queryId, pReq->user);
pConn->queryId = queryId;
mInfo("connId:%d, queryId:%d is killed by user:%s", killReq.connId, killReq.queryId, pReq->user);
pConn->queryId = killReq.queryId;
taosCacheRelease(pMgmt->cache, (void **)&pConn, false);
return 0;
}
......@@ -551,16 +550,19 @@ static int32_t mndProcessKillConnReq(SMnodeMsg *pReq) {
}
mndReleaseUser(pMnode, pUser);
SKillConnReq *pKill = pReq->rpcMsg.pCont;
int32_t connId = htonl(pKill->connId);
SKillConnReq killReq = {0};
if (tDeserializeSKillConnReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
SConnObj *pConn = taosCacheAcquireByKey(pMgmt->cache, &connId, sizeof(int32_t));
SConnObj *pConn = taosCacheAcquireByKey(pMgmt->cache, &killReq.connId, sizeof(int32_t));
if (pConn == NULL) {
mError("connId:%d, failed to kill connection, conn not exist", connId);
mError("connId:%d, failed to kill connection, conn not exist", killReq.connId);
terrno = TSDB_CODE_MND_INVALID_CONN_ID;
return -1;
} else {
mInfo("connId:%d, is killed by user:%s", connId, pReq->user);
mInfo("connId:%d, is killed by user:%s", killReq.connId, pReq->user);
pConn->killed = 1;
taosCacheRelease(pMgmt->cache, (void **)&pConn, false);
return TSDB_CODE_SUCCESS;
......
source/dnode/mnode/impl/src/mndQnode.c
浏览文件 @ 5e0a6d96
......@@ -187,17 +187,21 @@ static int32_t mndSetCreateQnodeCommitLogs(STrans *pTrans, SQnodeObj *pObj) {
}
static int32_t mndSetCreateQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQnodeObj *pObj) {
SDCreateQnodeReq *pReq = malloc(sizeof(SDCreateQnodeReq));
SDCreateQnodeReq createReq = {0};
createReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDCreateQnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_CREATE_QNODE;
action.acceptableCode = TSDB_CODE_DND_QNODE_ALREADY_DEPLOYED;
......@@ -210,17 +214,21 @@ static int32_t mndSetCreateQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
}
static int32_t mndSetCreateQnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, SQnodeObj *pObj) {
SDDropQnodeReq *pReq = malloc(sizeof(SDDropQnodeReq));
SDDropQnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDDropQnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_QNODE;
action.acceptableCode = TSDB_CODE_DND_QNODE_NOT_DEPLOYED;
......@@ -329,17 +337,21 @@ static int32_t mndSetDropQnodeCommitLogs(STrans *pTrans, SQnodeObj *pObj) {
}
static int32_t mndSetDropQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQnodeObj *pObj) {
SDDropQnodeReq *pReq = malloc(sizeof(SDDropQnodeReq));
SDDropQnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDDropQnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_QNODE;
action.acceptableCode = TSDB_CODE_DND_QNODE_NOT_DEPLOYED;
......
source/dnode/mnode/impl/src/mndSnode.c
浏览文件 @ 5e0a6d96
......@@ -187,17 +187,21 @@ static int32_t mndSetCreateSnodeCommitLogs(STrans *pTrans, SSnodeObj *pObj) {
}
static int32_t mndSetCreateSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SSnodeObj *pObj) {
SDCreateSnodeReq *pReq = malloc(sizeof(SDCreateSnodeReq));
SDCreateSnodeReq createReq = {0};
createReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &createReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &createReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDCreateSnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_CREATE_SNODE;
action.acceptableCode = TSDB_CODE_DND_SNODE_ALREADY_DEPLOYED;
......@@ -210,17 +214,21 @@ static int32_t mndSetCreateSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
}
static int32_t mndSetCreateSnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, SSnodeObj *pObj) {
SDDropSnodeReq *pReq = malloc(sizeof(SDDropSnodeReq));
SDDropSnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDDropSnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_SNODE;
action.acceptableCode = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
......@@ -302,7 +310,7 @@ static int32_t mndProcessCreateSnodeReq(SMnodeMsg *pReq) {
if (code == 0) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
CREATE_SNODE_OVER:
if(code != 0 && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
if (code != 0 && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
mError("snode:%d, failed to create since %s", createReq.dnodeId, terrstr());
return -1;
}
......@@ -331,17 +339,21 @@ static int32_t mndSetDropSnodeCommitLogs(STrans *pTrans, SSnodeObj *pObj) {
}
static int32_t mndSetDropSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SSnodeObj *pObj) {
SDDropSnodeReq *pReq = malloc(sizeof(SDDropSnodeReq));
SDDropSnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
int32_t contLen = tSerializeSMCreateDropQSBNodeReq(NULL, 0, &dropReq);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pReq->dnodeId = htonl(pDnode->id);
tSerializeSMCreateDropQSBNodeReq(pReq, contLen, &dropReq);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pReq;
action.contLen = sizeof(SDDropSnodeReq);
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_SNODE;
action.acceptableCode = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
......
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @ 5e0a6d96
......@@ -189,78 +189,95 @@ void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
sdbRelease(pSdb, pVgroup);
}
SCreateVnodeReq *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
SCreateVnodeReq *pCreate = calloc(1, sizeof(SCreateVnodeReq));
if (pCreate == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pCreate->vgId = htonl(pVgroup->vgId);
pCreate->dnodeId = htonl(pDnode->id);
memcpy(pCreate->db, pDb->name, TSDB_DB_FNAME_LEN);
pCreate->dbUid = htobe64(pDb->uid);
pCreate->vgVersion = htonl(pVgroup->version);
pCreate->cacheBlockSize = htonl(pDb->cfg.cacheBlockSize);
pCreate->totalBlocks = htonl(pDb->cfg.totalBlocks);
pCreate->daysPerFile = htonl(pDb->cfg.daysPerFile);
pCreate->daysToKeep0 = htonl(pDb->cfg.daysToKeep0);
pCreate->daysToKeep1 = htonl(pDb->cfg.daysToKeep1);
pCreate->daysToKeep2 = htonl(pDb->cfg.daysToKeep2);
pCreate->minRows = htonl(pDb->cfg.minRows);
pCreate->maxRows = htonl(pDb->cfg.maxRows);
pCreate->commitTime = htonl(pDb->cfg.commitTime);
pCreate->fsyncPeriod = htonl(pDb->cfg.fsyncPeriod);
pCreate->walLevel = pDb->cfg.walLevel;
pCreate->precision = pDb->cfg.precision;
pCreate->compression = pDb->cfg.compression;
pCreate->quorum = pDb->cfg.quorum;
pCreate->update = pDb->cfg.update;
pCreate->cacheLastRow = pDb->cfg.cacheLastRow;
pCreate->replica = pVgroup->replica;
pCreate->selfIndex = -1;
void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
SCreateVnodeReq createReq = {0};
createReq.vgId = pVgroup->vgId;
createReq.dnodeId = pDnode->id;
memcpy(createReq.db, pDb->name, TSDB_DB_FNAME_LEN);
createReq.dbUid = pDb->uid;
createReq.vgVersion = pVgroup->version;
createReq.cacheBlockSize = pDb->cfg.cacheBlockSize;
createReq.totalBlocks = pDb->cfg.totalBlocks;
createReq.daysPerFile = pDb->cfg.daysPerFile;
createReq.daysToKeep0 = pDb->cfg.daysToKeep0;
createReq.daysToKeep1 = pDb->cfg.daysToKeep1;
createReq.daysToKeep2 = pDb->cfg.daysToKeep2;
createReq.minRows = pDb->cfg.minRows;
createReq.maxRows = pDb->cfg.maxRows;
createReq.commitTime = pDb->cfg.commitTime;
createReq.fsyncPeriod = pDb->cfg.fsyncPeriod;
createReq.walLevel = pDb->cfg.walLevel;
createReq.precision = pDb->cfg.precision;
createReq.compression = pDb->cfg.compression;
createReq.quorum = pDb->cfg.quorum;
createReq.update = pDb->cfg.update;
createReq.cacheLastRow = pDb->cfg.cacheLastRow;
createReq.replica = pVgroup->replica;
createReq.selfIndex = -1;
for (int32_t v = 0; v < pVgroup->replica; ++v) {
SReplica *pReplica = &pCreate->replicas[v];
SReplica *pReplica = &createReq.replicas[v];
SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
SDnodeObj *pVgidDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pVgidDnode == NULL) {
free(pCreate);
return NULL;
}
pReplica->id = htonl(pVgidDnode->id);
pReplica->port = htons(pVgidDnode->port);
pReplica->id = pVgidDnode->id;
pReplica->port = pVgidDnode->port;
memcpy(pReplica->fqdn, pVgidDnode->fqdn, TSDB_FQDN_LEN);
mndReleaseDnode(pMnode, pVgidDnode);
if (pDnode->id == pVgid->dnodeId) {
pCreate->selfIndex = v;
createReq.selfIndex = v;
}
}
if (pCreate->selfIndex == -1) {
free(pCreate);
if (createReq.selfIndex == -1) {
terrno = TSDB_CODE_MND_APP_ERROR;
return NULL;
}
return pCreate;
int32_t contLen = tSerializeSCreateVnodeReq(NULL, 0, &createReq);
if (contLen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
tSerializeSCreateVnodeReq(pReq, contLen, &createReq);
*pContLen = contLen;
return pReq;
}
SDropVnodeReq *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
SDropVnodeReq *pDrop = calloc(1, sizeof(SDropVnodeReq));
if (pDrop == NULL) {
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup,
int32_t *pContLen) {
SDropVnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
dropReq.vgId = pVgroup->vgId;
memcpy(dropReq.db, pDb->name, TSDB_DB_FNAME_LEN);
dropReq.dbUid = pDb->uid;
int32_t contLen = tSerializeSDropVnodeReq(NULL, 0, &dropReq);
if (contLen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pDrop->dnodeId = htonl(pDnode->id);
pDrop->vgId = htonl(pVgroup->vgId);
memcpy(pDrop->db, pDb->name, TSDB_DB_FNAME_LEN);
pDrop->dbUid = htobe64(pDb->uid);
void *pReq = malloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
return pDrop;
tSerializeSDropVnodeReq(pReq, contLen, &dropReq);
*pContLen = contLen;
return pReq;
}
static bool mndResetDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
......
source/dnode/mnode/impl/src/mnode.c
浏览文件 @ 5e0a6d96
......@@ -60,11 +60,25 @@ void mndSendRedirectRsp(SMnode *pMnode, SRpcMsg *pMsg) {
}
}
static void *mndBuildTimerMsg(int32_t *pContLen) {
SMTimerReq timerReq = {0};
int32_t contLen = tSerializeSMTimerMsg(NULL, 0, &timerReq);
if (contLen <= 0) return NULL;
void *pReq = rpcMallocCont(contLen);
if (pReq == NULL) return NULL;
tSerializeSMTimerMsg(pReq, contLen, &timerReq);
*pContLen = contLen;
return pReq;
}
static void mndTransReExecute(void *param, void *tmrId) {
SMnode *pMnode = param;
if (mndIsMaster(pMnode)) {
STransReq *pMsg = rpcMallocCont(sizeof(STransReq));
SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS, .pCont = pMsg, .contLen = sizeof(STransReq)};
int32_t contLen = 0;
void *pReq = mndBuildTimerMsg(&contLen);
SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS, .pCont = pReq, .contLen = contLen};
pMnode->putReqToMWriteQFp(pMnode->pDnode, &rpcMsg);
}
......@@ -74,8 +88,9 @@ static void mndTransReExecute(void *param, void *tmrId) {
static void mndCalMqRebalance(void *param, void *tmrId) {
SMnode *pMnode = param;
if (mndIsMaster(pMnode)) {
SMqTmrMsg *pMsg = rpcMallocCont(sizeof(SMqTmrMsg));
SRpcMsg rpcMsg = {.msgType = TDMT_MND_MQ_TIMER, .pCont = pMsg, .contLen = sizeof(SMqTmrMsg)};
int32_t contLen = 0;
void *pReq = mndBuildTimerMsg(&contLen);
SRpcMsg rpcMsg = {.msgType = TDMT_MND_MQ_TIMER, .pCont = pReq, .contLen = contLen};
pMnode->putReqToMReadQFp(pMnode->pDnode, &rpcMsg);
}
......
source/dnode/mnode/impl/test/profile/profile.cpp
浏览文件 @ 5e0a6d96
......@@ -167,10 +167,12 @@ TEST_F(MndTestProfile, 05_KillConnMsg) {
// temporary remove since kill will use new heartbeat msg
#if 0
{
int32_t contLen = sizeof(SKillConnReq);
SKillConnReq killReq = {0};
killReq.connId = connId;
SKillConnReq* pReq = (SKillConnReq*)rpcMallocCont(contLen);
pReq->connId = htonl(connId);
int32_t contLen = tSerializeSKillConnReq(NULL, 0, &killReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSKillConnReq(pReq, contLen, &killReq);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_KILL_CONN, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -226,10 +228,12 @@ TEST_F(MndTestProfile, 05_KillConnMsg) {
}
TEST_F(MndTestProfile, 06_KillConnMsg_InvalidConn) {
int32_t contLen = sizeof(SKillConnReq);
SKillConnReq killReq = {0};
killReq.connId = 2345;
SKillConnReq* pReq = (SKillConnReq*)rpcMallocCont(contLen);
pReq->connId = htonl(2345);
int32_t contLen = tSerializeSKillConnReq(NULL, 0, &killReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSKillConnReq(pReq, contLen, &killReq);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_KILL_CONN, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -240,11 +244,13 @@ TEST_F(MndTestProfile, 07_KillQueryMsg) {
// temporary remove since kill will use new heartbeat msg
#if 0
{
int32_t contLen = sizeof(SKillQueryReq);
SKillQueryReq killReq = {0};
killReq.connId = connId;
killReq.queryId = 1234;
SKillQueryReq* pReq = (SKillQueryReq*)rpcMallocCont(contLen);
pReq->connId = htonl(connId);
pReq->queryId = htonl(1234);
int32_t contLen = tSerializeSKillQueryReq(NULL, 0, &killReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSKillQueryReq(pReq, contLen, &killReq);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_KILL_QUERY, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -291,11 +297,13 @@ TEST_F(MndTestProfile, 07_KillQueryMsg) {
}
TEST_F(MndTestProfile, 08_KillQueryMsg_InvalidConn) {
int32_t contLen = sizeof(SKillQueryReq);
SKillQueryReq killReq = {0};
killReq.connId = 2345;
killReq.queryId = 2345;
SKillQueryReq* pReq = (SKillQueryReq*)rpcMallocCont(contLen);
pReq->connId = htonl(2345);
pReq->queryId = htonl(1234);
int32_t contLen = tSerializeSKillQueryReq(NULL, 0, &killReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSKillQueryReq(pReq, contLen, &killReq);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_KILL_QUERY, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......
source/dnode/vnode/inc/tsdb.h
浏览文件 @ 5e0a6d96
......@@ -83,7 +83,7 @@ typedef struct {
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta, STfs *pTfs);
void tsdbClose(STsdb *);
void tsdbRemove(const char *path);
int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg, SSubmitRsp *pRsp);
int tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp);
int tsdbPrepareCommit(STsdb *pTsdb);
int tsdbCommit(STsdb *pTsdb);
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 5e0a6d96
......@@ -72,7 +72,7 @@ typedef struct {
int64_t ver;
int64_t tbUid;
SHashObj *tbIdHash;
const SSubmitMsg *pMsg;
const SSubmitReq *pMsg;
SSubmitBlk *pBlock;
SSubmitMsgIter msgIter;
SSubmitBlkIter blkIter;
......@@ -225,7 +225,7 @@ static FORCE_INLINE int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const S
return 0;
}
void tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitMsg *pMsg, int64_t ver);
void tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReadHandle *pHandle);
int tqRetrieveDataBlockInfo(STqReadHandle *pHandle, SDataBlockInfo *pBlockInfo);
// return SArray<SColumnInfoData>
......
source/dnode/vnode/src/inc/tsdbMemTable.h
浏览文件 @ 5e0a6d96
......@@ -54,7 +54,7 @@ typedef struct STsdbMemTable {
STsdbMemTable *tsdbNewMemTable(STsdb *pTsdb);
void tsdbFreeMemTable(STsdb *pTsdb, STsdbMemTable *pMemTable);
int tsdbMemTableInsert(STsdb *pTsdb, STsdbMemTable *pMemTable, SSubmitMsg *pMsg, SShellSubmitRsp *pRsp);
int tsdbMemTableInsert(STsdb *pTsdb, STsdbMemTable *pMemTable, SSubmitReq *pMsg, SSubmitRsp *pRsp);
int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols,
TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 5e0a6d96
......@@ -262,7 +262,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
}
pHead = pTopic->pReadhandle->pHead;
if (pHead->head.msgType == TDMT_VND_SUBMIT) {
SSubmitMsg* pCont = (SSubmitMsg*)&pHead->head.body;
SSubmitReq* pCont = (SSubmitReq*)&pHead->head.body;
qTaskInfo_t task = pTopic->buffer.output[pos].task;
qSetStreamInput(task, pCont);
SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
......@@ -409,7 +409,7 @@ int32_t tqProcessConsumeReqV0(STQ* pTq, SRpcMsg* pMsg) {
fetchOffset++;
}
if (skip == 1) continue;
SSubmitMsg* pCont = (SSubmitMsg*)&pHead->head.body;
SSubmitReq* pCont = (SSubmitReq*)&pHead->head.body;
qTaskInfo_t task = pTopic->buffer.output[pos].task;
printf("current fetch offset %ld\n", fetchOffset);
......
source/dnode/vnode/src/tq/tqRead.c
浏览文件 @ 5e0a6d96
......@@ -31,7 +31,7 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
return pReadHandle;
}
void tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitMsg* pMsg, int64_t ver) {
void tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t ver) {
pReadHandle->pMsg = pMsg;
pMsg->length = htonl(pMsg->length);
pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
......
source/dnode/vnode/src/tsdb/tsdbMemTable.c
浏览文件 @ 5e0a6d96
......@@ -15,7 +15,7 @@
#include "tsdbDef.h"
static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitMsg *pMsg);
static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg);
static int tsdbMemTableInsertTbData(STsdb *pRepo, SSubmitBlk *pBlock, int32_t *pAffectedRows);
static STbData *tsdbNewTbData(tb_uid_t uid);
static void tsdbFreeTbData(STbData *pTbData);
......@@ -73,7 +73,7 @@ void tsdbFreeMemTable(STsdb *pTsdb, STsdbMemTable *pMemTable) {
}
}
int tsdbMemTableInsert(STsdb *pTsdb, STsdbMemTable *pMemTable, SSubmitMsg *pMsg, SShellSubmitRsp *pRsp) {
int tsdbMemTableInsert(STsdb *pTsdb, STsdbMemTable *pMemTable, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
SSubmitBlk * pBlock = NULL;
SSubmitMsgIter msgIter = {0};
int32_t affectedrows = 0, numOfRows = 0;
......@@ -227,7 +227,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
return 0;
}
static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitMsg *pMsg) {
static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
ASSERT(pMsg != NULL);
// STsdbMeta * pMeta = pTsdb->tsdbMeta;
SSubmitMsgIter msgIter = {0};
......@@ -455,7 +455,7 @@ static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema *
/* ------------------------ REFACTORING ------------------------ */
#if 0
int tsdbInsertDataToMemTable(STsdbMemTable *pMemTable, SSubmitMsg *pMsg) {
int tsdbInsertDataToMemTable(STsdbMemTable *pMemTable, SSubmitReq *pMsg) {
SMemAllocator *pMA = pMemTable->pMA;
STbData * pTbData = (STbData *)TD_MA_MALLOC(pMA, sizeof(*pTbData));
if (pTbData == NULL) {
......@@ -496,9 +496,9 @@ static int tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables);
static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow* row);
static int tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter);
static STSRow* tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter);
static int tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg);
static int tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitReq *pMsg);
static int tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, int32_t *affectedrows);
static int tsdbInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter);
static int tsdbInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter);
static int tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock);
static int tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable);
static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, STSRow* row);
......
source/dnode/vnode/src/tsdb/tsdbWrite.c
浏览文件 @ 5e0a6d96
......@@ -15,7 +15,7 @@
#include "tsdbDef.h"
int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg, SSubmitRsp *pRsp) {
int tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
// Check if mem is there. If not, create one.
if (pTsdb->mem == NULL) {
pTsdb->mem = tsdbNewMemTable(pTsdb);
......
source/dnode/vnode/src/vnd/vnodeWrite.c
浏览文件 @ 5e0a6d96
......@@ -109,7 +109,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
// }
break;
case TDMT_VND_SUBMIT:
if (tsdbInsertData(pVnode->pTsdb, (SSubmitMsg *)ptr, NULL) < 0) {
if (tsdbInsertData(pVnode->pTsdb, (SSubmitReq *)ptr, NULL) < 0) {
// TODO: handle error
}
break;
......
source/libs/parser/src/dataBlockMgt.c
浏览文件 @ 5e0a6d96
......@@ -417,7 +417,7 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, SB
}
int32_t mergeTableDataBlocks(SHashObj* pHashObj, int8_t schemaAttached, uint8_t payloadType, SArray** pVgDataBlocks) {
const int INSERT_HEAD_SIZE = sizeof(SSubmitMsg);
const int INSERT_HEAD_SIZE = sizeof(SSubmitReq);
int code = 0;
bool isRawPayload = IS_RAW_PAYLOAD(payloadType);
SHashObj* pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
......
source/libs/parser/src/insertParser.c
浏览文件 @ 5e0a6d96
......@@ -121,7 +121,7 @@ static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pS
}
static void buildMsgHeader(SVgDataBlocks* blocks) {
SSubmitMsg* submit = (SSubmitMsg*)blocks->pData;
SSubmitReq* submit = (SSubmitReq*)blocks->pData;
submit->header.vgId = htonl(blocks->vg.vgId);
submit->header.contLen = htonl(blocks->size);
submit->length = submit->header.contLen;
......
source/libs/parser/test/insertParserTest.cpp
浏览文件 @ 5e0a6d96
......@@ -70,7 +70,7 @@ protected:
for (size_t i = 0; i < num; ++i) {
SVgDataBlocks* vg = (SVgDataBlocks*)taosArrayGetP(res_->pDataBlocks, i);
cout << "vgId:" << vg->vg.vgId << ", numOfTables:" << vg->numOfTables << ", dataSize:" << vg->size << endl;
SSubmitMsg* submit = (SSubmitMsg*)vg->pData;
SSubmitReq* submit = (SSubmitReq*)vg->pData;
cout << "length:" << ntohl(submit->length) << ", numOfBlocks:" << ntohl(submit->numOfBlocks) << endl;
int32_t numOfBlocks = ntohl(submit->numOfBlocks);
SSubmitBlk* blk = (SSubmitBlk*)(submit + 1);
......@@ -93,7 +93,7 @@ protected:
SVgDataBlocks* vg = (SVgDataBlocks*)taosArrayGetP(res_->pDataBlocks, i);
ASSERT_EQ(vg->numOfTables, numOfTables);
ASSERT_GE(vg->size, 0);
SSubmitMsg* submit = (SSubmitMsg*)vg->pData;
SSubmitReq* submit = (SSubmitReq*)vg->pData;
ASSERT_GE(ntohl(submit->length), 0);
ASSERT_GE(ntohl(submit->numOfBlocks), 0);
int32_t numOfBlocks = ntohl(submit->numOfBlocks);
......
source/libs/scheduler/src/scheduler.c
浏览文件 @ 5e0a6d96
......@@ -823,7 +823,7 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
SCH_ERR_RET(schProcessOnTaskFailure(pJob, pTask, rspCode));
}
SShellSubmitRsp *rsp = (SShellSubmitRsp *)msg;
SSubmitRsp *rsp = (SSubmitRsp *)msg;
if (rsp) {
pJob->resNumOfRows += rsp->affectedRows;
}
......
source/libs/scheduler/test/schedulerTests.cpp
浏览文件 @ 5e0a6d96
......@@ -273,7 +273,7 @@ void *schtSendRsp(void *param) {
while (pIter) {
SSchTask *task = *(SSchTask **)pIter;
SShellSubmitRsp rsp = {0};
SSubmitRsp rsp = {0};
rsp.affectedRows = 10;
schHandleResponseMsg(job, task, TDMT_VND_SUBMIT_RSP, (char *)&rsp, sizeof(rsp), 0);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或