trix.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 */

#include "Trix.hpp"

#include <string.h>
#include <kernel_types.h>
#include <NdbOut.hpp>

#include <signaldata/ReadNodesConf.hpp>
#include <signaldata/NodeFailRep.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <signaldata/GetTabInfo.hpp>
#include <signaldata/DictTabInfo.hpp>
#include <signaldata/BuildIndx.hpp>
#include <signaldata/SumaImpl.hpp>
#include <signaldata/UtilPrepare.hpp>
#include <signaldata/UtilExecute.hpp>
#include <signaldata/UtilRelease.hpp>
#include <SectionReader.hpp>
#include <AttributeHeader.hpp>

#define CONSTRAINT_VIOLATION 893

#define DEBUG(x) { ndbout << "TRIX::" << x << endl; }

/**
 *
 */
Trix::Trix(const Configuration & conf) :
  SimulatedBlock(TRIX, conf),
  c_theNodes(c_theNodeRecPool),
  c_masterNodeId(0),
  c_masterTrixRef(0),
  c_noNodesFailed(0),
  c_noActiveNodes(0),
  c_theSubscriptions(c_theSubscriptionRecPool)
{
  BLOCK_CONSTRUCTOR(Trix);

  // Add received signals
  addRecSignal(GSN_READ_CONFIG_REQ,  &Trix::execREAD_CONFIG_REQ);
  addRecSignal(GSN_STTOR,  &Trix::execSTTOR);
  addRecSignal(GSN_NDB_STTOR,  &Trix::execNDB_STTOR); // Forwarded from DICT
  addRecSignal(GSN_READ_NODESCONF, &Trix::execREAD_NODESCONF);
  addRecSignal(GSN_READ_NODESREF, &Trix::execREAD_NODESREF);
  addRecSignal(GSN_NODE_FAILREP, &Trix::execNODE_FAILREP);
  addRecSignal(GSN_INCL_NODEREQ, &Trix::execINCL_NODEREQ);
  addRecSignal(GSN_DUMP_STATE_ORD, &Trix::execDUMP_STATE_ORD);

  // Index build
  addRecSignal(GSN_BUILDINDXREQ, &Trix::execBUILDINDXREQ);
  // Dump testing
  addRecSignal(GSN_BUILDINDXCONF, &Trix::execBUILDINDXCONF);
  addRecSignal(GSN_BUILDINDXREF, &Trix::execBUILDINDXREF);


  addRecSignal(GSN_UTIL_PREPARE_CONF, &Trix::execUTIL_PREPARE_CONF);
  addRecSignal(GSN_UTIL_PREPARE_REF, &Trix::execUTIL_PREPARE_REF);
  addRecSignal(GSN_UTIL_EXECUTE_CONF, &Trix::execUTIL_EXECUTE_CONF);
  addRecSignal(GSN_UTIL_EXECUTE_REF, &Trix::execUTIL_EXECUTE_REF);
  addRecSignal(GSN_UTIL_RELEASE_CONF, &Trix::execUTIL_RELEASE_CONF);
  addRecSignal(GSN_UTIL_RELEASE_REF, &Trix::execUTIL_RELEASE_REF);


  // Suma signals
  addRecSignal(GSN_SUB_CREATE_CONF, &Trix::execSUB_CREATE_CONF);
  addRecSignal(GSN_SUB_CREATE_REF, &Trix::execSUB_CREATE_REF);
  addRecSignal(GSN_SUB_REMOVE_CONF, &Trix::execSUB_REMOVE_CONF);
  addRecSignal(GSN_SUB_REMOVE_REF, &Trix::execSUB_REMOVE_REF);
  addRecSignal(GSN_SUB_SYNC_CONF, &Trix::execSUB_SYNC_CONF);
  addRecSignal(GSN_SUB_SYNC_REF, &Trix::execSUB_SYNC_REF);
  addRecSignal(GSN_SUB_SYNC_CONTINUE_REQ, &Trix::execSUB_SYNC_CONTINUE_REQ);
  addRecSignal(GSN_SUB_META_DATA, &Trix::execSUB_META_DATA);
  addRecSignal(GSN_SUB_TABLE_DATA, &Trix::execSUB_TABLE_DATA);
}

