semisync_source_plugin.cc File Reference

#include <stddef.h>
#include <sys/types.h>
#include "my_inttypes.h"
#include "my_macros.h"
#include "my_psi_config.h"
#include "mysql/psi/mysql_memory.h"
#include "mysql/psi/mysql_stage.h"
#include "nulls.h"
#include "plugin/semisync/semisync_source.h"
#include "plugin/semisync/semisync_source_ack_receiver.h"
#include "sql/current_thd.h"
#include "sql/derror.h"
#include "sql/protocol_classic.h"
#include "sql/raii/sentry.h"
#include "sql/sql_class.h"
#include "sql/sql_lex.h"
#include "typelib.h"

Macros
#define	SHOW_FNAME(name)   rpl_semi_sync_source_show_##name
#define	DEF_SHOW_FUNC(name, show_type)

Enumerations
enum	enum_wait_point { WAIT_AFTER_SYNC , WAIT_AFTER_COMMIT }

Functions
static bool	is_semi_sync_dump ()
static int	repl_semi_report_binlog_update (Binlog_storage_param *, const char *log_file, my_off_t log_pos)
static int	repl_semi_report_binlog_sync (Binlog_storage_param *, const char *log_file, my_off_t log_pos)
static int	repl_semi_report_before_dml (Trans_param *, int &)
static int	repl_semi_report_before_commit (Trans_param *)
static int	repl_semi_report_before_rollback (Trans_param *)
static int	repl_semi_report_commit (Trans_param *param)
static int	repl_semi_report_rollback (Trans_param *param)
static int	repl_semi_report_begin (Trans_param *, int &)
static int	repl_semi_binlog_dump_start (Binlog_transmit_param *param, const char *log_file, my_off_t log_pos)
static int	repl_semi_binlog_dump_end (Binlog_transmit_param *param)
static int	repl_semi_reserve_header (Binlog_transmit_param *, unsigned char *header, unsigned long size, unsigned long *len)
static int	repl_semi_before_send_event (Binlog_transmit_param *param, unsigned char *packet, unsigned long, const char *log_file, my_off_t log_pos)
static int	repl_semi_after_send_event (Binlog_transmit_param *param, const char *event_buf, unsigned long, const char *skipped_log_file, my_off_t skipped_log_pos)
static int	repl_semi_reset_source (Binlog_transmit_param *)
static void	fix_rpl_semi_sync_source_timeout (MYSQL_THD thd, SYS_VAR *var, void *ptr, const void *val)
static void	fix_rpl_semi_sync_source_trace_level (MYSQL_THD thd, SYS_VAR *var, void *ptr, const void *val)
static void	fix_rpl_semi_sync_source_wait_no_replica (MYSQL_THD thd, SYS_VAR *var, void *ptr, const void *val)
static void	fix_rpl_semi_sync_source_enabled (MYSQL_THD thd, SYS_VAR *var, void *ptr, const void *val)
static void	fix_rpl_semi_sync_source_wait_for_replica_count (MYSQL_THD thd, SYS_VAR *var, void *ptr, const void *val)
static	MYSQL_SYSVAR_BOOL (enabled, rpl_semi_sync_source_enabled, PLUGIN_VAR_OPCMDARG, "Enable semi-synchronous replication source (disabled by default). ", nullptr, &fix_rpl_semi_sync_source_enabled, 0)
static	MYSQL_SYSVAR_ULONG (timeout, rpl_semi_sync_source_timeout, PLUGIN_VAR_OPCMDARG, "The timeout value (in milliseconds) for semi-synchronous replication on " "the source. If less than " "rpl_semi_sync_source_wait_for_replica_count " "replicas have replied after this amount of time, switch to asynchronous " "replication.", nullptr, fix_rpl_semi_sync_source_timeout, 10000, 0, ~0UL, 1)
static	MYSQL_SYSVAR_BOOL (wait_no_replica, rpl_semi_sync_source_wait_no_replica, PLUGIN_VAR_OPCMDARG, "If enabled, revert to asynchronous replication only if less " "than " "rpl_semi_sync_source_wait_for_replica_count " "replicas have replied when " "rpl_semi_sync_source_timeout " "seconds have passed. If disabled, revert to asynchronous " "replication also as soon as the number of connected replicas " "drops below " "rpl_semi_sync_source_wait_for_replica_count.", nullptr, &fix_rpl_semi_sync_source_wait_no_replica, 1)
static	MYSQL_SYSVAR_ULONG (trace_level, rpl_semi_sync_source_trace_level, PLUGIN_VAR_OPCMDARG, "The tracing level for semi-sync replication.", nullptr, &fix_rpl_semi_sync_source_trace_level, 32, 0, ~0UL, 1)
static	MYSQL_SYSVAR_ENUM (wait_point, rpl_semi_sync_source_wait_point, PLUGIN_VAR_OPCMDARG, "The semisync source plugin can wait for replica replies at one of two " "alternative points: AFTER_SYNC or AFTER_COMMIT. " "AFTER_SYNC is the default value. AFTER_SYNC means that the " "source-side semisynchronous plugin waits for the replies just after it " "has synced the binary log file (or would have synced, but may have " "skipped it, when sync_binlog!=1), but before it has committed in the " "engine on the source side. Therefore, it guarantees that no other " "sessions on the source can see the effects of the transaction before " "the replica has received it. " "AFTER_COMMIT means that the source-side semisynchronous plugin " "waits for the replies from the replica just after the source has " "committed the transaction in the engine, and before it sends an ACK " "packet to the client session. Other sessions may see the effects of " "the transaction before it has been replicated, even though the current " "session is still waiting for the replies from the replica.", nullptr, nullptr, WAIT_AFTER_SYNC, &wait_point_typelib)
static	MYSQL_SYSVAR_UINT (wait_for_replica_count, rpl_semi_sync_source_wait_for_replica_count, PLUGIN_VAR_OPCMDARG, "The number of replicas that need to acknowledge that they have " "received a transaction, before the transaction can complete on " "the source.", nullptr, &fix_rpl_semi_sync_source_wait_for_replica_count, 1, 1, 65535, 1)
static void	init_semisync_psi_keys (void)
static int	semi_sync_source_plugin_init (void *p)
static int	semi_sync_source_plugin_check_uninstall (void *)
static int	semi_sync_source_plugin_deinit (void *p)
	mysql_declare_plugin (semi_sync_source)

