ndbif.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 "NdbApiSignal.hpp"
#include "NdbImpl.hpp"
#include <NdbTransaction.hpp>
#include <NdbOperation.hpp>
#include <NdbIndexOperation.hpp>
#include <NdbScanOperation.hpp>
#include <NdbRecAttr.hpp>
#include <NdbReceiver.hpp>
#include "API.hpp"

#include <signaldata/TcCommit.hpp>
#include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/TcKeyConf.hpp>
#include <signaldata/TestOrd.hpp>
#include <signaldata/CreateIndx.hpp>
#include <signaldata/DropIndx.hpp>
#include <signaldata/TcIndx.hpp>
#include <signaldata/TransIdAI.hpp>
#include <signaldata/ScanFrag.hpp>
#include <signaldata/ScanTab.hpp>

#include <ndb_limits.h>
#include <NdbOut.hpp>
#include <NdbTick.h>


/******************************************************************************
 * int init( int aNrOfCon, int aNrOfOp );
 *
 * Return Value:   Return 0 : init was successful.
 *                Return -1: In all other case.  
 * Parameters:	aNrOfCon : Number of connections offered to the application.
 *		aNrOfOp : Number of operations offered to the application.
 * Remark:		Create pointers and idle list Synchronous.
 ****************************************************************************/ 
int
Ndb::init(int aMaxNoOfTransactions)
{
  DBUG_ENTER("Ndb::init");

  int i;
  int aNrOfCon;
  int aNrOfOp;
  int tMaxNoOfTransactions;
  NdbApiSignal* tSignal[16];	// Initiate free list of 16 signal objects
  if (theInitState != NotInitialised) {
    switch(theInitState){
    case InitConfigError:
      theError.code = 4117;
      break;
    default:
      theError.code = 4104;
      break;
    }
    DBUG_RETURN(-1);
  }//if
  theInitState = StartingInit;
  TransporterFacade * theFacade =  TransporterFacade::instance();
  theFacade->lock_mutex();
  
  const int tBlockNo = theFacade->open(this,
                                       executeMessage, 
                                       statusMessage);  
  if ( tBlockNo == -1 ) {
    theError.code = 4105;
    theFacade->unlock_mutex();
    DBUG_RETURN(-1); // no more free blocknumbers
  }//if
  
  theNdbBlockNumber = tBlockNo;

  theFacade->unlock_mutex();
  
  theDictionary->setTransporter(this, theFacade);
  
  aNrOfCon = theImpl->theNoOfDBnodes;
  aNrOfOp = 2*theImpl->theNoOfDBnodes;
  
  // Create connection object in a linked list 
  if((createConIdleList(aNrOfCon)) == -1){
    theError.code = 4000;
    goto error_handler;
  }
  
  // Create operations in a linked list
  if((createOpIdleList(aNrOfOp)) == -1){       
    theError.code = 4000;
    goto error_handler;
  }
  

  tMaxNoOfTransactions = aMaxNoOfTransactions;
  theMaxNoOfTransactions = tMaxNoOfTransactions;
  theRemainingStartTransactions= tMaxNoOfTransactions;  
  thePreparedTransactionsArray = new NdbTransaction* [tMaxNoOfTransactions];
  theSentTransactionsArray = new NdbTransaction* [tMaxNoOfTransactions];
  theCompletedTransactionsArray = new NdbTransaction* [tMaxNoOfTransactions];
  
  if ((thePreparedTransactionsArray == NULL) ||
      (theSentTransactionsArray == NULL) ||
      (theCompletedTransactionsArray == NULL)) {
    goto error_handler;
  }//if
  
  for (i = 0; i < tMaxNoOfTransactions; i++) {
    thePreparedTransactionsArray[i] = NULL;
    theSentTransactionsArray[i] = NULL;
    theCompletedTransactionsArray[i] = NULL;
  }//for     
  for (i = 0; i < 16; i++){
    tSignal[i] = getSignal();
    if(tSignal[i] == NULL) {
      theError.code = 4000;
      goto error_handler;
    }
  }
  for (i = 0; i < 16; i++)
    releaseSignal(tSignal[i]);
  theInitState = Initialised; 
  DBUG_RETURN(0);
  
error_handler:
  ndbout << "error_handler" << endl;
  releaseTransactionArrays();
  delete theDictionary;
  TransporterFacade::instance()->close(theNdbBlockNumber, 0);
  DBUG_RETURN(-1);
}

