Hi Jonas, Thanks for the patch! I read through the patch. I am not that familiar with the semi-sync plugin code, but it looks reasonable. But the patch causes a test failure in sys_vars.all_vars. This is because not all newly introduced variables have corresponding test cases in mysql-test/suite/sys_vars/t/*_basic.test. Also, do you have some documentation for the feature? (There is a bit in the Jira entry, but for example the new variable rpl_semi_sync_master_slave_lag_heartbeat_frequency_us seems to be not described at all... If you can fix the above two points (and if Buildbot does not point out more test suite issues), then I will push it into the MariaDB tree. Thanks, - Kristian.

patch has been ported to 10.1 you can use it under new (also called "3-clause") BSD license i have done internal benchmarks...but you are most welcome to do some too.

DESCRIPTION: no slave left behind

this patch implements master throttling based on slave lag, aka no slave left behind. the core feature works as follows 1) the semi-sync-reply is ammended to also report back SQL-thread position (aka exec position) 2) transactions are not removed from the "active-transaction-list" in the semi-sync-master plugin until atleast one slave has reported that it has executed this transaction. the slave lag can then be estimated by calculating how long the oldest transaction has been lingering in the active-transaction-list. 3) client-threads are forced to wait before commit until slave lag has decreased to acceptable value.

the following variables are introduced on master:

rpl_semi_sync_master_max_slave_lag (global) rpl_semi_sync_master_slave_lag_wait_timeout (session)

the following status variables are introduced on master:

rpl_semi_sync_master_slave_lag_wait_sessions rpl_semi_sync_master_estimated_slave_lag rpl_semi_sync_master_trx_slave_lag_wait_time rpl_semi_sync_master_trx_slave_lag_wait_num rpl_semi_sync_master_avg_trx_slave_lag_wait_time

the following variables are introduced on slave:

rpl_semi_sync_slave_lag_enabled (global)

in addition to this, 2 optimizations that decreases overhead of semi-sync is introduced. 1) the idea of this is that if when a slave should send and transaction, it checks if it should be semi-synced, but rather than semi-sync:ing each transaction (which is done currently) the code will skip semi-syncing transaction if there already is newer transactions committed. But, since this can mean that semi-syncing is delayed indefinitely a cap is set using 2 new master variables:

rpl_semi_sync_master_max_unacked_event_bytes (global) rpl_semi_sync_master_max_unacked_event_count (global) 2) rpl_semi_sync_master_group_commit which makes the semi-sync plugin only semi-sync the last transaction in a group commit.

diff --git a/mysql-test/suite/rpl/r/rpl_semi_sync_slave_lag.result b/mysql-test/suite/rpl/r/rpl_semi_sync_slave_lag.result new file mode 100644 index 0000000..f3545a4 --- /dev/null +++ b/mysql-test/suite/rpl/r/rpl_semi_sync_slave_lag.result @@ -0,0 +1,150 @@ +include/master-slave.inc +[connection master] +set global rpl_semi_sync_master_enabled = 1; +set global rpl_semi_sync_master_max_slave_lag = 10; +show variables like 'rpl_semi_sync_master_%slave_lag%'; +Variable_name Value +rpl_semi_sync_master_max_slave_lag 10 +rpl_semi_sync_master_slave_lag_heartbeat_frequency_us 500000 +rpl_semi_sync_master_slave_lag_wait_timeout 50 +include/stop_slave.inc +set global rpl_semi_sync_slave_enabled = 1; +set global rpl_semi_sync_slave_lag_enabled = 1; +include/start_slave.inc +# create non-root user for testing READ_ONLY +grant SELECT, INSERT on *.* to test@localhost; +CREATE TABLE t1 (i INT NOT NULL AUTO_INCREMENT PRIMARY KEY, f varchar(8)) +ENGINE=innodb; +# +# Check basic behaviour +# +INSERT INTO t1 (f) VALUES ('1'),('2'),('3'); +# Now wait for slave lag to decrease to 0 +# [ on slave ] +STOP SLAVE SQL_THREAD; +# [ on master ] +INSERT INTO t1 (f) VALUES ('4'),('5'),('6'); +# Now wait for slave lag to increase to > 0 +# [ on slave ] +START SLAVE SQL_THREAD; +# [ on master ] +# Now wait for slave lag to decrease to 0 +# [ on slave ] +STOP SLAVE SQL_THREAD; +# [ on master ] +set session rpl_semi_sync_master_slave_lag_wait_timeout = 5; +# First transaction should succeed. slave_lag is zero when it commits +INSERT INTO t1 (f) VALUES ('7'),('8'),('9'); +# Now wait for slave lag to increase to > 10s +# Check that estimated_slave_lag is > 10s +SELECT VARIABLE_VALUE > 10000000 as should_be_1 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +should_be_1 +1 +# Second transaction should now fail. slave_lag is >10s when it commits +INSERT INTO t1 (f) VALUES ('a'),('b'),('c'); +ERROR HY000: Slave-lag timeout +# [ on slave ] +START SLAVE SQL_THREAD; +# [ on master ] +# Now wait for slave lag to decrease to < 10s +# And now it should succeed again +INSERT INTO t1 (f) VALUES ('d'),('e'),('f'); +SELECT * +FROM t1 +ORDER BY 1; +i f +1 1 +2 2 +3 3 +4 4 +5 5 +6 6 +7 7 +8 8 +9 9 +13 d +14 e +15 f +# [ on slave ] +SELECT * +FROM t1 +ORDER BY 1; +i f +1 1 +2 2 +3 3 +4 4 +5 5 +6 6 +7 7 +8 8 +9 9 +13 d +14 e +15 f +# +# Test interaction with READ_ONLY +# +# [ on slave ] +STOP SLAVE SQL_THREAD; +# [ on master ] +INSERT INTO t1 (f) VALUES ('g'),('h'),('i'); +# Now wait for slave lag to increase to > 10s +# [ on con1 ] +BEGIN; +INSERT INTO t1 (f) VALUES ('g'),('h'),('i'); +# [ on master ] +set global read_only = 1; +# [ on con1 ] +set session rpl_semi_sync_master_slave_lag_wait_timeout = 5; +# read-only is check *before* slave lag +COMMIT; +ERROR HY000: The MariaDB server is running with the --read-only option so it cannot execute this statement +# [ on slave ] +START SLAVE SQL_THREAD; +# [ on master ] +# Now wait for slave lag to decrease to 0 +set global read_only = 0; +# +# check slave_lag > 0 but less than rpl_semi_sync_master_max_slave_lag +# +# [ on slave ] +STOP SLAVE SQL_THREAD; +# [ on master ] +INSERT INTO t1 (f) VALUES ('j'),('k'),('l'); +# Now wait for slave lag to increase to > 0 +# Capture rpl_semi_sync_master_tx_slave_lag_waits before transaction +select @count_before := VARIABLE_VALUE +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_tx_slave_lag_waits'; +# Set maximum allowed slave lag to 24h +set global rpl_semi_sync_master_max_slave_lag = 86400; +INSERT INTO t1 (f) VALUES ('m'),('n'),('o'); +# Capture rpl_semi_sync_master_tx_slave_lag_waits after transaction +select @count_after := VARIABLE_VALUE +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_tx_slave_lag_waits'; +# There should have been no wait, since maximum allowed is very high +select @count_before = @count_after as should_be_1; +should_be_1 +1 +# [ on slave ] +START SLAVE SQL_THREAD; +# [ on master ] +# Now wait for slave lag to decrease to 0 +# +# Clean up +# +# [ on master ] +set global rpl_semi_sync_master_max_slave_lag = default; +set session rpl_semi_sync_master_slave_lag_wait_timeout = default; +DROP USER test@localhost; +include/stop_slave.inc +set global rpl_semi_sync_slave_enabled = 0; +set global rpl_semi_sync_slave_lag_enabled = default; +set global rpl_semi_sync_master_enabled = 0; +include/start_slave.inc +DROP TABLE t1; +include/rpl_end.inc diff --git a/mysql-test/suite/rpl/t/rpl_semi_sync_slave_lag.test b/mysql-test/suite/rpl/t/rpl_semi_sync_slave_lag.test new file mode 100644 index 0000000..f9bdbd9 --- /dev/null +++ b/mysql-test/suite/rpl/t/rpl_semi_sync_slave_lag.test @@ -0,0 +1,252 @@ +source include/have_semisync.inc; +source include/not_embedded.inc; +source include/have_innodb.inc; +source include/master-slave.inc; + +connection master; +set global rpl_semi_sync_master_enabled = 1; +set global rpl_semi_sync_master_max_slave_lag = 10; + +show variables like 'rpl_semi_sync_master_%slave_lag%'; + +connection slave; +source include/stop_slave.inc; +set global rpl_semi_sync_slave_enabled = 1; +set global rpl_semi_sync_slave_lag_enabled = 1; + +source include/start_slave.inc; + +connection master; + +--echo # create non-root user for testing READ_ONLY +grant SELECT, INSERT on *.* to test@localhost; +connect (con1,localhost,test,,test); + +CREATE TABLE t1 (i INT NOT NULL AUTO_INCREMENT PRIMARY KEY, f varchar(8)) +ENGINE=innodb; + +--echo # +--echo # Check basic behaviour +--echo # +INSERT INTO t1 (f) VALUES ('1'),('2'),('3'); + +--echo # Now wait for slave lag to decrease to 0 +let $wait_condition= SELECT VARIABLE_VALUE = 0 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +--echo # [ on slave ] +connection slave; +STOP SLAVE SQL_THREAD; + +--echo # [ on master ] +connection master; +INSERT INTO t1 (f) VALUES ('4'),('5'),('6'); + +--echo # Now wait for slave lag to increase to > 0 +let $wait_condition= SELECT VARIABLE_VALUE > 0 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +--echo # [ on slave ] +connection slave; +START SLAVE SQL_THREAD; + +--echo # [ on master ] +connection master; + +--echo # Now wait for slave lag to decrease to 0 +let $wait_condition= SELECT VARIABLE_VALUE = 0 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +--echo # [ on slave ] +connection slave; +STOP SLAVE SQL_THREAD; + +--echo # [ on master ] +connection master; + +set session rpl_semi_sync_master_slave_lag_wait_timeout = 5; + +--echo # First transaction should succeed. slave_lag is zero when it commits +INSERT INTO t1 (f) VALUES ('7'),('8'),('9'); + +--echo # Now wait for slave lag to increase to > 10s +let $wait_condition= SELECT VARIABLE_VALUE > 10000000 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +--echo # Check that estimated_slave_lag is > 10s +SELECT VARIABLE_VALUE > 10000000 as should_be_1 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; + +--echo # Second transaction should now fail. slave_lag is >10s when it commits +--error ER_ERROR_DURING_COMMIT +INSERT INTO t1 (f) VALUES ('a'),('b'),('c'); + +--echo # [ on slave ] +connection slave; +START SLAVE SQL_THREAD; + +--echo # [ on master ] +connection master; + +--echo # Now wait for slave lag to decrease to < 10s +let $wait_condition= SELECT VARIABLE_VALUE < 10000000 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +--echo # And now it should succeed again +INSERT INTO t1 (f) VALUES ('d'),('e'),('f'); + +SELECT * +FROM t1 +ORDER BY 1; + +sync_slave_with_master; + +--echo # [ on slave ] +connection slave; + +SELECT * +FROM t1 +ORDER BY 1; + +--echo # +--echo # Test interaction with READ_ONLY +--echo # + +--echo # [ on slave ] +connection slave; +STOP SLAVE SQL_THREAD; + +--echo # [ on master ] +connection master; + +INSERT INTO t1 (f) VALUES ('g'),('h'),('i'); + +--echo # Now wait for slave lag to increase to > 10s +let $wait_condition= SELECT VARIABLE_VALUE > 10000000 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +connection con1; +--echo # [ on con1 ] +BEGIN; +INSERT INTO t1 (f) VALUES ('g'),('h'),('i'); + +connection master; +--echo # [ on master ] +set global read_only = 1; + +connection con1; +--echo # [ on con1 ] +set session rpl_semi_sync_master_slave_lag_wait_timeout = 5; + +--echo # read-only is check *before* slave lag +--error ER_OPTION_PREVENTS_STATEMENT +COMMIT; + +disconnect con1; + +--echo # [ on slave ] +connection slave; +START SLAVE SQL_THREAD; + +connection master; +--echo # [ on master ] + +--echo # Now wait for slave lag to decrease to 0 +let $wait_condition= SELECT VARIABLE_VALUE = 0 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +set global read_only = 0; + +--echo # +--echo # check slave_lag > 0 but less than rpl_semi_sync_master_max_slave_lag +--echo # + +--echo # [ on slave ] +connection slave; +STOP SLAVE SQL_THREAD; + +connection master; +--echo # [ on master ] +INSERT INTO t1 (f) VALUES ('j'),('k'),('l'); + +--echo # Now wait for slave lag to increase to > 0 +let $wait_condition= SELECT VARIABLE_VALUE > 0 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +--echo # Capture rpl_semi_sync_master_tx_slave_lag_waits before transaction +--disable_result_log +select @count_before := VARIABLE_VALUE +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_tx_slave_lag_waits'; +--enable_result_log + +--echo # Set maximum allowed slave lag to 24h +set global rpl_semi_sync_master_max_slave_lag = 86400; # 24h + +INSERT INTO t1 (f) VALUES ('m'),('n'),('o'); + +--echo # Capture rpl_semi_sync_master_tx_slave_lag_waits after transaction +--disable_result_log +select @count_after := VARIABLE_VALUE +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_tx_slave_lag_waits'; +--enable_result_log + +--echo # There should have been no wait, since maximum allowed is very high +select @count_before = @count_after as should_be_1; + +--echo # [ on slave ] +connection slave; +START SLAVE SQL_THREAD; + +connection master; +--echo # [ on master ] + +--echo # Now wait for slave lag to decrease to 0 +let $wait_condition= SELECT VARIABLE_VALUE = 0 +FROM INFORMATION_SCHEMA.GLOBAL_STATUS +WHERE VARIABLE_NAME = 'rpl_semi_sync_master_estimated_slave_lag'; +--source include/wait_condition.inc + +--echo # +--echo # Clean up +--echo # +connection master; +--echo # [ on master ] +set global rpl_semi_sync_master_max_slave_lag = default; +set session rpl_semi_sync_master_slave_lag_wait_timeout = default; +DROP USER test@localhost; + +connection slave; +source include/stop_slave.inc; +set global rpl_semi_sync_slave_enabled = 0; +set global rpl_semi_sync_slave_lag_enabled = default; + +connection master; +set global rpl_semi_sync_master_enabled = 0; + +connection slave; +source include/start_slave.inc; + +connection master; + +DROP TABLE t1; +sync_slave_with_master; +--source include/rpl_end.inc diff --git a/plugin/semisync/semisync.cc b/plugin/semisync/semisync.cc index 4a80360..fc02b9d 100644 --- a/plugin/semisync/semisync.cc +++ b/plugin/semisync/semisync.cc @@ -20,6 +20,7 @@

const unsigned char ReplSemiSyncBase::kPacketMagicNum = 0xef; const unsigned char ReplSemiSyncBase::kPacketFlagSync = 0x01; +const unsigned char ReplSemiSyncBase::kPacketFlagSyncAndReport = 0x02;

const unsigned long Trace::kTraceGeneral = 0x0001; @@ -29,3 +30,6 @@ const unsigned long Trace::kTraceFunction = 0x0040;

const unsigned char ReplSemiSyncBase::kSyncHeader[2] = {ReplSemiSyncBase::kPacketMagicNum, 0}; + +const char* const ReplSemiSyncBase::kRplSemiSyncSlaveReportExec = + "rpl_semi_sync_slave_report_exec"; diff --git a/plugin/semisync/semisync.h b/plugin/semisync/semisync.h index 2857729..78faba9 100644 --- a/plugin/semisync/semisync.h +++ b/plugin/semisync/semisync.h @@ -75,13 +75,23 @@ class ReplSemiSyncBase

/* Constants in network packet header. */ static const unsigned char kPacketMagicNum; + /* this event should be semisync acked */ static const unsigned char kPacketFlagSync; + /* this event should be semisync acked including the current SQL position */ + static const unsigned char kPacketFlagSyncAndReport; + + /* user variable for enabling exec-pos reporting */ + static const char* const kRplSemiSyncSlaveReportExec; };

