bk commit into 5.1 tree (tomas:1.2154) BUG#17610

[tomas]

Below is the list of changes that have just been committed into a local
5.1 repository of tomas. When tomas 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.2154 06/05/18 23:38:07 tomas@stripped +10 -0
      Bug #19395 mysqld does not always detect cluster shutdown
      Bug #17610 ndbapi: dropEventOperation leaves object behind, memleak

  storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp
    1.22 06/05/18 23:37:53 tomas@stripped +9 -5
    Bug #17610 ndbapi: dropEventOperation leaves object behind, memleak

  storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp
    1.58 06/05/18 23:37:53 tomas@stripped +230 -108
    Bug #17610 ndbapi: dropEventOperation leaves object behind, memleak

  storage/ndb/src/ndbapi/Ndb.cpp
    1.70 06/05/18 23:37:53 tomas@stripped +6 -0
    Bug #19395 mysqld does not always detect cluster shutdown

  storage/ndb/src/kernel/blocks/suma/Suma.cpp
    1.42 06/05/18 23:37:53 tomas@stripped +16 -1
    Bug #17610 ndbapi: dropEventOperation leaves object behind, memleak

  storage/ndb/include/ndbapi/NdbDictionary.hpp
    1.75 06/05/18 23:37:53 tomas@stripped +2 -1
    Bug #17610 ndbapi: dropEventOperation leaves object behind, memleak

  storage/ndb/include/ndbapi/Ndb.hpp
    1.50 06/05/18 23:37:53 tomas@stripped +1 -0
    Bug #19395 mysqld does not always detect cluster shutdown

  sql/ha_ndbcluster_binlog.cc
    1.55 06/05/18 23:37:53 tomas@stripped +3 -0
    Bug #19395 mysqld does not always detect cluster shutdown

  mysql-test/t/ndb_autodiscover3.test
    1.4 06/05/18 23:37:53 tomas@stripped +4 -0
    Bug #19395 mysqld does not always detect cluster shutdown

  mysql-test/t/disabled.def
    1.157 06/05/18 23:37:53 tomas@stripped +1 -1
    Bug #19395 mysqld does not always detect cluster shutdown

  mysql-test/r/ndb_autodiscover3.result
    1.5 06/05/18 23:37:53 tomas@stripped +4 -0
    Bug #19395 mysqld does not always detect cluster shutdown

# 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:	tomas
# Host:	poseidon.ndb.mysql.com
# Root:	/home/tomas/mysql-5.1-new-ndb

--- 1.156/mysql-test/t/disabled.def	2006-05-16 20:58:11 +02:00
+++ 1.157/mysql-test/t/disabled.def	2006-05-18 23:37:53 +02:00
@@ -16,7 +16,7 @@
 events_logs_tests        : BUG#17619 2006-05-16 andrey  Test case problems
 ndb_autodiscover         : BUG#18952 2006-02-16 jmiller Needs to be fixed w.r.t binlog
 ndb_autodiscover2        : BUG#18952 2006-02-16 jmiller Needs to be fixed w.r.t binlog
-ndb_binlog_discover      : BUG#19395 2006-04-28 tomas/knielsen mysqld does not always
detect cluster shutdown
+#ndb_binlog_discover      : BUG#19395 2006-04-28 tomas/knielsen mysqld does not always
detect cluster shutdown
 #ndb_cache2               : BUG#18597 2006-03-28 brian simultaneous drop table and ndb
statistics update triggers node failure
 #ndb_cache_multi2         : BUG#18597 2006-04-10 kent  simultaneous drop table and ndb
statistics update triggers node failure
 ndb_load                 : BUG#17233 2006-05-04 tomas failed load data from infile causes
mysqld dbug_assert, binlog not flushed