/**
 *
 */
Trix::~Trix()
{
}

void 
Trix::execREAD_CONFIG_REQ(Signal* signal)
{
  jamEntry();

  const ReadConfigReq * req = (ReadConfigReq*)signal->getDataPtr();

  Uint32 ref = req->senderRef;
  Uint32 senderData = req->senderData;

  const ndb_mgm_configuration_iterator * p = 
    theConfiguration.getOwnConfigIterator();
  ndbrequire(p != 0);

  // Allocate pool sizes
  c_theAttrOrderBufferPool.setSize(100);
  c_theSubscriptionRecPool.setSize(100);

  ArrayList<SubscriptionRecord> subscriptions(c_theSubscriptionRecPool);
  SubscriptionRecPtr subptr;
  while(subscriptions.seize(subptr) == true) {
    new (subptr.p) SubscriptionRecord(c_theAttrOrderBufferPool);
  }
  subscriptions.release();

  ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
  conf->senderRef = reference();
  conf->senderData = senderData;
  sendSignal(ref, GSN_READ_CONFIG_CONF, signal, 
	     ReadConfigConf::SignalLength, JBB);
}

/**
 *
 */
void Trix::execSTTOR(Signal* signal) 
{
  jamEntry();                            

  //const Uint32 startphase   = signal->theData[1];
  const Uint32 theSignalKey = signal->theData[6];
  
  signal->theData[0] = theSignalKey;
  signal->theData[3] = 1;
  signal->theData[4] = 255; // No more start phases from missra
  sendSignal(NDBCNTR_REF, GSN_STTORRY, signal, 5, JBB);
  return;
}//Trix::execSTTOR()

/**
 *
 */
void Trix::execNDB_STTOR(Signal* signal) 
{
  jamEntry();                            
  BlockReference ndbcntrRef = signal->theData[0];	 
  Uint16 startphase = signal->theData[2];      /* RESTART PHASE           */
  Uint16 mynode = signal->theData[1];		 
  //Uint16 restarttype = signal->theData[3];	 
  //UintR configInfo1 = signal->theData[6];     /* CONFIGRATION INFO PART 1 */
  //UintR configInfo2 = signal->theData[7];     /* CONFIGRATION INFO PART 2 */
  switch (startphase) {
  case 3:
    jam();
    /* SYMBOLIC START PHASE 4             */
    /* ABSOLUTE PHASE 5                   */
    /* REQUEST NODE IDENTITIES FROM DBDIH */
    signal->theData[0] = calcTrixBlockRef(mynode);
    sendSignal(ndbcntrRef, GSN_READ_NODESREQ, signal, 1, JBB);
    return;
    break;
  case 6:
    break;
  default:
    break;
  }
}

/**
 *
 */
void Trix::execREAD_NODESCONF(Signal* signal)
{
  jamEntry();

  ReadNodesConf * const  readNodes = (ReadNodesConf *)signal->getDataPtr();
  //Uint32 noOfNodes   = readNodes->noOfNodes;
  NodeRecPtr nodeRecPtr;

  c_masterNodeId = readNodes->masterNodeId;
  c_masterTrixRef = RNIL;
  c_noNodesFailed = 0;

  for(unsigned i = 0; i < MAX_NDB_NODES; i++) {
    jam();
    if(NodeBitmask::get(readNodes->allNodes, i)) {
      // Node is defined
      jam();
      ndbrequire(c_theNodes.seizeId(nodeRecPtr, i));
      nodeRecPtr.p->trixRef = calcTrixBlockRef(i);
      if (i == c_masterNodeId) {
        c_masterTrixRef = nodeRecPtr.p->trixRef;
      }
      if(NodeBitmask::get(readNodes->inactiveNodes, i)){
        // Node is not active
	jam();
	/**-----------------------------------------------------------------
	 * THIS NODE IS DEFINED IN THE CLUSTER BUT IS NOT ALIVE CURRENTLY.
	 * WE ADD THE NODE TO THE SET OF FAILED NODES AND ALSO SET THE
	 * BLOCKSTATE TO BUSY TO AVOID ADDING TRIGGERS OR INDEXES WHILE 
	 * NOT ALL NODES ARE ALIVE.
	 *------------------------------------------------------------------*/
	arrGuard(c_noNodesFailed, MAX_NDB_NODES);
	nodeRecPtr.p->alive = false;
	c_noNodesFailed++;
	c_blockState = Trix::NODE_FAILURE;
      }
      else {
        // Node is active
        jam();
        c_noActiveNodes++;
        nodeRecPtr.p->alive = true;
      }
    }
  }
  if (c_noNodesFailed == 0) {
    c_blockState = Trix::STARTED;
  }
}

