diff --git a/extra/yassl/src/ssl.cpp b/extra/yassl/src/ssl.cpp index 7069140..982d462 100644 --- a/extra/yassl/src/ssl.cpp +++ b/extra/yassl/src/ssl.cpp @@ -774,7 +774,6 @@ int SSL_CTX_load_verify_locations(SSL_CTX* ctx, const char* file, const char* path) { int ret = SSL_FAILURE; - const int HALF_PATH = 128; if (file) ret = read_file(ctx, file, SSL_FILETYPE_PEM, CA); diff --git a/mysys/mf_iocache.c b/mysys/mf_iocache.c index f891a22..8687c2e 100644 --- a/mysys/mf_iocache.c +++ b/mysys/mf_iocache.c @@ -1815,6 +1815,7 @@ int my_b_flush_io_cache(IO_CACHE *info, It's currently safe to call this if one has called init_io_cache() on the 'info' object, even if init_io_cache() failed. This function is also safe to call twice with the same handle. + Note that info->file is not reset as the caller may still use ut for my_close() RETURN 0 ok @@ -1850,10 +1851,12 @@ int end_io_cache(IO_CACHE *info) if (info->type == SEQ_READ_APPEND) { /* Destroy allocated mutex */ - info->type= TYPE_NOT_SET; mysql_mutex_destroy(&info->append_buffer_lock); } info->share= 0; + info->type= TYPE_NOT_SET; /* Ensure that flush_io_cache() does nothing */ + info->write_end= 0; /* Ensure that my_b_write() fails */ + info->write_function= 0; /* my_b_write will crash if used */ DBUG_RETURN(error); } /* end_io_cache */ diff --git a/sql/item_func.cc b/sql/item_func.cc index 3149ebb..2c43181 100644 --- a/sql/item_func.cc +++ b/sql/item_func.cc @@ -3958,12 +3958,7 @@ longlong Item_master_pos_wait::val_int() else connection_name= thd->variables.default_master_connection; - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index) // master_info_index is set to NULL on shutdown. - mi= master_info_index->get_master_info(&connection_name, - Sql_condition::WARN_LEVEL_WARN); - mysql_mutex_unlock(&LOCK_active_mi); - if (!mi) + if (!(mi= get_master_info(&connection_name, Sql_condition::WARN_LEVEL_WARN))) goto err; if ((event_count = mi->rli.wait_for_pos(thd, log_name, pos, timeout)) == -2) @@ -3971,6 +3966,7 @@ longlong Item_master_pos_wait::val_int() null_value = 1; event_count=0; } + mi->release(); #endif return event_count; diff --git a/sql/log.cc b/sql/log.cc index cd21582..e99eefe 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -2702,16 +2702,16 @@ int MYSQL_LOG::generate_new_name(char *new_name, const char *log_name) void MYSQL_QUERY_LOG::reopen_file() { char *save_name; - DBUG_ENTER("MYSQL_LOG::reopen_file"); + + mysql_mutex_lock(&LOCK_log); if (!is_open()) { DBUG_PRINT("info",("log is closed")); + mysql_mutex_unlock(&LOCK_log); DBUG_VOID_RETURN; } - mysql_mutex_lock(&LOCK_log); - save_name= name; name= 0; // Don't free name close(LOG_CLOSE_TO_BE_OPENED); @@ -2870,13 +2870,6 @@ bool MYSQL_QUERY_LOG::write(THD *thd, time_t current_time, DBUG_ENTER("MYSQL_QUERY_LOG::write"); mysql_mutex_lock(&LOCK_log); - - if (!is_open()) - { - mysql_mutex_unlock(&LOCK_log); - DBUG_RETURN(0); - } - if (is_open()) { // Safety agains reopen int tmp_errno= 0; @@ -3100,7 +3093,9 @@ void MYSQL_BIN_LOG::cleanup() } inited= 0; + mysql_mutex_lock(&LOCK_log); close(LOG_CLOSE_INDEX|LOG_CLOSE_STOP_EVENT); + mysql_mutex_unlock(&LOCK_log); delete description_event_for_queue; delete description_event_for_exec; @@ -3257,10 +3252,11 @@ bool MYSQL_BIN_LOG::open(const char *log_name, { File file= -1; xid_count_per_binlog *new_xid_list_entry= NULL, *b; - DBUG_ENTER("MYSQL_BIN_LOG::open"); DBUG_PRINT("enter",("log_type: %d",(int) log_type_arg)); + mysql_mutex_assert_owner(&LOCK_log); + if (!is_relay_log) { if (!binlog_state_recover_done) @@ -4129,7 +4125,7 @@ bool MYSQL_BIN_LOG::reset_logs(THD* thd, bool create_new_log, int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included) { - int error; + int error, errcode; char *to_purge_if_included= NULL; inuse_relaylog *ir; ulonglong log_space_reclaimed= 0; @@ -4200,7 +4196,8 @@ int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included) } /* Store where we are in the new file for the execution thread */ - flush_relay_log_info(rli); + if (flush_relay_log_info(rli)) + error= LOG_INFO_IO; DBUG_EXECUTE_IF("crash_before_purge_logs", DBUG_SUICIDE();); @@ -4216,11 +4213,13 @@ int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included) * Need to update the log pos because purge logs has been called * after fetching initially the log pos at the begining of the method. */ - if((error=find_log_pos(&rli->linfo, rli->event_relay_log_name, 0))) + if ((errcode= find_log_pos(&rli->linfo, rli->event_relay_log_name, 0))) { char buff[22]; + if (!error) + error= errcode; sql_print_error("next log error: %d offset: %s log: %s included: %d", - error, + errcode, llstr(rli->linfo.index_file_offset,buff), rli->group_relay_log_name, included); @@ -4839,21 +4838,21 @@ int MYSQL_BIN_LOG::new_file_impl(bool need_lock) bool delay_close= false; File old_file; LINT_INIT(old_file); - DBUG_ENTER("MYSQL_BIN_LOG::new_file_impl"); + + if (need_lock) + mysql_mutex_lock(&LOCK_log); + mysql_mutex_assert_owner(&LOCK_log); + if (!is_open()) { DBUG_PRINT("info",("log is closed")); + mysql_mutex_unlock(&LOCK_log); DBUG_RETURN(error); } - if (need_lock) - mysql_mutex_lock(&LOCK_log); mysql_mutex_lock(&LOCK_index); - mysql_mutex_assert_owner(&LOCK_log); - mysql_mutex_assert_owner(&LOCK_index); - /* Reuse old name if not binlog and not update log */ new_name_ptr= name; @@ -4986,9 +4985,9 @@ int MYSQL_BIN_LOG::new_file_impl(bool need_lock) new_name_ptr, errno); } + mysql_mutex_unlock(&LOCK_index); if (need_lock) mysql_mutex_unlock(&LOCK_log); - mysql_mutex_unlock(&LOCK_index); DBUG_RETURN(error); } @@ -7772,9 +7771,11 @@ int MYSQL_BIN_LOG::wait_for_update_bin_log(THD* thd, void MYSQL_BIN_LOG::close(uint exiting) { // One can't set log_type here! bool failed_to_save_state= false; - DBUG_ENTER("MYSQL_BIN_LOG::close"); DBUG_PRINT("enter",("exiting: %d", (int) exiting)); + + mysql_mutex_assert_owner(&LOCK_log); + if (log_state == LOG_OPENED) { #ifdef HAVE_REPLICATION diff --git a/sql/log_event.cc b/sql/log_event.cc index 14f6bb2..43e8df7 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -6244,9 +6244,11 @@ bool Rotate_log_event::write(IO_CACHE* file) @retval 0 ok + 1 error */ int Rotate_log_event::do_update_pos(rpl_group_info *rgi) { + int error= 0; Relay_log_info *rli= rgi->rli; DBUG_ENTER("Rotate_log_event::do_update_pos"); #ifndef DBUG_OFF @@ -6298,7 +6300,7 @@ int Rotate_log_event::do_update_pos(rpl_group_info *rgi) (ulong) rli->group_master_log_pos)); mysql_mutex_unlock(&rli->data_lock); rpl_global_gtid_slave_state->record_and_update_gtid(thd, rgi); - flush_relay_log_info(rli); + error= flush_relay_log_info(rli); /* Reset thd->variables.option_bits and sql_mode etc, because this could @@ -6316,8 +6318,7 @@ int Rotate_log_event::do_update_pos(rpl_group_info *rgi) else rgi->inc_event_relay_log_pos(); - - DBUG_RETURN(0); + DBUG_RETURN(error); } @@ -8174,6 +8175,7 @@ void Stop_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) int Stop_log_event::do_update_pos(rpl_group_info *rgi) { + int error= 0; Relay_log_info *rli= rgi->rli; DBUG_ENTER("Stop_log_event::do_update_pos"); /* @@ -8189,9 +8191,10 @@ int Stop_log_event::do_update_pos(rpl_group_info *rgi) { rpl_global_gtid_slave_state->record_and_update_gtid(thd, rgi); rli->inc_group_relay_log_pos(0, rgi); - flush_relay_log_info(rli); + if (flush_relay_log_info(rli)) + error= 1; } - DBUG_RETURN(0); + DBUG_RETURN(error); } #endif /* !MYSQL_CLIENT */ @@ -10178,8 +10181,8 @@ static int rows_event_stmt_cleanup(rpl_group_info *rgi, THD * thd) Rows_log_event::do_update_pos(rpl_group_info *rgi) { Relay_log_info *rli= rgi->rli; - DBUG_ENTER("Rows_log_event::do_update_pos"); int error= 0; + DBUG_ENTER("Rows_log_event::do_update_pos"); DBUG_PRINT("info", ("flags: %s", get_flags(STMT_END_F) ? "STMT_END_F " : "")); @@ -10191,7 +10194,7 @@ static int rows_event_stmt_cleanup(rpl_group_info *rgi, THD * thd) Step the group log position if we are not in a transaction, otherwise increase the event log position. */ - rli->stmt_done(log_pos, thd, rgi); + error= rli->stmt_done(log_pos, thd, rgi); /* Clear any errors in thd->net.last_err*. It is not known if this is needed or not. It is believed that any errors that may exist in diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc index 22ef970..8bdb81f 100644 --- a/sql/log_event_old.cc +++ b/sql/log_event_old.cc @@ -1843,8 +1843,8 @@ int Old_rows_log_event::do_apply_event(rpl_group_info *rgi) Old_rows_log_event::do_update_pos(rpl_group_info *rgi) { Relay_log_info *rli= rgi->rli; - DBUG_ENTER("Old_rows_log_event::do_update_pos"); int error= 0; + DBUG_ENTER("Old_rows_log_event::do_update_pos"); DBUG_PRINT("info", ("flags: %s", get_flags(STMT_END_F) ? "STMT_END_F " : "")); @@ -1856,7 +1856,7 @@ int Old_rows_log_event::do_apply_event(rpl_group_info *rgi) Step the group log position if we are not in a transaction, otherwise increase the event log position. */ - rli->stmt_done(log_pos, thd, rgi); + error= rli->stmt_done(log_pos, thd, rgi); /* Clear any errors in thd->net.last_err*. It is not known if this is needed or not. It is believed that any errors that may exist in diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 6e40168..9eeb577 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -702,12 +702,15 @@ THD *next_global_thread(THD *thd) LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_global_system_variables, - LOCK_user_conn, LOCK_slave_list, LOCK_active_mi, + LOCK_user_conn, LOCK_slave_list, LOCK_connection_count, LOCK_error_messages, LOCK_slave_init; mysql_mutex_t LOCK_stats, LOCK_global_user_client_stats, LOCK_global_table_stats, LOCK_global_index_stats; +/* This protects against changes in master_info_index */ +mysql_mutex_t LOCK_active_mi; + /** The below lock protects access to two global server variables: max_prepared_stmt_count and prepared_stmt_count. These variables @@ -860,7 +863,7 @@ THD *next_global_thread(THD *thd) key_LOCK_system_variables_hash, key_LOCK_thd_data, key_LOCK_user_conn, key_LOCK_uuid_short_generator, key_LOG_LOCK_log, key_master_info_data_lock, key_master_info_run_lock, - key_master_info_sleep_lock, + key_master_info_sleep_lock, key_master_info_start_stop_lock, key_mutex_slave_reporting_capability_err_lock, key_relay_log_info_data_lock, key_rpl_group_info_sleep_lock, key_relay_log_info_log_space_lock, key_relay_log_info_run_lock, @@ -930,6 +933,7 @@ THD *next_global_thread(THD *thd) { &key_LOCK_uuid_short_generator, "LOCK_uuid_short_generator", PSI_FLAG_GLOBAL}, { &key_LOG_LOCK_log, "LOG::LOCK_log", 0}, { &key_master_info_data_lock, "Master_info::data_lock", 0}, + { &key_master_info_start_stop_lock, "Master_info::start_stop_lock", 0}, { &key_master_info_run_lock, "Master_info::run_lock", 0}, { &key_master_info_sleep_lock, "Master_info::sleep_lock", 0}, { &key_mutex_slave_reporting_capability_err_lock, "Slave_reporting_capability::err_lock", 0}, @@ -1651,7 +1655,7 @@ static void close_connections(void) mysql_mutex_unlock(&LOCK_thread_count); // For unlink from list Events::deinit(); - end_slave(); + slave_prepare_for_shutdown(); /* Give threads time to die. @@ -1700,6 +1704,7 @@ static void close_connections(void) DBUG_PRINT("quit",("Unlocking LOCK_thread_count")); mysql_mutex_unlock(&LOCK_thread_count); } + end_slave(); /* All threads has now been aborted */ DBUG_PRINT("quit",("Waiting for threads to die (count=%u)",thread_count)); mysql_mutex_lock(&LOCK_thread_count); @@ -4923,9 +4928,17 @@ static int init_server_components() unireg_abort(1); } - if (opt_bin_log && mysql_bin_log.open(opt_bin_logname, LOG_BIN, 0, - WRITE_CACHE, max_binlog_size, 0, TRUE)) - unireg_abort(1); + if (opt_bin_log) + { + int error; + mysql_mutex_t *log_lock= mysql_bin_log.get_log_lock(); + mysql_mutex_lock(log_lock); + error= mysql_bin_log.open(opt_bin_logname, LOG_BIN, 0, + WRITE_CACHE, max_binlog_size, 0, TRUE); + mysql_mutex_unlock(log_lock); + if (error) + unireg_abort(1); + } #ifdef HAVE_REPLICATION if (opt_bin_log && expire_logs_days) @@ -7214,17 +7227,14 @@ static int show_slave_running(THD *thd, SHOW_VAR *var, char *buff) var->type= SHOW_MY_BOOL; var->value= buff; - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index) + + if ((mi= get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_NOTE))) { - mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_NOTE); - if (mi) - tmp= (my_bool) (mi->slave_running == MYSQL_SLAVE_RUN_READING && - mi->rli.slave_running != MYSQL_SLAVE_NOT_RUN); + tmp= (my_bool) (mi->slave_running == MYSQL_SLAVE_RUN_READING && + mi->rli.slave_running != MYSQL_SLAVE_NOT_RUN); + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); if (mi) *((my_bool *)buff)= tmp; else @@ -7256,38 +7266,26 @@ static int show_slaves_running(THD *thd, SHOW_VAR *var, char *buff) { var->type= SHOW_LONGLONG; var->value= buff; - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index) - *((longlong *)buff)= master_info_index->any_slave_sql_running(); - else - *((longlong *)buff)= 0; + *((longlong *)buff)= any_slave_sql_running(); - mysql_mutex_unlock(&LOCK_active_mi); return 0; } static int show_slave_received_heartbeats(THD *thd, SHOW_VAR *var, char *buff) { - Master_info *mi= NULL; - longlong tmp; - LINT_INIT(tmp); + Master_info *mi; var->type= SHOW_LONGLONG; var->value= buff; - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index) + + if ((mi= get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_NOTE))) { - mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_NOTE); - if (mi) - tmp= mi->received_heartbeats; + *((longlong *)buff)= mi->received_heartbeats; + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); - if (mi) - *((longlong *)buff)= tmp; else var->type= SHOW_UNDEF; return 0; @@ -7297,23 +7295,16 @@ static int show_slave_received_heartbeats(THD *thd, SHOW_VAR *var, char *buff) static int show_heartbeat_period(THD *thd, SHOW_VAR *var, char *buff) { Master_info *mi= NULL; - float tmp; - LINT_INIT(tmp); var->type= SHOW_CHAR; var->value= buff; - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index) + + if ((mi= get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_NOTE))) { - mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_NOTE); - if (mi) - tmp= mi->heartbeat_period; + sprintf(buff, "%.3f", mi->heartbeat_period); + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); - if (mi) - sprintf(buff, "%.3f", tmp); else var->type= SHOW_UNDEF; return 0; diff --git a/sql/mysqld.h b/sql/mysqld.h index 732f00c..a761ade 100644 --- a/sql/mysqld.h +++ b/sql/mysqld.h @@ -265,7 +265,7 @@ key_LOCK_thd_data, key_LOCK_user_conn, key_LOG_LOCK_log, key_master_info_data_lock, key_master_info_run_lock, - key_master_info_sleep_lock, + key_master_info_sleep_lock, key_master_info_start_stop_lock, key_mutex_slave_reporting_capability_err_lock, key_relay_log_info_data_lock, key_relay_log_info_log_space_lock, key_relay_log_info_run_lock, key_rpl_group_info_sleep_lock, diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc index 249bf76..cff3db6 100644 --- a/sql/rpl_mi.cc +++ b/sql/rpl_mi.cc @@ -40,7 +40,9 @@ sync_counter(0), heartbeat_period(0), received_heartbeats(0), master_id(0), prev_master_id(0), using_gtid(USE_GTID_NO), events_queued_since_last_gtid(0), - gtid_reconnect_event_skip_count(0), gtid_event_seen(false) + gtid_reconnect_event_skip_count(0), gtid_event_seen(false), + in_start_all_slaves(0), in_stop_all_slaves(0), + users(0), killed(0) { host[0] = 0; user[0] = 0; password[0] = 0; ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0; @@ -78,6 +80,8 @@ bzero((char*) &file, sizeof(file)); mysql_mutex_init(key_master_info_run_lock, &run_lock, MY_MUTEX_INIT_FAST); mysql_mutex_init(key_master_info_data_lock, &data_lock, MY_MUTEX_INIT_FAST); + mysql_mutex_init(key_master_info_start_stop_lock, &start_stop_lock, + MY_MUTEX_INIT_SLOW); mysql_mutex_setflags(&run_lock, MYF_NO_DEADLOCK_DETECTION); mysql_mutex_setflags(&data_lock, MYF_NO_DEADLOCK_DETECTION); mysql_mutex_init(key_master_info_sleep_lock, &sleep_lock, MY_MUTEX_INIT_FAST); @@ -87,8 +91,27 @@ mysql_cond_init(key_master_info_sleep_cond, &sleep_cond, NULL); } + +/** + Wait until no one is using Master_info +*/ + +void Master_info::wait_until_free() +{ + mysql_mutex_lock(&sleep_lock); + killed= 1; + while (users) + mysql_cond_wait(&sleep_cond, &sleep_lock); + mysql_mutex_unlock(&sleep_lock); +} + +/** + Delete master_info +*/ + Master_info::~Master_info() { + wait_until_free(); rpl_filters.delete_element(connection_name.str, connection_name.length, (void (*)(const char*, uchar*)) free_rpl_filter); my_free(connection_name.str); @@ -96,6 +119,7 @@ mysql_mutex_destroy(&run_lock); mysql_mutex_destroy(&data_lock); mysql_mutex_destroy(&sleep_lock); + mysql_mutex_destroy(&start_stop_lock); mysql_cond_destroy(&data_cond); mysql_cond_destroy(&start_cond); mysql_cond_destroy(&stop_cond); @@ -709,8 +733,13 @@ uchar *get_key_master_info(Master_info *mi, size_t *length, void free_key_master_info(Master_info *mi) { DBUG_ENTER("free_key_master_info"); + /* Ensure that we are not in reset_slave while this is done */ + mi->lock_slave_threads(); terminate_slave_threads(mi,SLAVE_FORCE_ALL); + /* We use 2 here instead of 1 just to make it easier when debugging */ + mi->killed= 2; end_master_info(mi); + mi->unlock_slave_threads(); delete mi; DBUG_VOID_RETURN; } @@ -867,9 +896,27 @@ void copy_filter_setting(Rpl_filter* dst_filter, Rpl_filter* src_filter) index_file.file= -1; } + +/** + Free all connection threads + + This is done during early stages of shutdown + to give connection threads and slave threads time + to die before ~Master_info_index is called +*/ + +void Master_info_index::free_connections() +{ + my_hash_reset(&master_info_hash); +} + + +/** + Free all connection threads and free structures +*/ + Master_info_index::~Master_info_index() { - /* This will close connection for all objects in the cache */ my_hash_free(&master_info_hash); end_io_cache(&index_file); if (index_file.file >= 0) @@ -892,7 +939,6 @@ bool Master_info_index::init_all_master_info() File index_file_nr; DBUG_ENTER("init_all_master_info"); - mysql_mutex_assert_owner(&LOCK_active_mi); DBUG_ASSERT(master_info_index); if ((index_file_nr= my_open(index_file_name, @@ -940,7 +986,7 @@ bool Master_info_index::init_all_master_info() DBUG_RETURN(1); } - lock_slave_threads(mi); + mi->lock_slave_threads(); init_thread_mask(&thread_mask,mi,0 /*not inverse*/); create_logfile_name_with_suffix(buf_master_info_file, @@ -968,14 +1014,14 @@ bool Master_info_index::init_all_master_info() if (master_info_index->add_master_info(mi, FALSE)) DBUG_RETURN(1); succ_num++; - unlock_slave_threads(mi); + mi->unlock_slave_threads(); } else { /* Master_info already in HASH */ sql_print_error(ER(ER_CONNECTION_ALREADY_EXISTS), (int) connection_name.length, connection_name.str); - unlock_slave_threads(mi); + mi->unlock_slave_threads(); delete mi; } continue; @@ -991,8 +1037,9 @@ bool Master_info_index::init_all_master_info() { /* Master_info was already registered */ sql_print_error(ER(ER_CONNECTION_ALREADY_EXISTS), + (int) connection_name.length, connection_name.str, (int) connection_name.length, connection_name.str); - unlock_slave_threads(mi); + mi->unlock_slave_threads(); delete mi; continue; } @@ -1001,7 +1048,6 @@ bool Master_info_index::init_all_master_info() if (master_info_index->add_master_info(mi, FALSE)) DBUG_RETURN(1); succ_num++; - unlock_slave_threads(mi); if (!opt_skip_slave_start) { @@ -1022,6 +1068,7 @@ bool Master_info_index::init_all_master_info() (int) connection_name.length, connection_name.str); } + mi->unlock_slave_threads(); } } @@ -1073,6 +1120,71 @@ bool Master_info_index::write_master_name_to_index_file(LEX_STRING *name, /** + Get Master_info for a connection and lock the object from deletion + + @param + connection_name Connection name + warning WARN_LEVEL_NOTE -> Don't print anything + WARN_LEVEL_WARN -> Issue warning if not exists + WARN_LEVEL_ERROR-> Issue error if not exists +*/ + +Master_info *get_master_info(LEX_STRING *connection_name, + Sql_condition::enum_warning_level warning) +{ + Master_info *mi; + DBUG_ENTER("get_master_info"); + + /* Protect against inserts into hash */ + mysql_mutex_lock(&LOCK_active_mi); + /* + The following can only be true during shutdown when slave has been killed + but some other threads are still trying to access slave statistics. + */ + if (unlikely(!master_info_index)) + { + if (warning != Sql_condition::WARN_LEVEL_NOTE) + my_error(WARN_NO_MASTER_INFO, + MYF(warning == Sql_condition::WARN_LEVEL_WARN ? + ME_JUST_WARNING : 0), + (int) connection_name->length, connection_name->str); + mysql_mutex_unlock(&LOCK_active_mi); + DBUG_RETURN(0); + } + if ((mi= master_info_index->get_master_info(connection_name, warning))) + { + /* + We have to use sleep_lock here. If we would use LOCK_active_mi + then we would take locks in wrong order in Master_info::release() + */ + mysql_mutex_lock(&mi->sleep_lock); + mi->users++; + DBUG_PRINT("info",("users: %d", mi->users)); + mysql_mutex_unlock(&mi->sleep_lock); + } + mysql_mutex_unlock(&LOCK_active_mi); + DBUG_RETURN(mi); +} + + +/** + Release master info. + Signals ~Master_info that it's now safe to delete it +*/ + +void Master_info::release() +{ + mysql_mutex_lock(&sleep_lock); + if (!--users && killed) + { + /* Signal ~Master_info that it's ok to now free it */ + mysql_cond_signal(&sleep_cond); + } + mysql_mutex_unlock(&sleep_lock); +} + + +/** Get Master_info for a connection @param @@ -1094,8 +1206,6 @@ bool Master_info_index::write_master_name_to_index_file(LEX_STRING *name, ("connection_name: '%.*s'", (int) connection_name->length, connection_name->str)); - mysql_mutex_assert_owner(&LOCK_active_mi); - /* Make name lower case for comparison */ res= strmake(buff, connection_name->str, connection_name->length); my_casedn_str(system_charset_info, buff); @@ -1187,105 +1297,123 @@ bool Master_info_index::add_master_info(Master_info *mi, bool write_to_file) atomic */ -bool Master_info_index::remove_master_info(LEX_STRING *name) +bool Master_info_index::remove_master_info(Master_info *mi) { - Master_info* mi; DBUG_ENTER("remove_master_info"); + mysql_mutex_assert_owner(&LOCK_active_mi); - if ((mi= get_master_info(name, Sql_condition::WARN_LEVEL_WARN))) + // Delete Master_info and rewrite others to file + if (!my_hash_delete(&master_info_hash, (uchar*) mi)) { - // Delete Master_info and rewrite others to file - if (!my_hash_delete(&master_info_hash, (uchar*) mi)) + File index_file_nr; + + // Close IO_CACHE and FILE handler fisrt + end_io_cache(&index_file); + my_close(index_file.file, MYF(MY_WME)); + + // Reopen File and truncate it + if ((index_file_nr= my_open(index_file_name, + O_RDWR | O_CREAT | O_TRUNC | O_BINARY , + MYF(MY_WME))) < 0 || + init_io_cache(&index_file, index_file_nr, + IO_SIZE, WRITE_CACHE, + my_seek(index_file_nr,0L,MY_SEEK_END,MYF(0)), + 0, MYF(MY_WME | MY_WAIT_IF_FULL))) { - File index_file_nr; - - // Close IO_CACHE and FILE handler fisrt - end_io_cache(&index_file); - my_close(index_file.file, MYF(MY_WME)); - - // Reopen File and truncate it - if ((index_file_nr= my_open(index_file_name, - O_RDWR | O_CREAT | O_TRUNC | O_BINARY , - MYF(MY_WME))) < 0 || - init_io_cache(&index_file, index_file_nr, - IO_SIZE, WRITE_CACHE, - my_seek(index_file_nr,0L,MY_SEEK_END,MYF(0)), - 0, MYF(MY_WME | MY_WAIT_IF_FULL))) - { - int error= my_errno; - if (index_file_nr >= 0) - my_close(index_file_nr,MYF(0)); - - sql_print_error("Create of Master Info Index file '%s' failed with " - "error: %M", - index_file_name, error); - DBUG_RETURN(TRUE); - } + int error= my_errno; + if (index_file_nr >= 0) + my_close(index_file_nr,MYF(0)); - // Rewrite Master_info.index - for (uint i= 0; i< master_info_hash.records; ++i) - { - Master_info *tmp_mi; - tmp_mi= (Master_info *) my_hash_element(&master_info_hash, i); - write_master_name_to_index_file(&tmp_mi->connection_name, 0); - } - my_sync(index_file_nr, MYF(MY_WME)); + sql_print_error("Create of Master Info Index file '%s' failed with " + "error: %M", + index_file_name, error); + DBUG_RETURN(TRUE); + } + + // Rewrite Master_info.index + for (uint i= 0; i< master_info_hash.records; ++i) + { + Master_info *tmp_mi; + tmp_mi= (Master_info *) my_hash_element(&master_info_hash, i); + write_master_name_to_index_file(&tmp_mi->connection_name, 0); } + if (my_sync(index_file_nr, MYF(MY_WME))) + DBUG_RETURN(TRUE); } DBUG_RETURN(FALSE); } /** - Master_info_index::give_error_if_slave_running() + give_error_if_slave_running() + + @param + already_locked 0 if we need to lock, 1 if we have LOCK_active_mi_locked @return TRUE If some slave is running. An error is printed FALSE No slave is running */ -bool Master_info_index::give_error_if_slave_running() +bool give_error_if_slave_running(bool already_locked) { + bool ret= 0; DBUG_ENTER("give_error_if_slave_running"); - mysql_mutex_assert_owner(&LOCK_active_mi); - for (uint i= 0; i< master_info_hash.records; ++i) + if (!already_locked) + mysql_mutex_lock(&LOCK_active_mi); + if (!master_info_index) { - Master_info *mi; - mi= (Master_info *) my_hash_element(&master_info_hash, i); - if (mi->rli.slave_running != MYSQL_SLAVE_NOT_RUN) + my_error(ER_SERVER_SHUTDOWN, MYF(0)); + ret= 1; + } + else + { + HASH *hash= &master_info_index->master_info_hash; + for (uint i= 0; i< hash->records; ++i) { - my_error(ER_SLAVE_MUST_STOP, MYF(0), (int) mi->connection_name.length, - mi->connection_name.str); - DBUG_RETURN(TRUE); + Master_info *mi; + mi= (Master_info *) my_hash_element(hash, i); + if (mi->rli.slave_running != MYSQL_SLAVE_NOT_RUN) + { + my_error(ER_SLAVE_MUST_STOP, MYF(0), (int) mi->connection_name.length, + mi->connection_name.str); + ret= 1; + break; + } } } - DBUG_RETURN(FALSE); + if (!already_locked) + mysql_mutex_unlock(&LOCK_active_mi); + DBUG_RETURN(ret); } /** - Master_info_index::any_slave_sql_running() - - The LOCK_active_mi must be held while calling this function. + any_slave_sql_running() @return 0 No Slave SQL thread is running # Number of slave SQL thread running */ -uint Master_info_index::any_slave_sql_running() +uint any_slave_sql_running() { uint count= 0; DBUG_ENTER("any_slave_sql_running"); - mysql_mutex_assert_owner(&LOCK_active_mi); - for (uint i= 0; i< master_info_hash.records; ++i) + mysql_mutex_lock(&LOCK_active_mi); + if (likely(master_info_index)) // Not shutdown { - Master_info *mi= (Master_info *)my_hash_element(&master_info_hash, i); - if (mi->rli.slave_running != MYSQL_SLAVE_NOT_RUN) - count++; + HASH *hash= &master_info_index->master_info_hash; + for (uint i= 0; i< hash->records; ++i) + { + Master_info *mi= (Master_info *)my_hash_element(hash, i); + if (mi->rli.slave_running != MYSQL_SLAVE_NOT_RUN) + count++; + } } + mysql_mutex_unlock(&LOCK_active_mi); DBUG_RETURN(count); } @@ -1298,15 +1426,25 @@ uint Master_info_index::any_slave_sql_running() @return TRUE Error FALSE Everything ok. + + This code is written so that we don't keep LOCK_active_mi active + while we are starting a slave. */ bool Master_info_index::start_all_slaves(THD *thd) { bool result= FALSE; - DBUG_ENTER("warn_if_slave_running"); + DBUG_ENTER("start_all_slaves"); mysql_mutex_assert_owner(&LOCK_active_mi); - for (uint i= 0; i< master_info_hash.records; ++i) + for (uint i= 0; i< master_info_hash.records; i++) + { + Master_info *mi; + mi= (Master_info *) my_hash_element(&master_info_hash, i); + mi->in_start_all_slaves= 0; + } + + for (uint i= 0; i< master_info_hash.records; ) { int error; Master_info *mi; @@ -1316,25 +1454,40 @@ bool Master_info_index::start_all_slaves(THD *thd) Try to start all slaves that are configured (host is defined) and are not already running */ - if ((mi->slave_running == MYSQL_SLAVE_NOT_RUN || - !mi->rli.slave_running) && *mi->host) + if (!((mi->slave_running == MYSQL_SLAVE_NOT_RUN || + !mi->rli.slave_running) && *mi->host) || + mi->in_start_all_slaves) { - if ((error= start_slave(thd, mi, 1))) - { - my_error(ER_CANT_START_STOP_SLAVE, MYF(0), - "START", - (int) mi->connection_name.length, - mi->connection_name.str); - result= 1; - if (error < 0) // fatal error - break; - } - else - push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE, - ER_SLAVE_STARTED, ER(ER_SLAVE_STARTED), - (int) mi->connection_name.length, - mi->connection_name.str); + i++; + continue; + } + mi->in_start_all_slaves= 1; + + mysql_mutex_lock(&mi->sleep_lock); + mi->users++; // Mark used + mysql_mutex_unlock(&mi->sleep_lock); + mysql_mutex_unlock(&LOCK_active_mi); + error= start_slave(thd, mi, 1); + mi->release(); + mysql_mutex_lock(&LOCK_active_mi); + if (error) + { + my_error(ER_CANT_START_STOP_SLAVE, MYF(0), + "START", + (int) mi->connection_name.length, + mi->connection_name.str); + result= 1; + if (error < 0) // fatal error + break; } + else + push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE, + ER_SLAVE_STARTED, ER(ER_SLAVE_STARTED), + (int) mi->connection_name.length, + mi->connection_name.str); + /* Restart from first element as master_info_hash may have changed */ + i= 0; + continue; } DBUG_RETURN(result); } @@ -1348,38 +1501,63 @@ bool Master_info_index::start_all_slaves(THD *thd) @return TRUE Error FALSE Everything ok. + + This code is written so that we don't keep LOCK_active_mi active + while we are stopping a slave. */ bool Master_info_index::stop_all_slaves(THD *thd) { bool result= FALSE; - DBUG_ENTER("warn_if_slave_running"); + DBUG_ENTER("stop_all_slaves"); mysql_mutex_assert_owner(&LOCK_active_mi); - for (uint i= 0; i< master_info_hash.records; ++i) + for (uint i= 0; i< master_info_hash.records; i++) + { + Master_info *mi; + mi= (Master_info *) my_hash_element(&master_info_hash, i); + mi->in_stop_all_slaves= 0; + } + + for (uint i= 0; i< master_info_hash.records ;) { int error; Master_info *mi; mi= (Master_info *) my_hash_element(&master_info_hash, i); - if ((mi->slave_running != MYSQL_SLAVE_NOT_RUN || - mi->rli.slave_running)) + if (!(mi->slave_running != MYSQL_SLAVE_NOT_RUN || + mi->rli.slave_running) || + mi->in_stop_all_slaves) { - if ((error= stop_slave(thd, mi, 1))) - { - my_error(ER_CANT_START_STOP_SLAVE, MYF(0), - "STOP", - (int) mi->connection_name.length, - mi->connection_name.str); - result= 1; - if (error < 0) // Fatal error - break; - } - else - push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE, - ER_SLAVE_STOPPED, ER(ER_SLAVE_STOPPED), - (int) mi->connection_name.length, - mi->connection_name.str); + i++; + continue; } + mi->in_stop_all_slaves= 1; // Protection for loops + + mysql_mutex_lock(&mi->sleep_lock); + mi->users++; // Mark used + mysql_mutex_unlock(&mi->sleep_lock); + mysql_mutex_unlock(&LOCK_active_mi); + error= stop_slave(thd, mi, 1); + mi->release(); + mysql_mutex_lock(&LOCK_active_mi); + if (error) + { + my_error(ER_CANT_START_STOP_SLAVE, MYF(0), + "STOP", + (int) mi->connection_name.length, + mi->connection_name.str); + result= 1; + if (error < 0) // Fatal error + break; + } + else + push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE, + ER_SLAVE_STOPPED, ER(ER_SLAVE_STOPPED), + (int) mi->connection_name.length, + mi->connection_name.str); + /* Restart from first element as master_info_hash may have changed */ + i= 0; + continue; } DBUG_RETURN(result); } diff --git a/sql/rpl_mi.h b/sql/rpl_mi.h index e58df01..5c42378 100644 --- a/sql/rpl_mi.h +++ b/sql/rpl_mi.h @@ -79,6 +79,10 @@ class Master_info : public Slave_reporting_capability { return opt_slave_parallel_threads > 0; } + void release(); + void wait_until_free(); + void lock_slave_threads(); + void unlock_slave_threads(); /* the variables below are needed because we can change masters on the fly */ char master_log_name[FN_REFLEN+6]; /* Room for multi-*/ @@ -97,7 +101,7 @@ class Master_info : public Slave_reporting_capability File fd; // we keep the file open, so we need to remember the file pointer IO_CACHE file; - mysql_mutex_t data_lock, run_lock, sleep_lock; + mysql_mutex_t data_lock, run_lock, sleep_lock, start_stop_lock; mysql_cond_t data_cond, start_cond, stop_cond, sleep_cond; THD *io_thd; MYSQL* mysql; @@ -182,7 +186,11 @@ class Master_info : public Slave_reporting_capability uint64 gtid_reconnect_event_skip_count; /* gtid_event_seen is false until we receive first GTID event from master. */ bool gtid_event_seen; + bool in_start_all_slaves, in_stop_all_slaves; + uint users; /* Active user for object */ + uint killed; }; + int init_master_info(Master_info* mi, const char* master_info_fname, const char* slave_info_fname, bool abort_if_no_master_info_file, @@ -218,13 +226,12 @@ class Master_info_index bool check_duplicate_master_info(LEX_STRING *connection_name, const char *host, uint port); bool add_master_info(Master_info *mi, bool write_to_file); - bool remove_master_info(LEX_STRING *connection_name); + bool remove_master_info(Master_info *mi); Master_info *get_master_info(LEX_STRING *connection_name, Sql_condition::enum_warning_level warning); - bool give_error_if_slave_running(); - uint any_slave_sql_running(); bool start_all_slaves(THD *thd); bool stop_all_slaves(THD *thd); + void free_connections(); }; @@ -237,6 +244,8 @@ class rpl_io_thread_info }; +Master_info *get_master_info(LEX_STRING *connection_name, + Sql_condition::enum_warning_level warning); bool check_master_connection_name(LEX_STRING *name); void create_logfile_name_with_suffix(char *res_file_name, size_t length, const char *info_file, @@ -246,7 +255,8 @@ void create_logfile_name_with_suffix(char *res_file_name, size_t length, uchar *get_key_master_info(Master_info *mi, size_t *length, my_bool not_used __attribute__((unused))); void free_key_master_info(Master_info *mi); - +uint any_slave_sql_running(); +bool give_error_if_slave_running(bool already_lock); #endif /* HAVE_REPLICATION */ #endif /* RPL_MI_H */ diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc index 23f61a8..0a2d745 100644 --- a/sql/rpl_parallel.cc +++ b/sql/rpl_parallel.cc @@ -1312,11 +1312,23 @@ static void signal_error_to_sql_driver_thread(THD *thd, rpl_group_info *rgi, } if (!in_event_group) { + /* If we are in a FLUSH TABLES FOR READ LOCK, wait for it */ + while (rpt->current_entry && rpt->pause_for_ftwrl) + { + mysql_mutex_lock(&rpt->current_entry->LOCK_parallel_entry); + mysql_mutex_unlock(&rpt->LOCK_rpl_thread); + if (rpt->pause_for_ftwrl) + mysql_cond_wait(&rpt->current_entry->COND_parallel_entry, + &rpt->current_entry->LOCK_parallel_entry); + mysql_mutex_unlock(&rpt->current_entry->LOCK_parallel_entry); + mysql_mutex_lock(&rpt->LOCK_rpl_thread); + } rpt->current_owner= NULL; /* Tell wait_for_done() that we are done, if it is waiting. */ if (likely(rpt->current_entry) && unlikely(rpt->current_entry->force_abort)) mysql_cond_broadcast(&rpt->COND_rpl_thread_stop); + rpt->current_entry= NULL; if (!rpt->stop) rpt->pool->release_thread(rpt); @@ -1370,6 +1382,21 @@ static void signal_error_to_sql_driver_thread(THD *thd, rpl_group_info *rgi, return res; /* + If we are about to delete pool, do an extra check that there are no new + slave threads running since we marked pool busy + */ + if (!new_count) + { + if (any_slave_sql_running()) + { + DBUG_PRINT("warning", + ("SQL threads running while trying to reset parallel pool")); + pool_mark_not_busy(pool); + return 1; + } + } + + /* Allocate the new list of threads up-front. That way, if we fail half-way, we only need to free whatever we managed to allocate, and will not be left with a half-functional thread pool. @@ -1382,7 +1409,7 @@ static void signal_error_to_sql_driver_thread(THD *thd, rpl_group_info *rgi, { my_error(ER_OUTOFMEMORY, MYF(0), (int(new_count*sizeof(*new_list) + new_count*sizeof(*rpt_array)))); - goto err;; + goto err; } for (i= 0; i < new_count; ++i) @@ -1503,6 +1530,20 @@ static void signal_error_to_sql_driver_thread(THD *thd, rpl_group_info *rgi, return 1; } +/* + Deactivate the parallel replication thread pool, if there are now no more + SQL threads running. +*/ + +int rpl_parallel_resize_pool_if_no_slaves(void) +{ + /* master_info_index is set to NULL on shutdown */ + if (opt_slave_parallel_threads > 0 && !any_slave_sql_running() && + master_info_index) + return rpl_parallel_inactivate_pool(&global_rpl_thread_pool); + return 0; +} + int rpl_parallel_activate_pool(rpl_parallel_thread_pool *pool) diff --git a/sql/rpl_parallel.h b/sql/rpl_parallel.h index a02c1af..0d7cd4f 100644 --- a/sql/rpl_parallel.h +++ b/sql/rpl_parallel.h @@ -342,6 +342,7 @@ struct rpl_parallel { extern struct rpl_parallel_thread_pool global_rpl_thread_pool; +extern int rpl_parallel_resize_pool_if_no_slaves(void); extern int rpl_parallel_activate_pool(rpl_parallel_thread_pool *pool); extern int rpl_parallel_inactivate_pool(rpl_parallel_thread_pool *pool); extern bool process_gtid_for_restart_pos(Relay_log_info *rli, rpl_gtid *gtid); diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index c570105..928fbd3 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -208,6 +208,7 @@ int init_relay_log_info(Relay_log_info* rli, Master_info* mi= rli->mi; char buf_relay_logname[FN_REFLEN], buf_relaylog_index_name_buff[FN_REFLEN]; char *buf_relaylog_index_name= opt_relaylog_index_name; + mysql_mutex_t *log_lock; create_logfile_name_with_suffix(buf_relay_logname, sizeof(buf_relay_logname), @@ -227,14 +228,18 @@ int init_relay_log_info(Relay_log_info* rli, note, that if open() fails, we'll still have index file open but a destructor will take care of that */ + log_lock= rli->relay_log.get_log_lock(); + mysql_mutex_lock(log_lock); if (rli->relay_log.open_index_file(buf_relaylog_index_name, ln, TRUE) || rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND, mi->rli.max_relay_log_size, 1, TRUE)) { + mysql_mutex_unlock(log_lock); mysql_mutex_unlock(&rli->data_lock); sql_print_error("Failed when trying to open logs for '%s' in init_relay_log_info(). Error: %M", ln, my_errno); DBUG_RETURN(1); } + mysql_mutex_unlock(log_lock); } /* if file does not exist */ @@ -432,7 +437,7 @@ int init_relay_log_info(Relay_log_info* rli, } rli->inited= 1; mysql_mutex_unlock(&rli->data_lock); - DBUG_RETURN(error); + DBUG_RETURN(0); err: sql_print_error("%s", msg); @@ -1304,9 +1309,10 @@ bool Relay_log_info::is_until_satisfied(THD *thd, Log_event *ev) } -void Relay_log_info::stmt_done(my_off_t event_master_log_pos, THD *thd, +bool Relay_log_info::stmt_done(my_off_t event_master_log_pos, THD *thd, rpl_group_info *rgi) { + int error= 0; DBUG_ENTER("Relay_log_info::stmt_done"); DBUG_ASSERT(rgi->rli == this); @@ -1358,10 +1364,11 @@ void Relay_log_info::stmt_done(my_off_t event_master_log_pos, THD *thd, } DBUG_EXECUTE_IF("inject_crash_before_flush_rli", DBUG_SUICIDE();); if (mi->using_gtid == Master_info::USE_GTID_NO) - flush_relay_log_info(this); + if (flush_relay_log_info(this)) + error= 1; DBUG_EXECUTE_IF("inject_crash_after_flush_rli", DBUG_SUICIDE();); } - DBUG_VOID_RETURN; + DBUG_RETURN(error); } diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index efcec83..7fc4178 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -416,7 +416,7 @@ class Relay_log_info : public Slave_reporting_capability relay log info and used to produce information for SHOW SLAVE STATUS. */ - void stmt_done(my_off_t event_log_pos, THD *thd, rpl_group_info *rgi); + bool stmt_done(my_off_t event_log_pos, THD *thd, rpl_group_info *rgi); int alloc_inuse_relaylog(const char *name); void free_inuse_relaylog(inuse_relaylog *ir); void reset_inuse_relaylog(); diff --git a/sql/slave.cc b/sql/slave.cc index c4817ef..2e5b301 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -228,16 +228,14 @@ void init_thread_mask(int* mask,Master_info* mi,bool inverse) /* - lock_slave_threads() + lock_slave_threads() against other threads doing STOP, START or RESET SLAVE + */ -void lock_slave_threads(Master_info* mi) +void Master_info::lock_slave_threads() { DBUG_ENTER("lock_slave_threads"); - - //TODO: see if we can do this without dual mutex - mysql_mutex_lock(&mi->run_lock); - mysql_mutex_lock(&mi->rli.run_lock); + mysql_mutex_lock(&start_stop_lock); DBUG_VOID_RETURN; } @@ -246,13 +244,10 @@ void lock_slave_threads(Master_info* mi) unlock_slave_threads() */ -void unlock_slave_threads(Master_info* mi) +void Master_info::unlock_slave_threads() { DBUG_ENTER("unlock_slave_threads"); - - //TODO: see if we can do this without dual mutex - mysql_mutex_unlock(&mi->rli.run_lock); - mysql_mutex_unlock(&mi->run_lock); + mysql_mutex_unlock(&start_stop_lock); DBUG_VOID_RETURN; } @@ -374,7 +369,6 @@ int init_slave() accepted. However bootstrap may conflict with us if it does START SLAVE. So it's safer to take the lock. */ - mysql_mutex_lock(&LOCK_active_mi); if (pthread_key_create(&RPL_MASTER_INFO, NULL)) goto err; @@ -383,7 +377,6 @@ int init_slave() if (!master_info_index || master_info_index->init_all_master_info()) { sql_print_error("Failed to initialize multi master structures"); - mysql_mutex_unlock(&LOCK_active_mi); DBUG_RETURN(1); } if (!(active_mi= new Master_info(&default_master_connection_name, @@ -441,7 +434,6 @@ int init_slave() } end: - mysql_mutex_unlock(&LOCK_active_mi); DBUG_RETURN(error); err: @@ -614,6 +606,7 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock) if (!mi->inited) DBUG_RETURN(0); /* successfully do nothing */ int error,force_all = (thread_mask & SLAVE_FORCE_ALL); + int retval= 0; mysql_mutex_t *sql_lock = &mi->rli.run_lock, *io_lock = &mi->run_lock; mysql_mutex_t *log_lock= mi->rli.relay_log.get_log_lock(); @@ -633,24 +626,19 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock) skip_lock)) && !force_all) DBUG_RETURN(error); + retval= error; mysql_mutex_lock(log_lock); DBUG_PRINT("info",("Flushing relay-log info file.")); if (current_thd) THD_STAGE_INFO(current_thd, stage_flushing_relay_log_info_file); - if (flush_relay_log_info(&mi->rli)) - DBUG_RETURN(ER_ERROR_DURING_FLUSH_LOGS); - - if (my_sync(mi->rli.info_fd, MYF(MY_WME))) - DBUG_RETURN(ER_ERROR_DURING_FLUSH_LOGS); + if (flush_relay_log_info(&mi->rli) || + my_sync(mi->rli.info_fd, MYF(MY_WME))) + retval= ER_ERROR_DURING_FLUSH_LOGS; mysql_mutex_unlock(log_lock); } - if (opt_slave_parallel_threads > 0 && - master_info_index &&// master_info_index is set to NULL on server shutdown - !master_info_index->any_slave_sql_running()) - rpl_parallel_inactivate_pool(&global_rpl_thread_pool); if (thread_mask & (SLAVE_IO|SLAVE_FORCE_ALL)) { DBUG_PRINT("info",("Terminating IO thread")); @@ -661,25 +649,26 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock) skip_lock)) && !force_all) DBUG_RETURN(error); + if (!retval) + retval= error; mysql_mutex_lock(log_lock); DBUG_PRINT("info",("Flushing relay log and master info file.")); if (current_thd) THD_STAGE_INFO(current_thd, stage_flushing_relay_log_and_master_info_repository); - if (flush_master_info(mi, TRUE, FALSE)) - DBUG_RETURN(ER_ERROR_DURING_FLUSH_LOGS); - + if (likely(mi->fd >= 0)) + { + if (flush_master_info(mi, TRUE, FALSE) || my_sync(mi->fd, MYF(MY_WME))) + retval= ER_ERROR_DURING_FLUSH_LOGS; + } if (mi->rli.relay_log.is_open() && my_sync(mi->rli.relay_log.get_log_file()->file, MYF(MY_WME))) - DBUG_RETURN(ER_ERROR_DURING_FLUSH_LOGS); - - if (my_sync(mi->fd, MYF(MY_WME))) - DBUG_RETURN(ER_ERROR_DURING_FLUSH_LOGS); + retval= ER_ERROR_DURING_FLUSH_LOGS; mysql_mutex_unlock(log_lock); } - DBUG_RETURN(0); + DBUG_RETURN(retval); } @@ -842,6 +831,15 @@ int start_slave_thread( mysql_mutex_unlock(start_lock); DBUG_RETURN(ER_SLAVE_THREAD); } + + /* + In the following loop we can't check for thd->killed as we have to + wait until THD structures for the slave thread are created + before we can return. + This should be ok as there is no major work done in the slave + threads before they signal that we can stop waiting. + */ + if (start_cond && cond_lock) // caller has cond_lock { THD* thd = current_thd; @@ -859,16 +857,9 @@ int start_slave_thread( registered, we could otherwise go waiting though thd->killed is set. */ - if (!thd->killed) - mysql_cond_wait(start_cond, cond_lock); + mysql_cond_wait(start_cond, cond_lock); thd->EXIT_COND(& saved_stage); mysql_mutex_lock(cond_lock); // re-acquire it as exit_cond() released - if (thd->killed) - { - if (start_lock) - mysql_mutex_unlock(start_lock); - DBUG_RETURN(thd->killed_errno()); - } } } if (start_lock) @@ -975,10 +966,18 @@ int start_slave_threads(bool need_slave_mutex, bool wait_for_start, /* - Release slave threads at time of executing shutdown. + Kill slaves preparing for shutdown +*/ - SYNOPSIS - end_slave() +void slave_prepare_for_shutdown() +{ + mysql_mutex_lock(&LOCK_active_mi); + master_info_index->free_connections(); + mysql_mutex_unlock(&LOCK_active_mi); +} + +/* + Release slave threads at time of executing shutdown. */ void end_slave() @@ -996,7 +995,10 @@ void end_slave() startup parameter to the server was wrong. */ mysql_mutex_lock(&LOCK_active_mi); - /* This will call terminate_slave_threads() on all connections */ + /* + master_info_index should not have any threads anymore as they where + killed as part of slave_prepare_for_shutdown() + */ delete master_info_index; master_info_index= 0; active_mi= 0; @@ -2657,7 +2659,9 @@ static bool send_show_master_info_data(THD *thd, Master_info *mi, bool full, mysql_mutex_lock(&mi->data_lock); mysql_mutex_lock(&mi->rli.data_lock); + /* err_lock is to protect mi->last_error() */ mysql_mutex_lock(&mi->err_lock); + /* err_lock is to protect mi->rli.last_error() */ mysql_mutex_lock(&mi->rli.err_lock); protocol->store(mi->host, &my_charset_bin); protocol->store(mi->user, &my_charset_bin); @@ -4779,8 +4783,15 @@ pthread_handler_t handle_slave_sql(void *arg) if (rli->mi->using_gtid != Master_info::USE_GTID_NO) { ulong domain_count; + my_bool save_log_all_errors= thd->log_all_errors; + /* + We don't need to check return value for flush_relay_log_info() + as any errors should be logged to stderr + */ + thd->log_all_errors= 1; flush_relay_log_info(rli); + thd->log_all_errors= save_log_all_errors; if (mi->using_parallel()) { /* @@ -4862,17 +4873,7 @@ pthread_handler_t handle_slave_sql(void *arg) DBUG_EXECUTE_IF("simulate_slave_delay_at_terminate_bug38694", sleep(5);); mysql_mutex_unlock(&rli->run_lock); // tell the world we are done - /* - Deactivate the parallel replication thread pool, if there are now no more - SQL threads running. Do this here, when we have released all locks, but - while our THD (and current_thd) is still valid. - */ - mysql_mutex_lock(&LOCK_active_mi); - if (opt_slave_parallel_threads > 0 && - master_info_index &&// master_info_index is set to NULL on server shutdown - !master_info_index->any_slave_sql_running()) - rpl_parallel_inactivate_pool(&global_rpl_thread_pool); - mysql_mutex_unlock(&LOCK_active_mi); + rpl_parallel_resize_pool_if_no_slaves(); mysql_mutex_lock(&LOCK_thread_count); delete thd; @@ -5982,6 +5983,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len) void end_relay_log_info(Relay_log_info* rli) { + mysql_mutex_t *log_lock; DBUG_ENTER("end_relay_log_info"); if (!rli->inited) @@ -5999,8 +6001,11 @@ void end_relay_log_info(Relay_log_info* rli) rli->cur_log_fd = -1; } rli->inited = 0; + log_lock= rli->relay_log.get_log_lock(); + mysql_mutex_lock(log_lock); rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT); rli->relay_log.harvest_bytes_written(&rli->log_space_total); + mysql_mutex_unlock(log_lock); /* Delete the slave's temporary tables from memory. In the future there will be other actions than this, to ensure persistance @@ -6716,9 +6721,12 @@ static Log_event* next_event(rpl_group_info *rgi, ulonglong *event_size) } rli->event_relay_log_pos = BIN_LOG_HEADER_SIZE; strmake_buf(rli->event_relay_log_name,rli->linfo.log_file_name); - flush_relay_log_info(rli); + if (flush_relay_log_info(rli)) + { + errmsg= "error flushing relay log"; + goto err; + } } - /* Now we want to open this next log. To know if it's a hot log (the one being written by the I/O thread now) or a cold log, we can use diff --git a/sql/slave.h b/sql/slave.h index c6b78b9..abcb4df 100644 --- a/sql/slave.h +++ b/sql/slave.h @@ -213,13 +213,12 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report, const char *print_slave_db_safe(const char *db); void skip_load_data_infile(NET* net); +void slave_prepare_for_shutdown(); void end_slave(); /* release slave threads */ void close_active_mi(); /* clean up slave threads data */ void clear_until_condition(Relay_log_info* rli); void clear_slave_error(Relay_log_info* rli); void end_relay_log_info(Relay_log_info* rli); -void lock_slave_threads(Master_info* mi); -void unlock_slave_threads(Master_info* mi); void init_thread_mask(int* mask,Master_info* mi,bool inverse); Format_description_log_event * read_relay_log_description_event(IO_CACHE *cur_log, ulonglong start_pos, diff --git a/sql/sql_class.cc b/sql/sql_class.cc index 21c8d55..6177210 100644 --- a/sql/sql_class.cc +++ b/sql/sql_class.cc @@ -6516,7 +6516,13 @@ bool THD::rgi_have_temporary_tables() So in this case, do a re-init of the mutex. In release builds, we want to avoid the overhead of a re-init though. + + To ensure that no one is locking the mutex, we take a lock of it first. + For full explanation, see wait_for_commit::~wait_for_commit() */ + mysql_mutex_lock(&LOCK_wait_commit); + mysql_mutex_unlock(&LOCK_wait_commit); + mysql_mutex_destroy(&LOCK_wait_commit); mysql_mutex_init(key_LOCK_wait_commit, &LOCK_wait_commit, MY_MUTEX_INIT_FAST); #endif diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc index 03bad46..0bbc99b 100644 --- a/sql/sql_parse.cc +++ b/sql/sql_parse.cc @@ -2207,7 +2207,7 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) int mysql_execute_command(THD *thd) { - int res= FALSE; + int res= 0; int up_result= 0; LEX *lex= thd->lex; /* first SELECT_LEX (have special meaning for many of non-SELECTcommands) */ @@ -2702,10 +2702,17 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) if (check_global_access(thd, SUPER_ACL)) goto error; + /* + In this code it's ok to use LOCK_active_mi as we are adding new things + into master_info_index + */ mysql_mutex_lock(&LOCK_active_mi); - if (!master_info_index) + { + mysql_mutex_unlock(&LOCK_active_mi); + my_error(ER_SERVER_SHUTDOWN, MYF(0)); goto error; + } mi= master_info_index->get_master_info(&lex_mi->connection_name, Sql_condition::WARN_LEVEL_NOTE); @@ -2734,7 +2741,7 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) If new master was not added, we still need to free mi. */ if (master_info_added) - master_info_index->remove_master_info(&lex_mi->connection_name); + master_info_index->remove_master_info(mi); else delete mi; } @@ -2752,22 +2759,24 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) /* Accept one of two privileges */ if (check_global_access(thd, SUPER_ACL | REPL_CLIENT_ACL)) goto error; - mysql_mutex_lock(&LOCK_active_mi); if (lex->verbose) + { + mysql_mutex_lock(&LOCK_active_mi); res= show_all_master_info(thd); + mysql_mutex_unlock(&LOCK_active_mi); + } else { LEX_MASTER_INFO *lex_mi= &thd->lex->mi; Master_info *mi; - mi= master_info_index->get_master_info(&lex_mi->connection_name, - Sql_condition::WARN_LEVEL_ERROR); - if (mi != NULL) + if ((mi= get_master_info(&lex_mi->connection_name, + Sql_condition::WARN_LEVEL_ERROR))) { res= show_master_info(thd, mi, 0); + mi->release(); } } - mysql_mutex_unlock(&LOCK_active_mi); break; } case SQLCOM_SHOW_MASTER_STAT: @@ -3091,22 +3100,23 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) load_error= rpl_load_gtid_slave_state(thd); - mysql_mutex_lock(&LOCK_active_mi); - - if ((mi= (master_info_index-> - get_master_info(&lex_mi->connection_name, - Sql_condition::WARN_LEVEL_ERROR)))) + /* + We don't need to ensure that only one user is using master_info + as start_slave is protected against simultaneous usage + */ + if ((mi= get_master_info(&lex_mi->connection_name, + Sql_condition::WARN_LEVEL_ERROR))) { if (load_error) { /* - We cannot start a slave using GTID if we cannot load the GTID position - from the mysql.gtid_slave_pos table. But we can allow non-GTID - replication (useful eg. during upgrade). + We cannot start a slave using GTID if we cannot load the + GTID position from the mysql.gtid_slave_pos table. But we + can allow non-GTID replication (useful eg. during upgrade). */ if (mi->using_gtid != Master_info::USE_GTID_NO) { - mysql_mutex_unlock(&LOCK_active_mi); + mi->release(); break; } else @@ -3114,8 +3124,8 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) } if (!start_slave(thd, mi, 1 /* net report*/)) my_ok(thd); + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); break; } case SQLCOM_SLAVE_STOP: @@ -3145,13 +3155,19 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) } lex_mi= &thd->lex->mi; - mysql_mutex_lock(&LOCK_active_mi); - if ((mi= (master_info_index-> - get_master_info(&lex_mi->connection_name, - Sql_condition::WARN_LEVEL_ERROR)))) - if (!stop_slave(thd, mi, 1/* net report*/)) + if ((mi= get_master_info(&lex_mi->connection_name, + Sql_condition::WARN_LEVEL_ERROR))) + { + if (stop_slave(thd, mi, 1/* net report*/)) + res= 1; + DBUG_ASSERT(res == 0 || thd->get_stmt_da()->is_error()); // QQ + mi->release(); + if (rpl_parallel_resize_pool_if_no_slaves()) + res= 1; + DBUG_ASSERT(res == 0 || thd->get_stmt_da()->is_error()); // QQ + if (!res) my_ok(thd); - mysql_mutex_unlock(&LOCK_active_mi); + } break; } case SQLCOM_SLAVE_ALL_START: @@ -4317,11 +4333,13 @@ static bool lock_tables_open_and_lock_tables(THD *thd, TABLE_LIST *tables) reload_acl_and_cache binlog interactions failed */ res= 1; - } + } if (!res) my_ok(thd); } + else + res= 1; // reload_acl_and_cache failed #ifdef HAVE_REPLICATION if (lex->type & REFRESH_READ_LOCK) rpl_unpause_after_ftwrl(thd); diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc index bcf2585..f9c5757 100644 --- a/sql/sql_reload.cc +++ b/sql/sql_reload.cc @@ -174,24 +174,20 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options, slave is not likely to have the same connection names. */ tmp_write_to_binlog= 0; - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index) + + if (!(mi= (get_master_info(&connection_name, + Sql_condition::WARN_LEVEL_ERROR)))) { - if (!(mi= (master_info_index-> - get_master_info(&connection_name, - Sql_condition::WARN_LEVEL_ERROR)))) - { - result= 1; - } - else - { - mysql_mutex_lock(&mi->data_lock); - if (rotate_relay_log(mi)) - *write_to_binlog= -1; - mysql_mutex_unlock(&mi->data_lock); - } + result= 1; + } + else + { + mysql_mutex_lock(&mi->data_lock); + if (rotate_relay_log(mi)) + *write_to_binlog= -1; + mysql_mutex_unlock(&mi->data_lock); + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); #endif } #ifdef HAVE_QUERY_CACHE @@ -349,27 +345,33 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options, LEX_MASTER_INFO* lex_mi= &thd->lex->mi; Master_info *mi; tmp_write_to_binlog= 0; - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index) + + if (!(mi= get_master_info(&lex_mi->connection_name, + Sql_condition::WARN_LEVEL_ERROR))) { - if (!(mi= (master_info_index-> - get_master_info(&lex_mi->connection_name, - Sql_condition::WARN_LEVEL_ERROR)))) - { - result= 1; - } - else if (reset_slave(thd, mi)) + result= 1; + } + else + { + /* The following will fail if slave is running */ + if (reset_slave(thd, mi)) { + mi->release(); /* NOTE: my_error() has been already called by reset_slave(). */ result= 1; } else if (mi->connection_name.length && thd->lex->reset_slave_info.all) { /* If not default connection and 'all' is used */ - master_info_index->remove_master_info(&mi->connection_name); + mi->release(); + mysql_mutex_lock(&LOCK_active_mi); + if (master_info_index->remove_master_info(mi)) + result= 1; + mysql_mutex_unlock(&LOCK_active_mi); } + else + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); } #endif if (options & REFRESH_USER_RESOURCES) diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 4505ad6..560b7ed 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -2833,7 +2833,16 @@ int start_slave(THD* thd , Master_info* mi, bool net_report) relay_log_info_file, 0, &mi->cmp_connection_name); - lock_slave_threads(mi); // this allows us to cleanly read slave_running + mi->lock_slave_threads(); + if (mi->killed) + { + /* connection was deleted while we waited for lock_slave_threads */ + mi->unlock_slave_threads(); + my_error(WARN_NO_MASTER_INFO, mi->connection_name.length, + mi->connection_name.str); + DBUG_RETURN(-1); + } + // Get a mask of _stopped_ threads init_thread_mask(&thread_mask,mi,1 /* inverse */); @@ -2968,7 +2977,7 @@ int start_slave(THD* thd , Master_info* mi, bool net_report) ER(ER_UNTIL_COND_IGNORED)); if (!slave_errno) - slave_errno = start_slave_threads(0 /*no mutex */, + slave_errno = start_slave_threads(1, 1 /* wait for start */, mi, master_info_file_tmp, @@ -2984,7 +2993,7 @@ int start_slave(THD* thd , Master_info* mi, bool net_report) } err: - unlock_slave_threads(mi); + mi->unlock_slave_threads(); if (slave_errno) { @@ -3024,8 +3033,12 @@ int stop_slave(THD* thd, Master_info* mi, bool net_report ) DBUG_RETURN(-1); THD_STAGE_INFO(thd, stage_killing_slave); int thread_mask; - lock_slave_threads(mi); - // Get a mask of _running_ threads + mi->lock_slave_threads(); + /* + Get a mask of _running_ threads. + We don't have to test for mi->killed as the thread_mask will take care + of checking if threads exists + */ init_thread_mask(&thread_mask,mi,0 /* not inverse*/); /* Below we will stop all running threads. @@ -3038,8 +3051,7 @@ int stop_slave(THD* thd, Master_info* mi, bool net_report ) if (thread_mask) { - slave_errno= terminate_slave_threads(mi,thread_mask, - 1 /*skip lock */); + slave_errno= terminate_slave_threads(mi,thread_mask, 0 /* get lock */); } else { @@ -3048,7 +3060,8 @@ int stop_slave(THD* thd, Master_info* mi, bool net_report ) push_warning(thd, Sql_condition::WARN_LEVEL_NOTE, ER_SLAVE_WAS_NOT_RUNNING, ER(ER_SLAVE_WAS_NOT_RUNNING)); } - unlock_slave_threads(mi); + + mi->unlock_slave_threads(); if (slave_errno) { @@ -3083,11 +3096,20 @@ int reset_slave(THD *thd, Master_info* mi) char relay_log_info_file_tmp[FN_REFLEN]; DBUG_ENTER("reset_slave"); - lock_slave_threads(mi); + mi->lock_slave_threads(); + if (mi->killed) + { + /* connection was deleted while we waited for lock_slave_threads */ + mi->unlock_slave_threads(); + my_error(WARN_NO_MASTER_INFO, mi->connection_name.length, + mi->connection_name.str); + DBUG_RETURN(-1); + } + init_thread_mask(&thread_mask,mi,0 /* not inverse */); if (thread_mask) // We refuse if any slave thread is running { - unlock_slave_threads(mi); + mi->unlock_slave_threads(); my_error(ER_SLAVE_MUST_STOP, MYF(0), (int) mi->connection_name.length, mi->connection_name.str); DBUG_RETURN(ER_SLAVE_MUST_STOP); @@ -3152,7 +3174,7 @@ int reset_slave(THD *thd, Master_info* mi) RUN_HOOK(binlog_relay_io, after_reset_slave, (thd, mi)); err: - unlock_slave_threads(mi); + mi->unlock_slave_threads(); if (error) my_error(sql_errno, MYF(0), errmsg); DBUG_RETURN(error); @@ -3265,8 +3287,8 @@ bool change_master(THD* thd, Master_info* mi, bool *master_info_added) LEX_MASTER_INFO* lex_mi= &thd->lex->mi; DBUG_ENTER("change_master"); - mysql_mutex_assert_owner(&LOCK_active_mi); DBUG_ASSERT(master_info_index); + mysql_mutex_assert_owner(&LOCK_active_mi); *master_info_added= false; /* @@ -3286,7 +3308,16 @@ bool change_master(THD* thd, Master_info* mi, bool *master_info_added) lex_mi->port)) DBUG_RETURN(TRUE); - lock_slave_threads(mi); + mi->lock_slave_threads(); + if (mi->killed) + { + /* connection was deleted while we waited for lock_slave_threads */ + mi->unlock_slave_threads(); + my_error(WARN_NO_MASTER_INFO, mi->connection_name.length, + mi->connection_name.str); + DBUG_RETURN(TRUE); + } + init_thread_mask(&thread_mask,mi,0 /*not inverse*/); if (thread_mask) // We refuse if any slave thread is running { @@ -3588,12 +3619,13 @@ bool change_master(THD* thd, Master_info* mi, bool *master_info_added) in-memory value at restart (thus causing errors, as the old relay log does not exist anymore). */ - flush_relay_log_info(&mi->rli); + if (flush_relay_log_info(&mi->rli)) + ret= 1; mysql_cond_broadcast(&mi->data_cond); mysql_mutex_unlock(&mi->rli.data_lock); err: - unlock_slave_threads(mi); + mi->unlock_slave_threads(); if (ret == FALSE) my_ok(thd); DBUG_RETURN(ret); @@ -3646,7 +3678,6 @@ bool mysql_show_binlog_events(THD* thd) int old_max_allowed_packet= thd->variables.max_allowed_packet; Master_info *mi= 0; LOG_INFO linfo; - DBUG_ENTER("mysql_show_binlog_events"); Log_event::init_show_field_list(&field_list); @@ -3674,13 +3705,9 @@ bool mysql_show_binlog_events(THD* thd) } else /* showing relay log contents */ { - mysql_mutex_lock(&LOCK_active_mi); - if (!master_info_index || - !(mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_ERROR))) + if (!(mi= get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_ERROR))) { - mysql_mutex_unlock(&LOCK_active_mi); DBUG_RETURN(TRUE); } binary_log= &(mi->rli.relay_log); @@ -3700,7 +3727,7 @@ bool mysql_show_binlog_events(THD* thd) if (mi) { /* We can unlock the mutex as we have a lock on the file */ - mysql_mutex_unlock(&LOCK_active_mi); + mi->release(); mi= 0; } @@ -3722,6 +3749,7 @@ bool mysql_show_binlog_events(THD* thd) goto err; } + /* These locks is here to enable syncronization with log_in_use() */ mysql_mutex_lock(&LOCK_thread_count); thd->current_linfo = &linfo; mysql_mutex_unlock(&LOCK_thread_count); @@ -3799,7 +3827,7 @@ bool mysql_show_binlog_events(THD* thd) mysql_mutex_unlock(log_lock); } else if (mi) - mysql_mutex_unlock(&LOCK_active_mi); + mi->release(); // Check that linfo is still on the function scope. DEBUG_SYNC(thd, "after_show_binlog_events"); @@ -3820,8 +3848,9 @@ bool mysql_show_binlog_events(THD* thd) else my_eof(thd); + /* These locks is here to enable syncronization with log_in_use() */ mysql_mutex_lock(&LOCK_thread_count); - thd->current_linfo = 0; + thd->current_linfo= 0; mysql_mutex_unlock(&LOCK_thread_count); thd->variables.max_allowed_packet= old_max_allowed_packet; DBUG_RETURN(ret); diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc index 4adc192..5ff79a2 100644 --- a/sql/sys_vars.cc +++ b/sql/sys_vars.cc @@ -1527,7 +1527,6 @@ static Sys_var_gtid_current_pos Sys_gtid_current_pos( Sys_var_gtid_slave_pos::do_check(THD *thd, set_var *var) { String str, *res; - bool running; DBUG_ASSERT(var->type == OPT_GLOBAL); @@ -1538,11 +1537,7 @@ static Sys_var_gtid_current_pos Sys_gtid_current_pos( return true; } - mysql_mutex_lock(&LOCK_active_mi); - running= (!master_info_index || - master_info_index->give_error_if_slave_running()); - mysql_mutex_unlock(&LOCK_active_mi); - if (running) + if (give_error_if_slave_running(0)) return true; if (!(res= var->value->val_str(&str))) return true; @@ -1580,7 +1575,7 @@ static Sys_var_gtid_current_pos Sys_gtid_current_pos( mysql_mutex_unlock(&LOCK_global_system_variables); mysql_mutex_lock(&LOCK_active_mi); - if (!master_info_index || master_info_index->give_error_if_slave_running()) + if (give_error_if_slave_running(1)) err= true; else err= rpl_gtid_pos_update(thd, var->save_result.string_value.str, @@ -1766,16 +1761,7 @@ static Sys_var_last_gtid Sys_last_gtid( static bool check_slave_parallel_threads(sys_var *self, THD *thd, set_var *var) { - bool running; - - mysql_mutex_lock(&LOCK_active_mi); - running= (!master_info_index || - master_info_index->give_error_if_slave_running()); - mysql_mutex_unlock(&LOCK_active_mi); - if (running) - return true; - - return false; + return give_error_if_slave_running(0); } static bool @@ -1784,10 +1770,7 @@ static Sys_var_last_gtid Sys_last_gtid( bool err; mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); - err= (!master_info_index || - master_info_index->give_error_if_slave_running()); - mysql_mutex_unlock(&LOCK_active_mi); + err= give_error_if_slave_running(0); mysql_mutex_lock(&LOCK_global_system_variables); return err; @@ -1810,16 +1793,7 @@ static Sys_var_ulong Sys_slave_parallel_threads( static bool check_slave_domain_parallel_threads(sys_var *self, THD *thd, set_var *var) { - bool running; - - mysql_mutex_lock(&LOCK_active_mi); - running= (!master_info_index || - master_info_index->give_error_if_slave_running()); - mysql_mutex_unlock(&LOCK_active_mi); - if (running) - return true; - - return false; + return give_error_if_slave_running(0); } static bool @@ -1828,13 +1802,10 @@ static Sys_var_ulong Sys_slave_parallel_threads( bool running; mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); - running= (!master_info_index || - master_info_index->give_error_if_slave_running()); - mysql_mutex_unlock(&LOCK_active_mi); + running= give_error_if_slave_running(0); mysql_mutex_lock(&LOCK_global_system_variables); - return running ? true : false; + return running; } @@ -1865,16 +1836,7 @@ static Sys_var_ulong Sys_slave_parallel_max_queued( static bool check_gtid_ignore_duplicates(sys_var *self, THD *thd, set_var *var) { - bool running; - - mysql_mutex_lock(&LOCK_active_mi); - running= (!master_info_index || - master_info_index->give_error_if_slave_running()); - mysql_mutex_unlock(&LOCK_active_mi); - if (running) - return true; - - return false; + return give_error_if_slave_running(0); } static bool @@ -1883,13 +1845,10 @@ static Sys_var_ulong Sys_slave_parallel_max_queued( bool running; mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); - running= (!master_info_index || - master_info_index->give_error_if_slave_running()); - mysql_mutex_unlock(&LOCK_active_mi); + running= give_error_if_slave_running(0); mysql_mutex_lock(&LOCK_global_system_variables); - return running ? true : false; + return running; } @@ -2837,10 +2796,8 @@ static Sys_var_mybool Sys_master_verify_checksum( DBUG_ENTER("Sys_var_replicate_events_marked_for_skip::global_update"); mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); - if (master_info_index && !master_info_index->give_error_if_slave_running()) + if (!give_error_if_slave_running(0)) result= Sys_var_enum::global_update(thd, var); - mysql_mutex_unlock(&LOCK_active_mi); mysql_mutex_lock(&LOCK_global_system_variables); DBUG_RETURN(result); } @@ -4112,19 +4069,16 @@ bool Sys_var_rpl_filter::global_update(THD *thd, set_var *var) Master_info *mi; mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); if (!var->base.length) // no base name { - mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_ERROR); + mi= get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_ERROR); } else // has base name { - mi= master_info_index-> - get_master_info(&var->base, - Sql_condition::WARN_LEVEL_WARN); + mi= get_master_info(&var->base, + Sql_condition::WARN_LEVEL_WARN); } if (mi) @@ -4132,17 +4086,17 @@ bool Sys_var_rpl_filter::global_update(THD *thd, set_var *var) if (mi->rli.slave_running) { my_error(ER_SLAVE_MUST_STOP, MYF(0), - mi->connection_name.length, - mi->connection_name.str); + mi->connection_name.length, + mi->connection_name.str); result= true; } else { result= set_filter_value(var->save_result.string_value.str, mi); } + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); mysql_mutex_lock(&LOCK_global_system_variables); return result; } @@ -4150,8 +4104,10 @@ bool Sys_var_rpl_filter::global_update(THD *thd, set_var *var) bool Sys_var_rpl_filter::set_filter_value(const char *value, Master_info *mi) { bool status= true; - Rpl_filter* rpl_filter= mi ? mi->rpl_filter : global_rpl_filter; + Rpl_filter* rpl_filter= mi->rpl_filter; + /* Proctect against other threads */ + mysql_mutex_lock(&LOCK_active_mi); switch (opt_id) { case OPT_REPLICATE_DO_DB: status= rpl_filter->set_do_db(value); @@ -4172,7 +4128,7 @@ bool Sys_var_rpl_filter::set_filter_value(const char *value, Master_info *mi) status= rpl_filter->set_wild_ignore_table(value); break; } - + mysql_mutex_unlock(&LOCK_active_mi); return status; } @@ -4185,29 +4141,24 @@ uchar *Sys_var_rpl_filter::global_value_ptr(THD *thd, LEX_STRING *base) Rpl_filter *rpl_filter; mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); if (!base->length) // no base name { - mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_ERROR); + mi= get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_ERROR); } else // has base name - { - mi= master_info_index-> - get_master_info(base, - Sql_condition::WARN_LEVEL_WARN); - } - mysql_mutex_lock(&LOCK_global_system_variables); + mi= get_master_info(base, Sql_condition::WARN_LEVEL_WARN); if (!mi) { - mysql_mutex_unlock(&LOCK_active_mi); + mysql_mutex_lock(&LOCK_global_system_variables); return 0; } + rpl_filter= mi->rpl_filter; tmp.length(0); + mysql_mutex_lock(&LOCK_active_mi); switch (opt_id) { case OPT_REPLICATE_DO_DB: rpl_filter->get_do_db(&tmp); @@ -4228,9 +4179,12 @@ uchar *Sys_var_rpl_filter::global_value_ptr(THD *thd, LEX_STRING *base) rpl_filter->get_wild_ignore_table(&tmp); break; } + mysql_mutex_unlock(&LOCK_active_mi); + mysql_mutex_lock(&LOCK_global_system_variables); + + mi->release(); ret= (uchar *) thd->strmake(tmp.ptr(), tmp.length()); - mysql_mutex_unlock(&LOCK_active_mi); return ret; } @@ -4301,17 +4255,12 @@ static Sys_var_uint Sys_slave_net_timeout( Master_info *mi; ulonglong res= 0; // Default value mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); - mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_WARN); - if (mi) + if ((mi= get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_WARN))) { - mysql_mutex_lock(&mi->rli.data_lock); res= *((ulonglong*) (((uchar*) mi) + master_info_offset)); - mysql_mutex_unlock(&mi->rli.data_lock); + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); mysql_mutex_lock(&LOCK_global_system_variables); return res; } @@ -4326,19 +4275,16 @@ bool update_multi_source_variable(sys_var *self_var, THD *thd, if (type == OPT_GLOBAL) mysql_mutex_unlock(&LOCK_global_system_variables); - mysql_mutex_lock(&LOCK_active_mi); - mi= master_info_index-> - get_master_info(&thd->variables.default_master_connection, - Sql_condition::WARN_LEVEL_ERROR); - if (mi) + if ((mi= (get_master_info(&thd->variables.default_master_connection, + Sql_condition::WARN_LEVEL_ERROR)))) { mysql_mutex_lock(&mi->rli.run_lock); mysql_mutex_lock(&mi->rli.data_lock); result= self->update_variable(thd, mi); mysql_mutex_unlock(&mi->rli.data_lock); mysql_mutex_unlock(&mi->rli.run_lock); + mi->release(); } - mysql_mutex_unlock(&LOCK_active_mi); if (type == OPT_GLOBAL) mysql_mutex_lock(&LOCK_global_system_variables); return result; diff --git a/storage/connect/ha_connect.cc b/storage/connect/ha_connect.cc index 33f02d8..e66f03e 100644 --- a/storage/connect/ha_connect.cc +++ b/storage/connect/ha_connect.cc @@ -5173,7 +5173,7 @@ static int connect_assisted_discovery(handlerton *, THD* thd, TABLE_SHARE *table_s, HA_CREATE_INFO *create_info) { - char v=0, spc= ',', qch= 0; + char v=0; const char *fncn= "?"; const char *user, *fn, *db, *host, *pwd, *sep, *tbl, *src; const char *col, *ocl, *rnk, *pic, *fcl, *skc, *zfn; @@ -5225,10 +5225,6 @@ static int connect_assisted_discovery(handlerton *, THD* thd, fncn= topt->catfunc; fnc= GetFuncID(fncn); sep= topt->separator; - spc= (!sep) ? ',' : *sep; - qch= topt->qchar ? *topt->qchar : (signed)topt->quoted >= 0 ? '"' : 0; - mul = (int)topt->multiple; - tbl= topt->tablist; col= topt->colist; if (topt->oplist) { diff --git a/storage/connect/ioapi.c b/storage/connect/ioapi.c index 7f5c191..a49da91 100644 --- a/storage/connect/ioapi.c +++ b/storage/connect/ioapi.c @@ -27,6 +27,7 @@ #include "ioapi.h" +#include "my_attribute.h" voidpf call_zopen64 (const zlib_filefunc64_32_def* pfilefunc,const void*filename,int mode) { @@ -92,7 +93,7 @@ void fill_zlib_filefunc64_32_def_from_filefunc32(zlib_filefunc64_32_def* p_filef static int ZCALLBACK fclose_file_func OF((voidpf opaque, voidpf stream)); static int ZCALLBACK ferror_file_func OF((voidpf opaque, voidpf stream)); -static voidpf ZCALLBACK fopen_file_func (voidpf opaque, const char* filename, int mode) +static voidpf ZCALLBACK fopen_file_func (voidpf opaque __attribute__((unused)), const char* filename, int mode) { FILE* file = NULL; const char* mode_fopen = NULL; @@ -110,7 +111,7 @@ static voidpf ZCALLBACK fopen_file_func (voidpf opaque, const char* filename, in return file; } -static voidpf ZCALLBACK fopen64_file_func (voidpf opaque, const void* filename, int mode) +static voidpf ZCALLBACK fopen64_file_func (voidpf opaque __attribute__((unused)), const void* filename, int mode) { FILE* file = NULL; const char* mode_fopen = NULL; @@ -129,21 +130,21 @@ static voidpf ZCALLBACK fopen64_file_func (voidpf opaque, const void* filename, } -static uLong ZCALLBACK fread_file_func (voidpf opaque, voidpf stream, void* buf, uLong size) +static uLong ZCALLBACK fread_file_func (voidpf opaque __attribute__((unused)), voidpf stream, void* buf, uLong size) { uLong ret; ret = (uLong)fread(buf, 1, (size_t)size, (FILE *)stream); return ret; } -static uLong ZCALLBACK fwrite_file_func (voidpf opaque, voidpf stream, const void* buf, uLong size) +static uLong ZCALLBACK fwrite_file_func (voidpf opaque __attribute__((unused)), voidpf stream, const void* buf, uLong size) { uLong ret; ret = (uLong)fwrite(buf, 1, (size_t)size, (FILE *)stream); return ret; } -static long ZCALLBACK ftell_file_func (voidpf opaque, voidpf stream) +static long ZCALLBACK ftell_file_func (voidpf opaque __attribute__((unused)), voidpf stream) { long ret; ret = ftell((FILE *)stream); @@ -151,14 +152,14 @@ static long ZCALLBACK ftell_file_func (voidpf opaque, voidpf stream) } -static ZPOS64_T ZCALLBACK ftell64_file_func (voidpf opaque, voidpf stream) +static ZPOS64_T ZCALLBACK ftell64_file_func (voidpf opaque __attribute__((unused)), voidpf stream) { ZPOS64_T ret; ret = FTELLO_FUNC((FILE *)stream); return ret; } -static long ZCALLBACK fseek_file_func (voidpf opaque, voidpf stream, uLong offset, int origin) +static long ZCALLBACK fseek_file_func (voidpf opaque __attribute__((unused)), voidpf stream, uLong offset, int origin) { int fseek_origin=0; long ret; @@ -181,7 +182,7 @@ static long ZCALLBACK fseek_file_func (voidpf opaque, voidpf stream, uLong offs return ret; } -static long ZCALLBACK fseek64_file_func (voidpf opaque, voidpf stream, ZPOS64_T offset, int origin) +static long ZCALLBACK fseek64_file_func (voidpf opaque __attribute__((unused)), voidpf stream, ZPOS64_T offset, int origin) { int fseek_origin=0; long ret; @@ -207,14 +208,14 @@ static long ZCALLBACK fseek64_file_func (voidpf opaque, voidpf stream, ZPOS64_T } -static int ZCALLBACK fclose_file_func (voidpf opaque, voidpf stream) +static int ZCALLBACK fclose_file_func (voidpf opaque __attribute__((unused)), voidpf stream) { int ret; ret = fclose((FILE *)stream); return ret; } -static int ZCALLBACK ferror_file_func (voidpf opaque, voidpf stream) +static int ZCALLBACK ferror_file_func (voidpf opaque __attribute__((unused)), voidpf stream) { int ret; ret = ferror((FILE *)stream); diff --git a/storage/connect/zip.c b/storage/connect/zip.c index ea54853..4bbe31a 100644 --- a/storage/connect/zip.c +++ b/storage/connect/zip.c @@ -637,7 +637,7 @@ local ZPOS64_T zip64local_SearchCentralDir64(const zlib_filefunc64_32_def* pzlib return relativeOffset; } -int LoadCentralDirectoryRecord(zip64_internal* pziinit) +static int LoadCentralDirectoryRecord(zip64_internal* pziinit) { int err=ZIP_OK; ZPOS64_T byte_before_the_zipfile;/* byte before the zipfile, (>0 for sfx)*/ @@ -846,7 +846,7 @@ int LoadCentralDirectoryRecord(zip64_internal* pziinit) /************************************************************/ -extern zipFile ZEXPORT zipOpen3 (const void *pathname, int append, zipcharpc* globalcomment, zlib_filefunc64_32_def* pzlib_filefunc64_32_def) +static zipFile zipOpen3 (const void *pathname, int append, zipcharpc* globalcomment, zlib_filefunc64_32_def* pzlib_filefunc64_32_def) { zip64_internal ziinit; zip64_internal* zi; @@ -955,7 +955,7 @@ extern zipFile ZEXPORT zipOpen64 (const void* pathname, int append) return zipOpen3(pathname,append,NULL,NULL); } -int Write_LocalFileHeader(zip64_internal* zi, const char* filename, uInt size_extrafield_local, const void* extrafield_local) +static int Write_LocalFileHeader(zip64_internal* zi, const char* filename, uInt size_extrafield_local, const void* extrafield_local) { /* write the local header */ int err; @@ -1752,7 +1752,7 @@ extern int ZEXPORT zipCloseFileInZip (zipFile file) return zipCloseFileInZipRaw (file,0,0); } -int Write_Zip64EndOfCentralDirectoryLocator(zip64_internal* zi, ZPOS64_T zip64eocd_pos_inzip) +static int Write_Zip64EndOfCentralDirectoryLocator(zip64_internal* zi, ZPOS64_T zip64eocd_pos_inzip) { int err = ZIP_OK; ZPOS64_T pos = zip64eocd_pos_inzip - zi->add_position_when_writting_offset; @@ -1774,7 +1774,7 @@ int Write_Zip64EndOfCentralDirectoryLocator(zip64_internal* zi, ZPOS64_T zip64eo return err; } -int Write_Zip64EndOfCentralDirectoryRecord(zip64_internal* zi, uLong size_centraldir, ZPOS64_T centraldir_pos_inzip) +static int Write_Zip64EndOfCentralDirectoryRecord(zip64_internal* zi, uLong size_centraldir, ZPOS64_T centraldir_pos_inzip) { int err = ZIP_OK; @@ -1813,7 +1813,7 @@ int Write_Zip64EndOfCentralDirectoryRecord(zip64_internal* zi, uLong size_centra } return err; } -int Write_EndOfCentralDirectoryRecord(zip64_internal* zi, uLong size_centraldir, ZPOS64_T centraldir_pos_inzip) +static int Write_EndOfCentralDirectoryRecord(zip64_internal* zi, uLong size_centraldir, ZPOS64_T centraldir_pos_inzip) { int err = ZIP_OK; @@ -1861,7 +1861,7 @@ int Write_EndOfCentralDirectoryRecord(zip64_internal* zi, uLong size_centraldir, return err; } -int Write_GlobalComment(zip64_internal* zi, const char* global_comment) +static int Write_GlobalComment(zip64_internal* zi, const char* global_comment) { int err = ZIP_OK; uInt size_global_comment = 0; diff --git a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.result b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.result index 42c8640..6b1d0a1 100644 --- a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.result +++ b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.result @@ -33,5 +33,5 @@ FROM `version_history` AS `v` INNER JOIN `db_history` AS `db` ON `db`.`nodeID` = WHERE `latch` = 'breadth_first' AND `origid` = '1' ORDER BY `weight` DESC LIMIT 1; version nodeID 0.0.3 3 -DROP TABLE db_history; DROP TABLE version_history; +DROP TABLE db_history; diff --git a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.test b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.test index 3a7a0e5..b8f0ae5 100644 --- a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.test +++ b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6282.test @@ -40,8 +40,8 @@ --disconnect con2 --connect (con3,localhost,root,,test) -DROP TABLE db_history; DROP TABLE version_history; +DROP TABLE db_history; --disconnect con3 diff --git a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.result b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.result index 8e680e1..68002ce 100644 --- a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.result +++ b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.result @@ -8,5 +8,5 @@ latch origid destid weight seq linkid breadth_first 1 6 NULL 0 1 breadth_first 1 6 1 1 2 breadth_first 1 6 1 2 6 -DROP TABLE IF EXISTS oq_backing; DROP TABLE IF EXISTS oq_graph; +DROP TABLE IF EXISTS oq_backing; diff --git a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.test b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.test index 72de926..fefd03a 100644 --- a/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.test +++ b/storage/oqgraph/mysql-test/oqgraph/regression_mdev6345.test @@ -14,6 +14,6 @@ CREATE TABLE oq_graph (latch VARCHAR(32) NULL, origid BIGINT UNSIGNED NULL, dest SELECT * FROM oq_graph WHERE latch='breadth_first' AND origid=1 AND destid=6; -DROP TABLE IF EXISTS oq_backing; DROP TABLE IF EXISTS oq_graph; +DROP TABLE IF EXISTS oq_backing; diff --git a/storage/xtradb/row/row0ftsort.cc b/storage/xtradb/row/row0ftsort.cc index ce13825..ef54c81 100644 --- a/storage/xtradb/row/row0ftsort.cc +++ b/storage/xtradb/row/row0ftsort.cc @@ -623,7 +623,7 @@ Create Full Text Index with (parallel) merge sort mem_heap_t* blob_heap = NULL; fts_doc_t doc; dict_table_t* table = psort_info->psort_common->new_table; - dict_field_t* idx_field; + dict_field_t* idx_field __attribute__((unused)); fts_tokenize_ctx_t t_ctx; ulint retried = 0; dberr_t error = DB_SUCCESS;