Merge poseidon.mysql.com:/home/tomas/mysql-5.1-telco-cga

into  poseidon.mysql.com:/home/tomas/mysql-5.1-new-ndb

1 files changed, 100 insertions, 9 deletions

diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index fe8050fc57a..c959d512764 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -371,6 +371,10 @@ void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
       strcmp(share->db, NDB_REP_DB) == 0 &&
       strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
     do_event_op= 1;
+  else if (!ndb_apply_status_share &&
+           strcmp(share->db, NDB_REP_DB) == 0 &&
+           strcmp(share->table_name, NDB_APPLY_TABLE) == 0)
+    do_event_op= 1;
 
   {
     int i, no_nodes= g_ndb_cluster_connection->no_db_nodes();
@@ -606,7 +610,14 @@ static int ndbcluster_binlog_end(THD *thd)
         DBUG_PRINT("share",
                    ("table->s->db.table_name: %s.%s",
                     share->table->s->db.str, share->table->s->table_name.str));
-      if (share->state != NSS_DROPPED && !--share->use_count)
+      /* ndb_share reference create free */
+      if (share->state != NSS_DROPPED)
+      {
+        DBUG_PRINT("NDB_SHARE", ("%s create free  use_count: %u",
+                                 share->key, share->use_count));
+        --share->use_count;
+      }
+      if (share->use_count == 0)
         ndbcluster_real_free_share(&share);
       else
       {
@@ -1716,15 +1727,25 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
   (void) pthread_cond_signal(&injector_cond);
 
   pthread_mutex_lock(&ndbcluster_mutex);
+  /* ndb_share reference binlog free */
+  DBUG_PRINT("NDB_SHARE", ("%s binlog free  use_count: %u",
+                           share->key, share->use_count));
   free_share(&share, TRUE);
   if (is_remote_change && share && share->state != NSS_DROPPED)
   {
     DBUG_PRINT("info", ("remote change"));
     share->state= NSS_DROPPED;
     if (share->use_count != 1)
+    {
+      /* open handler holding reference */
+      /* wait with freeing create ndb_share to below */
       do_close_cached_tables= TRUE;
+    }
     else
     {
+      /* ndb_share reference create free */
+      DBUG_PRINT("NDB_SHARE", ("%s create free  use_count: %u",
+                               share->key, share->use_count));
       free_share(&share, TRUE);
       share= 0;
     }
@@ -1747,6 +1768,9 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
     table_list.db= (char *)dbname;
     table_list.alias= table_list.table_name= (char *)tabname;
     close_cached_tables(thd, 0, &table_list);
+    /* ndb_share reference create free */
+    DBUG_PRINT("NDB_SHARE", ("%s create free  use_count: %u",
+                             share->key, share->use_count));
     free_share(&share);
   }
   DBUG_RETURN(0);
@@ -1814,7 +1838,13 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           char key[FN_REFLEN];
           build_table_filename(key, sizeof(key),
                                schema->db, schema->name, "", 0);
+          /* ndb_share reference temporary, free below */
           NDB_SHARE *share= get_share(key, 0, FALSE, FALSE);
+          if (share)
+          {
+            DBUG_PRINT("NDB_SHARE", ("%s temporary  use_count: %u",
+                                     share->key, share->use_count));
+          }
           // invalidation already handled by binlog thread
           if (!share || !share->op)
           {
@@ -1830,8 +1860,13 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
             table_list.alias= table_list.table_name= schema->name;
             close_cached_tables(thd, 0, &table_list, FALSE);
           }
+          /* ndb_share reference temporary free */
           if (share)
+          {
+            DBUG_PRINT("NDB_SHARE", ("%s temporary free  use_count: %u",
+                                     share->key, share->use_count));
             free_share(&share);
+          }
         }
         // fall through
         case SOT_CREATE_TABLE:
@@ -1940,6 +1975,10 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
 
       /* begin protect ndb_schema_share */
       pthread_mutex_lock(&ndb_schema_share_mutex);
+      /* ndb_share reference binlog extra free */
+      DBUG_PRINT("NDB_SHARE", ("%s binlog extra free  use_count: %u",
+                               ndb_schema_share->key,
+                               ndb_schema_share->use_count));
       free_share(&ndb_schema_share);
       ndb_schema_share= 0;
       pthread_mutex_unlock(&ndb_schema_share_mutex);
@@ -2066,7 +2105,13 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
         pthread_mutex_unlock(&ndbcluster_mutex);
         continue;
       }