/* The layout of a semisync slave reply packet: 1 byte for the magic num 8 bytes for the binlog positon - n bytes for the binlog filename, terminated with a '\0' + n bytes for the binlog filename, NOT terminated with a '\0' + [ optionally ] + 1 byte == 0 + 8 bytes for the sql-thread position + n bytes for the sql-thread filename, terminated with a '\0' */ #define REPLY_MAGIC_NUM_LEN 1 #define REPLY_BINLOG_POS_LEN 8 diff --git a/plugin/semisync/semisync_master.cc b/plugin/semisync/semisync_master.cc index c88c162..3ee6e26 100644 --- a/plugin/semisync/semisync_master.cc +++ b/plugin/semisync/semisync_master.cc @@ -22,6 +22,9 @@ #define TIME_MILLION 1000000 #define TIME_BILLION 1000000000

+/* thd_key for per slave thread state */ +static MYSQL_THD_KEY_T thd_key; + /* This indicates whether semi-synchronous replication is enabled. */ char rpl_semi_sync_master_enabled; unsigned long rpl_semi_sync_master_wait_point = @@ -45,6 +48,18 @@ unsigned long long rpl_semi_sync_master_net_wait_time = 0; unsigned long long rpl_semi_sync_master_trx_wait_time = 0; char rpl_semi_sync_master_wait_no_slave = 1;

+unsigned long rpl_semi_sync_master_max_unacked_event_count = 0; +unsigned long rpl_semi_sync_master_max_unacked_event_bytes = 4096; + +unsigned long rpl_semi_sync_master_slave_lag_clients = 0; +unsigned long long rpl_semi_sync_master_estimated_slave_lag = 0; +unsigned long rpl_semi_sync_master_slave_lag_heartbeat_frequency_us = 500000; +unsigned long rpl_semi_sync_master_max_slave_lag = 0; +unsigned long rpl_semi_sync_master_slave_lag_wait_sessions = 0; + +unsigned long rpl_semi_sync_master_avg_trx_slave_lag_wait_time = 0; +unsigned long long rpl_semi_sync_master_trx_slave_lag_wait_num = 0; +unsigned long long rpl_semi_sync_master_trx_slave_lag_wait_time = 0;

static int getWaitTime(const struct timespec& start_ts);

