ndb.cpp

mysql-5.0.22.tar.gz源码包

    theFirstTransId = ((tFirstTransId >> 32) << 32);      
  } else {
    theFirstTransId = tFirstTransId + 1;
  }//if
#ifdef VM_TRACE
  if (tConnection->theListState != NdbTransaction::NotInList) {
    printState("startTransactionLocal %x", tConnection);
    abort();
  }
#endif
  DBUG_RETURN(tConnection);
}//Ndb::startTransactionLocal()

/*****************************************************************************
void closeTransaction(NdbTransaction* aConnection);

Parameters:     aConnection: the connection used in the transaction.
Remark:         Close transaction by releasing the connection and all operations.
*****************************************************************************/
void
Ndb::closeTransaction(NdbTransaction* aConnection)
{
  DBUG_ENTER("Ndb::closeTransaction");
  NdbTransaction* tCon;
  NdbTransaction* tPreviousCon;

  if (aConnection == NULL) {
//-----------------------------------------------------
// closeTransaction called on NULL pointer, destructive
// application behaviour.
//-----------------------------------------------------
#ifdef VM_TRACE
    printf("NULL into closeTransaction\n");
#endif
    DBUG_VOID_RETURN;
  }//if
  CHECK_STATUS_MACRO_VOID;
  
  tCon = theTransactionList;
  theRemainingStartTransactions++;
  
  DBUG_PRINT("info",("close trans: 0x%x transid: 0x%llx",
		     aConnection, aConnection->getTransactionId()));
  DBUG_PRINT("info",("magic number: 0x%x TCConPtr: 0x%x theMyRef: 0x%x 0x%x",
		     aConnection->theMagicNumber, aConnection->theTCConPtr,
		     aConnection->theMyRef, getReference()));

  if (aConnection == tCon) {		// Remove the active connection object
    theTransactionList = tCon->next();	// from the transaction list.
  } else { 
    while (aConnection != tCon) {
      if (tCon == NULL) {
//-----------------------------------------------------
// closeTransaction called on non-existing transaction
//-----------------------------------------------------

	if(aConnection->theError.code == 4008){
	  /**
	   * When a SCAN timed-out, returning the NdbTransaction leads
	   * to reuse. And TC crashes when the API tries to reuse it to
	   * something else...
	   */
#ifdef VM_TRACE
	  printf("Scan timeout:ed NdbTransaction-> "
		 "not returning it-> memory leak\n");
#endif
	  DBUG_VOID_RETURN;
	}

#ifdef VM_TRACE
	printf("Non-existing transaction into closeTransaction\n");
	abort();
#endif
	DBUG_VOID_RETURN;
      }//if
      tPreviousCon = tCon;
      tCon = tCon->next();
    }//while
    tPreviousCon->next(tCon->next());
  }//if
  
  aConnection->release();
  
  if(aConnection->theError.code == 4008){
    /**
     * Something timed-out, returning the NdbTransaction leads
     * to reuse. And TC crashes when the API tries to reuse it to
     * something else...
     */
#ifdef VM_TRACE
    printf("Con timeout:ed NdbTransaction-> not returning it-> memory leak\n");
#endif
    DBUG_VOID_RETURN;
  }
  
  if (aConnection->theReleaseOnClose == false) {
    /**
     * Put it back in idle list for that node
     */
    Uint32 nodeId = aConnection->getConnectedNodeId();
    aConnection->theNext = theConnectionArray[nodeId];
    theConnectionArray[nodeId] = aConnection;
    DBUG_VOID_RETURN;
  } else {
    aConnection->theReleaseOnClose = false;
    releaseNdbCon(aConnection);
  }//if
  DBUG_VOID_RETURN;
}//Ndb::closeTransaction()

