dbtupgen.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 DBTUP_C
#include "Dbtup.hpp"
#include <RefConvert.hpp>
#include <ndb_limits.h>
#include <pc.hpp>
#include <AttributeDescriptor.hpp>
#include "AttributeOffset.hpp"
#include <AttributeHeader.hpp>
#include <Interpreter.hpp>
#include <signaldata/FsConf.hpp>
#include <signaldata/FsRemoveReq.hpp>
#include <signaldata/TupCommit.hpp>
#include <signaldata/TupKey.hpp>

#include <signaldata/DropTab.hpp>

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

#define ljam() { jamLine(24000 + __LINE__); }
#define ljamEntry() { jamEntryLine(24000 + __LINE__); }

void Dbtup::initData() 
{
  cnoOfAttrbufrec = ZNO_OF_ATTRBUFREC;
  cnoOfLcpRec = ZNO_OF_LCP_REC;
  cnoOfConcurrentOpenOp = ZNO_OF_CONCURRENT_OPEN_OP;
  cnoOfConcurrentWriteOp = ZNO_OF_CONCURRENT_WRITE_OP;
  cnoOfFragoprec = 2 * MAX_FRAG_PER_NODE;
  cnoOfPageRangeRec = ZNO_OF_PAGE_RANGE_REC;
  cnoOfParallellUndoFiles = ZNO_OF_PARALLELL_UNDO_FILES;
  cnoOfRestartInfoRec = ZNO_OF_RESTART_INFO_REC;
  c_maxTriggersPerTable = ZDEFAULT_MAX_NO_TRIGGERS_PER_TABLE;
  c_noOfBuildIndexRec = 32;

  attrbufrec = 0;
  checkpointInfo = 0;
  diskBufferSegmentInfo = 0;
  fragoperrec = 0;
  fragrecord = 0;
  hostBuffer = 0;
  localLogInfo = 0;
  operationrec = 0;
  page = 0;
  pageRange = 0;
  pendingFileOpenInfo = 0;
  restartInfoRecord = 0;
  tablerec = 0;
  tableDescriptor = 0;
  undoPage = 0;
  totNoOfPagesAllocated = 0;
  cnoOfAllocatedPages = 0;
  
  // Records with constant sizes
}//Dbtup::initData()

