Skip to content

M
mariadb
Commit e90f68c0 authored by unknown's avatar unknown
Browse files
Options

MDEV-5657: Parallel replication. 

Clean up and improve the parallel implementation code, mainly related to
scheduling of work to threads and handling of stop and errors.

Fix a lot of bugs in various corner cases that could lead to crashes or
corruption.

Fix that a single replication domain could easily grab all worker threads and
stall all other domains; now a configuration variable
--slave-domain-parallel-threads allows to limit the number of
workers.

Allow next event group to start as soon as previous group begins the commit
phase (as opposed to when it ends it); this allows multiple event groups on
the slave to participate in group commit, even when no other opportunities for
parallelism are available.

Various fixes:

 - Fix some races in the rpl.rpl_parallel test case.

 - Fix an old incorrect assertion in Log_event iocache read.

 - Fix repeated malloc/free of wait_for_commit and rpl_group_info objects.

 - Simplify wait_for_commit wakeup logic.

 - Fix one case in queue_for_group_commit() where killing one thread would
   fail to correctly signal the error to the next, causing loss of the
   transaction after slave restart.

 - Fix leaking of pthreads (and their allocated stack) due to missing
   PTHREAD_CREATE_DETACHED attribute.

 - Fix how one batch of group-committed transactions wait for the previous
   batch before starting to execute themselves. The old code had a very
   complex scheduling where the first transaction was handled differently,
   with subtle bugs in corner cases. Now each event group is always scheduled
   for a new worker (in a round-robin fashion amongst available workers).
   Keep a count of how many transactions have started to commit, and wait for
   that counter to reach the appropriate value.

 - Fix slave stop to wait for all workers to actually complete processing;
   before, the wait was for update of last_committed_sub_id, which happens a
   bit earlier, and could leave worker threads potentially accessing bits of
   the replication state that is no longer valid after slave stop.

 - Fix a couple of places where the test suite would kill a thread waiting
   inside enter_cond() in connection with debug_sync; debug_sync + kill can
   crash in rare cases due to a race with mysys_var_current_mutex in this
   case.

 - Fix some corner cases where we had enter_cond() but no exit_cond().

 - Fix that we could get failure in wait_for_prior_commit() but forget to flag
   the error with my_error().

 - Fix slave stop (both for normal stop and stop due to error). Now, at stop
   we pick a specific safe point (in terms of event groups executed) and make
   sure that all event groups before that point are executed to completion,
   and that no event group after start executing; this ensures a safe place to
   restart replication, even for non-transactional stuff/DDL. In error stop,
   make sure that all prior event groups are allowed to execute to completion,
   and that any later event groups that have started are rolled back, if
   possible. The old code could leave eg. T1 and T3 committed but T2 not, or
   it could even leave half a transaction not rolled back in some random
   worker, which would cause big problems when that worker was later reused
   after slave restart.

 - Fix the accounting of amount of events queued for one worker. Before, the
   amount was reduced immediately as soon as the events were dequeued (which
   happens all at once); this allowed twice the amount of events to be queued
   in memory for each single worker, which is not what users would expect.

 - Fix that an error set during execution of one event was sometimes not
   cleared before executing the next, causing problems with the error
   reporting.

 - Fix incorrect handling of thd->killed in worker threads.