void
Ndb::releaseTransactionArrays()
{
  DBUG_ENTER("Ndb::releaseTransactionArrays");
  if (thePreparedTransactionsArray != NULL) {
    delete [] thePreparedTransactionsArray;
  }//if
  if (theSentTransactionsArray != NULL) {
    delete [] theSentTransactionsArray;
  }//if
  if (theCompletedTransactionsArray != NULL) {
    delete [] theCompletedTransactionsArray;
  }//if
  DBUG_VOID_RETURN;
}//Ndb::releaseTransactionArrays()

void
Ndb::executeMessage(void* NdbObject,
                    NdbApiSignal * aSignal,
                    LinearSectionPtr ptr[3])
{
  Ndb* tNdb = (Ndb*)NdbObject;
  tNdb->handleReceivedSignal(aSignal, ptr);
}

void Ndb::connected(Uint32 ref)
{
  theMyRef= ref;
  Uint32 tmpTheNode= refToNode(ref);
  Uint64 tBlockNo= refToBlock(ref);
  if (theNdbBlockNumber >= 0){
    assert(theMyRef == numberToRef(theNdbBlockNumber, tmpTheNode));
  }
  
  TransporterFacade * theFacade =  TransporterFacade::instance();
  int i, n= 0;
  for (i = 1; i < MAX_NDB_NODES; i++){
    if (theFacade->getIsDbNode(i)){
      theImpl->theDBnodes[n] = i;
      n++;
    }
  }
  theImpl->theNoOfDBnodes= n;
  theFirstTransId = ((Uint64)tBlockNo << 52)+
    ((Uint64)tmpTheNode << 40);
  theFirstTransId += theFacade->m_max_trans_id;
  //      assert(0);
  DBUG_PRINT("info",("connected with ref=%x, id=%d, no_db_nodes=%d, first_trans_id=%lx",
		     theMyRef,
		     tmpTheNode,
		     theImpl->theNoOfDBnodes,
		     theFirstTransId));
  theCommitAckSignal = new NdbApiSignal(theMyRef);

  theDictionary->m_receiver.m_reference= theMyRef;
  theNode= tmpTheNode; // flag that Ndb object is initialized
}

void
Ndb::statusMessage(void* NdbObject, Uint32 a_node, bool alive, bool nfComplete)
{
  DBUG_ENTER("Ndb::statusMessage");
  Ndb* tNdb = (Ndb*)NdbObject;
  if (alive) {
    if (nfComplete) {
      tNdb->connected(a_node);
      DBUG_VOID_RETURN;
    }//if
  } else {
    if (nfComplete) {
      tNdb->report_node_failure_completed(a_node);
    } else {
      tNdb->report_node_failure(a_node);
    }//if
  }//if
  NdbDictInterface::execNodeStatus(&tNdb->theDictionary->m_receiver,
				   a_node, alive, nfComplete);
  DBUG_VOID_RETURN;
}

void
Ndb::report_node_failure(Uint32 node_id)
{
  /**
   * We can only set the state here since this object can execute 
   * simultaneously. 
   * 
   * This method is only called by ClusterMgr (via lots of methods)
   */
  theImpl->the_release_ind[node_id] = 1;
  // must come after
  theImpl->the_release_ind[0] = 1;
  theImpl->theWaiter.nodeFail(node_id);
  return;
}//Ndb::report_node_failure()


void
Ndb::report_node_failure_completed(Uint32 node_id)
{
  abortTransactionsAfterNodeFailure(node_id);

}//Ndb::report_node_failure_completed()

