MySQL数据库

开发平台：

Visual C++

DblqhMain.cpp：源码内容

This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#define DBLQH_C
#include "Dblqh.hpp"
#include <ndb_limits.h>
#include <md5_hash.hpp>
#include <ndb_version.h>
#include <signaldata/TuxBound.hpp>
#include <signaldata/AccScan.hpp>
#include <signaldata/CopyActive.hpp>
#include <signaldata/CopyFrag.hpp>
#include <signaldata/CreateTrig.hpp>
#include <signaldata/DropTrig.hpp>
#include <signaldata/EmptyLcp.hpp>
#include <signaldata/EventReport.hpp>
#include <signaldata/ExecFragReq.hpp>
#include <signaldata/GCPSave.hpp>
#include <signaldata/TcKeyRef.hpp>
#include <signaldata/LqhKey.hpp>
#include <signaldata/NextScan.hpp>
#include <signaldata/NFCompleteRep.hpp>
#include <signaldata/NodeFailRep.hpp>
#include <signaldata/ReadNodesConf.hpp>
#include <signaldata/RelTabMem.hpp>
#include <signaldata/ScanFrag.hpp>
#include <signaldata/SrFragidConf.hpp>
#include <signaldata/StartFragReq.hpp>
#include <signaldata/StartRec.hpp>
#include <signaldata/TupKey.hpp>
#include <signaldata/TupCommit.hpp>
#include <signaldata/LqhFrag.hpp>
#include <signaldata/AccFrag.hpp>
#include <signaldata/TupFrag.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <signaldata/PackedSignal.hpp>
#include <signaldata/PrepDropTab.hpp>
#include <signaldata/DropTab.hpp>
#include <signaldata/AlterTab.hpp>
#include <signaldata/LCP.hpp>
// Use DEBUG to print messages that should be
// seen only when we debug the product
#ifdef VM_TRACE
#define DEBUG(x) ndbout << "DBLQH: "<< x << endl;
NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::TransactionState state){
out << (int)state;
return out;
}
NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::LogWriteState state){
out << (int)state;
return out;
}
NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::ListState state){
out << (int)state;
return out;
}
NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::AbortState state){
out << (int)state;
return out;
}
NdbOut &
operator<<(NdbOut& out, Dblqh::ScanRecord::ScanState state){
out << (int)state;
return out;
}
NdbOut &
operator<<(NdbOut& out, Dblqh::LogFileOperationRecord::LfoState state){
out << (int)state;
return out;
}
NdbOut &
operator<<(NdbOut& out, Dblqh::ScanRecord::ScanType state){
out << (int)state;
return out;
}
#else
#define DEBUG(x)
#endif
//#define MARKER_TRACE 1
//#define TRACE_SCAN_TAKEOVER 1
const Uint32 NR_ScanNo = 0;
void Dblqh::execACC_COM_BLOCK(Signal* signal)
{
jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in ACC is in critical sector of being full.
/* ------------------------------------------------------------------------- */
cCounterAccCommitBlocked++;
caccCommitBlocked = true;
cCommitBlocked = true;
return;
}//Dblqh::execACC_COM_BLOCK()
void Dblqh::execACC_COM_UNBLOCK(Signal* signal)
{
jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in ACC ok again.
/* ------------------------------------------------------------------------- */
caccCommitBlocked = false;
if (ctupCommitBlocked == false) {
jam();
cCommitBlocked = false;
}//if
return;
}//Dblqh::execACC_COM_UNBLOCK()
void Dblqh::execTUP_COM_BLOCK(Signal* signal)
{
jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in TUP is in critical sector of being full.
/* ------------------------------------------------------------------------- */
cCounterTupCommitBlocked++;
ctupCommitBlocked = true;
cCommitBlocked = true;
return;
}//Dblqh::execTUP_COM_BLOCK()
void Dblqh::execTUP_COM_UNBLOCK(Signal* signal)
{
jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in TUP ok again.
/* ------------------------------------------------------------------------- */
ctupCommitBlocked = false;
if (caccCommitBlocked == false) {
jam();
cCommitBlocked = false;
}//if
return;
}//Dblqh::execTUP_COM_UNBLOCK()
/* ------------------------------------------------------------------------- */
/* ------- SEND SYSTEM ERROR ------- */
/* */
/* ------------------------------------------------------------------------- */
void Dblqh::systemError(Signal* signal)
{
progError(0, 0);
}//Dblqh::systemError()
/* *************** */
/* ACCSEIZEREF > */
/* *************** */
void Dblqh::execACCSEIZEREF(Signal* signal)
{
jamEntry();
ndbrequire(false);
}//Dblqh::execACCSEIZEREF()
/* ******************************************************>> */
/* THIS SIGNAL IS USED TO HANDLE REAL-TIME */
/* BREAKS THAT ARE NECESSARY TO ENSURE REAL-TIME */
/* OPERATION OF LQH. */
/* This signal is also used for signal loops, for example */
/* the timeout handling for writing logs every second. */
/* ******************************************************>> */
void Dblqh::execCONTINUEB(Signal* signal)
{
jamEntry();
Uint32 tcase = signal->theData[0];
Uint32 data0 = signal->theData[1];
Uint32 data1 = signal->theData[2];
Uint32 data2 = signal->theData[3];
#if 0
if (tcase == RNIL) {
tcConnectptr.i = data0;
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
ndbout << "State = " << tcConnectptr.p->transactionState;
ndbout << " seqNoReplica = " << tcConnectptr.p->seqNoReplica;
ndbout << " tcNodeFailrec = " << tcConnectptr.p->tcNodeFailrec;
ndbout << " activeCreat = " << tcConnectptr.p->activeCreat;
ndbout << endl;
ndbout << "tupkeyData0 = " << tcConnectptr.p->tupkeyData[0];
ndbout << "tupkeyData1 = " << tcConnectptr.p->tupkeyData[1];
ndbout << "tupkeyData2 = " << tcConnectptr.p->tupkeyData[2];
ndbout << "tupkeyData3 = " << tcConnectptr.p->tupkeyData[3];
ndbout << endl;
ndbout << "abortState = " << tcConnectptr.p->abortState;
ndbout << "listState = " << tcConnectptr.p->listState;
ndbout << endl;
return;
}//if
#endif
switch (tcase) {
case ZLOG_LQHKEYREQ:
if (cnoOfLogPages == 0) {
jam();
sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 10, 2);
return;
}//if
logPartPtr.i = data0;
ptrCheckGuard(logPartPtr, clogPartFileSize, logPartRecord);
logFilePtr.i = logPartPtr.p->currentLogfile;
ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
logPagePtr.i = logFilePtr.p->currentLogpage;
ptrCheckGuard(logPagePtr, clogPageFileSize, logPageRecord);
tcConnectptr.i = logPartPtr.p->firstLogQueue;
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
fragptr.i = tcConnectptr.p->fragmentptr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
if ((cCommitBlocked == true) &&
(fragptr.p->fragActiveStatus == ZTRUE)) {
jam();
sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 10, 2);
return;
}//if
logPartPtr.p->LogLqhKeyReqSent = ZFALSE;
getFirstInLogQueue(signal);
switch (tcConnectptr.p->transactionState) {
case TcConnectionrec::LOG_QUEUED:
if (tcConnectptr.p->abortState != TcConnectionrec::ABORT_IDLE) {
jam();
logNextStart(signal);
abortCommonLab(signal);
return;
} else {
jam();
/*------------------------------------------------------------*/
/* WE MUST SET THE STATE OF THE LOG PART TO IDLE TO */
/* ENSURE THAT WE ARE NOT QUEUED AGAIN ON THE LOG PART */
/* WE WILL SET THE LOG PART STATE TO ACTIVE IMMEDIATELY */
/* SO NO OTHER PROCESS WILL SEE THIS STATE. IT IS MERELY*/
/* USED TO ENABLE REUSE OF CODE. */
/*------------------------------------------------------------*/
if (logPartPtr.p->logPartState == LogPartRecord::ACTIVE) {
jam();
logPartPtr.p->logPartState = LogPartRecord::IDLE;
}//if
logLqhkeyreqLab(signal);
return;
}//if
break;
case TcConnectionrec::LOG_ABORT_QUEUED:
jam();
writeAbortLog(signal);
removeLogTcrec(signal);
logNextStart(signal);
continueAfterLogAbortWriteLab(signal);
return;
break;
case TcConnectionrec::LOG_COMMIT_QUEUED:
case TcConnectionrec::LOG_COMMIT_QUEUED_WAIT_SIGNAL:
jam();
writeCommitLog(signal, logPartPtr);
logNextStart(signal);
if (tcConnectptr.p->transactionState == TcConnectionrec::LOG_COMMIT_QUEUED) {
if (tcConnectptr.p->seqNoReplica != 0) {
jam();
commitReplyLab(signal);
} else {
jam();
localCommitLab(signal);
}//if
return;
} else {
jam();
tcConnectptr.p->transactionState = TcConnectionrec::LOG_COMMIT_WRITTEN_WAIT_SIGNAL;
return;
}//if
break;
case TcConnectionrec::COMMIT_QUEUED:
jam();
logNextStart(signal);
localCommitLab(signal);
break;
case TcConnectionrec::ABORT_QUEUED:
jam();
logNextStart(signal);
abortCommonLab(signal);
break;
default:
ndbrequire(false);
break;
}//switch
return;
break;
case ZSR_GCI_LIMITS:
jam();
signal->theData[0] = data0;
srGciLimits(signal);
return;
break;
case ZSR_LOG_LIMITS:
jam();
signal->theData[0] = data0;
signal->theData[1] = data1;
signal->theData[2] = data2;
srLogLimits(signal);
return;
break;
case ZSEND_EXEC_CONF:
jam();
signal->theData[0] = data0;
sendExecConf(signal);
return;
break;
case ZEXEC_SR:
jam();
signal->theData[0] = data0;
execSr(signal);
return;
break;
case ZSR_FOURTH_COMP:
jam();
signal->theData[0] = data0;
srFourthComp(signal);
return;
break;
case ZINIT_FOURTH:
jam();
signal->theData[0] = data0;
initFourth(signal);
return;
break;
case ZTIME_SUPERVISION:
jam();
signal->theData[0] = data0;
timeSup(signal);
return;
break;
case ZSR_PHASE3_START:
jam();
signal->theData[0] = data0;
srPhase3Start(signal);
return;
break;
case ZLQH_TRANS_NEXT:
jam();
tcNodeFailptr.i = data0;
ptrCheckGuard(tcNodeFailptr, ctcNodeFailrecFileSize, tcNodeFailRecord);
lqhTransNextLab(signal);
return;
break;
case ZSCAN_TC_CONNECT:
jam();
tabptr.i = data1;
ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
scanTcConnectLab(signal, data0, data2);
return;
break;
case ZINITIALISE_RECORDS:
jam();
initialiseRecordsLab(signal, data0, data2, signal->theData[4]);
return;
break;
case ZINIT_GCP_REC:
jam();
gcpPtr.i = 0;
ptrAss(gcpPtr, gcpRecord);
initGcpRecLab(signal);
return;
break;
case ZRESTART_OPERATIONS_AFTER_STOP:
jam();
tcConnectptr.i = data0;
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
if (tcConnectptr.p->listState != TcConnectionrec::WAIT_QUEUE_LIST) {
jam();
return;
}//if
releaseWaitQueue(signal);
linkActiveFrag(signal);
restartOperationsAfterStopLab(signal);
return;
break;
case ZCHECK_LCP_STOP_BLOCKED:
jam();
c_scanRecordPool.getPtr(scanptr, data0);
tcConnectptr.i = scanptr.p->scanTcrec;
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
fragptr.i = tcConnectptr.p->fragmentptr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
checkLcpStopBlockedLab(signal);
return;
case ZSCAN_MARKERS:
jam();
scanMarkers(signal, data0, data1, data2);
return;
break;
case ZOPERATION_EVENT_REP:
jam();
/* --------------------------------------------------------------------- */
// Report information about transaction activity once per second.
/* --------------------------------------------------------------------- */
if (signal->theData[1] == 0) {
signal->theData[0] = EventReport::OperationReportCounters;
signal->theData[1] = c_Counters.operations;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
}//if
c_Counters.clear();
signal->theData[0] = ZOPERATION_EVENT_REP;
signal->theData[1] = 0;
sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 5000, 2);
break;
case ZPREP_DROP_TABLE:
jam();
checkDropTab(signal);
return;
break;
default:
ndbrequire(false);
break;
}//switch
}//Dblqh::execCONTINUEB()
/* *********************************************************> */
/* Request from DBDIH to include a new node in the node list */
/* and so forth. */
/* *********************************************************> */
void Dblqh::execINCL_NODEREQ(Signal* signal)
{
jamEntry();
BlockReference retRef = signal->theData[0];
Uint32 nodeId = signal->theData[1];
cnewestGci = signal->theData[2];
cnewestCompletedGci = signal->theData[2] - 1;
ndbrequire(cnoOfNodes < MAX_NDB_NODES);
for (Uint32 i = 0; i < cnoOfNodes; i++) {
jam();
if (cnodeData[i] == nodeId) {
jam();
cnodeStatus[i] = ZNODE_UP;
}//if
}//for
signal->theData[0] = cownref;
sendSignal(retRef, GSN_INCL_NODECONF, signal, 1, JBB);
return;
}//Dblqh::execINCL_NODEREQ()
void Dblqh::execTUPSEIZEREF(Signal* signal)
{
jamEntry();
ndbrequire(false);
}//Dblqh::execTUPSEIZEREF()
/* ########################################################################## */
/* ####### START / RESTART MODULE ####### */
/* ########################################################################## */
/* ************************************************************************>> */
/* This is first signal that arrives in a start / restart. Sender is NDBCNTR_REF. */
/* ************************************************************************>> */
void Dblqh::execSTTOR(Signal* signal)
{
UintR tstartPhase;
jamEntry();
/* START CASE */
tstartPhase = signal->theData[1];
/* SYSTEM RESTART RANK */
csignalKey = signal->theData[6];
switch (tstartPhase) {
case ZSTART_PHASE1:
jam();
cstartPhase = tstartPhase;
sttorStartphase1Lab(signal);
return;
break;
default:
jam();
/*empty*/;
sendsttorryLab(signal);
return;
break;
}//switch
}//Dblqh::execSTTOR()
/* ***************************************> */
/* Restart phases 1 - 6, sender is Ndbcntr */
/* ***************************************> */
void Dblqh::execNDB_STTOR(Signal* signal)
{
jamEntry();
Uint32 ownNodeId = signal->theData[1]; /* START PHASE*/
cstartPhase = signal->theData[2]; /* MY NODE ID */
cstartType = signal->theData[3]; /* START TYPE */
switch (cstartPhase) {
case ZSTART_PHASE1:
jam();
preComputedRequestInfoMask = 0;
LqhKeyReq::setKeyLen(preComputedRequestInfoMask, RI_KEYLEN_MASK);
LqhKeyReq::setLastReplicaNo(preComputedRequestInfoMask, RI_LAST_REPL_MASK);
LqhKeyReq::setLockType(preComputedRequestInfoMask, RI_LOCK_TYPE_MASK);
// Dont LqhKeyReq::setApplicationAddressFlag
LqhKeyReq::setDirtyFlag(preComputedRequestInfoMask, 1);
// Dont LqhKeyReq::setInterpretedFlag
LqhKeyReq::setSimpleFlag(preComputedRequestInfoMask, 1);
LqhKeyReq::setOperation(preComputedRequestInfoMask, RI_OPERATION_MASK);
// Dont setAIInLqhKeyReq
// Dont setSeqNoReplica
// Dont setSameClientAndTcFlag
// Dont setReturnedReadLenAIFlag
// Dont setAPIVersion
LqhKeyReq::setMarkerFlag(preComputedRequestInfoMask, 1);
//preComputedRequestInfoMask = 0x003d7fff;
startphase1Lab(signal, /* dummy */ ~0, ownNodeId);
signal->theData[0] = ZOPERATION_EVENT_REP;
signal->theData[1] = 1;
sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 10, 2);
return;
break;
case ZSTART_PHASE2:
jam();
startphase2Lab(signal, /* dummy */ ~0);
return;
break;
case ZSTART_PHASE3:
jam();
startphase3Lab(signal);
return;
break;
case ZSTART_PHASE4:
jam();
startphase4Lab(signal);
return;
break;
case ZSTART_PHASE6:
jam();
startphase6Lab(signal);
return;
break;
default:
jam();
/*empty*/;
sendNdbSttorryLab(signal);
return;
break;
}//switch
}//Dblqh::execNDB_STTOR()
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++ START PHASE 1 +++++++ */
/* LOAD OUR BLOCK REFERENCE AND OUR PROCESSOR ID */
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::sttorStartphase1Lab(Signal* signal)
{
sendsttorryLab(signal);
return;
}//Dblqh::sttorStartphase1Lab()
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++ START PHASE 2 +++++++ */
/* */
/* INITIATE ALL RECORDS WITHIN THE BLOCK */
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::startphase1Lab(Signal* signal, Uint32 _dummy, Uint32 ownNodeId)
{
UintR Ti;
HostRecordPtr ThostPtr;
/* ------- INITIATE ALL RECORDS ------- */
cownNodeid = ownNodeId;
caccBlockref = calcAccBlockRef (cownNodeid);
ctupBlockref = calcTupBlockRef (cownNodeid);
ctuxBlockref = calcTuxBlockRef (cownNodeid);
cownref = calcLqhBlockRef (cownNodeid);
for (Ti = 0; Ti < chostFileSize; Ti++) {
ThostPtr.i = Ti;
ptrCheckGuard(ThostPtr, chostFileSize, hostRecord);
ThostPtr.p->hostLqhBlockRef = calcLqhBlockRef(ThostPtr.i);
ThostPtr.p->hostTcBlockRef = calcTcBlockRef(ThostPtr.i);
ThostPtr.p->inPackedList = false;
ThostPtr.p->noOfPackedWordsLqh = 0;
ThostPtr.p->noOfPackedWordsTc = 0;
}//for
cpackedListIndex = 0;
sendNdbSttorryLab(signal);
return;
}//Dblqh::startphase1Lab()
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++ START PHASE 2 +++++++ */
/* */
/* CONNECT LQH WITH ACC AND TUP. */
/* EVERY CONNECTION RECORD IN LQH IS ASSIGNED TO ONE ACC CONNECTION RECORD */
/* AND ONE TUP CONNECTION RECORD. */
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::startphase2Lab(Signal* signal, Uint32 _dummy)
{
cmaxWordsAtNodeRec = MAX_NO_WORDS_OUTSTANDING_COPY_FRAGMENT;
/* -- ACC AND TUP CONNECTION PROCESS -- */
tcConnectptr.i = 0;
ptrAss(tcConnectptr, tcConnectionrec);
moreconnectionsLab(signal);
return;
}//Dblqh::startphase2Lab()
void Dblqh::moreconnectionsLab(Signal* signal)
{
tcConnectptr.p->tcAccBlockref = caccBlockref;
// set TUX block here (no operation is seized in TUX)
tcConnectptr.p->tcTuxBlockref = ctuxBlockref;
/* NO STATE CHECKING IS PERFORMED, ASSUMED TO WORK */
/* *************** */
/* ACCSEIZEREQ < */
/* *************** */
signal->theData[0] = tcConnectptr.i;
signal->theData[1] = cownref;
sendSignal(caccBlockref, GSN_ACCSEIZEREQ, signal, 2, JBB);
return;
}//Dblqh::moreconnectionsLab()
/* ***************> */
/* ACCSEIZECONF > */
/* ***************> */
void Dblqh::execACCSEIZECONF(Signal* signal)
{
jamEntry();
tcConnectptr.i = signal->theData[0];
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
tcConnectptr.p->accConnectrec = signal->theData[1];
/* *************** */
/* TUPSEIZEREQ < */
/* *************** */
tcConnectptr.p->tcTupBlockref = ctupBlockref;
signal->theData[0] = tcConnectptr.i;
signal->theData[1] = cownref;
sendSignal(ctupBlockref, GSN_TUPSEIZEREQ, signal, 2, JBB);
return;
}//Dblqh::execACCSEIZECONF()
/* ***************> */
/* TUPSEIZECONF > */
/* ***************> */
void Dblqh::execTUPSEIZECONF(Signal* signal)
{
jamEntry();
tcConnectptr.i = signal->theData[0];
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
tcConnectptr.p->tupConnectrec = signal->theData[1];
/* ------- CHECK IF THERE ARE MORE CONNECTIONS TO BE CONNECTED ------- */
tcConnectptr.i = tcConnectptr.p->nextTcConnectrec;
if (tcConnectptr.i != RNIL) {
jam();
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
moreconnectionsLab(signal);
return;
}//if
/* ALL LQH_CONNECT RECORDS ARE CONNECTED TO ACC AND TUP ---- */
sendNdbSttorryLab(signal);
return;
}//Dblqh::execTUPSEIZECONF()
/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++ START PHASE 4 +++++++ */
/* */
/* CONNECT LQH WITH LQH. */
/* CONNECT EACH LQH WITH EVERY LQH IN THE DATABASE SYSTEM. */
/* IF INITIAL START THEN CREATE THE FRAGMENT LOG FILES */
/*IF SYSTEM RESTART OR NODE RESTART THEN OPEN THE FRAGMENT LOG FILES AND */
/*FIND THE END OF THE LOG FILES. */
/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* WAIT UNTIL ADD NODE PROCESSES ARE COMPLETED */
/* IF INITIAL START ALSO WAIT FOR LOG FILES TO INITIALISED */
/*START TIME SUPERVISION OF LOG FILES. WE HAVE TO WRITE LOG PAGES TO DISK */
/*EVEN IF THE PAGES ARE NOT FULL TO ENSURE THAT THEY COME TO DISK ASAP. */
/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::startphase3Lab(Signal* signal)
{
LogFileRecordPtr prevLogFilePtr;
LogFileRecordPtr zeroLogFilePtr;
caddNodeState = ZTRUE;
/* ***************<< */
/* READ_NODESREQ < */
/* ***************<< */
cinitialStartOngoing = ZTRUE;
ndbrequire(cnoLogFiles != 0);
for (logPartPtr.i = 0; logPartPtr.i < 4; logPartPtr.i++) {
jam();
ptrAss(logPartPtr, logPartRecord);
initLogpart(signal);
for (Uint32 fileNo = 0; fileNo < cnoLogFiles; fileNo++) {
seizeLogfile(signal);
if (fileNo != 0) {
jam();
prevLogFilePtr.p->nextLogFile = logFilePtr.i;
logFilePtr.p->prevLogFile = prevLogFilePtr.i;
} else {
jam();
logPartPtr.p->firstLogfile = logFilePtr.i;
logPartPtr.p->currentLogfile = logFilePtr.i;
zeroLogFilePtr.i = logFilePtr.i;
zeroLogFilePtr.p = logFilePtr.p;
}//if
prevLogFilePtr.i = logFilePtr.i;
prevLogFilePtr.p = logFilePtr.p;
initLogfile(signal, fileNo);
if ((cstartType == NodeState::ST_INITIAL_START) ||
(cstartType == NodeState::ST_INITIAL_NODE_RESTART)) {
if (logFilePtr.i == zeroLogFilePtr.i) {
jam();
/* ------------------------------------------------------------------------- */
/*IN AN INITIAL START WE START BY CREATING ALL LOG FILES AND SETTING THEIR */
/*PROPER SIZE AND INITIALISING PAGE ZERO IN ALL FILES. */
/*WE START BY CREATING FILE ZERO IN EACH LOG PART AND THEN PROCEED */
/*SEQUENTIALLY THROUGH ALL LOG FILES IN THE LOG PART. */
/* ------------------------------------------------------------------------- */
openLogfileInit(signal);
}//if
}//if
}//for
zeroLogFilePtr.p->prevLogFile = logFilePtr.i;
logFilePtr.p->nextLogFile = zeroLogFilePtr.i;
}//for
if (cstartType != NodeState::ST_INITIAL_START &&
cstartType != NodeState::ST_INITIAL_NODE_RESTART) {
jam();
ndbrequire(cstartType == NodeState::ST_NODE_RESTART ||
cstartType == NodeState::ST_SYSTEM_RESTART);
/** --------------------------------------------------------------------
* THIS CODE KICKS OFF THE SYSTEM RESTART AND NODE RESTART. IT STARTS UP
* THE RESTART BY FINDING THE END OF THE LOG AND FROM THERE FINDING THE
* INFO ABOUT THE GLOBAL CHECKPOINTS IN THE FRAGMENT LOG.
--------------------------------------------------------------------- */
for (logPartPtr.i = 0; logPartPtr.i < 4; logPartPtr.i++) {
jam();
LogFileRecordPtr locLogFilePtr;
ptrAss(logPartPtr, logPartRecord);
locLogFilePtr.i = logPartPtr.p->firstLogfile;
ptrCheckGuard(locLogFilePtr, clogFileFileSize, logFileRecord);
locLogFilePtr.p->logFileStatus = LogFileRecord::OPEN_SR_FRONTPAGE;
openFileRw(signal, locLogFilePtr);
}//for
}//if
signal->theData[0] = cownref;
sendSignal(NDBCNTR_REF, GSN_READ_NODESREQ, signal, 1, JBB);
return;
}//Dblqh::startphase3Lab()
/* ****************** */
/* READ_NODESCONF > */
/* ****************** */
void Dblqh::execREAD_NODESCONF(Signal* signal)
{
jamEntry();
ReadNodesConf * const readNodes = (ReadNodesConf *)&signal->theData[0];
cnoOfNodes = readNodes->noOfNodes;
unsigned ind = 0;
unsigned i = 0;
for (i = 1; i < MAX_NDB_NODES; i++) {
jam();
if (NodeBitmask::get(readNodes->allNodes, i)) {
jam();
cnodeData[ind] = i;
cnodeStatus[ind] = NodeBitmask::get(readNodes->inactiveNodes, i);
//readNodes->getVersionId(i, readNodes->theVersionIds) not used
ind++;
}//if
}//for
ndbrequire(ind == cnoOfNodes);
ndbrequire(cnoOfNodes >= 1 && cnoOfNodes < MAX_NDB_NODES);
ndbrequire(!(cnoOfNodes == 1 && cstartType == NodeState::ST_NODE_RESTART));
caddNodeState = ZFALSE;
if (cstartType == NodeState::ST_SYSTEM_RESTART) {
jam();
sendNdbSttorryLab(signal);
return;
}//if
checkStartCompletedLab(signal);
return;
}//Dblqh::execREAD_NODESCONF()
void Dblqh::checkStartCompletedLab(Signal* signal)
{
if (caddNodeState == ZFALSE) {
if (cinitialStartOngoing == ZFALSE) {
jam();
sendNdbSttorryLab(signal);
return;
}//if
}//if
return;
}//Dblqh::checkStartCompletedLab()
void Dblqh::startphase4Lab(Signal* signal)
{
sendNdbSttorryLab(signal);
return;
}//Dblqh::startphase4Lab()
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* SET CONCURRENCY OF LOCAL CHECKPOINTS TO BE USED AFTER SYSTEM RESTART. */
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::startphase6Lab(Signal* signal)
{
cstartPhase = ZNIL;
cstartType = ZNIL;
sendNdbSttorryLab(signal);
return;
}//Dblqh::startphase6Lab()
void Dblqh::sendNdbSttorryLab(Signal* signal)
{
signal->theData[0] = cownref;
sendSignal(NDBCNTR_REF, GSN_NDB_STTORRY, signal, 1, JBB);
return;
}//Dblqh::sendNdbSttorryLab()
void Dblqh::sendsttorryLab(Signal* signal)
{
/* *********<< */
/* STTORRY < */
/* *********<< */
signal->theData[0] = csignalKey; /* SIGNAL KEY */
signal->theData[1] = 3; /* BLOCK CATEGORY */
signal->theData[2] = 2; /* SIGNAL VERSION NUMBER */
signal->theData[3] = ZSTART_PHASE1;
signal->theData[4] = 255;
sendSignal(NDBCNTR_REF, GSN_STTORRY, signal, 5, JBB);
return;
}//Dblqh::sendsttorryLab()
/* ***************>> */
/* READ_NODESREF > */
/* ***************>> */
void Dblqh::execREAD_NODESREF(Signal* signal)
{
jamEntry();
ndbrequire(false);
}//Dblqh::execREAD_NODESREF()
/* *************** */
/* SIZEALT_REP > */
/* *************** */
void Dblqh::execREAD_CONFIG_REQ(Signal* signal)
{
const ReadConfigReq * req = (ReadConfigReq*)signal->getDataPtr();
Uint32 ref = req->senderRef;
Uint32 senderData = req->senderData;
ndbrequire(req->noOfParameters == 0);
jamEntry();
const ndb_mgm_configuration_iterator * p =
theConfiguration.getOwnConfigIterator();
ndbrequire(p != 0);
cnoLogFiles = 8;
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_DB_NO_REDOLOG_FILES,
&cnoLogFiles));
ndbrequire(cnoLogFiles > 0);
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_LQH_FRAG, &cfragrecFileSize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_LQH_TABLE, &ctabrecFileSize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_LQH_TC_CONNECT,
&ctcConnectrecFileSize));
clogFileFileSize = 4 * cnoLogFiles;
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_LQH_SCAN, &cscanrecFileSize));
cmaxAccOps = cscanrecFileSize * MAX_PARALLEL_OP_PER_SCAN;
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_DB_DISCLESS, &c_diskless));
initRecords();
initialiseRecordsLab(signal, 0, ref, senderData);
return;
}//Dblqh::execSIZEALT_REP()
/* ########################################################################## */
/* ####### ADD/DELETE FRAGMENT MODULE ####### */
/* THIS MODULE IS USED BY DICTIONARY TO CREATE NEW FRAGMENTS AND DELETE */
/* OLD FRAGMENTS. */
/* */
/* ########################################################################## */
/* -------------------------------------------------------------- */
/* FRAG REQ */
/* -------------------------------------------------------------- */
/* *********************************************************> */
/* LQHFRAGREQ: Create new fragments for a table. Sender DICT */
/* *********************************************************> */
// this unbelievable mess could be replaced by one signal to LQH
// and execute direct to local DICT to get everything at once
void Dblqh::execLQHFRAGREQ(Signal* signal)
{
jamEntry();
LqhFragReq * req = (LqhFragReq*)signal->getDataPtr();
Uint32 retPtr = req->senderData;
BlockReference retRef = req->senderRef;
Uint32 fragId = req->fragmentId;
Uint32 reqinfo = req->requestInfo;
tabptr.i = req->tableId;
Uint16 tlocalKeylen = req->localKeyLength;
Uint32 tmaxLoadFactor = req->maxLoadFactor;
Uint32 tminLoadFactor = req->minLoadFactor;
Uint8 tk = req->kValue;
Uint8 tlhstar = req->lh3DistrBits;
Uint8 tlh = req->lh3PageBits;
Uint32 tnoOfAttr = req->noOfAttributes;
Uint32 tnoOfNull = req->noOfNullAttributes;
Uint32 noOfAlloc = req->noOfPagesToPreAllocate;
Uint32 tschemaVersion = req->schemaVersion;
Uint32 ttupKeyLength = req->keyLength;
Uint32 nextLcp = req->nextLCP;
Uint32 noOfKeyAttr = req->noOfKeyAttr;
Uint32 noOfNewAttr = req->noOfNewAttr;
Uint32 checksumIndicator = req->checksumIndicator;
Uint32 noOfAttributeGroups = req->noOfAttributeGroups;
Uint32 gcpIndicator = req->GCPIndicator;
Uint32 startGci = req->startGci;
Uint32 tableType = req->tableType;
Uint32 primaryTableId = req->primaryTableId;
ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
bool tempTable = ((reqinfo & LqhFragReq::TemporaryTable) != 0);
/* Temporary tables set to defined in system restart */
if (tabptr.p->tableStatus == Tablerec::NOT_DEFINED){
tabptr.p->tableStatus = Tablerec::ADD_TABLE_ONGOING;
tabptr.p->tableType = tableType;
tabptr.p->primaryTableId = primaryTableId;
tabptr.p->schemaVersion = tschemaVersion;
}//if
if (tabptr.p->tableStatus != Tablerec::ADD_TABLE_ONGOING){
jam();
fragrefLab(signal, retRef, retPtr, ZTAB_STATE_ERROR);
return;
}//if
//--------------------------------------------------------------------
// We could arrive here if we create the fragment as part of a take
// over by a hot spare node. The table is then is already created
// and bit 31 is set, thus indicating that we are creating a fragment
// by copy creation. Also since the node has already been started we
// know that it is not a node restart ongoing.
//--------------------------------------------------------------------
if (getFragmentrec(signal, fragId)) {
jam();
fragrefLab(signal, retRef, retPtr, terrorCode);
return;
}//if
if (!insertFragrec(signal, fragId)) {
jam();
fragrefLab(signal, retRef, retPtr, terrorCode);
return;
}//if
Uint32 copyType = reqinfo & 3;
initFragrec(signal, tabptr.i, fragId, copyType);
fragptr.p->startGci = startGci;
fragptr.p->newestGci = startGci;
fragptr.p->tableType = tableType;
if (DictTabInfo::isOrderedIndex(tableType)) {
jam();
// find corresponding primary table fragment
TablerecPtr tTablePtr;
tTablePtr.i = primaryTableId;
ptrCheckGuard(tTablePtr, ctabrecFileSize, tablerec);
FragrecordPtr tFragPtr;
tFragPtr.i = RNIL;
for (Uint32 i = 0; i < MAX_FRAG_PER_NODE; i++) {
if (tTablePtr.p->fragid[i] == fragptr.p->fragId) {
jam();
tFragPtr.i = tTablePtr.p->fragrec[i];
break;
}
}
ndbrequire(tFragPtr.i != RNIL);
// store it
fragptr.p->tableFragptr = tFragPtr.i;
} else {
fragptr.p->tableFragptr = fragptr.i;
}
if (tempTable) {
//--------------------------------------------
// reqinfo bit 3-4 = 2 means temporary table
// without logging or checkpointing.
//--------------------------------------------
jam();
fragptr.p->logFlag = Fragrecord::STATE_FALSE;
fragptr.p->lcpFlag = Fragrecord::LCP_STATE_FALSE;
}//if
fragptr.p->nextLcp = nextLcp;
//----------------------------------------------
// For node restarts it is not necessarily zero
//----------------------------------------------
if (cfirstfreeAddfragrec == RNIL) {
jam();
deleteFragrec(fragId);
fragrefLab(signal, retRef, retPtr, ZNO_ADD_FRAGREC);
return;
}//if
seizeAddfragrec(signal);
addfragptr.p->addFragid = fragId;
addfragptr.p->fragmentPtr = fragptr.i;
addfragptr.p->dictBlockref = retRef;
addfragptr.p->dictConnectptr = retPtr;
addfragptr.p->m_senderAttrPtr = RNIL;
addfragptr.p->noOfAttr = tnoOfAttr;
addfragptr.p->noOfNull = tnoOfNull;
addfragptr.p->noOfAllocPages = noOfAlloc;
addfragptr.p->tabId = tabptr.i;
addfragptr.p->totalAttrReceived = 0;
addfragptr.p->attrSentToTup = ZNIL;/* TO FIND PROGRAMMING ERRORS QUICKLY */
addfragptr.p->schemaVer = tschemaVersion;
Uint32 tmp = (reqinfo & LqhFragReq::CreateInRunning);
addfragptr.p->fragCopyCreation = (tmp == 0 ? 0 : 1);
addfragptr.p->addfragErrorCode = 0;
addfragptr.p->noOfKeyAttr = noOfKeyAttr;
addfragptr.p->noOfNewAttr = noOfNewAttr;
addfragptr.p->checksumIndicator = checksumIndicator;
addfragptr.p->noOfAttributeGroups = noOfAttributeGroups;
addfragptr.p->GCPIndicator = gcpIndicator;
addfragptr.p->lh3DistrBits = tlhstar;
addfragptr.p->tableType = tableType;
addfragptr.p->primaryTableId = primaryTableId;
//
addfragptr.p->tup1Connectptr = RNIL;
addfragptr.p->tup2Connectptr = RNIL;
addfragptr.p->tux1Connectptr = RNIL;
addfragptr.p->tux2Connectptr = RNIL;
if (DictTabInfo::isTable(tableType) ||
DictTabInfo::isHashIndex(tableType)) {
jam();
AccFragReq* const accreq = (AccFragReq*)signal->getDataPtrSend();
accreq->userPtr = addfragptr.i;
accreq->userRef = cownref;
accreq->tableId = tabptr.i;
accreq->reqInfo = copyType << 4;
accreq->fragId = fragId;
accreq->localKeyLen = tlocalKeylen;
accreq->maxLoadFactor = tmaxLoadFactor;
accreq->minLoadFactor = tminLoadFactor;
accreq->kValue = tk;
accreq->lhFragBits = tlhstar;
accreq->lhDirBits = tlh;
accreq->keyLength = ttupKeyLength;
/* ----------------------------------------------------------------------- */
/* Send ACCFRAGREQ, when confirmation is received send 2 * TUPFRAGREQ to */
/* create 2 tuple fragments on this node. */
/* ----------------------------------------------------------------------- */
addfragptr.p->addfragStatus = AddFragRecord::ACC_ADDFRAG;
sendSignal(fragptr.p->accBlockref, GSN_ACCFRAGREQ,
signal, AccFragReq::SignalLength, JBB);
return;
}
if (DictTabInfo::isOrderedIndex(tableType)) {
jam();
// NOTE: next 2 lines stolen from ACC
addfragptr.p->fragid1 = (0 << tlhstar) | fragId;
addfragptr.p->fragid2 = (1 << tlhstar) | fragId;
addfragptr.p->addfragStatus = AddFragRecord::WAIT_TWO_TUP;
sendAddFragReq(signal);
return;
}
ndbrequire(false);
}//Dblqh::execLQHFRAGREQ()
/* *************** */
/* ACCFRAGCONF > */
/* *************** */
void Dblqh::execACCFRAGCONF(Signal* signal)
{
jamEntry();
addfragptr.i = signal->theData[0];
Uint32 taccConnectptr = signal->theData[1];
Uint32 fragId1 = signal->theData[2];
Uint32 fragId2 = signal->theData[3];
Uint32 accFragPtr1 = signal->theData[4];
Uint32 accFragPtr2 = signal->theData[5];
Uint32 hashCheckBit = signal->theData[6];
ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
ndbrequire(addfragptr.p->addfragStatus == AddFragRecord::ACC_ADDFRAG);
addfragptr.p->accConnectptr = taccConnectptr;
addfragptr.p->fragid1 = fragId1;
addfragptr.p->fragid2 = fragId2;
fragptr.i = addfragptr.p->fragmentPtr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
fragptr.p->accFragptr[0] = accFragPtr1;
fragptr.p->accFragptr[1] = accFragPtr2;
fragptr.p->hashCheckBit = hashCheckBit;
addfragptr.p->addfragStatus = AddFragRecord::WAIT_TWO_TUP;
sendAddFragReq(signal);
}//Dblqh::execACCFRAGCONF()
/* *************** */
/* TUPFRAGCONF > */
/* *************** */
void Dblqh::execTUPFRAGCONF(Signal* signal)
{
jamEntry();
addfragptr.i = signal->theData[0];
Uint32 tupConnectptr = signal->theData[1];
Uint32 tupFragPtr = signal->theData[2]; /* TUP FRAGMENT POINTER */
Uint32 localFragId = signal->theData[3]; /* LOCAL FRAGMENT ID */
ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
fragptr.i = addfragptr.p->fragmentPtr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
if (localFragId == addfragptr.p->fragid1) {
jam();
fragptr.p->tupFragptr[0] = tupFragPtr;
} else if (localFragId == addfragptr.p->fragid2) {
jam();
fragptr.p->tupFragptr[1] = tupFragPtr;
} else {
ndbrequire(false);
return;
}//if
switch (addfragptr.p->addfragStatus) {
case AddFragRecord::WAIT_TWO_TUP:
jam();
fragptr.p->tupFragptr[0] = tupFragPtr;
addfragptr.p->tup1Connectptr = tupConnectptr;
addfragptr.p->addfragStatus = AddFragRecord::WAIT_ONE_TUP;
sendAddFragReq(signal);
break;
case AddFragRecord::WAIT_ONE_TUP:
jam();
fragptr.p->tupFragptr[1] = tupFragPtr;
addfragptr.p->tup2Connectptr = tupConnectptr;
if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
addfragptr.p->addfragStatus = AddFragRecord::WAIT_TWO_TUX;
sendAddFragReq(signal);
break;
}
goto done_with_frag;
break;
case AddFragRecord::WAIT_TWO_TUX:
jam();
fragptr.p->tuxFragptr[0] = tupFragPtr;
addfragptr.p->tux1Connectptr = tupConnectptr;
addfragptr.p->addfragStatus = AddFragRecord::WAIT_ONE_TUX;
sendAddFragReq(signal);
break;
case AddFragRecord::WAIT_ONE_TUX:
jam();
fragptr.p->tuxFragptr[1] = tupFragPtr;
addfragptr.p->tux2Connectptr = tupConnectptr;
goto done_with_frag;
break;
done_with_frag:
/* ---------------------------------------------------------------- */
/* Finished create of fragments. Now ready for creating attributes. */
/* ---------------------------------------------------------------- */
addfragptr.p->addfragStatus = AddFragRecord::WAIT_ADD_ATTR;
{
LqhFragConf* conf = (LqhFragConf*)signal->getDataPtrSend();
conf->senderData = addfragptr.p->dictConnectptr;
conf->lqhFragPtr = addfragptr.i;
sendSignal(addfragptr.p->dictBlockref, GSN_LQHFRAGCONF,
signal, LqhFragConf::SignalLength, JBB);
}
break;
default:
ndbrequire(false);
break;
}
}//Dblqh::execTUPFRAGCONF()
/* *************** */
/* TUXFRAGCONF > */
/* *************** */
void Dblqh::execTUXFRAGCONF(Signal* signal)
{
jamEntry();
execTUPFRAGCONF(signal);
}//Dblqh::execTUXFRAGCONF
/*
* Add fragment in TUP or TUX. Called up to 4 times.
*/
void
Dblqh::sendAddFragReq(Signal* signal)
{
fragptr.i = addfragptr.p->fragmentPtr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
if (addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP ||
addfragptr.p->addfragStatus == AddFragRecord::WAIT_ONE_TUP) {
if (DictTabInfo::isTable(addfragptr.p->tableType) ||
DictTabInfo::isHashIndex(addfragptr.p->tableType)) {
jam();
signal->theData[0] = addfragptr.i;
signal->theData[1] = cownref;
signal->theData[2] = 0; /* ADD TABLE */
signal->theData[3] = addfragptr.p->tabId;
signal->theData[4] = addfragptr.p->noOfAttr;
signal->theData[5] =
addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP
? addfragptr.p->fragid1 : addfragptr.p->fragid2;
signal->theData[6] = (addfragptr.p->noOfAllocPages >> 1) + 1;
signal->theData[7] = addfragptr.p->noOfNull;
signal->theData[8] = addfragptr.p->schemaVer;
signal->theData[9] = addfragptr.p->noOfKeyAttr;
signal->theData[10] = addfragptr.p->noOfNewAttr;
signal->theData[11] = addfragptr.p->checksumIndicator;
signal->theData[12] = addfragptr.p->noOfAttributeGroups;
signal->theData[13] = addfragptr.p->GCPIndicator;
sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ,
signal, TupFragReq::SignalLength, JBB);
return;
}
if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
jam();
signal->theData[0] = addfragptr.i;
signal->theData[1] = cownref;
signal->theData[2] = 0; /* ADD TABLE */
signal->theData[3] = addfragptr.p->tabId;
signal->theData[4] = 1; /* ordered index: one array attr */
signal->theData[5] =
addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP
? addfragptr.p->fragid1 : addfragptr.p->fragid2;
signal->theData[6] = (addfragptr.p->noOfAllocPages >> 1) + 1;
signal->theData[7] = 0; /* ordered index: no nullable */
signal->theData[8] = addfragptr.p->schemaVer;
signal->theData[9] = 1; /* ordered index: one key */
signal->theData[10] = addfragptr.p->noOfNewAttr;
signal->theData[11] = addfragptr.p->checksumIndicator;
signal->theData[12] = addfragptr.p->noOfAttributeGroups;
signal->theData[13] = addfragptr.p->GCPIndicator;
sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ,
signal, TupFragReq::SignalLength, JBB);
return;
}
}
if (addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUX ||
addfragptr.p->addfragStatus == AddFragRecord::WAIT_ONE_TUX) {
if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
jam();
TuxFragReq* const tuxreq = (TuxFragReq*)signal->getDataPtrSend();
tuxreq->userPtr = addfragptr.i;
tuxreq->userRef = cownref;
tuxreq->reqInfo = 0; /* ADD TABLE */
tuxreq->tableId = addfragptr.p->tabId;
ndbrequire(addfragptr.p->noOfAttr >= 2);
tuxreq->noOfAttr = addfragptr.p->noOfAttr - 1; /* skip NDB$TNODE */
tuxreq->fragId =
addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUX
? addfragptr.p->fragid1 : addfragptr.p->fragid2;
tuxreq->fragOff = addfragptr.p->lh3DistrBits;
tuxreq->tableType = addfragptr.p->tableType;
tuxreq->primaryTableId = addfragptr.p->primaryTableId;
// pointer to index fragment in TUP
tuxreq->tupIndexFragPtrI =
addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUX ?
fragptr.p->tupFragptr[0] : fragptr.p->tupFragptr[1];
// pointers to table fragments in TUP and ACC
FragrecordPtr tFragPtr;
tFragPtr.i = fragptr.p->tableFragptr;
ptrCheckGuard(tFragPtr, cfragrecFileSize, fragrecord);
tuxreq->tupTableFragPtrI[0] = tFragPtr.p->tupFragptr[0];
tuxreq->tupTableFragPtrI[1] = tFragPtr.p->tupFragptr[1];
tuxreq->accTableFragPtrI[0] = tFragPtr.p->accFragptr[0];
tuxreq->accTableFragPtrI[1] = tFragPtr.p->accFragptr[1];
sendSignal(fragptr.p->tuxBlockref, GSN_TUXFRAGREQ,
signal, TuxFragReq::SignalLength, JBB);
return;
}
}
ndbrequire(false);
}//Dblqh::sendAddFragReq
/* ************************************************************************> */
/* LQHADDATTRREQ: Request from DICT to create attributes for the new table. */
/* ************************************************************************> */
void Dblqh::execLQHADDATTREQ(Signal* signal)
{
jamEntry();
LqhAddAttrReq * const req = (LqhAddAttrReq*)signal->getDataPtr();
addfragptr.i = req->lqhFragPtr;
const Uint32 tnoOfAttr = req->noOfAttributes;
const Uint32 senderData = req->senderData;
const Uint32 senderAttrPtr = req->senderAttrPtr;
ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
ndbrequire(addfragptr.p->addfragStatus == AddFragRecord::WAIT_ADD_ATTR);
ndbrequire((tnoOfAttr != 0) && (tnoOfAttr <= LqhAddAttrReq::MAX_ATTRIBUTES));
addfragptr.p->totalAttrReceived += tnoOfAttr;
ndbrequire(addfragptr.p->totalAttrReceived <= addfragptr.p->noOfAttr);
addfragptr.p->attrReceived = tnoOfAttr;
for (Uint32 i = 0; i < tnoOfAttr; i++) {
addfragptr.p->attributes[i] = req->attributes[i];
}//for
addfragptr.p->attrSentToTup = 0;
ndbrequire(addfragptr.p->dictConnectptr == senderData);
addfragptr.p->m_senderAttrPtr = senderAttrPtr;
addfragptr.p->addfragStatus = AddFragRecord::TUP_ATTR_WAIT1;
sendAddAttrReq(signal);
}//Dblqh::execLQHADDATTREQ()
/* *********************>> */
/* TUP_ADD_ATTCONF > */
/* *********************>> */
void Dblqh::execTUP_ADD_ATTCONF(Signal* signal)
{
jamEntry();
addfragptr.i = signal->theData[0];
// implies that operation was released on the other side
const bool lastAttr = signal->theData[1];
ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
switch (addfragptr.p->addfragStatus) {
case AddFragRecord::TUP_ATTR_WAIT1:
jam();
if (lastAttr)
addfragptr.p->tup1Connectptr = RNIL;
addfragptr.p->addfragStatus = AddFragRecord::TUP_ATTR_WAIT2;
sendAddAttrReq(signal);
break;
case AddFragRecord::TUP_ATTR_WAIT2:
jam();
if (lastAttr)
addfragptr.p->tup2Connectptr = RNIL;
if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
addfragptr.p->addfragStatus = AddFragRecord::TUX_ATTR_WAIT1;
sendAddAttrReq(signal);
break;
}
goto done_with_attr;
break;
case AddFragRecord::TUX_ATTR_WAIT1:
jam();
if (lastAttr)
addfragptr.p->tux1Connectptr = RNIL;
addfragptr.p->addfragStatus = AddFragRecord::TUX_ATTR_WAIT2;
sendAddAttrReq(signal);
break;
case AddFragRecord::TUX_ATTR_WAIT2:
jam();
if (lastAttr)
addfragptr.p->tux2Connectptr = RNIL;
goto done_with_attr;
break;
done_with_attr:
addfragptr.p->attrSentToTup = addfragptr.p->attrSentToTup + 1;
ndbrequire(addfragptr.p->attrSentToTup <= addfragptr.p->attrReceived);
ndbrequire(addfragptr.p->totalAttrReceived <= addfragptr.p->noOfAttr);
if (addfragptr.p->attrSentToTup < addfragptr.p->attrReceived) {
// more in this batch
jam();
addfragptr.p->addfragStatus = AddFragRecord::TUP_ATTR_WAIT1;
sendAddAttrReq(signal);
} else if (addfragptr.p->totalAttrReceived < addfragptr.p->noOfAttr) {
// more batches to receive
jam();
addfragptr.p->addfragStatus = AddFragRecord::WAIT_ADD_ATTR;
LqhAddAttrConf *const conf = (LqhAddAttrConf*)signal->getDataPtrSend();
conf->senderData = addfragptr.p->dictConnectptr;
conf->senderAttrPtr = addfragptr.p->m_senderAttrPtr;
conf->fragId = addfragptr.p->addFragid;
sendSignal(addfragptr.p->dictBlockref, GSN_LQHADDATTCONF,
signal, LqhAddAttrConf::SignalLength, JBB);
} else {
fragptr.i = addfragptr.p->fragmentPtr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
/* ------------------------------------------------------------------
* WE HAVE NOW COMPLETED ADDING THIS FRAGMENT. WE NOW NEED TO SET THE
* PROPER STATE IN FRAG_STATUS DEPENDENT ON IF WE ARE CREATING A NEW
* REPLICA OR IF WE ARE CREATING A TABLE. FOR FRAGMENTS IN COPY
* PROCESS WE DO NOT WANT LOGGING ACTIVATED.
* ----------------------------------------------------------------- */
if (addfragptr.p->fragCopyCreation == 1) {
jam();
if (! DictTabInfo::isOrderedIndex(addfragptr.p->tableType))
fragptr.p->fragStatus = Fragrecord::ACTIVE_CREATION;
else
fragptr.p->fragStatus = Fragrecord::FSACTIVE;
fragptr.p->logFlag = Fragrecord::STATE_FALSE;
} else {
jam();
fragptr.p->fragStatus = Fragrecord::FSACTIVE;
}//if
LqhAddAttrConf *const conf = (LqhAddAttrConf*)signal->getDataPtrSend();
conf->senderData = addfragptr.p->dictConnectptr;
conf->senderAttrPtr = addfragptr.p->m_senderAttrPtr;
conf->fragId = addfragptr.p->addFragid;
sendSignal(addfragptr.p->dictBlockref, GSN_LQHADDATTCONF, signal,
LqhAddAttrConf::SignalLength, JBB);
releaseAddfragrec(signal);
}//if
break;
default:
ndbrequire(false);
break;
}
}
/* **********************>> */
/* TUX_ADD_ATTRCONF > */
/* **********************>> */
void Dblqh::execTUX_ADD_ATTRCONF(Signal* signal)
{
jamEntry();
execTUP_ADD_ATTCONF(signal);
}//Dblqh::execTUX_ADD_ATTRCONF
/*
* Add attribute in TUP or TUX. Called up to 4 times.
*/
void
Dblqh::sendAddAttrReq(Signal* signal)
{
arrGuard(addfragptr.p->attrSentToTup, LqhAddAttrReq::MAX_ATTRIBUTES);
LqhAddAttrReq::Entry& entry =
addfragptr.p->attributes[addfragptr.p->attrSentToTup];
const Uint32 attrId = entry.attrId & 0xffff;
const Uint32 primaryAttrId = entry.attrId >> 16;
fragptr.i = addfragptr.p->fragmentPtr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
if (addfragptr.p->addfragStatus == AddFragRecord::TUP_ATTR_WAIT1 ||
addfragptr.p->addfragStatus == AddFragRecord::TUP_ATTR_WAIT2) {
if (DictTabInfo::isTable(addfragptr.p->tableType) ||
DictTabInfo::isHashIndex(addfragptr.p->tableType) ||
(DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
primaryAttrId == ZNIL)) {
jam();
TupAddAttrReq* const tupreq = (TupAddAttrReq*)signal->getDataPtrSend();
tupreq->tupConnectPtr =
addfragptr.p->addfragStatus == AddFragRecord::TUP_ATTR_WAIT1
? addfragptr.p->tup1Connectptr : addfragptr.p->tup2Connectptr;
tupreq->notused1 = 0;
tupreq->attrId = attrId;
tupreq->attrDescriptor = entry.attrDescriptor;
tupreq->extTypeInfo = entry.extTypeInfo;
sendSignal(fragptr.p->tupBlockref, GSN_TUP_ADD_ATTRREQ,
signal, TupAddAttrReq::SignalLength, JBB);
return;
}
if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
primaryAttrId != ZNIL) {
// this attribute is not for TUP
jam();
TupAddAttrConf* tupconf = (TupAddAttrConf*)signal->getDataPtrSend();
tupconf->userPtr = addfragptr.i;
tupconf->lastAttr = false;
sendSignal(reference(), GSN_TUP_ADD_ATTCONF,
signal, TupAddAttrConf::SignalLength, JBB);
return;
}
}
if (addfragptr.p->addfragStatus == AddFragRecord::TUX_ATTR_WAIT1 ||
addfragptr.p->addfragStatus == AddFragRecord::TUX_ATTR_WAIT2) {
jam();
if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
primaryAttrId != ZNIL) {
jam();
TuxAddAttrReq* const tuxreq = (TuxAddAttrReq*)signal->getDataPtrSend();
tuxreq->tuxConnectPtr =
addfragptr.p->addfragStatus == AddFragRecord::TUX_ATTR_WAIT1
? addfragptr.p->tux1Connectptr : addfragptr.p->tux2Connectptr;
tuxreq->notused1 = 0;
tuxreq->attrId = attrId;
tuxreq->attrDescriptor = entry.attrDescriptor;
tuxreq->extTypeInfo = entry.extTypeInfo;
tuxreq->primaryAttrId = primaryAttrId;
sendSignal(fragptr.p->tuxBlockref, GSN_TUX_ADD_ATTRREQ,
signal, TuxAddAttrReq::SignalLength, JBB);
return;
}
if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
primaryAttrId == ZNIL) {
// this attribute is not for TUX
jam();
TuxAddAttrConf* tuxconf = (TuxAddAttrConf*)signal->getDataPtrSend();
tuxconf->userPtr = addfragptr.i;
tuxconf->lastAttr = false;
sendSignal(reference(), GSN_TUX_ADD_ATTRCONF,
signal, TuxAddAttrConf::SignalLength, JBB);
return;
}
}
ndbrequire(false);
}//Dblqh::sendAddAttrReq
/* ************************************************************************>> */
/* TAB_COMMITREQ: Commit the new table for use in transactions. Sender DICT. */
/* ************************************************************************>> */
void Dblqh::execTAB_COMMITREQ(Signal* signal)
{
jamEntry();
Uint32 dihPtr = signal->theData[0];
BlockReference dihBlockref = signal->theData[1];
tabptr.i = signal->theData[2];
if (tabptr.i >= ctabrecFileSize) {
jam();
terrorCode = ZTAB_FILE_SIZE;
signal->theData[0] = dihPtr;
signal->theData[1] = cownNodeid;
signal->theData[2] = tabptr.i;
signal->theData[3] = terrorCode;
sendSignal(dihBlockref, GSN_TAB_COMMITREF, signal, 4, JBB);
return;
}//if
ptrAss(tabptr, tablerec);
if (tabptr.p->tableStatus != Tablerec::ADD_TABLE_ONGOING) {
jam();
terrorCode = ZTAB_STATE_ERROR;
signal->theData[0] = dihPtr;
signal->theData[1] = cownNodeid;
signal->theData[2] = tabptr.i;
signal->theData[3] = terrorCode;
signal->theData[4] = tabptr.p->tableStatus;
sendSignal(dihBlockref, GSN_TAB_COMMITREF, signal, 5, JBB);
ndbrequire(false);
return;
}//if
tabptr.p->usageCount = 0;
tabptr.p->tableStatus = Tablerec::TABLE_DEFINED;
signal->theData[0] = dihPtr;
signal->theData[1] = cownNodeid;
signal->theData[2] = tabptr.i;
sendSignal(dihBlockref, GSN_TAB_COMMITCONF, signal, 3, JBB);
return;
}//Dblqh::execTAB_COMMITREQ()
void Dblqh::fragrefLab(Signal* signal,
BlockReference fragBlockRef,
Uint32 fragConPtr,
Uint32 errorCode)
{
LqhFragRef * ref = (LqhFragRef*)signal->getDataPtrSend();
ref->senderData = fragConPtr;
ref->errorCode = errorCode;
sendSignal(fragBlockRef, GSN_LQHFRAGREF, signal,
LqhFragRef::SignalLength, JBB);
return;
}//Dblqh::fragrefLab()
/*
* Abort on-going ops.
*/
void Dblqh::abortAddFragOps(Signal* signal)
{
fragptr.i = addfragptr.p->fragmentPtr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
signal->theData[0] = (Uint32)-1;
if (addfragptr.p->tup1Connectptr != RNIL) {
jam();
signal->theData[1] = addfragptr.p->tup1Connectptr;
sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ, signal, 2, JBB);
addfragptr.p->tup1Connectptr = RNIL;
}
if (addfragptr.p->tup2Connectptr != RNIL) {
jam();
signal->theData[1] = addfragptr.p->tup2Connectptr;
sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ, signal, 2, JBB);
addfragptr.p->tup2Connectptr = RNIL;
}
if (addfragptr.p->tux1Connectptr != RNIL) {
jam();
signal->theData[1] = addfragptr.p->tux1Connectptr;
sendSignal(fragptr.p->tuxBlockref, GSN_TUXFRAGREQ, signal, 2, JBB);
addfragptr.p->tux1Connectptr = RNIL;
}
if (addfragptr.p->tux2Connectptr != RNIL) {
jam();
signal->theData[1] = addfragptr.p->tux2Connectptr;
sendSignal(fragptr.p->tuxBlockref, GSN_TUXFRAGREQ, signal, 2, JBB);
addfragptr.p->tux2Connectptr = RNIL;
}
}
/* ************>> */
/* ACCFRAGREF > */
/* ************>> */
void Dblqh::execACCFRAGREF(Signal* signal)
{
jamEntry();
addfragptr.i = signal->theData[0];
ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
terrorCode = signal->theData[1];
ndbrequire(addfragptr.p->addfragStatus == AddFragRecord::ACC_ADDFRAG);
addfragptr.p->addfragErrorCode = terrorCode;
const Uint32 ref = addfragptr.p->dictBlockref;
const Uint32 senderData = addfragptr.p->dictConnectptr;
const Uint32 errorCode = addfragptr.p->addfragErrorCode;
releaseAddfragrec(signal);
fragrefLab(signal, ref, senderData, errorCode);
return;
}//Dblqh::execACCFRAGREF()
/* ************>> */
/* TUPFRAGREF > */
/* ************>> */
void Dblqh::execTUPFRAGREF(Signal* signal)
{
jamEntry();
addfragptr.i = signal->theData[0];
ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
terrorCode = signal->theData[1];
fragptr.i = addfragptr.p->fragmentPtr;
ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
addfragptr.p->addfragErrorCode = terrorCode;
// no operation to release, just add some jams
switch (addfragptr.p->addfragStatus) {
case AddFragRecord::WAIT_TWO_TUP:
jam();
break;
case AddFragRecord::WAIT_ONE_TUP:
jam();
break;
case AddFragRecord::WAIT_TWO_TUX:
jam();
break;
case AddFragRecord::WAIT_ONE_TUX:
jam();
break;
default:
ndbrequire(false);
break;
}
abortAddFragOps(signal);
const Uint32 ref = addfragptr.p->dictBlockref;
const Uint32 senderData = addfragptr.p->dictConnectptr;
const Uint32 errorCode = addfragptr.p->addfragErrorCode;
releaseAddfragrec(signal);
fragrefLab(signal, ref, senderData, errorCode);
}//Dblqh::execTUPFRAGREF()
/* ************>> */
/* TUXFRAGREF > */
/* ************>> */
void Dblqh::execTUXFRAGREF(Signal* signal)
{
jamEntry();
execTUPFRAGREF(signal);
}//Dblqh::execTUXFRAGREF
/* *********************> */
/* TUP_ADD_ATTREF > */
/* *********************> */
void Dblqh::execTUP_ADD_ATTRREF(Signal* signal)
{
jamEntry();
addfragptr.i = signal->theData[0];
ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
terrorCode = signal->theData[1];
addfragptr.p->addfragErrorCode = terrorCode;
// operation was released on the other side
switch (addfragptr.p->addfragStatus) {
case AddFragRecord::TUP_ATTR_WAIT1:
jam();
ndbrequire(addfragptr.p->tup1Connectptr != RNIL);
addfragptr.p->tup1Connectptr = RNIL;
break;
case AddFragRecord::TUP_ATTR_WAIT2:
jam();
ndbrequire(addfragptr.p->tup2Connectptr != RNIL);
addfragptr.p->tup2Connectptr = RNIL;
break;
case AddFragRecord::TUX_ATTR_WAIT1:
jam();
ndbrequire(addfragptr.p->tux1Connectptr != RNIL);
addfragptr.p->tux1Connectptr = RNIL;
break;
case AddFragRecord::TUX_ATTR_WAIT2:
jam();
ndbrequire(addfragptr.p->tux2Connectptr != RNIL);
addfragptr.p->tux2Connectptr = RNIL;
break;
default:
ndbrequire(false);
break;
}
abortAddFragOps(signal);
const Uint32 Ref = addfragptr.p->dictBlockref;
const Uint32 senderData = addfragptr.p->dictConnectptr;
const Uint32 errorCode = addfragptr.p->addfragErrorCode;
releaseAddfragrec(signal);
LqhAddAttrRef *const ref = (LqhAddAttrRef*)signal->getDataPtrSend();
ref->senderData = senderData;
ref->errorCode = errorCode;
sendSignal(Ref, GSN_LQHADDATTREF, signal,
LqhAddAttrRef::SignalLength, JBB);
}//Dblqh::execTUP_ADD_ATTRREF()
/* **********************> */
/* TUX_ADD_ATTRREF > */
/* **********************> */
void Dblqh::execTUX_ADD_ATTRREF(Signal* signal)
{
jamEntry();
execTUP_ADD_ATTRREF(signal);
}//Dblqh::execTUX_ADD_ATTRREF
void
Dblqh::execPREP_DROP_TAB_REQ(Signal* signal){
jamEntry();
PrepDropTabReq* req = (PrepDropTabReq*)signal->getDataPtr();
Uint32 senderRef = req->senderRef;
Uint32 senderData = req->senderData;
TablerecPtr tabPtr;
tabPtr.i = req->tableId;
ptrCheckGuard(tabPtr, ctabrecFileSize, tablerec);
Uint32 errCode = 0;
errCode = checkDropTabState(tabPtr.p->tableStatus, GSN_PREP_DROP_TAB_REQ);
if(errCode != 0){
jam();
PrepDropTabRef* ref = (PrepDropTabRef*)signal->getDataPtrSend();
ref->senderRef = reference();
ref->senderData = senderData;
ref->tableId = tabPtr.i;
ref->errorCode = errCode;
sendSignal(senderRef, GSN_PREP_DROP_TAB_REF, signal,
PrepDropTabRef::SignalLength, JBB);
return;
}
tabPtr.p->tableStatus = Tablerec::PREP_DROP_TABLE_ONGOING;
tabPtr.p->waitingTC.clear();
tabPtr.p->waitingDIH.clear();
PrepDropTabConf * conf = (PrepDropTabConf*)signal->getDataPtrSend();
conf->tableId = tabPtr.i;
conf->senderRef = reference();
conf->senderData = senderData;
sendSignal(senderRef, GSN_PREP_DROP_TAB_CONF, signal,
PrepDropTabConf::SignalLength, JBB);
signal->theData[0] = ZPREP_DROP_TABLE;
signal->theData[1] = tabPtr.i;
signal->theData[2] = senderRef;
signal->theData[3] = senderData;
checkDropTab(signal);
}
void
Dblqh::checkDropTab(Signal* signal){
TablerecPtr tabPtr;
tabPtr.i = signal->theData[1];
ptrCheckGuard(tabPtr, ctabrecFileSize, tablerec);
ndbrequire(tabPtr.p->tableStatus == Tablerec::PREP_DROP_TABLE_ONGOING);
if(tabPtr.p->usageCount > 0){
jam();
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 100, 4);
return;
}
bool lcpDone = true;
lcpPtr.i = 0;
ptrAss(lcpPtr, lcpRecord);
if(lcpPtr.p->lcpState != LcpRecord::LCP_IDLE){
jam();
if(lcpPtr.p->currentFragment.lcpFragOrd.tableId == tabPtr.i){
jam();
lcpDone = false;
}
if(lcpPtr.p->lcpQueued &&
lcpPtr.p->queuedFragment.lcpFragOrd.tableId == tabPtr.i){
jam();
lcpDone = false;
}
}
if(!lcpDone){
jam();
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 100, 4);
return;
}
tabPtr.p->tableStatus = Tablerec::PREP_DROP_TABLE_DONE;
WaitDropTabConf * conf = (WaitDropTabConf*)signal->getDataPtrSend();
conf->tableId = tabPtr.i;
conf->senderRef = reference();
for(Uint32 i = 1; i<MAX_NDB_NODES; i++){
if(tabPtr.p->waitingTC.get(i)){
tabPtr.p->waitingTC.clear(i);
sendSignal(calcTcBlockRef(i), GSN_WAIT_DROP_TAB_CONF, signal,
WaitDropTabConf::SignalLength, JBB);
}
if(tabPtr.p->waitingDIH.get(i)){
tabPtr.p->waitingDIH.clear(i);
sendSignal(calcDihBlockRef(i), GSN_WAIT_DROP_TAB_CONF, signal,
WaitDropTabConf::SignalLength, JBB);
}
}
}
void
Dblqh::execWAIT_DROP_TAB_REQ(Signal* signal){
jamEntry();
WaitDropTabReq * req = (WaitDropTabReq*)signal->getDataPtr();
TablerecPtr tabPtr;
tabPtr.i = req->tableId;
ptrCheckGuard(tabPtr, ctabrecFileSize, tablerec);
Uint32 senderRef = req->senderRef;
Uint32 nodeId = refToNode(senderRef);
Uint32 blockNo = refToBlock(senderRef);
if(tabPtr.p->tableStatus == Tablerec::PREP_DROP_TABLE_ONGOING){
jam();
switch(blockNo){
case DBTC:
tabPtr.p->waitingTC.set(nodeId);
break;
case DBDIH:
tabPtr.p->waitingDIH.set(nodeId);
break;
default:
ndbrequire(false);
}
return;
}
if(tabPtr.p->tableStatus == Tablerec::PREP_DROP_TABLE_DONE){
jam();
WaitDropTabConf * conf = (WaitDropTabConf*)signal->getDataPtrSend();
conf->tableId = tabPtr.i;
conf->senderRef = reference();
sendSignal(senderRef, GSN_WAIT_DROP_TAB_CONF, signal,
WaitDropTabConf::SignalLength, JBB);
return;
}
WaitDropTabRef * ref = (WaitDropTabRef*)signal->getDataPtrSend();
ref->tableId = tabPtr.i;
ref->senderRef = reference();
bool ok = false;
switch(tabPtr.p->tableStatus){
case Tablerec::TABLE_DEFINED:
ok = true;
ref->errorCode = WaitDropTabRef::IllegalTableState;
break;
case Tablerec::NOT_DEFINED:
ok = true;
ref->errorCode = WaitDropTabRef::NoSuchTable;
break;
case Tablerec::ADD_TABLE_ONGOING:
ok = true;
ref->errorCode = WaitDropTabRef::IllegalTableState;
break;
case Tablerec::PREP_DROP_TABLE_ONGOING:
case Tablerec::PREP_DROP_TABLE_DONE:
// Should have been take care of above
ndbrequire(false);
}
ndbrequire(ok);
ref->tableStatus = tabPtr.p->tableStatus;
sendSignal(senderRef, GSN_WAIT_DROP_TAB_REF, signal,
WaitDropTabRef::SignalLength, JBB);
return;
}
void
Dblqh::execDROP_TAB_REQ(Signal* signal){
jamEntry();
DropTabReq* req = (DropTabReq*)signal->getDataPtr();
Uint32 senderRef = req->senderRef;
Uint32 senderData = req->senderData;
TablerecPtr tabPtr;
tabPtr.i = req->tableId;
ptrCheckGuard(tabPtr, ctabrecFileSize, tablerec);
do {
if(req->requestType == DropTabReq::RestartDropTab){
jam();
break;
}
if(req->requestType == DropTabReq::OnlineDropTab){
jam();
Uint32 errCode = 0;
errCode = checkDropTabState(tabPtr.p->tableStatus, GSN_DROP_TAB_REQ);
if(errCode != 0){
jam();
DropTabRef* ref = (DropTabRef*)signal->getDataPtrSend();
ref->senderRef = reference();
ref->senderData = senderData;
ref->tableId = tabPtr.i;
ref->errorCode = errCode;
sendSignal(senderRef, GSN_DROP_TAB_REF, signal,
DropTabRef::SignalLength, JBB);
return;
}
}
removeTable(tabPtr.i);
} while(false);
ndbrequire(tabPtr.p->usageCount == 0);
tabPtr.p->tableStatus = Tablerec::NOT_DEFINED;
DropTabConf * const dropConf = (DropTabConf *)signal->getDataPtrSend();
dropConf->senderRef = reference();
dropConf->senderData = senderData;
dropConf->tableId = tabPtr.i;
sendSignal(senderRef, GSN_DROP_TAB_CONF,
signal, DropTabConf::SignalLength, JBB);
}
Uint32
Dblqh::checkDropTabState(Tablerec::TableStatus status, Uint32 gsn) const{
if(gsn == GSN_PREP_DROP_TAB_REQ){
switch(status){
case Tablerec::NOT_DEFINED:
jam();
// Fall through
case Tablerec::ADD_TABLE_ONGOING:
jam();
return PrepDropTabRef::NoSuchTable;
break;
case Tablerec::PREP_DROP_TABLE_ONGOING:
jam();
return PrepDropTabRef::PrepDropInProgress;
break;
case Tablerec::PREP_DROP_TABLE_DONE:
jam();
return PrepDropTabRef::DropInProgress;
break;
case Tablerec::TABLE_DEFINED:
jam();
return 0;
break;
}
ndbrequire(0);
}
if(gsn == GSN_DROP_TAB_REQ){
switch(status){
case Tablerec::NOT_DEFINED:
jam();
// Fall through
case Tablerec::ADD_TABLE_ONGOING:
jam();
return DropTabRef::NoSuchTable;
break;
case Tablerec::PREP_DROP_TABLE_ONGOING:
jam();
return DropTabRef::PrepDropInProgress;
break;
case Tablerec::PREP_DROP_TABLE_DONE:
jam();
return 0;
break;
case Tablerec::TABLE_DEFINED:
jam();
return DropTabRef::DropWoPrep;
}
ndbrequire(0);
}
ndbrequire(0);
return RNIL;
}
void Dblqh::removeTable(Uint32 tableId)
{
tabptr.i = tableId;
ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
for (Uint32 i = (MAX_FRAG_PER_NODE - 1); (Uint32)~i; i--) {
jam();
if (tabptr.p->fragid[i] != ZNIL) {
jam();
deleteFragrec(tabptr.p->fragid[i]);
}//if
}//for
}//Dblqh::removeTable()
void
Dblqh::execALTER_TAB_REQ(Signal* signal)
{
jamEntry();
AlterTabReq* const req = (AlterTabReq*)signal->getDataPtr();
const Uint32 senderRef = req->senderRef;
const Uint32 senderData = req->senderData;
const Uint32 changeMask = req->changeMask;
const Uint32 tableId = req->tableId;
const Uint32 tableVersion = req->tableVersion;
const Uint32 gci = req->gci;
AlterTabReq::RequestType requestType =
(AlterTabReq::RequestType) req->requestType;
TablerecPtr tablePtr;
tablePtr.i = tableId;
ptrCheckGuard(tablePtr, ctabrecFileSize, tablerec);
tablePtr.p->schemaVersion = tableVersion;
// Request handled successfully
AlterTabConf * conf = (AlterTabConf*)signal->getDataPtrSend();
conf->senderRef = reference();
conf->senderData = senderData;
conf->changeMask = changeMask;
conf->tableId = tableId;
conf->tableVersion = tableVersion;
conf->gci = gci;
conf->requestType = requestType;
sendSignal(senderRef, GSN_ALTER_TAB_CONF, signal,
AlterTabConf::SignalLength, JBB);
}
/* ************************************************************************>>
* TIME_SIGNAL: Handles time-out of local operations. This is a clean-up
* handler. If no other measure has succeeded in cleaning up after time-outs
* or else then this routine will remove the transaction after 120 seconds of
* inactivity. The check is performed once per 10 second. Sender is QMGR.
* ************************************************************************>> */
void Dblqh::execTIME_SIGNAL(Signal* signal)
{
jamEntry();
cLqhTimeOutCount++;
cLqhTimeOutCheckCount++;
if ((cCounterAccCommitBlocked > 0) ||
(cCounterTupCommitBlocked > 0)) {
jam();
signal->theData[0] = EventReport::UndoLogBlocked;
signal->theData[1] = cCounterTupCommitBlocked;
signal->theData[2] = cCounterAccCommitBlocked;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 3, JBB);
cCounterTupCommitBlocked = 0;
cCounterAccCommitBlocked = 0;
}//if
if (cLqhTimeOutCheckCount < 10) {
jam();
return;
}//if
cLqhTimeOutCheckCount = 0;
#ifdef VM_TRACE
TcConnectionrecPtr tTcConptr;
for (tTcConptr.i = 0; tTcConptr.i < ctcConnectrecFileSize;
tTcConptr.i++) {
jam();
ptrAss(tTcConptr, tcConnectionrec);
if ((tTcConptr.p->tcTimer != 0) &&
((tTcConptr.p->tcTimer + 120) < cLqhTimeOutCount)) {
ndbout << "Dblqh::execTIME_SIGNAL"<<endl
<< "Timeout found in tcConnectRecord " <<tTcConptr.i<<endl
<< " cLqhTimeOutCount = " << cLqhTimeOutCount << endl
<< " tcTimer="<<tTcConptr.p->tcTimer<<endl
<< " tcTimer+120="<<tTcConptr.p->tcTimer + 120<<endl;
ndbout << " transactionState = " << tTcConptr.p->transactionState<<endl;
ndbout << " operation = " << tTcConptr.p->operation<<endl;
ndbout << " tcNodeFailrec = " << tTcConptr.p->tcNodeFailrec
<< " seqNoReplica = " << tTcConptr.p->seqNoReplica
<< " simpleRead = " << tTcConptr.p->simpleRead
<< endl;
ndbout << " replicaType = " << tTcConptr.p->replicaType
<< " reclenAiLqhkey = " << tTcConptr.p->reclenAiLqhkey
<< " opExec = " << tTcConptr.p->opExec
<< endl;
ndbout << " opSimple = " << tTcConptr.p->opSimple
<< " nextSeqNoReplica = " << tTcConptr.p->nextSeqNoReplica
<< " lockType = " << tTcConptr.p->lockType
<< " localFragptr = " << tTcConptr.p->localFragptr
<< endl;
ndbout << " lastReplicaNo = " << tTcConptr.p->lastReplicaNo
<< " indTakeOver = " << tTcConptr.p->indTakeOver
<< " dirtyOp = " << tTcConptr.p->dirtyOp
<< endl;
ndbout << " activeCreat = " << tTcConptr.p->activeCreat
<< " tcBlockref = " << hex << tTcConptr.p->tcBlockref
<< " reqBlockref = " << hex << tTcConptr.p->reqBlockref
<< " primKeyLen = " << tTcConptr.p->primKeyLen
<< endl;
ndbout << " nextReplica = " << tTcConptr.p->nextReplica
<< " tcBlockref = " << hex << tTcConptr.p->tcBlockref
<< " reqBlockref = " << hex << tTcConptr.p->reqBlockref
<< " primKeyLen = " << tTcConptr.p->primKeyLen
<< endl;
ndbout << " logStopPageNo = " << tTcConptr.p->logStopPageNo
<< " logStartPageNo = " << tTcConptr.p->logStartPageNo
<< " logStartPageIndex = " << tTcConptr.p->logStartPageIndex
<< endl;
ndbout << " errorCode = " << tTcConptr.p->errorCode
<< " clientBlockref = " << hex << tTcConptr.p->clientBlockref
<< " applRef = " << hex << tTcConptr.p->applRef
<< " totSendlenAi = " << tTcConptr.p->totSendlenAi
<< endl;
ndbout << " totReclenAi = " << tTcConptr.p->totReclenAi
<< " tcScanRec = " << tTcConptr.p->tcScanRec
<< " tcScanInfo = " << tTcConptr.p->tcScanInfo
<< " tcOprec = " << hex << tTcConptr.p->tcOprec
<< endl;
ndbout << " tableref = " << tTcConptr.p->tableref
<< " simpleTcConnect = " << tTcConptr.p->simpleTcConnect
<< " storedProcId = " << tTcConptr.p->storedProcId
<< " schemaVersion = " << tTcConptr.p->schemaVersion
<< endl;
ndbout << " reqinfo = " << tTcConptr.p->reqinfo
<< " reqRef = " << tTcConptr.p->reqRef
<< " readlenAi = " << tTcConptr.p->readlenAi
<< " prevTc = " << tTcConptr.p->prevTc
<< endl;
ndbout << " prevLogTcrec = " << tTcConptr.p->prevLogTcrec
<< " prevHashRec = " << tTcConptr.p->prevHashRec
<< " nodeAfterNext0 = " << tTcConptr.p->nodeAfterNext[0]
<< " nodeAfterNext1 = " << tTcConptr.p->nodeAfterNext[1]
<< endl;
ndbout << " nextTcConnectrec = " << tTcConptr.p->nextTcConnectrec
<< " nextTc = " << tTcConptr.p->nextTc
<< " nextTcLogQueue = " << tTcConptr.p->nextTcLogQueue
<< " nextLogTcrec = " << tTcConptr.p->nextLogTcrec
<< endl;
ndbout << " nextHashRec = " << tTcConptr.p->nextHashRec
<< " logWriteState = " << tTcConptr.p->logWriteState
<< " logStartFileNo = " << tTcConptr.p->logStartFileNo
<< " listState = " << tTcConptr.p->listState
<< endl;
ndbout << " lastAttrinbuf = " << tTcConptr.p->lastAttrinbuf
<< " lastTupkeybuf = " << tTcConptr.p->lastTupkeybuf
<< " hashValue = " << tTcConptr.p->hashValue
<< endl;
ndbout << " gci = " << tTcConptr.p->gci
<< " fragmentptr = " << tTcConptr.p->fragmentptr
<< " fragmentid = " << tTcConptr.p->fragmentid
<< " firstTupkeybuf = " << tTcConptr.p->firstTupkeybuf
<< endl;
ndbout << " firstAttrinbuf = " << tTcConptr.p->firstAttrinbuf
<< " currTupAiLen = " << tTcConptr.p->currTupAiLen
<< " currReclenAi = " << tTcConptr.p->currReclenAi
<< endl;
ndbout << " tcTimer = " << tTcConptr.p->tcTimer
<< " clientConnectrec = " << tTcConptr.p->clientConnectrec
<< " applOprec = " << hex << tTcConptr.p->applOprec
<< " abortState = " << tTcConptr.p->abortState
<< endl;
ndbout << " transid0 = " << hex << tTcConptr.p->transid[0]
<< " transid1 = " << hex << tTcConptr.p->transid[1]
<< " tupkeyData0 = " << tTcConptr.p->tupkeyData[0]
<< " tupkeyData1 = " << tTcConptr.p->tupkeyData[1]
<< endl;
ndbout << " tupkeyData2 = " << tTcConptr.p->tupkeyData[2]
<< " tupkeyData3 = " << tTcConptr.p->tupkeyData[3]
<< endl;
switch (tTcConptr.p->transactionState) {
case TcConnectionrec::SCAN_STATE_USED:
if (tTcConptr.p->tcScanRec < cscanrecFileSize){
ScanRecordPtr TscanPtr;
c_scanRecordPool.getPtr(TscanPtr, tTcConptr.p->tcScanRec);
ndbout << " scanState = " << TscanPtr.p->scanState << endl;
//TscanPtr.p->scanLocalref[2];
ndbout << " copyPtr="<<TscanPtr.p->copyPtr
<< " scanAccPtr="<<TscanPtr.p->scanAccPtr
<< " scanAiLength="<<TscanPtr.p->scanAiLength
<< endl;
ndbout << " m_curr_batch_size_rows="<<
TscanPtr.p->m_curr_batch_size_rows
<< " m_max_batch_size_rows="<<
TscanPtr.p->m_max_batch_size_rows
<< " scanErrorCounter="<<TscanPtr.p->scanErrorCounter
<< " scanLocalFragid="<<TscanPtr.p->scanLocalFragid
<< endl;
ndbout << " scanSchemaVersion="<<TscanPtr.p->scanSchemaVersion
<< " scanStoredProcId="<<TscanPtr.p->scanStoredProcId
<< " scanTcrec="<<TscanPtr.p->scanTcrec
<< endl;
ndbout << " scanType="<<TscanPtr.p->scanType
<< " scanApiBlockref="<<TscanPtr.p->scanApiBlockref
<< " scanNodeId="<<TscanPtr.p->scanNodeId
<< " scanCompletedStatus="<<TscanPtr.p->scanCompletedStatus
<< endl;
ndbout << " scanFlag="<<TscanPtr.p->scanFlag
<< " scanLockHold="<<TscanPtr.p->scanLockHold
<< " scanLockMode="<<TscanPtr.p->scanLockMode
<< " scanNumber="<<TscanPtr.p->scanNumber
<< endl;
ndbout << " scanReleaseCounter="<<TscanPtr.p->scanReleaseCounter
<< " scanTcWaiting="<<TscanPtr.p->scanTcWaiting
<< " scanKeyinfoFlag="<<TscanPtr.p->scanKeyinfoFlag
<< endl;
}else{
ndbout << "No connected scan record found" << endl;
}
break;
default:
break;
}//switch
// Reset the timer
tTcConptr.p->tcTimer = 0;
}//if
}//for
#endif
#ifdef VM_TRACE
for (lfoPtr.i = 0; lfoPtr.i < clfoFileSize; lfoPtr.i++) {
ptrAss(lfoPtr, logFileOperationRecord);
if ((lfoPtr.p->lfoTimer != 0) &&
((lfoPtr.p->lfoTimer + 120) < cLqhTimeOutCount)) {
ndbout << "We have lost LFO record" << endl;
ndbout << "index = " << lfoPtr.i;
ndbout << "State = " << lfoPtr.p->lfoState;
ndbout << " Page No = " << lfoPtr.p->lfoPageNo;
ndbout << " noPagesRw = " << lfoPtr.p->noPagesRw;
ndbout << "lfoWordWritten = " << lfoPtr.p->lfoWordWritten << endl;
lfoPtr.p->lfoTimer = cLqhTimeOutCount;
}//if
}//for
#endif
#if 0
LcpRecordPtr TlcpPtr;
// Print information about the current local checkpoint
TlcpPtr.i = 0;
ptrAss(TlcpPtr, lcpRecord);
ndbout << "Information about LCP in this LQH" << endl
<< " lcpState="<<TlcpPtr.p->lcpState<<endl
<< " firstLcpLocAcc="<<TlcpPtr.p->firstLcpLocAcc<<endl
<< " firstLcpLocTup="<<TlcpPtr.p->firstLcpLocTup<<endl
<< " lcpAccptr="<<TlcpPtr.p->lcpAccptr<<endl
<< " lastFragmentFlag="<<TlcpPtr.p->lastFragmentFlag<<endl
<< " lcpQueued="<<TlcpPtr.p->lcpQueued<<endl
<< " reportEmptyref="<< TlcpPtr.p->reportEmptyRef<<endl
<< " reportEmpty="<<TlcpPtr.p->reportEmpty<<endl;
#endif
}//Dblqh::execTIME_SIGNAL()
/* ######################################################################### */
/* ####### EXECUTION MODULE ####### */
/* THIS MODULE HANDLES THE RECEPTION OF LQHKEYREQ AND ALL PROCESSING */
/* OF OPERATIONS ON BEHALF OF THIS REQUEST. THIS DOES ALSO INVOLVE */
/* RECEPTION OF VARIOUS TYPES OF ATTRINFO AND KEYINFO. IT DOES ALSO */
/* INVOLVE COMMUNICATION WITH ACC AND TUP. */
/* ######################################################################### */
void Dblqh::noFreeRecordLab(Signal* signal,
const LqhKeyReq * lqhKeyReq,
Uint32 errCode)
{
jamEntry();
const Uint32 transid1 = lqhKeyReq->transId1;
const Uint32 transid2 = lqhKeyReq->transId2;
const Uint32 reqInfo = lqhKeyReq->requestInfo;
if(errCode == ZNO_FREE_MARKER_RECORDS_ERROR ||
errCode == ZNODE_SHUTDOWN_IN_PROGESS){
releaseTcrec(signal, tcConnectptr);
}
if (LqhKeyReq::getSimpleFlag(reqInfo) &&
LqhKeyReq::getOperation(reqInfo) == ZREAD){
jam();
ndbrequire(LqhKeyReq::getApplicationAddressFlag(reqInfo));
const Uint32 apiRef = lqhKeyReq->variableData[0];
const Uint32 apiOpRec = lqhKeyReq->variableData[1];
TcKeyRef * const tcKeyRef = (TcKeyRef *) signal->getDataPtrSend();
tcKeyRef->connectPtr = apiOpRec;
tcKeyRef->transId[0] = transid1;
tcKeyRef->transId[1] = transid2;
tcKeyRef->errorCode = errCode;
sendSignal(apiRef, GSN_TCKEYREF, signal, TcKeyRef::SignalLength, JBB);
} else {
jam();
const Uint32 clientPtr = lqhKeyReq->clientConnectPtr;
Uint32 TcOprec = clientPtr;
if(LqhKeyReq::getSameClientAndTcFlag(reqInfo) == 1){
if(LqhKeyReq::getApplicationAddressFlag(reqInfo))
TcOprec = lqhKeyReq->variableData[2];
else
TcOprec = lqhKeyReq->variableData[0];
}
LqhKeyRef * const ref = (LqhKeyRef*)signal->getDataPtrSend();
ref->userRef = clientPtr;
ref->connectPtr = TcOprec;
ref->errorCode = errCode;
ref->transId1 = transid1;
ref->transId2 = transid2;
sendSignal(signal->senderBlockRef(), GSN_LQHKEYREF, signal,
LqhKeyRef::SignalLength, JBB);
}//if
return;
}//Dblqh::noFreeRecordLab()
void Dblqh::LQHKEY_abort(Signal* signal, int errortype)
{
switch (errortype) {
case 0:
jam();
terrorCode = ZCOPY_NODE_ERROR;
break;
case 1:
jam();
terrorCode = ZNO_FREE_LQH_CONNECTION;
break;
case 2:
jam();
terrorCode = signal->theData[1];
break;
case 3:
jam();
ndbrequire((tcConnectptr.p->transactionState == TcConnectionrec::WAIT_ACC_ABORT) ||
(tcConnectptr.p->transactionState == TcConnectionrec::ABORT_STOPPED) ||
(tcConnectptr.p->transactionState == TcConnectionrec::ABORT_QUEUED));
return;
break;
case 4:
jam();
if(tabptr.p->tableStatus == Tablerec::NOT_DEFINED){
jam();
terrorCode = ZTABLE_NOT_DEFINED;
} else if (tabptr.p->tableStatus == Tablerec::PREP_DROP_TABLE_ONGOING ||
tabptr.p->tableStatus == Tablerec::PREP_DROP_TABLE_DONE){
jam();
terrorCode = ZDROP_TABLE_IN_PROGRESS;
} else {
ndbrequire(0);
}
break;
case 5:
jam();
terrorCode = ZINVALID_SCHEMA_VERSION;
break;
default:
ndbrequire(false);
break;
}//switch
abortErrorLab(signal);
}//Dblqh::LQHKEY_abort()
void Dblqh::LQHKEY_error(Signal* signal, int errortype)
{
switch (errortype) {
case 0:
jam();
break;
case 1:
jam();
break;
case 2:
jam();
break;
case 3:
jam();
break;
case 4:
jam();
break;
case 5:
jam();
break;
case 6:
jam();
break;
default:
jam();
break;
}//switch
ndbrequire(false);
}//Dblqh::LQHKEY_error()
void Dblqh::execLQHKEYREF(Signal* signal)
{
jamEntry();
tcConnectptr.i = signal->theData[0];
terrorCode = signal->theData[2];
Uint32 transid1 = signal->theData[3];
Uint32 transid2 = signal->theData[4];
if (tcConnectptr.i >= ctcConnectrecFileSize) {
errorReport(signal, 3);
return;
}//if
/*------------------------------------------------------------------*/
/* WE HAVE TO CHECK THAT THE SIGNAL DO NOT BELONG TO SOMETHING*/
/* REMOVED DUE TO A TIME-OUT. */
/*------------------------------------------------------------------*/
ptrAss(tcConnectptr, tcConnectionrec);
TcConnectionrec * const regTcPtr = tcConnectptr.p;
switch (regTcPtr->connectState) {
case TcConnectionrec::CONNECTED:
jam();
if ((regTcPtr->transid[0] != transid1) ||
(regTcPtr->transid[1] != transid2)) {
warningReport(signal, 14);
return;
}//if
if (regTcPtr->abortState != TcConnectionrec::ABORT_IDLE) {
warningReport(signal, 15);
return;
}//if
abortErrorLab(signal);
return;
break;
case TcConnectionrec::LOG_CONNECTED:
jam();
logLqhkeyrefLab(signal);
return;
break;
case TcConnectionrec::COPY_CONNECTED:
jam();
copyLqhKeyRefLab(signal);
return;
break;
default:
warningReport(signal, 16);
return;
break;
}//switch
}//Dblqh::execLQHKEYREF()
/* -------------------------------------------------------------------------- */
/* ------- ENTER PACKED_SIGNAL ------- */
/* Execution of packed signal. The packed signal can contain COMMIT, COMPLETE */
/* or LQHKEYCONF signals. These signals will be executed by their resp. exec */
/* functions. */
/* -------------------------------------------------------------------------- */
void Dblqh::execPACKED_SIGNAL(Signal* signal)
{
Uint32 Tstep = 0;
Uint32 Tlength;
Uint32 TpackedData[28];
Uint32 sig0, sig1, sig2, sig3 ,sig4, sig5, sig6;
jamEntry();
Tlength = signal->length();
ndbrequire(Tlength <= 25);
MEMCOPY_NO_WORDS(&TpackedData[0], &signal->theData[0], Tlength);
while (Tlength > Tstep) {
switch (TpackedData[Tstep] >> 28) {
case ZCOMMIT:
jam();
sig0 = TpackedData[Tstep + 0] & 0x0FFFFFFF;
sig1 = TpackedData[Tstep + 1];
sig2 = TpackedData[Tstep + 2];
sig3 = TpackedData[Tstep + 3];
signal->theData[0] = sig0;
signal->theData[1] = sig1;
signal->theData[2] = sig2;
signal->theData[3] = sig3;
signal->header.theLength = 4;
execCOMMIT(signal);
Tstep += 4;
break;
case ZCOMPLETE:
jam();
sig0 = TpackedData[Tstep + 0] & 0x0FFFFFFF;
sig1 = TpackedData[Tstep + 1];
sig2 = TpackedData[Tstep + 2];
signal->theData[0] = sig0;
signal->theData[1] = sig1;
signal->theData[2] = sig2;
signal->header.theLength = 3;
execCOMPLETE(signal);
Tstep += 3;
break;
case ZLQHKEYCONF: {
jam();
LqhKeyConf * const lqhKeyConf = (LqhKeyConf *)signal->getDataPtr();
sig0 = TpackedData[Tstep + 0] & 0x0FFFFFFF;
sig1 = TpackedData[Tstep + 1];
sig2 = TpackedData[Tstep + 2];
sig3 = TpackedData[Tstep + 3];
sig4 = TpackedData[Tstep + 4];
sig5 = TpackedData[Tstep + 5];
sig6 = TpackedData[Tstep + 6];
lqhKeyConf->connectPtr = sig0;
lqhKeyConf->opPtr = sig1;
lqhKeyConf->userRef = sig2;
lqhKeyConf->readLen = sig3;
lqhKeyConf->transId1 = sig4;
lqhKeyConf->transId2 = sig5;
lqhKeyConf->noFiredTriggers = sig6;
execLQHKEYCONF(signal);
Tstep += LqhKeyConf::SignalLength;
break;
}
case ZREMOVE_MARKER:
jam();
sig0 = TpackedData[Tstep + 1];
sig1 = TpackedData[Tstep + 2];
signal->theData[0] = sig0;
signal->theData[1] = sig1;
signal->header.theLength = 2;
execREMOVE_MARKER_ORD(signal);
Tstep += 3;
break;
default:
ndbrequire(false);
return;
}//switch
}//while
ndbrequire(Tlength == Tstep);
return;
}//Dblqh::execPACKED_SIGNAL()
void
Dblqh::execREMOVE_MARKER_ORD(Signal* signal)
{
CommitAckMarker key;
key.transid1 = signal->theData[0];
key.transid2 = signal->theData[1];
jamEntry();
CommitAckMarkerPtr removedPtr;
m_commitAckMarkerHash.release(removedPtr, key);
ndbrequire(removedPtr.i != RNIL);
#ifdef MARKER_TRACE
ndbout_c("Rem marker[%.8x %.8x]", key.transid1, key.transid2);
#endif
}
/* -------------------------------------------------------------------------- */
/* ------- ENTER SEND_PACKED ------- */
/* Used to force a packed signal to be sent if local signal buffer is not */
/* empty. */
/* -------------------------------------------------------------------------- */
void Dblqh::execSEND_PACKED(Signal* signal)
{
HostRecordPtr Thostptr;
UintR i;
UintR TpackedListIndex = cpackedListIndex;
jamEntry();
for (i = 0; i < TpackedListIndex; i++) {
Thostptr.i = cpackedList[i];
ptrAss(Thostptr, hostRecord);
jam();
ndbrequire(Thostptr.i - 1 < MAX_NDB_NODES - 1);
if (Thostptr.p->noOfPackedWordsLqh > 0) {
jam();
sendPackedSignalLqh(signal, Thostptr.p);
}//if
if (Thostptr.p->noOfPackedWordsTc > 0) {
jam();
sendPackedSignalTc(signal, Thostptr.p);
}//if
Thostptr.p->inPackedList = false;
}//for
cpackedListIndex = 0;
return;
}//Dblqh::execSEND_PACKED()
void
Dblqh::updatePackedList(Signal* signal, HostRecord * ahostptr, Uint16 hostId)
{
Uint32 TpackedListIndex = cpackedListIndex;
if (ahostptr->inPackedList == false) {
jam();
ahostptr->inPackedList = true;
cpackedList[TpackedListIndex] = hostId;
cpackedListIndex = TpackedListIndex + 1;
}//if
}//Dblqh::updatePackedList()
void
Dblqh::execREAD_PSUEDO_REQ(Signal* signal){
jamEntry();
TcConnectionrecPtr regTcPtr;
regTcPtr.i = signal->theData[0];
ptrCheckGuard(regTcPtr, ctcConnectrecFileSize, tcConnectionrec);
FragrecordPtr regFragptr;
regFragptr.i = regTcPtr.p->fragmentptr;
ptrCheckGuard(regFragptr, cfragrecFileSize, fragrecord);
signal->theData[0] = regFragptr.p->accFragptr[regTcPtr.p->localFragptr];
EXECUTE_DIRECT(DBACC, GSN_READ_PSUEDO_REQ, signal, 2);
}
/* ************>> */
/* TUPKEYCONF > */
/* ************>> */
void Dblqh::execTUPKEYCONF(Signal* signal)
{
TcConnectionrec *regTcConnectionrec = tcConnectionrec;
Uint32 ttcConnectrecFileSize = ctcConnectrecFileSize;
const TupKeyConf * const tupKeyConf = (TupKeyConf *)signal->getDataPtr();
Uint32 tcIndex = tupKeyConf->userPtr;
jamEntry();
tcConnectptr.i = tcIndex;
ptrCheckGuard(tcConnectptr, ttcConnectrecFileSize, regTcConnectionrec);
if (tcConnectptr.p->seqNoReplica == 0) // Primary replica
tcConnectptr.p->noFiredTriggers = tupKeyConf->noFiredTriggers;
switch (tcConnectptr.p->transactionState) {
case TcConnectionrec::WAIT_TUP:
jam();
tupkeyConfLab(signal);
break;
case TcConnectionrec::COPY_TUPKEY:
jam();
copyTupkeyConfLab(signal);
break;
case TcConnectionrec::SCAN_TUPKEY:
jam();
scanTupkeyConfLab(signal);
break;
case TcConnectionrec::WAIT_TUP_TO_ABORT:
jam();
/* ------------------------------------------------------------------------- */
// Abort was not ready to start until this signal came back. Now we are ready
// to start the abort.
/* ------------------------------------------------------------------------- */
releaseActiveFrag(signal);
abortCommonLab(signal);
break;
case TcConnectionrec::WAIT_ACC_ABORT:
case TcConnectionrec::ABORT_QUEUED:
jam();
/* -------------------------------------------------------------------------- */
/* IGNORE SINCE ABORT OF THIS OPERATION IS ONGOING ALREADY. */
/* -------------------------------------------------------------------------- */
break;
default:
ndbrequire(false);
break;
}//switch
}//Dblqh::execTUPKEYCONF()
/* ************> */
/* TUPKEYREF > */
/* ************> */
void Dblqh::execTUPKEYREF(Signal* signal)
{
const TupKeyRef * const tupKeyRef = (TupKeyRef *)signal->getDataPtr();
jamEntry();
tcConnectptr.i = tupKeyRef->userRef;
terrorCode = tupKeyRef->errorCode;
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
switch (tcConnectptr.p->transactionState) {
case TcConnectionrec::WAIT_TUP:
jam();
releaseActiveFrag(signal);
abortErrorLab(signal);
break;
case TcConnectionrec::COPY_TUPKEY:
ndbrequire(false);
break;
case TcConnectionrec::SCAN_TUPKEY:
jam();
scanTupkeyRefLab(signal);
break;
case TcConnectionrec::WAIT_TUP_TO_ABORT:
jam();
/* ------------------------------------------------------------------------- */
// Abort was not ready to start until this signal came back. Now we are ready
// to start the abort.
/* ------------------------------------------------------------------------- */
releaseActiveFrag(signal);
abortCommonLab(signal);
break;
case TcConnectionrec::WAIT_ACC_ABORT:
case TcConnectionrec::ABORT_QUEUED:
jam();
/* ------------------------------------------------------------------------- */
/* IGNORE SINCE ABORT OF THIS OPERATION IS ONGOING ALREADY. */
/* ------------------------------------------------------------------------- */
break;
default:
ndbrequire(false);
break;
}//switch
}//Dblqh::execTUPKEYREF()
void Dblqh::sendPackedSignalLqh(Signal* signal, HostRecord * ahostptr)
{
Uint32 noOfWords = ahostptr->noOfPackedWordsLqh;
BlockReference hostRef = ahostptr->hostLqhBlockRef;
MEMCOPY_NO_WORDS(&signal->theData[0],
&ahostptr->packedWordsLqh[0],
noOfWords);
sendSignal(hostRef, GSN_PACKED_SIGNAL, signal, noOfWords, JBB);
ahostptr->noOfPackedWordsLqh = 0;
}//Dblqh::sendPackedSignalLqh()
void Dblqh::sendPackedSignalTc(Signal* signal, HostRecord * ahostptr)
{
Uint32 noOfWords = ahostptr->noOfPackedWordsTc;
BlockReference hostRef = ahostptr->hostTcBlockRef;
MEMCOPY_NO_WORDS(&signal->theData[0],
&ahostptr->packedWordsTc[0],
noOfWords);
sendSignal(hostRef, GSN_PACKED_SIGNAL, signal, noOfWords, JBB);
ahostptr->noOfPackedWordsTc = 0;
}//Dblqh::sendPackedSignalTc()
void Dblqh::sendCommitLqh(Signal* signal, BlockReference alqhBlockref)
{
HostRecordPtr Thostptr;
Thostptr.i = refToNode(alqhBlockref);
ptrCheckGuard(Thostptr, chostFileSize, hostRecord);
if (Thostptr.p->noOfPackedWordsLqh > 21) {
jam();
sendPackedSignalLqh(signal, Thostptr.p);
} else {
jam();
updatePackedList(signal, Thostptr.p, Thostptr.i);
}//if
Uint32 pos = Thostptr.p->noOfPackedWordsLqh;
Uint32 ptrAndType = tcConnectptr.p->clientConnectrec | (ZCOMMIT << 28);
Uint32 gci = tcConnectptr.p->gci;
Uint32 transid1 = tcConnectptr.p->transid[0];
Uint32 transid2 = tcConnectptr.p->transid[1];
Thostptr.p->packedWordsLqh[pos] = ptrAndType;
Thostptr.p->packedWordsLqh[pos + 1] = gci;
Thostptr.p->packedWordsLqh[pos + 2] = transid1;
Thostptr.p->packedWordsLqh[pos + 3] = transid2;
Thostptr.p->noOfPackedWordsLqh = pos + 4;
}//Dblqh::sendCommitLqh()
void Dblqh::sendCompleteLqh(Signal* signal, BlockReference alqhBlockref)
{
HostRecordPtr Thostptr;
Thostptr.i = refToNode(alqhBlockref);
ptrCheckGuard(Thostptr, chostFileSize, hostRecord);
if (Thostptr.p->noOfPackedWordsLqh > 22) {
jam();
sendPackedSignalLqh(signal, Thostptr.p);
} else {
jam();
updatePackedList(signal, Thostptr.p, Thostptr.i);
}//if
Uint32 pos = Thostptr.p->noOfPackedWordsLqh;
Uint32 ptrAndType = tcConnectptr.p->clientConnectrec | (ZCOMPLETE << 28);
Uint32 transid1 = tcConnectptr.p->transid[0];
Uint32 transid2 = tcConnectptr.p->transid[1];
Thostptr.p->packedWordsLqh[pos] = ptrAndType;
Thostptr.p->packedWordsLqh[pos + 1] = transid1;
Thostptr.p->packedWordsLqh[pos + 2] = transid2;
Thostptr.p->noOfPackedWordsLqh = pos + 3;
}//Dblqh::sendCompleteLqh()
void Dblqh::sendCommittedTc(Signal* signal, BlockReference atcBlockref)
{
HostRecordPtr Thostptr;
Thostptr.i = refToNode(atcBlockref);
ptrCheckGuard(Thostptr, chostFileSize, hostRecord);
if (Thostptr.p->noOfPackedWordsTc > 22) {
jam();
sendPackedSignalTc(signal, Thostptr.p);
} else {
jam();
updatePackedList(signal, Thostptr.p, Thostptr.i);
}//if
Uint32 pos = Thostptr.p->noOfPackedWordsTc;
Uint32 ptrAndType = tcConnectptr.p->clientConnectrec | (ZCOMMITTED << 28);
Uint32 transid1 = tcConnectptr.p->transid[0];
Uint32 transid2 = tcConnectptr.p->transid[1];
Thostptr.p->packedWordsTc[pos] = ptrAndType;
Thostptr.p->packedWordsTc[pos + 1] = transid1;
Thostptr.p->packedWordsTc[pos + 2] = transid2;
Thostptr.p->noOfPackedWordsTc = pos + 3;
}//Dblqh::sendCommittedTc()
void Dblqh::sendCompletedTc(Signal* signal, BlockReference atcBlockref)
{
HostRecordPtr Thostptr;
Thostptr.i = refToNode(atcBlockref);
ptrCheckGuard(Thostptr, chostFileSize, hostRecord);
if (Thostptr.p->noOfPackedWordsTc > 22) {
jam();
sendPackedSignalTc(signal, Thostptr.p);
} else {
jam();
updatePackedList(signal, Thostptr.p, Thostptr.i);
}//if
Uint32 pos = Thostptr.p->noOfPackedWordsTc;
Uint32 ptrAndType = tcConnectptr.p->clientConnectrec | (ZCOMPLETED << 28);
Uint32 transid1 = tcConnectptr.p->transid[0];
Uint32 transid2 = tcConnectptr.p->transid[1];
Thostptr.p->packedWordsTc[pos] = ptrAndType;
Thostptr.p->packedWordsTc[pos + 1] = transid1;
Thostptr.p->packedWordsTc[pos + 2] = transid2;
Thostptr.p->noOfPackedWordsTc = pos + 3;
}//Dblqh::sendCompletedTc()
void Dblqh::sendLqhkeyconfTc(Signal* signal, BlockReference atcBlockref)
{
LqhKeyConf* lqhKeyConf;
HostRecordPtr Thostptr;
Thostptr.i = refToNode(atcBlockref);
ptrCheckGuard(Thostptr, chostFileSize, hostRecord);
if (refToBlock(atcBlockref) == DBTC) {
jam();
/*******************************************************************
// This signal was intended for DBTC as part of the normal transaction
// execution.
********************************************************************/
if (Thostptr.p->noOfPackedWordsTc > (25 - LqhKeyConf::SignalLength)) {
jam();
sendPackedSignalTc(signal, Thostptr.p);
} else {
jam();
updatePackedList(signal, Thostptr.p, Thostptr.i);
}//if
lqhKeyConf = (LqhKeyConf *)
&Thostptr.p->packedWordsTc[Thostptr.p->noOfPackedWordsTc];
Thostptr.p->noOfPackedWordsTc += LqhKeyConf::SignalLength;
} else {
jam();
/*******************************************************************
// This signal was intended for DBLQH as part of log execution or
// node recovery.
********************************************************************/
if (Thostptr.p->noOfPackedWordsLqh > (25 - LqhKeyConf::SignalLength)) {
jam();
sendPackedSignalLqh(signal, Thostptr.p);
} else {
jam();
updatePackedList(signal, Thostptr.p, Thostptr.i);
}//if
lqhKeyConf = (LqhKeyConf *)
&Thostptr.p->packedWordsLqh[Thostptr.p->noOfPackedWordsLqh];
Thostptr.p->noOfPackedWordsLqh += LqhKeyConf::SignalLength;
}//if
Uint32 ptrAndType = tcConnectptr.i | (ZLQHKEYCONF << 28);
Uint32 tcOprec = tcConnectptr.p->tcOprec;
Uint32 ownRef = cownref;
Uint32 readlenAi = tcConnectptr.p->readlenAi;
Uint32 transid1 = tcConnectptr.p->transid[0];
Uint32 transid2 = tcConnectptr.p->transid[1];
Uint32 noFiredTriggers = tcConnectptr.p->noFiredTriggers;
lqhKeyConf->connectPtr = ptrAndType;
lqhKeyConf->opPtr = tcOprec;
lqhKeyConf->userRef = ownRef;
lqhKeyConf->readLen = readlenAi;
lqhKeyConf->transId1 = transid1;
lqhKeyConf->transId2 = transid2;
lqhKeyConf->noFiredTriggers = noFiredTriggers;
}//Dblqh::sendLqhkeyconfTc()
/* ************************************************************************>>
* KEYINFO: Get tuple request from DBTC. Next step is to contact DBACC to get
* key to tuple if all key/attrinfo has been received, else for more attrinfo
* signals.
* ************************************************************************>> */
void Dblqh::execKEYINFO(Signal* signal)
{
Uint32 tcOprec = signal->theData[0];
Uint32 transid1 = signal->theData[1];
Uint32 transid2 = signal->theData[2];
jamEntry();
if (findTransaction(transid1, transid2, tcOprec) != ZOK) {
jam();
return;
}//if
TcConnectionrec * const regTcPtr = tcConnectptr.p;
TcConnectionrec::TransactionState state = regTcPtr->transactionState;
if (state != TcConnectionrec::WAIT_TUPKEYINFO &&
state != TcConnectionrec::WAIT_SCAN_AI)
{
jam();
/*****************************************************************************/
/* TRANSACTION WAS ABORTED, THIS IS MOST LIKELY A SIGNAL BELONGING TO THE */
/* ABORTED TRANSACTION. THUS IGNORE THE SIGNAL. */
/*****************************************************************************/
return;
}//if
Uint32 errorCode = handleLongTupKey(signal,
(Uint32)regTcPtr->save1,
(Uint32)regTcPtr->primKeyLen,
&signal->theData[3]);
if (errorCode != 0) {
if (errorCode == 1) {
jam();
return;
}//if
jam();
terrorCode = errorCode;
if(state == TcConnectionrec::WAIT_TUPKEYINFO)
abortErrorLab(signal);
else
abort_scan(signal, regTcPtr->tcScanRec, errorCode);
return;
}//if
if(state == TcConnectionrec::WAIT_TUPKEYINFO)
{
FragrecordPtr regFragptr;
regFragptr.i = regTcPtr->fragmentptr;
ptrCheckGuard(regFragptr, cfragrecFileSize, fragrecord);
fragptr = regFragptr;
endgettupkeyLab(signal);
}
return;
}//Dblqh::execKEYINFO()
/* ------------------------------------------------------------------------- */
/* FILL IN KEY DATA INTO DATA BUFFERS. */
/* ------------------------------------------------------------------------- */
Uint32 Dblqh::handleLongTupKey(Signal* signal,
Uint32 keyLength,
Uint32 primKeyLength,
Uint32* dataPtr)
{
TcConnectionrec * const regTcPtr = tcConnectptr.p;
Uint32 dataPos = 0;
while (true) {
keyLength += 4;
if (cfirstfreeDatabuf == RNIL) {
jam();
return ZGET_DATAREC_ERROR;
}//if
seizeTupkeybuf(signal);
Databuf * const regDataPtr = databufptr.p;
Uint32 data0 = dataPtr[dataPos];
Uint32 data1 = dataPtr[dataPos + 1];
Uint32 data2 = dataPtr[dataPos + 2];
Uint32 data3 = dataPtr[dataPos + 3];
regDataPtr->data[0] = data0;
regDataPtr->data[1] = data1;
regDataPtr->data[2] = data2;
regDataPtr->data[3] = data3;
dataPos += 4;
if (keyLength < primKeyLength) {
if (dataPos > 16) {
jam();
/* SAVE STATE AND WAIT FOR KEYINFO */
regTcPtr->save1 = keyLength;
return 1;
}//if
} else {
jam();
return 0;
}//if
}//while
}//Dblqh::handleLongTupKey()
/* ------------------------------------------------------------------------- */
/* ------- HANDLE ATTRINFO SIGNALS ------- */
/* */
/* ------------------------------------------------------------------------- */
/* ************************************************************************>> */
/* ATTRINFO: Continuation of KEYINFO signal (except for scans that do not use*/
/* any KEYINFO). When all key and attribute info is received we contact DBACC*/
/* for index handling. */
/* ************************************************************************>> */
void Dblqh::execATTRINFO(Signal* signal)
{
Uint32 tcOprec = signal->theData[0];
Uint32 transid1 = signal->theData[1];
Uint32 transid2 = signal->theData[2];
jamEntry();
if (findTransaction(transid1,
transid2,
tcOprec) != ZOK) {
jam();
return;
}//if
TcConnectionrec * const regTcPtr = tcConnectptr.p;
Uint32 length = signal->length() - 3;
Uint32 totReclenAi = regTcPtr->totReclenAi;
Uint32 currReclenAi = regTcPtr->currReclenAi + length;
Uint32* dataPtr = &signal->theData[3];
regTcPtr->currReclenAi = currReclenAi;
if (totReclenAi == currReclenAi) {
switch (regTcPtr->transactionState) {
case TcConnectionrec::WAIT_ATTR:
{
Fragrecord *regFragrecord = fragrecord;
Uint32 fragIndex = regTcPtr->fragmentptr;
Uint32 tfragrecFileSize = cfragrecFileSize;
jam();
fragptr.i = fragIndex;
ptrCheckGuard(fragptr, tfragrecFileSize, regFragrecord);
lqhAttrinfoLab(signal, dataPtr, length);
endgettupkeyLab(signal);
return;
break;
}
case TcConnectionrec::WAIT_SCAN_AI:
jam();
scanAttrinfoLab(signal, dataPtr, length);
return;
break;
case TcConnectionrec::WAIT_TUP_TO_ABORT:
case TcConnectionrec::LOG_ABORT_QUEUED:
case TcConnectionrec::ABORT_QUEUED:
case TcConnectionrec::ABORT_STOPPED:
case TcConnectionrec::WAIT_ACC_ABORT:
case TcConnectionrec::WAIT_AI_AFTER_ABORT:
jam();
aiStateErrorCheckLab(signal, dataPtr,length);
return;
break;
default:
jam();
ndbrequire(regTcPtr->abortState != TcConnectionrec::ABORT_IDLE);
break;
}//switch
} else if (currReclenAi < totReclenAi) {
jam();
switch (regTcPtr->transactionState) {
case TcConnectionrec::WAIT_ATTR:
jam();
lqhAttrinfoLab(signal, dataPtr, length);
return;
break;
case TcConnectionrec::WAIT_SCAN_AI:
jam();
scanAttrinfoLab(signal, dataPtr, length);
return;
break;
case TcConnectionrec::WAIT_TUP_TO_ABORT:
case TcConnectionrec::LOG_ABORT_QUEUED:
case TcConnectionrec::ABORT_QUEUED:
case TcConnectionrec::ABORT_STOPPED:
case TcConnectionrec::WAIT_ACC_ABORT:
case TcConnectionrec::WAIT_AI_AFTER_ABORT:
jam();
aiStateErrorCheckLab(signal, dataPtr, length);
return;
break;
default:
jam();
ndbrequire(regTcPtr->abortState != TcConnectionrec::ABORT_IDLE);
break;
}//switch
} else {
switch (regTcPtr->transactionState) {
case TcConnectionrec::WAIT_SCAN_AI:
jam();
scanAttrinfoLab(signal, dataPtr, length);
return;
break;
default:
ndbout_c("%d", regTcPtr->transactionState);
ndbrequire(false);