1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implementation of Multi-source replication (MDEV:253)

Browse Source 
Documentation of the feature can be found at: http://kb.askmonty.org/en/multi-source-replication/
This code is based on code from Taobao, developed by Plinux

BUILD/SETUP.sh:
  Added -Wno-invalid-offsetof to get rid of warning of offsetof() on C++ class (safe in the contex we use it)
client/mysqltest.cc:
  Added support for error names starting with 'W'
  Added connection_name support to --sync_with_master
cmake/maintainer.cmake:
  Added -Wno-invalid-offsetof to get rid of warning of offsetof() on C++ class (safe in the contex we use it)
mysql-test/r/mysqltest.result:
  Updated results
mysql-test/r/parser.result:
  Updated results
mysql-test/suite/multi_source/my.cnf:
  Setup of multi-master tests
mysql-test/suite/multi_source/simple.result:
  Simple basic test of multi-source functionality
mysql-test/suite/multi_source/simple.test:
  Simple basic test of multi-source functionality
mysql-test/suite/multi_source/syntax.result:
  Test of multi-source syntax
mysql-test/suite/multi_source/syntax.test:
  Test of multi-source syntax
mysql-test/suite/rpl/r/rpl_rotate_logs.result:
  Updated results because of new error messages
mysql-test/t/parser.test:
  Updated test as master_pos_wait() now takes more arguments than before
