rpl_gtid.h

/* Copyright (c) 2012, 2013, 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 RPL_GTID_H_INCLUDED
#define RPL_GTID_H_INCLUDED


#include <m_string.h>
#include <mysqld_error.h>
#include <my_global.h>
#ifdef MYSQL_SERVER
#include <mysqld.h>
#endif

#ifdef MYSQL_CLIENT
#define BINLOG_ERROR(MYSQLBINLOG_ERROR, SERVER_ERROR) error MYSQLBINLOG_ERROR
#else
#define BINLOG_ERROR(MYSQLBINLOG_ERROR, SERVER_ERROR) my_error SERVER_ERROR
#endif


#include "hash.h"
#include "lf.h"
#include "my_atomic.h"

#define SKIP_WHITESPACE() while (my_isspace(&my_charset_utf8_general_ci, *s)) s++
/*
  This macro must be used to filter out parts of the code that
  is not used now but may be useful in future. In other words,
  we want to keep such code until we make up our minds on whether
  it should be removed or not.
*/
#undef NON_DISABLED_GTID

/*
  This macro must be used to filter out parts of the code that
  is not used now but we are not sure if there is a bug around
  them. In other words, we want to keep such code until we have
  time to investigate it.
*/
#undef NON_ERROR_GTID

#ifndef MYSQL_CLIENT
class String;
class THD;
#endif // ifndef MYSQL_CLIENT


typedef int32 rpl_sidno;
typedef int64 rpl_gno;
typedef int64 rpl_binlog_pos;


enum enum_return_status
{
  RETURN_STATUS_OK= 0,
  RETURN_STATUS_UNREPORTED_ERROR= 1,
  RETURN_STATUS_REPORTED_ERROR= 2
};

#ifdef DBUG_OFF
#define __CHECK_RETURN_STATUS(STATUS, ACTION, STATUS_NAME, ALLOW_UNREPORTED)
#else
extern void check_return_status(enum_return_status status,
                                const char *action, const char *status_name,
                                int allow_unreported);
#define __CHECK_RETURN_STATUS(STATUS, ACTION, STATUS_NAME, ALLOW_UNREPORTED) \
  check_return_status(STATUS, ACTION, STATUS_NAME, ALLOW_UNREPORTED);
#endif