+      /* ndb_share reference temporary, free below */
       NDB_SHARE *share= get_share(key, 0, FALSE, FALSE);
+      if (share)
+      {
+        DBUG_PRINT("NDB_SHARE", ("%s temporary  use_count: %u",
+                                 share->key, share->use_count));
+      }
       switch (schema_type)
       {
       case SOT_DROP_DB:
@@ -2111,6 +2156,9 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
           */
           if (share)
           {
+            /* ndb_share reference temporary free */
+            DBUG_PRINT("NDB_SHARE", ("%s temporary free  use_count: %u",
+                                     share->key, share->use_count));
             free_share(&share);
             share= 0;
           }
@@ -2143,6 +2191,9 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
       }
       if (share)
       {
+        /* ndb_share reference temporary free */
+        DBUG_PRINT("NDB_SHARE", ("%s temporary free  use_count: %u",
+                                 share->key, share->use_count));
         free_share(&share);
         share= 0;
       }
@@ -2445,10 +2496,8 @@ int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
       pthread_mutex_unlock(&ndbcluster_mutex);
       DBUG_RETURN(1);
     }
-#ifdef NOT_YET
-    if (share->connect_count != 
+    if (!share_may_exist || share->connect_count != 
         g_ndb_cluster_connection->get_connect_count())
-#endif
     {
       handle_trailing_share(share);
       share= NULL;
@@ -2456,23 +2505,33 @@ int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
   }
 
   /* Create share which is needed to hold replication information */
-#ifdef NOT_YET
   if (share)
   {
+    /* ndb_share reference create */
     ++share->use_count;
+    DBUG_PRINT("NDB_SHARE", ("%s create  use_count: %u",
+                             share->key, share->use_count));
   }
-  else
-#endif
-  if (!(share= get_share(key, 0, TRUE, TRUE)))
+  /* ndb_share reference create */
+  else if (!(share= get_share(key, 0, TRUE, TRUE)))
   {
     sql_print_error("NDB Binlog: "
                     "allocating table share for %s failed", key);
   }
+  else
+  {
+    DBUG_PRINT("NDB_SHARE", ("%s create  use_count: %u",
+                             share->key, share->use_count));
+  }
 
   if (!ndb_schema_share &&
       strcmp(share->db, NDB_REP_DB) == 0 &&
       strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
     do_event_op= 1;
+  else if (!ndb_apply_status_share &&
+           strcmp(share->db, NDB_REP_DB) == 0 &&
+           strcmp(share->table_name, NDB_APPLY_TABLE) == 0)
+    do_event_op= 1;
 
   if (!do_event_op)
   {
@@ -2744,7 +2803,7 @@ ndbcluster_create_event_ops(NDB_SHARE *share, const NDBTAB *ndbtab,
   else if (!ndb_apply_status_share && strcmp(share->db, NDB_REP_DB) == 0 &&
            strcmp(share->table_name, NDB_APPLY_TABLE) == 0)
     do_ndb_apply_status_share= 1;
-  else if (!binlog_filter->db_ok(share->db))
+  else if (!binlog_filter->db_ok(share->db) || !ndb_binlog_running)
   {
     share->flags|= NSF_NO_BINLOG;
     DBUG_RETURN(0);
@@ -2756,6 +2815,9 @@ ndbcluster_create_event_ops(NDB_SHARE *share, const NDBTAB *ndbtab,
 
     DBUG_ASSERT(share->use_count > 1);
     sql_print_error("NDB Binlog: discover reusing old ev op");
+    /* ndb_share reference ToDo free */
+    DBUG_PRINT("NDB_SHARE", ("%s ToDo free  use_count: %u",
+                             share->key, share->use_count));
     free_share(&share); // old event op already has reference
     DBUG_RETURN(0);
   }
@@ -2898,15 +2960,24 @@ ndbcluster_create_event_ops(NDB_SHARE *share, const NDBTAB *ndbtab,
     break;
   }
 
+  /* ndb_share reference binlog */
   get_share(share);
+  DBUG_PRINT("NDB_SHARE", ("%s binlog  use_count: %u",
+                           share->key, share->use_count));
   if (do_ndb_apply_status_share)
   {
+    /* ndb_share reference binlog extra */
     ndb_apply_status_share= get_share(share);
+    DBUG_PRINT("NDB_SHARE", ("%s binlog extra  use_count: %u",
+                             share->key, share->use_count));
     (void) pthread_cond_signal(&injector_cond);
   }
   else if (do_ndb_schema_share)
   {
+    /* ndb_share reference binlog extra */
     ndb_schema_share= get_share(share);
+    DBUG_PRINT("NDB_SHARE", ("%s binlog extra  use_count: %u",
+                             share->key, share->use_count));
     (void) pthread_cond_signal(&injector_cond);
   }
 
@@ -3093,6 +3164,9 @@ ndb_binlog_thread_handle_non_data_event(THD *thd, Ndb *ndb,
           ndb_binlog_tables_inited && ndb_binlog_running)
         sql_print_information("NDB Binlog: ndb tables initially "
                               "read only on reconnect.");
+      /* ndb_share reference binlog extra free */
+      DBUG_PRINT("NDB_SHARE", ("%s binlog extra free  use_count: %u",
+                               share->key, share->use_count));
       free_share(&ndb_apply_status_share);
       ndb_apply_status_share= 0;
     }
@@ -3109,6 +3183,9 @@ ndb_binlog_thread_handle_non_data_event(THD *thd, Ndb *ndb,
           ndb_binlog_tables_inited && ndb_binlog_running)
         sql_print_information("NDB Binlog: ndb tables initially "
                               "read only on reconnect.");
+      /* ndb_share reference binlog extra free */
+      DBUG_PRINT("NDB_SHARE", ("%s binlog extra free  use_count: %u",
+                               share->key, share->use_count));
       free_share(&ndb_apply_status_share);
       ndb_apply_status_share= 0;
     }
@@ -4009,6 +4086,10 @@ err:
 
   if (ndb_apply_status_share)
   {
+    /* ndb_share reference binlog extra free */
+    DBUG_PRINT("NDB_SHARE", ("%s binlog extra free  use_count: %u",
+                             ndb_apply_status_share->key,
+                             ndb_apply_status_share->use_count));
     free_share(&ndb_apply_status_share);
     ndb_apply_status_share= 0;
   }
@@ -4016,6 +4097,10 @@ err:
   {
     /* begin protect ndb_schema_share */
     pthread_mutex_lock(&ndb_schema_share_mutex);
+    /* ndb_share reference binlog extra free */
+    DBUG_PRINT("NDB_SHARE", ("%s binlog extra free  use_count: %u",
+                             ndb_schema_share->key,
+                             ndb_schema_share->use_count));
     free_share(&ndb_schema_share);
     ndb_schema_share= 0;
     pthread_mutex_unlock(&ndb_schema_share_mutex);
@@ -4037,6 +4122,9 @@ err:
       DBUG_ASSERT(share->op == op ||
                   share->op_old == op);
       share->op= share->op_old= 0;
+      /* 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);
     }
@@ -4057,6 +4145,9 @@ err:
       DBUG_ASSERT(share->op == op ||
                   share->op_old == op);
       share->op= share->op_old= 0;
+      /* 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);
     }