MySQL数据库

开发平台：

Visual C++

DbdihMain.cpp：源码内容

#ifdef VM_TRACE
ndbout << "System crash due to GCP Stop in state = ";
ndbout << (Uint32) cgcpStatus << endl;
#endif
crashSystemAtGcpStop(signal);
return;
}//if
} else {
jam();
if (cgcpOrderBlocked == 0) {
jam();
cgcpSameCounter++;
if (cgcpSameCounter == 1200) {
jam();
#ifdef VM_TRACE
ndbout << "System crash due to GCP Stop in state = ";
ndbout << (Uint32) cgcpStatus << endl;
#endif
crashSystemAtGcpStop(signal);
return;
}//if
} else {
jam();
cgcpSameCounter = 0;
}//if
}//if
} else {
jam();
cgcpSameCounter = 0;
}//if
} else {
jam();
cgcpSameCounter = 0;
}//if
signal->theData[0] = DihContinueB::ZCHECK_GCP_STOP;
signal->theData[1] = coldGcpStatus;
signal->theData[2] = cgcpStatus;
signal->theData[3] = coldGcpId;
signal->theData[4] = cnewgcp;
signal->theData[5] = cgcpSameCounter;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 100, 6);
coldGcpStatus = cgcpStatus;
coldGcpId = cnewgcp;
return;
}//Dbdih::checkGcpStopLab()
void Dbdih::startGcpLab(Signal* signal, Uint32 aWaitTime)
{
if ((cgcpOrderBlocked == 1) ||
(c_nodeStartMaster.blockGcp == true) ||
(cfirstVerifyQueue != RNIL)) {
/*************************************************************************/
// 1: Global Checkpoint has been stopped by management command
// 2: Global Checkpoint is blocked by node recovery activity
// 3: Previous global checkpoint is not yet completed.
// All this means that global checkpoint cannot start now.
/*************************************************************************/
jam();
cgcpStartCounter++;
signal->theData[0] = DihContinueB::ZSTART_GCP;
signal->theData[1] = aWaitTime > 100 ? (aWaitTime - 100) : 0;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 100, 2);
return;
}//if
if (cstartGcpNow == false && aWaitTime > 100){
/*************************************************************************/
// We still have more than 100 milliseconds before we start the next and
// nobody has ordered immediate start of a global checkpoint.
// During initial start we will use continuos global checkpoints to
// speed it up since we need to complete a global checkpoint after
// inserting a lot of records.
/*************************************************************************/
jam();
cgcpStartCounter++;
signal->theData[0] = DihContinueB::ZSTART_GCP;
signal->theData[1] = (aWaitTime - 100);
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 100, 2);
return;
}//if
cgcpStartCounter = 0;
cstartGcpNow = false;
/***************************************************************************/
// Report the event that a global checkpoint has started.
/***************************************************************************/
signal->theData[0] = EventReport::GlobalCheckpointStarted; //Event type
signal->theData[1] = cnewgcp;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
CRASH_INSERTION(7000);
cnewgcp++;
signal->setTrace(TestOrd::TraceGlobalCheckpoint);
sendLoopMacro(GCP_PREPARE, sendGCP_PREPARE);
cgcpStatus = GCP_PREPARE_SENT;
}//Dbdih::startGcpLab()
void Dbdih::execGCP_PREPARECONF(Signal* signal)
{
jamEntry();
Uint32 senderNodeId = signal->theData[0];
Uint32 gci = signal->theData[1];
ndbrequire(gci == cnewgcp);
receiveLoopMacro(GCP_PREPARE, senderNodeId);
//-------------------------------------------------------------
// We have now received all replies. We are ready to continue
// with committing the global checkpoint.
//-------------------------------------------------------------
gcpcommitreqLab(signal);
}//Dbdih::execGCP_PREPARECONF()
void Dbdih::gcpcommitreqLab(Signal* signal)
{
CRASH_INSERTION(7001);
sendLoopMacro(GCP_COMMIT, sendGCP_COMMIT);
cgcpStatus = GCP_COMMIT_SENT;
return;
}//Dbdih::gcpcommitreqLab()
void Dbdih::execGCP_NODEFINISH(Signal* signal)
{
jamEntry();
const Uint32 senderNodeId = signal->theData[0];
const Uint32 gci = signal->theData[1];
const Uint32 failureNr = signal->theData[2];
if (!isMaster()) {
jam();
ndbrequire(failureNr > cfailurenr);
//-------------------------------------------------------------
// Another node thinks we are master. This could happen when he
// has heard of a node failure which I have not heard of. Ignore
// signal in this case since we will discover it by sending
// MASTER_GCPREQ to the node.
//-------------------------------------------------------------
return;
} else if (cmasterState == MASTER_TAKE_OVER_GCP) {
jam();
//-------------------------------------------------------------
// We are currently taking over as master. We will delay the
// signal until we have completed the take over gcp handling.
//-------------------------------------------------------------
sendSignalWithDelay(reference(), GSN_GCP_NODEFINISH, signal, 20, 3);
return;
} else {
ndbrequire(cmasterState == MASTER_ACTIVE);
}//if
ndbrequire(gci == coldgcp);
receiveLoopMacro(GCP_COMMIT, senderNodeId);
//-------------------------------------------------------------
// We have now received all replies. We are ready to continue
// with saving the global checkpoint to disk.
//-------------------------------------------------------------
CRASH_INSERTION(7002);
gcpsavereqLab(signal);
return;
}//Dbdih::execGCP_NODEFINISH()
void Dbdih::gcpsavereqLab(Signal* signal)
{
sendLoopMacro(GCP_SAVEREQ, sendGCP_SAVEREQ);
cgcpStatus = GCP_NODE_FINISHED;
}//Dbdih::gcpsavereqLab()
void Dbdih::execGCP_SAVECONF(Signal* signal)
{
jamEntry();
const GCPSaveConf * const saveConf = (GCPSaveConf*)&signal->theData[0];
ndbrequire(saveConf->gci == coldgcp);
ndbrequire(saveConf->nodeId == saveConf->dihPtr);
SYSFILE->lastCompletedGCI[saveConf->nodeId] = saveConf->gci;
GCP_SAVEhandling(signal, saveConf->nodeId);
}//Dbdih::execGCP_SAVECONF()
void Dbdih::execGCP_SAVEREF(Signal* signal)
{
jamEntry();
const GCPSaveRef * const saveRef = (GCPSaveRef*)&signal->theData[0];
ndbrequire(saveRef->gci == coldgcp);
ndbrequire(saveRef->nodeId == saveRef->dihPtr);
/**
* Only allow reason not to save
*/
ndbrequire(saveRef->errorCode == GCPSaveRef::NodeShutdownInProgress ||
saveRef->errorCode == GCPSaveRef::FakedSignalDueToNodeFailure ||
saveRef->errorCode == GCPSaveRef::NodeRestartInProgress);
GCP_SAVEhandling(signal, saveRef->nodeId);
}//Dbdih::execGCP_SAVEREF()
void Dbdih::GCP_SAVEhandling(Signal* signal, Uint32 nodeId)
{
receiveLoopMacro(GCP_SAVEREQ, nodeId);
/*-------------------------------------------------------------------------*/
// All nodes have replied. We are ready to update the system file.
/*-------------------------------------------------------------------------*/
cgcpStatus = GCP_SAVE_LQH_FINISHED;
CRASH_INSERTION(7003);
checkToCopy();
/**------------------------------------------------------------------------
* SET NEW RECOVERABLE GCI. ALSO RESET RESTART COUNTER TO ZERO.
* THIS INDICATES THAT THE SYSTEM HAS BEEN RECOVERED AND SURVIVED AT
* LEAST ONE GLOBAL CHECKPOINT PERIOD. WE WILL USE THIS PARAMETER TO
* SET BACK THE RESTART GCI IF WE ENCOUNTER MORE THAN ONE UNSUCCESSFUL
* RESTART.
*------------------------------------------------------------------------*/
SYSFILE->newestRestorableGCI = coldgcp;
if(Sysfile::getInitialStartOngoing(SYSFILE->systemRestartBits) &&
getNodeState().startLevel == NodeState::SL_STARTED){
jam();
#if 0
ndbout_c("Dbdih: Clearing initial start ongoing");
#endif
Sysfile::clearInitialStartOngoing(SYSFILE->systemRestartBits);
}
copyGciLab(signal, CopyGCIReq::GLOBAL_CHECKPOINT);
}//Dbdih::GCP_SAVEhandling()
/*
3.11 G L O B A L C H E C K P O I N T (N O T - M A S T E R)
*************************************************************
*/
void Dbdih::execGCP_PREPARE(Signal* signal)
{
jamEntry();
CRASH_INSERTION(7005);
Uint32 masterNodeId = signal->theData[0];
Uint32 gci = signal->theData[1];
BlockReference retRef = calcDihBlockRef(masterNodeId);
ndbrequire (cmasterdihref == retRef);
ndbrequire (cgcpParticipantState == GCP_PARTICIPANT_READY);
ndbrequire (gci == (currentgcp + 1));
cgckptflag = true;
cgcpParticipantState = GCP_PARTICIPANT_PREPARE_RECEIVED;
cnewgcp = gci;
signal->theData[0] = cownNodeId;
signal->theData[1] = gci;
sendSignal(retRef, GSN_GCP_PREPARECONF, signal, 2, JBA);
return;
}//Dbdih::execGCP_PREPARE()
void Dbdih::execGCP_COMMIT(Signal* signal)
{
jamEntry();
CRASH_INSERTION(7006);
Uint32 masterNodeId = signal->theData[0];
Uint32 gci = signal->theData[1];
ndbrequire(gci == (currentgcp + 1));
ndbrequire(masterNodeId = cmasterNodeId);
ndbrequire(cgcpParticipantState == GCP_PARTICIPANT_PREPARE_RECEIVED);
coldgcp = currentgcp;
currentgcp = cnewgcp;
cgckptflag = false;
emptyverificbuffer(signal, true);
cgcpParticipantState = GCP_PARTICIPANT_COMMIT_RECEIVED;
signal->theData[1] = coldgcp;
sendSignal(clocaltcblockref, GSN_GCP_NOMORETRANS, signal, 2, JBB);
return;
}//Dbdih::execGCP_COMMIT()
void Dbdih::execGCP_TCFINISHED(Signal* signal)
{
jamEntry();
CRASH_INSERTION(7007);
Uint32 gci = signal->theData[1];
ndbrequire(gci == coldgcp);
cgcpParticipantState = GCP_PARTICIPANT_TC_FINISHED;
signal->theData[0] = cownNodeId;
signal->theData[1] = coldgcp;
signal->theData[2] = cfailurenr;
sendSignal(cmasterdihref, GSN_GCP_NODEFINISH, signal, 3, JBB);
}//Dbdih::execGCP_TCFINISHED()
/*****************************************************************************/
//****** RECEIVING TAMPER REQUEST FROM NDBAPI ******
/*****************************************************************************/
void Dbdih::execDIHNDBTAMPER(Signal* signal)
{
jamEntry();
Uint32 tcgcpblocked = signal->theData[0];
/* ACTION TO BE TAKEN BY DIH */
Uint32 tuserpointer = signal->theData[1];
BlockReference tuserblockref = signal->theData[2];
switch (tcgcpblocked) {
case 1:
jam();
if (isMaster()) {
jam();
cgcpOrderBlocked = 1;
} else {
jam();
/* TRANSFER THE REQUEST */
/* TO MASTER*/
signal->theData[0] = tcgcpblocked;
signal->theData[1] = tuserpointer;
signal->theData[2] = tuserblockref;
sendSignal(cmasterdihref, GSN_DIHNDBTAMPER, signal, 3, JBB);
}//if
break;
case 2:
jam();
if (isMaster()) {
jam();
cgcpOrderBlocked = 0;
} else {
jam();
/* TRANSFER THE REQUEST */
/* TO MASTER*/
signal->theData[0] = tcgcpblocked;
signal->theData[1] = tuserpointer;
signal->theData[2] = tuserblockref;
sendSignal(cmasterdihref, GSN_DIHNDBTAMPER, signal, 3, JBB);
}//if
break;
case 3:
ndbrequire(false);
return;
break;
case 4:
jam();
signal->theData[0] = tuserpointer;
signal->theData[1] = crestartGci;
sendSignal(tuserblockref, GSN_DIHNDBTAMPER, signal, 2, JBB);
break;
#ifdef ERROR_INSERT
case 5:
jam();
/*----------------------------------------------------------------------*/
// Insert errors.
/*----------------------------------------------------------------------*/
if (tuserpointer < 1000) {
/*--------------------------------------------------------------------*/
// Insert errors into QMGR.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = QMGR_REF;
} else if (tuserpointer < 2000) {
/*--------------------------------------------------------------------*/
// Insert errors into NDBCNTR.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = NDBCNTR_REF;
} else if (tuserpointer < 3000) {
/*--------------------------------------------------------------------*/
// Insert errors into NDBFS.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = NDBFS_REF;
} else if (tuserpointer < 4000) {
/*--------------------------------------------------------------------*/
// Insert errors into DBACC.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = DBACC_REF;
} else if (tuserpointer < 5000) {
/*--------------------------------------------------------------------*/
// Insert errors into DBTUP.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = DBTUP_REF;
} else if (tuserpointer < 6000) {
/*---------------------------------------------------------------------*/
// Insert errors into DBLQH.
/*---------------------------------------------------------------------*/
jam();
tuserblockref = DBLQH_REF;
} else if (tuserpointer < 7000) {
/*---------------------------------------------------------------------*/
// Insert errors into DBDICT.
/*---------------------------------------------------------------------*/
jam();
tuserblockref = DBDICT_REF;
} else if (tuserpointer < 8000) {
/*---------------------------------------------------------------------*/
// Insert errors into DBDIH.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = DBDIH_REF;
} else if (tuserpointer < 9000) {
/*--------------------------------------------------------------------*/
// Insert errors into DBTC.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = DBTC_REF;
} else if (tuserpointer < 10000) {
/*--------------------------------------------------------------------*/
// Insert errors into CMVMI.
/*--------------------------------------------------------------------*/
jam();
tuserblockref = CMVMI_REF;
} else if (tuserpointer < 11000) {
jam();
tuserblockref = BACKUP_REF;
} else if (tuserpointer < 12000) {
// DBUTIL_REF ?
jam();
} else if (tuserpointer < 13000) {
jam();
tuserblockref = DBTUX_REF;
} else if (tuserpointer < 14000) {
jam();
tuserblockref = SUMA_REF;
} else if (tuserpointer < 15000) {
jam();
tuserblockref = DBDICT_REF;
} else if (tuserpointer < 30000) {
/*--------------------------------------------------------------------*/
// Ignore errors in the 20000-range.
/*--------------------------------------------------------------------*/
jam();
return;
} else if (tuserpointer < 40000) {
jam();
/*--------------------------------------------------------------------*/
// Redirect errors to master DIH in the 30000-range.
/*--------------------------------------------------------------------*/
tuserblockref = cmasterdihref;
tuserpointer -= 30000;
signal->theData[0] = 5;
signal->theData[1] = tuserpointer;
signal->theData[2] = tuserblockref;
sendSignal(tuserblockref, GSN_DIHNDBTAMPER, signal, 3, JBB);
return;
} else if (tuserpointer < 50000) {
NodeRecordPtr localNodeptr;
Uint32 Tfound = 0;
jam();
/*--------------------------------------------------------------------*/
// Redirect errors to non-master DIH in the 40000-range.
/*--------------------------------------------------------------------*/
tuserpointer -= 40000;
for (localNodeptr.i = 1;
localNodeptr.i < MAX_NDB_NODES;
localNodeptr.i++) {
jam();
ptrAss(localNodeptr, nodeRecord);
if ((localNodeptr.p->nodeStatus == NodeRecord::ALIVE) &&
(localNodeptr.i != cmasterNodeId)) {
jam();
tuserblockref = calcDihBlockRef(localNodeptr.i);
Tfound = 1;
break;
}//if
}//for
if (Tfound == 0) {
jam();
/*-------------------------------------------------------------------*/
// Ignore since no non-master node existed.
/*-------------------------------------------------------------------*/
return;
}//if
signal->theData[0] = 5;
signal->theData[1] = tuserpointer;
signal->theData[2] = tuserblockref;
sendSignal(tuserblockref, GSN_DIHNDBTAMPER, signal, 3, JBB);
return;
} else {
jam();
return;
}//if
signal->theData[0] = tuserpointer;
if (tuserpointer != 0) {
sendSignal(tuserblockref, GSN_NDB_TAMPER, signal, 1, JBB);
} else {
sendSignal(QMGR_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(NDBCNTR_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(NDBFS_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(DBACC_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(DBTUP_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(DBLQH_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(DBDICT_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(DBDIH_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(DBTC_REF, GSN_NDB_TAMPER, signal, 1, JBB);
sendSignal(CMVMI_REF, GSN_NDB_TAMPER, signal, 1, JBB);
}//if
break;
#endif
default:
ndbrequire(false);
break;
}//switch
return;
}//Dbdih::execDIHNDBTAMPER()
/*****************************************************************************/
/* ********** FILE HANDLING MODULE *************/
/*****************************************************************************/
void Dbdih::copyGciLab(Signal* signal, CopyGCIReq::CopyReason reason)
{
if(c_copyGCIMaster.m_copyReason != CopyGCIReq::IDLE){
/**
* There can currently only be one waiting
*/
ndbrequire(c_copyGCIMaster.m_waiting == CopyGCIReq::IDLE);
c_copyGCIMaster.m_waiting = reason;
return;
}
c_copyGCIMaster.m_copyReason = reason;
sendLoopMacro(COPY_GCIREQ, sendCOPY_GCIREQ);
}//Dbdih::copyGciLab()
/* ------------------------------------------------------------------------- */
/* COPY_GCICONF RESPONSE TO COPY_GCIREQ */
/* ------------------------------------------------------------------------- */
void Dbdih::execCOPY_GCICONF(Signal* signal)
{
jamEntry();
NodeRecordPtr senderNodePtr;
senderNodePtr.i = signal->theData[0];
receiveLoopMacro(COPY_GCIREQ, senderNodePtr.i);
CopyGCIReq::CopyReason waiting = c_copyGCIMaster.m_waiting;
CopyGCIReq::CopyReason current = c_copyGCIMaster.m_copyReason;
c_copyGCIMaster.m_copyReason = CopyGCIReq::IDLE;
c_copyGCIMaster.m_waiting = CopyGCIReq::IDLE;
bool ok = false;
switch(current){
case CopyGCIReq::RESTART:{
ok = true;
jam();
DictStartReq * req = (DictStartReq*)&signal->theData[0];
req->restartGci = SYSFILE->newestRestorableGCI;
req->senderRef = reference();
sendSignal(cdictblockref, GSN_DICTSTARTREQ,
signal, DictStartReq::SignalLength, JBB);
break;
}
case CopyGCIReq::LOCAL_CHECKPOINT:{
ok = true;
jam();
startLcpRoundLab(signal);
break;
}
case CopyGCIReq::GLOBAL_CHECKPOINT:
ok = true;
jam();
checkToCopyCompleted(signal);
/************************************************************************/
// Report the event that a global checkpoint has completed.
/************************************************************************/
signal->setTrace(0);
signal->theData[0] = EventReport::GlobalCheckpointCompleted; //Event type
signal->theData[1] = coldgcp;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
CRASH_INSERTION(7004);
emptyWaitGCPMasterQueue(signal);
cgcpStatus = GCP_READY;
signal->theData[0] = DihContinueB::ZSTART_GCP;
signal->theData[1] = cgcpDelay;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 100, 2);
if (c_nodeStartMaster.blockGcp == true) {
jam();
/* ------------------------------------------------------------------ */
/* A NEW NODE WANTS IN AND WE MUST ALLOW IT TO COME IN NOW SINCE THE */
/* GCP IS COMPLETED. */
/* ------------------------------------------------------------------ */
gcpBlockedLab(signal);
}//if
break;
case CopyGCIReq::INITIAL_START_COMPLETED:
ok = true;
jam();
initialStartCompletedLab(signal);
break;
case CopyGCIReq::IDLE:
ok = false;
jam();
}
ndbrequire(ok);
/**
* Pop queue
*/
if(waiting != CopyGCIReq::IDLE){
c_copyGCIMaster.m_copyReason = waiting;
signal->theData[0] = DihContinueB::ZCOPY_GCI;
signal->theData[1] = waiting;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
}
}//Dbdih::execCOPY_GCICONF()
void Dbdih::invalidateLcpInfoAfterSr()
{
NodeRecordPtr nodePtr;
SYSFILE->latestLCP_ID--;
Sysfile::clearLCPOngoing(SYSFILE->systemRestartBits);
for (nodePtr.i = 1; nodePtr.i < MAX_NDB_NODES; nodePtr.i++) {
jam();
ptrAss(nodePtr, nodeRecord);
if (!NdbNodeBitmask::get(SYSFILE->lcpActive, nodePtr.i)){
jam();
/* ------------------------------------------------------------------- */
// The node was not active in the local checkpoint.
// To avoid that we step the active status too fast to not
// active we step back one step from Sysfile::NS_ActiveMissed_x.
/* ------------------------------------------------------------------- */
switch (nodePtr.p->activeStatus) {
case Sysfile::NS_Active:
/* ----------------------------------------------------------------- */
// When not active in ongoing LCP and still active is a contradiction.
/* ----------------------------------------------------------------- */
ndbrequire(false);
case Sysfile::NS_ActiveMissed_1:
jam();
nodePtr.p->activeStatus = Sysfile::NS_Active;
break;
case Sysfile::NS_ActiveMissed_2:
jam();
nodePtr.p->activeStatus = Sysfile::NS_ActiveMissed_1;
break;
default:
jam();
break;
}//switch
}//if
}//for
setNodeRestartInfoBits();
}//Dbdih::invalidateLcpInfoAfterSr()
/* ------------------------------------------------------------------------- */
/* THE NEXT STEP IS TO WRITE THE FILE. */
/* ------------------------------------------------------------------------- */
void Dbdih::openingCopyGciSkipInitLab(Signal* signal, FileRecordPtr filePtr)
{
writeRestorableGci(signal, filePtr);
filePtr.p->reqStatus = FileRecord::WRITING_COPY_GCI;
return;
}//Dbdih::openingCopyGciSkipInitLab()
void Dbdih::writingCopyGciLab(Signal* signal, FileRecordPtr filePtr)
{
/* ----------------------------------------------------------------------- */
/* WE HAVE NOW WRITTEN THIS FILE. WRITE ALSO NEXT FILE IF THIS IS NOT */
/* ALREADY THE LAST. */
/* ----------------------------------------------------------------------- */
filePtr.p->reqStatus = FileRecord::IDLE;
if (filePtr.i == crestartInfoFile[0]) {
jam();
filePtr.i = crestartInfoFile[1];
ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
if (filePtr.p->fileStatus == FileRecord::OPEN) {
jam();
openingCopyGciSkipInitLab(signal, filePtr);
return;
}//if
openFileRw(signal, filePtr);
filePtr.p->reqStatus = FileRecord::OPENING_COPY_GCI;
return;
}//if
/* ----------------------------------------------------------------------- */
/* WE HAVE COMPLETED WRITING BOTH FILES SUCCESSFULLY. NOW REPORT OUR */
/* SUCCESS TO THE MASTER DIH. BUT FIRST WE NEED TO RESET A NUMBER OF */
/* VARIABLES USED BY THE LOCAL CHECKPOINT PROCESS (ONLY IF TRIGGERED */
/* BY LOCAL CHECKPOINT PROCESS. */
/* ----------------------------------------------------------------------- */
CopyGCIReq::CopyReason reason = c_copyGCISlave.m_copyReason;
if (reason == CopyGCIReq::GLOBAL_CHECKPOINT) {
jam();
cgcpParticipantState = GCP_PARTICIPANT_READY;
SubGcpCompleteRep * const rep = (SubGcpCompleteRep*)signal->getDataPtr();
rep->gci = coldgcp;
rep->senderData = 0;
sendSignal(SUMA_REF, GSN_SUB_GCP_COMPLETE_REP, signal,
SubGcpCompleteRep::SignalLength, JBB);
}
jam();
c_copyGCISlave.m_copyReason = CopyGCIReq::IDLE;
if(c_copyGCISlave.m_senderRef == cmasterdihref){
jam();
/**
* Only if same master
*/
signal->theData[0] = c_copyGCISlave.m_senderData;
sendSignal(c_copyGCISlave.m_senderRef, GSN_COPY_GCICONF, signal, 1, JBB);
}
return;
}//Dbdih::writingCopyGciLab()
void Dbdih::execSTART_LCP_REQ(Signal* signal){
StartLcpReq * req = (StartLcpReq*)signal->getDataPtr();
CRASH_INSERTION2(7021, isMaster());
CRASH_INSERTION2(7022, !isMaster());
ndbrequire(c_lcpState.m_masterLcpDihRef = req->senderRef);
c_lcpState.m_participatingDIH = req->participatingDIH;
c_lcpState.m_participatingLQH = req->participatingLQH;
c_lcpState.m_LCP_COMPLETE_REP_Counter_LQH = req->participatingLQH;
if(isMaster()){
jam();
ndbrequire(isActiveMaster());
c_lcpState.m_LCP_COMPLETE_REP_Counter_DIH = req->participatingDIH;
} else {
c_lcpState.m_LCP_COMPLETE_REP_Counter_DIH.clearWaitingFor();
}
c_lcpState.m_LCP_COMPLETE_REP_From_Master_Received = false;
c_lcpState.setLcpStatus(LCP_INIT_TABLES, __LINE__);
signal->theData[0] = DihContinueB::ZINIT_LCP;
signal->theData[1] = c_lcpState.m_masterLcpDihRef;
signal->theData[2] = 0;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
}
void Dbdih::initLcpLab(Signal* signal, Uint32 senderRef, Uint32 tableId)
{
TabRecordPtr tabPtr;
tabPtr.i = tableId;
if(c_lcpState.m_masterLcpDihRef != senderRef){
jam();
/**
* This is LCP master takeover
*/
#ifdef VM_TRACE
ndbout_c("initLcpLab aborted due to LCP master takeover - 1");
#endif
c_lcpState.setLcpStatus(LCP_STATUS_IDLE, __LINE__);
sendMASTER_LCPCONF(signal);
return;
}
if(c_lcpState.m_masterLcpDihRef != cmasterdihref){
jam();
/**
* Master take over but has not yet received MASTER_LCPREQ
*/
#ifdef VM_TRACE
ndbout_c("initLcpLab aborted due to LCP master takeover - 2");
#endif
return;
}
//const Uint32 lcpId = SYSFILE->latestLCP_ID;
for(; tabPtr.i < ctabFileSize; tabPtr.i++){
ptrAss(tabPtr, tabRecord);
if (tabPtr.p->tabStatus != TabRecord::TS_ACTIVE) {
jam();
tabPtr.p->tabLcpStatus = TabRecord::TLS_COMPLETED;
continue;
}
if (tabPtr.p->storedTable == 0) {
/**
* Temporary table
*/
jam();
tabPtr.p->tabLcpStatus = TabRecord::TLS_COMPLETED;
continue;
}
if (tabPtr.p->tabCopyStatus != TabRecord::CS_IDLE) {
/* ----------------------------------------------------------------- */
// We protect the updates of table data structures by this variable.
/* ----------------------------------------------------------------- */
jam();
signal->theData[0] = DihContinueB::ZINIT_LCP;
signal->theData[1] = senderRef;
signal->theData[2] = tabPtr.i;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 20, 3);
return;
}//if
/**
* Found a table
*/
tabPtr.p->tabLcpStatus = TabRecord::TLS_ACTIVE;
/**
* For each fragment
*/
for (Uint32 fragId = 0; fragId < tabPtr.p->totalfragments; fragId++) {
jam();
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
/**
* For each of replica record
*/
Uint32 replicaCount = 0;
ReplicaRecordPtr replicaPtr;
for(replicaPtr.i = fragPtr.p->storedReplicas; replicaPtr.i != RNIL;
replicaPtr.i = replicaPtr.p->nextReplica) {
jam();
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
Uint32 nodeId = replicaPtr.p->procNode;
if(c_lcpState.m_participatingLQH.get(nodeId)){
jam();
replicaCount++;
replicaPtr.p->lcpOngoingFlag = true;
}
}
fragPtr.p->noLcpReplicas = replicaCount;
}//for
signal->theData[0] = DihContinueB::ZINIT_LCP;
signal->theData[1] = senderRef;
signal->theData[2] = tabPtr.i + 1;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
}
/**
* No more tables
*/
jam();
if (c_lcpState.m_masterLcpDihRef != reference()){
jam();
ndbrequire(!isMaster());
c_lcpState.setLcpStatus(LCP_STATUS_ACTIVE, __LINE__);
} else {
jam();
ndbrequire(isMaster());
}
CRASH_INSERTION2(7023, isMaster());
CRASH_INSERTION2(7024, !isMaster());
jam();
StartLcpConf * conf = (StartLcpConf*)signal->getDataPtrSend();
conf->senderRef = reference();
sendSignal(c_lcpState.m_masterLcpDihRef, GSN_START_LCP_CONF, signal,
StartLcpConf::SignalLength, JBB);
return;
}//Dbdih::initLcpLab()
/* ------------------------------------------------------------------------- */
/* ERROR HANDLING FOR COPY RESTORABLE GCI FILE. */
/* ------------------------------------------------------------------------- */
void Dbdih::openingCopyGciErrorLab(Signal* signal, FileRecordPtr filePtr)
{
createFileRw(signal, filePtr);
/* ------------------------------------------------------------------------- */
/* ERROR IN OPENING FILE. WE WILL TRY BY CREATING FILE INSTEAD. */
/* ------------------------------------------------------------------------- */
filePtr.p->reqStatus = FileRecord::CREATING_COPY_GCI;
return;
}//Dbdih::openingCopyGciErrorLab()
/* ------------------------------------------------------------------------- */
/* ENTER DICTSTARTCONF WITH */
/* TBLOCKREF */
/* ------------------------------------------------------------------------- */
void Dbdih::dictStartConfLab(Signal* signal)
{
/* ----------------------------------------------------------------------- */
/* WE HAVE NOW RECEIVED ALL THE TABLES TO RESTART. */
/* ----------------------------------------------------------------------- */
signal->theData[0] = DihContinueB::ZSTART_FRAGMENT;
signal->theData[1] = 0; /* START WITH TABLE 0 */
signal->theData[2] = 0; /* AND FRAGMENT 0 */
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
}//Dbdih::dictStartConfLab()
void Dbdih::openingTableLab(Signal* signal, FileRecordPtr filePtr)
{
/* ---------------------------------------------------------------------- */
/* SUCCESSFULLY OPENED A FILE. READ THE FIRST PAGE OF THIS FILE. */
/* ---------------------------------------------------------------------- */
TabRecordPtr tabPtr;
PageRecordPtr pagePtr;
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
tabPtr.p->noPages = 1;
allocpage(pagePtr);
tabPtr.p->pageRef[0] = pagePtr.i;
readTabfile(signal, tabPtr.p, filePtr);
filePtr.p->reqStatus = FileRecord::READING_TABLE;
return;
}//Dbdih::openingTableLab()
void Dbdih::openingTableErrorLab(Signal* signal, FileRecordPtr filePtr)
{
TabRecordPtr tabPtr;
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
/* ---------------------------------------------------------------------- */
/* WE FAILED IN OPENING A FILE. IF THE FIRST FILE THEN TRY WITH THE */
/* DUPLICATE FILE, OTHERWISE WE REPORT AN ERROR IN THE SYSTEM RESTART. */
/* ---------------------------------------------------------------------- */
ndbrequire(filePtr.i == tabPtr.p->tabFile[0]);
filePtr.i = tabPtr.p->tabFile[1];
ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
openFileRw(signal, filePtr);
filePtr.p->reqStatus = FileRecord::OPENING_TABLE;
}//Dbdih::openingTableErrorLab()
void Dbdih::readingTableLab(Signal* signal, FileRecordPtr filePtr)
{
TabRecordPtr tabPtr;
PageRecordPtr pagePtr;
/* ---------------------------------------------------------------------- */
/* WE HAVE SUCCESSFULLY READ A NUMBER OF PAGES IN THE TABLE FILE. IF */
/* MORE PAGES EXIST IN THE FILE THEN READ ALL PAGES IN THE FILE. */
/* ---------------------------------------------------------------------- */
filePtr.p->reqStatus = FileRecord::IDLE;
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
pagePtr.i = tabPtr.p->pageRef[0];
ptrCheckGuard(pagePtr, cpageFileSize, pageRecord);
Uint32 noOfStoredPages = pagePtr.p->word[33];
if (tabPtr.p->noPages < noOfStoredPages) {
jam();
ndbrequire(noOfStoredPages <= 8);
for (Uint32 i = tabPtr.p->noPages; i < noOfStoredPages; i++) {
jam();
allocpage(pagePtr);
tabPtr.p->pageRef[i] = pagePtr.i;
}//for
tabPtr.p->noPages = noOfStoredPages;
readTabfile(signal, tabPtr.p, filePtr);
filePtr.p->reqStatus = FileRecord::READING_TABLE;
} else {
ndbrequire(tabPtr.p->noPages == pagePtr.p->word[33]);
ndbrequire(tabPtr.p->tabCopyStatus == TabRecord::CS_IDLE);
jam();
/* --------------------------------------------------------------------- */
/* WE HAVE READ ALL PAGES. NOW READ FROM PAGES INTO TABLE AND FRAGMENT */
/* DATA STRUCTURES. */
/* --------------------------------------------------------------------- */
tabPtr.p->tabCopyStatus = TabRecord::CS_SR_PHASE1_READ_PAGES;
signal->theData[0] = DihContinueB::ZREAD_PAGES_INTO_TABLE;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
}//if
return;
}//Dbdih::readingTableLab()
void Dbdih::readTableFromPagesLab(Signal* signal, TabRecordPtr tabPtr)
{
FileRecordPtr filePtr;
filePtr.i = tabPtr.p->tabFile[0];
ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
/* ---------------------------------------------------------------------- */
/* WE HAVE NOW COPIED TO OUR NODE. WE HAVE NOW COMPLETED RESTORING */
/* THIS TABLE. CONTINUE WITH THE NEXT TABLE. */
/* WE ALSO NEED TO CLOSE THE TABLE FILE. */
/* ---------------------------------------------------------------------- */
if (filePtr.p->fileStatus != FileRecord::OPEN) {
jam();
filePtr.i = tabPtr.p->tabFile[1];
ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
}//if
closeFile(signal, filePtr);
filePtr.p->reqStatus = FileRecord::CLOSING_TABLE_SR;
return;
}//Dbdih::readTableFromPagesLab()
void Dbdih::closingTableSrLab(Signal* signal, FileRecordPtr filePtr)
{
/**
* Update table/fragment info
*/
TabRecordPtr tabPtr;
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
resetReplicaSr(tabPtr);
signal->theData[0] = DihContinueB::ZCOPY_TABLE;
signal->theData[1] = filePtr.p->tabRef;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
}//Dbdih::closingTableSrLab()
void
Dbdih::resetReplicaSr(TabRecordPtr tabPtr){
const Uint32 newestRestorableGCI = SYSFILE->newestRestorableGCI;
for(Uint32 i = 0; i<tabPtr.p->totalfragments; i++){
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, i, fragPtr);
/**
* 1) Start by moving all replicas into oldStoredReplicas
*/
prepareReplicas(fragPtr);
/**
* 2) Move all "alive" replicas into storedReplicas
* + update noCrashedReplicas...
*/
ReplicaRecordPtr replicaPtr;
replicaPtr.i = fragPtr.p->oldStoredReplicas;
while (replicaPtr.i != RNIL) {
jam();
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
const Uint32 nextReplicaPtrI = replicaPtr.p->nextReplica;
NodeRecordPtr nodePtr;
nodePtr.i = replicaPtr.p->procNode;
ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
const Uint32 noCrashedReplicas = replicaPtr.p->noCrashedReplicas;
if (nodePtr.p->nodeStatus == NodeRecord::ALIVE) {
jam();
switch (nodePtr.p->activeStatus) {
case Sysfile::NS_Active:
case Sysfile::NS_ActiveMissed_1:
case Sysfile::NS_ActiveMissed_2:{
jam();
/* --------------------------------------------------------------- */
/* THE NODE IS ALIVE AND KICKING AND ACTIVE, LET'S USE IT. */
/* --------------------------------------------------------------- */
arrGuard(noCrashedReplicas, 8);
Uint32 lastGci = replicaPtr.p->replicaLastGci[noCrashedReplicas];
if(lastGci >= newestRestorableGCI){
jam();
/** -------------------------------------------------------------
* THE REPLICA WAS ALIVE AT THE SYSTEM FAILURE. WE WILL SET THE
* LAST REPLICA GCI TO MINUS ONE SINCE IT HASN'T FAILED YET IN THE
* NEW SYSTEM.
*-------------------------------------------------------------- */
replicaPtr.p->replicaLastGci[noCrashedReplicas] = (Uint32)-1;
} else {
jam();
/*--------------------------------------------------------------
* SINCE IT WAS NOT ALIVE AT THE TIME OF THE SYSTEM CRASH THIS IS
* A COMPLETELY NEW REPLICA. WE WILL SET THE CREATE GCI TO BE THE
* NEXT GCI TO BE EXECUTED.
*--------_----------------------------------------------------- */
const Uint32 nextCrashed = noCrashedReplicas + 1;
replicaPtr.p->noCrashedReplicas = nextCrashed;
arrGuard(nextCrashed, 8);
replicaPtr.p->createGci[nextCrashed] = newestRestorableGCI + 1;
ndbrequire(newestRestorableGCI + 1 != 0xF1F1F1F1);
replicaPtr.p->replicaLastGci[nextCrashed] = (Uint32)-1;
}//if
resetReplicaLcp(replicaPtr.p, newestRestorableGCI);
/* -----------------------------------------------------------------
* LINK THE REPLICA INTO THE STORED REPLICA LIST. WE WILL USE THIS
* NODE AS A STORED REPLICA.
* WE MUST FIRST LINK IT OUT OF THE LIST OF OLD STORED REPLICAS.
* --------------------------------------------------------------- */
removeOldStoredReplica(fragPtr, replicaPtr);
linkStoredReplica(fragPtr, replicaPtr);
}
default:
jam();
/*empty*/;
break;
}
}
replicaPtr.i = nextReplicaPtrI;
}//while
}
}
void
Dbdih::resetReplicaLcp(ReplicaRecord * replicaP, Uint32 stopGci){
Uint32 lcpNo = replicaP->nextLcp;
const Uint32 startLcpNo = lcpNo;
do {
lcpNo = prevLcpNo(lcpNo);
ndbrequire(lcpNo < MAX_LCP_STORED);
if (replicaP->lcpStatus[lcpNo] == ZVALID) {
if (replicaP->maxGciStarted[lcpNo] < stopGci) {
jam();
/* ----------------------------------------------------------------- */
/* WE HAVE FOUND A USEFUL LOCAL CHECKPOINT THAT CAN BE USED FOR */
/* RESTARTING THIS FRAGMENT REPLICA. */
/* ----------------------------------------------------------------- */
return ;
}//if
}//if
/**
* WE COULD NOT USE THIS LOCAL CHECKPOINT. IT WAS TOO
* RECENT OR SIMPLY NOT A VALID CHECKPOINT.
* WE SHOULD THUS REMOVE THIS LOCAL CHECKPOINT SINCE IT WILL NEVER
* AGAIN BE USED. SET LCP_STATUS TO INVALID.
*/
replicaP->nextLcp = lcpNo;
replicaP->lcpId[lcpNo] = 0;
replicaP->lcpStatus[lcpNo] = ZINVALID;
} while (lcpNo != startLcpNo);
replicaP->nextLcp = 0;
}
void Dbdih::readingTableErrorLab(Signal* signal, FileRecordPtr filePtr)
{
TabRecordPtr tabPtr;
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
/* ---------------------------------------------------------------------- */
/* READING THIS FILE FAILED. CLOSE IT AFTER RELEASING ALL PAGES. */
/* ---------------------------------------------------------------------- */
ndbrequire(tabPtr.p->noPages <= 8);
for (Uint32 i = 0; i < tabPtr.p->noPages; i++) {
jam();
releasePage(tabPtr.p->pageRef[i]);
}//for
closeFile(signal, filePtr);
filePtr.p->reqStatus = FileRecord::CLOSING_TABLE_CRASH;
return;
}//Dbdih::readingTableErrorLab()
void Dbdih::closingTableCrashLab(Signal* signal, FileRecordPtr filePtr)
{
TabRecordPtr tabPtr;
/* ---------------------------------------------------------------------- */
/* WE HAVE NOW CLOSED A FILE WHICH WE HAD A READ ERROR WITH. PROCEED */
/* WITH NEXT FILE IF NOT THE LAST OTHERWISE REPORT ERROR. */
/* ---------------------------------------------------------------------- */
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
ndbrequire(filePtr.i == tabPtr.p->tabFile[0]);
filePtr.i = tabPtr.p->tabFile[1];
ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
openFileRw(signal, filePtr);
filePtr.p->reqStatus = FileRecord::OPENING_TABLE;
}//Dbdih::closingTableCrashLab()
/*****************************************************************************/
/* ********** COPY TABLE MODULE *************/
/*****************************************************************************/
void Dbdih::execCOPY_TABREQ(Signal* signal)
{
CRASH_INSERTION(7172);
TabRecordPtr tabPtr;
PageRecordPtr pagePtr;
jamEntry();
BlockReference ref = signal->theData[0];
Uint32 reqinfo = signal->theData[1];
tabPtr.i = signal->theData[2];
Uint32 schemaVersion = signal->theData[3];
Uint32 noOfWords = signal->theData[4];
ndbrequire(ref == cmasterdihref);
ndbrequire(!isMaster());
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
if (reqinfo == 1) {
jam();
tabPtr.p->schemaVersion = schemaVersion;
initTableFile(tabPtr);
}//if
ndbrequire(tabPtr.p->noPages < 8);
if (tabPtr.p->noOfWords == 0) {
jam();
allocpage(pagePtr);
tabPtr.p->pageRef[tabPtr.p->noPages] = pagePtr.i;
tabPtr.p->noPages++;
} else {
jam();
pagePtr.i = tabPtr.p->pageRef[tabPtr.p->noPages - 1];
ptrCheckGuard(pagePtr, cpageFileSize, pageRecord);
}//if
ndbrequire(tabPtr.p->noOfWords + 15 < 2048);
ndbrequire(tabPtr.p->noOfWords < 2048);
MEMCOPY_NO_WORDS(&pagePtr.p->word[tabPtr.p->noOfWords], &signal->theData[5], 16);
tabPtr.p->noOfWords += 16;
if (tabPtr.p->noOfWords == 2048) {
jam();
tabPtr.p->noOfWords = 0;
}//if
if (noOfWords > 16) {
jam();
return;
}//if
tabPtr.p->noOfWords = 0;
ndbrequire(tabPtr.p->tabCopyStatus == TabRecord::CS_IDLE);
tabPtr.p->tabCopyStatus = TabRecord::CS_COPY_TAB_REQ;
signal->theData[0] = DihContinueB::ZREAD_PAGES_INTO_TABLE;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
}//Dbdih::execCOPY_TABREQ()
void
Dbdih::copyTabReq_complete(Signal* signal, TabRecordPtr tabPtr){
if (!isMaster()) {
jam();
//----------------------------------------------------------------------------
// In this particular case we do not release table pages if we are master. The
// reason is that the master could still be sending the table info to another
// node.
//----------------------------------------------------------------------------
releaseTabPages(tabPtr.i);
tabPtr.p->tabStatus = TabRecord::TS_ACTIVE;
for (Uint32 fragId = 0; fragId < tabPtr.p->totalfragments; fragId++) {
jam();
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
updateNodeInfo(fragPtr);
}//for
}//if
signal->theData[0] = cownNodeId;
signal->theData[1] = tabPtr.i;
sendSignal(cmasterdihref, GSN_COPY_TABCONF, signal, 2, JBB);
}
/*****************************************************************************/
/* ****** READ FROM A NUMBER OF PAGES INTO THE TABLE DATA STRUCTURES ********/
/*****************************************************************************/
void Dbdih::readPagesIntoTableLab(Signal* signal, Uint32 tableId)
{
RWFragment rf;
rf.wordIndex = 35;
rf.pageIndex = 0;
rf.rwfTabPtr.i = tableId;
ptrCheckGuard(rf.rwfTabPtr, ctabFileSize, tabRecord);
rf.rwfPageptr.i = rf.rwfTabPtr.p->pageRef[0];
ptrCheckGuard(rf.rwfPageptr, cpageFileSize, pageRecord);
rf.rwfTabPtr.p->totalfragments = readPageWord(&rf);
rf.rwfTabPtr.p->noOfBackups = readPageWord(&rf);
rf.rwfTabPtr.p->hashpointer = readPageWord(&rf);
rf.rwfTabPtr.p->kvalue = readPageWord(&rf);
rf.rwfTabPtr.p->mask = readPageWord(&rf);
ndbrequire(readPageWord(&rf) == TabRecord::HASH);
rf.rwfTabPtr.p->method = TabRecord::HASH;
/* ---------------------------------- */
/* Type of table, 2 = temporary table */
/* ---------------------------------- */
rf.rwfTabPtr.p->storedTable = readPageWord(&rf);
Uint32 noOfFrags = rf.rwfTabPtr.p->totalfragments;
ndbrequire(noOfFrags > 0);
ndbrequire((noOfFrags * (rf.rwfTabPtr.p->noOfBackups + 1)) <= cnoFreeReplicaRec);
allocFragments(noOfFrags, rf.rwfTabPtr);
signal->theData[0] = DihContinueB::ZREAD_PAGES_INTO_FRAG;
signal->theData[1] = rf.rwfTabPtr.i;
signal->theData[2] = 0;
signal->theData[3] = rf.pageIndex;
signal->theData[4] = rf.wordIndex;
sendSignal(reference(), GSN_CONTINUEB, signal, 5, JBB);
return;
}//Dbdih::readPagesIntoTableLab()
void Dbdih::readPagesIntoFragLab(Signal* signal, RWFragment* rf)
{
ndbrequire(rf->pageIndex < 8);
rf->rwfPageptr.i = rf->rwfTabPtr.p->pageRef[rf->pageIndex];
ptrCheckGuard(rf->rwfPageptr, cpageFileSize, pageRecord);
FragmentstorePtr fragPtr;
getFragstore(rf->rwfTabPtr.p, rf->fragId, fragPtr);
readFragment(rf, fragPtr);
readReplicas(rf, fragPtr);
rf->fragId++;
if (rf->fragId == rf->rwfTabPtr.p->totalfragments) {
jam();
switch (rf->rwfTabPtr.p->tabCopyStatus) {
case TabRecord::CS_SR_PHASE1_READ_PAGES:
jam();
releaseTabPages(rf->rwfTabPtr.i);
rf->rwfTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
signal->theData[0] = DihContinueB::ZREAD_TABLE_FROM_PAGES;
signal->theData[1] = rf->rwfTabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
case TabRecord::CS_COPY_TAB_REQ:
jam();
rf->rwfTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
if(getNodeState().getSystemRestartInProgress()){
jam();
copyTabReq_complete(signal, rf->rwfTabPtr);
return;
}
rf->rwfTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
rf->rwfTabPtr.p->tabUpdateState = TabRecord::US_COPY_TAB_REQ;
signal->theData[0] = DihContinueB::ZTABLE_UPDATE;
signal->theData[1] = rf->rwfTabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
default:
ndbrequire(false);
return;
break;
}//switch
} else {
jam();
signal->theData[0] = DihContinueB::ZREAD_PAGES_INTO_FRAG;
signal->theData[1] = rf->rwfTabPtr.i;
signal->theData[2] = rf->fragId;
signal->theData[3] = rf->pageIndex;
signal->theData[4] = rf->wordIndex;
sendSignal(reference(), GSN_CONTINUEB, signal, 5, JBB);
}//if
return;
}//Dbdih::readPagesIntoFragLab()
/*****************************************************************************/
/***** WRITING FROM TABLE DATA STRUCTURES INTO A SET OF PAGES ******/
// execCONTINUEB(ZPACK_TABLE_INTO_PAGES)
/*****************************************************************************/
void Dbdih::packTableIntoPagesLab(Signal* signal, Uint32 tableId)
{
RWFragment wf;
TabRecordPtr tabPtr;
allocpage(wf.rwfPageptr);
tabPtr.i = tableId;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
tabPtr.p->pageRef[0] = wf.rwfPageptr.i;
tabPtr.p->noPages = 1;
wf.wordIndex = 35;
wf.pageIndex = 0;
writePageWord(&wf, tabPtr.p->totalfragments);
writePageWord(&wf, tabPtr.p->noOfBackups);
writePageWord(&wf, tabPtr.p->hashpointer);
writePageWord(&wf, tabPtr.p->kvalue);
writePageWord(&wf, tabPtr.p->mask);
writePageWord(&wf, TabRecord::HASH);
writePageWord(&wf, tabPtr.p->storedTable);
signal->theData[0] = DihContinueB::ZPACK_FRAG_INTO_PAGES;
signal->theData[1] = tabPtr.i;
signal->theData[2] = 0;
signal->theData[3] = wf.pageIndex;
signal->theData[4] = wf.wordIndex;
sendSignal(reference(), GSN_CONTINUEB, signal, 5, JBB);
}//Dbdih::packTableIntoPagesLab()
/*****************************************************************************/
// execCONTINUEB(ZPACK_FRAG_INTO_PAGES)
/*****************************************************************************/
void Dbdih::packFragIntoPagesLab(Signal* signal, RWFragment* wf)
{
ndbrequire(wf->pageIndex < 8);
wf->rwfPageptr.i = wf->rwfTabPtr.p->pageRef[wf->pageIndex];
ptrCheckGuard(wf->rwfPageptr, cpageFileSize, pageRecord);
FragmentstorePtr fragPtr;
getFragstore(wf->rwfTabPtr.p, wf->fragId, fragPtr);
writeFragment(wf, fragPtr);
writeReplicas(wf, fragPtr.p->storedReplicas);
writeReplicas(wf, fragPtr.p->oldStoredReplicas);
wf->fragId++;
if (wf->fragId == wf->rwfTabPtr.p->totalfragments) {
jam();
PageRecordPtr pagePtr;
pagePtr.i = wf->rwfTabPtr.p->pageRef[0];
ptrCheckGuard(pagePtr, cpageFileSize, pageRecord);
pagePtr.p->word[33] = wf->rwfTabPtr.p->noPages;
pagePtr.p->word[34] = ((wf->rwfTabPtr.p->noPages - 1) * 2048) + wf->wordIndex;
switch (wf->rwfTabPtr.p->tabCopyStatus) {
case TabRecord::CS_SR_PHASE2_READ_TABLE:
/* -------------------------------------------------------------------*/
// We are performing a system restart and we are now ready to copy the
// table from this node (the master) to all other nodes.
/* -------------------------------------------------------------------*/
jam();
wf->rwfTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
signal->theData[0] = DihContinueB::ZSR_PHASE2_READ_TABLE;
signal->theData[1] = wf->rwfTabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
case TabRecord::CS_COPY_NODE_STATE:
jam();
tableCopyNodeLab(signal, wf->rwfTabPtr);
return;
break;
case TabRecord::CS_LCP_READ_TABLE:
jam();
signal->theData[0] = DihContinueB::ZTABLE_UPDATE;
signal->theData[1] = wf->rwfTabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
case TabRecord::CS_REMOVE_NODE:
case TabRecord::CS_INVALIDATE_NODE_LCP:
jam();
signal->theData[0] = DihContinueB::ZTABLE_UPDATE;
signal->theData[1] = wf->rwfTabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
case TabRecord::CS_ADD_TABLE_MASTER:
jam();
wf->rwfTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
signal->theData[0] = DihContinueB::ZADD_TABLE_MASTER_PAGES;
signal->theData[1] = wf->rwfTabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
case TabRecord::CS_ADD_TABLE_SLAVE:
jam();
wf->rwfTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
signal->theData[0] = DihContinueB::ZADD_TABLE_SLAVE_PAGES;
signal->theData[1] = wf->rwfTabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
default:
ndbrequire(false);
return;
break;
}//switch
} else {
jam();
signal->theData[0] = DihContinueB::ZPACK_FRAG_INTO_PAGES;
signal->theData[1] = wf->rwfTabPtr.i;
signal->theData[2] = wf->fragId;
signal->theData[3] = wf->pageIndex;
signal->theData[4] = wf->wordIndex;
sendSignal(reference(), GSN_CONTINUEB, signal, 5, JBB);
}//if
return;
}//Dbdih::packFragIntoPagesLab()
/*****************************************************************************/
/* ********** START FRAGMENT MODULE *************/
/*****************************************************************************/
void Dbdih::startFragment(Signal* signal, Uint32 tableId, Uint32 fragId)
{
Uint32 TloopCount = 0;
TabRecordPtr tabPtr;
while (true) {
if (TloopCount > 100) {
jam();
signal->theData[0] = DihContinueB::ZSTART_FRAGMENT;
signal->theData[1] = tableId;
signal->theData[2] = 0;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
}
if (tableId >= ctabFileSize) {
jam();
signal->theData[0] = DihContinueB::ZCOMPLETE_RESTART;
sendSignal(reference(), GSN_CONTINUEB, signal, 1, JBB);
return;
}//if
tabPtr.i = tableId;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
if (tabPtr.p->tabStatus != TabRecord::TS_ACTIVE){
jam();
TloopCount++;
tableId++;
fragId = 0;
continue;
}
if(tabPtr.p->storedTable == 0){
jam();
TloopCount++;
tableId++;
fragId = 0;
continue;
}
jam();
break;
}//while
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
/* ----------------------------------------------------------------------- */
/* WE NEED TO RESET THE REPLICA DATA STRUCTURES. THIS MEANS THAT WE */
/* MUST REMOVE REPLICAS THAT WAS NOT STARTED AT THE GCI TO RESTORE. WE */
/* NEED TO PUT ALL STORED REPLICAS ON THE LIST OF OLD STORED REPLICAS */
/* RESET THE NUMBER OF REPLICAS TO CREATE. */
/* ----------------------------------------------------------------------- */
cnoOfCreateReplicas = 0;
/* ----------------------------------------------------------------------- */
/* WE WILL NEVER START MORE THAN FOUR FRAGMENT REPLICAS WHATEVER THE */
/* DESIRED REPLICATION IS. */
/* ----------------------------------------------------------------------- */
ndbrequire(tabPtr.p->noOfBackups < 4);
/* ----------------------------------------------------------------------- */
/* SEARCH FOR STORED REPLICAS THAT CAN BE USED TO RESTART THE SYSTEM. */
/* ----------------------------------------------------------------------- */
searchStoredReplicas(fragPtr);
if (cnoOfCreateReplicas == 0) {
/* --------------------------------------------------------------------- */
/* THERE WERE NO STORED REPLICAS AVAILABLE THAT CAN SERVE AS REPLICA TO*/
/* RESTART THE SYSTEM FROM. IN A LATER RELEASE WE WILL ADD */
/* FUNCTIONALITY TO CHECK IF THERE ARE ANY STANDBY NODES THAT COULD DO */
/* THIS TASK INSTEAD IN THIS IMPLEMENTATION WE SIMPLY CRASH THE SYSTEM.*/
/* THIS WILL DECREASE THE GCI TO RESTORE WHICH HOPEFULLY WILL MAKE IT */
/* POSSIBLE TO RESTORE THE SYSTEM. */
/* --------------------------------------------------------------------- */
char buf[100];
BaseString::snprintf(buf, sizeof(buf),
"Unable to find restorable replica for "
"table: %d fragment: %d gci: %d",
tableId, fragId, SYSFILE->newestRestorableGCI);
progError(__LINE__,
ERR_SYSTEM_ERROR,
buf);
ndbrequire(false);
return;
}//if
/* ----------------------------------------------------------------------- */
/* WE HAVE CHANGED THE NODE TO BE PRIMARY REPLICA AND THE NODES TO BE */
/* BACKUP NODES. WE MUST UPDATE THIS NODES DATA STRUCTURE SINCE WE */
/* WILL NOT COPY THE TABLE DATA TO OURSELF. */
/* ----------------------------------------------------------------------- */
updateNodeInfo(fragPtr);
/* ----------------------------------------------------------------------- */
/* NOW WE HAVE COLLECTED ALL THE REPLICAS WE COULD GET. WE WILL NOW */
/* RESTART THE FRAGMENT REPLICAS WE HAVE FOUND IRRESPECTIVE OF IF THERE*/
/* ARE ENOUGH ACCORDING TO THE DESIRED REPLICATION. */
/* ----------------------------------------------------------------------- */
/* WE START BY SENDING ADD_FRAGREQ FOR THOSE REPLICAS THAT NEED IT. */
/* ----------------------------------------------------------------------- */
CreateReplicaRecordPtr createReplicaPtr;
for (createReplicaPtr.i = 0;
createReplicaPtr.i < cnoOfCreateReplicas;
createReplicaPtr.i++) {
jam();
ptrCheckGuard(createReplicaPtr, 4, createReplicaRecord);
createReplicaPtr.p->hotSpareUse = false;
}//for
sendStartFragreq(signal, tabPtr, fragId);
/**
* Don't wait for START_FRAGCONF
*/
fragId++;
if (fragId >= tabPtr.p->totalfragments) {
jam();
tabPtr.i++;
fragId = 0;
}//if
signal->theData[0] = DihContinueB::ZSTART_FRAGMENT;
signal->theData[1] = tabPtr.i;
signal->theData[2] = fragId;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
}//Dbdih::startFragmentLab()
/*****************************************************************************/
/* ********** COMPLETE RESTART MODULE *************/
/*****************************************************************************/
void Dbdih::completeRestartLab(Signal* signal)
{
sendLoopMacro(START_RECREQ, sendSTART_RECREQ);
}//completeRestartLab()
/* ------------------------------------------------------------------------- */
// SYSTEM RESTART:
/* A NODE HAS COMPLETED RESTORING ALL DATABASE FRAGMENTS. */
// NODE RESTART:
// THE STARTING NODE HAS PREPARED ITS LOG FILES TO ENABLE EXECUTION
// OF TRANSACTIONS.
// Precondition:
// This signal must be received by the master node.
/* ------------------------------------------------------------------------- */
void Dbdih::execSTART_RECCONF(Signal* signal)
{
jamEntry();
Uint32 senderNodeId = signal->theData[0];
ndbrequire(isMaster());
if (getNodeState().startLevel >= NodeState::SL_STARTED){
/* --------------------------------------------------------------------- */
// Since our node is already up and running this must be a node restart.
// This means that we should be the master node,
// otherwise we have a problem.
/* --------------------------------------------------------------------- */
jam();
ndbrequire(senderNodeId == c_nodeStartMaster.startNode);
nodeRestartStartRecConfLab(signal);
return;
} else {
/* --------------------------------------------------------------------- */
// This was the system restart case. We set the state indicating that the
// node has completed restoration of all fragments.
/* --------------------------------------------------------------------- */
receiveLoopMacro(START_RECREQ, senderNodeId);
signal->theData[0] = reference();
sendSignal(cntrlblockref, GSN_NDB_STARTCONF, signal, 1, JBB);
return;
}//if
}//Dbdih::execSTART_RECCONF()
void Dbdih::copyNodeLab(Signal* signal, Uint32 tableId)
{
/* ----------------------------------------------------------------------- */
// This code is executed by the master to assist a node restart in receiving
// the data in the master.
/* ----------------------------------------------------------------------- */
Uint32 TloopCount = 0;
if (!c_nodeStartMaster.activeState) {
jam();
/* --------------------------------------------------------------------- */
// Obviously the node crashed in the middle of its node restart. We will
// stop this process simply by returning after resetting the wait indicator.
/* ---------------------------------------------------------------------- */
c_nodeStartMaster.wait = ZFALSE;
return;
}//if
TabRecordPtr tabPtr;
tabPtr.i = tableId;
while (tabPtr.i < ctabFileSize) {
ptrAss(tabPtr, tabRecord);
if (tabPtr.p->tabStatus == TabRecord::TS_ACTIVE) {
/* -------------------------------------------------------------------- */
// The table is defined. We will start by packing the table into pages.
// The tabCopyStatus indicates to the CONTINUEB(ZPACK_TABLE_INTO_PAGES)
// who called it. After packing the table into page(s) it will be sent to
// the starting node by COPY_TABREQ signals. After returning from the
// starting node we will return to this subroutine and continue
// with the next table.
/* -------------------------------------------------------------------- */
ndbrequire(tabPtr.p->tabCopyStatus == TabRecord::CS_IDLE);
tabPtr.p->tabCopyStatus = TabRecord::CS_COPY_NODE_STATE;
signal->theData[0] = DihContinueB::ZPACK_TABLE_INTO_PAGES;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
} else {
jam();
if (TloopCount > 100) {
/* ------------------------------------------------------------------ */
// Introduce real-time break after looping through 100 not copied tables
/* ----------------------------------------------------------------- */
jam();
signal->theData[0] = DihContinueB::ZCOPY_NODE;
signal->theData[1] = tabPtr.i + 1;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
} else {
jam();
TloopCount++;
tabPtr.i++;
}//if
}//if
}//while
dihCopyCompletedLab(signal);
return;
}//Dbdih::copyNodeLab()
void Dbdih::tableCopyNodeLab(Signal* signal, TabRecordPtr tabPtr)
{
/* ----------------------------------------------------------------------- */
/* COPY PAGES READ TO STARTING NODE. */
/* ----------------------------------------------------------------------- */
if (!c_nodeStartMaster.activeState) {
jam();
releaseTabPages(tabPtr.i);
c_nodeStartMaster.wait = ZFALSE;
return;
}//if
NodeRecordPtr copyNodePtr;
PageRecordPtr pagePtr;
copyNodePtr.i = c_nodeStartMaster.startNode;
ptrCheckGuard(copyNodePtr, MAX_NDB_NODES, nodeRecord);
copyNodePtr.p->activeTabptr = tabPtr.i;
pagePtr.i = tabPtr.p->pageRef[0];
ptrCheckGuard(pagePtr, cpageFileSize, pageRecord);
signal->theData[0] = DihContinueB::ZCOPY_TABLE_NODE;
signal->theData[1] = tabPtr.i;
signal->theData[2] = copyNodePtr.i;
signal->theData[3] = 0;
signal->theData[4] = 0;
signal->theData[5] = pagePtr.p->word[34];
sendSignal(reference(), GSN_CONTINUEB, signal, 6, JBB);
}//Dbdih::tableCopyNodeLab()
/* ------------------------------------------------------------------------- */
// execCONTINUEB(ZCOPY_TABLE)
// This routine is used to copy the table descriptions from the master to
// other nodes. It is used in the system restart to copy from master to all
// starting nodes.
/* ------------------------------------------------------------------------- */
void Dbdih::copyTableLab(Signal* signal, Uint32 tableId)
{
TabRecordPtr tabPtr;
tabPtr.i = tableId;
ptrAss(tabPtr, tabRecord);
ndbrequire(tabPtr.p->tabCopyStatus == TabRecord::CS_IDLE);
tabPtr.p->tabCopyStatus = TabRecord::CS_SR_PHASE2_READ_TABLE;
signal->theData[0] = DihContinueB::ZPACK_TABLE_INTO_PAGES;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
}//Dbdih::copyTableLab()
/* ------------------------------------------------------------------------- */
// execCONTINUEB(ZSR_PHASE2_READ_TABLE)
/* ------------------------------------------------------------------------- */
void Dbdih::srPhase2ReadTableLab(Signal* signal, TabRecordPtr tabPtr)
{
/* ----------------------------------------------------------------------- */
// We set the sendCOPY_TABREQState to ZACTIVE for all nodes since it is a long
// process to send off all table descriptions. Thus we ensure that we do
// not encounter race conditions where one node is completed before the
// sending process is completed. This could lead to that we start off the
// system before we actually finished all copying of table descriptions
// and could lead to strange errors.
/* ----------------------------------------------------------------------- */
//sendLoopMacro(COPY_TABREQ, nullRoutine);
breakCopyTableLab(signal, tabPtr, cfirstAliveNode);
return;
}//Dbdih::srPhase2ReadTableLab()
/* ------------------------------------------------------------------------- */
/* COPY PAGES READ TO ALL NODES. */
/* ------------------------------------------------------------------------- */
void Dbdih::breakCopyTableLab(Signal* signal, TabRecordPtr tabPtr, Uint32 nodeId)
{
NodeRecordPtr nodePtr;
nodePtr.i = nodeId;
while (nodePtr.i != RNIL) {
jam();
ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
if (nodePtr.i == getOwnNodeId()){
jam();
/* ------------------------------------------------------------------- */
/* NOT NECESSARY TO COPY TO MY OWN NODE. I ALREADY HAVE THE PAGES. */
/* I DO HOWEVER NEED TO STORE THE TABLE DESCRIPTION ONTO DISK. */
/* ------------------------------------------------------------------- */
/* IF WE ARE MASTER WE ONLY NEED TO SAVE THE TABLE ON DISK. WE ALREADY */
/* HAVE THE TABLE DESCRIPTION IN THE DATA STRUCTURES. */
// AFTER COMPLETING THE WRITE TO DISK THE MASTER WILL ALSO SEND
// COPY_TABCONF AS ALL THE OTHER NODES.
/* ------------------------------------------------------------------- */
c_COPY_TABREQ_Counter.setWaitingFor(nodePtr.i);
tabPtr.p->tabUpdateState = TabRecord::US_COPY_TAB_REQ;
signal->theData[0] = DihContinueB::ZTABLE_UPDATE;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
nodePtr.i = nodePtr.p->nextNode;
} else {
PageRecordPtr pagePtr;
/* -------------------------------------------------------------------- */
// RATHER THAN SENDING ALL COPY_TABREQ IN PARALLEL WE WILL SERIALISE THIS
// ACTIVITY AND WILL THUS CALL breakCopyTableLab AGAIN WHEN COMPLETED THE
// SENDING OF COPY_TABREQ'S.
/* -------------------------------------------------------------------- */
jam();
tabPtr.p->tabCopyStatus = TabRecord::CS_SR_PHASE3_COPY_TABLE;
pagePtr.i = tabPtr.p->pageRef[0];
ptrCheckGuard(pagePtr, cpageFileSize, pageRecord);
signal->theData[0] = DihContinueB::ZCOPY_TABLE_NODE;
signal->theData[1] = tabPtr.i;
signal->theData[2] = nodePtr.i;
signal->theData[3] = 0;
signal->theData[4] = 0;
signal->theData[5] = pagePtr.p->word[34];
sendSignal(reference(), GSN_CONTINUEB, signal, 6, JBB);
return;
}//if
}//while
/* ----------------------------------------------------------------------- */
/* WE HAVE NOW SENT THE TABLE PAGES TO ALL NODES. EXIT AND WAIT FOR ALL */
/* REPLIES. */
/* ----------------------------------------------------------------------- */
return;
}//Dbdih::breakCopyTableLab()
/* ------------------------------------------------------------------------- */
// execCONTINUEB(ZCOPY_TABLE_NODE)
/* ------------------------------------------------------------------------- */
void Dbdih::copyTableNode(Signal* signal,
CopyTableNode* ctn, NodeRecordPtr nodePtr)
{
if (getNodeState().startLevel >= NodeState::SL_STARTED){
/* --------------------------------------------------------------------- */
// We are in the process of performing a node restart and are copying a
// table description to a starting node. We will check that no nodes have
// crashed in this process.
/* --------------------------------------------------------------------- */
if (!c_nodeStartMaster.activeState) {
jam();
/** ------------------------------------------------------------------
* The starting node crashed. We will release table pages and stop this
* copy process and allow new node restarts to start.
* ------------------------------------------------------------------ */
releaseTabPages(ctn->ctnTabPtr.i);
c_nodeStartMaster.wait = ZFALSE;
return;
}//if
}//if
ndbrequire(ctn->pageIndex < 8);
ctn->ctnPageptr.i = ctn->ctnTabPtr.p->pageRef[ctn->pageIndex];
ptrCheckGuard(ctn->ctnPageptr, cpageFileSize, pageRecord);
/**
* If first page & firstWord reqinfo = 1 (first signal)
*/
Uint32 reqinfo = (ctn->pageIndex == 0) && (ctn->wordIndex == 0);
if(reqinfo == 1){
c_COPY_TABREQ_Counter.setWaitingFor(nodePtr.i);
}
for (Uint32 i = 0; i < 16; i++) {
jam();
sendCopyTable(signal, ctn, calcDihBlockRef(nodePtr.i), reqinfo);
reqinfo = 0;
if (ctn->noOfWords <= 16) {
jam();
switch (ctn->ctnTabPtr.p->tabCopyStatus) {
case TabRecord::CS_SR_PHASE3_COPY_TABLE:
/* ------------------------------------------------------------------ */
// We have copied the table description to this node.
// We will now proceed
// with sending the table description to the next node in the node list.
/* ------------------------------------------------------------------ */
jam();
ctn->ctnTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
breakCopyTableLab(signal, ctn->ctnTabPtr, nodePtr.p->nextNode);
return;
break;
case TabRecord::CS_COPY_NODE_STATE:
jam();
ctn->ctnTabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
return;
break;
default:
ndbrequire(false);
break;
}//switch
} else {
jam();
ctn->wordIndex += 16;
if (ctn->wordIndex == 2048) {
jam();
ctn->wordIndex = 0;
ctn->pageIndex++;
ndbrequire(ctn->pageIndex < 8);
ctn->ctnPageptr.i = ctn->ctnTabPtr.p->pageRef[ctn->pageIndex];
ptrCheckGuard(ctn->ctnPageptr, cpageFileSize, pageRecord);
}//if
ctn->noOfWords -= 16;
}//if
}//for
signal->theData[0] = DihContinueB::ZCOPY_TABLE_NODE;
signal->theData[1] = ctn->ctnTabPtr.i;
signal->theData[2] = nodePtr.i;
signal->theData[3] = ctn->pageIndex;
signal->theData[4] = ctn->wordIndex;
signal->theData[5] = ctn->noOfWords;
sendSignal(reference(), GSN_CONTINUEB, signal, 6, JBB);
}//Dbdih::copyTableNodeLab()
void Dbdih::sendCopyTable(Signal* signal, CopyTableNode* ctn,
BlockReference ref, Uint32 reqinfo)
{
signal->theData[0] = reference();
signal->theData[1] = reqinfo;
signal->theData[2] = ctn->ctnTabPtr.i;
signal->theData[3] = ctn->ctnTabPtr.p->schemaVersion;
signal->theData[4] = ctn->noOfWords;
ndbrequire(ctn->wordIndex + 15 < 2048);
MEMCOPY_NO_WORDS(&signal->theData[5], &ctn->ctnPageptr.p->word[ctn->wordIndex], 16);
sendSignal(ref, GSN_COPY_TABREQ, signal, 21, JBB);
}//Dbdih::sendCopyTable()
void Dbdih::execCOPY_TABCONF(Signal* signal)
{
NodeRecordPtr nodePtr;
jamEntry();
nodePtr.i = signal->theData[0];
Uint32 tableId = signal->theData[1];
if (getNodeState().startLevel >= NodeState::SL_STARTED){
/* --------------------------------------------------------------------- */
// We are in the process of performing a node restart. Continue by copying
// the next table to the starting node.
/* --------------------------------------------------------------------- */
jam();
NodeRecordPtr nodePtr;
nodePtr.i = signal->theData[0];
ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
c_COPY_TABREQ_Counter.clearWaitingFor(nodePtr.i);
releaseTabPages(tableId);
signal->theData[0] = DihContinueB::ZCOPY_NODE;
signal->theData[1] = tableId + 1;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
} else {
/* --------------------------------------------------------------------- */
// We are in the process of performing a system restart. Check if all nodes
// have saved the new table description to file and then continue with the
// next table.
/* --------------------------------------------------------------------- */
receiveLoopMacro(COPY_TABREQ, nodePtr.i);
/* --------------------------------------------------------------------- */
/* WE HAVE NOW COPIED TO ALL NODES. WE HAVE NOW COMPLETED RESTORING */
/* THIS TABLE. CONTINUE WITH THE NEXT TABLE. */
/* WE NEED TO RELEASE THE PAGES IN THE TABLE IN THIS NODE HERE. */
/* WE ALSO NEED TO CLOSE THE TABLE FILE. */
/* --------------------------------------------------------------------- */
releaseTabPages(tableId);
TabRecordPtr tabPtr;
tabPtr.i = tableId;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
ConnectRecordPtr connectPtr;
connectPtr.i = tabPtr.p->connectrec;
ptrCheckGuard(connectPtr, cconnectFileSize, connectRecord);
sendAddFragreq(signal, connectPtr, tabPtr, 0);
return;
}//if
}//Dbdih::execCOPY_TABCONF()
/*
3.13 L O C A L C H E C K P O I N T (M A S T E R)
****************************************************
*/
/*****************************************************************************/
/* ********** LOCAL-CHECK-POINT-HANDLING MODULE *************/
/*****************************************************************************/
/* ------------------------------------------------------------------------- */
/* IT IS TIME TO CHECK IF IT IS TIME TO START A LOCAL CHECKPOINT. */
/* WE WILL EITHER START AFTER 1 MILLION WORDS HAVE ARRIVED OR WE WILL */
/* EXECUTE AFTER ABOUT 16 MINUTES HAVE PASSED BY. */
/* ------------------------------------------------------------------------- */
void Dbdih::checkTcCounterLab(Signal* signal)
{
CRASH_INSERTION(7009);
if (c_lcpState.lcpStatus != LCP_STATUS_IDLE) {
ndbout << "lcpStatus = " << (Uint32) c_lcpState.lcpStatus;
ndbout << "lcpStatusUpdatedPlace = " <<
c_lcpState.lcpStatusUpdatedPlace << endl;
ndbrequire(false);
return;
}//if
c_lcpState.ctimer += 32;
if ((c_nodeStartMaster.blockLcp == true) ||
((c_lcpState.lcpStartGcp + 1) > currentgcp)) {
jam();
/* --------------------------------------------------------------------- */
// No reason to start juggling the states and checking for start of LCP if
// we are blocked to start an LCP anyway.
// We also block LCP start if we have not completed one global checkpoints
// before starting another local checkpoint.
/* --------------------------------------------------------------------- */
signal->theData[0] = DihContinueB::ZCHECK_TC_COUNTER;
signal->theData[1] = __LINE__;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 1 * 100, 2);
return;
}//if
c_lcpState.setLcpStatus(LCP_TCGET, __LINE__);
c_lcpState.ctcCounter = c_lcpState.ctimer;
sendLoopMacro(TCGETOPSIZEREQ, sendTCGETOPSIZEREQ);
}//Dbdih::checkTcCounterLab()
void Dbdih::checkLcpStart(Signal* signal, Uint32 lineNo)
{
/* ----------------------------------------------------------------------- */
// Verify that we are not attempting to start another instance of the LCP
// when it is not alright to do so.
/* ----------------------------------------------------------------------- */
ndbrequire(c_lcpState.lcpStart == ZIDLE);
c_lcpState.lcpStart = ZACTIVE;
signal->theData[0] = DihContinueB::ZCHECK_TC_COUNTER;
signal->theData[1] = lineNo;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 1000, 2);
}//Dbdih::checkLcpStart()
/* ------------------------------------------------------------------------- */
/*TCGETOPSIZECONF HOW MUCH OPERATION SIZE HAVE BEEN EXECUTED BY TC */
/* ------------------------------------------------------------------------- */
void Dbdih::execTCGETOPSIZECONF(Signal* signal)
{
jamEntry();
Uint32 senderNodeId = signal->theData[0];
c_lcpState.ctcCounter += signal->theData[1];
receiveLoopMacro(TCGETOPSIZEREQ, senderNodeId);
ndbrequire(c_lcpState.lcpStatus == LCP_TCGET);
ndbrequire(c_lcpState.lcpStart == ZACTIVE);
/* ----------------------------------------------------------------------- */
// We are not actively starting another LCP, still we receive this signal.
// This is not ok.
/* ---------------------------------------------------------------------- */
/* ALL TC'S HAVE RESPONDED NOW. NOW WE WILL CHECK IF ENOUGH OPERATIONS */
/* HAVE EXECUTED TO ENABLE US TO START A NEW LOCAL CHECKPOINT. */
/* WHILE COPYING DICTIONARY AND DISTRIBUTION INFO TO A STARTING NODE */
/* WE WILL ALSO NOT ALLOW THE LOCAL CHECKPOINT TO PROCEED. */
/*----------------------------------------------------------------------- */
if (c_lcpState.immediateLcpStart == false) {
if ((c_lcpState.ctcCounter <
((Uint32)1 << c_lcpState.clcpDelay)) ||
(c_nodeStartMaster.blockLcp == true)) {
jam();
c_lcpState.setLcpStatus(LCP_STATUS_IDLE, __LINE__);
signal->theData[0] = DihContinueB::ZCHECK_TC_COUNTER;
signal->theData[1] = __LINE__;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 1 * 100, 2);
return;
}//if
}//if
c_lcpState.lcpStart = ZIDLE;
c_lcpState.immediateLcpStart = false;
/* -----------------------------------------------------------------------
* Now the initial lcp is started,
* we can reset the delay to its orginal value
* --------------------------------------------------------------------- */
CRASH_INSERTION(7010);
/* ----------------------------------------------------------------------- */
/* IF MORE THAN 1 MILLION WORDS PASSED THROUGH THE TC'S THEN WE WILL */
/* START A NEW LOCAL CHECKPOINT. CLEAR CTIMER. START CHECKPOINT */
/* ACTIVITY BY CALCULATING THE KEEP GLOBAL CHECKPOINT. */
// Also remember the current global checkpoint to ensure that we run at least
// one global checkpoints between each local checkpoint that we start up.
/* ----------------------------------------------------------------------- */
c_lcpState.ctimer = 0;
c_lcpState.keepGci = coldgcp;
c_lcpState.lcpStartGcp = currentgcp;
/* ----------------------------------------------------------------------- */
/* UPDATE THE NEW LATEST LOCAL CHECKPOINT ID. */
/* ----------------------------------------------------------------------- */
cnoOfActiveTables = 0;
c_lcpState.setLcpStatus(LCP_CALCULATE_KEEP_GCI, __LINE__);
c_lcpState.oldestRestorableGci = SYSFILE->oldestRestorableGCI;
ndbrequire(((int)c_lcpState.oldestRestorableGci) > 0);
if (ERROR_INSERTED(7011)) {
signal->theData[0] = EventReport::LCPStoppedInCalcKeepGci;
signal->theData[1] = 0;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
return;
}//if
signal->theData[0] = DihContinueB::ZCALCULATE_KEEP_GCI;
signal->theData[1] = 0; /* TABLE ID = 0 */
signal->theData[2] = 0; /* FRAGMENT ID = 0 */
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
}//Dbdih::execTCGETOPSIZECONF()
/* ------------------------------------------------------------------------- */
/* WE NEED TO CALCULATE THE OLDEST GLOBAL CHECKPOINT THAT WILL BE */
/* COMPLETELY RESTORABLE AFTER EXECUTING THIS LOCAL CHECKPOINT. */
/* ------------------------------------------------------------------------- */
void Dbdih::calculateKeepGciLab(Signal* signal, Uint32 tableId, Uint32 fragId)
{
TabRecordPtr tabPtr;
Uint32 TloopCount = 1;
tabPtr.i = tableId;
do {
if (tabPtr.i >= ctabFileSize) {
if (cnoOfActiveTables > 0) {
jam();
signal->theData[0] = DihContinueB::ZSTORE_NEW_LCP_ID;
sendSignal(reference(), GSN_CONTINUEB, signal, 1, JBB);
return;
} else {
jam();
/* ------------------------------------------------------------------ */
/* THERE ARE NO TABLES TO CHECKPOINT. WE STOP THE CHECKPOINT ALREADY */
/* HERE TO AVOID STRANGE PROBLEMS LATER. */
/* ------------------------------------------------------------------ */
c_lcpState.setLcpStatus(LCP_STATUS_IDLE, __LINE__);
checkLcpStart(signal, __LINE__);
return;
}//if
}//if
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
if (tabPtr.p->tabStatus != TabRecord::TS_ACTIVE ||
tabPtr.p->storedTable == 0) {
if (TloopCount > 100) {
jam();
signal->theData[0] = DihContinueB::ZCALCULATE_KEEP_GCI;
signal->theData[1] = tabPtr.i + 1;
signal->theData[2] = 0;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
} else {
jam();
TloopCount++;
tabPtr.i++;
}//if
} else {
jam();
TloopCount = 0;
}//if
} while (TloopCount != 0);
cnoOfActiveTables++;
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
checkKeepGci(fragPtr.p->storedReplicas);
fragId++;
if (fragId >= tabPtr.p->totalfragments) {
jam();
tabPtr.i++;
fragId = 0;
}//if
signal->theData[0] = DihContinueB::ZCALCULATE_KEEP_GCI;
signal->theData[1] = tabPtr.i;
signal->theData[2] = fragId;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
}//Dbdih::calculateKeepGciLab()
/* ------------------------------------------------------------------------- */
/* WE NEED TO STORE ON DISK THE FACT THAT WE ARE STARTING THIS LOCAL */
/* CHECKPOINT ROUND. THIS WILL INVALIDATE ALL THE LOCAL CHECKPOINTS */
/* THAT WILL EVENTUALLY BE OVERWRITTEN AS PART OF THIS LOCAL CHECKPOINT*/
/* ------------------------------------------------------------------------- */
void Dbdih::storeNewLcpIdLab(Signal* signal)
{
/***************************************************************************/
// Report the event that a local checkpoint has started.
/***************************************************************************/
signal->theData[0] = EventReport::LocalCheckpointStarted; //Event type
signal->theData[1] = SYSFILE->latestLCP_ID + 1;
signal->theData[2] = c_lcpState.keepGci;
signal->theData[3] = c_lcpState.oldestRestorableGci;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 4, JBB);
signal->setTrace(TestOrd::TraceLocalCheckpoint);
CRASH_INSERTION(7013);
SYSFILE->keepGCI = c_lcpState.keepGci;
//Uint32 lcpId = SYSFILE->latestLCP_ID;
SYSFILE->latestLCP_ID++;
SYSFILE->oldestRestorableGCI = c_lcpState.oldestRestorableGci;
const Uint32 oldestRestorableGCI = SYSFILE->oldestRestorableGCI;
//const Uint32 newestRestorableGCI = SYSFILE->newestRestorableGCI;
//ndbrequire(newestRestorableGCI >= oldestRestorableGCI);
Int32 val = oldestRestorableGCI;
ndbrequire(val > 0);
/* ----------------------------------------------------------------------- */
/* SET BIT INDICATING THAT LOCAL CHECKPOINT IS ONGOING. THIS IS CLEARED */
/* AT THE END OF A LOCAL CHECKPOINT. */
/* ----------------------------------------------------------------------- */
SYSFILE->setLCPOngoing(SYSFILE->systemRestartBits);
/* ---------------------------------------------------------------------- */
/* CHECK IF ANY NODE MUST BE TAKEN OUT OF SERVICE AND REFILLED WITH */
/* NEW FRESH DATA FROM AN ACTIVE NODE. */
/* ---------------------------------------------------------------------- */
setLcpActiveStatusStart(signal);
c_lcpState.setLcpStatus(LCP_COPY_GCI, __LINE__);
//#ifdef VM_TRACE
// infoEvent("LocalCheckpoint %d started", SYSFILE->latestLCP_ID);
// signal->theData[0] = 7012;
// execDUMP_STATE_ORD(signal);
//#endif
copyGciLab(signal, CopyGCIReq::LOCAL_CHECKPOINT);
}//Dbdih::storeNewLcpIdLab()
void Dbdih::startLcpRoundLab(Signal* signal) {
jam();
Mutex mutex(signal, c_mutexMgr, c_startLcpMutexHandle);
Callback c = { safe_cast(&Dbdih::startLcpMutex_locked), 0 };
ndbrequire(mutex.lock(c));
}
void
Dbdih::startLcpMutex_locked(Signal* signal, Uint32 senderData, Uint32 retVal){
jamEntry();
ndbrequire(retVal == 0);
StartLcpReq* req = (StartLcpReq*)signal->getDataPtrSend();
req->senderRef = reference();
req->lcpId = SYSFILE->latestLCP_ID;
req->participatingLQH = c_lcpState.m_participatingLQH;
req->participatingDIH = c_lcpState.m_participatingDIH;
sendLoopMacro(START_LCP_REQ, sendSTART_LCP_REQ);
}
void
Dbdih::sendSTART_LCP_REQ(Signal* signal, Uint32 nodeId){
BlockReference ref = calcDihBlockRef(nodeId);
sendSignal(ref, GSN_START_LCP_REQ, signal, StartLcpReq::SignalLength, JBB);
}
void
Dbdih::execSTART_LCP_CONF(Signal* signal){
StartLcpConf * conf = (StartLcpConf*)signal->getDataPtr();
Uint32 nodeId = refToNode(conf->senderRef);
receiveLoopMacro(START_LCP_REQ, nodeId);
Mutex mutex(signal, c_mutexMgr, c_startLcpMutexHandle);
Callback c = { safe_cast(&Dbdih::startLcpMutex_unlocked), 0 };
mutex.unlock(c);
}
void
Dbdih::startLcpMutex_unlocked(Signal* signal, Uint32 data, Uint32 retVal){
jamEntry();
ndbrequire(retVal == 0);
Mutex mutex(signal, c_mutexMgr, c_startLcpMutexHandle);
mutex.release();
CRASH_INSERTION(7014);
c_lcpState.setLcpStatus(LCP_TC_CLOPSIZE, __LINE__);
sendLoopMacro(TC_CLOPSIZEREQ, sendTC_CLOPSIZEREQ);
}
void Dbdih::execTC_CLOPSIZECONF(Signal* signal) {
jamEntry();
Uint32 senderNodeId = signal->theData[0];
receiveLoopMacro(TC_CLOPSIZEREQ, senderNodeId);
ndbrequire(c_lcpState.lcpStatus == LCP_TC_CLOPSIZE);
/* ----------------------------------------------------------------------- */
/* ALL TC'S HAVE CLEARED THEIR OPERATION SIZE COUNTERS. NOW PROCEED BY */
/* STARTING THE LOCAL CHECKPOINT IN EACH LQH. */
/* ----------------------------------------------------------------------- */
c_lcpState.m_LAST_LCP_FRAG_ORD = c_lcpState.m_participatingLQH;
CRASH_INSERTION(7015);
c_lcpState.setLcpStatus(LCP_START_LCP_ROUND, __LINE__);
startLcpRoundLoopLab(signal, 0, 0);
}//Dbdih::execTC_CLOPSIZECONF()
void Dbdih::startLcpRoundLoopLab(Signal* signal,
Uint32 startTableId, Uint32 startFragId)
{
NodeRecordPtr nodePtr;
for (nodePtr.i = 1; nodePtr.i < MAX_NDB_NODES; nodePtr.i++) {
ptrAss(nodePtr, nodeRecord);
if (nodePtr.p->nodeStatus == NodeRecord::ALIVE) {
ndbrequire(nodePtr.p->noOfStartedChkpt == 0);
ndbrequire(nodePtr.p->noOfQueuedChkpt == 0);
}//if
}//if
c_lcpState.currentFragment.tableId = startTableId;
c_lcpState.currentFragment.fragmentId = startFragId;
startNextChkpt(signal);
}//Dbdih::startLcpRoundLoopLab()
void Dbdih::startNextChkpt(Signal* signal)
{
Uint32 lcpId = SYSFILE->latestLCP_ID;
NdbNodeBitmask busyNodes;
busyNodes.clear();
const Uint32 lcpNodes = c_lcpState.m_participatingLQH.count();
bool save = true;
LcpState::CurrentFragment curr = c_lcpState.currentFragment;
while (curr.tableId < ctabFileSize) {
TabRecordPtr tabPtr;
tabPtr.i = curr.tableId;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
if ((tabPtr.p->tabStatus != TabRecord::TS_ACTIVE) ||
(tabPtr.p->tabLcpStatus != TabRecord::TLS_ACTIVE)) {
curr.tableId++;
curr.fragmentId = 0;
continue;
}//if
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, curr.fragmentId, fragPtr);
ReplicaRecordPtr replicaPtr;
for(replicaPtr.i = fragPtr.p->storedReplicas;
replicaPtr.i != RNIL ;
replicaPtr.i = replicaPtr.p->nextReplica){
jam();
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
NodeRecordPtr nodePtr;
nodePtr.i = replicaPtr.p->procNode;
ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
if (replicaPtr.p->lcpOngoingFlag &&
replicaPtr.p->lcpIdStarted < lcpId) {
jam();
//-------------------------------------------------------------------
// We have found a replica on a node that performs local checkpoint
// that is alive and that have not yet been started.
//-------------------------------------------------------------------
if (nodePtr.p->noOfStartedChkpt < 2) {
jam();
/**
* Send LCP_FRAG_ORD to LQH
*/
/**
* Mark the replica so with lcpIdStarted == true
*/
replicaPtr.p->lcpIdStarted = lcpId;
Uint32 i = nodePtr.p->noOfStartedChkpt;
nodePtr.p->startedChkpt[i].tableId = tabPtr.i;
nodePtr.p->startedChkpt[i].fragId = curr.fragmentId;
nodePtr.p->startedChkpt[i].replicaPtr = replicaPtr.i;
nodePtr.p->noOfStartedChkpt = i + 1;
sendLCP_FRAG_ORD(signal, nodePtr.p->startedChkpt[i]);
} else if (nodePtr.p->noOfQueuedChkpt < 2) {
jam();
/**
* Put LCP_FRAG_ORD "in queue"
*/
/**
* Mark the replica so with lcpIdStarted == true
*/
replicaPtr.p->lcpIdStarted = lcpId;
Uint32 i = nodePtr.p->noOfQueuedChkpt;
nodePtr.p->queuedChkpt[i].tableId = tabPtr.i;
nodePtr.p->queuedChkpt[i].fragId = curr.fragmentId;
nodePtr.p->queuedChkpt[i].replicaPtr = replicaPtr.i;
nodePtr.p->noOfQueuedChkpt = i + 1;
} else {
jam();
if(save){
/**
* Stop increasing value on first that was "full"
*/
c_lcpState.currentFragment = curr;
save = false;
}
busyNodes.set(nodePtr.i);
if(busyNodes.count() == lcpNodes){
/**
* There were no possibility to start the local checkpoint
* and it was not possible to queue it up. In this case we
* stop the start of local checkpoints until the nodes with a
* backlog have performed more checkpoints. We will return and
* will not continue the process of starting any more checkpoints.
*/
return;
}//if
}//if
}
}//while
curr.fragmentId++;
if (curr.fragmentId >= tabPtr.p->totalfragments) {
jam();
curr.fragmentId = 0;
curr.tableId++;
}//if
}//while
sendLastLCP_FRAG_ORD(signal);
}//Dbdih::startNextChkpt()
void Dbdih::sendLastLCP_FRAG_ORD(Signal* signal)
{
LcpFragOrd * const lcpFragOrd = (LcpFragOrd *)&signal->theData[0];
lcpFragOrd->tableId = RNIL;
lcpFragOrd->fragmentId = 0;
lcpFragOrd->lcpId = SYSFILE->latestLCP_ID;
lcpFragOrd->lcpNo = 0;
lcpFragOrd->keepGci = c_lcpState.keepGci;
lcpFragOrd->lastFragmentFlag = true;
NodeRecordPtr nodePtr;
for (nodePtr.i = 1; nodePtr.i < MAX_NDB_NODES; nodePtr.i++) {
jam();
ptrAss(nodePtr, nodeRecord);
if(nodePtr.p->noOfQueuedChkpt == 0 &&
nodePtr.p->noOfStartedChkpt == 0 &&
c_lcpState.m_LAST_LCP_FRAG_ORD.isWaitingFor(nodePtr.i)){
jam();
CRASH_INSERTION(7028);
/**
* Nothing queued or started <=> Complete on that node
*
*/
c_lcpState.m_LAST_LCP_FRAG_ORD.clearWaitingFor(nodePtr.i);
if(ERROR_INSERTED(7075)){
continue;
}
BlockReference ref = calcLqhBlockRef(nodePtr.i);
sendSignal(ref, GSN_LCP_FRAG_ORD, signal,LcpFragOrd::SignalLength, JBB);
}
}
if(ERROR_INSERTED(7075)){
if(c_lcpState.m_LAST_LCP_FRAG_ORD.done())
CRASH_INSERTION(7075);
}
}//Dbdih::sendLastLCP_FRAGORD()
/* ------------------------------------------------------------------------- */
/* A FRAGMENT REPLICA HAS COMPLETED EXECUTING ITS LOCAL CHECKPOINT. */
/* CHECK IF ALL REPLICAS IN THE TABLE HAVE COMPLETED. IF SO STORE THE */
/* THE TABLE DISTRIBUTION ON DISK. ALSO SEND LCP_REPORT TO ALL OTHER */
/* NODES SO THAT THEY CAN STORE THE TABLE ONTO DISK AS WELL. */
/* ------------------------------------------------------------------------- */
void Dbdih::execLCP_FRAG_REP(Signal* signal)
{
jamEntry();
ndbrequire(c_lcpState.lcpStatus != LCP_STATUS_IDLE);
#if 0
printLCP_FRAG_REP(stdout,
signal->getDataPtr(),
signal->length(), number());
#endif
LcpFragRep * const lcpReport = (LcpFragRep *)&signal->theData[0];
Uint32 nodeId = lcpReport->nodeId;
Uint32 tableId = lcpReport->tableId;
Uint32 fragId = lcpReport->fragId;
jamEntry();
CRASH_INSERTION2(7025, isMaster());
CRASH_INSERTION2(7016, !isMaster());
bool fromTimeQueue = (signal->senderBlockRef() == reference());
TabRecordPtr tabPtr;
tabPtr.i = tableId;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
if(tabPtr.p->tabCopyStatus != TabRecord::CS_IDLE) {
jam();
/*-----------------------------------------------------------------------*/
// If the table is currently copied to disk we also
// stop already here to avoid strange half-way updates
// of the table data structures.
/*-----------------------------------------------------------------------*/
/*
We need to send this signal without a delay since we have discovered
that we have run out of space in the short time queue. This problem
is very erunlikely to happen but it has and it results in a node crash.
This should be considered a "quick fix" and not a permanent solution.
A cleaner/better way would be to check the time queue if it is full or
not before sending this signal.
*/
sendSignal(reference(), GSN_LCP_FRAG_REP, signal, signal->length(), JBB);
/* Kept here for reference
sendSignalWithDelay(reference(), GSN_LCP_FRAG_REP,
signal, 20, signal->length());
*/
if(!fromTimeQueue){
c_lcpState.noOfLcpFragRepOutstanding++;
}
return;
}//if
if(fromTimeQueue){
jam();
ndbrequire(c_lcpState.noOfLcpFragRepOutstanding > 0);
c_lcpState.noOfLcpFragRepOutstanding--;
}
bool tableDone = reportLcpCompletion(lcpReport);
Uint32 started = lcpReport->maxGciStarted;
Uint32 completed = lcpReport->maxGciCompleted;
if(tableDone){
jam();
if(tabPtr.p->tabStatus == TabRecord::TS_DROPPING){
jam();
ndbout_c("TS_DROPPING - Neglecting to save Table: %d Frag: %d - ",
tableId,
fragId);
} else {
jam();
/**
* Write table description to file
*/
tabPtr.p->tabLcpStatus = TabRecord::TLS_WRITING_TO_FILE;
tabPtr.p->tabCopyStatus = TabRecord::CS_LCP_READ_TABLE;
tabPtr.p->tabUpdateState = TabRecord::US_LOCAL_CHECKPOINT;
signal->theData[0] = DihContinueB::ZPACK_TABLE_INTO_PAGES;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
checkLcpAllTablesDoneInLqh();
}
}
#ifdef VM_TRACE
/* --------------------------------------------------------------------- */
// REPORT that local checkpoint have completed this fragment.
/* --------------------------------------------------------------------- */
signal->theData[0] = EventReport::LCPFragmentCompleted;
signal->theData[1] = nodeId;
signal->theData[2] = tableId;
signal->theData[3] = fragId;
signal->theData[4] = started;
signal->theData[5] = completed;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 6, JBB);
#endif
bool ok = false;
switch(c_lcpMasterTakeOverState.state){
case LMTOS_IDLE:
ok = true;
jam();
/**
* Fall through
*/
break;
case LMTOS_WAIT_EMPTY_LCP: // LCP Take over waiting for EMPTY_LCPCONF
jam();
return;
case LMTOS_WAIT_LCP_FRAG_REP:
jam();
checkEmptyLcpComplete(signal);
return;
case LMTOS_INITIAL:
case LMTOS_ALL_IDLE:
case LMTOS_ALL_ACTIVE:
case LMTOS_LCP_CONCLUDING:
case LMTOS_COPY_ONGOING:
ndbrequire(false);
}
ndbrequire(ok);
/* ----------------------------------------------------------------------- */
// Check if there are more LCP's to start up.
/* ----------------------------------------------------------------------- */
if(isMaster()){
jam();
/**
* Remove from "running" array
*/
NodeRecordPtr nodePtr;
nodePtr.i = nodeId;
ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
const Uint32 outstanding = nodePtr.p->noOfStartedChkpt;
ndbrequire(outstanding > 0);
if(nodePtr.p->startedChkpt[0].tableId != tableId ||
nodePtr.p->startedChkpt[0].fragId != fragId){
jam();
ndbrequire(outstanding > 1);
ndbrequire(nodePtr.p->startedChkpt[1].tableId == tableId);
ndbrequire(nodePtr.p->startedChkpt[1].fragId == fragId);
} else {
jam();
nodePtr.p->startedChkpt[0] = nodePtr.p->startedChkpt[1];
}
nodePtr.p->noOfStartedChkpt--;
checkStartMoreLcp(signal, nodeId);
}
}
bool
Dbdih::checkLcpAllTablesDoneInLqh(){
TabRecordPtr tabPtr;
/**
* Check if finished with all tables
*/
for (tabPtr.i = 0; tabPtr.i < ctabFileSize; tabPtr.i++) {
jam();
ptrAss(tabPtr, tabRecord);
if ((tabPtr.p->tabStatus == TabRecord::TS_ACTIVE) &&
(tabPtr.p->tabLcpStatus == TabRecord::TLS_ACTIVE)) {
jam();
/**
* Nope, not finished with all tables
*/
return false;
}//if
}//for
CRASH_INSERTION2(7026, isMaster());
CRASH_INSERTION2(7017, !isMaster());
c_lcpState.setLcpStatus(LCP_TAB_COMPLETED, __LINE__);
return true;
}
void Dbdih::findReplica(ReplicaRecordPtr& replicaPtr,
Fragmentstore* fragPtrP, Uint32 nodeId)
{
replicaPtr.i = fragPtrP->storedReplicas;
while(replicaPtr.i != RNIL){
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
if (replicaPtr.p->procNode == nodeId) {
jam();
return;
} else {
jam();
replicaPtr.i = replicaPtr.p->nextReplica;
}//if
};
#ifdef VM_TRACE
ndbout_c("Fragment Replica(node=%d) not found", nodeId);
replicaPtr.i = fragPtrP->oldStoredReplicas;
while(replicaPtr.i != RNIL){
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
if (replicaPtr.p->procNode == nodeId) {
jam();
break;
} else {
jam();
replicaPtr.i = replicaPtr.p->nextReplica;
}//if
};
if(replicaPtr.i != RNIL){
ndbout_c("...But was found in oldStoredReplicas");
} else {
ndbout_c("...And wasn't found in oldStoredReplicas");
}
#endif
ndbrequire(false);
}//Dbdih::findReplica()
/**
* Return true if table is all fragment replicas have been checkpointed
* to disk (in all LQHs)
* false otherwise
*/
bool
Dbdih::reportLcpCompletion(const LcpFragRep* lcpReport)
{
Uint32 lcpNo = lcpReport->lcpNo;
Uint32 lcpId = lcpReport->lcpId;
Uint32 maxGciStarted = lcpReport->maxGciStarted;
Uint32 maxGciCompleted = lcpReport->maxGciCompleted;
Uint32 tableId = lcpReport->tableId;
Uint32 fragId = lcpReport->fragId;
Uint32 nodeId = lcpReport->nodeId;
TabRecordPtr tabPtr;
tabPtr.i = tableId;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
ReplicaRecordPtr replicaPtr;
findReplica(replicaPtr, fragPtr.p, nodeId);
ndbrequire(replicaPtr.p->lcpOngoingFlag == true);
if(lcpNo != replicaPtr.p->nextLcp){
ndbout_c("lcpNo = %d replicaPtr.p->nextLcp = %d",
lcpNo, replicaPtr.p->nextLcp);
ndbrequire(false);
}
ndbrequire(lcpNo == replicaPtr.p->nextLcp);
ndbrequire(lcpNo < MAX_LCP_STORED);
ndbrequire(replicaPtr.p->lcpId[lcpNo] != lcpId);
replicaPtr.p->lcpIdStarted = lcpId;
replicaPtr.p->lcpOngoingFlag = false;
removeOldCrashedReplicas(replicaPtr);
replicaPtr.p->lcpId[lcpNo] = lcpId;
replicaPtr.p->lcpStatus[lcpNo] = ZVALID;
replicaPtr.p->maxGciStarted[lcpNo] = maxGciStarted;
gth(maxGciStarted + 1, 0);
replicaPtr.p->maxGciCompleted[lcpNo] = maxGciCompleted;
replicaPtr.p->nextLcp = nextLcpNo(replicaPtr.p->nextLcp);
ndbrequire(fragPtr.p->noLcpReplicas > 0);
fragPtr.p->noLcpReplicas --;
if(fragPtr.p->noLcpReplicas > 0){
jam();
return false;
}
for (Uint32 fid = 0; fid < tabPtr.p->totalfragments; fid++) {
jam();
getFragstore(tabPtr.p, fid, fragPtr);
if (fragPtr.p->noLcpReplicas > 0){
jam();
/* ----------------------------------------------------------------- */
// Not all fragments in table have been checkpointed.
/* ----------------------------------------------------------------- */
if(0)
ndbout_c("reportLcpCompletion: fragment %d not ready", fid);
return false;
}//if
}//for
return true;
}//Dbdih::reportLcpCompletion()
void Dbdih::checkStartMoreLcp(Signal* signal, Uint32 nodeId)
{
ndbrequire(isMaster());
NodeRecordPtr nodePtr;
nodePtr.i = nodeId;
ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
ndbrequire(nodePtr.p->noOfStartedChkpt < 2);
if (nodePtr.p->noOfQueuedChkpt > 0) {
jam();
nodePtr.p->noOfQueuedChkpt--;
Uint32 i = nodePtr.p->noOfStartedChkpt;
nodePtr.p->startedChkpt[i] = nodePtr.p->queuedChkpt[0];
nodePtr.p->queuedChkpt[0] = nodePtr.p->queuedChkpt[1];
//-------------------------------------------------------------------
// We can send a LCP_FRAGORD to the node ordering it to perform a
// local checkpoint on this fragment replica.
//-------------------------------------------------------------------
nodePtr.p->noOfStartedChkpt = i + 1;
sendLCP_FRAG_ORD(signal, nodePtr.p->startedChkpt[i]);
}
/* ----------------------------------------------------------------------- */
// When there are no more outstanding LCP reports and there are no one queued
// in at least one node, then we are ready to make sure all nodes have at
// least two outstanding LCP requests per node and at least two queued for
// sending.
/* ----------------------------------------------------------------------- */
startNextChkpt(signal);
}//Dbdih::checkStartMoreLcp()
void
Dbdih::sendLCP_FRAG_ORD(Signal* signal,
NodeRecord::FragmentCheckpointInfo info){
ReplicaRecordPtr replicaPtr;
replicaPtr.i = info.replicaPtr;
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
BlockReference ref = calcLqhBlockRef(replicaPtr.p->procNode);
LcpFragOrd * const lcpFragOrd = (LcpFragOrd *)&signal->theData[0];
lcpFragOrd->tableId = info.tableId;
lcpFragOrd->fragmentId = info.fragId;
lcpFragOrd->lcpId = SYSFILE->latestLCP_ID;
lcpFragOrd->lcpNo = replicaPtr.p->nextLcp;
lcpFragOrd->keepGci = c_lcpState.keepGci;
lcpFragOrd->lastFragmentFlag = false;
sendSignal(ref, GSN_LCP_FRAG_ORD, signal, LcpFragOrd::SignalLength, JBB);
}
void Dbdih::checkLcpCompletedLab(Signal* signal)
{
if(c_lcpState.lcpStatus < LCP_TAB_COMPLETED){
jam();
return;
}
TabRecordPtr tabPtr;
for (tabPtr.i = 0; tabPtr.i < ctabFileSize; tabPtr.i++) {
jam();
ptrAss(tabPtr, tabRecord);
if (tabPtr.p->tabStatus == TabRecord::TS_ACTIVE) {
if (tabPtr.p->tabLcpStatus != TabRecord::TLS_COMPLETED) {
jam();
return;
}//if
}//if
}//for
CRASH_INSERTION2(7027, isMaster());
CRASH_INSERTION2(7018, !isMaster());
if(c_lcpState.lcpStatus == LCP_TAB_COMPLETED){
/**
* We'r done
*/
c_lcpState.setLcpStatus(LCP_TAB_SAVED, __LINE__);
sendLCP_COMPLETE_REP(signal);
return;
}
ndbrequire(c_lcpState.lcpStatus == LCP_TAB_SAVED);
allNodesLcpCompletedLab(signal);
return;
}//Dbdih::checkLcpCompletedLab()
void
Dbdih::sendLCP_COMPLETE_REP(Signal* signal){
jam();
LcpCompleteRep * rep = (LcpCompleteRep*)signal->getDataPtrSend();
rep->nodeId = getOwnNodeId();
rep->lcpId = SYSFILE->latestLCP_ID;
rep->blockNo = DBDIH;
sendSignal(c_lcpState.m_masterLcpDihRef, GSN_LCP_COMPLETE_REP, signal,
LcpCompleteRep::SignalLength, JBB);
}
/*-------------------------------------------------------------------------- */
/* COMP_LCP_ROUND A LQH HAS COMPLETED A LOCAL CHECKPOINT */
/*------------------------------------------------------------------------- */
void Dbdih::execLCP_COMPLETE_REP(Signal* signal)
{
jamEntry();
#if 0
ndbout_c("LCP_COMPLETE_REP");
printLCP_COMPLETE_REP(stdout,
signal->getDataPtr(),
signal->length(), number());
#endif
LcpCompleteRep * rep = (LcpCompleteRep*)signal->getDataPtr();
Uint32 lcpId = rep->lcpId;
Uint32 nodeId = rep->nodeId;
Uint32 blockNo = rep->blockNo;
if(c_lcpMasterTakeOverState.state > LMTOS_WAIT_LCP_FRAG_REP){
jam();
/**
* Don't allow LCP_COMPLETE_REP to arrive during
* LCP master take over
*/
ndbrequire(isMaster());
ndbrequire(blockNo == DBDIH);
sendSignalWithDelay(reference(), GSN_LCP_COMPLETE_REP, signal, 100,
signal->length());
return;
}
ndbrequire(c_lcpState.lcpStatus != LCP_STATUS_IDLE);
switch(blockNo){
case DBLQH:
jam();
c_lcpState.m_LCP_COMPLETE_REP_Counter_LQH.clearWaitingFor(nodeId);
ndbrequire(!c_lcpState.m_LAST_LCP_FRAG_ORD.isWaitingFor(nodeId));
break;
case DBDIH:
jam();
ndbrequire(isMaster());
c_lcpState.m_LCP_COMPLETE_REP_Counter_DIH.clearWaitingFor(nodeId);
break;
case 0:
jam();
ndbrequire(!isMaster());
ndbrequire(c_lcpState.m_LCP_COMPLETE_REP_From_Master_Received == false);
c_lcpState.m_LCP_COMPLETE_REP_From_Master_Received = true;
break;
default:
ndbrequire(false);
}
ndbrequire(lcpId == SYSFILE->latestLCP_ID);
allNodesLcpCompletedLab(signal);
return;
}
void Dbdih::allNodesLcpCompletedLab(Signal* signal)
{
jam();
if (c_lcpState.lcpStatus != LCP_TAB_SAVED) {
jam();
/**
* We have not sent LCP_COMPLETE_REP to master DIH yet
*/
return;
}//if
if (!c_lcpState.m_LCP_COMPLETE_REP_Counter_LQH.done()){
jam();
return;
}
if (!c_lcpState.m_LCP_COMPLETE_REP_Counter_DIH.done()){
jam();
return;
}
if (!isMaster() &&
c_lcpState.m_LCP_COMPLETE_REP_From_Master_Received == false){
jam();
/**
* Wait until master DIH has signaled lcp is complete
*/
return;
}
if(c_lcpMasterTakeOverState.state != LMTOS_IDLE){
jam();
#ifdef VM_TRACE
ndbout_c("Exiting from allNodesLcpCompletedLab");
#endif
return;
}
/*------------------------------------------------------------------------ */
/* WE HAVE NOW COMPLETED A LOCAL CHECKPOINT. WE ARE NOW READY TO WAIT */
/* FOR THE NEXT LOCAL CHECKPOINT. SEND WITHOUT TIME-OUT SINCE IT MIGHT */
/* BE TIME TO START THE NEXT LOCAL CHECKPOINT IMMEDIATELY. */
/* CLEAR BIT 3 OF SYSTEM RESTART BITS TO INDICATE THAT THERE IS NO */
/* LOCAL CHECKPOINT ONGOING. THIS WILL BE WRITTEN AT SOME LATER TIME */
/* DURING A GLOBAL CHECKPOINT. IT IS NOT NECESSARY TO WRITE IT */
/* IMMEDIATELY. WE WILL ALSO CLEAR BIT 2 OF SYSTEM RESTART BITS IF ALL */
/* CURRENTLY ACTIVE NODES COMPLETED THE LOCAL CHECKPOINT. */
/*------------------------------------------------------------------------ */
CRASH_INSERTION(7019);
signal->setTrace(0);
c_lcpState.setLcpStatus(LCP_STATUS_IDLE, __LINE__);
setLcpActiveStatusEnd();
Sysfile::clearLCPOngoing(SYSFILE->systemRestartBits);
if(!isMaster()){
jam();
/**
* We're not master, be content
*/
return;
}
// Send LCP_COMPLETE_REP to all other nodes
// allowing them to set their lcpStatus to LCP_STATUS_IDLE
LcpCompleteRep * rep = (LcpCompleteRep*)signal->getDataPtrSend();
rep->nodeId = getOwnNodeId();
rep->lcpId = SYSFILE->latestLCP_ID;
rep->blockNo = 0; // 0 = Sent from master
NodeRecordPtr nodePtr;
nodePtr.i = cfirstAliveNode;
do {
jam();
ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
if (nodePtr.i != cownNodeId){
BlockReference ref = calcDihBlockRef(nodePtr.i);
sendSignal(ref, GSN_LCP_COMPLETE_REP, signal,
LcpCompleteRep::SignalLength, JBB);
}
nodePtr.i = nodePtr.p->nextNode;
} while (nodePtr.i != RNIL);
jam();
/***************************************************************************/
// Report the event that a local checkpoint has completed.
/***************************************************************************/
signal->theData[0] = EventReport::LocalCheckpointCompleted; //Event type
signal->theData[1] = SYSFILE->latestLCP_ID;
sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
/**
* Start checking for next LCP
*/
checkLcpStart(signal, __LINE__);
if (cwaitLcpSr == true) {
jam();
cwaitLcpSr = false;
ndbsttorry10Lab(signal, __LINE__);
return;
}//if
if (c_nodeStartMaster.blockLcp == true) {
jam();
lcpBlockedLab(signal);
return;
}//if
return;
}//Dbdih::allNodesLcpCompletedLab()
/******************************************************************************/
/* ********** TABLE UPDATE MODULE *************/
/* ****************************************************************************/
/* ------------------------------------------------------------------------- */
/* THIS MODULE IS USED TO UPDATE THE TABLE DESCRIPTION. IT STARTS BY */
/* CREATING THE FIRST TABLE FILE, THEN UPDATES THIS FILE AND CLOSES IT.*/
/* AFTER THAT THE SAME HAPPENS WITH THE SECOND FILE. AFTER THAT THE */
/* TABLE DISTRIBUTION HAS BEEN UPDATED. */
/* */
/* THE REASON FOR CREATING THE FILE AND NOT OPENING IT IS TO ENSURE */
/* THAT WE DO NOT GET A MIX OF OLD AND NEW INFORMATION IN THE FILE IN */
/* ERROR SITUATIONS. */
/* ------------------------------------------------------------------------- */
void Dbdih::tableUpdateLab(Signal* signal, TabRecordPtr tabPtr) {
FileRecordPtr filePtr;
filePtr.i = tabPtr.p->tabFile[0];
ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
createFileRw(signal, filePtr);
filePtr.p->reqStatus = FileRecord::TABLE_CREATE;
return;
}//Dbdih::tableUpdateLab()
void Dbdih::tableCreateLab(Signal* signal, FileRecordPtr filePtr)
{
TabRecordPtr tabPtr;
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
writeTabfile(signal, tabPtr.p, filePtr);
filePtr.p->reqStatus = FileRecord::TABLE_WRITE;
return;
}//Dbdih::tableCreateLab()
void Dbdih::tableWriteLab(Signal* signal, FileRecordPtr filePtr)
{
closeFile(signal, filePtr);
filePtr.p->reqStatus = FileRecord::TABLE_CLOSE;
return;
}//Dbdih::tableWriteLab()
void Dbdih::tableCloseLab(Signal* signal, FileRecordPtr filePtr)
{
TabRecordPtr tabPtr;
tabPtr.i = filePtr.p->tabRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
if (filePtr.i == tabPtr.p->tabFile[0]) {
jam();
filePtr.i = tabPtr.p->tabFile[1];
ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
createFileRw(signal, filePtr);
filePtr.p->reqStatus = FileRecord::TABLE_CREATE;
return;
}//if
switch (tabPtr.p->tabUpdateState) {
case TabRecord::US_LOCAL_CHECKPOINT:
jam();
releaseTabPages(tabPtr.i);
signal->theData[0] = DihContinueB::ZCHECK_LCP_COMPLETED;
sendSignal(reference(), GSN_CONTINUEB, signal, 1, JBB);
tabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
tabPtr.p->tabUpdateState = TabRecord::US_IDLE;
tabPtr.p->tabLcpStatus = TabRecord::TLS_COMPLETED;
return;
break;
case TabRecord::US_REMOVE_NODE:
jam();
releaseTabPages(tabPtr.i);
for (Uint32 fragId = 0; fragId < tabPtr.p->totalfragments; fragId++) {
jam();
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
updateNodeInfo(fragPtr);
}//for
tabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
tabPtr.p->tabUpdateState = TabRecord::US_IDLE;
if (tabPtr.p->tabLcpStatus == TabRecord::TLS_WRITING_TO_FILE) {
jam();
tabPtr.p->tabLcpStatus = TabRecord::TLS_COMPLETED;
signal->theData[0] = DihContinueB::ZCHECK_LCP_COMPLETED;
sendSignal(reference(), GSN_CONTINUEB, signal, 1, JBB);
}//if
signal->theData[0] = DihContinueB::ZREMOVE_NODE_FROM_TABLE;
signal->theData[1] = tabPtr.p->tabRemoveNode;
signal->theData[2] = tabPtr.i + 1;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
break;
case TabRecord::US_INVALIDATE_NODE_LCP:
jam();
releaseTabPages(tabPtr.i);
tabPtr.p->tabCopyStatus = TabRecord::CS_IDLE;
tabPtr.p->tabUpdateState = TabRecord::US_IDLE;
signal->theData[0] = DihContinueB::ZINVALIDATE_NODE_LCP;
signal->theData[1] = tabPtr.p->tabRemoveNode;
signal->theData[2] = tabPtr.i + 1;
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
return;
case TabRecord::US_COPY_TAB_REQ:
jam();
tabPtr.p->tabUpdateState = TabRecord::US_IDLE;
copyTabReq_complete(signal, tabPtr);
return;
break;
case TabRecord::US_ADD_TABLE_MASTER:
jam();
releaseTabPages(tabPtr.i);
tabPtr.p->tabUpdateState = TabRecord::US_IDLE;
signal->theData[0] = DihContinueB::ZDIH_ADD_TABLE_MASTER;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
case TabRecord::US_ADD_TABLE_SLAVE:
jam();
releaseTabPages(tabPtr.i);
tabPtr.p->tabUpdateState = TabRecord::US_IDLE;
signal->theData[0] = DihContinueB::ZDIH_ADD_TABLE_SLAVE;
signal->theData[1] = tabPtr.i;
sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
return;
break;
default:
ndbrequire(false);
return;
break;
}//switch
}//Dbdih::tableCloseLab()
/**
* GCP stop detected,
* send SYSTEM_ERROR to all other alive nodes
*/
void Dbdih::crashSystemAtGcpStop(Signal* signal){
NodeRecordPtr nodePtr;
for (nodePtr.i = 1; nodePtr.i < MAX_NDB_NODES; nodePtr.i++) {
jam();
ptrAss(nodePtr, nodeRecord);
if (nodePtr.p->nodeStatus == NodeRecord::ALIVE) {
jam();
const BlockReference ref =
numberToRef(refToBlock(cntrlblockref), nodePtr.i);
SystemError * const sysErr = (SystemError*)&signal->theData[0];
sysErr->errorCode = SystemError::GCPStopDetected;
sysErr->errorRef = reference();
sysErr->data1 = cgcpStatus;
sysErr->data2 = cgcpOrderBlocked;
sendSignal(ref, GSN_SYSTEM_ERROR, signal,
SystemError::SignalLength, JBA);
}//if
}//for
return;
}//Dbdih::crashSystemAtGcpStop()
/*************************************************************************/
/* */
/* MODULE: ALLOCPAGE */
/* DESCRIPTION: THE SUBROUTINE IS CALLED WITH POINTER TO PAGE */
/* RECORD. A PAGE RECORD IS TAKEN FROM */
/* THE FREE PAGE LIST */
/*************************************************************************/
void Dbdih::allocpage(PageRecordPtr& pagePtr)
{
ndbrequire(cfirstfreepage != RNIL);
pagePtr.i = cfirstfreepage;
ptrCheckGuard(pagePtr, cpageFileSize, pageRecord);
cfirstfreepage = pagePtr.p->nextfreepage;
pagePtr.p->nextfreepage = RNIL;
}//Dbdih::allocpage()
/*************************************************************************/
/* */
/* MODULE: ALLOC_STORED_REPLICA */
/* DESCRIPTION: THE SUBROUTINE IS CALLED TO GET A REPLICA RECORD, */
/* TO INITIALISE IT AND TO LINK IT INTO THE FRAGMENT */
/* STORE RECORD. USED FOR STORED REPLICAS. */
/*************************************************************************/
void Dbdih::allocStoredReplica(FragmentstorePtr fragPtr,
ReplicaRecordPtr& newReplicaPtr,
Uint32 nodeId)
{
Uint32 i;
ReplicaRecordPtr arrReplicaPtr;
ReplicaRecordPtr arrPrevReplicaPtr;
seizeReplicaRec(newReplicaPtr);
for (i = 0; i < MAX_LCP_STORED; i++) {
newReplicaPtr.p->maxGciCompleted[i] = 0;
newReplicaPtr.p->maxGciStarted[i] = 0;
newReplicaPtr.p->lcpId[i] = 0;
newReplicaPtr.p->lcpStatus[i] = ZINVALID;
}//for
newReplicaPtr.p->noCrashedReplicas = 0;
newReplicaPtr.p->initialGci = currentgcp;
for (i = 0; i < 8; i++) {
newReplicaPtr.p->replicaLastGci[i] = (Uint32)-1;
newReplicaPtr.p->createGci[i] = 0;
}//for
newReplicaPtr.p->createGci[0] = currentgcp;
ndbrequire(currentgcp != 0xF1F1F1F1);
newReplicaPtr.p->nextLcp = 0;
newReplicaPtr.p->procNode = nodeId;
newReplicaPtr.p->lcpOngoingFlag = false;
newReplicaPtr.p->lcpIdStarted = 0;
arrPrevReplicaPtr.i = RNIL;
arrReplicaPtr.i = fragPtr.p->storedReplicas;
while (arrReplicaPtr.i != RNIL) {
jam();
ptrCheckGuard(arrReplicaPtr, creplicaFileSize, replicaRecord);
arrPrevReplicaPtr = arrReplicaPtr;
arrReplicaPtr.i = arrReplicaPtr.p->nextReplica;
}//while
if (arrPrevReplicaPtr.i == RNIL) {
jam();
fragPtr.p->storedReplicas = newReplicaPtr.i;
} else {
jam();
arrPrevReplicaPtr.p->nextReplica = newReplicaPtr.i;
}//if
fragPtr.p->noStoredReplicas++;
}//Dbdih::allocStoredReplica()
/*************************************************************************/
/* CALCULATE HOW MANY HOT SPARES THAT ARE TO BE ASSIGNED IN THIS SYSTEM */
/*************************************************************************/
void Dbdih::calculateHotSpare()
{
Uint32 tchsTmp;
Uint32 tchsNoNodes;
switch (cnoReplicas) {
case 1:
jam();
cnoHotSpare = 0;
break;
case 2:
case 3:
case 4:
jam();
if (csystemnodes > cnoReplicas) {
jam();
/* --------------------------------------------------------------------- */
/* WITH MORE NODES THAN REPLICAS WE WILL ALWAYS USE AT LEAST ONE HOT */
/* SPARE IF THAT HAVE BEEN REQUESTED BY THE CONFIGURATION FILE. THE */
/* NUMBER OF NODES TO BE USED FOR NORMAL OPERATION IS ALWAYS */
/* A MULTIPLE OF THE NUMBER OF REPLICAS SINCE WE WILL ORGANISE NODES */
/* INTO NODE GROUPS. THE REMAINING NODES WILL BE HOT SPARE NODES. */
/* --------------------------------------------------------------------- */
if ((csystemnodes - cnoReplicas) >= cminHotSpareNodes) {
jam();
/* --------------------------------------------------------------------- */
// We set the minimum number of hot spares according to users request
// through the configuration file.
/* --------------------------------------------------------------------- */
tchsNoNodes = csystemnodes - cminHotSpareNodes;
cnoHotSpare = cminHotSpareNodes;
} else if (cminHotSpareNodes > 0) {
jam();
/* --------------------------------------------------------------------- */
// The user requested at least one hot spare node and we will support him
// in that.
/* --------------------------------------------------------------------- */
tchsNoNodes = csystemnodes - 1;
cnoHotSpare = 1;
} else {
jam();
/* --------------------------------------------------------------------- */
// The user did not request any hot spare nodes so in this case we will
// only use hot spare nodes if the number of nodes is such that we cannot
// use all nodes as normal nodes.
/* --------------------------------------------------------------------- */
tchsNoNodes = csystemnodes;
cnoHotSpare = 0;
}//if
} else {
jam();
/* --------------------------------------------------------------------- */
// We only have enough to support the replicas. We will not have any hot
// spares.
/* --------------------------------------------------------------------- */
tchsNoNodes = csystemnodes;
cnoHotSpare = 0;
}//if
tchsTmp = tchsNoNodes - (cnoReplicas * (tchsNoNodes / cnoReplicas));
cnoHotSpare = cnoHotSpare + tchsTmp;
break;
default:
jam();
progError(0, 0);
break;
}//switch
}//Dbdih::calculateHotSpare()
/*************************************************************************/
/* CHECK IF THE NODE CRASH IS TO ESCALATE INTO A SYSTEM CRASH. WE COULD */
/* DO THIS BECAUSE ALL REPLICAS OF SOME FRAGMENT ARE LOST. WE COULD ALSO */
/* DO IT AFTER MANY NODE FAILURES THAT MAKE IT VERY DIFFICULT TO RESTORE */
/* DATABASE AFTER A SYSTEM CRASH. IT MIGHT EVEN BE IMPOSSIBLE AND THIS */
/* MUST BE AVOIDED EVEN MORE THAN AVOIDING SYSTEM CRASHES. */
/*************************************************************************/
void Dbdih::checkEscalation()
{
Uint32 TnodeGroup[MAX_NDB_NODES];
NodeRecordPtr nodePtr;
Uint32 i;
for (i = 0; i < MAX_NDB_NODES; i++) {
TnodeGroup[i] = ZFALSE;
}//for
for (nodePtr.i = 1; nodePtr.i < MAX_NDB_NODES; nodePtr.i++) {
jam();
ptrAss(nodePtr, nodeRecord);
if (nodePtr.p->nodeStatus == NodeRecord::ALIVE &&
nodePtr.p->activeStatus == Sysfile::NS_Active){
ndbrequire(nodePtr.p->nodeGroup < MAX_NDB_NODES);
TnodeGroup[nodePtr.p->nodeGroup] = ZTRUE;
}
}
for (i = 0; i < cnoOfNodeGroups; i++) {
jam();
if (TnodeGroup[i] == ZFALSE) {
jam();
progError(__LINE__, ERR_SYSTEM_ERROR, "Lost node group");
}//if
}//for
}//Dbdih::checkEscalation()
/*************************************************************************/
/* */
/* MODULE: CHECK_KEEP_GCI */
/* DESCRIPTION: CHECK FOR MINIMUM GCI RESTORABLE WITH NEW LOCAL */
/* CHECKPOINT. */
/*************************************************************************/
void Dbdih::checkKeepGci(Uint32 replicaStartIndex)
{
ReplicaRecordPtr ckgReplicaPtr;
ckgReplicaPtr.i = replicaStartIndex;
while (ckgReplicaPtr.i != RNIL) {
jam();
ptrCheckGuard(ckgReplicaPtr, creplicaFileSize, replicaRecord);
Uint32 keepGci;
Uint32 oldestRestorableGci;
findMinGci(ckgReplicaPtr, keepGci, oldestRestorableGci);
if (keepGci < c_lcpState.keepGci) {
jam();
/* ------------------------------------------------------------------- */
/* WE MUST KEEP LOG RECORDS SO THAT WE CAN USE ALL LOCAL CHECKPOINTS */
/* THAT ARE AVAILABLE. THUS WE NEED TO CALCULATE THE MINIMUM OVER ALL */
/* FRAGMENTS. */
/* ------------------------------------------------------------------- */
c_lcpState.keepGci = keepGci;
}//if
if (oldestRestorableGci > c_lcpState.oldestRestorableGci) {
jam();
c_lcpState.oldestRestorableGci = oldestRestorableGci;
ndbrequire(((int)c_lcpState.oldestRestorableGci) >= 0);
}//if
ckgReplicaPtr.i = ckgReplicaPtr.p->nextReplica;
}//while
}//Dbdih::checkKeepGci()
void Dbdih::closeFile(Signal* signal, FileRecordPtr filePtr)
{
signal->theData[0] = filePtr.p->fileRef;
signal->theData[1] = reference();
signal->theData[2] = filePtr.i;
signal->theData[3] = ZCLOSE_NO_DELETE;
sendSignal(NDBFS_REF, GSN_FSCLOSEREQ, signal, 4, JBA);
}//Dbdih::closeFile()
void Dbdih::closeFileDelete(Signal* signal, FileRecordPtr filePtr)
{
signal->theData[0] = filePtr.p->fileRef;
signal->theData[1] = reference();
signal->theData[2] = filePtr.i;
signal->theData[3] = ZCLOSE_DELETE;
sendSignal(NDBFS_REF, GSN_FSCLOSEREQ, signal, 4, JBA);
}//Dbdih::closeFileDelete()
void Dbdih::createFileRw(Signal* signal, FileRecordPtr filePtr)
{
signal->theData[0] = reference();
signal->theData[1] = filePtr.i;
signal->theData[2] = filePtr.p->fileName[0];
signal->theData[3] = filePtr.p->fileName[1];
signal->theData[4] = filePtr.p->fileName[2];
signal->theData[5] = filePtr.p->fileName[3];
signal->theData[6] = ZCREATE_READ_WRITE;
sendSignal(NDBFS_REF, GSN_FSOPENREQ, signal, 7, JBA);
}//Dbdih::createFileRw()
void Dbdih::emptyverificbuffer(Signal* signal, bool aContinueB)
{
if(cfirstVerifyQueue == RNIL){
jam();
return;
}//if
ApiConnectRecordPtr localApiConnectptr;
if(getBlockCommit() == false){
jam();
ndbrequire(cverifyQueueCounter > 0);
cverifyQueueCounter--;
localApiConnectptr.i = cfirstVerifyQueue;
ptrCheckGuard(localApiConnectptr, capiConnectFileSize, apiConnectRecord);
ndbrequire(localApiConnectptr.p->apiGci <= currentgcp);
cfirstVerifyQueue = localApiConnectptr.p->nextApi;
if (cfirstVerifyQueue == RNIL) {
jam();
ndbrequire(cverifyQueueCounter == 0);
clastVerifyQueue = RNIL;
}//if
signal->theData[0] = localApiConnectptr.i;
signal->theData[1] = currentgcp;
sendSignal(clocaltcblockref, GSN_DIVERIFYCONF, signal, 2, JBB);
if (aContinueB == true) {
jam();
//-----------------------------------------------------------------------
// This emptying happened as part of a take-out process by continueb signals.
// This ensures that we will empty the queue eventually. We will also empty
// one item every time we insert one item to ensure that the list doesn't
// grow when it is not blocked.
//-----------------------------------------------------------------------
signal->theData[0] = DihContinueB::ZEMPTY_VERIFY_QUEUE;
sendSignal(reference(), GSN_CONTINUEB, signal, 1, JBB);
}//if
} else {
jam();
//-----------------------------------------------------------------------
// We are blocked so it is no use in continuing the emptying of the
// verify buffer. Whenever the block is removed the emptying will
// restart.
//-----------------------------------------------------------------------
}
return;
}//Dbdih::emptyverificbuffer()
/*----------------------------------------------------------------*/
/* FIND A FREE HOT SPARE IF AVAILABLE AND ALIVE. */
/*----------------------------------------------------------------*/
Uint32 Dbdih::findHotSpare()
{
NodeRecordPtr nodePtr;
for (nodePtr.i = 1; nodePtr.i < MAX_NDB_NODES; nodePtr.i++) {
jam();
ptrAss(nodePtr, nodeRecord);
if (nodePtr.p->nodeStatus == NodeRecord::ALIVE) {
if (nodePtr.p->activeStatus == Sysfile::NS_HotSpare) {
jam();
return nodePtr.i;
}//if
}//if
}//for
return RNIL;
}//Dbdih::findHotSpare()
/*************************************************************************/
/* FIND THE NODES FROM WHICH WE CAN EXECUTE THE LOG TO RESTORE THE */
/* DATA NODE IN A SYSTEM RESTART. */
/*************************************************************************/
bool Dbdih::findLogNodes(CreateReplicaRecord* createReplica,
FragmentstorePtr fragPtr,
Uint32 startGci,
Uint32 stopGci)
{
ConstPtr<ReplicaRecord> flnReplicaPtr;
flnReplicaPtr.i = createReplica->replicaRec;
ptrCheckGuard(flnReplicaPtr, creplicaFileSize, replicaRecord);
/* --------------------------------------------------------------------- */
/* WE START BY CHECKING IF THE DATA NODE CAN HANDLE THE LOG ALL BY */
/* ITSELF. THIS IS THE DESIRED BEHAVIOUR. IF THIS IS NOT POSSIBLE */
/* THEN WE SEARCH FOR THE BEST POSSIBLE NODES AMONG THE NODES THAT */
/* ARE PART OF THIS SYSTEM RESTART. */
/* THIS CAN ONLY BE HANDLED BY THE LAST CRASHED REPLICA. */
/* The condition is that the replica was created before or at the */
/* time of the starting gci, in addition it must have been alive */
/* at the time of the stopping gci. This is checked by two */
/* conditions, the first checks replicaLastGci and the second */
/* checks that it is also smaller than the last gci the node was */
/* involved in. This is necessary to check since createGci is set */
/* Last + 1 and sometimes startGci = stopGci + 1 and in that case */
/* it could happen that replicaLastGci is set to -1 with CreateGci */
/* set to LastGci + 1. */
/* --------------------------------------------------------------------- */
arrGuard(flnReplicaPtr.p->noCrashedReplicas, 8);
const Uint32 noCrashed = flnReplicaPtr.p->noCrashedReplicas;
if (!(ERROR_INSERTED(7073) || ERROR_INSERTED(7074))&&
(startGci >= flnReplicaPtr.p->createGci[noCrashed]) &&
(stopGci <= flnReplicaPtr.p->replicaLastGci[noCrashed]) &&
(stopGci <= SYSFILE->lastCompletedGCI[flnReplicaPtr.p->procNode])) {
jam();
/* --------------------------------------------------------------------- */
/* WE FOUND ALL THE LOG RECORDS NEEDED IN THE DATA NODE. WE WILL */
/* USE THOSE. */
/* --------------------------------------------------------------------- */
createReplica->noLogNodes = 1;
createReplica->logStartGci[0] = startGci;
createReplica->logStopGci[0] = stopGci;
createReplica->logNodeId[0] = flnReplicaPtr.p->procNode;
return true;
}//if
Uint32 logNode = 0;
do {
Uint32 fblStopGci;
jam();
if(!findBestLogNode(createReplica,
fragPtr,
startGci,
stopGci,
logNode,
fblStopGci)){
jam();
return false;
}
logNode++;
if (fblStopGci >= stopGci) {
jam();
createReplica->noLogNodes = logNode;
return true;
}//if
startGci = fblStopGci + 1;
if (logNode >= 4) { // Why??
jam();
break;
}//if
} while (1);
/* --------------------------------------------------------------------- */
/* IT WAS NOT POSSIBLE TO RESTORE THE REPLICA. THIS CAN EITHER BE */
/* BECAUSE OF LACKING NODES OR BECAUSE OF A REALLY SERIOUS PROBLEM.*/
/* --------------------------------------------------------------------- */
return false;
}//Dbdih::findLogNodes()
/*************************************************************************/
/* FIND THE BEST POSSIBLE LOG NODE TO EXECUTE THE LOG AS SPECIFIED */
/* BY THE INPUT PARAMETERS. WE SCAN THROUGH ALL ALIVE REPLICAS. */
/* THIS MEANS STORED, OLD_STORED */
/*************************************************************************/
bool
Dbdih::findBestLogNode(CreateReplicaRecord* createReplica,
FragmentstorePtr fragPtr,
Uint32 startGci,
Uint32 stopGci,
Uint32 logNode,
Uint32& fblStopGci)
{
ConstPtr<ReplicaRecord> fblFoundReplicaPtr;
ConstPtr<ReplicaRecord> fblReplicaPtr;
/* --------------------------------------------------------------------- */
/* WE START WITH ZERO AS FOUND TO ENSURE THAT FIRST HIT WILL BE */
/* BETTER. */
/* --------------------------------------------------------------------- */
fblStopGci = 0;
fblReplicaPtr.i = fragPtr.p->storedReplicas;
while (fblReplicaPtr.i != RNIL) {
jam();
ptrCheckGuard(fblReplicaPtr, creplicaFileSize, replicaRecord);
if (checkNodeAlive(fblReplicaPtr.p->procNode)) {
jam();
Uint32 fliStopGci = findLogInterval(fblReplicaPtr, startGci);
if (fliStopGci > fblStopGci) {
jam();
fblStopGci = fliStopGci;
fblFoundReplicaPtr = fblReplicaPtr;
}//if
}//if
fblReplicaPtr.i = fblReplicaPtr.p->nextReplica;
}//while
fblReplicaPtr.i = fragPtr.p->oldStoredReplicas;
while (fblReplicaPtr.i != RNIL) {
jam();
ptrCheckGuard(fblReplicaPtr, creplicaFileSize, replicaRecord);
if (checkNodeAlive(fblReplicaPtr.p->procNode)) {
jam();
Uint32 fliStopGci = findLogInterval(fblReplicaPtr, startGci);
if (fliStopGci > fblStopGci) {
jam();
fblStopGci = fliStopGci;
fblFoundReplicaPtr = fblReplicaPtr;
}//if
}//if
fblReplicaPtr.i = fblReplicaPtr.p->nextReplica;
}//while
if (fblStopGci != 0) {
jam();
ndbrequire(logNode < MAX_LOG_EXEC);