Variables
ReplSemiSyncSource *	repl_semisync = nullptr
Ack_receiver *	ack_receiver = nullptr
static ulong	rpl_semi_sync_source_wait_point = WAIT_AFTER_COMMIT
thread_local bool	THR_RPL_SEMI_SYNC_DUMP = false
static const mysql_service_registry_t *	reg_srv = nullptr
const mysql_service_log_builtins_t *	log_bi = nullptr
	accessor built-ins More...
const mysql_service_log_builtins_string_t *	log_bs = nullptr
	string built-ins More...
static const char *	wait_point_names [] = {"AFTER_SYNC", "AFTER_COMMIT", NullS}
static TYPELIB	wait_point_typelib
static SYS_VAR *	semi_sync_source_system_vars []
Trans_observer	trans_observer
Binlog_storage_observer	storage_observer
Binlog_transmit_observer	transmit_observer
static SHOW_VAR	semi_sync_source_status_vars []
PSI_mutex_key	key_ss_mutex_LOCK_binlog_
PSI_mutex_key	key_ss_mutex_Ack_receiver_mutex
static PSI_mutex_info	all_semisync_mutexes []
PSI_cond_key	key_ss_cond_COND_binlog_send_
PSI_cond_key	key_ss_cond_Ack_receiver_cond
static PSI_cond_info	all_semisync_conds []
PSI_thread_key	key_ss_thread_Ack_receiver_thread
static PSI_thread_info	all_semisync_threads []
PSI_stage_info	stage_waiting_for_semi_sync_ack_from_replica
PSI_stage_info	stage_waiting_for_semi_sync_replica
PSI_stage_info	stage_reading_semi_sync_ack
PSI_memory_key	key_ss_memory_TranxNodeAllocator_block
PSI_stage_info *	all_semisync_stages []
PSI_memory_info	all_semisync_memory []
struct Mysql_replication	semi_sync_source_plugin
	mysql_declare_plugin_end

Macro Definition Documentation

DEF_SHOW_FUNC

#define DEF_SHOW_FUNC	(		name,
			show_type
	)

Value:

  static int SHOW_FNAME(name)(MYSQL_THD, SHOW_VAR * var, char *) { \
    repl_semisync->setExportStats();                               \
    var->type = show_type;                                         \
    var->value = (char *)&rpl_semi_sync_source_##name;             \
    return 0;                                                      \
  }

SHOW_FNAME

#define SHOW_FNAME

(

name

)

