From 8249fd6eef37dcbfc7f37359998c41e96aabee03 Mon Sep 17 00:00:00 2001 From: Mats Kindahl Date: Wed, 23 Sep 2009 11:43:43 +0200 Subject: BUG#29288: myisam transactions replicated to a transactional slave leaves slave unstable Problem: when replicating from non-transactional to transactional engine with autocommit off, no BEGIN/COMMIT is written to the binlog. When the slave replicates, it will start a transaction that never ends. Fix: Force autocommit=on on slave by always replicating autocommit=1 from the master. --- sql/log_event.cc | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) (limited to 'sql') diff --git a/sql/log_event.cc b/sql/log_event.cc index 0cda724b698..fb6a5230fda 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -2389,13 +2389,29 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg, charset_database_number= thd_arg->variables.collation_database->number; /* - If we don't use flags2 for anything else than options contained in - thd_arg->options, it would be more efficient to flags2=thd_arg->options - (OPTIONS_WRITTEN_TO_BIN_LOG would be used only at reading time). - But it's likely that we don't want to use 32 bits for 3 bits; in the future - we will probably want to reclaim the 29 bits. So we need the &. + We only replicate over the bits of flags2 that we need: the rest + are masked out by "& OPTIONS_WRITTEN_TO_BINLOG". + + We also force AUTOCOMMIT=1. Rationale (cf. BUG#29288): After + fixing BUG#26395, we always write BEGIN and COMMIT around all + transactions (even single statements in autocommit mode). This is + so that replication from non-transactional to transactional table + and error recovery from XA to non-XA table should work as + expected. The BEGIN/COMMIT are added in log.cc. However, there is + one exception: MyISAM bypasses log.cc and writes directly to the + binlog. So if autocommit is off, master has MyISAM, and slave has + a transactional engine, then the slave will just see one long + never-ending transaction. The only way to bypass explicit + BEGIN/COMMIT in the binlog is by using a non-transactional table. + So setting AUTOCOMMIT=1 will make this work as expected. + + Note: explicitly replicate AUTOCOMMIT=1 from master. We do not + assume AUTOCOMMIT=1 on slave; the slave still reads the state of + the autocommit flag as written by the master to the binlog. This + behavior may change after WL#4162 has been implemented. */ - flags2= (uint32) (thd_arg->options & OPTIONS_WRITTEN_TO_BIN_LOG); + flags2= (uint32) (thd_arg->options & + (OPTIONS_WRITTEN_TO_BIN_LOG & ~OPTION_NOT_AUTOCOMMIT)); DBUG_ASSERT(thd_arg->variables.character_set_client->number < 256*256); DBUG_ASSERT(thd_arg->variables.collation_connection->number < 256*256); DBUG_ASSERT(thd_arg->variables.collation_server->number < 256*256); -- cgit v1.2.1 From 124e830125a2964a60d01cf9df68ddc9caa70a0d Mon Sep 17 00:00:00 2001 From: Mats Kindahl Date: Wed, 23 Sep 2009 13:20:48 +0200 Subject: Bug #37221: SET AUTOCOMMIT=1 does not commit binary log When setting AUTOCOMMIT=1 after starting a transaction, the binary log did not commit the outstanding transaction. The reason was that the binary log commit function saw the values of the new settings, deciding that there were nothing to commit. Fixed the problem by moving the implicit commit to before the thread option flags were changed, so that the binary log sees the old values of the flags instead of the values they will take after the statement. mysql-test/extra/binlog_tests/implicit.test: New test file to check implicit commits both inside and outside transactions. mysql-test/suite/binlog/t/binlog_implicit_commit.test: Test for implicit commit of SET AUTOCOMMIT and LOCK/UNLOCK TABLES. sql/set_var.cc: Adding code to commit pending transaction before changing option flags. --- sql/set_var.cc | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) (limited to 'sql') diff --git a/sql/set_var.cc b/sql/set_var.cc index 0b89333ce03..5025356230c 100644 --- a/sql/set_var.cc +++ b/sql/set_var.cc @@ -3054,6 +3054,15 @@ static bool set_option_autocommit(THD *thd, set_var *var) ulonglong org_options= thd->options; + /* + If we are setting AUTOCOMMIT=1 and it was not already 1, then we + need to commit any outstanding transactions. + */ + if (var->save_result.ulong_value != 0 && + (thd->options & OPTION_NOT_AUTOCOMMIT) && + ha_commit(thd)) + return 1; + if (var->save_result.ulong_value != 0) thd->options&= ~((sys_var_thd_bit*) var->var)->bit_flag; else @@ -3067,8 +3076,6 @@ static bool set_option_autocommit(THD *thd, set_var *var) thd->options&= ~(ulonglong) (OPTION_BEGIN | OPTION_KEEP_LOG); thd->transaction.all.modified_non_trans_table= FALSE; thd->server_status|= SERVER_STATUS_AUTOCOMMIT; - if (ha_commit(thd)) - return 1; } else { -- cgit v1.2.1 From d47710c8dccd295428fa7547720f4a5d7311c413 Mon Sep 17 00:00:00 2001 From: Mats Kindahl Date: Wed, 23 Sep 2009 23:32:31 +0200 Subject: WL#5016: Fix header file include guards Adding header include file guards to files that are missing such. --- sql/authors.h | 5 +++++ sql/client_settings.h | 6 ++++++ sql/contributors.h | 5 +++++ sql/field.h | 5 +++++ sql/gstream.h | 5 +++++ sql/ha_ndbcluster.h | 5 +++++ sql/ha_ndbcluster_binlog.h | 5 +++++ sql/ha_ndbcluster_cond.h | 5 +++++ sql/ha_ndbcluster_tables.h | 5 +++++ sql/ha_partition.h | 5 +++++ sql/handler.h | 4 ++++ sql/item.h | 5 +++++ sql/item_cmpfunc.h | 5 +++++ sql/item_func.h | 4 ++++ sql/item_geofunc.h | 4 ++++ sql/item_row.h | 5 +++++ sql/item_strfunc.h | 4 ++++ sql/item_subselect.h | 5 ++++- sql/item_sum.h | 5 +++++ sql/item_timefunc.h | 5 +++++ sql/item_xmlfunc.h | 4 ++++ sql/lex.h | 5 +++++ sql/message.h | 7 +++++++ sql/mysqld_suffix.h | 4 ++++ sql/nt_servc.h | 5 +++++ sql/partition_element.h | 5 +++++ sql/partition_info.h | 5 +++++ sql/procedure.h | 5 +++++ sql/protocol.h | 4 ++++ sql/repl_failsafe.h | 4 ++++ sql/scheduler.h | 5 +++++ sql/set_var.h | 5 +++++ sql/sql_acl.h | 4 ++++ sql/sql_analyse.h | 5 +++++ sql/sql_array.h | 4 ++++ sql/sql_class.h | 4 ++++ sql/sql_crypt.h | 5 +++++ sql/sql_lex.h | 4 ++++ sql/sql_map.h | 5 +++++ sql/sql_partition.h | 4 ++++ sql/sql_repl.h | 4 ++++ sql/sql_select.h | 4 ++++ sql/sql_servers.h | 5 +++++ sql/sql_sort.h | 5 +++++ sql/sql_string.h | 5 +++++ sql/sql_trigger.h | 4 ++++ sql/sql_udf.h | 4 ++++ sql/sql_view.h | 4 ++++ sql/structs.h | 5 +++++ sql/table.h | 4 ++++ sql/tzfile.h | 5 +++++ sql/tztime.h | 4 ++++ sql/unireg.h | 5 +++-- 53 files changed, 245 insertions(+), 3 deletions(-) (limited to 'sql') diff --git a/sql/authors.h b/sql/authors.h index dfe3b143e2f..90cdae9c0a0 100644 --- a/sql/authors.h +++ b/sql/authors.h @@ -1,3 +1,6 @@ +#ifndef AUTHORS_INCLUDED +#define AUTHORS_INCLUDED + /* Copyright (C) 2005-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -150,3 +153,5 @@ struct show_table_authors_st show_table_authors[]= { "SHA1(), AES_ENCRYPT(), AES_DECRYPT(), bug fixing" }, {NULL, NULL, NULL} }; + +#endif /* AUTHORS_INCLUDED */ diff --git a/sql/client_settings.h b/sql/client_settings.h index 4f06c15a29e..fd50bfdbb88 100644 --- a/sql/client_settings.h +++ b/sql/client_settings.h @@ -14,6 +14,12 @@ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ +#ifndef CLIENT_SETTINGS_INCLUDED +#define CLIENT_SETTINGS_INCLUDED +#else +#error You have already included an client_settings.h and it should not be included twice +#endif /* CLIENT_SETTINGS_INCLUDED */ + #include #define CLIENT_CAPABILITIES (CLIENT_LONG_PASSWORD | CLIENT_LONG_FLAG | \ diff --git a/sql/contributors.h b/sql/contributors.h index 87001e29d88..6cf8bb88e3b 100644 --- a/sql/contributors.h +++ b/sql/contributors.h @@ -1,3 +1,6 @@ +#ifndef CONTRIBUTORS_INCLUDED +#define CONTRIBUTORS_INCLUDED + /* Copyright (C) 2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -37,3 +40,5 @@ struct show_table_contributors_st show_table_contributors[]= { {"Mark Shuttleworth", "London, UK.", "EFF contribution for UC2006 Auction"}, {NULL, NULL, NULL} }; + +#endif /* CONTRIBUTORS_INCLUDED */ diff --git a/sql/field.h b/sql/field.h index a9299256f88..36569428ee0 100644 --- a/sql/field.h +++ b/sql/field.h @@ -1,3 +1,6 @@ +#ifndef FIELD_INCLUDED +#define FIELD_INCLUDED + /* Copyright 2000-2008 MySQL AB, 2008 Sun Microsystems, Inc. This program is free software; you can redistribute it and/or modify @@ -2180,3 +2183,5 @@ int set_field_to_null_with_conversions(Field *field, bool no_conversions); #define f_no_default(x) (x & FIELDFLAG_NO_DEFAULT) #define f_bit_as_char(x) ((x) & FIELDFLAG_TREAT_BIT_AS_CHAR) #define f_is_hex_escape(x) ((x) & FIELDFLAG_HEX_ESCAPE) + +#endif /* FIELD_INCLUDED */ diff --git a/sql/gstream.h b/sql/gstream.h index 1ef90ad5bf0..ea7158ee1a3 100644 --- a/sql/gstream.h +++ b/sql/gstream.h @@ -1,3 +1,6 @@ +#ifndef GSTREAM_INCLUDED +#define GSTREAM_INCLUDED + /* Copyright (C) 2000-2004 MySQL AB This program is free software; you can redistribute it and/or modify @@ -73,3 +76,5 @@ protected: char *m_err_msg; CHARSET_INFO *m_charset; }; + +#endif /* GSTREAM_INCLUDED */ diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h index a17323d3fd6..662655c42b9 100644 --- a/sql/ha_ndbcluster.h +++ b/sql/ha_ndbcluster.h @@ -1,3 +1,6 @@ +#ifndef HA_NDBCLUSTER_INCLUDED +#define HA_NDBCLUSTER_INCLUDED + /* Copyright (C) 2000-2003 MySQL AB This program is free software; you can redistribute it and/or modify @@ -581,3 +584,5 @@ static const int ndbcluster_hton_name_length=sizeof(ndbcluster_hton_name)-1; extern int ndbcluster_terminating; extern int ndb_util_thread_running; extern pthread_cond_t COND_ndb_util_ready; + +#endif /* HA_NDBCLUSTER_INCLUDED */ diff --git a/sql/ha_ndbcluster_binlog.h b/sql/ha_ndbcluster_binlog.h index 1cad643e5ec..d80dfe9ee74 100644 --- a/sql/ha_ndbcluster_binlog.h +++ b/sql/ha_ndbcluster_binlog.h @@ -1,3 +1,6 @@ +#ifndef HA_NDBCLUSTER_BINLOG_INCLUDED +#define HA_NDBCLUSTER_BINLOG_INCLUDED + /* Copyright (C) 2000-2003 MySQL AB This program is free software; you can redistribute it and/or modify @@ -225,3 +228,5 @@ set_thd_ndb(THD *thd, Thd_ndb *thd_ndb) { thd_set_ha_data(thd, ndbcluster_hton, thd_ndb); } Ndb* check_ndb_in_thd(THD* thd); + +#endif /* HA_NDBCLUSTER_BINLOG_INCLUDED */ diff --git a/sql/ha_ndbcluster_cond.h b/sql/ha_ndbcluster_cond.h index 4401a93c9e1..4ccc7e062ec 100644 --- a/sql/ha_ndbcluster_cond.h +++ b/sql/ha_ndbcluster_cond.h @@ -1,3 +1,6 @@ +#ifndef HA_NDBCLUSTER_COND_INCLUDED +#define HA_NDBCLUSTER_COND_INCLUDED + /* Copyright (C) 2000-2007 MySQL AB This program is free software; you can redistribute it and/or modify @@ -486,3 +489,5 @@ private: Ndb_cond_stack *m_cond_stack; }; + +#endif /* HA_NDBCLUSTER_COND_INCLUDED */ diff --git a/sql/ha_ndbcluster_tables.h b/sql/ha_ndbcluster_tables.h index c6bc8f577f8..ba2e8ec251b 100644 --- a/sql/ha_ndbcluster_tables.h +++ b/sql/ha_ndbcluster_tables.h @@ -1,3 +1,6 @@ +#ifndef HA_NDBCLUSTER_TABLES_INCLUDED +#define HA_NDBCLUSTER_TABLES_INCLUDED + /* Copyright (C) 2000-2003 MySQL AB This program is free software; you can redistribute it and/or modify @@ -21,3 +24,5 @@ #define OLD_NDB_APPLY_TABLE "apply_status" #define NDB_SCHEMA_TABLE "ndb_schema" #define OLD_NDB_SCHEMA_TABLE "schema" + +#endif /* HA_NDBCLUSTER_TABLES_INCLUDED */ diff --git a/sql/ha_partition.h b/sql/ha_partition.h index 8a81a759e2a..804db028953 100644 --- a/sql/ha_partition.h +++ b/sql/ha_partition.h @@ -1,3 +1,6 @@ +#ifndef HA_PARTITION_INCLUDED +#define HA_PARTITION_INCLUDED + /* Copyright 2005-2008 MySQL AB, 2008 Sun Microsystems, Inc. This program is free software; you can redistribute it and/or modify @@ -1090,3 +1093,5 @@ public: virtual void append_create_info(String *packet) */ }; + +#endif /* HA_PARTITION_INCLUDED */ diff --git a/sql/handler.h b/sql/handler.h index f759239d66e..a281aaa0ad7 100644 --- a/sql/handler.h +++ b/sql/handler.h @@ -1,3 +1,6 @@ +#ifndef HANDLER_INCLUDED +#define HANDLER_INCLUDED + /* Copyright 2000-2008 MySQL AB, 2008 Sun Microsystems, Inc. This program is free software; you can redistribute it and/or modify @@ -2064,3 +2067,4 @@ int ha_binlog_end(THD *thd); #define ha_binlog_wait(a) do {} while (0) #define ha_binlog_end(a) do {} while (0) #endif +#endif /* HANDLER_INCLUDED */ diff --git a/sql/item.h b/sql/item.h index a2cff3ab3a9..b44e84f4b15 100644 --- a/sql/item.h +++ b/sql/item.h @@ -1,3 +1,6 @@ +#ifndef ITEM_INCLUDED +#define ITEM_INCLUDED + /* Copyright 2000-2008 MySQL AB, 2008 Sun Microsystems, Inc. This program is free software; you can redistribute it and/or modify @@ -3126,3 +3129,5 @@ extern Cached_item *new_Cached_item(THD *thd, Item *item); extern Item_result item_cmp_type(Item_result a,Item_result b); extern void resolve_const_item(THD *thd, Item **ref, Item *cmp_item); extern int stored_field_cmp_to_item(Field *field, Item *item); + +#endif /* ITEM_INCLUDED */ diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h index c2227fa04e0..3462bed94a2 100644 --- a/sql/item_cmpfunc.h +++ b/sql/item_cmpfunc.h @@ -1,3 +1,6 @@ +#ifndef ITEM_CMPFUNC_INCLUDED +#define ITEM_CMPFUNC_INCLUDED + /* Copyright (C) 2000-2003 MySQL AB This program is free software; you can redistribute it and/or modify @@ -1721,3 +1724,5 @@ inline Item *and_conds(Item *a, Item *b) } Item *and_expressions(Item *a, Item *b, Item **org_item); + +#endif /* ITEM_CMPFUNC_INCLUDED */ diff --git a/sql/item_func.h b/sql/item_func.h index fdbbff89e60..628878bcaed 100644 --- a/sql/item_func.h +++ b/sql/item_func.h @@ -1,3 +1,6 @@ +#ifndef ITEM_FUNC_INCLUDED +#define ITEM_FUNC_INCLUDED + /* Copyright 2000-2008 MySQL AB, 2008 Sun Microsystems, Inc. This program is free software; you can redistribute it and/or modify @@ -1718,3 +1721,4 @@ public: bool check_partition_func_processor(uchar *int_arg) {return FALSE;} }; +#endif /* ITEM_FUNC_INCLUDED */ diff --git a/sql/item_geofunc.h b/sql/item_geofunc.h index edbe104e307..9a55ea7d5b1 100644 --- a/sql/item_geofunc.h +++ b/sql/item_geofunc.h @@ -1,3 +1,6 @@ +#ifndef ITEM_GEOFUNC_INCLUDED +#define ITEM_GEOFUNC_INCLUDED + /* Copyright (C) 2000-2003 MySQL AB This program is free software; you can redistribute it and/or modify @@ -386,3 +389,4 @@ public: #endif +#endif /* ITEM_GEOFUNC_INCLUDED */ diff --git a/sql/item_row.h b/sql/item_row.h index 67441f49603..7c08c5888e0 100644 --- a/sql/item_row.h +++ b/sql/item_row.h @@ -1,3 +1,6 @@ +#ifndef ITEM_ROW_INCLUDED +#define ITEM_ROW_INCLUDED + /* Copyright (C) 2000 MySQL AB This program is free software; you can redistribute it and/or modify @@ -77,3 +80,5 @@ public: bool null_inside() { return with_null; }; void bring_value(); }; + +#endif /* ITEM_ROW_INCLUDED */ diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h index 2cdb45100ae..fc70b04ca68 100644 --- a/sql/item_strfunc.h +++ b/sql/item_strfunc.h @@ -1,3 +1,6 @@ +#ifndef ITEM_STRFUNC_INCLUDED +#define ITEM_STRFUNC_INCLUDED + /* Copyright (C) 2000-2003 MySQL AB This program is free software; you can redistribute it and/or modify @@ -842,3 +845,4 @@ public: String *val_str(String *); }; +#endif /* ITEM_STRFUNC_INCLUDED */ diff --git a/sql/item_subselect.h b/sql/item_subselect.h index d4aa621c083..ea59521aab1 100644 --- a/sql/item_subselect.h +++ b/sql/item_subselect.h @@ -1,3 +1,6 @@ +#ifndef ITEM_SUBSELECT_INCLUDED +#define ITEM_SUBSELECT_INCLUDED + /* Copyright (C) 2000 MySQL AB This program is free software; you can redistribute it and/or modify @@ -579,4 +582,4 @@ inline bool Item_subselect::is_uncacheable() const return engine->uncacheable(); } - +#endif /* ITEM_SUBSELECT_INCLUDED */ diff --git a/sql/item_sum.h b/sql/item_sum.h index d991327d847..993ec1597b4 100644 --- a/sql/item_sum.h +++ b/sql/item_sum.h @@ -1,3 +1,6 @@ +#ifndef ITEM_SUM_INCLUDED +#define ITEM_SUM_INCLUDED + /* Copyright (C) 2000-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -1279,3 +1282,5 @@ public: virtual bool change_context_processor(uchar *cntx) { context= (Name_resolution_context *)cntx; return FALSE; } }; + +#endif /* ITEM_SUM_INCLUDED */ diff --git a/sql/item_timefunc.h b/sql/item_timefunc.h index 9e3c2e8c89f..7f1b2ed3a53 100644 --- a/sql/item_timefunc.h +++ b/sql/item_timefunc.h @@ -1,3 +1,6 @@ +#ifndef ITEM_TIMEFUNC_INCLUDED +#define ITEM_TIMEFUNC_INCLUDED + /* Copyright (C) 2000-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -1026,3 +1029,5 @@ public: const char *func_name() const { return "last_day"; } bool get_date(MYSQL_TIME *res, uint fuzzy_date); }; + +#endif /* ITEM_TIMEFUNC_INCLUDED */ diff --git a/sql/item_xmlfunc.h b/sql/item_xmlfunc.h index dadbb5ccf42..6373bde0aab 100644 --- a/sql/item_xmlfunc.h +++ b/sql/item_xmlfunc.h @@ -1,3 +1,6 @@ +#ifndef ITEM_XMLFUNC_INCLUDED +#define ITEM_XMLFUNC_INCLUDED + /* Copyright (C) 2000-2005 MySQL AB This program is free software; you can redistribute it and/or modify @@ -61,3 +64,4 @@ public: String *val_str(String *); }; +#endif /* ITEM_XMLFUNC_INCLUDED */ diff --git a/sql/lex.h b/sql/lex.h index 0a85824f6f7..b199a79350b 100644 --- a/sql/lex.h +++ b/sql/lex.h @@ -1,3 +1,6 @@ +#ifndef LEX_INCLUDED +#define LEX_INCLUDED + /* Copyright (C) 2000-2002 MySQL AB This program is free software; you can redistribute it and/or modify @@ -634,3 +637,5 @@ static SYMBOL sql_functions[] = { { "VAR_POP", SYM(VARIANCE_SYM)}, { "VAR_SAMP", SYM(VAR_SAMP_SYM)}, }; + +#endif /* LEX_INCLUDED */ diff --git a/sql/message.h b/sql/message.h index 0e7c282d5a1..97d039352b4 100644 --- a/sql/message.h +++ b/sql/message.h @@ -1,3 +1,6 @@ +#ifndef MESSAGE_INCLUDED +#define MESSAGE_INCLUDED + /* To change or add messages mysqld writes to the Windows error log, run mc.exe message.mc @@ -6,6 +9,8 @@ mc.exe can be installed with Windows SDK, some Visual Studio distributions do not include it. */ + + // // Values are 32 bit values layed out as follows: // @@ -53,3 +58,5 @@ // #define MSG_DEFAULT 0xC0000064L +#endif /* MESSAGE_INCLUDED */ + diff --git a/sql/mysqld_suffix.h b/sql/mysqld_suffix.h index 654d7cf88c1..c7ab212f2a2 100644 --- a/sql/mysqld_suffix.h +++ b/sql/mysqld_suffix.h @@ -1,3 +1,6 @@ +#ifndef MYSQLD_SUFFIX_INCLUDED +#define MYSQLD_SUFFIX_INCLUDED + /* Copyright (C) 2000-2004 MySQL AB This program is free software; you can redistribute it and/or modify @@ -27,3 +30,4 @@ #else #define MYSQL_SERVER_SUFFIX_STR MYSQL_SERVER_SUFFIX_DEF #endif +#endif /* MYSQLD_SUFFIX_INCLUDED */ diff --git a/sql/nt_servc.h b/sql/nt_servc.h index 2f0d07df543..5bee42dedf0 100644 --- a/sql/nt_servc.h +++ b/sql/nt_servc.h @@ -1,3 +1,6 @@ +#ifndef NT_SERVC_INCLUDED +#define NT_SERVC_INCLUDED + /** @file @@ -98,3 +101,5 @@ class NTService }; /* ------------------------- the end -------------------------------------- */ + +#endif /* NT_SERVC_INCLUDED */ diff --git a/sql/partition_element.h b/sql/partition_element.h index 905bc38165b..d67259b9605 100644 --- a/sql/partition_element.h +++ b/sql/partition_element.h @@ -1,3 +1,6 @@ +#ifndef PARTITION_ELEMENT_INCLUDED +#define PARTITION_ELEMENT_INCLUDED + /* Copyright (C) 2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -97,3 +100,5 @@ public: } ~partition_element() {} }; + +#endif /* PARTITION_ELEMENT_INCLUDED */ diff --git a/sql/partition_info.h b/sql/partition_info.h index 9f438e8260b..b5a6c4a0961 100644 --- a/sql/partition_info.h +++ b/sql/partition_info.h @@ -1,3 +1,6 @@ +#ifndef PARTITION_INFO_INCLUDED +#define PARTITION_INFO_INCLUDED + /* Copyright 2006-2008 MySQL AB, 2008 Sun Microsystems, Inc. This program is free software; you can redistribute it and/or modify @@ -314,3 +317,5 @@ void init_all_partitions_iterator(partition_info *part_info, part_iter->ret_null_part= part_iter->ret_null_part_orig= FALSE; part_iter->get_next= get_next_partition_id_range; } + +#endif /* PARTITION_INFO_INCLUDED */ diff --git a/sql/procedure.h b/sql/procedure.h index ceb586766b1..c6f50493876 100644 --- a/sql/procedure.h +++ b/sql/procedure.h @@ -1,3 +1,6 @@ +#ifndef PROCEDURE_INCLUDED +#define PROCEDURE_INCLUDED + /* Copyright (C) 2000-2005 MySQL AB This program is free software; you can redistribute it and/or modify @@ -149,3 +152,5 @@ public: Procedure *setup_procedure(THD *thd,ORDER *proc_param,select_result *result, List &field_list,int *error); + +#endif /* PROCEDURE_INCLUDED */ diff --git a/sql/protocol.h b/sql/protocol.h index 251ba6fbc33..a39636a1595 100644 --- a/sql/protocol.h +++ b/sql/protocol.h @@ -1,3 +1,6 @@ +#ifndef PROTOCOL_INCLUDED +#define PROTOCOL_INCLUDED + /* Copyright (C) 2002-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -180,3 +183,4 @@ uchar *net_store_data(uchar *to,const uchar *from, size_t length); uchar *net_store_data(uchar *to,int32 from); uchar *net_store_data(uchar *to,longlong from); +#endif /* PROTOCOL_INCLUDED */ diff --git a/sql/repl_failsafe.h b/sql/repl_failsafe.h index 6ff78067aca..bce2c727050 100644 --- a/sql/repl_failsafe.h +++ b/sql/repl_failsafe.h @@ -1,3 +1,6 @@ +#ifndef REPL_FAILSAFE_INCLUDED +#define REPL_FAILSAFE_INCLUDED + /* Copyright (C) 2001-2005 MySQL AB & Sasha This program is free software; you can redistribute it and/or modify @@ -49,3 +52,4 @@ int register_slave(THD* thd, uchar* packet, uint packet_length); void unregister_slave(THD* thd, bool only_mine, bool need_mutex); #endif /* HAVE_REPLICATION */ +#endif /* REPL_FAILSAFE_INCLUDED */ diff --git a/sql/scheduler.h b/sql/scheduler.h index 46bbd300cbb..e7916031a27 100644 --- a/sql/scheduler.h +++ b/sql/scheduler.h @@ -1,3 +1,6 @@ +#ifndef SCHEDULER_INCLUDED +#define SCHEDULER_INCLUDED + /* Copyright (C) 2007 MySQL AB This program is free software; you can redistribute it and/or modify @@ -58,3 +61,5 @@ enum pool_command_op class thd_scheduler {}; + +#endif /* SCHEDULER_INCLUDED */ diff --git a/sql/set_var.h b/sql/set_var.h index 10e6e0f9c35..a3ed8e5be15 100644 --- a/sql/set_var.h +++ b/sql/set_var.h @@ -1,3 +1,6 @@ +#ifndef SET_VAR_INCLUDED +#define SET_VAR_INCLUDED + /* Copyright (C) 2002-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -1449,3 +1452,5 @@ void free_key_cache(const char *name, KEY_CACHE *key_cache); bool process_key_caches(process_key_cache_t func); void delete_elements(I_List *list, void (*free_element)(const char*, uchar*)); + +#endif /* SET_VAR_INCLUDED */ diff --git a/sql/sql_acl.h b/sql/sql_acl.h index a8090fba2e7..c0622bd747c 100644 --- a/sql/sql_acl.h +++ b/sql/sql_acl.h @@ -1,3 +1,6 @@ +#ifndef SQL_ACL_INCLUDED +#define SQL_ACL_INCLUDED + /* Copyright (C) 2000-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -273,3 +276,4 @@ bool is_acl_user(const char *host, const char *user); #define check_grant(A,B,C,D,E,F) 0 #define check_grant_db(A,B) 0 #endif +#endif /* SQL_ACL_INCLUDED */ diff --git a/sql/sql_analyse.h b/sql/sql_analyse.h index 8807b40857e..8f52b90c874 100644 --- a/sql/sql_analyse.h +++ b/sql/sql_analyse.h @@ -1,3 +1,6 @@ +#ifndef SQL_ANALYSE_INCLUDED +#define SQL_ANALYSE_INCLUDED + /* Copyright (C) 2000-2003, 2005 MySQL AB This program is free software; you can redistribute it and/or modify @@ -355,3 +358,5 @@ public: select_result *result, List &field_list); }; + +#endif /* SQL_ANALYSE_INCLUDED */ diff --git a/sql/sql_array.h b/sql/sql_array.h index e1b22921519..dfaa9b02947 100644 --- a/sql/sql_array.h +++ b/sql/sql_array.h @@ -1,3 +1,6 @@ +#ifndef SQL_ARRAY_INCLUDED +#define SQL_ARRAY_INCLUDED + /* Copyright (C) 2003 MySQL AB This program is free software; you can redistribute it and/or modify @@ -66,3 +69,4 @@ public: } }; +#endif /* SQL_ARRAY_INCLUDED */ diff --git a/sql/sql_class.h b/sql/sql_class.h index f52d5fae76f..49f9bc3fd5e 100644 --- a/sql/sql_class.h +++ b/sql/sql_class.h @@ -14,6 +14,9 @@ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ +#ifndef SQL_CLASS_INCLUDED +#define SQL_CLASS_INCLUDED + /* Classes in mysql */ #ifdef USE_PRAGMA_INTERFACE @@ -3003,3 +3006,4 @@ void add_diff_to_status(STATUS_VAR *to_var, STATUS_VAR *from_var, void mark_transaction_to_rollback(THD *thd, bool all); #endif /* MYSQL_SERVER */ +#endif /* SQL_CLASS_INCLUDED */ diff --git a/sql/sql_crypt.h b/sql/sql_crypt.h index a5a6bee8a58..8d5a761cbdf 100644 --- a/sql/sql_crypt.h +++ b/sql/sql_crypt.h @@ -1,3 +1,6 @@ +#ifndef SQL_CRYPT_INCLUDED +#define SQL_CRYPT_INCLUDED + /* Copyright (C) 2000-2001, 2005 MySQL AB This program is free software; you can redistribute it and/or modify @@ -35,3 +38,5 @@ class SQL_CRYPT :public Sql_alloc void encode(char *str, uint length); void decode(char *str, uint length); }; + +#endif /* SQL_CRYPT_INCLUDED */ diff --git a/sql/sql_lex.h b/sql/sql_lex.h index 76fd5354c51..6f9f667a75a 100644 --- a/sql/sql_lex.h +++ b/sql/sql_lex.h @@ -17,6 +17,9 @@ @defgroup Semantic_Analysis Semantic Analysis */ +#ifndef SQL_LEX_INCLUDED +#define SQL_LEX_INCLUDED + /* YACC and LEX Definitions */ /* These may not be declared yet */ @@ -1979,3 +1982,4 @@ extern bool is_lex_native_function(const LEX_STRING *name); int my_missing_function_error(const LEX_STRING &token, const char *name); #endif /* MYSQL_SERVER */ +#endif /* SQL_LEX_INCLUDED */ diff --git a/sql/sql_map.h b/sql/sql_map.h index a1efba0da6f..5ae260841e0 100644 --- a/sql/sql_map.h +++ b/sql/sql_map.h @@ -1,3 +1,6 @@ +#ifndef SQL_MAP_INCLUDED +#define SQL_MAP_INCLUDED + /* Copyright (C) 2000-2001, 2005 MySQL AB This program is free software; you can redistribute it and/or modify @@ -60,3 +63,5 @@ public: return file->map; } }; + +#endif /* SQL_MAP_INCLUDED */ diff --git a/sql/sql_partition.h b/sql/sql_partition.h index 282e24f1853..0c47340016c 100644 --- a/sql/sql_partition.h +++ b/sql/sql_partition.h @@ -1,3 +1,6 @@ +#ifndef SQL_PARTITION_INCLUDED +#define SQL_PARTITION_INCLUDED + /* Copyright (C) 2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -209,3 +212,4 @@ typedef int (*get_partitions_in_range_iter)(partition_info *part_info, #include "partition_info.h" +#endif /* SQL_PARTITION_INCLUDED */ diff --git a/sql/sql_repl.h b/sql/sql_repl.h index d5c9040f8dc..aa71ac96ff8 100644 --- a/sql/sql_repl.h +++ b/sql/sql_repl.h @@ -13,6 +13,9 @@ along with this program; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ +#ifndef SQL_REPL_INCLUDED +#define SQL_REPL_INCLUDED + #include "rpl_filter.h" #ifdef HAVE_REPLICATION @@ -65,3 +68,4 @@ int init_replication_sys_vars(); #endif /* HAVE_REPLICATION */ +#endif /* SQL_REPL_INCLUDED */ diff --git a/sql/sql_select.h b/sql/sql_select.h index a0366d47149..12cfebfc374 100644 --- a/sql/sql_select.h +++ b/sql/sql_select.h @@ -1,3 +1,6 @@ +#ifndef SQL_SELECT_INCLUDED +#define SQL_SELECT_INCLUDED + /* Copyright (C) 2000-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -736,3 +739,4 @@ inline bool optimizer_flag(THD *thd, uint flag) return (thd->variables.optimizer_switch & flag); } +#endif /* SQL_SELECT_INCLUDED */ diff --git a/sql/sql_servers.h b/sql/sql_servers.h index 63c691893d1..12855f8473c 100644 --- a/sql/sql_servers.h +++ b/sql/sql_servers.h @@ -1,3 +1,6 @@ +#ifndef SQL_SERVERS_INCLUDED +#define SQL_SERVERS_INCLUDED + /* Copyright (C) 2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -41,3 +44,5 @@ int alter_server(THD *thd, LEX_SERVER_OPTIONS *server_options); /* lookup functions */ FOREIGN_SERVER *get_server_by_name(MEM_ROOT *mem, const char *server_name, FOREIGN_SERVER *server_buffer); + +#endif /* SQL_SERVERS_INCLUDED */ diff --git a/sql/sql_sort.h b/sql/sql_sort.h index 1e9322f7f5b..102b3ef0a11 100644 --- a/sql/sql_sort.h +++ b/sql/sql_sort.h @@ -1,3 +1,6 @@ +#ifndef SQL_SORT_INCLUDED +#define SQL_SORT_INCLUDED + /* Copyright (C) 2000 MySQL AB This program is free software; you can redistribute it and/or modify @@ -87,3 +90,5 @@ int merge_buffers(SORTPARAM *param,IO_CACHE *from_file, BUFFPEK *lastbuff,BUFFPEK *Fb, BUFFPEK *Tb,int flag); void reuse_freed_buff(QUEUE *queue, BUFFPEK *reuse, uint key_length); + +#endif /* SQL_SORT_INCLUDED */ diff --git a/sql/sql_string.h b/sql/sql_string.h index d62908e5d66..7b10aafbff1 100644 --- a/sql/sql_string.h +++ b/sql/sql_string.h @@ -1,3 +1,6 @@ +#ifndef SQL_STRING_INCLUDED +#define SQL_STRING_INCLUDED + /* Copyright (C) 2000 MySQL AB This program is free software; you can redistribute it and/or modify @@ -389,3 +392,5 @@ static inline bool check_if_only_end_space(CHARSET_INFO *cs, char *str, { return str+ cs->cset->scan(cs, str, end, MY_SEQ_SPACES) == end; } + +#endif /* SQL_STRING_INCLUDED */ diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h index f6754a75284..b411acf2ac5 100644 --- a/sql/sql_trigger.h +++ b/sql/sql_trigger.h @@ -1,3 +1,6 @@ +#ifndef SQL_TRIGGER_INCLUDED +#define SQL_TRIGGER_INCLUDED + /* Copyright (C) 2004-2005 MySQL AB This program is free software; you can redistribute it and/or modify @@ -174,3 +177,4 @@ bool load_table_name_for_trigger(THD *thd, const LEX_STRING *trn_path, LEX_STRING *tbl_name); +#endif /* SQL_TRIGGER_INCLUDED */ diff --git a/sql/sql_udf.h b/sql/sql_udf.h index 4b8b492698e..95cb167869e 100644 --- a/sql/sql_udf.h +++ b/sql/sql_udf.h @@ -1,3 +1,6 @@ +#ifndef SQL_UDF_INCLUDED +#define SQL_UDF_INCLUDED + /* Copyright (C) 2000-2001, 2003-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -140,3 +143,4 @@ void free_udf(udf_func *udf); int mysql_create_function(THD *thd,udf_func *udf); int mysql_drop_function(THD *thd,const LEX_STRING *name); #endif +#endif /* SQL_UDF_INCLUDED */ diff --git a/sql/sql_view.h b/sql/sql_view.h index e08c2168e14..3de51c3264e 100644 --- a/sql/sql_view.h +++ b/sql/sql_view.h @@ -1,3 +1,6 @@ +#ifndef SQL_VIEW_INCLUDED +#define SQL_VIEW_INCLUDED + /* -*- C++ -*- */ /* Copyright (C) 2004 MySQL AB @@ -42,3 +45,4 @@ bool mysql_rename_view(THD *thd, const char *new_db, const char *new_name, #define VIEW_ANY_ACL (SELECT_ACL | UPDATE_ACL | INSERT_ACL | DELETE_ACL) +#endif /* SQL_VIEW_INCLUDED */ diff --git a/sql/structs.h b/sql/structs.h index a58c18f97c5..bcda7b1e787 100644 --- a/sql/structs.h +++ b/sql/structs.h @@ -1,3 +1,6 @@ +#ifndef STRUCTS_INCLUDED +#define STRUCTS_INCLUDED + /* Copyright (C) 2000-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -380,3 +383,5 @@ public: Discrete_interval* get_tail() const { return tail; }; Discrete_interval* get_current() const { return current; }; }; + +#endif /* STRUCTS_INCLUDED */ diff --git a/sql/table.h b/sql/table.h index 40372fa91cf..7c43ab339e5 100644 --- a/sql/table.h +++ b/sql/table.h @@ -1,3 +1,6 @@ +#ifndef TABLE_INCLUDED +#define TABLE_INCLUDED + /* Copyright 2000-2008 MySQL AB, 2008 Sun Microsystems, Inc. This program is free software; you can redistribute it and/or modify @@ -1729,3 +1732,4 @@ static inline void dbug_tmp_restore_column_maps(MY_BITMAP *read_set, size_t max_row_length(TABLE *table, const uchar *data); +#endif /* TABLE_INCLUDED */ diff --git a/sql/tzfile.h b/sql/tzfile.h index 1ff82d62329..1c1800ba1ed 100644 --- a/sql/tzfile.h +++ b/sql/tzfile.h @@ -1,3 +1,6 @@ +#ifndef TZFILE_INCLUDED +#define TZFILE_INCLUDED + /* Copyright (C) 2004 MySQL AB This program is free software; you can redistribute it and/or modify @@ -134,3 +137,5 @@ struct tzhead { */ #define isleap(y) (((y) % 4) == 0 && (((y) % 100) != 0 || ((y) % 400) == 0)) + +#endif diff --git a/sql/tztime.h b/sql/tztime.h index 9bf103519c4..9990e91f17b 100644 --- a/sql/tztime.h +++ b/sql/tztime.h @@ -1,3 +1,6 @@ +#ifndef TZTIME_INCLUDED +#define TZTIME_INCLUDED + /* Copyright (C) 2004 MySQL AB This program is free software; you can redistribute it and/or modify @@ -79,3 +82,4 @@ static const int MY_TZ_TABLES_COUNT= 4; #endif /* !defined(TESTTIME) && !defined(TZINFO2SQL) */ +#endif /* TZTIME_INCLUDED */ diff --git a/sql/unireg.h b/sql/unireg.h index 3ff7f058e3c..6c9080aea79 100644 --- a/sql/unireg.h +++ b/sql/unireg.h @@ -1,3 +1,6 @@ +#ifndef UNIREG_INCLUDED +#define UNIREG_INCLUDED + /* Copyright (C) 2000-2006 MySQL AB This program is free software; you can redistribute it and/or modify @@ -16,8 +19,6 @@ /* Extra functions used by unireg library */ -#ifndef _unireg_h - #ifndef NO_ALARM_LOOP #define NO_ALARM_LOOP /* lib5 and popen can't use alarm */ #endif -- cgit v1.2.1 From 5d2f79def9b67dd5400411070d47f8180ceb072b Mon Sep 17 00:00:00 2001 From: Mats Kindahl Date: Fri, 25 Sep 2009 11:47:15 +0200 Subject: Bug #47645: Segmentation fault when out of memory during handlerton initialization There is a missing check for memory allocation failure when allocating memory for the handlerton structure. If the handlerton init function tries to de-reference the pointer, it will cause a segmentation fault and crash the server. This patch fixes the problem by not calling the init function if memory allocation failed, and instead prints an informative error message and reports the error to the caller. sql/handler.cc: Add a check if memory allocation succeeded before calling the init function. If it failed, it is not necessary to free the memory, but the plugin->data is set to NULL to ensure that it can be checked for failure. --- sql/handler.cc | 9 +++++++++ 1 file changed, 9 insertions(+) (limited to 'sql') diff --git a/sql/handler.cc b/sql/handler.cc index e5c64452aaf..f966a9099ee 100644 --- a/sql/handler.cc +++ b/sql/handler.cc @@ -430,6 +430,14 @@ int ha_initialize_handlerton(st_plugin_int *plugin) hton= (handlerton *)my_malloc(sizeof(handlerton), MYF(MY_WME | MY_ZEROFILL)); + + if (hton == NULL) + { + sql_print_error("Unable to allocate memory for plugin '%s' handlerton.", + plugin->name.str); + goto err_no_hton_memory; + } + /* Historical Requirement */ plugin->data= hton; // shortcut for the future if (plugin->plugin->init && plugin->plugin->init(hton)) @@ -540,6 +548,7 @@ err_deinit: err: my_free((uchar*) hton, MYF(0)); +err_no_hton_memory: plugin->data= NULL; DBUG_RETURN(1); } -- cgit v1.2.1 From 623ed58cfda0aef6b6bf545a4200357a58a8a4cc Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Sat, 26 Sep 2009 12:49:49 +0800 Subject: Backporting WL#4398 WL#1720 Backporting BUG#44058 BUG#42244 BUG#45672 BUG#45673 Backporting BUG#45819 BUG#45973 BUG#39012 --- sql/CMakeLists.txt | 1 + sql/Makefile.am | 6 +- sql/handler.cc | 12 ++ sql/log.cc | 35 +++- sql/log.h | 16 +- sql/mysqld.cc | 10 ++ sql/replication.h | 490 ++++++++++++++++++++++++++++++++++++++++++++++++++++ sql/rpl_handler.cc | 493 +++++++++++++++++++++++++++++++++++++++++++++++++++++ sql/rpl_handler.h | 213 +++++++++++++++++++++++ sql/slave.cc | 124 +++++++++++++- sql/sql_class.cc | 36 ++++ sql/sql_class.h | 21 +-- sql/sql_parse.cc | 1 + sql/sql_plugin.cc | 17 +- sql/sql_plugin.h | 8 + sql/sql_repl.cc | 164 +++++++++++++----- 16 files changed, 1565 insertions(+), 82 deletions(-) create mode 100644 sql/replication.h create mode 100644 sql/rpl_handler.cc create mode 100644 sql/rpl_handler.h (limited to 'sql') diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt index 6f162f4d84d..18508468f60 100755 --- a/sql/CMakeLists.txt +++ b/sql/CMakeLists.txt @@ -75,6 +75,7 @@ SET (SQL_SOURCE rpl_rli.cc rpl_mi.cc sql_servers.cc sql_connect.cc scheduler.cc sql_profile.cc event_parse_data.cc + rpl_handler.cc ${PROJECT_SOURCE_DIR}/sql/sql_yacc.cc ${PROJECT_SOURCE_DIR}/sql/sql_yacc.h ${PROJECT_SOURCE_DIR}/include/mysqld_error.h diff --git a/sql/Makefile.am b/sql/Makefile.am index 2a12de2eaf6..afce7db59f2 100644 --- a/sql/Makefile.am +++ b/sql/Makefile.am @@ -76,7 +76,8 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \ sql_plugin.h authors.h event_parse_data.h \ event_data_objects.h event_scheduler.h \ sql_partition.h partition_info.h partition_element.h \ - contributors.h sql_servers.h + contributors.h sql_servers.h \ + rpl_handler.h replication.h mysqld_SOURCES = sql_lex.cc sql_handler.cc sql_partition.cc \ item.cc item_sum.cc item_buff.cc item_func.cc \ @@ -120,7 +121,8 @@ mysqld_SOURCES = sql_lex.cc sql_handler.cc sql_partition.cc \ event_queue.cc event_db_repository.cc events.cc \ sql_plugin.cc sql_binlog.cc \ sql_builtin.cc sql_tablespace.cc partition_info.cc \ - sql_servers.cc event_parse_data.cc + sql_servers.cc event_parse_data.cc \ + rpl_handler.cc nodist_mysqld_SOURCES = mini_client_errors.c pack.c client.c my_time.c my_user.c diff --git a/sql/handler.cc b/sql/handler.cc index e5c64452aaf..f17bb9f8036 100644 --- a/sql/handler.cc +++ b/sql/handler.cc @@ -24,6 +24,7 @@ #endif #include "mysql_priv.h" +#include "rpl_handler.h" #include "rpl_filter.h" #include #include @@ -221,6 +222,8 @@ handlerton *ha_checktype(THD *thd, enum legacy_db_type database_type, return NULL; } + RUN_HOOK(transaction, after_rollback, (thd, FALSE)); + switch (database_type) { #ifndef NO_HASH case DB_TYPE_HASH: @@ -1190,6 +1193,7 @@ int ha_commit_trans(THD *thd, bool all) if (cookie) tc_log->unlog(cookie, xid); DBUG_EXECUTE_IF("crash_commit_after", abort();); + RUN_HOOK(transaction, after_commit, (thd, FALSE)); end: if (rw_trans) start_waiting_global_read_lock(thd); @@ -1337,6 +1341,7 @@ int ha_rollback_trans(THD *thd, bool all) push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, ER_WARNING_NOT_COMPLETE_ROLLBACK, ER(ER_WARNING_NOT_COMPLETE_ROLLBACK)); + RUN_HOOK(transaction, after_rollback, (thd, FALSE)); DBUG_RETURN(error); } @@ -1371,7 +1376,14 @@ int ha_autocommit_or_rollback(THD *thd, int error) thd->variables.tx_isolation=thd->session_tx_isolation; } + else #endif + { + if (!error) + RUN_HOOK(transaction, after_commit, (thd, FALSE)); + else + RUN_HOOK(transaction, after_rollback, (thd, FALSE)); + } DBUG_RETURN(error); } diff --git a/sql/log.cc b/sql/log.cc index 1af2f3a4ddc..042431fc008 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -38,6 +38,7 @@ #endif #include +#include "rpl_handler.h" /* max size of the log message */ #define MAX_LOG_BUFFER_SIZE 1024 @@ -3683,9 +3684,11 @@ err: } -bool MYSQL_BIN_LOG::flush_and_sync() +bool MYSQL_BIN_LOG::flush_and_sync(bool *synced) { int err=0, fd=log_file.file; + if (synced) + *synced= 0; safe_mutex_assert_owner(&LOCK_log); if (flush_io_cache(&log_file)) return 1; @@ -3693,6 +3696,8 @@ bool MYSQL_BIN_LOG::flush_and_sync() { sync_binlog_counter= 0; err=my_sync(fd, MYF(MY_WME)); + if (synced) + *synced= 1; } return err; } @@ -3983,7 +3988,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd, if (file == &log_file) { - error= flush_and_sync(); + error= flush_and_sync(0); if (!error) { signal_update(); @@ -4169,8 +4174,16 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info) if (file == &log_file) // we are writing to the real log (disk) { - if (flush_and_sync()) + bool synced= 0; + if (flush_and_sync(&synced)) goto err; + + if (RUN_HOOK(binlog_storage, after_flush, + (thd, log_file_name, file->pos_in_file, synced))) { + sql_print_error("Failed to run 'after_flush' hooks"); + goto err; + } + signal_update(); rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED); } @@ -4425,7 +4438,7 @@ int MYSQL_BIN_LOG::write_cache(IO_CACHE *cache, bool lock_log, bool sync_log) DBUG_ASSERT(carry == 0); if (sync_log) - flush_and_sync(); + flush_and_sync(0); return 0; // All OK } @@ -4472,7 +4485,7 @@ bool MYSQL_BIN_LOG::write_incident(THD *thd, bool lock) ev.write(&log_file); if (lock) { - if (!error && !(error= flush_and_sync())) + if (!error && !(error= flush_and_sync(0))) { signal_update(); rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED); @@ -4560,7 +4573,8 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event, if (incident && write_incident(thd, FALSE)) goto err; - if (flush_and_sync()) + bool synced= 0; + if (flush_and_sync(&synced)) goto err; DBUG_EXECUTE_IF("half_binlogged_transaction", abort();); if (cache->error) // Error on read @@ -4569,6 +4583,15 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event, write_error=1; // Don't give more errors goto err; } + + if (RUN_HOOK(binlog_storage, after_flush, + (thd, log_file_name, log_file.pos_in_file, synced))) + { + sql_print_error("Failed to run 'after_flush' hooks"); + write_error=1; + goto err; + } + signal_update(); } diff --git a/sql/log.h b/sql/log.h index d306d6f7182..0550c921658 100644 --- a/sql/log.h +++ b/sql/log.h @@ -378,7 +378,21 @@ public: bool is_active(const char* log_file_name); int update_log_index(LOG_INFO* linfo, bool need_update_threads); void rotate_and_purge(uint flags); - bool flush_and_sync(); + + /** + Flush binlog cache and synchronize to disk. + + This function flushes events in binlog cache to binary log file, + it will do synchronizing according to the setting of system + variable 'sync_binlog'. If file is synchronized, @c synced will + be set to 1, otherwise 0. + + @param[out] synced if not NULL, set to 1 if file is synchronized, otherwise 0 + + @retval 0 Success + @retval other Failure + */ + bool flush_and_sync(bool *synced); int purge_logs(const char *to_log, bool included, bool need_mutex, bool need_update_threads, ulonglong *decrease_log_space); diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 7e9eb6e7291..2ae4ec9e9b6 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -31,6 +31,8 @@ #include "rpl_injector.h" +#include "rpl_handler.h" + #ifdef HAVE_SYS_PRCTL_H #include #endif @@ -1284,6 +1286,7 @@ void clean_up(bool print_message) ha_end(); if (tc_log) tc_log->close(); + delegates_destroy(); xid_cache_free(); delete_elements(&key_caches, (void (*)(const char*, uchar*)) free_key_cache); multi_keycache_free(); @@ -3760,6 +3763,13 @@ static int init_server_components() unireg_abort(1); } + /* initialize delegates for extension observers */ + if (delegates_init()) + { + sql_print_error("Initialize extension delegates failed"); + unireg_abort(1); + } + /* need to configure logging before initializing storage engines */ if (opt_update_log) { diff --git a/sql/replication.h b/sql/replication.h new file mode 100644 index 00000000000..6b7be58a5b1 --- /dev/null +++ b/sql/replication.h @@ -0,0 +1,490 @@ +/* Copyright (C) 2008 MySQL AB + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; version 2 of the License. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ + +#ifndef REPLICATION_H +#define REPLICATION_H + +#ifdef __cplusplus +extern "C" { +#endif + +/** + Transaction observer flags. +*/ +enum Trans_flags { + /** Transaction is a real transaction */ + TRANS_IS_REAL_TRANS = 1 +}; + +/** + Transaction observer parameter +*/ +typedef struct Trans_param { + uint32 server_id; + uint32 flags; + + /* + The latest binary log file name and position written by current + transaction, if binary log is disabled or no log event has been + written into binary log file by current transaction (events + written into transaction log cache are not counted), these two + member will be zero. + */ + const char *log_file; + my_off_t log_pos; +} Trans_param; + +/** + Observes and extends transaction execution +*/ +typedef struct Trans_observer { + uint32 len; + + /** + This callback is called after transaction commit + + This callback is called right after commit to storage engines for + transactional tables. + + For non-transactional tables, this is called at the end of the + statement, before sending statement status, if the statement + succeeded. + + @note The return value is currently ignored by the server. + + @param param The parameter for transaction observers + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_commit)(Trans_param *param); + + /** + This callback is called after transaction rollback + + This callback is called right after rollback to storage engines + for transactional tables. + + For non-transactional tables, this is called at the end of the + statement, before sending statement status, if the statement + failed. + + @note The return value is currently ignored by the server. + + @param param The parameter for transaction observers + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_rollback)(Trans_param *param); +} Trans_observer; + +/** + Binlog storage flags +*/ +enum Binlog_storage_flags { + /** Binary log was sync:ed */ + BINLOG_STORAGE_IS_SYNCED = 1 +}; + +/** + Binlog storage observer parameters + */ +typedef struct Binlog_storage_param { + uint32 server_id; +} Binlog_storage_param; + +/** + Observe binlog logging storage +*/ +typedef struct Binlog_storage_observer { + uint32 len; + + /** + This callback is called after binlog has been flushed + + This callback is called after cached events have been flushed to + binary log file. Whether the binary log file is synchronized to + disk is indicated by the bit BINLOG_STORAGE_IS_SYNCED in @a flags. + + @param param Observer common parameter + @param log_file Binlog file name been updated + @param log_pos Binlog position after update + @param flags flags for binlog storage + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_flush)(Binlog_storage_param *param, + const char *log_file, my_off_t log_pos, + uint32 flags); +} Binlog_storage_observer; + +/** + Replication binlog transmitter (binlog dump) observer parameter. +*/ +typedef struct Binlog_transmit_param { + uint32 server_id; + uint32 flags; +} Binlog_transmit_param; + +/** + Observe and extends the binlog dumping thread. +*/ +typedef struct Binlog_transmit_observer { + uint32 len; + + /** + This callback is called when binlog dumping starts + + + @param param Observer common parameter + @param log_file Binlog file name to transmit from + @param log_pos Binlog position to transmit from + + @retval 0 Sucess + @retval 1 Failure + */ + int (*transmit_start)(Binlog_transmit_param *param, + const char *log_file, my_off_t log_pos); + + /** + This callback is called when binlog dumping stops + + @param param Observer common parameter + + @retval 0 Sucess + @retval 1 Failure + */ + int (*transmit_stop)(Binlog_transmit_param *param); + + /** + This callback is called to reserve bytes in packet header for event transmission + + This callback is called when resetting transmit packet header to + reserve bytes for this observer in packet header. + + The @a header buffer is allocated by the server code, and @a size + is the size of the header buffer. Each observer can only reserve + a maximum size of @a size in the header. + + @param param Observer common parameter + @param header Pointer of the header buffer + @param size Size of the header buffer + @param len Header length reserved by this observer + + @retval 0 Sucess + @retval 1 Failure + */ + int (*reserve_header)(Binlog_transmit_param *param, + unsigned char *header, + unsigned long size, + unsigned long *len); + + /** + This callback is called before sending an event packet to slave + + @param param Observer common parameter + @param packet Binlog event packet to send + @param len Length of the event packet + @param log_file Binlog file name of the event packet to send + @param log_pos Binlog position of the event packet to send + + @retval 0 Sucess + @retval 1 Failure + */ + int (*before_send_event)(Binlog_transmit_param *param, + unsigned char *packet, unsigned long len, + const char *log_file, my_off_t log_pos ); + + /** + This callback is called after sending an event packet to slave + + @param param Observer common parameter + @param event_buf Binlog event packet buffer sent + @param len length of the event packet buffer + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_send_event)(Binlog_transmit_param *param, + const char *event_buf, unsigned long len); + + /** + This callback is called after resetting master status + + This is called when executing the command RESET MASTER, and is + used to reset status variables added by observers. + + @param param Observer common parameter + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_reset_master)(Binlog_transmit_param *param); +} Binlog_transmit_observer; + +/** + Binlog relay IO flags +*/ +enum Binlog_relay_IO_flags { + /** Binary relay log was sync:ed */ + BINLOG_RELAY_IS_SYNCED = 1 +}; + + +/** + Replication binlog relay IO observer parameter +*/ +typedef struct Binlog_relay_IO_param { + uint32 server_id; + + /* Master host, user and port */ + char *host; + char *user; + unsigned int port; + + char *master_log_name; + my_off_t master_log_pos; + + MYSQL *mysql; /* the connection to master */ +} Binlog_relay_IO_param; + +/** + Observes and extends the service of slave IO thread. +*/ +typedef struct Binlog_relay_IO_observer { + uint32 len; + + /** + This callback is called when slave IO thread starts + + @param param Observer common parameter + + @retval 0 Sucess + @retval 1 Failure + */ + int (*thread_start)(Binlog_relay_IO_param *param); + + /** + This callback is called when slave IO thread stops + + @param param Observer common parameter + + @retval 0 Sucess + @retval 1 Failure + */ + int (*thread_stop)(Binlog_relay_IO_param *param); + + /** + This callback is called before slave requesting binlog transmission from master + + This is called before slave issuing BINLOG_DUMP command to master + to request binlog. + + @param param Observer common parameter + @param flags binlog dump flags + + @retval 0 Sucess + @retval 1 Failure + */ + int (*before_request_transmit)(Binlog_relay_IO_param *param, uint32 flags); + + /** + This callback is called after read an event packet from master + + @param param Observer common parameter + @param packet The event packet read from master + @param len Length of the event packet read from master + @param event_buf The event packet return after process + @param event_len The length of event packet return after process + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_read_event)(Binlog_relay_IO_param *param, + const char *packet, unsigned long len, + const char **event_buf, unsigned long *event_len); + + /** + This callback is called after written an event packet to relay log + + @param param Observer common parameter + @param event_buf Event packet written to relay log + @param event_len Length of the event packet written to relay log + @param flags flags for relay log + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_queue_event)(Binlog_relay_IO_param *param, + const char *event_buf, unsigned long event_len, + uint32 flags); + + /** + This callback is called after reset slave relay log IO status + + @param param Observer common parameter + + @retval 0 Sucess + @retval 1 Failure + */ + int (*after_reset_slave)(Binlog_relay_IO_param *param); +} Binlog_relay_IO_observer; + + +/** + Register a transaction observer + + @param observer The transaction observer to register + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer already exists +*/ +int register_trans_observer(Trans_observer *observer, void *p); + +/** + Unregister a transaction observer + + @param observer The transaction observer to unregister + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer not exists +*/ +int unregister_trans_observer(Trans_observer *observer, void *p); + +/** + Register a binlog storage observer + + @param observer The binlog storage observer to register + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer already exists +*/ +int register_binlog_storage_observer(Binlog_storage_observer *observer, void *p); + +/** + Unregister a binlog storage observer + + @param observer The binlog storage observer to unregister + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer not exists +*/ +int unregister_binlog_storage_observer(Binlog_storage_observer *observer, void *p); + +/** + Register a binlog transmit observer + + @param observer The binlog transmit observer to register + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer already exists +*/ +int register_binlog_transmit_observer(Binlog_transmit_observer *observer, void *p); + +/** + Unregister a binlog transmit observer + + @param observer The binlog transmit observer to unregister + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer not exists +*/ +int unregister_binlog_transmit_observer(Binlog_transmit_observer *observer, void *p); + +/** + Register a binlog relay IO (slave IO thread) observer + + @param observer The binlog relay IO observer to register + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer already exists +*/ +int register_binlog_relay_io_observer(Binlog_relay_IO_observer *observer, void *p); + +/** + Unregister a binlog relay IO (slave IO thread) observer + + @param observer The binlog relay IO observer to unregister + @param p pointer to the internal plugin structure + + @retval 0 Sucess + @retval 1 Observer not exists +*/ +int unregister_binlog_relay_io_observer(Binlog_relay_IO_observer *observer, void *p); + +/** + Connect to master + + This function can only used in the slave I/O thread context, and + will use the same master information to do the connection. + + @code + MYSQL *mysql = mysql_init(NULL); + if (rpl_connect_master(mysql)) + { + // do stuff with the connection + } + mysql_close(mysql); // close the connection + @endcode + + @param mysql address of MYSQL structure to use, pass NULL will + create a new one + + @return address of MYSQL structure on success, NULL on failure +*/ +MYSQL *rpl_connect_master(MYSQL *mysql); + +/** + Set thread entering a condition + + This function should be called before putting a thread to wait for + a condition. @a mutex should be held before calling this + function. After being waken up, @f thd_exit_cond should be called. + + @param thd The thread entering the condition, NULL means current thread + @param cond The condition the thread is going to wait for + @param mutex The mutex associated with the condition, this must be + held before call this function + @param msg The new process message for the thread +*/ +const char* thd_enter_cond(MYSQL_THD thd, pthread_cond_t *cond, + pthread_mutex_t *mutex, const char *msg); + +/** + Set thread leaving a condition + + This function should be called after a thread being waken up for a + condition. + + @param thd The thread entering the condition, NULL means current thread + @param old_msg The process message, ususally this should be the old process + message before calling @f thd_enter_cond +*/ +void thd_exit_cond(MYSQL_THD thd, const char *old_msg); + + +#ifdef __cplusplus +} +#endif +#endif /* REPLICATION_H */ diff --git a/sql/rpl_handler.cc b/sql/rpl_handler.cc new file mode 100644 index 00000000000..aea838928b9 --- /dev/null +++ b/sql/rpl_handler.cc @@ -0,0 +1,493 @@ +/* Copyright (C) 2008 MySQL AB + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; version 2 of the License. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ + +#include "mysql_priv.h" + +#include "rpl_mi.h" +#include "sql_repl.h" +#include "log_event.h" +#include "rpl_filter.h" +#include +#include "rpl_handler.h" + +Trans_delegate *transaction_delegate; +Binlog_storage_delegate *binlog_storage_delegate; +#ifdef HAVE_REPLICATION +Binlog_transmit_delegate *binlog_transmit_delegate; +Binlog_relay_IO_delegate *binlog_relay_io_delegate; +#endif /* HAVE_REPLICATION */ + +/* + structure to save transaction log filename and position +*/ +typedef struct Trans_binlog_info { + my_off_t log_pos; + char log_file[FN_REFLEN]; +} Trans_binlog_info; + +static pthread_key(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO); + +int get_user_var_int(const char *name, + long long int *value, int *null_value) +{ + my_bool null_val; + user_var_entry *entry= + (user_var_entry*) hash_search(¤t_thd->user_vars, + (uchar*) name, strlen(name)); + if (!entry) + return 1; + *value= entry->val_int(&null_val); + if (null_value) + *null_value= null_val; + return 0; +} + +int get_user_var_real(const char *name, + double *value, int *null_value) +{ + my_bool null_val; + user_var_entry *entry= + (user_var_entry*) hash_search(¤t_thd->user_vars, + (uchar*) name, strlen(name)); + if (!entry) + return 1; + *value= entry->val_real(&null_val); + if (null_value) + *null_value= null_val; + return 0; +} + +int get_user_var_str(const char *name, char *value, + size_t len, unsigned int precision, int *null_value) +{ + String str; + my_bool null_val; + user_var_entry *entry= + (user_var_entry*) hash_search(¤t_thd->user_vars, + (uchar*) name, strlen(name)); + if (!entry) + return 1; + entry->val_str(&null_val, &str, precision); + strncpy(value, str.c_ptr(), len); + if (null_value) + *null_value= null_val; + return 0; +} + +int delegates_init() +{ + static unsigned char trans_mem[sizeof(Trans_delegate)]; + static unsigned char storage_mem[sizeof(Binlog_storage_delegate)]; +#ifdef HAVE_REPLICATION + static unsigned char transmit_mem[sizeof(Binlog_transmit_delegate)]; + static unsigned char relay_io_mem[sizeof(Binlog_relay_IO_delegate)]; +#endif + + if (!(transaction_delegate= new (trans_mem) Trans_delegate) + || (!transaction_delegate->is_inited()) + || !(binlog_storage_delegate= new (storage_mem) Binlog_storage_delegate) + || (!binlog_storage_delegate->is_inited()) +#ifdef HAVE_REPLICATION + || !(binlog_transmit_delegate= new (transmit_mem) Binlog_transmit_delegate) + || (!binlog_transmit_delegate->is_inited()) + || !(binlog_relay_io_delegate= new (relay_io_mem) Binlog_relay_IO_delegate) + || (!binlog_relay_io_delegate->is_inited()) +#endif /* HAVE_REPLICATION */ + ) + return 1; + + if (pthread_key_create(&RPL_TRANS_BINLOG_INFO, NULL)) + return 1; + return 0; +} + +void delegates_destroy() +{ + if (transaction_delegate) + transaction_delegate->~Trans_delegate(); + if (binlog_storage_delegate) + binlog_storage_delegate->~Binlog_storage_delegate(); +#ifdef HAVE_REPLICATION + if (binlog_transmit_delegate) + binlog_transmit_delegate->~Binlog_transmit_delegate(); + if (binlog_relay_io_delegate) + binlog_relay_io_delegate->~Binlog_relay_IO_delegate(); +#endif /* HAVE_REPLICATION */ +} + +/* + This macro is used by almost all the Delegate methods to iterate + over all the observers running given callback function of the + delegate . + + Add observer plugins to the thd->lex list, after each statement, all + plugins add to thd->lex will be automatically unlocked. + */ +#define FOREACH_OBSERVER(r, f, thd, args) \ + param.server_id= thd->server_id; \ + read_lock(); \ + Observer_info_iterator iter= observer_info_iter(); \ + Observer_info *info= iter++; \ + for (; info; info= iter++) \ + { \ + plugin_ref plugin= \ + my_plugin_lock(thd, &info->plugin); \ + if (!plugin) \ + { \ + r= 1; \ + break; \ + } \ + if (((Observer *)info->observer)->f \ + && ((Observer *)info->observer)->f args) \ + { \ + r= 1; \ + plugin_unlock(thd, plugin); \ + break; \ + } \ + plugin_unlock(thd, plugin); \ + } \ + unlock() + + +int Trans_delegate::after_commit(THD *thd, bool all) +{ + Trans_param param; + bool is_real_trans= (all || thd->transaction.all.ha_list == 0); + if (is_real_trans) + param.flags |= TRANS_IS_REAL_TRANS; + + Trans_binlog_info *log_info= + my_pthread_getspecific_ptr(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO); + + param.log_file= log_info ? log_info->log_file : 0; + param.log_pos= log_info ? log_info->log_pos : 0; + + int ret= 0; + FOREACH_OBSERVER(ret, after_commit, thd, (¶m)); + + /* + This is the end of a real transaction or autocommit statement, we + can free the memory allocated for binlog file and position. + */ + if (is_real_trans && log_info) + { + my_pthread_setspecific_ptr(RPL_TRANS_BINLOG_INFO, NULL); + my_free(log_info, MYF(0)); + } + return ret; +} + +int Trans_delegate::after_rollback(THD *thd, bool all) +{ + Trans_param param; + bool is_real_trans= (all || thd->transaction.all.ha_list == 0); + if (is_real_trans) + param.flags |= TRANS_IS_REAL_TRANS; + + Trans_binlog_info *log_info= + my_pthread_getspecific_ptr(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO); + + param.log_file= log_info ? log_info->log_file : 0; + param.log_pos= log_info ? log_info->log_pos : 0; + + int ret= 0; + FOREACH_OBSERVER(ret, after_commit, thd, (¶m)); + + /* + This is the end of a real transaction or autocommit statement, we + can free the memory allocated for binlog file and position. + */ + if (is_real_trans && log_info) + { + my_pthread_setspecific_ptr(RPL_TRANS_BINLOG_INFO, NULL); + my_free(log_info, MYF(0)); + } + return ret; +} + +int Binlog_storage_delegate::after_flush(THD *thd, + const char *log_file, + my_off_t log_pos, + bool synced) +{ + Binlog_storage_param param; + uint32 flags=0; + if (synced) + flags |= BINLOG_STORAGE_IS_SYNCED; + + Trans_binlog_info *log_info= + my_pthread_getspecific_ptr(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO); + + if (!log_info) + { + if(!(log_info= + (Trans_binlog_info *)my_malloc(sizeof(Trans_binlog_info), MYF(0)))) + return 1; + my_pthread_setspecific_ptr(RPL_TRANS_BINLOG_INFO, log_info); + } + + strcpy(log_info->log_file, log_file+dirname_length(log_file)); + log_info->log_pos = log_pos; + + int ret= 0; + FOREACH_OBSERVER(ret, after_flush, thd, + (¶m, log_info->log_file, log_info->log_pos, flags)); + return ret; +} + +#ifdef HAVE_REPLICATION +int Binlog_transmit_delegate::transmit_start(THD *thd, ushort flags, + const char *log_file, + my_off_t log_pos) +{ + Binlog_transmit_param param; + param.flags= flags; + + int ret= 0; + FOREACH_OBSERVER(ret, transmit_start, thd, (¶m, log_file, log_pos)); + return ret; +} + +int Binlog_transmit_delegate::transmit_stop(THD *thd, ushort flags) +{ + Binlog_transmit_param param; + param.flags= flags; + + int ret= 0; + FOREACH_OBSERVER(ret, transmit_stop, thd, (¶m)); + return ret; +} + +int Binlog_transmit_delegate::reserve_header(THD *thd, ushort flags, + String *packet) +{ + /* NOTE2ME: Maximum extra header size for each observer, I hope 32 + bytes should be enough for each Observer to reserve their extra + header. If later found this is not enough, we can increase this + /HEZX + */ +#define RESERVE_HEADER_SIZE 32 + unsigned char header[RESERVE_HEADER_SIZE]; + ulong hlen; + Binlog_transmit_param param; + param.flags= flags; + param.server_id= thd->server_id; + + int ret= 0; + read_lock(); + Observer_info_iterator iter= observer_info_iter(); + Observer_info *info= iter++; + for (; info; info= iter++) + { + plugin_ref plugin= + my_plugin_lock(thd, &info->plugin); + if (!plugin) + { + ret= 1; + break; + } + hlen= 0; + if (((Observer *)info->observer)->reserve_header + && ((Observer *)info->observer)->reserve_header(¶m, + header, + RESERVE_HEADER_SIZE, + &hlen)) + { + ret= 1; + plugin_unlock(thd, plugin); + break; + } + plugin_unlock(thd, plugin); + if (hlen == 0) + continue; + if (hlen > RESERVE_HEADER_SIZE || packet->append((char *)header, hlen)) + { + ret= 1; + break; + } + } + unlock(); + return ret; +} + +int Binlog_transmit_delegate::before_send_event(THD *thd, ushort flags, + String *packet, + const char *log_file, + my_off_t log_pos) +{ + Binlog_transmit_param param; + param.flags= flags; + + int ret= 0; + FOREACH_OBSERVER(ret, before_send_event, thd, + (¶m, (uchar *)packet->c_ptr(), + packet->length(), + log_file+dirname_length(log_file), log_pos)); + return ret; +} + +int Binlog_transmit_delegate::after_send_event(THD *thd, ushort flags, + String *packet) +{ + Binlog_transmit_param param; + param.flags= flags; + + int ret= 0; + FOREACH_OBSERVER(ret, after_send_event, thd, + (¶m, packet->c_ptr(), packet->length())); + return ret; +} + +int Binlog_transmit_delegate::after_reset_master(THD *thd, ushort flags) + +{ + Binlog_transmit_param param; + param.flags= flags; + + int ret= 0; + FOREACH_OBSERVER(ret, after_reset_master, thd, (¶m)); + return ret; +} + +void Binlog_relay_IO_delegate::init_param(Binlog_relay_IO_param *param, + Master_info *mi) +{ + param->mysql= mi->mysql; + param->user= mi->user; + param->host= mi->host; + param->port= mi->port; + param->master_log_name= mi->master_log_name; + param->master_log_pos= mi->master_log_pos; +} + +int Binlog_relay_IO_delegate::thread_start(THD *thd, Master_info *mi) +{ + Binlog_relay_IO_param param; + init_param(¶m, mi); + + int ret= 0; + FOREACH_OBSERVER(ret, thread_start, thd, (¶m)); + return ret; +} + + +int Binlog_relay_IO_delegate::thread_stop(THD *thd, Master_info *mi) +{ + + Binlog_relay_IO_param param; + init_param(¶m, mi); + + int ret= 0; + FOREACH_OBSERVER(ret, thread_stop, thd, (¶m)); + return ret; +} + +int Binlog_relay_IO_delegate::before_request_transmit(THD *thd, + Master_info *mi, + ushort flags) +{ + Binlog_relay_IO_param param; + init_param(¶m, mi); + + int ret= 0; + FOREACH_OBSERVER(ret, before_request_transmit, thd, (¶m, (uint32)flags)); + return ret; +} + +int Binlog_relay_IO_delegate::after_read_event(THD *thd, Master_info *mi, + const char *packet, ulong len, + const char **event_buf, + ulong *event_len) +{ + Binlog_relay_IO_param param; + init_param(¶m, mi); + + int ret= 0; + FOREACH_OBSERVER(ret, after_read_event, thd, + (¶m, packet, len, event_buf, event_len)); + return ret; +} + +int Binlog_relay_IO_delegate::after_queue_event(THD *thd, Master_info *mi, + const char *event_buf, + ulong event_len, + bool synced) +{ + Binlog_relay_IO_param param; + init_param(¶m, mi); + + uint32 flags=0; + if (synced) + flags |= BINLOG_STORAGE_IS_SYNCED; + + int ret= 0; + FOREACH_OBSERVER(ret, after_queue_event, thd, + (¶m, event_buf, event_len, flags)); + return ret; +} + +int Binlog_relay_IO_delegate::after_reset_slave(THD *thd, Master_info *mi) + +{ + Binlog_relay_IO_param param; + init_param(¶m, mi); + + int ret= 0; + FOREACH_OBSERVER(ret, after_reset_slave, thd, (¶m)); + return ret; +} +#endif /* HAVE_REPLICATION */ + +int register_trans_observer(Trans_observer *observer, void *p) +{ + return transaction_delegate->add_observer(observer, (st_plugin_int *)p); +} + +int unregister_trans_observer(Trans_observer *observer, void *p) +{ + return transaction_delegate->remove_observer(observer, (st_plugin_int *)p); +} + +int register_binlog_storage_observer(Binlog_storage_observer *observer, void *p) +{ + return binlog_storage_delegate->add_observer(observer, (st_plugin_int *)p); +} + +int unregister_binlog_storage_observer(Binlog_storage_observer *observer, void *p) +{ + return binlog_storage_delegate->remove_observer(observer, (st_plugin_int *)p); +} + +#ifdef HAVE_REPLICATION +int register_binlog_transmit_observer(Binlog_transmit_observer *observer, void *p) +{ + return binlog_transmit_delegate->add_observer(observer, (st_plugin_int *)p); +} + +int unregister_binlog_transmit_observer(Binlog_transmit_observer *observer, void *p) +{ + return binlog_transmit_delegate->remove_observer(observer, (st_plugin_int *)p); +} + +int register_binlog_relay_io_observer(Binlog_relay_IO_observer *observer, void *p) +{ + return binlog_relay_io_delegate->add_observer(observer, (st_plugin_int *)p); +} + +int unregister_binlog_relay_io_observer(Binlog_relay_IO_observer *observer, void *p) +{ + return binlog_relay_io_delegate->remove_observer(observer, (st_plugin_int *)p); +} +#endif /* HAVE_REPLICATION */ diff --git a/sql/rpl_handler.h b/sql/rpl_handler.h new file mode 100644 index 00000000000..4fb7b4e035b --- /dev/null +++ b/sql/rpl_handler.h @@ -0,0 +1,213 @@ +/* Copyright (C) 2008 MySQL AB + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; version 2 of the License. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ + +#ifndef RPL_HANDLER_H +#define RPL_HANDLER_H + +#include "mysql_priv.h" +#include "rpl_mi.h" +#include "rpl_rli.h" +#include "sql_plugin.h" +#include "replication.h" + +class Observer_info { +public: + void *observer; + st_plugin_int *plugin_int; + plugin_ref plugin; + + Observer_info(void *ob, st_plugin_int *p) + :observer(ob), plugin_int(p) + { + plugin= plugin_int_to_ref(plugin_int); + } +}; + +class Delegate { +public: + typedef List Observer_info_list; + typedef List_iterator Observer_info_iterator; + + int add_observer(void *observer, st_plugin_int *plugin) + { + int ret= FALSE; + if (!inited) + return TRUE; + write_lock(); + Observer_info_iterator iter(observer_info_list); + Observer_info *info= iter++; + while (info && info->observer != observer) + info= iter++; + if (!info) + { + info= new Observer_info(observer, plugin); + if (!info || observer_info_list.push_back(info, &memroot)) + ret= TRUE; + } + else + ret= TRUE; + unlock(); + return ret; + } + + int remove_observer(void *observer, st_plugin_int *plugin) + { + int ret= FALSE; + if (!inited) + return TRUE; + write_lock(); + Observer_info_iterator iter(observer_info_list); + Observer_info *info= iter++; + while (info && info->observer != observer) + info= iter++; + if (info) + iter.remove(); + else + ret= TRUE; + unlock(); + return ret; + } + + inline Observer_info_iterator observer_info_iter() + { + return Observer_info_iterator(observer_info_list); + } + + inline bool is_empty() + { + return observer_info_list.is_empty(); + } + + inline int read_lock() + { + if (!inited) + return TRUE; + return rw_rdlock(&lock); + } + + inline int write_lock() + { + if (!inited) + return TRUE; + return rw_wrlock(&lock); + } + + inline int unlock() + { + if (!inited) + return TRUE; + return rw_unlock(&lock); + } + + inline bool is_inited() + { + return inited; + } + + Delegate() + { + inited= FALSE; + if (my_rwlock_init(&lock, NULL)) + return; + init_sql_alloc(&memroot, 1024, 0); + inited= TRUE; + } + ~Delegate() + { + inited= FALSE; + rwlock_destroy(&lock); + free_root(&memroot, MYF(0)); + } + +private: + Observer_info_list observer_info_list; + rw_lock_t lock; + MEM_ROOT memroot; + bool inited; +}; + +class Trans_delegate + :public Delegate { +public: + typedef Trans_observer Observer; + int before_commit(THD *thd, bool all); + int before_rollback(THD *thd, bool all); + int after_commit(THD *thd, bool all); + int after_rollback(THD *thd, bool all); +}; + +class Binlog_storage_delegate + :public Delegate { +public: + typedef Binlog_storage_observer Observer; + int after_flush(THD *thd, const char *log_file, + my_off_t log_pos, bool synced); +}; + +#ifdef HAVE_REPLICATION +class Binlog_transmit_delegate + :public Delegate { +public: + typedef Binlog_transmit_observer Observer; + int transmit_start(THD *thd, ushort flags, + const char *log_file, my_off_t log_pos); + int transmit_stop(THD *thd, ushort flags); + int reserve_header(THD *thd, ushort flags, String *packet); + int before_send_event(THD *thd, ushort flags, + String *packet, const + char *log_file, my_off_t log_pos ); + int after_send_event(THD *thd, ushort flags, + String *packet); + int after_reset_master(THD *thd, ushort flags); +}; + +class Binlog_relay_IO_delegate + :public Delegate { +public: + typedef Binlog_relay_IO_observer Observer; + int thread_start(THD *thd, Master_info *mi); + int thread_stop(THD *thd, Master_info *mi); + int before_request_transmit(THD *thd, Master_info *mi, ushort flags); + int after_read_event(THD *thd, Master_info *mi, + const char *packet, ulong len, + const char **event_buf, ulong *event_len); + int after_queue_event(THD *thd, Master_info *mi, + const char *event_buf, ulong event_len, + bool synced); + int after_reset_slave(THD *thd, Master_info *mi); +private: + void init_param(Binlog_relay_IO_param *param, Master_info *mi); +}; +#endif /* HAVE_REPLICATION */ + +int delegates_init(); +void delegates_destroy(); + +extern Trans_delegate *transaction_delegate; +extern Binlog_storage_delegate *binlog_storage_delegate; +#ifdef HAVE_REPLICATION +extern Binlog_transmit_delegate *binlog_transmit_delegate; +extern Binlog_relay_IO_delegate *binlog_relay_io_delegate; +#endif /* HAVE_REPLICATION */ + +/* + if there is no observers in the delegate, we can return 0 + immediately. +*/ +#define RUN_HOOK(group, hook, args) \ + (group ##_delegate->is_empty() ? \ + 0 : group ##_delegate->hook args) + +#endif /* RPL_HANDLER_H */ diff --git a/sql/slave.cc b/sql/slave.cc index fac9ee214c5..4988886dce4 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -40,6 +40,7 @@ #include #include #include +#include "rpl_handler.h" #ifdef HAVE_REPLICATION @@ -69,6 +70,8 @@ ulonglong relay_log_space_limit = 0; int disconnect_slave_event_count = 0, abort_slave_event_count = 0; int events_till_abort = -1; +static pthread_key(Master_info*, RPL_MASTER_INFO); + enum enum_slave_reconnect_actions { SLAVE_RECON_ACT_REG= 0, @@ -231,6 +234,10 @@ int init_slave() TODO: re-write this to interate through the list of files for multi-master */ + + if (pthread_key_create(&RPL_MASTER_INFO, NULL)) + goto err; + active_mi= new Master_info; /* @@ -1868,17 +1875,22 @@ static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed, } -static int request_dump(MYSQL* mysql, Master_info* mi, - bool *suppress_warnings) +static int request_dump(THD *thd, MYSQL* mysql, Master_info* mi, + bool *suppress_warnings) { uchar buf[FN_REFLEN + 10]; int len; - int binlog_flags = 0; // for now + ushort binlog_flags = 0; // for now char* logname = mi->master_log_name; DBUG_ENTER("request_dump"); *suppress_warnings= FALSE; + if (RUN_HOOK(binlog_relay_io, + before_request_transmit, + (thd, mi, binlog_flags))) + DBUG_RETURN(1); + // TODO if big log files: Change next to int8store() int4store(buf, (ulong) mi->master_log_pos); int2store(buf + 4, binlog_flags); @@ -2532,6 +2544,16 @@ pthread_handler_t handle_slave_io(void *arg) mi->master_log_name, llstr(mi->master_log_pos,llbuff))); + /* This must be called before run any binlog_relay_io hooks */ + my_pthread_setspecific_ptr(RPL_MASTER_INFO, mi); + + if (RUN_HOOK(binlog_relay_io, thread_start, (thd, mi))) + { + mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, + ER(ER_SLAVE_FATAL_ERROR), "Failed to run 'thread_start' hook"); + goto err; + } + if (!(mi->mysql = mysql = mysql_init(NULL))) { mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, @@ -2621,7 +2643,7 @@ connected: while (!io_slave_killed(thd,mi)) { thd_proc_info(thd, "Requesting binlog dump"); - if (request_dump(mysql, mi, &suppress_warnings)) + if (request_dump(thd, mysql, mi, &suppress_warnings)) { sql_print_error("Failed on request_dump()"); if (check_io_slave_killed(thd, mi, "Slave I/O thread killed while \ @@ -2641,6 +2663,7 @@ requesting master dump") || goto err; goto connected; }); + const char *event_buf; DBUG_ASSERT(mi->last_error().number == 0); while (!io_slave_killed(thd,mi)) @@ -2697,14 +2720,37 @@ Stopping slave I/O thread due to out-of-memory error from master"); retry_count=0; // ok event, reset retry counter thd_proc_info(thd, "Queueing master event to the relay log"); - if (queue_event(mi,(const char*)mysql->net.read_pos + 1, - event_len)) + event_buf= (const char*)mysql->net.read_pos + 1; + if (RUN_HOOK(binlog_relay_io, after_read_event, + (thd, mi,(const char*)mysql->net.read_pos + 1, + event_len, &event_buf, &event_len))) + { + mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, + ER(ER_SLAVE_FATAL_ERROR), + "Failed to run 'after_read_event' hook"); + goto err; + } + + /* XXX: 'synced' should be updated by queue_event to indicate + whether event has been synced to disk */ + bool synced= 0; + if (queue_event(mi, event_buf, event_len)) { mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE, ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE), "could not queue event from master"); goto err; } + + if (RUN_HOOK(binlog_relay_io, after_queue_event, + (thd, mi, event_buf, event_len, synced))) + { + mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, + ER(ER_SLAVE_FATAL_ERROR), + "Failed to run 'after_queue_event' hook"); + goto err; + } + if (flush_master_info(mi, 1)) { sql_print_error("Failed to flush master info file"); @@ -2750,6 +2796,7 @@ err: // print the current replication position sql_print_information("Slave I/O thread exiting, read up to log '%s', position %s", IO_RPL_LOG_NAME, llstr(mi->master_log_pos,llbuff)); + RUN_HOOK(binlog_relay_io, thread_stop, (thd, mi)); thd->set_query(NULL, 0); thd->reset_db(NULL, 0); if (mysql) @@ -3906,6 +3953,71 @@ static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi, } +MYSQL *rpl_connect_master(MYSQL *mysql) +{ + THD *thd= current_thd; + Master_info *mi= my_pthread_getspecific_ptr(Master_info*, RPL_MASTER_INFO); + if (!mi) + { + sql_print_error("'rpl_connect_master' must be called in slave I/O thread context."); + return NULL; + } + + bool allocated= false; + + if (!mysql) + { + if(!(mysql= mysql_init(NULL))) + { + sql_print_error("rpl_connect_master: failed in mysql_init()"); + return NULL; + } + allocated= true; + } + + /* + XXX: copied from connect_to_master, this function should not + change the slave status, so we cannot use connect_to_master + directly + + TODO: make this part a seperate function to eliminate duplication + */ + mysql_options(mysql, MYSQL_OPT_CONNECT_TIMEOUT, (char *) &slave_net_timeout); + mysql_options(mysql, MYSQL_OPT_READ_TIMEOUT, (char *) &slave_net_timeout); + +#ifdef HAVE_OPENSSL + if (mi->ssl) + { + mysql_ssl_set(mysql, + mi->ssl_key[0]?mi->ssl_key:0, + mi->ssl_cert[0]?mi->ssl_cert:0, + mi->ssl_ca[0]?mi->ssl_ca:0, + mi->ssl_capath[0]?mi->ssl_capath:0, + mi->ssl_cipher[0]?mi->ssl_cipher:0); + mysql_options(mysql, MYSQL_OPT_SSL_VERIFY_SERVER_CERT, + &mi->ssl_verify_server_cert); + } +#endif + + mysql_options(mysql, MYSQL_SET_CHARSET_NAME, default_charset_info->csname); + /* This one is not strictly needed but we have it here for completeness */ + mysql_options(mysql, MYSQL_SET_CHARSET_DIR, (char *) charsets_dir); + + if (io_slave_killed(thd, mi) + || !mysql_real_connect(mysql, mi->host, mi->user, mi->password, 0, + mi->port, 0, 0)) + { + if (!io_slave_killed(thd, mi)) + sql_print_error("rpl_connect_master: error connecting to master: %s (server_error: %d)", + mysql_error(mysql), mysql_errno(mysql)); + + if (allocated) + mysql_close(mysql); // this will free the object + return NULL; + } + return mysql; +} + /* Store the file and position where the execute-slave thread are in the relay log. diff --git a/sql/sql_class.cc b/sql/sql_class.cc index 3f568566c89..755e60b4fc2 100644 --- a/sql/sql_class.cc +++ b/sql/sql_class.cc @@ -277,6 +277,42 @@ const char *set_thd_proc_info(THD *thd, const char *info, return old_info; } +extern "C" +const char* thd_enter_cond(MYSQL_THD thd, pthread_cond_t *cond, + pthread_mutex_t *mutex, const char *msg) +{ + if (!thd) + thd= current_thd; + + const char* old_msg = thd->proc_info; + safe_mutex_assert_owner(mutex); + thd->mysys_var->current_mutex = mutex; + thd->mysys_var->current_cond = cond; + thd->proc_info = msg; + return old_msg; +} + +extern "C" +void thd_exit_cond(MYSQL_THD thd, const char *old_msg) +{ + if (!thd) + thd= current_thd; + + /* + Putting the mutex unlock in thd_exit_cond() ensures that + mysys_var->current_mutex is always unlocked _before_ mysys_var->mutex is + locked (if that would not be the case, you'll get a deadlock if someone + does a THD::awake() on you). + */ + pthread_mutex_unlock(thd->mysys_var->current_mutex); + pthread_mutex_lock(&thd->mysys_var->mutex); + thd->mysys_var->current_mutex = 0; + thd->mysys_var->current_cond = 0; + thd->proc_info = old_msg; + pthread_mutex_unlock(&thd->mysys_var->mutex); + return; +} + extern "C" void **thd_ha_data(const THD *thd, const struct handlerton *hton) { diff --git a/sql/sql_class.h b/sql/sql_class.h index f52d5fae76f..3d7ff0ca0a1 100644 --- a/sql/sql_class.h +++ b/sql/sql_class.h @@ -22,6 +22,7 @@ #include "log.h" #include "rpl_tblmap.h" +#include "replication.h" /** An interface that is used to take an action when @@ -1940,27 +1941,11 @@ public: inline const char* enter_cond(pthread_cond_t *cond, pthread_mutex_t* mutex, const char* msg) { - const char* old_msg = proc_info; - safe_mutex_assert_owner(mutex); - mysys_var->current_mutex = mutex; - mysys_var->current_cond = cond; - proc_info = msg; - return old_msg; + return thd_enter_cond(this, cond, mutex, msg); } inline void exit_cond(const char* old_msg) { - /* - Putting the mutex unlock in exit_cond() ensures that - mysys_var->current_mutex is always unlocked _before_ mysys_var->mutex is - locked (if that would not be the case, you'll get a deadlock if someone - does a THD::awake() on you). - */ - pthread_mutex_unlock(mysys_var->current_mutex); - pthread_mutex_lock(&mysys_var->mutex); - mysys_var->current_mutex = 0; - mysys_var->current_cond = 0; - proc_info = old_msg; - pthread_mutex_unlock(&mysys_var->mutex); + thd_exit_cond(this, old_msg); } inline time_t query_start() { query_start_used=1; return start_time; } inline void set_time() diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc index ca27d476213..6019c385fb8 100644 --- a/sql/sql_parse.cc +++ b/sql/sql_parse.cc @@ -21,6 +21,7 @@ #include #include #include +#include "rpl_handler.h" #include "sp_head.h" #include "sp.h" diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc index da168d36429..8cf8c4cb81f 100644 --- a/sql/sql_plugin.cc +++ b/sql/sql_plugin.cc @@ -19,14 +19,6 @@ #define REPORT_TO_LOG 1 #define REPORT_TO_USER 2 -#ifdef DBUG_OFF -#define plugin_ref_to_int(A) A -#define plugin_int_to_ref(A) A -#else -#define plugin_ref_to_int(A) (A ? A[0] : NULL) -#define plugin_int_to_ref(A) &(A) -#endif - extern struct st_mysql_plugin *mysqld_builtins[]; /** @@ -54,7 +46,8 @@ const LEX_STRING plugin_type_names[MYSQL_MAX_PLUGIN_TYPE_NUM]= { C_STRING_WITH_LEN("STORAGE ENGINE") }, { C_STRING_WITH_LEN("FTPARSER") }, { C_STRING_WITH_LEN("DAEMON") }, - { C_STRING_WITH_LEN("INFORMATION SCHEMA") } + { C_STRING_WITH_LEN("INFORMATION SCHEMA") }, + { C_STRING_WITH_LEN("REPLICATION") }, }; extern int initialize_schema_table(st_plugin_int *plugin); @@ -93,7 +86,8 @@ static int min_plugin_info_interface_version[MYSQL_MAX_PLUGIN_TYPE_NUM]= MYSQL_HANDLERTON_INTERFACE_VERSION, MYSQL_FTPARSER_INTERFACE_VERSION, MYSQL_DAEMON_INTERFACE_VERSION, - MYSQL_INFORMATION_SCHEMA_INTERFACE_VERSION + MYSQL_INFORMATION_SCHEMA_INTERFACE_VERSION, + MYSQL_REPLICATION_INTERFACE_VERSION, }; static int cur_plugin_info_interface_version[MYSQL_MAX_PLUGIN_TYPE_NUM]= { @@ -101,7 +95,8 @@ static int cur_plugin_info_interface_version[MYSQL_MAX_PLUGIN_TYPE_NUM]= MYSQL_HANDLERTON_INTERFACE_VERSION, MYSQL_FTPARSER_INTERFACE_VERSION, MYSQL_DAEMON_INTERFACE_VERSION, - MYSQL_INFORMATION_SCHEMA_INTERFACE_VERSION + MYSQL_INFORMATION_SCHEMA_INTERFACE_VERSION, + MYSQL_REPLICATION_INTERFACE_VERSION, }; static bool initialized= 0; diff --git a/sql/sql_plugin.h b/sql/sql_plugin.h index 004d0d5abb7..c6ad846943c 100644 --- a/sql/sql_plugin.h +++ b/sql/sql_plugin.h @@ -18,6 +18,14 @@ class sys_var; +#ifdef DBUG_OFF +#define plugin_ref_to_int(A) A +#define plugin_int_to_ref(A) A +#else +#define plugin_ref_to_int(A) (A ? A[0] : NULL) +#define plugin_int_to_ref(A) &(A) +#endif + /* the following flags are valid for plugin_init() */ diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 0ec8d91214c..671f6785640 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -21,6 +21,7 @@ #include "log_event.h" #include "rpl_filter.h" #include +#include "rpl_handler.h" int max_binlog_dump_events = 0; // unlimited my_bool opt_sporadic_binlog_dump_fail = 0; @@ -80,6 +81,32 @@ static int fake_rotate_event(NET* net, String* packet, char* log_file_name, DBUG_RETURN(0); } +/* + Reset thread transmit packet buffer for event sending + + This function allocates header bytes for event transmission, and + should be called before store the event data to the packet buffer. +*/ +static int reset_transmit_packet(THD *thd, ushort flags, + ulong *ev_offset, const char **errmsg) +{ + int ret= 0; + String *packet= &thd->packet; + + /* reserve and set default header */ + packet->length(0); + packet->set("\0", 1, &my_charset_bin); + + if (RUN_HOOK(binlog_transmit, reserve_header, (thd, flags, packet))) + { + *errmsg= "Failed to run hook 'reserve_header'"; + my_errno= ER_UNKNOWN_ERROR; + ret= 1; + } + *ev_offset= packet->length(); + return ret; +} + static int send_file(THD *thd) { NET* net = &thd->net; @@ -346,6 +373,9 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos, LOG_INFO linfo; char *log_file_name = linfo.log_file_name; char search_file_name[FN_REFLEN], *name; + + ulong ev_offset; + IO_CACHE log; File file = -1; String* packet = &thd->packet; @@ -361,6 +391,14 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos, DBUG_PRINT("enter",("log_ident: '%s' pos: %ld", log_ident, (long) pos)); bzero((char*) &log,sizeof(log)); + sql_print_information("Start binlog_dump to slave_server(%d), pos(%s, %lu)", + thd->server_id, log_ident, (ulong)pos); + if (RUN_HOOK(binlog_transmit, transmit_start, (thd, flags, log_ident, pos))) + { + errmsg= "Failed to run hook 'transmit_start'"; + my_errno= ER_UNKNOWN_ERROR; + goto err; + } #ifndef DBUG_OFF if (opt_sporadic_binlog_dump_fail && (binlog_dump_count++ % 2)) @@ -416,11 +454,9 @@ impossible position"; goto err; } - /* - We need to start a packet with something other than 255 - to distinguish it from error - */ - packet->set("\0", 1, &my_charset_bin); /* This is the start of a new packet */ + /* reset transmit packet for the fake rotate event below */ + if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg)) + goto err; /* Tell the client about the log name with a fake Rotate event; @@ -460,7 +496,7 @@ impossible position"; my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG; goto err; } - packet->set("\0", 1, &my_charset_bin); + /* Adding MAX_LOG_EVENT_HEADER_LEN, since a binlog event can become this larger than the corresponding packet (query) sent @@ -476,6 +512,11 @@ impossible position"; log_lock = mysql_bin_log.get_log_lock(); if (pos > BIN_LOG_HEADER_SIZE) { + /* reset transmit packet for the event read from binary log + file */ + if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg)) + goto err; + /* Try to find a Format_description_log_event at the beginning of the binlog @@ -483,29 +524,30 @@ impossible position"; if (!(error = Log_event::read_log_event(&log, packet, log_lock))) { /* - The packet has offsets equal to the normal offsets in a binlog - event +1 (the first character is \0). + The packet has offsets equal to the normal offsets in a + binlog event + ev_offset (the first ev_offset characters are + the header (default \0)). */ DBUG_PRINT("info", ("Looked for a Format_description_log_event, found event type %d", - (*packet)[EVENT_TYPE_OFFSET+1])); - if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT) + (*packet)[EVENT_TYPE_OFFSET+ev_offset])); + if ((*packet)[EVENT_TYPE_OFFSET+ev_offset] == FORMAT_DESCRIPTION_EVENT) { - binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] & + binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+ev_offset] & LOG_EVENT_BINLOG_IN_USE_F); - (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F; + (*packet)[FLAGS_OFFSET+ev_offset] &= ~LOG_EVENT_BINLOG_IN_USE_F; /* mark that this event with "log_pos=0", so the slave should not increment master's binlog position (rli->group_master_log_pos) */ - int4store((char*) packet->ptr()+LOG_POS_OFFSET+1, 0); + int4store((char*) packet->ptr()+LOG_POS_OFFSET+ev_offset, 0); /* if reconnect master sends FD event with `created' as 0 to avoid destroying temp tables. */ int4store((char*) packet->ptr()+LOG_EVENT_MINIMAL_HEADER_LEN+ - ST_CREATED_OFFSET+1, (ulong) 0); + ST_CREATED_OFFSET+ev_offset, (ulong) 0); /* send it */ if (my_net_write(net, (uchar*) packet->ptr(), packet->length())) { @@ -531,8 +573,6 @@ impossible position"; Format_description_log_event will be found naturally if it is written. */ } - /* reset the packet as we wrote to it in any case */ - packet->set("\0", 1, &my_charset_bin); } /* end of if (pos > BIN_LOG_HEADER_SIZE); */ else { @@ -544,6 +584,12 @@ impossible position"; while (!net->error && net->vio != 0 && !thd->killed) { + Log_event_type event_type= UNKNOWN_EVENT; + + /* reset the transmit packet for the event read from binary log + file */ + if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg)) + goto err; while (!(error = Log_event::read_log_event(&log, packet, log_lock))) { #ifndef DBUG_OFF @@ -556,15 +602,25 @@ impossible position"; } #endif - if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT) + event_type= (Log_event_type)((*packet)[LOG_EVENT_OFFSET+ev_offset]); + if (event_type == FORMAT_DESCRIPTION_EVENT) { - binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] & + binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+ev_offset] & LOG_EVENT_BINLOG_IN_USE_F); - (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F; + (*packet)[FLAGS_OFFSET+ev_offset] &= ~LOG_EVENT_BINLOG_IN_USE_F; } - else if ((*packet)[EVENT_TYPE_OFFSET+1] == STOP_EVENT) + else if (event_type == STOP_EVENT) binlog_can_be_corrupted= FALSE; + pos = my_b_tell(&log); + if (RUN_HOOK(binlog_transmit, before_send_event, + (thd, flags, packet, log_file_name, pos))) + { + my_errno= ER_UNKNOWN_ERROR; + errmsg= "run 'before_send_event' hook failed"; + goto err; + } + if (my_net_write(net, (uchar*) packet->ptr(), packet->length())) { errmsg = "Failed on my_net_write()"; @@ -572,9 +628,8 @@ impossible position"; goto err; } - DBUG_PRINT("info", ("log event code %d", - (*packet)[LOG_EVENT_OFFSET+1] )); - if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT) + DBUG_PRINT("info", ("log event code %d", event_type)); + if (event_type == LOAD_EVENT) { if (send_file(thd)) { @@ -583,7 +638,17 @@ impossible position"; goto err; } } - packet->set("\0", 1, &my_charset_bin); + + if (RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet))) + { + errmsg= "Failed to run hook 'after_send_event'"; + my_errno= ER_UNKNOWN_ERROR; + goto err; + } + + /* reset transmit packet for next loop */ + if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg)) + goto err; } /* @@ -634,6 +699,11 @@ impossible position"; } #endif + /* reset the transmit packet for the event read from binary log + file */ + if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg)) + goto err; + /* No one will update the log while we are reading now, but we'll be quick and just read one record @@ -650,6 +720,7 @@ impossible position"; /* we read successfully, so we'll need to send it to the slave */ pthread_mutex_unlock(log_lock); read_packet = 1; + event_type= (Log_event_type)((*packet)[LOG_EVENT_OFFSET+ev_offset]); break; case LOG_READ_EOF: @@ -676,8 +747,17 @@ impossible position"; } if (read_packet) - { - thd_proc_info(thd, "Sending binlog event to slave"); + { + thd_proc_info(thd, "Sending binlog event to slave"); + pos = my_b_tell(&log); + if (RUN_HOOK(binlog_transmit, before_send_event, + (thd, flags, packet, log_file_name, pos))) + { + my_errno= ER_UNKNOWN_ERROR; + errmsg= "run 'before_send_event' hook failed"; + goto err; + } + if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) ) { errmsg = "Failed on my_net_write()"; @@ -685,7 +765,7 @@ impossible position"; goto err; } - if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT) + if (event_type == LOAD_EVENT) { if (send_file(thd)) { @@ -694,11 +774,13 @@ impossible position"; goto err; } } - packet->set("\0", 1, &my_charset_bin); - /* - No need to net_flush because we will get to flush later when - we hit EOF pretty quick - */ + + if (RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet))) + { + my_errno= ER_UNKNOWN_ERROR; + errmsg= "Failed to run hook 'after_send_event'"; + goto err; + } } if (fatal_error) @@ -734,6 +816,10 @@ impossible position"; end_io_cache(&log); (void) my_close(file, MYF(MY_WME)); + /* reset transmit packet for the possible fake rotate event */ + if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg)) + goto err; + /* Call fake_rotate_event() in case the previous log (the one which we have just finished reading) did not contain a Rotate event @@ -750,9 +836,6 @@ impossible position"; my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG; goto err; } - - packet->length(0); - packet->append('\0'); } } @@ -760,6 +843,7 @@ end: end_io_cache(&log); (void)my_close(file, MYF(MY_WME)); + RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags)); my_eof(thd); thd_proc_info(thd, "Waiting to finalize termination"); pthread_mutex_lock(&LOCK_thread_count); @@ -770,6 +854,7 @@ end: err: thd_proc_info(thd, "Waiting to finalize termination"); end_io_cache(&log); + RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags)); /* Exclude iteration through thread list this is needed for purge_logs() - it will iterate through @@ -1064,6 +1149,7 @@ int reset_slave(THD *thd, Master_info* mi) goto err; } + RUN_HOOK(binlog_relay_io, after_reset_slave, (thd, mi)); err: unlock_slave_threads(mi); if (error) @@ -1363,7 +1449,11 @@ int reset_master(THD* thd) ER(ER_FLUSH_MASTER_BINLOG_CLOSED), MYF(ME_BELL+ME_WAITTANG)); return 1; } - return mysql_bin_log.reset_logs(thd); + + if (mysql_bin_log.reset_logs(thd)) + return 1; + RUN_HOOK(binlog_transmit, after_reset_master, (thd, 0 /* flags */)); + return 0; } int cmp_master_pos(const char* log_file_name1, ulonglong log_pos1, @@ -1836,5 +1926,3 @@ int init_replication_sys_vars() } #endif /* HAVE_REPLICATION */ - - -- cgit v1.2.1 From f0886a4d9dfc36e92f7254c958ec73476704c4d4 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Tue, 29 Sep 2009 00:04:20 +0100 Subject: BUG#28777, WL#4293: SHOW BINLOG EVENTS does not work on relay log files NOTE: this is the backport to next-mr. SHOW BINLOG EVENTS does not work with relay log files. If issuing "SHOW BINLOG EVENTS IN 'relay-log.000001'" in a non-empty relay log file (relay-log.000001), mysql reports empty set. This patch addresses this issue by extending the SHOW command with RELAYLOG. Events in relay log files can now be inspected by issuing SHOW RELAYLOG EVENTS [IN 'log_name'] [FROM pos] [LIMIT [offset,] row_count]. mysql-test/extra/rpl_tests/rpl_show_relaylog_events.inc: Shared part of the test case. mysql-test/include/show_binlog_events.inc: Added options $binary_log_file, $binary_log_limit_row, $binary_log_limit_offset so that show_binlog_events can take same parameters as SHOW BINLOG EVENTS does. mysql-test/include/show_relaylog_events.inc: Clone of show_binlog_events for relaylog events. mysql-test/suite/rpl/t/rpl_row_show_relaylog_events.test: Test case for row based replication. mysql-test/suite/rpl/t/rpl_stm_mix_show_relaylog_events.test: Test case for statement and mixed mode replication. sql/lex.h: Added RELAYLOG symbol. sql/mysqld.cc: Added "show_relaylog_events" to status_vars. sql/sp_head.cc: Set SQLCOM_SHOW_RELAYLOG_EVENTS to return flags= sp_head::MULTI_RESULTS; in sp_get_flags_for_command as SQLCOM_SHOW_BINLOG_EVENTS does. sql/sql_lex.h: Added sql_command SQLCOM_SHOW_RELAYLOG_EVENTS to lex enum_sql_command. sql/sql_parse.cc: Added handling of SQLCOM_SHOW_RELAYLOG_EVENTS. sql/sql_repl.cc: mysql_show_binlog_events set to choose the log file to use based on the command issued (SHOW BINLOG|RELAYLOG EVENTS). sql/sql_yacc.yy: Added RELAYLOG to the grammar. --- sql/lex.h | 1 + sql/mysqld.cc | 1 + sql/sp_head.cc | 1 + sql/sql_lex.h | 2 +- sql/sql_parse.cc | 1 + sql/sql_repl.cc | 37 +++++++++++++++++++++++++++---------- sql/sql_yacc.yy | 7 +++++++ 7 files changed, 39 insertions(+), 11 deletions(-) (limited to 'sql') diff --git a/sql/lex.h b/sql/lex.h index b199a79350b..dd37d837045 100644 --- a/sql/lex.h +++ b/sql/lex.h @@ -429,6 +429,7 @@ static SYMBOL symbols[] = { { "REDUNDANT", SYM(REDUNDANT_SYM)}, { "REFERENCES", SYM(REFERENCES)}, { "REGEXP", SYM(REGEXP)}, + { "RELAYLOG", SYM(RELAYLOG_SYM)}, { "RELAY_LOG_FILE", SYM(RELAY_LOG_FILE_SYM)}, { "RELAY_LOG_POS", SYM(RELAY_LOG_POS_SYM)}, { "RELAY_THREAD", SYM(RELAY_THREAD)}, diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 7e9eb6e7291..9b70096eb73 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -3135,6 +3135,7 @@ SHOW_VAR com_status_vars[]= { {"show_processlist", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_PROCESSLIST]), SHOW_LONG_STATUS}, {"show_profile", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_PROFILE]), SHOW_LONG_STATUS}, {"show_profiles", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_PROFILES]), SHOW_LONG_STATUS}, + {"show_relaylog_events", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_RELAYLOG_EVENTS]), SHOW_LONG_STATUS}, {"show_slave_hosts", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_SLAVE_HOSTS]), SHOW_LONG_STATUS}, {"show_slave_status", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_SLAVE_STAT]), SHOW_LONG_STATUS}, {"show_status", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_STATUS]), SHOW_LONG_STATUS}, diff --git a/sql/sp_head.cc b/sql/sp_head.cc index 0736e5fc2a8..9fed5db6e3a 100644 --- a/sql/sp_head.cc +++ b/sql/sp_head.cc @@ -174,6 +174,7 @@ sp_get_flags_for_command(LEX *lex) case SQLCOM_SHOW_AUTHORS: case SQLCOM_SHOW_BINLOGS: case SQLCOM_SHOW_BINLOG_EVENTS: + case SQLCOM_SHOW_RELAYLOG_EVENTS: case SQLCOM_SHOW_CHARSETS: case SQLCOM_SHOW_COLLATIONS: case SQLCOM_SHOW_COLUMN_TYPES: diff --git a/sql/sql_lex.h b/sql/sql_lex.h index 6f9f667a75a..7db239cf9e3 100644 --- a/sql/sql_lex.h +++ b/sql/sql_lex.h @@ -121,7 +121,7 @@ enum enum_sql_command { SQLCOM_SHOW_CREATE_TRIGGER, SQLCOM_ALTER_DB_UPGRADE, SQLCOM_SHOW_PROFILE, SQLCOM_SHOW_PROFILES, - + SQLCOM_SHOW_RELAYLOG_EVENTS, /* When a command is added here, be sure it's also added in mysqld.cc in "struct show_var_st status_vars[]= {" ... diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc index ca27d476213..513b9230c37 100644 --- a/sql/sql_parse.cc +++ b/sql/sql_parse.cc @@ -2319,6 +2319,7 @@ mysql_execute_command(THD *thd) res = show_slave_hosts(thd); break; } + case SQLCOM_SHOW_RELAYLOG_EVENTS: /* fall through */ case SQLCOM_SHOW_BINLOG_EVENTS: { if (check_global_access(thd, REPL_SLAVE_ACL)) diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 0ec8d91214c..4d9b7410b88 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -1401,6 +1401,7 @@ bool mysql_show_binlog_events(THD* thd) bool ret = TRUE; IO_CACHE log; File file = -1; + MYSQL_BIN_LOG *binary_log= NULL; DBUG_ENTER("mysql_show_binlog_events"); Log_event::init_show_field_list(&field_list); @@ -1411,14 +1412,30 @@ bool mysql_show_binlog_events(THD* thd) Format_description_log_event *description_event= new Format_description_log_event(3); /* MySQL 4.0 by default */ - /* - Wait for handlers to insert any pending information - into the binlog. For e.g. ndb which updates the binlog asynchronously - this is needed so that the uses sees all its own commands in the binlog - */ - ha_binlog_wait(thd); + DBUG_ASSERT(thd->lex->sql_command == SQLCOM_SHOW_BINLOG_EVENTS || + thd->lex->sql_command == SQLCOM_SHOW_RELAYLOG_EVENTS); - if (mysql_bin_log.is_open()) + /* select wich binary log to use: binlog or relay */ + if ( thd->lex->sql_command == SQLCOM_SHOW_BINLOG_EVENTS ) + { + /* + Wait for handlers to insert any pending information + into the binlog. For e.g. ndb which updates the binlog asynchronously + this is needed so that the uses sees all its own commands in the binlog + */ + ha_binlog_wait(thd); + + binary_log= &mysql_bin_log; + } + else /* showing relay log contents */ + { + if (!active_mi) + DBUG_RETURN(TRUE); + + binary_log= &(active_mi->rli.relay_log); + } + + if (binary_log->is_open()) { LEX_MASTER_INFO *lex_mi= &thd->lex->mi; SELECT_LEX_UNIT *unit= &thd->lex->unit; @@ -1426,7 +1443,7 @@ bool mysql_show_binlog_events(THD* thd) my_off_t pos = max(BIN_LOG_HEADER_SIZE, lex_mi->pos); // user-friendly char search_file_name[FN_REFLEN], *name; const char *log_file_name = lex_mi->log_file_name; - pthread_mutex_t *log_lock = mysql_bin_log.get_log_lock(); + pthread_mutex_t *log_lock = binary_log->get_log_lock(); LOG_INFO linfo; Log_event* ev; @@ -1436,13 +1453,13 @@ bool mysql_show_binlog_events(THD* thd) name= search_file_name; if (log_file_name) - mysql_bin_log.make_log_name(search_file_name, log_file_name); + binary_log->make_log_name(search_file_name, log_file_name); else name=0; // Find first log linfo.index_file_offset = 0; - if (mysql_bin_log.find_log_pos(&linfo, name, 1)) + if (binary_log->find_log_pos(&linfo, name, 1)) { errmsg = "Could not find target log"; goto err; diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy index a18f57bf9cf..1b415fea52d 100644 --- a/sql/sql_yacc.yy +++ b/sql/sql_yacc.yy @@ -930,6 +930,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize); %token REDUNDANT_SYM %token REFERENCES /* SQL-2003-R */ %token REGEXP +%token RELAYLOG_SYM %token RELAY_LOG_FILE_SYM %token RELAY_LOG_POS_SYM %token RELAY_THREAD @@ -9947,6 +9948,11 @@ show_param: LEX *lex= Lex; lex->sql_command= SQLCOM_SHOW_BINLOG_EVENTS; } opt_limit_clause_init + | RELAYLOG_SYM EVENTS_SYM binlog_in binlog_from + { + LEX *lex= Lex; + lex->sql_command= SQLCOM_SHOW_RELAYLOG_EVENTS; + } opt_limit_clause_init | keys_or_index from_or_in table_ident opt_db where_clause { LEX *lex= Lex; @@ -11588,6 +11594,7 @@ keyword_sp: | REDO_BUFFER_SIZE_SYM {} | REDOFILE_SYM {} | REDUNDANT_SYM {} + | RELAYLOG_SYM {} | RELAY_LOG_FILE_SYM {} | RELAY_LOG_POS_SYM {} | RELAY_THREAD {} -- cgit v1.2.1 From 5983785ef4b04b865ea1d78c8d452642913a83f3 Mon Sep 17 00:00:00 2001 From: Andrei Elkin Date: Tue, 29 Sep 2009 14:16:23 +0300 Subject: WL#342 heartbeat backporting from 6.0 code base to 5.1. --- sql/lex.h | 1 + sql/log.cc | 53 ++++++++++++---- sql/log.h | 5 +- sql/log_event.cc | 14 ++++ sql/log_event.h | 57 +++++++++++++++++ sql/mysqld.cc | 36 +++++++++++ sql/rpl_mi.cc | 39 ++++++++++-- sql/rpl_mi.h | 2 + sql/slave.cc | 141 +++++++++++++++++++++++++++++++++++------ sql/slave.h | 12 +++- sql/sql_lex.h | 5 +- sql/sql_repl.cc | 190 ++++++++++++++++++++++++++++++++++++++++++++++++++----- sql/sql_yacc.yy | 49 +++++++++++++- 13 files changed, 541 insertions(+), 63 deletions(-) (limited to 'sql') diff --git a/sql/lex.h b/sql/lex.h index b199a79350b..790808a8c14 100644 --- a/sql/lex.h +++ b/sql/lex.h @@ -323,6 +323,7 @@ static SYMBOL symbols[] = { { "MASTER_SSL_KEY", SYM(MASTER_SSL_KEY_SYM)}, { "MASTER_SSL_VERIFY_SERVER_CERT", SYM(MASTER_SSL_VERIFY_SERVER_CERT_SYM)}, { "MASTER_USER", SYM(MASTER_USER_SYM)}, + { "MASTER_HEARTBEAT_PERIOD", SYM(MASTER_HEARTBEAT_PERIOD_SYM)}, { "MATCH", SYM(MATCH)}, { "MAX_CONNECTIONS_PER_HOUR", SYM(MAX_CONNECTIONS_PER_HOUR)}, { "MAX_QUERIES_PER_HOUR", SYM(MAX_QUERIES_PER_HOUR)}, diff --git a/sql/log.cc b/sql/log.cc index 1af2f3a4ddc..362df871ba9 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -2413,7 +2413,7 @@ const char *MYSQL_LOG::generate_name(const char *log_name, MYSQL_BIN_LOG::MYSQL_BIN_LOG() :bytes_written(0), prepared_xids(0), file_id(1), open_count(1), need_start_event(TRUE), m_table_map_version(0), - is_relay_log(0), + is_relay_log(0), signal_cnt(0), description_event_for_exec(0), description_event_for_queue(0) { /* @@ -4605,12 +4605,9 @@ err: /** - Wait until we get a signal that the binary log has been updated. + Wait until we get a signal that the relay log has been updated. @param thd Thread variable - @param is_slave If 0, the caller is the Binlog_dump thread from master; - if 1, the caller is the SQL thread from the slave. This - influences only thd->proc_info. @note One must have a lock on LOCK_log before calling this function. @@ -4618,22 +4615,53 @@ err: THD::enter_cond() (see NOTES in sql_class.h). */ -void MYSQL_BIN_LOG::wait_for_update(THD* thd, bool is_slave) +void MYSQL_BIN_LOG::wait_for_update_relay_log(THD* thd) { const char *old_msg; - DBUG_ENTER("wait_for_update"); + DBUG_ENTER("wait_for_update_relay_log"); old_msg= thd->enter_cond(&update_cond, &LOCK_log, - is_slave ? - "Has read all relay log; waiting for the slave I/O " - "thread to update it" : - "Has sent all binlog to slave; waiting for binlog " - "to be updated"); + "Slave has read all relay log; " + "waiting for the slave I/O " + "thread to update it" ); pthread_cond_wait(&update_cond, &LOCK_log); thd->exit_cond(old_msg); DBUG_VOID_RETURN; } +/** + Wait until we get a signal that the binary log has been updated. + Applies to master only. + + NOTES + @param[in] thd a THD struct + @param[in] timeout a pointer to a timespec; + NULL means to wait w/o timeout. + @retval 0 if got signalled on update + @retval non-0 if wait timeout elapsed + @note + LOCK_log must be taken before calling this function. + LOCK_log is being released while the thread is waiting. + LOCK_log is released by the caller. +*/ + +int MYSQL_BIN_LOG::wait_for_update_bin_log(THD* thd, + const struct timespec *timeout) +{ + int ret= 0; + const char* old_msg = thd->proc_info; + DBUG_ENTER("wait_for_update_bin_log"); + old_msg= thd->enter_cond(&update_cond, &LOCK_log, + "Master has sent all binlog to slave; " + "waiting for binlog to be updated"); + if (!timeout) + pthread_cond_wait(&update_cond, &LOCK_log); + else + ret= pthread_cond_timedwait(&update_cond, &LOCK_log, + const_cast(timeout)); + DBUG_RETURN(ret); +} + /** Close the log file. @@ -4846,6 +4874,7 @@ bool flush_error_log() void MYSQL_BIN_LOG::signal_update() { DBUG_ENTER("MYSQL_BIN_LOG::signal_update"); + signal_cnt++; pthread_cond_broadcast(&update_cond); DBUG_VOID_RETURN; } diff --git a/sql/log.h b/sql/log.h index d306d6f7182..8d6a90d8a35 100644 --- a/sql/log.h +++ b/sql/log.h @@ -284,7 +284,7 @@ public: /* This is relay log */ bool is_relay_log; - + ulong signal_cnt; // update of the counter is checked by heartbeat /* These describe the log's format. This is used only for relay logs. _for_exec is used by the SQL thread, _for_queue by the I/O thread. It's @@ -339,7 +339,8 @@ public: } void set_max_size(ulong max_size_arg); void signal_update(); - void wait_for_update(THD* thd, bool master_or_slave); + void wait_for_update_relay_log(THD* thd); + int wait_for_update_bin_log(THD* thd, const struct timespec * timeout); void set_need_start_event() { need_start_event = 1; } void init(bool no_auto_events_arg, ulong max_size); void init_pthread_objects(); diff --git a/sql/log_event.cc b/sql/log_event.cc index fb6a5230fda..6c240735d23 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -3643,6 +3643,7 @@ Format_description_log_event(uint8 binlog_ver, const char* server_ver) post_header_len[UPDATE_ROWS_EVENT-1]= post_header_len[DELETE_ROWS_EVENT-1]= 6;); post_header_len[INCIDENT_EVENT-1]= INCIDENT_HEADER_LEN; + post_header_len[HEARTBEAT_LOG_EVENT-1]= 0; // Sanity-check that all post header lengths are initialized. IF_DBUG({ @@ -9427,3 +9428,16 @@ st_print_event_info::st_print_event_info() open_cached_file(&body_cache, NULL, NULL, 0, flags); } #endif + + +#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) +Heartbeat_log_event::Heartbeat_log_event(const char* buf, uint event_len, + const Format_description_log_event* description_event) + :Log_event(buf, description_event) +{ + uint8 header_size= description_event->common_header_len; + ident_len = event_len - header_size; + set_if_smaller(ident_len,FN_REFLEN-1); + log_ident= buf + header_size; +} +#endif diff --git a/sql/log_event.h b/sql/log_event.h index 8202dddcc76..b481ae59502 100644 --- a/sql/log_event.h +++ b/sql/log_event.h @@ -250,6 +250,7 @@ struct sql_ex_info #define EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN (4 + 4 + 4 + 1) #define EXECUTE_LOAD_QUERY_HEADER_LEN (QUERY_HEADER_LEN + EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN) #define INCIDENT_HEADER_LEN 2 +#define HEARTBEAT_HEADER_LEN 0 /* Max number of possible extra bytes in a replication event compared to a packet (i.e. a query) sent from client to master; @@ -574,6 +575,12 @@ enum Log_event_type */ INCIDENT_EVENT= 26, + /* + Heartbeat event to be send by master at its idle time + to ensure master's online status to slave + */ + HEARTBEAT_LOG_EVENT= 27, + /* Add new events here - right above this comment! Existing events (except ENUM_END_EVENT) should never change their numbers @@ -689,6 +696,20 @@ typedef struct st_print_event_info } PRINT_EVENT_INFO; #endif +/** + the struct aggregates two paramenters that identify an event + uniquely in scope of communication of a particular master and slave couple. + I.e there can not be 2 events from the same staying connected master which + have the same coordinates. + @note + Such identifier is not yet unique generally as the event originating master + is resetable. Also the crashed master can be replaced with some other. +*/ +struct event_coordinates +{ + char * file_name; // binlog file name (directories stripped) + my_off_t pos; // event's position in the binlog file +}; /** @class Log_event @@ -3916,6 +3937,42 @@ static inline bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache, reinit_io_cache(cache, WRITE_CACHE, 0, FALSE, TRUE); } +#ifndef MYSQL_CLIENT +/***************************************************************************** + + Heartbeat Log Event class + + Replication event to ensure to slave that master is alive. + The event is originated by master's dump thread and sent straight to + slave without being logged. Slave itself does not store it in relay log + but rather uses a data for immediate checks and throws away the event. + + Two members of the class log_ident and Log_event::log_pos comprise + @see the event_coordinates instance. The coordinates that a heartbeat + instance carries correspond to the last event master has sent from + its binlog. + + ****************************************************************************/ +class Heartbeat_log_event: public Log_event +{ +public: + Heartbeat_log_event(const char* buf, uint event_len, + const Format_description_log_event* description_event); + Log_event_type get_type_code() { return HEARTBEAT_LOG_EVENT; } + bool is_valid() const + { + return (log_ident != NULL && + log_pos >= BIN_LOG_HEADER_SIZE); + } + const char * get_log_ident() { return log_ident; } + uint get_ident_len() { return ident_len; } + +private: + const char* log_ident; + uint ident_len; +}; +#endif + /** @} (end of group Replication) */ diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 7e9eb6e7291..4bbb49f47ff 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -7068,6 +7068,40 @@ static int show_slave_retried_trans(THD *thd, SHOW_VAR *var, char *buff) pthread_mutex_unlock(&LOCK_active_mi); return 0; } + +static int show_slave_received_heartbeats(THD *thd, SHOW_VAR *var, char *buff) +{ + pthread_mutex_lock(&LOCK_active_mi); + if (active_mi) + { + var->type= SHOW_LONGLONG; + var->value= buff; + pthread_mutex_lock(&active_mi->rli.data_lock); + *((longlong *)buff)= active_mi->received_heartbeats; + pthread_mutex_unlock(&active_mi->rli.data_lock); + } + else + var->type= SHOW_UNDEF; + pthread_mutex_unlock(&LOCK_active_mi); + return 0; +} + +static int show_heartbeat_period(THD *thd, SHOW_VAR *var, char *buff) +{ + pthread_mutex_lock(&LOCK_active_mi); + if (active_mi) + { + var->type= SHOW_CHAR; + var->value= buff; + my_sprintf(buff, (buff, "%.3f",active_mi->heartbeat_period)); + } + else + var->type= SHOW_UNDEF; + pthread_mutex_unlock(&LOCK_active_mi); + return 0; +} + + #endif /* HAVE_REPLICATION */ static int show_open_tables(THD *thd, SHOW_VAR *var, char *buff) @@ -7432,6 +7466,8 @@ SHOW_VAR status_vars[]= { {"Slave_open_temp_tables", (char*) &slave_open_temp_tables, SHOW_LONG}, #ifdef HAVE_REPLICATION {"Slave_retried_transactions",(char*) &show_slave_retried_trans, SHOW_FUNC}, + {"Slave_heartbeat_period", (char*) &show_heartbeat_period, SHOW_FUNC}, + {"Slave_received_heartbeats",(char*) &show_slave_received_heartbeats, SHOW_FUNC}, {"Slave_running", (char*) &show_slave_running, SHOW_FUNC}, #endif {"Slow_launch_threads", (char*) &slow_launch_threads, SHOW_LONG}, diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc index 5e46837e948..77f7b7e1929 100644 --- a/sql/rpl_mi.cc +++ b/sql/rpl_mi.cc @@ -26,12 +26,13 @@ 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, const char *default_val); +int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val); Master_info::Master_info() :Slave_reporting_capability("I/O"), ssl(0), ssl_verify_server_cert(0), fd(-1), io_thd(0), inited(0), - abort_slave(0),slave_running(0), - slave_run_id(0) + abort_slave(0),slave_running(0), slave_run_id(0), + heartbeat_period(0), received_heartbeats(0) { host[0] = 0; user[0] = 0; password[0] = 0; ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0; @@ -84,6 +85,17 @@ void init_master_info_with_options(Master_info* mi) strmake(mi->ssl_key, master_ssl_key, sizeof(mi->ssl_key)-1); /* Intentionally init ssl_verify_server_cert to 0, no option available */ mi->ssl_verify_server_cert= 0; + /* + always request heartbeat unless master_heartbeat_period is set + explicitly zero. Here is the default value for heartbeat period + if CHANGE MASTER did not specify it. (no data loss in conversion + as hb period has a max) + */ + mi->heartbeat_period= (float) min(SLAVE_MAX_HEARTBEAT_PERIOD, + (slave_net_timeout/2.0)); + DBUG_ASSERT(mi->heartbeat_period > (float) 0.001 + || mi->heartbeat_period == 0); + DBUG_VOID_RETURN; } @@ -94,8 +106,11 @@ enum { /* 5.1.16 added value of master_ssl_verify_server_cert */ LINE_FOR_MASTER_SSL_VERIFY_SERVER_CERT= 15, + /* 6.0 added value of master_heartbeat_period */ + LINE_FOR_MASTER_HEARTBEAT_PERIOD= 16, + /* Number of lines currently used when saving master info file */ - LINES_IN_MASTER_INFO= LINE_FOR_MASTER_SSL_VERIFY_SERVER_CERT + LINES_IN_MASTER_INFO= LINE_FOR_MASTER_HEARTBEAT_PERIOD }; int init_master_info(Master_info* mi, const char* master_info_fname, @@ -197,6 +212,7 @@ file '%s')", fname); mi->fd = fd; int port, connect_retry, master_log_pos, lines; int ssl= 0, ssl_verify_server_cert= 0; + float master_heartbeat_period= 0.0; char *first_non_digit; /* @@ -281,7 +297,13 @@ file '%s')", fname); if (lines >= LINE_FOR_MASTER_SSL_VERIFY_SERVER_CERT && init_intvar_from_file(&ssl_verify_server_cert, &mi->file, 0)) goto errwithmsg; - + /* + Starting from 6.0 master_heartbeat_period might be + in the file + */ + if (lines >= LINE_FOR_MASTER_HEARTBEAT_PERIOD && + init_floatvar_from_file(&master_heartbeat_period, &mi->file, 0.0)) + goto errwithmsg; } #ifndef HAVE_OPENSSL @@ -300,6 +322,7 @@ file '%s')", fname); mi->connect_retry= (uint) connect_retry; mi->ssl= (my_bool) ssl; mi->ssl_verify_server_cert= ssl_verify_server_cert; + mi->heartbeat_period= master_heartbeat_period; } DBUG_PRINT("master_info",("log_file_name: %s position: %ld", mi->master_log_name, @@ -378,16 +401,18 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) contents of file). But because of number of lines in the first line of file we don't care about this garbage. */ - + char heartbeat_buf[sizeof(mi->heartbeat_period) * 4]; // buffer to suffice always + my_sprintf(heartbeat_buf, (heartbeat_buf, "%.3f", mi->heartbeat_period)); my_b_seek(file, 0L); my_b_printf(file, - "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n", + "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n", LINES_IN_MASTER_INFO, mi->master_log_name, llstr(mi->master_log_pos, lbuf), mi->host, mi->user, mi->password, mi->port, mi->connect_retry, (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert, - mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert); + mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert, + heartbeat_buf); DBUG_RETURN(-flush_io_cache(file)); } diff --git a/sql/rpl_mi.h b/sql/rpl_mi.h index 93fb0a98198..35e18414932 100644 --- a/sql/rpl_mi.h +++ b/sql/rpl_mi.h @@ -83,6 +83,8 @@ class Master_info : public Slave_reporting_capability Relay_log_info rli; uint port; uint connect_retry; + float heartbeat_period; // interface with CHANGE MASTER or master.info + ulonglong received_heartbeats; // counter of received heartbeat events #ifndef DBUG_OFF int events_till_disconnect; #endif diff --git a/sql/slave.cc b/sql/slave.cc index fac9ee214c5..4a161a345eb 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -770,7 +770,6 @@ static bool sql_slave_killed(THD* thd, Relay_log_info* rli) DBUG_RETURN(0); } - /* skip_load_data_infile() @@ -860,6 +859,37 @@ int init_intvar_from_file(int* var, IO_CACHE* f, int default_val) DBUG_RETURN(1); } +int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val) +{ + char buf[16]; + DBUG_ENTER("init_floatvar_from_file"); + + + if (my_b_gets(f, buf, sizeof(buf))) + { + if (sscanf(buf, "%f", var) != 1) + DBUG_RETURN(1); + else + DBUG_RETURN(0); + } + else if (default_val != 0.0) + { + *var = default_val; + DBUG_RETURN(0); + } + DBUG_RETURN(1); +} + +static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info) +{ + if (io_slave_killed(thd, mi)) + { + if (info && global_system_variables.log_warnings) + sql_print_information(info); + return TRUE; + } + return FALSE; +} /* Check if the error is caused by network. @@ -1189,6 +1219,32 @@ when it try to get the value of TIME_ZONE global variable from master."; } } + if (mi->heartbeat_period != 0.0) + { + char llbuf[22]; + const char query_format[]= "SET @master_heartbeat_period= %s"; + char query[sizeof(query_format) - 2 + sizeof(llbuf)]; + /* + the period is an ulonglong of nano-secs. + */ + llstr((ulonglong) (mi->heartbeat_period*1000000000UL), llbuf); + my_sprintf(query, (query, query_format, llbuf)); + + if (mysql_real_query(mysql, query, strlen(query)) + && !check_io_slave_killed(mi->io_thd, mi, NULL)) + { + errmsg= "The slave I/O thread stops because querying master with '%s' " + "failed; error: '%s' "; + err_code= ER_SLAVE_FATAL_ERROR; + sprintf(err_buff, "%s Error: %s", errmsg, + query, mysql_error(mysql)); + mysql_free_result(mysql_store_result(mysql)); + goto err; + } + mysql_free_result(mysql_store_result(mysql)); + } + + err: if (errmsg) { @@ -2381,18 +2437,6 @@ on this slave.\ } -static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info) -{ - if (io_slave_killed(thd, mi)) - { - if (info && global_system_variables.log_warnings) - sql_print_information(info); - return TRUE; - } - return FALSE; -} - - /** @brief Try to reconnect slave IO thread. @@ -3552,6 +3596,7 @@ static int queue_old_event(Master_info *mi, const char *buf, static int queue_event(Master_info* mi,const char* buf, ulong event_len) { int error= 0; + String error_msg; ulong inc_pos; Relay_log_info *rli= &mi->rli; pthread_mutex_t *log_lock= rli->relay_log.get_log_lock(); @@ -3586,7 +3631,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) Rotate_log_event rev(buf,event_len,mi->rli.relay_log.description_event_for_queue); if (unlikely(process_io_rotate(mi,&rev))) { - error= 1; + error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE; goto err; } /* @@ -3613,7 +3658,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) Log_event::read_log_event(buf, event_len, &errmsg, mi->rli.relay_log.description_event_for_queue))) { - error= 2; + error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE; goto err; } delete mi->rli.relay_log.description_event_for_queue; @@ -3632,6 +3677,56 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) } break; + + case HEARTBEAT_LOG_EVENT: + { + /* + HB (heartbeat) cannot come before RL (Relay) + */ + char llbuf[22]; + Heartbeat_log_event hb(buf, event_len, mi->rli.relay_log.description_event_for_queue); + if (!hb.is_valid()) + { + error= ER_SLAVE_HEARTBEAT_FAILURE; + error_msg.append(STRING_WITH_LEN("inconsistent heartbeat event content;")); + error_msg.append(STRING_WITH_LEN("the event's data: log_file_name ")); + error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident())); + error_msg.append(STRING_WITH_LEN(" log_pos ")); + llstr(hb.log_pos, llbuf); + error_msg.append(llbuf, strlen(llbuf)); + goto err; + } + mi->received_heartbeats++; + /* + compare local and event's versions of log_file, log_pos. + + Heartbeat is sent only after an event corresponding to the corrdinates + the heartbeat carries. + Slave can not have a difference in coordinates except in the only + special case when mi->master_log_name, master_log_pos have never + been updated by Rotate event i.e when slave does not have any history + with the master (and thereafter mi->master_log_pos is NULL). + + TODO: handling `when' for SHOW SLAVE STATUS' snds behind + */ + if ((memcmp(mi->master_log_name, hb.get_log_ident(), hb.get_ident_len()) + && mi->master_log_name != NULL) + || mi->master_log_pos != hb.log_pos) + { + /* missed events of heartbeat from the past */ + error= ER_SLAVE_HEARTBEAT_FAILURE; + error_msg.append(STRING_WITH_LEN("heartbeat is not compatible with local info;")); + error_msg.append(STRING_WITH_LEN("the event's data: log_file_name ")); + error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident())); + error_msg.append(STRING_WITH_LEN(" log_pos ")); + llstr(hb.log_pos, llbuf); + error_msg.append(llbuf, strlen(llbuf)); + goto err; + } + goto skip_relay_logging; + } + break; + default: inc_pos= event_len; break; @@ -3692,15 +3787,23 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) rli->relay_log.harvest_bytes_written(&rli->log_space_total); } else - error= 3; + { + error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE; + } rli->ign_master_log_name_end[0]= 0; // last event is not ignored } pthread_mutex_unlock(log_lock); - +skip_relay_logging: + err: pthread_mutex_unlock(&mi->data_lock); DBUG_PRINT("info", ("error: %d", error)); + if (error) + mi->report(ERROR_LEVEL, error, ER(error), + (error == ER_SLAVE_RELAY_LOG_WRITE_FAILURE)? + "could not queue event from master" : + error_msg.ptr()); DBUG_RETURN(error); } @@ -4208,8 +4311,8 @@ static Log_event* next_event(Relay_log_info* rli) */ pthread_mutex_unlock(&rli->log_space_lock); pthread_cond_broadcast(&rli->log_space_cond); - // Note that wait_for_update unlocks lock_log ! - rli->relay_log.wait_for_update(rli->sql_thd, 1); + // Note that wait_for_update_relay_log unlocks lock_log ! + rli->relay_log.wait_for_update_relay_log(rli->sql_thd); // re-acquire data lock since we released it earlier pthread_mutex_lock(&rli->data_lock); rli->last_master_timestamp= save_timestamp; diff --git a/sql/slave.h b/sql/slave.h index a44a7eed83e..e8364090eb4 100644 --- a/sql/slave.h +++ b/sql/slave.h @@ -22,6 +22,17 @@ @file */ + +/** + Some of defines are need in parser even though replication is not + compiled in (embedded). +*/ + +/** + The maximum is defined as (ULONG_MAX/1000) with 4 bytes ulong +*/ +#define SLAVE_MAX_HEARTBEAT_PERIOD 4294967 + #ifdef HAVE_REPLICATION #include "log.h" @@ -33,7 +44,6 @@ #define MAX_SLAVE_ERROR 2000 - // Forward declarations class Relay_log_info; class Master_info; diff --git a/sql/sql_lex.h b/sql/sql_lex.h index 6f9f667a75a..f6effab93a4 100644 --- a/sql/sql_lex.h +++ b/sql/sql_lex.h @@ -206,14 +206,15 @@ typedef struct st_lex_master_info { char *host, *user, *password, *log_file_name; uint port, connect_retry; + float heartbeat_period; ulonglong pos; ulong server_id; /* Enum is used for making it possible to detect if the user changed variable or if it should be left at old value */ - enum {SSL_UNCHANGED, SSL_DISABLE, SSL_ENABLE} - ssl, ssl_verify_server_cert; + enum {LEX_MI_UNCHANGED, LEX_MI_DISABLE, LEX_MI_ENABLE} + ssl, ssl_verify_server_cert, heartbeat_opt; char *ssl_key, *ssl_cert, *ssl_ca, *ssl_capath, *ssl_cipher; char *relay_log_name; ulong relay_log_pos; diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 0ec8d91214c..cde713b1b40 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -336,6 +336,74 @@ Increase max_allowed_packet on master"; } +/** + An auxiliary function for calling in mysql_binlog_send + to initialize the heartbeat timeout in waiting for a binlogged event. + + @param[in] thd THD to access a user variable + + @return heartbeat period an ulonglong of nanoseconds + or zero if heartbeat was not demanded by slave +*/ +static ulonglong get_heartbeat_period(THD * thd) +{ + my_bool null_value; + LEX_STRING name= { C_STRING_WITH_LEN("master_heartbeat_period")}; + user_var_entry *entry= + (user_var_entry*) hash_search(&thd->user_vars, (uchar*) name.str, + name.length); + return entry? entry->val_int(&null_value) : 0; +} + +/* + Function prepares and sends repliation heartbeat event. + + @param net net object of THD + @param packet buffer to store the heartbeat instance + @param event_coordinates binlog file name and position of the last + real event master sent from binlog + + @note + Among three essential pieces of heartbeat data Log_event::when + is computed locally. + The error to send is serious and should force terminating + the dump thread. +*/ +static int send_heartbeat_event(NET* net, String* packet, + const struct event_coordinates *coord) +{ + DBUG_ENTER("send_heartbeat_event"); + char header[LOG_EVENT_HEADER_LEN]; + /* + 'when' (the timestamp) is set to 0 so that slave could distinguish between + real and fake Rotate events (if necessary) + */ + memset(header, 0, 4); // when + + header[EVENT_TYPE_OFFSET] = HEARTBEAT_LOG_EVENT; + + char* p= coord->file_name + dirname_length(coord->file_name); + + uint ident_len = strlen(p); + ulong event_len = ident_len + LOG_EVENT_HEADER_LEN; + int4store(header + SERVER_ID_OFFSET, server_id); + int4store(header + EVENT_LEN_OFFSET, event_len); + int2store(header + FLAGS_OFFSET, 0); + + int4store(header + LOG_POS_OFFSET, coord->pos); // log_pos + + packet->append(header, sizeof(header)); + packet->append(p, ident_len); // log_file_name + + if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) || + net_flush(net)) + { + DBUG_RETURN(-1); + } + packet->set("\0", 1, &my_charset_bin); + DBUG_RETURN(0); +} + /* TODO: Clean up loop to only have one call to send_file() */ @@ -361,7 +429,22 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos, DBUG_PRINT("enter",("log_ident: '%s' pos: %ld", log_ident, (long) pos)); bzero((char*) &log,sizeof(log)); - + /* + heartbeat_period from @master_heartbeat_period user variable + */ + ulonglong heartbeat_period= get_heartbeat_period(thd); + struct timespec heartbeat_buf; + struct event_coordinates coord_buf; + struct timespec *heartbeat_ts= NULL; + struct event_coordinates *coord= NULL; + if (heartbeat_period != LL(0)) + { + heartbeat_ts= &heartbeat_buf; + set_timespec_nsec(*heartbeat_ts, 0); + coord= &coord_buf; + coord->file_name= log_file_name; // initialization basing on what slave remembers + coord->pos= pos; + } #ifndef DBUG_OFF if (opt_sporadic_binlog_dump_fail && (binlog_dump_count++ % 2)) { @@ -555,6 +638,11 @@ impossible position"; goto err; } #endif + /* + log's filename does not change while it's active + */ + if (coord) + coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET); if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT) { @@ -650,26 +738,65 @@ impossible position"; /* we read successfully, so we'll need to send it to the slave */ pthread_mutex_unlock(log_lock); read_packet = 1; + if (coord) + coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET); break; case LOG_READ_EOF: + { + int ret; + ulong signal_cnt; DBUG_PRINT("wait",("waiting for data in binary log")); if (thd->server_id==0) // for mysqlbinlog (mysqlbinlog.server_id==0) { pthread_mutex_unlock(log_lock); goto end; } - if (!thd->killed) - { - /* Note that the following call unlocks lock_log */ - mysql_bin_log.wait_for_update(thd, 0); - } - else - pthread_mutex_unlock(log_lock); - DBUG_PRINT("wait",("binary log received update")); - break; - default: +#ifndef DBUG_OFF + ulong hb_info_counter= 0; +#endif + signal_cnt= mysql_bin_log.signal_cnt; + do + { + if (coord) + { + DBUG_ASSERT(heartbeat_ts && heartbeat_period != LL(0)); + set_timespec_nsec(*heartbeat_ts, heartbeat_period); + } + ret= mysql_bin_log.wait_for_update_bin_log(thd, heartbeat_ts); + DBUG_ASSERT(ret == 0 || heartbeat_period != LL(0) && coord != NULL); + if (ret == ETIMEDOUT || ret == ETIME) + { +#ifndef DBUG_OFF + if (hb_info_counter < 3) + { + sql_print_information("master sends heartbeat message"); + hb_info_counter++; + if (hb_info_counter == 3) + sql_print_information("the rest of heartbeat info skipped ..."); + } +#endif + if (send_heartbeat_event(net, packet, coord)) + { + errmsg = "Failed on my_net_write()"; + my_errno= ER_UNKNOWN_ERROR; + pthread_mutex_unlock(log_lock); + goto err; + } + } + else + { + DBUG_ASSERT(ret == 0 && signal_cnt != mysql_bin_log.signal_cnt || + thd->killed); + DBUG_PRINT("wait",("binary log received update")); + } + } while (signal_cnt == mysql_bin_log.signal_cnt && !thd->killed); + pthread_mutex_unlock(log_lock); + } + break; + + default: pthread_mutex_unlock(log_lock); fatal_error = 1; break; @@ -753,6 +880,8 @@ impossible position"; packet->length(0); packet->append('\0'); + if (coord) + coord->file_name= log_file_name; // reset to the next } } @@ -1195,13 +1324,18 @@ bool change_master(THD* thd, Master_info* mi) mi->port = lex_mi->port; if (lex_mi->connect_retry) mi->connect_retry = lex_mi->connect_retry; + if (lex_mi->heartbeat_opt != LEX_MASTER_INFO::LEX_MI_UNCHANGED) + mi->heartbeat_period = lex_mi->heartbeat_period; + else + mi->heartbeat_period= (float) min(SLAVE_MAX_HEARTBEAT_PERIOD, + (slave_net_timeout/2.0)); + mi->received_heartbeats= LL(0); // counter lives until master is CHANGEd + if (lex_mi->ssl != LEX_MASTER_INFO::LEX_MI_UNCHANGED) + mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::LEX_MI_ENABLE); - if (lex_mi->ssl != LEX_MASTER_INFO::SSL_UNCHANGED) - mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::SSL_ENABLE); - - if (lex_mi->ssl_verify_server_cert != LEX_MASTER_INFO::SSL_UNCHANGED) + if (lex_mi->ssl_verify_server_cert != LEX_MASTER_INFO::LEX_MI_UNCHANGED) mi->ssl_verify_server_cert= - (lex_mi->ssl_verify_server_cert == LEX_MASTER_INFO::SSL_ENABLE); + (lex_mi->ssl_verify_server_cert == LEX_MASTER_INFO::LEX_MI_ENABLE); if (lex_mi->ssl_ca) strmake(mi->ssl_ca, lex_mi->ssl_ca, sizeof(mi->ssl_ca)-1); @@ -1745,6 +1879,26 @@ public: bool update(THD *thd, set_var *var); }; +static void fix_slave_net_timeout(THD *thd, enum_var_type type) +{ + DBUG_ENTER("fix_slave_net_timeout"); +#ifdef HAVE_REPLICATION + pthread_mutex_lock(&LOCK_active_mi); + DBUG_PRINT("info",("slave_net_timeout=%lu mi->heartbeat_period=%.3f", + slave_net_timeout, + (active_mi? active_mi->heartbeat_period : 0.0))); + if (active_mi && slave_net_timeout < active_mi->heartbeat_period) + push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN, + ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE, + "The currect value for master_heartbeat_period" + " exceeds the new value of `slave_net_timeout' sec." + " A sensible value for the period should be" + " less than the timeout."); + pthread_mutex_unlock(&LOCK_active_mi); +#endif + DBUG_VOID_RETURN; +} + static sys_var_chain vars = { NULL, NULL }; static sys_var_const sys_log_slave_updates(&vars, "log_slave_updates", @@ -1770,7 +1924,8 @@ static sys_var_const sys_slave_load_tmpdir(&vars, "slave_load_tmpdir", OPT_GLOBAL, SHOW_CHAR_PTR, (uchar*) &slave_load_tmpdir); static sys_var_long_ptr sys_slave_net_timeout(&vars, "slave_net_timeout", - &slave_net_timeout); + &slave_net_timeout, + fix_slave_net_timeout); static sys_var_const sys_slave_skip_errors(&vars, "slave_skip_errors", OPT_GLOBAL, SHOW_CHAR, (uchar*) slave_skip_error_names); @@ -1835,6 +1990,7 @@ int init_replication_sys_vars() return 0; } + #endif /* HAVE_REPLICATION */ diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy index a18f57bf9cf..50395d386e8 100644 --- a/sql/sql_yacc.yy +++ b/sql/sql_yacc.yy @@ -814,6 +814,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize); %token MASTER_SSL_VERIFY_SERVER_CERT_SYM %token MASTER_SYM %token MASTER_USER_SYM +%token MASTER_HEARTBEAT_PERIOD_SYM %token MATCH /* SQL-2003-R */ %token MAX_CONNECTIONS_PER_HOUR %token MAX_QUERIES_PER_HOUR @@ -1592,7 +1593,7 @@ master_def: | MASTER_SSL_SYM EQ ulong_num { Lex->mi.ssl= $3 ? - LEX_MASTER_INFO::SSL_ENABLE : LEX_MASTER_INFO::SSL_DISABLE; + LEX_MASTER_INFO::LEX_MI_ENABLE : LEX_MASTER_INFO::LEX_MI_DISABLE; } | MASTER_SSL_CA_SYM EQ TEXT_STRING_sys { @@ -1617,9 +1618,51 @@ master_def: | MASTER_SSL_VERIFY_SERVER_CERT_SYM EQ ulong_num { Lex->mi.ssl_verify_server_cert= $3 ? - LEX_MASTER_INFO::SSL_ENABLE : LEX_MASTER_INFO::SSL_DISABLE; + LEX_MASTER_INFO::LEX_MI_ENABLE : LEX_MASTER_INFO::LEX_MI_DISABLE; } - | master_file_def + + | MASTER_HEARTBEAT_PERIOD_SYM EQ NUM_literal + { + Lex->mi.heartbeat_period= (float) $3->val_real(); + if (Lex->mi.heartbeat_period > SLAVE_MAX_HEARTBEAT_PERIOD || + Lex->mi.heartbeat_period < 0.0) + { + const char format[]= "%d seconds"; + char buf[4*sizeof(SLAVE_MAX_HEARTBEAT_PERIOD) + sizeof(format)]; + my_sprintf(buf, (buf, format, SLAVE_MAX_HEARTBEAT_PERIOD)); + my_error(ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE, + MYF(0), + " is negative or exceeds the maximum ", + buf); + MYSQL_YYABORT; + } + if (Lex->mi.heartbeat_period > slave_net_timeout) + { + push_warning_printf(YYTHD, MYSQL_ERROR::WARN_LEVEL_WARN, + ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE, + ER(ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE), + " exceeds the value of `slave_net_timeout' sec.", + " A sensible value for the period should be" + " less than the timeout."); + } + if (Lex->mi.heartbeat_period < 0.001) + { + if (Lex->mi.heartbeat_period != 0.0) + { + push_warning_printf(YYTHD, MYSQL_ERROR::WARN_LEVEL_WARN, + ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE, + ER(ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE), + " is less than 1 msec.", + " The period is reset to zero which means" + " no heartbeats will be sending"); + Lex->mi.heartbeat_period= 0.0; + } + Lex->mi.heartbeat_opt= LEX_MASTER_INFO::LEX_MI_DISABLE; + } + Lex->mi.heartbeat_opt= LEX_MASTER_INFO::LEX_MI_ENABLE; + } + | + master_file_def ; master_file_def: -- cgit v1.2.1 From c676cebbea734691b41bf7e8edcbe61e74a53f62 Mon Sep 17 00:00:00 2001 From: Andrei Elkin Date: Tue, 29 Sep 2009 14:37:52 +0300 Subject: WL#342 heartbeat Improving an error report generating. --- sql/slave.cc | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) (limited to 'sql') diff --git a/sql/slave.cc b/sql/slave.cc index 4a161a345eb..bbdfb8e633f 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -1233,11 +1233,10 @@ when it try to get the value of TIME_ZONE global variable from master."; if (mysql_real_query(mysql, query, strlen(query)) && !check_io_slave_killed(mi->io_thd, mi, NULL)) { - errmsg= "The slave I/O thread stops because querying master with '%s' " - "failed; error: '%s' "; + errmsg= "The slave I/O thread stops because SET @master_heartbeat_period " + "on master failed."; err_code= ER_SLAVE_FATAL_ERROR; - sprintf(err_buff, "%s Error: %s", errmsg, - query, mysql_error(mysql)); + sprintf(err_buff, "%s Error: %s", errmsg, mysql_error(mysql)); mysql_free_result(mysql_store_result(mysql)); goto err; } -- cgit v1.2.1 From 569ca8590c76e75a06d3ec2f320b7ea2b6960072 Mon Sep 17 00:00:00 2001 From: Alfranio Correia Date: Tue, 29 Sep 2009 14:38:32 +0100 Subject: BUG#44663 Unused replication options prevent server from starting. NOTE: Backporting the patch to next-mr. The use of option log_slave_updates without log_bin was preventing the server from starting. To fix the problem, we replaced the error message and the exit call by a warning message. --- sql/mysqld.cc | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) (limited to 'sql') diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 9b70096eb73..a57ee04081f 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -3819,9 +3819,8 @@ with --log-bin instead."); } if (opt_log_slave_updates && !opt_bin_log) { - sql_print_error("You need to use --log-bin to make " + sql_print_warning("You need to use --log-bin to make " "--log-slave-updates work."); - unireg_abort(1); } if (!opt_bin_log) { @@ -3851,11 +3850,17 @@ with --log-bin instead."); #ifdef HAVE_REPLICATION if (opt_log_slave_updates && replicate_same_server_id) { - sql_print_error("\ -using --replicate-same-server-id in conjunction with \ + if (opt_bin_log) + { + sql_print_error("using --replicate-same-server-id in conjunction with \ --log-slave-updates is impossible, it would lead to infinite loops in this \ server."); - unireg_abort(1); + unireg_abort(1); + } + else + sql_print_warning("using --replicate-same-server-id in conjunction with \ +--log-slave-updates would lead to infinite loops in this server. However this \ +will be ignored as the --log-bin option is not defined."); } #endif -- cgit v1.2.1 From b7f887652b22d49574860d6948e67f0ea7a83294 Mon Sep 17 00:00:00 2001 From: Alfranio Correia Date: Tue, 29 Sep 2009 14:55:36 +0100 Subject: WL#4828 and BUG#45747 NOTE: Backporting the patch to next-mr. WL#4828 Augment DBUG_ENTER/DBUG_EXIT to crash MySQL in different functions ------- The assessment of the replication code in the presence of faults is extremely import to increase reliability. In particular, one needs to know if servers will either correctly recovery or print out appropriate error messages thus avoiding unexpected problems in a production environment. In order to accomplish this, the current patch refactories the debug macros already provided in the source code and introduces three new macros that allows to inject faults, specifically crashes, while entering or exiting a function or method. For instance, to crash a server while returning from the init_slave function (see module sql/slave.cc), one needs to do what follows: 1 - Modify the source replacing DBUG_RETURN by DBUG_CRASH_RETURN; DBUG_CRASH_RETURN(0); 2 - Use the debug variable to activate dbug instructions: SET SESSION debug="+d,init_slave_crash_return"; The new macros are briefly described below: DBUG_CRASH_ENTER (function) is equivalent to DBUG_ENTER which registers the beginning of a function but in addition to it allows for crashing the server while entering the function if the appropriate dbug instruction is activate. In this case, the dbug instruction should be "+d,function_crash_enter". DBUG_CRASH_RETURN (value) is equivalent to DBUG_RETURN which notifies the end of a function but in addition to it allows for crashing the server while returning from the function if the appropriate dbug instruction is activate. In this case, the dbug instruction should be "+d,function_crash_return". Note that "function" should be the same string used by either the DBUG_ENTER or DBUG_CRASH_ENTER. DBUG_CRASH_VOID_RETURN (value) is equivalent to DBUG_VOID_RETURN which notifies the end of a function but in addition to it allows for crashing the server while returning from the function if the appropriate dbug instruction is activate. In this case, the dbug instruction should be "+d,function_crash_return". Note that "function" should be the same string used by either the DBUG_ENTER or DBUG_CRASH_ENTER. To inject other faults, for instance, wrong return values, one should rely on the macros already available. The current patch also removes a set of macros that were either not being used or were redundant as other macros could be used to provide the same feature. In the future, we also consider dynamic instrumentation of the code. BUG#45747 DBUG_CRASH_* is not setting the strict option --------- When combining DBUG_CRASH_* with "--debug=d:t:i:A,file" the server crashes due to a call to the abort function in the DBUG_CRASH_* macro althought the appropriate keyword has not been set. --- sql/mysql_priv.h | 94 ---------------------------------------------------- sql/sql_class.cc | 3 -- sql/sql_class.h | 3 -- sql/sql_partition.cc | 4 +++ 4 files changed, 4 insertions(+), 100 deletions(-) (limited to 'sql') diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h index 381a0313add..52cfa7934c3 100644 --- a/sql/mysql_priv.h +++ b/sql/mysql_priv.h @@ -938,100 +938,6 @@ struct Query_cache_query_flags #define query_cache_is_cacheable_query(L) 0 #endif /*HAVE_QUERY_CACHE*/ -/* - Error injector Macros to enable easy testing of recovery after failures - in various error cases. -*/ -#ifndef ERROR_INJECT_SUPPORT - -#define ERROR_INJECT(x) 0 -#define ERROR_INJECT_ACTION(x,action) 0 -#define ERROR_INJECT_CRASH(x) 0 -#define ERROR_INJECT_VALUE(x) 0 -#define ERROR_INJECT_VALUE_ACTION(x,action) 0 -#define ERROR_INJECT_VALUE_CRASH(x) 0 -#define SET_ERROR_INJECT_VALUE(x) - -#else - -inline bool check_and_unset_keyword(const char *dbug_str) -{ - const char *extra_str= "-d,"; - char total_str[200]; - if (_db_strict_keyword_ (dbug_str)) - { - strxmov(total_str, extra_str, dbug_str, NullS); - DBUG_SET(total_str); - return 1; - } - return 0; -} - - -inline bool -check_and_unset_inject_value(int value) -{ - THD *thd= current_thd; - if (thd->error_inject_value == (uint)value) - { - thd->error_inject_value= 0; - return 1; - } - return 0; -} - -/* - ERROR INJECT MODULE: - -------------------- - These macros are used to insert macros from the application code. - The event that activates those error injections can be activated - from SQL by using: - SET SESSION dbug=+d,code; - - After the error has been injected, the macros will automatically - remove the debug code, thus similar to using: - SET SESSION dbug=-d,code - from SQL. - - ERROR_INJECT_CRASH will inject a crash of the MySQL Server if code - is set when macro is called. ERROR_INJECT_CRASH can be used in - if-statements, it will always return FALSE unless of course it - crashes in which case it doesn't return at all. - - ERROR_INJECT_ACTION will inject the action specified in the action - parameter of the macro, before performing the action the code will - be removed such that no more events occur. ERROR_INJECT_ACTION - can also be used in if-statements and always returns FALSE. - ERROR_INJECT can be used in a normal if-statement, where the action - part is performed in the if-block. The macro returns TRUE if the - error was activated and otherwise returns FALSE. If activated the - code is removed. - - Sometimes it is necessary to perform error inject actions as a serie - of events. In this case one can use one variable on the THD object. - Thus one sets this value by using e.g. SET_ERROR_INJECT_VALUE(100). - Then one can later test for it by using ERROR_INJECT_CRASH_VALUE, - ERROR_INJECT_ACTION_VALUE and ERROR_INJECT_VALUE. This have the same - behaviour as the above described macros except that they use the - error inject value instead of a code used by DBUG macros. -*/ -#define SET_ERROR_INJECT_VALUE(x) \ - current_thd->error_inject_value= (x) -#define ERROR_INJECT_CRASH(code) \ - DBUG_EVALUATE_IF(code, (abort(), 0), 0) -#define ERROR_INJECT_ACTION(code, action) \ - (check_and_unset_keyword(code) ? ((action), 0) : 0) -#define ERROR_INJECT(code) \ - check_and_unset_keyword(code) -#define ERROR_INJECT_VALUE(value) \ - check_and_unset_inject_value(value) -#define ERROR_INJECT_VALUE_ACTION(value,action) \ - (check_and_unset_inject_value(value) ? (action) : 0) -#define ERROR_INJECT_VALUE_CRASH(value) \ - ERROR_INJECT_VALUE_ACTION(value, (abort(), 0)) - -#endif - void write_bin_log(THD *thd, bool clear_error, char const *query, ulong query_length); diff --git a/sql/sql_class.cc b/sql/sql_class.cc index 3f568566c89..6c2133ae6c8 100644 --- a/sql/sql_class.cc +++ b/sql/sql_class.cc @@ -586,9 +586,6 @@ THD::THD() limit_found_rows= 0; row_count_func= -1; statement_id_counter= 0UL; -#ifdef ERROR_INJECT_SUPPORT - error_inject_value= 0UL; -#endif // Must be reset to handle error with THD's created for init of mysqld lex->current_select= 0; start_time=(time_t) 0; diff --git a/sql/sql_class.h b/sql/sql_class.h index 49f9bc3fd5e..47b51559ed1 100644 --- a/sql/sql_class.h +++ b/sql/sql_class.h @@ -1720,9 +1720,6 @@ public: query_id_t query_id, warn_id; ulong col_access; -#ifdef ERROR_INJECT_SUPPORT - ulong error_inject_value; -#endif /* Statement id is thread-wide. This counter is used to generate ids */ ulong statement_id_counter; ulong rand_saved_seed1, rand_saved_seed2; diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc index 08ff2daacb9..a4c0f02e480 100644 --- a/sql/sql_partition.cc +++ b/sql/sql_partition.cc @@ -40,6 +40,10 @@ #ifdef WITH_PARTITION_STORAGE_ENGINE #include "ha_partition.h" + +#define ERROR_INJECT_CRASH(code) \ + DBUG_EVALUATE_IF(code, (abort(), 0), 0) + /* Partition related functions declarations and some static constants; */ -- cgit v1.2.1 From cc9e25af54ac0a00cfe9930253a0e6de70f0c668 Mon Sep 17 00:00:00 2001 From: Alfranio Correia Date: Tue, 29 Sep 2009 15:04:21 +0100 Subject: BUG#38173 Field doesn't have a default value with row-based replication NOTE: Backporting the patch to next-mr. The reason of the bug was incompatibile with the master side behaviour. INSERT query on the master is allowed to insert into a table without specifying values of DEFAULT-less fields if sql_mode is not strict. Fixed with checking sql_mode by the sql thread to decide how to react. Non-strict sql_mode should allow Write_rows event to complete. todo: warnings can be shown via show slave status, still this is a separate rather general issue how to show warnings for the slave threads. --- sql/log_event.cc | 5 ++++- sql/rpl_record.cc | 34 ++++++++++++++++++++++++---------- sql/rpl_record.h | 3 ++- 3 files changed, 30 insertions(+), 12 deletions(-) (limited to 'sql') diff --git a/sql/log_event.cc b/sql/log_event.cc index fb6a5230fda..d595f00bffd 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -8446,7 +8446,10 @@ Rows_log_event::write_row(const Relay_log_info *const rli, /* fill table->record[0] with default values */ if ((error= prepare_record(table, m_width, - TRUE /* check if columns have def. values */))) + table->file->ht->db_type != DB_TYPE_NDBCLUSTER, + (rli->sql_thd->variables.sql_mode & + (MODE_STRICT_TRANS_TABLES | + MODE_STRICT_ALL_TABLES))))) DBUG_RETURN(error); /* unpack row into table->record[0] */ diff --git a/sql/rpl_record.cc b/sql/rpl_record.cc index 14a80cbb4b6..f4768e2456a 100644 --- a/sql/rpl_record.cc +++ b/sql/rpl_record.cc @@ -305,13 +305,17 @@ unpack_row(Relay_log_info const *rli, @param table Table whose record[0] buffer is prepared. @param skip Number of columns for which default/nullable check should be skipped. - @param check Indicates if errors should be raised when checking - default/nullable field properties. + @param check Specifies if lack of default error needs checking. + @param abort_on_warning + Controls how to react on lack of a field's default. + The parameter mimics the master side one for + @c check_that_all_fields_are_given_values. @returns 0 on success or a handler level error code */ int prepare_record(TABLE *const table, - const uint skip, const bool check) + const uint skip, const bool check, + const bool abort_on_warning) { DBUG_ENTER("prepare_record"); @@ -326,17 +330,27 @@ int prepare_record(TABLE *const table, if (skip >= table->s->fields || !check) DBUG_RETURN(0); - /* Checking if exists default/nullable fields in the default values. */ - - for (Field **field_ptr= table->field+skip ; *field_ptr ; ++field_ptr) + /* + For fields the extra fields on the slave, we check if they have a default. + The check follows the same rules as the INSERT query without specifying an + explicit value for a field not having the explicit default + (@c check_that_all_fields_are_given_values()). + */ + for (Field **field_ptr= table->field+skip; *field_ptr; ++field_ptr) { uint32 const mask= NOT_NULL_FLAG | NO_DEFAULT_VALUE_FLAG; Field *const f= *field_ptr; - - if (((f->flags & mask) == mask)) + if ((f->flags & NO_DEFAULT_VALUE_FLAG) && + (f->real_type() != MYSQL_TYPE_ENUM)) { - my_error(ER_NO_DEFAULT_FOR_FIELD, MYF(0), f->field_name); - error = HA_ERR_ROWS_EVENT_APPLY; + push_warning_printf(current_thd, abort_on_warning? + MYSQL_ERROR::WARN_LEVEL_ERROR : + MYSQL_ERROR::WARN_LEVEL_WARN, + ER_NO_DEFAULT_FOR_FIELD, + ER(ER_NO_DEFAULT_FOR_FIELD), + f->field_name); + if (abort_on_warning) + error = HA_ERR_ROWS_EVENT_APPLY; } } diff --git a/sql/rpl_record.h b/sql/rpl_record.h index f9e64f0ab1d..ab2bcd382ca 100644 --- a/sql/rpl_record.h +++ b/sql/rpl_record.h @@ -30,7 +30,8 @@ int unpack_row(Relay_log_info const *rli, uchar const **const row_end, ulong *const master_reclength); // Fill table's record[0] with default values. -int prepare_record(TABLE *const, const uint =0, const bool =FALSE); +int prepare_record(TABLE *const table, const uint skip, const bool check, + const bool abort_on_warning= FALSE); #endif #endif -- cgit v1.2.1 From 01cdba58ba758e468d01770e87bca7f22790a184 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Tue, 29 Sep 2009 15:09:01 +0100 Subject: BUG#23300: Slow query log on slave does not log slow replicated statements NOTE: this is the backport to next-mr. When using replication, the slave will not log any slow query logs queries replicated from the master, even if the option "--log-slow-slave-statements" is set and these take more than "log_query_time" to execute. In order to log slow queries in replicated thread one needs to set the --log-slow-slave-statements, so that the SQL thread is initialized with the correct switch. Although setting this flag correctly configures the slave thread option to log slow queries, there is an issue with the condition that is used to check whether to log the slow query or not. When replaying binlog events the statement contains the SET TIMESTAMP clause which will force the slow logging condition check to fail. Consequently, the slow query logging will not take place. This patch addresses this issue by removing the second condition from the log_slow_statements as it prevents slow queries to be binlogged and seems to be deprecated. --- sql/log.cc | 1 + sql/sql_parse.cc | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) (limited to 'sql') diff --git a/sql/log.cc b/sql/log.cc index 1af2f3a4ddc..5c25d30bad0 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -961,6 +961,7 @@ bool LOGGER::slow_log_print(THD *thd, const char *query, uint query_length, uint user_host_len= 0; ulonglong query_utime, lock_utime; + DBUG_ASSERT(thd->enable_slow_log); /* Print the message to the buffer if we have slow log enabled */ diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc index 513b9230c37..d0897646cec 100644 --- a/sql/sql_parse.cc +++ b/sql/sql_parse.cc @@ -1615,9 +1615,9 @@ void log_slow_statement(THD *thd) /* Do not log administrative statements unless the appropriate option is - set; do not log into slow log if reading from backup. + set. */ - if (thd->enable_slow_log && !thd->user_time) + if (thd->enable_slow_log) { ulonglong end_utime_of_query= thd->current_utime(); thd_proc_info(thd, "logging slow query"); -- cgit v1.2.1 From b43d30e43a935ec53faea22fb7614550218ad42f Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Tue, 29 Sep 2009 15:09:46 +0100 Subject: BUG#28796: CHANGE MASTER TO MASTER_HOST="" leads to invalid master.info NOTE: this is the backport to next-mr. This patch addresses the bug reported by checking wether host argument is an empty string or not. If empty, an error is reported to the client, otherwise continue normally. This commit is based on the originally proposed patch and adds a test case as requested during review as well as refines comments, and makes test case result file less verbose (compared to previous patch). --- sql/sql_repl.cc | 13 +++++++++++++ 1 file changed, 13 insertions(+) (limited to 'sql') diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 4d9b7410b88..1cc21162022 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -1150,6 +1150,19 @@ bool change_master(THD* thd, Master_info* mi) thd_proc_info(thd, "Changing master"); LEX_MASTER_INFO* lex_mi= &thd->lex->mi; + /* + 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"); + unlock_slave_threads(mi); + DBUG_RETURN(TRUE); + } // TODO: see if needs re-write if (init_master_info(mi, master_info_file, relay_log_info_file, 0, thread_mask)) -- cgit v1.2.1 From b538536718389bcac283d875a7d92527d8ef57be Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Tue, 29 Sep 2009 15:10:37 +0100 Subject: Bug #30703 SHOW STATUS LIKE 'Slave_running' is not compatible with `SHOW SLAVE STATUS' NOTE: this is the backport to next-mr. SHOW SHOW STATUS LIKE 'Slave_running' command believes that if active_mi->slave_running != 0, then io thread is running normally. But it isn't so in fact. When some errors happen to make io thread try to reconnect master, then it will become transitional status (MYSQL_SLAVE_RUN_NOT_CONNECT == 1), which also doesn't equal 0. Yet, "SHOW SLAVE STATUS" believes that only if active_mi->slave_running == MYSQL_SLAVE_RUN_CONNECT, then io thread is running. So "SHOW SLAVE STATUS" can get the correct result. Fixed to make SHOW SHOW STATUS LIKE 'Slave_running' command have the same check condition with "SHOW SLAVE STATUS". It only believe that the io thread is running when active_mi->slave_running == MYSQL_SLAVE_RUN_CONNECT. --- sql/mysqld.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'sql') diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 9b70096eb73..9ae3db4762b 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -7043,7 +7043,8 @@ static int show_slave_running(THD *thd, SHOW_VAR *var, char *buff) var->type= SHOW_MY_BOOL; pthread_mutex_lock(&LOCK_active_mi); var->value= buff; - *((my_bool *)buff)= (my_bool) (active_mi && active_mi->slave_running && + *((my_bool *)buff)= (my_bool) (active_mi && + active_mi->slave_running == MYSQL_SLAVE_RUN_CONNECT && active_mi->rli.slave_running); pthread_mutex_unlock(&LOCK_active_mi); return 0; -- cgit v1.2.1 From b79b335033fae28f891217a28d898c018b380f03 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Tue, 29 Sep 2009 15:12:07 +0100 Subject: BUG#40611: MySQL cannot make a binary log after sequential number beyond unsigned long. BUG#44779: binlog.binlog_max_extension may be causing failure on next test in PB NOTE1: this is the backport to next-mr. NOTE2: already includes patch for BUG#44779. Binlog file extensions would turn into negative numbers once the variable used to hold the value reached maximum for signed long. Consequently, incrementing value to the next (negative) number would lead to .000000 extension, causing the server to fail. This patch addresses this issue by not allowing negative extensions and by returning an error on find_uniq_filename, when the limit is reached. Additionally, warnings are printed to the error log when the limit is approaching. FLUSH LOGS will also report warnings to the user, if the extension number has reached the limit. The limit has been set to 0x7FFFFFFF as the maximum. mysql-test/suite/binlog/t/binlog_max_extension.test: Test case added that checks the maximum available number for binlog extensions. sql/log.cc: Changes to find_uniq_filename and test_if_number. sql/log.h: Added macros with values for MAX_LOG_UNIQUE_FN_EXT and LOG_WARN_UNIQUE_FN_EXT_LEFT, as suggested in review. --- sql/log.cc | 75 ++++++++++++++++++++++++++++++++++++++++++++++++++------------ sql/log.h | 13 +++++++++++ 2 files changed, 74 insertions(+), 14 deletions(-) (limited to 'sql') diff --git a/sql/log.cc b/sql/log.cc index 5c25d30bad0..19b26fcf5ab 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -53,7 +53,7 @@ MYSQL_BIN_LOG mysql_bin_log; ulong sync_binlog_counter= 0; static bool test_if_number(const char *str, - long *res, bool allow_wildcards); + ulong *res, bool allow_wildcards); static int binlog_init(void *p); static int binlog_close_connection(handlerton *hton, THD *thd); static int binlog_savepoint_set(handlerton *hton, THD *thd, void *sv); @@ -1835,22 +1835,27 @@ static void setup_windows_event_source() /** Find a unique filename for 'filename.#'. - Set '#' to a number as low as possible. + Set '#' to the number next to the maximum found in the most + recent log file extension. + + This function will return nonzero if: (i) the generated name + exceeds FN_REFLEN; (ii) if the number of extensions is exhausted; + or (iii) some other error happened while examining the filesystem. @return - nonzero if not possible to get unique filename + nonzero if not possible to get unique filename. */ static int find_uniq_filename(char *name) { - long number; uint i; - char buff[FN_REFLEN]; + char buff[FN_REFLEN], ext_buf[FN_REFLEN]; struct st_my_dir *dir_info; reg1 struct fileinfo *file_info; - ulong max_found=0; + ulong max_found= 0, next= 0, number= 0; size_t buf_length, length; char *start, *end; + int error= 0; DBUG_ENTER("find_uniq_filename"); length= dirname_part(buff, name, &buf_length); @@ -1858,15 +1863,15 @@ static int find_uniq_filename(char *name) end= strend(start); *end='.'; - length= (size_t) (end-start+1); + length= (size_t) (end - start + 1); - if (!(dir_info = my_dir(buff,MYF(MY_DONT_SORT)))) + if (!(dir_info= my_dir(buff,MYF(MY_DONT_SORT)))) { // This shouldn't happen strmov(end,".1"); // use name+1 - DBUG_RETURN(0); + DBUG_RETURN(1); } file_info= dir_info->dir_entry; - for (i=dir_info->number_off_files ; i-- ; file_info++) + for (i= dir_info->number_off_files ; i-- ; file_info++) { if (bcmp((uchar*) file_info->name, (uchar*) start, length) == 0 && test_if_number(file_info->name+length, &number,0)) @@ -1876,9 +1881,44 @@ static int find_uniq_filename(char *name) } my_dirend(dir_info); + /* check if reached the maximum possible extension number */ + if ((max_found == MAX_LOG_UNIQUE_FN_EXT)) + { + sql_print_error("Log filename extension number exhausted: %06lu. \ +Please fix this by archiving old logs and \ +updating the index files.", max_found); + error= 1; + goto end; + } + + next= max_found + 1; + sprintf(ext_buf, "%06lu", next); *end++='.'; - sprintf(end,"%06ld",max_found+1); - DBUG_RETURN(0); + + /* + Check if the generated extension size + the file name exceeds the + buffer size used. If one did not check this, then the filename might be + truncated, resulting in error. + */ + if (((strlen(ext_buf) + (end - name)) >= FN_REFLEN)) + { + sql_print_error("Log filename too large: %s%s (%d). \ +Please fix this by archiving old logs and updating the \ +index files.", name, ext_buf, (strlen(ext_buf) + (end - name))); + error= 1; + goto end; + } + + sprintf(end, "%06lu", next); + + /* print warning if reaching the end of available extensions. */ + if ((next > (MAX_LOG_UNIQUE_FN_EXT - LOG_WARN_UNIQUE_FN_EXT_LEFT))) + sql_print_warning("Next log extension: %lu. \ +Remaining log filename extensions: %lu. \ +Please consider archiving some logs.", next, (MAX_LOG_UNIQUE_FN_EXT - next)); + +end: + DBUG_RETURN(error); } @@ -2077,6 +2117,13 @@ int MYSQL_LOG::generate_new_name(char *new_name, const char *log_name) { if (find_uniq_filename(new_name)) { + /* + This should be treated as error once propagation of error further + up in the stack gets proper handling. + */ + push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN, + ER_NO_UNIQUE_LOGFILE, ER(ER_NO_UNIQUE_LOGFILE), + log_name); sql_print_error(ER(ER_NO_UNIQUE_LOGFILE), log_name); return 1; } @@ -4739,11 +4786,11 @@ void MYSQL_BIN_LOG::set_max_size(ulong max_size_arg) @retval 1 String is a number @retval - 0 Error + 0 String is not a number */ static bool test_if_number(register const char *str, - long *res, bool allow_wildcards) + ulong *res, bool allow_wildcards) { reg2 int flag; const char *start; diff --git a/sql/log.h b/sql/log.h index d306d6f7182..3c2ccc521cc 100644 --- a/sql/log.h +++ b/sql/log.h @@ -121,6 +121,19 @@ extern TC_LOG_DUMMY tc_log_dummy; #define LOG_CLOSE_TO_BE_OPENED 2 #define LOG_CLOSE_STOP_EVENT 4 +/* + Maximum unique log filename extension. + Note: setting to 0x7FFFFFFF due to atol windows + overflow/truncate. + */ +#define MAX_LOG_UNIQUE_FN_EXT 0x7FFFFFFF + +/* + Number of warnings that will be printed to error log + before extension number is exhausted. +*/ +#define LOG_WARN_UNIQUE_FN_EXT_LEFT 1000 + class Relay_log_info; typedef struct st_log_info -- cgit v1.2.1 From ba98c882728a6ea3f03566f3548c7c40a21567b3 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Tue, 29 Sep 2009 15:12:43 +0100 Subject: BUG#42928: binlog-format setting prevents server from start if binary logging is disabled NOTE: this is the backport to next-mr. If one sets binlog-format but does NOT enable binary log, server refuses to start up. The following messages appears in the error log: 090217 12:47:14 [ERROR] You need to use --log-bin to make --binlog-format work. 090217 12:47:14 [ERROR] Aborting This patch addresses this by making the server not to bail out if the binlog-format is set without the log-bin option. Additionally, the specified binlog-format is stored, in the global system variable "binlog_format", and a warning is printed instead of an error. --- sql/mysqld.cc | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) (limited to 'sql') diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 9ae3db4762b..513302b6aec 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -3827,9 +3827,10 @@ with --log-bin instead."); { if (opt_binlog_format_id != BINLOG_FORMAT_UNSPEC) { - sql_print_error("You need to use --log-bin to make " - "--binlog-format work."); - unireg_abort(1); + sql_print_warning("You need to use --log-bin to make " + "--binlog-format work."); + + global_system_variables.binlog_format= opt_binlog_format_id; } else { -- cgit v1.2.1 From 25162d0166206ee4819ece2161d26caa01153723 Mon Sep 17 00:00:00 2001 From: Alfranio Correia Date: Tue, 29 Sep 2009 15:18:44 +0100 Subject: BUG#43789 different master/slave table defs cause crash: text/varchar null vs not null NOTE: Backporting the patch to next-mr. The replication was generating corrupted data, warning messages on Valgrind and aborting on debug mode while replicating a "null" to "not null" field. Specifically the unpack_row routine, was considering the slave's table definition and trying to retrieve a field value, where there was nothing to be retrieved, ignoring the fact that the value was defined as "null" by the master. To fix the problem, we proceed as follows: 1 - If it is not STRICT sql_mode, implicit default values are used, regardless if it is multi-row or single-row statement. 2 - However, if it is STRICT mode, then a we do what follows: 2.1 If it is a transactional engine, we do a rollback on the first NULL that is to be set into a NOT NULL column and return an error. 2.2 If it is a non-transactional engine and it is the first row to be inserted with multi-row, we also return the error. Otherwise, we proceed with the execution, use implicit default values and print out warning messages. Unfortunately, the current patch cannot mimic the behavior showed by the master for updates on multi-tables and multi-row inserts. This happens because such statements are unfolded in different row events. For instance, considering the following updates and strict mode: (master) create table t1 (a int); create table t2 (a int not null); insert into t1 values (1); insert into t2 values (2); update t1, t2 SET t1.a=10, t2.a=NULL; t1 would have (10) and t2 would have (0) as this would be handled as a multi-row update. On the other hand, if we had the following updates: (master) create table t1 (a int); create table t2 (a int); (slave) create table t1 (a int); create table t2 (a int not null); (master) insert into t1 values (1); insert into t2 values (2); update t1, t2 SET t1.a=10, t2.a=NULL; On the master t1 would have (10) and t2 would have (NULL). On the slave, t1 would have (10) but the update on t1 would fail. --- sql/log_event.cc | 17 +++++++++++------ sql/log_event.h | 8 ++++++-- sql/rpl_record.cc | 56 ++++++++++++++++++++++++++++++++++++++++++++++--------- sql/rpl_record.h | 6 ++++-- 4 files changed, 68 insertions(+), 19 deletions(-) (limited to 'sql') diff --git a/sql/log_event.cc b/sql/log_event.cc index d595f00bffd..eeed50f697d 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -8444,16 +8444,17 @@ Rows_log_event::write_row(const Relay_log_info *const rli, auto_afree_ptr key(NULL); /* fill table->record[0] with default values */ - + bool abort_on_warnings= (rli->sql_thd->variables.sql_mode & + (MODE_STRICT_TRANS_TABLES | MODE_STRICT_ALL_TABLES)); if ((error= prepare_record(table, m_width, table->file->ht->db_type != DB_TYPE_NDBCLUSTER, - (rli->sql_thd->variables.sql_mode & - (MODE_STRICT_TRANS_TABLES | - MODE_STRICT_ALL_TABLES))))) + abort_on_warnings, m_curr_row == m_rows_buf))) DBUG_RETURN(error); /* unpack row into table->record[0] */ - error= unpack_current_row(rli); // TODO: how to handle errors? + if ((error= unpack_current_row(rli, abort_on_warnings))) + DBUG_RETURN(error); + if (m_curr_row == m_rows_buf) { /* this is the first row to be inserted, we estimate the rows with @@ -9250,8 +9251,12 @@ Update_rows_log_event::do_exec_row(const Relay_log_info *const rli) store_record(m_table,record[1]); + bool abort_on_warnings= (rli->sql_thd->variables.sql_mode & + (MODE_STRICT_TRANS_TABLES | MODE_STRICT_ALL_TABLES)); m_curr_row= m_curr_row_end; - error= unpack_current_row(rli); // this also updates m_curr_row_end + /* this also updates m_curr_row_end */ + if ((error= unpack_current_row(rli, abort_on_warnings))) + return error; /* Now we have the right row to update. The old row (the one we're diff --git a/sql/log_event.h b/sql/log_event.h index 8202dddcc76..bbcca76741c 100644 --- a/sql/log_event.h +++ b/sql/log_event.h @@ -3541,12 +3541,16 @@ protected: int write_row(const Relay_log_info *const, const bool); // Unpack the current row into m_table->record[0] - int unpack_current_row(const Relay_log_info *const rli) + int unpack_current_row(const Relay_log_info *const rli, + const bool abort_on_warning= TRUE) { DBUG_ASSERT(m_table); + + bool first_row= (m_curr_row == m_rows_buf); ASSERT_OR_RETURN_ERROR(m_curr_row < m_rows_end, HA_ERR_CORRUPT_EVENT); int const result= ::unpack_row(rli, m_table, m_width, m_curr_row, &m_cols, - &m_curr_row_end, &m_master_reclength); + &m_curr_row_end, &m_master_reclength, + abort_on_warning, first_row); if (m_curr_row_end > m_rows_end) my_error(ER_SLAVE_CORRUPT_EVENT, MYF(0)); ASSERT_OR_RETURN_ERROR(m_curr_row_end <= m_rows_end, HA_ERR_CORRUPT_EVENT); diff --git a/sql/rpl_record.cc b/sql/rpl_record.cc index f4768e2456a..8e80620dd2c 100644 --- a/sql/rpl_record.cc +++ b/sql/rpl_record.cc @@ -180,7 +180,8 @@ int unpack_row(Relay_log_info const *rli, TABLE *table, uint const colcnt, uchar const *const row_data, MY_BITMAP const *cols, - uchar const **const row_end, ulong *const master_reclength) + uchar const **const row_end, ulong *const master_reclength, + const bool abort_on_warning, const bool first_row) { DBUG_ENTER("unpack_row"); DBUG_ASSERT(row_data); @@ -224,8 +225,35 @@ unpack_row(Relay_log_info const *rli, /* Field...::unpack() cannot return 0 */ DBUG_ASSERT(pack_ptr != NULL); - if ((null_bits & null_mask) && f->maybe_null()) - f->set_null(); + if (null_bits & null_mask) + { + if (f->maybe_null()) + { + DBUG_PRINT("debug", ("Was NULL; null mask: 0x%x; null bits: 0x%x", + null_mask, null_bits)); + f->set_null(); + } + else + { + MYSQL_ERROR::enum_warning_level error_type= + MYSQL_ERROR::WARN_LEVEL_NOTE; + if (abort_on_warning && (table->file->has_transactions() || + first_row)) + { + error = HA_ERR_ROWS_EVENT_APPLY; + error_type= MYSQL_ERROR::WARN_LEVEL_ERROR; + } + else + { + f->set_default(); + error_type= MYSQL_ERROR::WARN_LEVEL_WARN; + } + push_warning_printf(current_thd, error_type, + ER_BAD_NULL_ERROR, + ER(ER_BAD_NULL_ERROR), + f->field_name); + } + } else { f->set_notnull(); @@ -315,7 +343,7 @@ unpack_row(Relay_log_info const *rli, */ int prepare_record(TABLE *const table, const uint skip, const bool check, - const bool abort_on_warning) + const bool abort_on_warning, const bool first_row) { DBUG_ENTER("prepare_record"); @@ -343,14 +371,24 @@ int prepare_record(TABLE *const table, if ((f->flags & NO_DEFAULT_VALUE_FLAG) && (f->real_type() != MYSQL_TYPE_ENUM)) { - push_warning_printf(current_thd, abort_on_warning? - MYSQL_ERROR::WARN_LEVEL_ERROR : - MYSQL_ERROR::WARN_LEVEL_WARN, + + MYSQL_ERROR::enum_warning_level error_type= + MYSQL_ERROR::WARN_LEVEL_NOTE; + if (abort_on_warning && (table->file->has_transactions() || + first_row)) + { + error= HA_ERR_ROWS_EVENT_APPLY; + error_type= MYSQL_ERROR::WARN_LEVEL_ERROR; + } + else + { + f->set_default(); + error_type= MYSQL_ERROR::WARN_LEVEL_WARN; + } + push_warning_printf(current_thd, error_type, ER_NO_DEFAULT_FOR_FIELD, ER(ER_NO_DEFAULT_FOR_FIELD), f->field_name); - if (abort_on_warning) - error = HA_ERR_ROWS_EVENT_APPLY; } } diff --git a/sql/rpl_record.h b/sql/rpl_record.h index ab2bcd382ca..6e8838f82b3 100644 --- a/sql/rpl_record.h +++ b/sql/rpl_record.h @@ -27,11 +27,13 @@ size_t pack_row(TABLE* table, MY_BITMAP const* cols, int unpack_row(Relay_log_info const *rli, TABLE *table, uint const colcnt, uchar const *const row_data, MY_BITMAP const *cols, - uchar const **const row_end, ulong *const master_reclength); + uchar const **const row_end, ulong *const master_reclength, + const bool abort_on_warning= TRUE, const bool first_row= TRUE); // Fill table's record[0] with default values. int prepare_record(TABLE *const table, const uint skip, const bool check, - const bool abort_on_warning= FALSE); + const bool abort_on_warning= TRUE, + const bool first_row= TRUE); #endif #endif -- cgit v1.2.1 From 0110bd04d24503d84df93d31b444586c4137c98c Mon Sep 17 00:00:00 2001 From: Alfranio Correia Date: Tue, 29 Sep 2009 15:27:12 +0100 Subject: BUG#35542 Add option to sync master and relay log to disk after every event BUG#31665 sync_binlog should cause relay logs to be synchronized NOTE: Backporting the patch to next-mr. Add sync_relay_log option to server, this option works for relay log the same as option sync_binlog for binlog. This option also synchronize master info to disk when set to non-zero value. Original patches from Sinisa and Mark, with some modifications --- sql/log.cc | 35 +++++++++++++++++++++++------------ sql/log.h | 29 +++++++++++++++++++++++++++-- sql/mysql_priv.h | 3 ++- sql/mysqld.cc | 15 +++++++++++---- sql/rpl_mi.cc | 20 ++++++++++++++++---- sql/rpl_rli.cc | 3 ++- sql/set_var.cc | 17 +++++++++++++++++ sql/set_var.h | 21 +++++++++++++++++++++ sql/sql_repl.cc | 18 ++---------------- 9 files changed, 121 insertions(+), 40 deletions(-) (limited to 'sql') diff --git a/sql/log.cc b/sql/log.cc index 1af2f3a4ddc..12e57623add 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -49,8 +49,7 @@ LOGGER logger; -MYSQL_BIN_LOG mysql_bin_log; -ulong sync_binlog_counter= 0; +MYSQL_BIN_LOG mysql_bin_log(&sync_binlog_period); static bool test_if_number(const char *str, long *res, bool allow_wildcards); @@ -2410,9 +2409,10 @@ const char *MYSQL_LOG::generate_name(const char *log_name, -MYSQL_BIN_LOG::MYSQL_BIN_LOG() +MYSQL_BIN_LOG::MYSQL_BIN_LOG(uint *sync_period) :bytes_written(0), prepared_xids(0), file_id(1), open_count(1), need_start_event(TRUE), m_table_map_version(0), + sync_period_ptr(sync_period), is_relay_log(0), description_event_for_exec(0), description_event_for_queue(0) { @@ -3643,6 +3643,8 @@ bool MYSQL_BIN_LOG::append(Log_event* ev) } bytes_written+= ev->data_written; DBUG_PRINT("info",("max_size: %lu",max_size)); + if (flush_and_sync(0)) + goto err; if ((uint) my_b_append_tell(&log_file) > max_size) new_file_without_locking(); @@ -3673,6 +3675,8 @@ bool MYSQL_BIN_LOG::appendv(const char* buf, uint len,...) bytes_written += len; } while ((buf=va_arg(args,const char*)) && (len=va_arg(args,uint))); DBUG_PRINT("info",("max_size: %lu",max_size)); + if (flush_and_sync(0)) + goto err; if ((uint) my_b_append_tell(&log_file) > max_size) new_file_without_locking(); @@ -3682,17 +3686,21 @@ err: DBUG_RETURN(error); } - -bool MYSQL_BIN_LOG::flush_and_sync() +bool MYSQL_BIN_LOG::flush_and_sync(bool *synced) { int err=0, fd=log_file.file; + if (synced) + *synced= 0; safe_mutex_assert_owner(&LOCK_log); if (flush_io_cache(&log_file)) return 1; - if (++sync_binlog_counter >= sync_binlog_period && sync_binlog_period) + uint sync_period= get_sync_period(); + if (sync_period && ++sync_counter >= sync_period) { - sync_binlog_counter= 0; + sync_counter= 0; err=my_sync(fd, MYF(MY_WME)); + if (synced) + *synced= 1; } return err; } @@ -3983,7 +3991,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd, if (file == &log_file) { - error= flush_and_sync(); + error= flush_and_sync(0); if (!error) { signal_update(); @@ -4169,7 +4177,8 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info) if (file == &log_file) // we are writing to the real log (disk) { - if (flush_and_sync()) + bool synced; + if (flush_and_sync(&synced)) goto err; signal_update(); rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED); @@ -4425,7 +4434,7 @@ int MYSQL_BIN_LOG::write_cache(IO_CACHE *cache, bool lock_log, bool sync_log) DBUG_ASSERT(carry == 0); if (sync_log) - flush_and_sync(); + return flush_and_sync(0); return 0; // All OK } @@ -4472,7 +4481,8 @@ bool MYSQL_BIN_LOG::write_incident(THD *thd, bool lock) ev.write(&log_file); if (lock) { - if (!error && !(error= flush_and_sync())) + bool synced; + if (!error && !(error= flush_and_sync(&synced))) { signal_update(); rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED); @@ -4560,7 +4570,8 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event, if (incident && write_incident(thd, FALSE)) goto err; - if (flush_and_sync()) + bool synced; + if (flush_and_sync(&synced)) goto err; DBUG_EXECUTE_IF("half_binlogged_transaction", abort();); if (cache->error) // Error on read diff --git a/sql/log.h b/sql/log.h index d306d6f7182..53943b649ee 100644 --- a/sql/log.h +++ b/sql/log.h @@ -269,6 +269,18 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG ulonglong m_table_map_version; + /* pointer to the sync period variable, for binlog this will be + sync_binlog_period, for relay log this will be + sync_relay_log_period + */ + uint *sync_period_ptr; + uint sync_counter; + + inline uint get_sync_period() + { + return *sync_period_ptr; + } + int write_to_file(IO_CACHE *cache); /* This is used to start writing to a new log file. The difference from @@ -296,7 +308,7 @@ public: Format_description_log_event *description_event_for_exec, *description_event_for_queue; - MYSQL_BIN_LOG(); + MYSQL_BIN_LOG(uint *sync_period); /* note that there's no destructor ~MYSQL_BIN_LOG() ! The reason is that we don't want it to be automatically called @@ -378,7 +390,20 @@ public: bool is_active(const char* log_file_name); int update_log_index(LOG_INFO* linfo, bool need_update_threads); void rotate_and_purge(uint flags); - bool flush_and_sync(); + /** + Flush binlog cache and synchronize to disk. + + This function flushes events in binlog cache to binary log file, + it will do synchronizing according to the setting of system + variable 'sync_binlog'. If file is synchronized, @c synced will + be set to 1, otherwise 0. + + @param[out] synced if not NULL, set to 1 if file is synchronized, otherwise 0 + + @retval 0 Success + @retval other Failure + */ + bool flush_and_sync(bool *synced); int purge_logs(const char *to_log, bool included, bool need_mutex, bool need_update_threads, ulonglong *decrease_log_space); diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h index 52cfa7934c3..669942cc691 100644 --- a/sql/mysql_priv.h +++ b/sql/mysql_priv.h @@ -1868,7 +1868,8 @@ extern ulong MYSQL_PLUGIN_IMPORT specialflag; #endif /* MYSQL_SERVER || INNODB_COMPATIBILITY_HOOKS */ #ifdef MYSQL_SERVER extern ulong current_pid; -extern ulong expire_logs_days, sync_binlog_period, sync_binlog_counter; +extern ulong expire_logs_days; +extern uint sync_binlog_period, sync_relaylog_period; extern ulong opt_tc_log_size, tc_log_max_pages_used, tc_log_page_size; extern ulong tc_log_page_waits; extern my_bool relay_log_purge, opt_innodb_safe_binlog, opt_innodb; diff --git a/sql/mysqld.cc b/sql/mysqld.cc index a57ee04081f..8febc0bb7e5 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -552,7 +552,8 @@ ulong max_prepared_stmt_count; */ ulong prepared_stmt_count=0; ulong thread_id=1L,current_pid; -ulong slow_launch_threads = 0, sync_binlog_period; +ulong slow_launch_threads = 0; +uint sync_binlog_period= 0, sync_relaylog_period= 0; ulong expire_logs_days = 0; ulong rpl_recovery_rank=0; const char *log_output_str= "FILE"; @@ -5667,7 +5668,8 @@ enum options_mysqld OPT_SLAVE_EXEC_MODE, OPT_GENERAL_LOG_FILE, OPT_SLOW_QUERY_LOG_FILE, - OPT_IGNORE_BUILTIN_INNODB + OPT_IGNORE_BUILTIN_INNODB, + OPT_SYNC_RELAY_LOG }; @@ -6921,8 +6923,13 @@ The minimum value for this variable is 4096.", {"sync-binlog", OPT_SYNC_BINLOG, "Synchronously flush binary log to disk after every #th event. " "Use 0 (default) to disable synchronous flushing.", - (uchar**) &sync_binlog_period, (uchar**) &sync_binlog_period, 0, GET_ULONG, - REQUIRED_ARG, 0, 0, ULONG_MAX, 0, 1, 0}, + (uchar**) &sync_binlog_period, (uchar**) &sync_binlog_period, 0, GET_UINT, + REQUIRED_ARG, 0, 0, (longlong) UINT_MAX, 0, 1, 0}, + {"sync-relay-log", OPT_SYNC_RELAY_LOG, + "Synchronously flush relay log to disk after every #th event. " + "Use 0 (default) to disable synchronous flushing.", + (uchar**) &sync_relaylog_period, (uchar**) &sync_relaylog_period, 0, GET_UINT, + REQUIRED_ARG, 0, 0, (longlong) UINT_MAX, 0, 1, 0}, {"sync-frm", OPT_SYNC_FRM, "Sync .frm to disk on create. Enabled by default.", (uchar**) &opt_sync_frm, (uchar**) &opt_sync_frm, 0, GET_BOOL, NO_ARG, 1, 0, 0, 0, 0, 0}, diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc index 5e46837e948..1bca44ac613 100644 --- a/sql/rpl_mi.cc +++ b/sql/rpl_mi.cc @@ -342,6 +342,7 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) { IO_CACHE* file = &mi->file; char lbuf[22]; + int err= 0; DBUG_ENTER("flush_master_info"); DBUG_PRINT("enter",("master_pos: %ld", (long) mi->master_log_pos)); @@ -358,9 +359,17 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) When we come to this place in code, relay log may or not be initialized; the caller is responsible for setting 'flush_relay_log_cache' accordingly. */ - if (flush_relay_log_cache && - flush_io_cache(mi->rli.relay_log.get_log_file())) - DBUG_RETURN(2); + if (flush_relay_log_cache) + { + IO_CACHE *log_file= mi->rli.relay_log.get_log_file(); + if (flush_io_cache(log_file)) + DBUG_RETURN(2); + + /* Sync to disk if --sync-relay-log is set */ + if (sync_relaylog_period && + my_sync(log_file->file, MY_WME)) + DBUG_RETURN(2); + } /* We flushed the relay log BEFORE the master.info file, because if we crash @@ -388,7 +397,10 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) mi->password, mi->port, mi->connect_retry, (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert, mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert); - DBUG_RETURN(-flush_io_cache(file)); + err= flush_io_cache(file); + if (sync_relaylog_period && !err) + err= my_sync(mi->fd, MYF(MY_WME)); + DBUG_RETURN(-err); } diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index 18fbae9bb9d..37c0815fb8b 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -32,7 +32,8 @@ int init_strvar_from_file(char *var, int max_size, IO_CACHE *f, Relay_log_info::Relay_log_info() :Slave_reporting_capability("SQL"), no_storage(FALSE), replicate_same_server_id(::replicate_same_server_id), - info_fd(-1), cur_log_fd(-1), save_temporary_tables(0), + info_fd(-1), cur_log_fd(-1), relay_log(&sync_relaylog_period), + save_temporary_tables(0), #if HAVE_purify is_fake(FALSE), #endif diff --git a/sql/set_var.cc b/sql/set_var.cc index 5025356230c..f2b5201cf8b 100644 --- a/sql/set_var.cc +++ b/sql/set_var.cc @@ -1534,6 +1534,23 @@ static bool get_unsigned(THD *thd, set_var *var, ulonglong user_max, } +bool sys_var_int_ptr::check(THD *thd, set_var *var) +{ + var->save_result.ulong_value= (ulong) var->value->val_int(); + return 0; +} + +bool sys_var_int_ptr::update(THD *thd, set_var *var) +{ + *value= (uint) var->save_result.ulong_value; + return 0; +} + +void sys_var_int_ptr::set_default(THD *thd, enum_var_type type) +{ + *value= (uint) option_limits->def_value; +} + sys_var_long_ptr:: sys_var_long_ptr(sys_var_chain *chain, const char *name_arg, ulong *value_ptr_arg, sys_after_update_func after_update_arg) diff --git a/sql/set_var.h b/sql/set_var.h index a3ed8e5be15..02c87abed88 100644 --- a/sql/set_var.h +++ b/sql/set_var.h @@ -175,6 +175,27 @@ public: { return (uchar*) value; } }; +/** + Unsigned int system variable class + */ +class sys_var_int_ptr :public sys_var +{ +public: + sys_var_int_ptr(sys_var_chain *chain, const char *name_arg, + uint *value_ptr_arg, + sys_after_update_func after_update_arg= NULL) + :sys_var(name_arg, after_update_arg), + value(value_ptr_arg) + { chain_sys_var(chain); } + bool check(THD *thd, set_var *var); + bool update(THD *thd, set_var *var); + void set_default(THD *thd, enum_var_type type); + SHOW_TYPE show_type() { return SHOW_INT; } + uchar *value_ptr(THD *thd, enum_var_type type, LEX_STRING *base) + { return (uchar*) value; } +private: + uint *value; +}; /* A global ulong variable that is protected by LOCK_global_system_variables diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 4d9b7410b88..425d76c8b72 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -1753,15 +1753,6 @@ public: */ }; -class sys_var_sync_binlog_period :public sys_var_long_ptr -{ -public: - sys_var_sync_binlog_period(sys_var_chain *chain, const char *name_arg, - ulong *value_ptr) - :sys_var_long_ptr(chain, name_arg,value_ptr) {} - bool update(THD *thd, set_var *var); -}; - static sys_var_chain vars = { NULL, NULL }; static sys_var_const sys_log_slave_updates(&vars, "log_slave_updates", @@ -1793,7 +1784,8 @@ static sys_var_const sys_slave_skip_errors(&vars, "slave_skip_errors", (uchar*) slave_skip_error_names); static sys_var_long_ptr sys_slave_trans_retries(&vars, "slave_transaction_retries", &slave_trans_retries); -static sys_var_sync_binlog_period sys_sync_binlog_period(&vars, "sync_binlog", &sync_binlog_period); +static sys_var_int_ptr sys_sync_binlog_period(&vars, "sync_binlog", &sync_binlog_period); +static sys_var_int_ptr sys_sync_relaylog_period(&vars, "sync_relay_log", &sync_relaylog_period); static sys_var_slave_skip_counter sys_slave_skip_counter(&vars, "sql_slave_skip_counter"); @@ -1835,12 +1827,6 @@ bool sys_var_slave_skip_counter::update(THD *thd, set_var *var) } -bool sys_var_sync_binlog_period::update(THD *thd, set_var *var) -{ - sync_binlog_period= (ulong) var->save_result.ulonglong_value; - return 0; -} - int init_replication_sys_vars() { if (mysql_add_sys_var_chain(vars.first, my_long_options)) -- cgit v1.2.1 From 3ab71376ceb2d5da81d3b6fb092630d0b0929d76 Mon Sep 17 00:00:00 2001 From: Alfranio Correia Date: Tue, 29 Sep 2009 15:40:52 +0100 Subject: BUG#40337 Fsyncing master and relay log to disk after every event is too slow NOTE: Backporting the patch to next-mr. The fix proposed in BUG#35542 and BUG#31665 introduces a performance issue when fsyncing the master.info, relay.info and relay-log.bin* after #th events. Although such solution has been proposed to reduce the probability of corrupted files due to a slave-crash, the performance penalty introduced by it has made the approach impractical for highly intensive workloads. In a nutshell, the option --syn-relay-log proposed in BUG#35542 and BUG#31665 simultaneously fsyncs master.info, relay-log.info and relay-log.bin* and this is the main source of performance issues. This patch introduces new options that give more control to the user on what should be fsynced and how often: 1) (--sync-master-info, integer) which syncs the master.info after #th event; 2) (--sync-relay-log, integer) which syncs the relay-log.bin* after #th events. 3) (--sync-relay-log-info, integer) which syncs the relay.info after #th transactions. To provide both performance and increased reliability, we recommend the following setup: 1) --sync-master-info = 0 eventually the operating system will fsync it; 2) --sync-relay-log = 0 eventually the operating system will fsync it; 3) --sync-relay-log-info = 1 fsyncs it after every transaction; Notice, that the previous setup does not reduce the probability of corrupted master.info and relay-log.bin*. To overcome the issue, this patch also introduces a recovery mechanism that right after restart throws away relay-log.bin* retrieved from a master and updates the master.info based on the relay.info: 4) (--relay-log-recovery, boolean) which enables a recovery mechanism that throws away relay-log.bin* after a crash. However, it can only recover the incorrect binlog file and position in master.info, if other informations (host, port password, etc) are corrupted or incorrect, then this recovery mechanism will fail to work. --- sql/mysql_priv.h | 4 ++- sql/mysqld.cc | 26 ++++++++++++-- sql/rpl_mi.cc | 17 +++++---- sql/rpl_mi.h | 9 ++++- sql/rpl_rli.cc | 8 +++-- sql/rpl_rli.h | 15 +++++++- sql/set_var.cc | 8 ++--- sql/set_var.h | 4 +-- sql/slave.cc | 102 ++++++++++++++++++++++++++++++++++++++++++++++++++---- sql/sql_binlog.cc | 2 +- sql/sql_repl.cc | 12 +++++-- 11 files changed, 175 insertions(+), 32 deletions(-) (limited to 'sql') diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h index 669942cc691..435513832d0 100644 --- a/sql/mysql_priv.h +++ b/sql/mysql_priv.h @@ -1869,10 +1869,12 @@ extern ulong MYSQL_PLUGIN_IMPORT specialflag; #ifdef MYSQL_SERVER extern ulong current_pid; extern ulong expire_logs_days; -extern uint sync_binlog_period, sync_relaylog_period; +extern uint sync_binlog_period, sync_relaylog_period, + sync_relayloginfo_period, sync_masterinfo_period; extern ulong opt_tc_log_size, tc_log_max_pages_used, tc_log_page_size; extern ulong tc_log_page_waits; extern my_bool relay_log_purge, opt_innodb_safe_binlog, opt_innodb; +extern my_bool relay_log_recovery; extern uint test_flags,select_errors,ha_open_options; extern uint protocol_version, mysqld_port, dropping_tables; extern uint delay_key_write_options; diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 8febc0bb7e5..b8d09fd4e5a 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -477,6 +477,7 @@ extern const char *opt_ndb_distribution; extern enum ndb_distribution opt_ndb_distribution_id; #endif my_bool opt_readonly, use_temp_pool, relay_log_purge; +my_bool relay_log_recovery; my_bool opt_sync_frm, opt_allow_suspicious_udfs; my_bool opt_secure_auth= 0; char* opt_secure_file_priv= 0; @@ -553,7 +554,8 @@ ulong max_prepared_stmt_count; ulong prepared_stmt_count=0; ulong thread_id=1L,current_pid; ulong slow_launch_threads = 0; -uint sync_binlog_period= 0, sync_relaylog_period= 0; +uint sync_binlog_period= 0, sync_relaylog_period= 0, + sync_relayloginfo_period= 0, sync_masterinfo_period= 0; ulong expire_logs_days = 0; ulong rpl_recovery_rank=0; const char *log_output_str= "FILE"; @@ -5605,6 +5607,7 @@ enum options_mysqld OPT_QUERY_CACHE_TYPE, OPT_QUERY_CACHE_WLOCK_INVALIDATE, OPT_RECORD_BUFFER, OPT_RECORD_RND_BUFFER, OPT_DIV_PRECINCREMENT, OPT_RELAY_LOG_SPACE_LIMIT, OPT_RELAY_LOG_PURGE, + OPT_RELAY_LOG_RECOVERY, OPT_SLAVE_NET_TIMEOUT, OPT_SLAVE_COMPRESSED_PROTOCOL, OPT_SLOW_LAUNCH_TIME, OPT_SLAVE_TRANS_RETRIES, OPT_READONLY, OPT_DEBUGGING, OPT_SORT_BUFFER, OPT_TABLE_OPEN_CACHE, OPT_TABLE_DEF_CACHE, @@ -5669,7 +5672,9 @@ enum options_mysqld OPT_GENERAL_LOG_FILE, OPT_SLOW_QUERY_LOG_FILE, OPT_IGNORE_BUILTIN_INNODB, - OPT_SYNC_RELAY_LOG + OPT_SYNC_RELAY_LOG, + OPT_SYNC_RELAY_LOG_INFO, + OPT_SYNC_MASTER_INFO }; @@ -6889,6 +6894,13 @@ The minimum value for this variable is 4096.", (uchar**) &relay_log_purge, (uchar**) &relay_log_purge, 0, GET_BOOL, NO_ARG, 1, 0, 1, 0, 1, 0}, + {"relay_log_recovery", OPT_RELAY_LOG_RECOVERY, + "Enables automatic relay log recovery right after the database startup, " + "which means that the IO Thread starts re-fetching from the master " + "right after the last transaction processed.", + (uchar**) &relay_log_recovery, + (uchar**) &relay_log_recovery, 0, GET_BOOL, NO_ARG, + 0, 0, 1, 0, 1, 0}, {"relay_log_space_limit", OPT_RELAY_LOG_SPACE_LIMIT, "Maximum space to use for all relay logs.", (uchar**) &relay_log_space_limit, @@ -6930,6 +6942,16 @@ The minimum value for this variable is 4096.", "Use 0 (default) to disable synchronous flushing.", (uchar**) &sync_relaylog_period, (uchar**) &sync_relaylog_period, 0, GET_UINT, REQUIRED_ARG, 0, 0, (longlong) UINT_MAX, 0, 1, 0}, + {"sync-relay-log-info", OPT_SYNC_RELAY_LOG_INFO, + "Synchronously flush relay log info to disk after #th transaction. " + "Use 0 (default) to disable synchronous flushing.", + (uchar**) &sync_relayloginfo_period, (uchar**) &sync_relayloginfo_period, 0, GET_UINT, + REQUIRED_ARG, 0, 0, (longlong) UINT_MAX, 0, 1, 0}, + {"sync-master-info", OPT_SYNC_MASTER_INFO, + "Synchronously flush master info to disk after every #th event. " + "Use 0 (default) to disable synchronous flushing.", + (uchar**) &sync_masterinfo_period, (uchar**) &sync_masterinfo_period, 0, GET_UINT, + REQUIRED_ARG, 0, 0, (longlong) UINT_MAX, 0, 1, 0}, {"sync-frm", OPT_SYNC_FRM, "Sync .frm to disk on create. Enabled by default.", (uchar**) &opt_sync_frm, (uchar**) &opt_sync_frm, 0, GET_BOOL, NO_ARG, 1, 0, 0, 0, 0, 0}, diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc index 1bca44ac613..cec2eabdd20 100644 --- a/sql/rpl_mi.cc +++ b/sql/rpl_mi.cc @@ -27,11 +27,11 @@ 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, const char *default_val); -Master_info::Master_info() +Master_info::Master_info(bool is_slave_recovery) :Slave_reporting_capability("I/O"), ssl(0), ssl_verify_server_cert(0), fd(-1), io_thd(0), inited(0), - abort_slave(0),slave_running(0), - slave_run_id(0) + rli(is_slave_recovery), abort_slave(0), slave_running(0), + slave_run_id(0), sync_counter(0) { host[0] = 0; user[0] = 0; password[0] = 0; ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0; @@ -364,11 +364,6 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) IO_CACHE *log_file= mi->rli.relay_log.get_log_file(); if (flush_io_cache(log_file)) DBUG_RETURN(2); - - /* Sync to disk if --sync-relay-log is set */ - if (sync_relaylog_period && - my_sync(log_file->file, MY_WME)) - DBUG_RETURN(2); } /* @@ -398,8 +393,12 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert, mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert); err= flush_io_cache(file); - if (sync_relaylog_period && !err) + if (sync_masterinfo_period && !err && + ++(mi->sync_counter) >= sync_masterinfo_period) + { err= my_sync(mi->fd, MYF(MY_WME)); + mi->sync_counter= 0; + } DBUG_RETURN(-err); } diff --git a/sql/rpl_mi.h b/sql/rpl_mi.h index 93fb0a98198..c59dffefb7c 100644 --- a/sql/rpl_mi.h +++ b/sql/rpl_mi.h @@ -58,7 +58,7 @@ class Master_info : public Slave_reporting_capability { public: - Master_info(); + Master_info(bool is_slave_recovery); ~Master_info(); /* the variables below are needed because we can change masters on the fly */ @@ -100,6 +100,13 @@ class Master_info : public Slave_reporting_capability */ long clock_diff_with_master; + + /* + Keeps track of the number of events before fsyncing. + The option --sync-master-info determines how many + events should happen before fsyncing. + */ + uint sync_counter; }; void init_master_info_with_options(Master_info* mi); diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index 37c0815fb8b..3a12164a1cf 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -28,11 +28,11 @@ 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, const char *default_val); - -Relay_log_info::Relay_log_info() +Relay_log_info::Relay_log_info(bool is_slave_recovery) :Slave_reporting_capability("SQL"), no_storage(FALSE), replicate_same_server_id(::replicate_same_server_id), info_fd(-1), cur_log_fd(-1), relay_log(&sync_relaylog_period), + sync_counter(0), is_relay_log_recovery(is_slave_recovery), save_temporary_tables(0), #if HAVE_purify is_fake(FALSE), @@ -259,7 +259,8 @@ Failed to open the existing relay log info file '%s' (errno %d)", rli->group_relay_log_pos= rli->event_relay_log_pos= relay_log_pos; rli->group_master_log_pos= master_log_pos; - if (init_relay_log_pos(rli, + if (!rli->is_relay_log_recovery && + init_relay_log_pos(rli, rli->group_relay_log_name, rli->group_relay_log_pos, 0 /* no data lock*/, @@ -274,6 +275,7 @@ Failed to open the existing relay log info file '%s' (errno %d)", } #ifndef DBUG_OFF + if (!rli->is_relay_log_recovery) { char llbuf1[22], llbuf2[22]; DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s", diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index 171778d9675..a5410dd0c79 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -96,6 +96,19 @@ public: LOG_INFO linfo; IO_CACHE cache_buf,*cur_log; + /* + Keeps track of the number of transactions that commits + before fsyncing. The option --sync-relay-log-info determines + how many transactions should commit before fsyncing. + */ + uint sync_counter; + + /* + Identifies when the recovery process is going on. + See sql/slave.cc:init_recovery for further details. + */ + bool is_relay_log_recovery; + /* The following variables are safe to read any time */ /* IO_CACHE of the info file - set only during init or end */ @@ -267,7 +280,7 @@ public: char slave_patternload_file[FN_REFLEN]; size_t slave_patternload_file_size; - Relay_log_info(); + Relay_log_info(bool is_slave_recovery); ~Relay_log_info(); /* diff --git a/sql/set_var.cc b/sql/set_var.cc index f2b5201cf8b..dcc3954ff1e 100644 --- a/sql/set_var.cc +++ b/sql/set_var.cc @@ -1534,19 +1534,19 @@ static bool get_unsigned(THD *thd, set_var *var, ulonglong user_max, } -bool sys_var_int_ptr::check(THD *thd, set_var *var) +bool sys_var_uint_ptr::check(THD *thd, set_var *var) { - var->save_result.ulong_value= (ulong) var->value->val_int(); + var->save_result.ulong_value= (ulong) var->value->val_uint(); return 0; } -bool sys_var_int_ptr::update(THD *thd, set_var *var) +bool sys_var_uint_ptr::update(THD *thd, set_var *var) { *value= (uint) var->save_result.ulong_value; return 0; } -void sys_var_int_ptr::set_default(THD *thd, enum_var_type type) +void sys_var_uint_ptr::set_default(THD *thd, enum_var_type type) { *value= (uint) option_limits->def_value; } diff --git a/sql/set_var.h b/sql/set_var.h index 02c87abed88..0202a15836d 100644 --- a/sql/set_var.h +++ b/sql/set_var.h @@ -178,10 +178,10 @@ public: /** Unsigned int system variable class */ -class sys_var_int_ptr :public sys_var +class sys_var_uint_ptr :public sys_var { public: - sys_var_int_ptr(sys_var_chain *chain, const char *name_arg, + sys_var_uint_ptr(sys_var_chain *chain, const char *name_arg, uint *value_ptr_arg, sys_after_update_func after_update_arg= NULL) :sys_var(name_arg, after_update_arg), diff --git a/sql/slave.cc b/sql/slave.cc index fac9ee214c5..5edb47df8b5 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -129,6 +129,7 @@ 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_recovery(Master_info* mi); 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, @@ -220,6 +221,7 @@ void unlock_slave_threads(Master_info* mi) int init_slave() { DBUG_ENTER("init_slave"); + int error= 0; /* This is called when mysqld starts. Before client connections are @@ -231,7 +233,7 @@ int init_slave() TODO: re-write this to interate through the list of files for multi-master */ - active_mi= new Master_info; + active_mi= new Master_info(relay_log_recovery); /* If --slave-skip-errors=... was not used, the string value for the @@ -250,6 +252,7 @@ int init_slave() if (!active_mi) { sql_print_error("Failed to allocate memory for the master info structure"); + error= 1; goto err; } @@ -257,6 +260,13 @@ int init_slave() !master_host, (SLAVE_IO | SLAVE_SQL))) { sql_print_error("Failed to initialize the master info structure"); + error= 1; + goto err; + } + + if (active_mi->rli.is_relay_log_recovery && init_recovery(active_mi)) + { + error= 1; goto err; } @@ -275,18 +285,89 @@ int init_slave() SLAVE_IO | SLAVE_SQL)) { sql_print_error("Failed to create slave threads"); + error= 1; goto err; } } - pthread_mutex_unlock(&LOCK_active_mi); - DBUG_RETURN(0); err: + active_mi->rli.is_relay_log_recovery= FALSE; pthread_mutex_unlock(&LOCK_active_mi); - DBUG_RETURN(1); + DBUG_RETURN(error); } - +/* + Updates the master info based on the information stored in the + relay info and ignores relay logs previously retrieved by the IO + thread, which thus starts fetching again based on to the + group_master_log_pos and group_master_log_name. Eventually, the old + relay logs will be purged by the normal purge mechanism. + + In the feature, we should improve this routine in order to avoid throwing + away logs that are safely stored in the disk. Note also that this recovery + routine relies on the correctness of the relay-log.info and only tolerates + coordinate problems in master.info. + + In this function, there is no need for a mutex as the caller + (i.e. init_slave) already has one acquired. + + Specifically, the following structures are updated: + + 1 - mi->master_log_pos <-- rli->group_master_log_pos + 2 - mi->master_log_name <-- rli->group_master_log_name + 3 - It moves the relay log to the new relay log file, by + rli->group_relay_log_pos <-- BIN_LOG_HEADER_SIZE; + rli->event_relay_log_pos <-- BIN_LOG_HEADER_SIZE; + rli->group_relay_log_name <-- rli->relay_log.get_log_fname(); + rli->event_relay_log_name <-- rli->relay_log.get_log_fname(); + + If there is an error, it returns (1), otherwise returns (0). + */ +static int init_recovery(Master_info* mi) +{ + const char *errmsg= 0; + DBUG_ENTER("init_recovery"); + + Relay_log_info *rli= &mi->rli; + if (rli->group_master_log_name[0]) + { + mi->master_log_pos= max(BIN_LOG_HEADER_SIZE, + rli->group_master_log_pos); + strmake(mi->master_log_name, rli->group_master_log_name, + sizeof(mi->master_log_name)-1); + + sql_print_warning("Recovery from master pos %ld and file %s.", + (ulong) mi->master_log_pos, mi->master_log_name); + + strmake(rli->group_relay_log_name, rli->relay_log.get_log_fname(), + sizeof(rli->group_relay_log_name)-1); + strmake(rli->event_relay_log_name, rli->relay_log.get_log_fname(), + sizeof(mi->rli.event_relay_log_name)-1); + + rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE; + + if (init_relay_log_pos(rli, + rli->group_relay_log_name, + rli->group_relay_log_pos, + 0 /*no data lock*/, + &errmsg, 0)) + DBUG_RETURN(1); + + if (flush_master_info(mi, 0)) + { + sql_print_error("Failed to flush master info file"); + DBUG_RETURN(1); + } + if (flush_relay_log_info(rli)) + { + sql_print_error("Failed to flush relay info file"); + DBUG_RETURN(1); + } + } + + DBUG_RETURN(0); +} + /** Convert slave skip errors bitmap into a printable string. */ @@ -3959,7 +4040,14 @@ bool flush_relay_log_info(Relay_log_info* rli) error=1; if (flush_io_cache(file)) error=1; - + if (sync_relayloginfo_period && + !error && + ++(rli->sync_counter) >= sync_relayloginfo_period) + { + if (my_sync(rli->info_fd, MYF(MY_WME))) + error=1; + rli->sync_counter= 0; + } /* Flushing the relay log is done by the slave I/O thread */ DBUG_RETURN(error); } @@ -4366,6 +4454,8 @@ void rotate_relay_log(Master_info* mi) DBUG_ENTER("rotate_relay_log"); Relay_log_info* rli= &mi->rli; + DBUG_EXECUTE_IF("crash_before_rotate_relaylog", abort();); + /* We don't lock rli->run_lock. This would lead to deadlocks. */ pthread_mutex_lock(&mi->run_lock); diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc index 96e99b57e3c..531242f64d1 100644 --- a/sql/sql_binlog.cc +++ b/sql/sql_binlog.cc @@ -58,7 +58,7 @@ void mysql_client_binlog_statement(THD* thd) my_bool have_fd_event= TRUE; if (!thd->rli_fake) { - thd->rli_fake= new Relay_log_info; + thd->rli_fake= new Relay_log_info(FALSE); #ifdef HAVE_purify thd->rli_fake->is_fake= TRUE; #endif diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 425d76c8b72..6295dbb0e79 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -1769,6 +1769,16 @@ static sys_var_const sys_relay_log_info_file(&vars, "relay_log_info_file", (uchar*) &relay_log_info_file); static sys_var_bool_ptr sys_relay_log_purge(&vars, "relay_log_purge", &relay_log_purge); +static sys_var_bool_ptr sys_relay_log_recovery(&vars, "relay_log_recovery", + &relay_log_recovery); +static sys_var_uint_ptr sys_sync_binlog_period(&vars, "sync_binlog", + &sync_binlog_period); +static sys_var_uint_ptr sys_sync_relaylog_period(&vars, "sync_relay_log", + &sync_relaylog_period); +static sys_var_uint_ptr sys_sync_relayloginfo_period(&vars, "sync_relay_log_info", + &sync_relayloginfo_period); +static sys_var_uint_ptr sys_sync_masterinfo_period(&vars, "sync_master_info", + &sync_masterinfo_period); static sys_var_const sys_relay_log_space_limit(&vars, "relay_log_space_limit", OPT_GLOBAL, SHOW_LONGLONG, @@ -1784,8 +1794,6 @@ static sys_var_const sys_slave_skip_errors(&vars, "slave_skip_errors", (uchar*) slave_skip_error_names); static sys_var_long_ptr sys_slave_trans_retries(&vars, "slave_transaction_retries", &slave_trans_retries); -static sys_var_int_ptr sys_sync_binlog_period(&vars, "sync_binlog", &sync_binlog_period); -static sys_var_int_ptr sys_sync_relaylog_period(&vars, "sync_relay_log", &sync_relaylog_period); static sys_var_slave_skip_counter sys_slave_skip_counter(&vars, "sql_slave_skip_counter"); -- cgit v1.2.1 From 43fe9e045c4c84bdb84d61d0dc364576d1691a30 Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Wed, 30 Sep 2009 19:36:35 +0800 Subject: Backporting BUG#40244 Optimized build of mysqld crashes when built with Sun Studio on SPARC --- sql/rpl_handler.cc | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'sql') diff --git a/sql/rpl_handler.cc b/sql/rpl_handler.cc index aea838928b9..da7aade5b99 100644 --- a/sql/rpl_handler.cc +++ b/sql/rpl_handler.cc @@ -88,11 +88,11 @@ int get_user_var_str(const char *name, char *value, int delegates_init() { - static unsigned char trans_mem[sizeof(Trans_delegate)]; - static unsigned char storage_mem[sizeof(Binlog_storage_delegate)]; + static unsigned long trans_mem[sizeof(Trans_delegate) / sizeof(unsigned long) + 1]; + static unsigned long storage_mem[sizeof(Binlog_storage_delegate) / sizeof(unsigned long) + 1]; #ifdef HAVE_REPLICATION - static unsigned char transmit_mem[sizeof(Binlog_transmit_delegate)]; - static unsigned char relay_io_mem[sizeof(Binlog_relay_IO_delegate)]; + static unsigned long transmit_mem[sizeof(Binlog_transmit_delegate) / sizeof(unsigned long) + 1]; + static unsigned long relay_io_mem[sizeof(Binlog_relay_IO_delegate)/ sizeof(unsigned long) + 1]; #endif if (!(transaction_delegate= new (trans_mem) Trans_delegate) -- cgit v1.2.1 From 5c25d17c4e63325048340a76eb8d2fdf6ae475f4 Mon Sep 17 00:00:00 2001 From: Alfranio Correia Date: Wed, 30 Sep 2009 22:41:05 +0100 Subject: BUG#43075 rpl.rpl_sync fails sporadically on pushbuild NOTE: Backporting the patch to next-mr. The slave was crashing while failing to execute the init_slave() function. The issue stems from two different reasons: 1 - A failure while allocating the master info structure generated a segfault due to a NULL pointer. 2 - A failure while recovering generated a segfault due to a non-initialized relay log file. In other words, the mi->init and rli->init were both set to true before executing the recovery process thus creating an inconsistent state as the relay log file was not initialized. To circumvent such problems, we refactored the recovery process which is now executed while initializing the relay log. It is ensured that the master info structure is created before accessing it and any error is propagated thus avoiding to set mi->init and rli->init to true when for instance the relay log is not initialized or the relay info is not flushed. The changes related to the refactory are described below: 1 - Removed call to init_recovery from init_slave. 2 - Changed the signature of the function init_recovery. 3 - Removed flushes. They are called while initializing the relay log and master info. 4 - Made sure that if the relay info is not flushed the mi-init and rli-init are not set to true. In this patch, we also replaced the exit(1) in the fault injection by DBUG_ABORT() to make it compliant with the code guidelines. --- sql/rpl_mi.cc | 1 + sql/rpl_rli.cc | 12 ++++++++---- sql/slave.cc | 31 ++----------------------------- sql/slave.h | 1 + 4 files changed, 12 insertions(+), 33 deletions(-) (limited to 'sql') diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc index cec2eabdd20..fe005bdb2a3 100644 --- a/sql/rpl_mi.cc +++ b/sql/rpl_mi.cc @@ -310,6 +310,7 @@ file '%s')", fname); goto err; mi->inited = 1; + mi->rli.is_relay_log_recovery= FALSE; // now change cache READ -> WRITE - must do this before flush_master_info reinit_io_cache(&mi->file, WRITE_CACHE, 0L, 0, 1); if ((error=test(flush_master_info(mi, 1)))) diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index 3a12164a1cf..b3a1bbc31d2 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -259,8 +259,10 @@ Failed to open the existing relay log info file '%s' (errno %d)", rli->group_relay_log_pos= rli->event_relay_log_pos= relay_log_pos; rli->group_master_log_pos= master_log_pos; - if (!rli->is_relay_log_recovery && - init_relay_log_pos(rli, + if (rli->is_relay_log_recovery && init_recovery(rli->mi, &msg)) + goto err; + + if (init_relay_log_pos(rli, rli->group_relay_log_name, rli->group_relay_log_pos, 0 /* no data lock*/, @@ -275,7 +277,6 @@ Failed to open the existing relay log info file '%s' (errno %d)", } #ifndef DBUG_OFF - if (!rli->is_relay_log_recovery) { char llbuf1[22], llbuf2[22]; DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s", @@ -292,7 +293,10 @@ Failed to open the existing relay log info file '%s' (errno %d)", */ reinit_io_cache(&rli->info_file, WRITE_CACHE,0L,0,1); if ((error= flush_relay_log_info(rli))) - sql_print_error("Failed to flush relay log info file"); + { + msg= "Failed to flush relay log info file"; + goto err; + } if (count_relay_log_space(rli)) { msg="Error counting relay log space"; diff --git a/sql/slave.cc b/sql/slave.cc index 5edb47df8b5..a1097d67052 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -129,7 +129,6 @@ 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_recovery(Master_info* mi); 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, @@ -264,12 +263,6 @@ int init_slave() goto err; } - if (active_mi->rli.is_relay_log_recovery && init_recovery(active_mi)) - { - error= 1; - goto err; - } - if (server_id && !master_host && active_mi->host[0]) master_host= active_mi->host; @@ -291,7 +284,6 @@ int init_slave() } err: - active_mi->rli.is_relay_log_recovery= FALSE; pthread_mutex_unlock(&LOCK_active_mi); DBUG_RETURN(error); } @@ -323,9 +315,8 @@ err: If there is an error, it returns (1), otherwise returns (0). */ -static int init_recovery(Master_info* mi) +int init_recovery(Master_info* mi, const char** errmsg) { - const char *errmsg= 0; DBUG_ENTER("init_recovery"); Relay_log_info *rli= &mi->rli; @@ -345,26 +336,8 @@ static int init_recovery(Master_info* mi) sizeof(mi->rli.event_relay_log_name)-1); rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE; - - if (init_relay_log_pos(rli, - rli->group_relay_log_name, - rli->group_relay_log_pos, - 0 /*no data lock*/, - &errmsg, 0)) - DBUG_RETURN(1); - - if (flush_master_info(mi, 0)) - { - sql_print_error("Failed to flush master info file"); - DBUG_RETURN(1); - } - if (flush_relay_log_info(rli)) - { - sql_print_error("Failed to flush relay info file"); - DBUG_RETURN(1); - } } - + DBUG_RETURN(0); } diff --git a/sql/slave.h b/sql/slave.h index a44a7eed83e..86b28931f21 100644 --- a/sql/slave.h +++ b/sql/slave.h @@ -134,6 +134,7 @@ extern ulonglong relay_log_space_limit; #define SLAVE_FORCE_ALL 4 int init_slave(); +int init_recovery(Master_info* mi, const char** errmsg); void init_slave_skip_errors(const char* arg); bool flush_relay_log_info(Relay_log_info* rli); int register_slave_on_master(MYSQL* mysql); -- cgit v1.2.1 From d91aa57c38e9baf0bbf31a596e79bea2504e50bb Mon Sep 17 00:00:00 2001 From: Andrei Elkin Date: Thu, 1 Oct 2009 19:44:53 +0300 Subject: backporting bug@27808 fixes --- sql/lex.h | 1 + sql/rpl_mi.cc | 88 +++++++++++++++++++++++++--- sql/rpl_mi.h | 9 ++- sql/share/errmsg.txt | 3 + sql/slave.cc | 160 ++++++++++++++++++++++++++++++++++++++++++++++++--- sql/sql_lex.h | 3 +- sql/sql_repl.cc | 58 ++++++++++++++----- sql/sql_yacc.yy | 23 ++++++++ 8 files changed, 313 insertions(+), 32 deletions(-) (limited to 'sql') diff --git a/sql/lex.h b/sql/lex.h index 790808a8c14..0a61a92fd60 100644 --- a/sql/lex.h +++ b/sql/lex.h @@ -244,6 +244,7 @@ static SYMBOL symbols[] = { { "IDENTIFIED", SYM(IDENTIFIED_SYM)}, { "IF", SYM(IF)}, { "IGNORE", SYM(IGNORE_SYM)}, + { "IGNORE_SERVER_IDS", SYM(IGNORE_SERVER_IDS_SYM)}, { "IMPORT", SYM(IMPORT)}, { "IN", SYM(IN_SYM)}, { "INDEX", SYM(INDEX_SYM)}, diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc index 77f7b7e1929..a50f75d53e3 100644 --- a/sql/rpl_mi.cc +++ b/sql/rpl_mi.cc @@ -27,17 +27,19 @@ 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, const char *default_val); int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val); +int init_dynarray_intvar_from_file(DYNAMIC_ARRAY* arr, IO_CACHE* f); Master_info::Master_info() :Slave_reporting_capability("I/O"), ssl(0), ssl_verify_server_cert(0), fd(-1), io_thd(0), inited(0), abort_slave(0),slave_running(0), slave_run_id(0), - heartbeat_period(0), received_heartbeats(0) + heartbeat_period(0), received_heartbeats(0), master_id(0) { host[0] = 0; user[0] = 0; password[0] = 0; ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0; ssl_cipher[0]= 0; ssl_key[0]= 0; + my_init_dynamic_array(&ignore_server_ids, sizeof(::server_id), 16, 16); bzero((char*) &file, sizeof(file)); pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST); pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST); @@ -48,6 +50,7 @@ Master_info::Master_info() Master_info::~Master_info() { + delete_dynamic(&ignore_server_ids); pthread_mutex_destroy(&run_lock); pthread_mutex_destroy(&data_lock); pthread_cond_destroy(&data_cond); @@ -55,6 +58,43 @@ Master_info::~Master_info() pthread_cond_destroy(&stop_cond); } +/** + A comparison function to be supplied as argument to @c sort_dynamic() + and @c bsearch() + + @return -1 if first argument is less, 0 if it equal to, 1 if it is greater + than the second +*/ +int change_master_server_id_cmp(ulong *id1, ulong *id2) +{ + return *id1 < *id2? -1 : (*id1 > *id2? 1 : 0); +} + + +/** + Reports if the s_id server has been configured to ignore events + it generates with + + CHANGE MASTER IGNORE_SERVER_IDS= ( list of server ids ) + + Method is called from the io thread event receiver filtering. + + @param s_id the master server identifier + + @retval TRUE if s_id is in the list of ignored master servers, + @retval FALSE otherwise. + */ +bool Master_info::shall_ignore_server_id(ulong s_id) +{ + if (likely(ignore_server_ids.elements == 1)) + return (* (ulong*) dynamic_array_ptr(&ignore_server_ids, 0)) == s_id; + else + return bsearch((const ulong *) &s_id, + ignore_server_ids.buffer, + ignore_server_ids.elements, sizeof(ulong), + (int (*) (const void*, const void*)) change_master_server_id_cmp) + != NULL; +} void init_master_info_with_options(Master_info* mi) { @@ -105,12 +145,12 @@ enum { /* 5.1.16 added value of master_ssl_verify_server_cert */ LINE_FOR_MASTER_SSL_VERIFY_SERVER_CERT= 15, - /* 6.0 added value of master_heartbeat_period */ LINE_FOR_MASTER_HEARTBEAT_PERIOD= 16, - + /* 6.0 added value of master_ignore_server_id */ + LINE_FOR_REPLICATE_IGNORE_SERVER_IDS= 17, /* Number of lines currently used when saving master info file */ - LINES_IN_MASTER_INFO= LINE_FOR_MASTER_HEARTBEAT_PERIOD + LINES_IN_MASTER_INFO= LINE_FOR_REPLICATE_IGNORE_SERVER_IDS }; int init_master_info(Master_info* mi, const char* master_info_fname, @@ -304,6 +344,16 @@ file '%s')", fname); if (lines >= LINE_FOR_MASTER_HEARTBEAT_PERIOD && init_floatvar_from_file(&master_heartbeat_period, &mi->file, 0.0)) goto errwithmsg; + /* + Starting from 6.0 list of server_id of ignorable servers might be + in the file + */ + if (lines >= LINE_FOR_REPLICATE_IGNORE_SERVER_IDS && + init_dynarray_intvar_from_file(&mi->ignore_server_ids, &mi->file)) + { + sql_print_error("Failed to initialize master info ignore_server_ids"); + goto errwithmsg; + } } #ifndef HAVE_OPENSSL @@ -384,7 +434,29 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) if (flush_relay_log_cache && flush_io_cache(mi->rli.relay_log.get_log_file())) DBUG_RETURN(2); - + + /* + produce a line listing the total number and all the ignored server_id:s + */ + char* ignore_server_ids_buf; + { + ignore_server_ids_buf= + (char *) my_malloc((sizeof(::server_id) * 3 + 1) * + (1 + mi->ignore_server_ids.elements), MYF(MY_WME)); + if (!ignore_server_ids_buf) + DBUG_RETURN(1); + for (ulong i= 0, cur_len= my_sprintf(ignore_server_ids_buf, + (ignore_server_ids_buf, "%u", + mi->ignore_server_ids.elements)); + i < mi->ignore_server_ids.elements; i++) + { + ulong s_id; + get_dynamic(&mi->ignore_server_ids, (uchar*) &s_id, i); + cur_len +=my_sprintf(ignore_server_ids_buf + cur_len, + (ignore_server_ids_buf + cur_len, + " %lu", s_id)); + } + } /* We flushed the relay log BEFORE the master.info file, because if we crash now, we will get a duplicate event in the relay log at restart. If we @@ -405,14 +477,16 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache) my_sprintf(heartbeat_buf, (heartbeat_buf, "%.3f", mi->heartbeat_period)); my_b_seek(file, 0L); my_b_printf(file, - "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n", + "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n%s\n", LINES_IN_MASTER_INFO, mi->master_log_name, llstr(mi->master_log_pos, lbuf), mi->host, mi->user, mi->password, mi->port, mi->connect_retry, (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert, mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert, - heartbeat_buf); + heartbeat_buf, ignore_server_ids_buf); + + my_free(ignore_server_ids_buf, MYF(0)); DBUG_RETURN(-flush_io_cache(file)); } diff --git a/sql/rpl_mi.h b/sql/rpl_mi.h index 35e18414932..0caa6904da4 100644 --- a/sql/rpl_mi.h +++ b/sql/rpl_mi.h @@ -20,6 +20,7 @@ #include "rpl_rli.h" #include "rpl_reporting.h" +#include "my_sys.h" /***************************************************************************** @@ -60,6 +61,7 @@ class Master_info : public Slave_reporting_capability public: Master_info(); ~Master_info(); + bool shall_ignore_server_id(ulong s_id); /* the variables below are needed because we can change masters on the fly */ char master_log_name[FN_REFLEN]; @@ -83,8 +85,6 @@ class Master_info : public Slave_reporting_capability Relay_log_info rli; uint port; uint connect_retry; - float heartbeat_period; // interface with CHANGE MASTER or master.info - ulonglong received_heartbeats; // counter of received heartbeat events #ifndef DBUG_OFF int events_till_disconnect; #endif @@ -102,6 +102,10 @@ class Master_info : public Slave_reporting_capability */ long clock_diff_with_master; + float heartbeat_period; // interface with CHANGE MASTER or master.info + ulonglong received_heartbeats; // counter of received heartbeat events + DYNAMIC_ARRAY ignore_server_ids; + ulong master_id; }; void init_master_info_with_options(Master_info* mi); @@ -111,6 +115,7 @@ int init_master_info(Master_info* mi, const char* master_info_fname, int thread_mask); void end_master_info(Master_info* mi); int flush_master_info(Master_info* mi, bool flush_relay_log_cache); +int change_master_server_id_cmp(ulong *id1, ulong *id2); #endif /* HAVE_REPLICATION */ #endif /* RPL_MI_H */ diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt index 3aba434b284..18d3a41424a 100644 --- a/sql/share/errmsg.txt +++ b/sql/share/errmsg.txt @@ -6163,6 +6163,9 @@ ER_SLAVE_HEARTBEAT_FAILURE ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE eng "The requested value for the heartbeat period %s %s" +ER_SLAVE_IGNORE_SERVER_IDS + eng "The requested server id %d clashes with the slave startup option --replicate-same-server-id" + ER_NDB_REPLICATION_SCHEMA_ERROR eng "Bad schema for mysql.ndb_replication table. Message: %-.64s" ER_CONFLICT_FN_PARSE_ERROR diff --git a/sql/slave.cc b/sql/slave.cc index bbdfb8e633f..fe4a2f6ba0a 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -880,6 +880,95 @@ int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val) DBUG_RETURN(1); } + +/** + A master info read method + + This function is called from @c init_master_info() along with + relatives to restore some of @c active_mi members. + Particularly, this function is responsible for restoring + IGNORE_SERVER_IDS list of servers whose events the slave is + going to ignore (to not log them in the relay log). + Items being read are supposed to be decimal output of values of a + type shorter or equal of @c long and separated by the single space. + + @param arr @c DYNAMIC_ARRAY pointer to storage for servers id + @param f @c IO_CACHE pointer to the source file + + @retval 0 All OK + @retval non-zero An error +*/ + +int init_dynarray_intvar_from_file(DYNAMIC_ARRAY* arr, IO_CACHE* f) +{ + int ret= 0; + char buf[16 * (sizeof(long)*4 + 1)]; // static buffer to use most of times + char *buf_act= buf; // actual buffer can be dynamic if static is short + char *token, *last; + uint num_items; // number of items of `arr' + size_t read_size; + DBUG_ENTER("init_dynarray_intvar_from_file"); + + if ((read_size= my_b_gets(f, buf_act, sizeof(buf))) == 0) + { + return 0; // no line in master.info + } + if (read_size + 1 == sizeof(buf) && buf[sizeof(buf) - 2] != '\n') + { + /* + short read happend; allocate sufficient memory and make the 2nd read + */ + char buf_work[(sizeof(long)*3 + 1)*16]; + memcpy(buf_work, buf, sizeof(buf_work)); + num_items= atoi(strtok_r(buf_work, " ", &last)); + size_t snd_size; + /* + max size lower bound approximate estimation bases on the formula: + (the items number + items themselves) * + (decimal size + space) - 1 + `\n' + '\0' + */ + size_t max_size= (1 + num_items) * (sizeof(long)*3 + 1) + 1; + buf_act= (char*) my_malloc(max_size, MYF(MY_WME)); + memcpy(buf_act, buf, read_size); + snd_size= my_b_gets(f, buf_act + read_size, max_size - read_size); + if (snd_size == 0 || + (snd_size + 1 == max_size - read_size) && buf[max_size - 2] != '\n') + { + /* + failure to make the 2nd read or short read again + */ + ret= 1; + goto err; + } + } + token= strtok_r(buf_act, " ", &last); + if (token == NULL) + { + ret= 1; + goto err; + } + num_items= atoi(token); + for (uint i=0; i < num_items; i++) + { + token= strtok_r(NULL, " ", &last); + if (token == NULL) + { + ret= 1; + goto err; + } + else + { + ulong val= atol(token); + insert_dynamic(arr, (uchar *) &val); + } + } +err: + if (buf_act != buf) + my_free(buf_act, MYF(0)); + DBUG_RETURN(ret); +} + + static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info) { if (io_slave_killed(thd, mi)) @@ -1058,7 +1147,7 @@ static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi) (master_res= mysql_store_result(mysql)) && (master_row= mysql_fetch_row(master_res))) { - if ((::server_id == strtoul(master_row[1], 0, 10)) && + if ((::server_id == (mi->master_id= strtoul(master_row[1], 0, 10))) && !mi->rli.replicate_same_server_id) { errmsg= "The slave I/O thread stops because master and slave have equal \ @@ -1096,6 +1185,13 @@ maybe it is a *VERY OLD MASTER*."); mysql_free_result(master_res); master_res= NULL; } + if (mi->master_id == 0 && mi->ignore_server_ids.elements > 0) + { + errmsg= "Slave configured with server id filtering could not detect the master server id."; + err_code= ER_SLAVE_FATAL_ERROR; + sprintf(err_buff, ER(err_code), errmsg); + goto err; + } /* Check that the master's global character_set_server and ours are the same. @@ -1659,6 +1755,10 @@ bool show_master_info(THD* thd, Master_info* mi) field_list.push_back(new Item_empty_string("Last_IO_Error", 20)); field_list.push_back(new Item_return_int("Last_SQL_Errno", 4, MYSQL_TYPE_LONG)); field_list.push_back(new Item_empty_string("Last_SQL_Error", 20)); + field_list.push_back(new Item_empty_string("Replicate_Ignore_Server_Ids", + FN_REFLEN)); + field_list.push_back(new Item_return_int("Master_Server_Id", sizeof(ulong), + MYSQL_TYPE_LONG)); if (protocol->send_fields(&field_list, Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF)) @@ -1780,6 +1880,32 @@ bool show_master_info(THD* thd, Master_info* mi) protocol->store(mi->rli.last_error().number); // Last_SQL_Error protocol->store(mi->rli.last_error().message, &my_charset_bin); + // Replicate_Ignore_Server_Ids + { + char buff[FN_REFLEN]; + ulong i, cur_len; + for (i= 0, buff[0]= 0, cur_len= 0; + i < mi->ignore_server_ids.elements; i++) + { + ulong s_id, slen; + char sbuff[FN_REFLEN]; + get_dynamic(&mi->ignore_server_ids, (uchar*) &s_id, i); + slen= my_sprintf(sbuff, (sbuff, (i==0? "%lu" : ", %lu"), s_id)); + if (cur_len + slen + 4 > FN_REFLEN) + { + /* + break the loop whenever remained space could not fit + ellipses on the next cycle + */ + my_sprintf(buff + cur_len, (buff + cur_len, "...")); + break; + } + cur_len += my_sprintf(buff + cur_len, (buff + cur_len, "%s", sbuff)); + } + protocol->store(buff, &my_charset_bin); + } + // Master_Server_id + protocol->store((uint32) mi->master_id); pthread_mutex_unlock(&mi->rli.err_lock); pthread_mutex_unlock(&mi->err_lock); @@ -3599,6 +3725,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) ulong inc_pos; Relay_log_info *rli= &mi->rli; pthread_mutex_t *log_lock= rli->relay_log.get_log_lock(); + ulong s_id; DBUG_ENTER("queue_event"); LINT_INIT(inc_pos); @@ -3745,9 +3872,20 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) */ pthread_mutex_lock(log_lock); - - if ((uint4korr(buf + SERVER_ID_OFFSET) == ::server_id) && - !mi->rli.replicate_same_server_id) + s_id= uint4korr(buf + SERVER_ID_OFFSET); + if ((s_id == ::server_id && !mi->rli.replicate_same_server_id) || + /* + the following conjunction deals with IGNORE_SERVER_IDS, if set + If the master is on the ignore list, execution of + format description log events and rotate events is necessary. + */ + (mi->ignore_server_ids.elements > 0 && + mi->shall_ignore_server_id(s_id) && + /* everything is filtered out from non-master */ + (s_id != mi->master_id || + /* for the master meta information is necessary */ + buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT && + buf[EVENT_TYPE_OFFSET] != ROTATE_EVENT))) { /* Do not write it to the relay log. @@ -3762,10 +3900,14 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) But events which were generated by this slave and which do not exist in the master's binlog (i.e. Format_desc, Rotate & Stop) should not increment mi->master_log_pos. + If the event is originated remotely and is being filtered out by + IGNORE_SERVER_IDS it increments mi->master_log_pos + as well as rli->group_relay_log_pos. */ - if (buf[EVENT_TYPE_OFFSET]!=FORMAT_DESCRIPTION_EVENT && - buf[EVENT_TYPE_OFFSET]!=ROTATE_EVENT && - buf[EVENT_TYPE_OFFSET]!=STOP_EVENT) + if (!(s_id == ::server_id && !mi->rli.replicate_same_server_id) || + buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT && + buf[EVENT_TYPE_OFFSET] != ROTATE_EVENT && + buf[EVENT_TYPE_OFFSET] != STOP_EVENT) { mi->master_log_pos+= inc_pos; memcpy(rli->ign_master_log_name_end, mi->master_log_name, FN_REFLEN); @@ -3773,8 +3915,8 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) rli->ign_master_log_pos_end= mi->master_log_pos; } rli->relay_log.signal_update(); // the slave SQL thread needs to re-check - DBUG_PRINT("info", ("master_log_pos: %lu, event originating from the same server, ignored", - (ulong) mi->master_log_pos)); + DBUG_PRINT("info", ("master_log_pos: %lu, event originating from %u server, ignored", + (ulong) mi->master_log_pos, uint4korr(buf + SERVER_ID_OFFSET))); } else { diff --git a/sql/sql_lex.h b/sql/sql_lex.h index f6effab93a4..c4ab3091d6a 100644 --- a/sql/sql_lex.h +++ b/sql/sql_lex.h @@ -214,10 +214,11 @@ typedef struct st_lex_master_info changed variable or if it should be left at old value */ enum {LEX_MI_UNCHANGED, LEX_MI_DISABLE, LEX_MI_ENABLE} - ssl, ssl_verify_server_cert, heartbeat_opt; + ssl, ssl_verify_server_cert, heartbeat_opt, repl_ignore_server_ids_opt; char *ssl_key, *ssl_cert, *ssl_ca, *ssl_capath, *ssl_cipher; char *relay_log_name; ulong relay_log_pos; + DYNAMIC_ARRAY repl_ignore_server_ids; } LEX_MASTER_INFO; diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index cde713b1b40..eadb3362882 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -1266,26 +1266,27 @@ bool change_master(THD* thd, Master_info* mi) int thread_mask; const char* errmsg= 0; bool need_relay_log_purge= 1; + bool ret= FALSE; DBUG_ENTER("change_master"); 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)); - unlock_slave_threads(mi); - DBUG_RETURN(TRUE); + ret= TRUE; + goto err; } thd_proc_info(thd, "Changing master"); - LEX_MASTER_INFO* lex_mi= &thd->lex->mi; // TODO: see if needs re-write if (init_master_info(mi, master_info_file, relay_log_info_file, 0, thread_mask)) { my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0)); - unlock_slave_threads(mi); - DBUG_RETURN(TRUE); + ret= TRUE; + goto err; } /* @@ -1330,6 +1331,34 @@ bool change_master(THD* thd, Master_info* mi) mi->heartbeat_period= (float) min(SLAVE_MAX_HEARTBEAT_PERIOD, (slave_net_timeout/2.0)); mi->received_heartbeats= LL(0); // counter lives until master is CHANGEd + /* + reset the last time server_id list if the current CHANGE MASTER + is mentioning IGNORE_SERVER_IDS= (...) + */ + if (lex_mi->repl_ignore_server_ids_opt == LEX_MASTER_INFO::LEX_MI_ENABLE) + reset_dynamic(&mi->ignore_server_ids); + for (uint i= 0; i < lex_mi->repl_ignore_server_ids.elements; i++) + { + ulong s_id; + get_dynamic(&lex_mi->repl_ignore_server_ids, (uchar*) &s_id, i); + if (s_id == ::server_id && replicate_same_server_id) + { + my_error(ER_SLAVE_IGNORE_SERVER_IDS, MYF(0), s_id); + ret= TRUE; + goto err; + } + else + { + if (bsearch((const ulong *) &s_id, + mi->ignore_server_ids.buffer, + mi->ignore_server_ids.elements, sizeof(ulong), + (int (*) (const void*, const void*)) + change_master_server_id_cmp) == NULL) + insert_dynamic(&mi->ignore_server_ids, (uchar*) &s_id); + } + } + sort_dynamic(&mi->ignore_server_ids, (qsort_cmp) change_master_server_id_cmp); + if (lex_mi->ssl != LEX_MASTER_INFO::LEX_MI_UNCHANGED) mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::LEX_MI_ENABLE); @@ -1407,8 +1436,8 @@ bool change_master(THD* thd, Master_info* mi) if (flush_master_info(mi, 0)) { my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file"); - unlock_slave_threads(mi); - DBUG_RETURN(TRUE); + ret= TRUE; + goto err; } if (need_relay_log_purge) { @@ -1419,8 +1448,8 @@ bool change_master(THD* thd, Master_info* mi) &errmsg)) { my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg); - unlock_slave_threads(mi); - DBUG_RETURN(TRUE); + ret= TRUE; + goto err; } } else @@ -1435,8 +1464,8 @@ bool change_master(THD* thd, Master_info* mi) &msg, 0)) { my_error(ER_RELAY_LOG_INIT, MYF(0), msg); - unlock_slave_threads(mi); - DBUG_RETURN(TRUE); + ret= TRUE; + goto err; } } /* @@ -1473,10 +1502,13 @@ bool change_master(THD* thd, Master_info* mi) pthread_cond_broadcast(&mi->data_cond); pthread_mutex_unlock(&mi->rli.data_lock); +err: unlock_slave_threads(mi); thd_proc_info(thd, 0); - my_ok(thd); - DBUG_RETURN(FALSE); + if (ret == FALSE) + my_ok(thd); + delete_dynamic(&lex_mi->repl_ignore_server_ids); //freeing of parser-time alloc + DBUG_RETURN(ret); } diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy index 50395d386e8..7dff91befb0 100644 --- a/sql/sql_yacc.yy +++ b/sql/sql_yacc.yy @@ -740,6 +740,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize); %token IDENT_QUOTED %token IF %token IGNORE_SYM +%token IGNORE_SERVER_IDS_SYM %token IMPORT %token INDEXES %token INDEX_SYM @@ -1559,6 +1560,12 @@ change: LEX *lex = Lex; lex->sql_command = SQLCOM_CHANGE_MASTER; bzero((char*) &lex->mi, sizeof(lex->mi)); + /* + resetting flags that can left from the previous CHANGE MASTER + */ + lex->mi.repl_ignore_server_ids_opt= LEX_MASTER_INFO::LEX_MI_UNCHANGED; + my_init_dynamic_array(&Lex->mi.repl_ignore_server_ids, + sizeof(::server_id), 16, 16); } master_defs {} @@ -1661,10 +1668,26 @@ master_def: } Lex->mi.heartbeat_opt= LEX_MASTER_INFO::LEX_MI_ENABLE; } + | IGNORE_SERVER_IDS_SYM EQ '(' ignore_server_id_list ')' + { + Lex->mi.repl_ignore_server_ids_opt= LEX_MASTER_INFO::LEX_MI_ENABLE; + } | master_file_def ; +ignore_server_id_list: + /* Empty */ + | ignore_server_id + | ignore_server_id_list ',' ignore_server_id + ; + +ignore_server_id: + ulong_num + { + insert_dynamic(&Lex->mi.repl_ignore_server_ids, (uchar*) &($1)); + } + master_file_def: MASTER_LOG_FILE_SYM EQ TEXT_STRING_sys { -- cgit v1.2.1 From ebca60c1ff07ecb57cd2852052534640322ca576 Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Fri, 2 Oct 2009 13:59:42 +0800 Subject: Backport BUG#41013 main.bootstrap coredumps in 6.0-rpl When a storage engine failed to initialize before allocated slot number, the slot number would be 0, and when later finalizing this plugin, it would accidentally unplug the storage engine currently uses slot 0. This patch fixed this problem by add a new macro value HA_SLOT_UNDEF to distinguish undefined slot number from slot 0. --- sql/handler.cc | 9 ++++++++- sql/handler.h | 7 +++++++ 2 files changed, 15 insertions(+), 1 deletion(-) (limited to 'sql') diff --git a/sql/handler.cc b/sql/handler.cc index f966a9099ee..ac959cb62f2 100644 --- a/sql/handler.cc +++ b/sql/handler.cc @@ -413,7 +413,13 @@ int ha_finalize_handlerton(st_plugin_int *plugin) reuse an array slot. Otherwise the number of uninstall/install cycles would be limited. */ - hton2plugin[hton->slot]= NULL; + if (hton->slot != HA_SLOT_UNDEF) + { + /* Make sure we are not unpluging another plugin */ + DBUG_ASSERT(hton2plugin[hton->slot] == plugin); + DBUG_ASSERT(hton->slot < MAX_HA); + hton2plugin[hton->slot]= NULL; + } my_free((uchar*)hton, MYF(0)); @@ -438,6 +444,7 @@ int ha_initialize_handlerton(st_plugin_int *plugin) goto err_no_hton_memory; } + hton->slot= HA_SLOT_UNDEF; /* Historical Requirement */ plugin->data= hton; // shortcut for the future if (plugin->plugin->init && plugin->plugin->init(hton)) diff --git a/sql/handler.h b/sql/handler.h index a281aaa0ad7..ea0b134e53d 100644 --- a/sql/handler.h +++ b/sql/handler.h @@ -216,6 +216,13 @@ */ #define MAX_HA 15 +/* + Use this instead of 0 as the initial value for the slot number of + handlerton, so that we can distinguish uninitialized slot number + from slot 0. +*/ +#define HA_SLOT_UNDEF ((uint)-1) + /* Parameters for open() (in register form->filestat) HA_GET_INFO does an implicit HA_ABORT_IF_LOCKED -- cgit v1.2.1 From f509a3589653575f1e9dc9f05de11a86be710a68 Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Fri, 2 Oct 2009 16:07:33 +0800 Subject: Backport BUG#41613 Slave I/O status inconsistent between SHOW SLAVE STATUS and START SLAVE There are three internal status for slave I/O thread, both MYSQL_SLAVE_RUN_NOT_CONNECT and MYSQL_SLAVE_NOT_RUN are reported as 'No' for Slave_IO_running of command SHOW SLAVE STATUS. Change MYSQL_SLAVE_RUN_NOT_CONNECT to be reported as 'Connecting'. --- sql/slave.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'sql') diff --git a/sql/slave.cc b/sql/slave.cc index 4c13841875c..47842cce225 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -1848,7 +1848,8 @@ bool show_master_info(THD* thd, Master_info* mi) protocol->store((ulonglong) mi->rli.group_relay_log_pos); protocol->store(mi->rli.group_master_log_name, &my_charset_bin); protocol->store(mi->slave_running == MYSQL_SLAVE_RUN_CONNECT ? - "Yes" : "No", &my_charset_bin); + "Yes" : (mi->slave_running == MYSQL_SLAVE_RUN_NOT_CONNECT ? + "Connecting" : "No"), &my_charset_bin); protocol->store(mi->rli.slave_running ? "Yes":"No", &my_charset_bin); protocol->store(rpl_filter->get_do_db()); protocol->store(rpl_filter->get_ignore_db()); -- cgit v1.2.1 From 228ae2bf50c776e94decc731decefe6eaae0118f Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Fri, 2 Oct 2009 16:35:03 +0800 Subject: Backport BUG#12190 CHANGE MASTER has differ path requiremts on MASTER_LOG_FILE and RELAY_LOG_FILE CHANGE MASTER TO command required the value for RELAY_LOG_FILE to be an absolute path, which was different from the requirement of MASTER_LOG_FILE. This patch fixed the problem by changing the value for RELAY_LOG_FILE to be the basename of the log file as that for MASTER_LOG_FILE. --- sql/sql_repl.cc | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) (limited to 'sql') diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index a899c8b7551..2c373059bf7 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -1400,9 +1400,11 @@ bool change_master(THD* thd, Master_info* mi) if (lex_mi->relay_log_name) { need_relay_log_purge= 0; - strmake(mi->rli.group_relay_log_name,lex_mi->relay_log_name, + char relay_log_name[FN_REFLEN]; + mi->rli.relay_log.make_log_name(relay_log_name, lex_mi->relay_log_name); + strmake(mi->rli.group_relay_log_name, relay_log_name, sizeof(mi->rli.group_relay_log_name)-1); - strmake(mi->rli.event_relay_log_name,lex_mi->relay_log_name, + strmake(mi->rli.event_relay_log_name, relay_log_name, sizeof(mi->rli.event_relay_log_name)-1); } -- cgit v1.2.1 From 4381f7ed90f54be0be0c569e38d17aefe98b7d0b Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Fri, 2 Oct 2009 16:40:06 +0800 Subject: Backport post fix compiler warnings and test failures for BUG#25192 BUG#12190 --- sql/log.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'sql') diff --git a/sql/log.cc b/sql/log.cc index 16f54649d2d..a523c111e87 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -1901,7 +1901,7 @@ updating the index files.", max_found); */ if (((strlen(ext_buf) + (end - name)) >= FN_REFLEN)) { - sql_print_error("Log filename too large: %s%s (%d). \ + sql_print_error("Log filename too large: %s%s (%lu). \ Please fix this by archiving old logs and updating the \ index files.", name, ext_buf, (strlen(ext_buf) + (end - name))); error= 1; -- cgit v1.2.1 From dfbac1dd1dea74ad281f1bcfc98dc107a8180192 Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Fri, 2 Oct 2009 16:50:05 +0800 Subject: Backport BUG#34227 Replication permission error message is misleading According to Jon's comment, add (at least one of) to the error message. --- sql/share/errmsg.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'sql') diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt index 18d3a41424a..1773599c63d 100644 --- a/sql/share/errmsg.txt +++ b/sql/share/errmsg.txt @@ -4620,7 +4620,7 @@ ER_USER_LIMIT_REACHED 42000 swe "Användare '%-.64s' har överskridit '%s' (nuvarande värde: %ld)" ER_SPECIFIC_ACCESS_DENIED_ERROR 42000 nla "Toegang geweigerd. U moet het %-.128s privilege hebben voor deze operatie" - eng "Access denied; you need the %-.128s privilege for this operation" + eng "Access denied; you need (at least one of) the %-.128s privilege(s) for this operation" ger "Kein Zugriff. Hierfür wird die Berechtigung %-.128s benötigt" ita "Accesso non consentito. Serve il privilegio %-.128s per questa operazione" por "Acesso negado. Você precisa o privilégio %-.128s para essa operação" -- cgit v1.2.1 From 66b869d63740a146277346ba7f7ab06264cd6527 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Wed, 7 Oct 2009 22:13:07 +0100 Subject: BUG#40611, BUG#44779: reverted in mysql-5.1-rep+2. --- sql/log.cc | 75 ++++++++++++-------------------------------------------------- sql/log.h | 13 ----------- 2 files changed, 14 insertions(+), 74 deletions(-) (limited to 'sql') diff --git a/sql/log.cc b/sql/log.cc index 45d71379b9e..5926c2c036f 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -53,7 +53,7 @@ LOGGER logger; MYSQL_BIN_LOG mysql_bin_log(&sync_binlog_period); static bool test_if_number(const char *str, - ulong *res, bool allow_wildcards); + long *res, bool allow_wildcards); static int binlog_init(void *p); static int binlog_close_connection(handlerton *hton, THD *thd); static int binlog_savepoint_set(handlerton *hton, THD *thd, void *sv); @@ -1835,27 +1835,22 @@ static void setup_windows_event_source() /** Find a unique filename for 'filename.#'. - Set '#' to the number next to the maximum found in the most - recent log file extension. - - This function will return nonzero if: (i) the generated name - exceeds FN_REFLEN; (ii) if the number of extensions is exhausted; - or (iii) some other error happened while examining the filesystem. + Set '#' to a number as low as possible. @return - nonzero if not possible to get unique filename. + nonzero if not possible to get unique filename */ static int find_uniq_filename(char *name) { + long number; uint i; - char buff[FN_REFLEN], ext_buf[FN_REFLEN]; + char buff[FN_REFLEN]; struct st_my_dir *dir_info; reg1 struct fileinfo *file_info; - ulong max_found= 0, next= 0, number= 0; + ulong max_found=0; size_t buf_length, length; char *start, *end; - int error= 0; DBUG_ENTER("find_uniq_filename"); length= dirname_part(buff, name, &buf_length); @@ -1863,15 +1858,15 @@ static int find_uniq_filename(char *name) end= strend(start); *end='.'; - length= (size_t) (end - start + 1); + length= (size_t) (end-start+1); - if (!(dir_info= my_dir(buff,MYF(MY_DONT_SORT)))) + if (!(dir_info = my_dir(buff,MYF(MY_DONT_SORT)))) { // This shouldn't happen strmov(end,".1"); // use name+1 - DBUG_RETURN(1); + DBUG_RETURN(0); } file_info= dir_info->dir_entry; - for (i= dir_info->number_off_files ; i-- ; file_info++) + for (i=dir_info->number_off_files ; i-- ; file_info++) { if (bcmp((uchar*) file_info->name, (uchar*) start, length) == 0 && test_if_number(file_info->name+length, &number,0)) @@ -1881,44 +1876,9 @@ static int find_uniq_filename(char *name) } my_dirend(dir_info); - /* check if reached the maximum possible extension number */ - if ((max_found == MAX_LOG_UNIQUE_FN_EXT)) - { - sql_print_error("Log filename extension number exhausted: %06lu. \ -Please fix this by archiving old logs and \ -updating the index files.", max_found); - error= 1; - goto end; - } - - next= max_found + 1; - sprintf(ext_buf, "%06lu", next); *end++='.'; - - /* - Check if the generated extension size + the file name exceeds the - buffer size used. If one did not check this, then the filename might be - truncated, resulting in error. - */ - if (((strlen(ext_buf) + (end - name)) >= FN_REFLEN)) - { - sql_print_error("Log filename too large: %s%s (%lu). \ -Please fix this by archiving old logs and updating the \ -index files.", name, ext_buf, (strlen(ext_buf) + (end - name))); - error= 1; - goto end; - } - - sprintf(end, "%06lu", next); - - /* print warning if reaching the end of available extensions. */ - if ((next > (MAX_LOG_UNIQUE_FN_EXT - LOG_WARN_UNIQUE_FN_EXT_LEFT))) - sql_print_warning("Next log extension: %lu. \ -Remaining log filename extensions: %lu. \ -Please consider archiving some logs.", next, (MAX_LOG_UNIQUE_FN_EXT - next)); - -end: - DBUG_RETURN(error); + sprintf(end,"%06ld",max_found+1); + DBUG_RETURN(0); } @@ -2117,13 +2077,6 @@ int MYSQL_LOG::generate_new_name(char *new_name, const char *log_name) { if (find_uniq_filename(new_name)) { - /* - This should be treated as error once propagation of error further - up in the stack gets proper handling. - */ - push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN, - ER_NO_UNIQUE_LOGFILE, ER(ER_NO_UNIQUE_LOGFILE), - log_name); sql_print_error(ER(ER_NO_UNIQUE_LOGFILE), log_name); return 1; } @@ -4841,11 +4794,11 @@ void MYSQL_BIN_LOG::set_max_size(ulong max_size_arg) @retval 1 String is a number @retval - 0 String is not a number + 0 Error */ static bool test_if_number(register const char *str, - ulong *res, bool allow_wildcards) + long *res, bool allow_wildcards) { reg2 int flag; const char *start; diff --git a/sql/log.h b/sql/log.h index b22e22a2cf5..a31be6dcce6 100644 --- a/sql/log.h +++ b/sql/log.h @@ -121,19 +121,6 @@ extern TC_LOG_DUMMY tc_log_dummy; #define LOG_CLOSE_TO_BE_OPENED 2 #define LOG_CLOSE_STOP_EVENT 4 -/* - Maximum unique log filename extension. - Note: setting to 0x7FFFFFFF due to atol windows - overflow/truncate. - */ -#define MAX_LOG_UNIQUE_FN_EXT 0x7FFFFFFF - -/* - Number of warnings that will be printed to error log - before extension number is exhausted. -*/ -#define LOG_WARN_UNIQUE_FN_EXT_LEFT 1000 - class Relay_log_info; typedef struct st_log_info -- cgit v1.2.1 From 6a541923c390a81a64a373fc91aafaea6f748b85 Mon Sep 17 00:00:00 2001 From: Andrei Elkin Date: Fri, 9 Oct 2009 16:26:37 +0300 Subject: backporting fixes of bug@45940 to 5.1-rpl+2 to cover failures rpl_heartbeat_* as well. mysql-test/suite/rpl/r/rpl_stop_middle_group.result: the new result file mysql-test/suite/rpl/t/rpl_stop_middle_group.test: renamed from rpl_row_stop_middle_update and added a regression test for bug#45940. --- sql/log_event.cc | 53 ++++++++++----------- sql/log_event_old.cc | 59 +----------------------- sql/rpl_rli.cc | 1 - sql/rpl_rli.h | 10 ++-- sql/slave.cc | 128 ++++++++++++++++++++++++++++++++++++++++----------- 5 files changed, 132 insertions(+), 119 deletions(-) (limited to 'sql') diff --git a/sql/log_event.cc b/sql/log_event.cc index d8dd40c91de..e2f0029eb6e 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -3272,6 +3272,21 @@ Default database: '%s'. Query: '%s'", */ } /* End of if (db_ok(... */ + {/** + The following failure injecion works in cooperation with tests + setting @@global.debug= 'd,stop_slave_middle_group'. + The sql thread receives the killed status and will proceed + to shutdown trying to finish incomplete events group. + */ + DBUG_EXECUTE_IF("stop_slave_middle_group", + if (strcmp("COMMIT", query) != 0 && + strcmp("BEGIN", query) != 0) + { + if (thd->transaction.all.modified_non_trans_table) + const_cast(rli)->abort_slave= 1; + };); + } + end: /* Probably we have set thd->query, thd->db, thd->catalog to point to places @@ -7475,8 +7490,16 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli) thd->transaction.stmt.modified_non_trans_table= TRUE; } // row processing loop - DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event", - const_cast(rli)->abort_slave= 1;); + {/** + The following failure injecion works in cooperation with tests + setting @@global.debug= 'd,stop_slave_middle_group'. + The sql thread receives the killed status and will proceed + to shutdown trying to finish incomplete events group. + */ + DBUG_EXECUTE_IF("stop_slave_middle_group", + if (thd->transaction.all.modified_non_trans_table) + const_cast(rli)->abort_slave= 1;); + } if ((error= do_after_row_operations(rli, error)) && ignored_error_code(convert_handler_error(error, thd, table))) @@ -7513,32 +7536,6 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli) const_cast(rli)->cleanup_context(thd, error); thd->is_slave_error= 1; } - /* - This code would ideally be placed in do_update_pos() instead, but - since we have no access to table there, we do the setting of - last_event_start_time here instead. - */ - else if (table && (table->s->primary_key == MAX_KEY) && - !cache_stmt && get_flags(STMT_END_F) == RLE_NO_FLAGS) - { - /* - ------------ Temporary fix until WL#2975 is implemented --------- - - This event is not the last one (no STMT_END_F). If we stop now - (in case of terminate_slave_thread()), how will we restart? We - have to restart from Table_map_log_event, but as this table is - not transactional, the rows already inserted will still be - present, and idempotency is not guaranteed (no PK) so we risk - that repeating leads to double insert. So we desperately try to - continue, hope we'll eventually leave this buggy situation (by - executing the final Rows_log_event). If we are in a hopeless - wait (reached end of last relay log and nothing gets appended - there), we timeout after one minute, and notify DBA about the - problem. When WL#2975 is implemented, just remove the member - Relay_log_info::last_event_start_time and all its occurrences. - */ - const_cast(rli)->last_event_start_time= my_time(0); - } DBUG_RETURN(error); } diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc index 68cd2bf4673..a335210ce53 100644 --- a/sql/log_event_old.cc +++ b/sql/log_event_old.cc @@ -226,7 +226,7 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info row_start= row_end; } - DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event", + DBUG_EXECUTE_IF("stop_slave_middle_group", const_cast(rli)->abort_slave= 1;); error= do_after_row_operations(table, error); if (!ev->cache_stmt) @@ -269,34 +269,6 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info DBUG_RETURN(error); } - /* - This code would ideally be placed in do_update_pos() instead, but - since we have no access to table there, we do the setting of - last_event_start_time here instead. - */ - if (table && (table->s->primary_key == MAX_KEY) && - !ev->cache_stmt && - ev->get_flags(Old_rows_log_event::STMT_END_F) == Old_rows_log_event::RLE_NO_FLAGS) - { - /* - ------------ Temporary fix until WL#2975 is implemented --------- - - This event is not the last one (no STMT_END_F). If we stop now - (in case of terminate_slave_thread()), how will we restart? We - have to restart from Table_map_log_event, but as this table is - not transactional, the rows already inserted will still be - present, and idempotency is not guaranteed (no PK) so we risk - that repeating leads to double insert. So we desperately try to - continue, hope we'll eventually leave this buggy situation (by - executing the final Old_rows_log_event). If we are in a hopeless - wait (reached end of last relay log and nothing gets appended - there), we timeout after one minute, and notify DBA about the - problem. When WL#2975 is implemented, just remove the member - st_relay_log_info::last_event_start_time and all its occurences. - */ - const_cast(rli)->last_event_start_time= my_time(0); - } - DBUG_RETURN(0); } #endif @@ -1744,7 +1716,7 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli) } // row processing loop - DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event", + DBUG_EXECUTE_IF("stop_slave_middle_group", const_cast(rli)->abort_slave= 1;); error= do_after_row_operations(rli, error); if (!cache_stmt) @@ -1787,33 +1759,6 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli) DBUG_RETURN(error); } - /* - This code would ideally be placed in do_update_pos() instead, but - since we have no access to table there, we do the setting of - last_event_start_time here instead. - */ - if (table && (table->s->primary_key == MAX_KEY) && - !cache_stmt && get_flags(STMT_END_F) == RLE_NO_FLAGS) - { - /* - ------------ Temporary fix until WL#2975 is implemented --------- - - This event is not the last one (no STMT_END_F). If we stop now - (in case of terminate_slave_thread()), how will we restart? We - have to restart from Table_map_log_event, but as this table is - not transactional, the rows already inserted will still be - present, and idempotency is not guaranteed (no PK) so we risk - that repeating leads to double insert. So we desperately try to - continue, hope we'll eventually leave this buggy situation (by - executing the final Old_rows_log_event). If we are in a hopeless - wait (reached end of last relay log and nothing gets appended - there), we timeout after one minute, and notify DBA about the - problem. When WL#2975 is implemented, just remove the member - Relay_log_info::last_event_start_time and all its occurrences. - */ - const_cast(rli)->last_event_start_time= my_time(0); - } - DBUG_RETURN(0); } diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index b3a1bbc31d2..ec3adc79ca2 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -1195,7 +1195,6 @@ void Relay_log_info::cleanup_context(THD *thd, bool error) */ thd->options&= ~OPTION_NO_FOREIGN_KEY_CHECKS; thd->options&= ~OPTION_RELAXED_UNIQUE_CHECKS; - last_event_start_time= 0; DBUG_VOID_RETURN; } diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index a5410dd0c79..fd36d18adae 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -349,12 +349,10 @@ public: void clear_tables_to_lock(); /* - Used by row-based replication to detect that it should not stop at - this event, but give it a chance to send more events. The time - where the last event inside a group started is stored here. If the - variable is zero, we are not in a group (but may be in a - transaction). - */ + Used to defer stopping the SQL thread to give it a chance + to finish up the current group of events. + The timestamp is set and reset in @c sql_slave_killed(). + */ time_t last_event_start_time; /** diff --git a/sql/slave.cc b/sql/slave.cc index 44993601e78..834ed8ed933 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -49,6 +49,10 @@ #define FLAGSTR(V,F) ((V)&(F)?#F" ":"") #define MAX_SLAVE_RETRY_PAUSE 5 +/* + a parameter of sql_slave_killed() to defer the killed status +*/ +#define SLAVE_WAIT_GROUP_DONE 60 bool use_slave_mask = 0; MY_BITMAP slave_error_mask; char slave_skip_error_names[SHOW_VAR_FUNC_BUFF_SIZE]; @@ -790,46 +794,95 @@ static bool io_slave_killed(THD* thd, Master_info* mi) DBUG_RETURN(mi->abort_slave || abort_loop || thd->killed); } +/** + The function analyzes a possible killed status and makes + a decision whether to accept it or not. + Normally upon accepting the sql thread goes to shutdown. + In the event of deffering decision @rli->last_event_start_time waiting + timer is set to force the killed status be accepted upon its expiration. + @param thd pointer to a THD instance + @param rli pointer to Relay_log_info instance + + @return TRUE the killed status is recognized, FALSE a possible killed + status is deferred. +*/ static bool sql_slave_killed(THD* thd, Relay_log_info* rli) { + bool ret= FALSE; DBUG_ENTER("sql_slave_killed"); DBUG_ASSERT(rli->sql_thd == thd); DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun if (abort_loop || thd->killed || rli->abort_slave) { - if (rli->abort_slave && rli->is_in_group() && - thd->transaction.all.modified_non_trans_table) - DBUG_RETURN(0); - /* - If we are in an unsafe situation (stopping could corrupt replication), - we give one minute to the slave SQL thread of grace before really - terminating, in the hope that it will be able to read more events and - the unsafe situation will soon be left. Note that this one minute starts - from the last time anything happened in the slave SQL thread. So it's - really one minute of idleness, we don't timeout if the slave SQL thread - is actively working. - */ - if (rli->last_event_start_time == 0) - DBUG_RETURN(1); - DBUG_PRINT("info", ("Slave SQL thread is in an unsafe situation, giving " - "it some grace period")); - if (difftime(time(0), rli->last_event_start_time) > 60) + if (thd->transaction.all.modified_non_trans_table && rli->is_in_group()) { - rli->report(ERROR_LEVEL, 0, - "SQL thread had to stop in an unsafe situation, in " - "the middle of applying updates to a " - "non-transactional table without any primary key. " - "There is a risk of duplicate updates when the slave " - "SQL thread is restarted. Please check your tables' " - "contents after restart."); - DBUG_RETURN(1); + char msg_stopped[]= + "... The slave SQL is stopped, leaving the current group " + "of events unfinished with a non-transaction table changed. " + "If the group consists solely of Row-based events, you can try " + "restarting the slave with --slave-exec-mode=IDEMPOTENT, which " + "ignores duplicate key, key not found, and similar errors (see " + "documentation for details)."; + + if (rli->abort_slave) + { + DBUG_PRINT("info", ("Slave SQL thread is being stopped in the middle of" + " a group having updated a non-trans table, giving" + " it some grace period")); + + /* + Slave sql thread shutdown in face of unfinished group modified + Non-trans table is handled via a timer. The slave may eventually + give out to complete the current group and in that case there + might be issues at consequent slave restart, see the error message. + WL#2975 offers a robust solution requiring to store the last exectuted + event's coordinates along with the group's coordianates + instead of waiting with @c last_event_start_time the timer. + */ + + if (rli->last_event_start_time == 0) + rli->last_event_start_time= my_time(0); + ret= difftime(my_time(0), rli->last_event_start_time) <= + SLAVE_WAIT_GROUP_DONE ? FALSE : TRUE; + + DBUG_EXECUTE_IF("stop_slave_middle_group", + DBUG_EXECUTE_IF("incomplete_group_in_relay_log", + ret= TRUE;);); // time is over + + if (ret == 0) + { + rli->report(WARNING_LEVEL, 0, + "slave SQL thread is being stopped in the middle " + "of applying of a group having updated a non-transaction " + "table; waiting for the group completion ... "); + } + else + { + rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, + ER(ER_SLAVE_FATAL_ERROR), msg_stopped); + } + } + else + { + ret= TRUE; + rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, ER(ER_SLAVE_FATAL_ERROR), + msg_stopped); + } + } + else + { + ret= TRUE; } } - DBUG_RETURN(0); + if (ret) + rli->last_event_start_time= 0; + + DBUG_RETURN(ret); } + /* skip_load_data_infile() @@ -2525,6 +2578,27 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli) delete ev; DBUG_RETURN(1); } + + { /** + The following failure injecion works in cooperation with tests + setting @@global.debug= 'd,incomplete_group_in_relay_log'. + Xid or Commit events are not executed to force the slave sql + read hanging if the realy log does not have any more events. + */ + DBUG_EXECUTE_IF("incomplete_group_in_relay_log", + if ((ev->get_type_code() == XID_EVENT) || + ((ev->get_type_code() == QUERY_EVENT) && + strcmp("COMMIT", ((Query_log_event *) ev)->query) == 0)) + { + DBUG_ASSERT(thd->transaction.all.modified_non_trans_table); + rli->abort_slave= 1; + pthread_mutex_unlock(&rli->data_lock); + delete ev; + rli->inc_event_relay_log_pos(); + DBUG_RETURN(0); + };); + } + exec_res= apply_event_and_update_pos(ev, thd, rli, TRUE); /* @@ -3953,7 +4027,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) goto skip_relay_logging; } break; - + default: inc_pos= event_len; break; -- cgit v1.2.1 From 6ae50d8aedbded8dcdc7504d116f25bcd02535ff Mon Sep 17 00:00:00 2001 From: He Zhenxing Date: Wed, 14 Oct 2009 13:24:47 +0800 Subject: Postfix after merge semi-sync with heartbeat Use ev_offset instead of 1 as the packet header offset when getting log position from events for heartbeat call reset_transmit_packet before calling send_heartbeat_event sql/sql_repl.cc: Use ev_offset instead of 1 as the packet header offset when getting log position from events for heartbeat call reset_transmit_packet before calling send_heartbeat_event --- sql/sql_repl.cc | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) (limited to 'sql') diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 20ac7b01eda..9e4ef364408 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -427,7 +427,6 @@ static int send_heartbeat_event(NET* net, String* packet, { DBUG_RETURN(-1); } - packet->set("\0", 1, &my_charset_bin); DBUG_RETURN(0); } @@ -689,7 +688,7 @@ impossible position"; log's filename does not change while it's active */ if (coord) - coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET); + coord->pos= uint4korr(packet->ptr() + ev_offset + LOG_POS_OFFSET); event_type= (Log_event_type)((*packet)[LOG_EVENT_OFFSET+ev_offset]); if (event_type == FORMAT_DESCRIPTION_EVENT) @@ -849,6 +848,9 @@ impossible position"; sql_print_information("the rest of heartbeat info skipped ..."); } #endif + /* reset transmit packet for the heartbeat event */ + if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg)) + goto err; if (send_heartbeat_event(net, packet, coord)) { errmsg = "Failed on my_net_write()"; -- cgit v1.2.1