dbdihmain.cpp

mysql-5.0.22.tar.gz源码包

      jam();
      Uint32 tableId = signal->theData[1];
      Uint32 fragId = signal->theData[2];
      calculateKeepGciLab(signal, tableId, fragId);
      return;
    }
  case DihContinueB::ZSTORE_NEW_LCP_ID:
    jam();
    storeNewLcpIdLab(signal);
    return;
  case DihContinueB::ZTABLE_UPDATE:
    {
      TabRecordPtr tabPtr;
      jam();
      tabPtr.i = signal->theData[1];
      ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
      tableUpdateLab(signal, tabPtr);
      return;
    }
  case DihContinueB::ZCHECK_LCP_COMPLETED:
    {
      jam();
      checkLcpCompletedLab(signal);
      return;
    }
  case DihContinueB::ZINIT_LCP:
    {
      jam();
      Uint32 senderRef = signal->theData[1];
      Uint32 tableId = signal->theData[2];
      initLcpLab(signal, senderRef, tableId);
      return;
    }
  case DihContinueB::ZADD_TABLE_MASTER_PAGES:
    {
      TabRecordPtr tabPtr;
      jam();
      tabPtr.i = signal->theData[1];
      ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
      tabPtr.p->tabUpdateState = TabRecord::US_ADD_TABLE_MASTER;
      tableUpdateLab(signal, tabPtr);
      return;
      break;
    }
  case DihContinueB::ZDIH_ADD_TABLE_MASTER:
    {
      jam();
      addTable_closeConf(signal, signal->theData[1]);
      return;
    }
  case DihContinueB::ZADD_TABLE_SLAVE_PAGES:
    {
      TabRecordPtr tabPtr;
      jam();
      tabPtr.i = signal->theData[1];
      ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
      tabPtr.p->tabUpdateState = TabRecord::US_ADD_TABLE_SLAVE;
      tableUpdateLab(signal, tabPtr);
      return;
    }
  case DihContinueB::ZDIH_ADD_TABLE_SLAVE:
    {
      ndbrequire(false);
      return;
    }
  case DihContinueB::ZSTART_GCP:
    jam();
#ifndef NO_GCP
    startGcpLab(signal, signal->theData[1]);
#endif
    return;
    break;
  case DihContinueB::ZCOPY_GCI:{
    jam();
    CopyGCIReq::CopyReason reason = (CopyGCIReq::CopyReason)signal->theData[1];
    ndbrequire(c_copyGCIMaster.m_copyReason == reason);
    sendLoopMacro(COPY_GCIREQ, sendCOPY_GCIREQ);
    return;
  }
    break;
  case DihContinueB::ZEMPTY_VERIFY_QUEUE:
    jam();
    emptyverificbuffer(signal, true);
    return;
    break;
  case DihContinueB::ZCHECK_GCP_STOP:
    jam();
#ifndef NO_GCP
    checkGcpStopLab(signal);