rpl_semi_sync_source_show_##name

Enumeration Type Documentation

enum_wait_point

enum enum_wait_point

Enumerator
WAIT_AFTER_SYNC
WAIT_AFTER_COMMIT

Function Documentation

fix_rpl_semi_sync_source_enabled()

static void fix_rpl_semi_sync_source_enabled ( MYSQL_THD  thd, SYS_VAR *  var, void *  ptr, const void *  val  )

static

fix_rpl_semi_sync_source_timeout()

static void fix_rpl_semi_sync_source_timeout ( MYSQL_THD  thd, SYS_VAR *  var, void *  ptr, const void *  val  )

static

fix_rpl_semi_sync_source_trace_level()

static void fix_rpl_semi_sync_source_trace_level ( MYSQL_THD  thd, SYS_VAR *  var, void *  ptr, const void *  val  )

static

fix_rpl_semi_sync_source_wait_for_replica_count()

static void fix_rpl_semi_sync_source_wait_for_replica_count ( MYSQL_THD  thd, SYS_VAR *  var, void *  ptr, const void *  val  )

static

fix_rpl_semi_sync_source_wait_no_replica()

static void fix_rpl_semi_sync_source_wait_no_replica ( MYSQL_THD  thd, SYS_VAR *  var, void *  ptr, const void *  val  )

static

init_semisync_psi_keys()

static void init_semisync_psi_keys ( void  )

static

is_semi_sync_dump()

static bool is_semi_sync_dump ( )

inlinestatic

mysql_declare_plugin()

mysql_declare_plugin

(

semi_sync_source

)

MYSQL_SYSVAR_BOOL() [1/2]

static MYSQL_SYSVAR_BOOL ( enabled  , rpl_semi_sync_source_enabled  , PLUGIN_VAR_OPCMDARG  , "Enable semi-synchronous replication source (disabled by default). "  , nullptr  , &  fix_rpl_semi_sync_source_enabled, 0    )

static

MYSQL_SYSVAR_BOOL() [2/2]

static MYSQL_SYSVAR_BOOL ( wait_no_replica  , rpl_semi_sync_source_wait_no_replica  , PLUGIN_VAR_OPCMDARG  , "If  enabled, revert to asynchronous replication only if less " "than " "rpl_semi_sync_source_wait_for_replica_count " "replicas have replied when " "rpl_semi_sync_source_timeout " "seconds have passed. If  disabled, revert to asynchronous " "replication also as soon as the number of connected replicas " "drops below " "rpl_semi_sync_source_wait_for_replica_count."  , nullptr  , &  fix_rpl_semi_sync_source_wait_no_replica, 1    )

static

MYSQL_SYSVAR_ENUM()

static MYSQL_SYSVAR_ENUM ( wait_point  , rpl_semi_sync_source_wait_point  , PLUGIN_VAR_OPCMDARG  , "The semisync source plugin can wait for replica replies at one of two " "alternative points: AFTER_SYNC or AFTER_COMMIT. " "AFTER_SYNC is the default value. AFTER_SYNC means that the " "source-side semisynchronous plugin waits for the replies just after it " "has synced the binary log file   or would have synced, but may have " "skipped it, when sync_binlog!=1, but before it has committed in the " "engine on the source side.  Therefore, it guarantees that no other " "sessions on the source can see the effects of the transaction before " "the replica has received it. " "AFTER_COMMIT means that the source-side semisynchronous plugin " "waits for the replies from the replica just after the source has " "committed the transaction in the  engine, and before it sends an ACK " "packet to the client session. Other sessions may see the effects of " "the transaction before it has been  replicated, even though the current " "session is still waiting for the replies from the replica."  , nullptr  , nullptr  , WAIT_AFTER_SYNC  , &  wait_point_typelib  )

static

MYSQL_SYSVAR_UINT()

static MYSQL_SYSVAR_UINT ( wait_for_replica_count  , rpl_semi_sync_source_wait_for_replica_count  , PLUGIN_VAR_OPCMDARG  , "The number of replicas that need to acknowledge that they have " "received a  transaction, before the transaction can complete on " "the source."  , nullptr  , &  fix_rpl_semi_sync_source_wait_for_replica_count, 1  , 1  , 65535  , 1    )

static

MYSQL_SYSVAR_ULONG() [1/2]

