dbutil.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 <ndb_global.h>

#include "DbUtil.hpp"

#include <ndb_version.h>

#include <signaldata/WaitGCP.hpp>
#include <signaldata/KeyInfo.hpp>
#include <signaldata/AttrInfo.hpp>
#include <signaldata/TcKeyConf.hpp>
#include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/GetTabInfo.hpp>
#include <signaldata/DictTabInfo.hpp>

#include <signaldata/UtilSequence.hpp>
#include <signaldata/UtilPrepare.hpp>
#include <signaldata/UtilRelease.hpp>
#include <signaldata/UtilExecute.hpp>
#include <signaldata/UtilLock.hpp>

#include <SectionReader.hpp>
#include <Interpreter.hpp>
#include <AttributeHeader.hpp>

#include <NdbTick.h>


/**************************************************************************
 * ------------------------------------------------------------------------
 *  MODULE:       Startup
 * ------------------------------------------------------------------------
 * 
 *  Constructors, startup, initializations
 **************************************************************************/

DbUtil::DbUtil(const Configuration & conf) :
  SimulatedBlock(DBUTIL, conf),
  c_runningPrepares(c_preparePool),
  c_runningPreparedOperations(c_preparedOperationPool),
  c_seizingTransactions(c_transactionPool),
  c_runningTransactions(c_transactionPool),
  c_lockQueues(c_lockQueuePool)
{
  BLOCK_CONSTRUCTOR(DbUtil);
  
  // Add received signals
  addRecSignal(GSN_READ_CONFIG_REQ, &DbUtil::execREAD_CONFIG_REQ);
  addRecSignal(GSN_STTOR, &DbUtil::execSTTOR);
  addRecSignal(GSN_NDB_STTOR, &DbUtil::execNDB_STTOR);
  addRecSignal(GSN_DUMP_STATE_ORD, &DbUtil::execDUMP_STATE_ORD);
  addRecSignal(GSN_CONTINUEB, &DbUtil::execCONTINUEB);
  
  //addRecSignal(GSN_TCSEIZEREF, &DbUtil::execTCSEIZEREF);
  addRecSignal(GSN_TCSEIZECONF, &DbUtil::execTCSEIZECONF);
  addRecSignal(GSN_TCKEYCONF, &DbUtil::execTCKEYCONF);
  addRecSignal(GSN_TCKEYREF, &DbUtil::execTCKEYREF);
  addRecSignal(GSN_TCROLLBACKREP, &DbUtil::execTCROLLBACKREP);

  //addRecSignal(GSN_TCKEY_FAILCONF, &DbUtil::execTCKEY_FAILCONF);
  //addRecSignal(GSN_TCKEY_FAILREF, &DbUtil::execTCKEY_FAILREF);
  addRecSignal(GSN_TRANSID_AI, &DbUtil::execTRANSID_AI);

  /**
   *  Sequence Service
   */
  addRecSignal(GSN_UTIL_SEQUENCE_REQ, &DbUtil::execUTIL_SEQUENCE_REQ);
  // Debug
  addRecSignal(GSN_UTIL_SEQUENCE_REF, &DbUtil::execUTIL_SEQUENCE_REF);
  addRecSignal(GSN_UTIL_SEQUENCE_CONF, &DbUtil::execUTIL_SEQUENCE_CONF);

  /**
   * Locking
   */
  addRecSignal(GSN_UTIL_CREATE_LOCK_REQ,  &DbUtil::execUTIL_CREATE_LOCK_REQ);
  addRecSignal(GSN_UTIL_DESTROY_LOCK_REQ, &DbUtil::execUTIL_DESTORY_LOCK_REQ);
  addRecSignal(GSN_UTIL_LOCK_REQ,  &DbUtil::execUTIL_LOCK_REQ);
  addRecSignal(GSN_UTIL_UNLOCK_REQ, &DbUtil::execUTIL_UNLOCK_REQ);

  /**
   *  Backend towards Dict
   */
  addRecSignal(GSN_GET_TABINFOREF, &DbUtil::execGET_TABINFOREF);
  addRecSignal(GSN_GET_TABINFO_CONF, &DbUtil::execGET_TABINFO_CONF);

  /**
   *  Prepare / Execute / Release Services
   */
  addRecSignal(GSN_UTIL_PREPARE_REQ,  &DbUtil::execUTIL_PREPARE_REQ);
  addRecSignal(GSN_UTIL_PREPARE_CONF, &DbUtil::execUTIL_PREPARE_CONF);
  addRecSignal(GSN_UTIL_PREPARE_REF,  &DbUtil::execUTIL_PREPARE_REF);

  addRecSignal(GSN_UTIL_EXECUTE_REQ,  &DbUtil::execUTIL_EXECUTE_REQ);
  addRecSignal(GSN_UTIL_EXECUTE_CONF, &DbUtil::execUTIL_EXECUTE_CONF);
  addRecSignal(GSN_UTIL_EXECUTE_REF,  &DbUtil::execUTIL_EXECUTE_REF);

  addRecSignal(GSN_UTIL_RELEASE_REQ,  &DbUtil::execUTIL_RELEASE_REQ);
  addRecSignal(GSN_UTIL_RELEASE_CONF, &DbUtil::execUTIL_RELEASE_CONF);
  addRecSignal(GSN_UTIL_RELEASE_REF,  &DbUtil::execUTIL_RELEASE_REF);
}

