bzr commit into mysql-5.1 branch (DonKehndon.kehn:3102) Bug#33626

[DonKehndon.kehn]

#At file:///u02/devl/bzr-tree/mysql-5.1-telco-6.4/ based on
revid:jonas@stripped

 3102 Don Kehn	2008-11-17 [merge]
      [BUG #33626] added all the changes for the ndb recycle code in merge from 6.3
added:
  mysql-test/suite/ndb/r/ndb_reconnect.result
  mysql-test/suite/ndb/t/ndb_reconnect.test
modified:
  sql/ha_ndbcluster.cc
  sql/ha_ndbcluster.h
  sql/ha_ndbcluster_binlog.cc
  sql/ha_ndbcluster_binlog.h
  storage/ndb/src/ndbapi/NdbScanOperation.cpp

=== added file 'mysql-test/suite/ndb/r/ndb_reconnect.result'
--- a/mysql-test/suite/ndb/r/ndb_reconnect.result	1970-01-01 00:00:00 +0000
+++ b/mysql-test/suite/ndb/r/ndb_reconnect.result	2008-11-14 19:29:36 +0000
@@ -0,0 +1,27 @@
+CREATE TABLE t1(a int primary key, b varchar(255), c int) engine=ndb;
+select * from t1;
+a	b	c
+insert into t1 values (1, "row 1", 2);
+select * from t1;
+a	b	c
+1	row 1	2
+select * from t1;
+a	b	c
+1	row 1	2
+select * from t1;
+a	b	c
+1	row 1	2
+select * from t1;
+ERROR HY000: Got error 157 'Unknown error code' from NDBCLUSTER
+select * from t1;
+ERROR HY000: Got error 157 'Unknown error code' from NDBCLUSTER
+select a,b,c from t1;
+a	b	c
+1	row 1	2
+select * from t1;
+a	b	c
+1	row 1	2
+select * from t1;
+a	b	c
+1	row 1	2
+insert into t1 values (2, "row 1", 37);

=== added file 'mysql-test/suite/ndb/t/ndb_reconnect.test'
--- a/mysql-test/suite/ndb/t/ndb_reconnect.test	1970-01-01 00:00:00 +0000
+++ b/mysql-test/suite/ndb/t/ndb_reconnect.test	2008-11-14 19:29:36 +0000
@@ -0,0 +1,88 @@
+--source include/have_ndb.inc
+
+CREATE TABLE t1(a int primary key, b varchar(255), c int) engine=ndb;
+
+select * from t1;
+insert into t1 values (1, "row 1", 2);
+
+connect(con1, localhost, root,,);
+connect(con2, localhost, root,,);
+connect(con3, localhost, root,,);
+
+connection con1;
+select * from t1;
+connection con2;
+select * from t1;
+connection con3;
+select * from t1;
+
+# Restart cluster nodes "nostart"
+--exec $NDB_MGM --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT" -e "all
restart -n" >> $NDB_TOOLS_OUTPUT
+# Wait for all nodes to enter "nostart"
+--exec $NDB_TOOLS_DIR/ndb_waiter --no-defaults
--ndb-connectstring="localhost:$NDBCLUSTER_PORT" --not-started >> $NDB_TOOLS_OUTPUT
+
+connection con1;
+--error 1296
+select * from t1;
+
+connection con2;
+--error 1296
+select * from t1;
+
+# Don't do anything with the third connection
+#connection con3;
+
+
+# Start cluster nodes again
+--exec $NDB_MGM --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT" -e "all
start" >> $NDB_TOOLS_OUTPUT
+# Wait for all nodes to enter "started"
+--exec $NDB_TOOLS_DIR/ndb_waiter --no-defaults
--ndb-connectstring="localhost:$NDBCLUSTER_PORT" >> $NDB_TOOLS_OUTPUT
+
+
+#
+# Wait until the connection to the
+# cluster has been restored or timeout occurs
+#
+connection default;
+--disable_result_log
+--disable_query_log
+--disable_abort_on_error
+let $counter= 500;
+let $mysql_errno=37;
+while ($mysql_errno)
+{
+  select * from t1;
+
+  dec $counter;
+  if (!$counter)
+  {
+    --die Server failed to reconnect to cluster
+  }
+  --sleep 0.1
+}
+--enable_abort_on_error
+--enable_query_log
+--enable_result_log
+
+# Run selects to show that the cluster are back
+
+connection con1;
+select a,b,c from t1;
+
+connection con2;
+select * from t1;
+
+connection con3;
+select * from t1;
+
+#
+# Wait until mysqld has connected properly to cluster
+#
+--disable_result_log
+--disable_query_log
+source include/ndb_not_readonly.inc;
+--enable_query_log
+--enable_result_log
+
+# Do an insert to see table is writable
+insert into t1 values (2, "row 1", 37);

=== modified file 'sql/ha_ndbcluster.cc'
--- a/sql/ha_ndbcluster.cc	2008-11-14 11:01:37 +0000
+++ b/sql/ha_ndbcluster.cc	2008-11-17 19:03:20 +0000
@@ -618,6 +618,7 @@ uchar *thd_ndb_share_get_key(THD_NDB_SHA
 Thd_ndb::Thd_ndb()
 {
   connection= ndb_get_cluster_connection();
+  m_connect_count= connection->get_connect_count();
   ndb= new Ndb(connection, "");
   lock_count= 0;
   start_stmt_count= 0;
@@ -8203,6 +8204,52 @@ void ha_ndbcluster::release_thd_ndb(Thd_
 }
 
 
+bool Thd_ndb::recycle_ndb(THD* thd)
+{
+  DBUG_ENTER("recycle_ndb");
+  DBUG_PRINT("enter", ("ndb: 0x%lx", (long)ndb));
+
+  DBUG_ASSERT(global_schema_lock_trans == NULL);
+
+  delete ndb;
+  if ((ndb= new Ndb(connection, "")) == NULL)
+  {
+    DBUG_PRINT("error",("failed to allocate Ndb object"));
+    DBUG_RETURN(false);
+  }
+
+  if (ndb->init(max_transactions) != 0)
+  {
+    delete ndb;
+    ndb= NULL;
+    DBUG_PRINT("error", ("Ndb::init failed, %d  message: %s",
+                         ndb->getNdbError().code,
+                         ndb->getNdbError().message));
+    DBUG_RETURN(false);
+  }
+
+  changed_tables.empty();
+  trans= NULL;
+
+  DBUG_RETURN(true);
+}
+
+
+bool
+Thd_ndb::valid_ndb(void)
+{
+  // The ndb object should be valid as long as a
+  // global schema lock transaction is ongoing
+  if (global_schema_lock_trans)
+    return true;
+
+  if (unlikely(m_connect_count != connection->get_connect_count()))
+    return false;
+  return true;
+}
+
+
+
 /**
   If this thread already has a Thd_ndb object allocated
   in current THD, reuse it. Otherwise
@@ -8210,7 +8257,7 @@ void ha_ndbcluster::release_thd_ndb(Thd_
  
 */
 
-Ndb* check_ndb_in_thd(THD* thd)
+Ndb* check_ndb_in_thd(THD* thd, bool validate_ndb)
 {
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
   if (!thd_ndb)
@@ -8219,6 +8266,11 @@ Ndb* check_ndb_in_thd(THD* thd)
       return NULL;
     set_thd_ndb(thd, thd_ndb);
   }
+  else if (validate_ndb && !thd_ndb->valid_ndb())
+  {
+    if (!thd_ndb->recycle_ndb(thd))
+      return NULL;
+  }
   return thd_ndb->ndb;
 }
 
@@ -8229,7 +8281,7 @@ int ha_ndbcluster::check_ndb_connection(
   Ndb *ndb;
   DBUG_ENTER("check_ndb_connection");
   
-  if (!(ndb= check_ndb_in_thd(thd)))
+  if (!(ndb= check_ndb_in_thd(thd, true)))
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
   if (ndb->setDatabaseName(m_dbname))
   {
@@ -11144,6 +11196,11 @@ pthread_handler_t ndb_util_thread_func(v
       have been created.
       If not try to create it
     */
+		if (!check_ndb_in_thd(thd, false))
+    {
+      set_timespec(abstime, 1);
+      continue;
+    }
     if (!ndb_binlog_tables_inited)
       ndbcluster_setup_binlog_table_shares(thd);
 

=== modified file 'sql/ha_ndbcluster.h'
--- a/sql/ha_ndbcluster.h	2008-11-13 16:32:23 +0000
+++ b/sql/ha_ndbcluster.h	2008-11-17 19:03:20 +0000
@@ -335,6 +335,10 @@ class Thd_ndb 
   NdbTransaction *global_schema_lock_trans;
   uint global_schema_lock_count;
   uint global_schema_lock_error;
+
+  unsigned m_connect_count;
+  bool valid_ndb(void);
+  bool recycle_ndb(THD* thd);
 };
 
 int ndbcluster_commit(handlerton *hton, THD *thd, bool all);

=== modified file 'sql/ha_ndbcluster_binlog.cc'
--- a/sql/ha_ndbcluster_binlog.cc	2008-10-30 13:54:40 +0000
+++ b/sql/ha_ndbcluster_binlog.cc	2008-11-17 19:03:20 +0000
@@ -5378,6 +5378,42 @@ static void ndb_free_schema_object(NDB_S
 }
 
 
+static void
+remove_event_operations(Ndb* ndb)
+{
+  DBUG_ENTER("remove_event_operations");
+  NdbEventOperation *op;
+  while ((op= ndb->getEventOperation()))
+  {
+    DBUG_ASSERT(!IS_NDB_BLOB_PREFIX(op->getEvent()->getTable()->getName()));
+    DBUG_PRINT("info", ("removing event operation on %s",
+                        op->getEvent()->getName()));
+
+    Ndb_event_data *event_data= (Ndb_event_data *) op->getCustomData();
+    DBUG_ASSERT(event_data);
+
+    NDB_SHARE *share= event_data->share;
+    DBUG_ASSERT(share != NULL);
+    DBUG_ASSERT(share->op == op || share->new_op == op);
+
+    delete event_data;
+    op->setCustomData(NULL);
+
+    (void) pthread_mutex_lock(&share->mutex);
+    share->op= 0;
+    share->new_op= 0;
+    (void) pthread_mutex_unlock(&share->mutex);
+
+    DBUG_PRINT("NDB_SHARE", ("%s binlog free  use_count: %u",
+                             share->key, share->use_count));
+    free_share(&share);
+
+    ndb->dropEventOperation(op);
+  }
+  DBUG_VOID_RETURN;
+}
+
+
 pthread_handler_t ndb_binlog_thread_func(void *arg)
 {
   THD *thd; /* needs to be first for thread_stack */
@@ -5445,6 +5481,7 @@ pthread_handler_t ndb_binlog_thread_func
   pthread_mutex_unlock(&LOCK_thread_count);
   thd->lex->start_transaction_opt= 0;
 
+restart_cluster_failure:
   if (!(s_ndb= new Ndb(g_ndb_cluster_connection, "")) ||
       s_ndb->init())
   {
@@ -5757,6 +5794,32 @@ restart:
                                                 &post_epoch_log_list,
                                                 &post_epoch_unlock_list,
                                                 &mem_root);
+
+          if (unlikely(pOp->getEventType() == NDBEVENT::TE_CLUSTER_FAILURE))
+          {
+            sql_print_information("NDB Binlog: cluster failure detected");
+            pthread_mutex_lock(&LOCK_open);
+
+            pthread_mutex_lock(&injector_mutex);
+
+            remove_event_operations(s_ndb);
+            delete s_ndb;
+            s_ndb= 0;
+            schema_ndb= 0;
+
+            remove_event_operations(i_ndb);
+            delete i_ndb;
+            i_ndb= 0;
+            injector_ndb= 0;
+
+            hash_free(&ndb_schema_objects);
+
+            pthread_mutex_unlock(&LOCK_open);
+
+            sql_print_information("NDB Binlog: restarting");
+            goto restart_cluster_failure;
+          }
+
           DBUG_PRINT("info", ("s_ndb first: %s", s_ndb->getEventOperation() ?
                              
s_ndb->getEventOperation()->getEvent()->getTable()->getName() :
                               "<empty>"));
@@ -6147,63 +6210,13 @@ err:
   /* remove all event operations */
   if (s_ndb)
   {
-    NdbEventOperation *op;
-    DBUG_PRINT("info",("removing all event operations"));
-    while ((op= s_ndb->getEventOperation()))
-    {
-      DBUG_ASSERT(! IS_NDB_BLOB_PREFIX(op->getEvent()->getTable()->getName()));
-      DBUG_PRINT("info",("removing event operation on %s",
-                         op->getEvent()->getName()));
-      Ndb_event_data *event_data= (Ndb_event_data *) op->getCustomData();
-      NDB_SHARE *share= (event_data)?event_data->share:NULL;
-      DBUG_ASSERT(share != 0);
-      DBUG_ASSERT(share->op == op || share->new_op == op);
-      if (event_data)
-      {
-        delete event_data;
-        op->setCustomData(NULL);
-      }
-      (void) pthread_mutex_lock(&share->mutex);
-      share->op= 0;
-      share->new_op= 0;
-      (void) pthread_mutex_unlock(&share->mutex);
-      /* ndb_share reference binlog free */
-      DBUG_PRINT("NDB_SHARE", ("%s binlog free  use_count: %u",
-                               share->key, share->use_count));
-      free_share(&share);
-      s_ndb->dropEventOperation(op);
-    }
+    remove_event_operations(s_ndb);
     delete s_ndb;
     s_ndb= 0;
   }
   if (i_ndb)
   {
-    NdbEventOperation *op;
-    DBUG_PRINT("info",("removing all event operations"));
-    while ((op= i_ndb->getEventOperation()))
-    {
-      DBUG_ASSERT(! IS_NDB_BLOB_PREFIX(op->getEvent()->getTable()->getName()));
-      DBUG_PRINT("info",("removing event operation on %s",
-                         op->getEvent()->getName()));
-      Ndb_event_data *event_data= (Ndb_event_data *) op->getCustomData();
-      NDB_SHARE *share= (event_data)?event_data->share:NULL;
-      if (event_data)
-      {
-        delete event_data;
-        op->setCustomData(NULL);
-      }
-      DBUG_ASSERT(share != 0);
-      (void) pthread_mutex_lock(&share->mutex);
-      DBUG_ASSERT(share->op == op || share->new_op == op);
-      share->op= 0;
-      share->new_op= 0;
-      (void) pthread_mutex_unlock(&share->mutex);
-      /* ndb_share reference binlog free */
-      DBUG_PRINT("NDB_SHARE", ("%s binlog free  use_count: %u",
-                               share->key, share->use_count));
-      free_share(&share);
-      i_ndb->dropEventOperation(op);
-    }
+    remove_event_operations(i_ndb);
     delete i_ndb;
     i_ndb= 0;
   }

=== modified file 'sql/ha_ndbcluster_binlog.h'
--- a/sql/ha_ndbcluster_binlog.h	2008-10-30 13:27:52 +0000
+++ b/sql/ha_ndbcluster_binlog.h	2008-11-17 19:03:20 +0000
@@ -315,7 +315,7 @@ void
 set_thd_ndb(THD *thd, Thd_ndb *thd_ndb)
 { thd_set_ha_data(thd, ndbcluster_hton, thd_ndb); }
 
-Ndb* check_ndb_in_thd(THD* thd);
+Ndb* check_ndb_in_thd(THD* thd, bool validate_ndb= false);
 
 int ndbcluster_has_global_schema_lock(Thd_ndb *thd_ndb);
 int ndbcluster_no_global_schema_lock_abort(THD *thd, const char *msg);

=== modified file 'storage/ndb/src/ndbapi/NdbScanOperation.cpp'
--- a/storage/ndb/src/ndbapi/NdbScanOperation.cpp	2008-11-08 21:43:03 +0000
+++ b/storage/ndb/src/ndbapi/NdbScanOperation.cpp	2008-11-17 19:03:20 +0000
@@ -95,7 +95,10 @@ int
 NdbScanOperation::init(const NdbTableImpl* tab, NdbTransaction* myConnection)
 {
   m_transConnection = myConnection;
-  //NdbConnection* aScanConnection = theNdb->startTransaction(myConnection);
+
+  if(NdbOperation::init(tab, NULL, false) != 0)
+    return -1;
+
   theNdb->theRemainingStartTransactions++; // will be checked in hupp...
   NdbTransaction* aScanConnection = theNdb->hupp(myConnection);
   if (!aScanConnection){
@@ -105,11 +108,8 @@ NdbScanOperation::init(const NdbTableImp
   }
 
   // NOTE! The hupped trans becomes the owner of the operation
-  if(NdbOperation::init(tab, aScanConnection, false) != 0){
-    theNdb->theRemainingStartTransactions--;
-    return -1;
-  }
-  
+  theNdbCon= aScanConnection;
+
   initInterpreter();
   
   theStatus = GetValue;