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

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

/* **************************************************************** */
/* ********************* SYSTEM RESTART MANAGER ******************* */
/* **************************************************************** */
/***************************************************************/
/* CHECK RESTART STATE AND SET NEW STATE CALLED IN OPEN,       */
/* READ AND COPY STATES                                        */
/***************************************************************/
void Dbtup::execTUP_SRREQ(Signal* signal) 
{
  RestartInfoRecordPtr riPtr;
  PendingFileOpenInfoPtr pfoiPtr;

  ljamEntry();
  Uint32 userPtr = signal->theData[0];
  Uint32 userBlockref = signal->theData[1];
  Uint32 tableId = signal->theData[2];
  Uint32 fragId = signal->theData[3];
  Uint32 checkpointNumber = signal->theData[4];

  seizeRestartInfoRecord(riPtr);

  riPtr.p->sriUserptr = userPtr;
  riPtr.p->sriBlockref = userBlockref;
  riPtr.p->sriState = OPENING_DATA_FILE;
  riPtr.p->sriCheckpointVersion = checkpointNumber;
  riPtr.p->sriFragid = fragId;
  riPtr.p->sriTableId = tableId;

   /* OPEN THE DATA FILE IN THE FOLLOWING FORM              */
   /* D5/DBTUP/T<TABID>/F<FRAGID>/S<CHECKPOINT_NUMBER>.DATA */
  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 = 0;	                /* READ ONLY                     */

  seizePendingFileOpenInfoRecord(pfoiPtr);
  pfoiPtr.p->pfoOpenType = LCP_DATA_FILE_READ;
  pfoiPtr.p->pfoRestartInfoP = riPtr.i;

  signal->theData[0] = cownref;
  signal->theData[1] = pfoiPtr.i;
  signal->theData[2] = tableId;
  signal->theData[3] = fragId;
  signal->theData[4] = checkpointNumber;
  signal->theData[5] = fileType;
  signal->theData[6] = fileFlag;
  sendSignal(NDBFS_REF, GSN_FSOPENREQ, signal, 7, JBA);
  return;
}//Dbtup::execTUP_SRREQ()

void Dbtup::seizeRestartInfoRecord(RestartInfoRecordPtr& riPtr) 
{
  riPtr.i = cfirstfreeSri;
  ptrCheckGuard(riPtr, cnoOfRestartInfoRec, restartInfoRecord);
  cfirstfreeSri = riPtr.p->sriNextRec;
  riPtr.p->sriNextRec = RNIL;
}//Dbtup::seizeRestartInfoRecord()

void Dbtup::rfrReadRestartInfoLab(Signal* signal, RestartInfoRecordPtr riPtr) 
{
  DiskBufferSegmentInfoPtr dbsiPtr;

  seizeDiskBufferSegmentRecord(dbsiPtr);
  riPtr.p->sriDataBufferSegmentP = dbsiPtr.i;
  Uint32 retPageRef = RNIL;
  Uint32 noAllocPages = 1;
  Uint32 noOfPagesAllocated;
  {
    /**
     * Use low pages for 0-pages during SR
     *   bitmask of free pages is kept in c_sr_free_page_0
     */
    Uint32 tmp = c_sr_free_page_0;
    for(Uint32 i = 1; i<(1+MAX_PARALLELL_TUP_SRREQ); i++){
      if(tmp & (1 << i)){
	retPageRef = i;
	c_sr_free_page_0 = tmp & (~(1 << i));
	break;
      }
    }
    ndbrequire(retPageRef != RNIL);
  }
  
  dbsiPtr.p->pdxDataPage[0] = retPageRef;
  dbsiPtr.p->pdxNumDataPages = 1;
  dbsiPtr.p->pdxFilePage = 0;
  rfrReadNextDataSegment(signal, riPtr, dbsiPtr);
  dbsiPtr.p->pdxOperation = CHECKPOINT_DATA_READ_PAGE_ZERO;
}//Dbtup::rfrReadRestartInfoLab()

