summaryrefslogtreecommitdiffstats
path: root/sql/rpl_gtid.cc
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 18:07:14 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 18:07:14 +0000
commita175314c3e5827eb193872241446f2f8f5c9d33c (patch)
treecd3d60ca99ae00829c52a6ca79150a5b6e62528b /sql/rpl_gtid.cc
parentInitial commit. (diff)
downloadmariadb-10.5-a175314c3e5827eb193872241446f2f8f5c9d33c.tar.xz
mariadb-10.5-a175314c3e5827eb193872241446f2f8f5c9d33c.zip
Adding upstream version 1:10.5.12.upstream/1%10.5.12upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r--sql/rpl_gtid.cc2954
1 files changed, 2954 insertions, 0 deletions
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
new file mode 100644
index 00000000..bc06188a
--- /dev/null
+++ b/sql/rpl_gtid.cc
@@ -0,0 +1,2954 @@
+/* Copyright (c) 2013, Kristian Nielsen and MariaDB Services Ab.
+ Copyright (c) 2020, MariaDB Corporation.
+
+ 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-1335 USA */
+
+
+/* Definitions for MariaDB global transaction ID (GTID). */
+
+#include "mariadb.h"
+#include "sql_priv.h"
+#include "unireg.h"
+#include "mariadb.h"
+#include "sql_base.h"
+#include "sql_parse.h"
+#include "key.h"
+#include "rpl_gtid.h"
+#include "rpl_rli.h"
+#include "slave.h"
+#include "log_event.h"
+
+const LEX_CSTRING rpl_gtid_slave_state_table_name=
+ { STRING_WITH_LEN("gtid_slave_pos") };
+
+
+void
+rpl_slave_state::update_state_hash(uint64 sub_id, rpl_gtid *gtid, void *hton,
+ rpl_group_info *rgi)
+{
+ int err;
+ /*
+ Add the gtid to the HASH in the replication slave state.
+
+ We must do this only _after_ commit, so that for parallel replication,
+ there will not be an attempt to delete the corresponding table row before
+ it is even committed.
+ */
+ mysql_mutex_lock(&LOCK_slave_state);
+ err= update(gtid->domain_id, gtid->server_id, sub_id, gtid->seq_no, hton, rgi);
+ mysql_mutex_unlock(&LOCK_slave_state);
+ if (err)
+ {
+ sql_print_warning("Slave: Out of memory during slave state maintenance. "
+ "Some no longer necessary rows in table "
+ "mysql.%s may be left undeleted.",
+ rpl_gtid_slave_state_table_name.str);
+ /*
+ Such failure is not fatal. We will fail to delete the row for this
+ GTID, but it will do no harm and will be removed automatically on next
+ server restart.
+ */
+ }
+}
+
+
+int
+rpl_slave_state::record_and_update_gtid(THD *thd, rpl_group_info *rgi)
+{
+ DBUG_ENTER("rpl_slave_state::record_and_update_gtid");
+
+ /*
+ Update the GTID position, if we have it and did not already update
+ it in a GTID transaction.
+ */
+ if (rgi->gtid_pending)
+ {
+ uint64 sub_id= rgi->gtid_sub_id;
+ void *hton= NULL;
+
+ rgi->gtid_pending= false;
+ if (rgi->gtid_ignore_duplicate_state!=rpl_group_info::GTID_DUPLICATE_IGNORE)
+ {
+ if (record_gtid(thd, &rgi->current_gtid, sub_id, false, false, &hton))
+ DBUG_RETURN(1);
+ update_state_hash(sub_id, &rgi->current_gtid, hton, rgi);
+ }
+ rgi->gtid_ignore_duplicate_state= rpl_group_info::GTID_DUPLICATE_NULL;
+ }
+ DBUG_RETURN(0);
+}
+
+
+/*
+ Check GTID event execution when --gtid-ignore-duplicates.
+
+ The idea with --gtid-ignore-duplicates is that we allow multiple master
+ connections (in multi-source replication) to all receive the same GTIDs and
+ event groups. Only one instance of each is applied; we use the sequence
+ number in the GTID to decide whether a GTID has already been applied.
+
+ So if the seq_no of a GTID (or a higher sequence number) has already been
+ applied, then the event should be skipped. If not then the event should be
+ applied.
+
+ To avoid two master connections tring to apply the same event
+ simultaneously, only one is allowed to work in any given domain at any point
+ in time. The associated Relay_log_info object is called the owner of the
+ domain (and there can be multiple parallel worker threads working in that
+ domain for that Relay_log_info). Any other Relay_log_info/master connection
+ must wait for the domain to become free, or for their GTID to have been
+ applied, before being allowed to proceed.
+
+ Returns:
+ 0 This GTID is already applied, it should be skipped.
+ 1 The GTID is not yet applied; this rli is now the owner, and must apply
+ the event and release the domain afterwards.
+ -1 Error (out of memory to allocate a new element for the domain).
+*/
+int
+rpl_slave_state::check_duplicate_gtid(rpl_gtid *gtid, rpl_group_info *rgi)
+{
+ uint32 domain_id= gtid->domain_id;
+ uint64 seq_no= gtid->seq_no;
+ rpl_slave_state::element *elem;
+ int res;
+ bool did_enter_cond= false;
+ PSI_stage_info old_stage;
+ THD *UNINIT_VAR(thd);
+ Relay_log_info *rli= rgi->rli;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ if (!(elem= get_element(domain_id)))
+ {
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ res= -1;
+ goto err;
+ }
+ /*
+ Note that the elem pointer does not change once inserted in the hash. So
+ we can re-use the pointer without looking it up again in the hash after
+ each lock release and re-take.
+ */
+
+ for (;;)
+ {
+ if (elem->highest_seq_no >= seq_no)
+ {
+ /* This sequence number is already applied, ignore it. */
+ res= 0;
+ rgi->gtid_ignore_duplicate_state= rpl_group_info::GTID_DUPLICATE_IGNORE;
+ break;
+ }
+ if (!elem->owner_rli)
+ {
+ /* The domain became free, grab it and apply the event. */
+ elem->owner_rli= rli;
+ elem->owner_count= 1;
+ rgi->gtid_ignore_duplicate_state= rpl_group_info::GTID_DUPLICATE_OWNER;
+ res= 1;
+ break;
+ }
+ if (elem->owner_rli == rli)
+ {
+ /* Already own this domain, increment reference count and apply event. */
+ ++elem->owner_count;
+ rgi->gtid_ignore_duplicate_state= rpl_group_info::GTID_DUPLICATE_OWNER;
+ res= 1;
+ break;
+ }
+ thd= rgi->thd;
+ if (unlikely(thd->check_killed()))
+ {
+ res= -1;
+ break;
+ }
+ /*
+ Someone else is currently processing this GTID (or an earlier one).
+ Wait for them to complete (or fail), and then check again.
+ */
+ if (!did_enter_cond)
+ {
+ thd->ENTER_COND(&elem->COND_gtid_ignore_duplicates, &LOCK_slave_state,
+ &stage_gtid_wait_other_connection, &old_stage);
+ did_enter_cond= true;
+ }
+ mysql_cond_wait(&elem->COND_gtid_ignore_duplicates,
+ &LOCK_slave_state);
+ }
+
+err:
+ if (did_enter_cond)
+ thd->EXIT_COND(&old_stage);
+ else
+ mysql_mutex_unlock(&LOCK_slave_state);
+ return res;
+}
+
+
+void
+rpl_slave_state::release_domain_owner(rpl_group_info *rgi)
+{
+ element *elem= NULL;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ if (!(elem= get_element(rgi->current_gtid.domain_id)))
+ {
+ /*
+ We cannot really deal with error here, as we are already called in an
+ error handling case (transaction failure and rollback).
+
+ However, get_element() only fails if the element did not exist already
+ and could not be allocated due to out-of-memory - and if it did not
+ exist, then we would not get here in the first place.
+ */
+ mysql_mutex_unlock(&LOCK_slave_state);
+ return;
+ }
+
+ if (rgi->gtid_ignore_duplicate_state == rpl_group_info::GTID_DUPLICATE_OWNER)
+ {
+ uint32 count= elem->owner_count;
+ DBUG_ASSERT(count > 0);
+ DBUG_ASSERT(elem->owner_rli == rgi->rli);
+ --count;
+ elem->owner_count= count;
+ if (count == 0)
+ {
+ elem->owner_rli= NULL;
+ mysql_cond_broadcast(&elem->COND_gtid_ignore_duplicates);
+ }
+ }
+ rgi->gtid_ignore_duplicate_state= rpl_group_info::GTID_DUPLICATE_NULL;
+ mysql_mutex_unlock(&LOCK_slave_state);
+}
+
+
+static void
+rpl_slave_state_free_element(void *arg)
+{
+ struct rpl_slave_state::element *elem= (struct rpl_slave_state::element *)arg;
+ mysql_cond_destroy(&elem->COND_wait_gtid);
+ mysql_cond_destroy(&elem->COND_gtid_ignore_duplicates);
+ my_free(elem);
+}
+
+
+rpl_slave_state::rpl_slave_state()
+ : pending_gtid_count(0), last_sub_id(0), gtid_pos_tables(0), loaded(false)
+{
+ mysql_mutex_init(key_LOCK_slave_state, &LOCK_slave_state,
+ MY_MUTEX_INIT_SLOW);
+ my_hash_init(PSI_INSTRUMENT_ME, &hash, &my_charset_bin, 32, offsetof(element, domain_id),
+ sizeof(uint32), NULL, rpl_slave_state_free_element, HASH_UNIQUE);
+ my_init_dynamic_array(PSI_INSTRUMENT_ME, &gtid_sort_array, sizeof(rpl_gtid),
+ 8, 8, MYF(0));
+}
+
+
+rpl_slave_state::~rpl_slave_state()
+{
+ free_gtid_pos_tables(gtid_pos_tables.load(std::memory_order_relaxed));
+ truncate_hash();
+ my_hash_free(&hash);
+ delete_dynamic(&gtid_sort_array);
+ mysql_mutex_destroy(&LOCK_slave_state);
+}
+
+
+void
+rpl_slave_state::truncate_hash()
+{
+ uint32 i;
+
+ for (i= 0; i < hash.records; ++i)
+ {
+ element *e= (element *)my_hash_element(&hash, i);
+ list_element *l= e->list;
+ list_element *next;
+ while (l)
+ {
+ next= l->next;
+ my_free(l);
+ l= next;
+ }
+ /* The element itself is freed by the hash element free function. */
+ }
+ my_hash_reset(&hash);
+}
+
+
+int
+rpl_slave_state::update(uint32 domain_id, uint32 server_id, uint64 sub_id,
+ uint64 seq_no, void *hton, rpl_group_info *rgi)
+{
+ element *elem= NULL;
+ list_element *list_elem= NULL;
+
+ DBUG_ASSERT(hton || !loaded);
+ if (!(elem= get_element(domain_id)))
+ return 1;
+
+ if (seq_no > elem->highest_seq_no)
+ elem->highest_seq_no= seq_no;
+ if (elem->gtid_waiter && elem->min_wait_seq_no <= seq_no)
+ {
+ /*
+ Someone was waiting in MASTER_GTID_WAIT() for this GTID to appear.
+ Signal (and remove) them. The waiter will handle all the processing
+ of all pending MASTER_GTID_WAIT(), so we do not slow down the
+ replication SQL thread.
+ */
+ mysql_mutex_assert_owner(&LOCK_slave_state);
+ elem->gtid_waiter= NULL;
+ mysql_cond_broadcast(&elem->COND_wait_gtid);
+ }
+
+ if (rgi)
+ {
+ if (rgi->gtid_ignore_duplicate_state==rpl_group_info::GTID_DUPLICATE_OWNER)
+ {
+#ifdef DBUG_ASSERT_EXISTS
+ Relay_log_info *rli= rgi->rli;
+#endif
+ uint32 count= elem->owner_count;
+ DBUG_ASSERT(count > 0);
+ DBUG_ASSERT(elem->owner_rli == rli);
+ --count;
+ elem->owner_count= count;
+ if (count == 0)
+ {
+ elem->owner_rli= NULL;
+ mysql_cond_broadcast(&elem->COND_gtid_ignore_duplicates);
+ }
+ }
+ rgi->gtid_ignore_duplicate_state= rpl_group_info::GTID_DUPLICATE_NULL;
+ }
+
+ if (!(list_elem= (list_element *)my_malloc(PSI_INSTRUMENT_ME,
+ sizeof(*list_elem), MYF(MY_WME))))
+ return 1;
+ list_elem->domain_id= domain_id;
+ list_elem->server_id= server_id;
+ list_elem->sub_id= sub_id;
+ list_elem->seq_no= seq_no;
+ list_elem->hton= hton;
+
+ elem->add(list_elem);
+ if (last_sub_id < sub_id)
+ last_sub_id= sub_id;
+
+#ifdef HAVE_REPLICATION
+ ++pending_gtid_count;
+ if (pending_gtid_count >= opt_gtid_cleanup_batch_size)
+ {
+ pending_gtid_count = 0;
+ slave_background_gtid_pending_delete_request();
+ }
+#endif
+
+ return 0;
+}
+
+
+struct rpl_slave_state::element *
+rpl_slave_state::get_element(uint32 domain_id)
+{
+ struct element *elem;
+
+ elem= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0);
+ if (elem)
+ return elem;
+
+ if (!(elem= (element *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*elem), MYF(MY_WME))))
+ return NULL;
+ elem->list= NULL;
+ elem->domain_id= domain_id;
+ elem->highest_seq_no= 0;
+ elem->gtid_waiter= NULL;
+ elem->owner_rli= NULL;
+ elem->owner_count= 0;
+ mysql_cond_init(key_COND_wait_gtid, &elem->COND_wait_gtid, 0);
+ mysql_cond_init(key_COND_gtid_ignore_duplicates,
+ &elem->COND_gtid_ignore_duplicates, 0);
+ if (my_hash_insert(&hash, (uchar *)elem))
+ {
+ my_free(elem);
+ return NULL;
+ }
+ return elem;
+}
+
+
+int
+rpl_slave_state::put_back_list(list_element *list)
+{
+ element *e= NULL;
+ int err= 0;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ while (list)
+ {
+ list_element *next= list->next;
+
+ if ((!e || e->domain_id != list->domain_id) &&
+ !(e= (element *)my_hash_search(&hash, (const uchar *)&list->domain_id, 0)))
+ {
+ err= 1;
+ goto end;
+ }
+ e->add(list);
+ list= next;
+ }
+
+end:
+ mysql_mutex_unlock(&LOCK_slave_state);
+ return err;
+}
+
+
+int
+rpl_slave_state::truncate_state_table(THD *thd)
+{
+ TABLE_LIST tlist;
+ int err= 0;
+
+ tlist.init_one_table(&MYSQL_SCHEMA_NAME, &rpl_gtid_slave_state_table_name,
+ NULL, TL_WRITE);
+ tlist.mdl_request.set_type(MDL_EXCLUSIVE);
+ if (!(err= open_and_lock_tables(thd, &tlist, FALSE,
+ MYSQL_OPEN_IGNORE_LOGGING_FORMAT)))
+ {
+ DBUG_ASSERT(!tlist.table->file->row_logging);
+ tlist.table->s->tdc->flush(thd, true);
+ err= tlist.table->file->ha_truncate();
+
+ if (err)
+ {
+ ha_rollback_trans(thd, FALSE);
+ close_thread_tables(thd);
+ ha_rollback_trans(thd, TRUE);
+ }
+ else
+ {
+ ha_commit_trans(thd, FALSE);
+ close_thread_tables(thd);
+ ha_commit_trans(thd, TRUE);
+ }
+ thd->release_transactional_locks();
+ }
+ return err;
+}
+
+
+static const TABLE_FIELD_TYPE mysql_rpl_slave_state_coltypes[4]= {
+ { { STRING_WITH_LEN("domain_id") },
+ { STRING_WITH_LEN("int(10) unsigned") },
+ {NULL, 0} },
+ { { STRING_WITH_LEN("sub_id") },
+ { STRING_WITH_LEN("bigint(20) unsigned") },
+ {NULL, 0} },
+ { { STRING_WITH_LEN("server_id") },
+ { STRING_WITH_LEN("int(10) unsigned") },
+ {NULL, 0} },
+ { { STRING_WITH_LEN("seq_no") },
+ { STRING_WITH_LEN("bigint(20) unsigned") },
+ {NULL, 0} },
+};
+
+static const uint mysql_rpl_slave_state_pk_parts[]= {0, 1};
+
+static const TABLE_FIELD_DEF mysql_gtid_slave_pos_tabledef= {
+ array_elements(mysql_rpl_slave_state_coltypes),
+ mysql_rpl_slave_state_coltypes,
+ array_elements(mysql_rpl_slave_state_pk_parts),
+ mysql_rpl_slave_state_pk_parts
+};
+
+static Table_check_intact_log_error gtid_table_intact;
+
+/*
+ Check that the mysql.gtid_slave_pos table has the correct definition.
+*/
+int
+gtid_check_rpl_slave_state_table(TABLE *table)
+{
+ int err;
+
+ if ((err= gtid_table_intact.check(table, &mysql_gtid_slave_pos_tabledef)))
+ my_error(ER_GTID_OPEN_TABLE_FAILED, MYF(0), "mysql",
+ rpl_gtid_slave_state_table_name.str);
+ return err;
+}
+
+
+/*
+ Attempt to find a mysql.gtid_slave_posXXX table that has a storage engine
+ that is already in use by the current transaction, if any.
+*/
+void
+rpl_slave_state::select_gtid_pos_table(THD *thd, LEX_CSTRING *out_tablename)
+{
+ /*
+ See comments on rpl_slave_state::gtid_pos_tables for rules around proper
+ access to the list.
+ */
+ auto list= gtid_pos_tables.load(std::memory_order_acquire);
+
+ Ha_trx_info *ha_info;
+ uint count = 0;
+ for (ha_info= thd->transaction->all.ha_list; ha_info; ha_info= ha_info->next())
+ {
+ void *trx_hton= ha_info->ht();
+ auto table_entry= list;
+
+ if (!ha_info->is_trx_read_write() || trx_hton == binlog_hton)
+ continue;
+ while (table_entry)
+ {
+ if (table_entry->table_hton == trx_hton)
+ {
+ if (likely(table_entry->state == GTID_POS_AVAILABLE))
+ {
+ *out_tablename= table_entry->table_name;
+ /*
+ Check if this is a cross-engine transaction, so we can correctly
+ maintain the rpl_transactions_multi_engine status variable.
+ */
+ if (count >= 1)
+ statistic_increment(rpl_transactions_multi_engine, LOCK_status);
+ else
+ {
+ for (;;)
+ {
+ ha_info= ha_info->next();
+ if (!ha_info)
+ break;
+ if (ha_info->is_trx_read_write() && ha_info->ht() != binlog_hton)
+ {
+ statistic_increment(rpl_transactions_multi_engine, LOCK_status);
+ break;
+ }
+ }
+ }
+ return;
+ }
+ /*
+ This engine is marked to automatically create the table.
+ We cannot easily do this here (possibly in the middle of a
+ transaction). But we can request the slave background thread
+ to create it, and in a short while it should become available
+ for following transactions.
+ */
+#ifdef HAVE_REPLICATION
+ slave_background_gtid_pos_create_request(table_entry);
+#endif
+ break;
+ }
+ table_entry= table_entry->next;
+ }
+ ++count;
+ }
+ /*
+ If we cannot find any table whose engine matches an engine that is
+ already active in the transaction, or if there is no current transaction
+ engines available, we return the default gtid_slave_pos table.
+ */
+ *out_tablename=
+ default_gtid_pos_table.load(std::memory_order_acquire)->table_name;
+ /* Record in status that we failed to find a suitable gtid_pos table. */
+ if (count > 0)
+ {
+ statistic_increment(transactions_gtid_foreign_engine, LOCK_status);
+ if (count > 1)
+ statistic_increment(rpl_transactions_multi_engine, LOCK_status);
+ }
+}
+
+
+/*
+ Write a gtid to the replication slave state table.
+
+ Do it as part of the transaction, to get slave crash safety, or as a separate
+ transaction if !in_transaction (eg. MyISAM or DDL).
+
+ gtid The global transaction id for this event group.
+ sub_id Value allocated within the sub_id when the event group was
+ read (sub_id must be consistent with commit order in master binlog).
+
+ Note that caller must later ensure that the new gtid and sub_id is inserted
+ into the appropriate HASH element with rpl_slave_state.add(), so that it can
+ be deleted later. But this must only be done after COMMIT if in transaction.
+*/
+int
+rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
+ bool in_transaction, bool in_statement,
+ void **out_hton)
+{
+ TABLE_LIST tlist;
+ int err= 0, not_sql_thread;
+ bool table_opened= false;
+ TABLE *table;
+ ulonglong thd_saved_option= thd->variables.option_bits;
+ Query_tables_list lex_backup;
+ wait_for_commit* suspended_wfc;
+ void *hton= NULL;
+ LEX_CSTRING gtid_pos_table_name;
+ DBUG_ENTER("record_gtid");
+
+ *out_hton= NULL;
+ if (unlikely(!loaded))
+ {
+ /*
+ Probably the mysql.gtid_slave_pos table is missing (eg. upgrade) or
+ corrupt.
+
+ We already complained loudly about this, but we can try to continue
+ until the DBA fixes it.
+ */
+ DBUG_RETURN(0);
+ }
+
+ if (!in_statement)
+ thd->reset_for_next_command();
+
+ /*
+ Only the SQL thread can call select_gtid_pos_table without a mutex
+ Other threads needs to use a mutex and take into account that the
+ result may change during execution, so we have to make a copy.
+ */
+
+ if ((not_sql_thread= (thd->system_thread != SYSTEM_THREAD_SLAVE_SQL)))
+ mysql_mutex_lock(&LOCK_slave_state);
+ select_gtid_pos_table(thd, &gtid_pos_table_name);
+ if (not_sql_thread)
+ {
+ LEX_CSTRING *tmp= thd->make_clex_string(gtid_pos_table_name.str,
+ gtid_pos_table_name.length);
+ mysql_mutex_unlock(&LOCK_slave_state);
+ if (!tmp)
+ DBUG_RETURN(1);
+ gtid_pos_table_name= *tmp;
+ }
+
+ DBUG_EXECUTE_IF("gtid_inject_record_gtid",
+ {
+ my_error(ER_CANNOT_UPDATE_GTID_STATE, MYF(0));
+ DBUG_RETURN(1);
+ } );
+
+ /*
+ If we are applying a non-transactional event group, we will be committing
+ here a transaction, but that does not imply that the event group has
+ completed or has been binlogged. So we should not trigger
+ wakeup_subsequent_commits() here.
+
+ Note: An alternative here could be to put a call to mark_start_commit() in
+ stmt_done() before the call to record_and_update_gtid(). This would
+ prevent later calling mark_start_commit() after we have run
+ wakeup_subsequent_commits() from committing the GTID update transaction
+ (which must be avoided to avoid accessing freed group_commit_orderer
+ object). It would also allow following event groups to start slightly
+ earlier. And in the cases where record_gtid() is called without an active
+ transaction, the current statement should have been binlogged already, so
+ binlog order is preserved.
+
+ But this is rather subtle, and potentially fragile. And it does not really
+ seem worth it; non-transactional loads are unlikely to benefit much from
+ parallel replication in any case. So for now, we go with the simple
+ suspend/resume of wakeup_subsequent_commits() here in record_gtid().
+ */
+ suspended_wfc= thd->suspend_subsequent_commits();
+ thd->lex->reset_n_backup_query_tables_list(&lex_backup);
+ tlist.init_one_table(&MYSQL_SCHEMA_NAME, &gtid_pos_table_name, NULL, TL_WRITE);
+ if ((err= open_and_lock_tables(thd, &tlist, FALSE, 0)))
+ goto end;
+ table_opened= true;
+ table= tlist.table;
+ hton= table->s->db_type();
+ table->file->row_logging= 0; // No binary logging
+
+ if ((err= gtid_check_rpl_slave_state_table(table)))
+ goto end;
+
+#ifdef WITH_WSREP
+ /*
+ Updates in slave state table should not be appended to galera transaction
+ writeset.
+ */
+ thd->wsrep_ignore_table= true;
+#endif
+
+ if (!in_transaction)
+ {
+ DBUG_PRINT("info", ("resetting OPTION_BEGIN"));
+ thd->variables.option_bits&=
+ ~(ulonglong)(OPTION_NOT_AUTOCOMMIT |OPTION_BEGIN |OPTION_BIN_LOG |
+ OPTION_GTID_BEGIN);
+ }
+ else
+ thd->variables.option_bits&= ~(ulonglong)OPTION_BIN_LOG;
+
+ bitmap_set_all(table->write_set);
+ table->rpl_write_set= table->write_set;
+
+ table->field[0]->store((ulonglong)gtid->domain_id, true);
+ table->field[1]->store(sub_id, true);
+ table->field[2]->store((ulonglong)gtid->server_id, true);
+ table->field[3]->store(gtid->seq_no, true);
+ DBUG_EXECUTE_IF("inject_crash_before_write_rpl_slave_state", DBUG_SUICIDE(););
+ if ((err= table->file->ha_write_row(table->record[0])))
+ {
+ table->file->print_error(err, MYF(0));
+ goto end;
+ }
+ *out_hton= hton;
+
+ if(opt_bin_log &&
+ (err= mysql_bin_log.bump_seq_no_counter_if_needed(gtid->domain_id,
+ gtid->seq_no)))
+ {
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ goto end;
+ }
+end:
+
+#ifdef WITH_WSREP
+ thd->wsrep_ignore_table= false;
+#endif
+
+ if (table_opened)
+ {
+ if (err || (err= ha_commit_trans(thd, FALSE)))
+ ha_rollback_trans(thd, FALSE);
+
+ close_thread_tables(thd);
+ if (in_transaction)
+ thd->mdl_context.release_statement_locks();
+ else
+ thd->release_transactional_locks();
+ }
+ thd->lex->restore_backup_query_tables_list(&lex_backup);
+ thd->variables.option_bits= thd_saved_option;
+ thd->resume_subsequent_commits(suspended_wfc);
+ DBUG_EXECUTE_IF("inject_record_gtid_serverid_100_sleep",
+ {
+ if (gtid->server_id == 100)
+ my_sleep(500000);
+ });
+ DBUG_RETURN(err);
+}
+
+
+/*
+ Return a list of all old GTIDs in any mysql.gtid_slave_pos* table that are
+ no longer needed and can be deleted from the table.
+
+ Within each domain, we need to keep around the latest GTID (the one with the
+ highest sub_id), but any others in that domain can be deleted.
+*/
+rpl_slave_state::list_element *
+rpl_slave_state::gtid_grab_pending_delete_list()
+{
+ uint32 i;
+ list_element *full_list;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ full_list= NULL;
+ for (i= 0; i < hash.records; ++i)
+ {
+ element *elem= (element *)my_hash_element(&hash, i);
+ list_element *elist= elem->list;
+ list_element *last_elem, **best_ptr_ptr, *cur, *next;
+ uint64 best_sub_id;
+
+ if (!elist)
+ continue; /* Nothing here */
+
+ /* Delete any old stuff, but keep around the most recent one. */
+ cur= elist;
+ best_sub_id= cur->sub_id;
+ best_ptr_ptr= &elist;
+ last_elem= cur;
+ while ((next= cur->next)) {
+ last_elem= next;
+ if (next->sub_id > best_sub_id)
+ {
+ best_sub_id= next->sub_id;
+ best_ptr_ptr= &cur->next;
+ }
+ cur= next;
+ }
+ /*
+ Append the new elements to the full list. Note the order is important;
+ we do it here so that we do not break the list if best_sub_id is the
+ last of the new elements.
+ */
+ last_elem->next= full_list;
+ /*
+ Delete the highest sub_id element from the old list, and put it back as
+ the single-element new list.
+ */
+ cur= *best_ptr_ptr;
+ *best_ptr_ptr= cur->next;
+ cur->next= NULL;
+ elem->list= cur;
+
+ /*
+ Collect the full list so far here. Note that elist may have moved if we
+ deleted the first element, so order is again important.
+ */
+ full_list= elist;
+ }
+ mysql_mutex_unlock(&LOCK_slave_state);
+
+ return full_list;
+}
+
+
+/* Find the mysql.gtid_slave_posXXX table associated with a given hton. */
+LEX_CSTRING *
+rpl_slave_state::select_gtid_pos_table(void *hton)
+{
+ /*
+ See comments on rpl_slave_state::gtid_pos_tables for rules around proper
+ access to the list.
+ */
+ auto table_entry= gtid_pos_tables.load(std::memory_order_acquire);
+
+ while (table_entry)
+ {
+ if (table_entry->table_hton == hton)
+ {
+ if (likely(table_entry->state == GTID_POS_AVAILABLE))
+ return &table_entry->table_name;
+ }
+ table_entry= table_entry->next;
+ }
+
+ return &default_gtid_pos_table.load(std::memory_order_acquire)->table_name;
+}
+
+
+void
+rpl_slave_state::gtid_delete_pending(THD *thd,
+ rpl_slave_state::list_element **list_ptr)
+{
+ int err= 0;
+ ulonglong thd_saved_option;
+
+ if (unlikely(!loaded))
+ return;
+
+#ifdef WITH_WSREP
+ /*
+ Updates in slave state table should not be appended to galera transaction
+ writeset.
+ */
+ thd->wsrep_ignore_table= true;
+#endif
+
+ thd_saved_option= thd->variables.option_bits;
+ thd->variables.option_bits&=
+ ~(ulonglong)(OPTION_NOT_AUTOCOMMIT |OPTION_BEGIN |OPTION_BIN_LOG |
+ OPTION_GTID_BEGIN);
+
+ while (*list_ptr)
+ {
+ LEX_CSTRING *gtid_pos_table_name, *tmp_table_name;
+ Query_tables_list lex_backup;
+ TABLE_LIST tlist;
+ TABLE *table;
+ handler::Table_flags direct_pos= 0;
+ list_element *cur, **cur_ptr_ptr;
+ bool table_opened= false;
+ bool index_inited= false;
+ void *hton= (*list_ptr)->hton;
+
+ thd->reset_for_next_command();
+
+ /*
+ Only the SQL thread can call select_gtid_pos_table without a mutex
+ Other threads needs to use a mutex and take into account that the
+ result may change during execution, so we have to make a copy.
+ */
+ mysql_mutex_lock(&LOCK_slave_state);
+ tmp_table_name= select_gtid_pos_table(hton);
+ gtid_pos_table_name= thd->make_clex_string(tmp_table_name->str,
+ tmp_table_name->length);
+ mysql_mutex_unlock(&LOCK_slave_state);
+ if (!gtid_pos_table_name)
+ {
+ /* Out of memory - we can try again later. */
+ break;
+ }
+
+ thd->lex->reset_n_backup_query_tables_list(&lex_backup);
+ tlist.init_one_table(&MYSQL_SCHEMA_NAME, gtid_pos_table_name, NULL, TL_WRITE);
+ if ((err= open_and_lock_tables(thd, &tlist, FALSE, 0)))
+ goto end;
+ table_opened= true;
+ table= tlist.table;
+
+ if ((err= gtid_check_rpl_slave_state_table(table)))
+ goto end;
+
+ direct_pos= table->file->ha_table_flags() & HA_PRIMARY_KEY_REQUIRED_FOR_POSITION;
+ bitmap_set_all(table->write_set);
+ table->rpl_write_set= table->write_set;
+
+ /* Now delete any already committed GTIDs. */
+ bitmap_set_bit(table->read_set, table->field[0]->field_index);
+ bitmap_set_bit(table->read_set, table->field[1]->field_index);
+
+ if (!direct_pos)
+ {
+ if ((err= table->file->ha_index_init(0, 0)))
+ {
+ table->file->print_error(err, MYF(0));
+ goto end;
+ }
+ index_inited= true;
+ }
+
+ cur = *list_ptr;
+ cur_ptr_ptr = list_ptr;
+ do
+ {
+ uchar key_buffer[4+8];
+ list_element *next= cur->next;
+
+ if (cur->hton == hton)
+ {
+ int res;
+
+ table->field[0]->store((ulonglong)cur->domain_id, true);
+ table->field[1]->store(cur->sub_id, true);
+ if (direct_pos)
+ {
+ res= table->file->ha_rnd_pos_by_record(table->record[0]);
+ }
+ else
+ {
+ key_copy(key_buffer, table->record[0], &table->key_info[0], 0, false);
+ res= table->file->ha_index_read_map(table->record[0], key_buffer,
+ HA_WHOLE_KEY, HA_READ_KEY_EXACT);
+ }
+ DBUG_EXECUTE_IF("gtid_slave_pos_simulate_failed_delete",
+ { res= 1;
+ err= ENOENT;
+ sql_print_error("<DEBUG> Error deleting old GTID row");
+ });
+ if (res)
+ /* We cannot find the row, assume it is already deleted. */
+ ;
+ else if ((err= table->file->ha_delete_row(table->record[0])))
+ {
+ sql_print_error("Error deleting old GTID row: %s",
+ thd->get_stmt_da()->message());
+ /*
+ In case of error, we still discard the element from the list. We do
+ not want to endlessly error on the same element in case of table
+ corruption or such.
+ */
+ }
+ *cur_ptr_ptr= next;
+ my_free(cur);
+ }
+ else
+ {
+ /* Leave this one in the list until we get to the table for its hton. */
+ cur_ptr_ptr= &cur->next;
+ }
+ cur= next;
+ if (err)
+ break;
+ } while (cur);
+end:
+ if (table_opened)
+ {
+ DBUG_ASSERT(direct_pos || index_inited || err);
+ /*
+ Index may not be initialized if there was a failure during
+ 'ha_index_init'. Hence check if index initialization is successful and
+ then invoke ha_index_end(). Ending an index which is not initialized
+ will lead to assert.
+ */
+ if (index_inited)
+ table->file->ha_index_end();
+
+ if (err || (err= ha_commit_trans(thd, FALSE)))
+ ha_rollback_trans(thd, FALSE);
+ }
+ close_thread_tables(thd);
+ thd->release_transactional_locks();
+ thd->lex->restore_backup_query_tables_list(&lex_backup);
+
+ if (err)
+ break;
+ }
+ thd->variables.option_bits= thd_saved_option;
+
+#ifdef WITH_WSREP
+ thd->wsrep_ignore_table= false;
+#endif
+}
+
+
+uint64
+rpl_slave_state::next_sub_id(uint32 domain_id)
+{
+ uint64 sub_id= 0;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ sub_id= ++last_sub_id;
+ mysql_mutex_unlock(&LOCK_slave_state);
+
+ return sub_id;
+}
+
+/* A callback used in sorting of gtid list based on domain_id. */
+static int rpl_gtid_cmp_cb(const void *id1, const void *id2)
+{
+ uint32 d1= ((rpl_gtid *)id1)->domain_id;
+ uint32 d2= ((rpl_gtid *)id2)->domain_id;
+
+ if (d1 < d2)
+ return -1;
+ else if (d1 > d2)
+ return 1;
+ return 0;
+}
+
+/* Format the specified gtid and store it in the given string buffer. */
+bool
+rpl_slave_state_tostring_helper(String *dest, const rpl_gtid *gtid, bool *first)
+{
+ if (*first)
+ *first= false;
+ else
+ if (dest->append(",",1))
+ return true;
+ return
+ dest->append_ulonglong(gtid->domain_id) ||
+ dest->append("-",1) ||
+ dest->append_ulonglong(gtid->server_id) ||
+ dest->append("-",1) ||
+ dest->append_ulonglong(gtid->seq_no);
+}
+
+/*
+ Sort the given gtid list based on domain_id and store them in the specified
+ string.
+*/
+static bool
+rpl_slave_state_tostring_helper(DYNAMIC_ARRAY *gtid_dynarr, String *str)
+{
+ bool first= true, res= true;
+
+ sort_dynamic(gtid_dynarr, rpl_gtid_cmp_cb);
+
+ for (uint i= 0; i < gtid_dynarr->elements; i ++)
+ {
+ rpl_gtid *gtid= dynamic_element(gtid_dynarr, i, rpl_gtid *);
+ if (rpl_slave_state_tostring_helper(str, gtid, &first))
+ goto err;
+ }
+ res= false;
+
+err:
+ return res;
+}
+
+
+/* Sort the given gtid list based on domain_id and call cb for each gtid. */
+static bool
+rpl_slave_state_tostring_helper(DYNAMIC_ARRAY *gtid_dynarr,
+ int (*cb)(rpl_gtid *, void *),
+ void *data)
+{
+ rpl_gtid *gtid;
+ bool res= true;
+
+ sort_dynamic(gtid_dynarr, rpl_gtid_cmp_cb);
+
+ for (uint i= 0; i < gtid_dynarr->elements; i ++)
+ {
+ gtid= dynamic_element(gtid_dynarr, i, rpl_gtid *);
+ if ((*cb)(gtid, data))
+ goto err;
+ }
+ res= false;
+
+err:
+ return res;
+}
+
+int
+rpl_slave_state::iterate(int (*cb)(rpl_gtid *, void *), void *data,
+ rpl_gtid *extra_gtids, uint32 num_extra,
+ bool sort)
+{
+ uint32 i;
+ HASH gtid_hash;
+ uchar *rec;
+ rpl_gtid *gtid;
+ int res= 1;
+ bool locked= false;
+
+ my_hash_init(PSI_INSTRUMENT_ME, &gtid_hash, &my_charset_bin, 32,
+ offsetof(rpl_gtid, domain_id), sizeof(uint32), NULL, NULL,
+ HASH_UNIQUE);
+ for (i= 0; i < num_extra; ++i)
+ if (extra_gtids[i].server_id == global_system_variables.server_id &&
+ my_hash_insert(&gtid_hash, (uchar *)(&extra_gtids[i])))
+ goto err;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ locked= true;
+ reset_dynamic(&gtid_sort_array);
+
+ for (i= 0; i < hash.records; ++i)
+ {
+ uint64 best_sub_id;
+ rpl_gtid best_gtid;
+ element *e= (element *)my_hash_element(&hash, i);
+ list_element *l= e->list;
+
+ if (!l)
+ continue; /* Nothing here */
+
+ best_gtid.domain_id= e->domain_id;
+ best_gtid.server_id= l->server_id;
+ best_gtid.seq_no= l->seq_no;
+ best_sub_id= l->sub_id;
+ while ((l= l->next))
+ {
+ if (l->sub_id > best_sub_id)
+ {
+ best_sub_id= l->sub_id;
+ best_gtid.server_id= l->server_id;
+ best_gtid.seq_no= l->seq_no;
+ }
+ }
+
+ /* Check if we have something newer in the extra list. */
+ rec= my_hash_search(&gtid_hash, (const uchar *)&best_gtid.domain_id, 0);
+ if (rec)
+ {
+ gtid= (rpl_gtid *)rec;
+ if (gtid->seq_no > best_gtid.seq_no)
+ memcpy(&best_gtid, gtid, sizeof(best_gtid));
+ if (my_hash_delete(&gtid_hash, rec))
+ {
+ goto err;
+ }
+ }
+
+ if ((res= sort ? insert_dynamic(&gtid_sort_array,
+ (const void *) &best_gtid) :
+ (*cb)(&best_gtid, data)))
+ {
+ goto err;
+ }
+ }
+
+ /* Also add any remaining extra domain_ids. */
+ for (i= 0; i < gtid_hash.records; ++i)
+ {
+ gtid= (rpl_gtid *)my_hash_element(&gtid_hash, i);
+ if ((res= sort ? insert_dynamic(&gtid_sort_array, (const void *) gtid) :
+ (*cb)(gtid, data)))
+ {
+ goto err;
+ }
+ }
+
+ if (sort && rpl_slave_state_tostring_helper(&gtid_sort_array, cb, data))
+ {
+ goto err;
+ }
+
+ res= 0;
+
+err:
+ if (locked) mysql_mutex_unlock(&LOCK_slave_state);
+ my_hash_free(&gtid_hash);
+
+ return res;
+}
+
+
+struct rpl_slave_state_tostring_data {
+ String *dest;
+ bool first;
+};
+static int
+rpl_slave_state_tostring_cb(rpl_gtid *gtid, void *data)
+{
+ rpl_slave_state_tostring_data *p= (rpl_slave_state_tostring_data *)data;
+ return rpl_slave_state_tostring_helper(p->dest, gtid, &p->first);
+}
+
+
+/*
+ Prepare the current slave state as a string, suitable for sending to the
+ master to request to receive binlog events starting from that GTID state.
+
+ The state consists of the most recently applied GTID for each domain_id,
+ ie. the one with the highest sub_id within each domain_id.
+
+ Optinally, extra_gtids is a list of GTIDs from the binlog. This is used when
+ a server was previously a master and now needs to connect to a new master as
+ a slave. For each domain_id, if the GTID in the binlog was logged with our
+ own server_id _and_ has a higher seq_no than what is in the slave state,
+ then this should be used as the position to start replicating at. This
+ allows to promote a slave as new master, and connect the old master as a
+ slave with MASTER_GTID_POS=AUTO.
+*/
+int
+rpl_slave_state::tostring(String *dest, rpl_gtid *extra_gtids, uint32 num_extra)
+{
+ struct rpl_slave_state_tostring_data data;
+ data.first= true;
+ data.dest= dest;
+
+ return iterate(rpl_slave_state_tostring_cb, &data, extra_gtids,
+ num_extra, true);
+}
+
+
+/*
+ Lookup a domain_id in the current replication slave state.
+
+ Returns false if the domain_id has no entries in the slave state.
+ Otherwise returns true, and fills in out_gtid with the corresponding
+ GTID.
+*/
+bool
+rpl_slave_state::domain_to_gtid(uint32 domain_id, rpl_gtid *out_gtid)
+{
+ element *elem;
+ list_element *list;
+ uint64 best_sub_id;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ elem= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0);
+ if (!elem || !(list= elem->list))
+ {
+ mysql_mutex_unlock(&LOCK_slave_state);
+ return false;
+ }
+
+ out_gtid->domain_id= domain_id;
+ out_gtid->server_id= list->server_id;
+ out_gtid->seq_no= list->seq_no;
+ best_sub_id= list->sub_id;
+
+ while ((list= list->next))
+ {
+ if (best_sub_id > list->sub_id)
+ continue;
+ best_sub_id= list->sub_id;
+ out_gtid->server_id= list->server_id;
+ out_gtid->seq_no= list->seq_no;
+ }
+
+ mysql_mutex_unlock(&LOCK_slave_state);
+ return true;
+}
+
+
+/*
+ Parse a GTID at the start of a string, and update the pointer to point
+ at the first character after the parsed GTID.
+
+ Returns 0 on ok, non-zero on parse error.
+*/
+static int
+gtid_parser_helper(const char **ptr, const char *end, rpl_gtid *out_gtid)
+{
+ char *q;
+ const char *p= *ptr;
+ uint64 v1, v2, v3;
+ int err= 0;
+
+ q= (char*) end;
+ v1= (uint64)my_strtoll10(p, &q, &err);
+ if (err != 0 || v1 > (uint32)0xffffffff || q == end || *q != '-')
+ return 1;
+ p= q+1;
+ q= (char*) end;
+ v2= (uint64)my_strtoll10(p, &q, &err);
+ if (err != 0 || v2 > (uint32)0xffffffff || q == end || *q != '-')
+ return 1;
+ p= q+1;
+ q= (char*) end;
+ v3= (uint64)my_strtoll10(p, &q, &err);
+ if (err != 0)
+ return 1;
+
+ out_gtid->domain_id= (uint32) v1;
+ out_gtid->server_id= (uint32) v2;
+ out_gtid->seq_no= v3;
+ *ptr= q;
+ return 0;
+}
+
+
+rpl_gtid *
+gtid_parse_string_to_list(const char *str, size_t str_len, uint32 *out_len)
+{
+ const char *p= const_cast<char *>(str);
+ const char *end= p + str_len;
+ uint32 len= 0, alloc_len= 5;
+ rpl_gtid *list= NULL;
+
+ for (;;)
+ {
+ rpl_gtid gtid;
+
+ if (len >= (((uint32)1 << 28)-1) || gtid_parser_helper(&p, end, &gtid))
+ {
+ my_free(list);
+ return NULL;
+ }
+ if ((!list || len >= alloc_len) &&
+ !(list=
+ (rpl_gtid *)my_realloc(PSI_INSTRUMENT_ME, list,
+ (alloc_len= alloc_len*2) * sizeof(rpl_gtid),
+ MYF(MY_FREE_ON_ERROR|MY_ALLOW_ZERO_PTR))))
+ return NULL;
+ list[len++]= gtid;
+
+ if (p == end)
+ break;
+ if (*p != ',')
+ {
+ my_free(list);
+ return NULL;
+ }
+ ++p;
+ }
+ *out_len= len;
+ return list;
+}
+
+
+/*
+ Update the slave replication state with the GTID position obtained from
+ master when connecting with old-style (filename,offset) position.
+
+ If RESET is true then all existing entries are removed. Otherwise only
+ domain_ids mentioned in the STATE_FROM_MASTER are changed.
+
+ Returns 0 if ok, non-zero if error.
+*/
+int
+rpl_slave_state::load(THD *thd, const char *state_from_master, size_t len,
+ bool reset, bool in_statement)
+{
+ const char *end= state_from_master + len;
+
+ if (reset)
+ {
+ if (truncate_state_table(thd))
+ return 1;
+ truncate_hash();
+ }
+ if (state_from_master == end)
+ return 0;
+ for (;;)
+ {
+ rpl_gtid gtid;
+ uint64 sub_id;
+ void *hton= NULL;
+
+ if (gtid_parser_helper(&state_from_master, end, &gtid) ||
+ !(sub_id= next_sub_id(gtid.domain_id)) ||
+ record_gtid(thd, &gtid, sub_id, false, in_statement, &hton) ||
+ update(gtid.domain_id, gtid.server_id, sub_id, gtid.seq_no, hton, NULL))
+ return 1;
+ if (state_from_master == end)
+ break;
+ if (*state_from_master != ',')
+ return 1;
+ ++state_from_master;
+ }
+ return 0;
+}
+
+
+bool
+rpl_slave_state::is_empty()
+{
+ uint32 i;
+ bool result= true;
+
+ mysql_mutex_lock(&LOCK_slave_state);
+ for (i= 0; i < hash.records; ++i)
+ {
+ element *e= (element *)my_hash_element(&hash, i);
+ if (e->list)
+ {
+ result= false;
+ break;
+ }
+ }
+ mysql_mutex_unlock(&LOCK_slave_state);
+
+ return result;
+}
+
+
+void
+rpl_slave_state::free_gtid_pos_tables(struct rpl_slave_state::gtid_pos_table *list)
+{
+ struct gtid_pos_table *cur, *next;
+
+ cur= list;
+ while (cur)
+ {
+ next= cur->next;
+ my_free(cur);
+ cur= next;
+ }
+}
+
+
+/*
+ Replace the list of available mysql.gtid_slave_posXXX tables with a new list.
+ The caller must be holding LOCK_slave_state. Additionally, this function
+ must only be called while all SQL threads are stopped.
+*/
+void
+rpl_slave_state::set_gtid_pos_tables_list(rpl_slave_state::gtid_pos_table *new_list,
+ rpl_slave_state::gtid_pos_table *default_entry)
+{
+ mysql_mutex_assert_owner(&LOCK_slave_state);
+ auto old_list= gtid_pos_tables.load(std::memory_order_relaxed);
+ gtid_pos_tables.store(new_list, std::memory_order_release);
+ default_gtid_pos_table.store(default_entry, std::memory_order_release);
+ free_gtid_pos_tables(old_list);
+}
+
+
+void
+rpl_slave_state::add_gtid_pos_table(rpl_slave_state::gtid_pos_table *entry)
+{
+ mysql_mutex_assert_owner(&LOCK_slave_state);
+ entry->next= gtid_pos_tables.load(std::memory_order_relaxed);
+ gtid_pos_tables.store(entry, std::memory_order_release);
+}
+
+
+struct rpl_slave_state::gtid_pos_table *
+rpl_slave_state::alloc_gtid_pos_table(LEX_CSTRING *table_name, void *hton,
+ rpl_slave_state::gtid_pos_table_state state)
+{
+ struct gtid_pos_table *p;
+ char *allocated_str;
+
+ if (!my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME), &p, sizeof(*p),
+ &allocated_str, table_name->length+1, NULL))
+ {
+ my_error(ER_OUTOFMEMORY, MYF(0), (int)(sizeof(*p) + table_name->length+1));
+ return NULL;
+ }
+ memcpy(allocated_str, table_name->str, table_name->length+1); // Also copy '\0'
+ p->next = NULL;
+ p->table_hton= hton;
+ p->table_name.str= allocated_str;
+ p->table_name.length= table_name->length;
+ p->state= state;
+ return p;
+}
+
+
+void rpl_binlog_state::init()
+{
+ my_hash_init(PSI_INSTRUMENT_ME, &hash, &my_charset_bin, 32, offsetof(element, domain_id),
+ sizeof(uint32), NULL, my_free, HASH_UNIQUE);
+ my_init_dynamic_array(PSI_INSTRUMENT_ME, &gtid_sort_array, sizeof(rpl_gtid), 8, 8, MYF(0));
+ mysql_mutex_init(key_LOCK_binlog_state, &LOCK_binlog_state,
+ MY_MUTEX_INIT_SLOW);
+ initialized= 1;
+}
+
+void
+rpl_binlog_state::reset_nolock()
+{
+ uint32 i;
+
+ for (i= 0; i < hash.records; ++i)
+ my_hash_free(&((element *)my_hash_element(&hash, i))->hash);
+ my_hash_reset(&hash);
+}
+
+
+void
+rpl_binlog_state::reset()
+{
+ mysql_mutex_lock(&LOCK_binlog_state);
+ reset_nolock();
+ mysql_mutex_unlock(&LOCK_binlog_state);
+}
+
+
+void rpl_binlog_state::free()
+{
+ if (initialized)
+ {
+ initialized= 0;
+ reset_nolock();
+ my_hash_free(&hash);
+ delete_dynamic(&gtid_sort_array);
+ mysql_mutex_destroy(&LOCK_binlog_state);
+ }
+}
+
+
+bool
+rpl_binlog_state::load(struct rpl_gtid *list, uint32 count)
+{
+ uint32 i;
+ bool res= false;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ reset_nolock();
+ for (i= 0; i < count; ++i)
+ {
+ if (update_nolock(&(list[i]), false))
+ {
+ res= true;
+ break;
+ }
+ }
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+static int rpl_binlog_state_load_cb(rpl_gtid *gtid, void *data)
+{
+ rpl_binlog_state *self= (rpl_binlog_state *)data;
+ return self->update_nolock(gtid, false);
+}
+
+
+bool
+rpl_binlog_state::load(rpl_slave_state *slave_pos)
+{
+ bool res= false;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ reset_nolock();
+ if (slave_pos->iterate(rpl_binlog_state_load_cb, this, NULL, 0, false))
+ res= true;
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+rpl_binlog_state::~rpl_binlog_state()
+{
+ free();
+}
+
+
+/*
+ Update replication state with a new GTID.
+
+ If the (domain_id, server_id) pair already exists, then the new GTID replaces
+ the old one for that domain id. Else a new entry is inserted.
+
+ Returns 0 for ok, 1 for error.
+*/
+int
+rpl_binlog_state::update_nolock(const struct rpl_gtid *gtid, bool strict)
+{
+ element *elem;
+
+ if ((elem= (element *)my_hash_search(&hash,
+ (const uchar *)(&gtid->domain_id), 0)))
+ {
+ if (strict && elem->last_gtid && elem->last_gtid->seq_no >= gtid->seq_no)
+ {
+ my_error(ER_GTID_STRICT_OUT_OF_ORDER, MYF(0), gtid->domain_id,
+ gtid->server_id, gtid->seq_no, elem->last_gtid->domain_id,
+ elem->last_gtid->server_id, elem->last_gtid->seq_no);
+ return 1;
+ }
+ if (elem->seq_no_counter < gtid->seq_no)
+ elem->seq_no_counter= gtid->seq_no;
+ if (!elem->update_element(gtid))
+ return 0;
+ }
+ else if (!alloc_element_nolock(gtid))
+ return 0;
+
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ return 1;
+}
+
+
+int
+rpl_binlog_state::update(const struct rpl_gtid *gtid, bool strict)
+{
+ int res;
+ mysql_mutex_lock(&LOCK_binlog_state);
+ res= update_nolock(gtid, strict);
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+/*
+ Fill in a new GTID, allocating next sequence number, and update state
+ accordingly.
+*/
+int
+rpl_binlog_state::update_with_next_gtid(uint32 domain_id, uint32 server_id,
+ rpl_gtid *gtid)
+{
+ element *elem;
+ int res= 0;
+
+ gtid->domain_id= domain_id;
+ gtid->server_id= server_id;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ if ((elem= (element *)my_hash_search(&hash, (const uchar *)(&domain_id), 0)))
+ {
+ gtid->seq_no= ++elem->seq_no_counter;
+ if (!elem->update_element(gtid))
+ goto end;
+ }
+ else
+ {
+ gtid->seq_no= 1;
+ if (!alloc_element_nolock(gtid))
+ goto end;
+ }
+
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ res= 1;
+end:
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+/* Helper functions for update. */
+int
+rpl_binlog_state::element::update_element(const rpl_gtid *gtid)
+{
+ rpl_gtid *lookup_gtid;
+
+ /*
+ By far the most common case is that successive events within same
+ replication domain have the same server id (it changes only when
+ switching to a new master). So save a hash lookup in this case.
+ */
+ if (likely(last_gtid && last_gtid->server_id == gtid->server_id))
+ {
+ last_gtid->seq_no= gtid->seq_no;
+ return 0;
+ }
+
+ lookup_gtid= (rpl_gtid *)
+ my_hash_search(&hash, (const uchar *)&gtid->server_id, 0);
+ if (lookup_gtid)
+ {
+ lookup_gtid->seq_no= gtid->seq_no;
+ last_gtid= lookup_gtid;
+ return 0;
+ }
+
+ /* Allocate a new GTID and insert it. */
+ lookup_gtid= (rpl_gtid *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*lookup_gtid),
+ MYF(MY_WME));
+ if (!lookup_gtid)
+ return 1;
+ memcpy(lookup_gtid, gtid, sizeof(*lookup_gtid));
+ if (my_hash_insert(&hash, (const uchar *)lookup_gtid))
+ {
+ my_free(lookup_gtid);
+ return 1;
+ }
+ last_gtid= lookup_gtid;
+ return 0;
+}
+
+
+int
+rpl_binlog_state::alloc_element_nolock(const rpl_gtid *gtid)
+{
+ element *elem;
+ rpl_gtid *lookup_gtid;
+
+ /* First time we see this domain_id; allocate a new element. */
+ elem= (element *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*elem), MYF(MY_WME));
+ lookup_gtid= (rpl_gtid *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*lookup_gtid),
+ MYF(MY_WME));
+ if (elem && lookup_gtid)
+ {
+ elem->domain_id= gtid->domain_id;
+ my_hash_init(PSI_INSTRUMENT_ME, &elem->hash, &my_charset_bin, 32,
+ offsetof(rpl_gtid, server_id), sizeof(uint32), NULL, my_free,
+ HASH_UNIQUE);
+ elem->last_gtid= lookup_gtid;
+ elem->seq_no_counter= gtid->seq_no;
+ memcpy(lookup_gtid, gtid, sizeof(*lookup_gtid));
+ if (0 == my_hash_insert(&elem->hash, (const uchar *)lookup_gtid))
+ {
+ lookup_gtid= NULL; /* Do not free. */
+ if (0 == my_hash_insert(&hash, (const uchar *)elem))
+ return 0;
+ }
+ my_hash_free(&elem->hash);
+ }
+
+ /* An error. */
+ if (elem)
+ my_free(elem);
+ if (lookup_gtid)
+ my_free(lookup_gtid);
+ return 1;
+}
+
+
+/*
+ Check that a new GTID can be logged without creating an out-of-order
+ sequence number with existing GTIDs.
+*/
+bool
+rpl_binlog_state::check_strict_sequence(uint32 domain_id, uint32 server_id,
+ uint64 seq_no)
+{
+ element *elem;
+ bool res= 0;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ if ((elem= (element *)my_hash_search(&hash,
+ (const uchar *)(&domain_id), 0)) &&
+ elem->last_gtid && elem->last_gtid->seq_no >= seq_no)
+ {
+ my_error(ER_GTID_STRICT_OUT_OF_ORDER, MYF(0), domain_id, server_id, seq_no,
+ elem->last_gtid->domain_id, elem->last_gtid->server_id,
+ elem->last_gtid->seq_no);
+ res= 1;
+ }
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+/*
+ When we see a new GTID that will not be binlogged (eg. slave thread
+ with --log-slave-updates=0), then we need to remember to allocate any
+ GTID seq_no of our own within that domain starting from there.
+
+ Returns 0 if ok, non-zero if out-of-memory.
+*/
+int
+rpl_binlog_state::bump_seq_no_if_needed(uint32 domain_id, uint64 seq_no)
+{
+ element *elem;
+ int res;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ if ((elem= (element *)my_hash_search(&hash, (const uchar *)(&domain_id), 0)))
+ {
+ if (elem->seq_no_counter < seq_no)
+ elem->seq_no_counter= seq_no;
+ res= 0;
+ goto end;
+ }
+
+ /* We need to allocate a new, empty element to remember the next seq_no. */
+ if (!(elem= (element *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*elem),
+ MYF(MY_WME))))
+ {
+ res= 1;
+ goto end;
+ }
+
+ elem->domain_id= domain_id;
+ my_hash_init(PSI_INSTRUMENT_ME, &elem->hash, &my_charset_bin, 32,
+ offsetof(rpl_gtid, server_id), sizeof(uint32), NULL, my_free,
+ HASH_UNIQUE);
+ elem->last_gtid= NULL;
+ elem->seq_no_counter= seq_no;
+ if (0 == my_hash_insert(&hash, (const uchar *)elem))
+ {
+ res= 0;
+ goto end;
+ }
+
+ my_hash_free(&elem->hash);
+ my_free(elem);
+ res= 1;
+
+end:
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+/*
+ Write binlog state to text file, so we can read it in again without having
+ to scan last binlog file (normal shutdown/startup, not crash recovery).
+
+ The most recent GTID within each domain_id is written after any other GTID
+ within this domain.
+*/
+int
+rpl_binlog_state::write_to_iocache(IO_CACHE *dest)
+{
+ ulong i, j;
+ char buf[21];
+ int res= 0;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ for (i= 0; i < hash.records; ++i)
+ {
+ element *e= (element *)my_hash_element(&hash, i);
+ if (!e->last_gtid)
+ {
+ DBUG_ASSERT(e->hash.records == 0);
+ continue;
+ }
+ for (j= 0; j <= e->hash.records; ++j)
+ {
+ const rpl_gtid *gtid;
+ if (j < e->hash.records)
+ {
+ gtid= (const rpl_gtid *)my_hash_element(&e->hash, j);
+ if (gtid == e->last_gtid)
+ continue;
+ }
+ else
+ gtid= e->last_gtid;
+
+ longlong10_to_str(gtid->seq_no, buf, 10);
+ if (my_b_printf(dest, "%u-%u-%s\n", gtid->domain_id, gtid->server_id,
+ buf))
+ {
+ res= 1;
+ goto end;
+ }
+ }
+ }
+
+end:
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+int
+rpl_binlog_state::read_from_iocache(IO_CACHE *src)
+{
+ /* 10-digit - 10-digit - 20-digit \n \0 */
+ char buf[10+1+10+1+20+1+1];
+ const char *p, *end;
+ rpl_gtid gtid;
+ int res= 0;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ reset_nolock();
+ for (;;)
+ {
+ size_t len= my_b_gets(src, buf, sizeof(buf));
+ if (!len)
+ break;
+ p= buf;
+ end= buf + len;
+ if (gtid_parser_helper(&p, end, &gtid) ||
+ update_nolock(&gtid, false))
+ {
+ res= 1;
+ break;
+ }
+ }
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+rpl_gtid *
+rpl_binlog_state::find_nolock(uint32 domain_id, uint32 server_id)
+{
+ element *elem;
+ if (!(elem= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0)))
+ return NULL;
+ return (rpl_gtid *)my_hash_search(&elem->hash, (const uchar *)&server_id, 0);
+}
+
+rpl_gtid *
+rpl_binlog_state::find(uint32 domain_id, uint32 server_id)
+{
+ rpl_gtid *p;
+ mysql_mutex_lock(&LOCK_binlog_state);
+ p= find_nolock(domain_id, server_id);
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return p;
+}
+
+rpl_gtid *
+rpl_binlog_state::find_most_recent(uint32 domain_id)
+{
+ element *elem;
+ rpl_gtid *gtid= NULL;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ elem= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0);
+ if (elem && elem->last_gtid)
+ gtid= elem->last_gtid;
+ mysql_mutex_unlock(&LOCK_binlog_state);
+
+ return gtid;
+}
+
+
+uint32
+rpl_binlog_state::count()
+{
+ uint32 c= 0;
+ uint32 i;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ for (i= 0; i < hash.records; ++i)
+ c+= ((element *)my_hash_element(&hash, i))->hash.records;
+ mysql_mutex_unlock(&LOCK_binlog_state);
+
+ return c;
+}
+
+
+int
+rpl_binlog_state::get_gtid_list(rpl_gtid *gtid_list, uint32 list_size)
+{
+ uint32 i, j, pos;
+ int res= 0;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ pos= 0;
+ for (i= 0; i < hash.records; ++i)
+ {
+ element *e= (element *)my_hash_element(&hash, i);
+ if (!e->last_gtid)
+ {
+ DBUG_ASSERT(e->hash.records==0);
+ continue;
+ }
+ for (j= 0; j <= e->hash.records; ++j)
+ {
+ const rpl_gtid *gtid;
+ if (j < e->hash.records)
+ {
+ gtid= (rpl_gtid *)my_hash_element(&e->hash, j);
+ if (gtid == e->last_gtid)
+ continue;
+ }
+ else
+ gtid= e->last_gtid;
+
+ if (pos >= list_size)
+ {
+ res= 1;
+ goto end;
+ }
+ memcpy(&gtid_list[pos++], gtid, sizeof(*gtid));
+ }
+ }
+
+end:
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+
+/*
+ Get a list of the most recently binlogged GTID, for each domain_id.
+
+ This can be used when switching from being a master to being a slave,
+ to know where to start replicating from the new master.
+
+ The returned list must be de-allocated with my_free().
+
+ Returns 0 for ok, non-zero for out-of-memory.
+*/
+int
+rpl_binlog_state::get_most_recent_gtid_list(rpl_gtid **list, uint32 *size)
+{
+ uint32 i;
+ uint32 alloc_size, out_size;
+ int res= 0;
+
+ out_size= 0;
+ mysql_mutex_lock(&LOCK_binlog_state);
+ alloc_size= hash.records;
+ if (!(*list= (rpl_gtid *)my_malloc(PSI_INSTRUMENT_ME,
+ alloc_size * sizeof(rpl_gtid), MYF(MY_WME))))
+ {
+ res= 1;
+ goto end;
+ }
+ for (i= 0; i < alloc_size; ++i)
+ {
+ element *e= (element *)my_hash_element(&hash, i);
+ if (!e->last_gtid)
+ continue;
+ memcpy(&((*list)[out_size++]), e->last_gtid, sizeof(rpl_gtid));
+ }
+
+end:
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ *size= out_size;
+ return res;
+}
+
+bool
+rpl_binlog_state::append_pos(String *str)
+{
+ uint32 i;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ reset_dynamic(&gtid_sort_array);
+
+ for (i= 0; i < hash.records; ++i)
+ {
+ element *e= (element *)my_hash_element(&hash, i);
+ if (e->last_gtid &&
+ insert_dynamic(&gtid_sort_array, (const void *) e->last_gtid))
+ {
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return true;
+ }
+ }
+ rpl_slave_state_tostring_helper(&gtid_sort_array, str);
+ mysql_mutex_unlock(&LOCK_binlog_state);
+
+ return false;
+}
+
+
+bool
+rpl_binlog_state::append_state(String *str)
+{
+ uint32 i, j;
+ bool res= false;
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+ reset_dynamic(&gtid_sort_array);
+
+ for (i= 0; i < hash.records; ++i)
+ {
+ element *e= (element *)my_hash_element(&hash, i);
+ if (!e->last_gtid)
+ {
+ DBUG_ASSERT(e->hash.records==0);
+ continue;
+ }
+ for (j= 0; j <= e->hash.records; ++j)
+ {
+ const rpl_gtid *gtid;
+ if (j < e->hash.records)
+ {
+ gtid= (rpl_gtid *)my_hash_element(&e->hash, j);
+ if (gtid == e->last_gtid)
+ continue;
+ }
+ else
+ gtid= e->last_gtid;
+
+ if (insert_dynamic(&gtid_sort_array, (const void *) gtid))
+ {
+ res= true;
+ goto end;
+ }
+ }
+ }
+
+ rpl_slave_state_tostring_helper(&gtid_sort_array, str);
+
+end:
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ return res;
+}
+
+/**
+ Remove domains supplied by the first argument from binlog state.
+ Removal is done for any domain whose last gtids (from all its servers) match
+ ones in Gtid list event of the 2nd argument.
+
+ @param ids gtid domain id sequence, may contain dups
+ @param glev pointer to Gtid list event describing
+ the match condition
+ @param errbuf [out] pointer to possible error message array
+
+ @retval NULL as success when at least one domain is removed
+ @retval "" empty string to indicate ineffective call
+ when no domains removed
+ @retval NOT EMPTY string otherwise an error message
+*/
+const char*
+rpl_binlog_state::drop_domain(DYNAMIC_ARRAY *ids,
+ Gtid_list_log_event *glev,
+ char* errbuf)
+{
+ DYNAMIC_ARRAY domain_unique; // sequece (unsorted) of unique element*:s
+ rpl_binlog_state::element* domain_unique_buffer[16];
+ ulong k, l;
+ const char* errmsg= NULL;
+
+ DBUG_ENTER("rpl_binlog_state::drop_domain");
+
+ my_init_dynamic_array2(PSI_INSTRUMENT_ME, &domain_unique,
+ sizeof(element*), domain_unique_buffer,
+ sizeof(domain_unique_buffer) / sizeof(element*), 4, 0);
+
+ mysql_mutex_lock(&LOCK_binlog_state);
+
+ /*
+ Gtid list is supposed to come from a binlog's Gtid_list event and
+ therefore should be a subset of the current binlog state. That is
+ for every domain in the list the binlog state contains a gtid with
+ sequence number not less than that of the list.
+ Exceptions of this inclusion rule are:
+ A. the list may still refer to gtids from already deleted domains.
+ Files containing them must have been purged whereas the file
+ with the list is not yet.
+ B. out of order groups were injected
+ C. manually build list of binlog files violating the inclusion
+ constraint.
+ While A is a normal case (not necessarily distinguishable from C though),
+ B and C may require the user's attention so any (incl the A's suspected)
+ inconsistency is diagnosed and *warned*.
+ */
+ for (l= 0, errbuf[0]= 0; l < glev->count; l++, errbuf[0]= 0)
+ {
+ rpl_gtid* rb_state_gtid= find_nolock(glev->list[l].domain_id,
+ glev->list[l].server_id);
+ if (!rb_state_gtid)
+ sprintf(errbuf,
+ "missing gtids from the '%u-%u' domain-server pair which is "
+ "referred to in the gtid list describing an earlier state. Ignore "
+ "if the domain ('%u') was already explicitly deleted",
+ glev->list[l].domain_id, glev->list[l].server_id,
+ glev->list[l].domain_id);
+ else if (rb_state_gtid->seq_no < glev->list[l].seq_no)
+ sprintf(errbuf,
+ "having a gtid '%u-%u-%llu' which is less than "
+ "the '%u-%u-%llu' of the gtid list describing an earlier state. "
+ "The state may have been affected by manually injecting "
+ "a lower sequence number gtid or via replication",
+ rb_state_gtid->domain_id, rb_state_gtid->server_id,
+ rb_state_gtid->seq_no, glev->list[l].domain_id,
+ glev->list[l].server_id, glev->list[l].seq_no);
+ if (strlen(errbuf)) // use strlen() as cheap flag
+ push_warning_printf(current_thd, Sql_condition::WARN_LEVEL_WARN,
+ ER_BINLOG_CANT_DELETE_GTID_DOMAIN,
+ "The current gtid binlog state is incompatible with "
+ "a former one %s.", errbuf);
+ }
+
+ /*
+ For each domain_id from ids
+ when no such domain in binlog state
+ warn && continue
+ For each domain.server's last gtid
+ when not locate the last gtid in glev.list
+ error out binlog state can't change
+ otherwise continue
+ */
+ for (ulong i= 0; i < ids->elements; i++)
+ {
+ rpl_binlog_state::element *elem= NULL;
+ uint32 *ptr_domain_id;
+ bool not_match;
+
+ ptr_domain_id= (uint32*) dynamic_array_ptr(ids, i);
+ elem= (rpl_binlog_state::element *)
+ my_hash_search(&hash, (const uchar *) ptr_domain_id, 0);
+ if (!elem)
+ {
+ push_warning_printf(current_thd, Sql_condition::WARN_LEVEL_WARN,
+ ER_BINLOG_CANT_DELETE_GTID_DOMAIN,
+ "The gtid domain being deleted ('%lu') is not in "
+ "the current binlog state", *ptr_domain_id);
+ continue;
+ }
+
+ for (not_match= true, k= 0; k < elem->hash.records; k++)
+ {
+ rpl_gtid *d_gtid= (rpl_gtid *)my_hash_element(&elem->hash, k);
+ for (ulong l= 0; l < glev->count && not_match; l++)
+ not_match= !(*d_gtid == glev->list[l]);
+ }
+
+ if (not_match)
+ {
+ sprintf(errbuf, "binlog files may contain gtids from the domain ('%u') "
+ "being deleted. Make sure to first purge those files",
+ *ptr_domain_id);
+ errmsg= errbuf;
+ goto end;
+ }
+ // compose a sequence of unique pointers to domain object
+ for (k= 0; k < domain_unique.elements; k++)
+ {
+ if ((rpl_binlog_state::element*) dynamic_array_ptr(&domain_unique, k)
+ == elem)
+ break; // domain_id's elem has been already in
+ }
+ if (k == domain_unique.elements) // proven not to have duplicates
+ insert_dynamic(&domain_unique, (uchar*) &elem);
+ }
+
+ // Domain removal from binlog state
+ for (k= 0; k < domain_unique.elements; k++)
+ {
+ rpl_binlog_state::element *elem= *(rpl_binlog_state::element**)
+ dynamic_array_ptr(&domain_unique, k);
+ my_hash_free(&elem->hash);
+ my_hash_delete(&hash, (uchar*) elem);
+ }
+
+ DBUG_ASSERT(strlen(errbuf) == 0);
+
+ if (domain_unique.elements == 0)
+ errmsg= "";
+
+end:
+ mysql_mutex_unlock(&LOCK_binlog_state);
+ delete_dynamic(&domain_unique);
+
+ DBUG_RETURN(errmsg);
+}
+
+slave_connection_state::slave_connection_state()
+{
+ my_hash_init(PSI_INSTRUMENT_ME, &hash, &my_charset_bin, 32,
+ offsetof(entry, gtid) + offsetof(rpl_gtid, domain_id),
+ sizeof(uint32), NULL, my_free, HASH_UNIQUE);
+ my_init_dynamic_array(PSI_INSTRUMENT_ME, &gtid_sort_array, sizeof(rpl_gtid), 8, 8, MYF(0));
+}
+
+
+slave_connection_state::~slave_connection_state()
+{
+ my_hash_free(&hash);
+ delete_dynamic(&gtid_sort_array);
+}
+
+
+/*
+ Create a hash from the slave GTID state that is sent to master when slave
+ connects to start replication.
+
+ The state is sent as <GTID>,<GTID>,...,<GTID>, for example:
+
+ 0-2-112,1-4-1022
+
+ The state gives for each domain_id the GTID to start replication from for
+ the corresponding replication stream. So domain_id must be unique.
+
+ Returns 0 if ok, non-zero if error due to malformed input.
+
+ Note that input string is built by slave server, so it will not be incorrect
+ unless bug/corruption/malicious server. So we just need basic sanity check,
+ not fancy user-friendly error message.
+*/
+
+int
+slave_connection_state::load(const char *slave_request, size_t len)
+{
+ const char *p, *end;
+ uchar *rec;
+ rpl_gtid *gtid;
+ const entry *e;
+
+ reset();
+ p= slave_request;
+ end= slave_request + len;
+ if (p == end)
+ return 0;
+ for (;;)
+ {
+ if (!(rec= (uchar *)my_malloc(PSI_INSTRUMENT_ME, sizeof(entry), MYF(MY_WME))))
+ return 1;
+ gtid= &((entry *)rec)->gtid;
+ if (gtid_parser_helper(&p, end, gtid))
+ {
+ my_free(rec);
+ my_error(ER_INCORRECT_GTID_STATE, MYF(0));
+ return 1;
+ }
+ if ((e= (const entry *)
+ my_hash_search(&hash, (const uchar *)(&gtid->domain_id), 0)))
+ {
+ my_error(ER_DUPLICATE_GTID_DOMAIN, MYF(0), gtid->domain_id,
+ gtid->server_id, (ulonglong)gtid->seq_no, e->gtid.domain_id,
+ e->gtid.server_id, (ulonglong)e->gtid.seq_no, gtid->domain_id);
+ my_free(rec);
+ return 1;
+ }
+ ((entry *)rec)->flags= 0;
+ if (my_hash_insert(&hash, rec))
+ {
+ my_free(rec);
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ return 1;
+ }
+ if (p == end)
+ break; /* Finished. */
+ if (*p != ',')
+ {
+ my_error(ER_INCORRECT_GTID_STATE, MYF(0));
+ return 1;
+ }
+ ++p;
+ }
+
+ return 0;
+}
+
+
+int
+slave_connection_state::load(const rpl_gtid *gtid_list, uint32 count)
+{
+ uint32 i;
+
+ reset();
+ for (i= 0; i < count; ++i)
+ if (update(&gtid_list[i]))
+ return 1;
+ return 0;
+}
+
+
+static int
+slave_connection_state_load_cb(rpl_gtid *gtid, void *data)
+{
+ slave_connection_state *state= (slave_connection_state *)data;
+ return state->update(gtid);
+}
+
+
+/*
+ Same as rpl_slave_state::tostring(), but populates a slave_connection_state
+ instead.
+*/
+int
+slave_connection_state::load(rpl_slave_state *state,
+ rpl_gtid *extra_gtids, uint32 num_extra)
+{
+ reset();
+ return state->iterate(slave_connection_state_load_cb, this,
+ extra_gtids, num_extra, false);
+}
+
+
+slave_connection_state::entry *
+slave_connection_state::find_entry(uint32 domain_id)
+{
+ return (entry *) my_hash_search(&hash, (const uchar *)(&domain_id), 0);
+}
+
+
+rpl_gtid *
+slave_connection_state::find(uint32 domain_id)
+{
+ entry *e= find_entry(domain_id);
+ if (!e)
+ return NULL;
+ return &e->gtid;
+}
+
+
+int
+slave_connection_state::update(const rpl_gtid *in_gtid)
+{
+ entry *e;
+ uchar *rec= my_hash_search(&hash, (const uchar *)(&in_gtid->domain_id), 0);
+ if (rec)
+ {
+ e= (entry *)rec;
+ e->gtid= *in_gtid;
+ return 0;
+ }
+
+ if (!(e= (entry *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*e), MYF(MY_WME))))
+ return 1;
+ e->gtid= *in_gtid;
+ e->flags= 0;
+ if (my_hash_insert(&hash, (uchar *)e))
+ {
+ my_free(e);
+ return 1;
+ }
+
+ return 0;
+}
+
+
+void
+slave_connection_state::remove(const rpl_gtid *in_gtid)
+{
+ uchar *rec= my_hash_search(&hash, (const uchar *)(&in_gtid->domain_id), 0);
+#ifdef DBUG_ASSERT_EXISTS
+ bool err;
+ rpl_gtid *slave_gtid= &((entry *)rec)->gtid;
+ DBUG_ASSERT(rec /* We should never try to remove not present domain_id. */);
+ DBUG_ASSERT(slave_gtid->server_id == in_gtid->server_id);
+ DBUG_ASSERT(slave_gtid->seq_no == in_gtid->seq_no);
+ err=
+#endif
+ my_hash_delete(&hash, rec);
+ DBUG_ASSERT(!err);
+}
+
+
+void
+slave_connection_state::remove_if_present(const rpl_gtid *in_gtid)
+{
+ uchar *rec= my_hash_search(&hash, (const uchar *)(&in_gtid->domain_id), 0);
+ if (rec)
+ my_hash_delete(&hash, rec);
+}
+
+
+int
+slave_connection_state::to_string(String *out_str)
+{
+ out_str->length(0);
+ return append_to_string(out_str);
+}
+
+
+int
+slave_connection_state::append_to_string(String *out_str)
+{
+ uint32 i;
+ bool first;
+
+ first= true;
+ for (i= 0; i < hash.records; ++i)
+ {
+ const entry *e= (const entry *)my_hash_element(&hash, i);
+ if (rpl_slave_state_tostring_helper(out_str, &e->gtid, &first))
+ return 1;
+ }
+ return 0;
+}
+
+
+int
+slave_connection_state::get_gtid_list(rpl_gtid *gtid_list, uint32 list_size)
+{
+ uint32 i, pos;
+
+ pos= 0;
+ for (i= 0; i < hash.records; ++i)
+ {
+ entry *e;
+ if (pos >= list_size)
+ return 1;
+ e= (entry *)my_hash_element(&hash, i);
+ memcpy(&gtid_list[pos++], &e->gtid, sizeof(e->gtid));
+ }
+
+ return 0;
+}
+
+
+/*
+ Check if the GTID position has been reached, for mysql_binlog_send().
+
+ The position has not been reached if we have anything in the state, unless
+ it has either the START_ON_EMPTY_DOMAIN flag set (which means it does not
+ belong to this master at all), or the START_OWN_SLAVE_POS (which means that
+ we start on an old position from when the server was a slave with
+ --log-slave-updates=0).
+*/
+bool
+slave_connection_state::is_pos_reached()
+{
+ uint32 i;
+
+ for (i= 0; i < hash.records; ++i)
+ {
+ entry *e= (entry *)my_hash_element(&hash, i);
+ if (!(e->flags & (START_OWN_SLAVE_POS|START_ON_EMPTY_DOMAIN)))
+ return false;
+ }
+
+ return true;
+}
+
+
+/*
+ Execute a MASTER_GTID_WAIT().
+ The position to wait for is in gtid_str in string form.
+ The timeout in microseconds is in timeout_us, zero means no timeout.
+
+ Returns:
+ 1 for error.
+ 0 for wait completed.
+ -1 for wait timed out.
+*/
+int
+gtid_waiting::wait_for_pos(THD *thd, String *gtid_str, longlong timeout_us)
+{
+ int err;
+ rpl_gtid *wait_pos;
+ uint32 count, i;
+ struct timespec wait_until, *wait_until_ptr;
+ ulonglong before;
+
+ /* Wait for the empty position returns immediately. */
+ if (gtid_str->length() == 0)
+ {
+ status_var_increment(thd->status_var.master_gtid_wait_count);
+ return 0;
+ }
+
+ if (!(wait_pos= gtid_parse_string_to_list(gtid_str->ptr(), gtid_str->length(),
+ &count)))
+ {
+ my_error(ER_INCORRECT_GTID_STATE, MYF(0));
+ return 1;
+ }
+ status_var_increment(thd->status_var.master_gtid_wait_count);
+ before= microsecond_interval_timer();
+
+ if (timeout_us >= 0)
+ {
+ set_timespec_nsec(wait_until, (ulonglong)1000*timeout_us);
+ wait_until_ptr= &wait_until;
+ }
+ else
+ wait_until_ptr= NULL;
+ err= 0;
+ for (i= 0; i < count; ++i)
+ {
+ if ((err= wait_for_gtid(thd, &wait_pos[i], wait_until_ptr)))
+ break;
+ }
+ switch (err)
+ {
+ case -1:
+ status_var_increment(thd->status_var.master_gtid_wait_timeouts);
+ /* fall through */
+ case 0:
+ status_var_add(thd->status_var.master_gtid_wait_time,
+ static_cast<ulong>
+ (microsecond_interval_timer() - before));
+ }
+ my_free(wait_pos);
+ return err;
+}
+
+
+void
+gtid_waiting::promote_new_waiter(gtid_waiting::hash_element *he)
+{
+ queue_element *qe;
+
+ mysql_mutex_assert_owner(&LOCK_gtid_waiting);
+ if (queue_empty(&he->queue))
+ return;
+ qe= (queue_element *)queue_top(&he->queue);
+ qe->do_small_wait= true;
+ mysql_cond_signal(&qe->thd->COND_wakeup_ready);
+}
+
+void
+gtid_waiting::process_wait_hash(uint64 wakeup_seq_no,
+ gtid_waiting::hash_element *he)
+{
+ mysql_mutex_assert_owner(&LOCK_gtid_waiting);
+
+ for (;;)
+ {
+ queue_element *qe;
+
+ if (queue_empty(&he->queue))
+ break;
+ qe= (queue_element *)queue_top(&he->queue);
+ if (qe->wait_seq_no > wakeup_seq_no)
+ break;
+ DBUG_ASSERT(!qe->done);
+ queue_remove_top(&he->queue);
+ qe->done= true;;
+ mysql_cond_signal(&qe->thd->COND_wakeup_ready);
+ }
+}
+
+
+/*
+ Execute a MASTER_GTID_WAIT() for one specific domain.
+
+ The implementation is optimised primarily for (1) minimal performance impact
+ on the slave replication threads, and secondarily for (2) quick performance
+ of MASTER_GTID_WAIT() on a single GTID, which can be useful for consistent
+ read to clients in an async replication read-scaleout scenario.
+
+ To achieve (1), we have a "small" wait and a "large" wait. The small wait
+ contends with the replication threads on the lock on the gtid_slave_pos, so
+ only minimal processing is done under that lock, and only a single waiter at
+ a time does the small wait.
+
+ If there is already a small waiter, a new thread will either replace the
+ small waiter (if it needs to wait for an earlier sequence number), or
+ instead do a "large" wait.
+
+ Once awoken on the small wait, the waiting thread releases the lock shared
+ with the SQL threads quickly, and then processes all waiters currently doing
+ the large wait using a different lock that does not impact replication.
+
+ This way, the SQL threads only need to do a single check + possibly a
+ pthread_cond_signal() when updating the gtid_slave_state, and the time that
+ non-SQL threads contend for the lock on gtid_slave_state is minimized.
+
+ There is always at least one thread that has the responsibility to ensure
+ that there is a small waiter; this thread has queue_element::do_small_wait
+ set to true. This thread will do the small wait until it is done, at which
+ point it will make sure to pass on the responsibility to another thread.
+ Normally only one thread has do_small_wait==true, but it can occasionally
+ happen that there is more than one, when threads race one another for the
+ lock on the small wait (this results in slightly increased activity on the
+ small lock but is otherwise harmless).
+
+ Returns:
+ 0 Wait completed normally
+ -1 Wait completed due to timeout
+ 1 An error (my_error() will have been called to set the error in the da)
+*/
+int
+gtid_waiting::wait_for_gtid(THD *thd, rpl_gtid *wait_gtid,
+ struct timespec *wait_until)
+{
+ bool timed_out= false;
+#ifdef HAVE_REPLICATION
+ queue_element elem;
+ uint32 domain_id= wait_gtid->domain_id;
+ uint64 seq_no= wait_gtid->seq_no;
+ hash_element *he;
+ rpl_slave_state::element *slave_state_elem= NULL;
+ PSI_stage_info old_stage;
+ bool did_enter_cond= false;
+
+ elem.wait_seq_no= seq_no;
+ elem.thd= thd;
+ elem.done= false;
+
+ mysql_mutex_lock(&LOCK_gtid_waiting);
+ if (!(he= get_entry(wait_gtid->domain_id)))
+ {
+ mysql_mutex_unlock(&LOCK_gtid_waiting);
+ return 1;
+ }
+ /*
+ If there is already another waiter with seq_no no larger than our own,
+ we are sure that there is already a small waiter that will wake us up
+ (or later pass the small wait responsibility to us). So in this case, we
+ do not need to touch the small wait lock at all.
+ */
+ elem.do_small_wait=
+ (queue_empty(&he->queue) ||
+ ((queue_element *)queue_top(&he->queue))->wait_seq_no > seq_no);
+
+ if (register_in_wait_queue(thd, wait_gtid, he, &elem))
+ {
+ mysql_mutex_unlock(&LOCK_gtid_waiting);
+ return 1;
+ }
+ /*
+ Loop, doing either the small or large wait as appropriate, until either
+ the position waited for is reached, or we get a kill or timeout.
+ */
+ for (;;)
+ {
+ mysql_mutex_assert_owner(&LOCK_gtid_waiting);
+
+ if (elem.do_small_wait)
+ {
+ uint64 wakeup_seq_no;
+ queue_element *cur_waiter;
+
+ mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+ /*
+ The elements in the gtid_slave_state_hash are never re-allocated once
+ they enter the hash, so we do not need to re-do the lookup after releasing
+ and re-aquiring the lock.
+ */
+ if (!slave_state_elem &&
+ !(slave_state_elem= rpl_global_gtid_slave_state->get_element(domain_id)))
+ {
+ mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+ remove_from_wait_queue(he, &elem);
+ promote_new_waiter(he);
+ if (did_enter_cond)
+ thd->EXIT_COND(&old_stage);
+ else
+ mysql_mutex_unlock(&LOCK_gtid_waiting);
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ return 1;
+ }
+
+ if ((wakeup_seq_no= slave_state_elem->highest_seq_no) >= seq_no)
+ {
+ /*
+ We do not have to wait. (We will be removed from the wait queue when
+ we call process_wait_hash() below.
+ */
+ mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+ }
+ else if ((cur_waiter= slave_state_elem->gtid_waiter) &&
+ slave_state_elem->min_wait_seq_no <= seq_no)
+ {
+ /*
+ There is already a suitable small waiter, go do the large wait.
+ (Normally we would not have needed to check the small wait in this
+ case, but it can happen if we race with another thread for the small
+ lock).
+ */
+ elem.do_small_wait= false;
+ mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+ }
+ else
+ {
+ /*
+ We have to do the small wait ourselves (stealing it from any thread
+ that might already be waiting for a later seq_no).
+ */
+ slave_state_elem->gtid_waiter= &elem;
+ slave_state_elem->min_wait_seq_no= seq_no;
+ if (cur_waiter)
+ {
+ /* We stole the wait, so wake up the old waiting thread. */
+ mysql_cond_signal(&slave_state_elem->COND_wait_gtid);
+ }
+
+ /* Release the large lock, and do the small wait. */
+ if (did_enter_cond)
+ {
+ thd->EXIT_COND(&old_stage);
+ did_enter_cond= false;
+ }
+ else
+ mysql_mutex_unlock(&LOCK_gtid_waiting);
+ thd->ENTER_COND(&slave_state_elem->COND_wait_gtid,
+ &rpl_global_gtid_slave_state->LOCK_slave_state,
+ &stage_master_gtid_wait_primary, &old_stage);
+ do
+ {
+ if (unlikely(thd->check_killed(1)))
+ break;
+ else if (wait_until)
+ {
+ int err=
+ mysql_cond_timedwait(&slave_state_elem->COND_wait_gtid,
+ &rpl_global_gtid_slave_state->LOCK_slave_state,
+ wait_until);
+ if (err == ETIMEDOUT || err == ETIME)
+ {
+ timed_out= true;
+ break;
+ }
+ }
+ else
+ mysql_cond_wait(&slave_state_elem->COND_wait_gtid,
+ &rpl_global_gtid_slave_state->LOCK_slave_state);
+ } while (slave_state_elem->gtid_waiter == &elem);
+ wakeup_seq_no= slave_state_elem->highest_seq_no;
+ /*
+ If we aborted due to timeout or kill, remove us as waiter.
+
+ If we were replaced by another waiter with a smaller seq_no, then we
+ no longer have responsibility for the small wait.
+ */
+ if ((cur_waiter= slave_state_elem->gtid_waiter))
+ {
+ if (cur_waiter == &elem)
+ slave_state_elem->gtid_waiter= NULL;
+ else if (slave_state_elem->min_wait_seq_no <= seq_no)
+ elem.do_small_wait= false;
+ }
+ thd->EXIT_COND(&old_stage);
+
+ mysql_mutex_lock(&LOCK_gtid_waiting);
+ }
+
+ /*
+ Note that hash_entry pointers do not change once allocated, so we do
+ not need to lookup `he' again after re-aquiring LOCK_gtid_waiting.
+ */
+ process_wait_hash(wakeup_seq_no, he);
+ }
+ else
+ {
+ /* Do the large wait. */
+ if (!did_enter_cond)
+ {
+ thd->ENTER_COND(&thd->COND_wakeup_ready, &LOCK_gtid_waiting,
+ &stage_master_gtid_wait, &old_stage);
+ did_enter_cond= true;
+ }
+ while (!elem.done && likely(!thd->check_killed(1)))
+ {
+ thd_wait_begin(thd, THD_WAIT_BINLOG);
+ if (wait_until)
+ {
+ int err= mysql_cond_timedwait(&thd->COND_wakeup_ready,
+ &LOCK_gtid_waiting, wait_until);
+ if (err == ETIMEDOUT || err == ETIME)
+ timed_out= true;
+ }
+ else
+ mysql_cond_wait(&thd->COND_wakeup_ready, &LOCK_gtid_waiting);
+ thd_wait_end(thd);
+ if (elem.do_small_wait || timed_out)
+ break;
+ }
+ }
+
+ if ((thd->killed || timed_out) && !elem.done)
+ {
+ /* Aborted, so remove ourselves from the hash. */
+ remove_from_wait_queue(he, &elem);
+ elem.done= true;
+ }
+ if (elem.done)
+ {
+ /*
+ If our wait is done, but we have (or were passed) responsibility for
+ the small wait, then we need to pass on that task to someone else.
+ */
+ if (elem.do_small_wait)
+ promote_new_waiter(he);
+ break;
+ }
+ }
+
+ if (did_enter_cond)
+ thd->EXIT_COND(&old_stage);
+ else
+ mysql_mutex_unlock(&LOCK_gtid_waiting);
+ if (thd->killed)
+ thd->send_kill_message();
+#endif /* HAVE_REPLICATION */
+ return timed_out ? -1 : 0;
+}
+
+
+static void
+free_hash_element(void *p)
+{
+ gtid_waiting::hash_element *e= (gtid_waiting::hash_element *)p;
+ delete_queue(&e->queue);
+ my_free(e);
+}
+
+
+void
+gtid_waiting::init()
+{
+ my_hash_init(PSI_INSTRUMENT_ME, &hash, &my_charset_bin, 32,
+ offsetof(hash_element, domain_id), sizeof(uint32), NULL,
+ free_hash_element, HASH_UNIQUE);
+ mysql_mutex_init(key_LOCK_gtid_waiting, &LOCK_gtid_waiting, 0);
+}
+
+
+void
+gtid_waiting::destroy()
+{
+ mysql_mutex_destroy(&LOCK_gtid_waiting);
+ my_hash_free(&hash);
+}
+
+
+static int
+cmp_queue_elem(void *, uchar *a, uchar *b)
+{
+ uint64 seq_no_a= *(uint64 *)a;
+ uint64 seq_no_b= *(uint64 *)b;
+ if (seq_no_a < seq_no_b)
+ return -1;
+ else if (seq_no_a == seq_no_b)
+ return 0;
+ else
+ return 1;
+}
+
+
+gtid_waiting::hash_element *
+gtid_waiting::get_entry(uint32 domain_id)
+{
+ hash_element *e;
+
+ if ((e= (hash_element *)my_hash_search(&hash, (const uchar *)&domain_id, 0)))
+ return e;
+
+ if (!(e= (hash_element *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*e), MYF(MY_WME))))
+ return NULL;
+
+ if (init_queue(&e->queue, 8, offsetof(queue_element, wait_seq_no), 0,
+ cmp_queue_elem, NULL, 1+offsetof(queue_element, queue_idx), 1))
+ {
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ my_free(e);
+ return NULL;
+ }
+ e->domain_id= domain_id;
+ if (my_hash_insert(&hash, (uchar *)e))
+ {
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ delete_queue(&e->queue);
+ my_free(e);
+ return NULL;
+ }
+ return e;
+}
+
+
+int
+gtid_waiting::register_in_wait_queue(THD *thd, rpl_gtid *wait_gtid,
+ gtid_waiting::hash_element *he,
+ gtid_waiting::queue_element *elem)
+{
+ mysql_mutex_assert_owner(&LOCK_gtid_waiting);
+
+ if (queue_insert_safe(&he->queue, (uchar *)elem))
+ {
+ my_error(ER_OUT_OF_RESOURCES, MYF(0));
+ return 1;
+ }
+
+ return 0;
+}
+
+
+void
+gtid_waiting::remove_from_wait_queue(gtid_waiting::hash_element *he,
+ gtid_waiting::queue_element *elem)
+{
+ mysql_mutex_assert_owner(&LOCK_gtid_waiting);
+
+ queue_remove(&he->queue, elem->queue_idx);
+}