bk commit into 5.1 tree (pekka:1.2398)

[pekka]

Below is the list of changes that have just been committed into a local
5.1 repository of pekka. When pekka does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet
  1.2398 06/05/16 16:23:12 pekka@stripped +1 -0
  Merge mysql.com:/space/pekka/ndb/version/my50-bug14509
  into  mysql.com:/space/pekka/ndb/version/my51-bug14509

  storage/ndb/src/ndbapi/Ndb.cpp
    1.71 06/05/16 16:22:53 pekka@stripped +0 -0
    Auto merged

  storage/ndb/src/ndbapi/Ndb.cpp
    1.49.10.2 06/05/16 16:22:53 pekka@stripped +0 -0
    Merge rename: ndb/src/ndbapi/Ndb.cpp -> storage/ndb/src/ndbapi/Ndb.cpp

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	pekka
# Host:	orca.ndb.mysql.com
# Root:	/space/pekka/ndb/version/my51-bug14509/RESYNC

--- 1.49.10.1/ndb/src/ndbapi/Ndb.cpp	2006-05-16 16:14:36 +02:00
+++ 1.71/storage/ndb/src/ndbapi/Ndb.cpp	2006-05-16 16:22:53 +02:00
@@ -29,6 +29,7 @@
 #include <NdbOperation.hpp>
 #include <NdbTransaction.hpp>
 #include <NdbEventOperation.hpp>
+#include <NdbEventOperationImpl.hpp>
 #include <NdbRecAttr.hpp>
 #include <md5_hash.hpp>
 #include <NdbSleep.h>
@@ -138,7 +139,7 @@
 //***************************************************************************
   
   int	         tReturnCode;
-  TransporterFacade *tp = TransporterFacade::instance();
+  TransporterFacade *tp = theImpl->m_transporter_facade;
 
   DBUG_ENTER("Ndb::NDB_connect");
 
@@ -173,23 +174,9 @@
   tSignal->setData(theMyRef, 2);	// Set my block reference
   tNdbCon->Status(NdbTransaction::Connecting); // Set status to connecting
   Uint32 nodeSequence;
-  { // send and receive signal
-    Guard guard(tp->theMutexPtr);
-    nodeSequence = tp->getNodeSequence(tNode);
-    bool node_is_alive = tp->get_node_alive(tNode);
-    if (node_is_alive) { 
-      tReturnCode = tp->sendSignal(tSignal, tNode);  
-      releaseSignal(tSignal); 
-      if (tReturnCode != -1) {
-        theImpl->theWaiter.m_node = tNode;  
-        theImpl->theWaiter.m_state = WAIT_TC_SEIZE;  
-        tReturnCode = receiveResponse(); 
-      }//if
-    } else {
-      releaseSignal(tSignal);
-      tReturnCode = -1;
-    }//if
-  }
+  tReturnCode= sendRecSignal(tNode, WAIT_TC_SEIZE, tSignal,
+                             0, &nodeSequence);
+  releaseSignal(tSignal); 
   if ((tReturnCode == 0) && (tNdbCon->Status() == NdbTransaction::Connected))
{
     //************************************************
     // Send and receive was successful
@@ -230,10 +217,9 @@
 void 
 Ndb::doDisconnect()
 {
+  DBUG_ENTER("Ndb::doDisconnect");
   NdbTransaction* tNdbCon;
   CHECK_STATUS_MACRO_VOID;
-  /* DBUG_ENTER must be after CHECK_STATUS_MACRO_VOID because of 'return' */
-  DBUG_ENTER("Ndb::doDisconnect");
 
   Uint32 tNoOfDbNodes = theImpl->theNoOfDBnodes;
   Uint8 *theDBnodes= theImpl->theDBnodes;
@@ -588,6 +574,7 @@
 #ifdef CUSTOMER_RELEASE
   return -1;
 #else
+  DBUG_ENTER("Ndb::NdbTamper");
   CHECK_STATUS_MACRO;
   checkFailedNode();
 
@@ -609,20 +596,20 @@
 	break;
      default:
         theError.code = 4102;
-        return -1;
+        DBUG_RETURN(-1);
   }
 
   tNdbConn = getNdbCon();	// Get free connection object
   if (tNdbConn == NULL) {
     theError.code = 4000;
-    return -1;
+    DBUG_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();
+  TransporterFacade *tp = theImpl->m_transporter_facade;
   if (tAction == 3) {
     tp->lock_mutex();
     tp->sendSignal(&tSignal, aNode);
@@ -634,12 +621,12 @@
     if (tNode == 0) {
       theError.code = 4002;
       releaseNdbCon(tNdbConn);
-      return -1;
+      DBUG_RETURN(-1);
     }//if
     ret_code = tp->sendSignal(&tSignal,aNode);
     tp->unlock_mutex();
     releaseNdbCon(tNdbConn);
-    return ret_code;
+    DBUG_RETURN(ret_code);
   } else {
     do {
       tp->lock_mutex();
@@ -650,7 +637,7 @@
       if (tNode == 0) {
         theError.code = 4009;
         releaseNdbCon(tNdbConn);
-        return -1;
+        DBUG_RETURN(-1);
       }//if
       ret_code = sendRecSignal(tNode, WAIT_NDB_TAMPER, &tSignal, 0);
       if (ret_code == 0) {  
@@ -658,15 +645,15 @@
           theRestartGCI = 0;
         }//if
         releaseNdbCon(tNdbConn);
-        return theRestartGCI;
+        DBUG_RETURN(theRestartGCI);
       } else if ((ret_code == -5) || (ret_code == -2)) {
         TRACE_DEBUG("Continue DIHNDBTAMPER when node failed/stopping");
       } else {
-        return -1;
+        DBUG_RETURN(-1);
       }//if
     } while (1);
   }
-  return 0;
+  DBUG_RETURN(0);
 #endif
 }
 #if 0
