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

pullup trans when mnode restart

This commit is contained in:
Shengliang Guan 2022-01-05 20:22:33 -08:00
parent 3c3f32bdae
commit db2dbe5f51
7 changed files with 87 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -64,8 +64,8 @@ typedef enum {
TRN_STAGE_PREPARE = 0,
TRN_STAGE_REDO_LOG = 1,
TRN_STAGE_REDO_ACTION = 2,
TRN_STAGE_COMMIT_LOG = 3,
TRN_STAGE_COMMIT = 4,
TRN_STAGE_COMMIT = 3,
TRN_STAGE_COMMIT_LOG = 4,
TRN_STAGE_UNDO_ACTION = 5,
TRN_STAGE_UNDO_LOG = 6,
TRN_STAGE_ROLLBACK = 7,

3
source/dnode/mnode/impl/src/mndSync.c
View File

@ -54,7 +54,7 @@ static int32_t mndRestoreWal(SMnode *pMnode) {
int64_t first = walGetFirstVer(pWal);
int64_t last = walGetLastVer(pWal);
mDebug("restore sdb wal start, sdb ver:%" PRId64 ", wal first:%" PRId64 " last:%" PRId64, lastSdbVer, first, last);
mDebug("start to restore sdb wal, sdb ver:%" PRId64 ", wal first:%" PRId64 " last:%" PRId64, lastSdbVer, first, last);
first = MAX(lastSdbVer + 1, first);
for (int64_t ver = first; ver >= 0 && ver <= last; ++ver) {
@ -71,6 +71,7 @@ static int32_t mndRestoreWal(SMnode *pMnode) {
goto WAL_RESTORE_OVER;
}
mTrace("wal:%" PRId64 ", will be restored, content:%p", ver, pHead->head.body);
if (sdbWriteNotFree(pSdb, (void *)pHead->head.body) < 0) {
mError("failed to read wal from sdb since %s, ver:%" PRId64, terrstr(), ver);
goto WAL_RESTORE_OVER;

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

@ -112,6 +112,7 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
int32_t dataPos = 0;
SDB_SET_INT32(pRaw, dataPos, pTrans->id, TRANS_ENCODE_OVER)
SDB_SET_INT8(pRaw, dataPos, pTrans->policy, TRANS_ENCODE_OVER)
SDB_SET_INT8(pRaw, dataPos, pTrans->stage, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, redoLogNum, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, undoLogNum, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, commitLogNum, TRANS_ENCODE_OVER)
@ -216,6 +217,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32(pRaw, dataPos, &pTrans->id, TRANS_DECODE_OVER)
SDB_GET_INT8(pRaw, dataPos, (int8_t *)&pTrans->policy, TRANS_DECODE_OVER)
SDB_GET_INT8(pRaw, dataPos, (int8_t *)&pTrans->stage, TRANS_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &redoLogNum, TRANS_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &undoLogNum, TRANS_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &commitLogNum, TRANS_DECODE_OVER)
@ -315,6 +317,8 @@ static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans) {
}
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
if (pNew->stage == TRN_STAGE_COMMIT) pNew->stage = TRN_STAGE_COMMIT_LOG;
mTrace("trans:%d, perform update action, old row:%p stage:%d, new row:%p stage:%d", pOld->id, pOld, pOld->stage, pNew,
pNew->stage);
pOld->stage = pNew->stage;
@ -646,7 +650,7 @@ static int32_t mndTransSendActionMsg(SMnode *pMnode, STrans *pTrans, SArray *pAr
pAction->msgReceived = 0;
pAction->errCode = 0;
} else {
mDebug("trans:%d, action:%d not sent since %s", pTrans->id, action, terrstr());
mDebug("trans:%d, action:%d not send since %s", pTrans->id, action, terrstr());
return -1;
}
}

8
source/dnode/mnode/impl/test/bnode/bnode.cpp
View File

@ -186,7 +186,6 @@ TEST_F(MndTestBnode, 03_Create_Bnode_Rollback) {
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
@ -199,7 +198,6 @@ TEST_F(MndTestBnode, 03_Create_Bnode_Rollback) {
SMDropBnodeReq* pReq = (SMDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
@ -211,7 +209,7 @@ TEST_F(MndTestBnode, 03_Create_Bnode_Rollback) {
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateBnodeReq);
@ -250,7 +248,6 @@ TEST_F(MndTestBnode, 04_Drop_Bnode_Rollback) {
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
@ -263,7 +260,6 @@ TEST_F(MndTestBnode, 04_Drop_Bnode_Rollback) {
SMDropBnodeReq* pReq = (SMDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
@ -275,7 +271,7 @@ TEST_F(MndTestBnode, 04_Drop_Bnode_Rollback) {
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateBnodeReq);

8
source/dnode/mnode/impl/test/qnode/qnode.cpp
View File

@ -186,7 +186,6 @@ TEST_F(MndTestQnode, 03_Create_Qnode_Rollback) {
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
@ -199,7 +198,6 @@ TEST_F(MndTestQnode, 03_Create_Qnode_Rollback) {
SMDropQnodeReq* pReq = (SMDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
@ -211,7 +209,7 @@ TEST_F(MndTestQnode, 03_Create_Qnode_Rollback) {
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateQnodeReq);
@ -250,7 +248,6 @@ TEST_F(MndTestQnode, 04_Drop_Qnode_Rollback) {
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
@ -263,7 +260,6 @@ TEST_F(MndTestQnode, 04_Drop_Qnode_Rollback) {
SMDropQnodeReq* pReq = (SMDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
@ -275,7 +271,7 @@ TEST_F(MndTestQnode, 04_Drop_Qnode_Rollback) {
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateQnodeReq);

8
source/dnode/mnode/impl/test/snode/snode.cpp
View File

@ -186,7 +186,6 @@ TEST_F(MndTestSnode, 03_Create_Snode_Rollback) {
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
@ -199,7 +198,6 @@ TEST_F(MndTestSnode, 03_Create_Snode_Rollback) {
SMDropSnodeReq* pReq = (SMDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
@ -211,7 +209,7 @@ TEST_F(MndTestSnode, 03_Create_Snode_Rollback) {
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateSnodeReq);
@ -250,7 +248,6 @@ TEST_F(MndTestSnode, 04_Drop_Snode_Rollback) {
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
@ -263,7 +260,6 @@ TEST_F(MndTestSnode, 04_Drop_Snode_Rollback) {
SMDropSnodeReq* pReq = (SMDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
@ -275,7 +271,7 @@ TEST_F(MndTestSnode, 04_Drop_Snode_Rollback) {
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateSnodeReq);

75
source/dnode/mnode/impl/test/trans/trans.cpp
View File

@ -13,8 +13,18 @@
class MndTestTrans : public ::testing::Test {
protected:
static void SetUpTestSuite() { test.Init("/tmp/mnode_test_trans", 9013); }
static void TearDownTestSuite() { test.Cleanup(); }
static void SetUpTestSuite() {
test.Init("/tmp/mnode_test_trans", 9013);
const char* fqdn = "localhost";
const char* firstEp = "localhost:9013";
server2.Start("/tmp/mnode_test_trans2", fqdn, 9020, firstEp);
}
static void TearDownTestSuite() {
server2.Stop();
test.Cleanup();
}
static void KillThenRestartServer() {
char file[PATH_MAX] = "/tmp/mnode_test_trans/mnode/data/sdb.data";
FileFd fd = taosOpenFileRead(file);
@ -41,6 +51,7 @@ class MndTestTrans : public ::testing::Test {
}
static Testbase test;
static TestServer server2;
public:
void SetUp() override {}
@ -48,6 +59,7 @@ class MndTestTrans : public ::testing::Test {
};
Testbase MndTestTrans::test;
TestServer MndTestTrans::server2;
TEST_F(MndTestTrans, 01_Create_User_Crash) {
{
@ -84,7 +96,7 @@ TEST_F(MndTestTrans, 01_Create_User_Crash) {
CheckBinary("root", TSDB_USER_LEN);
}
TEST_F(MndTestTrans, 02_Create_Qnode_Crash) {
TEST_F(MndTestTrans, 02_Create_Qnode1_Crash) {
{
int32_t contLen = sizeof(SMCreateQnodeReq);
@ -102,9 +114,60 @@ TEST_F(MndTestTrans, 02_Create_Qnode_Crash) {
}
KillThenRestartServer();
{
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_MND_QNODE_ALREADY_EXIST);
test.SendShowMetaReq(TSDB_MGMT_TABLE_QNODE, "");
CHECK_META("show qnodes", 3);
test.SendShowRetrieveReq();
EXPECT_EQ(test.GetShowRows(), 1);
}
}
TEST_F(MndTestTrans, 03_Create_Qnode2_Crash) {
{
int32_t contLen = sizeof(SCreateDnodeMsg);
SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
strcpy(pReq->fqdn, "localhost");
pReq->port = htonl(9020);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_DNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
taosMsleep(1300);
test.SendShowMetaReq(TSDB_MGMT_TABLE_DNODE, "");
test.SendShowRetrieveReq();
EXPECT_EQ(test.GetShowRows(), 2);
}
{
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
KillThenRestartServer();
server2.DoStart();
{
int32_t retry = 0;
int32_t retryMax = 10;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateQnodeReq);
@ -118,9 +181,11 @@ TEST_F(MndTestTrans, 02_Create_Qnode_Crash) {
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
test.SendShowMetaReq(TSDB_MGMT_TABLE_QNODE, "");
CHECK_META("show qnodes", 3);
test.SendShowRetrieveReq();
EXPECT_EQ(test.GetShowRows(), 1);
EXPECT_EQ(test.GetShowRows(), 2);
}
}