#endif
    return;
    break;
  case DihContinueB::ZREMOVE_NODE_FROM_TABLE:
    {
      jam();
      Uint32 nodeId = signal->theData[1];
      Uint32 tableId = signal->theData[2];
      removeNodeFromTables(signal, nodeId, tableId);
      return;
    }
  case DihContinueB::ZCOPY_NODE:
    {
      jam();
      Uint32 tableId = signal->theData[1];
      copyNodeLab(signal, tableId);
      return;
    }
  case DihContinueB::ZSTART_TAKE_OVER:
    {
      jam();
      Uint32 takeOverPtrI = signal->theData[1];
      Uint32 startNode = signal->theData[2];
      Uint32 toNode = signal->theData[3];
      startTakeOver(signal, takeOverPtrI, startNode, toNode);
      return;
      break;
    }
  case DihContinueB::ZCHECK_START_TAKE_OVER:
    jam();
    checkStartTakeOver(signal);
    break;
  case DihContinueB::ZTO_START_COPY_FRAG:
    {
      jam();
      Uint32 takeOverPtrI = signal->theData[1];
      startNextCopyFragment(signal, takeOverPtrI);
      return;
    }
  case DihContinueB::ZINVALIDATE_NODE_LCP:
    {
      jam();
      const Uint32 nodeId = signal->theData[1];
      const Uint32 tableId = signal->theData[2];
      invalidateNodeLCP(signal, nodeId, tableId);
      return;
    }
  case DihContinueB::ZINITIALISE_RECORDS:
    jam();
    initialiseRecordsLab(signal, 
			 signal->theData[1], 
			 signal->theData[2], 
			 signal->theData[3]);
    return;
    break;
  case DihContinueB::ZSTART_PERMREQ_AGAIN:
    jam();
    nodeRestartPh2Lab(signal);
    return;
    break;
  case DihContinueB::SwitchReplica:
    {
      jam();
      const Uint32 nodeId = signal->theData[1];
      const Uint32 tableId = signal->theData[2];
      const Uint32 fragNo = signal->theData[3];
      switchReplica(signal, nodeId, tableId, fragNo);
      return;
    }
  case DihContinueB::ZSEND_START_TO:
    {
      jam();
      Uint32 takeOverPtrI = signal->theData[1];
      sendStartTo(signal, takeOverPtrI);
      return;
    }
  case DihContinueB::ZSEND_ADD_FRAG:
    {
      jam();
      Uint32 takeOverPtrI = signal->theData[1];
      toCopyFragLab(signal, takeOverPtrI);
      return;
    }
  case DihContinueB::ZSEND_UPDATE_TO:
    {
      jam();
      Uint32 takeOverPtrI = signal->theData[1];
      Uint32 updateState = signal->theData[4];
      sendUpdateTo(signal, takeOverPtrI, updateState);
      return;
    }
  case DihContinueB::ZSEND_END_TO:
    {
      jam();
      Uint32 takeOverPtrI = signal->theData[1];
      sendEndTo(signal, takeOverPtrI);
      return;
    }
  case DihContinueB::ZSEND_CREATE_FRAG:
    {
      jam();
      Uint32 takeOverPtrI = signal->theData[1];
      Uint32 storedType = signal->theData[2];
      Uint32 startGci = signal->theData[3];
      sendCreateFragReq(signal, startGci, storedType, takeOverPtrI);
      return;
    }
  case DihContinueB::WAIT_DROP_TAB_WRITING_TO_FILE:{
    jam();
    TabRecordPtr tabPtr;
    tabPtr.i = signal->theData[1];
    ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
    waitDropTabWritingToFile(signal, tabPtr);
    return;
  }
  case DihContinueB::CHECK_WAIT_DROP_TAB_FAILED_LQH:{
    jam();
    Uint32 nodeId = signal->theData[1];
    Uint32 tableId = signal->theData[2];
    checkWaitDropTabFailedLqh(signal, nodeId, tableId);
    return;
  }
  }//switch
  
  ndbrequire(false);
  return;
}//Dbdih::execCONTINUEB()

void Dbdih::execCOPY_GCIREQ(Signal* signal) 
{
  CopyGCIReq * const copyGCI = (CopyGCIReq *)&signal->theData[0];
  jamEntry();
  CopyGCIReq::CopyReason reason = (CopyGCIReq::CopyReason)copyGCI->copyReason;
  const Uint32 tstart = copyGCI->startWord;
  
  ndbrequire(cmasterdihref == signal->senderBlockRef()) ;
  ndbrequire(c_copyGCISlave.m_copyReason  == CopyGCIReq::IDLE);
  ndbrequire(c_copyGCISlave.m_expectedNextWord == tstart);
  ndbrequire(reason != CopyGCIReq::IDLE);
  
  arrGuard(tstart + CopyGCIReq::DATA_SIZE, sizeof(sysfileData)/4);
  for(Uint32 i = 0; i<CopyGCIReq::DATA_SIZE; i++)
    cdata[tstart+i] = copyGCI->data[i];
  
  if ((tstart + CopyGCIReq::DATA_SIZE) >= Sysfile::SYSFILE_SIZE32) {
    jam();
    c_copyGCISlave.m_expectedNextWord = 0;
  } else {
    jam();
    c_copyGCISlave.m_expectedNextWord += CopyGCIReq::DATA_SIZE;
    return;
  }//if
  
  memcpy(sysfileData, cdata, sizeof(sysfileData));
  
  c_copyGCISlave.m_copyReason = reason;
  c_copyGCISlave.m_senderRef  = signal->senderBlockRef();
  c_copyGCISlave.m_senderData = copyGCI->anyData;

  CRASH_INSERTION2(7020, reason==CopyGCIReq::LOCAL_CHECKPOINT);
  CRASH_INSERTION2(7008, reason==CopyGCIReq::GLOBAL_CHECKPOINT);

  /* -------------------------------------------------------------------------*/
  /*     WE SET THE REQUESTER OF THE COPY GCI TO THE CURRENT MASTER. IF THE   */
  /*     CURRENT MASTER WE DO NOT WANT THE NEW MASTER TO RECEIVE CONFIRM OF   */
  /*     SOMETHING HE HAS NOT SENT. THE TAKE OVER MUST BE CAREFUL.            */
  /* -------------------------------------------------------------------------*/
  bool ok = false;
  switch(reason){
  case CopyGCIReq::IDLE:
    ok = true;
    jam();
    ndbrequire(false);
    break;
  case CopyGCIReq::LOCAL_CHECKPOINT: {
    ok = true;
    jam();
    c_lcpState.setLcpStatus(LCP_COPY_GCI, __LINE__);
    c_lcpState.m_masterLcpDihRef = cmasterdihref;
    setNodeInfo(signal);
    break;
  }
  case CopyGCIReq::RESTART: {
    ok = true;
    jam();
    coldgcp = SYSFILE->newestRestorableGCI;
    crestartGci = SYSFILE->newestRestorableGCI;
    c_newest_restorable_gci = SYSFILE->newestRestorableGCI;
    Sysfile::setRestartOngoing(SYSFILE->systemRestartBits);
    currentgcp = coldgcp + 1;
    cnewgcp = coldgcp + 1;
    setNodeInfo(signal);
    if ((Sysfile::getLCPOngoing(SYSFILE->systemRestartBits))) {
      jam();
      /* -------------------------------------------------------------------- */
      //  IF THERE WAS A LOCAL CHECKPOINT ONGOING AT THE CRASH MOMENT WE WILL
      //    INVALIDATE THAT LOCAL CHECKPOINT.
      /* -------------------------------------------------------------------- */
      invalidateLcpInfoAfterSr();
    }//if
    break;
  }
  case CopyGCIReq::GLOBAL_CHECKPOINT: {
    ok = true;
    jam();
    cgcpParticipantState = GCP_PARTICIPANT_COPY_GCI_RECEIVED;
    c_newest_restorable_gci = SYSFILE->newestRestorableGCI;
    setNodeInfo(signal);
    break;
  }//if
  case CopyGCIReq::INITIAL_START_COMPLETED:
    ok = true;
    jam();
    break;
  }
  ndbrequire(ok);
  
  /* ----------------------------------------------------------------------- */
  /*     WE START BY TRYING TO OPEN THE FIRST RESTORABLE GCI FILE.           */
  /* ----------------------------------------------------------------------- */
  FileRecordPtr filePtr;
  filePtr.i = crestartInfoFile[0];
  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;
}//Dbdih::execCOPY_GCIREQ()