/*****************************************************************************
int* NdbTamper(int aAction, int aNode);

Parameters: aAction     Specifies what action to be taken
            1: Lock global checkpointing    Can only be sent to master DIH, Parameter aNode ignored.
            2: UnLock global checkpointing    Can only be sent to master DIH, Parameter aNode ignored.
	    3: Crash node

           aNode        Specifies which node the action will be taken
     	  -1: Master DIH 
       	0-16: Nodnumber

Return Value: -1 Error  .
                
Remark:         Sends a signal to DIH.
*****************************************************************************/ 
int 
Ndb::NdbTamper(TamperType aAction, int aNode)
{
  NdbTransaction*	tNdbConn;
  NdbApiSignal		tSignal(theMyRef);
  int			tNode;
  int                   tAction;
  int			ret_code;

#ifdef CUSTOMER_RELEASE
  return -1;
#else
  CHECK_STATUS_MACRO;
  checkFailedNode();

  theRestartGCI = 0;
  switch (aAction) {
// Translate enum to integer. This is done because the SCI layer
// expects integers. 
     case LockGlbChp:
        tAction = 1;
        break;
     case UnlockGlbChp:
        tAction = 2;
	break;
     case CrashNode:
        tAction = 3;
        break;
     case ReadRestartGCI:
	tAction = 4;
	break;
     default:
        theError.code = 4102;
        return -1;
  }

  tNdbConn = getNdbCon();	// Get free connection object
  if (tNdbConn == NULL) {
    theError.code = 4000;
    return -1;
  }
  tSignal.setSignal(GSN_DIHNDBTAMPER);
  tSignal.setData (tAction, 1);
  tSignal.setData(tNdbConn->ptr2int(),2);
  tSignal.setData(theMyRef,3);		// Set return block reference
  tNdbConn->Status(NdbTransaction::Connecting); // Set status to connecting
  TransporterFacade *tp = TransporterFacade::instance();
  if (tAction == 3) {
    tp->lock_mutex();
    tp->sendSignal(&tSignal, aNode);
    tp->unlock_mutex();
    releaseNdbCon(tNdbConn);
  } else if ( (tAction == 2) || (tAction == 1) ) {
    tp->lock_mutex();
    tNode = tp->get_an_alive_node();
    if (tNode == 0) {
      theError.code = 4002;
      releaseNdbCon(tNdbConn);
      return -1;
    }//if
    ret_code = tp->sendSignal(&tSignal,aNode);
    tp->unlock_mutex();
    releaseNdbCon(tNdbConn);
    return ret_code;
  } else {
    do {
      tp->lock_mutex();
      // Start protected area
      tNode = tp->get_an_alive_node();
      tp->unlock_mutex();
      // End protected area
      if (tNode == 0) {
        theError.code = 4009;
        releaseNdbCon(tNdbConn);
        return -1;
      }//if
      ret_code = sendRecSignal(tNode, WAIT_NDB_TAMPER, &tSignal, 0);
      if (ret_code == 0) {  
        if (tNdbConn->Status() != NdbTransaction::Connected) {
          theRestartGCI = 0;
        }//if
        releaseNdbCon(tNdbConn);
        return theRestartGCI;
      } else if ((ret_code == -5) || (ret_code == -2)) {
        TRACE_DEBUG("Continue DIHNDBTAMPER when node failed/stopping");
      } else {
        return -1;
      }//if
    } while (1);
  }
  return 0;
#endif
}
#if 0
/****************************************************************************
NdbSchemaCon* startSchemaTransaction();

Return Value:   Returns a pointer to a schema connection object.
                Return NULL otherwise.
Remark:         Start schema transaction. Synchronous.
****************************************************************************/ 
NdbSchemaCon* 
Ndb::startSchemaTransaction()
{
  NdbSchemaCon* tSchemaCon;
  if (theSchemaConToNdbList != NULL) {
    theError.code = 4321;
    return NULL;
  }//if
  tSchemaCon = new NdbSchemaCon(this);
  if (tSchemaCon == NULL) {
    theError.code = 4000;
    return NULL;
  }//if 
  theSchemaConToNdbList = tSchemaCon;
  return tSchemaCon;  
}
/*****************************************************************************
void closeSchemaTransaction(NdbSchemaCon* aSchemaCon);

Parameters:     aSchemaCon: the schemacon used in the transaction.
Remark:         Close transaction by releasing the schemacon and all schemaop.
*****************************************************************************/
void
Ndb::closeSchemaTransaction(NdbSchemaCon* aSchemaCon)
{
  if (theSchemaConToNdbList != aSchemaCon) {
    abort();
    return;
  }//if
  aSchemaCon->release();
  delete aSchemaCon;
  theSchemaConToNdbList = NULL;
  return;
}//Ndb::closeSchemaTransaction()
#endif

/*****************************************************************************
void RestartGCI(int aRestartGCI);

Remark:		Set theRestartGCI on the NDB object
*****************************************************************************/
void
Ndb::RestartGCI(int aRestartGCI)
{
  theRestartGCI = aRestartGCI;
}

/****************************************************************************
int getBlockNumber(void);

Remark:		
****************************************************************************/
int
Ndb::getBlockNumber()
{
  return theNdbBlockNumber;
}

NdbDictionary::Dictionary *
Ndb::getDictionary() const {
  return theDictionary;
}

/****************************************************************************
int getNodeId();

Remark:		
****************************************************************************/
int
Ndb::getNodeId()
{
  return theNode;
}

/****************************************************************************
Uint64 getTupleIdFromNdb( Uint32 aTableId, Uint32 cacheSize );

Parameters:     aTableId : The TableId.
                cacheSize: Prefetch this many values
Remark:		Returns a new TupleId to the application.
                The TupleId comes from SYSTAB_0 where SYSKEY_0 = TableId.
                It is initialized to (TableId << 48) + 1 in NdbcntrMain.cpp.
****************************************************************************/
Uint64
Ndb::getAutoIncrementValue(const char* aTableName, Uint32 cacheSize)
{
  DBUG_ENTER("getAutoIncrementValue");
  BaseString internal_tabname(internalize_table_name(aTableName));

  Ndb_local_table_info *info=
    theDictionary->get_local_table_info(internal_tabname, false);
  if (info == 0)
    DBUG_RETURN(~(Uint64)0);
  const NdbTableImpl *table= info->m_table_impl;
  Uint64 tupleId = getTupleIdFromNdb(table->m_tableId, cacheSize);
  DBUG_PRINT("info", ("value %ul", (ulong) tupleId));
  DBUG_RETURN(tupleId);
}

