dbtcmain.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 DBTC_C

#include "Dbtc.hpp"
#include "md5_hash.hpp"
#include <RefConvert.hpp>
#include <ndb_limits.h>
#include <my_sys.h>

#include <signaldata/EventReport.hpp>
#include <signaldata/TcKeyReq.hpp>
#include <signaldata/TcKeyConf.hpp>
#include <signaldata/TcKeyRef.hpp>
#include <signaldata/KeyInfo.hpp>
#include <signaldata/AttrInfo.hpp>
#include <signaldata/TransIdAI.hpp>
#include <signaldata/TcRollbackRep.hpp>
#include <signaldata/NodeFailRep.hpp>
#include <signaldata/ReadNodesConf.hpp>
#include <signaldata/NFCompleteRep.hpp>
#include <signaldata/LqhKey.hpp>
#include <signaldata/TcCommit.hpp>
#include <signaldata/TcContinueB.hpp>
#include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/AbortAll.hpp>
#include <signaldata/ScanFrag.hpp>
#include <signaldata/ScanTab.hpp>
#include <signaldata/PrepDropTab.hpp>
#include <signaldata/DropTab.hpp>
#include <signaldata/AlterTab.hpp>
#include <signaldata/CreateTrig.hpp>
#include <signaldata/DropTrig.hpp>
#include <signaldata/FireTrigOrd.hpp>
#include <signaldata/TrigAttrInfo.hpp>
#include <signaldata/CreateIndx.hpp>
#include <signaldata/DropIndx.hpp>
#include <signaldata/AlterIndx.hpp>
#include <signaldata/ScanTab.hpp>
#include <signaldata/SystemError.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <signaldata/DisconnectRep.hpp>
#include <signaldata/TcHbRep.hpp>

#include <signaldata/PrepDropTab.hpp>
#include <signaldata/DropTab.hpp>
#include <signaldata/TcIndx.hpp>
#include <signaldata/IndxKeyInfo.hpp>
#include <signaldata/IndxAttrInfo.hpp>
#include <signaldata/PackedSignal.hpp>
#include <AttributeHeader.hpp>
#include <signaldata/DictTabInfo.hpp>
#include <AttributeDescriptor.hpp>
#include <SectionReader.hpp>
#include <KeyDescriptor.hpp>

#include <NdbOut.hpp>
#include <DebuggerNames.hpp>

// Use DEBUG to print messages that should be
// seen only when we debug the product
#ifdef VM_TRACE
#define DEBUG(x) ndbout << "DBTC: "<< x << endl;
#else
#define DEBUG(x)
#endif
  
#define INTERNAL_TRIGGER_TCKEYREQ_JBA 0

#ifdef VM_TRACE
NdbOut &
operator<<(NdbOut& out, Dbtc::ConnectionState state){
  switch(state){
  case Dbtc::CS_CONNECTED: out << "CS_CONNECTED"; break;
  case Dbtc::CS_DISCONNECTED: out << "CS_DISCONNECTED"; break;
  case Dbtc::CS_STARTED: out << "CS_STARTED"; break;
  case Dbtc::CS_RECEIVING: out << "CS_RECEIVING"; break;
  case Dbtc::CS_PREPARED: out << "CS_PREPARED"; break;
  case Dbtc::CS_START_PREPARING: out << "CS_START_PREPARING"; break;
  case Dbtc::CS_REC_PREPARING: out << "CS_REC_PREPARING"; break;
  case Dbtc::CS_RESTART: out << "CS_RESTART"; break;
  case Dbtc::CS_ABORTING: out << "CS_ABORTING"; break;
  case Dbtc::CS_COMPLETING: out << "CS_COMPLETING"; break;
  case Dbtc::CS_COMPLETE_SENT: out << "CS_COMPLETE_SENT"; break;
  case Dbtc::CS_PREPARE_TO_COMMIT: out << "CS_PREPARE_TO_COMMIT"; break;
  case Dbtc::CS_COMMIT_SENT: out << "CS_COMMIT_SENT"; break;
  case Dbtc::CS_START_COMMITTING: out << "CS_START_COMMITTING"; break;
  case Dbtc::CS_COMMITTING: out << "CS_COMMITTING"; break;
  case Dbtc::CS_REC_COMMITTING: out << "CS_REC_COMMITTING"; break;
  case Dbtc::CS_WAIT_ABORT_CONF: out << "CS_WAIT_ABORT_CONF"; break;
  case Dbtc::CS_WAIT_COMPLETE_CONF: out << "CS_WAIT_COMPLETE_CONF"; break;
  case Dbtc::CS_WAIT_COMMIT_CONF: out << "CS_WAIT_COMMIT_CONF"; break;
  case Dbtc::CS_FAIL_ABORTING: out << "CS_FAIL_ABORTING"; break;
  case Dbtc::CS_FAIL_ABORTED: out << "CS_FAIL_ABORTED"; break;
  case Dbtc::CS_FAIL_PREPARED: out << "CS_FAIL_PREPARED"; break;
  case Dbtc::CS_FAIL_COMMITTING: out << "CS_FAIL_COMMITTING"; break;
  case Dbtc::CS_FAIL_COMMITTED: out << "CS_FAIL_COMMITTED"; break;
  case Dbtc::CS_FAIL_COMPLETED: out << "CS_FAIL_COMPLETED"; break;
  case Dbtc::CS_START_SCAN: out << "CS_START_SCAN"; break;
  default:
    out << "Unknown: " << (int)state; break;
  }
  return out;
}
NdbOut &
operator<<(NdbOut& out, Dbtc::OperationState state){
  out << (int)state;
  return out;
}
NdbOut &
operator<<(NdbOut& out, Dbtc::AbortState state){
  out << (int)state;
  return out;
}
NdbOut &
operator<<(NdbOut& out, Dbtc::ReturnSignal state){
  out << (int)state;
  return out;
}
NdbOut &
operator<<(NdbOut& out, Dbtc::ScanRecord::ScanState state){
  out << (int)state;
  return out;
}
NdbOut &
operator<<(NdbOut& out, Dbtc::ScanFragRec::ScanFragState state){
  out << (int)state;
  return out;
}
#endif

