/*****************************************************************************

Copyright (c) 2018, 2019, 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, version 2.0, as published by the
Free Software Foundation.

This program is also distributed with certain software (including but not
limited to OpenSSL) that is licensed under separate terms, as designated in a
particular file or component or in included license documentation. The authors
of MySQL hereby grant you an additional permission to link the program and
your derivative works with the separately licensed software that they have
included with MySQL.

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, version 2.0,
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

*****************************************************************************/

/** @file include/clone0repl.h
 GTID persistence interface

 *******************************************************/

#ifndef CLONE_REPL_INCLUDE
#define CLONE_REPL_INCLUDE

#include <vector>
#include "clone0monitor.h"
#include "os0thread-create.h"
#include "sql/rpl_gtid.h"
#include "srv0srv.h"
#include "srv0start.h"
#include "trx0sys.h"
#include "lizard0scn.h"

extern bool svr_write_non_innodb_gtids;

class Clone_persist_gtid;

/** Serialized GTID information size */
static const size_t GTID_INFO_SIZE = 64;

/** GTID format version. */
static const uint32_t GTID_VERSION = 1;

/** Serialized GTID */
using Gtid_info = std::array<unsigned char, GTID_INFO_SIZE>;

/** List of GTIDs */
using Gitd_info_list = std::vector<Gtid_info>;

/** GTID descriptor with version information. */
struct Gtid_desc {
  /** If GTID descriptor is set. */
  bool m_is_set;
  /** Serialized GTID information. */
  Gtid_info m_info;
  /* GTID version. */
  uint32_t m_version;
};

/** Persist GTID along with transaction commit */
class Clone_persist_gtid {
 public:
  /** Constructor: start gtid thread */
  Clone_persist_gtid() {
    m_event = os_event_create(0);
    /* No background is created yet. */
    m_thread_active.store(false);
    m_gtid_trx_scn.store(0);
    m_flush_number.store(0);
    m_explicit_request.store(false);
    m_active_number.store(m_flush_number.load() + 1);
    /* We accept GTID even before the background service is started. This
    is needed because we add GTIDs from undo log during recovery. */
    m_active.store(true);
    m_num_gtid_mem.store(0);
    m_flush_in_progress.store(false);
    m_close_thread.store(false);
  }

  /** Destructor: stop gtid thread */
  ~Clone_persist_gtid() {
    ut_ad(!m_thread_active.load());
    stop();
    os_event_destroy(m_event);
  }

  /** Start GTID persistence and background thread.
  @return true, if successful. */
  bool start();

  /* Stop GTID persistence. */
  void stop();

  /* Wait for immediate flush.
  @param[in]	wait		wait for completion
  @param[in]	compress_gtid	request GTID compression.
  @param[in]	early_timeout	don't wait long if flush is blocked.
  @param[in]	cbk		alert callback for long wait. */
  void wait_flush(bool wait, bool compress_gtid, bool early_timeout,
                  Clone_Alert_Func cbk);

  /** Flush immediately if transaction is operating on GTID table.
  @param[in,out]	thd	current session thread */
  void flush_if_implicit_gtid(THD *thd);

  /**@return true, if GTID persistence is active. */
  bool is_active() const { return (m_active.load()); }

  /**@return true, if GTID thread is active. */
  bool is_thread_active() const { return (m_thread_active.load()); }

  /** Get oldest transaction scn for which GTID is not persisted to table.
  Transactions committed after this point should not be purged.
  @return oldest transaction scn. */
  scn_t get_oldest_trx_scn() {
    scn_t ret_scn = m_gtid_trx_scn.load();
    /* Should never be zero. It can be set to max only before
    GTID persister is active and no GTID is persisted. */
    ut_ad(ret_scn > 0 || srv_force_recovery >= SRV_FORCE_NO_UNDO_LOG_SCAN);
    if (ret_scn == lizard::SCN_NULL) {
      ut_ad(!is_thread_active());
      ut_ad(m_num_gtid_mem.load() == 0);
    } else if (m_num_gtid_mem.load() == 0) {
      /* For all transactions that are committed before this function is called
      have their GTID flushed if flush is not in progress. "flush not in
      progress" is sufficient but not necessary condition here. This is mainly
      for cases when there is no GTID and purge doesn't need to wait. */
      if (!m_flush_in_progress.load()) {
        ret_scn = lizard::SCN_NULL;
      }
    }
    return (ret_scn);
  }