--- 1.49/storage/ndb/include/ndbapi/Ndb.hpp	2006-05-04 13:58:06 +02:00
+++ 1.50/storage/ndb/include/ndbapi/Ndb.hpp	2006-05-18 23:37:53 +02:00
@@ -1262,6 +1262,7 @@
   
 
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
+  int flushIncompleteEvents(Uint64 gci);
   NdbEventOperation *getEventOperation(NdbEventOperation* eventOp= 0);
   Uint64 getLatestGCI();
   void forceGCP();

--- 1.74/storage/ndb/include/ndbapi/NdbDictionary.hpp	2006-05-04 13:58:06 +02:00
+++ 1.75/storage/ndb/include/ndbapi/NdbDictionary.hpp	2006-05-18 23:37:53 +02:00
@@ -1132,7 +1132,8 @@
       _TE_NODE_FAILURE=10,
       _TE_SUBSCRIBE=11,
       _TE_UNSUBSCRIBE=12,
-      _TE_NUL=13 // internal (e.g. INS o DEL within same GCI)
+      _TE_NUL=13, // internal (e.g. INS o DEL within same GCI)
+      _TE_ACTIVE=14 // internal (node becomes active)
     };
 #endif
     /**

--- 1.41/storage/ndb/src/kernel/blocks/suma/Suma.cpp	2006-03-27 18:53:16 +02:00
+++ 1.42/storage/ndb/src/kernel/blocks/suma/Suma.cpp	2006-05-18 23:37:53 +02:00
@@ -2649,6 +2649,22 @@
                            SubscriptionPtr subPtr,
                            SubscriberPtr subbPtr)
 {
+  SubTableData * data  = (SubTableData*)signal->getDataPtrSend();
+
+  if (table_event == NdbDictionary::Event::_TE_SUBSCRIBE)
+  {
+    data->gci            = m_last_complete_gci + 1;
+    data->tableId        = subPtr.p->m_tableId;
+    data->operation      = NdbDictionary::Event::_TE_ACTIVE;
+    data->ndbd_nodeid    = refToNode(reference());
+    data->changeMask     = 0;
+    data->totalLen       = 0;
+    data->req_nodeid     = refToNode(subbPtr.p->m_senderRef);
+    data->senderData     = subbPtr.p->m_senderData;
+    sendSignal(subbPtr.p->m_senderRef, GSN_SUB_TABLE_DATA, signal,
+               SubTableData::SignalLength, JBB);
+  }
+
   if (!(subPtr.p->m_options & Subscription::REPORT_SUBSCRIBE))
   {
     return;
@@ -2663,7 +2679,6 @@
   ndbout_c("reportAllSubscribers  subPtr.i: %d  subPtr.p->n_subscribers: %d",
            subPtr.i, subPtr.p->n_subscribers);
 //#endif
-  SubTableData * data  = (SubTableData*)signal->getDataPtrSend();
   data->gci            = m_last_complete_gci + 1;
   data->tableId        = subPtr.p->m_tableId;
   data->operation      = table_event;

--- 1.69/storage/ndb/src/ndbapi/Ndb.cpp	2006-05-16 20:56:38 +02:00
+++ 1.70/storage/ndb/src/ndbapi/Ndb.cpp	2006-05-18 23:37:53 +02:00
@@ -1324,6 +1324,12 @@
   return theEventBuffer->pollEvents(aMillisecondNumber, latestGCI);
 }
 
+int
+Ndb::flushIncompleteEvents(Uint64 gci)
+{
+  return theEventBuffer->flushIncompleteEvents(gci);
+}
+
 NdbEventOperation *Ndb::nextEvent()
 {
   return theEventBuffer->nextEvent();

--- 1.57/storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp	2006-05-10 13:32:50 +02:00
+++ 1.58/storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp	2006-05-18 23:37:53 +02:00
@@ -153,11 +153,14 @@
 
   m_state= EO_CREATED;
 
+  m_node_bit_mask.clear();
 #ifdef ndb_event_stores_merge_events_flag
   m_mergeEvents = m_eventImpl->m_mergeEvents;
 #else
-   m_mergeEvents = false;
+  m_mergeEvents = false;
 #endif
+  m_ref_count = 0;
+  DBUG_PRINT("info", ("m_ref_count = 0 for op: %p", this));
 
   m_has_error= 0;
 
@@ -530,7 +533,11 @@
       }
     }
     if (r == 0)
+    {
+      m_ref_count++;
+      DBUG_PRINT("info", ("m_ref_count: %u for op: %p", m_ref_count, this));
       DBUG_RETURN(0);
+    }
   }
   //Error
   m_state= EO_ERROR;
@@ -657,80 +664,79 @@
 int
 NdbEventOperationImpl::receive_event()
 {
-  DBUG_ENTER_EVENT("NdbEventOperationImpl::receive_event");
-
   Uint32 operation= (Uint32)m_data_item->sdata->operation;
-  DBUG_PRINT_EVENT("info",("sdata->operation %u",operation));
-
-  if (operation == NdbDictionary::Event::_TE_ALTER)
+  if (unlikely(operation >= NdbDictionary::Event::_TE_FIRST_NON_DATA_EVENT))
   {
-    // Parse the new table definition and
-    // create a table object
-    NdbDictionary::Dictionary *myDict = m_ndb->getDictionary();
-    NdbDictionaryImpl *dict = & NdbDictionaryImpl::getImpl(*myDict);
-    NdbError error;
-    NdbDictInterface dif(error);
-    NdbTableImpl *at;
-    m_change_mask = m_data_item->sdata->changeMask;
-    error.code = dif.parseTableInfo(&at,
-                                    (Uint32*)m_buffer.get_data(), 
-                                    m_buffer.length() / 4, 
-                                    true);
-    m_buffer.clear();
-    if (at)
+    DBUG_ENTER("NdbEventOperationImpl::receive_event");
+    DBUG_PRINT("info",("sdata->operation %u  this: %p", operation, this));
+    if (operation == NdbDictionary::Event::_TE_ALTER)
+    {
+      // Parse the new table definition and
+      // create a table object
+      NdbDictionary::Dictionary *myDict = m_ndb->getDictionary();
+      NdbDictionaryImpl *dict = & NdbDictionaryImpl::getImpl(*myDict);
+      NdbError error;
+      NdbDictInterface dif(error);
+      NdbTableImpl *at;
+      m_change_mask = m_data_item->sdata->changeMask;
+      error.code = dif.parseTableInfo(&at,
+                                      (Uint32*)m_buffer.get_data(), 
+                                      m_buffer.length() / 4, 
+                                      true);
+      m_buffer.clear();
+      if (unlikely(!at))
+      {
+        DBUG_PRINT("info", ("Failed to parse DictTabInfo error %u", 
+                                  error.code));
+        ndbout_c("Failed to parse DictTabInfo error %u", error.code);
+        DBUG_RETURN(1);
+      }
       at->buildColumnHash();
-    else
-    {
-      DBUG_PRINT_EVENT("info", ("Failed to parse DictTabInfo error %u", 
-                                error.code));
-      DBUG_RETURN_EVENT(1);
-    }
-
-    NdbTableImpl *tmp_table_impl= m_eventImpl->m_tableImpl;
-    m_eventImpl->m_tableImpl = at;
-
-    DBUG_PRINT("info", ("switching table impl 0x%x -> 0x%x",
-                        tmp_table_impl, at));
-
-    // change the rec attrs to refer to the new table object
-    int i;
-    for (i = 0; i < 2; i++)
-    {
-      NdbRecAttr *p = theFirstPkAttrs[i];
-      while (p)
+      
+      NdbTableImpl *tmp_table_impl= m_eventImpl->m_tableImpl;
+      m_eventImpl->m_tableImpl = at;
+      
+      DBUG_PRINT("info", ("switching table impl 0x%x -> 0x%x",
+                          tmp_table_impl, at));
+      
+      // change the rec attrs to refer to the new table object
+      int i;
+      for (i = 0; i < 2; i++)
       {
-        int no = p->getColumn()->getColumnNo();
-        NdbColumnImpl *tAttrInfo = at->getColumn(no);
-        DBUG_PRINT("info", ("rec_attr: 0x%x  "
-                            "switching column impl 0x%x -> 0x%x",
-                            p, p->m_column, tAttrInfo));
-        p->m_column = tAttrInfo;
-        p = p->next();
+        NdbRecAttr *p = theFirstPkAttrs[i];
+        while (p)
+        {
+          int no = p->getColumn()->getColumnNo();
+          NdbColumnImpl *tAttrInfo = at->getColumn(no);
+          DBUG_PRINT("info", ("rec_attr: 0x%x  "
+                              "switching column impl 0x%x -> 0x%x",
+                              p, p->m_column, tAttrInfo));
+          p->m_column = tAttrInfo;
+          p = p->next();
+        }
       }
-    }
-    for (i = 0; i < 2; i++)
-    {
-      NdbRecAttr *p = theFirstDataAttrs[i];
-      while (p)
+      for (i = 0; i < 2; i++)
       {
-        int no = p->getColumn()->getColumnNo();
-        NdbColumnImpl *tAttrInfo = at->getColumn(no);
-        DBUG_PRINT("info", ("rec_attr: 0x%x  "
-                            "switching column impl 0x%x -> 0x%x",
-                            p, p->m_column, tAttrInfo));
-        p->m_column = tAttrInfo;
-        p = p->next();
+        NdbRecAttr *p = theFirstDataAttrs[i];
+        while (p)
+        {
+          int no = p->getColumn()->getColumnNo();
+          NdbColumnImpl *tAttrInfo = at->getColumn(no);
+          DBUG_PRINT("info", ("rec_attr: 0x%x  "
+                              "switching column impl 0x%x -> 0x%x",
+                              p, p->m_column, tAttrInfo));
+          p->m_column = tAttrInfo;
+          p = p->next();
+        }
       }
+      if (tmp_table_impl) 
+        delete tmp_table_impl;
     }
-    if (tmp_table_impl) 
-      delete tmp_table_impl;
-  }
-
-  if (unlikely(operation >= NdbDictionary::Event::_TE_FIRST_NON_DATA_EVENT))
-  {
-    DBUG_RETURN_EVENT(1);
+    DBUG_RETURN(1);
   }
 
+  DBUG_ENTER_EVENT("NdbEventOperationImpl::receive_event");
+  DBUG_PRINT_EVENT("info",("sdata->operation %u  this: %p", operation, this));
   // now move the data into the RecAttrs
     
   int is_update= operation == NdbDictionary::Event::_TE_UPDATE;
@@ -1089,6 +1095,32 @@
   return ret;
 }
 
+int
+NdbEventBuffer::flushIncompleteEvents(Uint64 gci)
+{
+  /**
+   *  Find min complete gci
+   */
+  Uint32 i;
+  Uint32 sz= m_active_gci.size();
+  Gci_container* array = (Gci_container*)m_active_gci.getBase();
+  for(i = 0; i < sz; i++)
+  {
+    Gci_container* tmp = array + i;
+    if (tmp->m_gci < gci)
+    {
+      // we have found an old not-completed gci, remove it
+      if(!tmp->m_data.is_empty())
+      {
+        free_list(tmp->m_data);
+      }
+      tmp->~Gci_container();
+      bzero(tmp, sizeof(Gci_container));
+    }
+  }
+  return 0;
+}
+
 NdbEventOperation *
 NdbEventBuffer::nextEvent()
 {
@@ -1157,7 +1189,10 @@
         }
         EventBufData_list::Gci_ops *gci_ops = m_available_data.first_gci_ops();
         while (gci_ops && op->getGCI() > gci_ops->m_gci)