DbUtil::~DbUtil()
{
}

BLOCK_FUNCTIONS(DbUtil)

void 
DbUtil::releasePrepare(PreparePtr prepPtr) {
  prepPtr.p->preparePages.release();
  c_runningPrepares.release(prepPtr);  // Automatic release in pool
}

void 
DbUtil::releasePreparedOperation(PreparedOperationPtr prepOpPtr) {
  prepOpPtr.p->attrMapping.release();
  prepOpPtr.p->attrInfo.release();
  prepOpPtr.p->rsInfo.release();
  prepOpPtr.p->pkBitmask.clear();
  c_preparedOperationPool.release(prepOpPtr);  // No list holding these structs
}

void
DbUtil::releaseTransaction(TransactionPtr transPtr){
  transPtr.p->executePages.release();
  OperationPtr opPtr;
  for(transPtr.p->operations.first(opPtr); opPtr.i != RNIL; 
      transPtr.p->operations.next(opPtr)){
    opPtr.p->attrInfo.release();
    opPtr.p->keyInfo.release();
    opPtr.p->rs.release();
    if (opPtr.p->prepOp != 0 && opPtr.p->prepOp_i != RNIL) {
      if (opPtr.p->prepOp->releaseFlag) {
	PreparedOperationPtr prepOpPtr;
	prepOpPtr.i = opPtr.p->prepOp_i;
	prepOpPtr.p = opPtr.p->prepOp;
	releasePreparedOperation(prepOpPtr);
      }
    }
  }
  transPtr.p->operations.release();
  c_runningTransactions.release(transPtr);
}

void 
DbUtil::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);

  c_pagePool.setSize(10);
  c_preparePool.setSize(1);            // one parallel prepare at a time
  c_preparedOperationPool.setSize(5);  // three hardcoded, two for test
  c_operationPool.setSize(64);         // 64 parallel operations
  c_transactionPool.setSize(32);       // 16 parallel transactions
  c_attrMappingPool.setSize(100);
  c_dataBufPool.setSize(6000);	       // 6000*11*4 = 264K > 8k+8k*16 = 256k
  {
    SLList<Prepare> tmp(c_preparePool);
    PreparePtr ptr;
    while(tmp.seize(ptr))
      new (ptr.p) Prepare(c_pagePool);
    tmp.release();
  }
  {
    SLList<Operation> tmp(c_operationPool);
    OperationPtr ptr;
    while(tmp.seize(ptr))
      new (ptr.p) Operation(c_dataBufPool, c_dataBufPool, c_dataBufPool);
    tmp.release();
  }
  {
    SLList<PreparedOperation> tmp(c_preparedOperationPool);
    PreparedOperationPtr ptr;
    while(tmp.seize(ptr))
      new (ptr.p) PreparedOperation(c_attrMappingPool, 
				    c_dataBufPool, c_dataBufPool);
    tmp.release();
  }
  {
    SLList<Transaction> tmp(c_transactionPool);
    TransactionPtr ptr;
    while(tmp.seize(ptr))
      new (ptr.p) Transaction(c_pagePool, c_operationPool);
    tmp.release();
  }

  c_lockQueuePool.setSize(5);
  c_lockElementPool.setSize(5);
  c_lockQueues.setSize(8);

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

void
DbUtil::execSTTOR(Signal* signal) 
{
  jamEntry();                            

  const Uint32 startphase = signal->theData[1];
  
  if(startphase == 1){
    c_transId[0] = (number() << 20) + (getOwnNodeId() << 8);
    c_transId[1] = 0;
  }
  
  if(startphase == 6){
    hardcodedPrepare();
    connectTc(signal);
  }
  
  signal->theData[0] = 0;
  signal->theData[3] = 1;
  signal->theData[4] = 6;
  signal->theData[5] = 255;
  sendSignal(NDBCNTR_REF, GSN_STTORRY, signal, 6, JBB);

  return;
}

void
DbUtil::execNDB_STTOR(Signal* signal) 
{
  (void)signal;  // Don't want compiler warning

  jamEntry();                            
}


/***************************
 *  Seize a number of TC records 
 *  to use for Util transactions
 */

void
DbUtil::connectTc(Signal* signal){
  
  TransactionPtr ptr;
  while(c_seizingTransactions.seize(ptr)){
    signal->theData[0] = ptr.i << 1; // See TcCommitConf
    signal->theData[1] = reference();
    sendSignal(DBTC_REF, GSN_TCSEIZEREQ, signal, 2, JBB);
  }  
}

void
DbUtil::execTCSEIZECONF(Signal* signal){
  jamEntry();
  
  TransactionPtr ptr;
  ptr.i = signal->theData[0] >> 1;
  c_seizingTransactions.getPtr(ptr, signal->theData[0] >> 1);
  ptr.p->connectPtr = signal->theData[1];
  
  c_seizingTransactions.release(ptr);
}