/***************************************************************************
void abortTransactionsAfterNodeFailure();

Remark:   Abort all transactions in theSentTransactionsArray after connection 
          to one node has failed
****************************************************************************/
void	
Ndb::abortTransactionsAfterNodeFailure(Uint16 aNodeId)
{  
  Uint32 tNoSentTransactions = theNoOfSentTransactions;
  for (int i = tNoSentTransactions - 1; i >= 0; i--) {
    NdbTransaction* localCon = theSentTransactionsArray[i];
    if (localCon->getConnectedNodeId() == aNodeId) {
      const NdbTransaction::SendStatusType sendStatus = localCon->theSendStatus;
      if (sendStatus == NdbTransaction::sendTC_OP || 
	  sendStatus == NdbTransaction::sendTC_COMMIT) {
        /*
        A transaction was interrupted in the prepare phase by a node
        failure. Since the transaction was not found in the phase
        after the node failure it cannot have been committed and
        we report a normal node failure abort.
        */
	localCon->setOperationErrorCodeAbort(4010);
        localCon->theCompletionStatus = NdbTransaction::CompletedFailure;
      } else if (sendStatus == NdbTransaction::sendTC_ROLLBACK) {
        /*
        We aimed for abort and abort we got even if it was by a node
        failure. We will thus report it as a success.
        */
        localCon->theCompletionStatus = NdbTransaction::CompletedSuccess;
      } else {
#ifdef VM_TRACE
        printState("abortTransactionsAfterNodeFailure %x", this);
        abort();
#endif
      }
      /*
      All transactions arriving here have no connection to the kernel
      intact since the node was failing and they were aborted. Thus we
      set commit state to Aborted and set state to release on close.
      */
      localCon->theReturnStatus = NdbTransaction::ReturnFailure;
      localCon->theCommitStatus = NdbTransaction::Aborted;
      localCon->theReleaseOnClose = true;
      completedTransaction(localCon);
    }
    else if(localCon->report_node_failure(aNodeId))
    {
      completedTransaction(localCon);
    }
  }//for
  return;
}//Ndb::abortTransactionsAfterNodeFailure()

