ndbtransaction.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 */

#include <ndb_global.h>
#include <NdbOut.hpp>
#include <NdbTransaction.hpp>
#include <NdbOperation.hpp>
#include <NdbScanOperation.hpp>
#include <NdbIndexScanOperation.hpp>
#include <NdbIndexOperation.hpp>
#include "NdbApiSignal.hpp"
#include "TransporterFacade.hpp"
#include "API.hpp"
#include "NdbBlob.hpp"

#include <signaldata/TcKeyConf.hpp>
#include <signaldata/TcIndx.hpp>
#include <signaldata/TcCommit.hpp>
#include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/TcHbRep.hpp>

/*****************************************************************************
NdbTransaction( Ndb* aNdb );

Return Value:  None
Parameters:    aNdb: Pointers to the Ndb object 
Remark:        Creates a connection object. 
*****************************************************************************/
NdbTransaction::NdbTransaction( Ndb* aNdb ) :
  theSendStatus(NotInit),
  theCallbackFunction(NULL),
  theCallbackObject(NULL),
  theTransArrayIndex(0),
  theStartTransTime(0),
  theErrorLine(0),
  theErrorOperation(NULL),
  theNdb(aNdb),
  theNext(NULL),
  theFirstOpInList(NULL),
  theLastOpInList(NULL),
  theFirstExecOpInList(NULL),
  theLastExecOpInList(NULL),
  theCompletedFirstOp(NULL),
  theCompletedLastOp(NULL),
  theNoOfOpSent(0),
  theNoOfOpCompleted(0),
  theNoOfOpFetched(0),
  theMyRef(0),
  theTCConPtr(0),
  theTransactionId(0),
  theGlobalCheckpointId(0),
  theStatus(NotConnected),
  theCompletionStatus(NotCompleted), 
  theCommitStatus(NotStarted),
  theMagicNumber(0xFE11DC),
  theTransactionIsStarted(false),
  theDBnode(0),
  theReleaseOnClose(false),
  // Scan operations
  m_waitForReply(true),
  m_theFirstScanOperation(NULL),
  m_theLastScanOperation(NULL),
  m_firstExecutedScanOp(NULL),
  // Scan operations
  theScanningOp(NULL),
  theBuddyConPtr(0xFFFFFFFF),
  theBlobFlag(false),
  thePendingBlobOps(0)
{
  theListState = NotInList;
  theError.code = 0;
  theId = theNdb->theImpl->theNdbObjectIdMap.map(this);

#define CHECK_SZ(mask, sz) assert((sizeof(mask)/sizeof(mask[0])) == sz)

  CHECK_SZ(m_db_nodes, NdbNodeBitmask::Size);
  CHECK_SZ(m_failed_db_nodes, NdbNodeBitmask::Size);
}//NdbTransaction::NdbTransaction()

/*****************************************************************************
~NdbTransaction();

Remark:        Deletes the connection object. 
*****************************************************************************/
NdbTransaction::~NdbTransaction()
{
  DBUG_ENTER("NdbTransaction::~NdbTransaction");
  theNdb->theImpl->theNdbObjectIdMap.unmap(theId, this);
  DBUG_VOID_RETURN;
}//NdbTransaction::~NdbTransaction()

/*****************************************************************************
void init();

Remark:         Initialise connection object for new transaction. 
*****************************************************************************/
void	
NdbTransaction::init()
{
  theListState            = NotInList;
  theInUseState           = true;
  theTransactionIsStarted = false;
  theNext		  = NULL;

  theFirstOpInList	  = NULL;
  theLastOpInList	  = NULL;

  theScanningOp            = NULL;

  theFirstExecOpInList	  = NULL;
  theLastExecOpInList	  = NULL;

  theCompletedFirstOp	  = NULL;
  theCompletedLastOp	  = NULL;

  theGlobalCheckpointId   = 0;
  theCommitStatus         = Started;
  theCompletionStatus     = NotCompleted;
  m_abortOption           = AbortOnError;

  theError.code		  = 0;
  theErrorLine		  = 0;
  theErrorOperation	  = NULL;

  theReleaseOnClose       = false;
  theSimpleState          = true;
  theSendStatus           = InitState;
  theMagicNumber          = 0x37412619;
  // Scan operations
  m_waitForReply            = true;
  m_theFirstScanOperation = NULL;
  m_theLastScanOperation  = NULL;
  m_firstExecutedScanOp   = 0;
  theBuddyConPtr            = 0xFFFFFFFF;
  //
  theBlobFlag = false;
  thePendingBlobOps = 0;
}//NdbTransaction::init()

