From 3f619478f796eddbba6e39502fe941b285dd97b1 Mon Sep 17 00:00:00 2001
From: Daniel Baumann <daniel.baumann@progress-linux.org>
Date: Sat, 4 May 2024 20:00:34 +0200
Subject: Adding upstream version 1:10.11.6.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
---
 sql/rpl_parallel.h | 479 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 479 insertions(+)
 create mode 100644 sql/rpl_parallel.h

(limited to 'sql/rpl_parallel.h')

diff --git a/sql/rpl_parallel.h b/sql/rpl_parallel.h
new file mode 100644
index 00000000..a9cfefcb
--- /dev/null
+++ b/sql/rpl_parallel.h
@@ -0,0 +1,479 @@
+#ifndef RPL_PARALLEL_H
+#define RPL_PARALLEL_H
+
+#include "log_event.h"
+
+
+struct rpl_parallel;
+struct rpl_parallel_entry;
+struct rpl_parallel_thread_pool;
+extern struct rpl_parallel_thread_pool pool_bkp_for_pfs;
+
+class Relay_log_info;
+struct inuse_relaylog;
+
+
+/*
+  Structure used to keep track of the parallel replication of a batch of
+  event-groups that group-committed together on the master.
+
+  It is used to ensure that every event group in one batch has reached the
+  commit stage before the next batch starts executing.
+
+  Note the lifetime of this structure:
+
+   - It is allocated when the first event in a new batch of group commits
+     is queued, from the free list rpl_parallel_entry::gco_free_list.
+
+   - The gco for the batch currently being queued is owned by
+     rpl_parallel_entry::current_gco. The gco for a previous batch that has
+     been fully queued is owned by the gco->prev_gco pointer of the gco for
+     the following batch.
+
+   - The worker thread waits on gco->COND_group_commit_orderer for
+     rpl_parallel_entry::count_committing_event_groups to reach wait_count
+     before starting; the first waiter links the gco into the next_gco
+     pointer of the gco of the previous batch for signalling.
+
+   - When an event group reaches the commit stage, it signals the
+     COND_group_commit_orderer if its gco->next_gco pointer is non-NULL and
+     rpl_parallel_entry::count_committing_event_groups has reached
+     gco->next_gco->wait_count.
+
+   - The gco lives until all its event groups have completed their commit.
+     This is detected by rpl_parallel_entry::last_committed_sub_id being
+     greater than or equal gco->last_sub_id. Once this happens, the gco is
+     freed. Note that since update of last_committed_sub_id can happen
+     out-of-order, the thread that frees a given gco can be for any later
+     event group, not necessarily an event group from the gco being freed.
+*/
+struct group_commit_orderer {
+  /* Wakeup condition, used with rpl_parallel_entry::LOCK_parallel_entry. */
+  mysql_cond_t COND_group_commit_orderer;
+  uint64 wait_count;
+  group_commit_orderer *prev_gco;
+  group_commit_orderer *next_gco;
+  /*
+    The sub_id of last event group in the previous GCO.
+    Only valid if prev_gco != NULL.
+  */
+  uint64 prior_sub_id;
+  /*
+    The sub_id of the last event group in this GCO. Only valid when next_gco
+    is non-NULL.
+  */
+  uint64 last_sub_id;
+  /*
+    This flag is set when this GCO has been installed into the next_gco pointer
+    of the previous GCO.
+  */
+  bool installed;
+
+  enum force_switch_bits
+  {
+    /*
+      This flag is set for a GCO in which we have event groups with multiple
+      different commit_id values from the master. This happens when we
+      optimistically try to execute in parallel transactions not known to be
+      conflict-free.
+
+      When this flag is set, in case of DDL we need to start a new GCO
+      regardless of current commit_id, as DDL is not safe to
+      speculatively apply in parallel with prior event groups.
+    */
+    MULTI_BATCH= 1,
+    /*
+      This flag is set for a GCO that contains DDL. If set, it forces
+      a switch to a new GCO upon seeing a new commit_id, as DDL is not
+      safe to speculatively replicate in parallel with subsequent
+      transactions.
+    */
+    FORCE_SWITCH= 2
+  };
+  uint8 flags;
+#ifndef DBUG_OFF
+  /*
+    Flag set when the GCO has been freed and entered the free list, to catch
+    (in debug) errors in the complex lifetime of this object.
+  */
+  bool gc_done;
+#endif
+};
+
+
+struct rpl_parallel_thread {
+  bool delay_start;
+  bool running;
+  bool stop;
+  bool pause_for_ftwrl;
+  /*
+    0  = No start alter assigned
+   >0 = Start alter assigned
+  */
+  uint64 current_start_alter_id;
+  uint32 current_start_alter_domain_id;
+  /*
+   This flag is true when Start Alter just needs to be binlogged only.
+   This scenario will happens when there is congestion , and we can not
+   allocate independent worker to start alter.
+  */
+  bool reserved_start_alter_thread;
+  mysql_mutex_t LOCK_rpl_thread;
+  mysql_cond_t COND_rpl_thread;
+  mysql_cond_t COND_rpl_thread_queue;
+  mysql_cond_t COND_rpl_thread_stop;
+  struct rpl_parallel_thread *next;             /* For free list. */
+  struct rpl_parallel_thread_pool *pool;
+  THD *thd;
+  /*
+    Who owns the thread, if any (it's a pointer into the
+    rpl_parallel_entry::rpl_threads array.
+  */
+  struct rpl_parallel_thread **current_owner;
+  /* The rpl_parallel_entry of the owner. */
+  rpl_parallel_entry *current_entry;
+  struct queued_event {
+    queued_event *next;
+    /*
+      queued_event can hold either an event to be executed, or just a binlog
+      position to be updated without any associated event.
+    */
+    enum queued_event_t {
+      QUEUED_EVENT,
+      QUEUED_POS_UPDATE,
+      QUEUED_MASTER_RESTART
+    } typ;
+    union {
+      Log_event *ev;                            /* QUEUED_EVENT */
+      rpl_parallel_entry *entry_for_queued;     /* QUEUED_POS_UPDATE and
+                                                   QUEUED_MASTER_RESTART */
+    };
+    rpl_group_info *rgi;
+    inuse_relaylog *ir;
+    ulonglong future_event_relay_log_pos;
+    char event_relay_log_name[FN_REFLEN];
+    char future_event_master_log_name[FN_REFLEN];
+    ulonglong event_relay_log_pos;
+    my_off_t future_event_master_log_pos;
+    size_t event_size;
+  } *event_queue, *last_in_queue;
+  uint64 queued_size;
+  /* These free lists are protected by LOCK_rpl_thread. */
+  queued_event *qev_free_list;
+  rpl_group_info *rgi_free_list;
+  group_commit_orderer *gco_free_list;
+  /*
+    These free lists are local to the thread, so need not be protected by any
+    lock. They are moved to the global free lists in batches in the function
+    batch_free(), to reduce LOCK_rpl_thread contention.
+
+    The lists are not NULL-terminated (as we do not need to traverse them).
+    Instead, if they are non-NULL, the loc_XXX_last_ptr_ptr points to the
+    `next' pointer of the last element, which is used to link into the front
+    of the global freelists.
+  */
+  queued_event *loc_qev_list, **loc_qev_last_ptr_ptr;
+  size_t loc_qev_size;
+  uint64 qev_free_pending;
+  rpl_group_info *loc_rgi_list, **loc_rgi_last_ptr_ptr;
+  group_commit_orderer *loc_gco_list, **loc_gco_last_ptr_ptr;
+  /* These keep track of batch update of inuse_relaylog refcounts. */
+  inuse_relaylog *accumulated_ir_last;
+  uint64 accumulated_ir_count;
+
+  char channel_name[MAX_CONNECTION_NAME];
+  uint channel_name_length;
+  rpl_gtid last_seen_gtid;
+  int last_error_number;
+  char last_error_message[MAX_SLAVE_ERRMSG];
+  ulonglong last_error_timestamp;
+  ulonglong worker_idle_time;
+  ulong last_trans_retry_count;
+  ulonglong start_time;
+  void start_time_tracker()
+  {
+    start_time= microsecond_interval_timer();
+  }
+  ulonglong compute_time_lapsed()
+  {
+    return (ulonglong)((microsecond_interval_timer() - start_time) / 1000000.0);
+  }
+  void add_to_worker_idle_time_and_reset()
+  {
+    worker_idle_time+= compute_time_lapsed();
+    start_time=0;
+  }
+  ulonglong get_worker_idle_time()
+  {
+    if (start_time)
+      return (worker_idle_time + compute_time_lapsed());
+    else
+      return worker_idle_time;
+  }
+  void enqueue(queued_event *qev)
+  {
+    if (last_in_queue)
+      last_in_queue->next= qev;
+    else
+      event_queue= qev;
+    last_in_queue= qev;
+    queued_size+= qev->event_size;
+  }
+
+  void dequeue1(queued_event *list)
+  {
+    DBUG_ASSERT(list == event_queue);
+    event_queue= last_in_queue= NULL;
+  }
+
+  void dequeue2(size_t dequeue_size)
+  {
+    queued_size-= dequeue_size;
+  }
+
+  queued_event *get_qev_common(Log_event *ev, ulonglong event_size);
+  queued_event *get_qev(Log_event *ev, ulonglong event_size,
+                        Relay_log_info *rli);
+  queued_event *retry_get_qev(Log_event *ev, queued_event *orig_qev,
+                              const char *relay_log_name,
+                              ulonglong event_pos, ulonglong event_size);
+  /*
+    Put a qev on the local free list, to be later released to the global free
+    list by batch_free().
+  */
+  void loc_free_qev(queued_event *qev);
+  /*
+    Release an rgi immediately to the global free list. Requires holding the
+    LOCK_rpl_thread mutex.
+  */
+  void free_qev(queued_event *qev);
+  rpl_group_info *get_rgi(Relay_log_info *rli, Gtid_log_event *gtid_ev,
+                          rpl_parallel_entry *e, ulonglong event_size);
+  /*
+    Put an gco on the local free list, to be later released to the global free
+    list by batch_free().
+  */
+  void loc_free_rgi(rpl_group_info *rgi);
+  /*
+    Release an rgi immediately to the global free list. Requires holding the
+    LOCK_rpl_thread mutex.
+  */
+  void free_rgi(rpl_group_info *rgi);
+  group_commit_orderer *get_gco(uint64 wait_count, group_commit_orderer *prev,
+                                uint64 first_sub_id);
+  /*
+    Put a gco on the local free list, to be later released to the global free
+    list by batch_free().
+  */
+  void loc_free_gco(group_commit_orderer *gco);
+  /*
+    Move all local free lists to the global ones. Requires holding
+    LOCK_rpl_thread.
+  */
+  void batch_free();
+  /* Update inuse_relaylog refcounts with what we have accumulated so far. */
+  void inuse_relaylog_refcount_update();
+  rpl_parallel_thread();
+};
+
+
+struct pool_bkp_for_pfs{
+  uint32 count;
+  bool inited, is_valid;
+  struct rpl_parallel_thread **rpl_thread_arr;
+  void init(uint32 thd_count)
+  {
+    DBUG_ASSERT(thd_count);
+    rpl_thread_arr= (rpl_parallel_thread **)
+                      my_malloc(PSI_INSTRUMENT_ME,
+                                thd_count * sizeof(rpl_parallel_thread*),
+                                MYF(MY_WME | MY_ZEROFILL));
+    for (uint i=0; i<thd_count; i++)
+      rpl_thread_arr[i]= (rpl_parallel_thread *)
+                          my_malloc(PSI_INSTRUMENT_ME, sizeof(rpl_parallel_thread),
+                                    MYF(MY_WME | MY_ZEROFILL));
+    count= thd_count;
+    inited= true;
+  }
+
+  void destroy()
+  {
+    if (inited)
+    {
+      for (uint i=0; i<count; i++)
+        my_free(rpl_thread_arr[i]);
+
+      my_free(rpl_thread_arr);
+      rpl_thread_arr= NULL;
+    }
+    inited= false;
+  }
+};
+
+struct rpl_parallel_thread_pool {
+  struct rpl_parallel_thread **threads;
+  struct rpl_parallel_thread *free_list;
+  mysql_mutex_t LOCK_rpl_thread_pool;
+  mysql_cond_t COND_rpl_thread_pool;
+  uint32 count;
+  bool inited;
+
+  /*
+    Lock first LOCK_rpl_thread_pool and then LOCK_rpl_thread to
+    update this variable.
+  */
+  uint32 current_start_alters;
+  /*
+    While FTWRL runs, this counter is incremented to make SQL thread or
+    STOP/START slave not try to start new activity while that operation
+    is in progress.
+  */
+  bool busy;
+  struct pool_bkp_for_pfs pfs_bkp;
+
+  rpl_parallel_thread_pool();
+  void copy_pool_for_pfs(Relay_log_info *rli);
+  int init(uint32 size);
+  void destroy();
+  void deactivate();
+  void destroy_cond_mutex();
+  struct rpl_parallel_thread *get_thread(rpl_parallel_thread **owner,
+                                         rpl_parallel_entry *entry);
+  void release_thread(rpl_parallel_thread *rpt);
+};
+
+
+struct rpl_parallel_entry {
+  mysql_mutex_t LOCK_parallel_entry;
+  mysql_cond_t COND_parallel_entry;
+  uint32 domain_id;
+  /*
+    Incremented by wait_for_workers_idle() and rpl_pause_for_ftwrl() to show
+    that they are waiting, so that finish_event_group knows to signal them
+    when last_committed_sub_id is increased.
+  */
+  uint32 need_sub_id_signal;
+  uint64 last_commit_id;
+  uint32 pending_start_alters;
+  bool active;
+  /*
+    Set when SQL thread is shutting down, and no more events can be processed,
+    so worker threads must force abort any current transactions without
+    waiting for event groups to complete.
+  */
+  bool force_abort;
+  /*
+   At STOP SLAVE (force_abort=true), we do not want to process all events in
+   the queue (which could unnecessarily delay stop, if a lot of events happen
+   to be queued). The stop_sub_id provides a safe point at which to stop, so
+   that everything before becomes committed and nothing after does. The value
+   corresponds to rpl_group_info::gtid_sub_id; if that is less than or equal
+   to stop_sub_id, we execute the associated event group, else we skip it (and
+   all following) and stop.
+  */
+  uint64 stop_sub_id;
+
+  /*
+    Cyclic array recording the last rpl_thread_max worker threads that we
+    queued event for. This is used to limit how many workers a single domain
+    can occupy (--slave-domain-parallel-threads).
+
+    Note that workers are never explicitly deleted from the array. Instead,
+    we need to check (under LOCK_rpl_thread) that the thread still belongs
+    to us before re-using (rpl_thread::current_owner).
+  */
+  rpl_parallel_thread **rpl_threads;
+  uint32 rpl_thread_max;
+  uint32 rpl_thread_idx;
+  /*
+    The sub_id of the last transaction to commit within this domain_id.
+    Must be accessed under LOCK_parallel_entry protection.
+
+    Event groups commit in order, so the rpl_group_info for an event group
+    will be alive (at least) as long as
+    rpl_group_info::gtid_sub_id > last_committed_sub_id. This can be used to
+    safely refer back to previous event groups if they are still executing,
+    and ignore them if they completed, without requiring explicit
+    synchronisation between the threads.
+  */
+  uint64 last_committed_sub_id;
+  /*
+    The sub_id of the last event group in this replication domain that was
+    queued for execution by a worker thread.
+  */
+  uint64 current_sub_id;
+  /*
+    The largest sub_id that has started its transaction. Protected by
+    LOCK_parallel_entry.
+
+    (Transactions can start out-of-order, so this value signifies that no
+    transactions with larger sub_id have started, but not necessarily that all
+    transactions with smaller sub_id have started).
+  */
+  uint64 largest_started_sub_id;
+  rpl_group_info *current_group_info;
+  /*
+    If we get an error in some event group, we set the sub_id of that event
+    group here. Then later event groups (with higher sub_id) can know not to
+    try to start (event groups that already started will be rolled back when
+    wait_for_prior_commit() returns error).
+    The value is ULONGLONG_MAX when no error occurred.
+  */
+  uint64 stop_on_error_sub_id;
+  /*
+    During FLUSH TABLES WITH READ LOCK, transactions with sub_id larger than
+    this value must not start, but wait until the global read lock is released.
+    The value is set to ULONGLONG_MAX when no FTWRL is pending.
+  */
+  uint64 pause_sub_id;
+  /* Total count of event groups queued so far. */
+  uint64 count_queued_event_groups;
+  /*
+    Count of event groups that have started (but not necessarily completed)
+    the commit phase. We use this to know when every event group in a previous
+    batch of master group commits have started committing on the slave, so
+    that it is safe to start executing the events in the following batch.
+  */
+  uint64 count_committing_event_groups;
+  /* The group_commit_orderer object for the events currently being queued. */
+  group_commit_orderer *current_gco;
+  /* Relay log info of replication source for this entry. */
+  Relay_log_info *rli;
+
+  rpl_parallel_thread * choose_thread(rpl_group_info *rgi, bool *did_enter_cond,
+                                      PSI_stage_info *old_stage,
+                                      Gtid_log_event *gtid_ev);
+  rpl_parallel_thread *
+  choose_thread_internal(uint idx, bool *did_enter_cond, rpl_group_info *rgi,
+                         PSI_stage_info *old_stage);
+  int queue_master_restart(rpl_group_info *rgi,
+                           Format_description_log_event *fdev);
+};
+struct rpl_parallel {
+  HASH domain_hash;
+  rpl_parallel_entry *current;
+  bool sql_thread_stopping;
+
+  rpl_parallel();
+  ~rpl_parallel();
+  void reset();
+  rpl_parallel_entry *find(uint32 domain_id, Relay_log_info *rli);
+  void wait_for_done(THD *thd, Relay_log_info *rli);
+  void stop_during_until();
+  bool workers_idle();
+  int wait_for_workers_idle(THD *thd);
+  int do_event(rpl_group_info *serial_rgi, Log_event *ev, ulonglong event_size);
+};
+
+
+extern struct rpl_parallel_thread_pool global_rpl_thread_pool;
+
+
+extern int rpl_parallel_resize_pool_if_no_slaves(void);
+extern int rpl_parallel_activate_pool(rpl_parallel_thread_pool *pool);
+extern int rpl_parallel_inactivate_pool(rpl_parallel_thread_pool *pool);
+extern bool process_gtid_for_restart_pos(Relay_log_info *rli, rpl_gtid *gtid);
+extern int rpl_pause_for_ftwrl(THD *thd);
+extern void rpl_unpause_after_ftwrl(THD *thd);
+
+#endif  /* RPL_PARALLEL_H */
-- 
cgit v1.2.3