/****************************************************************************
void handleReceivedSignal(NdbApiSignal* aSignal);

Parameters:     aSignal: The signal object.
Remark:         Send all operations belonging to this connection. 
*****************************************************************************/
void	
Ndb::handleReceivedSignal(NdbApiSignal* aSignal, LinearSectionPtr ptr[3])
{
  NdbOperation* tOp;
  NdbIndexOperation* tIndexOp;
  NdbTransaction* tCon;
  int tReturnCode = -1;
  const Uint32* tDataPtr = aSignal->getDataPtr();
  const Uint32 tWaitState = theImpl->theWaiter.m_state;
  const Uint32 tSignalNumber = aSignal->readSignalNumber();
  const Uint32 tFirstData = *tDataPtr;
  const Uint32 tLen = aSignal->getLength();
  void * tFirstDataPtr;

  /*
    In order to support 64 bit processes in the application we need to use
    id's rather than a direct pointer to the object used. It is also a good
    idea that one cannot corrupt the application code by sending a corrupt
    memory pointer.
    
    All signals received by the API requires the first data word to be such
    an id to the receiving object.
  */
  
  switch (tSignalNumber){
  case GSN_TCKEYCONF:
    {
      tFirstDataPtr = int2void(tFirstData);
      if (tFirstDataPtr == 0) goto InvalidSignal;

      const TcKeyConf * const keyConf = (TcKeyConf *)tDataPtr;
      const BlockReference aTCRef = aSignal->theSendersBlockRef;

      tCon = void2con(tFirstDataPtr);
      if ((tCon->checkMagicNumber() == 0) &&
          (tCon->theSendStatus == NdbTransaction::sendTC_OP)) {
        tReturnCode = tCon->receiveTCKEYCONF(keyConf, tLen);
        if (tReturnCode != -1) {
          completedTransaction(tCon);
        }//if

	if(TcKeyConf::getMarkerFlag(keyConf->confInfo)){
	  NdbTransaction::sendTC_COMMIT_ACK(theCommitAckSignal,
					   keyConf->transId1, 
					   keyConf->transId2,
					   aTCRef);
	}
      
	return;
      }//if
      goto InvalidSignal;
      
      return;
    }
  case GSN_TRANSID_AI:{
    tFirstDataPtr = int2void(tFirstData);
    NdbReceiver* tRec;
    if (tFirstDataPtr && (tRec = void2rec(tFirstDataPtr)) && 
	tRec->checkMagicNumber() && (tCon = tRec->getTransaction()) &&
	tCon->checkState_TransId(((const TransIdAI*)tDataPtr)->transId)){
      Uint32 com;
      if(aSignal->m_noOfSections > 0){
	com = tRec->execTRANSID_AI(ptr[0].p, ptr[0].sz);
      } else {
	com = tRec->execTRANSID_AI(tDataPtr + TransIdAI::HeaderLength, 
				   tLen - TransIdAI::HeaderLength);
      }

      if(com == 0)
	return;
      
      switch(tRec->getType()){
      case NdbReceiver::NDB_OPERATION:
      case NdbReceiver::NDB_INDEX_OPERATION:
	if(tCon->OpCompleteSuccess() != -1){
	  completedTransaction(tCon);
	}
	return;
      case NdbReceiver::NDB_SCANRECEIVER:
	tCon->theScanningOp->receiver_delivered(tRec);
	theImpl->theWaiter.m_state = (((WaitSignalType) tWaitState) == WAIT_SCAN ? 
				      (Uint32) NO_WAIT : tWaitState);
	break;
      default:
	goto InvalidSignal;
      }
      break;
    } else {
      /**
       * This is ok as transaction can have been aborted before TRANSID_AI
       * arrives (if TUP on  other node than TC)
       */
      return;
    }
  }
  case GSN_TCKEY_FAILCONF:
    {
      tFirstDataPtr = int2void(tFirstData);
      const TcKeyFailConf * failConf = (TcKeyFailConf *)tDataPtr;
      const BlockReference aTCRef = aSignal->theSendersBlockRef;
      if (tFirstDataPtr != 0){
	tOp = void2rec_op(tFirstDataPtr);
	
	if (tOp->checkMagicNumber(false) == 0) {
	  tCon = tOp->theNdbCon;
	  if (tCon != NULL) {
	    if ((tCon->theSendStatus == NdbTransaction::sendTC_OP) ||
		(tCon->theSendStatus == NdbTransaction::sendTC_COMMIT)) {
	      tReturnCode = tCon->receiveTCKEY_FAILCONF(failConf);
	      if (tReturnCode != -1) {
		completedTransaction(tCon);
	      }//if
	    }//if
	  }
	}
      } else {
#ifdef VM_TRACE
	ndbout_c("Recevied TCKEY_FAILCONF wo/ operation");
#endif
      }
      if(tFirstData & 1){
	NdbTransaction::sendTC_COMMIT_ACK(theCommitAckSignal,
					 failConf->transId1, 
					 failConf->transId2,
					 aTCRef);
      }
      return;
    }
  case GSN_TCKEY_FAILREF:
    {
      tFirstDataPtr = int2void(tFirstData);
      if(tFirstDataPtr != 0){
	tOp = void2rec_op(tFirstDataPtr);
	if (tOp->checkMagicNumber(false) == 0) {
	  tCon = tOp->theNdbCon;
	  if (tCon != NULL) {
	    if ((tCon->theSendStatus == NdbTransaction::sendTC_OP) ||
		(tCon->theSendStatus == NdbTransaction::sendTC_ROLLBACK)) {
	      tReturnCode = tCon->receiveTCKEY_FAILREF(aSignal);
	      if (tReturnCode != -1) {
		completedTransaction(tCon);
		return;