WL#7083 step 5.1. Anonymous transactions: Maintain a counter of ongoing anonymous transactions.

In order to prevent switching gtid_mode from ON_PERMISSIVE to ON when
there are ongoing anonymous transactions, we introduce a counter that
counts the number of ongoing anonymous transactions.

@atomic_class.h
- Auxiliary class to use C++ syntax to access atomic integers.

@rpl_gtid.h
- Add member functions in Gtid_state to acquire and release anonymous
ownership.

@rpl_gtid_execution.cc
- Acquire anonymous ownership when executing set gtid_next='anonymous'.

@rpl_gtid_state.cc
- Release anonymous ownership in update_gtid_impl.
- Acquire anonymous ownership in generate_automatic_gtid.

@binlog.cc:
- Decrease global counter when releasing anonymous ownership.
- When thread holds anonymous ownership, and gtid_end_transaction is
called, it should call update_on_commit instead of clear_owned_gtids, so
that it releases anonymous ownership.

Author: Sven Sandberg

sql/atomic_class.h

@@ -0,0 +1,117 @@
+/* Copyright (c) 2015, Oracle and/or its affiliates. All rights reserved.
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License as
+   published by the Free Software Foundation; version 2 of the
+   License.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+   02110-1301 USA */
+
+
+#ifndef ATOMIC_CLASS_H_INCLUDED
+#define ATOMIC_CLASS_H_INCLUDED
+
+#include "my_atomic.h"
+
+/**
+  Wrapper class to use C++ syntax to access atomic integers.
+*/
+#define DEFINE_ATOMIC_CLASS(NAME, SUFFIX, TYPE)                         \
+  class Atomic_##NAME                                                   \
+  {                                                                     \
+public:                                                                 \
+    /* Create a new Atomic_* object. */                                 \
+    Atomic_##NAME(TYPE n= 0)                                            \
+    {                                                                   \
+      my_atomic_store##SUFFIX(&value, n);                               \
+    }                                                                   \
+    /* Atomically read the value. */                                    \
+    TYPE atomic_get()                                                   \
+    {                                                                   \
+      return my_atomic_load##SUFFIX(&value);                            \
+    }                                                                   \
+    /* Atomically set the value. */                                     \
+    void atomic_set(TYPE n)                                             \
+    {                                                                   \
+      my_atomic_store##SUFFIX(&value, n);                               \
+    }                                                                   \
+    /* Atomically add to the value, and return the old value. */        \
+    TYPE atomic_add(TYPE n)                                             \
+    {                                                                   \
+      return my_atomic_add##SUFFIX(&value, n);                          \
+    }                                                                   \
+    /* Atomically set the value and return the old value. */            \
+    TYPE atomic_get_and_set(TYPE n)                                     \
+    {                                                                   \
+      return my_atomic_fas##SUFFIX(&value, n);                          \
+    }                                                                   \
+    /* If the old value is equal to *old, set the value to new and */   \
+    /* return true. Otherwise, store the old value in '*old' and */     \
+    /* return false. */                                                 \
+    bool atomic_compare_and_swap(TYPE *old, TYPE n)                     \
+    {                                                                   \
+      return my_atomic_cas##SUFFIX(&value, old, n);                     \
+    }                                                                   \
+    /* Read the value *non-atomically*. */                              \
+    TYPE non_atomic_get()                                               \
+    {                                                                   \
+      return value;                                                     \
+    }                                                                   \
+    /* Set the value *non-atomically*. */                               \
+    void non_atomic_set(TYPE n)                                         \
+    {                                                                   \
+      value= n;                                                         \
+    }                                                                   \
+    /* Add to the value *non-atomically*. */                            \
+    TYPE non_atomic_add(TYPE n)                                         \
+    {                                                                   \
+      TYPE ret= value;                                                  \
+      value+= n;                                                        \
+      return ret;                                                       \
+    }                                                                   \
+    /* Set the value to the greatest of (old, n). */                    \
+                                                                        \
+    /* The function will internally requires multiple atomic */         \
+    /* operations.  If the old value is known (or guessed), and less */ \
+    /* than n, it requires one atomic operation less. Therefore, */     \
+    /* the caller should set *guess to whatever is the likely value */  \
+    /* that the variable currently has, if such a guess is known. */    \
+                                                                        \
+    /* @return If the new value is changed to n, *guess is set to */    \
+    /* the old value and the the function returns true.  Otherwise, */  \
+    /* *guess is set to the current value (which is greater than or */  \
+    /* equal to n), and the function returns false. */                  \
+    bool atomic_set_to_max(TYPE n, TYPE *guess= NULL)                   \
+    {                                                                   \
+      TYPE _guess;                                                      \
+      if (guess == NULL)                                                \
+      {                                                                 \
+        _guess= n - 1;                                                  \
+        guess= &_guess;                                                 \
+      }                                                                 \
+      else                                                              \
+        DBUG_ASSERT(*guess < n);                                        \
+      bool ret;                                                         \
+      do {                                                              \
+        ret= atomic_compare_and_swap(guess, n);                         \
+      } while (!ret && *guess < n);                                     \
+      return ret;                                                       \
+    }                                                                   \
+                                                                        \
+private:                                                                \
+    volatile TYPE value;                                                \
+  }
+
+DEFINE_ATOMIC_CLASS(int32, 32, int32);
+//DEFINE_ATOMIC_CLASS(int64, 64, int64);
+//DEFINE_ATOMIC_CLASS(pointer, ptr, void *);
+
+#endif //ifndef ATOMIC_CLASS_H_INCLUDED