@@ -150,6 +165,15 @@ int ActiveTranx::insert_tranx_node(const char *log_file_name, ins_node->log_name_[FN_REFLEN-1] = 0; /* make sure it ends properly */ ins_node->log_pos_ = log_file_pos;

+ { + /** + * set trans commit time + * this is called when writing into binlog, which is not + * exactly right, but close enough for our purposes + */ + ins_node->tranx_commit_time_us = my_hrtime().val; + } + if (!trx_front_) { /* The list is empty. */ @@ -193,12 +217,11 @@ int ActiveTranx::insert_tranx_node(const char *log_file_name, return function_exit(kWho, result); }

-bool ActiveTranx::is_tranx_end_pos(const char *log_file_name, - my_off_t log_file_pos) +TranxNode* ActiveTranx::lookup_tranx_end_pos(const char *log_file_name, + my_off_t log_file_pos) { - const char *kWho = "ActiveTranx::is_tranx_end_pos"; + const char *kWho = "ActiveTranx::lookup_tranx_end_pos"; function_enter(kWho); - unsigned int hash_val = get_hash_value(log_file_name, log_file_pos); TranxNode *entry = trx_htb_[hash_val];

@@ -211,38 +234,24 @@ bool ActiveTranx::is_tranx_end_pos(const char *log_file_name, }

if (trace_level_ & kTraceDetail) - sql_print_information("%s: probe (%s, %lu) in entry(%u)", kWho, - log_file_name, (unsigned long)log_file_pos, hash_val); + sql_print_information("%s: probe (%s, %lu)", kWho, + log_file_name, (unsigned long)log_file_pos);

function_exit(kWho, (entry != NULL)); - return (entry != NULL); + return entry; }

-int ActiveTranx::clear_active_tranx_nodes(const char *log_file_name, - my_off_t log_file_pos) +int ActiveTranx::clear_active_tranx_nodes() { - const char *kWho = "ActiveTranx::::clear_active_tranx_nodes"; - TranxNode *new_front; + set_new_front(NULL); + return 0; +}

+void ActiveTranx::set_new_front(TranxNode *new_front) +{ + const char *kWho = "ActiveTranx::set_new_front"; function_enter(kWho);

- if (log_file_name != NULL) - { - new_front = trx_front_; - - while (new_front) - { - if (compare(new_front, log_file_name, log_file_pos) > 0) - break; - new_front = new_front->next_; - } - } - else - { - /* If log_file_name is NULL, clear everything. */ - new_front = NULL; - } - if (new_front == NULL) { /* No active transaction nodes after the call. */ @@ -257,7 +266,6 @@ int ActiveTranx::clear_active_tranx_nodes(const char *log_file_name, trx_front_ = NULL; trx_rear_ = NULL; } - if (trace_level_ & kTraceDetail) sql_print_information("%s: cleared all nodes", kWho); } @@ -291,14 +299,40 @@ int ActiveTranx::clear_active_tranx_nodes(const char *log_file_name,

trx_front_ = new_front; allocator_.free_nodes_before(trx_front_); - if (trace_level_ & kTraceDetail) sql_print_information("%s: cleared %d nodes back until pos (%s, %lu)", kWho, n_frees, trx_front_->log_name_, (unsigned long)trx_front_->log_pos_); } + function_exit(kWho, 0); +}

- return function_exit(kWho, 0); +bool ActiveTranx::prune_active_tranx_nodes( + LogPosPtr pos, + ulonglong *oldest_tranx_commit_time_us) +{ + TranxNode *old_front = trx_front_; + TranxNode *new_front; + + new_front = trx_front_; + while (new_front) + { + if (compare(new_front, pos.file_name, pos.file_pos) > 0) + break; + new_front = new_front->next_; + } + + set_new_front(new_front); + + if (oldest_tranx_commit_time_us) + { + if (trx_front_ == NULL) + *oldest_tranx_commit_time_us = 0; + else + *oldest_tranx_commit_time_us = trx_front_->tranx_commit_time_us; + } + + return ! (old_front == trx_front_); }

@@ -334,7 +368,8 @@ ReplSemiSyncMaster::ReplSemiSyncMaster() wait_file_pos_(0), master_enabled_(false), wait_timeout_(0L), - state_(0) + state_(0), + oldest_unapplied_tranx_commit_time_us_(0) { strcpy(reply_file_name_, ""); strcpy(wait_file_name_, ""); @@ -362,11 +397,19 @@ int ReplSemiSyncMaster::initObject() mysql_cond_init(key_ss_cond_COND_binlog_send_, &COND_binlog_send_, NULL);

+ /* Mutex initialization can only be done after MY_INIT(). */ + mysql_mutex_init(key_ss_mutex_LOCK_slave_lag_, + &LOCK_slave_lag_, MY_MUTEX_INIT_FAST); + mysql_cond_init(key_ss_cond_COND_slave_lag_, + &COND_slave_lag_, NULL); + if (rpl_semi_sync_master_enabled) result = enableMaster(); else result = disableMaster();

+ thd_key_create(&thd_key); + return result; }

@@ -437,6 +480,8 @@ void ReplSemiSyncMaster::cleanup() { mysql_mutex_destroy(&LOCK_binlog_); mysql_cond_destroy(&COND_binlog_send_); + mysql_mutex_destroy(&LOCK_slave_lag_); + mysql_cond_destroy(&COND_slave_lag_); init_done_= 0; }

@@ -473,7 +518,34 @@ void ReplSemiSyncMaster::add_slave() { lock(); rpl_semi_sync_master_clients++; + if (has_semi_sync_slave_lag()) + rpl_semi_sync_master_slave_lag_clients++; unlock(); + + if (has_semi_sync_slave_lag()) + { + int null_val = 0; + longlong new_val = + rpl_semi_sync_master_slave_lag_heartbeat_frequency_us * 1000; + longlong old_val = new_val + 1; + + get_user_var_int("master_heartbeat_period", &old_val, &null_val); + if (old_val > new_val || null_val) + { + /* if there no old value or it's bigger than what we want */ + int res = set_user_var_int("master_heartbeat_period",new_val, &old_val); + if (res == -1) + { + sql_print_error( + "Repl_semi_sync::failed to set master_heartbeat_period"); + } + } + } + + /** + * create per slave-state and store it in thread-local-storage */ + ReplSemiSyncMasterPerSlaveState *state = new ReplSemiSyncMasterPerSlaveState; + thd_setspecific(current_thd, thd_key, state); }

void ReplSemiSyncMaster::remove_slave() @@ -492,7 +564,31 @@ void ReplSemiSyncMaster::remove_slave() rpl_semi_sync_master_clients == 0) switch_off(); } + + bool no_slave_lag_clients = false; + if (has_semi_sync_slave_lag()) + { + if (--rpl_semi_sync_master_slave_lag_clients == 0) + { + no_slave_lag_clients = true; + } + } + unlock(); + + ReplSemiSyncMasterPerSlaveState *state = + (ReplSemiSyncMasterPerSlaveState*)thd_getspecific(current_thd, thd_key); + thd_setspecific(current_thd, thd_key, NULL); + + if (state != NULL) + { + delete state; + } + + if (no_slave_lag_clients) + { + wake_slave_lag_waiters(0); + } }

bool ReplSemiSyncMaster::is_semi_sync_slave() @@ -503,14 +599,115 @@ bool ReplSemiSyncMaster::is_semi_sync_slave() return val; }

+bool ReplSemiSyncMaster::has_semi_sync_slave_lag() +{ + int null_value; + long long val= 0; + get_user_var_int(kRplSemiSyncSlaveReportExec, &val, &null_value); + return val; +} + +int ReplSemiSyncMaster::checkSyncReq(const LogPosPtr *log_pos) +{ + if (log_pos == NULL) + { + /* heartbeat events does not have logpos (since they are not actually + * stored in the binlog). + */ + if (!has_semi_sync_slave_lag()) + { + /* don't semi-sync them if we haven't enabled slave-lag handling */ + return 0; + } + else + { + /* else ask for both IO and exec position */ + return 2; + } + } + + /** + * check if this log-pos is a candidate for semi-syncing event + */ + TranxNode *entry = active_tranxs_->lookup_tranx_end_pos(log_pos->file_name, + log_pos->file_pos); + + if (entry == NULL) + return 0; + + ReplSemiSyncMasterPerSlaveState *state = + (ReplSemiSyncMasterPerSlaveState*)thd_getspecific(current_thd, + thd_key); + do + { + state->unacked_event_count_++; + + if (active_tranxs_->is_rear(entry)) + { + /* always ask for ack on last event in tranx list */ + break; + } + + if (state->unacked_event_count_ >= + rpl_semi_sync_master_max_unacked_event_count) + { + /* enough events passed that it's time for another ack */ + break; + } + + if (!state->sync_req_pos_.IsInited()) + { + /* first event => time for ack */ + break; + } + + if (strcmp(log_pos->file_name, state->sync_req_pos_.file_name) != 0) + { + /* new file => time for ack */ + break; + } + + if (log_pos->file_pos >= (state->sync_req_pos_.file_pos + + rpl_semi_sync_master_max_unacked_event_bytes)) + { + /* enough bytes => time for ack */ + break; + } + + /* we skip asking for semi-sync ack on this event */ + return 0; + + } while (0); + + /* keep track on when we last asked for semi-sync-ack */ + state->unacked_event_count_ = 0; + state->sync_req_pos_.Assign(log_pos); + + /** + * check if this slave can report back exec position + */ + if (!has_semi_sync_slave_lag()) + { + /* slave can't report back SQL position */ + return 1; + } + + /* ask for both IO and SQL position */ + return 2; +} + int ReplSemiSyncMaster::reportReplyBinlog(uint32 server_id, const char *log_file_name, - my_off_t log_file_pos) + my_off_t log_file_pos, + const LogPos *exec_pos) { const char *kWho = "ReplSemiSyncMaster::reportReplyBinlog"; int cmp; bool can_release_threads = false; bool need_copy_send_pos = true; + bool pruned_trx_list = false; + ulonglong oldest_tranx_commit_time_us = 0; +

if (!(getMasterEnabled())) return 0; @@ -559,15 +756,29 @@ int ReplSemiSyncMaster::reportReplyBinlog(uint32 server_id, reply_file_pos_ = log_file_pos; reply_file_name_inited_ = true;

- /* Remove all active transaction nodes before this point. */ - assert(active_tranxs_ != NULL); - active_tranxs_->clear_active_tranx_nodes(log_file_name, log_file_pos); - if (trace_level_ & kTraceDetail) sql_print_information("%s: Got reply at (%s, %lu)", kWho, log_file_name, (unsigned long)log_file_pos); }

+ assert(active_tranxs_ != NULL); + if (exec_pos != NULL) + { + /* prune using exec_pos */ + LogPosPtr ptr(*exec_pos); + pruned_trx_list = active_tranxs_->prune_active_tranx_nodes( + ptr, &oldest_tranx_commit_time_us); + } + else if (rpl_semi_sync_master_slave_lag_clients == 0 && need_copy_send_pos) + { + /** + * if we don't have any slaves that can do exec_pos reporting, + * prune by IO position as "plain old semi sync" + */ + LogPosPtr ptr(log_file_name, log_file_pos); + active_tranxs_->prune_active_tranx_nodes(ptr, NULL); + } + if (rpl_semi_sync_master_wait_sessions > 0) { /* Let us check if some of the waiting threads doing a trx @@ -596,6 +807,15 @@ int ReplSemiSyncMaster::reportReplyBinlog(uint32 server_id, cond_broadcast(); }

+ if (pruned_trx_list) + { + /** + * if we did prune trx list, it might be that we should wake up + * threads waiting for slave-lag to decrease + */ + wake_slave_lag_waiters(oldest_tranx_commit_time_us); + } + return function_exit(kWho, 0); }