@@ -767,7 +754,7 @@
   BaseString internal_tabname(internalize_table_name(aTableName));
 
   Ndb_local_table_info *info=
-    theDictionary->get_local_table_info(internal_tabname, false);
+    theDictionary->get_local_table_info(internal_tabname);
   if (info == 0) {
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(~(Uint64)0);
@@ -885,7 +872,7 @@
   BaseString internal_tabname(internalize_table_name(aTableName));
 
   Ndb_local_table_info *info=
-    theDictionary->get_local_table_info(internal_tabname, false);
+    theDictionary->get_local_table_info(internal_tabname);
   if (info == 0) {
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(~(Uint64)0);
@@ -943,6 +930,27 @@
     DBUG_RETURN((opTupleIdOnNdb(info, val, 1)));
 }
 
+int Ndb::initAutoIncrement()
+{
+  if (m_sys_tab_0)
+    return 0;
+
+  BaseString currentDb(getDatabaseName());
+  BaseString currentSchema(getDatabaseSchemaName());
+
+  setDatabaseName("sys");
+  setDatabaseSchemaName("def");
+
+  m_sys_tab_0 = getDictionary()->getTableGlobal("SYSTAB_0");
+
+  // Restore current name space
+  setDatabaseName(currentDb.c_str());
+  setDatabaseSchemaName(currentSchema.c_str());
+
+
+  return (m_sys_tab_0 == NULL);
+}
+
 Uint64
 Ndb::opTupleIdOnNdb(Ndb_local_table_info* info, Uint64 opValue, Uint32 op)
 {
@@ -958,19 +966,14 @@
 
   CHECK_STATUS_MACRO_ZERO;
 
-  BaseString currentDb(getDatabaseName());
-  BaseString currentSchema(getDatabaseSchemaName());
+  if (initAutoIncrement())
+    goto error_return;
 
-  setDatabaseName("sys");
-  setDatabaseSchemaName("def");
   tConnection = this->startTransaction();
   if (tConnection == NULL)
     goto error_return;
 
-  if (usingFullyQualifiedNames())
-    tOperation = tConnection->getNdbOperation("SYSTAB_0");
-  else
-    tOperation = tConnection->getNdbOperation("sys/def/SYSTAB_0");
+  tOperation = tConnection->getNdbOperation(m_sys_tab_0);
   if (tOperation == NULL)
     goto error_handler;
 
@@ -1045,20 +1048,12 @@
 
   this->closeTransaction(tConnection);
 
-  // Restore current name space
-  setDatabaseName(currentDb.c_str());
-  setDatabaseSchemaName(currentSchema.c_str());
-
   DBUG_RETURN(ret);
 
   error_handler:
     theError.code = tConnection->theError.code;
     this->closeTransaction(tConnection);
   error_return:
-    // Restore current name space
-    setDatabaseName(currentDb.c_str());
-    setDatabaseSchemaName(currentSchema.c_str());
-
   DBUG_PRINT("error", ("ndb=%d con=%d op=%d",
              theError.code,
              tConnection ? tConnection->theError.code : -1,
@@ -1083,39 +1078,37 @@
   return Data;
 #endif
 }
+
+// <internal>
 const char * Ndb::getCatalogName() const
 {
   return theImpl->m_dbname.c_str();
 }
 
-
 void Ndb::setCatalogName(const char * a_catalog_name)
 {
-  if (a_catalog_name)
-  {
+  // TODO can table_name_separator be escaped?
+  if (a_catalog_name && ! strchr(a_catalog_name, table_name_separator)) {
     theImpl->m_dbname.assign(a_catalog_name);
     theImpl->update_prefix();
   }
 }
 
-
 const char * Ndb::getSchemaName() const
 {
   return theImpl->m_schemaname.c_str();
 }
 
-
 void Ndb::setSchemaName(const char * a_schema_name)
 {
-  if (a_schema_name) {
+  // TODO can table_name_separator be escaped?
+  if (a_schema_name && ! strchr(a_schema_name, table_name_separator)) {
     theImpl->m_schemaname.assign(a_schema_name);
     theImpl->update_prefix();
   }
 }
+// </internal>
  
-/*
-Deprecated functions
-*/
 const char * Ndb::getDatabaseName() const
 {
   return getCatalogName();
@@ -1135,6 +1128,26 @@
 {
   setSchemaName(a_schema_name);
 }
+
+int Ndb::setDatabaseAndSchemaName(const NdbDictionary::Table* t)
+{
+  const char* s0 = t->m_impl.m_internalName.c_str();
+  const char* s1 = strchr(s0, table_name_separator);
+  if (s1 && s1 != s0) {
+    const char* s2 = strchr(s1 + 1, table_name_separator);
+    if (s2 && s2 != s1 + 1) {
+      char buf[NAME_LEN + 1];
+      if (s1 - s0 <= NAME_LEN && s2 - (s1 + 1) <= NAME_LEN) {
+        sprintf(buf, "%.*s", s1 - s0, s0);
+        setDatabaseName(buf);
+        sprintf(buf, "%.*s", s2 - (s1 + 1), s1 + 1);
+        setDatabaseSchemaName(buf);
+        return 0;
+      }
+    }
+  }
+  return -1;
+}
  
 bool Ndb::usingFullyQualifiedNames()
 {
@@ -1197,9 +1210,16 @@
   if (fullyQualifiedNames)
   {
     /* Internal table name format <db>/<schema>/<table>
-       <db>/<schema> is already available in m_prefix
+       <db>/<schema>/ is already available in m_prefix
        so just concat the two strings
      */
+#ifdef VM_TRACE
+    // verify that m_prefix looks like abc/def/
+    const char* s0 = theImpl->m_prefix.c_str();
+    const char* s1 = s0 ? strchr(s0, table_name_separator) : 0;
+    const char* s2 = s1 ? strchr(s1 + 1, table_name_separator) : 0;
+    assert(s1 && s1 != s0 && s2 && s2 != s1 + 1 && *(s2 +
1) == 0);
+#endif
     ret.assfmt("%s%s",
                theImpl->m_prefix.c_str(),
                external_name);
@@ -1219,11 +1239,11 @@
   BaseString ret;
   DBUG_ENTER("internalize_index_name");
   DBUG_PRINT("enter", ("external_name: %s, table_id: %d",
-                       external_name, table ? table->m_tableId : ~0));
+                       external_name, table ? table->m_id : ~0));
   if (!table)
   {
     DBUG_PRINT("error", ("!table"));
-    return ret;
+    DBUG_RETURN(ret);
   }
 
   if (fullyQualifiedNames)
@@ -1231,7 +1251,7 @@
     /* Internal index name format <db>/<schema>/<tabid>/<table>
*/
     ret.assfmt("%s%d%c%s",
                theImpl->m_prefix.c_str(),
-               table->m_tableId,
+               table->m_id,
                table_name_separator,
                external_name);
   }
@@ -1278,49 +1298,102 @@
   return ret;
 }
 
-NdbEventOperation* Ndb::createEventOperation(const char* eventName,
-					     const int bufferLength)
+// ToDo set event buffer size
+NdbEventOperation* Ndb::createEventOperation(const char* eventName)
 {
-  NdbEventOperation* tOp;
+  DBUG_ENTER("Ndb::createEventOperation");
+  NdbEventOperation* tOp= theEventBuffer->createEventOperation(eventName,
+							       theError);
+  if (tOp)
+  {
+    // keep track of all event operations
+    NdbEventOperationImpl *op=
+      NdbEventBuffer::getEventOperationImpl(tOp);
+    op->m_next= theImpl->m_ev_op;
+    op->m_prev= 0;
+    theImpl->m_ev_op= op;
+    if (op->m_next)
+      op->m_next->m_prev= op;
+  }
+
+  DBUG_RETURN(tOp);
+}
+
+int Ndb::dropEventOperation(NdbEventOperation* tOp)
+{
+  DBUG_ENTER("Ndb::dropEventOperation");
+  // remove it from list
+  NdbEventOperationImpl *op=
+    NdbEventBuffer::getEventOperationImpl(tOp);
+  if (op->m_next)
+    op->m_next->m_prev= op->m_prev;
+  if (op->m_prev)
+    op->m_prev->m_next= op->m_next;
+  else
+    theImpl->m_ev_op= op->m_next;
 
-  tOp = new NdbEventOperation(this, eventName, bufferLength);
+  assert(theImpl->m_ev_op == 0 || theImpl->m_ev_op->m_prev == 0);
 
-  if (tOp == 0)
-  {
-    theError.code= 4000;
-    return NULL;
-  }
+  theEventBuffer->dropEventOperation(tOp);
+  DBUG_RETURN(0);
+}
 
-  if (tOp->getState() != NdbEventOperation::EO_CREATED) {
-    theError.code= tOp->getNdbError().code;
-    delete tOp;
-    tOp = NULL;
-  }
+NdbEventOperation *Ndb::getEventOperation(NdbEventOperation* tOp)
+{
+  NdbEventOperationImpl *op;
+  if (tOp)
+    op= NdbEventBuffer::getEventOperationImpl(tOp)->m_next;
+  else
+    op= theImpl->m_ev_op;
+  if (op)
+    return op->m_facade;
+  return 0;
+}
 
-  //now we have to look up this event in dict
+int
+Ndb::pollEvents(int aMillisecondNumber, Uint64 *latestGCI)
+{
+  return theEventBuffer->pollEvents(aMillisecondNumber, latestGCI);
+}
 
-  return tOp;
+NdbEventOperation *Ndb::nextEvent()
+{
+  return theEventBuffer->nextEvent();
 }
 
-int Ndb::dropEventOperation(NdbEventOperation* op) {
-  delete op;
-  return 0;
+const NdbEventOperation*
+Ndb::getGCIEventOperations(Uint32* iter, Uint32* event_types)
+{
+  NdbEventOperationImpl* op =
+    theEventBuffer->getGCIEventOperations(iter, event_types);
+  if (op != NULL)
+    return op->m_facade;
+  return NULL;
 }
 
-NdbGlobalEventBufferHandle* Ndb::getGlobalEventBufferHandle()
+Uint64 Ndb::getLatestGCI()
 {
-  return theGlobalEventBufferHandle;
+  return theEventBuffer->getLatestGCI();
 }
 
-//void Ndb::monitorEvent(NdbEventOperation *op, NdbEventCallback cb, void* rs)
-//{
-//}
+void Ndb::setReportThreshEventGCISlip(unsigned thresh)
+{
+ if (theEventBuffer->m_free_thresh != thresh)
+ {
+   theEventBuffer->m_free_thresh= thresh;
+   theEventBuffer->m_min_free_thresh= thresh;
+   theEventBuffer->m_max_free_thresh= 100;
+ }
+}
 
-int
-Ndb::pollEvents(int aMillisecondNumber)
+void Ndb::setReportThreshEventFreeMem(unsigned thresh)
 {
-  return NdbEventOperation::wait(theGlobalEventBufferHandle,
-				 aMillisecondNumber);
+  if (theEventBuffer->m_free_thresh != thresh)
+  {
+    theEventBuffer->m_free_thresh= thresh;
+    theEventBuffer->m_min_free_thresh= thresh;
+    theEventBuffer->m_max_free_thresh= 100;
+  }
 }
 
 #ifdef VM_TRACE