diff --git a/source/dnode/mnode/impl/inc/mndTrans.h b/source/dnode/mnode/impl/inc/mndTrans.h index 9fa59b4ac63f76f3527ff5f8b52a797361048570..ac078ce1a38d14ca883c61c239d69e07c7cf0b4e 100644 --- a/source/dnode/mnode/impl/inc/mndTrans.h +++ b/source/dnode/mnode/impl/inc/mndTrans.h @@ -70,7 +70,7 @@ int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendNullLog(STrans *pTrans); -int32_t mndTransAppendPrepareAction(STrans *pTrans, STransAction *pRaw); +int32_t mndTransAppendPrepareAction(STrans *pTrans, STransAction *pAction); int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction); int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction); void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen); diff --git a/source/dnode/mnode/impl/inc/mndVgroup.h b/source/dnode/mnode/impl/inc/mndVgroup.h index 0cd1228f2522f793cc2a3de9f0b45fa36402d34d..286a62438366919550c169b798a09eef0942bfdc 100644 --- a/source/dnode/mnode/impl/inc/mndVgroup.h +++ b/source/dnode/mnode/impl/inc/mndVgroup.h @@ -36,6 +36,7 @@ int64_t mndGetVgroupMemory(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup); SArray *mndBuildDnodesArray(SMnode *, int32_t exceptDnodeId); int32_t mndAllocSmaVgroup(SMnode *, SDbObj *pDb, SVgObj *pVgroup); int32_t mndAllocVgroup(SMnode *, SDbObj *pDb, SVgObj **ppVgroups); +int32_t mndAddPrepareNewVgAction(SMnode *, STrans *pTrans, SVgObj *pVg); int32_t mndAddCreateVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid); int32_t mndAddAlterVnodeConfirmAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup); int32_t mndAddAlterVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, tmsg_t msgType); diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c index 15374a8dddcb12fe0f6d377ec2f42aa85cf8b603..05acaea7bc0f8e079f07943d195eb16ededc22d3 100644 --- a/source/dnode/mnode/impl/src/mndDb.c +++ b/source/dnode/mnode/impl/src/mndDb.c @@ -414,6 +414,13 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) { if (pCfg->tsdbPageSize <= 0) pCfg->tsdbPageSize = TSDB_DEFAULT_TSDB_PAGESIZE; } +static int32_t mndSetPrepareNewVgActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) { + for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) { + if (mndAddPrepareNewVgAction(pMnode, pTrans, (pVgroups + v)) != 0) return -1; + } + return 0; +} + static int32_t mndSetCreateDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) { SSdbRaw *pDbRaw = mndDbActionEncode(pDb); if (pDbRaw == NULL) return -1; @@ -424,7 +431,7 @@ static int32_t mndSetCreateDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pD SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroups + v); if (pVgRaw == NULL) return -1; if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) return -1; - if (sdbSetRawStatus(pVgRaw, SDB_STATUS_CREATING) != 0) return -1; + if (sdbSetRawStatus(pVgRaw, SDB_STATUS_UPDATE) != 0) return -1; } return 0; @@ -575,6 +582,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate, if (mndTransCheckConflict(pMnode, pTrans) != 0) goto _OVER; mndTransSetOper(pTrans, MND_OPER_CREATE_DB); + if (mndSetPrepareNewVgActions(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER; if (mndSetCreateDbRedoLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER; if (mndSetCreateDbUndoLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER; if (mndSetCreateDbCommitLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER; diff --git a/source/dnode/mnode/impl/src/mndSma.c b/source/dnode/mnode/impl/src/mndSma.c index 996f6c64944bde21fd28bd83c627ee680e950cc2..c337d85b688da80266899376969d68dc5f2b6855 100644 --- a/source/dnode/mnode/impl/src/mndSma.c +++ b/source/dnode/mnode/impl/src/mndSma.c @@ -388,7 +388,7 @@ static int32_t mndSetCreateSmaVgroupRedoLogs(SMnode *pMnode, STrans *pTrans, SVg SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup); if (pVgRaw == NULL) return -1; if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) return -1; - if (sdbSetRawStatus(pVgRaw, SDB_STATUS_CREATING) != 0) return -1; + if (sdbSetRawStatus(pVgRaw, SDB_STATUS_UPDATE) != 0) return -1; return 0; } @@ -626,7 +626,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea mndTransSetSerial(pTrans); mInfo("trans:%d, used to create sma:%s stream:%s", pTrans->id, pCreate->name, streamObj.name); - + if (mndAddPrepareNewVgAction(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER; if (mndSetCreateSmaRedoLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER; if (mndSetCreateSmaVgroupRedoLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER; if (mndSetCreateSmaCommitLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER; diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index 69a6053ba6eef94f87adf911c3b232d2bad774fa..7822a596d42366d4d8efecff31e522c90b4d4381 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -1333,6 +1333,19 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans) static bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans) { bool continueExec = true; + int32_t code = 0; + + int32_t numOfActions = taosArrayGetSize(pTrans->prepareActions); + if (numOfActions == 0) goto _OVER; + + mInfo("trans:%d, execute %d prepare actions.", pTrans->id, numOfActions); + + for (int32_t action = 0; action < numOfActions; ++action) { + STransAction *pAction = taosArrayGet(pTrans->prepareActions, action); + mndTransExecSingleAction(pMnode, pTrans, pAction); + } + +_OVER: pTrans->stage = TRN_STAGE_REDO_ACTION; mInfo("trans:%d, stage from prepare to redoAction", pTrans->id); return continueExec; diff --git a/source/dnode/mnode/impl/src/mndVgroup.c b/source/dnode/mnode/impl/src/mndVgroup.c index a5a364023da63bf6afe65e9dffde013046dcedbd..58612efe48df5147917765eeb228842ed7ca002d 100644 --- a/source/dnode/mnode/impl/src/mndVgroup.c +++ b/source/dnode/mnode/impl/src/mndVgroup.c @@ -483,15 +483,15 @@ static void *mndBuildDisableVnodeWriteReq(SMnode *pMnode, SDbObj *pDb, int32_t v return pReq; } -static void *mndBuildAlterVnodeHashRangeReq(SMnode *pMnode, SVgObj *pVgroup, int32_t dstVgId, int32_t *pContLen) { +static void *mndBuildAlterVnodeHashRangeReq(SMnode *pMnode, int32_t srcVgId, SVgObj *pVgroup, int32_t *pContLen) { SAlterVnodeHashRangeReq alterReq = { - .srcVgId = pVgroup->vgId, - .dstVgId = dstVgId, + .srcVgId = srcVgId, + .dstVgId = pVgroup->vgId, .hashBegin = pVgroup->hashBegin, .hashEnd = pVgroup->hashEnd, }; - mInfo("vgId:%d, build alter vnode hashrange req, dstVgId:%d, hashrange:[%u, %u]", pVgroup->vgId, dstVgId, + mInfo("vgId:%d, build alter vnode hashrange req, dstVgId:%d, hashrange:[%u, %u]", srcVgId, pVgroup->vgId, pVgroup->hashBegin, pVgroup->hashEnd); int32_t contLen = tSerializeSAlterVnodeHashRangeReq(NULL, 0, &alterReq); if (contLen < 0) { @@ -1207,12 +1207,12 @@ int32_t mndAddAlterVnodeConfirmAction(SMnode *pMnode, STrans *pTrans, SDbObj *pD return 0; } -static int32_t mndAddAlterVnodeHashRangeAction(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, int32_t dstVgId) { +static int32_t mndAddAlterVnodeHashRangeAction(SMnode *pMnode, STrans *pTrans, int32_t srcVgId, SVgObj *pVgroup) { STransAction action = {0}; action.epSet = mndGetVgroupEpset(pMnode, pVgroup); int32_t contLen = 0; - void *pReq = mndBuildAlterVnodeHashRangeReq(pMnode, pVgroup, dstVgId, &contLen); + void *pReq = mndBuildAlterVnodeHashRangeReq(pMnode, srcVgId, pVgroup, &contLen); if (pReq == NULL) return -1; action.pCont = pReq; @@ -1247,6 +1247,21 @@ int32_t mndAddAlterVnodeConfigAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb return 0; } +int32_t mndAddPrepareNewVgAction(SMnode *pMnode, STrans *pTrans, SVgObj *pVg) { + SSdbRaw *pRaw = mndVgroupActionEncode(pVg); + if (pRaw == NULL) goto _err; + + STransAction action = {.pRaw = pRaw}; + if (mndTransAppendPrepareAction(pTrans, &action) != 0) goto _err; + (void)sdbSetRawStatus(pRaw, SDB_STATUS_CREATING); + pRaw = NULL; + return 0; + +_err: + sdbFreeRaw(pRaw); + return -1; +} + int32_t mndAddAlterVnodeReplicaAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, int32_t dnodeId) { SDnodeObj *pDnode = mndAcquireDnode(pMnode, dnodeId); if (pDnode == NULL) return -1; @@ -2313,12 +2328,16 @@ int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj *pVgro // alter vgId and hash range int32_t maxVgId = sdbGetMaxId(pMnode->pSdb, SDB_VGROUP); - if (mndAddAlterVnodeHashRangeAction(pMnode, pTrans, &newVg1, maxVgId) != 0) goto _OVER; + int32_t srcVgId = newVg1.vgId; newVg1.vgId = maxVgId; + if (mndAddPrepareNewVgAction(pMnode, pTrans, &newVg1) != 0) goto _OVER; + if (mndAddAlterVnodeHashRangeAction(pMnode, pTrans, srcVgId, &newVg1) != 0) goto _OVER; maxVgId++; - if (mndAddAlterVnodeHashRangeAction(pMnode, pTrans, &newVg2, maxVgId) != 0) goto _OVER; + srcVgId = newVg2.vgId; newVg2.vgId = maxVgId; + if (mndAddPrepareNewVgAction(pMnode, pTrans, &newVg2) != 0) goto _OVER; + if (mndAddAlterVnodeHashRangeAction(pMnode, pTrans, srcVgId, &newVg2) != 0) goto _OVER; if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg1) != 0) goto _OVER; diff --git a/source/dnode/mnode/sdb/inc/sdb.h b/source/dnode/mnode/sdb/inc/sdb.h index e9a9e425e3fd364a1702ca084edd8be4e0b4aeeb..3c96d8a2fd355de632392e4496a7a956b55bae45 100644 --- a/source/dnode/mnode/sdb/inc/sdb.h +++ b/source/dnode/mnode/sdb/inc/sdb.h @@ -122,6 +122,7 @@ typedef enum { SDB_STATUS_DROPPING = 2, SDB_STATUS_DROPPED = 3, SDB_STATUS_READY = 4, + SDB_STATUS_UPDATE = 5, } ESdbStatus; typedef enum { diff --git a/source/dnode/mnode/sdb/src/sdbHash.c b/source/dnode/mnode/sdb/src/sdbHash.c index f1cee6395b812c2bef8655530cd9f6ab211eedd3..258b22d8ee7f607b3239a2f5a278f942f7e5c428 100644 --- a/source/dnode/mnode/sdb/src/sdbHash.c +++ b/source/dnode/mnode/sdb/src/sdbHash.c @@ -256,6 +256,7 @@ int32_t sdbWriteWithoutFree(SSdb *pSdb, SSdbRaw *pRaw) { code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize); break; case SDB_STATUS_READY: + case SDB_STATUS_UPDATE: case SDB_STATUS_DROPPING: code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize); break;