maria-developers team mailing list archive
-
maria-developers team
-
Mailing list archive
-
Message #07757
Re: [Commits] Rev 4391: MDEV-6676: Speculative parallel replication. in http://bazaar.launchpad.net/~maria-captains/maria/10.0
how about also adding a mode = 'auto'
which currently can be == transaction,
but in future could be extended to based on statistics
flip actual modes back and forth ?
/Jonas
On Tue, Oct 7, 2014 at 4:20 PM, <knielsen@xxxxxxxxxxxxxxx> wrote:
> At http://bazaar.launchpad.net/~maria-captains/maria/10.0
>
> ------------------------------------------------------------
> revno: 4391
> revision-id: knielsen@xxxxxxxxxxxxxxx-20141007142037-y85xq099ltdiqno6
> parent: knielsen@xxxxxxxxxxxxxxx-20141003094323-nosx0y7vv0kjv20e
> committer: Kristian Nielsen <knielsen@xxxxxxxxxxxxxxx>
> branch nick: work-10.0-mdev6676
> timestamp: Tue 2014-10-07 16:20:37 +0200
> message:
> MDEV-6676: Speculative parallel replication.
>
> Better syntax for configuring the parallel mode. Now it is a CHANGE
> MASTER
> option instead of a system variable, which makes it possible to configure
> differently for different multi-source replication masters.
>
> Now also the domain-based replication mode can be disabled.
> === modified file 'mysql-test/include/check-testcase.test'
> --- a/mysql-test/include/check-testcase.test 2013-12-16 12:02:21 +0000
> +++ b/mysql-test/include/check-testcase.test 2014-10-07 14:20:37 +0000
> @@ -64,6 +64,7 @@ if ($tmp)
> --echo Master_SSL_Crlpath #
> --echo Using_Gtid No
> --echo Gtid_IO_Pos #
> + --echo Parallel_Mode domain,groupcommit
> }
> if (!$tmp) {
> # Note: after WL#5177, fields 13-18 shall not be filtered-out.
>
> === modified file 'mysql-test/r/mysqld--help.result'
> --- a/mysql-test/r/mysqld--help.result 2014-09-19 13:25:37 +0000
> +++ b/mysql-test/r/mysqld--help.result 2014-10-07 14:20:37 +0000
> @@ -915,17 +915,6 @@
> parallel replication thread when reading ahead in the
> relay log looking for opportunities for parallel
> replication. Only used when --slave-parallel-threads > 0.
> - --slave-parallel-mode=name
> - Controls what transactions are applied in parallel when
> - using --slave-parallel-threads. Syntax:
> - slave_paralle_mode=value[,value...], where "value" could
> - be one or more of: "domain", to apply different
> - replication domains in parallel; "groupcommit", to apply
> - in parallel transactions that group-committed together on
> - the master; "transactional", to optimistically try to
> - apply all transactional DML in parallel; and "waiting" to
> - extend "transactional" to even transactions that had to
> - wait on the master.
> --slave-parallel-threads=#
> If non-zero, number of threads to spawn to apply in
> parallel events on the slave that were group-committed on
> @@ -1320,7 +1309,6 @@ slave-exec-mode STRICT
> slave-max-allowed-packet 1073741824
> slave-net-timeout 3600
> slave-parallel-max-queued 131072
> -slave-parallel-mode domain,groupcommit
> slave-parallel-threads 0
> slave-skip-errors (No default value)
> slave-sql-verify-checksum TRUE
>
> === modified file 'mysql-test/suite/rpl/r/rpl_parallel.result'
> --- a/mysql-test/suite/rpl/r/rpl_parallel.result 2014-08-15
> 09:31:13 +0000
> +++ b/mysql-test/suite/rpl/r/rpl_parallel.result 2014-10-07
> 14:20:37 +0000
> @@ -1,5 +1,5 @@
> -include/rpl_init.inc [topology=1->2]
> -SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
> +include/master-slave.inc
> +[connection master]
> SET GLOBAL slave_parallel_threads=10;
> ERROR HY000: This operation cannot be performed as you have a running
> slave ''; run STOP SLAVE '' first
> include/stop_slave.inc
> @@ -923,8 +923,79 @@ SELECT * FROM t2 WHERE a >= 30 ORDER BY
> 32
> 33
> 34
> +*** MDEV-6676 - test syntax of CHANGE MASTER TO PARALLEL_MODE=xxx ***
> +Parallel_Mode = 'domain,groupcommit'
> include/stop_slave.inc
> -SET GLOBAL slave_parallel_threads=@old_parallel_threads;
> +CHANGE MASTER TO parallel_mode=(,domain,groupcommit);
> +ERROR 42000: You have an error in your SQL syntax; check the manual that
> corresponds to your MariaDB server version for the right syntax to use near
> 'domain,groupcommit)' at line 1
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit,transactional);
> +ERROR HY000: Invalid use of 'transactional' option for PARALLEL_MODE in
> CHANGE MASTER TO
> +CHANGE MASTER TO parallel_mode=(waiting,domain,transactional,waiting);
> +ERROR HY000: Invalid use of 'waiting' option for PARALLEL_MODE in CHANGE
> MASTER TO
> +CHANGE MASTER TO parallel_mode=(domain,domain);
> +ERROR HY000: Invalid use of 'domain' option for PARALLEL_MODE in CHANGE
> MASTER TO
> +CHANGE MASTER TO parallel_mode=(waiting,transactional,domain);
> +Parallel_Mode = 'domain,transactional,waiting'
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> +Parallel_Mode = 'domain,groupcommit'
> +*** MDEV-6676 - test that empty parallel_mode does not replicate in
> parallel ***
> +INSERT INTO t2 VALUES (40);
> +include/save_master_gtid.inc
> +CHANGE MASTER TO parallel_mode=();
> +SET @old_dbug= @@GLOBAL.debug_dbug;
> +SET GLOBAL debug_dbug="+d,slave_crash_if_parallel_apply";
> +include/start_slave.inc
> +include/sync_with_master_gtid.inc
> +SELECT * FROM t2 WHERE a >= 40 ORDER BY a;
> +a
> +40
> +include/stop_slave.inc
> +SET GLOBAL debug_dbug=@old_dbug;
> +*** MDEV-6676 - test disabling domain-based parallel replication ***
> +SET gtid_domain_id = 1;
> +INSERT INTO t2 VALUES (41);
> +INSERT INTO t2 VALUES (42);
> +INSERT INTO t2 VALUES (43);
> +INSERT INTO t2 VALUES (44);
> +INSERT INTO t2 VALUES (45);
> +INSERT INTO t2 VALUES (46);
> +DELETE FROM t2 WHERE a >= 41;
> +SET gtid_domain_id = 2;
> +INSERT INTO t2 VALUES (41);
> +INSERT INTO t2 VALUES (42);
> +INSERT INTO t2 VALUES (43);
> +INSERT INTO t2 VALUES (44);
> +INSERT INTO t2 VALUES (45);
> +INSERT INTO t2 VALUES (46);
> +SET gtid_domain_id = 0;
> +include/save_master_gtid.inc
> +CHANGE MASTER TO parallel_mode=(groupcommit);
> +include/start_slave.inc
> +include/sync_with_master_gtid.inc
> +SELECT * FROM t2 WHERE a >= 40 ORDER BY a;
> +a
> +40
> +41
> +42
> +43
> +44
> +45
> +46
> +include/stop_slave.inc
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> +include/start_slave.inc
> +*** MDEV-6676 - test that parallel mode is saved correctly in master.info
> across server restart ***
> +Parallel_Mode = 'domain,groupcommit'
> +include/stop_slave.inc
> +CHANGE MASTER TO parallel_mode=(transactional,waiting);
> +Parallel_Mode = 'transactional,waiting'
> +include/start_slave.inc
> +include/rpl_restart_server.inc [server_number=2]
> +Parallel_Mode = 'transactional,waiting'
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> +include/start_slave.inc
> +include/stop_slave.inc
> +SET GLOBAL slave_parallel_threads=0;
> include/start_slave.inc
> SET DEBUG_SYNC= 'RESET';
> DROP function foo;
>
> === modified file 'mysql-test/suite/rpl/r/rpl_parallel_optimistic.result'
> --- a/mysql-test/suite/rpl/r/rpl_parallel_optimistic.result 2014-09-24
> 19:25:10 +0000
> +++ b/mysql-test/suite/rpl/r/rpl_parallel_optimistic.result 2014-10-07
> 14:20:37 +0000
> @@ -2,11 +2,9 @@ include/rpl_init.inc [topology=1->2]
> ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
> CREATE TABLE t1 (a int PRIMARY KEY, b INT) ENGINE=InnoDB;
> SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
> -SET @old_mode=@@GLOBAL.slave_parallel_mode;
> include/stop_slave.inc
> SET GLOBAL slave_parallel_threads=10;
> -SET GLOBAL slave_parallel_mode="domain,transactional";
> -CHANGE MASTER TO master_use_gtid=slave_pos;
> +CHANGE MASTER TO master_use_gtid=slave_pos,
> parallel_mode=(domain,transactional);
> INSERT INTO t1 VALUES(1,1);
> BEGIN;
> INSERT INTO t1 VALUES(2,1);
> @@ -288,8 +286,8 @@ SET GLOBAL binlog_format= @old_format;
> SET GLOBAL tx_isolation= @old_isolation;
> include/start_slave.inc
> include/stop_slave.inc
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> SET GLOBAL slave_parallel_threads=@old_parallel_threads;
> -SET GLOBAL slave_parallel_mode=@old_mode;
> include/start_slave.inc
> DROP TABLE t1, t2, t3;
> include/rpl_end.inc
>
> === modified file
> 'mysql-test/suite/rpl/r/rpl_parallel_optimistic_nobinlog.result'
> --- a/mysql-test/suite/rpl/r/rpl_parallel_optimistic_nobinlog.result
> 2014-09-24 19:25:10 +0000
> +++ b/mysql-test/suite/rpl/r/rpl_parallel_optimistic_nobinlog.result
> 2014-10-07 14:20:37 +0000
> @@ -6,12 +6,10 @@ INSERT INTO t1 VALUES (1,0), (2,0), (3,0
> INSERT INTO t2 VALUES (1,0), (2,0);
> SET @old_isolation= @@GLOBAL.tx_isolation;
> SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
> -SET @old_mode=@@GLOBAL.slave_parallel_mode;
> include/stop_slave.inc
> SET GLOBAL TRANSACTION ISOLATION LEVEL READ COMMITTED;
> SET GLOBAL slave_parallel_threads=10;
> -SET GLOBAL slave_parallel_mode="domain,transactional,waiting";
> -CHANGE MASTER TO master_use_gtid=slave_pos;
> +CHANGE MASTER TO master_use_gtid=slave_pos,
> parallel_mode=(domain,transactional,waiting);
> *** Test that we replicate correctly when using READ COMMITTED and
> --log-slave-updates=0 on the slave ***
> INSERT INTO t1 SELECT 4, COUNT(*) FROM t2;
> INSERT INTO t2 SELECT 4, COUNT(*) FROM t1;
> @@ -78,8 +76,8 @@ a b
> 10 10
> include/stop_slave.inc
> SET GLOBAL tx_isolation= @old_isolation;
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> SET GLOBAL slave_parallel_threads=@old_parallel_threads;
> -SET GLOBAL slave_parallel_mode=@old_mode;
> include/start_slave.inc
> DROP TABLE t1, t2;
> include/rpl_end.inc
>
> === modified file 'mysql-test/suite/rpl/t/rpl_parallel.test'
> --- a/mysql-test/suite/rpl/t/rpl_parallel.test 2014-08-15 09:31:13 +0000
> +++ b/mysql-test/suite/rpl/t/rpl_parallel.test 2014-10-07 14:20:37 +0000
> @@ -1,13 +1,12 @@
> --source include/have_innodb.inc
> --source include/have_debug.inc
> --source include/have_debug_sync.inc
> ---let $rpl_topology=1->2
> ---source include/rpl_init.inc
> +--source include/master-slave.inc
>
> # Test various aspects of parallel replication.
>
> --connection server_2
> -SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
> +--let $old_parallel_threads=`SELECT @@GLOBAL.slave_parallel_threads`
> --error ER_SLAVE_MUST_STOP
> SET GLOBAL slave_parallel_threads=10;
> --source include/stop_slave.inc
> @@ -1466,9 +1465,100 @@ SET sql_slave_skip_counter= 1;
> SELECT * FROM t2 WHERE a >= 30 ORDER BY a;
>
>
> +--echo *** MDEV-6676 - test syntax of CHANGE MASTER TO PARALLEL_MODE=xxx
> ***
> +--connection server_2
> +
> +--let $status_items= Parallel_Mode
> +--source include/show_slave_status.inc
> +--source include/stop_slave.inc
> +--error ER_PARSE_ERROR
> +CHANGE MASTER TO parallel_mode=(,domain,groupcommit);
> +--error ER_INVALID_SLAVE_PARALLEL_MODE
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit,transactional);
> +--error ER_INVALID_SLAVE_PARALLEL_MODE
> +CHANGE MASTER TO parallel_mode=(waiting,domain,transactional,waiting);
> +--error ER_INVALID_SLAVE_PARALLEL_MODE
> +CHANGE MASTER TO parallel_mode=(domain,domain);
> +CHANGE MASTER TO parallel_mode=(waiting,transactional,domain);
> +--let $status_items= Parallel_Mode
> +--source include/show_slave_status.inc
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> +--let $status_items= Parallel_Mode
> +--source include/show_slave_status.inc
> +
> +
> +--echo *** MDEV-6676 - test that empty parallel_mode does not replicate
> in parallel ***
> +--connection server_1
> +INSERT INTO t2 VALUES (40);
> +--source include/save_master_gtid.inc
> +
> +--connection server_2
> +CHANGE MASTER TO parallel_mode=();
> +# Test that we do not use parallel apply, by injecting an unconditional
> +# crash in the parallel apply code.
> +SET @old_dbug= @@GLOBAL.debug_dbug;
> +SET GLOBAL debug_dbug="+d,slave_crash_if_parallel_apply";
> +--source include/start_slave.inc
> +--source include/sync_with_master_gtid.inc
> +SELECT * FROM t2 WHERE a >= 40 ORDER BY a;
> +--source include/stop_slave.inc
> +SET GLOBAL debug_dbug=@old_dbug;
> +
> +
> +--echo *** MDEV-6676 - test disabling domain-based parallel replication
> ***
> +--connection server_1
> +# Let's do a bunch of transactions that will conflict if run out-of-order
> in
> +# domain-based parallel replication mode.
> +SET gtid_domain_id = 1;
> +INSERT INTO t2 VALUES (41);
> +INSERT INTO t2 VALUES (42);
> +INSERT INTO t2 VALUES (43);
> +INSERT INTO t2 VALUES (44);
> +INSERT INTO t2 VALUES (45);
> +INSERT INTO t2 VALUES (46);
> +DELETE FROM t2 WHERE a >= 41;
> +SET gtid_domain_id = 2;
> +INSERT INTO t2 VALUES (41);
> +INSERT INTO t2 VALUES (42);
> +INSERT INTO t2 VALUES (43);
> +INSERT INTO t2 VALUES (44);
> +INSERT INTO t2 VALUES (45);
> +INSERT INTO t2 VALUES (46);
> +SET gtid_domain_id = 0;
> +--source include/save_master_gtid.inc
> +--connection server_2
> +CHANGE MASTER TO parallel_mode=(groupcommit);
> +--source include/start_slave.inc
> +--source include/sync_with_master_gtid.inc
> +SELECT * FROM t2 WHERE a >= 40 ORDER BY a;
> +--source include/stop_slave.inc
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> +--source include/start_slave.inc
> +
> +
> +--echo *** MDEV-6676 - test that parallel mode is saved correctly in
> master.info across server restart ***
> +--connection server_2
> +--let $status_items= Parallel_Mode
> +--source include/show_slave_status.inc
> +--source include/stop_slave.inc
> +CHANGE MASTER TO parallel_mode=(transactional,waiting);
> +--let $status_items= Parallel_Mode
> +--source include/show_slave_status.inc
> +--source include/start_slave.inc
> +
> +--let $rpl_server_number= 2
> +--source include/rpl_restart_server.inc
> +
> +--connection server_2
> +--let $status_items= Parallel_Mode
> +--source include/show_slave_status.inc
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> +--source include/start_slave.inc
> +
> +
> --connection server_2
> --source include/stop_slave.inc
> -SET GLOBAL slave_parallel_threads=@old_parallel_threads;
> +eval SET GLOBAL slave_parallel_threads=$old_parallel_threads;
> --source include/start_slave.inc
> SET DEBUG_SYNC= 'RESET';
>
>
> === modified file 'mysql-test/suite/rpl/t/rpl_parallel_optimistic.test'
> --- a/mysql-test/suite/rpl/t/rpl_parallel_optimistic.test 2014-09-24
> 19:25:10 +0000
> +++ b/mysql-test/suite/rpl/t/rpl_parallel_optimistic.test 2014-10-07
> 14:20:37 +0000
> @@ -11,11 +11,9 @@ CREATE TABLE t1 (a int PRIMARY KEY, b IN
> --connection server_2
> --sync_with_master
> SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
> -SET @old_mode=@@GLOBAL.slave_parallel_mode;
> --source include/stop_slave.inc
> SET GLOBAL slave_parallel_threads=10;
> -SET GLOBAL slave_parallel_mode="domain,transactional";
> -CHANGE MASTER TO master_use_gtid=slave_pos;
> +CHANGE MASTER TO master_use_gtid=slave_pos,
> parallel_mode=(domain,transactional);
>
>
> --connection server_1
> @@ -309,8 +307,8 @@ SET GLOBAL tx_isolation= @old_isolation;
>
> --connection server_2
> --source include/stop_slave.inc
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> SET GLOBAL slave_parallel_threads=@old_parallel_threads;
> -SET GLOBAL slave_parallel_mode=@old_mode;
> --source include/start_slave.inc
>
> --connection server_1
>
> === modified file
> 'mysql-test/suite/rpl/t/rpl_parallel_optimistic_nobinlog.test'
> --- a/mysql-test/suite/rpl/t/rpl_parallel_optimistic_nobinlog.test
> 2014-09-24 19:25:10 +0000
> +++ b/mysql-test/suite/rpl/t/rpl_parallel_optimistic_nobinlog.test
> 2014-10-07 14:20:37 +0000
> @@ -16,12 +16,10 @@ INSERT INTO t2 VALUES (1,0), (2,0);
> --sync_with_master
> SET @old_isolation= @@GLOBAL.tx_isolation;
> SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
> -SET @old_mode=@@GLOBAL.slave_parallel_mode;
> --source include/stop_slave.inc
> SET GLOBAL TRANSACTION ISOLATION LEVEL READ COMMITTED;
> SET GLOBAL slave_parallel_threads=10;
> -SET GLOBAL slave_parallel_mode="domain,transactional,waiting";
> -CHANGE MASTER TO master_use_gtid=slave_pos;
> +CHANGE MASTER TO master_use_gtid=slave_pos,
> parallel_mode=(domain,transactional,waiting);
>
>
> --echo *** Test that we replicate correctly when using READ COMMITTED and
> --log-slave-updates=0 on the slave ***
> @@ -63,8 +61,8 @@ SELECT * FROM t2 ORDER BY a;
> --connection server_2
> --source include/stop_slave.inc
> SET GLOBAL tx_isolation= @old_isolation;
> +CHANGE MASTER TO parallel_mode=(domain,groupcommit);
> SET GLOBAL slave_parallel_threads=@old_parallel_threads;
> -SET GLOBAL slave_parallel_mode=@old_mode;
> --source include/start_slave.inc
>
> --connection server_1
>
> === modified file 'sql/lex.h'
> --- a/sql/lex.h 2014-02-02 09:00:36 +0000
> +++ b/sql/lex.h 2014-10-07 14:20:37 +0000
> @@ -194,6 +194,7 @@ static SYMBOL symbols[] = {
> { "DISTINCTROW", SYM(DISTINCT)}, /* Access likes this */
> { "DIV", SYM(DIV_SYM)},
> { "DO", SYM(DO_SYM)},
> + { "DOMAIN", SYM(DOMAIN_SYM)},
> { "DOUBLE", SYM(DOUBLE_SYM)},
> { "DROP", SYM(DROP)},
> { "DUAL", SYM(DUAL_SYM)},
> @@ -257,6 +258,7 @@ static SYMBOL symbols[] = {
> { "GRANT", SYM(GRANT)},
> { "GRANTS", SYM(GRANTS)},
> { "GROUP", SYM(GROUP_SYM)},
> + { "GROUPCOMMIT", SYM(GROUPCOMMIT_SYM)},
> { "HANDLER", SYM(HANDLER_SYM)},
> { "HARD", SYM(HARD_SYM)},
> { "HASH", SYM(HASH_SYM)},
> @@ -429,6 +431,7 @@ static SYMBOL symbols[] = {
> { "PACK_KEYS", SYM(PACK_KEYS_SYM)},
> { "PAGE", SYM(PAGE_SYM)},
> { "PAGE_CHECKSUM", SYM(PAGE_CHECKSUM_SYM)},
> + { "PARALLEL_MODE", SYM(PARALLEL_MODE_SYM)},
> { "PARSER", SYM(PARSER_SYM)},
> { "PARSE_VCOL_EXPR", SYM(PARSE_VCOL_EXPR_SYM)},
> { "PARTIAL", SYM(PARTIAL)},
> @@ -650,6 +653,7 @@ static SYMBOL symbols[] = {
> { "VIEW", SYM(VIEW_SYM)},
> { "VIRTUAL", SYM(VIRTUAL_SYM)},
> { "WAIT", SYM(WAIT_SYM)},
> + { "WAITING", SYM(WAITING_SYM)},
> { "WARNINGS", SYM(WARNINGS)},
> { "WEEK", SYM(WEEK_SYM)},
> { "WEIGHT_STRING", SYM(WEIGHT_STRING_SYM)},
>
> === modified file 'sql/log_event.cc'
> --- a/sql/log_event.cc 2014-09-25 13:47:58 +0000
> +++ b/sql/log_event.cc 2014-10-07 14:20:37 +0000
> @@ -12728,7 +12728,7 @@ bool rpl_get_position_info(const char **
> return FALSE;
> #else
> const Relay_log_info *rli= &(active_mi->rli);
> - if (opt_slave_parallel_threads == 0)
> + if (!rli->mi->using_parallel())
> {
> *log_file_name= rli->group_master_log_name;
> *log_pos= rli->group_master_log_pos +
>
> === modified file 'sql/mysqld.cc'
> --- a/sql/mysqld.cc 2014-09-26 10:37:19 +0000
> +++ b/sql/mysqld.cc 2014-10-07 14:20:37 +0000
> @@ -562,8 +562,6 @@ ulong stored_program_cache_size= 0;
>
> ulong opt_slave_parallel_threads= 0;
> ulong opt_slave_domain_parallel_threads= 0;
> -ulonglong opt_slave_parallel_mode=
> - SLAVE_PARALLEL_DOMAIN | SLAVE_PARALLEL_GROUPCOMMIT;
> ulong opt_binlog_commit_wait_count= 0;
> ulong opt_binlog_commit_wait_usec= 0;
> ulong opt_slave_parallel_max_queued= 131072;
>
> === modified file 'sql/mysqld.h'
> --- a/sql/mysqld.h 2014-09-26 10:37:19 +0000
> +++ b/sql/mysqld.h 2014-10-07 14:20:37 +0000
> @@ -187,7 +187,6 @@ extern ulong stored_program_cache_size;
> extern ulong opt_slave_parallel_threads;
> extern ulong opt_slave_domain_parallel_threads;
> extern ulong opt_slave_parallel_max_queued;
> -extern ulonglong opt_slave_parallel_mode;
> extern ulong opt_binlog_commit_wait_count;
> extern ulong opt_binlog_commit_wait_usec;
> extern my_bool opt_gtid_ignore_duplicates;
>
> === modified file 'sql/rpl_mi.cc'
> --- a/sql/rpl_mi.cc 2014-09-02 12:07:01 +0000
> +++ b/sql/rpl_mi.cc 2014-10-07 14:20:37 +0000
> @@ -40,7 +40,8 @@ Master_info::Master_info(LEX_STRING *con
> heartbeat_period(0), received_heartbeats(0), master_id(0),
> prev_master_id(0),
> using_gtid(USE_GTID_NO), events_queued_since_last_gtid(0),
> - gtid_reconnect_event_skip_count(0), gtid_event_seen(false)
> + gtid_reconnect_event_skip_count(0), gtid_event_seen(false),
> + parallel_mode(SLAVE_PARALLEL_DOMAIN|SLAVE_PARALLEL_GROUPCOMMIT)
> {
> host[0] = 0; user[0] = 0; password[0] = 0;
> ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0;
> @@ -178,6 +179,7 @@ void init_master_log_pos(Master_info* mi
> mi->events_queued_since_last_gtid= 0;
> mi->gtid_reconnect_event_skip_count= 0;
> mi->gtid_event_seen= false;
> + mi->parallel_mode= SLAVE_PARALLEL_DOMAIN|SLAVE_PARALLEL_GROUPCOMMIT;
>
> /* Intentionally init ssl_verify_server_cert to 0, no option available
> */
> mi->ssl_verify_server_cert= 0;
> @@ -514,6 +516,22 @@ file '%s')", fname);
> else
> mi->using_gtid= Master_info::USE_GTID_NO;
> }
> + else if (!strncmp(buf, STRING_WITH_LEN("parallel_mode=")))
> + {
> + int val= atoi(buf + (sizeof("parallel_mode=") - 1));
> + mi->parallel_mode= val & (SLAVE_PARALLEL_DOMAIN |
> + SLAVE_PARALLEL_GROUPCOMMIT |
> + SLAVE_PARALLEL_TRX |
> + SLAVE_PARALLEL_WAITING);
> + }
> + else if (!strncmp(buf, STRING_WITH_LEN("END_MARKER")))
> + {
> + /*
> + Guard agaist extra left-overs at the end of file, in case a
> later
> + update causes the file to shrink compared to earlier
> contents.
> + */
> + break;
> + }
> }
> }
> }
> @@ -657,7 +675,9 @@ int flush_master_info(Master_info* mi,
> my_b_printf(file,
>
> "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n%s\n%s\n%s\n%d\n%s\n%s\n"
> "\n\n\n\n\n\n\n\n\n\n\n"
> - "using_gtid=%d\n",
> + "using_gtid=%d\n"
> + "parallel_mode=%u\n"
> + "END_MARKER\n",
> LINES_IN_MASTER_INFO,
> mi->master_log_name, llstr(mi->master_log_pos, lbuf),
> mi->host, mi->user,
> @@ -666,7 +686,7 @@ int flush_master_info(Master_info* mi,
> mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert,
> heartbeat_buf, "", ignore_server_ids_buf,
> "", 0,
> - mi->ssl_crl, mi->ssl_crlpath, mi->using_gtid);
> + mi->ssl_crl, mi->ssl_crlpath, mi->using_gtid,
> mi->parallel_mode);
> my_free(ignore_server_ids_buf);
> err= flush_io_cache(file);
> if (sync_masterinfo_period && !err &&
>
> === modified file 'sql/rpl_mi.h'
> --- a/sql/rpl_mi.h 2014-09-02 12:07:01 +0000
> +++ b/sql/rpl_mi.h 2014-10-07 14:20:37 +0000
> @@ -75,6 +75,10 @@ class Master_info : public Slave_reporti
> return connection_name.str == 0;
> }
> static const char *using_gtid_astext(enum enum_using_gtid arg);
> + bool using_parallel()
> + {
> + return opt_slave_parallel_threads > 0 && parallel_mode != 0;
> + }
>
> /* the variables below are needed because we can change masters on the
> fly */
> char master_log_name[FN_REFLEN+6]; /* Room for multi-*/
> @@ -178,6 +182,12 @@ class Master_info : public Slave_reporti
> uint64 gtid_reconnect_event_skip_count;
> /* gtid_event_seen is false until we receive first GTID event from
> master. */
> bool gtid_event_seen;
> + /*
> + The parallel replication modes, if any. A combination (binary OR) of
> any of
> + SLAVE_PARALLEL_DOMAIN, SLAVE_PARALLEL_GROUPCOMMIT,
> SLAVE_PARALLEL_TRX, and
> + SLAVE_PARALLEL_WAITING.
> + */
> + uint32 parallel_mode;
> };
> int init_master_info(Master_info* mi, const char* master_info_fname,
> const char* slave_info_fname,
>
> === modified file 'sql/rpl_parallel.cc'
> --- a/sql/rpl_parallel.cc 2014-10-03 09:43:23 +0000
> +++ b/sql/rpl_parallel.cc 2014-10-07 14:20:37 +0000
> @@ -1891,6 +1891,7 @@ rpl_parallel::do_event(rpl_group_info *s
> bool did_enter_cond= false;
> PSI_stage_info old_stage;
>
> + DBUG_EXECUTE_IF("slave_crash_if_parallel_apply", DBUG_SUICIDE(););
> /* Handle master log name change, seen in Rotate_log_event. */
> typ= ev->get_type_code();
> if (unlikely(typ == ROTATE_EVENT))
> @@ -1971,7 +1972,8 @@ rpl_parallel::do_event(rpl_group_info *s
> if (typ == GTID_EVENT)
> {
> Gtid_log_event *gtid_ev= static_cast<Gtid_log_event *>(ev);
> - uint32 domain_id= (rli->mi->using_gtid == Master_info::USE_GTID_NO ?
> + uint32 domain_id= (rli->mi->using_gtid == Master_info::USE_GTID_NO ||
> + !(rli->mi->parallel_mode & SLAVE_PARALLEL_DOMAIN) ?
> 0 : gtid_ev->domain_id);
> if (!(e= find(domain_id)))
> {
> @@ -2012,7 +2014,7 @@ rpl_parallel::do_event(rpl_group_info *s
> {
> Gtid_log_event *gtid_ev= static_cast<Gtid_log_event *>(ev);
> bool new_gco;
> - ulonglong mode= opt_slave_parallel_mode;
> + ulonglong mode= rli->mi->parallel_mode;
> uchar gtid_flags= gtid_ev->flags2;
> group_commit_orderer *gco;
> uint8 force_switch_flag;
>
> === modified file 'sql/rpl_parallel.h'
> --- a/sql/rpl_parallel.h 2014-10-03 09:43:23 +0000
> +++ b/sql/rpl_parallel.h 2014-10-07 14:20:37 +0000
> @@ -12,7 +12,11 @@ class Relay_log_info;
> struct inuse_relaylog;
>
>
> -/* Bit masks for the values in --slave-parallel-mode. */
> +/*
> + Bit masks for the values in --slave-parallel-mode.
> + Note that these values cannot be changed - they are stored in
> master.info,
> + so need to be possible to read back in a different version of the
> server.
> +*/
> #define SLAVE_PARALLEL_DOMAIN (1ULL << 0)
> #define SLAVE_PARALLEL_GROUPCOMMIT (1ULL << 1)
> #define SLAVE_PARALLEL_TRX (1ULL << 2)
>
> === modified file 'sql/share/errmsg-utf8.txt'
> --- a/sql/share/errmsg-utf8.txt 2014-07-08 17:38:26 +0000
> +++ b/sql/share/errmsg-utf8.txt 2014-10-07 14:20:37 +0000
> @@ -7111,3 +7111,5 @@ ER_SLAVE_SKIP_NOT_IN_GTID
> eng "When using GTID, @@sql_slave_skip_counter can not be used.
> Instead, setting @@gtid_slave_pos explicitly can be used to skip to after a
> given GTID position."
> ER_TABLE_DEFINITION_TOO_BIG
> eng "The definition for table %`s is too big"
> +ER_INVALID_SLAVE_PARALLEL_MODE
> + eng "Invalid use of '%s' option for PARALLEL_MODE in CHANGE
> MASTER TO"
>
> === modified file 'sql/slave.cc'
> --- a/sql/slave.cc 2014-09-02 12:07:01 +0000
> +++ b/sql/slave.cc 2014-10-07 14:20:37 +0000
> @@ -625,8 +625,7 @@ int terminate_slave_threads(Master_info*
> if (thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL))
> {
> DBUG_PRINT("info",("Terminating SQL thread"));
> - if (opt_slave_parallel_threads > 0 &&
> - mi->rli.abort_slave && mi->rli.stop_for_until)
> + if (mi->using_parallel() && mi->rli.abort_slave &&
> mi->rli.stop_for_until)
> {
> mi->rli.stop_for_until= false;
> mi->rli.parallel.stop_during_until();
> @@ -2576,6 +2575,8 @@ static bool send_show_master_info_header
> field_list.push_back(new Item_empty_string("Using_Gtid",
> sizeof("Current_Pos")-1));
> field_list.push_back(new Item_empty_string("Gtid_IO_Pos", 30));
> + field_list.push_back(new Item_empty_string("Parallel_Mode",
> + sizeof("domain,groupcommit,transactional,waiting")-1));
> if (full)
> {
> field_list.push_back(new Item_return_int("Retried_transactions",
> @@ -2708,8 +2709,7 @@ static bool send_show_master_info_data(T
> else
> {
> idle= mi->rli.sql_thread_caught_up;
> - if (opt_slave_parallel_threads > 0 && idle &&
> - !mi->rli.parallel.workers_idle())
> + if (mi->using_parallel() && idle &&
> !mi->rli.parallel.workers_idle())
> idle= false;
> }
> if (idle)
> @@ -2786,13 +2786,34 @@ static bool send_show_master_info_data(T
> protocol->store(mi->ssl_ca, &my_charset_bin);
> // Master_Ssl_Crlpath
> protocol->store(mi->ssl_capath, &my_charset_bin);
> + // Using_Gtid
> protocol->store(mi->using_gtid_astext(mi->using_gtid),
> &my_charset_bin);
> + // Gtid_IO_Pos
> {
> char buff[30];
> String tmp(buff, sizeof(buff), system_charset_info);
> mi->gtid_current_pos.to_string(&tmp);
> protocol->store(tmp.ptr(), tmp.length(), &my_charset_bin);
> }
> + // Parallel_Mode
> + {
> + /* Note how sizeof("domain") has room for "domain," due to traling
> 0. */
> + char buf[sizeof("domain") + sizeof("groupcommit") +
> + sizeof("transactional") + sizeof("waiting") + 1];
> + char *p= buf;
> + uint32 mode= mi->parallel_mode;
> + if (mode & SLAVE_PARALLEL_DOMAIN)
> + p= strmov(p, "domain,");
> + if (mode & SLAVE_PARALLEL_GROUPCOMMIT)
> + p= strmov(p, "groupcommit,");
> + if (mode & SLAVE_PARALLEL_TRX)
> + p= strmov(p, "transactional,");
> + if (mode & SLAVE_PARALLEL_WAITING)
> + p= strmov(p, "waiting,");
> + if (p != buf)
> + --p; // Discard last ','
> + protocol->store(buf, p-buf, &my_charset_bin);
> + }
> if (full)
> {
> protocol->store((uint32) mi->rli.retried_trans);
> @@ -3521,7 +3542,7 @@ static int exec_relay_log_event(THD* thd
>
> update_state_of_relay_log(rli, ev);
>
> - if (opt_slave_parallel_threads > 0)
> + if (rli->mi->using_parallel())
> {
> int res= rli->parallel.do_event(serial_rgi, ev, event_size);
> if (res >= 0)
> @@ -4667,7 +4688,7 @@ log '%s' at position %s, relay log '%s'
> }
> }
>
> - if (opt_slave_parallel_threads > 0)
> + if (mi->using_parallel())
> rli->parallel.wait_for_done(thd, rli);
>
> /* Thread stopped. Print the current replication position to the log */
> @@ -4693,7 +4714,7 @@ log '%s' at position %s, relay log '%s'
> (We want the first one to be before the printout of stop position to
> get the correct position printed.)
> */
> - if (opt_slave_parallel_threads > 0)
> + if (mi->using_parallel())
> rli->parallel.wait_for_done(thd, rli);
>
> /*
> @@ -6314,7 +6335,7 @@ static Log_event* next_event(rpl_group_i
> llstr(my_b_tell(cur_log),llbuf1),
> llstr(rli->event_relay_log_pos,llbuf2)));
> DBUG_ASSERT(my_b_tell(cur_log) >= BIN_LOG_HEADER_SIZE);
> - DBUG_ASSERT(opt_slave_parallel_threads > 0 ||
> + DBUG_ASSERT(rli->mi->using_parallel() ||
> my_b_tell(cur_log) == rli->event_relay_log_pos);
> }
> #endif
>
> === modified file 'sql/sql_lex.h'
> --- a/sql/sql_lex.h 2014-08-07 16:06:56 +0000
> +++ b/sql/sql_lex.h 2014-10-07 14:20:37 +0000
> @@ -218,11 +218,18 @@ struct LEX_MASTER_INFO
> uint port, connect_retry;
> float heartbeat_period;
> /*
> + Modes of parallel replication enabled, if any. A combination (binary
> OR) of
> + any of SLAVE_PARALLEL_DOMAIN, SLAVE_PARALLEL_GROUPCOMMIT,
> + SLAVE_PARALLEL_TRX, and SLAVE_PARALLEL_WAITING.
> + */
> + uint32 repl_parallel_mode;
> + /*
> Enum is used for making it possible to detect if the user
> changed variable or if it should be left at old value
> */
> enum {LEX_MI_UNCHANGED, LEX_MI_DISABLE, LEX_MI_ENABLE}
> - ssl, ssl_verify_server_cert, heartbeat_opt,
> repl_ignore_server_ids_opt;
> + ssl, ssl_verify_server_cert, heartbeat_opt,
> repl_ignore_server_ids_opt,
> + repl_parallel_mode_opt;
> enum {
> LEX_GTID_UNCHANGED, LEX_GTID_NO, LEX_GTID_CURRENT_POS,
> LEX_GTID_SLAVE_POS
> } use_gtid_opt;
> @@ -241,10 +248,11 @@ struct LEX_MASTER_INFO
> pos= relay_log_pos= server_id= port= connect_retry= 0;
> heartbeat_period= 0;
> ssl= ssl_verify_server_cert= heartbeat_opt=
> - repl_ignore_server_ids_opt= LEX_MI_UNCHANGED;
> + repl_ignore_server_ids_opt= repl_parallel_mode_opt=
> LEX_MI_UNCHANGED;
> gtid_pos_str.length= 0;
> gtid_pos_str.str= NULL;
> use_gtid_opt= LEX_GTID_UNCHANGED;
> + repl_parallel_mode= 0;
> }
> };
>
>
> === modified file 'sql/sql_repl.cc'
> --- a/sql/sql_repl.cc 2014-08-07 16:06:56 +0000
> +++ b/sql/sql_repl.cc 2014-10-07 14:20:37 +0000
> @@ -3434,6 +3434,9 @@ bool change_master(THD* thd, Master_info
> lex_mi->relay_log_name || lex_mi->relay_log_pos)
> mi->using_gtid= Master_info::USE_GTID_NO;
>
> + if (lex_mi->repl_parallel_mode_opt != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
> + mi->parallel_mode= lex_mi->repl_parallel_mode;
> +
> /*
> If user did specify neither host nor port nor any log name nor any log
> pos, i.e. he specified only user/password/master_connect_retry, he
> probably
>
> === modified file 'sql/sql_yacc.yy'
> --- a/sql/sql_yacc.yy 2014-09-03 14:24:31 +0000
> +++ b/sql/sql_yacc.yy 2014-10-07 14:20:37 +0000
> @@ -1130,6 +1130,7 @@ bool my_yyoverflow(short **a, YYSTYPE **
> %token DIV_SYM
> %token DOUBLE_SYM /* SQL-2003-R */
> %token DO_SYM
> +%token DOMAIN_SYM
> %token DROP /* SQL-2003-R */
> %token DUAL_SYM
> %token DUMPFILE
> @@ -1194,6 +1195,7 @@ bool my_yyoverflow(short **a, YYSTYPE **
> %token GRANT /* SQL-2003-R */
> %token GRANTS
> %token GROUP_SYM /* SQL-2003-R */
> +%token GROUPCOMMIT_SYM
> %token GROUP_CONCAT_SYM
> %token GT_SYM /* OPERATOR */
> %token HANDLER_SYM
> @@ -1376,6 +1378,7 @@ bool my_yyoverflow(short **a, YYSTYPE **
> %token PACK_KEYS_SYM
> %token PAGE_SYM
> %token PAGE_CHECKSUM_SYM
> +%token PARALLEL_MODE_SYM
> %token PARAM_MARKER
> %token PARSER_SYM
> %token PARSE_VCOL_EXPR_SYM
> @@ -1604,6 +1607,7 @@ bool my_yyoverflow(short **a, YYSTYPE **
> %token VIEW_SYM /* SQL-2003-N */
> %token VIRTUAL_SYM
> %token WAIT_SYM
> +%token WAITING_SYM
> %token WARNINGS
> %token WEEK_SYM
> %token WEIGHT_STRING_SYM
> @@ -2244,6 +2248,14 @@ rule: <-- starts at col 1
> {
> Lex->mi.repl_ignore_server_ids_opt=
> LEX_MASTER_INFO::LEX_MI_ENABLE;
> }
> + | PARALLEL_MODE_SYM EQ '(' ')'
> + {
> + Lex->mi.repl_parallel_mode_opt=
> LEX_MASTER_INFO::LEX_MI_DISABLE;
> + }
> + | PARALLEL_MODE_SYM EQ '(' parallel_mode_list ')'
> + {
> + Lex->mi.repl_parallel_mode_opt=
> LEX_MASTER_INFO::LEX_MI_ENABLE;
> + }
> |
> master_file_def
> ;
> @@ -2260,6 +2272,52 @@ rule: <-- starts at col 1
> insert_dynamic(&Lex->mi.repl_ignore_server_ids, (uchar*)
> &($1));
> }
>
> +parallel_mode_list:
> + parallel_mode_option
> + | parallel_mode_list ',' parallel_mode_option
> + ;
> +
> +parallel_mode_option:
> + DOMAIN_SYM
> + {
> + if (Lex->mi.repl_parallel_mode & SLAVE_PARALLEL_DOMAIN)
> + {
> + my_error(ER_INVALID_SLAVE_PARALLEL_MODE, MYF(0),
> "domain");
> + MYSQL_YYABORT;
> + }
> + Lex->mi.repl_parallel_mode|= SLAVE_PARALLEL_DOMAIN;
> + }
> + | GROUPCOMMIT_SYM
> + {
> + if (Lex->mi.repl_parallel_mode &
> + (SLAVE_PARALLEL_GROUPCOMMIT|SLAVE_PARALLEL_TRX))
> + {
> + my_error(ER_INVALID_SLAVE_PARALLEL_MODE, MYF(0),
> "groupcommit");
> + MYSQL_YYABORT;
> + }
> + Lex->mi.repl_parallel_mode|= SLAVE_PARALLEL_GROUPCOMMIT;
> + }
> + | TRANSACTIONAL_SYM
> + {
> + if (Lex->mi.repl_parallel_mode &
> + (SLAVE_PARALLEL_GROUPCOMMIT|SLAVE_PARALLEL_TRX))
> + {
> + my_error(ER_INVALID_SLAVE_PARALLEL_MODE, MYF(0),
> "transactional");
> + MYSQL_YYABORT;
> + }
> + Lex->mi.repl_parallel_mode|= SLAVE_PARALLEL_TRX;
> + }
> + | WAITING_SYM
> + {
> + if (Lex->mi.repl_parallel_mode & SLAVE_PARALLEL_WAITING)
> + {
> + my_error(ER_INVALID_SLAVE_PARALLEL_MODE, MYF(0),
> "waiting");
> + MYSQL_YYABORT;
> + }
> + Lex->mi.repl_parallel_mode|= SLAVE_PARALLEL_WAITING;
> + }
> +
> +
> master_file_def:
> MASTER_LOG_FILE_SYM EQ TEXT_STRING_sys
> {
> @@ -14098,6 +14156,7 @@ user: user_maybe_role
> | DISABLE_SYM {}
> | DISCARD {}
> | DISK_SYM {}
> + | DOMAIN_SYM {}
> | DUMPFILE {}
> | DUPLICATE_SYM {}
> | DYNAMIC_SYM {}
> @@ -14131,6 +14190,7 @@ user: user_maybe_role
> | GET_FORMAT {}
> | GRANTS {}
> | GLOBAL_SYM {}
> + | GROUPCOMMIT_SYM {}
> | HASH_SYM {}
> | HARD_SYM {}
> | HOSTS_SYM {}
> @@ -14221,6 +14281,7 @@ user: user_maybe_role
> | ONLY_SYM {}
> | PACK_KEYS_SYM {}
> | PAGE_SYM {}
> + | PARALLEL_MODE_SYM {}
> | PARTIAL {}
> | PARTITIONING_SYM {}
> | PARTITIONS_SYM {}
> @@ -14336,6 +14397,7 @@ user: user_maybe_role
> | VALUE_SYM {}
> | WARNINGS {}
> | WAIT_SYM {}
> + | WAITING_SYM {}
> | WEEK_SYM {}
> | WEIGHT_STRING_SYM {}
> | WORK_SYM {}
>
> === modified file 'sql/sys_vars.cc'
> --- a/sql/sys_vars.cc 2014-09-19 13:25:37 +0000
> +++ b/sql/sys_vars.cc 2014-10-07 14:20:37 +0000
> @@ -1834,31 +1834,11 @@ static Sys_var_ulong Sys_slave_parallel_
> VALID_RANGE(0,2147483647), DEFAULT(131072), BLOCK_SIZE(1));
>
>
> -static const char *slave_parallel_mode_names[] = {
> - "domain", "groupcommit", "transactional", "waiting", NULL
> -};
> -
> -static Sys_var_set Sys_slave_parallel_mode(
> - "slave_parallel_mode",
> - "Controls what transactions are applied in parallel when using "
> - "--slave-parallel-threads. Syntax:
> slave_paralle_mode=value[,value...], "
> - "where \"value\" could be one or more of: \"domain\", to apply "
> - "different replication domains in parallel; \"groupcommit\", to
> apply "
> - "in parallel transactions that group-committed together on the
> master; "
> - "\"transactional\", to optimistically try to apply all
> transactional "
> - "DML in parallel; and \"waiting\" to extend \"transactional\" to "
> - "even transactions that had to wait on the master.",
> - GLOBAL_VAR(opt_slave_parallel_mode), CMD_LINE(REQUIRED_ARG),
> - slave_parallel_mode_names,
> - DEFAULT(SLAVE_PARALLEL_DOMAIN |
> - SLAVE_PARALLEL_GROUPCOMMIT));
> -
> -
> static Sys_var_bit Sys_replicate_allow_parallel(
> "replicate_allow_parallel",
> "If set when a transaction is written to the binlog, that
> transaction "
> - "is allowed to replicate in parallel on a slave where "
> - "slave_parallel_mode is set to \"transactional\". Can be cleared
> for "
> + "is allowed to replicate in parallel on a slave where
> parallel_mode "
> + "is set to \"transactional\" (in CHANGE MASTER). Can be cleared
> for "
> "transactions that are likely to cause a conflict if replicated in
> "
> "parallel, to avoid unnecessary rollback and retry.",
> SESSION_ONLY(option_bits), NO_CMD_LINE, OPTION_RPL_ALLOW_PARALLEL,
>
> _______________________________________________
> commits mailing list
> commits@xxxxxxxxxxx
> https://lists.askmonty.org/cgi-bin/mailman/listinfo/commits
>
Follow ups
