Merge mskold@bk-internal.mysql.com:/home/bk/mysql-5.0-ndb

into mysql.com:/usr/local/home/marty/MySQL/test/mysql-5.0-ndb

ndb/include/mgmapi/mgmapi.h

@@ -519,6 +519,13 @@ extern "C" {
    */
   int ndb_mgm_connect(NdbMgmHandle handle, int no_retries,
 		      int retry_delay_in_seconds, int verbose);
+  /**
+   * Return true if connected.
+   *
+   * @param   handle        Management handle
+   * @return  0 if not connected, non-zero if connected.
+   */
+  int ndb_mgm_is_connected(NdbMgmHandle handle);
 
   /**
    * Disconnects from a management server

ndb/src/common/transporter/TransporterRegistry.cpp

@@ -1211,11 +1211,34 @@ TransporterRegistry::start_clients_thread()
       switch(performStates[nodeId]){
       case CONNECTING:
 	if(!t->isConnected() && !t->isServer) {
+	  int result= 0;
+	  /**
+	   * First, we try to connect (if we have a port number).
+	   */
+	  if (t->get_s_port())
+	    result= t->connect_client();
+
+
+	  if (result<0 && t->get_s_port()!=0)
+	    g_eventLogger.warning("Error while trying to make connection "
+				  "(Node %u to %u via port %u) "
+				  "error: %d. Retrying...",
+				  t->getRemoteNodeId(),
+				  t->getLocalNodeId(),
+				  t->get_s_port());
+	  
+	  /**
+	   * If dynamic, get the port for connecting from the management server
+	   */
 	  if(t->get_s_port() <= 0) {		// Port is dynamic
 	    int server_port= 0;
 	    struct ndb_mgm_reply mgm_reply;
 	    int res;
 
+	    if(!ndb_mgm_is_connected(m_mgm_handle))
+	      if(ndb_mgm_connect(m_mgm_handle, 0, 0, 0)<0)
+		ndbout_c("Failed to reconnect to management server");
+
 	    res= ndb_mgm_get_connection_int_parameter(m_mgm_handle,
 						     t->getRemoteNodeId(),
 						     t->getLocalNodeId(),
@@ -1231,17 +1254,6 @@ TransporterRegistry::start_clients_thread()
 	    else
 	      ndbout_c("Failed to get dynamic port to connect to: %d", res);
 	  }
-	  if (theTransporterTypes[nodeId] != tt_TCP_TRANSPORTER
-             || t->get_s_port() > 0) {
-	    int result = t->connect_client();
-	    if (result<0)
-	      ndbout_c("Error while trying to make connection (Node %u to"
-		       " %u via port %u) error: %d. Retrying...",
-		       t->getRemoteNodeId(),
-		       t->getLocalNodeId(),
-		       t->get_s_port());
-	  } else
-	    NdbSleep_MilliSleep(400); // wait before retrying
 	}
 	break;
       case DISCONNECTING:

ndb/src/kernel/blocks/suma/Suma.cpp

@@ -610,6 +610,19 @@ Suma::execSIGNAL_DROPPED_REP(Signal* signal){
  *
  */
 
+static unsigned
+count_subscribers(const DLList<SumaParticipant::Subscriber> &subs)
+{
+  unsigned n= 0;
+  SumaParticipant::SubscriberPtr i_subbPtr;
+  subs.first(i_subbPtr);
+  while(!i_subbPtr.isNull()){
+    n++;
+    subs.next(i_subbPtr);
+  }
+  return n;
+}
+
 void
 Suma::execDUMP_STATE_ORD(Signal* signal){
   jamEntry();
@@ -664,6 +677,15 @@ Suma::execDUMP_STATE_ORD(Signal* signal){
     infoEvent("Suma: c_dataBufferPool  size: %d free: %d",
 	      c_dataBufferPool.getSize(),
 	      c_dataBufferPool.getNoOfFree());
+
+    infoEvent("Suma: c_metaSubscribers count: %d",
+	      count_subscribers(c_metaSubscribers));
+    infoEvent("Suma: c_dataSubscribers count: %d",
+	      count_subscribers(c_dataSubscribers));
+    infoEvent("Suma: c_prepDataSubscribers count: %d",
+	      count_subscribers(c_prepDataSubscribers));
+    infoEvent("Suma: c_removeDataSubscribers count: %d",
+	      count_subscribers(c_removeDataSubscribers));
   }
 }
 

ndb/src/mgmapi/mgmapi.cpp

@@ -334,6 +334,17 @@ ndb_mgm_call(NdbMgmHandle handle, const ParserRow<ParserDummy> *command_reply,
   return p;
 }
 
+/**
+ * Returns true if connected
+ */
+extern "C"
+int ndb_mgm_is_connected(NdbMgmHandle handle)
+{
+  if(!handle)
+    return 0;
+  return handle->connected;
+}
+
 /**
  * Connect to a management server
  */

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -2530,6 +2530,7 @@ int
 NdbDictInterface::executeSubscribeEvent(class Ndb & ndb,
 				 NdbEventImpl & evnt)
 {
+  DBUG_ENTER("NdbDictInterface::executeSubscribeEvent");
   NdbApiSignal tSignal(m_reference);
   //  tSignal.theReceiversBlockNumber = SUMA;
   tSignal.theReceiversBlockNumber = DBDICT;
@@ -2544,7 +2545,7 @@ NdbDictInterface::executeSubscribeEvent(class Ndb & ndb,
   sumaStart->subscriberData   = evnt.m_bufferId & 0xFF;
   sumaStart->subscriberRef    = m_reference;
 
-  return executeSubscribeEvent(&tSignal, NULL);
+  DBUG_RETURN(executeSubscribeEvent(&tSignal, NULL));
 }
 
 int

ndb/src/ndbapi/NdbEventOperationImpl.cpp

@@ -92,10 +92,7 @@ NdbEventOperationImpl::NdbEventOperationImpl(NdbEventOperation &N,
 NdbEventOperationImpl::~NdbEventOperationImpl()
 {
   int i;
-  if (sdata) NdbMem_Free(sdata);
-  for (i=0 ; i<3; i++) {
-    if (ptr[i].p) NdbMem_Free(ptr[i].p);
-  }
+  if (sdata) NdbMem_Free((char*)sdata);
   for (i=0 ; i<2; i++) {
     NdbRecAttr *p = theFirstRecAttrs[i];
     while (p) {
@@ -856,39 +853,46 @@ void
 NdbGlobalEventBuffer::real_init (NdbGlobalEventBufferHandle *h,
 				 int MAX_NUMBER_ACTIVE_EVENTS)
 {
-  if (m_handlers.size() == 0) { // First init
+  DBUG_ENTER("NdbGlobalEventBuffer::real_init");
+  DBUG_PRINT("enter",("m_handles.size()=%u %u", m_handlers.size(), h));
+  if (m_handlers.size() == 0)
+  { // First init
+    DBUG_PRINT("info",("first to come"));
     m_max = MAX_NUMBER_ACTIVE_EVENTS;
     m_buf = new BufItem[m_max];
-      // (BufItem *)NdbMem_Allocate(m_max*sizeof(BufItem));
-
     for (int i=0; i<m_max; i++) {
       m_buf[i].gId= 0;
     }
   }
+  assert(m_max == MAX_NUMBER_ACTIVE_EVENTS);
   // TODO make sure we don't hit roof
-  //  m_handlers[m_nhandlers] = h;
   m_handlers.push_back(h);
-  //  ndbout_c("NdbGlobalEventBuffer::real_init(), m_handles=%u %u", m_nhandlers, h);
+  DBUG_VOID_RETURN;
 }
 void
 NdbGlobalEventBuffer::real_remove(NdbGlobalEventBufferHandle *h)
 {
-  //  ndbout_c("NdbGlobalEventBuffer::real_init_remove(), m_handles=%u %u", m_nhandlers, h);
-  for (Uint32 i=0 ; i < m_handlers.size(); i++) {
-    //    ndbout_c("%u %u %u", i, m_handlers[i], h);
-    if (m_handlers[i] == h) {
+  DBUG_ENTER("NdbGlobalEventBuffer::real_remove");
+  DBUG_PRINT("enter",("m_handles.size()=%u %u", m_handlers.size(), h));
+  for (Uint32 i=0 ; i < m_handlers.size(); i++)
+  {
+    DBUG_PRINT("info",("m_handlers[%u] %u", i, m_handlers[i]));
+    if (m_handlers[i] == h)
+    {
       m_handlers.erase(i);
-      if (m_handlers.size() == 0) {
-	//	ndbout_c("last to go");
+      if (m_handlers.size() == 0)
+      {
+	DBUG_PRINT("info",("last to go"));
 	delete[] m_buf;
 	m_buf = NULL;
-	// NdbMem_Free((char*)m_buf);
       }
-      return;
+      DBUG_VOID_RETURN;
     }
   }
-  ndbout_c("NdbGlobalEventBuffer::real_init_remove() non-existing handle");
-  exit(-1);
+  ndbout_c("NdbGlobalEventBuffer::real_remove() non-existing handle");
+  DBUG_PRINT("error",("non-existing handle"));
+  abort();
+  DBUG_VOID_RETURN;
 }
 
 int
@@ -1231,6 +1235,9 @@ int NdbGlobalEventBuffer::real_getDataL(const int bufferId,
     DBUG_RETURN(0); // nothing to get
   }
 
+  DBUG_PRINT("info",("ID(bufferId) %d NO(bufferId) %d e.b %d",
+		     ID(bufferId), NO(bufferId), e.b));
+
   if (copy_data_alloc(b.data[e.b].sdata, b.data[e.b].ptr,
 		      sdata, ptr))
   {
@@ -1255,26 +1262,29 @@ NdbGlobalEventBuffer::copy_data_alloc(const SubTableData * const f_sdata,
 				      LinearSectionPtr t_ptr[3])
 {
   DBUG_ENTER("NdbGlobalEventBuffer::copy_data_alloc");
-  if (t_sdata == NULL) {
-    t_sdata = (SubTableData *)NdbMem_Allocate(sizeof(SubTableData));
-  }
+  unsigned sz4= (sizeof(SubTableData)+3)>>2;
+  Uint32 *ptr= (Uint32*)NdbMem_Allocate((sz4 +
+					 f_ptr[0].sz +
+					 f_ptr[1].sz +
+					 f_ptr[2].sz) * sizeof(Uint32));
+  if (t_sdata)
+    NdbMem_Free((char*)t_sdata);
+  t_sdata= (SubTableData *)ptr;
   memcpy(t_sdata,f_sdata,sizeof(SubTableData));
+  ptr+= sz4;
+
   for (int i = 0; i < 3; i++) {
     LinearSectionPtr & f_p = f_ptr[i];
     LinearSectionPtr & t_p = t_ptr[i];
     if (f_p.sz > 0) {
-      if (t_p.p == NULL) {
-	t_p.p = (Uint32 *)NdbMem_Allocate(sizeof(Uint32)*f_p.sz);
-      } else if (t_p.sz != f_p.sz) {
-	NdbMem_Free(t_p.p);
-	t_p.p = (Uint32 *)NdbMem_Allocate(sizeof(Uint32)*f_p.sz);
-      }
+      t_p.p= (Uint32 *)ptr;
       memcpy(t_p.p, f_p.p, sizeof(Uint32)*f_p.sz);
-    } else if (t_p.p != NULL) {
-      NdbMem_Free(t_p.p);
-      t_p.p = NULL;
+      ptr+= f_p.sz;
+      t_p.sz= f_p.sz;
+    } else {
+      t_p.p= NULL;
+      t_p.sz= 0;
     }
-    t_p.sz = f_p.sz;
   }
   DBUG_RETURN(0);
 }

ndb/test/ndbapi/test_event.cpp

@@ -83,7 +83,7 @@ int runEventOperation(NDBT_Context* ctx, NDBT_Step* step)
 
   EventOperationStats stats;
 
-  g_info << "***** Id " << tId << endl;
+  g_info << "***** start Id " << tId << endl;
 
   //  sleep(tId);
 
@@ -102,12 +102,13 @@ int runEventOperation(NDBT_Context* ctx, NDBT_Step* step)
     ret = NDBT_FAILED;
 
   if (ret == NDBT_FAILED) {
-    ndbout << "n_inserts =           " << stats.n_inserts << endl;
-    ndbout << "n_deletes =           " << stats.n_deletes << endl;
-    ndbout << "n_updates =           " << stats.n_updates << endl;
-    ndbout << "n_consecutive =       " << stats.n_consecutive << endl;
-    ndbout << "n_duplicates =        " << stats.n_duplicates << endl;
-    ndbout << "n_inconsistent_gcis = " << stats.n_inconsistent_gcis << endl;
+    g_info << "***** end Id " << tId << endl;
+    ndbout_c("n_inserts =           %d (%d)", stats.n_inserts, records);
+    ndbout_c("n_deletes =           %d (%d)", stats.n_deletes, records);
+    ndbout_c("n_updates =           %d (%d)", stats.n_updates, records);
+    ndbout_c("n_consecutive =       %d (%d)", stats.n_consecutive, 3);
+    ndbout_c("n_duplicates =        %d (%d)", stats.n_duplicates, 0);
+    ndbout_c("n_inconsistent_gcis = %d (%d)", stats.n_inconsistent_gcis, 0);
   }
 
   return ret;
@@ -156,9 +157,6 @@ TESTCASE("BasicEventOperation",
 	 "NOTE! No errors are allowed!" ){
   INITIALIZER(runCreateEvent);
   STEP(runEventOperation);
-  STEP(runEventOperation);
-  STEP(runEventOperation);
-  STEP(runEventOperation);
   STEP(runEventLoad);
   FINALIZER(runDropEvent);
 }
@@ -169,19 +167,16 @@ TESTCASE("CreateDropEventOperation",
   STEP(runCreateDropEventOperation);
   FINALIZER(runDropEvent);
 }
-NDBT_TESTSUITE_END(test_event);
-
-#if 0
-NDBT_TESTSUITE(test_event);
 TESTCASE("ParallellEventOperation", 
-	 "Verify that we can listen to Events in Parallell"
+	 "Verify that we can listen to Events in parallell"
 	 "NOTE! No errors are allowed!" ){
-  INITIALIZER(runCreateAllEvent);
+  INITIALIZER(runCreateEvent);
+  STEP(runEventOperation);
   STEP(runEventOperation);
+  STEP(runEventLoad);
   FINALIZER(runDropEvent);
 }
 NDBT_TESTSUITE_END(test_event);
-#endif
 
 int main(int argc, const char** argv){
   ndb_init();