/**
 *
 */
void Trix::execREAD_NODESREF(Signal* signal)
{
  // NYI
}

/**
 *
 */
void Trix::execNODE_FAILREP(Signal* signal)
{
  jamEntry();
  NodeFailRep * const  nodeFail = (NodeFailRep *) signal->getDataPtr();

  //Uint32 failureNr    = nodeFail->failNo;
  //Uint32 numberNodes  = nodeFail->noOfNodes;
  Uint32 masterNodeId = nodeFail->masterNodeId;

  NodeRecPtr nodeRecPtr;

  for(c_theNodes.first(nodeRecPtr); 
      nodeRecPtr.i != RNIL; 
      c_theNodes.next(nodeRecPtr)) {
    if(NodeBitmask::get(nodeFail->theNodes, nodeRecPtr.i)) {
      nodeRecPtr.p->alive = false;
      c_noNodesFailed++;
      c_noActiveNodes--;      
    }
  }
  if (c_masterNodeId != masterNodeId) {
    c_masterNodeId = masterNodeId;
    NodeRecord* nodeRec = c_theNodes.getPtr(masterNodeId);
    c_masterTrixRef = nodeRec->trixRef;
  }
}

/**
 *
 */
void Trix::execINCL_NODEREQ(Signal* signal)
{
  jamEntry();
  UintR node_id = signal->theData[1];
  NodeRecord* nodeRec = c_theNodes.getPtr(node_id);
  nodeRec->alive = true;
  c_noNodesFailed--;
  c_noActiveNodes++;      
  nodeRec->trixRef = calcTrixBlockRef(node_id);
  if (c_noNodesFailed == 0) {
    c_blockState = Trix::STARTED;
  }  
}

// Debugging
void
Trix::execDUMP_STATE_ORD(Signal* signal)
{
  jamEntry();

  DumpStateOrd * dumpStateOrd = (DumpStateOrd *)signal->getDataPtr();

  switch(dumpStateOrd->args[0]) {
  case(300): {// ok
    // index2 -T; index2 -I -n10000; index2 -c
    // all dump 300 0 0 0 0 0 4 2
    // select_count INDEX0000
    BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
		     signal->theData + 1, 
		     BuildIndxReq::SignalLength);
    buildIndxReq->setUserRef(reference()); // return to me
    buildIndxReq->setParallelism(10);
    Uint32 indexColumns[1] = {1};
    Uint32 keyColumns[1] = {0};
    struct LinearSectionPtr orderPtr[2];
    buildIndxReq->setColumnOrder(indexColumns, 1, keyColumns, 1, orderPtr);
    sendSignal(reference(),
	       GSN_BUILDINDXREQ,
	       signal,
	       BuildIndxReq::SignalLength,
	       JBB,
	       orderPtr,
	       BuildIndxReq::NoOfSections);
    break;
  }
  case(301): { // ok
    // index2 -T; index2 -I -n10000; index2 -c -p
    // all dump 301 0 0 0 0 0 4 2
    // select_count INDEX0000
    BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
		     signal->theData + 1, 
		     BuildIndxReq::SignalLength);
    buildIndxReq->setUserRef(reference()); // return to me
    buildIndxReq->setParallelism(10);
    Uint32 indexColumns[2] = {0, 1};
    Uint32 keyColumns[1] = {0};
    struct LinearSectionPtr orderPtr[2];
    buildIndxReq->setColumnOrder(indexColumns, 2, keyColumns, 1, orderPtr);