/***************************************************************/
/* THE RESTART INFORMATION IS NOW READ INTO THE DATA BUFFER    */
/* USE THE RESTART INFORMATION TO INITIATE THE RESTART RECORD  */
/***************************************************************/
void 
Dbtup::rfrInitRestartInfoLab(Signal* signal, DiskBufferSegmentInfoPtr dbsiPtr) 
{
  Uint32 TzeroDataPage[64];
  Uint32 Ti;
  FragrecordPtr regFragPtr;
  LocalLogInfoPtr lliPtr;
  PagePtr pagePtr;
  RestartInfoRecordPtr riPtr;
  TablerecPtr regTabPtr;

  riPtr.i = dbsiPtr.p->pdxRestartInfoP;
  ptrCheckGuard(riPtr, cnoOfRestartInfoRec, restartInfoRecord);
  
  regTabPtr.i = riPtr.p->sriTableId;
  ptrCheckGuard(regTabPtr, cnoOfTablerec, tablerec);

  Uint32 fragId = riPtr.p->sriFragid;
  getFragmentrec(regFragPtr, fragId, regTabPtr.p);
  riPtr.p->sriFragP = regFragPtr.i;
  
  /* ----- PAGE ALLOCATION --- */
  /* ALLOCATE PAGES TO FRAGMENT, INSERT THEM INTO PAGE RANGE TABLE AND       */
  /* ALSO CONVERT THEM INTO EMPTY PAGES AND INSERT THEM INTO THE EMPTY LIST  */
  /* OF THE FRAGMENT. SET ALL LISTS OF FREE PAGES TO RNIL                    */

  ndbrequire(cfirstfreerange != RNIL);
  pagePtr.i = dbsiPtr.p->pdxDataPage[0];
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  for (Ti = 0; Ti < 63; Ti++) {
    /***************************************************************/
    // Save Important content from Page zero in stack variable so
    // that we can immediately release page zero.
    /***************************************************************/
    TzeroDataPage[Ti] = pagePtr.p->pageWord[Ti];
  }//for
  /************************************************************************/
  /* NOW WE DON'T NEED THE RESTART INFO BUFFER PAGE ANYMORE               */
  /* LETS REMOVE IT AND REUSE THE SEGMENT FOR REAL DATA PAGES             */
  /* REMOVE ONE PAGE ONLY, PAGEP IS ALREADY SET TO THE RESTART INFO PAGE */
  /************************************************************************/
  {
    ndbrequire(pagePtr.i > 0 && pagePtr.i <= MAX_PARALLELL_TUP_SRREQ);
    c_sr_free_page_0 |= (1 << pagePtr.i);
  }

  Uint32 undoFileVersion = TzeroDataPage[ZSRI_UNDO_FILE_VER];
  lliPtr.i = (undoFileVersion << 2) + (regTabPtr.i & 0x3);
  ptrCheckGuard(lliPtr, cnoOfParallellUndoFiles, localLogInfo);
  riPtr.p->sriLocalLogInfoP = lliPtr.i;

  ndbrequire(regFragPtr.p->fragTableId == regTabPtr.i);
  ndbrequire(regFragPtr.p->fragmentId == fragId);
  
  regFragPtr.p->fragStatus = SYSTEM_RESTART;

  regFragPtr.p->noCopyPagesAlloc = TzeroDataPage[ZSRI_NO_COPY_PAGES_ALLOC];

  riPtr.p->sriCurDataPageFromBuffer = 0;
  riPtr.p->sriNumDataPages = TzeroDataPage[ZSRI_NO_OF_FRAG_PAGES_POS];

  ndbrequire(riPtr.p->sriNumDataPages >= regFragPtr.p->noOfPages);
  const Uint32 pageCount = riPtr.p->sriNumDataPages - regFragPtr.p->noOfPages;
  if(pageCount > 0){
    Uint32 noAllocPages = allocFragPages(regFragPtr.p, pageCount);
    ndbrequireErr(noAllocPages == pageCount, NDBD_EXIT_SR_OUT_OF_DATAMEMORY);
  }//if
  ndbrequire(getNoOfPages(regFragPtr.p) == riPtr.p->sriNumDataPages);

/***************************************************************/
// Set the variables on fragment record which might have been
// affected by allocFragPages.
/***************************************************************/

  regFragPtr.p->emptyPrimPage = TzeroDataPage[ZSRI_EMPTY_PRIM_PAGE];
  regFragPtr.p->thFreeFirst = TzeroDataPage[ZSRI_TH_FREE_FIRST];
  regFragPtr.p->thFreeCopyFirst = TzeroDataPage[ZSRI_TH_FREE_COPY_FIRST];

/***************************************************************/
/* THE RESTART INFORMATION IS NOW READ INTO THE DATA BUFFER    */
/* USE THE RESTART INFORMATION TO INITIATE THE FRAGMENT        */
/***************************************************************/
  /**
   * IF THIS UNDO FILE IS NOT OPEN, IT WILL BE OPENED HERE AND THE EXECUTION 
   * WILL CONTINUE WHEN THE FSOPENCONF IS ENTERED.
   * IF IT'S ALREADY IN USE THE EXECUTION WILL CONTINUE BY A 
   * CONTINUE B SIGNAL
   */
  if (lliPtr.p->lliActiveLcp == 0) {
    PendingFileOpenInfoPtr pfoiPtr;
    ljam();
/***************************************************************/
/* OPEN THE UNDO FILE FOR READ                                 */
/* THE FILE HANDLE WILL BE SET IN THE LOCAL_LOG_INFO_REC       */
/* UPON FSOPENCONF                                             */
/***************************************************************/
    cnoOfLocalLogInfo++;
                                                /* 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 = 0;	                /* READ ONLY                     */

    seizePendingFileOpenInfoRecord(pfoiPtr);
    pfoiPtr.p->pfoOpenType = LCP_UNDO_FILE_READ;
    pfoiPtr.p->pfoRestartInfoP = riPtr.i;

    signal->theData[0] = cownref;
    signal->theData[1] = pfoiPtr.i;
    signal->theData[2] = lliPtr.i;
    signal->theData[3] = 0xFFFFFFFF;
    signal->theData[4] = undoFileVersion;
    signal->theData[5] = fileType;
    signal->theData[6] = fileFlag;
    sendSignal(NDBFS_REF, GSN_FSOPENREQ, signal, 7, JBA);

    lliPtr.p->lliPrevRecordId = 0;
    lliPtr.p->lliActiveLcp = 1;
    lliPtr.p->lliNumFragments = 1;
  } else {
    ljam();
    signal->theData[0] = ZCONT_LOAD_DP;
    signal->theData[1] = riPtr.i;
    sendSignal(cownref, GSN_CONTINUEB, signal, 2, JBB);
    lliPtr.p->lliNumFragments++;
  }//if
   /* RETAIN THE HIGH- AND LOWSCORE ID:S OF THE LOGRECORD POSITIONS. WE HAVE TO EXECUTE THE  */
   /* UNDO LOG BETWEEN THE END AND START RECORDS FOR ALL RECORDS THAT INCLUDE FRAGMENTS OF   */
   /* THE RIGHT CHECKPOINT VERSION TO COMPLETE THE OPERATION WE HAVE TO RUN ALL LOGS THAT    */
   /* HAS THE NUMBER OF LCP ELEMENT GREATER THAN 0, I.E. IS INCLUDED.                        */
  if (TzeroDataPage[ZSRI_UNDO_LOG_END_REC_ID] > lliPtr.p->lliPrevRecordId) {
    ljam();
    lliPtr.p->lliPrevRecordId = TzeroDataPage[ZSRI_UNDO_LOG_END_REC_ID];
    lliPtr.p->lliEndPageId = TzeroDataPage[ZSRI_UNDO_LOG_END_PAGE_ID];
  }//if
  return;
}//Dbtup::rfrInitRestartInfoLab()