Dbtup::Dbtup(const class Configuration & conf)
  : SimulatedBlock(DBTUP, conf),
  c_storedProcPool(),
  c_buildIndexList(c_buildIndexPool)
{
  BLOCK_CONSTRUCTOR(Dbtup);

  addRecSignal(GSN_DEBUG_SIG, &Dbtup::execDEBUG_SIG);
  addRecSignal(GSN_CONTINUEB, &Dbtup::execCONTINUEB);

  addRecSignal(GSN_DUMP_STATE_ORD, &Dbtup::execDUMP_STATE_ORD);
  addRecSignal(GSN_SEND_PACKED, &Dbtup::execSEND_PACKED);
  addRecSignal(GSN_ATTRINFO, &Dbtup::execATTRINFO);
  addRecSignal(GSN_STTOR, &Dbtup::execSTTOR);
  addRecSignal(GSN_TUP_LCPREQ, &Dbtup::execTUP_LCPREQ);
  addRecSignal(GSN_END_LCPREQ, &Dbtup::execEND_LCPREQ);
  addRecSignal(GSN_START_RECREQ, &Dbtup::execSTART_RECREQ);
  addRecSignal(GSN_MEMCHECKREQ, &Dbtup::execMEMCHECKREQ);
  addRecSignal(GSN_TUPKEYREQ, &Dbtup::execTUPKEYREQ);
  addRecSignal(GSN_TUPSEIZEREQ, &Dbtup::execTUPSEIZEREQ);
  addRecSignal(GSN_TUPRELEASEREQ, &Dbtup::execTUPRELEASEREQ);
  addRecSignal(GSN_STORED_PROCREQ, &Dbtup::execSTORED_PROCREQ);
  addRecSignal(GSN_TUPFRAGREQ, &Dbtup::execTUPFRAGREQ);
  addRecSignal(GSN_TUP_ADD_ATTRREQ, &Dbtup::execTUP_ADD_ATTRREQ);
  addRecSignal(GSN_TUP_COMMITREQ, &Dbtup::execTUP_COMMITREQ);
  addRecSignal(GSN_TUP_ABORTREQ, &Dbtup::execTUP_ABORTREQ);
  addRecSignal(GSN_TUP_SRREQ, &Dbtup::execTUP_SRREQ);
  addRecSignal(GSN_TUP_PREPLCPREQ, &Dbtup::execTUP_PREPLCPREQ);
  addRecSignal(GSN_FSOPENCONF, &Dbtup::execFSOPENCONF);
  addRecSignal(GSN_FSCLOSECONF, &Dbtup::execFSCLOSECONF);
  addRecSignal(GSN_FSWRITECONF, &Dbtup::execFSWRITECONF);
  addRecSignal(GSN_FSREADCONF, &Dbtup::execFSREADCONF);
  addRecSignal(GSN_NDB_STTOR, &Dbtup::execNDB_STTOR);
  addRecSignal(GSN_READ_CONFIG_REQ, &Dbtup::execREAD_CONFIG_REQ, true);
  addRecSignal(GSN_SET_VAR_REQ,  &Dbtup::execSET_VAR_REQ);

  // Trigger Signals
  addRecSignal(GSN_CREATE_TRIG_REQ, &Dbtup::execCREATE_TRIG_REQ);
  addRecSignal(GSN_DROP_TRIG_REQ,  &Dbtup::execDROP_TRIG_REQ);

  addRecSignal(GSN_DROP_TAB_REQ, &Dbtup::execDROP_TAB_REQ);
  addRecSignal(GSN_FSREMOVECONF, &Dbtup::execFSREMOVECONF);

  addRecSignal(GSN_TUP_ALLOCREQ, &Dbtup::execTUP_ALLOCREQ);
  addRecSignal(GSN_TUP_DEALLOCREQ, &Dbtup::execTUP_DEALLOCREQ);
  addRecSignal(GSN_TUP_WRITELOG_REQ, &Dbtup::execTUP_WRITELOG_REQ);

  // Ordered index related
  addRecSignal(GSN_BUILDINDXREQ, &Dbtup::execBUILDINDXREQ);

  // Tup scan
  addRecSignal(GSN_ACC_SCANREQ, &Dbtup::execACC_SCANREQ);
  addRecSignal(GSN_NEXT_SCANREQ, &Dbtup::execNEXT_SCANREQ);
  addRecSignal(GSN_ACC_CHECK_SCAN, &Dbtup::execACC_CHECK_SCAN);

  initData();

  attrbufrec = 0;  
  checkpointInfo = 0;  
  diskBufferSegmentInfo = 0;  
  fragoperrec = 0;
  fragrecord = 0;  
  hostBuffer = 0;  
  localLogInfo = 0;  
  operationrec = 0;
  page = 0;  
  pageRange = 0;
  pendingFileOpenInfo = 0;  
  restartInfoRecord = 0;  
  tablerec = 0;  
  tableDescriptor = 0;  
  undoPage = 0;
}//Dbtup::Dbtup()

