dbtuplcp.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 <signaldata/FsConf.hpp>
#include <signaldata/FsRef.hpp>

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

/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
/* -------------------- LOCAL CHECKPOINT MODULE ------------------- */
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
void Dbtup::execTUP_PREPLCPREQ(Signal* signal) 
{
  CheckpointInfoPtr ciPtr;
  DiskBufferSegmentInfoPtr dbsiPtr;
  FragrecordPtr regFragPtr;
  LocalLogInfoPtr lliPtr;
  TablerecPtr regTabPtr;

  ljamEntry();
  Uint32 userptr = signal->theData[0];
  BlockReference userblockref = signal->theData[1];
  regTabPtr.i = signal->theData[2];
  ptrCheckGuard(regTabPtr, cnoOfTablerec, tablerec);
  Uint32 fragId = signal->theData[3];
  Uint32 checkpointNumber = signal->theData[4];
  cundoFileVersion = signal->theData[5];

  getFragmentrec(regFragPtr, fragId, regTabPtr.p);
  ndbrequire(regTabPtr.i != RNIL);
  seizeCheckpointInfoRecord(ciPtr);

  lliPtr.i = (cundoFileVersion << 2) + (regTabPtr.i & 0x3);
  ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
  cnoOfDataPagesToDiskWithoutSynch = 0;

  ciPtr.p->lcpDataFileHandle = RNIL;
  ciPtr.p->lcpCheckpointVersion = checkpointNumber;
  ciPtr.p->lcpLocalLogInfoP = lliPtr.i;
  ciPtr.p->lcpFragmentP = regFragPtr.i;	/* SET THE FRAGMENT              */
  ciPtr.p->lcpFragmentId = fragId;	/* SAVE THE FRAGMENT IDENTITY    */
  ciPtr.p->lcpTabPtr = regTabPtr.i;	/* SET THE TABLE POINTER         */
  ciPtr.p->lcpBlockref = userblockref;	/* SET THE BLOCK REFERENCE       */
  ciPtr.p->lcpUserptr = userptr;	/* SET THE USERPOINTER           */

   /***************************************************************/
   /* OPEN THE UNDO FILE FOR WRITE                                */
   /* UPON FSOPENCONF                                             */
   /***************************************************************/
  if (lliPtr.p->lliActiveLcp == 0) {                  /* IS THE UNDO LOG FILE OPEN?    */
    PendingFileOpenInfoPtr undoPfoiPtr;
    UndoPagePtr regUndoPagePtr;

    ljam();
    lliPtr.p->lliPrevRecordId = 0;
    lliPtr.p->lliLogFilePage = 0;
    lliPtr.p->lliUndoPagesToDiskWithoutSynch = 0;
    lliPtr.p->lliUndoWord = ZUNDO_PAGE_HEADER_SIZE;

    seizeUndoBufferSegment(signal, regUndoPagePtr);
    seizeDiskBufferSegmentRecord(dbsiPtr);
    dbsiPtr.p->pdxBuffertype = UNDO_PAGES;
    for (Uint32 i = 0; i < ZUB_SEGMENT_SIZE; i++) {
      dbsiPtr.p->pdxDataPage[i] = regUndoPagePtr.i + i;
    }//for
    dbsiPtr.p->pdxFilePage = lliPtr.p->lliLogFilePage;
    lliPtr.p->lliUndoPage = regUndoPagePtr.i;
    lliPtr.p->lliUndoBufferSegmentP = dbsiPtr.i;
                                                /* F LEVEL NOT USED              */
    Uint32 fileType = 1;	                /* VERSION                       */
    fileType = (fileType << 8) | 2;	        /* .LOCLOG                       */
    fileType = (fileType << 8) | 6;	        /* D6                            */
    fileType = (fileType << 8) | 0xff;	        /* DON'T USE P DIRECTORY LEVEL   */
    Uint32 fileFlag = 0x301;	                /* CREATE, WRITE ONLY, TRUNCATE  */

    seizePendingFileOpenInfoRecord(undoPfoiPtr);
    undoPfoiPtr.p->pfoOpenType = LCP_UNDO_FILE_WRITE;
    undoPfoiPtr.p->pfoCheckpointInfoP = ciPtr.i;

    signal->theData[0] = cownref;
    signal->theData[1] = undoPfoiPtr.i;
    signal->theData[2] = lliPtr.i;
    signal->theData[3] = 0xFFFFFFFF;
    signal->theData[4] = cundoFileVersion;
    signal->theData[5] = fileType;
    signal->theData[6] = fileFlag;
    sendSignal(NDBFS_REF, GSN_FSOPENREQ, signal, 7, JBA);
  }//if
   /***************************************************************/
   /* OPEN THE DATA FILE FOR WRITE                                */
   /* THE FILE HANDLE WILL BE SET IN THE CHECKPOINT_INFO_RECORD   */
   /* UPON FSOPENCONF                                             */
   /***************************************************************/
   /* OPEN THE DATA FILE IN THE FOLLOWING FORM              */
   /* D5/DBTUP/T<TABID>/F<FRAGID>/S<CHECKPOINT_NUMBER>.DATA */

  PendingFileOpenInfoPtr dataPfoiPtr;

  Uint32 fileType = 1;	                /* VERSION                       */
  fileType = (fileType << 8) | 0;	/* .DATA                         */
  fileType = (fileType << 8) | 5;	/* D5                            */
  fileType = (fileType << 8) | 0xff;	/* DON'T USE P DIRECTORY LEVEL   */
  Uint32 fileFlag = 0x301;	        /* CREATE, WRITE ONLY, TRUNCATE  */

  seizePendingFileOpenInfoRecord(dataPfoiPtr);	/* SEIZE A NEW FILE OPEN INFO    */
  if (lliPtr.p->lliActiveLcp == 0) {
    ljam();
    dataPfoiPtr.p->pfoOpenType = LCP_DATA_FILE_WRITE_WITH_UNDO;
  } else {
    ljam();
    dataPfoiPtr.p->pfoOpenType = LCP_DATA_FILE_WRITE;
  }//if
  dataPfoiPtr.p->pfoCheckpointInfoP = ciPtr.i;

   /* LET'S OPEN THE DATA FILE FOR WRITE */
   /* INCREASE NUMBER OF ACTIVE CHECKPOINTS */
  lliPtr.p->lliActiveLcp = 1;
  signal->theData[0] = cownref;
  signal->theData[1] = dataPfoiPtr.i;
  signal->theData[2] = ciPtr.p->lcpTabPtr;
  signal->theData[3] = ciPtr.p->lcpFragmentId;
  signal->theData[4] = ciPtr.p->lcpCheckpointVersion;
  signal->theData[5] = fileType;
  signal->theData[6] = fileFlag;
  sendSignal(NDBFS_REF, GSN_FSOPENREQ, signal, 7, JBA);
  return;
}//Dbtup::execTUP_PREPLCPREQ()