+        {
+          deleteUsedEventOperations();
           gci_ops = m_available_data.next_gci_ops();
+        }
         assert(gci_ops && (op->getGCI() == gci_ops->m_gci));
 	DBUG_RETURN_EVENT(op->m_facade);
       }
@@ -1177,7 +1212,10 @@
   // free all "per gci unique" collected operations
   EventBufData_list::Gci_ops *gci_ops = m_available_data.first_gci_ops();
   while (gci_ops)
+  {
+    deleteUsedEventOperations();
     gci_ops = m_available_data.next_gci_ops();
+  }
   DBUG_RETURN_EVENT(0);
 }
 
@@ -1191,31 +1229,38 @@
     EventBufData_list::Gci_op g = gci_ops->m_gci_op_list[(*iter)++];
     if (event_types != NULL)
       *event_types = g.event_types;
-    DBUG_PRINT("info", ("gci: %d", (unsigned)gci_ops->m_gci));
+    DBUG_PRINT("info", ("gci: %d  g.op: %x  g.event_types: %x",
+                        (unsigned)gci_ops->m_gci, g.op, g.event_types));
     DBUG_RETURN(g.op);
   }
   DBUG_RETURN(NULL);
 }
 
 void
-NdbEventBuffer::lock()
-{
-  NdbMutex_Lock(m_mutex);
-}
-void
-NdbEventBuffer::unlock()
-{
-  NdbMutex_Unlock(m_mutex);
-}
-void
-NdbEventBuffer::add_drop_lock()
-{
-  NdbMutex_Lock(p_add_drop_mutex);
-}
-void
-NdbEventBuffer::add_drop_unlock()
+NdbEventBuffer::deleteUsedEventOperations()
 {
-  NdbMutex_Unlock(p_add_drop_mutex);
+  Uint32 iter= 0;
+  const NdbEventOperation *op_f;
+  while ((op_f= getGCIEventOperations(&iter, NULL)) != NULL)
+  {
+    NdbEventOperationImpl *op = &op_f->m_impl;
+    DBUG_ASSERT(op->m_ref_count > 0);
+    op->m_ref_count--;
+    DBUG_PRINT("info", ("m_ref_count: %u for op: %p", op->m_ref_count, op));
+    if (op->m_ref_count == 0)
+    {
+      DBUG_PRINT("info", ("deleting op: %p", op));
+      DBUG_ASSERT(op->m_node_bit_mask.isclear());
+      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
+        m_dropped_ev_op = op->m_next;
+      ndbout_c("deleting NdbEventOperation %p", op->m_facade);
+      delete op->m_facade;
+    }
+  }
 }
 
 static
