dblqhmain.cpp

来自「MySQL源码文件5.X系列, 可自已编译到服务器」· C++ 代码 · 共 1,915 行 · 第 1/5 页

  case ZINIT_GCP_REC:
    jam();
    gcpPtr.i = 0;
    ptrAss(gcpPtr, gcpRecord);
    initGcpRecLab(signal);
    return;
    break;
  case ZRESTART_OPERATIONS_AFTER_STOP:
    jam();
    tcConnectptr.i = data0;
    ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
    if (tcConnectptr.p->listState != TcConnectionrec::WAIT_QUEUE_LIST) {
      jam();
      return;
    }//if
    releaseWaitQueue(signal);
    linkActiveFrag(signal);
    restartOperationsAfterStopLab(signal);
    return;
    break;
  case ZCHECK_LCP_STOP_BLOCKED:
    jam();
    c_scanRecordPool.getPtr(scanptr, data0);
    tcConnectptr.i = scanptr.p->scanTcrec;
    ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
    fragptr.i = tcConnectptr.p->fragmentptr;
    ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
    checkLcpStopBlockedLab(signal);
    return;
  case ZSCAN_MARKERS:
    jam();
    scanMarkers(signal, data0, data1, data2);
    return;
    break;

  case ZOPERATION_EVENT_REP:
    jam();
    /* --------------------------------------------------------------------- */
    // Report information about transaction activity once per second.
    /* --------------------------------------------------------------------- */
    if (signal->theData[1] == 0) {
      signal->theData[0] = NDB_LE_OperationReportCounters;
      signal->theData[1] = c_Counters.operations;
      sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
    }//if
    c_Counters.clear();
    signal->theData[0] = ZOPERATION_EVENT_REP;
    signal->theData[1] = 0;
    sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 5000, 2);
    break;
  case ZPREP_DROP_TABLE:
    jam();
    checkDropTab(signal);
    return;
    break;
  case ZENABLE_EXPAND_CHECK:
  {
    jam();
    fragptr.i = signal->theData[1];
    if (fragptr.i != RNIL)
    {
      jam();
      ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
      signal->theData[0] = fragptr.p->tabRef;
      signal->theData[1] = fragptr.p->fragId;
      sendSignal(DBACC_REF, GSN_EXPANDCHECK2, signal, 2, JBB);
      
      signal->theData[0] = ZENABLE_EXPAND_CHECK;
      signal->theData[1] = fragptr.p->nextFrag;
      sendSignal(DBLQH_REF, GSN_CONTINUEB, signal, 2, JBB);	
      return;
    }
    else
    {
      jam();
      StartRecConf * conf = (StartRecConf*)signal->getDataPtrSend();
      conf->startingNodeId = getOwnNodeId();
      sendSignal(cmasterDihBlockref, GSN_START_RECCONF, signal, 
		 StartRecConf::SignalLength, JBB);
      return;
    }
  }
  case ZRETRY_TCKEYREF:
  {
    jam();
    Uint32 cnt = signal->theData[1];
    Uint32 ref = signal->theData[2];
    if (cnt < (10 * 60 * 5))
    {
      jam();
      /**
       * Only retry for 5 minutes...then hope that API has handled it..somehow
       */
      memmove(signal->theData, signal->theData+3, 4*TcKeyRef::SignalLength);
      sendTCKEYREF(signal, ref, 0, cnt);
    }
    return;
  }
  default:
    ndbrequire(false);
    break;
  }//switch
}//Dblqh::execCONTINUEB()

/* *********************************************************> */
/*  Request from DBDIH to include a new node in the node list */
/*  and so forth.                                             */
/* *********************************************************> */
void Dblqh::execINCL_NODEREQ(Signal* signal) 
{
  jamEntry();
  BlockReference retRef = signal->theData[0];
  Uint32 nodeId = signal->theData[1];
  cnewestGci = signal->theData[2];
  cnewestCompletedGci = signal->theData[2] - 1;
  ndbrequire(cnoOfNodes < MAX_NDB_NODES);
  for (Uint32 i = 0; i < cnoOfNodes; i++) {
    jam();
    if (cnodeData[i] == nodeId) {
      jam();
      cnodeStatus[i] = ZNODE_UP;
    }//if
  }//for
  signal->theData[0] = cownref; 
  sendSignal(retRef, GSN_INCL_NODECONF, signal, 1, JBB);
  return;
}//Dblqh::execINCL_NODEREQ()

void Dblqh::execTUPSEIZEREF(Signal* signal) 
{
  jamEntry();
  ndbrequire(false);
}//Dblqh::execTUPSEIZEREF()