/**************************************************************************
 * ------------------------------------------------------------------------
 *  MODULE:       Misc
 * ------------------------------------------------------------------------
 *
 *  ContinueB, Dump
 **************************************************************************/

void
DbUtil::execCONTINUEB(Signal* signal){
  jamEntry();
  const Uint32 Tdata0 = signal->theData[0];
  
  switch(Tdata0){
  default:
    ndbrequire(0);
  }
}

void
DbUtil::execDUMP_STATE_ORD(Signal* signal){
  jamEntry();

  /****************************************************************************
   *  SEQUENCE SERVICE
   * 
   *  200 : Simple test of Public Sequence Interface
   *  ----------------------------------------------
   *  - Sends a SEQUENCE_REQ signal to Util (itself)
   */
  const Uint32 tCase = signal->theData[0];
  if(tCase == 200){
    jam()
    ndbout << "--------------------------------------------------" << endl;
    UtilSequenceReq * req = (UtilSequenceReq*)signal->getDataPtrSend();
    Uint32 seqId = 1;
    Uint32 reqTy = UtilSequenceReq::CurrVal;

    if(signal->length() > 1) seqId = signal->theData[1];
    if(signal->length() > 2) reqTy = signal->theData[2];
    
    req->senderData = 12;
    req->sequenceId = seqId;
    req->requestType = reqTy;

    sendSignal(DBUTIL_REF, GSN_UTIL_SEQUENCE_REQ, 
	       signal, UtilSequenceReq::SignalLength, JBB);
  }

  /****************************************************************************/
  /* // Obsolete tests, should be rewritten for long signals!!
  if(tCase == 210){
    jam();
    ndbout << "--------------------------------------------------" << endl;
    const Uint32 pageSizeInWords = 128;
    Uint32 propPage[pageSizeInWords];
    LinearWriter w(&propPage[0], 128);
    w.first();
    w.add(UtilPrepareReq::NoOfOperations, 1);
    w.add(UtilPrepareReq::OperationType, UtilPrepareReq::Delete);
    w.add(UtilPrepareReq::TableName,      "sys/def/SYSTAB_0");
    w.add(UtilPrepareReq::AttributeName,  "SYSKEY_0"); // AttrNo = 0
    Uint32 length = w.getWordsUsed();
    ndbassert(length <= pageSizeInWords);

    sendUtilPrepareReqSignals(signal, propPage, length);
  }
  if(tCase == 211){
    jam();
    ndbout << "--------------------------------------------------" << endl;
    const Uint32 pageSizeInWords = 128;
    Uint32 propPage[pageSizeInWords];
    LinearWriter w(&propPage[0],128);
    w.first();
    w.add(UtilPrepareReq::NoOfOperations, 1);
    w.add(UtilPrepareReq::OperationType,  UtilPrepareReq::Insert);
    w.add(UtilPrepareReq::TableName,      "sys/def/SYSTAB_0");
    w.add(UtilPrepareReq::AttributeName,  "SYSKEY_0");  // AttrNo = 0
    w.add(UtilPrepareReq::AttributeName,  "NEXTID");    // AttrNo = 1
    Uint32 length = w.getWordsUsed();
    ndbassert(length <= pageSizeInWords);

    sendUtilPrepareReqSignals(signal, propPage, length);
  }
  if(tCase == 212){
    jam();
    ndbout << "--------------------------------------------------" << endl;
    const Uint32 pageSizeInWords = 128;
    Uint32 propPage[pageSizeInWords];
    LinearWriter w(&propPage[0],128);
    w.first();
    w.add(UtilPrepareReq::NoOfOperations, 1);
    w.add(UtilPrepareReq::OperationType,  UtilPrepareReq::Update);
    w.add(UtilPrepareReq::TableName,      "sys/def/SYSTAB_0");
    w.add(UtilPrepareReq::AttributeName,  "SYSKEY_0");  // AttrNo = 0
    w.add(UtilPrepareReq::AttributeName,  "NEXTID");    // AttrNo = 1
    Uint32 length = w.getWordsUsed();
    ndbassert(length <= pageSizeInWords);

    sendUtilPrepareReqSignals(signal, propPage, length);
  }
  if(tCase == 213){
    jam();
    ndbout << "--------------------------------------------------" << endl;
    const Uint32 pageSizeInWords = 128;
    Uint32 propPage[pageSizeInWords];
    LinearWriter w(&propPage[0],128);
    w.first();
    w.add(UtilPrepareReq::NoOfOperations, 1);
    w.add(UtilPrepareReq::OperationType, UtilPrepareReq::Read);
    w.add(UtilPrepareReq::TableName,      "sys/def/SYSTAB_0");
    w.add(UtilPrepareReq::AttributeName,  "SYSKEY_0"); // AttrNo = 0
    Uint32 length = w.getWordsUsed();
    ndbassert(length <= pageSizeInWords);

    sendUtilPrepareReqSignals(signal, propPage, length);
  }
  if(tCase == 214){
    jam();
    ndbout << "--------------------------------------------------" << endl;