sql/event_scheduler.cc:
  No reason to initialize slave_thread (it's guaranteed to be zero here)
sql/item_create.cc:
  Added connection_name argument to master_pos_wait()
  Simplified code
sql/item_func.cc:
  Added connection_name argument to master_pos_wait()
sql/item_func.h:
  Added connection_name argument to master_pos_wait()
sql/log.cc:
  Added tag "Master 'connection_name'" to slave errors that has a connection name.
sql/mysqld.cc:
  Added variable mysqld_server_initialized so that other functions can test if server is fully initialized.
  Free all slave data in one place (fewer ifdef's)
  Removed not needed call to close_active_mi()
  Initialize slaves() later in startup to ensure that everthing is really initialized when slaves start.
  Made status variable slave_running multi-source safe
sql/mysqld.h:
  Added mysqld_server_initialized
sql/rpl_mi.cc:
  Store connection name and cmp_connection_name (only used for show full slave status) in Master_info
  Added code for Master_info_index, which handles storage of multi-master information
  Don't write the empty "" connection_name to multi-master.info file. This is handled by the original code.
sql/rpl_mi.h:
  Added connection_name and Master_info_index
sql/rpl_rli.cc:
  Added connection_name to relay log files.
sql/rpl_rli.h:
  Fixed type of slave_skip_counter as we now access it directly in sys_vars.cc, so it must be uint
sql/share/errmsg-utf8.txt:
  Added new error messages needed for multi-source
  Added multi-source name to error ER_MASTER_INFO and WARN_NO_MASTER_INFO
sql/slave.cc:
  Moved things a bit around to make it easier to handle error conditions.
  Create a global master_info_index and add the "" connection to it
  Ensure that new Master_info doesn't fail.
  Don't call terminate_slave_threads(active_mi..) on end_slave() as this is now done automaticly when deleting master_info_index.
  Delete not needed function close_active_mi(). One can achive same thing by calling end_slave().
  Added support for SHOW FULL SLAVE STATUS (show status for all master connections with connection_name as first column)
sql/slave.h:
  Added new prototypes
sql/sql_base.cc:
  More DBUG_PRINT
sql/sql_class.cc:
  Reset thd->connection_name and thd-->default_master_connection
sql/sql_class.h:
  Added thd->connection_name and thd-->default_master_connection
  Added slave_skip_count to variables to make changing the @@sql_slave_skip_count variable thread safe
sql/sql_const.h:
  Added MAX_CONNECTION_NAME
sql/sql_lex.cc:
  Reset 'lex->verbose' (to simplify some sql_yacc.yy code)
sql/sql_lex.h:
  Added connection_name
sql/sql_parse.cc:
  Added support for connection_name to all SLAVE commands.
  - Instead of using active_mi, we now get the current Master_info from master_info_index.
  - Create new replication threads with CHANGE MASTER
  - Added support for show_all_master_info()
sql/sql_reload.cc:
  Made reset/full slave use master_info_index->get_master_info() instead of active_mi.
  If one uses 'RESET SLAVE "connection_name" all' the connection is removed from master_info_index.
sql/sql_repl.cc:
  sql_slave_skip_counter is moved to thd->variables to make it thread safe and fix some bugs with it
  Add connection name to relay log files.
  Added connection name to errors.
  Added some logging for multi-master if log_warnings > 1
  stop_slave():
  - Don't check if thd is set. It's guaranteed to always be set.
  change_master():
  - Check for duplicate connection names in change_master()
  - Check for wrong arguments first in file (to simplify error handling)
  - Register new connections in master_info_index
sql/sql_yacc.yy:
  Added optional connection_name to a all relevant master/slave commands
sql/strfunc.cc:
  my_global.h shoud always be included first.
sql/sys_vars.cc:
  Added variable default_master_connection
  Made variable sql_slave_skip_counter multi-source safe
sql/sys_vars.h:
  Added Sys_var_session_lexstring (needed for default_master_connection)
  Added Sys_var_multi_source_uint (needed for sql_slave_skip_counter).
This commit is contained in:
Michael Widenius 2012-09-28 02:06:56 +03:00
parent 620d14f8c3
commit 1864d9596d
41 changed files with 1833 additions and 234 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
BUILD/SETUP.sh
View File

@ -139,7 +139,7 @@ else
# C warnings
c_warnings="$warnings"
# C++ warnings
cxx_warnings="$warnings -Wno-unused-parameter"
cxx_warnings="$warnings -Wno-unused-parameter -Wno-invalid-offsetof"
# cxx_warnings="$cxx_warnings -Woverloaded-virtual -Wsign-promo"
cxx_warnings="$cxx_warnings -Wnon-virtual-dtor"
debug_extra_cflags="-O0 -g3 -gdwarf-2"

40
client/mysqltest.cc
View File

@ -253,6 +253,8 @@ static void init_re(void);
static int match_re(my_regex_t *, char *);
static void free_re(void);
static char *get_string(char **to_ptr, char **from_ptr,
struct st_command *command);
static int replace(DYNAMIC_STRING *ds_str,
const char *search_str, ulong search_len,
const char *replace_str, ulong replace_len);
@ -4580,7 +4582,8 @@ void do_wait_for_slave_to_stop(struct st_command *c __attribute__((unused)))
}
void do_sync_with_master2(struct st_command *command, long offset)
void do_sync_with_master2(struct st_command *command, long offset,
const char *connection_name)
{
MYSQL_RES *res;
MYSQL_ROW row;
@ -4591,8 +4594,9 @@ void do_sync_with_master2(struct st_command *command, long offset)
if (!master_pos.file[0])
die("Calling 'sync_with_master' without calling 'save_master_pos'");
sprintf(query_buf, "select master_pos_wait('%s', %ld, %d)",
master_pos.file, master_pos.pos + offset, timeout);
sprintf(query_buf, "select master_pos_wait('%s', %ld, %d, '%s')",
master_pos.file, master_pos.pos + offset, timeout,
connection_name);
if (mysql_query(mysql, query_buf))
die("failed in '%s': %d: %s", query_buf, mysql_errno(mysql),
@ -4652,16 +4656,32 @@ void do_sync_with_master(struct st_command *command)
long offset= 0;
char *p= command->first_argument;
const char *offset_start= p;
char *start, *buff= 0;
start= (char*) "";
if (*offset_start)
{
for (; my_isdigit(charset_info, *p); p++)
offset = offset * 10 + *p - '0';
if(*p && !my_isspace(charset_info, *p))
if (*p && !my_isspace(charset_info, *p) && *p != ',')
die("Invalid integer argument \"%s\"", offset_start);
while (*p && my_isspace(charset_info, *p))
p++;
if (*p == ',')
{
p++;
while (*p && my_isspace(charset_info, *p))
p++;
start= buff= (char*)my_malloc(strlen(p)+1,MYF(MY_WME | MY_FAE));
get_string(&buff, &p, command);
}
command->last_argument= p;
}
do_sync_with_master2(command, offset);
do_sync_with_master2(command, offset, start);
if (buff)
my_free(start);
return;
}
@ -5286,7 +5306,7 @@ void do_get_errcodes(struct st_command *command)
{
die("The sqlstate definition must start with an uppercase S");
}
else if (*p == 'E')
else if (*p == 'E' || *p == 'W')
{
/* Error name string */
@ -5295,9 +5315,9 @@ void do_get_errcodes(struct st_command *command)
to->type= ERR_ERRNO;
DBUG_PRINT("info", ("ERR_ERRNO: %d", to->code.errnum));
}
else if (*p == 'e')
else if (*p == 'e' || *p == 'w')
{
die("The error name definition must start with an uppercase E");
die("The error name definition must start with an uppercase E or W");
}
else
{
@ -5360,7 +5380,7 @@ void do_get_errcodes(struct st_command *command)
If string is a '$variable', return the value of the variable.
*/
char *get_string(char **to_ptr, char **from_ptr,
static char *get_string(char **to_ptr, char **from_ptr,
struct st_command *command)
{
char c, sep;
@ -9288,7 +9308,7 @@ int main(int argc, char **argv)
select_connection(command);
else
select_connection_name("slave");
do_sync_with_master2(command, 0);
do_sync_with_master2(command, 0, "");
break;
}
case Q_COMMENT:

2
cmake/maintainer.cmake
View File

@ -18,7 +18,7 @@ INCLUDE(CheckCCompilerFlag)
# Setup GCC (GNU C compiler) warning options.
MACRO(SET_MYSQL_MAINTAINER_GNU_C_OPTIONS)
SET(MY_MAINTAINER_WARNINGS
"-Wall -Wextra -Wunused -Wwrite-strings -Wno-strict-aliasing -DFORCE_INIT_OF_VARS")
"-Wall -Wextra -Wunused -Wwrite-strings -Wno-strict-aliasing -Wno-invalid-offsetof -DFORCE_INIT_OF_VARS")
CHECK_C_COMPILER_FLAG("-Wno-missing-field-initializers"
HAVE_NO_MISSING_FIELD_INITIALIZERS)

2
mysql-test/r/mysqltest.result
View File

@ -232,7 +232,7 @@ mysqltest: At line 2: Spurious text after `query` expression
mysqltest: At line 1: Missing argument(s) to 'error'
mysqltest: At line 1: Missing argument(s) to 'error'
mysqltest: At line 1: The sqlstate definition must start with an uppercase S
mysqltest: At line 1: The error name definition must start with an uppercase E
mysqltest: At line 1: The error name definition must start with an uppercase E or W
mysqltest: At line 1: Invalid argument to error: '9eeeee' - the errno may only consist of digits[0-9]
mysqltest: At line 1: Invalid argument to error: '1sssss' - the errno may only consist of digits[0-9]
mysqltest: At line 1: The sqlstate must be exactly 5 chars long

2
mysql-test/r/parser.result
View File

@ -441,7 +441,7 @@ select master_pos_wait();
ERROR 42000: Incorrect parameter count in the call to native function 'master_pos_wait'
select master_pos_wait(1);
ERROR 42000: Incorrect parameter count in the call to native function 'master_pos_wait'
select master_pos_wait(1, 2, 3, 4);
select master_pos_wait(1, 2, 3, 4, 5);
ERROR 42000: Incorrect parameter count in the call to native function 'master_pos_wait'
select rand(1, 2, 3);
ERROR 42000: Incorrect parameter count in the call to native function 'rand'

22
mysql-test/suite/multi_source/my.cnf Normal file
View File

@ -0,0 +1,22 @@
# cat t/multisource1.cnf
!include include/default_mysqld.cnf
!include include/default_client.cnf
[mysqld.1]
server-id=1
log-bin=master-bin
[mysqld.2]
server-id=2
log-bin=master-bin
[mysqld.3]
server-id=3
[ENV]
SERVER_MYPORT_1= @mysqld.1.port
SERVER_MYSOCK_1= @mysqld.1.socket
SERVER_MYPORT_2= @mysqld.2.port
SERVER_MYSOCK_2= @mysqld.2.socket
SERVER_MYPORT_3= @mysqld.3.port
SERVER_MYSOCK_3= @mysqld.3.socket

64
mysql-test/suite/multi_source/simple.result Normal file
View File

@ -0,0 +1,64 @@
change master 'slave1' to master_port=MYPORT_1, master_host='127.0.0.1', master_user='root';
change master 'slave2' to master_port=MYPORT_2, master_host='127.0.0.1', master_user='root';
start slave 'slave1';
start slave 'slave2';
show full slave status;
Connection_name Slave_IO_State Master_Host Master_User Master_Port Connect_Retry Master_Log_File Read_Master_Log_Pos Relay_Log_File Relay_Log_Pos Relay_Master_Log_File Slave_IO_Running Slave_SQL_Running Replicate_Do_DB Replicate_Ignore_DB Replicate_Do_Table Replicate_Ignore_Table Replicate_Wild_Do_Table Replicate_Wild_Ignore_Table Last_Errno Last_Error Skip_Counter Exec_Master_Log_Pos Relay_Log_Space Until_Condition Until_Log_File Until_Log_Pos Master_SSL_Allowed Master_SSL_CA_File Master_SSL_CA_Path Master_SSL_Cert Master_SSL_Cipher Master_SSL_Key Seconds_Behind_Master Master_SSL_Verify_Server_Cert Last_IO_Errno Last_IO_Error Last_SQL_Errno Last_SQL_Error Replicate_Ignore_Server_Ids Master_Server_Id
slave1 Waiting for master to send event 127.0.0.1 root MYPORT_1 60 master-bin.000001 286 mysqld-relay-bin-slave1.000002 572 master-bin.000001 Yes Yes 0 0 286 875 None 0 No 0 No 0 0 1
slave2 Waiting for master to send event 127.0.0.1 root MYPORT_2 60 master-bin.000001 286 mysqld-relay-bin-slave2.000002 572 master-bin.000001 Yes Yes 0 0 286 875 None 0 No 0 No 0 0 2
stop slave 'slave1';
show slave 'slave1' status;
Slave_IO_State
Master_Host 127.0.0.1
Master_User root
Master_Port MYPORT_1
Connect_Retry 60
Master_Log_File master-bin.000001
Read_Master_Log_Pos 286
Relay_Log_File mysqld-relay-bin-slave1.000002
Relay_Log_Pos 572
Relay_Master_Log_File master-bin.000001
Slave_IO_Running No
Slave_SQL_Running No
Replicate_Do_DB
Replicate_Ignore_DB
Replicate_Do_Table
Replicate_Ignore_Table
Replicate_Wild_Do_Table
Replicate_Wild_Ignore_Table
Last_Errno 0
Last_Error
Skip_Counter 0
Exec_Master_Log_Pos 286
Relay_Log_Space 875
Until_Condition None
Until_Log_File
Until_Log_Pos 0
Master_SSL_Allowed No
Master_SSL_CA_File
Master_SSL_CA_Path
Master_SSL_Cert
Master_SSL_Cipher
Master_SSL_Key
Seconds_Behind_Master NULL
Master_SSL_Verify_Server_Cert No
Last_IO_Errno 0
Last_IO_Error
Last_SQL_Errno 0
Last_SQL_Error
Replicate_Ignore_Server_Ids
Master_Server_Id 1
reset slave 'slave1';
show full slave status;
Connection_name Slave_IO_State Master_Host Master_User Master_Port Connect_Retry Master_Log_File Read_Master_Log_Pos Relay_Log_File Relay_Log_Pos Relay_Master_Log_File Slave_IO_Running Slave_SQL_Running Replicate_Do_DB Replicate_Ignore_DB Replicate_Do_Table Replicate_Ignore_Table Replicate_Wild_Do_Table Replicate_Wild_Ignore_Table Last_Errno Last_Error Skip_Counter Exec_Master_Log_Pos Relay_Log_Space Until_Condition Until_Log_File Until_Log_Pos Master_SSL_Allowed Master_SSL_CA_File Master_SSL_CA_Path Master_SSL_Cert Master_SSL_Cipher Master_SSL_Key Seconds_Behind_Master Master_SSL_Verify_Server_Cert Last_IO_Errno Last_IO_Error Last_SQL_Errno Last_SQL_Error Replicate_Ignore_Server_Ids Master_Server_Id
slave1 127.0.0.1 root MYPORT_1 60 4 mysqld-relay-bin-slave1.000001 4 No No 0 0 0 265 None 0 No NULL No 0 0 1
slave2 Waiting for master to send event 127.0.0.1 root MYPORT_2 60 master-bin.000001 286 mysqld-relay-bin-slave2.000002 572 master-bin.000001 Yes Yes 0 0 286 875 None 0 No 0 No 0 0 2
reset slave 'slave1' all;
show full slave status;
Connection_name Slave_IO_State Master_Host Master_User Master_Port Connect_Retry Master_Log_File Read_Master_Log_Pos Relay_Log_File Relay_Log_Pos Relay_Master_Log_File Slave_IO_Running Slave_SQL_Running Replicate_Do_DB Replicate_Ignore_DB Replicate_Do_Table Replicate_Ignore_Table Replicate_Wild_Do_Table Replicate_Wild_Ignore_Table Last_Errno Last_Error Skip_Counter Exec_Master_Log_Pos Relay_Log_Space Until_Condition Until_Log_File Until_Log_Pos Master_SSL_Allowed Master_SSL_CA_File Master_SSL_CA_Path Master_SSL_Cert Master_SSL_Cipher Master_SSL_Key Seconds_Behind_Master Master_SSL_Verify_Server_Cert Last_IO_Errno Last_IO_Error Last_SQL_Errno Last_SQL_Error Replicate_Ignore_Server_Ids Master_Server_Id
slave2 Waiting for master to send event 127.0.0.1 root MYPORT_2 60 master-bin.000001 286 mysqld-relay-bin-slave2.000002 572 master-bin.000001 Yes Yes 0 0 286 875 None 0 No 0 No 0 0 2
stop slave 'slave2';
show full slave status;
Connection_name Slave_IO_State Master_Host Master_User Master_Port Connect_Retry Master_Log_File Read_Master_Log_Pos Relay_Log_File Relay_Log_Pos Relay_Master_Log_File Slave_IO_Running Slave_SQL_Running Replicate_Do_DB Replicate_Ignore_DB Replicate_Do_Table Replicate_Ignore_Table Replicate_Wild_Do_Table Replicate_Wild_Ignore_Table Last_Errno Last_Error Skip_Counter Exec_Master_Log_Pos Relay_Log_Space Until_Condition Until_Log_File Until_Log_Pos Master_SSL_Allowed Master_SSL_CA_File Master_SSL_CA_Path Master_SSL_Cert Master_SSL_Cipher Master_SSL_Key Seconds_Behind_Master Master_SSL_Verify_Server_Cert Last_IO_Errno Last_IO_Error Last_SQL_Errno Last_SQL_Error Replicate_Ignore_Server_Ids Master_Server_Id
slave2 127.0.0.1 root MYPORT_2 60 master-bin.000001 286 mysqld-relay-bin-slave2.000002 572 master-bin.000001 No No 0 0 286 875 None 0 No NULL No 0 0 2
reset slave 'slave2' all;

38
mysql-test/suite/multi_source/simple.test Normal file
View File

@ -0,0 +1,38 @@
#
# Simple multi-master test
#
--connect (slave,127.0.0.1,root,,,$SERVER_MYPORT_3)
--replace_result $SERVER_MYPORT_1 MYPORT_1
eval change master 'slave1' to master_port=$SERVER_MYPORT_1, master_host='127.0.0.1', master_user='root';
--replace_result $SERVER_MYPORT_2 MYPORT_2
eval change master 'slave2' to master_port=$SERVER_MYPORT_2, master_host='127.0.0.1', master_user='root';
start slave 'slave1';
start slave 'slave2';
--sleep 5
--replace_result $SERVER_MYPORT_1 MYPORT_1 $SERVER_MYPORT_2 MYPORT_2
show full slave status;
stop slave 'slave1';
--replace_result $SERVER_MYPORT_1 MYPORT_1 $SERVER_MYPORT_2 MYPORT_2
query_vertical show slave 'slave1' status;
reset slave 'slave1';
--replace_result $SERVER_MYPORT_1 MYPORT_1 $SERVER_MYPORT_2 MYPORT_2
show full slave status;
reset slave 'slave1' all;
--replace_result $SERVER_MYPORT_1 MYPORT_1 $SERVER_MYPORT_2 MYPORT_2
show full slave status;
stop slave 'slave2';
--replace_result $SERVER_MYPORT_1 MYPORT_1 $SERVER_MYPORT_2 MYPORT_2
show full slave status;
#
# clean up
#
reset slave 'slave2' all;

81
mysql-test/suite/multi_source/skip_counter.result Normal file
View File

@ -0,0 +1,81 @@
connect master1,127.0.0.1,root,,,$SERVER_MYPORT_1;
drop database if exists db;
create database db;
create table db.t1 (i int) engine=MyISAM;
connect master2,127.0.0.1,root,,,$SERVER_MYPORT_2;
drop database if exists db;
create database db;
create table db.t2 (i int) engine=MyISAM;
connect slave,127.0.0.1,root,,,$SERVER_MYPORT_3;
change master 'master1' to
master_port=MYPORT_1,
master_host='127.0.0.1',
master_user='root';
start slave 'master1';
set default_master_connection = 'master1';
include/wait_for_slave_to_start.inc
set default_master_connection = 'master2';
change master 'master2' to
master_port=MYPORT_2,
master_host='127.0.0.1',
master_user='root';
set global sql_slave_skip_counter = 2;
select @@global.sql_slave_skip_counter;
@@global.sql_slave_skip_counter
2
select @@session.sql_slave_skip_counter;
@@session.sql_slave_skip_counter
2
set session sql_slave_skip_counter = 3;
select @@global.sql_slave_skip_counter;
@@global.sql_slave_skip_counter
3
select @@session.sql_slave_skip_counter;
@@session.sql_slave_skip_counter
3
set global sql_slave_skip_counter= default;
select @@global.sql_slave_skip_counter;
@@global.sql_slave_skip_counter
0
select @@session.sql_slave_skip_counter;
@@session.sql_slave_skip_counter
0
set global sql_slave_skip_counter= 3;
set default_master_connection = 'master1';
select @@session.sql_slave_skip_counter;
@@session.sql_slave_skip_counter
0
set default_master_connection = 'qqq';
select @@session.sql_slave_skip_counter;
@@session.sql_slave_skip_counter
0
Warnings:
Warning 1617 There is no master connection 'qqq'
set default_master_connection = 'master2';
select @@session.sql_slave_skip_counter;
@@session.sql_slave_skip_counter
3
start slave 'master2';
include/wait_for_slave_to_start.inc
set default_master_connection = '';
connection master2;
connection slave;
show tables in db;
Tables_in_db
t1
t2
drop database db;
set default_master_connection = 'master1';
stop slave;
include/wait_for_slave_to_stop.inc
set default_master_connection = 'master2';
stop slave;
include/wait_for_slave_to_stop.inc
set global sql_slave_skip_counter = 0;
disconnect slave;
connection master1;
drop database db;
disconnect master1;
connection master2;
drop database db;
disconnect master2;

126
mysql-test/suite/multi_source/skip_counter.test Normal file
View File

@ -0,0 +1,126 @@
#
# Test of sql_slave_skip_counter
#
--enable_connect_log
# Create a schema and a table i
# on the 1st master
--connect (master1,127.0.0.1,root,,,$SERVER_MYPORT_1)
--disable_warnings
drop database if exists db;
--enable_warnings
create database db;
create table db.t1 (i int) engine=MyISAM;
--save_master_pos
# Create the same schema and another table
# on the 2nd master
--connect (master2,127.0.0.1,root,,,$SERVER_MYPORT_2)
--disable_warnings
drop database if exists db;
--enable_warnings
create database db;
create table db.t2 (i int) engine=MyISAM;
--connect (slave,127.0.0.1,root,,,$SERVER_MYPORT_3)
--disable_connect_log
# Start replication from the first master
--replace_result $SERVER_MYPORT_1 MYPORT_1
eval change master 'master1' to
master_port=$SERVER_MYPORT_1,
master_host='127.0.0.1',
master_user='root';
start slave 'master1';
set default_master_connection = 'master1';
--source include/wait_for_slave_to_start.inc
--sync_with_master 0,'master1'
# Start replication from the second master
set default_master_connection = 'master2';
--replace_result $SERVER_MYPORT_2 MYPORT_2
eval change master 'master2' to
master_port=$SERVER_MYPORT_2,
master_host='127.0.0.1',
master_user='root';
# the schema creation will be replicated from the 1st master,
# so we want to skip it in the second replication connection.
# Normally it should have been 2 events, but
# currently Binlog_checkpoint also counts. Maybe we'll need
# to modify the test later
--let $skip_counter_saved = `select @@global.sql_slave_skip_counter`
set global sql_slave_skip_counter = 2;
select @@global.sql_slave_skip_counter;
select @@session.sql_slave_skip_counter;
set session sql_slave_skip_counter = 3;
select @@global.sql_slave_skip_counter;
select @@session.sql_slave_skip_counter;
set global sql_slave_skip_counter= default;
select @@global.sql_slave_skip_counter;
select @@session.sql_slave_skip_counter;
set global sql_slave_skip_counter= 3;
set default_master_connection = 'master1';
select @@session.sql_slave_skip_counter;
set default_master_connection = 'qqq';
select @@session.sql_slave_skip_counter;
set default_master_connection = 'master2';
select @@session.sql_slave_skip_counter;
start slave 'master2';
--source include/wait_for_slave_to_start.inc
set default_master_connection = '';
--enable_connect_log
--connection master2
--save_master_pos
--connection slave
--disable_connect_log
--sync_with_master 0,'master2'
# If the skip_counter worked as expected, we should
# get here (replication shouldn't have broken)
# and should see both tables here
# (drop database which came from master2 shoudn't have been executed
# so t1 should still exist)
show tables in db;
# Cleanup
drop database db;
set default_master_connection = 'master1';
stop slave;
--source include/wait_for_slave_to_stop.inc
set default_master_connection = 'master2';
stop slave;
--source include/wait_for_slave_to_stop.inc
--eval set global sql_slave_skip_counter = $skip_counter_saved
--enable_connect_log
--disconnect slave
--connection master1
drop database db;
--disconnect master1
--connection master2
drop database db;
--disconnect master2

89
mysql-test/suite/multi_source/syntax.result Normal file
View File

@ -0,0 +1,89 @@
include/master-slave.inc
[connection master]
show slave status;
Slave_IO_State Master_Host Master_User Master_Port Connect_Retry Master_Log_File Read_Master_Log_Pos Relay_Log_File Relay_Log_Pos Relay_Master_Log_File Slave_IO_Running Slave_SQL_Running Replicate_Do_DB Replicate_Ignore_DB Replicate_Do_Table Replicate_Ignore_Table Replicate_Wild_Do_Table Replicate_Wild_Ignore_Table Last_Errno Last_Error Skip_Counter Exec_Master_Log_Pos Relay_Log_Space Until_Condition Until_Log_File Until_Log_Pos Master_SSL_Allowed Master_SSL_CA_File Master_SSL_CA_Path Master_SSL_Cert Master_SSL_Cipher Master_SSL_Key Seconds_Behind_Master Master_SSL_Verify_Server_Cert Last_IO_Errno Last_IO_Error Last_SQL_Errno Last_SQL_Error Replicate_Ignore_Server_Ids Master_Server_Id
show slave '' status;
Slave_IO_State Master_Host Master_User Master_Port Connect_Retry Master_Log_File Read_Master_Log_Pos Relay_Log_File Relay_Log_Pos Relay_Master_Log_File Slave_IO_Running Slave_SQL_Running Replicate_Do_DB Replicate_Ignore_DB Replicate_Do_Table Replicate_Ignore_Table Replicate_Wild_Do_Table Replicate_Wild_Ignore_Table Last_Errno Last_Error Skip_Counter Exec_Master_Log_Pos Relay_Log_Space Until_Condition Until_Log_File Until_Log_Pos Master_SSL_Allowed Master_SSL_CA_File Master_SSL_CA_Path Master_SSL_Cert Master_SSL_Cipher Master_SSL_Key Seconds_Behind_Master Master_SSL_Verify_Server_Cert Last_IO_Errno Last_IO_Error Last_SQL_Errno Last_SQL_Error Replicate_Ignore_Server_Ids Master_Server_Id
show full slave status;
Connection_name Slave_IO_State Master_Host Master_User Master_Port Connect_Retry Master_Log_File Read_Master_Log_Pos Relay_Log_File Relay_Log_Pos Relay_Master_Log_File Slave_IO_Running Slave_SQL_Running Replicate_Do_DB Replicate_Ignore_DB Replicate_Do_Table Replicate_Ignore_Table Replicate_Wild_Do_Table Replicate_Wild_Ignore_Table Last_Errno Last_Error Skip_Counter Exec_Master_Log_Pos Relay_Log_Space Until_Condition Until_Log_File Until_Log_Pos Master_SSL_Allowed Master_SSL_CA_File Master_SSL_CA_Path Master_SSL_Cert Master_SSL_Cipher Master_SSL_Key Seconds_Behind_Master Master_SSL_Verify_Server_Cert Last_IO_Errno Last_IO_Error Last_SQL_Errno Last_SQL_Error Replicate_Ignore_Server_Ids Master_Server_Id
#
# Check error handling
#
show slave 'qqq' status;
ERROR HY000: There is no master connection 'qqq'
show slave 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc' status;
ERROR HY000: There is no master connection 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc'
show slave 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc' status;
ERROR HY000: Incorrect arguments to MASTER_CONNECTION_NAME
change master 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc' to master_host='dummy';
ERROR HY000: Incorrect arguments to MASTER_CONNECTION_NAME
start slave 'qqq';
ERROR HY000: There is no master connection 'qqq'
stop slave 'qqq';
ERROR HY000: There is no master connection 'qqq'
slave 'qqq' start;
ERROR HY000: There is no master connection 'qqq'
slave 'qqq' stop;
ERROR HY000: There is no master connection 'qqq'
flush slave 'qqq';
ERROR HY000: There is no master connection 'qqq'
reset slave 'qqq';
ERROR HY000: There is no master connection 'qqq'
select master_pos_wait('master-bin.999999',0,2,'qqq');
master_pos_wait('master-bin.999999',0,2,'qqq')
NULL
Warnings:
Warning 1617 There is no master connection 'qqq'
select master_pos_wait('master-bin.999999',0,2,'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc');
master_pos_wait('master-bin.999999',0,2,'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc')
NULL
Warnings:
Warning 1210 Incorrect arguments to MASTER_CONNECTION_NAME
#
# checking usage of default_master_connection;
#
select @@default_master_connection;
@@default_master_connection
select @@global.default_master_connection;
ERROR HY000: Variable 'default_master_connection' is a SESSION variable
set @@global.default_master_connection='qqq';
ERROR HY000: Variable 'default_master_connection' is a SESSION variable and can't be used with SET GLOBAL
set @@default_master_connection='aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc';
ERROR 42000: Variable 'default_master_connection' can't be set to the value of 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc'
select @@default_master_connection;
@@default_master_connection
set @@default_master_connection='qqq';
select @@default_master_connection;
@@default_master_connection
qqq
show variables like "default_master_connection";
Variable_name Value
default_master_connection qqq
show slave status;
ERROR HY000: There is no master connection 'qqq'
select master_pos_wait('master-bin.999999',0,2);
master_pos_wait('master-bin.999999',0,2)
NULL
Warnings:
Warning 1617 There is no master connection 'qqq'
set @@default_master_connection='';
select master_pos_wait('master-bin.999999',0,2);
master_pos_wait('master-bin.999999',0,2)
-1
set @@default_master_connection='';
#
# checking variables
#
show status like "Slave_running";
Variable_name Value
Slave_running ON
set @@default_master_connection='qqq';
show status like "Slave_running";
Variable_name Value
Slave_running OFF
Warnings:
Warning 1617 There is no master connection 'qqq'
set @@default_master_connection='';
include/rpl_end.inc

77
mysql-test/suite/multi_source/syntax.test Normal file
View File

@ -0,0 +1,77 @@
# Test multi master syntax
source include/master-slave.inc;
# Check syntax of multi source replication
show slave status;
show slave '' status;
show full slave status;
--echo #
--echo # Check error handling
--echo #
--error WARN_NO_MASTER_INFO
show slave 'qqq' status;
--error WARN_NO_MASTER_INFO
show slave 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc' status;
--error ER_WRONG_ARGUMENTS
show slave 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc' status;
--error ER_WRONG_ARGUMENTS
change master 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc' to master_host='dummy';
--error WARN_NO_MASTER_INFO
start slave 'qqq';
--error WARN_NO_MASTER_INFO
stop slave 'qqq';
--error WARN_NO_MASTER_INFO
slave 'qqq' start;
--error WARN_NO_MASTER_INFO
slave 'qqq' stop;
--error WARN_NO_MASTER_INFO
flush slave 'qqq';
--error WARN_NO_MASTER_INFO
reset slave 'qqq';
select master_pos_wait('master-bin.999999',0,2,'qqq');
select master_pos_wait('master-bin.999999',0,2,'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc');
save_master_pos;
connection slave;
sync_with_master 0,'';
sync_with_master 0 ,'';
sync_with_master 0, '';
--echo #
--echo # checking usage of default_master_connection;
--echo #
select @@default_master_connection;
--error 1238
select @@global.default_master_connection;
--error 1228
set @@global.default_master_connection='qqq';
--error 1231
set @@default_master_connection='aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc';
select @@default_master_connection;
set @@default_master_connection='qqq';
select @@default_master_connection;
show variables like "default_master_connection";
--error WARN_NO_MASTER_INFO
show slave status;
select master_pos_wait('master-bin.999999',0,2);
set @@default_master_connection='';
select master_pos_wait('master-bin.999999',0,2);
set @@default_master_connection='';
--echo #
--echo # checking variables
--echo #
show status like "Slave_running";
set @@default_master_connection='qqq';
show status like "Slave_running";
set @@default_master_connection='';
--source include/rpl_end.inc

4
mysql-test/suite/rpl/r/rpl_rotate_logs.result
View File

@ -2,9 +2,9 @@ CALL mtr.add_suppression("Unsafe statement written to the binary log using state
start slave;
Got one of the listed errors
start slave;
ERROR HY000: Could not initialize master info structure; more error messages can be found in the MariaDB error log
ERROR HY000: Could not initialize master info structure for ''; more error messages can be found in the MariaDB error log
change master to master_host='127.0.0.1',master_port=MASTER_PORT, master_user='root';
ERROR HY000: Could not initialize master info structure; more error messages can be found in the MariaDB error log
ERROR HY000: Could not initialize master info structure for ''; more error messages can be found in the MariaDB error log
reset slave;
change master to master_host='127.0.0.1',master_port=MASTER_PORT, master_user='root';
reset master;

2
mysql-test/t/parser.test
View File

@ -553,7 +553,7 @@ select master_pos_wait();
-- error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
select master_pos_wait(1);
-- error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
select master_pos_wait(1, 2, 3, 4);
select master_pos_wait(1, 2, 3, 4, 5);
-- error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
select rand(1, 2, 3);

1
sql/event_scheduler.cc
View File

@ -190,7 +190,6 @@ pre_init_event_thread(THD* thd)
my_net_init(&thd->net, NULL);
thd->security_ctx->set_user((char*)"event_scheduler");
thd->net.read_timeout= slave_net_timeout;
thd->slave_thread= 0;
thd->variables.option_bits|= OPTION_AUTO_IS_NULL;
thd->client_capabilities|= CLIENT_MULTI_RESULTS;
mysql_mutex_lock(&LOCK_thread_count);

23
sql/item_create.cc
View File

@ -4393,27 +4393,36 @@ Create_func_master_pos_wait::create_native(THD *thd, LEX_STRING name,
if (item_list != NULL)
arg_count= item_list->elements;
if (arg_count < 2 || arg_count >4)
{
my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name.str);
return func;
}
thd->lex->safe_to_cache_query= 0;
Item *param_1= item_list->pop();
Item *param_2= item_list->pop();
switch (arg_count) {
case 2:
{
Item *param_1= item_list->pop();
Item *param_2= item_list->pop();
func= new (thd->mem_root) Item_master_pos_wait(param_1, param_2);
thd->lex->safe_to_cache_query= 0;
break;
}
case 3:
{
Item *param_1= item_list->pop();
Item *param_2= item_list->pop();
Item *param_3= item_list->pop();
func= new (thd->mem_root) Item_master_pos_wait(param_1, param_2, param_3);
thd->lex->safe_to_cache_query= 0;
break;
}
default:
case 4:
{
my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name.str);
Item *param_3= item_list->pop();
Item *param_4= item_list->pop();
func= new (thd->mem_root) Item_master_pos_wait(param_1, param_2, param_3,
param_4);
thd->lex->safe_to_cache_query= 0;
break;
}
}

34
sql/item_func.cc
View File

@ -3866,13 +3866,45 @@ longlong Item_master_pos_wait::val_int()
#ifdef HAVE_REPLICATION
longlong pos = (ulong)args[1]->val_int();
longlong timeout = (arg_count==3) ? args[2]->val_int() : 0 ;
if ((event_count = active_mi->rli.wait_for_pos(thd, log_name, pos, timeout)) == -2)
String connection_name_buff;
LEX_STRING connection_name;
Master_info *mi;
if (arg_count == 4)
{
String *con;
if (!(con= args[3]->val_str(&connection_name_buff)))
goto err;
connection_name.str= (char*) con->ptr();
connection_name.length= con->length();
if (check_master_connection_name(&connection_name))
{
my_error(ER_WRONG_ARGUMENTS, MYF(ME_JUST_WARNING),
"MASTER_CONNECTION_NAME");
goto err;
}
}
else
connection_name= thd->variables.default_master_connection;
if (!(mi= master_info_index->get_master_info(&connection_name,
MYSQL_ERROR::WARN_LEVEL_WARN)))
goto err;
if ((event_count = mi->rli.wait_for_pos(thd, log_name, pos, timeout)) == -2)
{
null_value = 1;
event_count=0;
}
#endif
return event_count;
#ifdef HAVE_REPLICATION
err:
{
null_value = 1;
return 0;
}
#endif
}

3
sql/item_func.h
View File

@ -496,6 +496,8 @@ public:
{ collation.set_numeric(); fix_char_length(21); }
Item_int_func(Item *a,Item *b,Item *c) :Item_func(a,b,c)
{ collation.set_numeric(); fix_char_length(21); }
Item_int_func(Item *a,Item *b,Item *c, Item *d) :Item_func(a,b,c,d)
{ collation.set_numeric(); fix_char_length(21); }
Item_int_func(List<Item> &list) :Item_func(list)
{ collation.set_numeric(); fix_char_length(21); }
Item_int_func(THD *thd, Item_int_func *item) :Item_func(thd, item)
@ -1522,6 +1524,7 @@ class Item_master_pos_wait :public Item_int_func
public:
Item_master_pos_wait(Item *a,Item *b) :Item_int_func(a,b) {}
Item_master_pos_wait(Item *a,Item *b,Item *c) :Item_int_func(a,b,c) {}
Item_master_pos_wait(Item *a,Item *b, Item *c, Item *d) :Item_int_func(a,b,c,d) {}
longlong val_int();
const char *func_name() const { return "master_pos_wait"; }
void fix_length_and_dec() { max_length=21; maybe_null=1;}

21
sql/log.cc
View File

@ -40,6 +40,7 @@
#include "rpl_rli.h"
#include "sql_audit.h"
#include "log_slow.h"
#include "mysqld.h"
#include <my_dir.h>
#include <stdarg.h>
@ -6980,16 +6981,33 @@ static void print_buffer_to_file(enum loglevel level, const char *buffer,
time_t skr;
struct tm tm_tmp;
struct tm *start;
THD *thd;
int tag_length= 0;
char tag[NAME_LEN];
DBUG_ENTER("print_buffer_to_file");
DBUG_PRINT("enter",("buffer: %s", buffer));
if (mysqld_server_initialized && (thd= current_thd))
{
if (thd->connection_name.length)
{
/*
Add tag for slaves so that the user can see from which connection
the error originates.
*/
tag_length= my_snprintf(tag, sizeof(tag), ER(ER_MASTER_LOG_PREFIX),
(int) thd->connection_name.length,
thd->connection_name.str);
}
}
mysql_mutex_lock(&LOCK_error_log);
skr= my_time(0);
localtime_r(&skr, &tm_tmp);
start=&tm_tmp;
fprintf(stderr, "%02d%02d%02d %2d:%02d:%02d [%s] %.*s\n",
fprintf(stderr, "%02d%02d%02d %2d:%02d:%02d [%s] %.*s%.*s\n",
start->tm_year % 100,
start->tm_mon+1,
start->tm_mday,
@ -6998,6 +7016,7 @@ static void print_buffer_to_file(enum loglevel level, const char *buffer,
start->tm_sec,
(level == ERROR_LEVEL ? "ERROR" : level == WARNING_LEVEL ?
"Warning" : "Note"),
tag_length, tag,
(int) length, buffer);
fflush(stderr);

68
sql/mysqld.cc
View File

@ -673,8 +673,7 @@ pthread_attr_t connection_attrib;
mysql_mutex_t LOCK_server_started;
mysql_cond_t COND_server_started;
int mysqld_server_started= 0;
int mysqld_server_started=0, mysqld_server_initialized= 0;
File_parser_dummy_hook file_parser_dummy_hook;
/* replication parameters, if master_host is not NULL, we are a slave */
@ -1761,7 +1760,12 @@ void clean_up(bool print_message)
if (cleanup_done++)
return; /* purecov: inspected */
close_active_mi();
#ifdef HAVE_REPLICATION
// We must call end_slave() as clean_up may have been called during startup
end_slave();
if (use_slave_mask)
bitmap_free(&slave_error_mask);
#endif
stop_handle_manager();
release_ddl_log();
@ -1776,10 +1780,6 @@ void clean_up(bool print_message)
injector::free_instance();
mysql_bin_log.cleanup();
#ifdef HAVE_REPLICATION
if (use_slave_mask)
bitmap_free(&slave_error_mask);
#endif
my_tz_free();
my_dboptions_cache_free();
#ifndef NO_EMBEDDED_ACCESS_CHECKS
@ -4626,6 +4626,7 @@ int mysqld_main(int argc, char **argv)
}
#endif
mysqld_server_started= mysqld_server_initialized= 0;
orig_argc= argc;
orig_argv= argv;
my_getopt_use_args_separator= TRUE;
@ -4890,16 +4891,6 @@ int mysqld_main(int argc, char **argv)
opt_skip_slave_start= 1;
binlog_unsafe_map_init();
/*
init_slave() must be called after the thread keys are created.
Some parts of the code (e.g. SHOW STATUS LIKE 'slave_running' and other
places) assume that active_mi != 0, so let's fail if it's 0 (out of
memory); a message has already been printed.
*/
if (init_slave() && !active_mi)
{
unireg_abort(1);
}
#ifdef WITH_PERFSCHEMA_STORAGE_ENGINE
initialize_performance_schema_acl(opt_bootstrap);
@ -4917,9 +4908,17 @@ int mysqld_main(int argc, char **argv)
execute_ddl_log_recovery();
/*
We must have LOCK_open before LOCK_global_system_variables because
LOCK_open is hold while sql_plugin.c::intern_sys_var_ptr() is called.
*/
mysql_mutex_record_order(&LOCK_open, &LOCK_global_system_variables);
if (Events::init(opt_noacl || opt_bootstrap))
unireg_abort(1);
mysqld_server_initialized= 1;
if (opt_bootstrap)
{
select_thread_in_use= 0; // Allow 'kill' to work
@ -4932,21 +4931,27 @@ int mysqld_main(int argc, char **argv)
exit(0);
}
}
create_shutdown_thread();
start_handle_manager();
/*
init_slave() must be called after the thread keys are created.
Some parts of the code (e.g. SHOW STATUS LIKE 'slave_running' and other
places) assume that active_mi != 0, so let's fail if it's 0 (out of
memory); a message has already been printed.
*/
if (init_slave() && !active_mi)
{
unireg_abort(1);
}
if (opt_init_file && *opt_init_file)
{
if (read_init_file(opt_init_file))
unireg_abort(1);
}
/*
We must have LOCK_open before LOCK_global_system_variables because
LOCK_open is hold while sql_plugin.c::intern_sys_var_ptr() is called.
*/
mysql_mutex_record_order(&LOCK_open, &LOCK_global_system_variables);
create_shutdown_thread();
start_handle_manager();
sql_print_information(ER_DEFAULT(ER_STARTUP),my_progname,server_version,
((unix_sock == INVALID_SOCKET) ? (char*) ""
: mysqld_unix_port),
@ -6524,12 +6529,17 @@ static int show_rpl_status(THD *thd, SHOW_VAR *var, char *buff)
static int show_slave_running(THD *thd, SHOW_VAR *var, char *buff)
{
Master_info *mi;
var->type= SHOW_MY_BOOL;
mysql_mutex_lock(&LOCK_active_mi);
var->value= buff;
*((my_bool *)buff)= (my_bool) (active_mi &&
active_mi->slave_running == MYSQL_SLAVE_RUN_CONNECT &&
active_mi->rli.slave_running);
mi= master_info_index->
get_master_info(&thd->variables.default_master_connection,
MYSQL_ERROR::WARN_LEVEL_WARN);
*((my_bool *)buff)= (my_bool) (mi &&
mi->slave_running ==
MYSQL_SLAVE_RUN_CONNECT &&
mi->rli.slave_running);
mysql_mutex_unlock(&LOCK_active_mi);
return 0;
}

2
sql/mysqld.h
View File

@ -198,7 +198,7 @@ extern handlerton *myisam_hton;
extern handlerton *heap_hton;
extern const char *load_default_groups[];
extern struct my_option my_long_options[];
extern int mysqld_server_started;
extern int mysqld_server_started, mysqld_server_initialized;
extern "C" MYSQL_PLUGIN_IMPORT int orig_argc;
extern "C" MYSQL_PLUGIN_IMPORT char **orig_argv;
extern pthread_attr_t connection_attrib;

472
sql/rpl_mi.cc
View File

@ -20,6 +20,7 @@
#include "unireg.h" // REQUIRED by other includes
#include "rpl_mi.h"
#include "slave.h" // SLAVE_MAX_HEARTBEAT_PERIOD
#include "strfunc.h"
#ifdef HAVE_REPLICATION
@ -27,7 +28,8 @@
static void init_master_log_pos(Master_info* mi);
Master_info::Master_info(bool is_slave_recovery)
Master_info::Master_info(LEX_STRING *connection_name_arg,
bool is_slave_recovery)
:Slave_reporting_capability("I/O"),
ssl(0), ssl_verify_server_cert(1), fd(-1), io_thd(0),
rli(is_slave_recovery), port(MYSQL_PORT),
@ -40,6 +42,21 @@ Master_info::Master_info(bool is_slave_recovery)
ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0;
ssl_cipher[0]= 0; ssl_key[0]= 0;
/* Store connection name and lower case connection name */
connection_name.length= cmp_connection_name.length=
connection_name_arg->length;
if ((connection_name.str= (char*) my_malloc(connection_name_arg->length*2+2,
MYF(MY_WME))))
{
cmp_connection_name.str= (connection_name.str +
connection_name_arg->length+1);
strmake(connection_name.str, connection_name_arg->str,
connection_name.length);
memcpy(cmp_connection_name.str, connection_name_arg->str,
connection_name.length+1);
my_casedn_str(system_charset_info, cmp_connection_name.str);
}
my_init_dynamic_array(&ignore_server_ids, sizeof(::server_id), 16, 16);
bzero((char*) &file, sizeof(file));
mysql_mutex_init(key_master_info_run_lock, &run_lock, MY_MUTEX_INIT_FAST);
@ -55,6 +72,7 @@ Master_info::Master_info(bool is_slave_recovery)
Master_info::~Master_info()
{
my_free(connection_name.str);
delete_dynamic(&ignore_server_ids);
mysql_mutex_destroy(&run_lock);
mysql_mutex_destroy(&data_lock);
@ -407,7 +425,7 @@ file '%s')", fname);
mi->master_log_name,
(ulong) mi->master_log_pos));
mi->rli.mi = mi;
mi->rli.mi= mi;
if (init_relay_log_info(&mi->rli, slave_info_fname))
goto err;
@ -560,5 +578,455 @@ void end_master_info(Master_info* mi)
DBUG_VOID_RETURN;
}
/* Multi-Master By P.Linux */
uchar *get_key_master_info(Master_info *mi, size_t *length,
my_bool not_used __attribute__((unused)))
{
/* Return lower case name */
*length= mi->cmp_connection_name.length;
return (uchar*) mi->cmp_connection_name.str;
}
void free_key_master_info(Master_info *mi)
{
DBUG_ENTER("free_key_master_info");
terminate_slave_threads(mi,SLAVE_FORCE_ALL);
end_master_info(mi);
delete mi;
DBUG_VOID_RETURN;
}
/**
Check if connection name for master_info is valid.
It's valid if it's a valid system name, is less than
MAX_CONNECTION_NAME.
@return
0 ok
1 error
*/
bool check_master_connection_name(LEX_STRING *name)
{
if (name->length >= MAX_CONNECTION_NAME)
return 1;
return 0;
}
/**
Create a log file with a signed suffix.
@param
res_file_name Store result here
length Length of res_file_name buffer
info_file Original file name (prefix)
separator Separator character
suffix Suffix
@note
If suffix is an empty string, then we don't add any suffix.
This is to allow one to use this function also to generate old
file names without a prefix.
*/
void create_signed_file_name(char *res_file_name, uint length,
const char *info_file,
char separator, LEX_STRING *suffix)
{
char buff[MAX_CONNECTION_NAME+1], res[MAX_CONNECTION_NAME+1], *p;
p= strmake(res_file_name, info_file, length);
if (suffix->length != 0 && p != info_file + length)
{
uint errors;
size_t res_length;
*p++= separator;
/* Create null terminated string */
strmake(buff, suffix->str, suffix->length);
/* Convert to lower case */
my_casedn_str(system_charset_info, buff);
/* Convert to characters usable in a file name */
res_length= strconvert(system_charset_info, buff,
&my_charset_filename, res, sizeof(res), &errors);
strmake(p, res, min(length - (p - res_file_name), res_length));
}
}
Master_info_index::Master_info_index()
{
index_file_name[0] = 0;
bzero((char*) &index_file, sizeof(index_file));
}
Master_info_index::~Master_info_index()
{
/* This will close connection for all objects in the cache */
my_hash_free(&master_info_hash);
end_io_cache(&index_file);
if (index_file.file > 0)
my_close(index_file.file, MYF(MY_WME));
}
/* Load All Master_info from master.info.index File
* RETURN:
* 0 - All Success
* 1 - All Fail
* 2 - Some Success, Some Fail
*/
bool Master_info_index::init_all_master_info()
{
int thread_mask;
int err_num= 0, succ_num= 0; // The number of success read Master_info
char sign[MAX_CONNECTION_NAME];
File index_file_nr;
size_t filename_length, dir_length;
DBUG_ENTER("init_all_master_info");
/*
Create the Master_info index file by prepending 'multi-' before
the master_info_file file name.
*/
fn_format(index_file_name, master_info_file, mysql_data_home,
"", MY_UNPACK_FILENAME);
filename_length= strlen(index_file_name) + 1; /* Count 0 byte */
dir_length= dirname_length(index_file_name);
bmove_upp((uchar*) index_file_name + filename_length + 6,
(uchar*) index_file_name + filename_length,
filename_length - dir_length);
memcpy(index_file_name + dir_length, "multi-", 6);
if ((index_file_nr= my_open(index_file_name,
O_RDWR | O_CREAT | O_BINARY ,
MYF(MY_WME | ME_NOREFRESH))) < 0 ||
my_sync(index_file_nr, MYF(MY_WME)) ||
init_io_cache(&index_file, index_file_nr,
IO_SIZE, READ_CACHE,
my_seek(index_file_nr,0L,MY_SEEK_END,MYF(0)),
0, MYF(MY_WME | MY_WAIT_IF_FULL)))
{
if (index_file_nr >= 0)
my_close(index_file_nr,MYF(0));
sql_print_error("Creation of Master_info index file '%s' failed",
index_file_name);
DBUG_RETURN(1);
}
/* Initialize Master_info Hash Table */
if (my_hash_init(&master_info_hash, system_charset_info,
MAX_REPLICATION_THREAD, 0, 0,
(my_hash_get_key) get_key_master_info,
(my_hash_free_key)free_key_master_info, HASH_UNIQUE))
{
sql_print_error("Initializing Master_info hash table failed");
DBUG_RETURN(1);
}
reinit_io_cache(&index_file, READ_CACHE, 0L,0,0);
while (!init_strvar_from_file(sign, sizeof(sign),
&index_file, NULL))
{
LEX_STRING connection_name;
Master_info *mi;
char buf_master_info_file[FN_REFLEN];
char buf_relay_log_info_file[FN_REFLEN];
connection_name.str= sign;
connection_name.length= strlen(sign);
if (!(mi= new Master_info(&connection_name, relay_log_recovery)) ||
mi->error())
{
delete mi;
DBUG_RETURN(1);
}
lock_slave_threads(mi);
init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
create_signed_file_name(buf_master_info_file, sizeof(buf_master_info_file),
master_info_file, '.', &connection_name);
create_signed_file_name(buf_relay_log_info_file,
sizeof(buf_relay_log_info_file),
relay_log_info_file, '.', &connection_name);
if (global_system_variables.log_warnings > 1)
sql_print_information("Reading Master_info: '%s' Relay_info:'%s'",
buf_master_info_file, buf_relay_log_info_file);
if (init_master_info(mi, buf_master_info_file, buf_relay_log_info_file,
0, thread_mask))
{
err_num++;
sql_print_error("Initialized Master_info from '%s' failed",
buf_master_info_file);
if (!master_info_index->get_master_info(&connection_name,
MYSQL_ERROR::WARN_LEVEL_NOTE))
{
/* Master_info is not in HASH; Add it */
if (master_info_index->add_master_info(mi, FALSE))
return 1;
succ_num++;
unlock_slave_threads(mi);
}
else
{
/* Master_info already in HASH */
sql_print_error(ER(ER_CONNECTION_ALREADY_EXISTS),
(int) connection_name.length, connection_name.str);
unlock_slave_threads(mi);
delete mi;
}
continue;
}
else
{
/* Initialization of Master_info succeded. Add it to HASH */
if (global_system_variables.log_warnings > 1)
sql_print_information("Initialized Master_info from '%s'",
buf_master_info_file);
if (master_info_index->get_master_info(&connection_name,
MYSQL_ERROR::WARN_LEVEL_NOTE))
{
/* Master_info was already registered */
sql_print_error(ER(ER_CONNECTION_ALREADY_EXISTS),
(int) connection_name.length, connection_name.str);
unlock_slave_threads(mi);
delete mi;
continue;
}
/* Master_info was not registered; add it */
if (master_info_index->add_master_info(mi, FALSE))
return 1;
succ_num++;
unlock_slave_threads(mi);
if (!opt_skip_slave_start)
{
if (start_slave_threads(1 /* need mutex */,
0 /* no wait for start*/,
mi,
buf_master_info_file,
buf_relay_log_info_file,
SLAVE_IO | SLAVE_SQL))
{
sql_print_error("Failed to create slave threads for connection %.*s",
(int) connection_name.length,
connection_name.str);
continue;
}
if (global_system_variables.log_warnings)
sql_print_information("Started replication for '%.*s'",
(int) connection_name.length,
connection_name.str);
}
}
}
if (!err_num) // No Error on read Master_info
{
if (global_system_variables.log_warnings > 1)
sql_print_information("Reading of all Master_info entries succeded");
DBUG_RETURN(0);
}
else if (succ_num) // Have some Error and some Success
{
sql_print_warning("Reading of some Master_info entries failed");
DBUG_RETURN(2);
}
else // All failed
{
sql_print_error("Reading of all Master_info entries failed!");
DBUG_RETURN(1);
}
}
/* Write new master.info to master.info.index File */
bool Master_info_index::write_master_name_to_index_file(LEX_STRING *name,
bool do_sync)
{
DBUG_ASSERT(my_b_inited(&index_file) != 0);
DBUG_ENTER("write_master_name_to_index_file");
/* Don't write default slave to master_info.index */
if (name->length == 0)
DBUG_RETURN(0);
reinit_io_cache(&index_file, WRITE_CACHE,
my_b_filelength(&index_file), 0, 0);
if (my_b_write(&index_file, (uchar*) name->str, name->length) ||
my_b_write(&index_file, (uchar*) "\n", 1) ||
flush_io_cache(&index_file) ||
(do_sync && my_sync(index_file.file, MYF(MY_WME))))
{
sql_print_error("Write of new Master_info for '%.*s' to index file failed",
(int) name->length, name->str);
DBUG_RETURN(1);
}
DBUG_RETURN(0);
}
/**
Get Master_info for a connection
@param
connection_name Connection name
warning WARN_LEVEL_NOTE -> Don't print anything
WARN_LEVEL_WARN -> Issue warning if not exists
WARN_LEVEL_ERROR-> Issue error if not exists
*/
Master_info *
Master_info_index::get_master_info(LEX_STRING *connection_name,
MYSQL_ERROR::enum_warning_level warning)
{
Master_info *mi;
char buff[MAX_CONNECTION_NAME+1], *res;
uint buff_length;
/* Make name lower case for comparison */
res= strmake(buff, connection_name->str, connection_name->length);
my_casedn_str(system_charset_info, buff);
buff_length= (size_t) (res-buff);
mi= (Master_info*) my_hash_search(&master_info_hash,
(uchar*) buff, buff_length);
if (!mi && warning != MYSQL_ERROR::WARN_LEVEL_NOTE)
{
my_error(WARN_NO_MASTER_INFO,
MYF(warning == MYSQL_ERROR::WARN_LEVEL_WARN ? ME_JUST_WARNING :
0),
(int) connection_name->length,
connection_name->str);
}
return mi;
}
/* Check Master_host & Master_port is duplicated or not */
bool Master_info_index::check_duplicate_master_info(LEX_STRING *name_arg,
const char *host,
uint port)
{
Master_info *mi;
/* Get full host and port name */
if ((mi= master_info_index->get_master_info(name_arg,
MYSQL_ERROR::WARN_LEVEL_NOTE)))
{
if (!host)
host= mi->host;
if (!port)
port= mi->port;
}
if (!host || !port)
return FALSE; // Not comparable yet
for (uint i= 0; i < master_info_hash.records; ++i)
{
Master_info *tmp_mi;
tmp_mi= (Master_info *) my_hash_element(&master_info_hash, i);
if (tmp_mi == mi)
continue; // Current connection
if (!strcasecmp(host, tmp_mi->host) && port == tmp_mi->port)
{
sql_print_error(ER(ER_CONNECTION_ALREADY_EXISTS),
(int) tmp_mi->connection_name.length,
tmp_mi->connection_name.str);
return TRUE;
}
}
return FALSE;
}
/* Add a Master_info class to Hash Table */
bool Master_info_index::add_master_info(Master_info *mi, bool write_to_file)
{
if (!my_hash_insert(&master_info_hash, (uchar*) mi))
{
if (global_system_variables.log_warnings > 1)
sql_print_information("Added new Master_info '%.*s' to hash table",
(int) mi->connection_name.length,
mi->connection_name.str);
if (write_to_file)
return write_master_name_to_index_file(&mi->connection_name, 1);
return FALSE;
}
/* Impossible error (EOM) ? */
sql_print_error("Adding new entry '%.*s' to master_info failed",
(int) mi->connection_name.length,
mi->connection_name.str);
return TRUE;
}
/**
Remove a Master_info class From Hash Table
TODO: Change this to use my_rename() to make the file name creation
atomic
*/
bool Master_info_index::remove_master_info(LEX_STRING *name)
{
Master_info* mi;
DBUG_ENTER("remove_master_info");
if ((mi= get_master_info(name, MYSQL_ERROR::WARN_LEVEL_WARN)))
{
// Delete Master_info and rewrite others to file
if (!my_hash_delete(&master_info_hash, (uchar*) mi))
{
File index_file_nr;
// Close IO_CACHE and FILE handler fisrt
end_io_cache(&index_file);
my_close(index_file.file, MYF(MY_WME));
// Reopen File and truncate it
fn_format(index_file_name, master_info_file, mysql_data_home,
".index", MY_UNPACK_FILENAME | MY_APPEND_EXT);
if ((index_file_nr= my_open(index_file_name,
O_RDWR | O_CREAT | O_TRUNC | O_BINARY ,
MYF(MY_WME))) < 0 ||
init_io_cache(&index_file, index_file_nr,
IO_SIZE, WRITE_CACHE,
my_seek(index_file_nr,0L,MY_SEEK_END,MYF(0)),
0, MYF(MY_WME | MY_WAIT_IF_FULL)))
{
int error= my_errno;
if (index_file_nr >= 0)
my_close(index_file_nr,MYF(0));
sql_print_error("Create of Master Info Index file '%s' failed with "
"error: %M",
index_file_name, error);
DBUG_RETURN(TRUE);
}
// Rewrite Master_info.index
uint i;
for (i= 0; i< master_info_hash.records; ++i)
{
Master_info *tmp_mi;
tmp_mi= (Master_info *) my_hash_element(&master_info_hash, i);
write_master_name_to_index_file(&tmp_mi->connection_name, 0);
}
my_sync(index_file_nr, MYF(MY_WME));
}
}
DBUG_RETURN(FALSE);
}
#endif /* HAVE_REPLICATION */

51
sql/rpl_mi.h
View File

@ -59,16 +59,23 @@ typedef struct st_mysql MYSQL;
class Master_info : public Slave_reporting_capability
{
public:
Master_info(bool is_slave_recovery);
Master_info(LEX_STRING *connection_name, bool is_slave_recovery);
~Master_info();
bool shall_ignore_server_id(ulong s_id);
void clear_in_memory_info(bool all);
bool error()
{
/* If malloc() in initialization failed */
return connection_name.str == 0;
}
/* the variables below are needed because we can change masters on the fly */
char master_log_name[FN_REFLEN];
char master_log_name[FN_REFLEN+6]; /* Place for multi-*/
char host[HOSTNAME_LENGTH+1];
char user[USERNAME_LENGTH+1];
char password[MAX_PASSWORD_LENGTH+1];
LEX_STRING connection_name; /* User supplied connection name */
LEX_STRING cmp_connection_name; /* Connection name in lower case */
bool ssl; // enables use of SSL connection if true
char ssl_ca[FN_REFLEN], ssl_capath[FN_REFLEN], ssl_cert[FN_REFLEN];
char ssl_cipher[FN_REFLEN], ssl_key[FN_REFLEN];
@ -130,5 +137,45 @@ int flush_master_info(Master_info* mi,
bool need_lock_relay_log);
int change_master_server_id_cmp(ulong *id1, ulong *id2);
/*
Multi master are handled trough this struct.
Changes to this needs to be protected by LOCK_active_mi;
*/
class Master_info_index
{
private:
IO_CACHE index_file;
char index_file_name[FN_REFLEN];
public:
Master_info_index();
~Master_info_index();
HASH master_info_hash;
bool init_all_master_info();
bool write_master_name_to_index_file(LEX_STRING *connection_name,
bool do_sync);
bool check_duplicate_master_info(LEX_STRING *connection_name,
const char *host, uint port);
bool add_master_info(Master_info *mi, bool write_to_file);
bool remove_master_info(LEX_STRING *connection_name);
Master_info *get_master_info(LEX_STRING *connection_name,
MYSQL_ERROR::enum_warning_level warning);
};
bool check_master_connection_name(LEX_STRING *name);
void create_signed_file_name(char *res_file_name, uint length,
const char *info_file,
char separator,
LEX_STRING *suffix);
uchar *get_key_master_info(Master_info *mi, size_t *length,
my_bool not_used __attribute__((unused)));
void free_key_master_info(Master_info *mi);
#endif /* HAVE_REPLICATION */
#endif /* RPL_MI_H */

22
sql/rpl_rli.cc
View File

@ -206,19 +206,37 @@ a file name for --relay-log-index option", opt_relaylog_index_name);
name_warning_sent= 1;
}
/* For multimaster, add connection name to relay log filenames */
Master_info* mi= rli->mi;
char buf_relay_logname[FN_REFLEN], buf_relaylog_index_name_buff[FN_REFLEN];
char *buf_relaylog_index_name= opt_relaylog_index_name;
create_signed_file_name(buf_relay_logname, sizeof(buf_relay_logname),
ln, '-', &mi->connection_name);
ln= buf_relay_logname;
if (opt_relaylog_index_name)
{
buf_relaylog_index_name= buf_relaylog_index_name_buff;
create_signed_file_name(buf_relaylog_index_name_buff,
sizeof(buf_relaylog_index_name_buff),
opt_relaylog_index_name, '-',
&mi->connection_name);
}
rli->relay_log.is_relay_log= TRUE;
/*
note, that if open() fails, we'll still have index file open
but a destructor will take care of that
*/
if (rli->relay_log.open_index_file(opt_relaylog_index_name, ln, TRUE) ||
if (rli->relay_log.open_index_file(buf_relaylog_index_name, ln, TRUE) ||
rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND,
(max_relay_log_size ? max_relay_log_size :
max_binlog_size), 1, TRUE))
{
mysql_mutex_unlock(&rli->data_lock);
sql_print_error("Failed in open_log() called from init_relay_log_info()");
sql_print_error("Failed when trying to open logs for '%s' in init_relay_log_info(). Error: %M", ln, my_errno);
DBUG_RETURN(1);
}
}

