mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-16 03:52:35 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions
mariadb/sql/semisync_master.cc
Robin Newhouse dc38d8ea80 Minimize unsafe C functions with safe_strcpy() 
Similar to #2480.
567b681 introduced safe_strcpy() to minimize the use of C with
potentially unsafe memory overflow with strcpy() whose use is
discouraged.
Replace instances of strcpy() with safe_strcpy() where possible, limited
here to files in the `sql/` directory.

All new code of the whole pull request, including one or several files
that are either new files or modified ones, are contributed under the
BSD-new license. I am contributing on behalf of my employer
Amazon Web Services, Inc.
2024-05-17 13:33:16 +01:00

1420 lines
41 KiB
C++
Raw Blame History

/* Copyright (C) 2007 Google Inc.
Copyright (c) 2008, 2013, Oracle and/or its affiliates.
Copyright (c) 2011, 2022, MariaDB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; version 2 of the License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA */
#include <my_global.h>
#include "semisync_master.h"
#include <algorithm>
#include <mysql_com.h>
#define TIME_THOUSAND 1000
#define TIME_MILLION 1000000
#define TIME_BILLION 1000000000
/* This indicates whether semi-synchronous replication is enabled. */
my_bool rpl_semi_sync_master_enabled= 0;
unsigned long long rpl_semi_sync_master_request_ack = 0;
unsigned long long rpl_semi_sync_master_get_ack = 0;
my_bool rpl_semi_sync_master_wait_no_slave = 1;
my_bool rpl_semi_sync_master_status = 0;
ulong rpl_semi_sync_master_wait_point =
SEMI_SYNC_MASTER_WAIT_POINT_AFTER_STORAGE_COMMIT;
ulong rpl_semi_sync_master_timeout;
ulong rpl_semi_sync_master_trace_level;
ulong rpl_semi_sync_master_yes_transactions = 0;
ulong rpl_semi_sync_master_no_transactions = 0;
ulong rpl_semi_sync_master_off_times = 0;
ulong rpl_semi_sync_master_timefunc_fails = 0;
ulong rpl_semi_sync_master_wait_timeouts = 0;
ulong rpl_semi_sync_master_wait_sessions = 0;
ulong rpl_semi_sync_master_wait_pos_backtraverse = 0;
ulong rpl_semi_sync_master_avg_trx_wait_time = 0;
ulonglong rpl_semi_sync_master_trx_wait_num = 0;
ulong rpl_semi_sync_master_avg_net_wait_time = 0;
ulonglong rpl_semi_sync_master_net_wait_num = 0;
ulong rpl_semi_sync_master_clients = 0;
ulonglong rpl_semi_sync_master_net_wait_time = 0;
ulonglong rpl_semi_sync_master_trx_wait_time = 0;
Repl_semi_sync_master repl_semisync_master;
Ack_receiver ack_receiver;
/*
structure to save transaction log filename and position
*/
typedef struct Trans_binlog_info {
my_off_t log_pos;
char log_file[FN_REFLEN];
} Trans_binlog_info;
static int get_wait_time(const struct timespec& start_ts);
static ulonglong timespec_to_usec(const struct timespec *ts)
{
return (ulonglong) ts->tv_sec * TIME_MILLION + ts->tv_nsec / TIME_THOUSAND;
}
/*******************************************************************************
*
* <Active_tranx> class : manage all active transaction nodes
*
******************************************************************************/
Active_tranx::Active_tranx(mysql_mutex_t *lock,
ulong trace_level)
: Trace(trace_level), m_allocator(max_connections),
m_num_entries(max_connections << 1), /* Transaction hash table size
* is set to double the size
* of max_connections */
m_lock(lock)
{
/* No transactions are in the list initially. */
m_trx_front = NULL;
m_trx_rear = NULL;
/* Create the hash table to find a transaction's ending event. */
m_trx_htb = new Tranx_node *[m_num_entries];
for (int idx = 0; idx < m_num_entries; ++idx)
m_trx_htb[idx] = NULL;
sql_print_information("Semi-sync replication initialized for transactions.");
}
Active_tranx::~Active_tranx()
{
delete [] m_trx_htb;
m_trx_htb = NULL;
m_num_entries = 0;
}
unsigned int Active_tranx::calc_hash(const unsigned char *key, size_t length)
{
unsigned int nr = 1, nr2 = 4;
/* The hash implementation comes from calc_hashnr() in mysys/hash.c. */
while (length--)
{
nr ^= (((nr & 63)+nr2)*((unsigned int) (unsigned char) *key++))+ (nr << 8);
nr2 += 3;
}
return((unsigned int) nr);
}
unsigned int Active_tranx::get_hash_value(const char *log_file_name,
my_off_t log_file_pos)
{
unsigned int hash1 = calc_hash((const unsigned char *)log_file_name,
strlen(log_file_name));
unsigned int hash2 = calc_hash((const unsigned char *)(&log_file_pos),
sizeof(log_file_pos));
return (hash1 + hash2) % m_num_entries;
}
int Active_tranx::compare(const char *log_file_name1, my_off_t log_file_pos1,
const char *log_file_name2, my_off_t log_file_pos2)
{
int cmp = strcmp(log_file_name1, log_file_name2);
if (cmp != 0)
return cmp;
if (log_file_pos1 > log_file_pos2)
return 1;
else if (log_file_pos1 < log_file_pos2)
return -1;
return 0;
}
int Active_tranx::insert_tranx_node(const char *log_file_name,
my_off_t log_file_pos)
{
Tranx_node *ins_node;
int result = 0;
unsigned int hash_val;
DBUG_ENTER("Active_tranx:insert_tranx_node");
ins_node = m_allocator.allocate_node();
if (!ins_node)
{
sql_print_error("%s: transaction node allocation failed for: (%s, %lu)",
"Active_tranx:insert_tranx_node",
log_file_name, (ulong)log_file_pos);
result = -1;
goto l_end;
}
/* insert the binlog position in the active transaction list. */
strncpy(ins_node->log_name, log_file_name, FN_REFLEN-1);
ins_node->log_name[FN_REFLEN-1] = 0; /* make sure it ends properly */
ins_node->log_pos = log_file_pos;
if (!m_trx_front)
{
/* The list is empty. */
m_trx_front = m_trx_rear = ins_node;
}
else
{
int cmp = compare(ins_node, m_trx_rear);
if (cmp > 0)
{
/* Compare with the tail first. If the transaction happens later in
* binlog, then make it the new tail.
*/
m_trx_rear->next = ins_node;
m_trx_rear = ins_node;
}
else
{
/* Otherwise, it is an error because the transaction should hold the
* mysql_bin_log.LOCK_log when appending events.
*/
sql_print_error("%s: binlog write out-of-order, tail (%s, %lu), "
"new node (%s, %lu)", "Active_tranx:insert_tranx_node",
m_trx_rear->log_name, (ulong)m_trx_rear->log_pos,
ins_node->log_name, (ulong)ins_node->log_pos);
result = -1;
goto l_end;
}
}
hash_val = get_hash_value(ins_node->log_name, ins_node->log_pos);
ins_node->hash_next = m_trx_htb[hash_val];
m_trx_htb[hash_val] = ins_node;
DBUG_PRINT("semisync", ("%s: insert (%s, %lu) in entry(%u)",
"Active_tranx:insert_tranx_node",
ins_node->log_name, (ulong)ins_node->log_pos,
hash_val));
l_end:
DBUG_RETURN(result);
}
bool Active_tranx::is_tranx_end_pos(const char *log_file_name,
my_off_t log_file_pos)
{
DBUG_ENTER("Active_tranx::is_tranx_end_pos");
unsigned int hash_val = get_hash_value(log_file_name, log_file_pos);
Tranx_node *entry = m_trx_htb[hash_val];
while (entry != NULL)
{
if (compare(entry, log_file_name, log_file_pos) == 0)
break;
entry = entry->hash_next;
}
DBUG_PRINT("semisync", ("%s: probe (%s, %lu) in entry(%u)",
"Active_tranx::is_tranx_end_pos",
log_file_name, (ulong)log_file_pos, hash_val));
DBUG_RETURN(entry != NULL);
}
void Active_tranx::clear_active_tranx_nodes(const char *log_file_name,
my_off_t log_file_pos)
{
Tranx_node *new_front;
DBUG_ENTER("Active_tranx::::clear_active_tranx_nodes");
if (log_file_name != NULL)
{
new_front = m_trx_front;
while (new_front)
{
if (compare(new_front, log_file_name, log_file_pos) > 0)
break;
new_front = new_front->next;
}
}
else
{
/* If log_file_name is NULL, clear everything. */
new_front = NULL;
}
if (new_front == NULL)
{
/* No active transaction nodes after the call. */
/* Clear the hash table. */
memset(m_trx_htb, 0, m_num_entries * sizeof(Tranx_node *));
m_allocator.free_all_nodes();
/* Clear the active transaction list. */
if (m_trx_front != NULL)
{
m_trx_front = NULL;
m_trx_rear = NULL;
}
DBUG_PRINT("semisync", ("%s: cleared all nodes",
"Active_tranx::::clear_active_tranx_nodes"));
}
else if (new_front != m_trx_front)
{
Tranx_node *curr_node, *next_node;
/* Delete all transaction nodes before the confirmation point. */
#ifdef DBUG_TRACE
int n_frees = 0;
#endif
curr_node = m_trx_front;
while (curr_node != new_front)
{
next_node = curr_node->next;
#ifdef DBUG_TRACE
n_frees++;
#endif
/* Remove the node from the hash table. */
unsigned int hash_val = get_hash_value(curr_node->log_name, curr_node->log_pos);
Tranx_node **hash_ptr = &(m_trx_htb[hash_val]);
while ((*hash_ptr) != NULL)
{
if ((*hash_ptr) == curr_node)
{
(*hash_ptr) = curr_node->hash_next;
break;
}
hash_ptr = &((*hash_ptr)->hash_next);
}
curr_node = next_node;
}
m_trx_front = new_front;
m_allocator.free_nodes_before(m_trx_front);
DBUG_PRINT("semisync", ("%s: cleared %d nodes back until pos (%s, %lu)",
"Active_tranx::::clear_active_tranx_nodes",
n_frees,
m_trx_front->log_name, (ulong)m_trx_front->log_pos));
}
DBUG_VOID_RETURN;
}
/*******************************************************************************
*
* <Repl_semi_sync_master> class: the basic code layer for semisync master.
* <Repl_semi_sync_slave> class: the basic code layer for semisync slave.
*
* The most important functions during semi-syn replication listed:
*
* Master:
* . report_reply_binlog(): called by the binlog dump thread when it receives
* the slave's status information.
* . update_sync_header(): based on transaction waiting information, decide
* whether to request the slave to reply.
* . write_tranx_in_binlog(): called by the transaction thread when it finishes
* writing all transaction events in binlog.
* . commit_trx(): transaction thread wait for the slave reply.
*
* Slave:
* . slave_read_sync_header(): read the semi-sync header from the master, get
* the sync status and get the payload for events.
* . slave_reply(): reply to the master about the replication progress.
*
******************************************************************************/
Repl_semi_sync_master::Repl_semi_sync_master()
: m_active_tranxs(NULL),
m_init_done(false),
m_reply_file_name_inited(false),
m_reply_file_pos(0L),
m_wait_file_name_inited(false),
m_wait_file_pos(0),
m_master_enabled(false),
m_wait_timeout(0L),
m_state(0),
m_wait_point(0)
{
m_reply_file_name[0]= '\0';
m_wait_file_name[0]= '\0';
}
int Repl_semi_sync_master::init_object()
{
int result= 0;
m_init_done = true;
/* References to the parameter works after set_options(). */
set_wait_timeout(rpl_semi_sync_master_timeout);
set_trace_level(rpl_semi_sync_master_trace_level);
set_wait_point(rpl_semi_sync_master_wait_point);
/* Mutex initialization can only be done after MY_INIT(). */
mysql_mutex_init(key_LOCK_rpl_semi_sync_master_enabled,
&LOCK_rpl_semi_sync_master_enabled, MY_MUTEX_INIT_FAST);
mysql_mutex_init(key_LOCK_binlog,
&LOCK_binlog, MY_MUTEX_INIT_FAST);
mysql_cond_init(key_COND_binlog_send,
&COND_binlog_send, NULL);
if (rpl_semi_sync_master_enabled)
{
result = enable_master();
if (!result)
{
result= ack_receiver.start(); /* Start the ACK thread. */
/*
If rpl_semi_sync_master_wait_no_slave is disabled, let's temporarily
switch off semisync to avoid hang if there's none active slave.
*/
if (!rpl_semi_sync_master_wait_no_slave)
switch_off();
}
}
else
{
disable_master();
}
return result;
}
int Repl_semi_sync_master::enable_master()
{
int result = 0;
/* Must have the lock when we do enable of disable. */
lock();
if (!get_master_enabled())
{
m_active_tranxs = new Active_tranx(&LOCK_binlog, m_trace_level);
if (m_active_tranxs != NULL)
{
m_commit_file_name_inited = false;
m_reply_file_name_inited = false;
m_wait_file_name_inited = false;
set_master_enabled(true);
m_state = true;
sql_print_information("Semi-sync replication enabled on the master.");
}
else
{
sql_print_error("Cannot allocate memory to enable semi-sync on the master.");
result = -1;
}
}
unlock();
return result;
}
void Repl_semi_sync_master::disable_master()
{
/* Must have the lock when we do enable of disable. */
lock();
if (get_master_enabled())
{
/* Switch off the semi-sync first so that waiting transaction will be
* waken up.
*/
switch_off();
assert(m_active_tranxs != NULL);
delete m_active_tranxs;
m_active_tranxs = NULL;
m_reply_file_name_inited = false;
m_wait_file_name_inited = false;
m_commit_file_name_inited = false;
set_master_enabled(false);
sql_print_information("Semi-sync replication disabled on the master.");
}
unlock();
}
void Repl_semi_sync_master::cleanup()
{
if (m_init_done)
{
mysql_mutex_destroy(&LOCK_rpl_semi_sync_master_enabled);
mysql_mutex_destroy(&LOCK_binlog);
mysql_cond_destroy(&COND_binlog_send);
m_init_done= 0;
}
delete m_active_tranxs;
}
int Repl_semi_sync_master::sync_get_master_wait_sessions()
{
int wait_sessions;
lock();
wait_sessions= rpl_semi_sync_master_wait_sessions;
unlock();
return wait_sessions;
}
void Repl_semi_sync_master::create_timeout(struct timespec *out,
struct timespec *start_arg)
{
struct timespec *start_ts;
struct timespec now_ts;
if (!start_arg)
{
set_timespec(now_ts, 0);
start_ts= &now_ts;
}
else
{
start_ts= start_arg;
}
long diff_secs= (long) (m_wait_timeout / TIME_THOUSAND);
long diff_nsecs= (long) ((m_wait_timeout % TIME_THOUSAND) * TIME_MILLION);
long nsecs= start_ts->tv_nsec + diff_nsecs;
out->tv_sec= start_ts->tv_sec + diff_secs + nsecs / TIME_BILLION;
out->tv_nsec= nsecs % TIME_BILLION;
}
void Repl_semi_sync_master::lock()
{
mysql_mutex_lock(&LOCK_binlog);
}
void Repl_semi_sync_master::unlock()
{
mysql_mutex_unlock(&LOCK_binlog);
}
void Repl_semi_sync_master::cond_broadcast()
{
mysql_cond_broadcast(&COND_binlog_send);
}
int Repl_semi_sync_master::cond_timewait(struct timespec *wait_time)
{
int wait_res;
DBUG_ENTER("Repl_semi_sync_master::cond_timewait()");
wait_res= mysql_cond_timedwait(&COND_binlog_send,
&LOCK_binlog, wait_time);
DBUG_RETURN(wait_res);
}
void Repl_semi_sync_master::add_slave()
{
lock();
rpl_semi_sync_master_clients++;
unlock();
}
void Repl_semi_sync_master::remove_slave()
{
lock();
rpl_semi_sync_master_clients--;
/* Only switch off if semi-sync is enabled and is on */
if (get_master_enabled() && is_on())
{
/* If user has chosen not to wait if no semi-sync slave available
and the last semi-sync slave exits, turn off semi-sync on master
immediately.
*/
if (!rpl_semi_sync_master_wait_no_slave &&
rpl_semi_sync_master_clients == 0)
switch_off();
}
unlock();
}
int Repl_semi_sync_master::report_reply_packet(uint32 server_id,
const uchar *packet,
ulong packet_len)
{
int result= -1;
char log_file_name[FN_REFLEN+1];
my_off_t log_file_pos;
ulong log_file_len = 0;
DBUG_ENTER("Repl_semi_sync_master::report_reply_packet");
DBUG_EXECUTE_IF("semisync_corrupt_magic",
const_cast<uchar*>(packet)[REPLY_MAGIC_NUM_OFFSET]= 0;);
if (unlikely(packet[REPLY_MAGIC_NUM_OFFSET] !=
Repl_semi_sync_master::k_packet_magic_num))
{
sql_print_error("Read semi-sync reply magic number error");
goto l_end;
}
if (unlikely(packet_len < REPLY_BINLOG_NAME_OFFSET))
{
sql_print_error("Read semi-sync reply length error: packet is too small");
goto l_end;
}
log_file_pos = uint8korr(packet + REPLY_BINLOG_POS_OFFSET);
log_file_len = packet_len - REPLY_BINLOG_NAME_OFFSET;
if (unlikely(log_file_len >= FN_REFLEN))
{
sql_print_error("Read semi-sync reply binlog file length too large");
goto l_end;
}
strncpy(log_file_name, (const char*)packet + REPLY_BINLOG_NAME_OFFSET, log_file_len);
log_file_name[log_file_len] = 0;
DBUG_ASSERT(dirname_length(log_file_name) == 0);
DBUG_PRINT("semisync", ("%s: Got reply(%s, %lu) from server %u",
"Repl_semi_sync_master::report_reply_packet",
log_file_name, (ulong)log_file_pos, server_id));
rpl_semi_sync_master_get_ack++;
report_reply_binlog(server_id, log_file_name, log_file_pos);
result= 0;
l_end:
if (result == -1)
{
char buf[256];
octet2hex(buf, (const char*) packet, std::min(static_cast<ulong>(sizeof(buf)-1),
packet_len));
sql_print_information("First bytes of the packet from semisync slave "
"server-id %d: %s", server_id, buf);
}
DBUG_RETURN(result);
}
int Repl_semi_sync_master::report_reply_binlog(uint32 server_id,
const char *log_file_name,
my_off_t log_file_pos)
{
int cmp;
bool can_release_threads = false;
bool need_copy_send_pos = true;
DBUG_ENTER("Repl_semi_sync_master::report_reply_binlog");
if (!(get_master_enabled()))
DBUG_RETURN(0);
lock();
/* This is the real check inside the mutex. */
if (!get_master_enabled())
goto l_end;
if (!is_on())
/* We check to see whether we can switch semi-sync ON. */
try_switch_on(server_id, log_file_name, log_file_pos);
/* The position should increase monotonically, if there is only one
* thread sending the binlog to the slave.
* In reality, to improve the transaction availability, we allow multiple
* sync replication slaves. So, if any one of them get the transaction,
* the transaction session in the primary can move forward.
*/
if (m_reply_file_name_inited)
{
cmp = Active_tranx::compare(log_file_name, log_file_pos,
m_reply_file_name, m_reply_file_pos);
/* If the requested position is behind the sending binlog position,
* would not adjust sending binlog position.
* We based on the assumption that there are multiple semi-sync slave,
* and at least one of them shou/ld be up to date.
* If all semi-sync slaves are behind, at least initially, the primary
* can find the situation after the waiting timeout. After that, some
* slaves should catch up quickly.
*/
if (cmp < 0)
{
/* If the position is behind, do not copy it. */
need_copy_send_pos = false;
}
}
if (need_copy_send_pos)
{
strmake_buf(m_reply_file_name, log_file_name);
m_reply_file_pos = log_file_pos;
m_reply_file_name_inited = true;
/* Remove all active transaction nodes before this point. */
assert(m_active_tranxs != NULL);
m_active_tranxs->clear_active_tranx_nodes(log_file_name, log_file_pos);
DBUG_PRINT("semisync", ("%s: Got reply at (%s, %lu)",
"Repl_semi_sync_master::report_reply_binlog",
log_file_name, (ulong)log_file_pos));
}
if (rpl_semi_sync_master_wait_sessions > 0)
{
/* Let us check if some of the waiting threads doing a trx
* commit can now proceed.
*/
cmp = Active_tranx::compare(m_reply_file_name, m_reply_file_pos,
m_wait_file_name, m_wait_file_pos);
if (cmp >= 0)
{
/* Yes, at least one waiting thread can now proceed:
* let us release all waiting threads with a broadcast
*/
can_release_threads = true;
m_wait_file_name_inited = false;
}
}
l_end:
unlock();
if (can_release_threads)
{
DBUG_PRINT("semisync", ("%s: signal all waiting threads.",
"Repl_semi_sync_master::report_reply_binlog"));
cond_broadcast();
}
DBUG_RETURN(0);
}
int Repl_semi_sync_master::wait_after_sync(const char *log_file, my_off_t log_pos)
{
if (!get_master_enabled())
return 0;
int ret= 0;
if(log_pos &&
wait_point() == SEMI_SYNC_MASTER_WAIT_POINT_AFTER_BINLOG_SYNC)
ret= commit_trx(log_file + dirname_length(log_file), log_pos);
return ret;
}
int Repl_semi_sync_master::wait_after_commit(THD* thd, bool all)
{
if (!get_master_enabled())
return 0;
int ret= 0;
const char *log_file;
my_off_t log_pos;
bool is_real_trans=
(all || thd->transaction->all.ha_list == 0);
/*
The coordinates are propagated to this point having been computed
in report_binlog_update
*/
Trans_binlog_info *log_info= thd->semisync_info;
log_file= log_info && log_info->log_file[0] ? log_info->log_file : 0;
log_pos= log_info ? log_info->log_pos : 0;
DBUG_ASSERT(!log_file || dirname_length(log_file) == 0);
if (is_real_trans &&
log_pos &&
wait_point() == SEMI_SYNC_MASTER_WAIT_POINT_AFTER_STORAGE_COMMIT)
ret= commit_trx(log_file, log_pos);
if (is_real_trans && log_info)
{
log_info->log_file[0]= 0;
log_info->log_pos= 0;
}
return ret;
}
int Repl_semi_sync_master::wait_after_rollback(THD *thd, bool all)
{
return wait_after_commit(thd, all);
}
/**
The method runs after flush to binary log is done.
*/
int Repl_semi_sync_master::report_binlog_update(THD* thd, const char *log_file,
my_off_t log_pos)
{
if (get_master_enabled())
{
Trans_binlog_info *log_info;
if (!(log_info= thd->semisync_info))
{
if(!(log_info= (Trans_binlog_info*)my_malloc(PSI_INSTRUMENT_ME,
sizeof(Trans_binlog_info), MYF(0))))
return 1;
thd->semisync_info= log_info;
}
safe_strcpy(log_info->log_file, sizeof(log_info->log_file),
log_file + dirname_length(log_file));
log_info->log_pos = log_pos;
return write_tranx_in_binlog(log_info->log_file, log_pos);
}
return 0;
}
int Repl_semi_sync_master::dump_start(THD* thd,
const char *log_file,
my_off_t log_pos)
{
if (!thd->semi_sync_slave)
return 0;
if (ack_receiver.add_slave(thd))
{
sql_print_error("Failed to register slave to semi-sync ACK receiver "
"thread. Turning off semisync");
thd->semi_sync_slave= 0;
return 1;
}
add_slave();
report_reply_binlog(thd->variables.server_id,
log_file + dirname_length(log_file), log_pos);
sql_print_information("Start semi-sync binlog_dump to slave "
"(server_id: %ld), pos(%s, %lu)",
(long) thd->variables.server_id, log_file,
(ulong) log_pos);
return 0;
}
void Repl_semi_sync_master::dump_end(THD* thd)
{
if (!thd->semi_sync_slave)
return;
sql_print_information("Stop semi-sync binlog_dump to slave (server_id: %ld)",
(long) thd->variables.server_id);
remove_slave();
ack_receiver.remove_slave(thd);
}
int Repl_semi_sync_master::commit_trx(const char* trx_wait_binlog_name,
my_off_t trx_wait_binlog_pos)
{
DBUG_ENTER("Repl_semi_sync_master::commit_trx");
if (get_master_enabled() && trx_wait_binlog_name)
{
struct timespec start_ts;
struct timespec abstime;
int wait_result;
PSI_stage_info old_stage;
THD *thd= current_thd;
set_timespec(start_ts, 0);
DEBUG_SYNC(thd, "rpl_semisync_master_commit_trx_before_lock");
/* Acquire the mutex. */
lock();
/* This must be called after acquired the lock */
THD_ENTER_COND(thd, &COND_binlog_send, &LOCK_binlog,
& stage_waiting_for_semi_sync_ack_from_slave,
& old_stage);
/* This is the real check inside the mutex. */
if (!get_master_enabled() || !is_on())
goto l_end;
DBUG_PRINT("semisync", ("%s: wait pos (%s, %lu), repl(%d)",
"Repl_semi_sync_master::commit_trx",
trx_wait_binlog_name, (ulong)trx_wait_binlog_pos,
(int)is_on()));
while (is_on() && !thd_killed(thd))
{
if (m_reply_file_name_inited)
{
int cmp = Active_tranx::compare(m_reply_file_name, m_reply_file_pos,
trx_wait_binlog_name,
trx_wait_binlog_pos);
if (cmp >= 0)
{
/* We have already sent the relevant binlog to the slave: no need to
* wait here.
*/
DBUG_PRINT("semisync", ("%s: Binlog reply is ahead (%s, %lu),",
"Repl_semi_sync_master::commit_trx",
m_reply_file_name,
(ulong)m_reply_file_pos));
break;
}
}
/* Let us update the info about the minimum binlog position of waiting
* threads.
*/
if (m_wait_file_name_inited)
{
int cmp = Active_tranx::compare(trx_wait_binlog_name,
trx_wait_binlog_pos,
m_wait_file_name, m_wait_file_pos);
if (cmp <= 0)
{
/* This thd has a lower position, let's update the minimum info. */
strmake_buf(m_wait_file_name, trx_wait_binlog_name);
m_wait_file_pos = trx_wait_binlog_pos;
rpl_semi_sync_master_wait_pos_backtraverse++;
DBUG_PRINT("semisync", ("%s: move back wait position (%s, %lu),",
"Repl_semi_sync_master::commit_trx",
m_wait_file_name, (ulong)m_wait_file_pos));
}
}
else
{
strmake_buf(m_wait_file_name, trx_wait_binlog_name);
m_wait_file_pos = trx_wait_binlog_pos;
m_wait_file_name_inited = true;
DBUG_PRINT("semisync", ("%s: init wait position (%s, %lu),",
"Repl_semi_sync_master::commit_trx",
m_wait_file_name, (ulong)m_wait_file_pos));
}
/* In semi-synchronous replication, we wait until the binlog-dump
* thread has received the reply on the relevant binlog segment from the
* replication slave.
*
* Let us suspend this thread to wait on the condition;
* when replication has progressed far enough, we will release
* these waiting threads.
*/
rpl_semi_sync_master_wait_sessions++;
/* We keep track of when this thread is awaiting an ack to ensure it is
* not killed while awaiting an ACK if a shutdown is issued.
*/
set_thd_awaiting_semisync_ack(thd, TRUE);
DBUG_PRINT("semisync", ("%s: wait %lu ms for binlog sent (%s, %lu)",
"Repl_semi_sync_master::commit_trx",
m_wait_timeout,
m_wait_file_name, (ulong)m_wait_file_pos));
create_timeout(&abstime, &start_ts);
wait_result = cond_timewait(&abstime);
set_thd_awaiting_semisync_ack(thd, FALSE);
rpl_semi_sync_master_wait_sessions--;
if (wait_result != 0)
{
/* This is a real wait timeout. */
sql_print_warning("Timeout waiting for reply of binlog (file: %s, pos: %lu), "
"semi-sync up to file %s, position %lu.",
trx_wait_binlog_name, (ulong)trx_wait_binlog_pos,
m_reply_file_name, (ulong)m_reply_file_pos);
rpl_semi_sync_master_wait_timeouts++;
/* switch semi-sync off */
switch_off();
}
else
{
int wait_time;
wait_time = get_wait_time(start_ts);
if (wait_time < 0)
{
DBUG_PRINT("semisync", ("Replication semi-sync getWaitTime fail at "
"wait position (%s, %lu)",
trx_wait_binlog_name,
(ulong)trx_wait_binlog_pos));
rpl_semi_sync_master_timefunc_fails++;
}
else
{
rpl_semi_sync_master_trx_wait_num++;
rpl_semi_sync_master_trx_wait_time += wait_time;
}
}
}
/*
At this point, the binlog file and position of this transaction
must have been removed from Active_tranx.
m_active_tranxs may be NULL if someone disabled semi sync during
cond_timewait()
*/
assert(thd_killed(thd) || !m_active_tranxs ||
!m_active_tranxs->is_tranx_end_pos(trx_wait_binlog_name,
trx_wait_binlog_pos));
l_end:
/* Update the status counter. */
if (is_on())
rpl_semi_sync_master_yes_transactions++;
else
rpl_semi_sync_master_no_transactions++;
/* The lock held will be released by thd_exit_cond, so no need to
call unlock() here */
THD_EXIT_COND(thd, &old_stage);
}
DBUG_RETURN(0);
}
/* Indicate that semi-sync replication is OFF now.
*
* What should we do when it is disabled? The problem is that we want
* the semi-sync replication enabled again when the slave catches up
* later. But, it is not that easy to detect that the slave has caught
* up. This is caused by the fact that MySQL's replication protocol is
* asynchronous, meaning that if the master does not use the semi-sync
* protocol, the slave would not send anything to the master.
* Still, if the master is sending (N+1)-th event, we assume that it is
* an indicator that the slave has received N-th event and earlier ones.
*
* If semi-sync is disabled, all transactions still update the wait
* position with the last position in binlog. But no transactions will
* wait for confirmations and the active transaction list would not be
* maintained. In binlog dump thread, update_sync_header() checks whether
* the current sending event catches up with last wait position. If it
* does match, semi-sync will be switched on again.
*/
void Repl_semi_sync_master::switch_off()
{
DBUG_ENTER("Repl_semi_sync_master::switch_off");
m_state = false;
/* Clear the active transaction list. */
assert(m_active_tranxs != NULL);
m_active_tranxs->clear_active_tranx_nodes(NULL, 0);
rpl_semi_sync_master_off_times++;
m_wait_file_name_inited = false;
m_reply_file_name_inited = false;
sql_print_information("Semi-sync replication switched OFF.");
cond_broadcast(); /* wake up all waiting threads */
DBUG_VOID_RETURN;
}
int Repl_semi_sync_master::try_switch_on(int server_id,
const char *log_file_name,
my_off_t log_file_pos)
{
bool semi_sync_on = false;
DBUG_ENTER("Repl_semi_sync_master::try_switch_on");
/* If the current sending event's position is larger than or equal to the
* 'largest' commit transaction binlog position, the slave is already
* catching up now and we can switch semi-sync on here.
* If m_commit_file_name_inited indicates there are no recent transactions,
* we can enable semi-sync immediately.
*/
if (m_commit_file_name_inited)
{
int cmp = Active_tranx::compare(log_file_name, log_file_pos,
m_commit_file_name, m_commit_file_pos);
semi_sync_on = (cmp >= 0);
}
else
{
semi_sync_on = true;
}
if (semi_sync_on)
{
/* Switch semi-sync replication on. */
m_state = true;
sql_print_information("Semi-sync replication switched ON with slave (server_id: %d) "
"at (%s, %lu)",
server_id, log_file_name,
(ulong)log_file_pos);
}
DBUG_RETURN(0);
}
int Repl_semi_sync_master::reserve_sync_header(String* packet)
{
DBUG_ENTER("Repl_semi_sync_master::reserve_sync_header");
/* Set the magic number and the sync status. By default, no sync
* is required.
*/
packet->append(reinterpret_cast<const char*>(k_sync_header),
sizeof(k_sync_header));
DBUG_RETURN(0);
}
int Repl_semi_sync_master::update_sync_header(THD* thd, unsigned char *packet,
const char *log_file_name,
my_off_t log_file_pos,
bool* need_sync)
{
int cmp = 0;
bool sync = false;
DBUG_ENTER("Repl_semi_sync_master::update_sync_header");
/* If the semi-sync master is not enabled, or the slave is not a semi-sync
* target, do not request replies from the slave.
*/
if (!get_master_enabled() || !thd->semi_sync_slave)
{
*need_sync = false;
DBUG_RETURN(0);
}
lock();
/* This is the real check inside the mutex. */
if (!get_master_enabled())
{
assert(sync == false);
goto l_end;
}
if (is_on())
{
/* semi-sync is ON */
sync = false; /* No sync unless a transaction is involved. */
if (m_reply_file_name_inited)
{
cmp = Active_tranx::compare(log_file_name, log_file_pos,
m_reply_file_name, m_reply_file_pos);
if (cmp <= 0)
{
/* If we have already got the reply for the event, then we do
* not need to sync the transaction again.
*/
goto l_end;
}
}
if (m_wait_file_name_inited)
{
cmp = Active_tranx::compare(log_file_name, log_file_pos,
m_wait_file_name, m_wait_file_pos);
}
else
{
cmp = 1;
}
/* If we are already waiting for some transaction replies which
* are later in binlog, do not wait for this one event.
*/
if (cmp >= 0)
{
/*
* We only wait if the event is a transaction's ending event.
*/
assert(m_active_tranxs != NULL);
sync = m_active_tranxs->is_tranx_end_pos(log_file_name,
log_file_pos);
}
}
else
{
if (m_commit_file_name_inited)
{
int cmp = Active_tranx::compare(log_file_name, log_file_pos,
m_commit_file_name, m_commit_file_pos);
sync = (cmp >= 0);
}
else
{
sync = true;
}
}
DBUG_PRINT("semisync", ("%s: server(%lu), (%s, %lu) sync(%d), repl(%d)",
"Repl_semi_sync_master::update_sync_header",
thd->variables.server_id, log_file_name,
(ulong)log_file_pos, sync, (int)is_on()));
*need_sync= sync;
l_end:
unlock();
/* We do not need to clear sync flag because we set it to 0 when we
* reserve the packet header.
*/
if (sync)
{
(packet)[2] = k_packet_flag_sync;
}
DBUG_RETURN(0);
}
int Repl_semi_sync_master::write_tranx_in_binlog(const char* log_file_name,
my_off_t log_file_pos)
{
int result = 0;
DBUG_ENTER("Repl_semi_sync_master::write_tranx_in_binlog");
lock();
/* This is the real check inside the mutex. */
if (!get_master_enabled())
goto l_end;
/* Update the 'largest' transaction commit position seen so far even
* though semi-sync is switched off.
* It is much better that we update m_commit_file* here, instead of
* inside commit_trx(). This is mostly because update_sync_header()
* will watch for m_commit_file* to decide whether to switch semi-sync
* on. The detailed reason is explained in function update_sync_header().
*/
if (m_commit_file_name_inited)
{
int cmp = Active_tranx::compare(log_file_name, log_file_pos,
m_commit_file_name, m_commit_file_pos);
if (cmp > 0)
{
/* This is a larger position, let's update the maximum info. */
strncpy(m_commit_file_name, log_file_name, FN_REFLEN-1);
m_commit_file_name[FN_REFLEN-1] = 0; /* make sure it ends properly */
m_commit_file_pos = log_file_pos;
}
}
else
{
strncpy(m_commit_file_name, log_file_name, FN_REFLEN-1);
m_commit_file_name[FN_REFLEN-1] = 0; /* make sure it ends properly */
m_commit_file_pos = log_file_pos;
m_commit_file_name_inited = true;
}
if (is_on())
{
assert(m_active_tranxs != NULL);
if(m_active_tranxs->insert_tranx_node(log_file_name, log_file_pos))
{
/*
if insert tranx_node failed, print a warning message
and turn off semi-sync
*/
sql_print_warning("Semi-sync failed to insert tranx_node for binlog file: %s, position: %lu",
log_file_name, (ulong)log_file_pos);
switch_off();
}
else
{
rpl_semi_sync_master_request_ack++;
}
}
l_end:
unlock();
DBUG_RETURN(result);
}
int Repl_semi_sync_master::flush_net(THD *thd,
const char *event_buf)
{
int result = -1;
NET* net= &thd->net;
DBUG_ENTER("Repl_semi_sync_master::flush_net");
assert((unsigned char)event_buf[1] == k_packet_magic_num);
if ((unsigned char)event_buf[2] != k_packet_flag_sync)
{
/* current event does not require reply */
result = 0;
goto l_end;
}
/* We flush to make sure that the current event is sent to the network,
* instead of being buffered in the TCP/IP stack.
*/
if (net_flush(net))
{
sql_print_error("Semi-sync master failed on net_flush() "
"before waiting for slave reply");
goto l_end;
}
net_clear(net, 0);
net->pkt_nr++;
net->compress_pkt_nr++;
result = 0;
rpl_semi_sync_master_net_wait_num++;
l_end:
thd->clear_error();
DBUG_RETURN(result);
}
int Repl_semi_sync_master::after_reset_master()
{
int result = 0;
DBUG_ENTER("Repl_semi_sync_master::after_reset_master");
if (rpl_semi_sync_master_enabled)
{
sql_print_information("Enable Semi-sync Master after reset master");
enable_master();
}
lock();
if (rpl_semi_sync_master_clients == 0 &&
!rpl_semi_sync_master_wait_no_slave)
m_state = 0;
else
m_state = get_master_enabled()? 1 : 0;
m_wait_file_name_inited = false;
m_reply_file_name_inited = false;
m_commit_file_name_inited = false;
rpl_semi_sync_master_yes_transactions = 0;
rpl_semi_sync_master_no_transactions = 0;
rpl_semi_sync_master_off_times = 0;
rpl_semi_sync_master_timefunc_fails = 0;
rpl_semi_sync_master_wait_sessions = 0;
rpl_semi_sync_master_wait_pos_backtraverse = 0;
rpl_semi_sync_master_trx_wait_num = 0;
rpl_semi_sync_master_trx_wait_time = 0;
rpl_semi_sync_master_net_wait_num = 0;
rpl_semi_sync_master_net_wait_time = 0;
unlock();
DBUG_RETURN(result);
}
int Repl_semi_sync_master::before_reset_master()
{
int result = 0;
DBUG_ENTER("Repl_semi_sync_master::before_reset_master");
if (rpl_semi_sync_master_enabled)
disable_master();
DBUG_RETURN(result);
}
void Repl_semi_sync_master::check_and_switch()
{
lock();
if (get_master_enabled() && is_on())
{
if (!rpl_semi_sync_master_wait_no_slave
&& rpl_semi_sync_master_clients == 0)
switch_off();
}
unlock();
}
void Repl_semi_sync_master::set_export_stats()
{
lock();
rpl_semi_sync_master_status = m_state;
rpl_semi_sync_master_avg_trx_wait_time=
((rpl_semi_sync_master_trx_wait_num) ?
(ulong)((double)rpl_semi_sync_master_trx_wait_time /
((double)rpl_semi_sync_master_trx_wait_num)) : 0);
rpl_semi_sync_master_avg_net_wait_time=
((rpl_semi_sync_master_net_wait_num) ?
(ulong)((double)rpl_semi_sync_master_net_wait_time /
((double)rpl_semi_sync_master_net_wait_num)) : 0);
unlock();
}
void Repl_semi_sync_master::await_slave_reply()
{
struct timespec abstime;
DBUG_ENTER("Repl_semi_sync_master::::await_slave_reply");
lock();
/* Just return if there is nothing to wait for */
if (!rpl_semi_sync_master_wait_sessions)
goto end;
create_timeout(&abstime, NULL);
cond_timewait(&abstime);
end:
unlock();
DBUG_VOID_RETURN;
}
/* Get the waiting time given the wait's staring time.
*
* Return:
* >= 0: the waiting time in microsecons(us)
* < 0: error in get time or time back traverse
*/
static int get_wait_time(const struct timespec& start_ts)
{
ulonglong start_usecs, end_usecs;
struct timespec end_ts;
/* Starting time in microseconds(us). */
start_usecs = timespec_to_usec(&start_ts);
/* Get the wait time interval. */
set_timespec(end_ts, 0);
/* Ending time in microseconds(us). */
end_usecs = timespec_to_usec(&end_ts);
if (end_usecs < start_usecs)
return -1;
return (int)(end_usecs - start_usecs);
}
void semi_sync_master_deinit()
{
repl_semisync_master.cleanup();
ack_receiver.cleanup();
}
Reference in a new issue View git blame Copy permalink