dblqhmain.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 DBLQH_C
#include "Dblqh.hpp"
#include <ndb_limits.h>
#include <md5_hash.hpp>

#include <ndb_version.h>
#include <signaldata/TuxBound.hpp>
#include <signaldata/AccScan.hpp>
#include <signaldata/CopyActive.hpp>
#include <signaldata/CopyFrag.hpp>
#include <signaldata/CreateTrig.hpp>
#include <signaldata/DropTrig.hpp>
#include <signaldata/EmptyLcp.hpp>
#include <signaldata/EventReport.hpp>
#include <signaldata/ExecFragReq.hpp>
#include <signaldata/GCPSave.hpp>
#include <signaldata/TcKeyRef.hpp>
#include <signaldata/LqhKey.hpp>
#include <signaldata/NextScan.hpp>
#include <signaldata/NFCompleteRep.hpp>
#include <signaldata/NodeFailRep.hpp>
#include <signaldata/ReadNodesConf.hpp>
#include <signaldata/RelTabMem.hpp>
#include <signaldata/ScanFrag.hpp>
#include <signaldata/SrFragidConf.hpp>
#include <signaldata/StartFragReq.hpp>
#include <signaldata/StartRec.hpp>
#include <signaldata/TupKey.hpp>
#include <signaldata/TupCommit.hpp>
#include <signaldata/LqhFrag.hpp>
#include <signaldata/AccFrag.hpp>
#include <signaldata/TupFrag.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <signaldata/PackedSignal.hpp>

#include <signaldata/PrepDropTab.hpp>
#include <signaldata/DropTab.hpp>

#include <signaldata/AlterTab.hpp>

#include <signaldata/LCP.hpp>
#include <KeyDescriptor.hpp>

// Use DEBUG to print messages that should be
// seen only when we debug the product
#ifdef VM_TRACE
#define DEBUG(x) ndbout << "DBLQH: "<< x << endl;
NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::TransactionState state){
  out << (int)state;
  return out;
}

NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::LogWriteState state){
  out << (int)state;
  return out;
}

NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::ListState state){
  out << (int)state;
  return out;
}

NdbOut &
operator<<(NdbOut& out, Dblqh::TcConnectionrec::AbortState state){
  out << (int)state;
  return out;
}

NdbOut &
operator<<(NdbOut& out, Dblqh::ScanRecord::ScanState state){
  out << (int)state;
  return out;
}

NdbOut &
operator<<(NdbOut& out, Dblqh::LogFileOperationRecord::LfoState state){
  out << (int)state;
  return out;
}

NdbOut &
operator<<(NdbOut& out, Dblqh::ScanRecord::ScanType state){
  out << (int)state;
  return out;
}

#else
#define DEBUG(x)
#endif

//#define MARKER_TRACE 1
//#define TRACE_SCAN_TAKEOVER 1

const Uint32 NR_ScanNo = 0;

void Dblqh::execACC_COM_BLOCK(Signal* signal)
{
  jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in ACC is in critical sector of being full.
/* ------------------------------------------------------------------------- */
  cCounterAccCommitBlocked++;
  caccCommitBlocked = true;
  cCommitBlocked = true;
  return;
}//Dblqh::execACC_COM_BLOCK()

void Dblqh::execACC_COM_UNBLOCK(Signal* signal)
{
  jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in ACC ok again.
/* ------------------------------------------------------------------------- */
  caccCommitBlocked = false;
  if (ctupCommitBlocked == false) {
    jam();
    cCommitBlocked = false;
  }//if
  return;
}//Dblqh::execACC_COM_UNBLOCK()

void Dblqh::execTUP_COM_BLOCK(Signal* signal)
{
  jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in TUP is in critical sector of being full.
/* ------------------------------------------------------------------------- */
  cCounterTupCommitBlocked++;
  ctupCommitBlocked = true;
  cCommitBlocked = true;
  return;
}//Dblqh::execTUP_COM_BLOCK()