@@ -1469,6 +1514,10 @@
 void
 NdbEventBuffer::report_node_failure(Uint32 node_id)
 {
+  NdbEventOperation* op= m_ndb->getEventOperation(0);
+  if (op == 0)
+    return;
+
   DBUG_ENTER("NdbEventBuffer::report_node_failure");
   SubTableData data;
   LinearSectionPtr ptr[3];
@@ -1484,12 +1533,20 @@
   /**
    * Insert this event for each operation
    */
-  NdbEventOperation* op= 0;
-  while((op = m_ndb->getEventOperation(op)))
   {
-    NdbEventOperationImpl* impl= &op->m_impl;
-    data.senderData = impl->m_oid;
-    insertDataL(impl, &data, ptr); 
+    // no need to lock()/unlock(), receive thread calls this
+    NdbEventOperationImpl* impl = &op->m_impl;
+    do if (!impl->m_node_bit_mask.isclear())
+    {
+      data.senderData = impl->m_oid;
+      insertDataL(impl, &data, ptr);
+    } while((impl = impl->m_next));
+    for (impl = m_dropped_ev_op; impl; impl = impl->m_next)
+      if (!impl->m_node_bit_mask.isclear())
+      {
+        data.senderData = impl->m_oid;
+        insertDataL(impl, &data, ptr);
+      }
   }
   DBUG_VOID_RETURN;
 }