Dbtup::~Dbtup() 
{
  // Records with dynamic sizes
  deallocRecord((void **)&attrbufrec,"Attrbufrec", 
		sizeof(Attrbufrec), 
		cnoOfAttrbufrec);
  
  deallocRecord((void **)&checkpointInfo,"CheckpointInfo",
		sizeof(CheckpointInfo), 
		cnoOfLcpRec);
  
  deallocRecord((void **)&diskBufferSegmentInfo,
		"DiskBufferSegmentInfo",
		sizeof(DiskBufferSegmentInfo), 
		cnoOfConcurrentWriteOp);
  
  deallocRecord((void **)&fragoperrec,"Fragoperrec",
		sizeof(Fragoperrec),
		cnoOfFragoprec);
  
  deallocRecord((void **)&fragrecord,"Fragrecord",
		sizeof(Fragrecord), 
		cnoOfFragrec);
  
  deallocRecord((void **)&hostBuffer,"HostBuffer",
		sizeof(HostBuffer), 
		MAX_NODES);
  
  deallocRecord((void **)&localLogInfo,"LocalLogInfo",
		sizeof(LocalLogInfo), 
		cnoOfParallellUndoFiles);
  
  deallocRecord((void **)&operationrec,"Operationrec",
		sizeof(Operationrec),
		cnoOfOprec);

  deallocRecord((void **)&page,"Page", 
		sizeof(Page), 
		cnoOfPage);
  
  deallocRecord((void **)&pageRange,"PageRange",
		sizeof(PageRange), 
		cnoOfPageRangeRec);

  deallocRecord((void **)&pendingFileOpenInfo,
		"PendingFileOpenInfo",
		sizeof(PendingFileOpenInfo), 
		cnoOfConcurrentOpenOp);
  
  deallocRecord((void **)&restartInfoRecord,
		"RestartInfoRecord",
		sizeof(RestartInfoRecord),
		cnoOfRestartInfoRec);
  
  deallocRecord((void **)&tablerec,"Tablerec",
		sizeof(Tablerec), 
		cnoOfTablerec);
  
  deallocRecord((void **)&tableDescriptor, "TableDescriptor",
		sizeof(TableDescriptor),
		cnoOfTabDescrRec);
  
  deallocRecord((void **)&undoPage,"UndoPage",
		sizeof(UndoPage),
		cnoOfUndoPage);

}//Dbtup::~Dbtup()

BLOCK_FUNCTIONS(Dbtup)

/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* ----- GENERAL SIGNAL MULTIPLEXER (FS + CONTINUEB) -------------- */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
void Dbtup::execFSCLOSECONF(Signal* signal) 
{
  PendingFileOpenInfoPtr pfoPtr;
  ljamEntry();
  pfoPtr.i = signal->theData[0];
  ptrCheckGuard(pfoPtr, cnoOfConcurrentOpenOp, pendingFileOpenInfo);
  switch (pfoPtr.p->pfoOpenType) {
  case LCP_DATA_FILE_CLOSE:
  {
    CheckpointInfoPtr ciPtr;
    ljam();
    ciPtr.i = pfoPtr.p->pfoCheckpointInfoP;
    ptrCheckGuard(ciPtr, cnoOfLcpRec, checkpointInfo);
    ciPtr.p->lcpDataFileHandle = RNIL;
    lcpClosedDataFileLab(signal, ciPtr);
    break;
  }
  case LCP_UNDO_FILE_CLOSE:
  {
    LocalLogInfoPtr lliPtr;
    ljam();
    lliPtr.i = pfoPtr.p->pfoCheckpointInfoP;
    ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
    lliPtr.p->lliUndoFileHandle = RNIL;
    lcpEndconfLab(signal);
    break;
  }
  case LCP_UNDO_FILE_READ:
    ljam();
    endExecUndoLogLab(signal, pfoPtr.p->pfoCheckpointInfoP);
    break;
  case LCP_DATA_FILE_READ:
    ljam();
    rfrClosedDataFileLab(signal, pfoPtr.p->pfoCheckpointInfoP);
    break;
  default:
    ndbrequire(false);
    break;
  }//switch
  releasePendingFileOpenInfoRecord(pfoPtr);
}//Dbtup::execFSCLOSECONF()