static MYSQL_SYSVAR_ULONG ( timeout  , rpl_semi_sync_source_timeout  , PLUGIN_VAR_OPCMDARG  , "The timeout value (in milliseconds) for semi-synchronous replication on " "the source. If less than " "rpl_semi_sync_source_wait_for_replica_count " "replicas have replied after this amount of  time, switch to asynchronous " "replication."  , nullptr  , fix_rpl_semi_sync_source_timeout  , 10000  , 0  , ~  0UL, 1    )

static

MYSQL_SYSVAR_ULONG() [2/2]

static MYSQL_SYSVAR_ULONG ( trace_level  , rpl_semi_sync_source_trace_level  , PLUGIN_VAR_OPCMDARG  , "The tracing level for semi-sync replication."  , nullptr  , &  fix_rpl_semi_sync_source_trace_level, 32  , 0  , ~  0UL, 1    )

static

repl_semi_after_send_event()

static int repl_semi_after_send_event ( Binlog_transmit_param *  param, const char *  event_buf, unsigned long  , const char *  skipped_log_file, my_off_t  skipped_log_pos  )

static

repl_semi_before_send_event()

static int repl_semi_before_send_event ( Binlog_transmit_param *  param, unsigned char *  packet, unsigned long  , const char *  log_file, my_off_t  log_pos  )

static

repl_semi_binlog_dump_end()

static int repl_semi_binlog_dump_end ( Binlog_transmit_param *  param )

static

repl_semi_binlog_dump_start()

static int repl_semi_binlog_dump_start ( Binlog_transmit_param *  param, const char *  log_file, my_off_t  log_pos  )

static

repl_semi_report_before_commit()

static int repl_semi_report_before_commit ( Trans_param *  )

static

repl_semi_report_before_dml()

static int repl_semi_report_before_dml ( Trans_param *  , int &    )

static

repl_semi_report_before_rollback()

static int repl_semi_report_before_rollback ( Trans_param *  )

static

repl_semi_report_begin()

static int repl_semi_report_begin ( Trans_param *  , int &    )

static

repl_semi_report_binlog_sync()

static int repl_semi_report_binlog_sync ( Binlog_storage_param *  , const char *  log_file, my_off_t  log_pos  )

static

repl_semi_report_binlog_update()

static int repl_semi_report_binlog_update ( Binlog_storage_param *  , const char *  log_file, my_off_t  log_pos  )

static

repl_semi_report_commit()

static int repl_semi_report_commit ( Trans_param *  param )

static

repl_semi_report_rollback()

static int repl_semi_report_rollback ( Trans_param *  param )

static

repl_semi_reserve_header()

static int repl_semi_reserve_header ( Binlog_transmit_param *  , unsigned char *  header, unsigned long  size, unsigned long *  len  )

static

repl_semi_reset_source()

static int repl_semi_reset_source ( Binlog_transmit_param *  )

static

semi_sync_source_plugin_check_uninstall()

static int semi_sync_source_plugin_check_uninstall ( void *  )

static

semi_sync_source_plugin_deinit()

static int semi_sync_source_plugin_deinit ( void *  p )

static

semi_sync_source_plugin_init()

static int semi_sync_source_plugin_init ( void *  p )

static

Variable Documentation

ack_receiver

Ack_receiver* ack_receiver = nullptr

all_semisync_conds

PSI_cond_info all_semisync_conds[]

static

Initial value:

= {
    {&key_ss_cond_COND_binlog_send_, "COND_binlog_send_", 0, 0,
     PSI_DOCUMENT_ME},
    {&key_ss_cond_Ack_receiver_cond, "Ack_receiver::m_cond", 0, 0,
     PSI_DOCUMENT_ME}}

all_semisync_memory

PSI_memory_info all_semisync_memory[]

Initial value:

= {
    {&key_ss_memory_TranxNodeAllocator_block, "TranxNodeAllocator::block", 0, 0,
     PSI_DOCUMENT_ME}}

all_semisync_mutexes

PSI_mutex_info all_semisync_mutexes[]

static

Initial value:

= {
    {&key_ss_mutex_LOCK_binlog_, "LOCK_binlog_", 0, 0, PSI_DOCUMENT_ME},
    {&key_ss_mutex_Ack_receiver_mutex, "Ack_receiver::m_mutex", 0, 0,
     PSI_DOCUMENT_ME}}

all_semisync_stages

PSI_stage_info* all_semisync_stages[]

Initial value:

= {
    &stage_waiting_for_semi_sync_ack_from_replica,
    &stage_waiting_for_semi_sync_replica, &stage_reading_semi_sync_ack}