void
Dbtc::updateBuddyTimer(ApiConnectRecordPtr apiPtr)
{
  if (apiPtr.p->buddyPtr != RNIL) {
    jam();
    ApiConnectRecordPtr buddyApiPtr;
    buddyApiPtr.i = apiPtr.p->buddyPtr;
    ptrCheckGuard(buddyApiPtr, capiConnectFilesize, apiConnectRecord);
    if (getApiConTimer(buddyApiPtr.i) != 0) {
      if ((apiPtr.p->transid[0] == buddyApiPtr.p->transid[0]) &&
          (apiPtr.p->transid[1] == buddyApiPtr.p->transid[1])) {
        jam();
        setApiConTimer(buddyApiPtr.i, ctcTimer, __LINE__);
      } else {
        jam();
        // Not a buddy anymore since not the same transid
        apiPtr.p->buddyPtr = RNIL;
      }//if
    }//if
  }//if
}

void Dbtc::execCONTINUEB(Signal* signal) 
{
  UintR tcase;

  jamEntry();
  tcase = signal->theData[0];
  UintR Tdata0 = signal->theData[1];
  UintR Tdata1 = signal->theData[2];
  UintR Tdata2 = signal->theData[3];
  switch (tcase) {
  case TcContinueB::ZRETURN_FROM_QUEUED_DELIVERY:
    jam();
    ndbrequire(false);
    return;
  case TcContinueB::ZCOMPLETE_TRANS_AT_TAKE_OVER:
    jam();
    tcNodeFailptr.i = Tdata0;
    ptrCheckGuard(tcNodeFailptr, 1, tcFailRecord);
    completeTransAtTakeOverLab(signal, Tdata1);
    return;
  case TcContinueB::ZCONTINUE_TIME_OUT_CONTROL:
    jam();
    timeOutLoopStartLab(signal, Tdata0);
    return;
  case TcContinueB::ZNODE_TAKE_OVER_COMPLETED:
    jam();
    tnodeid = Tdata0;
    tcNodeFailptr.i = 0;
    ptrAss(tcNodeFailptr, tcFailRecord);
    nodeTakeOverCompletedLab(signal);
    return;
  case TcContinueB::ZINITIALISE_RECORDS:
    jam();
    initialiseRecordsLab(signal, Tdata0, Tdata2, signal->theData[4]);
    return;
  case TcContinueB::ZSEND_COMMIT_LOOP:
    jam();
    apiConnectptr.i = Tdata0;
    ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
    tcConnectptr.i = Tdata1;
    ptrCheckGuard(tcConnectptr, ctcConnectFilesize, tcConnectRecord);
    commit020Lab(signal);
    return;
  case TcContinueB::ZSEND_COMPLETE_LOOP:
    jam();
    apiConnectptr.i = Tdata0;
    ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
    tcConnectptr.i = Tdata1;
    ptrCheckGuard(tcConnectptr, ctcConnectFilesize, tcConnectRecord);
    complete010Lab(signal);
    return;
  case TcContinueB::ZHANDLE_FAILED_API_NODE:
    jam();
    handleFailedApiNode(signal, Tdata0, Tdata1);
    return;
  case TcContinueB::ZTRANS_EVENT_REP:
    jam();
    /* -------------------------------------------------------------------- */
    // Report information about transaction activity once per second.
    /* -------------------------------------------------------------------- */
    if (c_counters.c_trans_status == TransCounters::Timer){
      Uint32 len = c_counters.report(signal);
      sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, len, JBB);
      
      c_counters.reset();
      signal->theData[0] = TcContinueB::ZTRANS_EVENT_REP;
      sendSignalWithDelay(cownref, GSN_CONTINUEB, signal, 5000, 1);
    }
    return;
  case TcContinueB::ZCONTINUE_TIME_OUT_FRAG_CONTROL:
    jam();
    timeOutLoopStartFragLab(signal, Tdata0);
    return;
  case TcContinueB::ZABORT_BREAK:
    jam();
    tcConnectptr.i = Tdata0;
    apiConnectptr.i = Tdata1;
    ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
    apiConnectptr.p->counter--;
    abort015Lab(signal);
    return;
  case TcContinueB::ZABORT_TIMEOUT_BREAK:
    jam();
    tcConnectptr.i = Tdata0;
    apiConnectptr.i = Tdata1;
    ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
    apiConnectptr.p->counter--;
    sendAbortedAfterTimeout(signal, 1);
    return;
  case TcContinueB::ZHANDLE_FAILED_API_NODE_REMOVE_MARKERS:
    jam();
    removeMarkerForFailedAPI(signal, Tdata0, Tdata1);
    return;
  case TcContinueB::ZWAIT_ABORT_ALL:
    jam();
    checkAbortAllTimeout(signal, Tdata0);
    return;
  case TcContinueB::ZCHECK_SCAN_ACTIVE_FAILED_LQH:
    jam();
    checkScanActiveInFailedLqh(signal, Tdata0, Tdata1);
    return;
  case TcContinueB::ZNF_CHECK_TRANSACTIONS:
    jam();
    nodeFailCheckTransactions(signal, Tdata0, Tdata1);
    return;
  case TcContinueB::CHECK_WAIT_DROP_TAB_FAILED_LQH:
    jam();
    checkWaitDropTabFailedLqh(signal, Tdata0, Tdata1);
    return;
  case TcContinueB::TRIGGER_PENDING:
    jam();
    ApiConnectRecordPtr transPtr;
    transPtr.i = Tdata0;
    ptrCheckGuard(transPtr, capiConnectFilesize, apiConnectRecord);
    transPtr.p->triggerPending = false;
    executeTriggers(signal, &transPtr);
    return;
  case TcContinueB::DelayTCKEYCONF:
    jam();
    apiConnectptr.i = Tdata0;
    ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
    sendtckeyconf(signal, Tdata1);
    return;
  default:
    ndbrequire(false);
  }//switch
}