/***************************************************************/
/* LOAD THE NEXT DATA PAGE SEGMENT INTO MEMORY                 */
/***************************************************************/
void Dbtup::rfrLoadDataPagesLab(Signal* signal, RestartInfoRecordPtr riPtr, DiskBufferSegmentInfoPtr dbsiPtr) 
{
  FragrecordPtr regFragPtr;

  if (riPtr.p->sriCurDataPageFromBuffer >= riPtr.p->sriNumDataPages) {
    ljam();
    rfrCompletedLab(signal, riPtr);
    return;
  }//if
  Uint32 startPage = riPtr.p->sriCurDataPageFromBuffer;
  Uint32 endPage;
  if ((startPage + ZDB_SEGMENT_SIZE) < riPtr.p->sriNumDataPages) {
    ljam();
    endPage = startPage + ZDB_SEGMENT_SIZE;
  } else {
    ljam();
    endPage = riPtr.p->sriNumDataPages;
  }//if
  regFragPtr.i = riPtr.p->sriFragP;
  ptrCheckGuard(regFragPtr, cnoOfFragrec, fragrecord);
  ndbrequire((endPage - startPage) <= 16);
  Uint32 i = 0;
  for (Uint32 pageId = startPage; pageId < endPage; pageId++) {
    ljam();
    dbsiPtr.p->pdxDataPage[i] = getRealpid(regFragPtr.p, pageId);
    i++;
  }//for
  dbsiPtr.p->pdxNumDataPages = endPage - startPage;	/* SET THE NUMBER OF DATA PAGES */
  riPtr.p->sriCurDataPageFromBuffer = endPage;
  dbsiPtr.p->pdxFilePage = startPage + 1;
  rfrReadNextDataSegment(signal, riPtr, dbsiPtr);
  return;
}//Dbtup::rfrLoadDataPagesLab()

