| List: | Commits | « Previous MessageNext Message » | |
| From: | jonas | Date: | February 6 2008 9:15pm |
| Subject: | bk commit into 5.1 tree (jonas:1.2851) | ||
| View as plain text | |||
Below is the list of changes that have just been committed into a local 5.1 repository of jonas. When jonas does a push these changes will be propagated to the main repository and, within 24 hours after the push, to the public repository. For information on how to access the public repository see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html ChangeSet@stripped, 2008-02-06 21:15:14+01:00, jonas@stripped +7 -0 ndb - wl3600 [ 11 / 11 ] Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... --- storage/ndb/include/kernel/signaldata/DictSignal.hpp | 12 storage/ndb/include/kernel/signaldata/SchemaTrans.hpp | 3 storage/ndb/include/kernel/signaldata/SchemaTransImpl.hpp | 89 storage/ndb/src/common/debugger/signaldata/SchemaTransImpl.cpp | 8 storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp | 2378 ++++------ storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp | 363 - storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp | 1 7 files changed, 1056 insertions(+), 1798 deletions(-) storage/ndb/include/kernel/signaldata/DictSignal.hpp@stripped, 2008-02-06 21:15:12+01:00, jonas@stripped +0 -12 Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... storage/ndb/include/kernel/signaldata/SchemaTrans.hpp@stripped, 2008-02-06 21:15:12+01:00, jonas@stripped +2 -1 Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... storage/ndb/include/kernel/signaldata/SchemaTransImpl.hpp@stripped, 2008-02-06 21:15:12+01:00, jonas@stripped +11 -78 Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... storage/ndb/src/common/debugger/signaldata/SchemaTransImpl.cpp@stripped, 2008-02-06 21:15:12+01:00, jonas@stripped +5 -3 Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp@stripped, 2008-02-06 21:15:12+01:00, jonas@stripped +900 -1332 Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp@stripped, 2008-02-06 21:15:12+01:00, jonas@stripped +65 -298 Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp@stripped, 2008-02-06 21:15:12+01:00, jonas@stripped +0 -1 Trans-replace This patch removes iterator and "old" transaction machinery And replaces it with a switch-based version with 1 "flow" for each phase of a transaction Interesting that it seems to be a code reduce... diff -Nrup a/storage/ndb/include/kernel/signaldata/DictSignal.hpp b/storage/ndb/include/kernel/signaldata/DictSignal.hpp --- a/storage/ndb/include/kernel/signaldata/DictSignal.hpp 2008-02-06 21:10:05 +01:00 +++ b/storage/ndb/include/kernel/signaldata/DictSignal.hpp 2008-02-06 21:15:12 +01:00 @@ -83,16 +83,6 @@ public: */ RF_NO_BUILD = (1 << 17), - // local - - /* - * Client does not create a schema trans. Client sends a single - * operation request with this flag set. DICT creates an internal - * trans and runs the op (and any sub-ops) within it. - * - * NOTE: not yet implemented - */ - RF_SIMPLE_TRANS = (1 << 24) }; static void @@ -113,8 +103,6 @@ public: strcat(buf, " LOCAL_TRANS"); if (info & RF_NO_BUILD) strcat(buf, " NO_BUILD"); - if (info & RF_SIMPLE_TRANS) - strcat(buf, " SIMPLE_TRANS"); return &buf[1]; } diff -Nrup a/storage/ndb/include/kernel/signaldata/SchemaTrans.hpp b/storage/ndb/include/kernel/signaldata/SchemaTrans.hpp --- a/storage/ndb/include/kernel/signaldata/SchemaTrans.hpp 2007-12-25 17:05:07 +01:00 +++ b/storage/ndb/include/kernel/signaldata/SchemaTrans.hpp 2008-02-06 21:15:12 +01:00 @@ -57,7 +57,8 @@ struct SchemaTransEndReq { // must match NdbDictionary::Dictionary::SchemaTransFlag enum Flag { SchemaTransAbort = 1, - SchemaTransBackground = 2 + SchemaTransBackground = 2, + SchemaTransPrepare = 4 // Only run prepare }; STATIC_CONST( SignalLength = 5 ); Uint32 clientRef; diff -Nrup a/storage/ndb/include/kernel/signaldata/SchemaTransImpl.hpp b/storage/ndb/include/kernel/signaldata/SchemaTransImpl.hpp --- a/storage/ndb/include/kernel/signaldata/SchemaTransImpl.hpp 2008-02-06 21:13:49 +01:00 +++ b/storage/ndb/include/kernel/signaldata/SchemaTransImpl.hpp 2008-02-06 21:15:12 +01:00 @@ -37,96 +37,26 @@ struct SchemaTransImplReq STATIC_CONST( SignalLength = 9 ); Uint32 senderRef; + Uint32 transId; Uint32 transKey; - Uint32 opKey; - Uint32 phaseInfo; // mode | phase | piggy-backed gsn - Uint32 requestInfo; // 0 | op extra | global flags | local flags - Uint32 operationInfo; // op index | op depth - Uint32 iteratorInfo; // list id | list index | repeat | 0 + Uint32 requestInfo; // request type | op extra | global flags | local flags Uint32 clientRef; - Uint32 transId; - - Uint32 requestType; - - // phaseInfo - static Uint32 getMode(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 0, 8); - } - static void setMode(Uint32& info, Uint32 val) { - assert(val < (1 << 8)); - BitmaskImpl::setField(1, &info, 0, 8, val); - } - static Uint32 getPhase(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 8, 4); - } - static void setPhase(Uint32& info, Uint32 val) { - assert(val < (1 << 4)); - BitmaskImpl::setField(1, &info, 8, 4, val); - } - static Uint32 getGsn(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 16, 16); - } - static void setGsn(Uint32& info, Uint32 val) { - assert(val < (1 << 16)); - BitmaskImpl::setField(1, &info, 16, 16, val); - } - - // requestInfo is defined by DictSignal - - // operation info - static Uint32 getOpIndex(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 0, 24); - } - static void setOpIndex(Uint32& info, Uint32 val) { - BitmaskImpl::setField(1, &info, 0, 24, val); - } - static Uint32 getOpDepth(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 24, 8); - } - static void setOpDepth(Uint32& info, Uint32 val) { - assert(val < (1 << 8)); - BitmaskImpl::setField(1, &info, 24, 8, val); - } - - // iteratorInfo - static Uint32 getListId(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 0, 8); - } - static void setListId(Uint32& info, Uint32 val) { - assert(val < (1 << 8)); - BitmaskImpl::setField(1, &info, 0, 8, val); - } - static Uint32 getListIndex(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 8, 8); - } - static void setListIndex(Uint32& info, Uint32 val) { - assert(val < (1 << 8)); - BitmaskImpl::setField(1, &info, 8, 8, val); - } - static Uint32 getItRepeat(const Uint32& info) { - return BitmaskImpl::getField(1, &info, 16, 8); - } - static void setItRepeat(Uint32& info, Uint32 val) { - assert(val < (1 << 8)); - BitmaskImpl::setField(1, &info, 16, 8, val); - } + Uint32 opKey; + Uint32 gsn; }; struct SchemaTransImplConf { - enum { - IT_REPEAT = (1 << 1) - }; - - STATIC_CONST( SignalLength = 3 ); + STATIC_CONST( SignalLength = 4 ); Uint32 senderRef; Uint32 transKey; - Uint32 itFlags; + Uint32 opKey; + Uint32 requestType; }; struct SchemaTransImplRef { - STATIC_CONST( SignalLength = 6 ); + STATIC_CONST( SignalLength = 8 ); STATIC_CONST( GSN = GSN_SCHEMA_TRANS_IMPL_REF ); enum ErrorCode { NoError = 0, @@ -135,11 +65,14 @@ struct SchemaTransImplRef InvalidTransKey = 781, InvalidTransId = 782, TooManySchemaOps = 783, + SeizeFailed = 783, InvalidTransState = 784, NF_FakeErrorREF = 1 }; Uint32 senderRef; union { Uint32 transKey, senderData; }; + Uint32 opKey; + Uint32 requestType; Uint32 errorCode; Uint32 errorLine; Uint32 errorNodeId; diff -Nrup a/storage/ndb/src/common/debugger/signaldata/SchemaTransImpl.cpp b/storage/ndb/src/common/debugger/signaldata/SchemaTransImpl.cpp --- a/storage/ndb/src/common/debugger/signaldata/SchemaTransImpl.cpp 2008-02-06 21:10:05 +01:00 +++ b/storage/ndb/src/common/debugger/signaldata/SchemaTransImpl.cpp 2008-02-06 21:15:12 +01:00 @@ -20,8 +20,10 @@ #include <DebuggerNames.hpp> bool -printSCHEMA_TRANS_IMPL_REQ(FILE* output, const Uint32* theData, Uint32 len, Uint16 rbn) +printSCHEMA_TRANS_IMPL_REQ(FILE* output, const Uint32* theData, + Uint32 len, Uint16 rbn) { +#if 0 const SchemaTransImplReq* sig = (const SchemaTransImplReq*)theData; const Uint32 phaseInfo = sig->phaseInfo; Uint32 mode = SchemaTransImplReq::getMode(phaseInfo); @@ -106,7 +108,8 @@ printSCHEMA_TRANS_IMPL_REQ(FILE* output, break; } } - return true; +#endif + return false; } bool @@ -115,7 +118,6 @@ printSCHEMA_TRANS_IMPL_CONF(FILE* output const SchemaTransImplConf* sig = (const SchemaTransImplConf*)theData; fprintf(output, " senderRef: 0x%x", sig->senderRef); fprintf(output, " transKey: %u", sig->transKey); - fprintf(output, " itFlags: 0x%x", sig->itFlags); fprintf(output, "\n"); return true; } diff -Nrup a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp --- a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp 2008-02-06 21:13:49 +01:00 +++ b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp 2008-02-06 21:15:12 +01:00 @@ -2241,7 +2241,7 @@ void Dbdict::execREAD_CONFIG_REQ(Signal* c_opSectionBufferPool.setSize(1024); // units OpSectionSegmentSize c_schemaOpPool.setSize(256); c_schemaOpHash.setSize(256); - c_schemaTransPool.setSize(1); + c_schemaTransPool.setSize(5); c_schemaTransHash.setSize(2); c_txHandlePool.setSize(2); c_txHandleHash.setSize(2); @@ -4229,9 +4229,6 @@ void Dbdict::execNODE_FAILREP(Signal* si NodeRecordPtr nodePtr; c_nodes.getPtr(nodePtr, i); - // update schema transactions (does not time-slice) - handleTransSlaveFail(signal, i); - nodePtr.p->nodeState = NodeRecord::NDB_NODE_DEAD; NFCompleteRep * const nfCompRep = (NFCompleteRep *)&signal->theData[0]; nfCompRep->blockNo = DBDICT; @@ -5863,7 +5860,6 @@ Dbdict::createTable_commit(Signal* signa { jam(); SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); CreateTableRecPtr createTabPtr; getOpRec(op_ptr, createTabPtr); @@ -5872,7 +5868,7 @@ Dbdict::createTable_commit(Signal* signa TableRecordPtr tabPtr; c_tableRecordPool.getPtr(tabPtr, tableId); - D("createTable_commit" << V(itRepeat) << *op_ptr.p); + D("createTable_commit" << *op_ptr.p); bool savetodisk = !(tabPtr.p->m_bits & TableRecord::TR_Temporary); @@ -5976,7 +5972,7 @@ Dbdict::createTab_alterComplete(Signal* #endif } - sendTransConf(signal, op_ptr, 0); + sendTransConf(signal, op_ptr); } void @@ -5990,19 +5986,16 @@ Dbdict::createTable_abortParse(Signal* s D("createTable_abortParse" << V(tableId) << *op_ptr.p); do { - if (op_ptr.p->m_abortPrepareDone) { + if (createTabPtr.p->m_abortPrepareDone) + { jam(); D("done by abort prepare"); break; } - const TransPhase::Value phase = op_ptr.p->m_step.m_phase; - ndbrequire(phase <= TransPhase::Parse); - if (tableId == RNIL) { jam(); D("no table allocated"); - ndbrequire(phase < TransPhase::Parse); break; } @@ -6019,8 +6012,8 @@ Dbdict::createTable_abortParse(Signal* s releaseTableObject(tableId, true); } - if (phase == TransPhase::Parse) { - jam(); + if (createTabPtr.p->m_curr_entry.m_tableState != SchemaFile::INIT) + { // parse was completed so schema entry must be restored SchemaFile::TableEntry& te = createTabPtr.p->m_orig_entry; bool savetodisk = false; @@ -6155,6 +6148,7 @@ Dbdict::createTable_abortWriteSchemaConf releaseTableObject(tableId); + createTabPtr.p->m_abortPrepareDone = true; sendTransConf(signal, op_ptr); } @@ -6500,12 +6494,11 @@ void Dbdict::dropTable_prepare(Signal* signal, SchemaOpPtr op_ptr) { SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); DropTableRecPtr dropTabPtr; getOpRec(op_ptr, dropTabPtr); - D("dropTable_prepare" << V(itRepeat) << *op_ptr.p); + D("dropTable_prepare" << *op_ptr.p); dropTabPtr.p->m_block = 0; @@ -6760,12 +6753,11 @@ void Dbdict::dropTable_commit(Signal* signal, SchemaOpPtr op_ptr) { SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); DropTableRecPtr dropTabPtr; getOpRec(op_ptr, dropTabPtr); - D("dropTable_commit" << V(itRepeat) << *op_ptr.p); + D("dropTable_commit" << *op_ptr.p); TableRecordPtr tablePtr; c_tableRecordPool.getPtr(tablePtr, dropTabPtr.p->m_request.tableId); @@ -7012,7 +7004,8 @@ Dbdict::dropTable_abortParse(Signal* sig Uint32 tableId = impl_req->tableId; // update schema state in memory XXX temp - if (dropTabPtr.p->m_orig_entry.m_tableVersion != 0) { + if (dropTabPtr.p->m_curr_entry.m_tableState != SchemaFile::INIT) + { XSchemaFile * xsf = &c_schemaFile[c_schemaRecord.schemaPage != 0]; SchemaFile::TableEntry * tableEntry = getTableEntry(xsf, tableId); *tableEntry = dropTabPtr.p->m_orig_entry; @@ -7316,7 +7309,7 @@ Dbdict::alterTable_parse(Signal* signal, { XSchemaFile * xsf = &c_schemaFile[c_schemaRecord.schemaPage != 0]; SchemaFile::TableEntry * tableEntry = getTableEntry(xsf, impl_req->tableId); - alterTabPtr.p->m_oldTableEntry = *tableEntry; + alterTabPtr.p->m_orig_entry = *tableEntry; } // master rewrites DictTabInfo (it is re-parsed only on slaves) @@ -7391,13 +7384,12 @@ Dbdict::alterTable_prepare(Signal* signa { jam(); SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); AlterTableRecPtr alterTabPtr; getOpRec(op_ptr, alterTabPtr); //const AlterTabReq* impl_req = &alterTabPtr.p->m_request; - D("alterTable_prepare" << V(itRepeat) << *op_ptr.p); + D("alterTable_prepare" << *op_ptr.p); Mutex mutex(signal, c_mutexMgr, alterTabPtr.p->m_define_backup_mutex); Callback c = { @@ -7641,13 +7633,12 @@ void Dbdict::alterTable_commit(Signal* signal, SchemaOpPtr op_ptr) { SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); AlterTableRecPtr alterTabPtr; getOpRec(op_ptr, alterTabPtr); //const AlterTabReq* impl_req = &alterTabPtr.p->m_request; - D("alterTable_commit" << V(itRepeat) << *op_ptr.p); + D("alterTable_commit" << *op_ptr.p); // wl3600_todo hmm there is no prepare to disk @@ -9511,12 +9502,11 @@ void Dbdict::dropIndex_prepare(Signal* signal, SchemaOpPtr op_ptr) { SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); DropIndexRecPtr dropIndexPtr; getOpRec(op_ptr, dropIndexPtr); - D("dropIndex_prepare" << V(itRepeat) << *op_ptr.p); + D("dropIndex_prepare" << *op_ptr.p); sendTransConf(signal, op_ptr); } @@ -9527,12 +9517,11 @@ void Dbdict::dropIndex_commit(Signal* signal, SchemaOpPtr op_ptr) { SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); DropIndexRecPtr dropIndexPtr; getOpRec(op_ptr, dropIndexPtr); - D("dropIndex_commit" << V(itRepeat) << *op_ptr.p); + D("dropIndex_commit" << *op_ptr.p); sendTransConf(signal, op_ptr); } @@ -10008,7 +9997,6 @@ Dbdict::alterIndex_toCreateTrigger(Signa Uint32 requestInfo = 0; DictSignal::setRequestType(requestInfo, CreateTrigReq::CreateTriggerOnline); - DictSignal::setRequestExtra(requestInfo, AlterIndex_atCreateTrigger); DictSignal::addRequestFlagsGlobal(requestInfo, op_ptr.p->m_requestInfo); req->clientRef = reference(); @@ -10126,7 +10114,6 @@ Dbdict::alterIndex_toDropTrigger(Signal* Uint32 requestInfo = 0; DictSignal::setRequestType(requestInfo, 0); - DictSignal::setRequestExtra(requestInfo, AlterIndex_atDropTrigger); DictSignal::addRequestFlagsGlobal(requestInfo, op_ptr.p->m_requestInfo); req->clientRef = reference(); @@ -10303,7 +10290,6 @@ void Dbdict::alterIndex_prepare(Signal* signal, SchemaOpPtr op_ptr) { SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); AlterIndexRecPtr alterIndexPtr; getOpRec(op_ptr, alterIndexPtr); @@ -10313,7 +10299,7 @@ Dbdict::alterIndex_prepare(Signal* signa TableRecordPtr indexPtr; c_tableRecordPool.getPtr(indexPtr, impl_req->indexId); - D("alterIndex_prepare" << V(itRepeat) << *op_ptr.p); + D("alterIndex_prepare" << *op_ptr.p); /* * Create table creates index table in all blocks. DBTC has @@ -10440,12 +10426,11 @@ Dbdict::alterIndex_commit(Signal* signal { jam(); SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - Uint32 itRepeat = getIteratorRepeat(trans_ptr); AlterIndexRecPtr alterIndexPtr; getOpRec(op_ptr, alterIndexPtr); - D("alterIndex_commit" << V(itRepeat) << *op_ptr.p); + D("alterIndex_commit" << *op_ptr.p); sendTransConf(signal, op_ptr); } @@ -10830,7 +10815,6 @@ Dbdict::buildIndex_toCreateConstraint(Si Uint32 requestInfo = 0; DictSignal::setRequestType(requestInfo, CreateTrigReq::CreateTriggerOnline); - DictSignal::setRequestExtra(requestInfo, BuildIndex_atCreateConstraint); DictSignal::addRequestFlagsGlobal(requestInfo, op_ptr.p->m_requestInfo); req->clientRef = reference(); @@ -11003,7 +10987,6 @@ Dbdict::buildIndex_toDropConstraint(Sign Uint32 requestInfo = 0; DictSignal::setRequestType(requestInfo, 0); - DictSignal::setRequestExtra(requestInfo, BuildIndex_atDropConstraint); DictSignal::addRequestFlagsGlobal(requestInfo, op_ptr.p->m_requestInfo); req->clientRef = reference(); @@ -11227,8 +11210,7 @@ Dbdict::buildIndex_fromLocalBuild(Signal if (ret == 0) { jam(); - Uint32 itFlags = 0; - sendTransConf(signal, op_ptr, itFlags); + sendTransConf(signal, op_ptr); } else { jam(); setError(op_ptr, ret, __LINE__); @@ -13947,6 +13929,14 @@ Dbdict::createTrigger_parse(Signal* sign setError(error, 9124, __LINE__); return; } + + if (impl_req->indexId != RNIL) + { + TableRecordPtr indexPtr; + c_tableRecordPool.getPtr(indexPtr, impl_req->indexId); + triggerPtr.p->indexId = impl_req->indexId; + indexPtr.p->triggerId = impl_req->triggerId; + } } void @@ -14244,6 +14234,14 @@ Dbdict::createTrigger_abortParse(Signal* triggerPtr.p->triggerState = TriggerRecord::TS_NOT_DEFINED; } + if (triggerPtr.p->indexId != RNIL) + { + TableRecordPtr indexPtr; + c_tableRecordPool.getPtr(indexPtr, triggerPtr.p->indexId); + triggerPtr.p->indexId = RNIL; + indexPtr.p->triggerId = RNIL; + } + // ignore Feedback for now (referencing object will be dropped too) if (hasDictObject(op_ptr)) @@ -14856,6 +14854,14 @@ Dbdict::dropTrigger_commit(Signal* signa TriggerRecordPtr triggerPtr; c_triggerRecordPool.getPtr(triggerPtr, triggerId); + if (triggerPtr.p->indexId != RNIL) + { + TableRecordPtr indexPtr; + c_tableRecordPool.getPtr(indexPtr, triggerPtr.p->indexId); + triggerPtr.p->indexId = RNIL; + indexPtr.p->triggerId = RNIL; + } + // remove trigger releaseDictObject(op_ptr); triggerPtr.p->triggerState = TriggerRecord::TS_NOT_DEFINED; @@ -18230,62 +18236,6 @@ Dbdict::findOpInfo(Uint32 gsn) return 0; } -// FeedbackTable - -const Dbdict::FeedbackEntry -Dbdict::g_feedbackTable[] = { - { AlterIndex_atCreateTrigger, - TransPhase::Parse, - &Dbdict::alterIndex_atCreateTrigger - }, - { AlterIndex_atDropTrigger, - TransPhase::Commit, - &Dbdict::alterIndex_atDropTrigger - }, - { BuildIndex_atCreateConstraint, - TransPhase::Parse, - &Dbdict::buildIndex_atCreateConstraint - }, - { BuildIndex_atDropConstraint, - TransPhase::Commit, - &Dbdict::buildIndex_atDropConstraint - }, - { 0, TransPhase::Undef, 0 } -}; - -const Dbdict::FeedbackEntry* -Dbdict::findFeedbackEntry(Uint32 id, TransPhase::Value phase) -{ - Uint32 i = 0; - while (1) { - const FeedbackEntry& e = g_feedbackTable[i]; - if (e.m_id == 0) - break; - if (e.m_id == id && e.m_phase == phase) - return &e; - i++; - } - return 0; -} - -void -Dbdict::runFeedbackMethod(Signal* signal, SchemaOpPtr op_ptr) -{ - ndbrequire(!op_ptr.isNull()); - SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - - Uint32 id = DictSignal::getRequestExtra(op_ptr.p->m_requestInfo); - if (id != 0) { - jam(); - const FeedbackEntry* e = findFeedbackEntry(id, tLoc.m_phase); - if (e != 0) { - jam(); - (this->*(e->m_method))(signal, op_ptr); - } - } -} - // OpRec // OpSection @@ -18396,8 +18346,6 @@ Dbdict::seizeSchemaOp(SchemaOpPtr& op_pt ERROR_INSERTED(6114) && ( info.m_impl_req_gsn == GSN_CREATE_TRIG_IMPL_REQ || info.m_impl_req_gsn == GSN_DROP_TRIG_IMPL_REQ) || - ERROR_INSERTED(6115) && ( - info.m_impl_req_gsn == GSN_ALTER_TRIG_IMPL_REQ) || ERROR_INSERTED(6116) && ( info.m_impl_req_gsn == GSN_BUILD_INDX_IMPL_REQ)) { jam(); @@ -18520,31 +18468,19 @@ Dbdict::releaseOpSection(SchemaOpPtr op_ void Dbdict::addSchemaOp(SchemaTransPtr trans_ptr, SchemaOpPtr& op_ptr) { - TransLoc& tLoc = trans_ptr.p->m_transLoc; - iteratorAddOp(tLoc, op_ptr); + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + list.addLast(op_ptr); - const Uint32 opDepth = trans_ptr.p->m_opDepth; op_ptr.p->m_trans_ptr = trans_ptr; - op_ptr.p->m_opDepth = opDepth; + // jonas_todo REMOVE side effect // add global flags from trans const Uint32& src_info = trans_ptr.p->m_requestInfo; DictSignal::addRequestFlagsGlobal(op_ptr.p->m_requestInfo, src_info); - - D("addSchemaOp" << *op_ptr.p); } // update op step after successful execution -void -Dbdict::updateSchemaOpStep(SchemaTransPtr trans_ptr, SchemaOpPtr op_ptr) -{ - ndbrequire(!op_ptr.isNull()); - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - D("updateSchemaOpStep" << *op_ptr.p << tLoc); - new (&op_ptr.p->m_step) TransStep(tLoc); -} - // the link SchemaOp -> DictObject -> SchemaTrans // check if link to dict object exists @@ -18711,10 +18647,9 @@ Dbdict::findDictObjectOp(SchemaOpPtr& op break; } D("found" << *trans_ptr.p); - TransLoc& tLoc = trans_ptr.p->m_transLoc; { - LocalDLFifoList<SchemaOp> list(c_schemaOpPool, tLoc.m_opList.m_head); + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); SchemaOpPtr loop_ptr; list.first(loop_ptr); while (!loop_ptr.isNull()) { @@ -18733,366 +18668,6 @@ Dbdict::findDictObjectOp(SchemaOpPtr& op // trans state -bool -Dbdict::getOpPtr(const TransLoc& tLoc, SchemaOpPtr& op_ptr) -{ - if (tLoc.m_opKey == 0) { - jam(); - op_ptr.setNull(); - } else { - jam(); - bool ok = findSchemaOp(op_ptr, tLoc.m_opKey); - ndbrequire(ok); - } - return !op_ptr.isNull(); -} - -const Dbdict::TransPhaseEntry& -Dbdict::getPhaseEntry(const TransLoc& tLoc) -{ - ndbrequire(tLoc.m_phaseList != 0); - const TransPhaseList& phaseList = *tLoc.m_phaseList; - const Uint32 phaseIndex = tLoc.m_phaseIndex; - ndbrequire(phaseIndex > 0 && phaseIndex < phaseList.m_length); - const TransPhaseEntry& phaseEntry = phaseList.m_phaseEntry[phaseIndex]; - return phaseEntry; -} - -void -Dbdict::iteratorAddOp(TransLoc& tLoc, SchemaOpPtr op_ptr) -{ - LocalDLFifoList<SchemaOp> list(c_schemaOpPool, tLoc.m_opList.m_head); -#ifdef VM_TRACE - // check for duplicates - { - SchemaOpPtr loop_ptr; - Uint32 size = 0; - list.first(loop_ptr); - while (!loop_ptr.isNull()) { - jam(); - ndbrequire(loop_ptr.i != op_ptr.i); - ndbrequire(loop_ptr.p->op_key != op_ptr.p->op_key); - list.next(loop_ptr); - size++; - } - ndbrequire(tLoc.m_opList.m_size == size); - } -#endif - // add it to end - list.addLast(op_ptr); - op_ptr.p->m_opIndex = tLoc.m_opList.m_size; - tLoc.m_opList.m_size += 1; - // position iterator at the new op - tLoc.m_opKey = op_ptr.p->op_key; -} - -void -Dbdict::iteratorRemoveLastOp(TransLoc& tLoc, SchemaOpPtr& op_ptr) -{ - D("iteratorRemoveOp" << *op_ptr.p); - if (tLoc.m_opKey == op_ptr.p->op_key) { - jam(); - // move to previous op or start of list - iteratorNext(tLoc, -1); - } - // verify op is last and remove it - LocalDLFifoList<SchemaOp> list(c_schemaOpPool, tLoc.m_opList.m_head); - bool isLast = !list.hasNext(op_ptr); - ndbrequire(isLast); - list.remove(op_ptr); - ndbrequire(tLoc.m_opList.m_size == op_ptr.p->m_opIndex + 1); - tLoc.m_opList.m_size -= 1; - releaseSchemaOp(op_ptr); -} - -bool -Dbdict::iteratorFirstOp(TransLoc& tLoc, int dir) -{ - if (tLoc.m_opList.m_size == 0) { - jam(); - return false; - } - LocalDLFifoList<SchemaOp> list(c_schemaOpPool, tLoc.m_opList.m_head); - SchemaOpPtr op_ptr; - bool ok = dir > 0 ? list.first(op_ptr) : list.last(op_ptr); - ndbrequire(ok); - ndbrequire(op_ptr.p->op_key != 0); - tLoc.m_opKey = op_ptr.p->op_key; - return true; -} - -bool -Dbdict::iteratorNextOp(TransLoc& tLoc, int dir) -{ - if (tLoc.m_opList.m_size == 0) { - jam(); - return false; - } - LocalDLFifoList<SchemaOp> list(c_schemaOpPool, tLoc.m_opList.m_head); - SchemaOpPtr op_ptr; - ndbrequire(tLoc.m_opKey != 0); - bool ok = findSchemaOp(op_ptr, tLoc.m_opKey); - ndbrequire(ok); - if (!(dir > 0 ? list.next(op_ptr) : list.prev(op_ptr))) { - jam(); - return false; - } - ndbrequire(op_ptr.p->op_key != 0); - tLoc.m_opKey = op_ptr.p->op_key; - return true; -} - -bool -Dbdict::iteratorFirst(TransLoc& tLoc, int dir) -{ - ndbrequire(tLoc.m_phaseList != 0); - const TransPhaseList& phaseList = *tLoc.m_phaseList; - ndbrequire(phaseList.m_length >= 2); - const Uint32 i = dir > 0 ? 1 : phaseList.m_length - 1; - const TransPhaseEntry& phaseEntry = phaseList.m_phaseEntry[i]; - // end points must be simple (no ops) - ndbrequire(phaseEntry.m_phaseFlags & TransPhaseFlag::Simple); - tLoc.m_phaseIndex = i; - tLoc.m_phase = phaseEntry.m_phase; - tLoc.m_repeat = 0; - tLoc.m_opKey = 0; - return true; -} - -bool -Dbdict::iteratorNext(TransLoc& tLoc, int dir) -{ - ndbrequire(tLoc.m_phaseList != 0); - const TransPhaseList& phaseList = *tLoc.m_phaseList; - bool loop_one = true; - Uint32 i = tLoc.m_phaseIndex; - for (; i > 0 && i < phaseList.m_length; i += dir, loop_one = false) { - const TransPhaseEntry& phaseEntry = phaseList.m_phaseEntry[i]; - if (phaseEntry.m_phaseFlags & TransPhaseFlag::Simple || - tLoc.m_opList.m_size == 0) { - jam(); - tLoc.m_opKey = 0; - if (loop_one) { - jam(); - continue; - } - } else { - if (tLoc.m_opKey == 0) { - jam(); - bool ok = iteratorFirstOp(tLoc, dir); - ndbrequire(ok); - } else { - jam(); - if (!iteratorNextOp(tLoc, dir)) { - jam(); - tLoc.m_opKey = 0; - continue; - } - } - } - tLoc.m_phaseIndex = i; - tLoc.m_phase = phaseEntry.m_phase; - tLoc.m_repeat = 0; - return true; - } - return false; -} - -bool -Dbdict::iteratorInit(TransLoc& tLoc) -{ - tLoc.m_phaseList = &g_defaultPhaseList; - tLoc.m_mode = TransMode::Normal; - return iteratorFirst(tLoc, +1); -} - -bool -Dbdict::iteratorMove(TransLoc& tLoc, bool repeatFlag) -{ - D("iteratorMove <" << tLoc << V(repeatFlag)); - bool ret = false; - if (repeatFlag) { - jam(); - tLoc.m_repeat += 1; - ret = true; - } - else { - if (tLoc.m_hold) { - jam(); - tLoc.m_hold = false; - ret = true; - } - else if (tLoc.m_mode == TransMode::Normal) { - jam(); - ret = iteratorNext(tLoc, +1); - } - else if (tLoc.m_mode == TransMode::Rollback) { - jam(); - ret = iteratorNext(tLoc, -1); - } - else if (tLoc.m_mode == TransMode::Abort) { - jam(); - ret = iteratorNext(tLoc, -1); - } - else { - ndbrequire(false); - } - tLoc.m_repeat = 0; - } - D("iteratorMove >" << tLoc << V(ret)); - return ret; -} - -void -Dbdict::iteratorCopy(TransLoc& tLoc, Uint32 mode, - Uint32 phaseIndex, Uint32 op_key, Uint32 repeat) -{ - D("iteratorCopy <" << tLoc); - tLoc.m_phaseList = &g_defaultPhaseList; - tLoc.m_mode = (TransMode::Value)mode; - tLoc.m_phaseIndex = phaseIndex; - const TransPhaseEntry& phaseEntry = getPhaseEntry(tLoc); - tLoc.m_phase = phaseEntry.m_phase; - tLoc.m_repeat = repeat; - tLoc.m_opKey = op_key; - D("iteratorCopy >" << tLoc); -} - -Dbdict::TransGlob& -Dbdict::getTransGlob(SchemaTransPtr trans_ptr, Uint32 nodeId) -{ - ndbrequire(nodeId > 0 && nodeId < MAX_NDB_NODES); - TransGlob& tGlob = trans_ptr.p->m_transGlob[nodeId]; - return tGlob; -} - -void -Dbdict::setGlobState(SchemaTransPtr trans_ptr, - Uint32 nodeId, TransGlob::State state) -{ - TransGlob& tGlob = getTransGlob(trans_ptr, nodeId); - TransGlob tGlob2 = tGlob; - tGlob2.state(state); - D("setGlobState " << nodeId << ":" << tGlob << " ->" << tGlob2); - tGlob = tGlob2; -} - -void -Dbdict::setGlobState(SchemaTransPtr trans_ptr, - const NdbNodeBitmask& nodes, TransGlob::State state) -{ - Uint32 nodeId; - for (nodeId = 1; nodeId < MAX_NDB_NODES; nodeId++) { - if (nodes.get(nodeId)) { - setGlobState(trans_ptr, nodeId, state); - } - } -} - -void -Dbdict::setGlobState(SchemaTransPtr trans_ptr, - TransGlob::State oldstate, TransGlob::State state) -{ - Uint32 nodeId; - for (nodeId = 1; nodeId < MAX_NDB_NODES; nodeId++) { - if (trans_ptr.p->m_initNodes.get(nodeId)) { - const TransGlob& tGlob = getTransGlob(trans_ptr, nodeId); - if (tGlob.state() == oldstate) { - jam(); - setGlobState(trans_ptr, nodeId, state); - } - } - } -} - -void -Dbdict::setGlobFlags(SchemaTransPtr trans_ptr, - Uint32 nodeId, Uint32 flags, int op) -{ - TransGlob& tGlob = getTransGlob(trans_ptr, nodeId); - TransGlob tGlob2 = tGlob; - if (op < 0) - tGlob2.m_flags &= ~flags; - else if (op > 0) - tGlob2.m_flags |= flags; - else - tGlob2.m_flags = flags; - D("setGlobFlags " << nodeId << ":" << tGlob << " ->" << tGlob2); - tGlob = tGlob2; -} - -void -Dbdict::setGlobFlags(SchemaTransPtr trans_ptr, - const NdbNodeBitmask& nodes, Uint32 flags, int op) -{ - Uint32 nodeId; - for (nodeId = 1; nodeId < MAX_NDB_NODES; nodeId++) { - if (nodes.get(nodeId)) { - setGlobFlags(trans_ptr, nodeId, flags, op); - } - } -} - -Uint32 -Dbdict::getGlobFlags(SchemaTransPtr trans_ptr, - const NdbNodeBitmask& nodes) -{ - Uint32 flags = 0; - Uint32 nodeId; - for (nodeId = 1; nodeId < MAX_NDB_NODES; nodeId++) { - if (nodes.get(nodeId)) { - const TransGlob& tGlob = getTransGlob(trans_ptr, nodeId); - flags |= tGlob.m_flags; - } - } - return flags; -} - -const Dbdict::TransPhaseEntry -Dbdict::g_defaultPhaseEntry[] = { - { TransPhase::Undef, - 0, - 0, - 0 - }, - { - TransPhase::Begin, - TransPhaseFlag::Simple, - 0, - 0 - }, - { - TransPhase::Parse, - 0, - 0, - 0 - }, - { - TransPhase::Prepare, - 0, - 0, - 0 - }, - { - TransPhase::Commit, - 0, - 0, - 0 - }, - { - TransPhase::End, - TransPhaseFlag::Simple, - 0, - 0 - } -}; - -const Dbdict::TransPhaseList -Dbdict::g_defaultPhaseList = { - 0, - sizeof(g_defaultPhaseEntry) / sizeof(g_defaultPhaseEntry[0]), - g_defaultPhaseEntry -}; // SchemaTrans @@ -19154,8 +18729,7 @@ Dbdict::releaseSchemaTrans(SchemaTransPt Uint32 trans_key = trans_ptr.p->trans_key; D("releaseSchemaTrans" << V(trans_key)); - TransLoc& tLoc = trans_ptr.p->m_transLoc; - LocalDLFifoList<SchemaOp> list(c_schemaOpPool, tLoc.m_opList.m_head); + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); SchemaOpPtr op_ptr; while (list.first(op_ptr)) { list.remove(op_ptr); @@ -19178,13 +18752,6 @@ Dbdict::releaseSchemaTrans(SchemaTransPt #endif } -Uint32 -Dbdict::getIteratorRepeat(SchemaTransPtr trans_ptr) -{ - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - return tLoc.m_repeat; -} - // client requests void @@ -19220,35 +18787,27 @@ Dbdict::execSCHEMA_TRANS_BEGIN_REQ(Signa trans_ptr.p->m_clientRef = clientRef; trans_ptr.p->m_transId = transId; trans_ptr.p->m_requestInfo = requestInfo; - if (!localTrans) { - trans_ptr.p->m_initNodes = c_aliveNodes; - } else { - trans_ptr.p->m_initNodes.clear(); - trans_ptr.p->m_initNodes.set(getOwnNodeId()); + if (!localTrans) + { + jam(); + trans_ptr.p->m_nodes = c_aliveNodes; + } + else + { + jam(); + trans_ptr.p->m_nodes.clear(); + trans_ptr.p->m_nodes.set(getOwnNodeId()); } trans_ptr.p->m_clientState = TransClient::BeginReq; - // initialize local state and iterator - TransLoc& tLoc = trans_ptr.p->m_transLoc; - iteratorInit(tLoc); - - // initialize global state - const Uint32 ownNodeId = getOwnNodeId(); - setGlobState(trans_ptr, ownNodeId, TransGlob::Ok); - - // other nodes have no trans record yet - NdbNodeBitmask nodes = trans_ptr.p->m_initNodes; - nodes.bitANDC(trans_ptr.p->m_failedNodes); - nodes.clear(ownNodeId); - setGlobState(trans_ptr, nodes, TransGlob::NeedTrans); - // lock DictLockReq& lockReq = trans_ptr.p->m_lockReq; lockReq.userPtr = trans_ptr.p->trans_key; lockReq.userRef = reference(); lockReq.lockType = DictLockReq::SchemaTransLock; int lockError = dict_lock_trylock(&lockReq); - if (lockError != 0) { + if (lockError != 0) + { // remove the trans releaseSchemaTrans(trans_ptr); setError(error, lockError, __LINE__); @@ -19256,9 +18815,33 @@ Dbdict::execSCHEMA_TRANS_BEGIN_REQ(Signa } // begin tx on all participants - SectionHandle handle(this); - handle.m_cnt = 0; - sendTransReq(signal, trans_ptr, handle); + trans_ptr.p->m_state = SchemaTrans::TS_STARTING; + + /** + * Send RT_START + */ + { + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = RNIL; + req->requestInfo = SchemaTransImplReq::RT_START; + req->clientRef = trans_ptr.p->m_clientRef; + req->transId = trans_ptr.p->m_transId; + req->gsn = GSN_SCHEMA_TRANS_BEGIN_REQ; + + trans_ptr.p->m_ref_nodes.clear(); + NodeReceiverGroup rg(DBDICT, trans_ptr.p->m_nodes); + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + } + + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); + } + if (ERROR_INSERTED(6102)) { jam(); CLEAR_ERROR_INSERT_VALUE; @@ -19280,9 +18863,71 @@ Dbdict::execSCHEMA_TRANS_BEGIN_REQ(Signa } void +Dbdict::trans_start_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) +{ + jam(); + + switch(trans_ptr.p->m_state){ + case SchemaTrans::TS_STARTING: + if (hasError(trans_ptr.p->m_error)) + { + jam(); + + trans_ptr.p->m_state = SchemaTrans::TS_ABORT_START; + + /** + * Abort before replying to client + */ + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = RNIL; + req->requestInfo = SchemaTransImplReq::RT_COMPLETE; + req->clientRef = trans_ptr.p->m_clientRef; + req->transId = trans_ptr.p->m_transId; + req->gsn = GSN_SCHEMA_TRANS_BEGIN_REQ; + + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; + nodes.bitANDC(trans_ptr.p->m_ref_nodes); + NodeReceiverGroup rg(DBDICT, nodes); + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + } + + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); + return; + } + else + { + jam(); + sendTransClientReply(signal, trans_ptr); + return; + } + break; + case SchemaTrans::TS_ABORT_START:{ + jam(); + sendTransClientReply(signal, trans_ptr); + + const DictLockReq& lockReq = trans_ptr.p->m_lockReq; + dict_lock_unlock(signal, &lockReq); + releaseSchemaTrans(trans_ptr); + return; + } + default: + jamLine(trans_ptr.p->m_state); + ndbrequire(false); + } +} + +void Dbdict::execSCHEMA_TRANS_END_REQ(Signal* signal) { jamEntry(); + const SchemaTransEndReq* req = (const SchemaTransEndReq*)signal->getDataPtr(); Uint32 clientRef = req->clientRef; @@ -19291,63 +18936,58 @@ Dbdict::execSCHEMA_TRANS_END_REQ(Signal* Uint32 requestInfo = req->requestInfo; Uint32 flags = req->flags; - bool localTrans = (requestInfo & DictSignal::RF_LOCAL_TRANS); - SchemaTransPtr trans_ptr; ErrorInfo error; do { - if (getOwnNodeId() != c_masterNodeId && !localTrans) { - jam(); - // future when MNF is handled - setError(error, SchemaTransEndRef::NotMaster, __LINE__); - break; - } - findSchemaTrans(trans_ptr, trans_key); if (trans_ptr.isNull()) { jam(); + ndbassert(false); setError(error, SchemaTransEndRef::InvalidTransKey, __LINE__); break; } if (trans_ptr.p->m_transId != transId) { jam(); + ndbassert(false); setError(error, SchemaTransEndRef::InvalidTransId, __LINE__); break; } - if (trans_ptr.p->m_clientState != TransClient::BeginReply && - trans_ptr.p->m_clientState != TransClient::ParseReply) { + bool localTrans = (trans_ptr.p->m_requestInfo & DictSignal::RF_LOCAL_TRANS); + + if (getOwnNodeId() != c_masterNodeId && !localTrans) { jam(); - setError(error, SchemaTransEndRef::InvalidTransState, __LINE__); + // future when MNF is handled + ndbassert(false); + setError(error, SchemaTransEndRef::NotMaster, __LINE__); break; } + //XXX Check state + + if (hasError(trans_ptr.p->m_error)) { - Uint32 requestInfo2 = trans_ptr.p->m_requestInfo; - bool localTrans2 = (requestInfo2 & DictSignal::RF_LOCAL_TRANS); - ndbrequire(localTrans == localTrans2); + jam(); + ndbassert(false); + setError(error, SchemaTransEndRef::InvalidTransState, __LINE__); + break; } - ndbrequire(!hasError(trans_ptr.p->m_error)); - trans_ptr.p->m_clientState = TransClient::EndReq; - - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - if (tLoc.m_mode == TransMode::Rollback) { + bool localTrans2 = requestInfo & DictSignal::RF_LOCAL_TRANS; + if (localTrans != localTrans2) + { jam(); - D("continuing after rollback"); - resetError(trans_ptr); - setTransMode(trans_ptr, TransMode::Normal, false); + ndbassert(false); + setError(error, SchemaTransEndRef::InvalidTransState, __LINE__); + break; } - ndbrequire(tLoc.m_mode == TransMode::Normal); - D("execSCHEMA_TRANS_END_REQ" << *trans_ptr.p << tLoc.m_opList); - - setGlobState(trans_ptr, TransGlob::NeedOp, TransGlob::NoOp); - setGlobState(trans_ptr, TransGlob::NeedTrans, TransGlob::NoTrans); + trans_ptr.p->m_clientState = TransClient::EndReq; const bool doBackground = flags & SchemaTransEndReq::SchemaTransBackground; - if (doBackground) { + if (doBackground) + { jam(); // send reply to original client and restore EndReq state sendTransClientReply(signal, trans_ptr); @@ -19358,14 +18998,19 @@ Dbdict::execSCHEMA_TRANS_END_REQ(Signal* takeOverTransClient(signal, trans_ptr); } - const bool doAbort = flags & SchemaTransEndReq::SchemaTransAbort; - if (doAbort) { + if (flags & SchemaTransEndReq::SchemaTransAbort) + { jam(); - setTransMode(trans_ptr, TransMode::Abort, true); + trans_abort_prepare_start(signal, trans_ptr); + return; + } + else if ((flags & SchemaTransEndReq::SchemaTransPrepare) == 0) + { + jam(); + trans_ptr.p->m_clientFlags |= TransClient::Commit; } - ndbrequire(!hasError(trans_ptr.p->m_error)); - runTransMaster(signal, trans_ptr); + trans_prepare_start(signal, trans_ptr); return; } while (0); @@ -19390,6 +19035,17 @@ Dbdict::handleClientReq(Signal* signal, SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; + if (trans_ptr.p->m_state == SchemaTrans::TS_SUBOP) + { + jam(); + SchemaOpPtr baseOp; + c_schemaOpPool.getPtr(baseOp, trans_ptr.p->m_curr_op_ptr_i); + op_ptr.p->m_base_op_ptr_i = baseOp.i; + } + + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + op_ptr.p->m_state = SchemaOp::OS_PARSE_MASTER; + ErrorInfo error; const OpInfo& info = getOpInfo(op_ptr); (this->*(info.m_parse))(signal, true, op_ptr, handle, error); @@ -19397,128 +19053,41 @@ Dbdict::handleClientReq(Signal* signal, if (hasError(error)) { jam(); - setError(op_ptr, error); - setTransMode(trans_ptr, TransMode::Rollback, true); - - // other nodes will not acquire an operation record - const Uint32 ownNodeId = getOwnNodeId(); - NdbNodeBitmask nodes = trans_ptr.p->m_initNodes; - nodes.bitANDC(trans_ptr.p->m_failedNodes); - nodes.clear(ownNodeId); - setGlobState(trans_ptr, nodes, TransGlob::NoOp); - + setError(trans_ptr, error); releaseSections(handle); - runTransMaster(signal, trans_ptr); + trans_rollback_sp_start(signal, trans_ptr); return; } - updateSchemaOpStep(trans_ptr, op_ptr); - sendTransParseReq(signal, op_ptr, handle); -} + trans_ptr.p->m_state = SchemaTrans::TS_PARSING; + op_ptr.p->m_state = SchemaOp::OS_PARSING; -void -Dbdict::sendTransReq(Signal* signal, SchemaTransPtr trans_ptr, - SectionHandle& handle) -{ - D("sendTransReq" << *trans_ptr.p); + Uint32 gsn = info.m_impl_req_gsn; + const Uint32* src = op_ptr.p->m_oprec_ptr.p->m_impl_req_data; - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - SchemaOpPtr op_ptr; - getOpPtr(tLoc, op_ptr); - Uint32 gsn = 0; - - // piggy-backed signal in parse phase - Uint32 extra_length = 0; - if (tLoc.m_mode == TransMode::Normal && - tLoc.m_phase == TransPhase::Parse) { - jam(); - const OpInfo& info = getOpInfo(op_ptr); - gsn = info.m_impl_req_gsn; - const Uint32* src = op_ptr.p->m_oprec_ptr.p->m_impl_req_data; - - Uint32* data = signal->getDataPtrSend(); - Uint32 skip = SchemaTransImplReq::SignalLength; - extra_length = info.m_impl_req_length; - ndbrequire(skip + extra_length <= 25); - - Uint32 i; - for (i = 0; i < extra_length; i++) - data[skip + i] = src[i]; - } + Uint32* data = signal->getDataPtrSend(); + Uint32 skip = SchemaTransImplReq::SignalLength; + Uint32 extra_length = info.m_impl_req_length; + ndbrequire(skip + extra_length <= 25); - SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); - - Uint32 phaseInfo = 0; - SchemaTransImplReq::setMode(phaseInfo, tLoc.m_mode); - SchemaTransImplReq::setPhase(phaseInfo, tLoc.m_phase); - SchemaTransImplReq::setGsn(phaseInfo, gsn); + Uint32 i; + for (i = 0; i < extra_length; i++) + data[skip + i] = src[i]; Uint32 requestInfo = 0; - if (!op_ptr.isNull()) - requestInfo = op_ptr.p->m_requestInfo; - else - requestInfo = trans_ptr.p->m_requestInfo; - - Uint32 operationInfo = 0; - if (!op_ptr.isNull()) { - SchemaTransImplReq::setOpIndex(operationInfo, op_ptr.p->m_opIndex); - SchemaTransImplReq::setOpDepth(operationInfo, op_ptr.p->m_opDepth); - } - - Uint32 iteratorInfo = 0; - SchemaTransImplReq::setListId(iteratorInfo, tLoc.m_phaseList->m_id); - SchemaTransImplReq::setListIndex(iteratorInfo, tLoc.m_phaseIndex); - SchemaTransImplReq::setItRepeat(iteratorInfo, tLoc.m_repeat); - + DictSignal::setRequestType(requestInfo, SchemaTransImplReq::RT_PARSE); + DictSignal::addRequestFlags(requestInfo, op_ptr.p->m_requestInfo); + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); req->senderRef = reference(); req->transKey = trans_ptr.p->trans_key; - req->opKey = !op_ptr.isNull() ? op_ptr.p->op_key : 0; - req->phaseInfo = phaseInfo; + req->opKey = op_ptr.p->op_key; req->requestInfo = requestInfo; - req->operationInfo = operationInfo; - req->iteratorInfo = iteratorInfo; req->clientRef = trans_ptr.p->m_clientRef; req->transId = trans_ptr.p->m_transId; + req->gsn = gsn; - NdbNodeBitmask nodes; - Uint32 nodeId; - - for (nodeId = 1; nodeId < MAX_NDB_NODES; nodeId++) { - if (!trans_ptr.p->m_initNodes.get(nodeId)) { - jam(); - continue; - } - - const TransGlob& tGlob = getTransGlob(trans_ptr, nodeId); - bool itRepeat = tGlob.m_flags & SchemaTransImplConf::IT_REPEAT; - setGlobFlags(trans_ptr, nodeId, SchemaTransImplConf::IT_REPEAT, -1); - - { - bool b1 = tGlob.state() == TransGlob::NodeFail; - bool b2 = trans_ptr.p->m_failedNodes.get(nodeId); - ndbrequire(b1 == b2); - } - - if (tGlob.state() == TransGlob::NodeFail || - tGlob.state() == TransGlob::NoTrans || - tGlob.state() == TransGlob::NoOp) { - jam(); - D("skip send (state)" << V(nodeId) << tGlob); - continue; - } - - if (tLoc.m_repeat && !itRepeat) { - D("skip send (no repeat)" << V(nodeId)); - jam(); - continue; - } - - jam(); - nodes.set(nodeId); - } - - ndbrequire(!nodes.isclear());//wl3600_todo got crash here after NF - + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; NodeReceiverGroup rg(DBDICT, nodes); { SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); @@ -19532,17 +19101,38 @@ Dbdict::sendTransReq(Signal* signal, Sch } void -Dbdict::sendTransParseReq(Signal* signal, SchemaOpPtr op_ptr, - SectionHandle& handle) +Dbdict::trans_parse_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) { - D("sendTransParseReq"); + SchemaOpPtr op_ptr; + c_schemaOpPool.getPtr(op_ptr, trans_ptr.p->m_curr_op_ptr_i); - SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - ndbrequire(tLoc.m_mode == TransMode::Normal && - tLoc.m_phase == TransPhase::Parse); - sendTransReq(signal, trans_ptr, handle); - releaseSections(handle); + op_ptr.p->m_state = SchemaOp::OS_PARSED; + + if (hasError(trans_ptr.p->m_error)) + { + jam(); + trans_rollback_sp_start(signal, trans_ptr); + return; + } + + const OpInfo& info = getOpInfo(op_ptr); + if ((this->*(info.m_subOps))(signal, op_ptr)) + { + jam(); + // more sub-ops on the way + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + trans_ptr.p->m_state = SchemaTrans::TS_SUBOP; + return; + } + + /** + * Reply to client + */ + ErrorInfo error; + (this->*(info.m_reply))(signal, op_ptr, error); + + trans_ptr.p->m_clientState = TransClient::ParseReply; + trans_ptr.p->m_state = SchemaTrans::TS_STARTED; } void @@ -19550,7 +19140,25 @@ Dbdict::execSCHEMA_TRANS_IMPL_CONF(Signa { jamEntry(); ndbrequire(signal->getNoOfSections() == 0); - recvTransReply(signal, true); + + const SchemaTransImplConf* conf = + (const SchemaTransImplConf*)signal->getDataPtr(); + + SchemaTransPtr trans_ptr; + ndbrequire(findSchemaTrans(trans_ptr, conf->transKey)); + + Uint32 senderRef = conf->senderRef; + Uint32 nodeId = refToNode(senderRef); + + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + if (!sc.clearWaitingFor(nodeId)) { + jam(); + return; + } + } + + trans_recv_reply(signal, trans_ptr); } void @@ -19558,87 +19166,44 @@ Dbdict::execSCHEMA_TRANS_IMPL_REF(Signal { jamEntry(); ndbrequire(signal->getNoOfSections() == 0); - recvTransReply(signal, false); -} -void -Dbdict::recvTransReply(Signal* signal, bool isConf) -{ - Uint32 trans_key; - Uint32 senderRef; - Uint32 itFlags = 0; - ErrorInfo error; - - if (isConf) { - jam(); - const SchemaTransImplConf* conf = - (const SchemaTransImplConf*)signal->getDataPtr(); - senderRef = conf->senderRef; - trans_key = conf->transKey; - itFlags = conf->itFlags; - } else { - jam(); - const SchemaTransImplRef* ref = - (const SchemaTransImplRef*)signal->getDataPtr(); - senderRef = ref->senderRef; - trans_key = ref->transKey; - setError(error, ref); - } - - Uint32 nodeId = refToNode(senderRef); - D("recvTransReply" << V(isConf) << V(nodeId) << V(trans_key)); + const SchemaTransImplRef* ref = + (const SchemaTransImplRef*)signal->getDataPtr(); SchemaTransPtr trans_ptr; - findSchemaTrans(trans_ptr, trans_key); - ndbrequire(!trans_ptr.isNull()); - ndbrequire(trans_ptr.p->m_initNodes.get(nodeId)); - - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - const TransGlob& tGlob = getTransGlob(trans_ptr, nodeId); - setGlobFlags(trans_ptr, nodeId, itFlags, +1); - - // clear previous round + ndbrequire(findSchemaTrans(trans_ptr, ref->transKey)); - setGlobState(trans_ptr, TransGlob::NoTrans, TransGlob::Ok); - setGlobState(trans_ptr, TransGlob::NoOp, TransGlob::Ok); - - // set remote state according to error type + Uint32 senderRef = ref->senderRef; + Uint32 nodeId = refToNode(senderRef); - if (!hasError(error)) { - jam(); - setGlobState(trans_ptr, nodeId, TransGlob::Ok); - } - else if (error.errorCode == SchemaTransImplRef::NF_FakeErrorREF) { - jam(); - // exclude the node permanently from this tx - D("node marked dead" << V(nodeId)); - setGlobState(trans_ptr, nodeId, TransGlob::NodeFail); - trans_ptr.p->m_failedNodes.set(nodeId); - resetError(error); - } - else if (error.errorCode == SchemaTransImplRef::TooManySchemaTrans) { - jam(); - D("node failed to seize trans record" << V(nodeId)); - ndbrequire(tLoc.m_mode == TransMode::Normal); - ndbrequire(tLoc.m_phase == TransPhase::Begin); - ndbrequire(tGlob.state() == TransGlob::NeedTrans); - setGlobState(trans_ptr, nodeId, TransGlob::NoTrans); - setError(trans_ptr, error); - } - else if (error.errorCode == SchemaTransImplRef::TooManySchemaOps) { + if (ref->errorCode == SchemaTransImplRef::NF_FakeErrorREF) + { jam(); - D("node failed to seize op record" << V(nodeId)); - ndbrequire(tLoc.m_mode == TransMode::Normal); - ndbrequire(tLoc.m_phase == TransPhase::Parse); - ndbrequire(tGlob.state() == TransGlob::NeedOp); - setGlobState(trans_ptr, nodeId, TransGlob::NoOp); - setError(trans_ptr, error); + // trans_ptr.p->m_nodes.clear(nodeId); + // No need to clear, will be cleared when next REQ is set } - else { + else + { jam(); - D("node returned error" << V(nodeId) << V(error.errorCode)); - setGlobState(trans_ptr, nodeId, TransGlob::Error); + ErrorInfo error; + setError(error, ref); setError(trans_ptr, error); + switch(trans_ptr.p->m_state){ + case SchemaTrans::TS_STARTING: + jam(); + trans_ptr.p->m_ref_nodes.set(nodeId); + break; + case SchemaTrans::TS_PARSING: + jam(); + if (ref->errorCode == SchemaTransImplRef::SeizeFailed) + { + jam(); + trans_ptr.p->m_ref_nodes.set(nodeId); + } + break; + default: + jam(); + } } { @@ -19649,10 +19214,59 @@ Dbdict::recvTransReply(Signal* signal, b } } - // all participants have replied to this round - handleTransReply(signal, trans_ptr); + trans_recv_reply(signal, trans_ptr); +} + +void +Dbdict::trans_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) +{ + switch(trans_ptr.p->m_state){ + case SchemaTrans::TS_INITIAL: + ndbrequire(false); + case SchemaTrans::TS_STARTING: + jam(); + trans_start_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_ABORT_START: + jam(); + trans_start_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_PARSING: + jam(); + trans_parse_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_SUBOP: + ndbrequire(false); + case SchemaTrans::TS_ROLLBACK_SP: + trans_rollback_sp_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_PREPARING: + jam(); + trans_prepare_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_ABORTING_PREPARE: + jam(); + trans_abort_prepare_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_ABORTING_PARSE: + jam(); + trans_abort_parse_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_COMMITTING: + trans_commit_recv_reply(signal, trans_ptr); + return; + case SchemaTrans::TS_COMPLETING: + trans_complete_recv_reply(signal, trans_ptr); + return; + + case SchemaTrans::TS_STARTED: // These states are waiting for client + jam(); // And should not get a "internal" reply + case SchemaTrans::TS_COMMITTED: + ndbrequire(false); + } } +#if 0 void Dbdict::handleTransReply(Signal* signal, SchemaTransPtr trans_ptr) { @@ -19778,6 +19392,7 @@ Dbdict::handleTransReply(Signal* signal, ndbrequire(false); } } +#endif void Dbdict::createSubOps(Signal* signal, SchemaOpPtr op_ptr, bool first) @@ -19785,31 +19400,21 @@ Dbdict::createSubOps(Signal* signal, Sch D("createSubOps" << *op_ptr.p); SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - if (first) { - jam(); - // raise operation depth for any sub-ops which may be created - trans_ptr.p->m_opDepth += 1; - } const OpInfo& info = getOpInfo(op_ptr); if ((this->*(info.m_subOps))(signal, op_ptr)) { jam(); // more sub-ops on the way + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + trans_ptr.p->m_state = SchemaTrans::TS_SUBOP; return; } ErrorInfo error; (this->*(info.m_reply))(signal, op_ptr, error); - // restore operation depth - ndbrequire(trans_ptr.p->m_opDepth != 0); - trans_ptr.p->m_opDepth -= 1; - - if (trans_ptr.p->m_opDepth == 0) { - jam(); - resetError(trans_ptr); - trans_ptr.p->m_clientState = TransClient::ParseReply; - } + trans_ptr.p->m_clientState = TransClient::ParseReply; + trans_ptr.p->m_state = SchemaTrans::TS_STARTED; } // a sub-op create failed, roll back and send REF to client @@ -19817,116 +19422,438 @@ void Dbdict::abortSubOps(Signal* signal, SchemaOpPtr op_ptr, ErrorInfo error) { D("abortSubOps" << *op_ptr.p << error); - SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - //const OpInfo& info = getOpInfo(op_ptr); + setError(trans_ptr, error); + trans_rollback_sp_start(signal, trans_ptr); +} + +void +Dbdict::trans_prepare_start(Signal* signal, SchemaTransPtr trans_ptr) +{ + trans_ptr.p->m_state = SchemaTrans::TS_PREPARING; - // restore operation depth - ndbrequire(trans_ptr.p->m_opDepth != 0); - trans_ptr.p->m_opDepth -= 1; + SchemaOpPtr op_ptr; + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + if (list.first(op_ptr)) + { + jam(); + trans_prepare_next(signal, trans_ptr, op_ptr); + return; + } + } - // roll back all subops and this op (the parent) - ndbrequire(hasError(error)); - setError(op_ptr, error); - setTransMode(trans_ptr, TransMode::Rollback, true); - runTransMaster(signal, trans_ptr); + trans_prepare_done(signal, trans_ptr); } void -Dbdict::runTransMaster(Signal* signal, SchemaTransPtr trans_ptr) +Dbdict::trans_prepare_next(Signal* signal, + SchemaTransPtr trans_ptr, + SchemaOpPtr op_ptr) { - D("runTransMaster"); - TransLoc& tLoc = trans_ptr.p->m_transLoc; + ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_PREPARING); - bool is_prepare = tLoc.m_phase == TransPhase::Prepare; + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + op_ptr.p->m_state = SchemaOp::OS_PREPARING; - if (tLoc.m_phase == TransPhase::Begin) { - jam(); - // skip trans with no operations - if (tLoc.m_mode == TransMode::Normal) { - while (tLoc.m_phase != TransPhase::End) { - jam(); - iteratorMove(tLoc, false); - } - } - } else { - jam(); - NdbNodeBitmask nodes = trans_ptr.p->m_initNodes; - nodes.bitANDC(trans_ptr.p->m_failedNodes); - Uint32 flags = getGlobFlags(trans_ptr, nodes); - bool repeatFlag = (flags & SchemaTransImplConf::IT_REPEAT); - bool found = iteratorMove(tLoc, repeatFlag); - ndbrequire(found); - } + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = op_ptr.p->op_key; + req->requestInfo = SchemaTransImplReq::RT_PREPARE; + req->clientRef = 0; + req->transId = trans_ptr.p->m_transId; + req->gsn = 0; - if (is_prepare && tLoc.m_phase == TransPhase::Commit) + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; + NodeReceiverGroup rg(DBDICT, nodes); { - /** - * This is commit moment... - */ - jam(); - trans_commit_start(signal, trans_ptr); - return; + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); } - // always send to all in order to keep iterators synced - SectionHandle handle(this); - handle.m_cnt = 0; - sendTransReq(signal, trans_ptr, handle); + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); } void -Dbdict::trans_prepare_start(Signal* signal, SchemaTransPtr trans_ptr) +Dbdict::trans_prepare_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) { - trans_ptr.p->m_state = SchemaTrans::TS_PREPARING; -} + jam(); -void -Dbdict::trans_prepare_next(Signal* signal, SchemaTransPtr trans_ptr) -{ - ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_PREPARING); + SchemaOpPtr op_ptr; + c_schemaOpPool.getPtr(op_ptr, trans_ptr.p->m_curr_op_ptr_i); + if (hasError(trans_ptr.p->m_error)) + { + jam(); + trans_ptr.p->m_state = SchemaTrans::TS_ABORTING_PREPARE; + trans_abort_prepare_next(signal, trans_ptr, op_ptr); + return; + } + + op_ptr.p->m_state = SchemaOp::OS_PREPARED; + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + if (list.next(op_ptr)) + { + jam(); + trans_prepare_next(signal, trans_ptr, op_ptr); + return; + } + } + + trans_prepare_done(signal, trans_ptr); + return; } void Dbdict::trans_prepare_done(Signal* signal, SchemaTransPtr trans_ptr) { ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_PREPARING); + + if (trans_ptr.p->m_clientFlags & TransClient::Commit) + { + jam(); + trans_commit_start(signal, trans_ptr); + return; + } + + // prepare not currently implemted (fully) + ndbrequire(false); } void Dbdict::trans_abort_parse_start(Signal* signal, SchemaTransPtr trans_ptr) { trans_ptr.p->m_state = SchemaTrans::TS_ABORTING_PARSE; + + SchemaOpPtr op_ptr; + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + if (list.last(op_ptr)) + { + jam(); + trans_abort_parse_next(signal, trans_ptr, op_ptr); + return; + } + } + + trans_abort_parse_done(signal, trans_ptr); + +} + +void +Dbdict::trans_abort_parse_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) +{ + SchemaOpPtr op_ptr; + c_schemaOpPool.getPtr(op_ptr, trans_ptr.p->m_curr_op_ptr_i); + + { + SchemaOpPtr last_op = op_ptr; + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + bool prev = list.prev(op_ptr); + list.remove(last_op); // Release aborted op + releaseSchemaOp(last_op); + + if (prev) + { + jam(); + trans_abort_parse_next(signal, trans_ptr, op_ptr); + return; + } + } + + trans_abort_parse_done(signal, trans_ptr); } void -Dbdict::trans_abort_parse_next(Signal* signal, SchemaTransPtr trans_ptr) +Dbdict::trans_abort_parse_next(Signal* signal, + SchemaTransPtr trans_ptr, + SchemaOpPtr op_ptr) { ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_ABORTING_PARSE); + + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + op_ptr.p->m_state = SchemaOp::OS_ABORTING_PARSE; + + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = op_ptr.p->op_key; + req->requestInfo = SchemaTransImplReq::RT_ABORT_PARSE; + req->clientRef = 0; + req->transId = trans_ptr.p->m_transId; + req->gsn = 0; + + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; + nodes.bitANDC(trans_ptr.p->m_ref_nodes); + trans_ptr.p->m_ref_nodes.clear(); + NodeReceiverGroup rg(DBDICT, nodes); + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + } + + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); } void Dbdict::trans_abort_parse_done(Signal* signal, SchemaTransPtr trans_ptr) { ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_ABORTING_PARSE); + + // unlock + const DictLockReq& lockReq = trans_ptr.p->m_lockReq; + dict_lock_unlock(signal, &lockReq); + + trans_complete_start(signal, trans_ptr); } void Dbdict::trans_abort_prepare_start(Signal* signal, SchemaTransPtr trans_ptr) { trans_ptr.p->m_state = SchemaTrans::TS_ABORTING_PREPARE; + + bool last = false; + SchemaOpPtr op_ptr; + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + last = list.last(op_ptr); + } + + if (last) + { + jam(); + trans_abort_prepare_next(signal, trans_ptr, op_ptr); + } + else + { + jam(); + trans_abort_prepare_done(signal, trans_ptr); + } } void -Dbdict::trans_abort_prepare_next(Signal* signal, SchemaTransPtr trans_ptr) +Dbdict::trans_abort_prepare_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) +{ + // XXX error states + + SchemaOpPtr op_ptr; + c_schemaOpPool.getPtr(op_ptr, trans_ptr.p->m_curr_op_ptr_i); + + op_ptr.p->m_state = SchemaOp::OS_ABORTED_PREPARE; + + bool prev = false; + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + prev = list.prev(op_ptr); + } + + if (prev) + { + jam(); + trans_abort_prepare_next(signal, trans_ptr, op_ptr); + } + else + { + jam(); + trans_abort_prepare_done(signal, trans_ptr); + } +} + +void +Dbdict::trans_abort_prepare_next(Signal* signal, + SchemaTransPtr trans_ptr, + SchemaOpPtr op_ptr) { ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_ABORTING_PREPARE); + + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + + switch(op_ptr.p->m_state){ + case SchemaOp::OS_PARSED: + jam(); + /** + * Operation has not beed prepared... + * continue with next + */ + trans_abort_prepare_recv_reply(signal, trans_ptr); + return; + case SchemaOp::OS_PREPARING: + case SchemaOp::OS_PREPARED: + break; + case SchemaOp::OS_INTIAL: + case SchemaOp::OS_PARSE_MASTER: + case SchemaOp::OS_PARSING: + case SchemaOp::OS_ABORTING_PREPARE: + case SchemaOp::OS_ABORTED_PREPARE: + case SchemaOp::OS_ABORTING_PARSE: + //case SchemaOp::OS_ABORTED_PARSE: + case SchemaOp::OS_COMMITTING: + case SchemaOp::OS_COMMITTED: +#ifndef VM_TRACE + default: +#endif + jamLine(op_ptr.p->m_state); + ndbrequire(false); + } + + op_ptr.p->m_state = SchemaOp::OS_ABORTING_PREPARE; + + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = op_ptr.p->op_key; + req->requestInfo = SchemaTransImplReq::RT_ABORT_PREPARE; + req->clientRef = 0; + req->transId = trans_ptr.p->m_transId; + req->gsn = 0; + + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; + NodeReceiverGroup rg(DBDICT, nodes); + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + } + + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); } void Dbdict::trans_abort_prepare_done(Signal* signal, SchemaTransPtr trans_ptr) { ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_ABORTING_PREPARE); + + /** + * Now run abort parse + */ + trans_abort_parse_start(signal, trans_ptr); +} + +/** + * FLOW: Rollback SP + * abort parse each operation (backwards) until op which is not subop + */ +void +Dbdict::trans_rollback_sp_start(Signal* signal, SchemaTransPtr trans_ptr) +{ + SchemaOpPtr op_ptr; + + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + ndbrequire(list.last(op_ptr)); + } + + trans_ptr.p->m_state = SchemaTrans::TS_ROLLBACK_SP; + + if (op_ptr.p->m_state == SchemaOp::OS_PARSE_MASTER) + { + jam(); + /** + * This op is only parsed at master.. + * + */ + NdbNodeBitmask nodes; + nodes.set(getOwnNodeId()); + NodeReceiverGroup rg(DBDICT, nodes); + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + + const OpInfo& info = getOpInfo(op_ptr); + (this->*(info.m_abortParse))(signal, op_ptr); + return; + } + + trans_rollback_sp_next(signal, trans_ptr, op_ptr); +} + +void +Dbdict::trans_rollback_sp_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) +{ + // TODO split trans error from op_error...? + + SchemaOpPtr op_ptr; + c_schemaOpPool.getPtr(op_ptr, trans_ptr.p->m_curr_op_ptr_i); + + if (op_ptr.p->m_base_op_ptr_i == RNIL) + { + /** + * SP + */ + trans_rollback_sp_done(signal, trans_ptr, op_ptr); + return; + } + + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + + SchemaOpPtr last_op = op_ptr; + ndbrequire(list.prev(op_ptr)); // Must have prev, as not SP + list.remove(last_op); // Release aborted op + releaseSchemaOp(last_op); + } + + trans_rollback_sp_next(signal, trans_ptr, op_ptr); +} + +void +Dbdict::trans_rollback_sp_next(Signal* signal, + SchemaTransPtr trans_ptr, + SchemaOpPtr op_ptr) +{ + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = op_ptr.p->op_key; + req->requestInfo = SchemaTransImplReq::RT_ABORT_PARSE; + req->clientRef = trans_ptr.p->m_clientRef; + req->transId = trans_ptr.p->m_transId; + req->gsn = GSN_SCHEMA_TRANS_BEGIN_REQ; + + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; + nodes.bitANDC(trans_ptr.p->m_ref_nodes); // Note: uses m_ref_nodes + trans_ptr.p->m_ref_nodes.clear(); + NodeReceiverGroup rg(DBDICT, nodes); + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + } + + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); + +} + +void +Dbdict::trans_rollback_sp_done(Signal* signal, + SchemaTransPtr trans_ptr, + SchemaOpPtr op_ptr) +{ + + ErrorInfo error = trans_ptr.p->m_error; + const OpInfo info = getOpInfo(op_ptr); + (this->*(info.m_reply))(signal, op_ptr, error); + + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + list.remove(op_ptr); + releaseSchemaOp(op_ptr); + + resetError(trans_ptr); + trans_ptr.p->m_clientState = TransClient::ParseReply; + trans_ptr.p->m_state = SchemaTrans::TS_STARTED; } void @@ -19958,30 +19885,97 @@ Dbdict::trans_commit_mutex_locked(Signal ndbrequire(trans_ptr.p->m_state == SchemaTrans::TS_COMMITTING); - SectionHandle handle(this); - handle.m_cnt = 0; - sendTransReq(signal, trans_ptr, handle); + bool first = false; + SchemaOpPtr op_ptr; + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + first = list.first(op_ptr); + } + + if (first) + { + jam(); + trans_commit_next(signal, trans_ptr, op_ptr); + } + else + { + jam(); + trans_commit_done(signal, trans_ptr); + } } void -Dbdict::trans_commit_done(Signal* signal, SchemaTransPtr trans_ptr) +Dbdict::trans_commit_next(Signal* signal, + SchemaTransPtr trans_ptr, + SchemaOpPtr op_ptr) { - ndbout_c("trans_commit_done"); - Mutex mutex(signal, c_mutexMgr, trans_ptr.p->m_commit_mutex); - Callback c = { safe_cast(&Dbdict::trans_commit_mutex_unlocked), trans_ptr.i }; - if (mutex.isNull()) + op_ptr.p->m_state = SchemaOp::OS_COMMITTING; + trans_ptr.p->m_curr_op_ptr_i = op_ptr.i; + + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = op_ptr.p->op_key; + req->requestInfo = SchemaTransImplReq::RT_COMMIT; + req->clientRef = trans_ptr.p->m_clientRef; + req->transId = trans_ptr.p->m_transId; + req->gsn = 0; + + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; + NodeReceiverGroup rg(DBDICT, nodes); + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + } + + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); +} + +void +Dbdict::trans_commit_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) +{ + if (hasError(trans_ptr.p->m_error)) { - /** - * Argh...this is a temporary fix until trans-phases are "clear" - */ jam(); - trans_commit_mutex_unlocked(signal, trans_ptr.i, 0); + // kill nodes that failed COMMIT + ndbrequire(false); + return; + } + + SchemaOpPtr op_ptr; + c_schemaOpPool.getPtr(op_ptr, trans_ptr.p->m_curr_op_ptr_i); + op_ptr.p->m_state = SchemaOp::OS_COMMITTED; + + bool next = false; + { + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + next = list.next(op_ptr); + } + + if (next) + { + jam(); + trans_commit_next(signal, trans_ptr, op_ptr); + return; } else { jam(); - mutex.unlock(c); + trans_commit_done(signal, trans_ptr); } + return; +} + +void +Dbdict::trans_commit_done(Signal* signal, SchemaTransPtr trans_ptr) +{ + ndbout_c("trans_commit_done"); + Mutex mutex(signal, c_mutexMgr, trans_ptr.p->m_commit_mutex); + Callback c = { safe_cast(&Dbdict::trans_commit_mutex_unlocked), trans_ptr.i }; + mutex.unlock(c); } void @@ -19994,45 +19988,59 @@ Dbdict::trans_commit_mutex_unlocked(Sign SchemaTransPtr trans_ptr; c_schemaTransPool.getPtr(trans_ptr, transPtrI); - if (trans_ptr.p->m_commit_mutex.isNull()) - { - jam(); - /** - * Argh...this is a temporary fix until trans-phases are "clear" - */ - } - else - { - jam(); - trans_ptr.p->m_commit_mutex.release(c_mutexMgr); - } + trans_ptr.p->m_commit_mutex.release(c_mutexMgr); - sendTransClientReply(signal, trans_ptr); // unlock const DictLockReq& lockReq = trans_ptr.p->m_lockReq; dict_lock_unlock(signal, &lockReq); - releaseSchemaTrans(trans_ptr); + + /** + * Here we should wait for SCHEMA_TRANS_COMMIT_ACK + * + * But for now, we proceed and complete transaction + */ + trans_complete_start(signal, trans_ptr); } void -Dbdict::setTransMode(SchemaTransPtr trans_ptr, - TransMode::Value new_mode, - bool hold) +Dbdict::trans_complete_start(Signal * signal, SchemaTransPtr trans_ptr) { - TransLoc& tLoc = trans_ptr.p->m_transLoc; - const TransMode::Value old_mode = tLoc.m_mode; + trans_ptr.p->m_state = SchemaTrans::TS_COMPLETING; + + SchemaTransImplReq* req = (SchemaTransImplReq*)signal->getDataPtrSend(); + req->senderRef = reference(); + req->transKey = trans_ptr.p->trans_key; + req->opKey = RNIL; + req->requestInfo = SchemaTransImplReq::RT_COMPLETE; + req->clientRef = trans_ptr.p->m_clientRef; + req->transId = trans_ptr.p->m_transId; + req->gsn = 0; - D("setTransMode" - << *trans_ptr.p - << " old:" << DictSignal::getTransModeName(old_mode) - << " new:" << DictSignal::getTransModeName(new_mode) - << V(hold)); - - tLoc.m_mode = new_mode; - tLoc.m_hold = hold; - NdbNodeBitmask nodes = trans_ptr.p->m_initNodes; - setGlobFlags(trans_ptr, nodes, SchemaTransImplConf::IT_REPEAT, -1); - tLoc.m_repeat = 0; + trans_ptr.p->m_nodes.bitAND(c_aliveNodes); + NdbNodeBitmask nodes = trans_ptr.p->m_nodes; + NodeReceiverGroup rg(DBDICT, nodes); + { + SafeCounter sc(c_counterMgr, trans_ptr.p->m_counter); + bool ok = sc.init<SchemaTransImplRef>(rg, trans_ptr.p->trans_key); + ndbrequire(ok); + } + + sendSignal(rg, GSN_SCHEMA_TRANS_IMPL_REQ, signal, + SchemaTransImplReq::SignalLength, JBB); +} + +void +Dbdict::trans_complete_recv_reply(Signal* signal, SchemaTransPtr trans_ptr) +{ + trans_complete_done(signal, trans_ptr); +} + +void +Dbdict::trans_complete_done(Signal* signal, SchemaTransPtr trans_ptr) +{ + sendTransClientReply(signal, trans_ptr); + + releaseSchemaTrans(trans_ptr); } // participant @@ -20050,8 +20058,8 @@ Dbdict::execSCHEMA_TRANS_IMPL_REQ(Signal *(const SchemaTransImplReq*)signal->getDataPtr(); const SchemaTransImplReq *req = &reqCopy; - Uint32 i = ~0; - if (i == SchemaTransImplReq::RT_START) + const Uint32 rt = DictSignal::getRequestType(req->requestInfo); + if (rt == SchemaTransImplReq::RT_START) { jam(); slave_run_start(signal, req); @@ -20071,14 +20079,14 @@ Dbdict::execSCHEMA_TRANS_IMPL_REQ(Signal /** * Check *transaction* request */ - switch(i){ + switch(rt){ case SchemaTransImplReq::RT_PARSE: jam(); slave_run_parse(signal, trans_ptr, req); return; case SchemaTransImplReq::RT_COMPLETE: jam(); - //slave_run_complete(signal, trans_ptr); + slave_run_complete(signal, trans_ptr); return; case SchemaTransImplReq::RT_FLUSH_SCHEMA: jam(); @@ -20099,7 +20107,7 @@ Dbdict::execSCHEMA_TRANS_IMPL_REQ(Signal { const OpInfo info = getOpInfo(op_ptr); - switch(i){ + switch(rt){ case SchemaTransImplReq::RT_START: case SchemaTransImplReq::RT_PARSE: case SchemaTransImplReq::RT_COMPLETE: @@ -20111,7 +20119,17 @@ Dbdict::execSCHEMA_TRANS_IMPL_REQ(Signal return; case SchemaTransImplReq::RT_ABORT_PARSE: jam(); + ndbrequire(op_ptr.p->nextList == RNIL); (this->*(info.m_abortParse))(signal, op_ptr); + if (!trans_ptr.p->m_isMaster) + { + /** + * Remove op (except at coordinator + */ + LocalDLFifoList<SchemaOp> list(c_schemaOpPool, trans_ptr.p->m_op_list); + list.remove(op_ptr); + releaseSchemaOp(op_ptr); + } return; case SchemaTransImplReq::RT_ABORT_PREPARE: jam(); @@ -20124,7 +20142,6 @@ Dbdict::execSCHEMA_TRANS_IMPL_REQ(Signal } } - recvTransReq(signal); return; err: ndbrequire(false); @@ -20147,16 +20164,26 @@ Dbdict::slave_run_start(Signal *signal, } trans_ptr.p->m_clientRef = req->clientRef; trans_ptr.p->m_transId = req->transId; + trans_ptr.p->m_isMaster = false; + trans_ptr.p->m_masterRef = req->senderRef; + trans_ptr.p->m_requestInfo = req->requestInfo; } else { jam(); + // this branch does nothing but is convenient for signal pong ndbrequire(findSchemaTrans(trans_ptr, req->transKey)); } sendTransConf(signal, trans_ptr); return; err: + SchemaTrans tmp_trans; + trans_ptr.i = RNIL; + trans_ptr.p = &tmp_trans; + trans_ptr.p->trans_key = trans_key; + trans_ptr.p->m_masterRef = req->senderRef; + setError(trans_ptr.p->m_error, error); sendTransRef(signal, trans_ptr); } @@ -20169,13 +20196,12 @@ Dbdict::slave_run_parse(Signal *signal, D("slave_run_parse"); const Uint32 op_key = req->opKey; - const Uint32 phaseInfo = req->phaseInfo; - const Uint32 gsn = SchemaTransImplReq::getGsn(phaseInfo); + const Uint32 gsn = req->gsn; const Uint32 requestInfo = req->requestInfo; const OpInfo& info = *findOpInfo(gsn); // signal data contains impl_req - const Uint32* src = signal->getDataPtr(); + const Uint32* src = signal->getDataPtr() + SchemaTransImplReq::SignalLength; const Uint32 len = info.m_impl_req_length; SectionHandle handle(this, signal); @@ -20208,12 +20234,8 @@ Dbdict::slave_run_parse(Signal *signal, Uint32* dst = oprec_ptr.p->m_impl_req_data; memcpy(dst, src, len << 2); + addSchemaOp(trans_ptr, op_ptr); (this->*(info.m_parse))(signal, false, op_ptr, handle, error); - if (!hasError(error)) - { - jam(); - addSchemaOp(trans_ptr, op_ptr); - } } else { jam(); setError(error, SchemaTransImplRef::TooManySchemaOps, __LINE__); @@ -20226,6 +20248,7 @@ Dbdict::slave_run_parse(Signal *signal, if (hasError(error)) { jam(); + setError(trans_ptr, error); sendTransRef(signal, trans_ptr); return; } @@ -20236,347 +20259,31 @@ void Dbdict::slave_flush_schema(Signal *signal, SchemaTransPtr trans_ptr) { -} -void -Dbdict::recvTransReq(Signal* signal) -{ - const SchemaTransImplReq* req = - (const SchemaTransImplReq*)signal->getDataPtr(); - - D("recvTransReq" << V(signal->getLength())); - - const Uint32 senderRef = req->senderRef; - const Uint32 trans_key = req->transKey; - const Uint32 op_key = req->opKey; - const Uint32 phaseInfo = req->phaseInfo; - const Uint32 requestInfo = req->requestInfo; - const Uint32 operationInfo = req->operationInfo; - const Uint32 iteratorInfo = req->iteratorInfo; - const Uint32 clientRef = req->clientRef; - const Uint32 transId = req->transId; - - // unpack phaseInfo - const Uint32 mode = SchemaTransImplReq::getMode(phaseInfo); - const Uint32 phase = SchemaTransImplReq::getPhase(phaseInfo); - const Uint32 gsn = SchemaTransImplReq::getGsn(phaseInfo); - - // unpack operation info - const Uint32 opDepth = SchemaTransImplReq::getOpDepth(operationInfo); - - // unpack iterator info - const Uint32 phaseIndex = SchemaTransImplReq::getListIndex(iteratorInfo); - const Uint32 repeat = SchemaTransImplReq::getItRepeat(iteratorInfo); - - // cannot receive a new master before this signal arrives - // enable when RF_LOCAL_TRANS is in req - //ndbrequire(c_masterNodeId == refToNode(senderRef)); - - SchemaTransPtr trans_ptr; - ErrorInfo error; - do { - const bool isMaster = (senderRef == reference()); - - const bool seizeTrans = - !isMaster && mode == TransMode::Normal && phase == TransPhase::Begin; - - const bool releaseTrans = - !isMaster && mode == TransMode::Normal && phase == TransPhase::End || - !isMaster && mode != TransMode::Normal && phase == TransPhase::Begin; - - const bool sendConf = - phase == TransPhase::Begin || phase == TransPhase::End; - - if (seizeTrans) { - jam(); - if (!seizeSchemaTrans(trans_ptr, trans_key)) { - jam(); - setError(error, SchemaTransImplRef::TooManySchemaTrans, __LINE__); - break; - } - trans_ptr.p->m_clientRef = clientRef; - trans_ptr.p->m_transId = transId; - } else { - jam(); - findSchemaTrans(trans_ptr, trans_key); - ndbrequire(!trans_ptr.isNull()); - } - - TransLoc& tLoc = trans_ptr.p->m_transLoc; - - if (isMaster) { - ndbrequire(trans_ptr.p->m_isMaster == true); - ndbrequire(trans_ptr.p->m_masterRef == senderRef); - ndbrequire((Uint32)tLoc.m_mode == mode); - ndbrequire((Uint32)tLoc.m_phase == phase); - } else { - jam(); - if (trans_ptr.p->m_masterRef == 0) { - jam(); - // first time - trans_ptr.p->m_isMaster = false; - trans_ptr.p->m_masterRef = senderRef; - } else { - // until MNF anyway - ndbrequire(trans_ptr.p->m_isMaster == false); - ndbrequire(trans_ptr.p->m_masterRef == senderRef); - } - trans_ptr.p->m_opDepth = opDepth; - iteratorCopy(tLoc, mode, phaseIndex, op_key, repeat); - ndbrequire((Uint32)tLoc.m_phase == phase); - } - - if (tLoc.m_mode == TransMode::Normal) { - if (tLoc.m_phase == TransPhase::Begin) { - jam(); - DictSignal::addRequestFlags(trans_ptr.p->m_requestInfo, requestInfo); - } - else if (tLoc.m_phase == TransPhase::Parse) { - jam(); - const OpInfo* info = findOpInfo(gsn); - ndbrequire(info != 0); - Uint32 length = info->m_impl_req_length; - Uint32 skip = SchemaTransImplReq::SignalLength; - ndbrequire(skip + length <= 25); - - Uint32* data = signal->getDataPtrSend(); - Uint32 i = 0; - for (i = 0; i < length; i++) - data[i] = data[skip + i]; - - recvTransParseReq(signal, trans_ptr, op_key, *info, requestInfo); - } - else if (tLoc.m_phase == TransPhase::Prepare) { - jam(); - runTransSlave(signal, trans_ptr); - } - else if (tLoc.m_phase == TransPhase::Commit) { - jam(); - runTransSlave(signal, trans_ptr); - } - else if (tLoc.m_phase == TransPhase::End) { - jam(); - } - else { - ndbrequire(false); - } - } - else if (tLoc.m_mode == TransMode::Rollback) { - if (tLoc.m_phase == TransPhase::Parse) { - jam(); - /* - * Rolling back sub-operation on participant. - * Non-master removes the op before replying. - */ - runTransSlave(signal, trans_ptr); - } - else { - ndbrequire(false); - } - } - else if (tLoc.m_mode == TransMode::Abort) { - if (tLoc.m_phase == TransPhase::Begin) { - jam(); - } - else if (tLoc.m_phase == TransPhase::Parse) { - jam(); - runTransSlave(signal, trans_ptr); - } - else if (tLoc.m_phase == TransPhase::Prepare) { - jam(); - runTransSlave(signal, trans_ptr); - } - else { - ndbrequire(false); - } - } - else { - ndbrequire(false); - } - - if (sendConf) { - jam(); - sendTransConf(signal, trans_ptr); - } - if (releaseTrans) { - jam(); - releaseSchemaTrans(trans_ptr); - } - return; - } while (0); - - SchemaTrans tmp_trans; - trans_ptr.i = RNIL; - trans_ptr.p = &tmp_trans; - trans_ptr.p->trans_key = trans_key; - trans_ptr.p->m_masterRef = senderRef; - setError(trans_ptr.p->m_error, error); - sendTransRef(signal, trans_ptr); } -/* - * Parse on master has been done before this request was sent. - * Create and parse the operation on slave participant. - */ void -Dbdict::recvTransParseReq(Signal* signal, SchemaTransPtr trans_ptr, - Uint32 op_key, const OpInfo& info, - Uint32 requestInfo) +Dbdict::slave_run_complete(Signal* signal, + SchemaTransPtr trans_ptr) { - SchemaOpPtr op_ptr; - D("recvTransParseReq"); - - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - D("before add op" << *trans_ptr.p << tLoc << tLoc.m_opList); - - // signal data contains impl_req - const Uint32* src = signal->getDataPtr(); - const Uint32 len = info.m_impl_req_length; - - SectionHandle handle(this, signal); - - ndbrequire(op_key != RNIL); - ErrorInfo error; - if (trans_ptr.p->m_isMaster) - { - jam(); - // this branch does nothing but is convenient for signal pong - findSchemaOp(op_ptr, op_key); - ndbrequire(!op_ptr.isNull()); - - OpRecPtr oprec_ptr = op_ptr.p->m_oprec_ptr; - const Uint32* dst = oprec_ptr.p->m_impl_req_data; - ndbrequire(memcmp(dst, src, len << 2) == 0); - } - else - { - if (seizeSchemaOp(op_ptr, op_key, info)) - { - jam(); - - DictSignal::addRequestExtra(op_ptr.p->m_requestInfo, requestInfo); - DictSignal::addRequestFlags(op_ptr.p->m_requestInfo, requestInfo); - - addSchemaOp(trans_ptr, op_ptr); - - OpRecPtr oprec_ptr = op_ptr.p->m_oprec_ptr; - Uint32* dst = oprec_ptr.p->m_impl_req_data; - memcpy(dst, src, len << 2); - - (this->*(info.m_parse))(signal, false, op_ptr, handle, error); - if (!hasError(error)) - { - jam(); - updateSchemaOpStep(trans_ptr, op_ptr); - } - } else { - jam(); - setError(error, SchemaTransImplRef::TooManySchemaOps, __LINE__); - } - } - - // parse must consume but not release signal sections - releaseSections(handle); - - if (hasError(error)) + sendTransConf(signal, trans_ptr); + if (!trans_ptr.p->m_isMaster) { jam(); - setError(trans_ptr, error); - sendTransRef(signal, trans_ptr); - return; - } - sendTransConf(signal, op_ptr); -} - -void -Dbdict::runTransSlave(Signal* signal, SchemaTransPtr trans_ptr) -{ - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - SchemaOpPtr op_ptr; - getOpPtr(tLoc, op_ptr); - - D("runTransSlave"); - - const TransPhaseEntry& phaseEntry = getPhaseEntry(tLoc); - const bool isMaster = (trans_ptr.p->m_masterRef == reference()); - - // master phase on slave does nothing more - if (!isMaster && (phaseEntry.m_phaseFlags & TransPhaseFlag::Master)) { - jam(); - sendTransConf(signal, trans_ptr); - return; - } - - if (phaseEntry.m_phaseFlags & TransPhaseFlag::Simple) { - jam(); - ndbrequire(op_ptr.isNull()); - if (tLoc.m_mode == TransMode::Normal) { - jam(); - ndbrequire(phaseEntry.m_run != 0); - (this->*(phaseEntry.m_run))(signal, trans_ptr); - } else { - jam(); - ndbrequire(phaseEntry.m_abort != 0); - (this->*(phaseEntry.m_abort))(signal, trans_ptr); - } - } else { - jam(); - ndbrequire(!op_ptr.isNull()); - // wl3600_todo verify op_key - const OpInfo& info = getOpInfo(op_ptr); - if (tLoc.m_mode == TransMode::Normal) { - switch (tLoc.m_phase) { - case TransPhase::Prepare: - (this->*(info.m_prepare))(signal, op_ptr); - break; - case TransPhase::Commit: - (this->*(info.m_commit))(signal, op_ptr); - break; - default: - ndbrequire(false); - break; - } - } else { - switch (tLoc.m_phase) { - case TransPhase::Parse: - (this->*(info.m_abortParse))(signal, op_ptr); - break; - case TransPhase::Prepare: - (this->*(info.m_abortPrepare))(signal, op_ptr); - // mark it in advance - op_ptr.p->m_abortPrepareDone = true; - break; - default: - ndbrequire(false); - break; - } - } + releaseSchemaTrans(trans_ptr); } } void -Dbdict::sendTransConf(Signal* signal, SchemaOpPtr op_ptr, Uint32 itFlags) +Dbdict::sendTransConf(Signal* signal, SchemaOpPtr op_ptr) { SchemaTransPtr trans_ptr = op_ptr.p->m_trans_ptr; - TransLoc& tLoc = trans_ptr.p->m_transLoc; - - if (!op_ptr.isNull() && - tLoc.m_mode == TransMode::Normal) { - jam(); - runFeedbackMethod(signal, op_ptr); - } - if (tLoc.m_mode == TransMode::Rollback && - !trans_ptr.p->m_isMaster) { - jam(); - iteratorRemoveLastOp(tLoc, op_ptr); - } - - sendTransConf(signal, trans_ptr, itFlags); + sendTransConf(signal, trans_ptr); } void -Dbdict::sendTransConf(Signal* signal, SchemaTransPtr trans_ptr, Uint32 itFlags) +Dbdict::sendTransConf(Signal* signal, SchemaTransPtr trans_ptr) { ndbrequire(!trans_ptr.isNull()); ndbrequire(signal->getNoOfSections() == 0); @@ -20585,7 +20292,6 @@ Dbdict::sendTransConf(Signal* signal, Sc (SchemaTransImplConf*)signal->getDataPtrSend(); conf->senderRef = reference(); conf->transKey = trans_ptr.p->trans_key; - conf->itFlags = itFlags; const Uint32 masterRef = trans_ptr.p->m_masterRef; @@ -20625,24 +20331,6 @@ Dbdict::sendTransRef(Signal* signal, Sch ndbrequire(hasError(trans_ptr.p->m_error)); // trans is not defined on RT_BEGIN REF - if (trans_ptr.i != RNIL) { - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - - if (tLoc.m_mode == TransMode::Normal) { - switch (tLoc.m_phase) { - case TransPhase::Commit: - case TransPhase::End: - // fail commit must crash - ndbrequire(false); - break; - default: - break; - } - } else { - // fail abort must crash - ndbrequire(false); - } - } SchemaTransImplRef* ref = (SchemaTransImplRef*)signal->getDataPtrSend(); @@ -20730,42 +20418,6 @@ Dbdict::sendTransClientReply(Signal* sig ndbrequire(false); } -/* - * Slave node failure. If we are waiting for reply then we get it - * via NF_FakeErrorREF. Otherwise we get it only via NODE_FAILREP. - * This second case is handled here. - */ - -void -Dbdict::handleTransSlaveFail(Signal* signal, Uint32 failedNode) -{ - D("handleTransSlaveFail" << V(failedNode)); - ndbrequire(failedNode < MAX_NDB_NODES); - - SchemaTransPtr trans_ptr; - c_schemaTransList.first(trans_ptr); - while (trans_ptr.i != RNIL) { - jam(); - const TransGlob& tGlob = getTransGlob(trans_ptr, failedNode); - D("check" << *trans_ptr.p << tGlob); - - switch (tGlob.state()) { - case TransGlob::Undef: - D("not part of this trans"); - break; - case TransGlob::NodeFail: - D("node failure seen already"); - ndbrequire(trans_ptr.p->m_failedNodes.get(failedNode)); - break; - default: - setGlobState(trans_ptr, failedNode, TransGlob::NodeFail); - trans_ptr.p->m_failedNodes.set(failedNode); - break; - } - - c_schemaTransList.next(trans_ptr); - } -} // DICT as schema trans client @@ -20949,24 +20601,28 @@ Dbdict::handleApiFail(Signal* signal, c_schemaTransList.first(trans_ptr); while (trans_ptr.i != RNIL) { jam(); - const TransLoc& tLoc = trans_ptr.p->m_transLoc; - D("check" << *trans_ptr.p << tLoc << tLoc.m_opList); + D("check" << *trans_ptr.p); Uint32 clientRef = trans_ptr.p->m_clientRef; - if (refToNode(clientRef) == failedApiNode) { + if (refToNode(clientRef) == failedApiNode) + { jam(); D("failed" << hex << V(clientRef)); ndbrequire(!(trans_ptr.p->m_clientFlags & TransClient::ApiFail)); trans_ptr.p->m_clientFlags |= TransClient::ApiFail; - if (trans_ptr.p->m_isMaster) { + if (trans_ptr.p->m_isMaster) + { jam(); - if (trans_ptr.p->m_clientFlags & TransClient::TakeOver) { + if (trans_ptr.p->m_clientFlags & TransClient::TakeOver) + { // maybe already running in background jam(); ndbrequire(trans_ptr.p->m_clientFlags & TransClient::Background); - } else { + } + else + { takeOverTransClient(signal, trans_ptr); } @@ -21258,99 +20914,21 @@ Dbdict::SchemaOp::print(NdbOut& out) con out << " (SchemaOp"; out << " " << info.m_opType; out << dec << V(op_key); - out << dec << V(m_opIndex); - out << dec << V(m_opDepth); if (m_error.errorCode != 0) out << m_error; if (m_sections != 0) out << dec << V(m_sections); - out << ")"; -} - -// OpList - -NdbOut& -operator<<(NdbOut& out, const Dbdict::OpList& a) -{ - a.print(out); - return out; -} - -void -Dbdict::OpList::print(NdbOut& out) const -{ - Dbdict* dict = (Dbdict*)globalData.getBlock(DBDICT); - Dbdict::OpList a_copy = *this; - LocalDLFifoList<SchemaOp> list(dict->c_schemaOpPool, a_copy.m_head); - Dbdict::SchemaOpPtr op_ptr; - out << " (OpList"; - Uint32 depth = 0; - out << " ["; - list.first(op_ptr); - while (!op_ptr.isNull()) { - if (op_ptr.p->m_opDepth > depth) { - assert(op_ptr.p->m_opDepth > depth); - out << " ["; - } - if (op_ptr.p->m_opDepth < depth) { - Uint32 i; - for (i = op_ptr.p->m_opDepth; i < depth; i++) - out << "]"; - } - const Dbdict::OpInfo& info = dict->getOpInfo(op_ptr); - out << " " << dec << op_ptr.p->op_key << "-" << info.m_opType; - depth = op_ptr.p->m_opDepth; - list.next(op_ptr); - } - while (depth != 0) { - out << "]"; - depth--; + if (m_base_op_ptr_i == RNIL) + out << "-> RNIL"; + else + { + out << "-> " << m_base_op_ptr_i; } - out << "]"; - out << ")"; -} - -// TransLoc -NdbOut& -operator<<(NdbOut& out, const Dbdict::TransLoc& a) -{ - a.print(out); - return out; -} - -void -Dbdict::TransLoc::print(NdbOut& out) const -{ - out << " (TransLoc"; - out << dec << V(m_mode) - << " [" << DictSignal::getTransModeName(m_mode) << "]"; - out << dec << V(m_phaseIndex); - out << dec << V(m_phase) - << " [" << DictSignal::getTransPhaseName(m_phase) << "]"; - out << dec << V(m_repeat); - out << dec << V(m_opKey); out << ")"; } -// TransGlob - -NdbOut& -operator<<(NdbOut& out, const Dbdict::TransGlob& a) -{ - a.print(out); - return out; -} -void -Dbdict::TransGlob::print(NdbOut& out) const -{ - out << " "; - out << dec << (Uint32)m_state; - if (m_flags != 0) - out << dec << "," << m_flags; - out << " [" << DictSignal::getTransStateName(m_state) << "]"; -} // SchemaTrans @@ -21365,22 +20943,13 @@ void Dbdict::SchemaTrans::print(NdbOut& out) const { out << " (SchemaTrans"; + out << dec << V(m_state); out << dec << V(trans_key); out << dec << V(m_isMaster); out << hex << V(m_clientRef); out << hex << V(m_transId); out << dec << V(m_clientState); out << hex << V(m_clientFlags); - out << dec << V(m_opDepth); - out << " State"; - Uint32 i; - for (i = 1; i < MAX_NDB_NODES; i++) { - if (m_initNodes.get(i)) { - out << dec << " " << i << ":"; - TransGlob::State state = m_transGlob[i].state(); - out << DictSignal::getTransStateName(state); - } - } out << ")"; } @@ -21416,7 +20985,6 @@ void Dbdict::check_consistency() { D("check_consistency"); - ndbrequire(c_schemaTransCount == 0); // schema file entries // mis-named "tables" TableRecordPtr tablePtr; diff -Nrup a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp --- a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp 2008-02-06 21:13:49 +01:00 +++ b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp 2008-02-06 21:15:12 +01:00 @@ -1244,65 +1244,6 @@ private: void resetError(SchemaTransPtr); void resetError(TxHandlePtr); - // transaction mode and phase - - struct TransMode { - enum Value { - Undef = 0, - Normal, - Rollback, // rolling back current op and sub-ops - Abort // rolling back entire transaction - }; - }; - - struct TransPhase { - enum Value { - Undef = 0, - Begin, - Parse, - Prepare, - Commit, - Complete, - End - }; - }; - - struct TransStep { - TransMode::Value m_mode; - TransPhase::Value m_phase; - Uint32 m_repeat; - TransStep() { - m_mode = TransMode::Undef; - m_phase = TransPhase::Undef; - m_repeat = 0; - } - }; - - struct TransPhaseFlag { - enum Value { - Master = (1 << 0), // no-op on slave participant - Simple = (1 << 1) // not iterated over operations - }; - }; - - struct TransPhaseEntry { - TransPhase::Value m_phase; - Uint32 m_phaseFlags; - // for Simple phase - void (Dbdict::*m_run)(Signal*, SchemaTransPtr); - void (Dbdict::*m_abort)(Signal*, SchemaTransPtr); - }; - - struct TransPhaseList { - Uint32 m_id; - Uint32 m_length; - const TransPhaseEntry* const m_phaseEntry; - }; - - // default (and only) phase list - static const TransPhaseEntry g_defaultPhaseEntry[]; - static const TransPhaseList g_defaultPhaseList; - // OpInfo struct OpInfo { @@ -1335,29 +1276,12 @@ private: static const OpInfo* g_opInfoList[]; const OpInfo* findOpInfo(Uint32 gsn); - // Feedback is a sub-op callback for a parent op on all nodes - - enum FeedbackId { - AlterIndex_atCreateTrigger = 1, - AlterIndex_atDropTrigger = 2, - BuildIndex_atCreateConstraint = 3, - BuildIndex_atDropConstraint = 4 - }; - - struct FeedbackEntry { - Uint32 m_id; - TransPhase::Value m_phase; - void (Dbdict::*m_method)(Signal*, SchemaOpPtr); - }; - - static const FeedbackEntry g_feedbackTable[]; - const FeedbackEntry* findFeedbackEntry(Uint32 id, TransPhase::Value phase); - void runFeedbackMethod(Signal*, SchemaOpPtr); - // OpRec struct OpRec { + char m_opType[4]; + Uint32 nextPool; // reference to the static member in subclass @@ -1373,8 +1297,6 @@ private: SchemaFile::TableEntry m_orig_entry; SchemaFile::TableEntry m_curr_entry; - char m_opType[4]; - OpRec(const OpInfo& info, Uint32* impl_req_data) : m_opInfo(info), m_impl_req_data(impl_req_data) { @@ -1429,20 +1351,23 @@ private: enum OpState { - OS_PARSING = 1, - OS_PARSED = 2, - OS_PREPARING = 3, - OS_PREPARED = 4, - OS_ABORTING_PREPARE = 5, - OS_ABORTED_PREPARE = 6, - OS_ABORTING_PARSE = 7, - //OS_ABORTED_PARSE = 8 // Not used, op released - OS_COMMITTING = 9, - //OS_COMMITTED = 10 // Not used, op released + OS_INTIAL = 0, + OS_PARSE_MASTER = 1, + OS_PARSING = 2, + OS_PARSED = 3, + OS_PREPARING = 4, + OS_PREPARED = 5, + OS_ABORTING_PREPARE = 6, + OS_ABORTED_PREPARE = 7, + OS_ABORTING_PARSE = 8, + //OS_ABORTED_PARSE = 9, // Not used, op released + OS_COMMITTING = 10, + OS_COMMITTED = 11 // TODO: Not used, op released }; Uint32 m_state; Uint32 op_key; + Uint32 m_base_op_ptr_i; Uint32 nextHash; Uint32 prevHash; Uint32 hashValue() const { @@ -1475,22 +1400,10 @@ private: // callback for use with sub-operations Callback m_callback; - // index (0-based) of operation in schema trans op list - Uint32 m_opIndex; - - // sub-operation tree structure expressed as depth (top level = 0) - Uint32 m_opDepth; - // link to an extra "helper" op and link back from it SchemaOpPtr m_oplnk_ptr; SchemaOpPtr m_opbck_ptr; - // last successfully completed step - TransStep m_step; - - // flag if this op has been aborted in RT_PREPARE phase - bool m_abortPrepareDone; - // error always propagates to trans level ErrorInfo m_error; @@ -1507,12 +1420,10 @@ private: m_sections = 0; m_callback.m_callbackFunction = 0; m_callback.m_callbackData = 0; - m_opIndex = ~(Uint32)0; - m_opDepth = 0; m_oplnk_ptr.setNull(); m_opbck_ptr.setNull(); - m_abortPrepareDone = false; m_magic = 0; + m_base_op_ptr_i = RNIL; } SchemaOp(Uint32 the_op_key) { @@ -1657,113 +1568,6 @@ private: void releaseDictObject(SchemaOpPtr); void findDictObjectOp(SchemaOpPtr&, DictObjectPtr); - // list of SchemaOp (sans pool) - struct OpList { - DLFifoList<SchemaOp>::Head m_head; - Uint32 m_size; - OpList() { - m_size = 0; - } -#ifdef VM_TRACE - void print(NdbOut&) const; -#endif - }; - - /* - * Transaction state, location, and iteration. Maintained - * by master and copied to slaves on each round. - * - * Iteration is over phases, operations, and repeats. - */ - - struct TransLoc : public TransStep { - const TransPhaseList* m_phaseList; - Uint32 m_phaseIndex; - OpList m_opList; - Uint32 m_opKey; - bool m_hold; // master-local temp flag - TransLoc () { - m_phaseIndex = 0; - m_opKey = 0; - m_hold = false; - } -#ifdef VM_TRACE - void print(NdbOut&) const; -#endif - }; - - bool getOpPtr(const TransLoc&, SchemaOpPtr&); - const TransPhaseEntry& getPhaseEntry(const TransLoc&); - - void iteratorAddOp(TransLoc&, SchemaOpPtr); - void iteratorRemoveLastOp(TransLoc&, SchemaOpPtr&); - bool iteratorFirstOp(TransLoc&, int dir); - bool iteratorNextOp(TransLoc&, int dir); - - bool iteratorFirst(TransLoc&, int dir); - bool iteratorNext(TransLoc&, int dir); - bool iteratorInit(TransLoc&); - bool iteratorMove(TransLoc&, bool repeatFlag); - void iteratorCopy(TransLoc&, Uint32 mode, - Uint32 phaseIndex, Uint32 op_key, Uint32 repeat); - - /* - * Transaction state of each node tracked by master only. - * Also records requests for repeat after each round. - */ - - struct TransGlob { - enum State { - Undef = 0, - Ok, - Error, - NodeFail, - // handle cases where slave has no free record - NeedTrans, - NoTrans, - NeedOp, - NoOp, - // end valid values - End - }; - // pack State to save space - Uint8 m_state; - Uint8 m_flags; - TransGlob() { - m_state = Undef; - m_flags = 0; - } - TransGlob(State state, Uint32 flags) { - m_state = (Uint8)state; - m_flags = flags; - } - State state() const { - return (State)m_state; - } - void state(State state) { - assert(state > 0 && state < End); - m_state = (Uint8)state; - } -#ifdef VM_TRACE - void print(NdbOut&) const; -#endif - }; - - TransGlob& getTransGlob(SchemaTransPtr, Uint32 nodeId); - - void setGlobState(SchemaTransPtr, - Uint32 nodeId, TransGlob::State state); - void setGlobState(SchemaTransPtr, - const NdbNodeBitmask&, TransGlob::State state); - void setGlobState(SchemaTransPtr, - TransGlob::State oldstate, TransGlob::State state); - void setGlobFlags(SchemaTransPtr, - Uint32 nodeId, Uint32 flags, int op); - void setGlobFlags(SchemaTransPtr, - const NdbNodeBitmask&, Uint32 flags, int op); - Uint32 getGlobFlags(SchemaTransPtr trans_ptr, - const NdbNodeBitmask& nodes); - /* * Trans client is the API client (not us, for recursive ops). * Its state is shared by SchemaTrans / TxHandle (for takeover). @@ -1781,7 +1585,8 @@ private: enum Flag { ApiFail = 1, Background = 2, - TakeOver = 4 + TakeOver = 4, + Commit = 8 }; }; @@ -1793,17 +1598,19 @@ private: enum TransState { + TS_INITIAL = 0, TS_STARTING = 1, // Starting at participants - TS_STARTED = 2, // Started (potentially with parsed ops) - TS_PARSING = 3, // Parsing at participants - TS_SUBOP = 4, // Creating subop - TS_ROLLBACK_SP = 5, // Rolling back to SP (supported before prepare) - TS_PREPARING = 6, // Preparing operations - TS_ABORTING_PREPARE = 7, // Aborting prepared operations - TS_ABORTING_PARSE = 8, // Aborting parsed operations - TS_COMMITTING = 9, // Committing - TS_COMMITTED = 10,// Committed - TS_COMPLETING = 11,// Completing + TS_ABORT_START = 2, // Endreq on all START_CONF + TS_STARTED = 3, // Started (potentially with parsed ops) + TS_PARSING = 4, // Parsing at participants + TS_SUBOP = 5, // Creating subop + TS_ROLLBACK_SP = 6, // Rolling back to SP (supported before prepare) + TS_PREPARING = 7, // Preparing operations + TS_ABORTING_PREPARE = 8, // Aborting prepared operations + TS_ABORTING_PARSE = 9, // Aborting parsed operations + TS_COMMITTING = 10,// Committing + TS_COMMITTED = 11,// Committed + TS_COMPLETING = 12,// Completing }; Uint32 m_state; @@ -1834,20 +1641,16 @@ private: Uint32 m_clientFlags; Uint32 m_takeOverTxKey; - // local and global transaction state - TransLoc m_transLoc; - TransGlob m_transGlob[MAX_NDB_NODES]; + NdbNodeBitmask m_nodes; // Nodes part of transaction + NdbNodeBitmask m_ref_nodes; // Nodes replying REF to req + SafeCounterHandle m_counter; // Outstanding REQ's - NdbNodeBitmask m_initNodes; - NdbNodeBitmask m_failedNodes; - SafeCounterHandle m_counter; + Uint32 m_curr_op_ptr_i; + DLFifoList<SchemaOp>::Head m_op_list; // request for lock/unlock DictLockReq m_lockReq; - // operation depth during parse (increased for sub-ops) - Uint32 m_opDepth; - // callback (not yet used) Callback m_callback; @@ -1864,6 +1667,7 @@ private: Uint32 m_magic; SchemaTrans() { + m_state = TS_INITIAL; m_isMaster = false; m_masterRef = 0; m_requestInfo = 0; @@ -1872,10 +1676,7 @@ private: m_clientState = TransClient::StateUndef; m_clientFlags = 0; m_takeOverTxKey = 0; - m_initNodes.clear(); - m_failedNodes.clear(); bzero(&m_lockReq, sizeof(m_lockReq)); - m_opDepth = 0; m_callback.m_callbackFunction = 0; m_callback.m_callbackData = 0; m_magic = 0; @@ -1900,49 +1701,59 @@ private: bool findSchemaTrans(SchemaTransPtr&, Uint32 trans_key); void releaseSchemaTrans(SchemaTransPtr&); - Uint32 getIteratorRepeat(SchemaTransPtr trans_ptr); - // coordinator - void sendTransReq(Signal*, SchemaTransPtr, SectionHandle&); - void sendTransParseReq(Signal*, SchemaOpPtr, SectionHandle&); - void recvTransReply(Signal*, bool isConf); - void handleTransReply(Signal*, SchemaTransPtr); void createSubOps(Signal*, SchemaOpPtr, bool first = false); void abortSubOps(Signal*, SchemaOpPtr, ErrorInfo); - void runTransMaster(Signal*, SchemaTransPtr); - void setTransMode(SchemaTransPtr, TransMode::Value, bool hold); + + void trans_recv_reply(Signal*, SchemaTransPtr); + void trans_start_recv_reply(Signal*, SchemaTransPtr); + void trans_parse_recv_reply(Signal*, SchemaTransPtr); void trans_prepare_start(Signal*, SchemaTransPtr); - void trans_prepare_next(Signal*, SchemaTransPtr); + void trans_prepare_recv_reply(Signal*, SchemaTransPtr); + void trans_prepare_next(Signal*, SchemaTransPtr, SchemaOpPtr); void trans_prepare_done(Signal*, SchemaTransPtr); void trans_abort_prepare_start(Signal*, SchemaTransPtr); - void trans_abort_prepare_next(Signal*, SchemaTransPtr); + void trans_abort_prepare_recv_reply(Signal*, SchemaTransPtr); + void trans_abort_prepare_next(Signal*, SchemaTransPtr, SchemaOpPtr); void trans_abort_prepare_done(Signal*, SchemaTransPtr); void trans_abort_parse_start(Signal*, SchemaTransPtr); - void trans_abort_parse_next(Signal*, SchemaTransPtr); + void trans_abort_parse_recv_reply(Signal*, SchemaTransPtr); + void trans_abort_parse_next(Signal*, SchemaTransPtr, SchemaOpPtr); void trans_abort_parse_done(Signal*, SchemaTransPtr); + void trans_rollback_sp_start(Signal* signal, SchemaTransPtr); + void trans_rollback_sp_recv_reply(Signal* signal, SchemaTransPtr); + void trans_rollback_sp_next(Signal* signal, SchemaTransPtr, SchemaOpPtr); + void trans_rollback_sp_done(Signal* signal, SchemaTransPtr, SchemaOpPtr); + void trans_commit_start(Signal*, SchemaTransPtr); void trans_commit_mutex_locked(Signal*, Uint32, Uint32); - void trans_commit_next(Signal*, SchemaTransPtr); + void trans_commit_recv_reply(Signal*, SchemaTransPtr); + void trans_commit_next(Signal*, SchemaTransPtr, SchemaOpPtr); void trans_commit_done(Signal* signal, SchemaTransPtr); void trans_commit_mutex_unlocked(Signal*, Uint32, Uint32); + void trans_complete_start(Signal* signal, SchemaTransPtr); + void trans_complete_recv_reply(Signal*, SchemaTransPtr); + void trans_complete_done(Signal*, SchemaTransPtr); + // participant void recvTransReq(Signal*); void recvTransParseReq(Signal*, SchemaTransPtr, Uint32 op_key, const OpInfo& info, Uint32 requestInfo); void runTransSlave(Signal*, SchemaTransPtr); - void sendTransConf(Signal*, SchemaOpPtr, Uint32 itFlags = 0); - void sendTransConf(Signal*, SchemaTransPtr, Uint32 itFlags = 0); + void sendTransConf(Signal*, SchemaOpPtr); + void sendTransConf(Signal*, SchemaTransPtr); void sendTransRef(Signal*, SchemaOpPtr); void sendTransRef(Signal*, SchemaTransPtr); void slave_run_start(Signal*, const SchemaTransImplReq*); void slave_run_parse(Signal*, SchemaTransPtr, const SchemaTransImplReq*); + void slave_run_complete(Signal*, SchemaTransPtr); void slave_flush_schema(Signal*, SchemaTransPtr); // reply to trans client for begin/end trans @@ -1981,8 +1792,7 @@ private: return; } - TransLoc& tLoc = trans_ptr.p->m_transLoc; - D("before add op" << *trans_ptr.p << tLoc << tLoc.m_opList); + D("before add op" << *trans_ptr.p); if (trans_ptr.p->m_transId != req->transId) { jam(); @@ -1990,45 +1800,12 @@ private: return; } - if (trans_ptr.p->m_opDepth == 0) { - jam(); - if (trans_ptr.p->m_clientState != TransClient::BeginReply && - trans_ptr.p->m_clientState != TransClient::ParseReply) { - jam(); - setError(error, SchemaTransImplRef::InvalidTransState, __LINE__); - return; - } - } - - if (tLoc.m_phase != TransPhase::Begin && - tLoc.m_phase != TransPhase::Parse) { - jam(); - setError(error, SchemaTransImplRef::InvalidTransState, __LINE__); - return; - } - - ndbrequire(!hasError(trans_ptr.p->m_error)); - - if (tLoc.m_mode == TransMode::Rollback) { - jam(); - D("continuing after rollback"); - resetError(trans_ptr); - setTransMode(trans_ptr, TransMode::Normal, false); - } - ndbrequire(tLoc.m_mode == TransMode::Normal); - if (!seizeSchemaOp(op_ptr, t_ptr)) { jam(); setError(error, SchemaTransImplRef::TooManySchemaOps, __LINE__); return; } - // switch to parse phase on first op - if (tLoc.m_phase == TransPhase::Begin) { - jam(); - iteratorMove(tLoc, false); - } - trans_ptr.p->m_clientState = TransClient::ParseReq; DictSignal::setRequestExtra(op_ptr.p->m_requestInfo, requestExtra); @@ -2047,16 +1824,6 @@ private: // client of this REQ (trans client or us, recursively) op_ptr.p->m_clientRef = req->clientRef; op_ptr.p->m_clientData = req->clientData; - - const Uint32 ownNodeId = getOwnNodeId(); - const TransGlob& tGlob = getTransGlob(trans_ptr, ownNodeId); - ndbrequire(tGlob.state() == TransGlob::Ok); - - // other nodes have no op record yet - NdbNodeBitmask nodes = trans_ptr.p->m_initNodes; - nodes.bitANDC(trans_ptr.p->m_failedNodes); - nodes.clear(ownNodeId); - setGlobState(trans_ptr, nodes, TransGlob::NeedOp); } /* @@ -2211,6 +1978,9 @@ private: // who is using local create tab Callback m_callback; + // flag if this op has been aborted in RT_PREPARE phase + bool m_abortPrepareDone; + CreateTableRec() : OpRec(g_opInfo, (Uint32*)&m_request) { memset(&m_request, 0, sizeof(m_request)); @@ -2218,6 +1988,7 @@ private: m_fragmentsPtrI = RNIL; m_dihAddFragPtr = RNIL; m_lqhFragPtr = RNIL; + m_abortPrepareDone = false; } #ifdef VM_TRACE @@ -2342,7 +2113,6 @@ private: TableRecordPtr m_newTablePtr; // before image - SchemaFile::TableEntry m_oldTableEntry; RopeHandle m_oldTableName; RopeHandle m_oldFrmData; @@ -3413,9 +3183,6 @@ public: friend NdbOut& operator<<(NdbOut& out, const DictObject&); friend NdbOut& operator<<(NdbOut& out, const ErrorInfo&); friend NdbOut& operator<<(NdbOut& out, const SchemaOp&); - friend NdbOut& operator<<(NdbOut& out, const OpList&); - friend NdbOut& operator<<(NdbOut& out, const TransLoc&); - friend NdbOut& operator<<(NdbOut& out, const TransGlob&); friend NdbOut& operator<<(NdbOut& out, const SchemaTrans&); friend NdbOut& operator<<(NdbOut& out, const TxHandle&); void check_consistency(); diff -Nrup a/storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp b/storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp --- a/storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp 2008-01-11 09:49:08 +01:00 +++ b/storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp 2008-02-06 21:15:12 +01:00 @@ -520,7 +520,6 @@ public: Uint32 requestFlags() const { Uint32 flags = 0; // not yet supported in DICT - flags |= m_transOn ? 0 : DictSignal::RF_SIMPLE_TRANS; return flags; } };
| Thread | ||
|---|---|---|
| • bk commit into 5.1 tree (jonas:1.2851) | jonas | 6 Feb |
© 1995-2008 MySQL AB, 2008-2009 Sun Microsystems, Inc.
Page generated in 1.387 sec. using MySQL 5.1.14-beta-log