From 010971a761c0dd0435d05d1e8b81ebeadadd9b3c Mon Sep 17 00:00:00 2001 From: unknown Date: Wed, 23 Apr 2014 16:06:06 +0200 Subject: MDEV-6156: Parallel replication incorrectly caches charset between worker threads Replication caches the character sets used in a query, to be able to quickly reuse them for the next query in the common case of them not having changed. In parallel replication, this caching needs to be per-worker-thread. The code was not modified to handle this correctly, so the caching in one worker could cause another worker to run a query using the wrong character set, causing replication corruption. --- mysql-test/suite/rpl/r/rpl_parallel_charset.result | 87 ++++++++++++++++++++++ mysql-test/suite/rpl/t/rpl_parallel_charset.test | 56 ++++++++++++++ sql/log_event.cc | 6 +- sql/rpl_rli.cc | 48 ++++++------ sql/rpl_rli.h | 21 +++--- sql/slave.cc | 12 +-- sql/slave.h | 3 +- 7 files changed, 186 insertions(+), 47 deletions(-) create mode 100644 mysql-test/suite/rpl/r/rpl_parallel_charset.result create mode 100644 mysql-test/suite/rpl/t/rpl_parallel_charset.test diff --git a/mysql-test/suite/rpl/r/rpl_parallel_charset.result b/mysql-test/suite/rpl/r/rpl_parallel_charset.result new file mode 100644 index 00000000000..8929c209934 --- /dev/null +++ b/mysql-test/suite/rpl/r/rpl_parallel_charset.result @@ -0,0 +1,87 @@ +include/rpl_init.inc [topology=1->2] +*** MDEV-6156: Parallel replication incorrectly caches charset between worker threads *** +SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads; +include/stop_slave.inc +SET GLOBAL slave_parallel_threads=5; +include/start_slave.inc +CREATE TABLE t1 (a INT PRIMARY KEY, b VARCHAR(100) CHARACTER SET utf8); +SHOW CREATE TABLE t1; +Table Create Table +t1 CREATE TABLE `t1` ( + `a` int(11) NOT NULL, + `b` varchar(100) CHARACTER SET utf8 DEFAULT NULL, + PRIMARY KEY (`a`) +) ENGINE=MyISAM DEFAULT CHARSET=latin1 +SET character_set_client=latin1; +INSERT INTO t1 VALUES (1, 'Rřdgrřd med flřde 1'); +INSERT INTO t1 VALUES (2, 'Rřdgrřd med flřde 2'); +INSERT INTO t1 VALUES (3, 'Rřdgrřd med flřde 3'); +INSERT INTO t1 VALUES (4, 'Rřdgrřd med flřde 4'); +INSERT INTO t1 VALUES (5, 'Rřdgrřd med flřde 5'); +INSERT INTO t1 VALUES (6, 'Rřdgrřd med flřde 6'); +INSERT INTO t1 VALUES (7, 'Rřdgrřd med flřde 7'); +INSERT INTO t1 VALUES (8, 'Rřdgrřd med flřde 8'); +INSERT INTO t1 VALUES (9, 'Rřdgrřd med flřde 9'); +INSERT INTO t1 VALUES (10, 'Rřdgrřd med flřde 10'); +SET character_set_client=utf8; +INSERT INTO t1 VALUES (11, 'Rødgrød med fløde 1'); +INSERT INTO t1 VALUES (12, 'Rødgrød med fløde 2'); +INSERT INTO t1 VALUES (13, 'Rødgrød med fløde 3'); +INSERT INTO t1 VALUES (14, 'Rødgrød med fløde 4'); +INSERT INTO t1 VALUES (15, 'Rødgrød med fløde 5'); +INSERT INTO t1 VALUES (16, 'Rødgrød med fløde 6'); +INSERT INTO t1 VALUES (17, 'Rødgrød med fløde 7'); +INSERT INTO t1 VALUES (18, 'Rødgrød med fløde 8'); +INSERT INTO t1 VALUES (19, 'Rødgrød med fløde 9'); +INSERT INTO t1 VALUES (20, 'Rødgrød med fløde 10'); +SET character_set_results=utf8; +SELECT * FROM t1 ORDER BY a; +a b +1 Rødgrød med fløde 1 +2 Rødgrød med fløde 2 +3 Rødgrød med fløde 3 +4 Rødgrød med fløde 4 +5 Rødgrød med fløde 5 +6 Rødgrød med fløde 6 +7 Rødgrød med fløde 7 +8 Rødgrød med fløde 8 +9 Rødgrød med fløde 9 +10 Rødgrød med fløde 10 +11 Rødgrød med fløde 1 +12 Rødgrød med fløde 2 +13 Rødgrød med fløde 3 +14 Rødgrød med fløde 4 +15 Rødgrød med fløde 5 +16 Rødgrød med fløde 6 +17 Rødgrød med fløde 7 +18 Rødgrød med fløde 8 +19 Rødgrød med fløde 9 +20 Rødgrød med fløde 10 +SET character_set_results=utf8; +SELECT * FROM t1 ORDER BY a; +a b +1 Rødgrød med fløde 1 +2 Rødgrød med fløde 2 +3 Rødgrød med fløde 3 +4 Rødgrød med fløde 4 +5 Rødgrød med fløde 5 +6 Rødgrød med fløde 6 +7 Rødgrød med fløde 7 +8 Rødgrød med fløde 8 +9 Rødgrød med fløde 9 +10 Rødgrød med fløde 10 +11 Rødgrød med fløde 1 +12 Rødgrød med fløde 2 +13 Rødgrød med fløde 3 +14 Rødgrød med fløde 4 +15 Rødgrød med fløde 5 +16 Rødgrød med fløde 6 +17 Rødgrød med fløde 7 +18 Rødgrød med fløde 8 +19 Rødgrød med fløde 9 +20 Rødgrød med fløde 10 +include/stop_slave.inc +SET GLOBAL slave_parallel_threads=@old_parallel_threads; +include/start_slave.inc +DROP TABLE t1; +include/rpl_end.inc diff --git a/mysql-test/suite/rpl/t/rpl_parallel_charset.test b/mysql-test/suite/rpl/t/rpl_parallel_charset.test new file mode 100644 index 00000000000..3e0f4913886 --- /dev/null +++ b/mysql-test/suite/rpl/t/rpl_parallel_charset.test @@ -0,0 +1,56 @@ +--source include/have_binlog_format_statement.inc +--let $rpl_topology=1->2 +--source include/rpl_init.inc + +--echo *** MDEV-6156: Parallel replication incorrectly caches charset between worker threads *** + +--connection server_2 +SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads; +--source include/stop_slave.inc +SET GLOBAL slave_parallel_threads=5; +--source include/start_slave.inc + +--connection server_1 +CREATE TABLE t1 (a INT PRIMARY KEY, b VARCHAR(100) CHARACTER SET utf8); +SHOW CREATE TABLE t1; +SET character_set_client=latin1; +INSERT INTO t1 VALUES (1, 'Rřdgrřd med flřde 1'); +INSERT INTO t1 VALUES (2, 'Rřdgrřd med flřde 2'); +INSERT INTO t1 VALUES (3, 'Rřdgrřd med flřde 3'); +INSERT INTO t1 VALUES (4, 'Rřdgrřd med flřde 4'); +INSERT INTO t1 VALUES (5, 'Rřdgrřd med flřde 5'); +INSERT INTO t1 VALUES (6, 'Rřdgrřd med flřde 6'); +INSERT INTO t1 VALUES (7, 'Rřdgrřd med flřde 7'); +INSERT INTO t1 VALUES (8, 'Rřdgrřd med flřde 8'); +INSERT INTO t1 VALUES (9, 'Rřdgrřd med flřde 9'); +INSERT INTO t1 VALUES (10, 'Rřdgrřd med flřde 10'); +SET character_set_client=utf8; +INSERT INTO t1 VALUES (11, 'Rødgrød med fløde 1'); +INSERT INTO t1 VALUES (12, 'Rødgrød med fløde 2'); +INSERT INTO t1 VALUES (13, 'Rødgrød med fløde 3'); +INSERT INTO t1 VALUES (14, 'Rødgrød med fløde 4'); +INSERT INTO t1 VALUES (15, 'Rødgrød med fløde 5'); +INSERT INTO t1 VALUES (16, 'Rødgrød med fløde 6'); +INSERT INTO t1 VALUES (17, 'Rødgrød med fløde 7'); +INSERT INTO t1 VALUES (18, 'Rødgrød med fløde 8'); +INSERT INTO t1 VALUES (19, 'Rødgrød med fløde 9'); +INSERT INTO t1 VALUES (20, 'Rødgrød med fløde 10'); +SET character_set_results=utf8; +SELECT * FROM t1 ORDER BY a; +--save_master_pos + +--connection server_2 +--sync_with_master +SET character_set_results=utf8; +SELECT * FROM t1 ORDER BY a; + + +--connection server_2 +--source include/stop_slave.inc +SET GLOBAL slave_parallel_threads=@old_parallel_threads; +--source include/start_slave.inc + +--connection server_1 +DROP TABLE t1; + +--source include/rpl_end.inc diff --git a/sql/log_event.cc b/sql/log_event.cc index 6d2854a9f11..0a24d45113b 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -4153,7 +4153,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi, (sql_mode & ~(ulong) MODE_NO_DIR_IN_CREATE)); if (charset_inited) { - if (rli->cached_charset_compare(charset)) + if (rgi->cached_charset_compare(charset)) { /* Verify that we support the charsets found in the event. */ if (!(thd->variables.character_set_client= @@ -4169,7 +4169,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi, stop with EE_UNKNOWN_CHARSET in compare_errors (unless set to ignore this error). */ - set_slave_thread_default_charset(thd, rli); + set_slave_thread_default_charset(thd, rgi); goto compare_errors; } thd->update_charset(); // for the charset change to take effect @@ -6211,7 +6211,7 @@ int Rotate_log_event::do_update_pos(rpl_group_info *rgi) master is 4.0 then the events are in the slave's format (conversion). */ set_slave_thread_options(thd); - set_slave_thread_default_charset(thd, rli); + set_slave_thread_default_charset(thd, rgi); thd->variables.sql_mode= global_system_variables.sql_mode; thd->variables.auto_increment_increment= thd->variables.auto_increment_offset= 1; diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index 0d0c8c9df70..8104d0ff553 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -82,7 +82,6 @@ Relay_log_info::Relay_log_info(bool is_slave_recovery) max_relay_log_size= global_system_variables.max_relay_log_size; bzero((char*) &info_file, sizeof(info_file)); bzero((char*) &cache_buf, sizeof(cache_buf)); - cached_charset_invalidate(); mysql_mutex_init(key_relay_log_info_run_lock, &run_lock, MY_MUTEX_INIT_FAST); mysql_mutex_init(key_relay_log_info_data_lock, &data_lock, MY_MUTEX_INIT_FAST); @@ -1200,29 +1199,6 @@ bool Relay_log_info::is_until_satisfied(THD *thd, Log_event *ev) } -void Relay_log_info::cached_charset_invalidate() -{ - DBUG_ENTER("Relay_log_info::cached_charset_invalidate"); - - /* Full of zeroes means uninitialized. */ - bzero(cached_charset, sizeof(cached_charset)); - DBUG_VOID_RETURN; -} - - -bool Relay_log_info::cached_charset_compare(char *charset) const -{ - DBUG_ENTER("Relay_log_info::cached_charset_compare"); - - if (memcmp(cached_charset, charset, sizeof(cached_charset))) - { - memcpy(const_cast(cached_charset), charset, sizeof(cached_charset)); - DBUG_RETURN(1); - } - DBUG_RETURN(0); -} - - void Relay_log_info::stmt_done(my_off_t event_master_log_pos, time_t event_creation_time, THD *thd, rpl_group_info *rgi) @@ -1503,6 +1479,7 @@ rpl_group_info::rpl_group_info(Relay_log_info *rli) deferred_events(NULL), m_annotate_event(0), is_parallel_exec(false) { reinit(rli); + cached_charset_invalidate(); bzero(¤t_gtid, sizeof(current_gtid)); mysql_mutex_init(key_rpl_group_info_sleep_lock, &sleep_lock, MY_MUTEX_INIT_FAST); @@ -1585,6 +1562,29 @@ delete_or_keep_event_post_apply(rpl_group_info *rgi, } +void rpl_group_info::cached_charset_invalidate() +{ + DBUG_ENTER("rpl_group_info::cached_charset_invalidate"); + + /* Full of zeroes means uninitialized. */ + bzero(cached_charset, sizeof(cached_charset)); + DBUG_VOID_RETURN; +} + + +bool rpl_group_info::cached_charset_compare(char *charset) const +{ + DBUG_ENTER("rpl_group_info::cached_charset_compare"); + + if (memcmp(cached_charset, charset, sizeof(cached_charset))) + { + memcpy(const_cast(cached_charset), charset, sizeof(cached_charset)); + DBUG_RETURN(1); + } + DBUG_RETURN(0); +} + + void rpl_group_info::cleanup_context(THD *thd, bool error) { DBUG_ENTER("Relay_log_info::cleanup_context"); diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index 48193afce4d..bcb237b6ced 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -295,7 +295,6 @@ public: /* Condition for UNTIL master_gtid_pos. */ slave_connection_state until_gtid_pos; - char cached_charset[6]; /* retried_trans is a cumulative counter: how many times the slave has retried a transaction (any) since slave started. @@ -371,15 +370,6 @@ public: group_relay_log_pos); } - /* - Last charset (6 bytes) seen by slave SQL thread is cached here; it helps - the thread save 3 get_charset() per Query_log_event if the charset is not - changing from event to event (common situation). - When the 6 bytes are equal to 0 is used to mean "cache is invalidated". - */ - void cached_charset_invalidate(); - bool cached_charset_compare(char *charset) const; - /** Helper function to do after statement completion. @@ -546,6 +536,8 @@ struct rpl_group_info mysql_mutex_t sleep_lock; mysql_cond_t sleep_cond; + char cached_charset[6]; + /* trans_retries varies between 0 to slave_transaction_retries and counts how many times the slave has retried the present transaction; gets reset to 0 @@ -679,6 +671,15 @@ struct rpl_group_info return false; } + /* + Last charset (6 bytes) seen by slave SQL thread is cached here; it helps + the thread save 3 get_charset() per Query_log_event if the charset is not + changing from event to event (common situation). + When the 6 bytes are equal to 0 is used to mean "cache is invalidated". + */ + void cached_charset_invalidate(); + bool cached_charset_compare(char *charset) const; + void clear_tables_to_lock(); void cleanup_context(THD *, bool); void slave_close_thread_tables(THD *); diff --git a/sql/slave.cc b/sql/slave.cc index 43c97d0af9c..b5d8758a405 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -2879,7 +2879,7 @@ void set_slave_thread_options(THD* thd) DBUG_VOID_RETURN; } -void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli) +void set_slave_thread_default_charset(THD* thd, rpl_group_info *rgi) { DBUG_ENTER("set_slave_thread_default_charset"); @@ -2891,13 +2891,7 @@ void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli) global_system_variables.collation_server; thd->update_charset(); - /* - We use a const cast here since the conceptual (and externally - visible) behavior of the function is to set the default charset of - the thread. That the cache has to be invalidated is a secondary - effect. - */ - const_cast(rli)->cached_charset_invalidate(); + rgi->cached_charset_invalidate(); DBUG_VOID_RETURN; } @@ -4682,7 +4676,7 @@ err_during_init: mysql_cond_broadcast(&rli->data_cond); rli->ignore_log_space_limit= 0; /* don't need any lock */ /* we die so won't remember charset - re-update them on next thread start */ - rli->cached_charset_invalidate(); + serial_rgi->cached_charset_invalidate(); /* TODO: see if we can do this conditionally in next_event() instead diff --git a/sql/slave.h b/sql/slave.h index 3981a9d4f2c..aa3976f6e6c 100644 --- a/sql/slave.h +++ b/sql/slave.h @@ -51,6 +51,7 @@ class Relay_log_info; class Master_info; class Master_info_index; +struct rpl_group_info; struct rpl_parallel_thread; int init_intvar_from_file(int* var, IO_CACHE* f, int default_val); @@ -226,7 +227,7 @@ int init_relay_log_pos(Relay_log_info* rli,const char* log,ulonglong pos, int purge_relay_logs(Relay_log_info* rli, THD *thd, bool just_reset, const char** errmsg); void set_slave_thread_options(THD* thd); -void set_slave_thread_default_charset(THD *thd, Relay_log_info const *rli); +void set_slave_thread_default_charset(THD *thd, rpl_group_info *rgi); int rotate_relay_log(Master_info* mi); int apply_event_and_update_pos(Log_event* ev, THD* thd, struct rpl_group_info *rgi, -- cgit v1.2.1