[MYSQL] error日志文件大量的Multi-threaded slave: Coordinator has waited磁盘占满了

2025-05-28T17:36:06.790418+08:00 34 [Note] Multi-threaded slave: Coordinator has waited 4881 times hitting slave_pending_jobs_size_max; current event size = 8205.
2025-05-28T17:36:06.797475+08:00 34 [Note] Multi-threaded slave: Coordinator has waited 4891 times hitting slave_pending_jobs_size_max; current event size = 8205.
2025-05-28T17:36:06.804947+08:00 34 [Note] Multi-threaded slave: Coordinator has waited 4901 times hitting slave_pending_jobs_size_max; current event size = 8205.

// sql/sys_vars.cc
static Sys_var_ulonglong Sys_mts_pending_jobs_size_max(
       "slave_pending_jobs_size_max",
       "Max size of Slave Worker queues holding yet not applied events."
       "The least possible value must be not less than the master side "
       "max_allowed_packet.",
       GLOBAL_VAR(opt_mts_pending_jobs_size_max), CMD_LINE(REQUIRED_ARG),
       VALID_RANGE(1024, (ulonglong)~(intptr)0), DEFAULT(16 * 1024*1024),
       BLOCK_SIZE(1024), ON_CHECK(0));

// sql/rpl_slave.cc
  // length of WQ is actually constant though can be made configurable
  rli->mts_slave_worker_queue_len_max= mts_slave_worker_queue_len_max;
  rli->mts_pending_jobs_size= 0;
  rli->mts_pending_jobs_size_max= ::opt_mts_pending_jobs_size_max; // 这个参数
  rli->mts_wq_underrun_w_id= MTS_WORKER_UNDEF;
  rli->mts_wq_excess_cnt= 0;
  rli->mts_wq_overrun_cnt= 0;
  rli->mts_wq_oversize= FALSE;
  rli->mts_coordinator_basic_nap= mts_coordinator_basic_nap;
  rli->mts_worker_underrun_level= mts_worker_underrun_level;
  rli->curr_group_seen_begin= rli->curr_group_seen_gtid= false;
  rli->curr_group_isolated= FALSE;
  rli->checkpoint_seqno= 0;
  rli->mts_last_online_stat= my_time(0);
  rli->mts_group_status= Relay_log_info::MTS_NOT_IN_GROUP;

// sql/rpl_rli_pdb.cc
size_t ev_size= job_item->data->common_header->data_written;
new_pend_size= rli->mts_pending_jobs_size + ev_size;
bool big_event= (ev_size > rli->mts_pending_jobs_size_max);
  while ( (!big_event && new_pend_size > rli->mts_pending_jobs_size_max)
          || (big_event && rli->mts_pending_jobs_size != 0 ))
  {
    rli->mts_wq_oversize= TRUE;
    rli->wq_size_waits_cnt++; // waiting due to the total size
    thd->ENTER_COND(&rli->pending_jobs_cond, &rli->pending_jobs_lock,
                    &stage_slave_waiting_worker_to_free_events, &old_stage);
    mysql_cond_wait(&rli->pending_jobs_cond, &rli->pending_jobs_lock);
    mysql_mutex_unlock(&rli->pending_jobs_lock);
    thd->EXIT_COND(&old_stage);
    if (thd->killed)
      return true;
    if (rli->wq_size_waits_cnt % 10 == 1)
      sql_print_information("Multi-threaded slave: Coordinator has waited "
                            "%lu times hitting slave_pending_jobs_size_max; "
                            "current event size = %zu.",
                            rli->wq_size_waits_cnt, ev_size);
    mysql_mutex_lock(&rli->pending_jobs_lock);

    new_pend_size= rli->mts_pending_jobs_size + ev_size;
  }

-- 从库修改slave_pending_jobs_size_max=1024并重启复制进程
set global slave_pending_jobs_size_max=1024;
stop slave;
start slave;

-- 主库随便来点超过1024字节的event
delete from sbtest5 limit 10000;

#!/usr/bin/env python
import struct
import time
import sys
import os

def print_event_info(filename):
	f = open(filename,'rb')
	if f.read(4) != b'\xfebin':
		f.seek(0,0)
	filesize = os.path.getsize(filename)
	last_offset = f.tell()
	current_offset = f.tell()
	while True:
		last_offset = current_offset
		t = struct.unpack('<LBLLLH',f.read(19))
		current_offset = t[4]
		print(f'TIME:{time.strftime("%Y%m%d %H:%M:%S", time.localtime(t[0]))} TYPE:{t[1]} SERVER_ID:{t[2]} SIZE:{t[3]} LOG_POS:{t[4]} FLAGS:{t[5]} OFFSET:{last_offset} --> {current_offset}')
		if t[4] >= filesize:
			break
		f.seek(t[4],0)
	f.close()

if len(sys.argv) != 2:
	print(f'USAGE: python {sys.argv[0]} mysql-bin.0000x')
	sys.exit(1)
filename = sys.argv[1]
if not os.path.exists(filename):
	print(f"{filename} not exists")
	sys.exit(2)

print_event_info(filename)

mysql -e "flush error logs"
mv mysql.err mysql.err.to_rm
mysql -e "flush error logs"
rm mysql.err.to_rm

gdb -p `pdiof mysqld`
gdb> break append_item_to_jobs
gdb> continue

# 主库触发大事务

gdb> info threads