Uint64
Ndb::getAutoIncrementValue(const NdbDictionary::Table * aTable, Uint32 cacheSize)
{
  DBUG_ENTER("getAutoIncrementValue");
  if (aTable == 0)
    DBUG_RETURN(~(Uint64)0);
  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
  Uint64 tupleId = getTupleIdFromNdb(table->m_tableId, cacheSize);
  DBUG_PRINT("info", ("value %ul", (ulong) tupleId));
  DBUG_RETURN(tupleId);
}

Uint64 
Ndb::getTupleIdFromNdb(const char* aTableName, Uint32 cacheSize)
{
  const NdbTableImpl* table = theDictionary->getTable(aTableName);
  if (table == 0)
    return ~(Uint64)0;
  return getTupleIdFromNdb(table->m_tableId, cacheSize);
}

Uint64
Ndb::getTupleIdFromNdb(Uint32 aTableId, Uint32 cacheSize)
{
  DBUG_ENTER("getTupleIdFromNdb");
  if ( theFirstTupleId[aTableId] != theLastTupleId[aTableId] )
  {
    theFirstTupleId[aTableId]++;
    DBUG_PRINT("info", ("next cached value %ul", 
                        (ulong) theFirstTupleId[aTableId]));
    DBUG_RETURN(theFirstTupleId[aTableId]);
  }
  else // theFirstTupleId == theLastTupleId
  {
    DBUG_PRINT("info",("reading %u values from database", 
                       (cacheSize == 0) ? 1 : cacheSize));
    DBUG_RETURN(opTupleIdOnNdb(aTableId, (cacheSize == 0) ? 1 : cacheSize, 0));
  }
}

Uint64
Ndb::readAutoIncrementValue(const char* aTableName)
{
  DBUG_ENTER("readAutoIncrementValue");
  const NdbTableImpl* table = theDictionary->getTable(aTableName);
  if (table == 0) {
    theError= theDictionary->getNdbError();
    DBUG_RETURN(~(Uint64)0);
  }
  Uint64 tupleId = readTupleIdFromNdb(table->m_tableId);
  DBUG_PRINT("info", ("value %ul", (ulong) tupleId));
  DBUG_RETURN(tupleId);
}

Uint64
Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable)
{
  DBUG_ENTER("readAutoIncrementValue");
  if (aTable == 0)
    DBUG_RETURN(~(Uint64)0);
  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
  Uint64 tupleId = readTupleIdFromNdb(table->m_tableId);
  DBUG_PRINT("info", ("value %ul", (ulong) tupleId));
  DBUG_RETURN(tupleId);
}

Uint64
Ndb::readTupleIdFromNdb(Uint32 aTableId)
{
  if ( theFirstTupleId[aTableId] == theLastTupleId[aTableId] )
    // Cache is empty, check next in database
    return opTupleIdOnNdb(aTableId, 0, 3);

  return theFirstTupleId[aTableId] + 1;
}

bool
Ndb::setAutoIncrementValue(const char* aTableName, Uint64 val, bool increase)
{
  DBUG_ENTER("setAutoIncrementValue");
  BaseString internal_tabname(internalize_table_name(aTableName));

  Ndb_local_table_info *info=
    theDictionary->get_local_table_info(internal_tabname, false);
  if (info == 0) {
    theError= theDictionary->getNdbError();
    DBUG_RETURN(false);
  }
  const NdbTableImpl* table= info->m_table_impl;
  DBUG_RETURN(setTupleIdInNdb(table->m_tableId, val, increase));
}

bool
Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable, Uint64 val, bool increase)
{
  DBUG_ENTER("setAutoIncrementValue");
  if (aTable == 0)
    DBUG_RETURN(~(Uint64)0);
  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
  DBUG_RETURN(setTupleIdInNdb(table->m_tableId, val, increase));
}

bool
Ndb::setTupleIdInNdb(const char* aTableName, Uint64 val, bool increase )
{
  DBUG_ENTER("setTupleIdInNdb(const char*, ...)");
  const NdbTableImpl* table = theDictionary->getTable(aTableName);
  if (table == 0) {
    theError= theDictionary->getNdbError();
    DBUG_RETURN(false);
  }
  DBUG_RETURN(setTupleIdInNdb(table->m_tableId, val, increase));
}

bool
Ndb::setTupleIdInNdb(Uint32 aTableId, Uint64 val, bool increase )
{
  DBUG_ENTER("setTupleIdInNdb(Uint32, ...)");
  if (increase)
  {
    if (theFirstTupleId[aTableId] != theLastTupleId[aTableId])
    {
      // We have a cache sequence
      if (val <= theFirstTupleId[aTableId]+1)
	DBUG_RETURN(false);
      if (val <= theLastTupleId[aTableId])