@@ -1515,12 +1572,21 @@
   /**
    * Insert this event for each operation
    */
-  do
   {
-    NdbEventOperationImpl* impl= &op->m_impl;
-    data.senderData = impl->m_oid;
-    insertDataL(impl, &data, ptr);
-  } while((op = m_ndb->getEventOperation(op)));
+    // no need to lock()/unlock(), receive thread calls this
+    NdbEventOperationImpl* impl = &op->m_impl;
+    do if (!impl->m_node_bit_mask.isclear())
+    {
+      data.senderData = impl->m_oid;
+      insertDataL(impl, &data, ptr);
+    } while((impl = impl->m_next));
+    for (impl = m_dropped_ev_op; impl; impl = impl->m_next)
+      if (!impl->m_node_bit_mask.isclear())
+      {
+        data.senderData = impl->m_oid;
+        insertDataL(impl, &data, ptr);
+      }
+  }
   
   /**
    *  Release all GCI's with m_gci > gci
@@ -1565,7 +1631,11 @@
     }
   }
 
-  assert(bucket != 0);
+  if (bucket == 0)
+  {
+    // no bucket to complete
+    DBUG_VOID_RETURN;
+  }
 
   const Uint32 cnt= bucket->m_gcp_complete_rep_count = 1; 
   bucket->m_gci = gci;
@@ -1595,6 +1665,40 @@
 {
   DBUG_ENTER_EVENT("NdbEventBuffer::insertDataL");
   Uint64 gci= sdata->gci;
+  const bool is_data_event =
+    sdata->operation < NdbDictionary::Event::_TE_FIRST_NON_DATA_EVENT;
+
+  if (!is_data_event)
+  {
+    switch (sdata->operation)
+    {
+    case NdbDictionary::Event::_TE_NODE_FAILURE:
+      op->m_node_bit_mask.clear(sdata->ndbd_nodeid);
+      break;
+    case NdbDictionary::Event::_TE_ACTIVE:
+      op->m_node_bit_mask.set(sdata->ndbd_nodeid);
+      // internal event, do not relay to user
+      DBUG_RETURN_EVENT(0);
+      break;
+    case NdbDictionary::Event::_TE_CLUSTER_FAILURE:
+      op->m_node_bit_mask.clear();
+      DBUG_ASSERT(op->m_ref_count > 0);
+      op->m_ref_count--;
+      DBUG_PRINT("info", ("m_ref_count: %u for op: %p", op->m_ref_count, op));
+      break;
+    case NdbDictionary::Event::_TE_STOP:
+      op->m_node_bit_mask.clear(sdata->ndbd_nodeid);
+      if (op->m_node_bit_mask.isclear())
+      {
+        DBUG_ASSERT(op->m_ref_count > 0);
+        op->m_ref_count--;
+        DBUG_PRINT("info", ("m_ref_count: %u for op: %p", op->m_ref_count, op));
+      }
+      break;
+    default:
+      break;
+    }
+  }
 
   if ( likely((Uint32)op->mi_type & (1 << (Uint32)sdata->operation)) )
   {
@@ -1615,8 +1719,6 @@
     }
     
     const bool is_blob_event = (op->theMainOp != NULL);
-    const bool is_data_event =
-      sdata->operation < NdbDictionary::Event::_TE_FIRST_NON_DATA_EVENT;
     const bool use_hash =  op->m_mergeEvents && is_data_event;
 
     if (! is_data_event && is_blob_event)
@@ -2244,6 +2346,8 @@
 void
 EventBufData_list::add_gci_op(Gci_op g, bool del)
 {
+  DBUG_ENTER_EVENT("EventBufData_list::add_gci_op");
+  DBUG_PRINT_EVENT("info", ("p.op: %p  g.event_types: %x", g.op, g.event_types));
   assert(g.op != NULL);
   Uint32 i;
   for (i = 0; i < m_gci_op_count; i++) {
@@ -2273,8 +2377,15 @@
     }
     assert(m_gci_op_count < m_gci_op_alloc);
     assert(! del);
+#ifndef DBUG_OFF
+    i = m_gci_op_count;
+#endif
+    g.op->m_ref_count++;
+    DBUG_PRINT("info", ("m_ref_count: %u for op: %p", g.op->m_ref_count, g.op));
     m_gci_op_list[m_gci_op_count++] = g;
   }
+  DBUG_PRINT_EVENT("exit", ("m_gci_op_list[%u].event_types: %x", i,
m_gci_op_list[i].event_types));
+  DBUG_VOID_RETURN_EVENT;
 }
 
 void
@@ -2337,6 +2448,9 @@
     delete tOp;
     DBUG_RETURN(NULL);
   }
+  getEventOperationImpl(tOp)->m_ref_count = 1;
+  DBUG_PRINT("info", ("m_ref_count: %u for op: %p",
+                      getEventOperationImpl(tOp)->m_ref_count,
getEventOperationImpl(tOp)));
   DBUG_RETURN(tOp);
 }
 
@@ -2362,16 +2476,10 @@
 void
 NdbEventBuffer::dropEventOperation(NdbEventOperation* tOp)
 {
+  DBUG_ENTER("NdbEventBuffer::dropEventOperation");
   NdbEventOperationImpl* op= getEventOperationImpl(tOp);
 
   op->stop();
-
-  op->m_next= m_dropped_ev_op;
-  op->m_prev= 0;
-  if (m_dropped_ev_op)
-    m_dropped_ev_op->m_prev= op;
-  m_dropped_ev_op= op;
- 
   // stop blob event ops
   if (op->theMainOp == NULL)
   {
@@ -2391,11 +2499,25 @@
     }
   }
 
-  // ToDo, take care of these to be deleted at the
-  // appropriate time, after we are sure that there
-  // are _no_ more events coming
-
-  //  delete tOp;
+  DBUG_ASSERT(op->m_ref_count > 0);
+  op->m_ref_count--;
+  DBUG_PRINT("info", ("m_ref_count: %u for op: %p", op->m_ref_count, op));
+  if (op->m_ref_count == 0)
+  {
+    DBUG_PRINT("info", ("deleting op: %p", op));
+    DBUG_ASSERT(op->m_node_bit_mask.isclear());
+    ndbout_c("deleting NdbEventOperation %p", op->m_facade);
+    delete op->m_facade;
+  }
+  else
+  {
+    op->m_next= m_dropped_ev_op;
+    op->m_prev= 0;
+    if (m_dropped_ev_op)
+      m_dropped_ev_op->m_prev= op;
+    m_dropped_ev_op= op;
+  }
+  DBUG_VOID_RETURN;
 }
 
 void

--- 1.21/storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp	2006-05-09 21:08:01 +02:00
+++ 1.22/storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp	2006-05-18 23:37:53 +02:00
@@ -25,7 +25,7 @@
 #include <UtilBuffer.hpp>
 
 #define NDB_EVENT_OP_MAGIC_NUMBER 0xA9F301B4
-//#define EVENT_DEBUG
+#define EVENT_DEBUG
 #ifdef EVENT_DEBUG
 #define DBUG_ENTER_EVENT(A) DBUG_ENTER(A)
 #define DBUG_RETURN_EVENT(A) DBUG_RETURN(A)
@@ -367,6 +367,8 @@
   Uint32 m_eventId;
   Uint32 m_oid;
 
+  Bitmask<(unsigned int)_NDB_NODE_BITMASK_SIZE> m_node_bit_mask;
+  int m_ref_count;
   bool m_mergeEvents;
   
   EventBufData *m_data_item;
@@ -406,10 +408,10 @@
   void dropEventOperation(NdbEventOperation *);
   static NdbEventOperationImpl* getEventOperationImpl(NdbEventOperation* tOp);
 
-  void add_drop_lock();
-  void add_drop_unlock();
-  void lock();
-  void unlock();
+  void add_drop_lock() { NdbMutex_Lock(p_add_drop_mutex); }
+  void add_drop_unlock() { NdbMutex_Unlock(p_add_drop_mutex); }
+  void lock() { NdbMutex_Lock(m_mutex); }
+  void unlock() { NdbMutex_Unlock(m_mutex); }
 
   void add_op();
   void remove_op();
@@ -430,9 +432,11 @@
   Uint32 getEventId(int bufferId);
 
   int pollEvents(int aMillisecondNumber, Uint64 *latestGCI= 0);
+  int flushIncompleteEvents(Uint64 gci);
   NdbEventOperation *nextEvent();
   NdbEventOperationImpl* getGCIEventOperations(Uint32* iter,
                                                Uint32* event_types);
+  void NdbEventBuffer::deleteUsedEventOperations();
 
   NdbEventOperationImpl *move_data();
 

--- 1.4/mysql-test/r/ndb_autodiscover3.result	2006-05-17 08:34:41 +02:00
+++ 1.5/mysql-test/r/ndb_autodiscover3.result	2006-05-18 23:37:53 +02:00
@@ -18,6 +18,7 @@
 ERROR 42S02: Table 'test.t2' doesn't exist
 show tables like 't2';
 Tables_in_test (t2)
+reset master;
 create table t2 (a int key) engine=ndbcluster;
 insert into t2 values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
 select * from t2 order by a limit 3;
@@ -30,10 +31,12 @@
 1
 2
 3
+reset master;
 select * from t2;
 ERROR 42S02: Table 'test.t2' doesn't exist
 show tables like 't2';
 Tables_in_test (t2)
+reset master;
 create table t2 (a int key) engine=ndbcluster;
 insert into t2 values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
 select * from t2 order by a limit 3;
@@ -46,4 +49,5 @@
 1
 2
 3
+reset master;
 drop table t2;

--- 1.3/mysql-test/t/ndb_autodiscover3.test	2006-05-17 08:34:41 +02:00
+++ 1.4/mysql-test/t/ndb_autodiscover3.test	2006-05-18 23:37:53 +02:00
@@ -43,6 +43,7 @@
 --error ER_NO_SUCH_TABLE
 select * from t2;
 show tables like 't2';
+reset master;
 create table t2 (a int key) engine=ndbcluster;
 insert into t2 values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
 select * from t2 order by a limit 3;
@@ -50,6 +51,7 @@
 # server 1 should have a stale cache, and in this case wrong frm, transaction must be
retried
 --connection server1
 select * from t2 order by a limit 3;
+reset master;
 
 --exec $NDB_MGM --no-defaults -e "all restart -i" >> $NDB_TOOLS_OUTPUT
 --exec $NDB_TOOLS_DIR/ndb_waiter --no-defaults >> $NDB_TOOLS_OUTPUT
@@ -60,6 +62,7 @@
 --error ER_NO_SUCH_TABLE
 select * from t2;
 show tables like 't2';
+reset master;
 create table t2 (a int key) engine=ndbcluster;
 insert into t2 values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
 select * from t2 order by a limit 3;
@@ -67,6 +70,7 @@
 # server 2 should have a stale cache, but with right frm, transaction need not be retried
 --connection server2
 select * from t2 order by a limit 3;
+reset master;
 
 drop table t2;
 # End of 4.1 tests

--- 1.54/sql/ha_ndbcluster_binlog.cc	2006-05-17 08:34:41 +02:00
+++ 1.55/sql/ha_ndbcluster_binlog.cc	2006-05-18 23:37:53 +02:00
@@ -3321,6 +3321,9 @@
     DBUG_PRINT("info", ("schema_res: %d  schema_gci: %d", schema_res, schema_gci));
     if (schema_res > 0)
     {
+      i_ndb->pollEvents(0);
+      i_ndb->flushIncompleteEvents(schema_gci);
+      s_ndb->flushIncompleteEvents(schema_gci);
       if (schema_gci < ndb_latest_handled_binlog_epoch)
       {
         sql_print_error("NDB Binlog: cluster has been restarted --initial or with older
filesystem. "