/*****************************************************************************
setOperationErrorCode(int error);

Remark:        Sets an error code on the connection object from an 
               operation object. 
*****************************************************************************/
void
NdbTransaction::setOperationErrorCode(int error)
{
  DBUG_ENTER("NdbTransaction::setOperationErrorCode");
  setErrorCode(error);
  DBUG_VOID_RETURN;
}

/*****************************************************************************
setOperationErrorCodeAbort(int error);

Remark:        Sets an error code on the connection object from an 
               operation object. 
*****************************************************************************/
void
NdbTransaction::setOperationErrorCodeAbort(int error, int abortOption)
{
  DBUG_ENTER("NdbTransaction::setOperationErrorCodeAbort");
  if (abortOption == -1)
    abortOption = m_abortOption;
  if (theTransactionIsStarted == false) {
    theCommitStatus = Aborted;
  } else if ((abortOption == AbortOnError) && 
	     (theCommitStatus != Committed) &&
             (theCommitStatus != Aborted)) {
    theCommitStatus = NeedAbort;
  }//if
  setErrorCode(error);
  DBUG_VOID_RETURN;
}

/*****************************************************************************
setErrorCode(int anErrorCode);

Remark:        Sets an error indication on the connection object. 
*****************************************************************************/
void
NdbTransaction::setErrorCode(int error)
{
  DBUG_ENTER("NdbTransaction::setErrorCode");
  DBUG_PRINT("enter", ("error: %d, theError.code: %d", error, theError.code));

  if (theError.code == 0)
    theError.code = error;

  DBUG_VOID_RETURN;
}//NdbTransaction::setErrorCode()

int
NdbTransaction::restart(){
  DBUG_ENTER("NdbTransaction::restart");
  if(theCompletionStatus == CompletedSuccess){
    releaseCompletedOperations();
    Uint64 tTransid = theNdb->theFirstTransId;
    theTransactionId = tTransid;
    if ((tTransid & 0xFFFFFFFF) == 0xFFFFFFFF) {
      theNdb->theFirstTransId = (tTransid >> 32) << 32;
    } else {
      theNdb->theFirstTransId = tTransid + 1;
    }
    theCommitStatus = Started;
    theCompletionStatus = NotCompleted;
    theTransactionIsStarted = false;
    DBUG_RETURN(0);
  }
  DBUG_PRINT("error",("theCompletionStatus != CompletedSuccess"));
  DBUG_RETURN(-1);
}

/*****************************************************************************
void handleExecuteCompletion(void);

Remark:        Handle time-out on a transaction object. 
*****************************************************************************/
void
NdbTransaction::handleExecuteCompletion()
{
  /***************************************************************************
   *	  Move the NdbOperation objects from the list of executing 
   *      operations to list of completed
   **************************************************************************/
  NdbOperation* tFirstExecOp = theFirstExecOpInList;
  NdbOperation* tLastExecOp = theLastExecOpInList;
  if (tLastExecOp != NULL) {
    tLastExecOp->next(theCompletedFirstOp);
    theCompletedFirstOp = tFirstExecOp;
    if (theCompletedLastOp == NULL)
      theCompletedLastOp = tLastExecOp;
    theFirstExecOpInList = NULL;
    theLastExecOpInList = NULL;
  }//if
  theSendStatus = InitState;
  return;
}//NdbTransaction::handleExecuteCompletion()