sql/binlog.cc

@@ -740,8 +740,12 @@ class binlog_cache_mngr {
           thd->owned_gtid.sidno == THD::OWNED_SIDNO_ANONYMOUS)
       {
         thd->clear_owned_gtids();
+        global_sid_lock->rdlock();
+        gtid_state->release_anonymous_ownership();
+        global_sid_lock->unlock();
       }
     }
+    DEBUG_SYNC(thd, "after_flush_stm_cache_before_flush_trx_cache");
     if (int error= trx_cache.flush(thd, &trx_bytes, wrote_xid))
       return error;
     *bytes_written= stmt_bytes + trx_bytes;
@@ -1242,7 +1246,7 @@ int MYSQL_BIN_LOG::gtid_end_transaction(THD *thd)
   }
   else if (thd->owned_gtid.sidno == THD::OWNED_SIDNO_ANONYMOUS)
   {
-    thd->clear_owned_gtids();
+    gtid_state->update_on_commit(thd);
   }
   DBUG_RETURN(0);
 }

sql/rpl_gtid.h

@@ -30,6 +30,7 @@
 #endif
 
 #include <list>
+#include "atomic_class.h"
 
 using binary_log::Uuid;
 /**
@@ -2373,6 +2374,50 @@ class Gtid_state
     @param thd Thread for which owned groups are updated.
   */
   void update_on_rollback(THD *thd);