void Dbtup::execFSOPENCONF(Signal* signal) 
{
  PendingFileOpenInfoPtr pfoPtr;

  ljamEntry();
  pfoPtr.i = signal->theData[0];
  Uint32 fileHandle = signal->theData[1];
  ptrCheckGuard(pfoPtr, cnoOfConcurrentOpenOp, pendingFileOpenInfo);
  switch (pfoPtr.p->pfoOpenType) {
  case LCP_DATA_FILE_READ:
  {
    RestartInfoRecordPtr riPtr;
    ljam();
    riPtr.i = pfoPtr.p->pfoRestartInfoP;
    ptrCheckGuard(riPtr, cnoOfRestartInfoRec, restartInfoRecord);
    riPtr.p->sriDataFileHandle = fileHandle;
    rfrReadRestartInfoLab(signal, riPtr);
    break;
  }
  case LCP_UNDO_FILE_READ:
  {
    RestartInfoRecordPtr riPtr;
    LocalLogInfoPtr lliPtr;
    DiskBufferSegmentInfoPtr dbsiPtr;

    ljam();
    riPtr.i = pfoPtr.p->pfoRestartInfoP;
    ptrCheckGuard(riPtr, cnoOfRestartInfoRec, restartInfoRecord);
    lliPtr.i = riPtr.p->sriLocalLogInfoP;
    ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
    lliPtr.p->lliUndoFileHandle = fileHandle;
    dbsiPtr.i = riPtr.p->sriDataBufferSegmentP;
    ptrCheckGuard(dbsiPtr, cnoOfConcurrentWriteOp, diskBufferSegmentInfo);
    rfrLoadDataPagesLab(signal, riPtr, dbsiPtr);
    break;
  }
  case LCP_DATA_FILE_WRITE_WITH_UNDO:
  {
    CheckpointInfoPtr ciPtr;
    LocalLogInfoPtr lliPtr;

    ljam();
    ciPtr.i = pfoPtr.p->pfoCheckpointInfoP;
    ptrCheckGuard(ciPtr, cnoOfLcpRec, checkpointInfo);
    lliPtr.i = ciPtr.p->lcpLocalLogInfoP;
    ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
    ciPtr.p->lcpDataFileHandle = fileHandle;
    if (lliPtr.p->lliUndoFileHandle != RNIL) {
      ljam();
      signal->theData[0] = ciPtr.p->lcpUserptr;
      signal->theData[1] = ciPtr.i;
      sendSignal(ciPtr.p->lcpBlockref, GSN_TUP_PREPLCPCONF, signal, 2, JBB);
    }//if
    break;
  }
  case LCP_DATA_FILE_WRITE:
  {
    CheckpointInfoPtr ciPtr;

    ljam();
    ciPtr.i = pfoPtr.p->pfoCheckpointInfoP;
    ptrCheckGuard(ciPtr, cnoOfLcpRec, checkpointInfo);
    ciPtr.p->lcpDataFileHandle = fileHandle;
    signal->theData[0] = ciPtr.p->lcpUserptr;
    signal->theData[1] = ciPtr.i;
    sendSignal(ciPtr.p->lcpBlockref, GSN_TUP_PREPLCPCONF, signal, 2, JBB);
    break;
  }
  case LCP_UNDO_FILE_WRITE:
  {
    CheckpointInfoPtr ciPtr;
    LocalLogInfoPtr lliPtr;

    ljam();
    ciPtr.i = pfoPtr.p->pfoCheckpointInfoP;
    ptrCheckGuard(ciPtr, cnoOfLcpRec, checkpointInfo);
    lliPtr.i = ciPtr.p->lcpLocalLogInfoP;
    ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
    lliPtr.p->lliUndoFileHandle = fileHandle;
    if (ciPtr.p->lcpDataFileHandle != RNIL) {
      ljam();
      signal->theData[0] = ciPtr.p->lcpUserptr;
      signal->theData[1] = ciPtr.i;
      sendSignal(ciPtr.p->lcpBlockref, GSN_TUP_PREPLCPCONF, signal, 2, JBB);
    }//if
    break;
  }
  default:
    ndbrequire(false);
    break;
  }//switch
  releasePendingFileOpenInfoRecord(pfoPtr);
}//Dbtup::execFSOPENCONF()

