rpl_source.cc File Reference

#include "sql/rpl_source.h"
#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <memory>
#include <unordered_map>
#include <utility>
#include "map_helpers.h"
#include "mutex_lock.h"
#include "my_byteorder.h"
#include "my_command.h"
#include "my_dbug.h"
#include "my_io.h"
#include "my_macros.h"
#include "my_psi_config.h"
#include "my_sys.h"
#include "mysql/components/services/bits/mysql_mutex_bits.h"
#include "mysql/components/services/bits/psi_bits.h"
#include "mysql/components/services/bits/psi_mutex_bits.h"
#include "mysql/components/services/log_builtins.h"
#include "mysql/my_loglevel.h"
#include "mysql/psi/mysql_file.h"
#include "mysql/psi/mysql_mutex.h"
#include "mysql/service_mysql_alloc.h"
#include "mysql/strings/m_ctype.h"
#include "mysqld_error.h"
#include "sql/auth/auth_acls.h"
#include "sql/auth/auth_common.h"
#include "sql/binlog.h"
#include "sql/current_thd.h"
#include "sql/debug_sync.h"
#include "sql/item.h"
#include "sql/item_func.h"
#include "sql/log.h"
#include "sql/mysqld.h"
#include "sql/mysqld_thd_manager.h"
#include "sql/protocol.h"
#include "sql/protocol_classic.h"
#include "sql/psi_memory_key.h"
#include "sql/rpl_binlog_sender.h"
#include "sql/rpl_filter.h"
#include "sql/rpl_group_replication.h"
#include "sql/rpl_gtid.h"
#include "sql/rpl_handler.h"
#include "sql/rpl_utility.h"
#include "sql/sql_class.h"
#include "sql/sql_lex.h"
#include "sql/sql_list.h"
#include "sql/system_variables.h"
#include "sql_string.h"
#include "strmake.h"
#include "thr_mutex.h"
#include "typelib.h"

Classes
class	Find_zombie_dump_thread
	Callback function used by kill_zombie_dump_threads() function to to find zombie dump thread from the thd list. More...

Macros
#define	get_object(p, obj, msg)
#define	CHECK_PACKET_SIZE(BYTES)
	If there are less than BYTES bytes left to read in the packet, report error. More...
#define	READ(DECODE, BYTES)
	Auxiliary macro used to define READ_INT and READ_STRING. More...
#define	READ_INT(VAR, BYTES)    READ(VAR = uint##BYTES##korr(packet_position), BYTES)
	Check that there are at least BYTES more bytes to read, then read the bytes and decode them into the given integer VAR, then advance the reading position. More...
#define	READ_STRING(VAR, BYTES, BUFFER_SIZE)
	Check that there are at least BYTES more bytes to read and that BYTES+1 is not greater than BUFFER_SIZE, then read the bytes into the given variable VAR, then advance the reading position. More...

Functions
static std::unique_ptr< uchar[]>	build_debug_com_binlog_dump_gtid_encoding (size_t *buffer_size, uint32 *declared_data_size)
resource_blocker::Resource &	get_dump_thread_resource ()
static bool	check_and_report_dump_thread_blocked (resource_blocker::User &rpl_user)
static bool	same_authenticated_source_account (const THD *lhs, const THD *rhs)
	Check whether two dump-thread connections belong to the same authenticated replication account. More...
static void	report_conflicting_dump_thread_owner (const char *replica_uuid, uint32 replica_server_id)
	Report that a dump thread with the requested replica identity belongs to a different authenticated account. More...
static bool	reject_conflicting_dump_thread_owner (THD *thd, const THD *existing_dump_thread, const char *replica_uuid)
	Reject a dump-thread identity owned by a different authenticated account. More...
static THD_ptr	find_dump_thread_with_same_replica_identity (const String &replica_uuid)
	Find an existing dump thread with the same replica identity as the current connection. More...
int	register_replica (THD *thd, uchar *packet, size_t packet_length)
	Register slave in 'slave_list' hash table. More...
void	unregister_replica (THD *thd, bool only_mine, bool need_lock_slave_list)
bool	show_replicas (THD *thd)
	Execute a SHOW REPLICAS statement. More...
bool	com_binlog_dump (THD *thd, char *packet, size_t packet_length)
	Process a COM_BINLOG_DUMP packet. More...
bool	com_binlog_dump_gtid (THD *thd, char *packet, size_t packet_length)
	Process a COM_BINLOG_DUMP_GTID packet. More...
void	mysql_binlog_send (THD *thd, char *log_ident, my_off_t pos, Gtid_set *slave_gtid_executed, uint32 flags)
	Low-level function where the dump thread iterates over the binary log and sends events to the slave. More...
const user_var_entry *	get_user_var_from_alternatives (const THD *thd, const std::string alt1, const std::string alt2)
	Read a user variable that may exist under two different names. More...
String *	get_replica_uuid (THD *thd, String *value)
	An auxiliary function extracts slave UUID. More...
bool	kill_zombie_dump_threads (THD *thd)
bool	reset_binary_logs_and_gtids (THD *thd, bool unlock_global_read_lock)
	Execute a RESET BINARY LOGS AND GTIDS statement. More...