3
sql/rpl_rli.h
View File

@ -25,7 +25,6 @@
struct RPL_TABLE_LIST;
class Master_info;
extern uint sql_slave_skip_counter;
/****************************************************************************
@ -230,7 +229,7 @@ public:
skipping one or more events in the master log that have caused
errors, and have been manually applied by DBA already.
*/
volatile uint32 slave_skip_counter;
volatile uint slave_skip_counter; /* Must be uint */
volatile ulong abort_pos_wait; /* Incremented on change master */
volatile ulong slave_run_id; /* Incremented on slave start */
mysql_mutex_t log_space_lock;

19
sql/share/errmsg-utf8.txt
View File

@ -4310,11 +4310,11 @@ ER_BAD_SLAVE
swe "Servern är inte konfigurerade som en replikationsslav. Ändra konfigurationsfilen eller gör CHANGE MASTER TO"
ukr "Сервер не зконфігуровано як підлеглий, виправте це у файлі конфігурації або з CHANGE MASTER TO"
ER_MASTER_INFO
eng "Could not initialize master info structure; more error messages can be found in the MariaDB error log"
fre "Impossible d'initialiser les structures d'information de maître, vous trouverez des messages d'erreur supplémentaires dans le journal des erreurs de MariaDB"
ger "Konnte Master-Info-Struktur nicht initialisieren. Weitere Fehlermeldungen können im MariaDB-Error-Log eingesehen werden"
serbian "Nisam mogao da inicijalizujem informacionu strukturu glavnog servera, proverite da li imam privilegije potrebne za pristup file-u 'master.info'"
swe "Kunde inte initialisera replikationsstrukturerna. See MariaDB fel fil för mera information"
eng "Could not initialize master info structure for '%.*s'; more error messages can be found in the MariaDB error log"
fre "Impossible d'initialiser les structures d'information de maître '%.*s', vous trouverez des messages d'erreur supplémentaires dans le journal des erreurs de MariaDB"
ger "Konnte Master-Info-Struktur '%.*s' nicht initialisieren. Weitere Fehlermeldungen können im MariaDB-Error-Log eingesehen werden"
serbian "Nisam mogao da inicijalizujem informacionu strukturu glavnog servera, proverite da li imam privilegije potrebne za pristup file-u 'master.info' '%.*s'"
swe "Kunde inte initialisera replikationsstrukturerna för '%.*s'. See MariaDB fel fil för mera information"
ER_SLAVE_THREAD
dan "Kunne ikke danne en slave-tråd; check systemressourcerne"
nla "Kon slave thread niet aanmaken, controleer systeem resources"
@ -6156,8 +6156,8 @@ ER_DELAYED_NOT_SUPPORTED
eng "DELAYED option not supported for table '%-.192s'"
ger "Die DELAYED-Option wird für Tabelle '%-.192s' nicht unterstützt"
WARN_NO_MASTER_INFO
eng "The master info structure does not exist"
ger "Die Master-Info-Struktur existiert nicht"
eng "There is no master connection '%.*s'"
ger "Die Master-Info-Struktur existiert nicht '%.*s'"
WARN_OPTION_IGNORED
eng "<%-.64s> option ignored"
ger "Option <%-.64s> ignoriert"
@ -6588,3 +6588,8 @@ ER_QUERY_EXCEEDED_ROWS_EXAMINED_LIMIT
ER_NO_SUCH_TABLE_IN_ENGINE 42S02
eng "Table '%-.192s.%-.192s' doesn't exist in engine"
swe "Det finns ingen tabell som heter '%-.192s.%-.192s' i handlern"
ER_CONNECTION_ALREADY_EXISTS
eng "Connection '%.*s' already exists"
ER_MASTER_LOG_PREFIX
eng "Master '%.*s': "