/*****************************************************************************
int execute(ExecType aTypeOfExec, CommitType aTypeOfCommit, int forceSend);

Return Value:  Return 0 : execute was successful.
               Return -1: In all other case.  
Parameters :   aTypeOfExec: Type of execute.
Remark:        Initialise connection object for new transaction. 
*****************************************************************************/
int 
NdbTransaction::execute(ExecType aTypeOfExec, 
		       AbortOption abortOption,
		       int forceSend)
{
  NdbError savedError= theError;
  DBUG_ENTER("NdbTransaction::execute");
  DBUG_PRINT("enter", ("aTypeOfExec: %d, abortOption: %d", 
		       aTypeOfExec, abortOption));

  if (! theBlobFlag)
    DBUG_RETURN(executeNoBlobs(aTypeOfExec, abortOption, forceSend));

  /*
   * execute prepared ops in batches, as requested by blobs
   * - blob error does not terminate execution
   * - blob error sets error on operation
   * - if error on operation skip blob calls
   */

  ExecType tExecType;
  NdbOperation* tPrepOp;
  NdbOperation* tCompletedFirstOp = NULL;
  NdbOperation* tCompletedLastOp = NULL;

  int ret = 0;
  do {
    tExecType = aTypeOfExec;
    tPrepOp = theFirstOpInList;
    while (tPrepOp != NULL) {
      if (tPrepOp->theError.code == 0) {
        bool batch = false;
        NdbBlob* tBlob = tPrepOp->theBlobList;
        while (tBlob != NULL) {
          if (tBlob->preExecute(tExecType, batch) == -1)
	  {
            ret = -1;
	    if(savedError.code==0)
	      savedError= theError;
	  }
          tBlob = tBlob->theNext;
        }
        if (batch) {
          // blob asked to execute all up to here now
          tExecType = NoCommit;
          break;
        }
      }
      tPrepOp = tPrepOp->next();
    }

    // save rest of prepared ops if batch
    NdbOperation* tRestOp= 0;
    NdbOperation* tLastOp= 0;
    if (tPrepOp != NULL) {
      tRestOp = tPrepOp->next();
      tPrepOp->next(NULL);
      tLastOp = theLastOpInList;
      theLastOpInList = tPrepOp;
    }

    if (tExecType == Commit) {
      NdbOperation* tOp = theCompletedFirstOp;
      while (tOp != NULL) {
        if (tOp->theError.code == 0) {
          NdbBlob* tBlob = tOp->theBlobList;
          while (tBlob != NULL) {
            if (tBlob->preCommit() == -1)
	    {
	      ret = -1;
	      if(savedError.code==0)
		savedError= theError;
	    }
            tBlob = tBlob->theNext;
          }
        }
        tOp = tOp->next();
      }
    }

    // completed ops are in unspecified order
    if (theCompletedFirstOp != NULL) {
      if (tCompletedFirstOp == NULL) {
        tCompletedFirstOp = theCompletedFirstOp;
        tCompletedLastOp = theCompletedLastOp;
      } else {
        tCompletedLastOp->next(theCompletedFirstOp);
        tCompletedLastOp = theCompletedLastOp;
      }
      theCompletedFirstOp = NULL;
      theCompletedLastOp = NULL;
    }

    if (executeNoBlobs(tExecType, abortOption, forceSend) == -1)
    {
      ret = -1;
      if(savedError.code==0)
	savedError= theError;
    }

#ifdef ndb_api_crash_on_complex_blob_abort
    assert(theFirstOpInList == NULL && theLastOpInList == NULL);
#else
    theFirstOpInList = theLastOpInList = NULL;
#endif

    {
      NdbOperation* tOp = theCompletedFirstOp;
      while (tOp != NULL) {
        if (tOp->theError.code == 0) {
          NdbBlob* tBlob = tOp->theBlobList;
          while (tBlob != NULL) {
            // may add new operations if batch
            if (tBlob->postExecute(tExecType) == -1)
	    {
              ret = -1;
	      if(savedError.code==0)
		savedError= theError;
	    }
            tBlob = tBlob->theNext;
          }
        }
        tOp = tOp->next();
      }
    }

    // add saved prepared ops if batch
    if (tPrepOp != NULL && tRestOp != NULL) {
      if (theFirstOpInList == NULL)