extra logging and asserts to try to track down the spurious completeCluster failure assetion

    corrected error in invalidation of dict cache which may be causing the valgrind warning at times


sql/ha_ndbcluster_binlog.cc:
  extra logging and asserts to try to track down the spurious completeCluster failure assetion
storage/ndb/src/ndbapi/DictCache.cpp:
  corrected error in invalidation of dict cache which may be causing the valgrind warning at times
storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp:
  extra logging and asserts to try to track down the spurious completeCluster failure assetion
storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp:
  extra logging and asserts to try to track down the spurious completeCluster failure assetion

sql/ha_ndbcluster_binlog.cc

@@ -3466,6 +3466,11 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
         ndb_latest_applied_binlog_epoch= 0;
         ndb_latest_received_binlog_epoch= 0;
       }
+      if (ndb_extra_logging)
+      {
+        sql_print_information("NDB Binlog: starting log at epoch %u",
+                              (unsigned)schema_gci);
+      }
     }
   }
   {

storage/ndb/src/ndbapi/DictCache.cpp

@@ -312,12 +312,15 @@ GlobalDictCache::invalidate_all()
     if (vers->size())
     {
       TableVersion * ver = & vers->back();
-      ver->m_impl->m_status = NdbDictionary::Object::Invalid;
-      ver->m_status = DROPPED;
-      if (ver->m_refCount == 0)
+      if (ver->m_status != RETREIVING)
       {
-        delete ver->m_impl;
-        vers->erase(vers->size() - 1);
+        ver->m_impl->m_status = NdbDictionary::Object::Invalid;
+        ver->m_status = DROPPED;
+        if (ver->m_refCount == 0)
+        {
+          delete ver->m_impl;
+          vers->erase(vers->size() - 1);
+        }
       }
     }
     curr = m_tableHash.getNext(curr);

storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp

@@ -942,6 +942,7 @@ NdbEventBuffer::NdbEventBuffer(Ndb *ndb) :
 {
 #ifdef VM_TRACE
   m_latest_command= "NdbEventBuffer::NdbEventBuffer";
+  m_flush_gci = 0;
 #endif
 
   if ((p_cond = NdbCondition_Create()) ==  NULL) {
@@ -1109,6 +1110,8 @@ NdbEventBuffer::flushIncompleteEvents(Uint64 gci)
   /**
    *  Find min complete gci
    */
+  // called by user thread, so we need to lock the data
+  lock();
   Uint32 i;
   Uint32 sz= m_active_gci.size();
   Gci_container* array = (Gci_container*)m_active_gci.getBase();
@@ -1127,6 +1130,10 @@ NdbEventBuffer::flushIncompleteEvents(Uint64 gci)
       bzero(tmp, sizeof(Gci_container));
     }
   }
+#ifdef VM_TRACE
+  m_flush_gci = gci;
+#endif
+  unlock();
   return 0;
 }
 
@@ -1301,7 +1308,11 @@ operator<<(NdbOut& out, const Gci_container_pod& gci)
 
 static
 Gci_container*
-find_bucket_chained(Vector<Gci_container_pod> * active, Uint64 gci)
+find_bucket_chained(Vector<Gci_container_pod> * active, Uint64 gci
+#ifdef VM_TRACE
+                    ,Uint64 flush_gci
+#endif
+                    )
 {
   Uint32 pos = (gci & ACTIVE_GCI_MASK);
   Gci_container *bucket= ((Gci_container*)active->getBase()) + pos;
@@ -1322,6 +1333,13 @@ find_bucket_chained(Vector<Gci_container_pod> * active, Uint64 gci)
 	bzero(bucket, sizeof(Gci_container));
 	bucket->m_gci = gci;
 	bucket->m_gcp_complete_rep_count = ~(Uint32)0;
+#ifdef VM_TRACE
+        if (gci < flush_gci)
+        {
+          ndbout_c("received old gci %llu < flush gci %llu", gci, flush_gci);
+          assert(false);
+        }
+#endif
 	return bucket;
       }
       move_pos += ACTIVE_GCI_DIRECTORY_SIZE;
@@ -1336,7 +1354,16 @@ find_bucket_chained(Vector<Gci_container_pod> * active, Uint64 gci)
       bucket += ACTIVE_GCI_DIRECTORY_SIZE;
       
       if(bucket->m_gci == gci)
+      {
+#ifdef VM_TRACE
+        if (gci < flush_gci)
+        {
+          ndbout_c("received old gci %llu < flush gci %llu", gci, flush_gci);
+          assert(false);
+        }
+#endif
 	return bucket;
+      }
       
     } while(pos < size);
     
@@ -1346,14 +1373,22 @@ find_bucket_chained(Vector<Gci_container_pod> * active, Uint64 gci)
 
 inline
 Gci_container*
-find_bucket(Vector<Gci_container_pod> * active, Uint64 gci)
+find_bucket(Vector<Gci_container_pod> * active, Uint64 gci
+#ifdef VM_TRACE
+            ,Uint64 flush_gci
+#endif
+            )
 {
   Uint32 pos = (gci & ACTIVE_GCI_MASK);
   Gci_container *bucket= ((Gci_container*)active->getBase()) + pos;
   if(likely(gci == bucket->m_gci))
     return bucket;
 
-  return find_bucket_chained(active,gci);
+  return find_bucket_chained(active,gci
+#ifdef VM_TRACE
+                             , flush_gci
+#endif
+                             );
 }
 
 static
@@ -1386,7 +1421,11 @@ NdbEventBuffer::execSUB_GCP_COMPLETE_REP(const SubGcpCompleteRep * const rep)
   const Uint64 gci= rep->gci;
   const Uint32 cnt= rep->gcp_complete_rep_count;
 
-  Gci_container *bucket = find_bucket(&m_active_gci, gci);
+  Gci_container *bucket = find_bucket(&m_active_gci, gci
+#ifdef VM_TRACE
+                                      , m_flush_gci
+#endif
+                                      );
 
   if (unlikely(bucket == 0))
   {
@@ -1752,7 +1791,11 @@ NdbEventBuffer::insertDataL(NdbEventOperationImpl *op,
 
   if ( likely((Uint32)op->mi_type & (1 << (Uint32)sdata->operation)) )
   {
-    Gci_container* bucket= find_bucket(&m_active_gci, gci);
+    Gci_container* bucket= find_bucket(&m_active_gci, gci
+#ifdef VM_TRACE
+                                       , m_flush_gci
+#endif
+                                       );
       
     DBUG_PRINT_EVENT("info", ("data insertion in eventId %d", op->m_eventId));
     DBUG_PRINT_EVENT("info", ("gci=%d tab=%d op=%d node=%d",

storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp

@@ -463,6 +463,7 @@ public:
 
 #ifdef VM_TRACE
   const char *m_latest_command;
+  Uint64 m_flush_gci;
 #endif
 
   Ndb *m_ndb;