revision-id: d9913834cebe4ed8494b3d76762cd882788a8fd5 (mariadb-10.1.33-28-gd9913834ceb) parent(s): c1698e8dc50f0c342c1a6886426ba1d43396cd1e author: Andrei Elkin committer: Andrei Elkin timestamp: 2018-05-30 20:06:44 +0300 message: MDEV-14014 Multi-Slave Replication Fail: bogus data in log event MDEV-7257 made a dump thread to read from binlog concurrently with writers as long as the read bytes are below a water-mark (MYSQL_BIN_LOG::binlog_end_pos). However it appeared to be possible a dump thread reader reach out for bytes past the water mark through a feature of IO_CACHE that fills in the internal buffer and while doing so it could read what the reader is not supposed to see (the bytes above MYSQL_BIN_LOG::binlog_end_pos). The issue is fixed with constraining the IO_CACHE buffer fill to respect the watermark. An added test simulates potentially unconstrained buffer fill and an assert guards this is not the case anymore. --- include/my_sys.h | 11 ++++++ .../r/rpl_mdev14014_dirty_read_dump_thread.result | 14 +++++++ .../t/rpl_mdev14014_dirty_read_dump_thread.test | 44 ++++++++++++++++++++++ mysys/mf_iocache.c | 9 +++-- sql/log.cc | 2 + sql/sql_repl.cc | 37 +++++++++++++++++- 6 files changed, 113 insertions(+), 4 deletions(-) diff --git a/include/my_sys.h b/include/my_sys.h index 110a2ee9af3..41c1579b116 100644 --- a/include/my_sys.h +++ b/include/my_sys.h @@ -495,6 +495,17 @@ typedef struct st_io_cache /* Used when cacheing files */ my_off_t aio_read_pos; my_aio_result aio_result; #endif + /* + An offset for read caches that intend to access + associated files lock-free way and concurrently with possible + writers. The cache's buffer must not contain bytes beyond this margin, at where + reading must return as if the physical eof is reached. + The zero value designates the cache is ordinary so reading into its + buffer is constrained "as usual" only by the buffer size. + Non-zero value - typically produced by a file writer and assigned before attempting to + read the file - affects decision of how many bytes to pump into the buffer. + */ + my_off_t end_of_read; } IO_CACHE; typedef int (*qsort2_cmp)(const void *, const void *, const void *); diff --git a/mysql-test/suite/rpl/r/rpl_mdev14014_dirty_read_dump_thread.result b/mysql-test/suite/rpl/r/rpl_mdev14014_dirty_read_dump_thread.result new file mode 100644 index 00000000000..e8c5f49565c --- /dev/null +++ b/mysql-test/suite/rpl/r/rpl_mdev14014_dirty_read_dump_thread.result @@ -0,0 +1,14 @@ +include/rpl_init.inc [topology=1->2] +CREATE TABLE t1 (a INT); +include/rpl_sync.inc +SET @save_debug= @@GLOBAL.debug_dbug; +SET @@GLOBAL.debug_dbug="+d,dump_thread_wait_around_read"; +INSERT INTO t1 SET a=1; +SET debug_sync="at_after_write_to_binlog SIGNAL dump_go_on_reading"; +INSERT INTO t1 SET a=1; +include/rpl_sync.inc +SET debug_sync='reset'; +*** Clean up *** +SET @@GLOBAL.debug_dbug= @save_debug; +DROP TABLE t1; +include/rpl_end.inc diff --git a/mysql-test/suite/rpl/t/rpl_mdev14014_dirty_read_dump_thread.test b/mysql-test/suite/rpl/t/rpl_mdev14014_dirty_read_dump_thread.test new file mode 100644 index 00000000000..98e77439702 --- /dev/null +++ b/mysql-test/suite/rpl/t/rpl_mdev14014_dirty_read_dump_thread.test @@ -0,0 +1,44 @@ +# ==== Purpose ==== +# +# Check out if there are any dirty read by dump threads of the master (server_1) +# +# ==== Related worklogs ==== +# +# MDEV-14014 +# MDEV-7257 + +--source include/have_debug_sync.inc +--source include/have_debug.inc + +--let $rpl_topology= 1->2 +--source include/rpl_init.inc + +--connection server_1 +CREATE TABLE t1 (a INT); +--source include/rpl_sync.inc + + +--connection server_1 +SET @save_debug= @@GLOBAL.debug_dbug; +SET @@GLOBAL.debug_dbug="+d,dump_thread_wait_around_read"; + +--connect(con1, localhost, root) +INSERT INTO t1 SET a=1; # incr BEP and breaks waiting to reading +#SET debug_sync="at_after_write_to_binlog SIGNAL dump_go_on_reading WAIT_FOR user_thread_go"; +SET debug_sync="at_after_write_to_binlog SIGNAL dump_go_on_reading"; +INSERT INTO t1 SET a=1; # the 2nd group is the extra, it must not be seen by the dump thread + +--source include/rpl_sync.inc + +--connection con1 +SET debug_sync='reset'; + + +# +--echo *** Clean up *** +# +--connection server_1 +SET @@GLOBAL.debug_dbug= @save_debug; +DROP TABLE t1; + +--source include/rpl_end.inc diff --git a/mysys/mf_iocache.c b/mysys/mf_iocache.c index 56b1ae3fc6e..4e484015159 100644 --- a/mysys/mf_iocache.c +++ b/mysys/mf_iocache.c @@ -312,6 +312,7 @@ int init_io_cache(IO_CACHE *info, File file, size_t cachesize, /* End_of_file may be changed by user later */ info->end_of_file= end_of_file; + info->end_of_read= 0; info->error=0; info->type= type; init_functions(info); @@ -648,10 +649,12 @@ int _my_b_cache_read(IO_CACHE *info, uchar *Buffer, size_t Count) within a block already. So we will reach new alignment. */ max_length= info->read_length-diff_length; - /* We will not read past end of file. */ + /* We will not read past end of file or the end-of-read marker when set. */ if (info->type != READ_FIFO && - max_length > (info->end_of_file - pos_in_file)) - max_length= (size_t) (info->end_of_file - pos_in_file); + max_length > ((info->end_of_read > 0 ? + info->end_of_read : info->end_of_file) - pos_in_file)) + max_length= (size_t) ((info->end_of_read > 0 ? info->end_of_read : + info->end_of_file) - pos_in_file); /* If there is nothing left to read, we either are done, or we failed to fulfill the request. diff --git a/sql/log.cc b/sql/log.cc index 29f8c5639cf..f72238d39ce 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -7857,6 +7857,8 @@ MYSQL_BIN_LOG::trx_group_commit_leader(group_commit_entry *leader) bool any_error= false; bool all_error= true; + DEBUG_SYNC(leader->thd, "at_after_write_to_binlog"); + mysql_mutex_assert_not_owner(&LOCK_prepare_ordered); mysql_mutex_assert_owner(&LOCK_log); mysql_mutex_assert_not_owner(&LOCK_after_binlog_sync); diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 569c3d2c4ef..9b150e7e8e7 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -2529,6 +2529,13 @@ static my_off_t get_binlog_end_pos(binlog_send_info *info, return 0; } +static my_off_t my_b_set_read_limit(IO_CACHE *log, my_off_t end_pos) +{ + DBUG_ASSERT(log->end_of_read < end_pos); + + return (log->end_of_read= end_pos); +} + /** * This function sends events from one binlog file * but only up until end_pos @@ -2541,11 +2548,14 @@ static int send_events(binlog_send_info *info, IO_CACHE* log, LOG_INFO* linfo, { int error; ulong ev_offset; - +#ifndef DBUG_OFF + ulonglong dbug_events= 0; +#endif String *packet= info->packet; linfo->pos= my_b_tell(log); info->last_pos= my_b_tell(log); + (void) my_b_set_read_limit(log, end_pos); while (linfo->pos < end_pos) { if (should_stop(info)) @@ -2556,12 +2566,37 @@ static int send_events(binlog_send_info *info, IO_CACHE* log, LOG_INFO* linfo, if (reset_transmit_packet(info, info->flags, &ev_offset, &info->errmsg)) return 1; +#ifdef ENABLED_DEBUG_SYNC + DBUG_EXECUTE_IF("dump_thread_wait_around_read", + if (dbug_events++ == 0) + { + const char act[]= + "now " + "WAIT_FOR dump_go_on_reading"; + DBUG_ASSERT(debug_sync_service); + DBUG_ASSERT(!debug_sync_set_action( + info->thd, + STRING_WITH_LEN(act))); + } + ); +#endif + info->last_pos= linfo->pos; error= Log_event::read_log_event(log, packet, info->fdev, opt_master_verify_checksum ? info->current_checksum_alg : BINLOG_CHECKSUM_ALG_OFF); linfo->pos= my_b_tell(log); +#ifdef ENABLED_DEBUG_SYNC + DBUG_EXECUTE_IF("dump_thread_wait_around_read", + if (dbug_events == 1) + { + DBUG_ASSERT(log->pos_in_file + + (size_t) (log->read_end - log->buffer) + <= end_pos); + } + ); +#endif if (error) { set_read_error(info, error);