void Dbtup::execFSREADCONF(Signal* signal) 
{
  DiskBufferSegmentInfoPtr dbsiPtr;
  ljamEntry();
  dbsiPtr.i = signal->theData[0];
  ptrCheckGuard(dbsiPtr, cnoOfConcurrentWriteOp, diskBufferSegmentInfo);
  switch (dbsiPtr.p->pdxOperation) {
  case CHECKPOINT_DATA_READ:
  {
    RestartInfoRecordPtr riPtr;
    ljam();
    riPtr.i = dbsiPtr.p->pdxRestartInfoP;
    ptrCheckGuard(riPtr, cnoOfRestartInfoRec, restartInfoRecord);
/************************************************************/
/*       VERIFY THAT THE PAGES ARE CORRECT, HAVE A CORRECT  */
/*       STATE AND A CORRECT PAGE ID.                       */
/************************************************************/
    ndbrequire(dbsiPtr.p->pdxNumDataPages <= 16);
    for (Uint32 i = 0; i < dbsiPtr.p->pdxNumDataPages; i++) {
      PagePtr pagePtr;
      ljam();
      pagePtr.i = dbsiPtr.p->pdxDataPage[i];
      ptrCheckGuard(pagePtr, cnoOfPage, page);
      ndbrequire(pagePtr.p->pageWord[ZPAGE_STATE_POS] != 0);
      ndbrequire(pagePtr.p->pageWord[ZPAGE_STATE_POS] <= ZAC_MM_FREE_COPY);
      ndbrequire(pagePtr.p->pageWord[ZPAGE_FRAG_PAGE_ID_POS] == ((dbsiPtr.p->pdxFilePage - 1) + i));
    }//for
    rfrLoadDataPagesLab(signal, riPtr, dbsiPtr);
    break;
  }
  case CHECKPOINT_DATA_READ_PAGE_ZERO:
  {
    ljam();
    rfrInitRestartInfoLab(signal, dbsiPtr);
    break;
  }
  case CHECKPOINT_UNDO_READ:
  {
    LocalLogInfoPtr lliPtr;
    ljam();
    lliPtr.i = dbsiPtr.p->pdxCheckpointInfoP;
    ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
    xlcGetNextRecordLab(signal, dbsiPtr, lliPtr);
    break;
  }
  case CHECKPOINT_UNDO_READ_FIRST:
    ljam();
    rfrReadSecondUndoLogLab(signal, dbsiPtr);
    break;
  default:
    ndbrequire(false);
    break;
  }//switch
}//Dbtup::execFSREADCONF()

void Dbtup::execFSWRITECONF(Signal* signal) 
{
  DiskBufferSegmentInfoPtr dbsiPtr;

  ljamEntry();
  dbsiPtr.i = signal->theData[0];
  ptrCheckGuard(dbsiPtr, cnoOfConcurrentWriteOp, diskBufferSegmentInfo);
  switch (dbsiPtr.p->pdxOperation) {
  case CHECKPOINT_DATA_WRITE:
    ljam();
    lcpSaveDataPageLab(signal, dbsiPtr.p->pdxCheckpointInfoP);
    break;
  case CHECKPOINT_DATA_WRITE_LAST:
  {
    CheckpointInfoPtr ciPtr;
    ljam();
    ciPtr.i = dbsiPtr.p->pdxCheckpointInfoP;
    ptrCheckGuard(ciPtr, cnoOfLcpRec, checkpointInfo);
    lcpFlushLogLab(signal, ciPtr);
    break;
  }
  case CHECKPOINT_DATA_WRITE_FLUSH:
  {
    ljam();
    Uint32 ciIndex = dbsiPtr.p->pdxCheckpointInfoP;
    freeDiskBufferSegmentRecord(signal, dbsiPtr);
    lcpCompletedLab(signal, ciIndex);
    break;
  }
  case CHECKPOINT_UNDO_WRITE_FLUSH:
  {
    ljam();
    Uint32 ciIndex = dbsiPtr.p->pdxCheckpointInfoP;
    freeDiskBufferSegmentRecord(signal, dbsiPtr);
    lcpFlushRestartInfoLab(signal, ciIndex);
    break;
  }