dbdihmain.cpp

mysql-5.0.22.tar.gz源码包

/* Copyright (C) 2003 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#define DBDIH_C
#include <ndb_limits.h>
#include <ndb_version.h>
#include <NdbOut.hpp>

#include "Dbdih.hpp"
#include "Configuration.hpp"

#include <signaldata/BlockCommitOrd.hpp>
#include <signaldata/CheckNodeGroups.hpp>
#include <signaldata/CreateFrag.hpp>
#include <signaldata/CopyActive.hpp>
#include <signaldata/CopyFrag.hpp>
#include <signaldata/CopyGCIReq.hpp>
#include <signaldata/DiAddTab.hpp>
#include <signaldata/DictStart.hpp>
#include <signaldata/DiGetNodes.hpp>
#include <signaldata/DihContinueB.hpp>
#include <signaldata/DihSwitchReplica.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <signaldata/EmptyLcp.hpp>
#include <signaldata/EndTo.hpp>
#include <signaldata/EventReport.hpp>
#include <signaldata/GCPSave.hpp>
#include <signaldata/HotSpareRep.hpp>
#include <signaldata/MasterGCP.hpp>
#include <signaldata/MasterLCP.hpp>
#include <signaldata/NFCompleteRep.hpp>
#include <signaldata/NodeFailRep.hpp>
#include <signaldata/ReadNodesConf.hpp>
#include <signaldata/StartFragReq.hpp>
#include <signaldata/StartInfo.hpp>
#include <signaldata/StartMe.hpp>
#include <signaldata/StartPerm.hpp>
#include <signaldata/StartRec.hpp>
#include <signaldata/StartTo.hpp>
#include <signaldata/StopPerm.hpp>
#include <signaldata/StopMe.hpp>
#include <signaldata/TestOrd.hpp>
#include <signaldata/UpdateTo.hpp>
#include <signaldata/WaitGCP.hpp>
#include <signaldata/DihStartTab.hpp>
#include <signaldata/LCP.hpp>
#include <signaldata/SystemError.hpp>

#include <signaldata/DropTab.hpp>
#include <signaldata/AlterTab.hpp>
#include <signaldata/PrepDropTab.hpp>
#include <signaldata/SumaImpl.hpp>
#include <signaldata/DictTabInfo.hpp>
#include <signaldata/CreateFragmentation.hpp>
#include <signaldata/LqhFrag.hpp>
#include <signaldata/FsOpenReq.hpp>
#include <DebuggerNames.hpp>

#include <EventLogger.hpp>
extern EventLogger g_eventLogger;

#define SYSFILE ((Sysfile *)&sysfileData[0])

#define RETURN_IF_NODE_NOT_ALIVE(node) \
  if (!checkNodeAlive((node))) { \
    jam(); \
    return; \
  } \

#define RETURN_IF_TAKE_OVER_INTERRUPTED(takeOverIndex, regTOPtr) \
  regTOPtr.i = takeOverIndex; \
  ptrCheckGuard(regTOPtr, MAX_NDB_NODES, takeOverRecord); \
  if (checkToInterrupted(regTOPtr)) { \
    jam(); \
    return; \
  } \

#define receiveLoopMacro(sigName, receiveNodeId)\
{                                                \
  c_##sigName##_Counter.clearWaitingFor(receiveNodeId); \
  if(c_##sigName##_Counter.done() == false){     \
     jam();                                      \
     return;                                     \
  }                                              \
}

#define sendLoopMacro(sigName, signalRoutine)  \
{                                                                       \
  c_##sigName##_Counter.clearWaitingFor();                              \
  NodeRecordPtr specNodePtr;                                            \
  specNodePtr.i = cfirstAliveNode;                                      \
  do {                                                                  \
    jam();                                                              \
    ptrCheckGuard(specNodePtr, MAX_NDB_NODES, nodeRecord);              \
    c_##sigName##_Counter.setWaitingFor(specNodePtr.i);                 \
    signalRoutine(signal, specNodePtr.i);                               \
    specNodePtr.i = specNodePtr.p->nextNode;                            \
  } while (specNodePtr.i != RNIL);                                      \
}

static
Uint32
prevLcpNo(Uint32 lcpNo){
  if(lcpNo == 0)
    return MAX_LCP_STORED - 1;
  return lcpNo - 1;
}

static
Uint32
nextLcpNo(Uint32 lcpNo){
  lcpNo++;
  if(lcpNo == MAX_LCP_STORED)
    return 0;
  return lcpNo;
}

#define gth(x, y) ndbrequire(((int)x)>((int)y))

void Dbdih::nullRoutine(Signal* signal, Uint32 nodeId)
{
}//Dbdih::nullRoutine()

void Dbdih::sendCOPY_GCIREQ(Signal* signal, Uint32 nodeId) 
{
  ndbrequire(c_copyGCIMaster.m_copyReason != CopyGCIReq::IDLE);
  
  const BlockReference ref = calcDihBlockRef(nodeId);
  const Uint32 wordPerSignal = CopyGCIReq::DATA_SIZE;
  const Uint32 noOfSignals = ((Sysfile::SYSFILE_SIZE32 + (wordPerSignal - 1)) /
			      wordPerSignal);
  
  CopyGCIReq * const copyGCI = (CopyGCIReq *)&signal->theData[0];  
  copyGCI->anyData = nodeId;
  copyGCI->copyReason = c_copyGCIMaster.m_copyReason;
  copyGCI->startWord = 0;
  
  for(Uint32 i = 0; i < noOfSignals; i++) {
    jam();
    { // Do copy
      const int startWord = copyGCI->startWord;
      for(Uint32 j = 0; j < wordPerSignal; j++) {
        copyGCI->data[j] = sysfileData[j+startWord];
      }//for
    }
    sendSignal(ref, GSN_COPY_GCIREQ, signal, 25, JBB);
    copyGCI->startWord += wordPerSignal;
  }//for
}//Dbdih::sendCOPY_GCIREQ()


void Dbdih::sendDIH_SWITCH_REPLICA_REQ(Signal* signal, Uint32 nodeId)
{
  const BlockReference ref    = calcDihBlockRef(nodeId);
  sendSignal(ref, GSN_DIH_SWITCH_REPLICA_REQ, signal, 
             DihSwitchReplicaReq::SignalLength, JBB);
}//Dbdih::sendDIH_SWITCH_REPLICA_REQ()

void Dbdih::sendEMPTY_LCP_REQ(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcLqhBlockRef(nodeId);
  sendSignal(ref, GSN_EMPTY_LCP_REQ, signal, EmptyLcpReq::SignalLength, JBB);
}//Dbdih::sendEMPTY_LCPREQ()

void Dbdih::sendEND_TOREQ(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcDihBlockRef(nodeId);
  sendSignal(ref, GSN_END_TOREQ, signal, EndToReq::SignalLength, JBB);
}//Dbdih::sendEND_TOREQ()

void Dbdih::sendGCP_COMMIT(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcDihBlockRef(nodeId);
  signal->theData[0] = cownNodeId;
  signal->theData[1] = cnewgcp;
  sendSignal(ref, GSN_GCP_COMMIT, signal, 2, JBA);
}//Dbdih::sendGCP_COMMIT()

void Dbdih::sendGCP_PREPARE(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcDihBlockRef(nodeId);
  signal->theData[0] = cownNodeId;
  signal->theData[1] = cnewgcp;
  sendSignal(ref, GSN_GCP_PREPARE, signal, 2, JBA);
}//Dbdih::sendGCP_PREPARE()

void Dbdih::sendGCP_SAVEREQ(Signal* signal, Uint32 nodeId)
{
  GCPSaveReq * const saveReq = (GCPSaveReq*)&signal->theData[0];
  BlockReference ref = calcLqhBlockRef(nodeId);
  saveReq->dihBlockRef = reference();
  saveReq->dihPtr = nodeId;
  saveReq->gci = coldgcp;
  sendSignal(ref, GSN_GCP_SAVEREQ, signal, GCPSaveReq::SignalLength, JBB);
}//Dbdih::sendGCP_SAVEREQ()

void Dbdih::sendINCL_NODEREQ(Signal* signal, Uint32 nodeId)
{
  BlockReference nodeDihRef = calcDihBlockRef(nodeId);
  signal->theData[0] = reference();
  signal->theData[1] = c_nodeStartMaster.startNode;
  signal->theData[2] = c_nodeStartMaster.failNr;
  signal->theData[3] = 0;
  signal->theData[4] = currentgcp;  
  sendSignal(nodeDihRef, GSN_INCL_NODEREQ, signal, 5, JBA);
}//Dbdih::sendINCL_NODEREQ()

void Dbdih::sendMASTER_GCPREQ(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcDihBlockRef(nodeId);
  sendSignal(ref, GSN_MASTER_GCPREQ, signal, MasterGCPReq::SignalLength, JBB);
}//Dbdih::sendMASTER_GCPREQ()

void Dbdih::sendMASTER_LCPREQ(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcDihBlockRef(nodeId);
  sendSignal(ref, GSN_MASTER_LCPREQ, signal, MasterLCPReq::SignalLength, JBB);
}//Dbdih::sendMASTER_LCPREQ()

void Dbdih::sendSTART_INFOREQ(Signal* signal, Uint32 nodeId)
{
  const BlockReference ref = calcDihBlockRef(nodeId);
  sendSignal(ref, GSN_START_INFOREQ, signal, StartInfoReq::SignalLength, JBB);
}//sendSTART_INFOREQ()

void Dbdih::sendSTART_RECREQ(Signal* signal, Uint32 nodeId)
{
  StartRecReq * const req = (StartRecReq*)&signal->theData[0];
  BlockReference ref = calcLqhBlockRef(nodeId);
  req->receivingNodeId = nodeId;
  req->senderRef = reference();
  req->keepGci = SYSFILE->keepGCI;
  req->lastCompletedGci = SYSFILE->lastCompletedGCI[nodeId];
  req->newestGci = SYSFILE->newestRestorableGCI;
  sendSignal(ref, GSN_START_RECREQ, signal, StartRecReq::SignalLength, JBB);

  signal->theData[0] = NDB_LE_StartREDOLog;
  signal->theData[1] = nodeId;
  signal->theData[2] = SYSFILE->keepGCI;
  signal->theData[3] = SYSFILE->lastCompletedGCI[nodeId];
  signal->theData[4] = SYSFILE->newestRestorableGCI;
  sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 5, JBB);
}//Dbdih::sendSTART_RECREQ()

void Dbdih::sendSTART_TOREQ(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcDihBlockRef(nodeId);
  sendSignal(ref, GSN_START_TOREQ, signal, StartToReq::SignalLength, JBB);
}//Dbdih::sendSTART_TOREQ()

void Dbdih::sendSTOP_ME_REQ(Signal* signal, Uint32 nodeId)
{
  if (nodeId != getOwnNodeId()) {
    jam();
    const BlockReference ref = calcDihBlockRef(nodeId);
    sendSignal(ref, GSN_STOP_ME_REQ, signal, StopMeReq::SignalLength, JBB);
  }//if
}//Dbdih::sendSTOP_ME_REQ()

void Dbdih::sendTC_CLOPSIZEREQ(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcTcBlockRef(nodeId);
  signal->theData[0] = nodeId;
  signal->theData[1] = reference();
  sendSignal(ref, GSN_TC_CLOPSIZEREQ, signal, 2, JBB);
}//Dbdih::sendTC_CLOPSIZEREQ()

void Dbdih::sendTCGETOPSIZEREQ(Signal* signal, Uint32 nodeId)
{
  BlockReference ref = calcTcBlockRef(nodeId);
  signal->theData[0] = nodeId;
  signal->theData[1] = reference();
  sendSignal(ref, GSN_TCGETOPSIZEREQ, signal, 2, JBB);
}//Dbdih::sendTCGETOPSIZEREQ()

void Dbdih::sendUPDATE_TOREQ(Signal* signal, Uint32 nodeId)
{
  const BlockReference ref = calcDihBlockRef(nodeId);
  sendSignal(ref, GSN_UPDATE_TOREQ, signal, UpdateToReq::SignalLength, JBB);
}//sendUPDATE_TOREQ()

void Dbdih::execCONTINUEB(Signal* signal)
{
  jamEntry();
  switch ((DihContinueB::Type)signal->theData[0]) {
  case DihContinueB::ZPACK_TABLE_INTO_PAGES:
    {
      jam();
      Uint32 tableId = signal->theData[1];
      packTableIntoPagesLab(signal, tableId);
      return;
      break;
    }
  case DihContinueB::ZPACK_FRAG_INTO_PAGES:
    {
      RWFragment wf;
      jam();
      wf.rwfTabPtr.i = signal->theData[1];
      ptrCheckGuard(wf.rwfTabPtr, ctabFileSize, tabRecord);
      wf.fragId = signal->theData[2];
      wf.pageIndex = signal->theData[3];
      wf.wordIndex = signal->theData[4];
      packFragIntoPagesLab(signal, &wf);
      return;
      break;
    }
  case DihContinueB::ZREAD_PAGES_INTO_TABLE:
    {
      jam();
      Uint32 tableId = signal->theData[1];
      readPagesIntoTableLab(signal, tableId);
      return;
      break;
    }
  case DihContinueB::ZREAD_PAGES_INTO_FRAG:
    {
      RWFragment rf;
      jam();
      rf.rwfTabPtr.i = signal->theData[1];
      ptrCheckGuard(rf.rwfTabPtr, ctabFileSize, tabRecord);
      rf.fragId = signal->theData[2];
      rf.pageIndex = signal->theData[3];
      rf.wordIndex = signal->theData[4];
      readPagesIntoFragLab(signal, &rf);
      return;
      break;
    }
  case DihContinueB::ZCOPY_TABLE:
    {
      jam();
      Uint32 tableId = signal->theData[1];
      copyTableLab(signal, tableId);
      return;
    }
  case DihContinueB::ZCOPY_TABLE_NODE:
    {
      NodeRecordPtr nodePtr;
      CopyTableNode ctn;
      jam();
      ctn.ctnTabPtr.i = signal->theData[1];
      ptrCheckGuard(ctn.ctnTabPtr, ctabFileSize, tabRecord);
      nodePtr.i = signal->theData[2];
      ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
      ctn.pageIndex = signal->theData[3];
      ctn.wordIndex = signal->theData[4];
      ctn.noOfWords = signal->theData[5];
      copyTableNode(signal, &ctn, nodePtr);
      return;
    }
  case DihContinueB::ZSTART_FRAGMENT:
    {
      jam();
      Uint32 tableId = signal->theData[1];
      Uint32 fragId = signal->theData[2];
      startFragment(signal, tableId, fragId);
      return;
    }
  case DihContinueB::ZCOMPLETE_RESTART:
    jam();
    completeRestartLab(signal);
    return;
  case DihContinueB::ZREAD_TABLE_FROM_PAGES:
    {
      TabRecordPtr tabPtr;
      jam();
      tabPtr.i = signal->theData[1];
      ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
      readTableFromPagesLab(signal, tabPtr);
      return;
    }
  case DihContinueB::ZSR_PHASE2_READ_TABLE:
    {
      TabRecordPtr tabPtr;
      jam();
      tabPtr.i = signal->theData[1];
      ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
      srPhase2ReadTableLab(signal, tabPtr);
      return;
    }
  case DihContinueB::ZCHECK_TC_COUNTER:
    jam();
#ifndef NO_LCP
    checkTcCounterLab(signal);
#endif
    return;
  case DihContinueB::ZCALCULATE_KEEP_GCI:
    {