/* ---------------------------------------------------------------- */
/* ------------------------ START CHECKPOINT  --------------------- */
/* ---------------------------------------------------------------- */
void Dbtup::execTUP_LCPREQ(Signal* signal) 
{
  CheckpointInfoPtr ciPtr;
  DiskBufferSegmentInfoPtr dbsiPtr;
  FragrecordPtr regFragPtr;
  LocalLogInfoPtr lliPtr;

  ljamEntry();
//  Uint32 userptr = signal->theData[0];
//  BlockReference userblockref = signal->theData[1];
  ciPtr.i = signal->theData[2];

  ptrCheckGuard(ciPtr, cnoOfLcpRec, checkpointInfo);
  regFragPtr.i = ciPtr.p->lcpFragmentP;
  ptrCheckGuard(regFragPtr, cnoOfFragrec, fragrecord);

/* ---------------------------------------------------------------- */
/*       ASSIGNING A VALUE DIFFERENT FROM RNIL TO CHECKPOINT VERSION*/
/*       TRIGGERS THAT UNDO LOGGING WILL START FOR THIS FRAGMENT.   */
/*       WE ASSIGN IT THE POINTER TO THE CHECKPOINT RECORD FOR      */
/*       OPTIMISATION OF THE WRITING OF THE UNDO LOG.               */
/* ---------------------------------------------------------------- */
  regFragPtr.p->checkpointVersion = ciPtr.p->lcpLocalLogInfoP;	/* MARK START OF UNDO LOGGING */

  regFragPtr.p->maxPageWrittenInCheckpoint = getNoOfPages(regFragPtr.p);
  regFragPtr.p->minPageNotWrittenInCheckpoint = 0;
  ndbrequire(getNoOfPages(regFragPtr.p) > 0);
  allocDataBufferSegment(signal, dbsiPtr);

  dbsiPtr.p->pdxNumDataPages =  0;
  dbsiPtr.p->pdxFilePage = 1;
  ciPtr.p->lcpDataBufferSegmentP = dbsiPtr.i;
  dbsiPtr.p->pdxCheckpointInfoP = ciPtr.i;
  ciPtr.p->lcpNoOfPages = getNoOfPages(regFragPtr.p);
  ciPtr.p->lcpNoCopyPagesAlloc = regFragPtr.p->noCopyPagesAlloc;
  ciPtr.p->lcpEmptyPrimPage = regFragPtr.p->emptyPrimPage;
  ciPtr.p->lcpThFreeFirst = regFragPtr.p->thFreeFirst;
  ciPtr.p->lcpThFreeCopyFirst = regFragPtr.p->thFreeCopyFirst;
  lliPtr.i = ciPtr.p->lcpLocalLogInfoP;
  ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
/* ---------------------------------------------------------------- */
/* --- PERFORM A COPY OF THE TABLE DESCRIPTOR FOR THIS FRAGMENT --- */
/* ---------------------------------------------------------------- */
  cprAddLogHeader(signal,
                  lliPtr.p,
                  ZTABLE_DESCRIPTOR,
                  ciPtr.p->lcpTabPtr,
                  ciPtr.p->lcpFragmentId);

/* ---------------------------------------------------------------- */
/*       CONTINUE WITH SAVING ACTIVE OPERATIONS AFTER A REAL-TIME   */
/*       BREAK.                                                     */
/* ---------------------------------------------------------------- */
  ciPtr.p->lcpTmpOperPtr = regFragPtr.p->firstusedOprec;
  lcpSaveCopyListLab(signal, ciPtr);
  return;
}//Dbtup::execTUP_LCPREQ()