parent 50323f12
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 1144 additions and 450 deletions
mysql-test/r/mysqld--help.result
View file @ e90f68c0
...@@ -782,6 +782,14 @@ The following options may be given as the first argument: ...@@ -782,6 +782,14 @@ The following options may be given as the first argument:
is already the default. is already the default.
--slave-compressed-protocol --slave-compressed-protocol
Use compression on master/slave protocol Use compression on master/slave protocol
--slave-domain-parallel-threads=#
Maximum number of parallel threads to use on slave for
events in a single replication domain. When using
multiple domains, this can be used to limit a single
domain from grabbing all threads and thus stalling other
domains. The default of 0 means to allow a domain to grab
as many threads as it wants, up to the value of
slave_parallel_threads.
--slave-exec-mode=name --slave-exec-mode=name
Modes for how replication events should be executed. Modes for how replication events should be executed.
Legal values are STRICT (default) and IDEMPOTENT. In Legal values are STRICT (default) and IDEMPOTENT. In
...@@ -1155,6 +1163,7 @@ skip-networking FALSE ...@@ -1155,6 +1163,7 @@ skip-networking FALSE
skip-show-database FALSE skip-show-database FALSE
skip-slave-start FALSE skip-slave-start FALSE
slave-compressed-protocol FALSE slave-compressed-protocol FALSE
slave-domain-parallel-threads 0
slave-exec-mode STRICT slave-exec-mode STRICT
slave-max-allowed-packet 1073741824 slave-max-allowed-packet 1073741824
slave-net-timeout 3600 slave-net-timeout 3600
......
mysql-test/suite/perfschema/r/dml_setup_instruments.result
View file @ e90f68c0
...@@ -37,6 +37,7 @@ where name like 'Wait/Synch/Cond/sql/%' ...@@ -37,6 +37,7 @@ where name like 'Wait/Synch/Cond/sql/%'
order by name limit 10; order by name limit 10;
NAME ENABLED TIMED NAME ENABLED TIMED
wait/synch/cond/sql/COND_flush_thread_cache YES YES wait/synch/cond/sql/COND_flush_thread_cache YES YES
wait/synch/cond/sql/COND_group_commit_orderer YES YES
wait/synch/cond/sql/COND_manager YES YES wait/synch/cond/sql/COND_manager YES YES
wait/synch/cond/sql/COND_parallel_entry YES YES wait/synch/cond/sql/COND_parallel_entry YES YES
wait/synch/cond/sql/COND_prepare_ordered YES YES wait/synch/cond/sql/COND_prepare_ordered YES YES
...@@ -44,8 +45,7 @@ wait/synch/cond/sql/COND_queue_state YES YES ...@@ -44,8 +45,7 @@ wait/synch/cond/sql/COND_queue_state YES YES
wait/synch/cond/sql/COND_rpl_status YES YES wait/synch/cond/sql/COND_rpl_status YES YES
wait/synch/cond/sql/COND_rpl_thread YES YES wait/synch/cond/sql/COND_rpl_thread YES YES
wait/synch/cond/sql/COND_rpl_thread_pool YES YES wait/synch/cond/sql/COND_rpl_thread_pool YES YES
wait/synch/cond/sql/COND_server_started YES YES wait/synch/cond/sql/COND_rpl_thread_queue YES YES
wait/synch/cond/sql/COND_thread_cache YES YES
select * from performance_schema.setup_instruments select * from performance_schema.setup_instruments
where name='Wait'; where name='Wait';
select * from performance_schema.setup_instruments select * from performance_schema.setup_instruments
......
mysql-test/suite/rpl/r/rpl_parallel.result
View file @ e90f68c0
...@@ -115,6 +115,7 @@ SET GLOBAL slave_parallel_threads=10; ...@@ -115,6 +115,7 @@ SET GLOBAL slave_parallel_threads=10;
SET debug_sync='RESET'; SET debug_sync='RESET';
include/start_slave.inc include/start_slave.inc
*** Test that group-committed transactions on the master can replicate in parallel on the slave. *** *** Test that group-committed transactions on the master can replicate in parallel on the slave. ***
SET debug_sync='RESET';
FLUSH LOGS; FLUSH LOGS;
CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB; CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
INSERT INTO t3 VALUES (1,1), (3,3), (5,5), (7,7); INSERT INTO t3 VALUES (1,1), (3,3), (5,5), (7,7);
...@@ -141,6 +142,7 @@ INSERT INTO t3 VALUES (6, foo(16, ...@@ -141,6 +142,7 @@ INSERT INTO t3 VALUES (6, foo(16,
'')); ''));
SET debug_sync='now WAIT_FOR master_queued3'; SET debug_sync='now WAIT_FOR master_queued3';
SET debug_sync='now SIGNAL master_cont1'; SET debug_sync='now SIGNAL master_cont1';
SET debug_sync='RESET';
SELECT * FROM t3 ORDER BY a; SELECT * FROM t3 ORDER BY a;
a b a b
1 1 1 1
...@@ -213,6 +215,9 @@ slave-bin.000003 # Query # # use `test`; INSERT INTO t3 VALUES (6, foo(16, ...@@ -213,6 +215,9 @@ slave-bin.000003 # Query # # use `test`; INSERT INTO t3 VALUES (6, foo(16,
slave-bin.000003 # Xid # # COMMIT /* XID */ slave-bin.000003 # Xid # # COMMIT /* XID */
*** Test STOP SLAVE in parallel mode *** *** Test STOP SLAVE in parallel mode ***
include/stop_slave.inc include/stop_slave.inc
SET debug_sync='RESET';
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
SET binlog_direct_non_transactional_updates=0; SET binlog_direct_non_transactional_updates=0;
SET sql_log_bin=0; SET sql_log_bin=0;
CALL mtr.add_suppression("Statement is unsafe because it accesses a non-transactional table after accessing a transactional table within the same transaction"); CALL mtr.add_suppression("Statement is unsafe because it accesses a non-transactional table after accessing a transactional table within the same transaction");
...@@ -227,10 +232,15 @@ INSERT INTO t3 VALUES(21, 21); ...@@ -227,10 +232,15 @@ INSERT INTO t3 VALUES(21, 21);
INSERT INTO t3 VALUES(22, 22); INSERT INTO t3 VALUES(22, 22);
SET binlog_format=@old_format; SET binlog_format=@old_format;
BEGIN; BEGIN;
INSERT INTO t2 VALUES (21); INSERT INTO t2 VALUES (21);
START SLAVE; START SLAVE;
SET @old_dbug= @@GLOBAL.debug_dbug;
SET GLOBAL debug_dbug="+d,rpl_parallel_wait_for_done_trigger";
STOP SLAVE; STOP SLAVE;
SET debug_sync='now WAIT_FOR wait_for_done_waiting';
ROLLBACK; ROLLBACK;
SET GLOBAL debug_dbug=@old_dbug;
SET debug_sync='RESET';
include/wait_for_slave_to_stop.inc include/wait_for_slave_to_stop.inc
SELECT * FROM t1 WHERE a >= 20 ORDER BY a; SELECT * FROM t1 WHERE a >= 20 ORDER BY a;
a a
...@@ -292,6 +302,7 @@ a b ...@@ -292,6 +302,7 @@ a b
32 32 32 32
33 33 33 33
34 34 34 34
SET debug_sync='RESET';
SET sql_log_bin=0; SET sql_log_bin=0;
CALL mtr.add_suppression("Query execution was interrupted"); CALL mtr.add_suppression("Query execution was interrupted");
CALL mtr.add_suppression("Commit failed due to failure of an earlier commit on which this one depends"); CALL mtr.add_suppression("Commit failed due to failure of an earlier commit on which this one depends");
...@@ -308,6 +319,7 @@ STOP SLAVE IO_THREAD; ...@@ -308,6 +319,7 @@ STOP SLAVE IO_THREAD;
SELECT * FROM t3 WHERE a >= 30 ORDER BY a; SELECT * FROM t3 WHERE a >= 30 ORDER BY a;
a b a b
31 31 31 31
SET debug_sync='RESET';
SET GLOBAL slave_parallel_threads=0; SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10; SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0; SET sql_log_bin=0;
...@@ -379,6 +391,7 @@ a b ...@@ -379,6 +391,7 @@ a b
42 42 42 42
43 43 43 43
44 44 44 44
SET debug_sync='RESET';
SET debug_sync='now WAIT_FOR t2_query'; SET debug_sync='now WAIT_FOR t2_query';
SET debug_sync='now SIGNAL t2_cont'; SET debug_sync='now SIGNAL t2_cont';
SET debug_sync='now WAIT_FOR t1_ready'; SET debug_sync='now WAIT_FOR t1_ready';
...@@ -386,9 +399,7 @@ KILL THD_ID; ...@@ -386,9 +399,7 @@ KILL THD_ID;
SET debug_sync='now WAIT_FOR t2_killed'; SET debug_sync='now WAIT_FOR t2_killed';
SET debug_sync='now SIGNAL t1_cont'; SET debug_sync='now SIGNAL t1_cont';
include/wait_for_slave_sql_error.inc [errno=1317,1963] include/wait_for_slave_sql_error.inc [errno=1317,1963]
SELECT * FROM t3 WHERE a >= 40 ORDER BY a; SET debug_sync='RESET';
a b
41 41
SET GLOBAL slave_parallel_threads=0; SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10; SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0; SET sql_log_bin=0;
...@@ -463,6 +474,7 @@ a b ...@@ -463,6 +474,7 @@ a b
52 52 52 52
53 53 53 53
54 54 54 54
SET debug_sync='RESET';
SET debug_sync='now WAIT_FOR t2_query'; SET debug_sync='now WAIT_FOR t2_query';
SET debug_sync='now SIGNAL t2_cont'; SET debug_sync='now SIGNAL t2_cont';
SET debug_sync='now WAIT_FOR t1_ready'; SET debug_sync='now WAIT_FOR t1_ready';
...@@ -473,6 +485,7 @@ include/wait_for_slave_sql_error.inc [errno=1317,1963] ...@@ -473,6 +485,7 @@ include/wait_for_slave_sql_error.inc [errno=1317,1963]
SELECT * FROM t3 WHERE a >= 50 ORDER BY a; SELECT * FROM t3 WHERE a >= 50 ORDER BY a;
a b a b
51 51 51 51
SET debug_sync='RESET';
SET GLOBAL slave_parallel_threads=0; SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10; SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0; SET sql_log_bin=0;
...@@ -514,14 +527,18 @@ include/start_slave.inc ...@@ -514,14 +527,18 @@ include/start_slave.inc
include/stop_slave.inc include/stop_slave.inc
SET GLOBAL binlog_format=@old_format; SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0; SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=3; SET GLOBAL slave_parallel_threads=4;
include/start_slave.inc include/start_slave.inc
*** 4. Test killing thread that is waiting to start transaction until previous transaction commits *** *** 4. Test killing thread that is waiting to start transaction until previous transaction commits ***
SET binlog_format=statement; SET binlog_format=statement;
SET gtid_domain_id=2; SET gtid_domain_id=2;
BEGIN;
INSERT INTO t3 VALUES (70, foo(70,
'rpl_parallel_start_waiting_for_prior SIGNAL t4_waiting', ''));
INSERT INTO t3 VALUES (60, foo(60, INSERT INTO t3 VALUES (60, foo(60,
'ha_write_row_end SIGNAL d2_query WAIT_FOR d2_cont2', 'ha_write_row_end SIGNAL d2_query WAIT_FOR d2_cont2',
'rpl_parallel_end_of_group SIGNAL d2_done WAIT_FOR d2_cont')); 'rpl_parallel_end_of_group SIGNAL d2_done WAIT_FOR d2_cont'));
COMMIT;
SET gtid_domain_id=0; SET gtid_domain_id=0;
SET debug_sync='now WAIT_FOR d2_query'; SET debug_sync='now WAIT_FOR d2_query';
SET gtid_domain_id=1; SET gtid_domain_id=1;
...@@ -540,15 +557,27 @@ INSERT INTO t3 VALUES (63, foo(63, ...@@ -540,15 +557,27 @@ INSERT INTO t3 VALUES (63, foo(63,
'ha_write_row_end SIGNAL d0_query WAIT_FOR d0_cont2', 'ha_write_row_end SIGNAL d0_query WAIT_FOR d0_cont2',
'rpl_parallel_end_of_group SIGNAL d0_done WAIT_FOR d0_cont')); 'rpl_parallel_end_of_group SIGNAL d0_done WAIT_FOR d0_cont'));
SET debug_sync='now WAIT_FOR d0_query'; SET debug_sync='now WAIT_FOR d0_query';
SET gtid_domain_id=3;
BEGIN;
INSERT INTO t3 VALUES (68, foo(68,
'rpl_parallel_start_waiting_for_prior SIGNAL t2_waiting', ''));
INSERT INTO t3 VALUES (69, foo(69,
'ha_write_row_end SIGNAL d3_query WAIT_FOR d3_cont2',
'rpl_parallel_end_of_group SIGNAL d3_done WAIT_FOR d3_cont'));
COMMIT;
SET gtid_domain_id=0;
SET debug_sync='now WAIT_FOR d3_query';
SET debug_sync='now SIGNAL d2_cont2'; SET debug_sync='now SIGNAL d2_cont2';
SET debug_sync='now WAIT_FOR d2_done'; SET debug_sync='now WAIT_FOR d2_done';
SET debug_sync='now SIGNAL d1_cont2'; SET debug_sync='now SIGNAL d1_cont2';
SET debug_sync='now WAIT_FOR d1_done'; SET debug_sync='now WAIT_FOR d1_done';
SET debug_sync='now SIGNAL d0_cont2'; SET debug_sync='now SIGNAL d0_cont2';
SET debug_sync='now WAIT_FOR d0_done'; SET debug_sync='now WAIT_FOR d0_done';
SET debug_sync='now SIGNAL d3_cont2';
SET debug_sync='now WAIT_FOR d3_done';
SET binlog_format=statement; SET binlog_format=statement;
INSERT INTO t3 VALUES (64, foo(64, INSERT INTO t3 VALUES (64, foo(64,
'commit_before_prepare_ordered SIGNAL t1_waiting WAIT_FOR t1_cont', '')); 'rpl_parallel_before_mark_start_commit SIGNAL t1_waiting WAIT_FOR t1_cont', ''));
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued2 WAIT_FOR master_cont2'; SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued2 WAIT_FOR master_cont2';
INSERT INTO t3 VALUES (65, foo(65, '', '')); INSERT INTO t3 VALUES (65, foo(65, '', ''));
SET debug_sync='now WAIT_FOR master_queued2'; SET debug_sync='now WAIT_FOR master_queued2';
...@@ -569,23 +598,34 @@ a b ...@@ -569,23 +598,34 @@ a b
65 65 65 65
66 66 66 66
67 67 67 67
68 68
69 69
70 70
SET debug_sync='RESET';
SET debug_sync='now SIGNAL d0_cont'; SET debug_sync='now SIGNAL d0_cont';
SET debug_sync='now WAIT_FOR t1_waiting'; SET debug_sync='now WAIT_FOR t1_waiting';
SET debug_sync='now SIGNAL d3_cont';
SET debug_sync='now WAIT_FOR t2_waiting';
SET debug_sync='now SIGNAL d1_cont'; SET debug_sync='now SIGNAL d1_cont';
SET debug_sync='now WAIT_FOR t3_waiting'; SET debug_sync='now WAIT_FOR t3_waiting';
SET debug_sync='now SIGNAL d2_cont'; SET debug_sync='now SIGNAL d2_cont';
SET debug_sync='now WAIT_FOR t4_waiting';
KILL THD_ID; KILL THD_ID;
SET debug_sync='now WAIT_FOR t3_killed'; SET debug_sync='now WAIT_FOR t3_killed';
SET debug_sync='now SIGNAL t1_cont'; SET debug_sync='now SIGNAL t1_cont';
include/wait_for_slave_sql_error.inc [errno=1317,1927,1963] include/wait_for_slave_sql_error.inc [errno=1317,1927,1963]
STOP SLAVE IO_THREAD; STOP SLAVE IO_THREAD;
SELECT * FROM t3 WHERE a >= 60 ORDER BY a; SELECT * FROM t3 WHERE a >= 60 AND a != 65 ORDER BY a;
a b a b
60 60 60 60
61 61 61 61
62 62 62 62
63 63 63 63
64 64 64 64
68 68
69 69
70 70
SET debug_sync='RESET';
SET GLOBAL slave_parallel_threads=0; SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10; SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0; SET sql_log_bin=0;
...@@ -597,11 +637,11 @@ RETURN x; ...@@ -597,11 +637,11 @@ RETURN x;
END END
|| ||
SET sql_log_bin=1; SET sql_log_bin=1;
INSERT INTO t3 VALUES (69,0); UPDATE t3 SET b=b+1 WHERE a=60;
include/start_slave.inc include/start_slave.inc
SELECT * FROM t3 WHERE a >= 60 ORDER BY a; SELECT * FROM t3 WHERE a >= 60 ORDER BY a;
a b a b
60 60 60 61
61 61 61 61
62 62 62 62
63 63 63 63
...@@ -609,7 +649,9 @@ a b ...@@ -609,7 +649,9 @@ a b
65 65 65 65
66 66 66 66
67 67 67 67
69 0 68 68
69 69
70 70
SET sql_log_bin=0; SET sql_log_bin=0;
DROP FUNCTION foo; DROP FUNCTION foo;
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500)) CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
...@@ -634,37 +676,31 @@ include/start_slave.inc ...@@ -634,37 +676,31 @@ include/start_slave.inc
SET @old_max_queued= @@GLOBAL.slave_parallel_max_queued; SET @old_max_queued= @@GLOBAL.slave_parallel_max_queued;
SET GLOBAL slave_parallel_max_queued=9000; SET GLOBAL slave_parallel_max_queued=9000;
SET binlog_format=statement; SET binlog_format=statement;
INSERT INTO t3 VALUES (70, foo(0, INSERT INTO t3 VALUES (80, foo(0,
'ha_write_row_end SIGNAL query_waiting WAIT_FOR query_cont', '')); 'ha_write_row_end SIGNAL query_waiting WAIT_FOR query_cont', ''));
SET debug_sync='now WAIT_FOR query_waiting'; SET debug_sync='now WAIT_FOR query_waiting';
SET @old_dbug= @@GLOBAL.debug_dbug; SET @old_dbug= @@GLOBAL.debug_dbug;
SET GLOBAL debug_dbug="+d,rpl_parallel_wait_queue_max"; SET GLOBAL debug_dbug="+d,rpl_parallel_wait_queue_max";
INSERT INTO t3 VALUES (72, 0); SELECT * FROM t3 WHERE a >= 80 ORDER BY a;
SELECT * FROM t3 WHERE a >= 70 ORDER BY a;
a b a b
70 0 80 0
71 10000 81 10000
72 0
SET debug_sync='now WAIT_FOR wait_queue_ready'; SET debug_sync='now WAIT_FOR wait_queue_ready';
KILL THD_ID; KILL THD_ID;
SET debug_sync='now WAIT_FOR wait_queue_killed'; SET debug_sync='now WAIT_FOR wait_queue_killed';
SET debug_sync='now SIGNAL query_cont'; SET debug_sync='now SIGNAL query_cont';
include/wait_for_slave_sql_error.inc [errno=1317,1927,1963] include/wait_for_slave_sql_error.inc [errno=1317,1927,1963]
STOP SLAVE IO_THREAD; STOP SLAVE IO_THREAD;
SELECT * FROM t3 WHERE a >= 70 ORDER BY a;
a b
70 0
71 10000
SET GLOBAL debug_dbug=@old_dbug; SET GLOBAL debug_dbug=@old_dbug;
SET GLOBAL slave_parallel_max_queued= @old_max_queued; SET GLOBAL slave_parallel_max_queued= @old_max_queued;
INSERT INTO t3 VALUES (73,0); INSERT INTO t3 VALUES (82,0);
SET debug_sync='RESET';
include/start_slave.inc include/start_slave.inc
SELECT * FROM t3 WHERE a >= 70 ORDER BY a; SELECT * FROM t3 WHERE a >= 80 ORDER BY a;
a b a b
70 0 80 0
71 10000 81 10000
72 0 82 0
73 0
include/stop_slave.inc include/stop_slave.inc
SET GLOBAL binlog_format=@old_format; SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0; SET GLOBAL slave_parallel_threads=0;
......
mysql-test/suite/rpl/t/rpl_parallel.test
View file @ e90f68c0
This diff is collapsed.
mysql-test/suite/sys_vars/r/slave_domain_parallel_threads_basic.result 0 → 100644
View file @ e90f68c0
SET @save_slave_domain_parallel_threads= @@GLOBAL.slave_domain_parallel_threads;
SELECT @@GLOBAL.slave_domain_parallel_threads as 'must be zero because of default';
must be zero because of default
0
SELECT @@SESSION.slave_domain_parallel_threads as 'no session var';
ERROR HY000: Variable 'slave_domain_parallel_threads' is a GLOBAL variable
SET GLOBAL slave_domain_parallel_threads= 0;
SET GLOBAL slave_domain_parallel_threads= DEFAULT;
SET GLOBAL slave_domain_parallel_threads= 10;
SELECT @@GLOBAL.slave_domain_parallel_threads;
@@GLOBAL.slave_domain_parallel_threads
10
SET GLOBAL slave_domain_parallel_threads = @save_slave_domain_parallel_threads;
mysql-test/suite/sys_vars/t/slave_domain_parallel_threads_basic.test 0 → 100644
View file @ e90f68c0
--source include/not_embedded.inc
SET @save_slave_domain_parallel_threads= @@GLOBAL.slave_domain_parallel_threads;
SELECT @@GLOBAL.slave_domain_parallel_threads as 'must be zero because of default';
--error ER_INCORRECT_GLOBAL_LOCAL_VAR
SELECT @@SESSION.slave_domain_parallel_threads as 'no session var';
SET GLOBAL slave_domain_parallel_threads= 0;
SET GLOBAL slave_domain_parallel_threads= DEFAULT;
SET GLOBAL slave_domain_parallel_threads= 10;
SELECT @@GLOBAL.slave_domain_parallel_threads;
SET GLOBAL slave_domain_parallel_threads = @save_slave_domain_parallel_threads;
mysys/mf_iocache.c
View file @ e90f68c0
...@@ -1281,10 +1281,6 @@ read_append_buffer: ...@@ -1281,10 +1281,6 @@ read_append_buffer:
size_t transfer_len; size_t transfer_len;
DBUG_ASSERT(info->append_read_pos <= info->write_pos); DBUG_ASSERT(info->append_read_pos <= info->write_pos);
/*
TODO: figure out if the assert below is needed or correct.
*/
DBUG_ASSERT(pos_in_file == info->end_of_file);
copy_len=min(Count, len_in_buff); copy_len=min(Count, len_in_buff);
memcpy(Buffer, info->append_read_pos, copy_len); memcpy(Buffer, info->append_read_pos, copy_len);
info->append_read_pos += copy_len; info->append_read_pos += copy_len;
......
sql/log.cc
View file @ e90f68c0
...@@ -6644,13 +6644,15 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry) ...@@ -6644,13 +6644,15 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
*/ */
wfc= orig_entry->thd->wait_for_commit_ptr; wfc= orig_entry->thd->wait_for_commit_ptr;
orig_entry->queued_by_other= false; orig_entry->queued_by_other= false;
if (wfc && wfc->waiting_for_commit) if (wfc && wfc->waitee)
{ {
mysql_mutex_lock(&wfc->LOCK_wait_commit); mysql_mutex_lock(&wfc->LOCK_wait_commit);
/* Do an extra check here, this time safely under lock. */ /* Do an extra check here, this time safely under lock. */
if (wfc->waiting_for_commit) if (wfc->waitee)
{ {
const char *old_msg; const char *old_msg;
wait_for_commit *loc_waitee;
/* /*
By setting wfc->opaque_pointer to our own entry, we mark that we are By setting wfc->opaque_pointer to our own entry, we mark that we are
ready to commit, but waiting for another transaction to commit before ready to commit, but waiting for another transaction to commit before
...@@ -6661,21 +6663,20 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry) ...@@ -6661,21 +6663,20 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
queued_by_other flag is set. queued_by_other flag is set.
*/ */
wfc->opaque_pointer= orig_entry; wfc->opaque_pointer= orig_entry;
DEBUG_SYNC(orig_entry->thd, "group_commit_waiting_for_prior");
old_msg= old_msg=
orig_entry->thd->enter_cond(&wfc->COND_wait_commit, orig_entry->thd->enter_cond(&wfc->COND_wait_commit,
&wfc->LOCK_wait_commit, &wfc->LOCK_wait_commit,
"Waiting for prior transaction to commit"); "Waiting for prior transaction to commit");
DEBUG_SYNC(orig_entry->thd, "group_commit_waiting_for_prior"); while ((loc_waitee= wfc->waitee) && !orig_entry->thd->check_killed())
while (wfc->waiting_for_commit && !orig_entry->thd->check_killed())
mysql_cond_wait(&wfc->COND_wait_commit, &wfc->LOCK_wait_commit); mysql_cond_wait(&wfc->COND_wait_commit, &wfc->LOCK_wait_commit);
wfc->opaque_pointer= NULL; wfc->opaque_pointer= NULL;
DBUG_PRINT("info", ("After waiting for prior commit, queued_by_other=%d", DBUG_PRINT("info", ("After waiting for prior commit, queued_by_other=%d",
orig_entry->queued_by_other)); orig_entry->queued_by_other));
if (wfc->waiting_for_commit) if (loc_waitee)
{ {
/* Wait terminated due to kill. */ /* Wait terminated due to kill. */
wait_for_commit *loc_waitee= wfc->waitee;
mysql_mutex_lock(&loc_waitee->LOCK_wait_commit); mysql_mutex_lock(&loc_waitee->LOCK_wait_commit);
if (loc_waitee->wakeup_subsequent_commits_running || if (loc_waitee->wakeup_subsequent_commits_running ||
orig_entry->queued_by_other) orig_entry->queued_by_other)
...@@ -6685,13 +6686,14 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry) ...@@ -6685,13 +6686,14 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
do do
{ {
mysql_cond_wait(&wfc->COND_wait_commit, &wfc->LOCK_wait_commit); mysql_cond_wait(&wfc->COND_wait_commit, &wfc->LOCK_wait_commit);
} while (wfc->waiting_for_commit); } while (wfc->waitee);
} }
else else
{ {
/* We were killed, so remove us from the list of waitee. */ /* We were killed, so remove us from the list of waitee. */
wfc->remove_from_list(&loc_waitee->subsequent_commits_list); wfc->remove_from_list(&loc_waitee->subsequent_commits_list);
mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit); mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit);
wfc->waitee= NULL;
orig_entry->thd->exit_cond(old_msg); orig_entry->thd->exit_cond(old_msg);
/* Interrupted by kill. */ /* Interrupted by kill. */
...@@ -6707,12 +6709,11 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry) ...@@ -6707,12 +6709,11 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
} }
else else
mysql_mutex_unlock(&wfc->LOCK_wait_commit); mysql_mutex_unlock(&wfc->LOCK_wait_commit);
}
if (wfc->wakeup_error) if (wfc && wfc->wakeup_error)
{ {
my_error(ER_PRIOR_COMMIT_FAILED, MYF(0)); my_error(ER_PRIOR_COMMIT_FAILED, MYF(0));
DBUG_RETURN(-1); DBUG_RETURN(-1);
}
} }
/* /*
...@@ -9043,7 +9044,7 @@ start_binlog_background_thread() ...@@ -9043,7 +9044,7 @@ start_binlog_background_thread()
array_elements(all_binlog_threads)); array_elements(all_binlog_threads));
#endif #endif
if (mysql_thread_create(key_thread_binlog, &th, NULL, if (mysql_thread_create(key_thread_binlog, &th, &connection_attrib,
binlog_background_thread, NULL)) binlog_background_thread, NULL))
return 1; return 1;
......
sql/mysqld.cc
View file @ e90f68c0
...@@ -546,6 +546,7 @@ ulong rpl_recovery_rank=0; ...@@ -546,6 +546,7 @@ ulong rpl_recovery_rank=0;
ulong stored_program_cache_size= 0; ulong stored_program_cache_size= 0;
ulong opt_slave_parallel_threads= 0; ulong opt_slave_parallel_threads= 0;
ulong opt_slave_domain_parallel_threads= 0;
ulong opt_binlog_commit_wait_count= 0; ulong opt_binlog_commit_wait_count= 0;
ulong opt_binlog_commit_wait_usec= 0; ulong opt_binlog_commit_wait_usec= 0;
ulong opt_slave_parallel_max_queued= 131072; ulong opt_slave_parallel_max_queued= 131072;
...@@ -895,8 +896,10 @@ PSI_cond_key key_RELAYLOG_update_cond, key_COND_wakeup_ready, ...@@ -895,8 +896,10 @@ PSI_cond_key key_RELAYLOG_update_cond, key_COND_wakeup_ready,
key_COND_wait_commit; key_COND_wait_commit;
PSI_cond_key key_RELAYLOG_COND_queue_busy; PSI_cond_key key_RELAYLOG_COND_queue_busy;
PSI_cond_key key_TC_LOG_MMAP_COND_queue_busy; PSI_cond_key key_TC_LOG_MMAP_COND_queue_busy;
PSI_cond_key key_COND_rpl_thread, key_COND_rpl_thread_pool, PSI_cond_key key_COND_rpl_thread_queue, key_COND_rpl_thread,
key_COND_parallel_entry, key_COND_prepare_ordered; key_COND_rpl_thread_pool,
key_COND_parallel_entry, key_COND_group_commit_orderer,
key_COND_prepare_ordered;
PSI_cond_key key_COND_wait_gtid; PSI_cond_key key_COND_wait_gtid;
static PSI_cond_info all_server_conds[]= static PSI_cond_info all_server_conds[]=
...@@ -941,8 +944,10 @@ static PSI_cond_info all_server_conds[]= ...@@ -941,8 +944,10 @@ static PSI_cond_info all_server_conds[]=
{ &key_COND_thread_cache, "COND_thread_cache", PSI_FLAG_GLOBAL}, { &key_COND_thread_cache, "COND_thread_cache", PSI_FLAG_GLOBAL},
{ &key_COND_flush_thread_cache, "COND_flush_thread_cache", PSI_FLAG_GLOBAL}, { &key_COND_flush_thread_cache, "COND_flush_thread_cache", PSI_FLAG_GLOBAL},
{ &key_COND_rpl_thread, "COND_rpl_thread", 0}, { &key_COND_rpl_thread, "COND_rpl_thread", 0},
{ &key_COND_rpl_thread_queue, "COND_rpl_thread_queue", 0},
{ &key_COND_rpl_thread_pool, "COND_rpl_thread_pool", 0}, { &key_COND_rpl_thread_pool, "COND_rpl_thread_pool", 0},
{ &key_COND_parallel_entry, "COND_parallel_entry", 0}, { &key_COND_parallel_entry, "COND_parallel_entry", 0},
{ &key_COND_group_commit_orderer, "COND_group_commit_orderer", 0},
{ &key_COND_prepare_ordered, "COND_prepare_ordered", 0}, { &key_COND_prepare_ordered, "COND_prepare_ordered", 0},
{ &key_COND_wait_gtid, "COND_wait_gtid", 0} { &key_COND_wait_gtid, "COND_wait_gtid", 0}
}; };
......
sql/mysqld.h
View file @ e90f68c0
...@@ -177,6 +177,7 @@ extern ulong opt_binlog_rows_event_max_size; ...@@ -177,6 +177,7 @@ extern ulong opt_binlog_rows_event_max_size;
extern ulong rpl_recovery_rank, thread_cache_size; extern ulong rpl_recovery_rank, thread_cache_size;
extern ulong stored_program_cache_size; extern ulong stored_program_cache_size;
extern ulong opt_slave_parallel_threads; extern ulong opt_slave_parallel_threads;
extern ulong opt_slave_domain_parallel_threads;
extern ulong opt_slave_parallel_max_queued; extern ulong opt_slave_parallel_max_queued;
extern ulong opt_binlog_commit_wait_count; extern ulong opt_binlog_commit_wait_count;
extern ulong opt_binlog_commit_wait_usec; extern ulong opt_binlog_commit_wait_usec;
...@@ -284,8 +285,9 @@ extern PSI_cond_key key_RELAYLOG_update_cond, key_COND_wakeup_ready, ...@@ -284,8 +285,9 @@ extern PSI_cond_key key_RELAYLOG_update_cond, key_COND_wakeup_ready,
key_COND_wait_commit; key_COND_wait_commit;
extern PSI_cond_key key_RELAYLOG_COND_queue_busy; extern PSI_cond_key key_RELAYLOG_COND_queue_busy;
extern PSI_cond_key key_TC_LOG_MMAP_COND_queue_busy; extern PSI_cond_key key_TC_LOG_MMAP_COND_queue_busy;
extern PSI_cond_key key_COND_rpl_thread, key_COND_rpl_thread_pool, extern PSI_cond_key key_COND_rpl_thread, key_COND_rpl_thread_queue,
key_COND_parallel_entry; key_COND_rpl_thread_pool,
key_COND_parallel_entry, key_COND_group_commit_orderer;
extern PSI_cond_key key_COND_wait_gtid; extern PSI_cond_key key_COND_wait_gtid;
extern PSI_thread_key key_thread_bootstrap, key_thread_delayed_insert, extern PSI_thread_key key_thread_bootstrap, key_thread_delayed_insert,
......
sql/rpl_parallel.cc
View file @ e90f68c0
This diff is collapsed.
sql/rpl_parallel.h
View file @ e90f68c0
...@@ -9,16 +9,66 @@ struct rpl_parallel_entry; ...@@ -9,16 +9,66 @@ struct rpl_parallel_entry;
struct rpl_parallel_thread_pool; struct rpl_parallel_thread_pool;
class Relay_log_info; class Relay_log_info;
/*
Structure used to keep track of the parallel replication of a batch of
event-groups that group-committed together on the master.
It is used to ensure that every event group in one batch has reached the
commit stage before the next batch starts executing.
Note the lifetime of this structure:
- It is allocated when the first event in a new batch of group commits
is queued, from the free list rpl_parallel_entry::gco_free_list.
- The gco for the batch currently being queued is owned by
rpl_parallel_entry::current_gco. The gco for a previous batch that has
been fully queued is owned by the gco->prev_gco pointer of the gco for
the following batch.
- The worker thread waits on gco->COND_group_commit_orderer for
rpl_parallel_entry::count_committing_event_groups to reach wait_count
before starting; the first waiter links the gco into the next_gco
pointer of the gco of the previous batch for signalling.
- When an event group reaches the commit stage, it signals the
COND_group_commit_orderer if its gco->next_gco pointer is non-NULL and
rpl_parallel_entry::count_committing_event_groups has reached
gco->next_gco->wait_count.
- When gco->wait_count is reached for a worker and the wait completes,
the worker frees gco->prev_gco; at this point it is guaranteed not to
be needed any longer.
*/
struct group_commit_orderer {
/* Wakeup condition, used with rpl_parallel_entry::LOCK_parallel_entry. */
mysql_cond_t COND_group_commit_orderer;
uint64 wait_count;
group_commit_orderer *prev_gco;
group_commit_orderer *next_gco;
bool installed;
};
struct rpl_parallel_thread { struct rpl_parallel_thread {
bool delay_start; bool delay_start;
bool running; bool running;
bool stop; bool stop;
mysql_mutex_t LOCK_rpl_thread; mysql_mutex_t LOCK_rpl_thread;
mysql_cond_t COND_rpl_thread; mysql_cond_t COND_rpl_thread;
mysql_cond_t COND_rpl_thread_queue;
struct rpl_parallel_thread *next; /* For free list. */ struct rpl_parallel_thread *next; /* For free list. */
struct rpl_parallel_thread_pool *pool; struct rpl_parallel_thread_pool *pool;
THD *thd; THD *thd;
struct rpl_parallel_entry *current_entry; /*
Who owns the thread, if any (it's a pointer into the
rpl_parallel_entry::rpl_threads array.
*/
struct rpl_parallel_thread **current_owner;
/* The rpl_parallel_entry of the owner. */
rpl_parallel_entry *current_entry;
struct queued_event { struct queued_event {
queued_event *next; queued_event *next;
Log_event *ev; Log_event *ev;
...@@ -31,6 +81,9 @@ struct rpl_parallel_thread { ...@@ -31,6 +81,9 @@ struct rpl_parallel_thread {
size_t event_size; size_t event_size;
} *event_queue, *last_in_queue; } *event_queue, *last_in_queue;
uint64 queued_size; uint64 queued_size;
queued_event *qev_free_list;
rpl_group_info *rgi_free_list;
group_commit_orderer *gco_free_list;
void enqueue(queued_event *qev) void enqueue(queued_event *qev)
{ {
...@@ -42,15 +95,25 @@ struct rpl_parallel_thread { ...@@ -42,15 +95,25 @@ struct rpl_parallel_thread {
queued_size+= qev->event_size; queued_size+= qev->event_size;
} }
void dequeue(queued_event *list) void dequeue1(queued_event *list)
{ {
queued_event *tmp;
DBUG_ASSERT(list == event_queue); DBUG_ASSERT(list == event_queue);
event_queue= last_in_queue= NULL; event_queue= last_in_queue= NULL;
for (tmp= list; tmp; tmp= tmp->next)
queued_size-= tmp->event_size;
} }
void dequeue2(size_t dequeue_size)
{
queued_size-= dequeue_size;
}
queued_event *get_qev(Log_event *ev, ulonglong event_size,
Relay_log_info *rli);
void free_qev(queued_event *qev);
rpl_group_info *get_rgi(Relay_log_info *rli, Gtid_log_event *gtid_ev,
rpl_parallel_entry *e);
void free_rgi(rpl_group_info *rgi);
group_commit_orderer *get_gco(uint64 wait_count, group_commit_orderer *prev);
void free_gco(group_commit_orderer *gco);
}; };
...@@ -66,14 +129,16 @@ struct rpl_parallel_thread_pool { ...@@ -66,14 +129,16 @@ struct rpl_parallel_thread_pool {
rpl_parallel_thread_pool(); rpl_parallel_thread_pool();
int init(uint32 size); int init(uint32 size);
void destroy(); void destroy();
struct rpl_parallel_thread *get_thread(rpl_parallel_entry *entry); struct rpl_parallel_thread *get_thread(rpl_parallel_thread **owner,
rpl_parallel_entry *entry);
void release_thread(rpl_parallel_thread *rpt);
}; };
struct rpl_parallel_entry { struct rpl_parallel_entry {
mysql_mutex_t LOCK_parallel_entry;
mysql_cond_t COND_parallel_entry;
uint32 domain_id; uint32 domain_id;
uint32 last_server_id;
uint64 last_seq_no;
uint64 last_commit_id; uint64 last_commit_id;
bool active; bool active;
/* /*
...@@ -82,15 +147,41 @@ struct rpl_parallel_entry { ...@@ -82,15 +147,41 @@ struct rpl_parallel_entry {
waiting for event groups to complete. waiting for event groups to complete.
*/ */
bool force_abort; bool force_abort;
/*
At STOP SLAVE (force_abort=true), we do not want to process all events in
the queue (which could unnecessarily delay stop, if a lot of events happen
to be queued). The stop_count provides a safe point at which to stop, so
that everything before becomes committed and nothing after does. The value
corresponds to group_commit_orderer::wait_count; if wait_count is less than
or equal to stop_count, we execute the associated event group, else we
skip it (and all following) and stop.
*/
uint64 stop_count;
rpl_parallel_thread *rpl_thread; /*
Cyclic array recording the last rpl_thread_max worker threads that we
queued event for. This is used to limit how many workers a single domain
can occupy (--slave-domain-parallel-threads).
Note that workers are never explicitly deleted from the array. Instead,
we need to check (under LOCK_rpl_thread) that the thread still belongs
to us before re-using (rpl_thread::current_owner).
*/
rpl_parallel_thread **rpl_threads;
uint32 rpl_thread_max;
uint32 rpl_thread_idx;
/* /*
The sub_id of the last transaction to commit within this domain_id. The sub_id of the last transaction to commit within this domain_id.
Must be accessed under LOCK_parallel_entry protection. Must be accessed under LOCK_parallel_entry protection.
Event groups commit in order, so the rpl_group_info for an event group
will be alive (at least) as long as
rpl_grou_info::gtid_sub_id > last_committed_sub_id. This can be used to
safely refer back to previous event groups if they are still executing,
and ignore them if they completed, without requiring explicit
synchronisation between the threads.
*/ */
uint64 last_committed_sub_id; uint64 last_committed_sub_id;
mysql_mutex_t LOCK_parallel_entry;
mysql_cond_t COND_parallel_entry;
/* /*
The sub_id of the last event group in this replication domain that was The sub_id of the last event group in this replication domain that was
queued for execution by a worker thread. queued for execution by a worker thread.
...@@ -98,14 +189,29 @@ struct rpl_parallel_entry { ...@@ -98,14 +189,29 @@ struct rpl_parallel_entry {
uint64 current_sub_id; uint64 current_sub_id;
rpl_group_info *current_group_info; rpl_group_info *current_group_info;
/* /*
The sub_id of the last event group in the previous batch of group-committed If we get an error in some event group, we set the sub_id of that event
transactions. group here. Then later event groups (with higher sub_id) can know not to
try to start (event groups that already started will be rolled back when
When we spawn parallel worker threads for the next group-committed batch, wait_for_prior_commit() returns error).
they first need to wait for this sub_id to be committed before it is safe The value is ULONGLONG_MAX when no error occured.
to start executing them. */
uint64 stop_on_error_sub_id;
/* Total count of event groups queued so far. */
uint64 count_queued_event_groups;
/*
Count of event groups that have started (but not necessarily completed)
the commit phase. We use this to know when every event group in a previous
batch of master group commits have started committing on the slave, so
that it is safe to start executing the events in the following batch.
*/ */
uint64 prev_groupcommit_sub_id; uint64 count_committing_event_groups;
/* The group_commit_orderer object for the events currently being queued. */
group_commit_orderer *current_gco;
rpl_parallel_thread * choose_thread(Relay_log_info *rli, bool *did_enter_cond,
const char **old_msg, bool reuse);
group_commit_orderer *get_gco();
void free_gco(group_commit_orderer *gco);
}; };
struct rpl_parallel { struct rpl_parallel {
HASH domain_hash; HASH domain_hash;
...@@ -116,7 +222,7 @@ struct rpl_parallel { ...@@ -116,7 +222,7 @@ struct rpl_parallel {
~rpl_parallel(); ~rpl_parallel();
void reset(); void reset();
rpl_parallel_entry *find(uint32 domain_id); rpl_parallel_entry *find(uint32 domain_id);
void wait_for_done(); void wait_for_done(THD *thd);
bool workers_idle(); bool workers_idle();
bool do_event(rpl_group_info *serial_rgi, Log_event *ev, bool do_event(rpl_group_info *serial_rgi, Log_event *ev,
ulonglong event_size); ulonglong event_size);
......
sql/rpl_rli.cc
View file @ e90f68c0
...@@ -1479,14 +1479,27 @@ end: ...@@ -1479,14 +1479,27 @@ end:
} }
rpl_group_info::rpl_group_info(Relay_log_info *rli_) void
: rli(rli_), thd(0), gtid_sub_id(0), wait_commit_sub_id(0), rpl_group_info::reinit(Relay_log_info *rli)
wait_commit_group_info(0), wait_start_sub_id(0), parallel_entry(0), {
deferred_events(NULL), m_annotate_event(0), tables_to_lock(0), this->rli= rli;
tables_to_lock_count(0), trans_retries(0), last_event_start_time(0), tables_to_lock= NULL;
is_parallel_exec(false), is_error(false), tables_to_lock_count= 0;
row_stmt_start_timestamp(0), long_find_row_note_printed(false) trans_retries= 0;
last_event_start_time= 0;
is_error= false;
row_stmt_start_timestamp= 0;
long_find_row_note_printed= false;
did_mark_start_commit= false;
commit_orderer.reinit();
}
rpl_group_info::rpl_group_info(Relay_log_info *rli)
: thd(0), gtid_sub_id(0), wait_commit_sub_id(0),
wait_commit_group_info(0), parallel_entry(0),
deferred_events(NULL), m_annotate_event(0), is_parallel_exec(false)
{ {
reinit(rli);
bzero(&current_gtid, sizeof(current_gtid)); bzero(&current_gtid, sizeof(current_gtid));
mysql_mutex_init(key_rpl_group_info_sleep_lock, &sleep_lock, mysql_mutex_init(key_rpl_group_info_sleep_lock, &sleep_lock,
MY_MUTEX_INIT_FAST); MY_MUTEX_INIT_FAST);
...@@ -1706,4 +1719,40 @@ void rpl_group_info::slave_close_thread_tables(THD *thd) ...@@ -1706,4 +1719,40 @@ void rpl_group_info::slave_close_thread_tables(THD *thd)
} }
static void
mark_start_commit_inner(rpl_parallel_entry *e, group_commit_orderer *gco)
{
uint64 count= ++e->count_committing_event_groups;
if (gco->next_gco && gco->next_gco->wait_count == count)
mysql_cond_broadcast(&gco->next_gco->COND_group_commit_orderer);
}
void
rpl_group_info::mark_start_commit_no_lock()
{
if (did_mark_start_commit)
return;
mark_start_commit_inner(parallel_entry, gco);
did_mark_start_commit= true;
}
void
rpl_group_info::mark_start_commit()
{
rpl_parallel_entry *e;
if (did_mark_start_commit)
return;
e= this->parallel_entry;
mysql_mutex_lock(&e->LOCK_parallel_entry);
mark_start_commit_inner(e, gco);
mysql_mutex_unlock(&e->LOCK_parallel_entry);
did_mark_start_commit= true;
}
#endif #endif
sql/rpl_rli.h
View file @ e90f68c0
...@@ -481,6 +481,7 @@ private: ...@@ -481,6 +481,7 @@ private:
struct rpl_group_info struct rpl_group_info
{ {
rpl_group_info *next; /* For free list in rpl_parallel_thread */
Relay_log_info *rli; Relay_log_info *rli;
THD *thd; THD *thd;
/* /*
...@@ -510,14 +511,15 @@ struct rpl_group_info ...@@ -510,14 +511,15 @@ struct rpl_group_info
uint64 wait_commit_sub_id; uint64 wait_commit_sub_id;
rpl_group_info *wait_commit_group_info; rpl_group_info *wait_commit_group_info;
/* /*
If non-zero, the event group must wait for this sub_id to be committed This holds a pointer to a struct that keeps track of the need to wait
before the execution of the event group is allowed to start. for the previous batch of event groups to reach the commit stage, before
this batch can start to execute.
(When we execute in parallel the transactions that group committed (When we execute in parallel the transactions that group committed
together on the master, we still need to wait for any prior transactions together on the master, we still need to wait for any prior transactions
to have commtted). to have reached the commit stage).
*/ */
uint64 wait_start_sub_id; group_commit_orderer *gco;
struct rpl_parallel_entry *parallel_entry; struct rpl_parallel_entry *parallel_entry;
...@@ -567,18 +569,22 @@ struct rpl_group_info ...@@ -567,18 +569,22 @@ struct rpl_group_info
char future_event_master_log_name[FN_REFLEN]; char future_event_master_log_name[FN_REFLEN];
bool is_parallel_exec; bool is_parallel_exec;
bool is_error; bool is_error;
/*
Set true when we signalled that we reach the commit phase. Used to avoid
counting one event group twice.
*/
bool did_mark_start_commit;
private:
/* /*
Runtime state for printing a note when slave is taking Runtime state for printing a note when slave is taking
too long while processing a row event. too long while processing a row event.
*/ */
time_t row_stmt_start_timestamp; time_t row_stmt_start_timestamp;
bool long_find_row_note_printed; bool long_find_row_note_printed;
public:
rpl_group_info(Relay_log_info *rli_); rpl_group_info(Relay_log_info *rli_);
~rpl_group_info(); ~rpl_group_info();
void reinit(Relay_log_info *rli);
/* /*
Returns true if the argument event resides in the containter; Returns true if the argument event resides in the containter;
...@@ -661,6 +667,8 @@ public: ...@@ -661,6 +667,8 @@ public:
void clear_tables_to_lock(); void clear_tables_to_lock();
void cleanup_context(THD *, bool); void cleanup_context(THD *, bool);
void slave_close_thread_tables(THD *); void slave_close_thread_tables(THD *);
void mark_start_commit_no_lock();
void mark_start_commit();
time_t get_row_stmt_start_timestamp() time_t get_row_stmt_start_timestamp()
{ {
......
sql/slave.cc
View file @ e90f68c0
...@@ -330,7 +330,7 @@ run_slave_init_thread() ...@@ -330,7 +330,7 @@ run_slave_init_thread()
pthread_t th; pthread_t th;
slave_init_thread_running= true; slave_init_thread_running= true;
if (mysql_thread_create(key_thread_slave_init, &th, NULL, if (mysql_thread_create(key_thread_slave_init, &th, &connection_attrib,
handle_slave_init, NULL)) handle_slave_init, NULL))
{ {
sql_print_error("Failed to create thread while initialising slave"); sql_print_error("Failed to create thread while initialising slave");
...@@ -4526,7 +4526,7 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME, ...@@ -4526,7 +4526,7 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME,
} }
if (opt_slave_parallel_threads > 0) if (opt_slave_parallel_threads > 0)
rli->parallel.wait_for_done(); rli->parallel.wait_for_done(thd);
/* Thread stopped. Print the current replication position to the log */ /* Thread stopped. Print the current replication position to the log */
{ {
...@@ -4552,7 +4552,7 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME, ...@@ -4552,7 +4552,7 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME,
get the correct position printed.) get the correct position printed.)
*/ */
if (opt_slave_parallel_threads > 0) if (opt_slave_parallel_threads > 0)
rli->parallel.wait_for_done(); rli->parallel.wait_for_done(thd);
/* /*
Some events set some playgrounds, which won't be cleared because thread Some events set some playgrounds, which won't be cleared because thread
......
sql/sql_class.cc
View file @ e90f68c0
...@@ -5667,14 +5667,23 @@ bool THD::rgi_have_temporary_tables() ...@@ -5667,14 +5667,23 @@ bool THD::rgi_have_temporary_tables()
} }
void
wait_for_commit::reinit()
{
subsequent_commits_list= NULL;
next_subsequent_commit= NULL;
waitee= NULL;
opaque_pointer= NULL;
wakeup_error= 0;
wakeup_subsequent_commits_running= false;
}
wait_for_commit::wait_for_commit() wait_for_commit::wait_for_commit()
: subsequent_commits_list(0), next_subsequent_commit(0), waitee(0),
opaque_pointer(0),
waiting_for_commit(false), wakeup_error(0),
wakeup_subsequent_commits_running(false)
{ {
mysql_mutex_init(key_LOCK_wait_commit, &LOCK_wait_commit, MY_MUTEX_INIT_FAST); mysql_mutex_init(key_LOCK_wait_commit, &LOCK_wait_commit, MY_MUTEX_INIT_FAST);
mysql_cond_init(key_COND_wait_commit, &COND_wait_commit, 0); mysql_cond_init(key_COND_wait_commit, &COND_wait_commit, 0);
reinit();
} }
...@@ -5722,7 +5731,7 @@ wait_for_commit::wakeup(int wakeup_error) ...@@ -5722,7 +5731,7 @@ wait_for_commit::wakeup(int wakeup_error)
*/ */
mysql_mutex_lock(&LOCK_wait_commit); mysql_mutex_lock(&LOCK_wait_commit);
waiting_for_commit= false; waitee= NULL;
this->wakeup_error= wakeup_error; this->wakeup_error= wakeup_error;
/* /*
Note that it is critical that the mysql_cond_signal() here is done while Note that it is critical that the mysql_cond_signal() here is done while
...@@ -5754,9 +5763,8 @@ wait_for_commit::wakeup(int wakeup_error) ...@@ -5754,9 +5763,8 @@ wait_for_commit::wakeup(int wakeup_error)
void void
wait_for_commit::register_wait_for_prior_commit(wait_for_commit *waitee) wait_for_commit::register_wait_for_prior_commit(wait_for_commit *waitee)
{ {
waiting_for_commit= true;
wakeup_error= 0;
DBUG_ASSERT(!this->waitee /* No prior registration allowed */); DBUG_ASSERT(!this->waitee /* No prior registration allowed */);
wakeup_error= 0;
this->waitee= waitee; this->waitee= waitee;
mysql_mutex_lock(&waitee->LOCK_wait_commit); mysql_mutex_lock(&waitee->LOCK_wait_commit);
...@@ -5766,7 +5774,7 @@ wait_for_commit::register_wait_for_prior_commit(wait_for_commit *waitee) ...@@ -5766,7 +5774,7 @@ wait_for_commit::register_wait_for_prior_commit(wait_for_commit *waitee)
see comments on wakeup_subsequent_commits2() for details. see comments on wakeup_subsequent_commits2() for details.
*/ */
if (waitee->wakeup_subsequent_commits_running) if (waitee->wakeup_subsequent_commits_running)
waiting_for_commit= false; this->waitee= NULL;
else else
{ {
/* /*
...@@ -5795,9 +5803,9 @@ wait_for_commit::wait_for_prior_commit2(THD *thd) ...@@ -5795,9 +5803,9 @@ wait_for_commit::wait_for_prior_commit2(THD *thd)
DEBUG_SYNC(thd, "wait_for_prior_commit_waiting"); DEBUG_SYNC(thd, "wait_for_prior_commit_waiting");
old_msg= thd->enter_cond(&COND_wait_commit, &LOCK_wait_commit, old_msg= thd->enter_cond(&COND_wait_commit, &LOCK_wait_commit,
"Waiting for prior transaction to commit"); "Waiting for prior transaction to commit");
while (waiting_for_commit && !thd->check_killed()) while ((loc_waitee= this->waitee) && !thd->check_killed())
mysql_cond_wait(&COND_wait_commit, &LOCK_wait_commit); mysql_cond_wait(&COND_wait_commit, &LOCK_wait_commit);
if (!waiting_for_commit) if (!loc_waitee)
{ {
if (wakeup_error) if (wakeup_error)
my_error(ER_PRIOR_COMMIT_FAILED, MYF(0)); my_error(ER_PRIOR_COMMIT_FAILED, MYF(0));
...@@ -5810,7 +5818,6 @@ wait_for_commit::wait_for_prior_commit2(THD *thd) ...@@ -5810,7 +5818,6 @@ wait_for_commit::wait_for_prior_commit2(THD *thd)
waiter as to whether we succeed or fail (eg. we may roll back but waitee waiter as to whether we succeed or fail (eg. we may roll back but waitee
might attempt to commit both us and any subsequent commits waiting for us). might attempt to commit both us and any subsequent commits waiting for us).
*/ */
loc_waitee= this->waitee;
mysql_mutex_lock(&loc_waitee->LOCK_wait_commit); mysql_mutex_lock(&loc_waitee->LOCK_wait_commit);
if (loc_waitee->wakeup_subsequent_commits_running) if (loc_waitee->wakeup_subsequent_commits_running)
{ {
...@@ -5819,21 +5826,29 @@ wait_for_commit::wait_for_prior_commit2(THD *thd) ...@@ -5819,21 +5826,29 @@ wait_for_commit::wait_for_prior_commit2(THD *thd)
do do
{ {
mysql_cond_wait(&COND_wait_commit, &LOCK_wait_commit); mysql_cond_wait(&COND_wait_commit, &LOCK_wait_commit);
} while (waiting_for_commit); } while (this->waitee);
if (wakeup_error)
my_error(ER_PRIOR_COMMIT_FAILED, MYF(0));
goto end; goto end;
} }
remove_from_list(&loc_waitee->subsequent_commits_list); remove_from_list(&loc_waitee->subsequent_commits_list);
mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit); mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit);
this->waitee= NULL;
DEBUG_SYNC(thd, "wait_for_prior_commit_killed");
wakeup_error= thd->killed_errno(); wakeup_error= thd->killed_errno();
if (!wakeup_error) if (!wakeup_error)
wakeup_error= ER_QUERY_INTERRUPTED; wakeup_error= ER_QUERY_INTERRUPTED;
my_message(wakeup_error, ER(wakeup_error), MYF(0)); my_message(wakeup_error, ER(wakeup_error), MYF(0));
thd->exit_cond(old_msg);
/*
Must do the DEBUG_SYNC() _after_ exit_cond(), as DEBUG_SYNC is not safe to
use within enter_cond/exit_cond.
*/
DEBUG_SYNC(thd, "wait_for_prior_commit_killed");
return wakeup_error;
end: end:
thd->exit_cond(old_msg); thd->exit_cond(old_msg);
waitee= NULL;
return wakeup_error; return wakeup_error;
} }
...@@ -5916,10 +5931,11 @@ wait_for_commit::wakeup_subsequent_commits2(int wakeup_error) ...@@ -5916,10 +5931,11 @@ wait_for_commit::wakeup_subsequent_commits2(int wakeup_error)
void void
wait_for_commit::unregister_wait_for_prior_commit2() wait_for_commit::unregister_wait_for_prior_commit2()
{ {
wait_for_commit *loc_waitee;
mysql_mutex_lock(&LOCK_wait_commit); mysql_mutex_lock(&LOCK_wait_commit);
if (waiting_for_commit) if ((loc_waitee= this->waitee))
{ {
wait_for_commit *loc_waitee= this->waitee;
mysql_mutex_lock(&loc_waitee->LOCK_wait_commit); mysql_mutex_lock(&loc_waitee->LOCK_wait_commit);
if (loc_waitee->wakeup_subsequent_commits_running) if (loc_waitee->wakeup_subsequent_commits_running)
{ {
...@@ -5931,7 +5947,7 @@ wait_for_commit::unregister_wait_for_prior_commit2() ...@@ -5931,7 +5947,7 @@ wait_for_commit::unregister_wait_for_prior_commit2()
See comments on wakeup_subsequent_commits2() for more details. See comments on wakeup_subsequent_commits2() for more details.
*/ */
mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit); mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit);
while (waiting_for_commit) while (this->waitee)
mysql_cond_wait(&COND_wait_commit, &LOCK_wait_commit); mysql_cond_wait(&COND_wait_commit, &LOCK_wait_commit);
} }
else else
...@@ -5939,10 +5955,10 @@ wait_for_commit::unregister_wait_for_prior_commit2() ...@@ -5939,10 +5955,10 @@ wait_for_commit::unregister_wait_for_prior_commit2()
/* Remove ourselves from the list in the waitee. */ /* Remove ourselves from the list in the waitee. */
remove_from_list(&loc_waitee->subsequent_commits_list); remove_from_list(&loc_waitee->subsequent_commits_list);
mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit); mysql_mutex_unlock(&loc_waitee->LOCK_wait_commit);
this->waitee= NULL;
} }
} }
mysql_mutex_unlock(&LOCK_wait_commit); mysql_mutex_unlock(&LOCK_wait_commit);
this->waitee= NULL;
} }
......
sql/sql_class.h
View file @ e90f68c0
...@@ -1598,8 +1598,8 @@ struct wait_for_commit ...@@ -1598,8 +1598,8 @@ struct wait_for_commit
{ {
/* /*
The LOCK_wait_commit protects the fields subsequent_commits_list and The LOCK_wait_commit protects the fields subsequent_commits_list and
wakeup_subsequent_commits_running (for a waitee), and the flag wakeup_subsequent_commits_running (for a waitee), and the pointer
waiting_for_commit and associated COND_wait_commit (for a waiter). waiterr and associated COND_wait_commit (for a waiter).
*/ */
mysql_mutex_t LOCK_wait_commit; mysql_mutex_t LOCK_wait_commit;
mysql_cond_t COND_wait_commit; mysql_cond_t COND_wait_commit;
...@@ -1607,7 +1607,13 @@ struct wait_for_commit ...@@ -1607,7 +1607,13 @@ struct wait_for_commit
wait_for_commit *subsequent_commits_list; wait_for_commit *subsequent_commits_list;
/* Link field for entries in subsequent_commits_list. */ /* Link field for entries in subsequent_commits_list. */
wait_for_commit *next_subsequent_commit; wait_for_commit *next_subsequent_commit;
/* Our waitee, if we did register_wait_for_prior_commit(), else NULL. */ /*
Our waitee, if we did register_wait_for_prior_commit(), and were not
yet woken up. Else NULL.
When this is cleared for wakeup, the COND_wait_commit condition is
signalled.
*/
wait_for_commit *waitee; wait_for_commit *waitee;
/* /*
Generic pointer for use by the transaction coordinator to optimise the Generic pointer for use by the transaction coordinator to optimise the
...@@ -1618,12 +1624,6 @@ struct wait_for_commit ...@@ -1618,12 +1624,6 @@ struct wait_for_commit
used by another transaction coordinator for similar purposes. used by another transaction coordinator for similar purposes.
*/ */
void *opaque_pointer; void *opaque_pointer;
/*
The waiting_for_commit flag is cleared when a waiter has been woken
up. The COND_wait_commit condition is signalled when this has been
cleared.
*/
bool waiting_for_commit;
/* The wakeup error code from the waitee. 0 means no error. */ /* The wakeup error code from the waitee. 0 means no error. */
int wakeup_error; int wakeup_error;
/* /*
...@@ -1639,10 +1639,14 @@ struct wait_for_commit ...@@ -1639,10 +1639,14 @@ struct wait_for_commit
Quick inline check, to avoid function call and locking in the common case Quick inline check, to avoid function call and locking in the common case
where no wakeup is registered, or a registered wait was already signalled. where no wakeup is registered, or a registered wait was already signalled.
*/ */
if (waiting_for_commit) if (waitee)
return wait_for_prior_commit2(thd); return wait_for_prior_commit2(thd);
else else
{
if (wakeup_error)
my_error(ER_PRIOR_COMMIT_FAILED, MYF(0));
return wakeup_error; return wakeup_error;
}
} }
void wakeup_subsequent_commits(int wakeup_error) void wakeup_subsequent_commits(int wakeup_error)
{ {
...@@ -1663,7 +1667,7 @@ struct wait_for_commit ...@@ -1663,7 +1667,7 @@ struct wait_for_commit
} }
void unregister_wait_for_prior_commit() void unregister_wait_for_prior_commit()
{ {
if (waiting_for_commit) if (waitee)
unregister_wait_for_prior_commit2(); unregister_wait_for_prior_commit2();
} }
/* /*
...@@ -1683,7 +1687,7 @@ struct wait_for_commit ...@@ -1683,7 +1687,7 @@ struct wait_for_commit
} }
next_ptr_ptr= &cur->next_subsequent_commit; next_ptr_ptr= &cur->next_subsequent_commit;
} }
waiting_for_commit= false; waitee= NULL;
} }
void wakeup(int wakeup_error); void wakeup(int wakeup_error);
...@@ -1694,6 +1698,7 @@ struct wait_for_commit ...@@ -1694,6 +1698,7 @@ struct wait_for_commit
wait_for_commit(); wait_for_commit();
~wait_for_commit(); ~wait_for_commit();
void reinit();
}; };
......
sql/sys_vars.cc
View file @ e90f68c0
...@@ -1609,6 +1609,49 @@ static Sys_var_ulong Sys_slave_parallel_threads( ...@@ -1609,6 +1609,49 @@ static Sys_var_ulong Sys_slave_parallel_threads(
ON_UPDATE(fix_slave_parallel_threads)); ON_UPDATE(fix_slave_parallel_threads));
static bool
check_slave_domain_parallel_threads(sys_var *self, THD *thd, set_var *var)
{
bool running;
mysql_mutex_lock(&LOCK_active_mi);
running= master_info_index->give_error_if_slave_running();
mysql_mutex_unlock(&LOCK_active_mi);
if (running)
return true;
return false;
}
static bool
fix_slave_domain_parallel_threads(sys_var *self, THD *thd, enum_var_type type)
{
bool running;
mysql_mutex_unlock(&LOCK_global_system_variables);
mysql_mutex_lock(&LOCK_active_mi);
running= master_info_index->give_error_if_slave_running();
mysql_mutex_unlock(&LOCK_active_mi);
mysql_mutex_lock(&LOCK_global_system_variables);
return running ? true : false;
}
static Sys_var_ulong Sys_slave_domain_parallel_threads(
"slave_domain_parallel_threads",
"Maximum number of parallel threads to use on slave for events in a "
"single replication domain. When using multiple domains, this can be "
"used to limit a single domain from grabbing all threads and thus "
"stalling other domains. The default of 0 means to allow a domain to "
"grab as many threads as it wants, up to the value of "
"slave_parallel_threads.",
GLOBAL_VAR(opt_slave_domain_parallel_threads), CMD_LINE(REQUIRED_ARG),
VALID_RANGE(0,16383), DEFAULT(0), BLOCK_SIZE(1), NO_MUTEX_GUARD,
NOT_IN_BINLOG, ON_CHECK(check_slave_domain_parallel_threads),
ON_UPDATE(fix_slave_domain_parallel_threads));
static Sys_var_ulong Sys_slave_parallel_max_queued( static Sys_var_ulong Sys_slave_parallel_max_queued(
"slave_parallel_max_queued", "slave_parallel_max_queued",
"Limit on how much memory SQL threads should use per parallel " "Limit on how much memory SQL threads should use per parallel "
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7