+
+  /**
+    Acquire anonymous ownership.
+
+    The caller must hold either sid_lock.rdlock or
+    sid_lock.wrlock. (The caller must have taken the lock and checked
+    that gtid_mode!=ON before calling this function, or else the
+    gtid_mode could have changed to ON by a concurrent SET GTID_MODE.)
+  */
+  void acquire_anonymous_ownership()
+  {
+    DBUG_ENTER("Gtid_state::acquire_anonymous_ownership");
+    sid_lock->assert_some_lock();
+    DBUG_ASSERT(get_gtid_mode(GTID_MODE_LOCK_SID) != GTID_MODE_ON);
+#ifndef DBUG_OFF
+    int32 old_value=
+#endif
+      anonymous_gtid_count.atomic_add(1);
+    DBUG_PRINT("info", ("anonymous_gtid_count increased to %d", old_value + 1));
+    DBUG_ASSERT(old_value >= 0);
+    DBUG_VOID_RETURN;
+  }
+
+  /// Release anonymous ownership.
+  void release_anonymous_ownership()
+  {
+    DBUG_ENTER("Gtid_state::release_anonymous_ownership");
+    sid_lock->assert_some_lock();
+    DBUG_ASSERT(get_gtid_mode(GTID_MODE_LOCK_SID) != GTID_MODE_ON);
+#ifndef DBUG_OFF
+    int32 old_value=
+#endif
+      anonymous_gtid_count.atomic_add(-1);
+    DBUG_PRINT("info", ("anonymous_gtid_count decreased to %d", old_value - 1));
+    DBUG_ASSERT(old_value >= 1);
+    DBUG_VOID_RETURN;
+  }
+
+  /// Return the number of clients that hold anonymous ownership.
+  int32 get_anonymous_ownership_count()
+  {
+    return anonymous_gtid_count.atomic_get();
+  }
+
 #endif // ifndef MYSQL_CLIENT
 private:
   /**
@@ -2706,6 +2751,9 @@ class Gtid_state
   /// The SIDNO for this server.
   rpl_sidno server_sidno;
 
+  /// The number of anonymous transactions owned by any client.
+  Atomic_int32 anonymous_gtid_count;
+
   /// Used by unit tests that need to access private members.
 #ifdef FRIEND_OF_GTID_STATE
   friend FRIEND_OF_GTID_STATE;

sql/rpl_gtid_execution.cc

@@ -75,10 +75,11 @@ bool set_gtid_next(THD *thd, const Gtid_specification &spec)
       my_error(ER_CANT_SET_GTID_NEXT_TO_ANONYMOUS_WHEN_GTID_MODE_IS_ON, MYF(0));
       goto err;
     }
-    // @todo WL#7083: increase atomic counter here
 
-    thd->owned_gtid.sidno= THD::OWNED_SIDNO_ANONYMOUS;
     thd->variables.gtid_next.set_anonymous();
+    thd->owned_gtid.sidno= THD::OWNED_SIDNO_ANONYMOUS;
+    thd->owned_gtid.gno= 0;
+    gtid_state->acquire_anonymous_ownership();
   }
   else
   {

sql/rpl_gtid_state.cc

@@ -179,6 +179,49 @@ void Gtid_state::update_gtids_impl(THD *thd, bool is_commit)
 {
   DBUG_ENTER("Gtid_state::update_gtids_impl");
 
+  /*
+    This variable is true for anonymous transactions, when the
+    'transaction' has been split into multiple transactions in the
+    binlog, and the present transaction is not the last one.
+
+    This means two things:
+
+    - We should not release anonymous ownership in case
+      gtid_next=anonymous.  If we did, it would be possible for user
+      to set GTID_MODE=ON from a concurrent transaction, making it
+      impossible to commit the current transaction.
+
+    - We should not decrease the counters for GTID-violating
+      statements.  If we did, it would be possible for a concurrent
+      client to set ENFORCE_GTID_CONSISTENCY=ON despite there is an
+      ongoing transaction that violates GTID consistency.
+
+    The flag is set in two cases:
+
+     1. We are committing the statement cache when there are more
+        changes in the transaction cache.
+
+        This happens either because a single statement in the
+        beginning of a transaction updates both transactional and
+        non-transactional tables, or because we are committing a
+        non-transactional update in the middle of a transaction when
+        binlog_direct_non_transactional_updates=1.
+
+        In this case, the flag is set further down in this function.
+
+     2. The statement is one of the special statements that may
+        generate multiple transactions: CREATE...SELECT, DROP TABLE,
+        DROP DATABASE. See comment for THD::owned_gtid in
+        sql/sql_class.h.
+
+        In this case, the THD::is_commit_in_middle_of_statement flag
+        is set by the caller and the flag becomes true here.
+  */
+  bool more_transactions_with_same_gtid_next=
+    thd->is_commit_in_middle_of_statement;
+  DBUG_PRINT("info", ("thd->is_commit_in_middle_of_statement=%d",
+                      thd->is_commit_in_middle_of_statement));
+
   // Caller must take global_sid_lock.
   global_sid_lock->assert_some_lock();
 
@@ -244,42 +287,72 @@ void Gtid_state::update_gtids_impl(THD *thd, bool is_commit)
         gtids_only_in_table._add_gtid(thd->owned_gtid);
       }
     }