@@ -743,14 +963,6 @@ int ReplSemiSyncMaster::commitTrx(const char* trx_wait_binlog_name, } }

- /* - At this point, the binlog file and position of this transaction - must have been removed from ActiveTranx. - */ - assert(thd_killed(NULL) || - !active_tranxs_->is_tranx_end_pos(trx_wait_binlog_name, - trx_wait_binlog_pos)); - l_end: /* Update the status counter. */ if (is_on()) @@ -794,7 +1006,7 @@ int ReplSemiSyncMaster::switch_off()

/* Clear the active transaction list. */ assert(active_tranxs_ != NULL); - result = active_tranxs_->clear_active_tranx_nodes(NULL, 0); + result = active_tranxs_->clear_active_tranx_nodes();

rpl_semi_sync_master_off_times++; wait_file_name_inited_ = false; @@ -884,7 +1096,7 @@ int ReplSemiSyncMaster::updateSyncHeader(unsigned char *packet, { const char *kWho = "ReplSemiSyncMaster::updateSyncHeader"; int cmp = 0; - bool sync = false; + int sync = 0;

/* If the semi-sync master is not enabled, or the slave is not a semi-sync * target, do not request replies from the slave. @@ -905,6 +1117,13 @@ int ReplSemiSyncMaster::updateSyncHeader(unsigned char *packet, /* semi-sync is ON */ /* sync= false; No sync unless a transaction is involved. */

+ if (log_file_name == NULL) + { + /* this is heartbeat, request io_pos and exec_pos */ + sync = checkSyncReq(0); + goto l_end; + } + if (reply_file_name_inited_) { cmp = ActiveTranx::compare(log_file_name, log_file_pos, @@ -933,12 +1152,12 @@ int ReplSemiSyncMaster::updateSyncHeader(unsigned char *packet, */ if (cmp >= 0) { - /* + /* * We only wait if the event is a transaction's ending event. */ assert(active_tranxs_ != NULL); - sync = active_tranxs_->is_tranx_end_pos(log_file_name, - log_file_pos); + LogPosPtr pos(log_file_name, log_file_pos); + sync = checkSyncReq(&pos); } } else @@ -951,7 +1170,7 @@ int ReplSemiSyncMaster::updateSyncHeader(unsigned char *packet, } else { - sync = true; + sync = 1; } }

@@ -966,10 +1185,14 @@ int ReplSemiSyncMaster::updateSyncHeader(unsigned char *packet, /* We do not need to clear sync flag because we set it to 0 when we * reserve the packet header. */ - if (sync) + if (sync == 1) { (packet)[2] = kPacketFlagSync; } + else if (sync == 2) + { + (packet)[2] = kPacketFlagSyncAndReport; + }

return function_exit(kWho, 0); } @@ -1018,7 +1241,8 @@ int ReplSemiSyncMaster::writeTranxInBinlog(const char* log_file_name, if (is_on()) { assert(active_tranxs_ != NULL); - if(active_tranxs_->insert_tranx_node(log_file_name, log_file_pos)) + bool empty = active_tranxs_->is_empty(); + if (active_tranxs_->insert_tranx_node(log_file_name, log_file_pos)) { /* if insert tranx_node failed, print a warning message @@ -1028,6 +1252,14 @@ int ReplSemiSyncMaster::writeTranxInBinlog(const char* log_file_name, log_file_name, (ulong)log_file_pos); switch_off(); } + else if (empty && rpl_semi_sync_master_slave_lag_clients > 0) + { + /* if the list of transactions was empty, + * we need to init the oldest_tranx_commit_time_us + */ + oldest_unapplied_tranx_commit_time_us_ = + active_tranxs_->get_oldest_tranx_commit_time_us(); + } }

l_end: @@ -1037,10 +1269,10 @@ int ReplSemiSyncMaster::writeTranxInBinlog(const char* log_file_name, }

int ReplSemiSyncMaster::readSlaveReply(NET *net, uint32 server_id, - const char *event_buf) + const char *event_buf_) { const char *kWho = "ReplSemiSyncMaster::readSlaveReply"; - const unsigned char *packet; + const unsigned char *packet, *packet_start; char log_file_name[FN_REFLEN]; my_off_t log_file_pos; ulong log_file_len = 0; @@ -1048,12 +1280,15 @@ int ReplSemiSyncMaster::readSlaveReply(NET *net, uint32 server_id, int result = -1; struct timespec start_ts; ulong trc_level = trace_level_; + const unsigned char *event_buf = (const unsigned char*)event_buf_; + bool exec_pos_present = false; // is SQL exec pos present in reply + LogPos exec_pos; // position of SQL thread LINT_INIT_STRUCT(start_ts);

function_enter(kWho);

- assert((unsigned char)event_buf[1] == kPacketMagicNum); - if ((unsigned char)event_buf[2] != kPacketFlagSync) + assert(event_buf[1] == kPacketMagicNum); + if ((event_buf[2] & (kPacketFlagSync | kPacketFlagSyncAndReport)) == 0) { /* current event does not require reply */ result = 0; @@ -1111,28 +1346,60 @@ int ReplSemiSyncMaster::readSlaveReply(NET *net, uint32 server_id, goto l_end; }

- packet = net->read_pos; + packet_start = packet = net->read_pos; if (packet[REPLY_MAGIC_NUM_OFFSET] != ReplSemiSyncMaster::kPacketMagicNum) { sql_print_error("Read semi-sync reply magic number error"); goto l_end; }

+ /* we determine if this semisync ack contains a sql-thread exec-pos + * by checking if last byte == 0, since the packet then contains + * \0-terminated filenames */ + exec_pos_present = packet[packet_len - 1] == 0; + log_file_pos = uint8korr(packet + REPLY_BINLOG_POS_OFFSET); - log_file_len = packet_len - REPLY_BINLOG_NAME_OFFSET; + if (exec_pos_present == false) + { + log_file_len = packet_len - REPLY_BINLOG_NAME_OFFSET; + } + else + { + log_file_len = strnlen((char*)packet + REPLY_BINLOG_NAME_OFFSET, + MY_MIN((ulong)FN_REFLEN, + packet_len - REPLY_BINLOG_NAME_OFFSET)); + } if (log_file_len >= FN_REFLEN) { sql_print_error("Read semi-sync reply binlog file length too large"); goto l_end; } - strncpy(log_file_name, (const char*)packet + REPLY_BINLOG_NAME_OFFSET, log_file_len); + packet+= REPLY_BINLOG_NAME_OFFSET; + + strncpy(log_file_name, (const char*)packet, log_file_len); log_file_name[log_file_len] = 0;

+ if (exec_pos_present) + { + packet += log_file_len + 1; + if (packet + 8 + 1 >= (packet_start + packet_len)) + { + sql_print_error("Read semi-sync reply binlog. " + "Packet to short to contain exec-position!"); + goto l_end; + } + exec_pos.file_pos = uint8korr(packet); + packet += 8; + strncpy(exec_pos.file_name, (char*)packet, + (packet_start + packet_len) - packet); + } + if (trc_level & kTraceDetail) sql_print_information("%s: Got reply (%s, %lu)", kWho, log_file_name, (ulong)log_file_pos);

- result = reportReplyBinlog(server_id, log_file_name, log_file_pos); + result = reportReplyBinlog(server_id, log_file_name, log_file_pos, + exec_pos_present ? &exec_pos : NULL);

l_end: return function_exit(kWho, result); @@ -1154,6 +1421,16 @@ int ReplSemiSyncMaster::resetMaster() wait_file_name_inited_ = false; reply_file_name_inited_ = false; commit_file_name_inited_ = false; + if (active_tranxs_ != NULL) + { + /** + * make sure to empty transaction hash/list + * with slave-lag reporting this container does + * not have to be empty even if no transaction is + * currently running + */ + active_tranxs_->clear_active_tranx_nodes(); + }

rpl_semi_sync_master_yes_transactions = 0; rpl_semi_sync_master_no_transactions = 0; @@ -1168,6 +1445,13 @@ int ReplSemiSyncMaster::resetMaster()

unlock();

+ mysql_mutex_lock(&LOCK_slave_lag_); + rpl_semi_sync_master_slave_lag_wait_sessions = 0; + oldest_unapplied_tranx_commit_time_us_ = 0; + rpl_semi_sync_master_trx_slave_lag_wait_num = 0; + rpl_semi_sync_master_trx_slave_lag_wait_time = 0; + mysql_mutex_unlock(&LOCK_slave_lag_); + return function_exit(kWho, result); }

@@ -1186,6 +1470,29 @@ void ReplSemiSyncMaster::setExportStats() ((double)rpl_semi_sync_master_net_wait_num)) : 0);

unlock(); + + if (oldest_unapplied_tranx_commit_time_us_ != 0) + { + rpl_semi_sync_master_estimated_slave_lag = my_hrtime().val - + oldest_unapplied_tranx_commit_time_us_; + } + else + { + rpl_semi_sync_master_estimated_slave_lag = 0; + } + + mysql_mutex_lock(&LOCK_slave_lag_); + if (rpl_semi_sync_master_trx_slave_lag_wait_num) + { + rpl_semi_sync_master_avg_trx_slave_lag_wait_time = + (unsigned long)((double)rpl_semi_sync_master_trx_slave_lag_wait_time / + (double)rpl_semi_sync_master_trx_slave_lag_wait_num); + } + else + { + rpl_semi_sync_master_avg_trx_slave_lag_wait_time = 0; + } + mysql_mutex_unlock(&LOCK_slave_lag_); }

/* Get the waiting time given the wait's staring time. @@ -1213,3 +1520,117 @@ static int getWaitTime(const struct timespec& start_ts)

return (int)(end_usecs - start_usecs); } + +void ReplSemiSyncMaster::wake_slave_lag_waiters( + ulonglong oldest_unapplied_tranx_commit_time_us) +{ + mysql_mutex_lock(&LOCK_slave_lag_); + oldest_unapplied_tranx_commit_time_us_ = + oldest_unapplied_tranx_commit_time_us; + + if (rpl_semi_sync_master_slave_lag_wait_sessions > 0) + { + mysql_cond_broadcast(&COND_slave_lag_); + } + mysql_mutex_unlock(&LOCK_slave_lag_); +} + +int ReplSemiSyncMaster::wait_slave_lag(ulong timeout_sec) +{ + int error = 0; + PSI_stage_info old_stage; + + /* slave lag waiting not enabled, return directly */ + if (rpl_semi_sync_master_max_slave_lag == 0) + return 0; + + /* there is no slave that can report slave lag, return directly */ + if (rpl_semi_sync_master_slave_lag_clients == 0) + return 0; + + /* compute start_time and end_time */ + struct timespec end_time; + set_timespec(end_time, 0); + ulonglong start_time_us = timespec_to_usec(&end_time); + end_time.tv_sec += timeout_sec; + + mysql_mutex_lock(&LOCK_slave_lag_); + + if (oldest_unapplied_tranx_commit_time_us_ == 0) + { + /* no slave lag, atleast one slave is up to date */ + mysql_mutex_unlock(&LOCK_slave_lag_); + return 0; + } + + if (rpl_semi_sync_master_max_slave_lag == 0) + { + /* slave lag waiting not enabled */ + mysql_mutex_unlock(&LOCK_slave_lag_); + return 0; + } + + /* This must be called after acquired the lock */ + THD_ENTER_COND(NULL, &COND_slave_lag_, &LOCK_slave_lag_, + &stage_waiting_for_semi_sync_slave_lag, + &old_stage); + + bool waited = false; + ulonglong lag = 0; + ulonglong max_lag = 0; + while (oldest_unapplied_tranx_commit_time_us_ != 0) + { + /* check kill_level after THD_ENTER_COND but *before* cond_wait + * to avoid missing kills */ + if (! (getMasterEnabled() && is_on() && + thd_kill_level(NULL) == THD_IS_NOT_KILLED)) + break; + + lag = start_time_us - oldest_unapplied_tranx_commit_time_us_; + max_lag = 1000000 * rpl_semi_sync_master_max_slave_lag; + if (lag <= max_lag) + break; + + waited = true; + rpl_semi_sync_master_slave_lag_wait_sessions++; + int wait_result = mysql_cond_timedwait(&COND_slave_lag_, &LOCK_slave_lag_, + &end_time); + rpl_semi_sync_master_slave_lag_wait_sessions--; + + bool thd_was_killed = thd_kill_level(NULL) != THD_IS_NOT_KILLED; + if (wait_result != 0 || thd_was_killed) + { + break; + } + } + + if (thd_kill_level(NULL) != THD_IS_NOT_KILLED) + { + /* Return error to client. */ + error = 1; + my_printf_error(ER_ERROR_DURING_COMMIT, + "Killed while waiting for replication semi-sync slave-lag.", + MYF(0)); + } + else if (lag > max_lag) + { + /* Return error to client. */ + error = 1; + my_printf_error(ER_ERROR_DURING_COMMIT, + "Slave-lag timeout", + MYF(0)); + } + + if (waited) + { + rpl_semi_sync_master_trx_slave_lag_wait_num++; + rpl_semi_sync_master_trx_slave_lag_wait_time += + (my_hrtime().val - start_time_us); + } + + /* The lock held will be released by thd_exit_cond, so no need to + call unlock() here */ + THD_EXIT_COND(NULL, & old_stage); + + return error; +} diff --git a/plugin/semisync/semisync_master.h b/plugin/semisync/semisync_master.h index d9dc4ce..e5c0de6 100644 --- a/plugin/semisync/semisync_master.h +++ b/plugin/semisync/semisync_master.h @@ -24,17 +24,101 @@ #ifdef HAVE_PSI_INTERFACE extern PSI_mutex_key key_ss_mutex_LOCK_binlog_; extern PSI_cond_key key_ss_cond_COND_binlog_send_; + +extern PSI_mutex_key key_ss_mutex_LOCK_slave_lag_; +extern PSI_cond_key key_ss_cond_COND_slave_lag_; #endif

extern PSI_stage_info stage_waiting_for_semi_sync_ack_from_slave; +extern PSI_stage_info stage_waiting_for_semi_sync_slave_lag;

struct TranxNode { char log_name_[FN_REFLEN]; - my_off_t log_pos_; + my_off_t log_pos_; + ulonglong tranx_commit_time_us; struct TranxNode *next_; /* the next node in the sorted list */ struct TranxNode *hash_next_; /* the next node during hash collision */ };

+struct LogPos; + +/* This represent a log position */ +struct LogPosPtr { + LogPosPtr() { Uninit();} + LogPosPtr(const char *name, my_off_t pos) : file_name(name), file_pos(pos){} + explicit LogPosPtr(const LogPos& pos) { Assign(&pos); } + + const char *file_name; + my_off_t file_pos; + + LogPosPtr& Assign(const LogPosPtr *src) { + file_name = src->file_name; + file_pos = src->file_pos; + return *this; + } + + LogPosPtr& Assign(const LogPos *src); + + void Uninit() { file_name = NULL;} + bool IsInited() const { return file_name != NULL; } +}; + +struct LogPos { + char file_name[FN_REFLEN]; + my_off_t file_pos; + + LogPos() { Uninit(); } + + LogPosPtr ToLogPosPtr() const { + if (IsInited()){ + LogPosPtr p(file_name, file_pos); + return p; + } else { + LogPosPtr p; + return p; + } + } + + LogPos& Assign(const LogPosPtr *src) { + if (src->IsInited()) { + strcpy(file_name, src->file_name); + file_pos = src->file_pos; + } else { + Uninit(); + } + return *this; + } + + LogPos& Assign(const LogPos* src) { + LogPosPtr p = src->ToLogPosPtr(); + Assign(&p); + return *this; + } + + void Uninit() { file_name[0] = 0; } + bool IsInited() const { return file_name[0] != 0; } +}; + +inline LogPosPtr& LogPosPtr::Assign(const LogPos* src) { + LogPosPtr p = src->ToLogPosPtr(); + Assign(&p); + return *this; +} + +inline int CompareLogPos(const LogPosPtr *pos1, const LogPosPtr *pos2) { + int cmp = strcmp(pos1->file_name, pos2->file_name); + + if (cmp != 0) + return cmp; + + if (pos1->file_pos > pos2->file_pos) + return 1; + else if (pos1->file_pos < pos2->file_pos) + return -1; + else + return 0; +} + /** @class TranxNodeAllocator

@@ -329,10 +413,14 @@ class ActiveTranx node2->log_name_, node2->log_pos_); }

+ void set_new_front(TranxNode* new_front); + public: ActiveTranx(mysql_mutex_t *lock, unsigned long trace_level); ~ActiveTranx();

+ bool is_empty() const { return trx_front_ == NULL; } + /* Insert an active transaction node with the specified position. * * Return: @@ -340,21 +428,42 @@ class ActiveTranx */ int insert_tranx_node(const char *log_file_name, my_off_t log_file_pos);

- /* Clear the active transaction nodes until(inclusive) the specified - * position. - * If log_file_name is NULL, everything will be cleared: the sorted + /* Clear the active transaction + * Everything will be cleared: the sorted * list and the hash table will be reset to empty. - * + * * Return: - * 0: success; non-zero: error + * 0 success; non-zero: error + */ + int clear_active_tranx_nodes(); + + /* Prune the active transaction nodes until the specified + * position (inclusive). + * + * Return: + * true if any transaction was removed + * false if list was left unchanged */ - int clear_active_tranx_nodes(const char *log_file_name, - my_off_t log_file_pos); + bool prune_active_tranx_nodes(LogPosPtr logpos, + ulonglong *oldest_tranx_commit_time_us);

- /* Given a position, check to see whether the position is an active - * transaction's ending position by probing the hash table. + /* Lookup a transaction's ending position by probing the hash table. + * + * return entry if found or NULL otherwise */ - bool is_tranx_end_pos(const char *log_file_name, my_off_t log_file_pos); + TranxNode* lookup_tranx_end_pos(const char *log_file_name, + my_off_t log_file_pos); + + + /* Check if an entry is rear (i.e last) */ + bool is_rear(TranxNode* entry) const { return entry == trx_rear_; } + + /** + * return timestamp of oldest transaction in list + */ + ulonglong get_oldest_tranx_commit_time_us() const { + return trx_front_->tranx_commit_time_us; + }

/* Given two binlog positions, compare which one is bigger based on * (file_name, file_position). @@ -365,6 +474,24 @@ class ActiveTranx };

/** + * State that semisync master keeps per slave + */ +struct ReplSemiSyncMasterPerSlaveState +{ + ReplSemiSyncMasterPerSlaveState() : unacked_event_count_(0) {} + + /** + * No of events that has not been semi-sync acked + */ + unsigned unacked_event_count_; + + /** + * Position of last event that was semisync'ed + */ + LogPos sync_req_pos_; +}; + +/** The extension class for the master of semi-synchronous replication */ class ReplSemiSyncMaster @@ -432,6 +559,16 @@ class ReplSemiSyncMaster

bool state_; /* whether semi-sync is switched */

+ /* This cond variable is signaled when slave lag has decreased */ + mysql_cond_t COND_slave_lag_; + + /* Mutex that protects oldest_unapplied_tranx_commit_time_us */ + mysql_mutex_t LOCK_slave_lag_; + + /* this is commit time of oldest transaction that has not been applied + * on any slave */ + ulonglong oldest_unapplied_tranx_commit_time_us_; + void lock(); void unlock(); void cond_broadcast(); @@ -493,6 +630,9 @@ class ReplSemiSyncMaster /* Is the slave servered by the thread requested semi-sync */ bool is_semi_sync_slave();

+ /* Does this slave have slave lag reporting capabilities */ + bool has_semi_sync_slave_lag(); + /* In semi-sync replication, reports up to which binlog position we have * received replies from the slave indicating that it already get the events. * @@ -501,13 +641,15 @@ class ReplSemiSyncMaster * log_file_name - (IN) binlog file name * end_offset - (IN) the offset in the binlog file up to which we have * the replies from the slave + * exec_position - (IN) position of SQL thread or NULL if not present * * Return: * 0: success; non-zero: error */ int reportReplyBinlog(uint32 server_id, const char* log_file_name, - my_off_t end_offset); + my_off_t end_offset, + const LogPos *exec_position);