/* ########################################################################## */
/* #######                  START / RESTART MODULE                    ####### */
/* ########################################################################## */
/* ************************************************************************>> */
/*  This is first signal that arrives in a start / restart. Sender is NDBCNTR_REF. */
/* ************************************************************************>> */
void Dblqh::execSTTOR(Signal* signal) 
{
  UintR tstartPhase;

  jamEntry();
                                                  /* START CASE */
  tstartPhase = signal->theData[1];
                                                  /* SYSTEM RESTART RANK */
  csignalKey = signal->theData[6];
  switch (tstartPhase) {
  case ZSTART_PHASE1:
    jam();
    cstartPhase = tstartPhase;
    sttorStartphase1Lab(signal);
    c_tup = (Dbtup*)globalData.getBlock(DBTUP);
    ndbrequire(c_tup != 0);
    return;
    break;
  default:
    jam();
    /*empty*/;
    sendsttorryLab(signal);
    return;
    break;
  }//switch
}//Dblqh::execSTTOR()

/* ***************************************> */
/*  Restart phases 1 - 6, sender is Ndbcntr */
/* ***************************************> */
void Dblqh::execNDB_STTOR(Signal* signal) 
{
  jamEntry();
  Uint32 ownNodeId = signal->theData[1];   /* START PHASE*/
  cstartPhase = signal->theData[2];  /* MY NODE ID */
  cstartType = signal->theData[3];   /* START TYPE */

  switch (cstartPhase) {
  case ZSTART_PHASE1:
    jam();
    preComputedRequestInfoMask = 0;
    LqhKeyReq::setKeyLen(preComputedRequestInfoMask, RI_KEYLEN_MASK);
    LqhKeyReq::setLastReplicaNo(preComputedRequestInfoMask, RI_LAST_REPL_MASK);
    LqhKeyReq::setLockType(preComputedRequestInfoMask, RI_LOCK_TYPE_MASK);
    // Dont LqhKeyReq::setApplicationAddressFlag
    LqhKeyReq::setDirtyFlag(preComputedRequestInfoMask, 1);
    // Dont LqhKeyReq::setInterpretedFlag
    LqhKeyReq::setSimpleFlag(preComputedRequestInfoMask, 1);
    LqhKeyReq::setOperation(preComputedRequestInfoMask, RI_OPERATION_MASK);
    // Dont setAIInLqhKeyReq
    // Dont setSeqNoReplica
    // Dont setSameClientAndTcFlag
    // Dont setReturnedReadLenAIFlag
    // Dont setAPIVersion
    LqhKeyReq::setMarkerFlag(preComputedRequestInfoMask, 1);
    //preComputedRequestInfoMask = 0x003d7fff;
    startphase1Lab(signal, /* dummy */ ~0, ownNodeId);

    signal->theData[0] = ZOPERATION_EVENT_REP;
    signal->theData[1] = 1;
    sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 10, 2);
    return;
    break;
  case ZSTART_PHASE2:
    jam();
    startphase2Lab(signal, /* dummy */ ~0);
    return;
    break;
  case ZSTART_PHASE3:
    jam();
    startphase3Lab(signal);
    return;
    break;
  case ZSTART_PHASE4:
    jam();
    startphase4Lab(signal);
    return;
    break;
  case ZSTART_PHASE6:
    jam();
    startphase6Lab(signal);
    return;
    break;
  default:
    jam();
    /*empty*/;
    sendNdbSttorryLab(signal);
    return;
    break;
  }//switch
}//Dblqh::execNDB_STTOR()

/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++                     START PHASE 1                          +++++++ */
/*               LOAD OUR BLOCK REFERENCE AND OUR PROCESSOR ID                */
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::sttorStartphase1Lab(Signal* signal) 
{
  sendsttorryLab(signal);
  return;
}//Dblqh::sttorStartphase1Lab()

/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++                           START PHASE 2                    +++++++ */
/*                                                                            */
/*               INITIATE ALL RECORDS WITHIN THE BLOCK                        */
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::startphase1Lab(Signal* signal, Uint32 _dummy, Uint32 ownNodeId) 
{
  UintR Ti;
  HostRecordPtr ThostPtr;

/* ------- INITIATE ALL RECORDS ------- */
  cownNodeid    = ownNodeId;
  caccBlockref  = calcAccBlockRef (cownNodeid);
  ctupBlockref  = calcTupBlockRef (cownNodeid);
  ctuxBlockref  = calcTuxBlockRef (cownNodeid);
  cownref       = calcLqhBlockRef (cownNodeid);
  for (Ti = 0; Ti < chostFileSize; Ti++) {
    ThostPtr.i = Ti;
    ptrCheckGuard(ThostPtr, chostFileSize, hostRecord);
    ThostPtr.p->hostLqhBlockRef = calcLqhBlockRef(ThostPtr.i);
    ThostPtr.p->hostTcBlockRef  = calcTcBlockRef(ThostPtr.i);
    ThostPtr.p->inPackedList = false;
    ThostPtr.p->noOfPackedWordsLqh = 0;
    ThostPtr.p->noOfPackedWordsTc  = 0;
  }//for
  cpackedListIndex = 0;
  sendNdbSttorryLab(signal);
  return;
}//Dblqh::startphase1Lab()

