dbaccmain.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 DBACC_C
#include "Dbacc.hpp"
#include <my_sys.h>

#include <AttributeHeader.hpp>
#include <signaldata/AccFrag.hpp>
#include <signaldata/AccScan.hpp>
#include <signaldata/AccLock.hpp>
#include <signaldata/EventReport.hpp>
#include <signaldata/FsConf.hpp>
#include <signaldata/FsRef.hpp>
#include <signaldata/FsRemoveReq.hpp>
#include <signaldata/DropTab.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <KeyDescriptor.hpp>


// TO_DO_RONM is a label for comments on what needs to be improved in future versions
// when more time is given.

#ifdef VM_TRACE
#define DEBUG(x) ndbout << "DBACC: "<< x << endl;
#else
#define DEBUG(x)
#endif


Uint32
Dbacc::remainingUndoPages(){  
  Uint32 HeadPage = cundoposition >> ZUNDOPAGEINDEXBITS;
  Uint32 TailPage = clastUndoPageIdWritten;

  // Head must be larger or same as tail
  ndbrequire(HeadPage>=TailPage);

  Uint32 UsedPages = HeadPage - TailPage;
  Int32 Remaining = cundopagesize - UsedPages;

  // There can not be more than cundopagesize remaining
  if (Remaining <= 0){
    // No more undolog, crash node
    progError(__LINE__, NDBD_EXIT_NO_MORE_UNDOLOG,
	      "There are more than 1Mbyte undolog writes outstanding");
  }
  return Remaining;
}

void
Dbacc::updateLastUndoPageIdWritten(Signal* signal, Uint32 aNewValue){
  if (remainingUndoPages() < ZMIN_UNDO_PAGES_AT_COMMIT) {
    clastUndoPageIdWritten = aNewValue;
    if (remainingUndoPages() >= ZMIN_UNDO_PAGES_AT_COMMIT) {
      jam();
      EXECUTE_DIRECT(DBLQH, GSN_ACC_COM_UNBLOCK, signal, 1);
      jamEntry();
    }//if
  } else {
    clastUndoPageIdWritten = aNewValue;
  }//if
}//Dbacc::updateLastUndoPageIdWritten()

void 
Dbacc::updateUndoPositionPage(Signal* signal, Uint32 aNewValue){
  if (remainingUndoPages() >= ZMIN_UNDO_PAGES_AT_COMMIT) {
    cundoposition = aNewValue;
    if (remainingUndoPages() < ZMIN_UNDO_PAGES_AT_COMMIT) {
      jam();
      EXECUTE_DIRECT(DBLQH, GSN_ACC_COM_BLOCK, signal, 1);
      jamEntry();
    }//if
  } else {
    cundoposition = aNewValue;
  }//if
}//Dbacc::updateUndoPositionPage()

// Signal entries and statement blocks
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/*                                                                                   */
/*       COMMON SIGNAL RECEPTION MODULE                                              */
/*                                                                                   */
/* --------------------------------------------------------------------------------- */

/* --------------------------------------------------------------------------------- */
/* ******************--------------------------------------------------------------- */
/* CONTINUEB                                       CONTINUE SIGNAL                   */
/* ******************------------------------------+                                 */
/*   SENDER: ACC,    LEVEL B       */
void Dbacc::execCONTINUEB(Signal* signal) 
{
  Uint32 tcase;

  jamEntry();
  tcase = signal->theData[0];
  tdata0 = signal->theData[1];
  tresult = 0;
  switch (tcase) {
  case ZLOAD_BAL_LCP_TIMER:
    if (clblPageOver == 0) {
      jam();
      clblPageCounter = clblPagesPerTick;
    } else {
      if (clblPageOver > clblPagesPerTick) {
        jam();
        clblPageOver = clblPageOver - clblPagesPerTick;
      } else {
        jam();
        clblPageOver = 0;
        clblPageCounter = clblPagesPerTick - clblPageOver;
      }//if
    }//if
    signal->theData[0] = ZLOAD_BAL_LCP_TIMER;
    sendSignalWithDelay(cownBlockref, GSN_CONTINUEB, signal, 100, 1);
    return;
    break;
  case ZINITIALISE_RECORDS:
    jam();
    initialiseRecordsLab(signal, signal->theData[3], signal->theData[4]);
    return;
    break;
  case ZSR_READ_PAGES_ALLOC:
    jam();
    fragrecptr.i = tdata0;
    ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
    srReadPagesAllocLab(signal);
    return;
    break;
  case ZSTART_UNDO:
    jam();
    startUndoLab(signal);
    return;
    break;
  case ZSEND_SCAN_HBREP:
    jam();
    sendScanHbRep(signal, tdata0);
    break;
  case ZREL_ROOT_FRAG:
    {
      jam();
      Uint32 tableId = signal->theData[1];
      releaseRootFragResources(signal, tableId);
      break;
    }
  case ZREL_FRAG:
    {
      jam();
      Uint32 fragIndex = signal->theData[1];
      releaseFragResources(signal, fragIndex);
      break;
    }
  case ZREL_DIR:
    {
      jam();
      Uint32 fragIndex = signal->theData[1];
      Uint32 dirIndex = signal->theData[2];
      Uint32 startIndex = signal->theData[3];
      releaseDirResources(signal, fragIndex, dirIndex, startIndex);
      break;
    }
  case ZREPORT_MEMORY_USAGE:{
    jam();
    static int c_currentMemUsed = 0;
    int now = (cnoOfAllocatedPages * 100)/cpagesize;
    const int thresholds[] = { 99, 90, 80, 0};
    
    Uint32 i = 0;
    const Uint32 sz = sizeof(thresholds)/sizeof(thresholds[0]);
    for(i = 0; i<sz; i++){
      if(now >= thresholds[i]){
	now = thresholds[i];
	break;
      }
    }
    
    if(now != c_currentMemUsed){
      reportMemoryUsage(signal, now > c_currentMemUsed ? 1 : -1);
    }
    
    c_currentMemUsed = now;
    
    signal->theData[0] = ZREPORT_MEMORY_USAGE;
    sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 2000, 1);    
    return;
  }

  case ZLCP_OP_WRITE_RT_BREAK:
  {
    operationRecPtr.i= signal->theData[1];
    fragrecptr.i= signal->theData[2];
    lcpConnectptr.i= signal->theData[3];
    ptrCheckGuard(operationRecPtr, coprecsize, operationrec);
    ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
    ptrCheckGuard(lcpConnectptr, clcpConnectsize, lcpConnectrec);
    lcp_write_op_to_undolog(signal);
    return;
  }
  default:
    ndbrequire(false);
    break;
  }//switch
  return;
}//Dbacc::execCONTINUEB()