void Dbtc::execDIGETNODESREF(Signal* signal) 
{
  jamEntry();
  terrorCode = signal->theData[1];
  releaseAtErrorLab(signal);
}

void Dbtc::execINCL_NODEREQ(Signal* signal) 
{
  jamEntry();
  tblockref = signal->theData[0];
  hostptr.i = signal->theData[1];
  ptrCheckGuard(hostptr, chostFilesize, hostRecord);
  hostptr.p->hostStatus = HS_ALIVE;
  signal->theData[0] = cownref;
  c_alive_nodes.set(hostptr.i);
  sendSignal(tblockref, GSN_INCL_NODECONF, signal, 1, JBB);
}

void Dbtc::execREAD_NODESREF(Signal* signal) 
{
  jamEntry();
  ndbrequire(false);
}

void Dbtc::execTC_SCHVERREQ(Signal* signal) 
{
  jamEntry();
  if (! assembleFragments(signal)) {
    jam();
    return;
  }
  tabptr.i = signal->theData[0];
  ptrCheckGuard(tabptr, ctabrecFilesize, tableRecord);
  tabptr.p->currentSchemaVersion = signal->theData[1];
  tabptr.p->storedTable = (bool)signal->theData[2];
  BlockReference retRef = signal->theData[3];
  tabptr.p->tableType = (Uint8)signal->theData[4];
  BlockReference retPtr = signal->theData[5];
  Uint32 noOfKeyAttr = signal->theData[6];
  ndbrequire(noOfKeyAttr <= MAX_ATTRIBUTES_IN_INDEX);

  const KeyDescriptor* desc = g_key_descriptor_pool.getPtr(tabptr.i);
  ndbrequire(noOfKeyAttr == desc->noOfKeyAttr);

  ndbrequire(tabptr.p->enabled == false);
  tabptr.p->enabled = true;
  tabptr.p->dropping = false;
  tabptr.p->noOfKeyAttr = desc->noOfKeyAttr;
  tabptr.p->hasCharAttr = desc->hasCharAttr;
  tabptr.p->noOfDistrKeys = desc->noOfDistrKeys;
  
  signal->theData[0] = tabptr.i;
  signal->theData[1] = retPtr;
  sendSignal(retRef, GSN_TC_SCHVERCONF, signal, 2, JBB);
}//Dbtc::execTC_SCHVERREQ()