void Dbdih::execDICTSTARTCONF(Signal* signal) 
{
  jamEntry();
  Uint32 nodeId = refToNode(signal->getSendersBlockRef());
  if (nodeId != getOwnNodeId()) {
    jam();
    nodeDictStartConfLab(signal);
  } else {
    jam();
    dictStartConfLab(signal);
  }//if
}//Dbdih::execDICTSTARTCONF()

void Dbdih::execFSCLOSECONF(Signal* signal) 
{
  FileRecordPtr filePtr;
  jamEntry();
  filePtr.i = signal->theData[0];
  ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
  filePtr.p->fileStatus = FileRecord::CLOSED;
  FileRecord::ReqStatus status = filePtr.p->reqStatus;
  filePtr.p->reqStatus = FileRecord::IDLE;
  switch (status) {
  case FileRecord::CLOSING_GCP:
    jam();
    closingGcpLab(signal, filePtr);
    break;
  case FileRecord::CLOSING_GCP_CRASH:
    jam();
    closingGcpCrashLab(signal, filePtr);
    break;
  case FileRecord::CLOSING_TABLE_CRASH:
    jam();
    closingTableCrashLab(signal, filePtr);
    break;
  case FileRecord::CLOSING_TABLE_SR:
    jam();
    closingTableSrLab(signal, filePtr);
    break;
  case FileRecord::TABLE_CLOSE:
    jam();
    tableCloseLab(signal, filePtr);
    break;
  case FileRecord::TABLE_CLOSE_DELETE:
    jam();
    tableDeleteLab(signal, filePtr);
    break;
  default:
    ndbrequire(false);
    break;
  }//switch
  return;
}//Dbdih::execFSCLOSECONF()

void Dbdih::execFSCLOSEREF(Signal* signal) 
{
  FileRecordPtr filePtr;
  jamEntry();
  filePtr.i = signal->theData[0];
  ptrCheckGuard(filePtr, cfileFileSize, fileRecord);
  FileRecord::ReqStatus status = filePtr.p->reqStatus;
  filePtr.p->reqStatus = FileRecord::IDLE;
  switch (status) {
  case FileRecord::CLOSING_GCP:
    jam();
    break;
  case FileRecord::CLOSING_GCP_CRASH:
    jam();
    closingGcpCrashLab(signal, filePtr);
    return;
  case FileRecord::CLOSING_TABLE_CRASH:
    jam();
    closingTableCrashLab(signal, filePtr);
    return;
  case FileRecord::CLOSING_TABLE_SR:
    jam();
    break;
  case FileRecord::TABLE_CLOSE: