dbtupcommit.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/TupCommit.hpp>

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

void Dbtup::execTUP_WRITELOG_REQ(Signal* signal)
{
  jamEntry();
  OperationrecPtr loopOpPtr;
  loopOpPtr.i = signal->theData[0];
  Uint32 gci = signal->theData[1];
  ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
  while (loopOpPtr.p->nextActiveOp != RNIL) {
    ljam();
    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
  }//while
  do {
    Uint32 blockNo = refToBlock(loopOpPtr.p->userblockref);
    ndbrequire(loopOpPtr.p->transstate == STARTED);
    signal->theData[0] = loopOpPtr.p->userpointer;
    signal->theData[1] = gci;
    if (loopOpPtr.p->prevActiveOp == RNIL) {
      ljam();
      EXECUTE_DIRECT(blockNo, GSN_LQH_WRITELOG_REQ, signal, 2);
      return;
    }//if
    ljam();
    EXECUTE_DIRECT(blockNo, GSN_LQH_WRITELOG_REQ, signal, 2);
    jamEntry();
    loopOpPtr.i = loopOpPtr.p->prevActiveOp;
    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
  } while (true);
}//Dbtup::execTUP_WRITELOG_REQ()

void Dbtup::execTUP_DEALLOCREQ(Signal* signal)
{
  TablerecPtr regTabPtr;
  FragrecordPtr regFragPtr;

  jamEntry();

  Uint32 fragId = signal->theData[0];
  regTabPtr.i = signal->theData[1];
  Uint32 fragPageId = signal->theData[2];
  Uint32 pageIndex = signal->theData[3];

  ptrCheckGuard(regTabPtr, cnoOfTablerec, tablerec);
  getFragmentrec(regFragPtr, fragId, regTabPtr.p);
  ndbrequire(regFragPtr.p != NULL);

  PagePtr pagePtr;
  pagePtr.i = getRealpid(regFragPtr.p, fragPageId);
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 pageIndexScaled = pageIndex >> 1;
  ndbrequire((pageIndex & 1) == 0);
  Uint32 pageOffset = ZPAGE_HEADER_SIZE + 
                     (regTabPtr.p->tupheadsize * pageIndexScaled);
//---------------------------------------------------
/* --- Deallocate a tuple as requested by ACC  --- */
//---------------------------------------------------
  if (isUndoLoggingNeeded(regFragPtr.p, fragPageId)) {
    ljam();
    cprAddUndoLogRecord(signal,
                        ZLCPR_TYPE_INSERT_TH,
                        fragPageId,
                        pageIndex,
                        regTabPtr.i,
                        fragId,
                        regFragPtr.p->checkpointVersion);
    cprAddData(signal,
               regFragPtr.p,
               pagePtr.i,
               regTabPtr.p->tupheadsize,
               pageOffset);
  }//if
  {
    freeTh(regFragPtr.p,
           regTabPtr.p,
           signal,
           pagePtr.p,
           pageOffset);
  }
}

/* ---------------------------------------------------------------- */
/* ------------ PERFORM A COMMIT ON AN UPDATE OPERATION  ---------- */
/* ---------------------------------------------------------------- */
void Dbtup::commitUpdate(Signal* signal,
                         Operationrec*  const regOperPtr,
                         Fragrecord* const regFragPtr,
                         Tablerec* const regTabPtr)
{
  if (regOperPtr->realPageIdC != RNIL) {
    if (isUndoLoggingNeeded(regFragPtr, regOperPtr->fragPageIdC)) {
/* ------------------------------------------------------------------------ */
/* IF THE COPY WAS CREATED WITHIN THIS CHECKPOINT WE ONLY HAVE              */
/* TO LOG THE CREATION OF THE COPY. IF HOWEVER IT WAS CREATED BEFORE  SAVE  */
/* THIS CHECKPOINT, WE HAVE TO THE DATA AS WELL.                            */
/* ------------------------------------------------------------------------ */
      if (regOperPtr->undoLogged) {
        ljam();
        cprAddUndoLogRecord(signal,
                            ZLCPR_TYPE_INSERT_TH_NO_DATA,
                            regOperPtr->fragPageIdC,
                            regOperPtr->pageIndexC,
                            regOperPtr->tableRef,
                            regOperPtr->fragId,
                            regFragPtr->checkpointVersion);
      } else {
        ljam();
        cprAddUndoLogRecord(signal,
                            ZLCPR_TYPE_INSERT_TH,
                            regOperPtr->fragPageIdC,
                            regOperPtr->pageIndexC,
                            regOperPtr->tableRef,
                            regOperPtr->fragId,
                            regFragPtr->checkpointVersion);
        cprAddData(signal,
                   regFragPtr,
                   regOperPtr->realPageIdC,
                   regTabPtr->tupheadsize,
                   regOperPtr->pageOffsetC);
      }//if
    }//if

    PagePtr copyPagePtr;
    copyPagePtr.i = regOperPtr->realPageIdC;
    ptrCheckGuard(copyPagePtr, cnoOfPage, page);
    freeTh(regFragPtr,
           regTabPtr,
           signal,
           copyPagePtr.p,
           (Uint32)regOperPtr->pageOffsetC);
    regOperPtr->realPageIdC = RNIL;
    regOperPtr->fragPageIdC = RNIL;
    regOperPtr->pageOffsetC = ZNIL;
    regOperPtr->pageIndexC = ZNIL;
  }//if
}//Dbtup::commitUpdate()

void
Dbtup::commitSimple(Signal* signal,
                    Operationrec* const regOperPtr,
                    Fragrecord* const regFragPtr,
                    Tablerec* const regTabPtr)
{
  operPtr.p = regOperPtr;
  fragptr.p = regFragPtr;
  tabptr.p = regTabPtr;

  // Checking detached triggers
  checkDetachedTriggers(signal,
                        regOperPtr,
                        regTabPtr);

  removeActiveOpList(regOperPtr);
  if (regOperPtr->optype == ZUPDATE) {
    ljam();
    commitUpdate(signal, regOperPtr, regFragPtr, regTabPtr);
    if (regTabPtr->GCPIndicator) {
      updateGcpId(signal, regOperPtr, regFragPtr, regTabPtr);
    }//if
  } else if (regOperPtr->optype == ZINSERT) {
    ljam();
    if (regTabPtr->GCPIndicator) {
      updateGcpId(signal, regOperPtr, regFragPtr, regTabPtr);
    }//if
  } else {
    ndbrequire(regOperPtr->optype == ZDELETE);
  }//if
}//Dbtup::commitSimple()

void Dbtup::removeActiveOpList(Operationrec*  const regOperPtr)
{
  if (regOperPtr->inActiveOpList == ZTRUE) {
    OperationrecPtr raoOperPtr;
    regOperPtr->inActiveOpList = ZFALSE;
    if (regOperPtr->prevActiveOp != RNIL) {
      ljam();
      raoOperPtr.i = regOperPtr->prevActiveOp;
      ptrCheckGuard(raoOperPtr, cnoOfOprec, operationrec);
      raoOperPtr.p->nextActiveOp = regOperPtr->nextActiveOp;
    } else {
      ljam();
      PagePtr pagePtr;
      pagePtr.i = regOperPtr->realPageId;
      ptrCheckGuard(pagePtr, cnoOfPage, page);
      ndbrequire(regOperPtr->pageOffset < ZWORDS_ON_PAGE);
      pagePtr.p->pageWord[regOperPtr->pageOffset] = regOperPtr->nextActiveOp;
    }//if
    if (regOperPtr->nextActiveOp != RNIL) {
      ljam();
      raoOperPtr.i = regOperPtr->nextActiveOp;
      ptrCheckGuard(raoOperPtr, cnoOfOprec, operationrec);
      raoOperPtr.p->prevActiveOp = regOperPtr->prevActiveOp;
    }//if
    regOperPtr->prevActiveOp = RNIL;
    regOperPtr->nextActiveOp = RNIL;
  }//if
}//Dbtup::removeActiveOpList()

/* ---------------------------------------------------------------- */
/* INITIALIZATION OF ONE CONNECTION RECORD TO PREPARE FOR NEXT OP.  */
/* ---------------------------------------------------------------- */
void Dbtup::initOpConnection(Operationrec* regOperPtr,
			     Fragrecord * fragPtrP)
{
  Uint32 RinFragList = regOperPtr->inFragList;
  regOperPtr->transstate = IDLE;
  regOperPtr->currentAttrinbufLen = 0;
  regOperPtr->optype = ZREAD;
  if (RinFragList == ZTRUE) {
    OperationrecPtr tropNextLinkPtr;
    OperationrecPtr tropPrevLinkPtr;
/*----------------------------------------------------------------- */
/*       TO ENSURE THAT WE HAVE SUCCESSFUL ABORTS OF FOLLOWING      */
/*       OPERATIONS WHICH NEVER STARTED WE SET THE OPTYPE TO READ.  */
/*----------------------------------------------------------------- */
/*       REMOVE IT FROM THE DOUBLY LINKED LIST ON THE FRAGMENT      */
/*----------------------------------------------------------------- */
    tropPrevLinkPtr.i = regOperPtr->prevOprecInList;
    tropNextLinkPtr.i = regOperPtr->nextOprecInList;
    regOperPtr->inFragList = ZFALSE;
    if (tropPrevLinkPtr.i == RNIL) {
      ljam();
      fragPtrP->firstusedOprec = tropNextLinkPtr.i;
    } else {
      ljam();
      ptrCheckGuard(tropPrevLinkPtr, cnoOfOprec, operationrec);
      tropPrevLinkPtr.p->nextOprecInList = tropNextLinkPtr.i;
    }//if
    if (tropNextLinkPtr.i == RNIL) {
      fragPtrP->lastusedOprec = tropPrevLinkPtr.i;
    } else {
      ptrCheckGuard(tropNextLinkPtr, cnoOfOprec, operationrec);
      tropNextLinkPtr.p->prevOprecInList = tropPrevLinkPtr.i;
    }
    regOperPtr->prevOprecInList = RNIL;
    regOperPtr->nextOprecInList = RNIL;
  }//if
}//Dbtup::initOpConnection()

/* ----------------------------------------------------------------- */
/* --------------- COMMIT THIS PART OF A TRANSACTION --------------- */
/* ----------------------------------------------------------------- */
void Dbtup::execTUP_COMMITREQ(Signal* signal) 
{
  FragrecordPtr regFragPtr;
  OperationrecPtr regOperPtr;
  TablerecPtr regTabPtr;

  TupCommitReq * const tupCommitReq = (TupCommitReq *)signal->getDataPtr();

  ljamEntry();
  regOperPtr.i = tupCommitReq->opPtr;
  ptrCheckGuard(regOperPtr, cnoOfOprec, operationrec);

  ndbrequire(regOperPtr.p->transstate == STARTED);
  regOperPtr.p->gci = tupCommitReq->gci;
  regOperPtr.p->hashValue = tupCommitReq->hashValue;

  regFragPtr.i = regOperPtr.p->fragmentPtr;
  ptrCheckGuard(regFragPtr, cnoOfFragrec, fragrecord);

  regTabPtr.i = regOperPtr.p->tableRef;
  ptrCheckGuard(regTabPtr, cnoOfTablerec, tablerec);

  if (!regTabPtr.p->tuxCustomTriggers.isEmpty()) {
    ljam();
    executeTuxCommitTriggers(signal,
                             regOperPtr.p,
                             regTabPtr.p);
  }