  /** Set oldest transaction scn for which GTID is not persisted to table.
  This is set during recovery from persisted value.
  @param[in]	max_trx_scn	transaction scn */
  void set_oldest_trx_scn_recovery(scn_t max_trx_scn) {
    ib::info(ER_IB_CLONE_GTID_PERSIST)
        << "GTID recovery trx_scn: " << max_trx_scn;
    /* Zero is special value. It is from old database without GTID
    persistence. */
    if (max_trx_scn == 0) {
      max_trx_scn = lizard::SCN_NULL;
    }
    m_gtid_trx_scn.store(max_trx_scn);
  }

  /** Get transaction GTID information.
  @param[in,out]	trx		innodb transaction
  @param[out]		gtid_desc	descriptor with serialized GTID */
  void get_gtid_info(trx_t *trx, Gtid_desc &gtid_desc);

  /** Set transaction flag to persist GTID and check if space need to be
  allocated for GTID.
  @param[in,out]	trx		current innodb transaction
  @param[in]		prepare		if operation is Prepare
  @param[in]		rollback	if operation is Rollback
  @return true, if undo space needs to be allocated. */
  bool trx_check_set(trx_t *trx, bool prepare, bool rollback);

  /** Check if current transaction has GTID.
  @param[in]		trx		innodb transaction
  @param[in,out]	thd		session THD
  @param[out]		passed_check	true if transaction is good for GTID
  @return true, if transaction has valid GTID. */
  bool has_gtid(trx_t *trx, THD *&thd, bool &passed_check);

  /** Check if GTID persistence is set
  @param[in]	trx	current innnodb transaction
  @return GTID storage type. */
  trx_undo_t::Gtid_storage persists_gtid(const trx_t *trx);

  /** Set or reset GTID persist flag in THD.
  @param[in,out]	trx	current innnodb transaction
  @param[in]		set	true, if need to set */
  void set_persist_gtid(trx_t *trx, bool set);

  /** Add GTID to in memory list.
  @param[in]	gtid_desc	Descriptor with serialized GTID */
  void add(const Gtid_desc &gtid_desc);

  /** Write GTIDs periodically to disk table. */
  void periodic_write();

  /** Write GTIDs of non Innodb transactions to table. */
  int write_other_gtids();

  /** Disable copy construction */
  Clone_persist_gtid(Clone_persist_gtid const &) = delete;

  /** Disable assignment */
  Clone_persist_gtid &operator=(Clone_persist_gtid const &) = delete;

 private:
  /** Check if GTID needs to persist at XA prepare.
  @param[in]		thd		session THD
  @param[in,out]	trx		current innnodb transaction
  @param[in]		found_gtid	session is owning GTID
  @param[in,out]	alloc		in:transaction checks are passed
                                        out:GTID space need to be allocated
  @return true, if GTID needs to be persisted */
  bool check_gtid_prepare(THD *thd, trx_t *trx, bool found_gtid, bool &alloc);

  /** Check if GTID needs to persist at commit.
  @param[in]		thd		session THD
  @param[in]		found_gtid	session is owning GTID
  @return true, if GTID needs to be persisted */
  bool check_gtid_commit(THD *thd, bool found_gtid);

  /** Check if GTID needs to persist at rollback.
  @param[in]		thd		session THD
  @param[in,out]	trx		current innnodb transaction
  @param[in]		found_gtid	session is owning GTID
  @return true, if GTID needs to be persisted */
  bool check_gtid_rollback(THD *thd, trx_t *trx, bool found_gtid);

  /** Wait for gtid thread to start, finish or flush.
  @param[in]	start		if waiting for start
  @param[in]	flush		wait for immediate flush
  @param[in]	flush_number	wait flush to reach this number
  @param[in]	compress	wait also for compression
  @param[in]	early_timeout	don't wait long if flush is blocked
  @param[in]	cbk		alert callback for long wait
  @return true if successful. */
  bool wait_thread(bool start, bool flush, uint64_t flush_number, bool compress,
                   bool early_timeout, Clone_Alert_Func cbk);

  /** @return current active GTID list */
  Gitd_info_list &get_active_list() {
    ut_ad(trx_sys_mutex_own());
    return (get_list(m_active_number));
  }

  /** @return GTID list by number.
  @param[in]	list_number	list number
  @return GTID list reference. */
  Gitd_info_list &get_list(uint64_t list_number) {
    int list_index = (list_number & static_cast<uint64_t>(1));
    return (m_gtids[list_index]);
  }

  /** Check if we need to skip write or compression based on debug variables.
  @param[in]	compression	check for compression
  @return true, if we should skip. */
  bool debug_skip_write(bool compression);