void Dbtup::allocDataBufferSegment(Signal* signal, DiskBufferSegmentInfoPtr& dbsiPtr) 
{
  UndoPagePtr regUndoPagePtr;

  seizeDiskBufferSegmentRecord(dbsiPtr);
  dbsiPtr.p->pdxBuffertype = COMMON_AREA_PAGES;
  ndbrequire(cfirstfreeUndoSeg != RNIL);
  if (cnoFreeUndoSeg == ZMIN_PAGE_LIMIT_TUP_COMMITREQ) {
    EXECUTE_DIRECT(DBLQH, GSN_TUP_COM_BLOCK, signal, 1);
    ljamEntry();
  }//if
  cnoFreeUndoSeg--;
  ndbrequire(cnoFreeUndoSeg >= 0);

  regUndoPagePtr.i = cfirstfreeUndoSeg;
  ptrCheckGuard(regUndoPagePtr, cnoOfUndoPage, undoPage);
  cfirstfreeUndoSeg = regUndoPagePtr.p->undoPageWord[ZPAGE_NEXT_POS];
  regUndoPagePtr.p->undoPageWord[ZPAGE_NEXT_POS] = RNIL;
  for (Uint32 i = 0; i < ZUB_SEGMENT_SIZE; i++) {
    dbsiPtr.p->pdxDataPage[i] = regUndoPagePtr.i + i;
  }//for
}//Dbtup::allocDataBufferSegment()

/* ---------------------------------------------------------------- */
/* --- PERFORM A COPY OF THE ACTIVE OPERATIONS FOR THIS FRAGMENT -- */
/* ---------------------------------------------------------------- */
void Dbtup::lcpSaveCopyListLab(Signal* signal, CheckpointInfoPtr ciPtr) 
{
  FragrecordPtr regFragPtr;
  LocalLogInfoPtr lliPtr;
  OperationrecPtr regOpPtr;

  regFragPtr.i = ciPtr.p->lcpFragmentP;
  ptrCheckGuard(regFragPtr, cnoOfFragrec, fragrecord);
  lliPtr.i = ciPtr.p->lcpLocalLogInfoP;
  ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
  regOpPtr.i = ciPtr.p->lcpTmpOperPtr;

/* -------------------------------------------------------------------------------- */
/* TRAVERSE THE ENTIRE BLOCK OF OPERATIONS. CHECK IF THERE ARE EXISTING COPYS OF    */
/* TUPLES IN THE CHECKPOINTED FRAGMENT. SAVE THOSE IN A LIST IN THE FOLLOWING FORM: */
/*                                                                                  */
/*    SOURCE PAGE                                                                   */
/*    SOURCE INDEX                                                                  */
/*    COPY PAGE                                                                     */
/*    COPY INDEX                                                                    */
/* -------------------------------------------------------------------------------- */
  Uint32 loopCount = 0;
  while ((regOpPtr.i != RNIL) && (loopCount < 50)) {
    ljam();
    ptrCheckGuard(regOpPtr, cnoOfOprec, operationrec);
    if (regOpPtr.p->realPageId != RNIL) {
/* ---------------------------------------------------------------- */
// We ensure that we have actually allocated the tuple header and
// also found it. Otherwise we will fill the undo log with garbage.
/* ---------------------------------------------------------------- */
      if (regOpPtr.p->optype == ZUPDATE || 
	  (regOpPtr.p->optype == ZINSERT && regOpPtr.p->deleteInsertFlag)) {
        ljam();
        if (regOpPtr.p->realPageIdC != RNIL) {
/* ---------------------------------------------------------------- */
// We ensure that we have actually allocated the tuple header copy.
// Otherwise we will fill the undo log with garbage.
/* ---------------------------------------------------------------- */
          cprAddLogHeader(signal,
                          lliPtr.p,
                          ZLCPR_ABORT_UPDATE,
                          ciPtr.p->lcpTabPtr,
                          ciPtr.p->lcpFragmentId);
          cprAddAbortUpdate(signal, lliPtr.p, regOpPtr.p);
        }//if
      } else if (regOpPtr.p->optype == ZINSERT) {
        ljam();
        cprAddUndoLogRecord(signal,
                            ZLCPR_ABORT_INSERT,
                            regOpPtr.p->fragPageId,
                            regOpPtr.p->pageIndex,
                            regOpPtr.p->tableRef,
                            regOpPtr.p->fragId,
                            regFragPtr.p->checkpointVersion);
      } else {
        ndbrequire(regOpPtr.p->optype == ZDELETE);
        ljam();
        cprAddUndoLogRecord(signal,
                            ZINDICATE_NO_OP_ACTIVE,
                            regOpPtr.p->fragPageId,
                            regOpPtr.p->pageIndex,
                            regOpPtr.p->tableRef,