bool	show_binary_log_status (THD *thd)
	Execute a SHOW BINARY LOG STATUS statement. More...
bool	show_binlogs (THD *thd)
	Execute a SHOW BINARY LOGS statement. More...

Variables
int	max_binlog_dump_events = 0
bool	opt_sporadic_binlog_dump_fail = false
malloc_unordered_map< uint32, unique_ptr_my_free< REPLICA_INFO > >	slave_list
TYPELIB	binlog_checksum_typelib

Macro Definition Documentation

CHECK_PACKET_SIZE

#define CHECK_PACKET_SIZE

(

BYTES

)

Value:

  do {                                                          \
    if (packet_bytes_todo < BYTES) goto error_malformed_packet; \
  } while (0)

If there are less than BYTES bytes left to read in the packet, report error.

get_object

#define get_object	(		p,
			obj,
			msg
	)

Value:

  {                                                                     \
    if (p >= p_end) {                                                   \
      my_error(ER_MALFORMED_PACKET, MYF(0));                            \
      return 1;                                                         \
    }                                                                   \
    /* net_field_length_ll() assumes a complete length field. */        \
    const size_t length_size = net_field_length_size(p);                \
    if (length_size > static_cast<size_t>(p_end - p)) {                 \
      my_error(ER_MALFORMED_PACKET, MYF(0));                            \
      return 1;                                                         \
    }                                                                   \
    const uint64_t len = net_field_length_ll(&p);                       \
    if (len > static_cast<uint64_t>(p_end - p) || len >= sizeof(obj)) { \
      errmsg = msg;                                                     \
      goto err;                                                         \
    }                                                                   \
    const size_t object_length = static_cast<size_t>(len);              \
    strmake(obj, (char *)p, object_length);                             \
    p += object_length;                                                 \
  }

READ

#define READ	(		DECODE,
			BYTES
	)

Value:

  do {                          \
    CHECK_PACKET_SIZE(BYTES);   \
    DECODE;                     \
    packet_position += BYTES;   \
    packet_bytes_todo -= BYTES; \
  } while (0)

Auxiliary macro used to define READ_INT and READ_STRING.

Check that there are at least BYTES more bytes to read, then read the bytes using the given DECODER, then advance the reading position.

READ_INT

#define READ_INT	(		VAR,
			BYTES
	)		READ(VAR = uint##BYTES##korr(packet_position), BYTES)

Check that there are at least BYTES more bytes to read, then read the bytes and decode them into the given integer VAR, then advance the reading position.

READ_STRING

#define READ_STRING	(		VAR,
			BYTES,
			BUFFER_SIZE
	)

Value:

  do {                                                     \
    if (BUFFER_SIZE <= BYTES) goto error_malformed_packet; \
    READ(memcpy(VAR, packet_position, BYTES), BYTES);      \
    VAR[BYTES] = '\0';                                     \
  } while (0)

Check that there are at least BYTES more bytes to read and that BYTES+1 is not greater than BUFFER_SIZE, then read the bytes into the given variable VAR, then advance the reading position.

Function Documentation

build_debug_com_binlog_dump_gtid_encoding()

static std::unique_ptr< uchar[]> build_debug_com_binlog_dump_gtid_encoding ( size_t *  buffer_size, uint32 *  declared_data_size  )

static

check_and_report_dump_thread_blocked()

static bool check_and_report_dump_thread_blocked ( resource_blocker::User &  rpl_user )

static

com_binlog_dump()

bool com_binlog_dump	(	THD *	thd,
		char *	packet,
		size_t	packet_length
	)

Process a COM_BINLOG_DUMP packet.

This function parses the packet and then calls mysql_binlog_send.

Parameters

thd	The dump thread.
packet	The COM_BINLOG_DUMP packet.
packet_length	The length of the packet in bytes.

Return values

true	Error
false	Success

com_binlog_dump_gtid()

bool com_binlog_dump_gtid	(	THD *	thd,
		char *	packet,
		size_t	packet_length
	)

Process a COM_BINLOG_DUMP_GTID packet.

This function parses the packet and then calls mysql_binlog_send.

Parameters

thd	The dump thread.
packet	The COM_BINLOG_DUMP_GTID packet.
packet_length	The length of the packet in bytes.

Return values

true	Error
false	Success

find_dump_thread_with_same_replica_identity()

static THD_ptr find_dump_thread_with_same_replica_identity ( const String &  replica_uuid )

static

Find an existing dump thread with the same replica identity as the current connection.

A replica provides its UUID through the replica_uuid or legacy slave_uuid session user variable. An empty UUID occurs for direct dump clients that do not initialize either variable; those clients are matched by server ID.

Parameters

replica_uuid

Replica UUID, or an empty string when unavailable.

Return values

THD_ptr	holding the matching dump thread and its LOCK_thd_data.
THD_ptr{nullptr}	if no dump thread has the same identity.

get_dump_thread_resource()

resource_blocker::Resource & get_dump_thread_resource

(

)

