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

serialize node msg

This commit is contained in:
Shengliang Guan 2022-02-16 17:33:07 +08:00
parent bec7a0d533
commit f0e697479f
13 changed files with 267 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/client/src/clientHb.c
View File

@ -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;

22
source/dnode/mgmt/impl/src/dndBnode.c
View File

@ -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;

22
source/dnode/mgmt/impl/src/dndQnode.c
View File

@ -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;

22
source/dnode/mgmt/impl/src/dndSnode.c
View File

@ -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;

73
source/dnode/mgmt/impl/test/bnode/dbnode.cpp
View File

@ -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);

72
source/dnode/mgmt/impl/test/qnode/dqnode.cpp
View File

@ -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);

72
source/dnode/mgmt/impl/test/snode/dsnode.cpp
View File

@ -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);

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

@ -47,7 +47,7 @@ int32_t mndRetriveAuth(SMnode *pMnode, char *user, char *spi, char *encrypt, cha
static int32_t mndProcessAuthReq(SMnodeMsg *pReq) {
SAuthReq authReq = {0};
if (tDeserializeSAuthReq(pReq->pCont, pReq->contLen, &authReq) != 0) {
if (tDeserializeSAuthReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &authReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}

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

@ -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;

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

@ -517,7 +517,7 @@ static int32_t mndProcessKillQueryReq(SMnodeMsg *pReq) {
mndReleaseUser(pMnode, pUser);
SKillQueryReq killReq = {0};
if (tDeserializeSKillQueryReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
if (tDeserializeSKillQueryReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
@ -551,7 +551,7 @@ static int32_t mndProcessKillConnReq(SMnodeMsg *pReq) {
mndReleaseUser(pMnode, pUser);
SKillConnReq killReq = {0};
if (tDeserializeSKillConnReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
if (tDeserializeSKillConnReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}

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

@ -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;

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

@ -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;

36
source/dnode/mnode/impl/test/profile/profile.cpp
View File

@ -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);