all_semisync_threads

PSI_thread_info all_semisync_threads[]

static

Initial value:

= {
    {&key_ss_thread_Ack_receiver_thread, "Ack_receiver", "ss_ack",
     PSI_FLAG_SINGLETON | PSI_FLAG_THREAD_SYSTEM, 0, PSI_DOCUMENT_ME}}

key_ss_cond_Ack_receiver_cond

PSI_cond_key key_ss_cond_Ack_receiver_cond

key_ss_cond_COND_binlog_send_

PSI_cond_key key_ss_cond_COND_binlog_send_

key_ss_memory_TranxNodeAllocator_block

PSI_memory_key key_ss_memory_TranxNodeAllocator_block

key_ss_mutex_Ack_receiver_mutex

PSI_mutex_key key_ss_mutex_Ack_receiver_mutex

key_ss_mutex_LOCK_binlog_

PSI_mutex_key key_ss_mutex_LOCK_binlog_

key_ss_thread_Ack_receiver_thread

PSI_thread_key key_ss_thread_Ack_receiver_thread

log_bi

const mysql_service_log_builtins_t* log_bi = nullptr

accessor built-ins

accessor built-ins

log_bs

const mysql_service_log_builtins_string_t* log_bs = nullptr

string built-ins

mysql_declare_plugin_end

mysql_declare_plugin_end

reg_srv

const mysql_service_registry_t* reg_srv = nullptr

static

repl_semisync

ReplSemiSyncSource* repl_semisync = nullptr

rpl_semi_sync_source_wait_point

ulong rpl_semi_sync_source_wait_point = WAIT_AFTER_COMMIT

static

semi_sync_source_plugin

struct Mysql_replication semi_sync_source_plugin

Initial value:

= {
    MYSQL_REPLICATION_INTERFACE_VERSION}

semi_sync_source_status_vars

SHOW_VAR semi_sync_source_status_vars[]

static

semi_sync_source_system_vars

SYS_VAR* semi_sync_source_system_vars[]

static

Initial value:

= {
    MYSQL_SYSVAR(enabled),
    MYSQL_SYSVAR(timeout),
    MYSQL_SYSVAR(wait_no_replica),
    MYSQL_SYSVAR(trace_level),
    MYSQL_SYSVAR(wait_point),
    MYSQL_SYSVAR(wait_for_replica_count),
    nullptr,
}

stage_reading_semi_sync_ack

PSI_stage_info stage_reading_semi_sync_ack

Initial value:

= {
    0, "Reading semi-sync ACK from replica", 0, PSI_DOCUMENT_ME}

stage_waiting_for_semi_sync_ack_from_replica

PSI_stage_info stage_waiting_for_semi_sync_ack_from_replica

Initial value:

= {
    0, "Waiting for semi-sync ACK from replica", 0, PSI_DOCUMENT_ME}

stage_waiting_for_semi_sync_replica

PSI_stage_info stage_waiting_for_semi_sync_replica

Initial value:

= {
    0, "Waiting for semi-sync replica connection", 0, PSI_DOCUMENT_ME}

storage_observer

Binlog_storage_observer storage_observer

Initial value:

= {
    sizeof(Binlog_storage_observer),  
 
    repl_semi_report_binlog_update,  
    repl_semi_report_binlog_sync,    
}

THR_RPL_SEMI_SYNC_DUMP

thread_local bool THR_RPL_SEMI_SYNC_DUMP = false

trans_observer

Trans_observer trans_observer

Initial value:

= {
    sizeof(Trans_observer),  
 
    repl_semi_report_before_dml,       
    repl_semi_report_before_commit,    
    repl_semi_report_before_rollback,  
    repl_semi_report_commit,           
    repl_semi_report_rollback,         
    repl_semi_report_begin,            
}

transmit_observer

Binlog_transmit_observer transmit_observer

Initial value:

= {
    sizeof(Binlog_transmit_observer),  
 
    repl_semi_binlog_dump_start,  
    repl_semi_binlog_dump_end,    
    repl_semi_reserve_header,     
    repl_semi_before_send_event,  
    repl_semi_after_send_event,   
    repl_semi_reset_source,       
}

wait_point_names

const char* wait_point_names[] = {"AFTER_SYNC", "AFTER_COMMIT", NullS}

static

wait_point_typelib

TYPELIB wait_point_typelib

static

Initial value:

= {array_elements(wait_point_names) - 1, "",
                                     wait_point_names, nullptr}