database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2026-10-23 18:49:29 +03:00
Code Activity

MDEV-41217: Inefficient memory management by parallel slave

Browse Source 
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 <knielsen@knielsen-hq.org>
This commit is contained in:
Kristian Nielsen
2026-10-12 20:31:36 +02:00
parent 6f60728075
commit 6e55da9f49
7 changed files with 82 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
sql/log.cc
View File

@@ -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

7
sql/log_event.h
View File

@@ -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:

65
sql/rpl_parallel.cc
View File

@@ -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,

5
sql/rpl_parallel.h
View File

@@ -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,

6
sql/rpl_rli.cc
View File

@@ -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));

12
sql/rpl_rli.h
View File

@@ -490,7 +490,8 @@ public:
SLAVE STATUS</code>.
*/
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<int64> dequeued_count;
/* Set when all events have been read from a relaylog. */
std::atomic<bool> 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);
}
};

23
sql/slave.cc
View File

@@ -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