199
sql/slave.cc
View File

@ -71,8 +71,10 @@ char slave_skip_error_names[SHOW_VAR_FUNC_BUFF_SIZE];
char* slave_load_tmpdir = 0;
Master_info *active_mi= 0;
Master_info_index *master_info_index;
my_bool replicate_same_server_id;
ulonglong relay_log_space_limit = 0;
LEX_STRING default_master_connection_name= { (char*) "", 0 };
/*
When slave thread exits, we need to remember the temporary tables so we
@ -144,7 +146,8 @@ static int process_io_create_file(Master_info* mi, Create_file_log_event* cev);
static bool wait_for_relay_log_space(Relay_log_info* rli);
static inline bool io_slave_killed(THD* thd,Master_info* mi);
static inline bool sql_slave_killed(THD* thd,Relay_log_info* rli);
static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type);
static int init_slave_thread(THD* thd, Master_info *mi,
SLAVE_THD_TYPE thd_type);
static void print_slave_skip_errors(void);
static int safe_connect(THD* thd, MYSQL* mysql, Master_info* mi);
static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi,
@ -159,6 +162,8 @@ static int terminate_slave_thread(THD *thd,
volatile uint *slave_running,
bool skip_lock);
static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info);
static bool send_show_master_info_header(THD *thd, Master_info *mi, bool full);
static bool send_show_master_info_data(THD *thd, Master_info *mi, bool full);
/*
Find out which replications threads are running
@ -263,15 +268,33 @@ int init_slave()
So it's safer to take the lock.
*/
mysql_mutex_lock(&LOCK_active_mi);
/*
TODO: re-write this to interate through the list of files
for multi-master
*/
active_mi= new Master_info(relay_log_recovery);
if (pthread_key_create(&RPL_MASTER_INFO, NULL))
goto err;
master_info_index= new Master_info_index;
if (!master_info_index || master_info_index->init_all_master_info())
{
sql_print_error("Failed to initialize multi master structures");
mysql_mutex_unlock(&LOCK_active_mi);
DBUG_RETURN(1);
}
if (!(active_mi= new Master_info(&default_master_connection_name,
relay_log_recovery)) ||
active_mi->error())
{
delete active_mi;
active_mi= 0;
goto err;
}
if (master_info_index->add_master_info(active_mi, FALSE))
{
delete active_mi;
active_mi= 0;
goto err;
}
/*
If --slave-skip-errors=... was not used, the string value for the
system variable has not been set up yet. Do it now.
@ -286,18 +309,11 @@ int init_slave()
If master_host is specified, create the master_info file if it doesn't
exists.
*/
if (!active_mi)
{
sql_print_error("Failed to allocate memory for the master info structure");
error= 1;
goto err;
}
if (init_master_info(active_mi,master_info_file,relay_log_info_file,
1, (SLAVE_IO | SLAVE_SQL)))
{
sql_print_error("Failed to initialize the master info structure");
error= 1;
goto err;
}
@ -313,14 +329,18 @@ int init_slave()
SLAVE_IO | SLAVE_SQL))
{
sql_print_error("Failed to create slave threads");
error= 1;
goto err;
}
}
err:
end:
mysql_mutex_unlock(&LOCK_active_mi);
DBUG_RETURN(error);
err:
sql_print_error("Failed to allocate memory for the Master Info structure");
error= 1;
goto end;
}
/*
@ -820,42 +840,19 @@ void end_slave()
running presently. If a START SLAVE was in progress, the mutex lock below
will make us wait until slave threads have started, and START SLAVE
returns, then we terminate them here.
We can also be called by cleanup(), which only happens if some
startup parameter to the server was wrong.
*/
mysql_mutex_lock(&LOCK_active_mi);
if (active_mi)
{
/*
TODO: replace the line below with
list_walk(&master_list, (list_walk_action)end_slave_on_walk,0);
once multi-master code is ready.
*/
terminate_slave_threads(active_mi,SLAVE_FORCE_ALL);
}
/* This will call terminate_slave_threads() on all connections */
delete master_info_index;
master_info_index= 0;
active_mi= 0;
mysql_mutex_unlock(&LOCK_active_mi);
DBUG_VOID_RETURN;
}
/**
Free all resources used by slave threads at time of executing shutdown.
The routine must be called after all possible users of @c active_mi
have left.
SYNOPSIS
close_active_mi()
*/
void close_active_mi()
{
mysql_mutex_lock(&LOCK_active_mi);
if (active_mi)
{
end_master_info(active_mi);
delete active_mi;
active_mi= 0;
}
mysql_mutex_unlock(&LOCK_active_mi);
}
static bool io_slave_killed(THD* thd, Master_info* mi)
{
DBUG_ENTER("io_slave_killed");
@ -2022,12 +2019,28 @@ int register_slave_on_master(MYSQL* mysql, Master_info *mi,
@retval FALSE success
@retval TRUE failure
*/
bool show_master_info(THD* thd, Master_info* mi)
bool show_master_info(THD *thd, Master_info *mi, bool full)
{
DBUG_ENTER("show_master_info");
if (send_show_master_info_header(thd, mi, full))
DBUG_RETURN(TRUE);
if (send_show_master_info_data(thd, mi, full))
DBUG_RETURN(TRUE);
my_eof(thd);
DBUG_RETURN(FALSE);
}
static bool send_show_master_info_header(THD *thd, Master_info *mi, bool full)
{
// TODO: fix this for multi-master
List<Item> field_list;
Protocol *protocol= thd->protocol;
DBUG_ENTER("show_master_info");
DBUG_ENTER("show_master_info_header");
if (full)
field_list.push_back(new Item_empty_string("Connection_name",
MAX_CONNECTION_NAME));
field_list.push_back(new Item_empty_string("Slave_IO_State",
14));
@ -2097,17 +2110,29 @@ bool show_master_info(THD* thd, Master_info* mi)
if (protocol->send_result_set_metadata(&field_list,
Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
DBUG_RETURN(TRUE);
DBUG_RETURN(FALSE);
}
static bool send_show_master_info_data(THD *thd, Master_info *mi, bool full)
{
DBUG_ENTER("send_show_master_info_data");
if (mi->host[0])
{
DBUG_PRINT("info",("host is set: '%s'", mi->host));
String *packet= &thd->packet;
Protocol *protocol= thd->protocol;
protocol->prepare_for_resend();
/*
slave_running can be accessed without run_lock but not other
non-volotile members like mi->io_thd, which is guarded by the mutex.
*/
if (full)
protocol->store(mi->connection_name.str, mi->connection_name.length,
&my_charset_bin);
mysql_mutex_lock(&mi->run_lock);
protocol->store(mi->io_thd ? mi->io_thd->proc_info : "", &my_charset_bin);
mysql_mutex_unlock(&mi->run_lock);
@ -2250,6 +2275,65 @@ bool show_master_info(THD* thd, Master_info* mi)
if (my_net_write(&thd->net, (uchar*) thd->packet.ptr(), packet->length()))
DBUG_RETURN(TRUE);
}
DBUG_RETURN(FALSE);
}
/* Used to sort connections by name */
static int cmp_mi_by_name(const Master_info **arg1,
const Master_info **arg2)
{
return my_strcasecmp(system_charset_info, (*arg1)->connection_name.str,
(*arg2)->connection_name.str);
}
/**
Execute a SHOW FULL SLAVE STATUS statement.
@param thd Pointer to THD object for the client thread executing the
statement.
Elements are sorted according to the original connection_name.
@retval FALSE success
@retval TRUE failure
*/
bool show_all_master_info(THD* thd)
{
uint i, elements;
Master_info **tmp;
DBUG_ENTER("show_master_info");
if (send_show_master_info_header(thd, active_mi, 1))
DBUG_RETURN(TRUE);
if (!(elements= master_info_index->master_info_hash.records))
goto end;
/*
Sort lines to get them into a predicted order
(needed for test cases and to not confuse users)
*/
if (!(tmp= (Master_info**) thd->alloc(sizeof(Master_info*) * elements)))
DBUG_RETURN(TRUE);
for (i= 0; i < elements; i++)
{
tmp[i]= (Master_info *) my_hash_element(&master_info_index->
master_info_hash, i);
}
my_qsort(tmp, elements, sizeof(Master_info*), (qsort_cmp) cmp_mi_by_name);
for (i= 0; i < elements; i++)
{
if (send_show_master_info_data(thd, tmp[i], 1))
DBUG_RETURN(TRUE);
}
end:
my_eof(thd);
DBUG_RETURN(FALSE);
}
@ -2303,7 +2387,8 @@ void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli)
init_slave_thread()
*/
static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
static int init_slave_thread(THD* thd, Master_info *mi,
SLAVE_THD_TYPE thd_type)
{
DBUG_ENTER("init_slave_thread");
#if !defined(DBUG_OFF)
@ -2319,7 +2404,8 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
than the corresponding packet (query) sent from client to master.
*/
thd->variables.max_allowed_packet= slave_max_allowed_packet;
thd->slave_thread = 1;
thd->slave_thread= 1;
thd->connection_name= mi->connection_name;
thd->enable_slow_log= opt_log_slow_slave_statements;
thd->variables.log_slow_filter= global_system_variables.log_slow_filter;
set_slave_thread_options(thd);
@ -2635,7 +2721,10 @@ int apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli)
int reason= ev->shall_skip(rli);
if (reason == Log_event::EVENT_SKIP_COUNT)
sql_slave_skip_counter= --rli->slave_skip_counter;
{
DBUG_ASSERT(rli->slave_skip_counter > 0);
rli->slave_skip_counter--;
}
mysql_mutex_unlock(&rli->data_lock);
if (reason == Log_event::EVENT_SKIP_NOT)
exec_res= ev->apply_event(rli);
@ -3058,7 +3147,7 @@ pthread_handler_t handle_slave_io(void *arg)
pthread_detach_this_thread();
thd->thread_stack= (char*) &thd; // remember where our stack is
mi->clear_error();
if (init_slave_thread(thd, SLAVE_THD_IO))
if (init_slave_thread(thd, mi, SLAVE_THD_IO))
{
mysql_cond_broadcast(&mi->start_cond);
sql_print_error("Failed during slave I/O thread initialization");
@ -3457,7 +3546,8 @@ pthread_handler_t handle_slave_sql(void *arg)
my_off_t UNINIT_VAR(saved_log_pos);
my_off_t UNINIT_VAR(saved_master_log_pos);
my_off_t saved_skip= 0;
Relay_log_info* rli = &((Master_info*)arg)->rli;
Master_info *mi= ((Master_info*)arg);
Relay_log_info* rli = &mi->rli;
const char *errmsg;
// needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
@ -3471,6 +3561,7 @@ pthread_handler_t handle_slave_sql(void *arg)
thd->thread_stack = (char*)&thd; // remember where our stack is
DBUG_ASSERT(rli->inited);
DBUG_ASSERT(rli->mi == mi);
mysql_mutex_lock(&rli->run_lock);
DBUG_ASSERT(!rli->slave_running);
errmsg= 0;
@ -3485,7 +3576,7 @@ pthread_handler_t handle_slave_sql(void *arg)
rli->slave_running = 1;
pthread_detach_this_thread();
if (init_slave_thread(thd, SLAVE_THD_SQL))
if (init_slave_thread(thd, mi, SLAVE_THD_SQL))
{
/*
TODO: this is currently broken - slave start and change master

9
sql/slave.h
View File

@ -45,9 +45,12 @@
#define MAX_SLAVE_ERROR 2000
#define MAX_REPLICATION_THREAD 64
// Forward declarations
class Relay_log_info;
class Master_info;
class Master_info_index;
int init_intvar_from_file(int* var, IO_CACHE* f, int default_val);
int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
@ -197,7 +200,8 @@ int mysql_table_dump(THD* thd, const char* db,
int fetch_master_table(THD* thd, const char* db_name, const char* table_name,
Master_info* mi, MYSQL* mysql, bool overwrite);
bool show_master_info(THD* thd, Master_info* mi);
bool show_master_info(THD* thd, Master_info* mi, bool full);
bool show_all_master_info(THD* thd);
bool show_binlog_info(THD* thd);
bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
bool (*pred)(const void *), const void *param);
@ -231,6 +235,9 @@ bool net_request_file(NET* net, const char* fname);
extern bool volatile abort_loop;
extern Master_info main_mi, *active_mi; /* active_mi for multi-master */
extern Master_info *default_master_info; /* To replace active_mi */
extern Master_info_index *master_info_index;
extern LEX_STRING default_master_connection_name;
extern LIST master_list;
extern my_bool replicate_same_server_id;

5
sql/sql_base.cc
View File

@ -396,6 +396,7 @@ bool table_def_init(void)
void table_def_start_shutdown(void)
{
DBUG_ENTER("table_def_start_shutdown");
if (table_def_inited)
{
mysql_mutex_lock(&LOCK_open);
@ -410,6 +411,7 @@ void table_def_start_shutdown(void)
/* Free all cached but unused TABLEs and TABLE_SHAREs. */
close_cached_tables(NULL, NULL, FALSE, LONG_TIMEOUT);
}
DBUG_VOID_RETURN;
}
@ -1086,6 +1088,9 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
mysql_mutex_unlock(&LOCK_open);
DBUG_PRINT("info", ("open table definitions: %d",
(int) table_def_cache.records));
if (!wait_for_refresh)
DBUG_RETURN(result);

10
sql/sql_class.cc
View File

@ -799,6 +799,9 @@ THD::THD()
progress.max_counter= 0;
current_linfo = 0;
slave_thread = 0;
connection_name.str= 0;
connection_name.length= 0;
bzero(&variables, sizeof(variables));
thread_id= 0;
one_shot_set= 0;
@ -1166,7 +1169,14 @@ void THD::init(void)
avoid temporary tables replication failure.
*/
variables.pseudo_thread_id= thread_id;
variables.default_master_connection.str= default_master_connection_buff;
::strmake(variables.default_master_connection.str,
global_system_variables.default_master_connection.str,
variables.default_master_connection.length);
mysql_mutex_unlock(&LOCK_global_system_variables);
server_status= SERVER_STATUS_AUTOCOMMIT;
if (variables.sql_mode & MODE_NO_BACKSLASH_ESCAPES)
server_status|= SERVER_STATUS_NO_BACKSLASH_ESCAPES;

10
sql/sql_class.h
View File

@ -533,6 +533,11 @@ typedef struct system_variables
thread the query is being run to replicate temp tables properly
*/
my_thread_id pseudo_thread_id;
/**
Place holder to store sql_slave_skip_counter in sys_var.cc during
update and show of variables.
*/
uint slave_skip_counter;
my_bool low_priority_updates;
my_bool query_cache_wlock_invalidate;
@ -557,6 +562,9 @@ typedef struct system_variables
CHARSET_INFO *collation_database;
CHARSET_INFO *collation_connection;
/* Names. These will be allocated in buffers in thd */
LEX_STRING default_master_connection;
/* Error messages */
MY_LOCALE *lc_messages;
/* Locale Support */
@ -2192,6 +2200,8 @@ public:
/* scramble - random string sent to client on handshake */
char scramble[SCRAMBLE_LENGTH+1];
LEX_STRING connection_name; /* If slave */
char default_master_connection_buff[MAX_CONNECTION_NAME+1];
bool slave_thread, one_shot_set;
bool extra_port; /* If extra connection */

1
sql/sql_const.h
View File

@ -38,6 +38,7 @@
#define MAX_REFLENGTH 4 /* Max length for record ref */
#endif
#define MAX_HOSTNAME 61 /* len+1 in mysql.user */
#define MAX_CONNECTION_NAME NAME_LEN
#define MAX_MBWIDTH 3 /* Max multibyte sequence */
#define MAX_FIELD_CHARLENGTH 255

1
sql/sql_lex.cc
View File

@ -503,6 +503,7 @@ void lex_start(THD *thd)
lex->expr_allows_subselect= TRUE;
lex->use_only_table_context= FALSE;
lex->parse_vcol_expr= FALSE;
lex->verbose= 0;
lex->name.str= 0;
lex->name.length= 0;

1
sql/sql_lex.h
View File

@ -286,6 +286,7 @@ struct LEX_MASTER_INFO
char *host, *user, *password, *log_file_name;
char *ssl_key, *ssl_cert, *ssl_ca, *ssl_capath, *ssl_cipher;
char *relay_log_name;
LEX_STRING connection_name;
ulonglong pos;
ulong relay_log_pos;
ulong server_id;

75
sql/sql_parse.cc
View File

@ -82,6 +82,7 @@
#include <myisam.h>
#include <my_dir.h>
#include "rpl_handler.h"
#include "rpl_mi.h"
#include "sp_head.h"
#include "sp.h"
@ -2343,10 +2344,41 @@ case SQLCOM_PREPARE:
#ifdef HAVE_REPLICATION
case SQLCOM_CHANGE_MASTER:
{
LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
Master_info *mi;
bool new_master= 0;
if (check_global_access(thd, SUPER_ACL))
goto error;
mysql_mutex_lock(&LOCK_active_mi);
res = change_master(thd,active_mi);
mi= master_info_index->get_master_info(&lex_mi->connection_name,
MYSQL_ERROR::WARN_LEVEL_NOTE);
if (mi == NULL)
{
/* New replication created */
mi= new Master_info(&lex_mi->connection_name, relay_log_recovery);
if (!mi || mi->error())
{
delete mi;
res= 1;
mysql_mutex_unlock(&LOCK_active_mi);
break;
}
new_master= 1;
}
res= change_master(thd, mi);
if (res && new_master)
{
/*
The new master was added by change_master(). Remove it as it didn't
work.
*/
master_info_index->remove_master_info(&lex_mi->connection_name);
}
mysql_mutex_unlock(&LOCK_active_mi);
break;
}
@ -2356,15 +2388,19 @@ case SQLCOM_PREPARE:
if (check_global_access(thd, SUPER_ACL | REPL_CLIENT_ACL))
goto error;
mysql_mutex_lock(&LOCK_active_mi);
if (active_mi != NULL)
{
res = show_master_info(thd, active_mi);
}
if (lex->verbose)
res= show_all_master_info(thd);
else
{
push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
WARN_NO_MASTER_INFO, ER(WARN_NO_MASTER_INFO));
my_ok(thd);
LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
Master_info *mi;
mi= master_info_index->get_master_info(&lex_mi->connection_name,
MYSQL_ERROR::WARN_LEVEL_ERROR);
if (mi != NULL)
{
res= show_master_info(thd, mi, 0);
}
}
mysql_mutex_unlock(&LOCK_active_mi);
break;
@ -2664,12 +2700,21 @@ end_with_restore_list:
#ifdef HAVE_REPLICATION
case SQLCOM_SLAVE_START:
{
LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
Master_info *mi;
mysql_mutex_lock(&LOCK_active_mi);
start_slave(thd,active_mi,1 /* net report*/);
if ((mi= (master_info_index->
get_master_info(&lex_mi->connection_name,
MYSQL_ERROR::WARN_LEVEL_ERROR))))
start_slave(thd, mi, 1 /* net report*/);
mysql_mutex_unlock(&LOCK_active_mi);
break;
}
case SQLCOM_SLAVE_STOP:
{
LEX_MASTER_INFO *lex_mi;
Master_info *mi;
/*
If the client thread has locked tables, a deadlock is possible.
Assume that
@ -2684,20 +2729,24 @@ end_with_restore_list:
client thread has locked tables
*/
if (thd->locked_tables_mode ||
thd->in_active_multi_stmt_transaction() || thd->global_read_lock.is_acquired())
thd->in_active_multi_stmt_transaction() ||
thd->global_read_lock.is_acquired())
{
my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
goto error;
}
{
lex_mi= &thd->lex->mi;
mysql_mutex_lock(&LOCK_active_mi);
stop_slave(thd,active_mi,1/* net report*/);
if ((mi= (master_info_index->
get_master_info(&lex_mi->connection_name,
MYSQL_ERROR::WARN_LEVEL_ERROR))))
stop_slave(thd, mi, 1/* net report*/);
mysql_mutex_unlock(&LOCK_active_mi);
break;
}
#endif /* HAVE_REPLICATION */
case SQLCOM_RENAME_TABLE:
{
if (execute_rename_table(thd, first_table, all_tables))

17
sql/sql_reload.cc
View File

@ -26,6 +26,7 @@
#include "sql_repl.h" // reset_master, reset_slave
#include "rpl_mi.h" // Master_info::data_lock
#include "debug_sync.h"
#include "rpl_mi.h"
static void disable_checkpoints(THD *thd);
@ -314,13 +315,27 @@ bool reload_acl_and_cache(THD *thd, unsigned long options,
#ifdef HAVE_REPLICATION
if (options & REFRESH_SLAVE)
{
LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
Master_info *mi;
tmp_write_to_binlog= 0;
mysql_mutex_lock(&LOCK_active_mi);
if (reset_slave(thd, active_mi))
if (!(mi= (master_info_index->
get_master_info(&lex_mi->connection_name,
MYSQL_ERROR::WARN_LEVEL_ERROR))))
{
result= 1;
}
else if (reset_slave(thd, mi))
{
/* NOTE: my_error() has been already called by reset_slave(). */
result= 1;
}
else if (mi->connection_name.length && thd->lex->reset_slave_info.all)
{
/* If not default connection and 'all' is used */
master_info_index->remove_master_info(&mi->connection_name);
}
mysql_mutex_unlock(&LOCK_active_mi);
}
#endif

121
sql/sql_repl.cc
View File

@ -34,14 +34,6 @@ my_bool opt_sporadic_binlog_dump_fail = 0;
static int binlog_dump_count = 0;
#endif
/**
a copy of active_mi->rli->slave_skip_counter, for showing in SHOW VARIABLES,
INFORMATION_SCHEMA.GLOBAL_VARIABLES and @@sql_slave_skip_counter without
taking all the mutexes needed to access active_mi->rli->slave_skip_counter
properly.
*/
uint sql_slave_skip_counter;
extern TYPELIB binlog_checksum_typelib;
/*
@ -1310,8 +1302,17 @@ int start_slave(THD* thd , Master_info* mi, bool net_report)
{
int slave_errno= 0;
int thread_mask;
char master_info_file_tmp[FN_REFLEN];
char relay_log_info_file_tmp[FN_REFLEN];
DBUG_ENTER("start_slave");
create_signed_file_name(master_info_file_tmp,
sizeof(master_info_file_tmp),
master_info_file, '.', &mi->connection_name);
create_signed_file_name(relay_log_info_file_tmp,
sizeof(relay_log_info_file_tmp),
relay_log_info_file, '.', &mi->connection_name);
if (check_access(thd, SUPER_ACL, any_db, NULL, NULL, 0, 0))
DBUG_RETURN(1);
lock_slave_threads(mi); // this allows us to cleanly read slave_running
@ -1327,7 +1328,7 @@ int start_slave(THD* thd , Master_info* mi, bool net_report)
thread_mask&= thd->lex->slave_thd_opt;
if (thread_mask) //some threads are stopped, start them
{
if (init_master_info(mi,master_info_file,relay_log_info_file, 0,
if (init_master_info(mi,master_info_file_tmp,relay_log_info_file_tmp, 0,
thread_mask))
slave_errno=ER_MASTER_INFO;
else if (server_id_supplied && *mi->host)
@ -1403,7 +1404,8 @@ int start_slave(THD* thd , Master_info* mi, bool net_report)
slave_errno = start_slave_threads(0 /*no mutex */,
1 /* wait for start */,
mi,
master_info_file,relay_log_info_file,
master_info_file_tmp,
relay_log_info_file_tmp,
thread_mask);
}
else
@ -1421,7 +1423,9 @@ int start_slave(THD* thd , Master_info* mi, bool net_report)
if (slave_errno)
{
if (net_report)
my_message(slave_errno, ER(slave_errno), MYF(0));
my_error(slave_errno, MYF(0),
(int) mi->connection_name.length,
mi->connection_name.str);
DBUG_RETURN(1);
}
else if (net_report)
@ -1446,11 +1450,9 @@ int start_slave(THD* thd , Master_info* mi, bool net_report)
*/
int stop_slave(THD* thd, Master_info* mi, bool net_report )
{
DBUG_ENTER("stop_slave");
int slave_errno;
if (!thd)
thd = current_thd;
DBUG_ENTER("stop_slave");
DBUG_PRINT("enter",("Connection: %s", mi->connection_name.str));
if (check_access(thd, SUPER_ACL, any_db, NULL, NULL, 0, 0))
DBUG_RETURN(1);
@ -1514,6 +1516,8 @@ int reset_slave(THD *thd, Master_info* mi)
int thread_mask= 0, error= 0;
uint sql_errno=ER_UNKNOWN_ERROR;
const char* errmsg= "Unknown error occured while reseting slave";
char master_info_file_tmp[FN_REFLEN];
char relay_log_info_file_tmp[FN_REFLEN];
DBUG_ENTER("reset_slave");
lock_slave_threads(mi);
@ -1549,22 +1553,35 @@ int reset_slave(THD *thd, Master_info* mi)
// close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
end_master_info(mi);
// and delete these two files
fn_format(fname, master_info_file, mysql_data_home, "", 4+32);
create_signed_file_name(master_info_file_tmp,
sizeof(master_info_file_tmp),
master_info_file, '.', &mi->connection_name);
create_signed_file_name(relay_log_info_file_tmp,
sizeof(relay_log_info_file_tmp),
relay_log_info_file, '.', &mi->connection_name);
fn_format(fname, master_info_file_tmp, mysql_data_home, "", 4+32);
if (mysql_file_stat(key_file_master_info, fname, &stat_area, MYF(0)) &&
mysql_file_delete(key_file_master_info, fname, MYF(MY_WME)))
{
error=1;
goto err;
}
else if (global_system_variables.log_warnings > 1)
sql_print_information("Deleted Master_info file '%s'.", fname);
// delete relay_log_info_file
fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32);
fn_format(fname, relay_log_info_file_tmp, mysql_data_home, "", 4+32);
if (mysql_file_stat(key_file_relay_log_info, fname, &stat_area, MYF(0)) &&
mysql_file_delete(key_file_relay_log_info, fname, MYF(MY_WME)))
{
error=1;
goto err;
}
else if (global_system_variables.log_warnings > 1)
sql_print_information("Deleted Master_info file '%s'.", fname);
RUN_HOOK(binlog_relay_io, after_reset_slave, (thd, mi));
err:
@ -1644,12 +1661,36 @@ bool change_master(THD* thd, Master_info* mi)
char saved_host[HOSTNAME_LENGTH + 1];
uint saved_port;
char saved_log_name[FN_REFLEN];
char master_info_file_tmp[FN_REFLEN];
char relay_log_info_file_tmp[FN_REFLEN];
my_off_t saved_log_pos;
LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
DBUG_ENTER("change_master");
/*
We need to check if there is an empty master_host. Otherwise
change master succeeds, a master.info file is created containing
empty master_host string and when issuing: start slave; an error
is thrown stating that the server is not configured as slave.
(See BUG#28796).
*/
if (lex_mi->host && !*lex_mi->host)
{
my_error(ER_WRONG_ARGUMENTS, MYF(0), "MASTER_HOST");
DBUG_RETURN(TRUE);
}
if (master_info_index->check_duplicate_master_info(&lex_mi->connection_name,
lex_mi->host,
lex_mi->port))
{
my_error(ER_MASTER_INFO, MYF(0),
(int) lex_mi->connection_name.length,
lex_mi->connection_name.str);
DBUG_RETURN(TRUE);
}
lock_slave_threads(mi);
init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
if (thread_mask) // We refuse if any slave thread is running
{
my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
@ -1658,24 +1699,40 @@ bool change_master(THD* thd, Master_info* mi)
}
thd_proc_info(thd, "Changing master");
/*
We need to check if there is an empty master_host. Otherwise
change master succeeds, a master.info file is created containing
empty master_host string and when issuing: start slave; an error
is thrown stating that the server is not configured as slave.
(See BUG#28796).
*/
if(lex_mi->host && !*lex_mi->host)
create_signed_file_name(master_info_file_tmp,
sizeof(master_info_file_tmp),
master_info_file, '.', &mi->connection_name);
create_signed_file_name(relay_log_info_file_tmp,
sizeof(relay_log_info_file_tmp),
relay_log_info_file, '.', &mi->connection_name);
/* if new Master_info doesn't exists, add it */
if (!master_info_index->get_master_info(&mi->connection_name,
MYSQL_ERROR::WARN_LEVEL_NOTE))
{
my_error(ER_WRONG_ARGUMENTS, MYF(0), "MASTER_HOST");
unlock_slave_threads(mi);
DBUG_RETURN(TRUE);
if (master_info_index->add_master_info(mi, TRUE))
{
my_error(ER_MASTER_INFO, MYF(0),
(int) lex_mi->connection_name.length,
lex_mi->connection_name.str);
ret= TRUE;
goto err;
}
// TODO: see if needs re-write
if (init_master_info(mi, master_info_file, relay_log_info_file, 0,
}
if (global_system_variables.log_warnings > 1)
sql_print_information("Master: '%.*s' Master_info_file: '%s' "
"Relay_info_file: '%s'",
(int) mi->connection_name.length,
mi->connection_name.str,
master_info_file_tmp, relay_log_info_file_tmp);
if (init_master_info(mi, master_info_file_tmp, relay_log_info_file_tmp, 0,
thread_mask))
{
my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
my_error(ER_MASTER_INFO, MYF(0),
(int) lex_mi->connection_name.length,
lex_mi->connection_name.str);
ret= TRUE;
goto err;
}

53
sql/sql_yacc.yy
View File

@ -65,6 +65,7 @@
#include <myisammrg.h>
#include "keycaches.h"
#include "set_var.h"
#include "rpl_mi.h"
/* this is to get the bison compilation windows warnings out */
#ifdef _MSC_VER
@ -1904,7 +1905,7 @@ help:
/* change master */
change:
CHANGE MASTER_SYM TO_SYM
CHANGE MASTER_SYM optional_connection_name TO_SYM
{
Lex->sql_command = SQLCOM_CHANGE_MASTER;
}
@ -2053,6 +2054,29 @@ master_file_def:
}
;
optional_connection_name:
/* empty */
{
THD *thd= YYTHD;
LEX *lex= thd->lex;
lex->mi.connection_name= thd->variables.default_master_connection;
}
| connection_name;
;
connection_name:
TEXT_STRING_sys
{
Lex->mi.connection_name= $1;
#ifdef HAVE_REPLICATION
if (check_master_connection_name(&$1))
{
my_error(ER_WRONG_ARGUMENTS, MYF(0), "MASTER_CONNECTION_NAME");
MYSQL_YYABORT;
}
#endif
}
/* create a table */
create:
@ -7074,7 +7098,7 @@ opt_to:
*/
slave:
START_SYM SLAVE slave_thread_opts
START_SYM SLAVE optional_connection_name slave_thread_opts
{
LEX *lex=Lex;
lex->sql_command = SQLCOM_SLAVE_START;
@ -7083,14 +7107,14 @@ slave:
}
slave_until
{}
| STOP_SYM SLAVE slave_thread_opts
| STOP_SYM SLAVE optional_connection_name slave_thread_opts
{
LEX *lex=Lex;
lex->sql_command = SQLCOM_SLAVE_STOP;
lex->type = 0;
/* If you change this code don't forget to update SLAVE STOP too */
}
| SLAVE START_SYM slave_thread_opts
| SLAVE optional_connection_name START_SYM slave_thread_opts
{
LEX *lex=Lex;
lex->sql_command = SQLCOM_SLAVE_START;
@ -7098,7 +7122,7 @@ slave:
}
slave_until
{}
| SLAVE STOP_SYM slave_thread_opts
| SLAVE optional_connection_name STOP_SYM slave_thread_opts
{
LEX *lex=Lex;
lex->sql_command = SQLCOM_SLAVE_STOP;
@ -11558,9 +11582,23 @@ show_param:
{
Lex->sql_command = SQLCOM_SHOW_MASTER_STAT;
}
| SLAVE STATUS_SYM
| FULL SLAVE STATUS_SYM
{
Lex->sql_command = SQLCOM_SHOW_SLAVE_STAT;
Lex->verbose= 1;
}
| SLAVE STATUS_SYM
{
THD *thd= YYTHD;
LEX *lex= thd->lex;
lex->mi.connection_name= thd->variables.default_master_connection;
lex->sql_command = SQLCOM_SHOW_SLAVE_STAT;
lex->verbose= 0;
}
| SLAVE connection_name STATUS_SYM
{
Lex->sql_command = SQLCOM_SHOW_SLAVE_STAT;
Lex->verbose= 0;
}
| CLIENT_STATS_SYM
{
@ -11824,7 +11862,7 @@ flush_option:
{ Lex->type|= REFRESH_LOG; }
| STATUS_SYM
{ Lex->type|= REFRESH_STATUS; }
| SLAVE
| SLAVE optional_connection_name
{
Lex->type|= REFRESH_SLAVE;
Lex->reset_slave_info.all= false;
@ -11871,6 +11909,7 @@ reset_options:
reset_option:
SLAVE { Lex->type|= REFRESH_SLAVE; }
optional_connection_name
slave_reset_options { }
| MASTER_SYM { Lex->type|= REFRESH_MASTER; }
| QUERY_SYM CACHE_SYM { Lex->type|= REFRESH_QUERY_CACHE;}

1
sql/strfunc.cc
View File

@ -15,6 +15,7 @@
/* Some useful string utility functions used by the MySQL server */
#include <my_global.h>
#include "sql_priv.h"
#include "unireg.h"
#include "strfunc.h"

110
sql/sys_vars.cc
View File

@ -796,6 +796,30 @@ static Sys_var_lexstring Sys_init_connect(
DEFAULT(""), &PLock_sys_init_connect, NOT_IN_BINLOG,
ON_CHECK(check_init_string));
#ifdef HAVE_REPLICATION
static bool check_master_connection(sys_var *self, THD *thd, set_var *var)
{
LEX_STRING tmp;
tmp.str= var->save_result.string_value.str;
tmp.length= var->save_result.string_value.length;
if (check_master_connection_name(&tmp))
{
my_error(ER_WRONG_ARGUMENTS, MYF(ME_JUST_WARNING),
"DEFAULT_MASTER_CONNECTION");
return true;
}
return false;
}
static Sys_var_session_lexstring Sys_default_master_connection(
"default_master_connection",
"Master connection to use for all slave variables and slave commands",
SESSION_ONLY(default_master_connection),
NO_CMD_LINE, IN_SYSTEM_CHARSET,
DEFAULT(""), MAX_CONNECTION_NAME, ON_CHECK(check_master_connection),
ON_UPDATE(0));
#endif
static Sys_var_charptr Sys_init_file(
"init_file", "Read SQL commands from this file at startup",
READ_ONLY GLOBAL_VAR(opt_init_file),
@ -3454,47 +3478,67 @@ static Sys_var_uint Sys_slave_net_timeout(
NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(0),
ON_UPDATE(fix_slave_net_timeout));
static bool check_slave_skip_counter(sys_var *self, THD *thd, set_var *var)
/*
Access a multi_source variable
Return 0 + warning if it doesn't exist
*/
uint Sys_var_multi_source_uint::
get_master_info_uint_value(THD *thd, ptrdiff_t offset)
{
bool result= false;
Master_info *mi;
uint res= 0; // Default value
mysql_mutex_lock(&LOCK_active_mi);
mysql_mutex_lock(&active_mi->rli.run_lock);
if (active_mi->rli.slave_running)
mi= master_info_index->
get_master_info(&thd->variables.default_master_connection,
MYSQL_ERROR::WARN_LEVEL_WARN);
if (mi)
{
my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
result= true;
mysql_mutex_lock(&mi->rli.data_lock);
res= *((uint*) (((uchar*) mi) + master_info_offset));
mysql_mutex_unlock(&mi->rli.data_lock);
}
mysql_mutex_unlock(&LOCK_active_mi);
return res;
}
static bool update_slave_skip_counter(sys_var *self, THD *thd,
enum_var_type type)
{
bool result= true;
Master_info *mi;
mysql_mutex_lock(&LOCK_active_mi);
mi= master_info_index->
get_master_info(&thd->variables.default_master_connection,
MYSQL_ERROR::WARN_LEVEL_ERROR);
if (mi)
{
mysql_mutex_lock(&mi->rli.run_lock);
if (mi->rli.slave_running)
my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
else
{
result= false; // ok
mysql_mutex_lock(&mi->rli.data_lock);
/* The value was stored temporarly in thd */
mi->rli.slave_skip_counter= thd->variables.slave_skip_counter;
mysql_mutex_unlock(&mi->rli.data_lock);
}
mysql_mutex_unlock(&mi->rli.run_lock);
}
mysql_mutex_unlock(&active_mi->rli.run_lock);
mysql_mutex_unlock(&LOCK_active_mi);
return result;
}
static bool fix_slave_skip_counter(sys_var *self, THD *thd, enum_var_type type)
{
mysql_mutex_unlock(&LOCK_global_system_variables);
mysql_mutex_lock(&LOCK_active_mi);
mysql_mutex_lock(&active_mi->rli.run_lock);
/*
The following test should normally never be true as we test this
in the check function; To be safe against multiple
SQL_SLAVE_SKIP_COUNTER request, we do the check anyway
*/
if (!active_mi->rli.slave_running)
{
mysql_mutex_lock(&active_mi->rli.data_lock);
active_mi->rli.slave_skip_counter= sql_slave_skip_counter;
mysql_mutex_unlock(&active_mi->rli.data_lock);
}
mysql_mutex_unlock(&active_mi->rli.run_lock);
mysql_mutex_unlock(&LOCK_active_mi);
mysql_mutex_lock(&LOCK_global_system_variables);
return 0;
}
static Sys_var_uint Sys_slave_skip_counter(
"sql_slave_skip_counter", "sql_slave_skip_counter",
GLOBAL_VAR(sql_slave_skip_counter), NO_CMD_LINE,
static Sys_var_multi_source_uint
Sys_slave_skip_counter("sql_slave_skip_counter",
"Skip the next N events from the master log",
SESSION_VAR(slave_skip_counter),
offsetof(Master_info, rli.slave_skip_counter),
VALID_RANGE(0, UINT_MAX), DEFAULT(0), BLOCK_SIZE(1),
NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_slave_skip_counter),
ON_UPDATE(fix_slave_skip_counter));
ON_UPDATE(update_slave_skip_counter));
static Sys_var_charptr Sys_slave_skip_errors(
"slave_skip_errors", "Tells the slave thread to continue "

142
sql/sys_vars.h
View File

@ -633,6 +633,88 @@ public:
}
};
/*
A LEX_STRING stored only in thd->variables
Only to be used for small buffers
*/
class Sys_var_session_lexstring: public sys_var
{
size_t max_length;
public:
Sys_var_session_lexstring(const char *name_arg,
const char *comment, int flag_args,
ptrdiff_t off, size_t size, CMD_LINE getopt,
enum charset_enum is_os_charset_arg,
const char *def_val, size_t max_length_arg,
on_check_function on_check_func=0,
on_update_function on_update_func=0)
: sys_var(&all_sys_vars, name_arg, comment, flag_args, off, getopt.id,
getopt.arg_type, SHOW_CHAR, (intptr)def_val,
0, VARIABLE_NOT_IN_BINLOG, on_check_func, on_update_func,
0, 0),max_length(max_length_arg)
{
option.var_type= GET_NO_ARG;
SYSVAR_ASSERT(scope() == ONLY_SESSION)
*const_cast<SHOW_TYPE*>(&show_val_type)= SHOW_LEX_STRING;
}
bool do_check(THD *thd, set_var *var)
{
char buff[STRING_BUFFER_USUAL_SIZE];
String str(buff, sizeof(buff), system_charset_info), *res;
if (!(res=var->value->val_str(&str)))
var->save_result.string_value.str= const_cast<char*>("");
else
{
if (res->length() > max_length)
{
my_error(ER_WRONG_STRING_LENGTH, MYF(0),
res->ptr(), name.str, (int) max_length);
return true;
}
var->save_result.string_value.str= thd->strmake(res->ptr(),
res->length());
var->save_result.string_value.length= res->length();
}
return false;
}
bool session_update(THD *thd, set_var *var)
{
LEX_STRING *tmp= &session_var(thd, LEX_STRING);
tmp->length= var->save_result.string_value.length;
/* Store as \0 terminated string (just to be safe) */
strmake(tmp->str, var->save_result.string_value.str, tmp->length);
return false;
}
bool global_update(THD *thd, set_var *var)
{
DBUG_ASSERT(FALSE);
return false;
}
void session_save_default(THD *thd, set_var *var)
{
char *ptr= (char*)(intptr)option.def_value;
var->save_result.string_value.str= ptr;
}
void global_save_default(THD *thd, set_var *var)
{
DBUG_ASSERT(FALSE);
}
uchar *session_value_ptr(THD *thd, LEX_STRING *base)
{
return (uchar*) &session_var(thd, LEX_STRING);
}
uchar *global_value_ptr(THD *thd, LEX_STRING *base)
{
DBUG_ASSERT(FALSE);
}
bool check_update_type(Item_result type)
{ return type != STRING_RESULT; }
};
#ifndef DBUG_OFF
/**
@@session.dbug and @@global.dbug variables.
@ -1389,6 +1471,7 @@ public:
};
#endif /* defined(ENABLED_DEBUG_SYNC) */
/**
The class for bit variables - a variant of boolean that stores the value
in a bit.
@ -1862,6 +1945,65 @@ public:
bool global_update(THD *thd, set_var *var);
};
/*
Class for handing multi-source replication variables
Variable values are store in Master_info, but to make it possible to
access variable without locks we also store it thd->variables.
These can be used as GLOBAL or SESSION, but both points to the same
variable. This is to make things compatible with MySQL 5.5 where variables
like sql_slave_skip_counter are GLOBAL.
*/
class Sys_var_multi_source_uint :public Sys_var_uint
{
ptrdiff_t master_info_offset;
public:
Sys_var_multi_source_uint(const char *name_arg,
const char *comment, int flag_args,
ptrdiff_t off, size_t size,
ptrdiff_t master_info_offset_arg,
uint min_val, uint max_val, uint def_val,
uint block_size,
on_update_function on_update_func)
:Sys_var_uint(name_arg, comment, flag_args, off, size,
NO_CMD_LINE, min_val, max_val, def_val, block_size,
0, VARIABLE_NOT_IN_BINLOG, 0, on_update_func),
master_info_offset(master_info_offset_arg)
{
/* No global storage of variables. Cause a crash if we try an update */
option.value= (uchar**)1;
}
bool session_update(THD *thd, set_var *var)
{
session_var(thd, uint)= (uint) (var->save_result.ulonglong_value);
/* Value should be moved to multi_master in on_update_func */
return false;
}
bool global_update(THD *thd, set_var *var)
{
return session_update(thd, var);
}
void session_save_default(THD *thd, set_var *var)
{
/* Use value given in variable declaration */
global_save_default(thd, var);
}
uchar *session_value_ptr(THD *thd,LEX_STRING *base)
{
uint *tmp, res;
tmp= (uint*) (((uchar*)&(thd->variables)) + offset);
res= get_master_info_uint_value(thd, master_info_offset);
*tmp= res;
return (uchar*) tmp;
}
uchar *global_value_ptr(THD *thd, LEX_STRING *base)
{
return session_value_ptr(thd, base);
}
uint get_master_info_uint_value(THD *thd, ptrdiff_t offset);
};
/****************************************************************************
Used templates
****************************************************************************/