/* ******************--------------------------------------------------------------- */
/* FSCLOSECONF                                       CLOSE FILE CONF                 */
/* ******************------------------------------+                                 */
/* SENDER: FS,     LEVEL B         */
void Dbacc::execFSCLOSECONF(Signal* signal) 
{
  jamEntry();
  fsConnectptr.i = signal->theData[0];
  ptrCheckGuard(fsConnectptr, cfsConnectsize, fsConnectrec);
  tresult = 0;
  switch (fsConnectptr.p->fsState) {
  case WAIT_CLOSE_UNDO:
    jam();
    releaseFsConnRec(signal);
    break;
  case LCP_CLOSE_DATA:
    jam();
    checkSyncUndoPagesLab(signal);
    return;
    break;
  case SR_CLOSE_DATA:
    jam();
    sendaccSrconfLab(signal);
    return;
    break;
  default:
    ndbrequire(false);
    break;
  }//switch
  return;
}//Dbacc::execFSCLOSECONF()


/* ******************--------------------------------------------------------------- */
/* FSOPENCONF                                         OPENFILE CONF                  */
/* ******************------------------------------+                                 */
/*   SENDER: FS,     LEVEL B       */
void Dbacc::execFSOPENCONF(Signal* signal) 
{
  jamEntry();
  fsConnectptr.i = signal->theData[0];
  ptrCheckGuard(fsConnectptr, cfsConnectsize, fsConnectrec);
  tuserptr = signal->theData[1];
  tresult = 0;	/* RESULT CHECK VALUE              */
  switch (fsConnectptr.p->fsState) {
  case WAIT_OPEN_UNDO_LCP:
    jam();
    lcpOpenUndofileConfLab(signal);
    return;
    break;
  case WAIT_OPEN_UNDO_LCP_NEXT:
    jam();
    fsConnectptr.p->fsPtr = tuserptr;
    return;
    break;
  case OPEN_UNDO_FILE_SR:
    jam();
    fsConnectptr.p->fsPtr = tuserptr;
    srStartUndoLab(signal);
    return;
    break;
  case WAIT_OPEN_DATA_FILE_FOR_WRITE:
    jam();
    lcpFsOpenConfLab(signal);
    return;
    break;
  case WAIT_OPEN_DATA_FILE_FOR_READ:
    jam();
    fsConnectptr.p->fsPtr = tuserptr;
    srFsOpenConfLab(signal);
    return;
    break;
  default:
    ndbrequire(false);
    break;
  }//switch
  return;
}//Dbacc::execFSOPENCONF()


/* ******************--------------------------------------------------------------- */
/* FSREADCONF                                          OPENFILE CONF                 */
/* ******************------------------------------+                                 */
/*   SENDER: FS,     LEVEL B       */
void Dbacc::execFSREADCONF(Signal* signal) 
{
  jamEntry();
  fsConnectptr.i = signal->theData[0];
  ptrCheckGuard(fsConnectptr, cfsConnectsize, fsConnectrec);
  tresult = 0;	/* RESULT CHECK VALUE              */
  switch (fsConnectptr.p->fsState) {
  case WAIT_READ_PAGE_ZERO:
    jam();
    fragrecptr.i = fsConnectptr.p->fragrecPtr;
    ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
    srReadPageZeroLab(signal);
    return;
    break;
  case WAIT_READ_DATA:
    jam();
    fragrecptr.i = fsConnectptr.p->fragrecPtr;
    ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
    storeDataPageInDirectoryLab(signal);
    return;
    break;
  case READ_UNDO_PAGE:
    jam();
    srDoUndoLab(signal);
    return;
    break;
  case READ_UNDO_PAGE_AND_CLOSE:
    jam();
    fsConnectptr.p->fsState = WAIT_CLOSE_UNDO;
    /* ************************ */
    /* FSCLOSEREQ               */
    /* ************************ */
    signal->theData[0] = fsConnectptr.p->fsPtr;
    signal->theData[1] = cownBlockref;
    signal->theData[2] = fsConnectptr.i;
    signal->theData[3] = 0;
    sendSignal(NDBFS_REF, GSN_FSCLOSEREQ, signal, 4, JBA);
    /* FLAG = DO NOT DELETE FILE */
    srDoUndoLab(signal);
    return;
    break;
  default:
    ndbrequire(false);
    break;
  }//switch
  return;
}//Dbacc::execFSREADCONF()


/* ******************--------------------------------------------------------------- */
/* FSWRITECONF                                         OPENFILE CONF                 */
/* ******************------------------------------+                                 */
/*   SENDER: FS,     LEVEL B       */
void Dbacc::execFSWRITECONF(Signal* signal) 
{
  jamEntry();
  fsOpptr.i = signal->theData[0];
  ptrCheckGuard(fsOpptr, cfsOpsize, fsOprec);