ndb.cpp

mysql-5.0.24源码包

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.
****************************************************************************/
int
Ndb::getAutoIncrementValue(const char* aTableName,
                           Uint64 & tupleId, Uint32 cacheSize)
{
  DBUG_ENTER("Ndb::getAutoIncrementValue");
  BaseString internal_tabname(internalize_table_name(aTableName));

  Ndb_local_table_info *info=
    theDictionary->get_local_table_info(internal_tabname, false);
  if (info == 0) {
    theError.code = theDictionary->getNdbError().code;
    DBUG_RETURN(-1);
  }
  if (getTupleIdFromNdb(info, tupleId, cacheSize) == -1)
    DBUG_RETURN(-1);
  DBUG_PRINT("info", ("value %llu", (ulonglong)tupleId));
  DBUG_RETURN(0);
}

int
Ndb::getAutoIncrementValue(const NdbDictionary::Table * aTable,
                           Uint64 & tupleId, Uint32 cacheSize)
{
  DBUG_ENTER("Ndb::getAutoIncrementValue");
  assert(aTable != 0);
  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
  const BaseString& internal_tabname = table->m_internalName;

  Ndb_local_table_info *info=
    theDictionary->get_local_table_info(internal_tabname, false);
  if (info == 0) {
    theError.code = theDictionary->getNdbError().code;
    DBUG_RETURN(-1);
  }
  if (getTupleIdFromNdb(info, tupleId, cacheSize) == -1)
    DBUG_RETURN(-1);
  DBUG_PRINT("info", ("value %llu", (ulonglong)tupleId));
  DBUG_RETURN(0);
}

int
Ndb::getTupleIdFromNdb(Ndb_local_table_info* info,
                       Uint64 & tupleId, Uint32 cacheSize)
{
  DBUG_ENTER("Ndb::getTupleIdFromNdb");
  if (info->m_first_tuple_id != info->m_last_tuple_id)
  {
    assert(info->m_first_tuple_id < info->m_last_tuple_id);
    tupleId = ++info->m_first_tuple_id;
    DBUG_PRINT("info", ("next cached value %llu", (ulonglong)tupleId));
  }
  else
  {
    if (cacheSize == 0)
      cacheSize = 1;
    DBUG_PRINT("info", ("reading %u values from database", (uint)cacheSize));
    /*
     * reserve next cacheSize entries in db.  adds cacheSize to NEXTID
     * and returns first tupleId in the new range.
     */
    Uint64 opValue = cacheSize;
    if (opTupleIdOnNdb(info, opValue, 0) == -1)
      DBUG_RETURN(-1);
    tupleId = opValue;
  }
  DBUG_RETURN(0);
}

int
Ndb::readAutoIncrementValue(const char* aTableName,
                            Uint64 & tupleId)
{
  DBUG_ENTER("Ndb::readAutoIncrementValue");
  BaseString internal_tabname(internalize_table_name(aTableName));

  Ndb_local_table_info *info=
    theDictionary->get_local_table_info(internal_tabname, false);
  if (info == 0) {
    theError.code = theDictionary->getNdbError().code;
    DBUG_RETURN(-1);
  }
  if (readTupleIdFromNdb(info, tupleId) == -1)
    DBUG_RETURN(-1);
  DBUG_PRINT("info", ("value %llu", (ulonglong)tupleId));
  DBUG_RETURN(0);
}

int
Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable,
                            Uint64 & tupleId)
{
  DBUG_ENTER("Ndb::readAutoIncrementValue");
  assert(aTable != 0);
  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
  const BaseString& internal_tabname = table->m_internalName;

  Ndb_local_table_info *info=
    theDictionary->get_local_table_info(internal_tabname, false);
  if (info == 0) {
    theError.code = theDictionary->getNdbError().code;
    DBUG_RETURN(-1);
  }
  if (readTupleIdFromNdb(info, tupleId) == -1)
    DBUG_RETURN(-1);
  DBUG_PRINT("info", ("value %llu", (ulonglong)tupleId));
  DBUG_RETURN(0);
}

int
Ndb::readTupleIdFromNdb(Ndb_local_table_info* info,
                        Uint64 & tupleId)
{
  DBUG_ENTER("Ndb::readTupleIdFromNdb");
  if (info->m_first_tuple_id != info->m_last_tuple_id)
  {
    assert(info->m_first_tuple_id < info->m_last_tuple_id);
    tupleId = info->m_first_tuple_id + 1;
  }
  else
  {
    /*
     * peek at NEXTID.  does not reserve it so the value is valid
     * only if no other transactions are allowed.
     */
    Uint64 opValue = 0;
    if (opTupleIdOnNdb(info, opValue, 3) == -1)
      DBUG_RETURN(-1);
    tupleId = opValue;
  }
  DBUG_RETURN(0);
}

int
Ndb::setAutoIncrementValue(const char* aTableName,
                           Uint64 tupleId, bool increase)
{
  DBUG_ENTER("Ndb::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.code = theDictionary->getNdbError().code;
    DBUG_RETURN(-1);
  }
  if (setTupleIdInNdb(info, tupleId, increase) == -1)
    DBUG_RETURN(-1);
  DBUG_RETURN(0);
}

int
Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable,
                           Uint64 tupleId, bool increase)
{
  DBUG_ENTER("Ndb::setAutoIncrementValue");
  assert(aTable != 0);
  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
  const BaseString& internal_tabname = table->m_internalName;

  Ndb_local_table_info *info=
    theDictionary->get_local_table_info(internal_tabname, false);
  if (info == 0) {
    theError.code = theDictionary->getNdbError().code;
    DBUG_RETURN(-1);
  }
  if (setTupleIdInNdb(info, tupleId, increase) == -1)
    DBUG_RETURN(-1);
  DBUG_RETURN(0);
}

int
Ndb::setTupleIdInNdb(Ndb_local_table_info* info,
                     Uint64 tupleId, bool increase)
{
  DBUG_ENTER("Ndb::setTupleIdInNdb");
  if (increase)
  {
    if (info->m_first_tuple_id != info->m_last_tuple_id)
    {
      assert(info->m_first_tuple_id < info->m_last_tuple_id);