void Dblqh::execTUP_COM_UNBLOCK(Signal* signal)
{
  jamEntry();
/* ------------------------------------------------------------------------- */
// Undo log buffer in TUP ok again.
/* ------------------------------------------------------------------------- */
  ctupCommitBlocked = false;
  if (caccCommitBlocked == false) {
    jam();
    cCommitBlocked = false;
  }//if
  return;
}//Dblqh::execTUP_COM_UNBLOCK()

/* ------------------------------------------------------------------------- */
/* -------               SEND SYSTEM ERROR                           ------- */
/*                                                                           */
/* ------------------------------------------------------------------------- */
void Dblqh::systemError(Signal* signal, int line)
{
  signal->theData[0] = 2304;
  execDUMP_STATE_ORD(signal);
  progError(line, NDBD_EXIT_NDBREQUIRE);
}//Dblqh::systemError()

/* *************** */
/*  ACCSEIZEREF  > */
/* *************** */
void Dblqh::execACCSEIZEREF(Signal* signal) 
{
  jamEntry();
  ndbrequire(false);
}//Dblqh::execACCSEIZEREF()

/* ******************************************************>> */
/* THIS SIGNAL IS USED TO HANDLE REAL-TIME                  */
/* BREAKS THAT ARE NECESSARY TO ENSURE REAL-TIME            */
/* OPERATION OF LQH.                                        */
/* This signal is also used for signal loops, for example   */
/* the timeout handling for writing logs every second.      */
/* ******************************************************>> */
void Dblqh::execCONTINUEB(Signal* signal) 
{
  jamEntry();
  Uint32 tcase = signal->theData[0];
  Uint32 data0 = signal->theData[1];
  Uint32 data1 = signal->theData[2];
  Uint32 data2 = signal->theData[3];
#if 0
  if (tcase == RNIL) {
    tcConnectptr.i = data0;
    ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
    ndbout << "State = " << tcConnectptr.p->transactionState;
    ndbout << " seqNoReplica = " << tcConnectptr.p->seqNoReplica;
    ndbout << " tcNodeFailrec = " << tcConnectptr.p->tcNodeFailrec;
    ndbout << " activeCreat = " << tcConnectptr.p->activeCreat;
    ndbout << endl;
    ndbout << "tupkeyData0 = " << tcConnectptr.p->tupkeyData[0];
    ndbout << "tupkeyData1 = " << tcConnectptr.p->tupkeyData[1];
    ndbout << "tupkeyData2 = " << tcConnectptr.p->tupkeyData[2];
    ndbout << "tupkeyData3 = " << tcConnectptr.p->tupkeyData[3];
    ndbout << endl;
    ndbout << "abortState = " << tcConnectptr.p->abortState;
    ndbout << "listState = " << tcConnectptr.p->listState;
    ndbout << endl;
    return;
  }//if
#endif
  switch (tcase) {
  case ZLOG_LQHKEYREQ:
    if (cnoOfLogPages == 0) {
      jam();
      sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 10, 2);
      return;
    }//if
    logPartPtr.i = data0;
    ptrCheckGuard(logPartPtr, clogPartFileSize, logPartRecord);
    logFilePtr.i = logPartPtr.p->currentLogfile;
    ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
    logPagePtr.i = logFilePtr.p->currentLogpage;
    ptrCheckGuard(logPagePtr, clogPageFileSize, logPageRecord);

    tcConnectptr.i = logPartPtr.p->firstLogQueue;
    ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
    fragptr.i = tcConnectptr.p->fragmentptr;
    ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
    if ((cCommitBlocked == true) &&
        (fragptr.p->fragActiveStatus == ZTRUE)) {
      jam();
      sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 10, 2);
      return;
    }//if
    logPartPtr.p->LogLqhKeyReqSent = ZFALSE;
    getFirstInLogQueue(signal);

    switch (tcConnectptr.p->transactionState) {
    case TcConnectionrec::LOG_QUEUED:
      if (tcConnectptr.p->abortState != TcConnectionrec::ABORT_IDLE) {
        jam();
        logNextStart(signal);
        abortCommonLab(signal);
        return;
      } else {
        jam();
/*------------------------------------------------------------*/
/*       WE MUST SET THE STATE OF THE LOG PART TO IDLE TO     */
/*       ENSURE THAT WE ARE NOT QUEUED AGAIN ON THE LOG PART  */
/*       WE WILL SET THE LOG PART STATE TO ACTIVE IMMEDIATELY */
/*       SO NO OTHER PROCESS WILL SEE THIS STATE. IT IS MERELY*/
/*       USED TO ENABLE REUSE OF CODE.                        */
/*------------------------------------------------------------*/
        if (logPartPtr.p->logPartState == LogPartRecord::ACTIVE) {
          jam();
          logPartPtr.p->logPartState = LogPartRecord::IDLE;
        }//if
        logLqhkeyreqLab(signal);
        return;
      }//if
      break;
    case TcConnectionrec::LOG_ABORT_QUEUED:
      jam();
      writeAbortLog(signal);
      removeLogTcrec(signal);
      logNextStart(signal);
      continueAfterLogAbortWriteLab(signal);
      return;
      break;
    case TcConnectionrec::LOG_COMMIT_QUEUED:
    case TcConnectionrec::LOG_COMMIT_QUEUED_WAIT_SIGNAL:
      jam();
      writeCommitLog(signal, logPartPtr);
      logNextStart(signal);
      if (tcConnectptr.p->transactionState == TcConnectionrec::LOG_COMMIT_QUEUED) {
        if (tcConnectptr.p->seqNoReplica != 0) {
          jam();
          commitReplyLab(signal);
        } else {
          jam();
          localCommitLab(signal);
        }//if
        return;
      } else {
        jam();
        tcConnectptr.p->transactionState = TcConnectionrec::LOG_COMMIT_WRITTEN_WAIT_SIGNAL;
        return;
      }//if
      break;
    case TcConnectionrec::COMMIT_QUEUED:
      jam();
      logNextStart(signal);
      localCommitLab(signal);
      break;
    case TcConnectionrec::ABORT_QUEUED:
      jam();
      logNextStart(signal);
      abortCommonLab(signal);
      break;
    default:
      ndbrequire(false);
      break;
    }//switch
    return;
    break;
  case ZSR_GCI_LIMITS:
    jam();
    signal->theData[0] = data0;
    srGciLimits(signal);
    return;
    break;
  case ZSR_LOG_LIMITS:
    jam();
    signal->theData[0] = data0;
    signal->theData[1] = data1;
    signal->theData[2] = data2;
    srLogLimits(signal);
    return;
    break;
  case ZSEND_EXEC_CONF:
    jam();
    signal->theData[0] = data0;
    sendExecConf(signal);
    return;
    break;
  case ZEXEC_SR:
    jam();
    signal->theData[0] = data0;
    execSr(signal);
    return;
    break;
  case ZSR_FOURTH_COMP:
    jam();
    signal->theData[0] = data0;
    srFourthComp(signal);
    return;
    break;
  case ZINIT_FOURTH:
    jam();
    signal->theData[0] = data0;
    initFourth(signal);
    return;
    break;
  case ZTIME_SUPERVISION:
    jam();
    signal->theData[0] = data0;
    timeSup(signal);
    return;
    break;
  case ZSR_PHASE3_START:
    jam();
    signal->theData[0] = data0;
    srPhase3Start(signal);
    return;
    break;
  case ZLQH_TRANS_NEXT:
    jam();
    tcNodeFailptr.i = data0;
    ptrCheckGuard(tcNodeFailptr, ctcNodeFailrecFileSize, tcNodeFailRecord);
    lqhTransNextLab(signal);
    return;
    break;
  case ZSCAN_TC_CONNECT:
    jam();
    tabptr.i = data1;
    ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
    scanTcConnectLab(signal, data0, data2);
    return;
    break;
  case ZINITIALISE_RECORDS:
    jam();
    initialiseRecordsLab(signal, data0, data2, signal->theData[4]);
    return;
    break;
  case ZINIT_GCP_REC:
    jam();
    gcpPtr.i = 0;
    ptrAss(gcpPtr, gcpRecord);
    initGcpRecLab(signal);
    return;