  /** Request immediate flush of all GTIDs accumulated.
  @param[in]	compress	request compression of GTID table
  @return flush list number to track and wait for flush to complete. */
  uint64_t request_immediate_flush(bool compress) {
    trx_sys_mutex_enter();
    /* We want to flush all GTIDs. */
    uint64_t request_number = m_active_number.load();
    /* If no GTIDs added to active, wait for previous index. */
    if (m_num_gtid_mem.load() == 0) {
      ut_a(request_number > 0);
      --request_number;
    }
    m_flush_request_number = request_number;
    trx_sys_mutex_exit();

    if (compress) {
      m_explicit_request.store(true);
    }
    return (request_number);
  }

  /** Check if flush has finished up to a list number.
  @param[in]	request_number	flush request number
  @return true, if it is already flushed. */
  bool check_flushed(uint64_t request_number) const {
    return (m_flush_number >= request_number);
  }

  /** @return true, iff background needs to flush immediately. */
  bool flush_immediate() const {
    return (m_flush_number < m_flush_request_number || m_explicit_request);
  }

  /** Check if GTID compression is necessary based on threshold.
  @return true, if GTID table needs to be compressed. */
  bool check_compress();

  /** Switch active GTID list. */
  uint64_t switch_active_list() {
    /* Switch active list under transaction system mutex. */
    ut_ad(trx_sys_mutex_own());
    uint64_t flush_number = m_active_number;
    ++m_active_number;
    m_compression_gtid_counter += m_num_gtid_mem;
    m_num_gtid_mem.store(0);
#ifdef UNIV_DEBUG
    /* The new active list must have no elements. */
    auto &active_list = get_active_list();
    ut_ad(active_list.size() == 0);
#endif
    return (flush_number);
  }

  /** Persist GTID to gtid_executed table.
  @param[in]		flush_list_number	list number to flush
  @param[in,out]	table_gtid_set		GTIDs in table during recovery
  @param[in,out]	sid_map			SID map for GTIDs
  @return mysql error code. */
  int write_to_table(uint64_t flush_list_number, Gtid_set &table_gtid_set,
                     Sid_map &sid_map);

  /** Update transaction scn upto which GTIDs are flushed to table.
  @param[in]	new_gtid_trx_scn	GTID transaction scn */
  void update_gtid_trx_scn(scn_t new_gtid_trx_scn);

  /** Write all GTIDs to table and update GTID transaction number.
  @param[in,out]	thd	current session thread */
  void flush_gtids(THD *thd);

 private:
  /** Time threshold to trigger persisting GTID. Insert GTID once per 1k
  transactions or every 100 millisecond. */
  const static uint32_t s_time_threshold_ms = 100;

  /** Threshold for the count for compressing GTID. */
  const static uint32_t s_compression_threshold = 50;

  /** Number of transaction/GTID threshold for writing to disk table. */
  const static int s_gtid_threshold = 1024;

  /** Two lists of GTID. One of them is active where running transactions
  add their GTIDs. Other list is used to persist them to table from time
  to time. */
  Gitd_info_list m_gtids[2];

  /** Number of the current GTID list. Increased when list is switched */
  std::atomic<uint64_t> m_active_number;

  /** Number up to which GTIDs are flushed. Increased when list is flushed.*/
  std::atomic<uint64_t> m_flush_number;

  /** If explicit request to flush is made. */
  std::atomic<bool> m_explicit_request;

  /** Number for which last flush request was made. */
  uint64_t m_flush_request_number{0};

  /** Event for GTID background thread. */
  os_event_t m_event;

  /** Counter to keep track of the number of writes till it reaches
  compression threshold. */
  uint32_t m_compression_counter{0};

  /** Counter to keep number of GTIDs flushed before compression. */
  uint32_t m_compression_gtid_counter{0};

  /* Oldest transaction number for which GTID is not persisted. */
  std::atomic<scn_t> m_gtid_trx_scn;

  /** Number of GTID accumulated in memory */
  std::atomic<int> m_num_gtid_mem;

  /** Flush of GTID is in progress. */
  std::atomic<bool> m_flush_in_progress;

  /** Set to true, when the background thread is asked to exit. */
  std::atomic<bool> m_close_thread;

  /** TRUE, if background thread is active.*/
  std::atomic<bool> m_thread_active;

  /** TRUE, if GTID persistence is active.*/
  std::atomic<bool> m_active;

  unsigned long m_thread_id{0};

 public:
  void add_undo_gtids(const Gtid_desc &gtid_desc);
  void print_undo_gtids();

 private:
  Gitd_info_list m_undo_gtids_list;
};

#endif /* CLONE_REPL_INCLUDE */