/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++                           START PHASE 2                    +++++++ */
/*                                                                            */
/* CONNECT LQH WITH ACC AND TUP.                                              */
/* EVERY CONNECTION RECORD IN LQH IS ASSIGNED TO ONE ACC CONNECTION RECORD    */
/*       AND ONE TUP CONNECTION RECORD.                                       */
/* ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::startphase2Lab(Signal* signal, Uint32 _dummy) 
{
  cmaxWordsAtNodeRec = MAX_NO_WORDS_OUTSTANDING_COPY_FRAGMENT;
/* -- ACC AND TUP CONNECTION PROCESS -- */
  tcConnectptr.i = 0;
  ptrAss(tcConnectptr, tcConnectionrec);
  moreconnectionsLab(signal);
  return;
}//Dblqh::startphase2Lab()

void Dblqh::moreconnectionsLab(Signal* signal) 
{
  tcConnectptr.p->tcAccBlockref = caccBlockref;
  // set TUX block here (no operation is seized in TUX)
  tcConnectptr.p->tcTuxBlockref = ctuxBlockref;
/* NO STATE CHECKING IS PERFORMED, ASSUMED TO WORK */
/* *************** */
/*  ACCSEIZEREQ  < */
/* *************** */
  signal->theData[0] = tcConnectptr.i;
  signal->theData[1] = cownref;
  sendSignal(caccBlockref, GSN_ACCSEIZEREQ, signal, 2, JBB);
  return;
}//Dblqh::moreconnectionsLab()

/* ***************> */
/*  ACCSEIZECONF  > */
/* ***************> */
void Dblqh::execACCSEIZECONF(Signal* signal) 
{
  jamEntry();
  tcConnectptr.i = signal->theData[0];
  ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
  tcConnectptr.p->accConnectrec = signal->theData[1];
/* *************** */
/*  TUPSEIZEREQ  < */
/* *************** */
  tcConnectptr.p->tcTupBlockref = ctupBlockref;
  signal->theData[0] = tcConnectptr.i;
  signal->theData[1] = cownref;
  sendSignal(ctupBlockref, GSN_TUPSEIZEREQ, signal, 2, JBB);
  return;
}//Dblqh::execACCSEIZECONF()

/* ***************> */
/*  TUPSEIZECONF  > */
/* ***************> */
void Dblqh::execTUPSEIZECONF(Signal* signal) 
{
  jamEntry();
  tcConnectptr.i = signal->theData[0];
  ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
  tcConnectptr.p->tupConnectrec = signal->theData[1];
/* ------- CHECK IF THERE ARE MORE CONNECTIONS TO BE CONNECTED ------- */
  tcConnectptr.i = tcConnectptr.p->nextTcConnectrec;
  if (tcConnectptr.i != RNIL) {
    jam();
    ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
    moreconnectionsLab(signal);
    return;
  }//if
/* ALL LQH_CONNECT RECORDS ARE CONNECTED TO ACC AND TUP ---- */
  sendNdbSttorryLab(signal);
  return;
}//Dblqh::execTUPSEIZECONF()

/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* +++++++                    START PHASE 4                          +++++++ */
/*                                                                           */
/*       CONNECT LQH WITH LQH.                                               */
/*       CONNECT EACH LQH WITH EVERY LQH IN THE DATABASE SYSTEM.             */
/*       IF INITIAL START THEN CREATE THE FRAGMENT LOG FILES                 */
/*IF SYSTEM RESTART OR NODE RESTART THEN OPEN THE FRAGMENT LOG FILES AND     */
/*FIND THE END OF THE LOG FILES.                                             */
/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/*        WAIT UNTIL ADD NODE PROCESSES ARE COMPLETED                        */
/*        IF INITIAL START ALSO WAIT FOR LOG FILES TO INITIALISED            */
/*START TIME SUPERVISION OF LOG FILES. WE HAVE TO WRITE LOG PAGES TO DISK    */
/*EVEN IF THE PAGES ARE NOT FULL TO ENSURE THAT THEY COME TO DISK ASAP.      */
/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
void Dblqh::startphase3Lab(Signal* signal) 
{
  LogFileRecordPtr prevLogFilePtr;
  LogFileRecordPtr zeroLogFilePtr;

  caddNodeState = ZTRUE;
/* ***************<< */
/*  READ_NODESREQ  < */
/* ***************<< */
  cinitialStartOngoing = ZTRUE;
  ndbrequire(cnoLogFiles != 0);

  for (logPartPtr.i = 0; logPartPtr.i < 4; logPartPtr.i++) {
    jam();
    ptrAss(logPartPtr, logPartRecord);
    initLogpart(signal);
    for (Uint32 fileNo = 0; fileNo < cnoLogFiles; fileNo++) {
      seizeLogfile(signal);
      if (fileNo != 0) {
        jam();
        prevLogFilePtr.p->nextLogFile = logFilePtr.i;
        logFilePtr.p->prevLogFile = prevLogFilePtr.i;
      } else {
        jam();