/* Commit a transaction in the final step. This function is called from * InnoDB before returning from the low commit. If semi-sync is switch on, @@ -540,6 +682,16 @@ class ReplSemiSyncMaster */ int reserveSyncHeader(unsigned char *header, unsigned long size);

+ /* + * check if an event should be semi synced and optionally + * if it should report back position of SQL thread on slave + * + * return 0 - no semi sync + * 1 - semi sync + * 2 - semi sync and report exec position + */ + int checkSyncReq(const LogPosPtr *log_pos); + /* Update the sync bit in the packet header to indicate to the slave whether * the master will wait for the reply of the event. If semi-sync is switched * off and we detect that the slave is catching up, we switch semi-sync on. @@ -592,6 +744,21 @@ class ReplSemiSyncMaster * go off for that. */ int resetMaster(); + + /** + * wake potential slave-lag waiters + * called by binlog dump-thread(s) + */ + void wake_slave_lag_waiters(ulonglong oldest_unapplied_tranx_commit_time_us); + + /** + * wait for slave lag to get below threshold + * called by user-thread(s) + * + * return 0 - success + * 1 - timeout + */ + int wait_slave_lag(ulong max_wait_time_sec); };

enum rpl_semi_sync_master_wait_point_t { @@ -621,6 +788,17 @@ extern unsigned long long rpl_semi_sync_master_trx_wait_num; extern unsigned long long rpl_semi_sync_master_net_wait_time; extern unsigned long long rpl_semi_sync_master_trx_wait_time;

+extern unsigned long rpl_semi_sync_master_max_unacked_event_count; +extern unsigned long rpl_semi_sync_master_max_unacked_event_bytes; +extern unsigned long rpl_semi_sync_master_max_slave_lag; +extern unsigned long rpl_semi_sync_master_slave_lag_heartbeat_frequency_us; +extern unsigned long rpl_semi_sync_master_slave_lag_wait_sessions; +extern unsigned long long rpl_semi_sync_master_estimated_slave_lag; + +extern unsigned long rpl_semi_sync_master_avg_trx_slave_lag_wait_time; +extern unsigned long long rpl_semi_sync_master_trx_slave_lag_wait_num; +extern unsigned long long rpl_semi_sync_master_trx_slave_lag_wait_time; + /* This indicates whether we should keep waiting if no semi-sync slave is available. diff --git a/plugin/semisync/semisync_master_plugin.cc b/plugin/semisync/semisync_master_plugin.cc index 7bb0eea..3282158 100644 --- a/plugin/semisync/semisync_master_plugin.cc +++ b/plugin/semisync/semisync_master_plugin.cc @@ -21,6 +21,11 @@

static ReplSemiSyncMaster repl_semisync;

+// forward declaration +static inline ulong get_slave_lag_wait_timeout(THD* thd); + +static char rpl_semi_sync_master_group_commit = 0; + C_MODE_START

int repl_semi_report_binlog_update(Binlog_storage_param *param, @@ -31,6 +36,13 @@ int repl_semi_report_binlog_update(Binlog_storage_param *param,

if (repl_semisync.getMasterEnabled()) { + if (rpl_semi_sync_master_group_commit && + ((flags & BINLOG_GROUP_COMMIT_TRAILER) == 0)) + { + /** there are transactions more coming... */ + return 0; + } + /* Let us store the binlog file name and the position, so that we know how long to wait for the binlog to the replicated to @@ -43,8 +55,11 @@ int repl_semi_report_binlog_update(Binlog_storage_param *param, return error; }

-int repl_semi_request_commit(Trans_param *param) +int repl_semi_before_commit(Trans_param *param, int *error) { + *error = repl_semisync.wait_slave_lag( + get_slave_lag_wait_timeout(current_thd)); + return 0; }

@@ -53,6 +68,14 @@ int repl_semi_report_binlog_sync(Binlog_storage_param *param, my_off_t log_pos, uint32 flags) { int error= 0; + + if (rpl_semi_sync_master_group_commit && + ((flags & BINLOG_GROUP_COMMIT_TRAILER) == 0)) + { + /** there are transactions more coming... */ + return 0; + } + if (rpl_semi_sync_master_wait_point == SEMI_SYNC_MASTER_WAIT_POINT_AFTER_BINLOG_SYNC) { @@ -100,7 +123,7 @@ int repl_semi_binlog_dump_start(Binlog_transmit_param *param, Let's assume this semi-sync slave has already received all binlog events before the filename and position it requests. */ - repl_semisync.reportReplyBinlog(param->server_id, log_file, log_pos); + repl_semisync.reportReplyBinlog(param->server_id, log_file, log_pos, NULL); } sql_print_information("Start %s binlog_dump to slave (server_id: %d), pos(%s, %lu)", semi_sync_slave ? "semi-sync" : "asynchronous", @@ -242,15 +265,72 @@ static MYSQL_SYSVAR_ULONG(trace_level, rpl_semi_sync_master_trace_level, &fix_rpl_semi_sync_master_trace_level, // update 32, 0, ~0UL, 1);

+static MYSQL_SYSVAR_ULONG(max_unacked_event_count, + rpl_semi_sync_master_max_unacked_event_count, + PLUGIN_VAR_OPCMDARG, + "Maximum unacked replication events", + NULL, // check + NULL, // update + rpl_semi_sync_master_max_unacked_event_count, 0, ~0UL, 1); + +static MYSQL_SYSVAR_ULONG(max_unacked_event_bytes, + rpl_semi_sync_master_max_unacked_event_bytes, + PLUGIN_VAR_OPCMDARG, + "Maximum unacked replication bytes", + NULL, // check + NULL, // update + rpl_semi_sync_master_max_unacked_event_bytes, 0, ~0UL, 1); + +static MYSQL_SYSVAR_ULONG(max_slave_lag, rpl_semi_sync_master_max_slave_lag, + PLUGIN_VAR_OPCMDARG, + "Maximum allowed lag of fastest semi-sync slave (in seconds), " + "checked before commit.", + NULL, // check + NULL, // update + rpl_semi_sync_master_max_slave_lag, 0, ~0UL, 1); + +static MYSQL_THDVAR_ULONG(slave_lag_wait_timeout, + PLUGIN_VAR_RQCMDARG, + "Timeout in seconds a rw-transaction may wait for max slave lag before " + "being rolled back.", + NULL, NULL, 50, 1, 1024 * 1024 * 1024, 0); + +static MYSQL_SYSVAR_ULONG( + slave_lag_heartbeat_frequency_us, + rpl_semi_sync_master_slave_lag_heartbeat_frequency_us, + PLUGIN_VAR_RQCMDARG, + "Heartbeat frequency when slave-lag is enabled (in microseconds).", + NULL, // check + NULL, // update + 500000, /* 500 ms */ + 1, ~0UL, 1); + +static MYSQL_SYSVAR_BOOL(group_commit, rpl_semi_sync_master_group_commit, + PLUGIN_VAR_OPCMDARG, + "Group commit for semi sync", + NULL, // check + NULL, + 0); + static SYS_VAR* semi_sync_master_system_vars[]= { MYSQL_SYSVAR(enabled), MYSQL_SYSVAR(wait_point), MYSQL_SYSVAR(timeout), MYSQL_SYSVAR(wait_no_slave), MYSQL_SYSVAR(trace_level), + MYSQL_SYSVAR(max_unacked_event_count), + MYSQL_SYSVAR(max_unacked_event_bytes), + MYSQL_SYSVAR(max_slave_lag), + MYSQL_SYSVAR(slave_lag_wait_timeout), + MYSQL_SYSVAR(slave_lag_heartbeat_frequency_us), + MYSQL_SYSVAR(group_commit), NULL, };

+static inline ulong get_slave_lag_wait_timeout(THD* thd) +{ + return THDVAR(thd, slave_lag_wait_timeout); +}

static void fix_rpl_semi_sync_master_timeout(MYSQL_THD thd, SYS_VAR *var, @@ -297,6 +377,7 @@ Trans_observer trans_observer = {

repl_semi_report_commit, // after_commit repl_semi_report_rollback, // after_rollback + repl_semi_before_commit, // before commit };

Binlog_storage_observer storage_observer = { @@ -339,7 +420,11 @@ DEF_SHOW_FUNC(net_wait_time, SHOW_LONGLONG) DEF_SHOW_FUNC(net_wait_num, SHOW_LONGLONG) DEF_SHOW_FUNC(avg_net_wait_time, SHOW_LONG) DEF_SHOW_FUNC(avg_trx_wait_time, SHOW_LONG) - +DEF_SHOW_FUNC(slave_lag_wait_sessions, SHOW_LONG) +DEF_SHOW_FUNC(estimated_slave_lag, SHOW_LONGLONG) +DEF_SHOW_FUNC(trx_slave_lag_wait_time, SHOW_LONGLONG) +DEF_SHOW_FUNC(trx_slave_lag_wait_num, SHOW_LONGLONG) +DEF_SHOW_FUNC(avg_trx_slave_lag_wait_time, SHOW_LONG)

/* plugin status variables */ static SHOW_VAR semi_sync_master_status_vars[]= { @@ -385,32 +470,55 @@ static SHOW_VAR semi_sync_master_status_vars[]= { {"Rpl_semi_sync_master_net_avg_wait_time", (char*) &SHOW_FNAME(avg_net_wait_time), SHOW_SIMPLE_FUNC}, + {"Rpl_semi_sync_master_slave_lag_wait_sessions", + (char*) &SHOW_FNAME(slave_lag_wait_sessions), + SHOW_SIMPLE_FUNC}, + {"Rpl_semi_sync_master_estimated_slave_lag", + (char*) &SHOW_FNAME(estimated_slave_lag), + SHOW_SIMPLE_FUNC}, + {"Rpl_semi_sync_master_tx_slave_lag_wait_time", + (char*) &SHOW_FNAME(trx_slave_lag_wait_time), + SHOW_SIMPLE_FUNC}, + {"Rpl_semi_sync_master_tx_slave_lag_waits", + (char*) &SHOW_FNAME(trx_slave_lag_wait_num), + SHOW_SIMPLE_FUNC}, + {"Rpl_semi_sync_master_tx_avg_slave_lag_wait_time", + (char*) &SHOW_FNAME(avg_trx_slave_lag_wait_time), + SHOW_SIMPLE_FUNC}, {NULL, NULL, SHOW_LONG}, };

#ifdef HAVE_PSI_INTERFACE PSI_mutex_key key_ss_mutex_LOCK_binlog_; +PSI_mutex_key key_ss_mutex_LOCK_slave_lag_;

static PSI_mutex_info all_semisync_mutexes[]= { - { &key_ss_mutex_LOCK_binlog_, "LOCK_binlog_", 0} + { &key_ss_mutex_LOCK_binlog_, "LOCK_binlog_", 0 }, + { &key_ss_mutex_LOCK_slave_lag_, "LOCK_slave_lag_", 0 } };

PSI_cond_key key_ss_cond_COND_binlog_send_; +PSI_cond_key key_ss_cond_COND_slave_lag_;

static PSI_cond_info all_semisync_conds[]= { - { &key_ss_cond_COND_binlog_send_, "COND_binlog_send_", 0} + { &key_ss_cond_COND_binlog_send_, "COND_binlog_send_", 0 }, + { &key_ss_cond_COND_slave_lag_, "COND_slave_lag_", 0 } }; #endif /* HAVE_PSI_INTERFACE */

PSI_stage_info stage_waiting_for_semi_sync_ack_from_slave= { 0, "Waiting for semi-sync ACK from slave", 0};

+PSI_stage_info stage_waiting_for_semi_sync_slave_lag= +{ 0, "Waiting for semi-sync slave lag", 0}; + #ifdef HAVE_PSI_INTERFACE PSI_stage_info *all_semisync_stages[]= { - & stage_waiting_for_semi_sync_ack_from_slave + & stage_waiting_for_semi_sync_ack_from_slave, + & stage_waiting_for_semi_sync_slave_lag };

static void init_semisync_psi_keys(void) @@ -492,4 +600,3 @@ maria_declare_plugin(semisync_master) MariaDB_PLUGIN_MATURITY_GAMMA } maria_declare_plugin_end; - diff --git a/plugin/semisync/semisync_slave.cc b/plugin/semisync/semisync_slave.cc index 5f98472..839e0cc 100644 --- a/plugin/semisync/semisync_slave.cc +++ b/plugin/semisync/semisync_slave.cc @@ -20,6 +20,7 @@ char rpl_semi_sync_slave_enabled; char rpl_semi_sync_slave_status= 0; unsigned long rpl_semi_sync_slave_trace_level; +char rpl_semi_sync_slave_lag_enabled= 0;

int ReplSemiSyncSlave::initObject() { @@ -42,7 +43,7 @@ int ReplSemiSyncSlave::initObject()

int ReplSemiSyncSlave::slaveReadSyncHeader(const char *header, unsigned long total_len, - bool *need_reply, + unsigned char *need_reply, const char **payload, unsigned long *payload_len) { @@ -52,7 +53,7 @@ int ReplSemiSyncSlave::slaveReadSyncHeader(const char *header,

if ((unsigned char)(header[0]) == kPacketMagicNum) { - *need_reply = (header[1] & kPacketFlagSync); + *need_reply = (header[1] & (kPacketFlagSync | kPacketFlagSyncAndReport)); *payload_len = total_len - 2; *payload = header + 2;

@@ -95,16 +96,20 @@ int ReplSemiSyncSlave::slaveStop(Binlog_relay_IO_param *param) return 0; }

-int ReplSemiSyncSlave::slaveReply(MYSQL *mysql, - const char *binlog_filename, - my_off_t binlog_filepos) +int ReplSemiSyncSlave::slaveReply(unsigned char header_byte, + MYSQL *mysql, + const char *binlog_filename, + my_off_t binlog_filepos, + Master_info * mi) { const char *kWho = "ReplSemiSyncSlave::slaveReply"; NET *net= &mysql->net; - uchar reply_buffer[REPLY_MAGIC_NUM_LEN - + REPLY_BINLOG_POS_LEN - + REPLY_BINLOG_NAME_LEN]; + uchar reply_buffer[REPLY_MAGIC_NUM_LEN + + 2 * ( REPLY_BINLOG_POS_LEN + + REPLY_BINLOG_NAME_LEN + + /* '\0' */ 1) ]; int reply_res, name_len = strlen(binlog_filename); + int msg_len = name_len + REPLY_BINLOG_NAME_OFFSET;

function_enter(kWho);

@@ -119,10 +124,29 @@ int ReplSemiSyncSlave::slaveReply(MYSQL *mysql, sql_print_information("%s: reply (%s, %lu)", kWho, binlog_filename, (ulong)binlog_filepos);

+ if (header_byte & kPacketFlagSyncAndReport) + { + /** + * master requests that we also report back SQL-thread position + */ + + // where to store sql filename/position + char *bufptr = (char*)reply_buffer + msg_len; + bufptr[0] = 0; // '\0' terminate previous filename + bufptr++; + + my_off_t sql_file_pos; + // get file/position and store the filename directly info bufptr+8 + size_t name_len2 = get_master_log_pos(mi, bufptr + 8, &sql_file_pos); + int8store(bufptr, sql_file_pos); // store position + + msg_len += /* '\0' */ 1 + /* position */ 8 + name_len2 + /* '\0' */ 1; + } + net_clear(net, 0); /* Send the reply. */ - reply_res = my_net_write(net, reply_buffer, - name_len + REPLY_BINLOG_NAME_OFFSET); + reply_res = my_net_write(net, reply_buffer, msg_len); + if (!reply_res) { reply_res = net_flush(net); diff --git a/plugin/semisync/semisync_slave.h b/plugin/semisync/semisync_slave.h index 1bf8cf3..c91847d 100644 --- a/plugin/semisync/semisync_slave.h +++ b/plugin/semisync/semisync_slave.h @@ -60,23 +60,30 @@ class ReplSemiSyncSlave * Return: * 0: success; non-zero: error */ - int slaveReadSyncHeader(const char *header, unsigned long total_len, bool *need_reply, + int slaveReadSyncHeader(const char *header, unsigned long total_len, + unsigned char *need_reply_byte, const char **payload, unsigned long *payload_len);

/* A slave replies to the master indicating its replication process. It * indicates that the slave has received all events before the specified * binlog position. - * + * * Input: + * need_reply_byte - (IN) the header byte * mysql - (IN) the mysql network connection * binlog_filename - (IN) the reply point's binlog file name * binlog_filepos - (IN) the reply point's binlog file offset + * master_info - (IN) the master info struct so that we can get more + * info if needed * * Return: * 0: success; non-zero: error */ - int slaveReply(MYSQL *mysql, const char *binlog_filename, - my_off_t binlog_filepos); + int slaveReply(unsigned char need_reply_byte, + MYSQL *mysql, + const char *binlog_filename, + my_off_t binlog_filepos, + Master_info* master_info);

int slaveStart(Binlog_relay_IO_param *param); int slaveStop(Binlog_relay_IO_param *param); @@ -93,5 +100,6 @@ class ReplSemiSyncSlave extern char rpl_semi_sync_slave_enabled; extern unsigned long rpl_semi_sync_slave_trace_level; extern char rpl_semi_sync_slave_status; +extern char rpl_semi_sync_slave_lag_enabled;

#endif /* SEMISYNC_SLAVE_H */ diff --git a/plugin/semisync/semisync_slave_plugin.cc b/plugin/semisync/semisync_slave_plugin.cc index 572ead2..0bf03be 100644 --- a/plugin/semisync/semisync_slave_plugin.cc +++ b/plugin/semisync/semisync_slave_plugin.cc @@ -28,7 +28,7 @@ static ReplSemiSyncSlave repl_semisync; event read is the last event of a transaction. And the value is checked in repl_semi_slave_queue_event. */ -bool semi_sync_need_reply= false; +unsigned char semi_sync_need_reply= 0;

C_MODE_START

@@ -81,6 +81,23 @@ int repl_semi_slave_request_dump(Binlog_relay_IO_param *param, return 1; } mysql_free_result(mysql_store_result(mysql)); + + if (rpl_semi_sync_slave_lag_enabled) + { + char buf[100]; + /* + Tell master that we can do exec-position reporting + */ + snprintf(buf, sizeof(buf), "SET @%s= 1", + ReplSemiSyncBase::kRplSemiSyncSlaveReportExec); + if (mysql_real_query(mysql, buf, strlen(buf))) + { + sql_print_error("query: %s on master failed", buf); + return 1; + } + mysql_free_result(mysql_store_result(mysql)); + } + rpl_semi_sync_slave_status= 1; return 0; } @@ -110,9 +127,11 @@ int repl_semi_slave_queue_event(Binlog_relay_IO_param *param, should not cause the slave IO thread to stop, and the error messages are already reported. */ - (void) repl_semisync.slaveReply(param->mysql, + (void) repl_semisync.slaveReply(semi_sync_need_reply, + param->mysql, param->master_log_name, - param->master_log_pos); + param->master_log_pos, + param->mi); } return 0; } @@ -164,9 +183,17 @@ static MYSQL_SYSVAR_ULONG(trace_level, rpl_semi_sync_slave_trace_level, &fix_rpl_semi_sync_trace_level, // update 32, 0, ~0UL, 1);

+static MYSQL_SYSVAR_BOOL(lag_enabled, rpl_semi_sync_slave_lag_enabled, + PLUGIN_VAR_OPCMDARG, + "Enable semi-synchronous replication slave lag reporting. ", + NULL, // check + NULL, // update + 0); + static SYS_VAR* semi_sync_slave_system_vars[]= { MYSQL_SYSVAR(enabled), MYSQL_SYSVAR(trace_level), + MYSQL_SYSVAR(lag_enabled), NULL, };

@@ -230,4 +257,3 @@ maria_declare_plugin(semisync_slave) MariaDB_PLUGIN_MATURITY_GAMMA } maria_declare_plugin_end; - diff --git a/sql/handler.cc b/sql/handler.cc index 3ca9ec3..3e6cd65 100644 --- a/sql/handler.cc +++ b/sql/handler.cc @@ -1364,12 +1364,30 @@ int ha_commit_trans(THD *thd, bool all) uint rw_ha_count= ha_check_and_coalesce_trx_read_only(thd, ha_info, all); /* rw_trans is TRUE when we in a transaction changing data */ bool rw_trans= is_real_trans && (rw_ha_count > 0); + bool mdl_request_initialized= false; MDL_request mdl_request; DBUG_PRINT("info", ("is_real_trans: %d rw_trans: %d rw_ha_count: %d", is_real_trans, rw_trans, rw_ha_count));

if (rw_trans) { + /* check READ-ONLY just before before_commit hook to decrease likelihood + * of having threads hanging waiting for slave-lag only to be aborted + * due to read-only. + */ + if (opt_readonly && + !(thd->security_ctx->master_access & SUPER_ACL) && + !thd->slave_thread) + { + my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only"); + goto err; + } + + if (RUN_HOOK(transaction, before_commit, (thd))) + { + goto err; + } + /* Acquire a metadata lock which will ensure that COMMIT is blocked by an active FLUSH TABLES WITH READ LOCK (and vice versa: @@ -1378,6 +1396,7 @@ int ha_commit_trans(THD *thd, bool all) We allow the owner of FTWRL to COMMIT; we assume that it knows what it does. */ + mdl_request_initialized= true; mdl_request.init(MDL_key::COMMIT, "", "", MDL_INTENTION_EXCLUSIVE, MDL_EXPLICIT);

@@ -1486,7 +1505,7 @@ int ha_commit_trans(THD *thd, bool all) ha_rollback_trans(thd, all);

end: - if (rw_trans && mdl_request.ticket) + if (rw_trans && mdl_request_initialized && mdl_request.ticket) { /* We do not always immediately release transactional locks diff --git a/sql/replication.h b/sql/replication.h index 4731c22..309bdb4 100644 --- a/sql/replication.h +++ b/sql/replication.h @@ -110,6 +110,21 @@ typedef struct Trans_observer { @retval 1 Failure */ int (*after_rollback)(Trans_param *param); + + /** + This callback is called before transaction commit + If function does not return *error == 0 transaction will + not be committed but error code will be returned to client + + @note *error!=0 and return code 0 shall be used by plugin to signal + that transaction should be aborted. + If returning non-zero transaction will also be aborted and an error + will be printed to error log. + + @retval 0 Sucess + @retval non-zero error + */ + int (*before_commit)(Trans_param *param, int *error); } Trans_observer;

/** @@ -294,6 +309,8 @@ enum Binlog_relay_IO_flags { };

+class Master_info; + /** Replication binlog relay IO observer parameter */ @@ -309,8 +326,20 @@ typedef struct Binlog_relay_IO_param { my_off_t master_log_pos;

MYSQL *mysql; /* the connection to master */ + + Master_info * mi; /* master info handle */ } Binlog_relay_IO_param;

+ +/* get the master log given a Master_info + * and store it in filename_buf/filepos + * return length of filename (excluding \0) + * + * note: filename_buf should be a minimum FN_REFLEN + */ +size_t get_master_log_pos(const Master_info *mi, + char *filename_buf, my_off_t *filepos); + /** Observes and extends the service of slave IO thread. */ @@ -561,7 +590,21 @@ int get_user_var_str(const char *name, char *value, unsigned long len, unsigned int precision, int *null_value);

- + +/** + Set or replace the value of user variable as to an ulonglong + + @param name user variable name + @param value the value + @param old_value pointer to where old value will be stored (or NULL) + + @retval 0 Success, no prior value found + @retval 1 Success, old_value populated + @retval -1 Fail +*/ +int set_user_var_int(const char *name, + long long int value, + long long int *old_value);

#ifdef __cplusplus } diff --git a/sql/rpl_handler.cc b/sql/rpl_handler.cc index 3962600..209a809 100644 --- a/sql/rpl_handler.cc +++ b/sql/rpl_handler.cc @@ -23,6 +23,7 @@ #include "rpl_filter.h" #include #include "rpl_handler.h" +#include "sql_prepare.h"

Trans_delegate *transaction_delegate; Binlog_storage_delegate *binlog_storage_delegate; @@ -88,6 +89,42 @@ int get_user_var_str(const char *name, char *value, return 0; }

+int set_user_var_int(const char *name, + long long int value, + long long int *old_value) +{ + THD* thd= current_thd; + bool null_val; + user_var_entry *entry= + (user_var_entry*) my_hash_search(&thd->user_vars, + (uchar*) name, strlen(name)); + if (entry != NULL) + { + if (old_value != NULL) + *old_value= entry->val_int(&null_val); + } + + Ed_connection con(thd); + + char buf[256]; + int res= snprintf(buf, sizeof(buf), "SET @%s=%lld", name, value); + if (/* error */ res < 0 || + /* truncated */ res >= sizeof(buf)) + { + return -1; + } + + LEX_STRING str; + lex_string_set(&str, buf); + + if (con.execute_direct(str)) + { + return -1; + } + + return entry == NULL ? 0 : 1; +} + int delegates_init() { static my_aligned_storage trans_mem; @@ -249,6 +286,17 @@ int Trans_delegate::after_rollback(THD *thd, bool all) return ret; }

+int Trans_delegate::before_commit(THD *thd) +{ + int ret= 0, error= 0; + Trans_param param; + param.flags= 0; + param.log_file= 0; + param.log_pos= 0; + FOREACH_OBSERVER(ret, before_commit, thd, (¶m, &error)); + return error; +} + int Binlog_storage_delegate::after_flush(THD *thd, const char *log_file, my_off_t log_pos, @@ -374,17 +422,19 @@ int Binlog_transmit_delegate::reserve_header(THD *thd, ushort flags,

int Binlog_transmit_delegate::before_send_event(THD *thd, ushort flags, String *packet, - const char *log_file, + const char *log_file_path, my_off_t log_pos) { Binlog_transmit_param param; param.flags= flags;

int ret= 0; + const char* log_file_name= log_file_path != NULL ? + log_file_path + dirname_length(log_file_path) : NULL; FOREACH_OBSERVER(ret, before_send_event, false, (¶m, (uchar *)packet->c_ptr(), packet->length(), - log_file+dirname_length(log_file), log_pos)); + log_file_name, log_pos)); return ret; }

@@ -414,6 +464,7 @@ int Binlog_transmit_delegate::after_reset_master(THD *thd, ushort flags) void Binlog_relay_IO_delegate::init_param(Binlog_relay_IO_param *param, Master_info *mi) { + param->mi = mi; param->mysql= mi->mysql; param->user= mi->user; param->host= mi->host; @@ -540,6 +591,20 @@ int unregister_binlog_relay_io_observer(Binlog_relay_IO_observer *observer, void { return binlog_relay_io_delegate->remove_observer(observer, (st_plugin_int *)p); } + +/* get master log pos for a Master_info struct */ +size_t get_master_log_pos(const Master_info* mi, + char *filename_buf, my_off_t *filepos) +{ + mysql_mutex_t *mutex= &mi->rli.data_lock; + + mysql_mutex_lock(mutex); + *filepos= mi->rli.group_master_log_pos; + strncpy(filename_buf, mi->rli.group_master_log_name, FN_REFLEN); + mysql_mutex_unlock(mutex); + return strnlen(filename_buf, FN_REFLEN); +} + #else int register_binlog_transmit_observer(Binlog_transmit_observer *observer, void *p) { @@ -560,4 +625,13 @@ int unregister_binlog_relay_io_observer(Binlog_relay_IO_observer *observer, void { return 0; } + +size_t get_master_log_pos(const Master_info* mi, + char *filename_buf, my_off_t *filepos) +{ + *filepos= 0; + filename_buf[0]= 0; + return 0; +} + #endif /* HAVE_REPLICATION */ diff --git a/sql/rpl_handler.h b/sql/rpl_handler.h index afcfd9d..5119ee4 100644 --- a/sql/rpl_handler.h +++ b/sql/rpl_handler.h @@ -142,7 +142,7 @@ class Trans_delegate :public Delegate { public: typedef Trans_observer Observer; - int before_commit(THD *thd, bool all); + int before_commit(THD *thd); int before_rollback(THD *thd, bool all); int after_commit(THD *thd, bool all); int after_rollback(THD *thd, bool all); diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index 56300c6..d60a122 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -154,7 +154,9 @@ class Relay_log_info : public Slave_reporting_capability standard lock acquisition order to avoid deadlocks: run_lock, data_lock, relay_log.LOCK_log, relay_log.LOCK_index */ - mysql_mutex_t data_lock, run_lock; + mutable mysql_mutex_t data_lock; + mysql_mutex_t run_lock; + /* start_cond is broadcast when SQL thread is started stop_cond - when stopped diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index d9ae6ca..9662058 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -818,6 +818,13 @@ static int send_heartbeat_event(binlog_send_info *info, packet->append(b, sizeof(b)); }

+ if (RUN_HOOK(binlog_transmit, before_send_event, + (info->thd, info->flags, packet, 0, 0))) + { + info->error= ER_UNKNOWN_ERROR; + DBUG_RETURN(-1); + } + if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) || net_flush(net)) { @@ -825,6 +832,13 @@ static int send_heartbeat_event(binlog_send_info *info, DBUG_RETURN(-1); }

+ if (RUN_HOOK(binlog_transmit, after_send_event, + (info->thd, info->flags, packet))) + { + info->error= ER_UNKNOWN_ERROR; + DBUG_RETURN(-1); + } + DBUG_RETURN(0); }