#define __PROPAGATE_ERROR(STATUS, RETURN_VALUE, ALLOW_UNREPORTED)       \
  do                                                                    \
  {                                                                     \
    enum_return_status __propagate_error_status= STATUS;                \
    if (__propagate_error_status != RETURN_STATUS_OK) {                 \
      __CHECK_RETURN_STATUS(__propagate_error_status, "Propagating",    \
                            #STATUS, ALLOW_UNREPORTED);                 \
      DBUG_RETURN(RETURN_VALUE);                                        \
    }                                                                   \
  } while (0)

#define __RETURN_STATUS(STATUS, ALLOW_UNREPORTED)                       \
  do                                                                    \
  {                                                                     \
    enum_return_status __return_status_status= STATUS;                  \
    __CHECK_RETURN_STATUS(__return_status_status, "Returning",          \
                          #STATUS, ALLOW_UNREPORTED);                   \
    DBUG_RETURN(__return_status_status);                                \
  } while (0)

#define PROPAGATE_ERROR(STATUS)                                 \
  __PROPAGATE_ERROR(STATUS, __propagate_error_status, true)

#define PROPAGATE_REPORTED_ERROR(STATUS)                        \
  __PROPAGATE_ERROR(STATUS, __propagate_error_status, false)

#define PROPAGATE_REPORTED_ERROR_INT(STATUS)    \
  __PROPAGATE_ERROR(STATUS, 1, false)

#define RETURN_STATUS(STATUS) __RETURN_STATUS(STATUS, true)

#define RETURN_REPORTED_STATUS(STATUS) __RETURN_STATUS(STATUS, false)

#define RETURN_OK DBUG_RETURN(RETURN_STATUS_OK)

#define RETURN_REPORTED_ERROR RETURN_STATUS(RETURN_STATUS_REPORTED_ERROR)

#define RETURN_UNREPORTED_ERROR RETURN_STATUS(RETURN_STATUS_UNREPORTED_ERROR)


const rpl_gno MAX_GNO= LONGLONG_MAX;
const int MAX_GNO_TEXT_LENGTH= 19;
const int MAX_THREAD_ID_TEXT_LENGTH= 19;


rpl_gno parse_gno(const char **s);
int format_gno(char *s, rpl_gno gno);


struct Uuid
{
  enum_return_status parse(const char *string);
  void clear() { memset(bytes, 0, BYTE_LENGTH); }
  void copy_from(const uchar *data) { memcpy(bytes, data, BYTE_LENGTH); }
  void copy_from(const Uuid &data) { copy_from(data.bytes); }
  void copy_to(uchar *data) const { memcpy(data, bytes, BYTE_LENGTH); }
  bool equals(const Uuid &other) const
  { return memcmp(bytes, other.bytes, BYTE_LENGTH) == 0; }
  size_t to_string(char *buf) const;
  static size_t to_string(const uchar* bytes_arg, char *buf);
#ifndef DBUG_OFF

  void print() const
  {
    char buf[TEXT_LENGTH + 1];
    to_string(buf);
    printf("%s\n", buf);
  }
#endif

  void dbug_print(const char *text= "") const
  {
#ifndef DBUG_OFF
    char buf[TEXT_LENGTH + 1];
    to_string(buf);
    DBUG_PRINT("info", ("%s%s%s", text, *text ? ": " : "", buf));
#endif
  }
  static bool is_valid(const char *string);
  static const size_t TEXT_LENGTH= 36;
  static const size_t BYTE_LENGTH= 16;
  static const size_t BIT_LENGTH= 128;
  uchar bytes[BYTE_LENGTH];
private:
  static const int NUMBER_OF_SECTIONS= 5;
  static const int bytes_per_section[NUMBER_OF_SECTIONS];
  static const int hex_to_byte[256];
};


typedef Uuid rpl_sid;


class Checkable_rwlock
{
public:
  Checkable_rwlock()
  {
#ifndef DBUG_OFF
    my_atomic_rwlock_init(&atomic_lock);
    lock_state= 0;
#else
    is_write_lock= false;
#endif
#ifdef MYSQL_SERVER
    mysql_rwlock_init(key_rwlock_global_sid_lock, &rwlock);
#else
    mysql_rwlock_init(0, &rwlock);
#endif
  }
  ~Checkable_rwlock()
  {
#ifndef DBUG_OFF
    my_atomic_rwlock_destroy(&atomic_lock);
#endif
    mysql_rwlock_destroy(&rwlock);
  }

  inline void rdlock()
  {
    mysql_rwlock_rdlock(&rwlock);
    assert_no_wrlock();
#ifndef DBUG_OFF
    my_atomic_rwlock_wrlock(&atomic_lock);
    my_atomic_add32(&lock_state, 1);
    my_atomic_rwlock_wrunlock(&atomic_lock);
#endif
  }
  inline void wrlock()
  {
    mysql_rwlock_wrlock(&rwlock);
    assert_no_lock();
#ifndef DBUG_OFF
    my_atomic_rwlock_wrlock(&atomic_lock);
    my_atomic_store32(&lock_state, -1);
    my_atomic_rwlock_wrunlock(&atomic_lock);
#else
    is_write_lock= true;
#endif
  }
  inline void unlock()
  {
    assert_some_lock();
#ifndef DBUG_OFF
    my_atomic_rwlock_wrlock(&atomic_lock);
    int val= my_atomic_load32(&lock_state);
    if (val > 0)
      my_atomic_add32(&lock_state, -1);
    else if (val == -1)
      my_atomic_store32(&lock_state, 0);
    else
      DBUG_ASSERT(0);
    my_atomic_rwlock_wrunlock(&atomic_lock);
#else
    is_write_lock= false;
#endif
    mysql_rwlock_unlock(&rwlock);
  }
  inline bool is_wrlock()
  {
    assert_some_lock();
#ifndef DBUG_OFF
    return get_state() == -1;
#else
    return is_write_lock;
#endif
  }

  inline void assert_some_lock() const
  { DBUG_ASSERT(get_state() != 0); }
  inline void assert_some_rdlock() const
  { DBUG_ASSERT(get_state() > 0); }
  inline void assert_some_wrlock() const
  { DBUG_ASSERT(get_state() == -1); }
  inline void assert_no_wrlock() const
  { DBUG_ASSERT(get_state() >= 0); }
  inline void assert_no_rdlock() const
  { DBUG_ASSERT(get_state() <= 0); }
  inline void assert_no_lock() const
  { DBUG_ASSERT(get_state() == 0); }

private:
#ifndef DBUG_OFF

  volatile int32 lock_state;
  mutable my_atomic_rwlock_t atomic_lock;
  inline int32 get_state() const
  {
    int32 ret;
    my_atomic_rwlock_rdlock(&atomic_lock);
    ret= my_atomic_load32(const_cast<volatile int32*>(&lock_state));
    my_atomic_rwlock_rdunlock(&atomic_lock);
    return ret;
  }
#else
  bool is_write_lock;
#endif

  mysql_rwlock_t rwlock;
};


extern Checkable_rwlock *global_sid_lock;


class Sid_map
{
public:
  Sid_map(Checkable_rwlock *sid_lock);
  ~Sid_map();
#ifdef NON_DISABLED_GTID

  enum_return_status clear();
#endif

  rpl_sidno add_sid(const rpl_sid &sid);
  rpl_sidno sid_to_sidno(const rpl_sid &sid) const
  {
    if (sid_lock != NULL)
      sid_lock->assert_some_lock();
    Node *node= (Node *)my_hash_search(&_sid_to_sidno, sid.bytes,
                                       rpl_sid::BYTE_LENGTH);
    if (node == NULL)
      return 0;
    return node->sidno;
  }
  const rpl_sid &sidno_to_sid(rpl_sidno sidno) const
  {
    if (sid_lock != NULL)
      sid_lock->assert_some_lock();
    DBUG_ASSERT(sidno >= 1 && sidno <= get_max_sidno());
    return (*dynamic_element(&_sidno_to_sid, sidno - 1, Node **))->sid;
  }
  rpl_sidno get_sorted_sidno(rpl_sidno n) const
  {
    if (sid_lock != NULL)
      sid_lock->assert_some_lock();
    rpl_sidno ret= *dynamic_element(&_sorted, n, rpl_sidno *);
    return ret;
  }
  rpl_sidno get_max_sidno() const
  {
    if (sid_lock != NULL)
      sid_lock->assert_some_lock();
    return _sidno_to_sid.elements;
  }

private:
  struct Node
  {
    rpl_sidno sidno;
    rpl_sid sid;
  };

  enum_return_status add_node(rpl_sidno sidno, const rpl_sid &sid);

  mutable Checkable_rwlock *sid_lock;

  DYNAMIC_ARRAY _sidno_to_sid;
  HASH _sid_to_sidno;
  DYNAMIC_ARRAY _sorted;
};


extern Sid_map *global_sid_map;


class Mutex_cond_array
{
public:
  Mutex_cond_array(Checkable_rwlock *global_lock);
  ~Mutex_cond_array();
  inline void lock(int n) const
  {
    assert_not_owner(n);
    mysql_mutex_lock(&get_mutex_cond(n)->mutex);
  }
  inline void unlock(int n) const
  {
    assert_owner(n);
    mysql_mutex_unlock(&get_mutex_cond(n)->mutex);
  }
  inline void broadcast(int n) const
  {
    mysql_cond_broadcast(&get_mutex_cond(n)->cond);
  }
  inline void assert_owner(int n) const
  {
#ifndef DBUG_OFF
    mysql_mutex_assert_owner(&get_mutex_cond(n)->mutex);
#endif
  }
  inline void assert_not_owner(int n) const
  {
#ifndef DBUG_OFF
    mysql_mutex_assert_not_owner(&get_mutex_cond(n)->mutex);
#endif
  }
  inline void wait(int n) const
  {
    DBUG_ENTER("Mutex_cond_array::wait");
    Mutex_cond *mutex_cond= get_mutex_cond(n);
    global_lock->unlock();
    mysql_mutex_assert_owner(&mutex_cond->mutex);
    mysql_cond_wait(&mutex_cond->cond, &mutex_cond->mutex);
    mysql_mutex_assert_owner(&mutex_cond->mutex);
    DBUG_VOID_RETURN;
  }
#ifndef MYSQL_CLIENT

  void enter_cond(THD *thd, int n, PSI_stage_info *stage,
                  PSI_stage_info *old_stage) const;
#endif // ifndef MYSQL_CLIENT

  inline int get_max_index() const
  {
    global_lock->assert_some_lock();
    return array.elements - 1;
  }
  enum_return_status ensure_index(int n);
private:
  struct Mutex_cond
  {
    mysql_mutex_t mutex;
    mysql_cond_t cond;
  };
  inline Mutex_cond *get_mutex_cond(int n) const
  {
    global_lock->assert_some_lock();
    DBUG_ASSERT(n <= get_max_index());
    Mutex_cond *ret= *dynamic_element(&array, n, Mutex_cond **);
    DBUG_ASSERT(ret);
    return ret;
  }
  mutable Checkable_rwlock *global_lock;
  DYNAMIC_ARRAY array;
};


struct Gtid
{
  rpl_sidno sidno;
  rpl_gno gno;

  void clear() { sidno= 0; gno= 0; }
  static const int MAX_TEXT_LENGTH= Uuid::TEXT_LENGTH + 1 + MAX_GNO_TEXT_LENGTH;
  static bool is_valid(const char *text);
  int to_string(const rpl_sid &sid, char *buf) const;
  int to_string(const Sid_map *sid_map, char *buf) const;
  bool equals(const Gtid &other) const
  { return sidno == other.sidno && gno == other.gno; }
  enum_return_status parse(Sid_map *sid_map, const char *text);

#ifndef DBUG_OFF

  void print(const Sid_map *sid_map) const
  {
    char buf[MAX_TEXT_LENGTH + 1];
    to_string(sid_map, buf);
    printf("%s\n", buf);
  }
#endif

  void dbug_print(const Sid_map *sid_map, const char *text= "") const
  {
#ifndef DBUG_OFF
    char buf[MAX_TEXT_LENGTH + 1];
    to_string(sid_map, buf);
    DBUG_PRINT("info", ("%s%s%s", text, *text ? ": " : "", buf));
#endif
  }
};


class Gtid_set
{
public:
  Gtid_set(Sid_map *sid_map, Checkable_rwlock *sid_lock= NULL);
  Gtid_set(Sid_map *sid_map, const char *text, enum_return_status *status,
           Checkable_rwlock *sid_lock= NULL);
private:
  void init();
public:
  ~Gtid_set();
  void clear();
  enum_return_status _add_gtid(rpl_sidno sidno, rpl_gno gno)
  {
    DBUG_ENTER("Gtid_set::_add_gtid(sidno, gno)");
    Interval_iterator ivit(this, sidno);
    Free_intervals_lock lock(this);
    enum_return_status ret= add_gno_interval(&ivit, gno, gno + 1, &lock);
    DBUG_RETURN(ret);
  }
  enum_return_status _remove_gtid(rpl_sidno sidno, rpl_gno gno)
  {
    DBUG_ENTER("Gtid_set::_remove_gtid(rpl_sidno, rpl_gno)");
    if (sidno <= get_max_sidno())
    {
      Interval_iterator ivit(this, sidno);
      Free_intervals_lock lock(this);
      enum_return_status ret= remove_gno_interval(&ivit, gno, gno + 1, &lock);
      DBUG_RETURN(ret);
    }
    RETURN_OK;
  }
  enum_return_status _add_gtid(const Gtid &gtid)
  { return _add_gtid(gtid.sidno, gtid.gno); }
  enum_return_status add_gtid_set(const Gtid_set *other);
  enum_return_status remove_gtid_set(const Gtid_set *other);
  enum_return_status add_gtid_text(const char *text, bool *anonymous= NULL);
  enum_return_status add_gtid_encoding(const uchar *encoded, size_t length,
                                       size_t *actual_length= NULL);
  bool contains_gtid(rpl_sidno sidno, rpl_gno gno) const;
  bool contains_gtid(const Gtid &gtid) const
  { return contains_gtid(gtid.sidno, gtid.gno); }
  rpl_sidno get_max_sidno() const
  {
    if (sid_lock)
      sid_lock->assert_some_lock();
    return intervals.elements;
  }
  enum_return_status ensure_sidno(rpl_sidno sidno);
  bool is_subset(const Gtid_set *super) const;
  bool is_intersection_nonempty(const Gtid_set *other) const;
  enum_return_status intersection(const Gtid_set *other, Gtid_set *result);
  bool is_empty() const
  {
    Gtid_iterator git(this);
    return git.get().sidno == 0;
  }
  bool contains_sidno(rpl_sidno sidno) const
  {
    DBUG_ASSERT(sidno >= 1);
    if (sidno > get_max_sidno())
      return false;
    Const_interval_iterator ivit(this, sidno);
    return ivit.get() != NULL;
  }
  static bool is_valid(const char *text);
  char *to_string() const
  {
    char *str= (char *)my_malloc(get_string_length() + 1, MYF(MY_WME));
    if (str != NULL)
      to_string(str);
    return str;
  }
#ifndef DBUG_OFF

  void print() const
  {
    char *str= to_string();
    printf("%s\n", str);
    my_free(str);
  }
#endif

  void dbug_print(const char *text= "") const
  {
#ifndef DBUG_OFF
    char *str= to_string();
    DBUG_PRINT("info", ("%s%s'%s'", text, *text ? ": " : "", str));
    my_free(str);
#endif
  }

  struct String_format
  {
    const char *begin;
    const char *end;
    const char *sid_gno_separator;
    const char *gno_start_end_separator;
    const char *gno_gno_separator;
    const char *gno_sid_separator;
    const char *empty_set_string;
    const int begin_length;
    const int end_length;
    const int sid_gno_separator_length;
    const int gno_start_end_separator_length;
    const int gno_gno_separator_length;
    const int gno_sid_separator_length;
    const int empty_set_string_length;
  };
  int get_string_length(const String_format *string_format= NULL) const;
  int to_string(char *buf, const String_format *string_format= NULL) const;

  int to_string(char **buf, const String_format *string_format= NULL) const;

  static const String_format default_string_format;
  static const String_format sql_string_format;
  static const String_format commented_string_format;

  Sid_map *get_sid_map() const { return sid_map; }

  struct Interval
  {
  public:
    rpl_gno start;
    rpl_gno end;
    bool equals(const Interval &other) const
    {
      return start == other.start && end == other.end;
    }
    Interval *next;
  };

  void add_interval_memory(int n_intervals, Interval *intervals_param)
  {
    if (sid_lock != NULL)
      mysql_mutex_lock(&free_intervals_mutex);
    add_interval_memory_lock_taken(n_intervals, intervals_param);
    if (sid_lock != NULL)
      mysql_mutex_unlock(&free_intervals_mutex);
  }


  template<typename Gtid_set_p, typename Interval_p> class Interval_iterator_base
  {
  public:
    Interval_iterator_base(Gtid_set_p gtid_set, rpl_sidno sidno)
    {
      DBUG_ASSERT(sidno >= 1 && sidno <= gtid_set->get_max_sidno());
      init(gtid_set, sidno);
    }
    Interval_iterator_base(Gtid_set_p gtid_set)
    { p= const_cast<Interval_p *>(&gtid_set->free_intervals); }
    inline void init(Gtid_set_p gtid_set, rpl_sidno sidno)
    { p= dynamic_element(&gtid_set->intervals, sidno - 1, Interval_p *); }
    inline void next()
    {
      DBUG_ASSERT(*p != NULL);
      p= const_cast<Interval_p *>(&(*p)->next);
    }
    inline Interval_p get() const { return *p; }
  protected:
    Interval_p *p;
  };

  class Const_interval_iterator
    : public Interval_iterator_base<const Gtid_set *, const Interval *>
  {
  public:
    Const_interval_iterator(const Gtid_set *gtid_set, rpl_sidno sidno)
      : Interval_iterator_base<const Gtid_set *, const Interval *>(gtid_set, sidno) {}
    Const_interval_iterator(const Gtid_set *gtid_set)
      : Interval_iterator_base<const Gtid_set *, const Interval *>(gtid_set) {}
  };

  class Interval_iterator
    : public Interval_iterator_base<Gtid_set *, Interval *>
  {
  public:
    Interval_iterator(Gtid_set *gtid_set, rpl_sidno sidno)
      : Interval_iterator_base<Gtid_set *, Interval *>(gtid_set, sidno) {}
    Interval_iterator(Gtid_set *gtid_set)
      : Interval_iterator_base<Gtid_set *, Interval *>(gtid_set) {}
  private:
    inline void set(Interval *iv) { *p= iv; }
    inline void insert(Interval *iv) { iv->next= *p; set(iv); }
    inline void remove(Gtid_set *gtid_set)
    {
      DBUG_ASSERT(get() != NULL);
      Interval *next= (*p)->next;
      gtid_set->put_free_interval(*p);
      set(next);
    }
    friend class Gtid_set;
  };


  class Gtid_iterator
  {
  public:
    Gtid_iterator(const Gtid_set *gs)
      : gtid_set(gs), sidno(0), ivit(gs)
    {
      if (gs->sid_lock != NULL)
        gs->sid_lock->assert_some_wrlock();
      next_sidno();
    }
    inline void next()
    {
      DBUG_ASSERT(gno > 0 && sidno > 0);
      // go to next group in current interval
      gno++;
      // end of interval? then go to next interval for this sidno
      if (gno == ivit.get()->end)
      {
        ivit.next();
        const Interval *iv= ivit.get();
        // last interval for this sidno? then go to next sidno
        if (iv == NULL)
        {
          next_sidno();
          // last sidno? then don't try more
          if (sidno == 0)
            return;
          iv= ivit.get();
        }
        gno= iv->start;
      }
    }
    inline Gtid get() const
    {
      Gtid ret= { sidno, gno };
      return ret;
    }
  private:
    inline void next_sidno()
    {
      const Interval *iv;
      do
      {
        sidno++;
        if (sidno > gtid_set->get_max_sidno())
        {
          sidno= 0;
          gno= 0;
          return;
        }
        ivit.init(gtid_set, sidno);
        iv= ivit.get();
      } while (iv == NULL);
      gno= iv->start;
    }
    const Gtid_set *gtid_set;
    rpl_sidno sidno;
    rpl_gno gno;
    Const_interval_iterator ivit;
  };

public:

  void encode(uchar *buf) const;
  size_t get_encoded_length() const;

private:
  struct Interval_chunk
  {
    Interval_chunk *next;
    Interval intervals[1];
  };
  static const int CHUNK_GROW_SIZE= 8;

/*
  Functions sidno_equals() and equals() are only used by unitests
*/
#ifdef NON_DISABLED_UNITTEST_GTID

  bool sidno_equals(rpl_sidno sidno,
                    const Gtid_set *other, rpl_sidno other_sidno) const;
  bool equals(const Gtid_set *other) const;
#endif

  int get_n_intervals(rpl_sidno sidno) const
  {
    Const_interval_iterator ivit(this, sidno);
    int ret= 0;
    while (ivit.get() != NULL)
    {
      ret++;
      ivit.next();
    }
    return ret;
  }
  int get_n_intervals() const
  {
    if (sid_lock != NULL)
      sid_lock->assert_some_wrlock();
    rpl_sidno max_sidno= get_max_sidno();
    int ret= 0;
    for (rpl_sidno sidno= 1; sidno < max_sidno; sidno++)
      ret+= get_n_intervals(sidno);
    return ret;
  }
  enum_return_status create_new_chunk(int size);
  enum_return_status get_free_interval(Interval **out);
  void put_free_interval(Interval *iv);
  void add_interval_memory_lock_taken(int n_ivs, Interval *ivs);

  mutable Checkable_rwlock *sid_lock;
  mysql_mutex_t free_intervals_mutex;
  class Free_intervals_lock
  {
  public:
    Free_intervals_lock(Gtid_set *_gtid_set)
      : gtid_set(_gtid_set), locked(false) {}
    void lock_if_not_locked()
    {
      if (gtid_set->sid_lock && !locked)
      {
        mysql_mutex_lock(&gtid_set->free_intervals_mutex);
        locked= true;
      }
    }
    void unlock_if_locked()
    {
      if (gtid_set->sid_lock && locked)
      {
        mysql_mutex_unlock(&gtid_set->free_intervals_mutex);
        locked= false;
      }
    }
    ~Free_intervals_lock()
    {
      unlock_if_locked();
    }
  private:
    Gtid_set *gtid_set;
    bool locked;
  };
  void assert_free_intervals_locked()
  {
    if (sid_lock != NULL)
      mysql_mutex_assert_owner(&free_intervals_mutex);
  }

  enum_return_status add_gno_interval(Interval_iterator *ivitp,
                                      rpl_gno start, rpl_gno end,
                                      Free_intervals_lock *lock);
  enum_return_status remove_gno_interval(Interval_iterator *ivitp,
                                         rpl_gno start, rpl_gno end,
                                         Free_intervals_lock *lock);
  enum_return_status add_gno_intervals(rpl_sidno sidno,
                                       Const_interval_iterator ivit,
                                       Free_intervals_lock *lock);
  enum_return_status remove_gno_intervals(rpl_sidno sidno,
                                          Const_interval_iterator ivit,
                                          Free_intervals_lock *lock);

  static bool is_interval_subset(Const_interval_iterator *sub,
                                 Const_interval_iterator *super);
  static bool is_interval_intersection_nonempty(Const_interval_iterator *ivit1,
                                                Const_interval_iterator *ivit2);

  Sid_map *sid_map;
  DYNAMIC_ARRAY intervals;
  Interval *free_intervals;
  Interval_chunk *chunks;
  mutable int cached_string_length;
  mutable const String_format *cached_string_format;
#ifndef DBUG_OFF

  int n_chunks;
#endif

#ifdef FRIEND_OF_GTID_SET
  friend FRIEND_OF_GTID_SET;
#endif

  friend class Gtid_set::Free_intervals_lock;
};


struct Gtid_set_or_null
{
  Gtid_set *gtid_set;
  bool is_non_null;
  inline Gtid_set *get_gtid_set() const
  {
    DBUG_ASSERT(!(is_non_null && gtid_set == NULL));
    return is_non_null ? gtid_set : NULL;
  }
  Gtid_set *set_non_null(Sid_map *sm)
  {
    if (!is_non_null)
    {
      if (gtid_set == NULL)
        gtid_set= new Gtid_set(sm);
      else
        gtid_set->clear();
    }
    is_non_null= (gtid_set != NULL);
    return gtid_set;
  }
  inline void set_null() { is_non_null= false; }
};


class Owned_gtids
{
public:
  Owned_gtids(Checkable_rwlock *sid_lock);
  ~Owned_gtids();
  enum_return_status add_gtid_owner(const Gtid &gtid, my_thread_id owner);
  my_thread_id get_owner(const Gtid &gtid) const;
  void remove_gtid(const Gtid &gtid);
  enum_return_status ensure_sidno(rpl_sidno sidno);
  bool is_intersection_nonempty(const Gtid_set *other) const;
  bool is_empty() const
  {
    Gtid_iterator git(this);
    return git.get().sidno == 0;
  }
  rpl_sidno get_max_sidno() const
  {
    sid_lock->assert_some_lock();
    return sidno_to_hash.elements;
  }

  int to_string(char *out) const
  {
    char *p= out;
    rpl_sidno max_sidno= get_max_sidno();
    rpl_sidno sid_map_max_sidno= global_sid_map->get_max_sidno();
    for (rpl_sidno sid_i= 0; sid_i < sid_map_max_sidno; sid_i++)
    {
      rpl_sidno sidno= global_sid_map->get_sorted_sidno(sid_i);
      if (sidno > max_sidno)
        continue;
      HASH *hash= get_hash(sidno);
      bool printed_sid= false;
      for (uint i= 0; i < hash->records; i++)
      {
        Node *node= (Node *)my_hash_element(hash, i);
        DBUG_ASSERT(node != NULL);
        if (!printed_sid)
        {
          p+= global_sid_map->sidno_to_sid(sidno).to_string(p);
          printed_sid= true;
        }
        p+= sprintf(p, ":%lld#%lu", node->gno, node->owner);
      }
    }
    *p= 0;
    return (int)(p - out);
  }

  size_t get_max_string_length() const
  {
    rpl_sidno max_sidno= get_max_sidno();
    size_t ret= 0;
    for (rpl_sidno sidno= 1; sidno <= max_sidno; sidno++)
    {
      HASH *hash= get_hash(sidno);
      if (hash->records > 0)
        ret+= rpl_sid::TEXT_LENGTH +
          hash->records * (1 + MAX_GNO_TEXT_LENGTH +
                           1 + MAX_THREAD_ID_TEXT_LENGTH);
    }
    return 1 + ret;
  }

  bool thread_owns_anything(my_thread_id thd_id) const
  {
    Gtid_iterator git(this);
    Node *node= git.get_node();
    while (node != NULL)
    {
      if (node->owner == thd_id)
        return true;
      git.next();
      node= git.get_node();
    }
    return false;
  }

#ifndef DBUG_OFF

  char *to_string() const
  {
    char *str= (char *)my_malloc(get_max_string_length(), MYF(MY_WME));
    DBUG_ASSERT(str != NULL);
    to_string(str);
    return str;
  }
  void print() const
  {
    char *str= to_string();
    printf("%s\n", str);
    my_free(str);
  }
#endif

  void dbug_print(const char *text= "") const
  {
#ifndef DBUG_OFF
    char *str= to_string();
    DBUG_PRINT("info", ("%s%s%s", text, *text ? ": " : "", str));
    my_free(str);
#endif
  }
private:
  struct Node
  {
    rpl_gno gno;
    my_thread_id owner;
  };
  mutable Checkable_rwlock *sid_lock;
  HASH *get_hash(rpl_sidno sidno) const
  {
    DBUG_ASSERT(sidno >= 1 && sidno <= get_max_sidno());
    sid_lock->assert_some_lock();
    return *dynamic_element(&sidno_to_hash, sidno - 1, HASH **);
  }
  Node *get_node(const HASH *hash, rpl_gno gno) const
  {
    sid_lock->assert_some_lock();
    return (Node *)my_hash_search(hash, (const uchar *)&gno, sizeof(rpl_gno));
  }
  Node *get_node(const Gtid &gtid) const
  { return get_node(get_hash(gtid.sidno), gtid.gno); };
  bool contains_gtid(const Gtid &gtid) const { return get_node(gtid) != NULL; }
  DYNAMIC_ARRAY sidno_to_hash;

public:
  class Gtid_iterator
  {
  public:
    Gtid_iterator(const Owned_gtids* og)
      : owned_gtids(og), sidno(1), hash(NULL), node_index(0), node(NULL)
    {
      max_sidno= owned_gtids->get_max_sidno();
      if (sidno <= max_sidno)
        hash= owned_gtids->get_hash(sidno);
      next();
    }
    inline void next()
    {
#ifndef DBUG_OFF
      if (owned_gtids->sid_lock)
        owned_gtids->sid_lock->assert_some_wrlock();
#endif

      while (sidno <= max_sidno)
      {
        DBUG_ASSERT(hash != NULL);
        if (node_index < hash->records)
        {
          node= (Node *)my_hash_element(hash, node_index);
          DBUG_ASSERT(node != NULL);
          // Jump to next node on next iteration.
          node_index++;
          return;
        }

        node_index= 0;
        // hash is initialized on constructor or in previous iteration
        // for current SIDNO, so we must increment for next iteration.
        sidno++;
        if (sidno <= max_sidno)
          hash= owned_gtids->get_hash(sidno);
      }
      node= NULL;
    }
    inline Gtid get() const
    {
      Gtid ret= { 0, 0 };
      if (node)
      {
        ret.sidno= sidno;
        ret.gno= node->gno;
      }
      return ret;
    }
    inline Node* get_node() const
    {
      return node;
    }
  private:
    const Owned_gtids *owned_gtids;
    rpl_sidno sidno;
    rpl_sidno max_sidno;
    HASH *hash;
    uint node_index;
    Node *node;
  };
};


class Gtid_state
{
public:
  Gtid_state(Checkable_rwlock *_sid_lock, Sid_map *_sid_map)
    : sid_lock(_sid_lock),
    sid_map(_sid_map),
    sid_locks(sid_lock),
    logged_gtids(sid_map, sid_lock),
    lost_gtids(sid_map, sid_lock),
    owned_gtids(sid_lock) {}
  int init();
  void clear();
  bool is_logged(const Gtid &gtid) const
  {
    DBUG_ENTER("Gtid_state::is_logged");
    bool ret= logged_gtids.contains_gtid(gtid);
    DBUG_RETURN(ret);
  }
  my_thread_id get_owner(const Gtid &gtid) const
  { return owned_gtids.get_owner(gtid); }
#ifndef MYSQL_CLIENT

  enum_return_status acquire_ownership(THD *thd, const Gtid &gtid);
  enum_return_status update_on_flush(THD *thd);
  void update_on_commit(THD *thd);
  void update_on_rollback(THD *thd);
#endif // ifndef MYSQL_CLIENT

  rpl_gno get_automatic_gno(rpl_sidno sidno) const;
  void lock_sidno(rpl_sidno sidno) { sid_locks.lock(sidno); }
  void unlock_sidno(rpl_sidno sidno) { sid_locks.unlock(sidno); }
  void broadcast_sidno(rpl_sidno sidno) { sid_locks.broadcast(sidno); }
  void assert_sidno_lock_owner(rpl_sidno sidno)
  { sid_locks.assert_owner(sidno); }
#ifndef MYSQL_CLIENT

  void wait_for_gtid(THD *thd, const Gtid &gtid);
#endif // ifndef MYSQL_CLIENT
#ifdef HAVE_NDB_BINLOG

  void lock_sidnos(const Gtid_set *set);
  void unlock_sidnos(const Gtid_set *set);
  void broadcast_sidnos(const Gtid_set *set);
#endif // ifdef HAVE_NDB_BINLOG

  enum_return_status ensure_sidno();
  enum_return_status add_lost_gtids(const char *text);
  const Gtid_set *get_logged_gtids() const { return &logged_gtids; }
  const Gtid_set *get_lost_gtids() const { return &lost_gtids; }
  const Owned_gtids *get_owned_gtids() const { return &owned_gtids; }
  rpl_sidno get_server_sidno() const { return server_sidno; }
#ifndef DBUG_OFF

  size_t get_max_string_length() const
  {
    return owned_gtids.get_max_string_length() +
      logged_gtids.get_string_length() +
      lost_gtids.get_string_length() +
      100;
  }
  int to_string(char *buf) const
  {
    char *p= buf;
    p+= sprintf(p, "Logged GTIDs:\n");
    p+= logged_gtids.to_string(p);
    p+= sprintf(p, "\nOwned GTIDs:\n");
    p+= owned_gtids.to_string(p);
    p+= sprintf(p, "\nLost GTIDs:\n");
    p+= lost_gtids.to_string(p);
    return (int)(p - buf);
  }
  char *to_string() const
  {
    char *str= (char *)my_malloc(get_max_string_length(), MYF(MY_WME));
    to_string(str);
    return str;
  }
  void print() const
  {
    char *str= to_string();
    printf("%s", str);
    my_free(str);
  }
#endif

  void dbug_print(const char *text= "") const
  {
#ifndef DBUG_OFF
    sid_lock->assert_some_wrlock();
    char *str= to_string();
    DBUG_PRINT("info", ("%s%s%s", text, *text ? ": " : "", str));
    my_free(str);
#endif
  }
private:
#ifdef HAVE_NDB_BINLOG

  void lock_owned_sidnos(const THD *thd);
#endif

  void unlock_owned_sidnos(const THD *thd);
  void broadcast_owned_sidnos(const THD *thd);
  void update_owned_gtids_impl(THD *thd, bool is_commit);


  mutable Checkable_rwlock *sid_lock;
  mutable Sid_map *sid_map;
  Mutex_cond_array sid_locks;
  Gtid_set logged_gtids;
  Gtid_set lost_gtids;
  Owned_gtids owned_gtids;
  rpl_sidno server_sidno;

#ifdef FRIEND_OF_GTID_STATE
  friend FRIEND_OF_GTID_STATE;
#endif
};


extern Gtid_state *gtid_state;


enum enum_group_type
{
  AUTOMATIC_GROUP= 0, GTID_GROUP, ANONYMOUS_GROUP, INVALID_GROUP, UNDEFINED_GROUP
};


struct Gtid_specification
{
  enum_group_type type;
  Gtid gtid;
  void set(rpl_sidno sidno, rpl_gno gno)
  { type= GTID_GROUP; gtid.sidno= sidno; gtid.gno= gno; }
  void set(const Gtid &gtid_param) { set(gtid_param.sidno, gtid_param.gno); }
  void set_automatic()
  {
    type= AUTOMATIC_GROUP;
  }
  void set_undefined()
  {
    if (type == GTID_GROUP)
      type= UNDEFINED_GROUP;
  }
  void clear() { set(0, 0); }
  bool equals(const Gtid_specification &other) const
  {
    return (type == other.type &&
            (type != GTID_GROUP || gtid.equals(other.gtid)));
  }
  bool equals(const Gtid &other_gtid) const
  { return type == GTID_GROUP && gtid.equals(other_gtid); }
#ifndef MYSQL_CLIENT

  enum_return_status parse(Sid_map *sid_map, const char *text);
  static enum_group_type get_type(const char *text);
  static bool is_valid(const char *text)
  { return Gtid_specification::get_type(text) != INVALID_GROUP; }
#endif
  static const int MAX_TEXT_LENGTH= Uuid::TEXT_LENGTH + 1 + MAX_GNO_TEXT_LENGTH;
  int to_string(const Sid_map *sid_map, char *buf) const;
  int to_string(const rpl_sid *sid, char *buf) const;
#ifndef DBUG_OFF

  void print() const
  {
    char buf[MAX_TEXT_LENGTH + 1];
    to_string(global_sid_map, buf);
    printf("%s\n", buf);
  }
#endif

  void dbug_print(const char *text= "") const
  {
#ifndef DBUG_OFF
    char buf[MAX_TEXT_LENGTH + 1];
    to_string(global_sid_map, buf);
    DBUG_PRINT("info", ("%s%s%s", text, *text ? ": " : "", buf));
#endif
  }
};


struct Cached_group
{
  Gtid_specification spec;
  rpl_binlog_pos binlog_offset;
};


class Group_cache
{
public:
  Group_cache();
  ~Group_cache();
  void clear();
  inline int get_n_groups() const { return groups.elements; }
  inline bool is_empty() const { return get_n_groups() == 0; }
  enum enum_add_group_status
  {
    EXTEND_EXISTING_GROUP, APPEND_NEW_GROUP, ERROR
  };
#ifndef MYSQL_CLIENT
  enum_add_group_status
    add_logged_group(const THD *thd, my_off_t binlog_offset);
#endif // ifndef MYSQL_CLIENT
#ifdef NON_DISABLED_GTID

  enum_add_group_status add_empty_group(const Gtid &gtid);
#endif // ifdef NON_DISABLED_GTID
#ifndef MYSQL_CLIENT

  enum_return_status write_to_gtid_state() const;
  enum_return_status generate_automatic_gno(THD *thd);
#endif // ifndef MYSQL_CLIENT

  bool contains_gtid(const Gtid &gtid) const;
  enum_return_status get_gtids(Gtid_set *gs) const;

#ifndef DBUG_OFF

  size_t to_string(const Sid_map *sm, char *buf) const
  {
    int n_groups= get_n_groups();
    char *s= buf;

    s += sprintf(s, "%d groups = {\n", n_groups);
    for (int i= 0; i < n_groups; i++)
    {
      Cached_group *group= get_unsafe_pointer(i);
      char uuid[Uuid::TEXT_LENGTH + 1]= "[]";
      if (group->spec.gtid.sidno)
        sm->sidno_to_sid(group->spec.gtid.sidno).to_string(uuid);
      s += sprintf(s, "  %s:%lld [offset %lld] %s\n",
                   uuid, group->spec.gtid.gno, group->binlog_offset,
                   group->spec.type == GTID_GROUP ? "GTID" :
                   group->spec.type == ANONYMOUS_GROUP ? "ANONYMOUS" :
                   group->spec.type == AUTOMATIC_GROUP ? "AUTOMATIC" :
                   "INVALID-GROUP-TYPE");
    }
    sprintf(s, "}\n");
    return s - buf;
  }
  size_t get_max_string_length() const
  {
    return (2 + Uuid::TEXT_LENGTH + 1 + MAX_GNO_TEXT_LENGTH + 4 + 2 +
            40 + 10 + 21 + 1 + 100/*margin*/) * get_n_groups() + 100/*margin*/;
  }
  char *to_string(const Sid_map *sm) const
  {
    char *str= (char *)my_malloc(get_max_string_length(), MYF(MY_WME));
    if (str)
      to_string(sm, str);
    return str;
  }
  void print(const Sid_map *sm) const
  {
    char *str= to_string(sm);
    printf("%s\n", str);
    my_free(str);
  }
#endif

  void dbug_print(const Sid_map *sid_map, const char *text= "") const
  {
#ifndef DBUG_OFF
    char *str= to_string(sid_map);
    DBUG_PRINT("info", ("%s%s%s", text, *text ? ": " : "", str));
    my_free(str);
#endif
  }

  inline Cached_group *get_unsafe_pointer(int index) const
  {
    DBUG_ASSERT(index >= 0 && index < get_n_groups());
    return dynamic_element(&groups, index, Cached_group *);
  }

private:
  DYNAMIC_ARRAY groups;

  Cached_group *get_last_group()
  {
    int n_groups= get_n_groups();
    return n_groups == 0 ? NULL : get_unsafe_pointer(n_groups - 1);
  }

  Cached_group *allocate_group()
  {
    Cached_group *ret= (Cached_group *)alloc_dynamic(&groups);
    if (ret == NULL)
      BINLOG_ERROR(("Out of memory."), (ER_OUT_OF_RESOURCES, MYF(0)));
    return ret;
  }

  enum_return_status add_group(const Cached_group *group);
  enum_return_status
    write_to_log_prepare(Group_cache *trx_group_cache,
                         rpl_binlog_pos offset_after_last_statement,
                         Cached_group **last_non_empty_group);

#ifdef FRIEND_OF_GROUP_CACHE
  friend FRIEND_OF_GROUP_CACHE;
#endif
};

enum enum_gtid_statement_status
{
  GTID_STATEMENT_EXECUTE,
  GTID_STATEMENT_CANCEL,
  GTID_STATEMENT_SKIP
};


#ifndef MYSQL_CLIENT

enum_gtid_statement_status
gtid_before_statement(THD *thd, Group_cache *gsc, Group_cache *gtc);

enum_gtid_statement_status gtid_pre_statement_checks(const THD *thd);

void gtid_post_statement_checks(THD *thd);

int gtid_rollback(THD *thd);

int gtid_acquire_ownership_single(THD *thd);
#ifdef HAVE_NDB_BINLOG
int gtid_acquire_ownership_multiple(THD *thd);
#endif

#endif // ifndef MYSQL_CLIENT

#endif /* RPL_GTID_H_INCLUDED */