get_replica_uuid()

String * get_replica_uuid	(	THD *	thd,
		String *	value
	)

An auxiliary function extracts slave UUID.

Parameters

[in]	thd	THD to access a user variable
[out]	value	String to return UUID value.

Returns

if success value is returned else NULL is returned.

get_user_var_from_alternatives()

const user_var_entry * get_user_var_from_alternatives	(	const THD *	thd,
		const std::string	alt1,
		const std::string	alt2
	)

Read a user variable that may exist under two different names.

Parameters

thd	The session to read from.
alt1	The first variable name alternative.
alt2	The second variable name alternative.

Return values

If	there exists a user variable in the current session with the first name, return that. Otherwise, if the second one exists, return that. Otherwise, return NULL.

kill_zombie_dump_threads()

bool kill_zombie_dump_threads

(

THD *

thd

)

mysql_binlog_send()

void mysql_binlog_send	(	THD *	thd,
		char *	log_ident,
		my_off_t	pos,
		Gtid_set *	gtid_set,
		uint32	flags
	)

Low-level function where the dump thread iterates over the binary log and sends events to the slave.

This function is common for both COM_BINLOG_DUMP and COM_BINLOG_DUMP_GTID.

Parameters

thd	The dump thread.
log_ident	The filename of the binary log, as given in the COM_BINLOG_DUMP[_GTID] packet. If this is is an empty string (first character is '\0'), we start with the oldest binary log.
pos	The offset in the binary log, as given in the COM_BINLOG_DUMP[_GTID] packet. This must be at least 4 and at most the size of the binary log file.
gtid_set	The gtid_set that the slave sent, or NULL if the protocol is COM_BINLOG_DUMP.
flags	flags in COM_BINLOG_DUMP[_GTID] packets.

Note

This function will start reading at the given (filename, offset), or from the oldest log if filename[0]==0. It will send all events from that position; but if gtid_set!=NULL, it will skip all events in that set.

register_replica()

int register_replica	(	THD *	thd,
		uchar *	packet,
		size_t	packet_length
	)

Register slave in 'slave_list' hash table.

Returns

0 ok

1 Error. Error message sent to client

reject_conflicting_dump_thread_owner()

static bool reject_conflicting_dump_thread_owner ( THD *  thd, const THD *  existing_dump_thread, const char *  replica_uuid  )

static

Reject a dump-thread identity owned by a different authenticated account.

Parameters

thd	New connection requesting the replica identity.
existing_dump_thread	Existing dump thread with the same identity.
replica_uuid	Null-terminated replica UUID, or an empty string when unavailable.

Return values

true	The existing dump thread belongs to a different account and an error was reported.
false	Account affinity is disabled or the authenticated accounts match.

report_conflicting_dump_thread_owner()

static void report_conflicting_dump_thread_owner ( const char *  replica_uuid, uint32  replica_server_id  )

static

Report that a dump thread with the requested replica identity belongs to a different authenticated account.

Parameters

replica_uuid	Null-terminated replica UUID. This is empty when the client did not provide a UUID.
replica_server_id	Replica server ID used as the identity fallback when replica_uuid is empty.

reset_binary_logs_and_gtids()

bool reset_binary_logs_and_gtids	(	THD *	thd,
		bool	unlock_global_read_lock
	)

Execute a RESET BINARY LOGS AND GTIDS statement.

Parameters

thd	Pointer to THD object of the client thread executing the statement.
unlock_global_read_lock	Unlock the global read lock aquired by RESET BINARY LOGS AND GTIDS.

Return values

false	success
true	error

same_authenticated_source_account()

static bool same_authenticated_source_account ( const THD *  lhs, const THD *  rhs  )

static

Check whether two dump-thread connections belong to the same authenticated replication account.

Parameters

lhs	First connection to compare.
rhs	Second connection to compare.

Return values

true	The authenticated account matches on both connections.
false	The authenticated account differs or is unavailable.

show_binary_log_status()

bool show_binary_log_status

(

THD *

thd

)

Execute a SHOW BINARY LOG STATUS statement.

Parameters

thd	Pointer to THD object for the client thread executing the statement.

Return values

false	success
true	failure

show_binlogs()

bool show_binlogs

(

THD *

thd

)

Execute a SHOW BINARY LOGS statement.

Parameters

thd	Pointer to THD object for the client thread executing the statement.

Return values

false	success
true	failure

show_replicas()

bool show_replicas

(

THD *

thd

)

Execute a SHOW REPLICAS statement.

Parameters

thd	Pointer to THD object for the client thread executing the statement.

Return values

false	success
true	failure

unregister_replica()

void unregister_replica	(	THD *	thd,
		bool	only_mine,
		bool	need_lock_slave_list
	)

Variable Documentation

max_binlog_dump_events

int max_binlog_dump_events = 0

opt_sporadic_binlog_dump_fail

bool opt_sporadic_binlog_dump_fail = false

slave_list

malloc_unordered_map<uint32, unique_ptr_my_free<REPLICA_INFO> > slave_list

Initial value:

{
    key_memory_REPLICA_INFO}