void Dbtup::rfrCompletedLab(Signal* signal, RestartInfoRecordPtr riPtr) 
{
  PendingFileOpenInfoPtr pfoPtr;
/* ---------------------------------------------------------------------- */
/*       CLOSE THE DATA FILE BEFORE SENDING TUP_SRCONF                    */
/* ---------------------------------------------------------------------- */
  seizePendingFileOpenInfoRecord(pfoPtr);
  pfoPtr.p->pfoOpenType = LCP_DATA_FILE_READ;
  pfoPtr.p->pfoCheckpointInfoP = riPtr.i;

  signal->theData[0] = riPtr.p->sriDataFileHandle;
  signal->theData[1] = cownref;
  signal->theData[2] = pfoPtr.i;
  signal->theData[3] = 0;
  sendSignal(NDBFS_REF, GSN_FSCLOSEREQ, signal, 4, JBA);
}//Dbtup::rfrCompletedLab()

void Dbtup::rfrClosedDataFileLab(Signal* signal, Uint32 restartIndex) 
{
  RestartInfoRecordPtr riPtr;
  DiskBufferSegmentInfoPtr dbsiPtr;

  riPtr.i = restartIndex;
  ptrCheckGuard(riPtr, cnoOfRestartInfoRec, restartInfoRecord);
  riPtr.p->sriDataFileHandle = RNIL;
  dbsiPtr.i = riPtr.p->sriDataBufferSegmentP;
  ptrCheckGuard(dbsiPtr, cnoOfConcurrentWriteOp, diskBufferSegmentInfo);
  releaseDiskBufferSegmentRecord(dbsiPtr);
  signal->theData[0] = riPtr.p->sriUserptr;
  signal->theData[1] = riPtr.p->sriFragP;
  sendSignal(riPtr.p->sriBlockref, GSN_TUP_SRCONF, signal, 2, JBB);
  releaseRestartInfoRecord(riPtr);
}//Dbtup::rfrClosedDataFileLab()

/* ---------------------------------------------------------------- */
/* ---------------------- EXECUTE LOCAL LOG  ---------------------- */
/* ---------------------------------------------------------------- */
void Dbtup::execSTART_RECREQ(Signal* signal) 
{
  ljamEntry();
  clqhUserpointer = signal->theData[0];