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

Merge pull request #25456 from taosdata/FIX/TD-29704-3.0 

fix: execute undo actions of TRN_EXEC_SERIAL trans sequentially
This commit is contained in:
Hongze Cheng 2024-04-24 18:09:56 +08:00 committed by GitHub
parent ab2c9aaf5b 9f8e970a32
commit c085efb4f8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 42 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -166,7 +166,7 @@ typedef struct {
int32_t failedTimes; int32_t failedTimes;
void* rpcRsp; void* rpcRsp;
int32_t rpcRspLen; int32_t rpcRspLen;
int32_t redoActionPos; int32_t actionPos;
SArray* prepareActions; SArray* prepareActions;
SArray* redoActions; SArray* redoActions;
SArray* undoActions; SArray* undoActions;

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

@ -169,7 +169,7 @@ SSdbRaw *mndTransEncode(STrans *pTrans) {
SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER) SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER)
SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER) SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
SDB_SET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER) SDB_SET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->redoActionPos, _OVER) SDB_SET_INT32(pRaw, dataPos, pTrans->actionPos, _OVER)
int32_t prepareActionNum = taosArrayGetSize(pTrans->prepareActions); int32_t prepareActionNum = taosArrayGetSize(pTrans->prepareActions);
int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions); int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions);
@ -317,7 +317,7 @@ SSdbRow *mndTransDecode(SSdbRaw *pRaw) {
SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER) SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER) SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
SDB_GET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER) SDB_GET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pTrans->redoActionPos, _OVER) SDB_GET_INT32(pRaw, dataPos, &pTrans->actionPos, _OVER)
if (sver > TRANS_VER1_NUMBER) { if (sver > TRANS_VER1_NUMBER) {
SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER) SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
@ -525,7 +525,7 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
mndTransUpdateActions(pOld->undoActions, pNew->undoActions); mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
mndTransUpdateActions(pOld->commitActions, pNew->commitActions); mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
pOld->stage = pNew->stage; pOld->stage = pNew->stage;
pOld->redoActionPos = pNew->redoActionPos; pOld->actionPos = pNew->actionPos;
if (pOld->stage == TRN_STAGE_COMMIT) { if (pOld->stage == TRN_STAGE_COMMIT) {
pOld->stage = TRN_STAGE_COMMIT_ACTION; pOld->stage = TRN_STAGE_COMMIT_ACTION;
@ -1360,22 +1360,19 @@ static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool
return code; return code;
} }
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) { static int32_t mndTransExecuteActionsSerial(SMnode *pMnode, STrans *pTrans, SArray *pActions, bool topHalf) {
int32_t code = 0; int32_t code = 0;
int32_t numOfActions = taosArrayGetSize(pTrans->redoActions); int32_t numOfActions = taosArrayGetSize(pActions);
if (numOfActions == 0) return code; if (numOfActions == 0) return code;
taosThreadMutexLock(&pTrans->mutex); if (pTrans->actionPos >= numOfActions) {
if (pTrans->redoActionPos >= numOfActions) {
taosThreadMutexUnlock(&pTrans->mutex);
return code; return code;
} }
mInfo("trans:%d, execute %d actions serial, current redoAction:%d", pTrans->id, numOfActions, pTrans->redoActionPos); mInfo("trans:%d, execute %d actions serial, current redoAction:%d", pTrans->id, numOfActions, pTrans->actionPos);
for (int32_t action = pTrans->redoActionPos; action < numOfActions; ++action) { for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->redoActionPos); STransAction *pAction = taosArrayGet(pActions, pTrans->actionPos);
code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf); code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
if (code == 0) { if (code == 0) {
@ -1409,14 +1406,14 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans,
if (code == 0) { if (code == 0) {
pTrans->code = 0; pTrans->code = 0;
pTrans->redoActionPos++; pTrans->actionPos++;
mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage), mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
pAction->id); pAction->id);
taosThreadMutexUnlock(&pTrans->mutex); taosThreadMutexUnlock(&pTrans->mutex);
code = mndTransSync(pMnode, pTrans); code = mndTransSync(pMnode, pTrans);
taosThreadMutexLock(&pTrans->mutex); taosThreadMutexLock(&pTrans->mutex);
if (code != 0) { if (code != 0) {
pTrans->redoActionPos--; pTrans->actionPos--;
pTrans->code = terrno; pTrans->code = terrno;
mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id, mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
mndTransStr(pAction->stage), pAction->id, terrstr()); mndTransStr(pAction->stage), pAction->id, terrstr());
@ -1442,8 +1439,26 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans,
} }
} }
taosThreadMutexUnlock(&pTrans->mutex); return code;
}
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
taosThreadMutexLock(&pTrans->mutex);
if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
}
taosThreadMutexUnlock(&pTrans->mutex);
return code;
}
static int32_t mndTransExecuteUndoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
taosThreadMutexLock(&pTrans->mutex);
if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->undoActions, topHalf);
}
taosThreadMutexUnlock(&pTrans->mutex);
return code; return code;
} }
@ -1563,13 +1578,22 @@ static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, boo
static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) { static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
bool continueExec = true; bool continueExec = true;
int32_t code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf); int32_t code = 0;
if (pTrans->exec == TRN_EXEC_SERIAL) {
code = mndTransExecuteUndoActionsSerial(pMnode, pTrans, topHalf);
} else {
code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf);
}
if (mndCannotExecuteTransAction(pMnode, topHalf)) return false;
terrno = code;
if (code == 0) { if (code == 0) {
pTrans->stage = TRN_STAGE_PRE_FINISH; pTrans->stage = TRN_STAGE_PRE_FINISH;
mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id); mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id);
continueExec = true; continueExec = true;
} else if (code == TSDB_CODE_ACTION_IN_PROGRESS) { } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code)); mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
continueExec = false; continueExec = false;
} else { } else {