From 6e55da9f491772fb12d3f7d2be1eedb6c70bd605 Mon Sep 17 00:00:00 2001 From: Kristian Nielsen Date: Mon, 12 Oct 2026 20:31:36 +0200 Subject: [PATCH] MDEV-41217: Inefficient memory management by parallel slave Remove from rpl_parallel_thread::queued_event the names of relaylog and master binlog. Each one is an array of size FN_REFLEN, thus we save 1024 bytes of memory for each event queued for parallel replication worker threads. This can be a very large saving when the slave is lagging behind and many events are queued for worker threads. Instead, reference these names from the inuse_relaylog object. This already contains the relaylog name; this patch adds also the master binlog name in inuse_relaylog. Since we rotate the slave relaylog whenever we see a master binlog rotate, there can only ever be one master binlog file name associated with events in one relaylog. Signed-off-by: Kristian Nielsen --- sql/log.cc | 6 ++--- sql/log_event.h | 7 +++++ sql/rpl_parallel.cc | 65 ++++++++++++++++++++++++++++++--------------- sql/rpl_parallel.h | 5 ++-- sql/rpl_rli.cc | 6 +++-- sql/rpl_rli.h | 12 ++++++--- sql/slave.cc | 23 ++++++++++------ 7 files changed, 82 insertions(+), 42 deletions(-) diff --git a/sql/log.cc b/sql/log.cc index 64ac04198a2..c2e5b6a3431 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -5854,14 +5854,14 @@ int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included) included= false; break; } - if (!included && !strcmp(ir->name, rli->group_relay_log_name)) + if (!included && !strcmp(ir->relaylog_name, rli->group_relay_log_name)) break; if (!next) { rli->last_inuse_relaylog= NULL; included= 1; to_purge_if_included= my_strdup(key_memory_Relay_log_info_group_relay_log_name, - ir->name, MYF(0)); + ir->relaylog_name, MYF(0)); } rli->free_inuse_relaylog(ir); ir= next; @@ -5869,7 +5869,7 @@ int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included) rli->inuse_relaylog_list= ir; if (ir) to_purge_if_included= my_strdup(key_memory_Relay_log_info_group_relay_log_name, - ir->name, MYF(0)); + ir->relaylog_name, MYF(0)); /* Read the next log file name from the index file and pass it back to diff --git a/sql/log_event.h b/sql/log_event.h index bf3f1718691..326fd6d05dc 100644 --- a/sql/log_event.h +++ b/sql/log_event.h @@ -3352,6 +3352,13 @@ public: bool is_valid() const override { return new_log_ident != 0; } #ifdef MYSQL_SERVER bool write(Log_event_writer *writer) override; + /* + Mark that this is an artificial event generated on-the-fly when the SQL + driver thread reaches the (temporary) EOF of the hot log, to allow to + update the currently reached position. + */ + void set_is_hot_log_rotate() { server_id= 0; } + bool is_hot_log_rotate() { return (server_id == 0); } #endif private: diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc index 589a5ec74ee..3e16e784c51 100644 --- a/sql/rpl_parallel.cc +++ b/sql/rpl_parallel.cc @@ -54,12 +54,13 @@ rpt_handle_event(rpl_group_info *rgi, rpl_parallel_thread::queued_event *qev, ev->thd= thd; safe_strcpy(rgi->event_relay_log_name_buf, sizeof(rgi->event_relay_log_name_buf), - qev->event_relay_log_name); + qev->ir->relaylog_name); rgi->event_relay_log_name= rgi->event_relay_log_name_buf; rgi->event_relay_log_pos= qev->event_relay_log_pos; rgi->future_event_relay_log_pos= qev->future_event_relay_log_pos; - safe_strcpy(rgi->future_event_master_log_name, sizeof(rgi->future_event_master_log_name), - qev->future_event_master_log_name); + safe_strcpy(rgi->future_event_master_log_name, + sizeof(rgi->future_event_master_log_name), + qev->ir->master_binlog_name); if (event_can_update_last_master_timestamp(ev)) rgi->last_master_timestamp= ev->when + ev->exec_time; err= apply_event_and_update_pos_for_parallel(ev, thd, rgi); @@ -115,20 +116,21 @@ handle_queued_pos_update(THD *thd, rpl_parallel_thread::queued_event *qev) #endif mysql_mutex_lock(&rli->data_lock); - cmp= compare_log_name(rli->group_relay_log_name, qev->event_relay_log_name); + cmp= compare_log_name(rli->group_relay_log_name, qev->ir->relaylog_name); if (cmp < 0) { rli->group_relay_log_pos= qev->future_event_relay_log_pos; - strmake_buf(rli->group_relay_log_name, qev->event_relay_log_name); + strmake_buf(rli->group_relay_log_name, qev->ir->relaylog_name); } else if (cmp == 0 && rli->group_relay_log_pos < qev->future_event_relay_log_pos) rli->group_relay_log_pos= qev->future_event_relay_log_pos; - cmp= compare_log_name(rli->group_master_log_name, qev->future_event_master_log_name); + cmp= compare_log_name(rli->group_master_log_name, + qev->ir->master_binlog_name); if (cmp < 0) { safe_strcpy(rli->group_master_log_name, sizeof(rli->group_master_log_name), - qev->future_event_master_log_name); + qev->ir->master_binlog_name); rli->group_master_log_pos= qev->future_event_master_log_pos; } else if (cmp == 0 @@ -1044,7 +1046,7 @@ do_retry: WSREP_DEBUG("parallel slave retry, after trx start"); } #endif /* WITH_WSREP */ - strmake_buf(log_name, ir->name); + strmake_buf(log_name, ir->relaylog_name); if ((fd= open_binlog(&rlog, log_name, &errmsg)) <0) { err= 1; @@ -2081,7 +2083,8 @@ rpl_parallel_thread::inuse_relaylog_refcount_update() rpl_parallel_thread::queued_event * -rpl_parallel_thread::get_qev_common(Log_event *ev, ulonglong event_size) +rpl_parallel_thread::get_qev_common(Log_event *ev, ulonglong event_size, + inuse_relaylog *ir) { queued_event *qev; mysql_mutex_assert_owner(&LOCK_rpl_thread); @@ -2094,6 +2097,7 @@ rpl_parallel_thread::get_qev_common(Log_event *ev, ulonglong event_size) } qev->typ= rpl_parallel_thread::queued_event::QUEUED_EVENT; qev->ev= ev; + qev->ir= ir; qev->event_size= (size_t)event_size; qev->next= NULL; return qev; @@ -2104,16 +2108,12 @@ rpl_parallel_thread::queued_event * rpl_parallel_thread::get_qev(Log_event *ev, ulonglong event_size, Relay_log_info *rli) { - queued_event *qev= get_qev_common(ev, event_size); + queued_event *qev= get_qev_common(ev, event_size, rli->last_inuse_relaylog); if (!qev) return NULL; - safe_strcpy(qev->event_relay_log_name, sizeof(qev->event_relay_log_name), - rli->event_relay_log_name); qev->event_relay_log_pos= rli->event_relay_log_pos; qev->future_event_relay_log_pos= rli->future_event_relay_log_pos; - safe_strcpy(qev->future_event_master_log_name, - sizeof(qev->future_event_master_log_name), - rli->future_event_master_log_name); + /* The relaylog and master binlog names are available in qev->ir. */ return qev; } @@ -2123,17 +2123,13 @@ rpl_parallel_thread::retry_get_qev(Log_event *ev, queued_event *orig_qev, const char *relay_log_name, ulonglong event_pos, ulonglong event_size) { - queued_event *qev= get_qev_common(ev, event_size); + queued_event *qev= get_qev_common(ev, event_size, orig_qev->ir); if (!qev) return NULL; qev->rgi_q= orig_qev->rgi_q; - safe_strcpy(qev->event_relay_log_name, sizeof(qev->event_relay_log_name), - relay_log_name); qev->event_relay_log_pos= event_pos; qev->future_event_relay_log_pos= event_pos+event_size; - safe_strcpy(qev->future_event_master_log_name, - sizeof(qev->future_event_master_log_name), - orig_qev->future_event_master_log_name); + /* The relaylog and master binlog names are available in qev->ir. */ return qev; } @@ -3286,6 +3282,32 @@ rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev, !rev->is_relay_log_event() && !rli->is_in_group()) { + /* + We rotate the relaylog everytime we see a rotate event from the master. + So we should only ever see one master rotate event that causes the + master binlog name in inuse_relaylog to be set. + + When we reach the temporary EOF of the hot relay log however, we + generate an artificial rotate event, this can occur any number of times + and should not cause us to set the master name in inuse_relaylog again. + */ + if (!rev->is_hot_log_rotate()) + { + /* + The rotate event, if any, comes near the start of the relay log file + and contains the master binlog name of the events contained in this + relay log file. + + We overwrite here the original name set when the inuse_relaylog was + allocated at the last relaylog rotation. If the rotation was due to + this master rotate, then the previous value will be one behind and + we correct it here, and we rely on the rotate event coming before + any following events in the relaylog to be queued for the worker + threads. + */ + memcpy(rli->last_inuse_relaylog->master_binlog_name, + rev->new_log_ident, rev->ident_len+1); + } memcpy(rli->future_event_master_log_name, rev->new_log_ident, rev->ident_len+1); rli->notify_group_master_log_name_update(); @@ -3665,7 +3687,6 @@ rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev, /* Queue the event for processing. */ - qev->ir= rli->last_inuse_relaylog; qev->ir->queued_count.fetch_add(1, std::memory_order_relaxed); cur_thread->enqueue(qev); unlock_or_exit_cond(rli->sql_driver_thd, &cur_thread->LOCK_rpl_thread, diff --git a/sql/rpl_parallel.h b/sql/rpl_parallel.h index 653f50a36ce..16ac51c619d 100644 --- a/sql/rpl_parallel.h +++ b/sql/rpl_parallel.h @@ -153,8 +153,6 @@ struct rpl_parallel_thread { rgi_queued_part *rgi_q; inuse_relaylog *ir; ulonglong future_event_relay_log_pos; - char event_relay_log_name[FN_REFLEN]; - char future_event_master_log_name[FN_REFLEN]; ulonglong event_relay_log_pos; my_off_t future_event_master_log_pos; size_t event_size; @@ -233,7 +231,8 @@ struct rpl_parallel_thread { queued_size-= dequeue_size; } - queued_event *get_qev_common(Log_event *ev, ulonglong event_size); + queued_event *get_qev_common(Log_event *ev, ulonglong event_size, + inuse_relaylog *ir); queued_event *get_qev(Log_event *ev, ulonglong event_size, Relay_log_info *rli); queued_event *retry_get_qev(Log_event *ev, queued_event *orig_qev, diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index 0ce77b80f62..7a553264543 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -1428,7 +1428,8 @@ bool Relay_log_info::stmt_done(my_off_t event_master_log_pos, THD *thd, int -Relay_log_info::alloc_inuse_relaylog(const char *name) +Relay_log_info::alloc_inuse_relaylog(const char *relaylog_name, + const char *master_binlog_name) { inuse_relaylog *ir; uint32 gtid_count; @@ -1441,7 +1442,8 @@ Relay_log_info::alloc_inuse_relaylog(const char *name) my_error(ER_OUTOFMEMORY, MYF(0), (int)sizeof(*gtid_list)*gtid_count); return 1; } - if (!(ir= new inuse_relaylog(this, gtid_list, gtid_count, name))) + if (!(ir= new inuse_relaylog(this, gtid_list, gtid_count, relaylog_name, + master_binlog_name))) { my_free(gtid_list); my_error(ER_OUTOFMEMORY, MYF(0), (int) sizeof(*ir)); diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index f8d37c6c2bd..c6fccd175c0 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -490,7 +490,8 @@ public: SLAVE STATUS. */ bool stmt_done(my_off_t event_log_pos, THD *thd, rpl_group_info *rgi); - int alloc_inuse_relaylog(const char *name); + int alloc_inuse_relaylog(const char *relaylog_name, + const char *master_binlog_name); void free_inuse_relaylog(inuse_relaylog *ir); void reset_inuse_relaylog(); int update_relay_log_state(rpl_gtid *gtid_list, uint32 count); @@ -684,16 +685,19 @@ struct inuse_relaylog { std::atomic dequeued_count; /* Set when all events have been read from a relaylog. */ std::atomic completed; - char name[FN_REFLEN]; + char relaylog_name[FN_REFLEN]; + char master_binlog_name[FN_REFLEN]; inuse_relaylog(Relay_log_info *rli_arg, rpl_gtid *relay_log_state_arg, uint32 relay_log_state_count_arg, - const char *name_arg): + const char *relaylog_name_arg, + const char *master_binlog_name_arg): next(0), rli(rli_arg), relay_log_state(relay_log_state_arg), relay_log_state_count(relay_log_state_count_arg), queued_count(0), dequeued_count(0), completed(false) { - strmake_buf(name, name_arg); + strmake_buf(relaylog_name, relaylog_name_arg); + strmake_buf(master_binlog_name, master_binlog_name_arg); } }; diff --git a/sql/slave.cc b/sql/slave.cc index b6e48b0fea4..11bcb5529ab 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -5374,7 +5374,8 @@ pthread_handler_t handle_slave_sql(void *arg) goto err_before_start; } rli->reset_inuse_relaylog(); - if (rli->alloc_inuse_relaylog(rli->group_relay_log_name)) + if (rli->alloc_inuse_relaylog(rli->group_relay_log_name, + rli->group_master_log_name)) { mysql_mutex_unlock(&rli->data_lock); goto err_before_start; @@ -5686,7 +5687,7 @@ pthread_handler_t handle_slave_sql(void *arg) ++gtid; --count; } - strmake_buf(rli->group_relay_log_name, ir->name); + strmake_buf(rli->group_relay_log_name, ir->relaylog_name); rli->group_relay_log_pos= BIN_LOG_HEADER_SIZE; rli->relay_log_state.load(ir->relay_log_state, ir->relay_log_state_count); } @@ -7463,19 +7464,20 @@ static Log_event* next_event(rpl_group_info *rgi, ulonglong *event_size) { /* We generate and return a Rotate, to make our positions advance */ DBUG_PRINT("info",("seeing an ignored end segment")); - ev= new Rotate_log_event(rli->ign_master_log_name_end, + Rotate_log_event *rev= + new Rotate_log_event(rli->ign_master_log_name_end, 0, rli->ign_master_log_pos_end, Rotate_log_event::DUP_NAME); rli->ign_master_log_name_end[0]= 0; mysql_mutex_unlock(log_lock); - if (unlikely(!ev)) + if (unlikely(!rev)) { errmsg= "Slave SQL thread failed to create a Rotate event " "(out of memory?), SHOW SLAVE STATUS may be inaccurate"; goto err; } - ev->server_id= 0; // don't be ignored by slave SQL thread - DBUG_RETURN(ev); + rev->set_is_hot_log_rotate(); // don't be ignored by slave SQL thread + DBUG_RETURN(rev); } if (rli->ign_gtids.count() && !rli->is_in_group()) @@ -7646,6 +7648,9 @@ static Log_event* next_event(rpl_group_info *rgi, ulonglong *event_size) To guard against this, we need to have LOCK_log. */ + const char *master_binlog_name= + (rli->mi->using_parallel() ? + rli->future_event_master_log_name : rli->group_master_log_name); DBUG_PRINT("info",("hot_log: %d",hot_log)); if (!hot_log) /* if hot_log, we already have this mutex */ mysql_mutex_lock(log_lock); @@ -7722,7 +7727,8 @@ static Log_event* next_event(rpl_group_info *rgi, ulonglong *event_size) mysql_mutex_unlock(log_lock); goto err; } - if (rli->alloc_inuse_relaylog(rli->linfo.log_file_name)) + if (rli->alloc_inuse_relaylog(rli->linfo.log_file_name, + master_binlog_name)) { if (!hot_log) mysql_mutex_unlock(log_lock); @@ -7743,7 +7749,8 @@ static Log_event* next_event(rpl_group_info *rgi, ulonglong *event_size) if ((rli->cur_log_fd=open_binlog(cur_log,rli->linfo.log_file_name, &errmsg)) <0) goto err; - if (rli->alloc_inuse_relaylog(rli->linfo.log_file_name)) + if (rli->alloc_inuse_relaylog(rli->linfo.log_file_name, + master_binlog_name)) goto err; } else