void
Dbtc::execPREP_DROP_TAB_REQ(Signal* signal)
{
  jamEntry();
  
  PrepDropTabReq* req = (PrepDropTabReq*)signal->getDataPtr();
  
  TableRecordPtr tabPtr;
  tabPtr.i = req->tableId;
  ptrCheckGuard(tabPtr, ctabrecFilesize, tableRecord);
  
  Uint32 senderRef = req->senderRef;
  Uint32 senderData = req->senderData;
  
  if(!tabPtr.p->enabled){
    jam();
    PrepDropTabRef* ref = (PrepDropTabRef*)signal->getDataPtrSend();
    ref->senderRef = reference();
    ref->senderData = senderData;
    ref->tableId = tabPtr.i;
    ref->errorCode = PrepDropTabRef::NoSuchTable;
    sendSignal(senderRef, GSN_PREP_DROP_TAB_REF, signal,
	       PrepDropTabRef::SignalLength, JBB);
    return;
  }

  if(tabPtr.p->dropping){
    jam();
    PrepDropTabRef* ref = (PrepDropTabRef*)signal->getDataPtrSend();
    ref->senderRef = reference();
    ref->senderData = senderData;
    ref->tableId = tabPtr.i;
    ref->errorCode = PrepDropTabRef::DropInProgress;
    sendSignal(senderRef, GSN_PREP_DROP_TAB_REF, signal,
	       PrepDropTabRef::SignalLength, JBB);
    return;
  }
  
  tabPtr.p->dropping = true;
  tabPtr.p->dropTable.senderRef = senderRef;
  tabPtr.p->dropTable.senderData = senderData;

  {
    WaitDropTabReq * req = (WaitDropTabReq*)signal->getDataPtrSend();
    req->tableId = tabPtr.i;
    req->senderRef = reference();
    
    HostRecordPtr hostPtr;
    tabPtr.p->dropTable.waitDropTabCount.clearWaitingFor();
    for (hostPtr.i = 1; hostPtr.i < MAX_NDB_NODES; hostPtr.i++) {
      jam();
      ptrAss(hostPtr, hostRecord);
      if (hostPtr.p->hostStatus == HS_ALIVE) {
	jam();
	tabPtr.p->dropTable.waitDropTabCount.setWaitingFor(hostPtr.i);
	sendSignal(calcLqhBlockRef(hostPtr.i), GSN_WAIT_DROP_TAB_REQ,
		   signal, WaitDropTabReq::SignalLength, JBB);
      }//for
    }//if
    
    ndbrequire(tabPtr.p->dropTable.waitDropTabCount.done() != true);
  }
}

void
Dbtc::execWAIT_DROP_TAB_CONF(Signal* signal)
{
  jamEntry();
  WaitDropTabConf * conf = (WaitDropTabConf*)signal->getDataPtr();

  TableRecordPtr tabPtr;
  tabPtr.i = conf->tableId;
  ptrCheckGuard(tabPtr, ctabrecFilesize, tableRecord);
  
  ndbrequire(tabPtr.p->dropping == true);
  Uint32 nodeId = refToNode(conf->senderRef);
  tabPtr.p->dropTable.waitDropTabCount.clearWaitingFor(nodeId);
  
  if(!tabPtr.p->dropTable.waitDropTabCount.done()){