+
+    broadcast_owned_sidnos(thd);
+    unlock_owned_sidnos(thd);
+
+    thd->clear_owned_gtids();
+    if (thd->variables.gtid_next.type == GTID_GROUP)
+    {
+      DBUG_ASSERT(!more_transactions_with_same_gtid_next);
+      thd->variables.gtid_next.set_undefined();
+    }
+    else
+    {
+      /*
+        Can be UNDEFINED for statements where
+        gtid_pre_statement_checks skips the test for undefined,
+        e.g. ROLLBACK.
+      */
+      DBUG_ASSERT(thd->variables.gtid_next.type == AUTOMATIC_GROUP ||
+                  thd->variables.gtid_next.type == UNDEFINED_GROUP);
+    }
   }
+  else if (thd->owned_gtid.sidno == THD::OWNED_SIDNO_ANONYMOUS)
+  {
+    DBUG_ASSERT(thd->variables.gtid_next.type == ANONYMOUS_GROUP ||
+                thd->variables.gtid_next.type == AUTOMATIC_GROUP);
+    /*
+      If there is more in the transaction cache, set
+      more_transactions_with_same_gtid_next to indicate this.
 
-  /*
-    There may be commands that cause implicit commits, e.g.
-    SET AUTOCOMMIT=1 may cause the previous statements to commit
-    without executing a COMMIT command or be on auto-commit mode.
-  */
-  broadcast_owned_sidnos(thd);
-  unlock_owned_sidnos(thd);
-  if (thd->variables.gtid_next.type == GTID_GROUP)
-    thd->variables.gtid_next.set_undefined();
-  /*
-    This early return prevents releasing anonymous ownership when a
-    non-transactional statement is flushed to the binary log in the
-    middle of a transaction.  If we would release ownership in the
-    middle of a transaction when gtid_next.type==ANONYMOUS_GROUP, it
-    would be possible for a concurrent transaction to change GTID_MODE
-    to ON in the middle of a transaction, making it impossible to
-    commit.
-  */
-  if (opt_bin_log && thd->variables.gtid_next.type == ANONYMOUS_GROUP)
+      See comment for the declaration of
+      more_transactions_with_same_gtid_next.
+    */
+    if (opt_bin_log)
+    {
+      // Needed before is_binlog_cache_empty.
+      thd->binlog_setup_trx_data();
+      if (!thd->is_binlog_cache_empty(true))
+      {
+        more_transactions_with_same_gtid_next= true;
+        DBUG_PRINT("info", ("Transaction cache is non-empty: setting "
+                            "more_transaction_with_same_gtid_next="
+                            "true."));
+      }
+    }
+    if (!(more_transactions_with_same_gtid_next &&
+          thd->variables.gtid_next.type == ANONYMOUS_GROUP))
+    {
+      release_anonymous_ownership();
+      thd->clear_owned_gtids();
+    }
+  }
+  else
   {
-    // Needed before is_binlog_cache_empty.
-    thd->binlog_setup_trx_data();
-    if (!thd->is_binlog_cache_empty(true))
-      DBUG_VOID_RETURN;
+    // Nothing is owned. Then it must be a rollback of an automatic
+    // transaction.
+    DBUG_ASSERT(!is_commit);
+    DBUG_ASSERT(thd->variables.gtid_next.type == AUTOMATIC_GROUP);
   }
-  if (!(thd->variables.gtid_next.type == ANONYMOUS_GROUP &&
-        thd->is_commit_in_middle_of_statement))
-    thd->clear_owned_gtids();
+
   thd->owned_gtid.dbug_print(NULL,
                              "set owned_gtid (clear) in update_gtids_impl");
 
   DBUG_VOID_RETURN;
 }
 
+
 int Gtid_state::wait_for_gtid_set(THD* thd, String* gtid_set_text, longlong timeout)
 {
   int error= 0;
@@ -440,6 +513,7 @@ enum_return_status Gtid_state::generate_automatic_gtid(THD *thd,
     // using an anonymous transaction.
     thd->owned_gtid.sidno= THD::OWNED_SIDNO_ANONYMOUS;
     thd->owned_gtid.gno= 0;
+    acquire_anonymous_ownership();
     thd->owned_gtid.dbug_print(NULL,
                                "set owned_gtid (anonymous) in generate_automatic_gtid");
   }