xcom_base.cc File Reference

xcom/xcom_base.c The new version of xcom is a major rewrite to allow transmission of multiple messages from several sources simultaneously without collision. More...

#include <assert.h>
#include <errno.h>
#include <inttypes.h>
#include <limits.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <sys/types.h>
#include <poll.h>
#include "xcom/xcom_profile.h"
#include "my_compiler.h"
#include "xcom/x_platform.h"
#include <arpa/inet.h>
#include <net/if.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/sockio.h>
#include <memory>
#include "xcom/app_data.h"
#include "xcom/get_synode_app_data.h"
#include "xcom/node_no.h"
#include "xcom/server_struct.h"
#include "xcom/simset.h"
#include "xcom/site_struct.h"
#include "xcom/task.h"
#include "xcom/task_net.h"
#include "xcom/task_os.h"
#include "xcom/xcom_base.h"
#include "xcom/xcom_common.h"
#include "xcom/xcom_detector.h"
#include "xcom/xcom_transport.h"
#include "xcom/xdr_utils.h"
#include "xdr_gen/xcom_vp.h"
#include "xcom/bitset.h"
#include "xcom/leader_info_data.h"
#include "xcom/node_list.h"
#include "xcom/node_set.h"
#include "xcom/pax_msg.h"
#include "xcom/site_def.h"
#include "xcom/sock_probe.h"
#include "xcom/synode_no.h"
#include "xcom/task_debug.h"
#include "xcom/xcom_cache.h"
#include "xcom/xcom_cfg.h"
#include "xcom/xcom_interface.h"
#include "xcom/xcom_memory.h"
#include "xcom/xcom_msg_queue.h"
#include "xcom/xcom_recover.h"
#include "xcom/xcom_statistics.h"
#include "xcom/xcom_vp_str.h"
#include "xcom/network/xcom_network_provider.h"
#include <openssl/ssl.h>
#include <chrono>
#include <future>
#include <queue>
#include <tuple>
#include "xcom/retry.h"
#include <sys/utsname.h>
#include <netinet/in.h>

Classes
struct	synode_pool
struct	execute_context
struct	fp_name
struct	xcom_fsm_state

Macros
#define	__STDC_FORMAT_MACROS
#define	SYS_STRERROR_SIZE   512
#define	XCOM_SEND_APP_WAIT_TIMEOUT   20
#define	PROTOVERSION_WARNING_TIMEOUT   600.0 /** Every 10 minutes */
#define	IS_CONS_ALL(p)    ((p)->proposer.msg->a ? (p)->proposer.msg->a->consensus == cons_all : 0)
#define	PROP_ITER
#define	FNVSTART   0x811c9dc5
#define	GOTO(x)
#define	FIFO_SIZE   1000
#define	NAME(f)    { f, #f }
#define	reply_msg(m)
#define	CREATE_REPLY(x)
#define	SEND_REPLY
#define	PING_GATHERING_TIME_WINDOW   5.0
#define	PINGS_GATHERED_BEFORE_CONNECTION_SHUTDOWN   3
#define	BAL_FMT   "ballot {cnt %d node %d}"
#define	BAL_MEM(x)   (x).cnt, (x).node
#define	SERIALIZE_REPLY(msg)
#define	WRITE_REPLY
#define	X(b)   #b
#define	X_FSM_STATE(s)    { s, #s }
#define	SET_X_FSM_STATE(s)
#define	CONNECT_FAIL
#define	TICK_PERIOD   0.01
#define	X(b)   #b

Typedefs
typedef bool(*	unsafe_node_check) (node_address const *node)
typedef enum allow_event_horizon_result	allow_event_horizon_result
typedef struct execute_context	execute_context
typedef void(*	exec_fp) (execute_context *xc)
typedef struct xcom_fsm_state	xcom_fsm_state
typedef int(*	xcom_fsm_fp) (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
typedef enum xcom_send_app_wait_result	xcom_send_app_wait_result
typedef void(*	paxos_state_action) (pax_machine *paxos, site_def const *site, pax_msg *mess)

Enumerations
enum class	synode_allocation_type { todo = 0 , local , remote , global }
enum class	synode_reservation_status : int { number_ok , no_nodes , delivery_timeout }
enum	allow_event_horizon_result { EVENT_HORIZON_ALLOWED , EVENT_HORIZON_INVALID , EVENT_HORIZON_UNCHANGEABLE }
	Check if we can reconfigure the event horizon. More...
enum	{ TAG_START = 313 }
enum	xcom_send_app_wait_result {   SEND_REQUEST_FAILED = 0 , RECEIVE_REQUEST_FAILED , REQUEST_BOTCHED , RETRIES_EXCEEDED ,   REQUEST_OK_RECEIVED , REQUEST_FAIL_RECEIVED , REQUEST_OK_REDIRECT }
enum	{ paxos_timer_range = 1000 }

Functions
long	xcom_unique_long (void)
static int64_t	socket_write (connection_descriptor *wfd, void *_buf, uint32_t n, connnection_write_method write_function=con_write)
static double	wakeup_delay (double old)
static void	note_snapshot (node_no node)
static int	proposer_task (task_arg arg)
static int	executor_task (task_arg arg)
static int	sweeper_task (task_arg arg)
int	alive_task (task_arg arg)
int	cache_manager_task (task_arg arg)
int	detector_task (task_arg arg)
static int	finished (pax_machine *p)
static int	accepted (pax_machine *p)
static int	started (pax_machine *p)
static synode_no	first_free_synode_local (synode_no msgno)
static void	free_forced_config_site_def ()
static void	activate_sweeper ()
static void	force_pax_machine (pax_machine *p, int enforcer)
static void	propose_noop_2p (synode_no find, pax_machine *p)
static void	handle_need_snapshot (linkage *reply_queue, pax_msg *pm)
static void	handle_skip (site_def const *site, pax_machine *p, pax_msg *m)
static void	paxos_fsm (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static bool	is_leader (site_def *site)
int	is_active_leader (node_no x, site_def *site)
static msg_handler *	primary_dispatch_table ()
static msg_handler *	secondary_dispatch_table ()
static void	recompute_node_sets (site_def const *old_site, site_def *new_site)
void	recompute_timestamps (detector_state const old_timestamp, node_list const *old_nodes, detector_state new_timestamp, node_list const *new_nodes)
void	analyze_leaders (site_def *site)
static int	ignore_message (synode_no x, site_def *site, char const *dbg)
static void	init_proposers ()
void	initialize_lsn (uint64_t n)
void	init_base_vars ()
uint32_t	get_my_xcom_id ()
synode_no	get_current_message ()
synode_no	get_max_synode ()
static bool_t	is_latest_config (site_def const *const config)
static site_def const *	first_event_horizon_reconfig ()
	Get the first pending configuration that reconfigures the event horizon. More...
static site_def const *	latest_event_horizon_reconfig ()
	Get the latest pending configuration that reconfigures the event horizon. More...
static synode_no	add_event_horizon (synode_no s)
	Add the event horizon to the given base synod s. More...
void	set_group (uint32_t id)
	Set node group. More...
static void	bury_site (uint32_t id)
static bool_t	is_dead_site (uint32_t id)
node_set *	init_node_set (node_set *set, u_int n)
node_set *	alloc_node_set (node_set *set, u_int n)
static synode_no	incr_msgno (synode_no msgno)
synode_no	incr_synode (synode_no synode)
static void	skip_value (pax_msg *p)
static int	ignoresig (int signum)
static int	recently_active (pax_machine *p)
int	pm_finished (pax_machine *p)
static int	accepted_noop (pax_machine *p)
static int	noop_match (pax_machine *p, pax_msg *pm)
void	set_last_received_config (synode_no received_config_change)
static node_no	max_check (site_def const *site)
static int	is_forcing_node (pax_machine const *p)
static int	majority (bit_set const *nodeset, site_def const *s, int all, int delay, int force)
static int	prep_majority (site_def const *site, pax_machine const *p)
static int	prop_majority (site_def const *site, pax_machine const *p)
site_def const *	get_executor_site ()
site_def *	get_executor_site_rw ()
site_def const *	get_proposer_site ()
synode_no	get_delivered_msg ()
synode_no	get_last_delivered_msg ()
void	init_xcom_base ()
static void	init_tasks ()
void	xcom_thread_init ()
static void	empty_prop_input_queue ()
static void	empty_synode_number_pool ()
void	xcom_thread_deinit ()
static void	create_proposers ()
static void	add_proposer_synode (int i, synode_no *syn_ptr)
static void	remove_proposer_synode (int i)
static synode_no	get_proposer_synode (int i)
static synode_no	min_proposer_synode ()
static void	terminate_proposers ()
static void	set_proposer_startpoint ()
void	set_xcom_run_cb (xcom_state_change_cb x)
void	set_xcom_exit_cb (xcom_state_change_cb x)
void	set_xcom_comms_cb (xcom_state_change_cb x)
void	set_xcom_expel_cb (xcom_state_change_cb x)
void	set_xcom_input_try_pop_cb (xcom_input_try_pop_cb pop)
static bool_t	xcom_input_signal_connection_shutdown_ssl_wait_for_peer ()
static bool_t	xcom_input_signal_connection_shutdown_ssl ()
bool_t	xcom_input_new_signal_connection (char const *address, xcom_port port)
bool_t	xcom_input_signal ()
void	xcom_input_free_signal_connection ()
static int	local_server_shutdown_ssl (connection_descriptor *con, void *buf, int n, int *ret)
int	local_server (task_arg arg)
static bool_t	local_server_is_setup ()
static void	init_time_queue ()
static int	paxos_timer_task (task_arg arg)
int	xcom_taskmain2 (xcom_port listen_port)
static void	prepare (pax_msg *p, pax_op op)
void	init_prepare_msg (pax_msg *p)
	Initializes the message p as a Prepare message, as in the message for Phase 1 (a) of the Paxos protocol. More...
static int	prepare_msg (pax_msg *p)
pax_msg *	create_noop (pax_msg *p)
	Initializes the message p as a Prepare message for a no-op, as in the message for Phase 1 (a) of the Paxos protocol. More...
static pax_msg *	create_read (site_def const *site, pax_msg *p)
static int	skip_msg (pax_msg *p)
static void	brand_app_data (pax_msg *p)
static synode_no	my_unique_id (synode_no synode)
static void	set_unique_id (pax_msg *msg, synode_no synode)
void	init_propose_msg (pax_msg *p)
	Initializes the message p as an Accept message, as in the message for Phase 2 (a) of the Paxos protocol. More...
static int	send_propose_msg (pax_msg *p)
static int	propose_msg (pax_msg *p)
static void	set_learn_type (pax_msg *p)
static void	init_learn_msg (pax_msg *p)
static int	send_learn_msg (site_def const *site, pax_msg *p)
static pax_msg *	create_tiny_learn_msg (pax_machine *pm, pax_msg *p)
static int	send_tiny_learn_msg (site_def const *site, pax_msg *p)
void	prepare_push_3p (site_def const *site, pax_machine *p, pax_msg *msg, synode_no msgno, pax_msg_type msg_type)
	Initializes the message msg to go through a 3-phase, regular Paxos. More...
void	prepare_push_2p (site_def const *site, pax_machine *p)
	Initializes the proposer's message to go through a 2-phase Paxos on the proposer's reserved ballot (0,_). More...
static void	push_msg_2p (site_def const *site, pax_machine *p)
static void	push_msg_3p (site_def const *site, pax_machine *p, pax_msg *msg, synode_no msgno, pax_msg_type msg_type)
static void	brand_client_msg (pax_msg *msg, synode_no msgno)
void	xcom_send (app_data_ptr a, pax_msg *msg)
static uint32_t	fnv_hash (unsigned char *buf, size_t length, uint32_t sum)
uint32_t	new_id ()
	Create a new (hopefully unique) ID. More...
static synode_no	getstart (app_data_ptr a)
void	site_install_action (site_def *site, cargo_type operation)
static void	active_leaders (site_def *site, leader_array *leaders)
void	synthesize_leaders (leader_array *leaders)
static bool	leaders_set_by_client (site_def const *site)
static site_def *	create_site_def_with_start (app_data_ptr a, synode_no start)
static site_def *	install_ng_with_start (app_data_ptr a, synode_no start)
site_def *	install_node_group (app_data_ptr a)
void	set_max_synode (synode_no synode)
static int	is_busy (synode_no s)
bool_t	match_my_msg (pax_msg *learned, pax_msg *mine)
static uint64_t	assign_lsn ()
	Assign the next log sequence number (lsn) for a message. More...
static void	propose_noop (synode_no find, pax_machine *p)
static uint64_t	too_far_threshold (xcom_event_horizon active_event_horizon)
	Checks if the given synod s is outside the event horizon. More...
static uint64_t	too_far_threshold_new_event_horizon_pending (site_def const *new_config)
static int	too_far (synode_no s)
static int	is_view (cargo_type x)
static int	is_config (cargo_type x)
static int	wait_for_cache (pax_machine **pm, synode_no synode, double timeout)
static synode_no	local_synode_allocator (synode_no synode)
static synode_no	global_synode_allocator (site_def *site, synode_no synode)
static node_no	remote_synode_allocator (site_def *site, app_data const &a)
static int	reserve_synode_number (synode_allocation_type *synode_allocation, site_def **site, synode_no *msgno, int *remote_retry, app_data *a, synode_reservation_status *ret)
static constexpr bool	should_ignore_forced_config_or_view (xcom_proto protocol_version)
static node_no	get_leader (site_def const *s)
int	iamthegreatest (site_def const *s)
static site_def *	update_site (site_def *site, node_set const *ns, synode_no boot_key, synode_no start)
void	execute_msg (site_def *site, pax_machine *pma, pax_msg *p)
static void	read_missing_values (int n)
static void	propose_missing_values (int n)
static void	find_value (site_def const *site, unsigned int *wait, int n)
static void	dump_debug_exec_state ()
int	get_xcom_message (pax_machine **p, synode_no msgno, int n)
synode_no	set_executed_msg (synode_no msgno)
synode_no	set_current_message (synode_no msgno)
static void	update_max_synode (pax_msg *p)
static int	match_leader (char const *addr, leader_array const leaders)
static bool	alive_node (site_def const *site, u_int i)
node_no	found_active_leaders (site_def *site)
static void	debug_loser (synode_no x)
static void	send_value (site_def const *site, node_no to, synode_no synode)
static synode_no	compute_delay (synode_no start, xcom_event_horizon event_horizon)
	Returns the message number where it is safe for nodes in previous configuration to exit. More...
static void	inform_removed (int index, int all)
static bool_t	backwards_compatible (xcom_event_horizon event_horizon)
static bool_t	reconfigurable_event_horizon (xcom_proto protocol_version)
static bool_t	add_node_unsafe_against_ipv4_old_nodes (app_data_ptr a)
static bool_t	add_node_adding_own_address (app_data_ptr a)
	This will test if we are receiving a boot request that contains ourselves. More...
static bool	unsafe_against_event_horizon (node_address const *node)
	Check if a node is compatible with the group's event horizon. More...
static bool	check_if_add_node_is_unsafe (app_data_ptr a, unsafe_node_check unsafe)
static bool	check_if_add_node_is_unsafe_against_event_horizon (app_data_ptr a)
void	recompute_node_set (node_set const *old_set, node_list const *old_nodes, node_set *new_set, node_list const *new_nodes)
static bool	incompatible_proto_and_max_leaders (xcom_proto x_proto, node_no max_leaders)
static bool	incompatible_proto_and_leaders (xcom_proto x_proto)
static bool	incompatible_proto_and_max_leaders (node_address const *node)
static bool	incompatible_proto_and_leaders (node_address const *node)
bool	unsafe_leaders (app_data *a)
static void	set_start_and_boot (site_def *new_config, app_data_ptr a)
site_def *	handle_add_node (app_data_ptr a)
	Reconfigure the group membership: add new member(s). More...
static void	log_event_horizon_reconfiguration_failure (allow_event_horizon_result error_code, xcom_event_horizon attempted_event_horizon)
static allow_event_horizon_result	allow_event_horizon (xcom_event_horizon event_horizon)
static bool_t	is_unsafe_event_horizon_reconfiguration (app_data_ptr a)
static bool_t	is_unsafe_max_leaders_reconfiguration (app_data_ptr a)
static bool_t	is_unsafe_set_leaders_reconfiguration (app_data_ptr a)
static bool_t	is_unsafe_leaders_reconfiguration (app_data_ptr a)
static bool_t	are_there_dead_nodes_in_new_config (app_data_ptr a)
bool_t	handle_event_horizon (app_data_ptr a)
	Reconfigure the event horizon. More...
static bool_t	handle_max_leaders (site_def *new_config, app_data_ptr a)
bool_t	handle_max_leaders (app_data_ptr a)
static void	zero_leader_array (leader_array *l)
static void	move_leader_array (leader_array *target, leader_array *source)
static bool_t	handle_set_leaders (site_def *new_config, app_data_ptr a)
bool_t	handle_set_leaders (app_data_ptr a)
bool_t	handle_leaders (app_data_ptr a)
void	terminate_and_exit ()
static int	is_empty_site (site_def const *s)
site_def *	handle_remove_node (app_data_ptr a)
static void	log_ignored_forced_config (app_data_ptr a, char const *const caller_name)
bool_t	handle_config (app_data_ptr a, bool const forced)
static int	is_member (site_def const *site)
static int	addone (int i)
static int	fifo_empty ()
static int	fifo_full ()
static void	fifo_insert (synode_no s)
static synode_no	fifo_extract ()
static synode_no	fifo_front ()
static void	dump_exec_state (execute_context *xc, long dbg)
static int	x_check_exit (execute_context *xc)
static int	x_check_execute_inform (execute_context *xc)
static void	x_fetch (execute_context *xc)
static void	x_execute (execute_context *xc)
static void	x_check_increment_fetch (execute_context *xc)
static void	x_check_increment_execute (execute_context *xc)
static void	x_terminate (execute_context *xc)
static void	setup_exit_handling (execute_context *xc, site_def *site)
static synode_no	get_sweep_start ()
static bool	allow_channel_takeover (site_def const *site)
static void	broadcast_noop (synode_no find, pax_machine *p)
static site_def const *	init_noop (synode_no find, pax_machine *p)
static void	send_read (synode_no find)
static int	ok_to_propose (pax_machine *p)
bool_t	safe_app_data_copy (pax_msg **target, app_data_ptr source)
	Copies app data source into target and checks if the copy succeeded. More...
static pax_msg *	create_learn_msg_for_ignorant_node (pax_machine *p, pax_msg *pm, synode_no synode)
static void	teach_ignorant_node (site_def const *site, pax_machine *p, pax_msg *pm, synode_no synode, linkage *reply_queue)
static void	handle_read (site_def const *site, pax_machine *p, linkage *reply_queue, pax_msg *pm)
static pax_msg *	create_ack_prepare_msg (pax_machine *p, pax_msg *pm, synode_no synode)
pax_msg *	handle_simple_prepare (pax_machine *p, pax_msg *pm, synode_no synode)
	Process the incoming Prepare message from a Proposer, as in the message for Phase 1 (a) of the Paxos protocol. More...
static void	handle_prepare (site_def const *site, pax_machine *p, linkage *reply_queue, pax_msg *pm)
bool_t	check_propose (site_def const *site, pax_machine *p)
static bool	learn_ok (site_def const *site, pax_machine const *p)
static pax_msg *	check_learn (site_def const *site, pax_machine *p)
static void	do_learn (site_def const *site, pax_machine *p, pax_msg *m)
bool_t	handle_simple_ack_prepare (site_def const *site, pax_machine *p, pax_msg *m)
	Process the incoming acknowledge from an Acceptor to a sent Prepare, as in the message for Phase 1 (b) of the Paxos protocol. More...
static void	handle_ack_prepare (site_def const *site, pax_machine *p, pax_msg *m)
static pax_msg *	create_ack_accept_msg (pax_msg *m, synode_no synode)
pax_msg *	handle_simple_accept (pax_machine *p, pax_msg *m, synode_no synode)
	Process the incoming Accept from a Proposer, as in the message for Phase 2 (a) of the Paxos protocol. More...
static void	handle_accept (site_def const *site, pax_machine *p, linkage *reply_queue, pax_msg *m)
pax_msg *	handle_simple_ack_accept (site_def const *site, pax_machine *p, pax_msg *m)
	Process the incoming acknowledge from an Acceptor to a sent Accept, as in the message for Phase 2 (b) of the Paxos protocol. More...
static void	handle_ack_accept (site_def const *site, pax_machine *p, pax_msg *m)
void	handle_tiny_learn (site_def const *site, pax_machine *pm, pax_msg *p)
	Process the incoming tiny, i.e. More...
static void	force_interval (synode_no start, synode_no end, int enforcer)
static void	start_force_config (site_def *s, int enforcer)
void	handle_learn (site_def const *site, pax_machine *p, pax_msg *m)
	Process the incoming Learn message. More...
static void	handle_client_msg (pax_msg *p)
static void	handle_boot (site_def const *site, linkage *reply_queue, pax_msg *p)
bool_t	should_handle_need_boot (site_def const *site, pax_msg *p)
void	init_need_boot_op (pax_msg *p, node_address *identity)
	Initializes the message p as a need_boot_op message. More...
int	pre_process_incoming_ping (site_def const *site, pax_msg const *pm, int has_client_already_booted, double current_time)
	Process incoming are_you_alive (i.e. More...
static void	handle_alive (site_def const *site, linkage *reply_queue, pax_msg *pm)
xcom_event_horizon	xcom_get_minimum_event_horizon ()
xcom_event_horizon	xcom_get_maximum_event_horizon ()
static client_reply_code	xcom_get_event_horizon (xcom_event_horizon *event_horizon)
	Retrieves the latest event horizon. More...
static bool	add_node_test_connectivity_to_added_nodes (node_address *nodes_to_change, u_int number_of_nodes_to_change)
static u_int	allow_add_node (app_data_ptr a)
static u_int	allow_remove_node (app_data_ptr a)
static void	log_cfgchange_wrong_group (app_data_ptr a, const char *const message_fmt)
	Logs the fact that an add/remove node request is aimed at another group. More...
static client_reply_code	can_execute_cfgchange (pax_msg *p)
	Validates if a configuration command can be executed. More...
void	dispatch_get_event_horizon (site_def const *site, pax_msg *p, linkage *reply_queue)
static reply_data *	new_leader_info (site_def *site)
void	dispatch_get_leaders (site_def *site, pax_msg *p, linkage *reply_queue)
static void	log_get_synode_app_data_failure (xcom_get_synode_app_data_result error_code)
void	dispatch_get_synode_app_data (site_def const *site, pax_msg *p, linkage *reply_queue)
static int	can_send_snapshot ()
static void	process_client_msg (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_prepare_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static int	abort_processing (pax_msg *p)
static void	process_ack_prepare_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_accept_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_ack_accept_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_learn_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_recover_learn_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_skip_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_i_am_alive_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_are_you_alive_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_need_boot_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_die_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_read_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_gcs_snapshot_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_tiny_learn_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_synode_request (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	process_synode_allocated (site_def const *site, pax_msg *p, linkage *reply_queue)
static msg_handler *	clone_dispatch_table (msg_handler const *proto)
pax_msg *	dispatch_op (site_def const *site, pax_msg *p, linkage *reply_queue)
static void	update_srv (server **target, server *srv)
static int	harmless (pax_msg const *p)
static bool_t	should_poll_cache (pax_op op)
int	acceptor_learner_task (task_arg arg)
static void	server_handle_need_snapshot (server *srv, site_def const *s, node_no node)
int	reply_handler_task (task_arg arg)
void	xcom_sleep (unsigned int seconds)
app_data_ptr	init_config_with_group (app_data *a, node_list *nl, cargo_type type, uint32_t group_id)
app_data_ptr	init_set_event_horizon_msg (app_data *a, uint32_t group_id, xcom_event_horizon event_horizon)
app_data_ptr	init_get_msg (app_data *a, uint32_t group_id, cargo_type const t)
app_data_ptr	init_get_leaders_msg (app_data *a, uint32_t group_id)
app_data_ptr	init_get_event_horizon_msg (app_data *a, uint32_t group_id)
app_data_ptr	init_app_msg (app_data *a, char *payload, u_int payload_size)
static app_data_ptr	init_get_synode_app_data_msg (app_data *a, uint32_t group_id, synode_no_array *const synodes)
app_data_ptr	init_set_cache_size_msg (app_data *a, uint64_t cache_limit)
app_data_ptr	init_convert_into_local_server_msg (app_data *a)
static void	server_send_snapshot (server *srv, site_def const *s, gcs_snapshot *gcs_snap, node_no node)
static void	server_push_log (server *srv, synode_no push, node_no node)
static void	reply_push_log (synode_no push, linkage *reply_queue)
static gcs_snapshot *	create_snapshot ()
static int	xcom_timer (task_arg arg)
static void	stop_x_timer ()
static void	start_x_timer (double t)
static int	x_fsm_completion_task (task_arg arg)
void	send_x_fsm_complete ()
static void	reset_snapshot_mask ()
static int	got_all_snapshots ()
static int	better_snapshot (gcs_snapshot *gcs)
static void	handle_x_snapshot (gcs_snapshot *gcs)
static void	update_best_snapshot (gcs_snapshot *gcs)
static void	send_need_boot ()
void	set_log_end (gcs_snapshot *gcs)
static int	xcom_fsm_init (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_start_enter (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_start (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_snapshot_wait_enter (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_snapshot_wait (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_recover_wait_enter (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_recover_wait (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_run_enter (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	xcom_fsm_run (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)
static int	handle_fsm_net_boot (task_arg fsmargs, xcom_fsm_state *ctxt, int cont)
static int	handle_fsm_snapshot (task_arg fsmargs, xcom_fsm_state *ctxt)
static int	handle_fsm_snapshot_wait (xcom_fsm_state *ctxt)
static void	handle_fsm_exit ()
static int	handle_local_snapshot (task_arg fsmargs, xcom_fsm_state *ctxt)
static int	handle_snapshot (task_arg fsmargs, xcom_fsm_state *ctxt)
static int	handle_fsm_terminate (task_arg fsmargs, xcom_fsm_state *ctxt)
static void	handle_fsm_force_config (task_arg fsmargs)
xcom_fsm_state *	xcom_fsm_impl (xcom_actions action, task_arg fsmargs)
char const *	xcom_fsm (xcom_actions action, task_arg fsmargs)
void	set_app_snap_handler (app_snap_handler x)
void	set_app_snap_getter (app_snap_getter x)
static result	socket_read (connection_descriptor *rfd, void *buf, int n)
static int64_t	socket_read_bytes (connection_descriptor *rfd, char *p, uint32_t n)
connection_descriptor *	xcom_open_client_connection (char const *server, xcom_port port)
static int	xcom_send_proto (connection_descriptor *con, xcom_proto x_proto, x_msg_type x_type, unsigned int tag)
static int	xcom_recv_proto (connection_descriptor *rfd, xcom_proto *x_proto, x_msg_type *x_type, unsigned int *tag)
static int	is_cargo_type (app_data_ptr a, cargo_type t)
	Checks if a given app_data is from a given cargo_type. More...
static char *	get_add_node_address (app_data_ptr a, unsigned int *member)
	Retrieves the address that was used in the add_node request. More...
int	is_node_v4_reachable_with_info (struct addrinfo *retrieved_addr_info)
int	is_node_v4_reachable (char *node_address)
int	are_we_allowed_to_upgrade_to_v6 (app_data_ptr a)
int64_t	xcom_send_client_app_data (connection_descriptor *fd, app_data_ptr a, int force)
int64_t	xcom_client_send_die (connection_descriptor *fd)
void	warn_protoversion_mismatch (connection_descriptor *rfd)
static pax_msg *	socket_read_msg (connection_descriptor *rfd, pax_msg *p)
static xcom_send_app_wait_result	xcom_send_app_wait_and_get (connection_descriptor *fd, app_data *a, int force, pax_msg *p, leader_info_data *leaders)
	Send a message and wait for response. More...
static int	xcom_send_app_wait (connection_descriptor *fd, app_data *a, int force, leader_info_data *leaders)
int	xcom_send_cfg_wait (connection_descriptor *fd, node_list *nl, uint32_t group_id, cargo_type ct, int force)
int	xcom_client_add_node (connection_descriptor *fd, node_list *nl, uint32_t group_id)
int	xcom_client_remove_node (connection_descriptor *fd, node_list *nl, uint32_t group_id)
static int	xcom_check_reply (int const res)
int	xcom_client_get_synode_app_data (connection_descriptor *const fd, uint32_t group_id, synode_no_array *const synodes, synode_app_data_array *const reply)
int	xcom_client_enable_arbitrator (connection_descriptor *fd)
int	xcom_client_disable_arbitrator (connection_descriptor *fd)
int	xcom_client_set_cache_limit (connection_descriptor *fd, uint64_t cache_limit)
int	xcom_client_convert_into_local_server (connection_descriptor *const fd)
void	init_set_max_leaders (uint32_t group_id, app_data *a, node_no max_leaders)
int	xcom_client_set_max_leaders (connection_descriptor *fd, node_no max_leaders, uint32_t group_id)
leader_array	new_leader_array (u_int n, char const *names[])
void	init_set_leaders (uint32_t group_id, app_data *a, leader_array const leaders)
void	init_set_leaders (uint32_t group_id, app_data *a, u_int n, char const *names[])
void	init_set_leaders (uint32_t group_id, app_data *leader_app, leader_array const leaders, app_data *max_app, node_no max_leaders)
void	init_set_leaders (uint32_t group_id, app_data *leader_app, u_int n, char const *names[], app_data *max_app, node_no max_leaders)
int	xcom_client_set_leaders (connection_descriptor *fd, u_int n, char const *names[], uint32_t group_id)
std::unique_ptr< Network_provider_management_interface >	get_network_management_interface ()
std::unique_ptr< Network_provider_operations_interface >	get_network_operations_interface ()
int	xcom_client_set_leaders (connection_descriptor *fd, u_int n, char const *names[], node_no max_leaders, uint32_t group_id)
int	xcom_client_get_leaders (connection_descriptor *fd, uint32_t group_id, leader_info_data *leaders)
static void	paxos_twait (pax_machine *p, unsigned int t)
static void	paxos_twait_cancel (pax_machine *p)
static void	paxos_wakeup (unsigned int t)
static void	paxos_timer_advance ()
static int	paxos_fsm_p1_master_enter (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p1_master_wait (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p2_master_enter (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p2_master_wait (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p2_slave_enter (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p2_slave_wait (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p3_master_wait (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p3_slave_enter (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_p3_slave_wait (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	paxos_fsm_finished (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
static int	accept_new_prepare (pax_machine *paxos, pax_msg *mess)
static int	accept_new_accept (pax_machine *paxos, pax_msg *mess)
static int	own_message (pax_msg *mess, site_def const *site)
static void	action_paxos_prepare (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_paxos_accept (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_paxos_learn (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_paxos_start (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_new_prepare (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_ack_prepare (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_new_accept (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_ack_accept (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	action_ignorant (pax_machine *paxos, site_def const *site, pax_msg *mess)
static void	dispatch_p_event (paxos_state_action *vtbl, pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)
int	paxos_fsm_idle (pax_machine *paxos, site_def const *site, paxos_event event, pax_msg *mess)

Variables
static double	protoversion_warning_time
int const	threephase = 0
	Timestamp of previous protoversion warning. More...
int	ARBITRATOR_HACK = 0
static int const	no_duplicate_payload = 1
static int	use_buffered_read = 1
int	oom_abort = 0
int	xcom_shutdown = 0
synode_no	executed_msg
synode_no	max_synode
task_env *	boot = nullptr
task_env *	detector = nullptr
task_env *	killer = nullptr
task_env *	net_boot = nullptr
task_env *	net_recover = nullptr
void *	xcom_thread_input = nullptr
long	xcom_debug_mask
long	xcom_dbg_stack [DBG_STACK_SIZE]
int	xcom_dbg_stack_top = 0
static task_env *	executor = nullptr
static task_env *	sweeper = nullptr
static task_env *	retry = nullptr
static task_env *	proposer [PROPOSERS]
static task_env *	alive_t = nullptr
static task_env *	cache_task = nullptr
static uint32_t	my_id = 0
static synode_no	current_message
static synode_no	last_config_modification_id
static uint64_t	lsn = 0
static channel	prop_input_queue
synode_pool	synode_number_pool
int	client_boot_done
int	netboot_ok
static linkage	exec_wait
linkage	detector_wait
struct {
int   n
unsigned long   id [MAX_DEAD]
}	dead_sites
static site_def *	forced_config = nullptr
static int	wait_forced_config = 0
static site_def *	executor_site = nullptr
static site_def *	proposer_site = nullptr
static synode_no	delivered_msg = NULL_SYNODE
static synode_no	last_delivered_msg = NULL_SYNODE
static synode_no *	proposer_synodes [PROPOSERS]
static xcom_state_change_cb	xcom_run_cb = nullptr
static xcom_state_change_cb	xcom_terminate_cb = nullptr
static xcom_state_change_cb	xcom_comms_cb = nullptr
static xcom_state_change_cb	xcom_exit_cb = nullptr
static xcom_state_change_cb	xcom_expel_cb = nullptr
static xcom_input_try_pop_cb	xcom_try_pop_from_input_cb = nullptr
static xcom_recovery_cb	recovery_begin_cb = nullptr
static xcom_recovery_cb	recovery_restart_cb = nullptr
static xcom_recovery_cb	recovery_init_cb = nullptr
static xcom_recovery_cb	recovery_end_cb = nullptr
static connection_descriptor *	input_signal_connection {nullptr}
	These fields are used to signal XCom's request queue. More...
connection_descriptor *	input_signal_connection_pipe {nullptr}
int	pipe_signal_connections [2] = {-1, -1}
static constexpr xcom_proto	single_writer_support = x_1_9
static int	prop_started = 0
static int	prop_finished = 0
static constexpr xcom_proto	first_protocol_that_ignores_intermediate_forced_configs_or_views
static xcom_proto const	first_event_horizon_aware_protocol = x_1_4
struct {
int   n
int   front
int   rear
synode_no   q [FIFO_SIZE]
}	delay_fifo
static struct fp_name	oblist []
static execute_context *	debug_xc
static double	sent_alive = 0.0
static int	clicnt = 0
static synode_no	start_config = NULL_SYNODE
static msg_handler	dispatch_table [LAST_OP]
static app_snap_getter	get_app_snap_cb
static app_snap_handler	handle_app_snap_cb
static task_env *	x_timer = nullptr
const char *	xcom_actions_name [] = {x_actions}
static int	snapshots [NSERVERS]
static synode_no	log_start_max
static synode_no	log_end_max
static unsigned int	current_tick = 0
static linkage	time_queue [paxos_timer_range]
unsigned constexpr int const	paxos_default_timeout = 100
paxos_state_action	p1_idle_vtbl [last_p_event]
paxos_state_action	p1_master_enter_vtbl [last_p_event]
paxos_state_action	p1_master_wait_vtbl [last_p_event]
paxos_state_action	p2_master_enter_vtbl [last_p_event]
paxos_state_action	p2_master_wait_vtbl [last_p_event]
paxos_state_action	p2_slave_wait_vtbl [last_p_event]
paxos_state_action	p3_master_wait_vtbl [last_p_event]
paxos_state_action	p3_slave_wait_vtbl [last_p_event]
paxos_state_action	p_finished_vtbl [last_p_event]
const char *	paxos_event_name [] = {p_events}

Detailed Description

xcom/xcom_base.c The new version of xcom is a major rewrite to allow transmission of multiple messages from several sources simultaneously without collision.

The interface to xcom is largely intact, one notable change is that xcom will consider the message delivered as soon as it has got a majority. Consequently, the VP set will not necessarily show all nodes which will actually receive the message.

OHKFIX Add wait for complete last known node set to mimic the old semantics.

IMPORTANT: What xcom does and what it does not do:

xcom messages are received in the same order on all nodes.

xcom guarantees that if a message is delivered to one node, it will eventually be seen on all other nodes as well.

xcom messages are available to a crashed node when it comes up again if at least one node which knows the value of the message has not crashed. The size of the message cache is configurable.

OHKFIX Add logging to disk to make messages durable across system crash and to increase the number of messages which may be cached.

There is no guarantee whatsoever about the order of messages from different nodes, not even the order of multiple messages from the same node. It is up to the client to impose such an order by waiting on a message before it sends the next.

xcom can notify the client that a message has timed out, and in that case will try to cancel the message, but it cannot guarantee that a message which has timed out will not be delivered.

xcom attaches a node set to each message as it is delivered to the client. This node set reflects the current node set that xcom believes is active, it does not mean that the message has been delivered yet to all nodes in the set. Neither does it mean that the message has not been delivered to the nodes not in the set.

A cache of Paxos state machines is central to the new design. The purpose of the cache is both to store a window of messages, and to decouple the different parts of xcom, like message proposal, message delivery and execution, and recovery. The old cache was limited to caching messages, and a single state machine ran the combined VP and Paxos algorithm. This constrained xcom to deliver only a single message at a time.

Each instance of the Paxos state machine implements the basic Paxos protocol. Unlike the cache in the old system, it is not cleared when a site is deleted. This removes some problems related to message delivery during site deletion. The cache is a classic fixed size LRU with a hash index.

Some extensions to the basic Paxos algorithm has been implemented:

A node has ownership to all synodes with its own node number. Only a node with node number N can propose a value for synode {X N}, where X is the sequence number, and N is the node number. Other nodes can only propose the special value no_op for synode {X N}. The reason for this is to retain the leaderless Paxos algorithm, but to avoid collisions between nodes which are competing for the same synode number. With this scheme, each node has its own unique number series during normal operation. The scheme has the following implications:

1. If a node N has not already proposed a value for the synode {X N}, it may at any time send a LEARN message to the other nodes with the reserved value no_op, without going through phase 1 and 2 of Paxos. This is because the other nodes are constrained to propose no_op for this synode, so the final outcome will always be no_op. To avoid unnecessary message transmission, a node will try to broadcast the no_op LEARN messages by piggybacking the information on the messages of the basic Paxos protocol.
2. Other nodes which want to find the value of synode {X N} may do so by trying to get the value no_op accepted by following the basic Paxos algorithm. The result will be the actual value proposed by node N if it has done so, otherwise no_op. This will typically only be necessary when a node is down, and the other nodes need to find the values from the missing node in order to be able to continue execution.

Messages are delivered in order to the client, and the order is determined by the sequence number and the node number, with the sequence number as the most significant part.

The xcom network interface has been redesigned and is now implemented directly on top of TCP, and has so far been completely trouble free. We use poll() or select() to implement non-blocking send and receive, but libev could equally well have been used.

Multicast is implemented on top of unicast as before, but the implementation is prepared to use real multicast with relatively minor changes.

The roles of proposer, acceptor/learner, and executor are now directly mapped to unique task types which interact with the Paxos state machines, whereas the previous implementation folded all the roles into a single event driven state machine.

The following terminology will be used:

A node is an instance of the xcom thread. There is only one instance of the xcom thread in the agent. A client is the application which is using xcom to send messages. A thread is a real OS thread. A task is a logical process. It is implemented by coroutines and an explicit stack.

The implementation of tasks and non-blocking socket operations is isolated in task.h and task.c.

A node will open a tcp connection to each of the other nodes. This connection is used for all communication initiated by the node, and replies to messages will arrive on the connection on which it was sent.

static int tcp_server(task_arg);

The tcp_server listens on the xcom port and starts an acceptor_learner_task whenever a new connection is detected.

static int tcp_reaper_task(task_arg);

Closes tcp connection which have been unused for too long.

static int sender_task(task_arg);

The sender_task waits for tcp messages on its input queue and sends it on the tcp socket. If the socket is closed for any reason, the sender_task will reconnect the socket. There is one sender_task for each socket. The sender task exists mainly to simplify the logic in the other tasks, but it could have been replaced with a coroutine which handles the connection logic after having reserved the socket for its client task.

static int generator_task(task_arg);

The generator_task reads messages from the client queue and moves them into the input queue of the proposer_task.

OHKFIX Use a tcp socket instead of the client queue. We can then remove the generator_task and let the acceptor_learner_task do the dispatching.

static int proposer_task(task_arg);

Assign a message number to an outgoing message and try to get it accepted. There may be several proposer tasks on each node working in parallel. If there are multiple proposer tasks, xcom can not guarantee that the messages will be sent in the same order as received from the client.

static int acceptor_learner_task(task_arg);

This is the server part of the xcom thread. There is one acceptor_learner_task for each node in the system. The acceptor learner_task reads messages from the socket, finds the correct Paxos state machine, and dispatches to the correct message handler with the state machine and message as arguments.

static int reply_handler_task(task_arg);

The reply_handler_task does the same job as the acceptor_learner_task, but listens on the socket which the node uses to send messages, so it will handle only replies on that socket.

static int executor_task(task_arg);

The ececutor_task waits for a Paxos message to be accpeted. When the message is accepted, it is delivered to the client, unless it is a no-op. In either case, the executor_task steps to the next message and repeats the wait. If it times out waiting for a message, it will try to get a no-op accepted.

static int alive_task(task_arg);

Sends i-am-alive to other nodes if there has been no normal traffic for a while. It also pings nodes which seem to be inactive.

static int detector_task(task_arg);

The detector_task periodically scans the set of connections from other nodes and sees if there has been any activity. If there has been no activity for some time, it will assume that the node is dead, and send a view message to the client.

Reconfiguration:

The xcom reconfiguration process is essentially the one described in "Reconfiguring a State Machine" by Lamport et al. as the R-alpha algorithm. We execute the reconfiguration command immediately, but the config is only valid after a delay of alpha messages. The parameter alpha is the same as EVENT_HORIZON in this implementation. :/static.*too_far All tcp messages from beyond the event horizon will be ignored.

Macro Definition Documentation

__STDC_FORMAT_MACROS

#define __STDC_FORMAT_MACROS

BAL_FMT

#define BAL_FMT   "ballot {cnt %d node %d}"

BAL_MEM

#define BAL_MEM

(

x

)

(x).cnt, (x).node

CONNECT_FAIL

#define CONNECT_FAIL

Value:

  ret_fd = -1;       \
  goto end

CREATE_REPLY

#define CREATE_REPLY

(

x

)

Value:

  pax_msg *reply = NULL; \
  CLONE_PAX_MSG(reply, x)

FIFO_SIZE

#define FIFO_SIZE   1000

FNVSTART

#define FNVSTART   0x811c9dc5

GOTO

#define GOTO

(

x

)

Value:

  {                                             \
    IFDBG(D_NONE, STRLIT("goto "); STRLIT(#x)); \
    goto x;                                     \
  }

IS_CONS_ALL

#define IS_CONS_ALL

(

p

)

((p)->proposer.msg->a ? (p)->proposer.msg->a->consensus == cons_all : 0)

NAME

#define NAME

(

f

)

{ f, #f }

PING_GATHERING_TIME_WINDOW

#define PING_GATHERING_TIME_WINDOW   5.0

PINGS_GATHERED_BEFORE_CONNECTION_SHUTDOWN

#define PINGS_GATHERED_BEFORE_CONNECTION_SHUTDOWN   3

PROP_ITER

#define PROP_ITER

Value:

  int i;          \
  for (i = 0; i < PROPOSERS; i++)

PROTOVERSION_WARNING_TIMEOUT

#define PROTOVERSION_WARNING_TIMEOUT   600.0 /** Every 10 minutes */

reply_msg

#define reply_msg

(

m

)

Value:

  {                                                               \
    if (is_local_node((m)->from, site)) {                         \
      dispatch_op(site, m, NULL);                                 \
    } else {                                                      \
      link_into(&(msg_link_new((m), (m)->from)->l), reply_queue); \
    }                                                             \
  }

SEND_REPLY

#define SEND_REPLY

Value:

  reply_msg(reply); \
  replace_pax_msg(&reply, NULL)

SERIALIZE_REPLY

#define SERIALIZE_REPLY

(

msg

)

Value:

  msg->to = ep->p->from;                    \
  msg->from = ep->p->to;                    \
  msg->delivered_msg = get_delivered_msg(); \
  msg->max_synode = get_max_synode();       \
  serialize_msg(msg, ep->rfd->x_proto, &ep->buflen, &ep->buf);

SET_X_FSM_STATE

#define SET_X_FSM_STATE

(

s

)

Value:

  do {                     \
    ctxt->state_fp = s;    \
    ctxt->state_name = #s; \
  } while (0)

SYS_STRERROR_SIZE

#define SYS_STRERROR_SIZE   512

TICK_PERIOD

#define TICK_PERIOD   0.01

WRITE_REPLY

#define WRITE_REPLY

Value:

  if (ep->buflen) {                                                    \
    int64_t sent;                                                      \
    IFDBG(D_TRANSPORT, FN; STRLIT("task_write "); NDBG(ep->rfd.fd, d); \
          NDBG(ep->buflen, u));                                        \
    TASK_CALL(task_write(ep->rfd, ep->buf, ep->buflen, &sent));        \
    send_count[ep->p->op]++;                                           \
    send_bytes[ep->p->op] += ep->buflen;                               \
    X_FREE(ep->buf);                                                   \
  }                                                                    \
  ep->buf = NULL;

X [1/2]

#define X

(

b

)

#b

X [2/2]

#define X

(

b

)

#b

X_FSM_STATE

#define X_FSM_STATE

(

s

)

{ s, #s }

XCOM_SEND_APP_WAIT_TIMEOUT

#define XCOM_SEND_APP_WAIT_TIMEOUT   20

Typedef Documentation

allow_event_horizon_result

typedef enum allow_event_horizon_result allow_event_horizon_result

exec_fp

typedef void(* exec_fp) (execute_context *xc)

execute_context

typedef struct execute_context execute_context

paxos_state_action

typedef void(* paxos_state_action) (pax_machine *paxos, site_def const *site, pax_msg *mess)

unsafe_node_check

typedef bool(* unsafe_node_check) (node_address const *node)

xcom_fsm_fp

typedef int(* xcom_fsm_fp) (xcom_actions action, task_arg fsmargs, xcom_fsm_state *ctxt)

xcom_fsm_state

typedef struct xcom_fsm_state xcom_fsm_state

xcom_send_app_wait_result

typedef enum xcom_send_app_wait_result xcom_send_app_wait_result

Enumeration Type Documentation

anonymous enum

anonymous enum

Enumerator
TAG_START

anonymous enum

anonymous enum

Enumerator
paxos_timer_range

allow_event_horizon_result

enum allow_event_horizon_result

Check if we can reconfigure the event horizon.

We can reconfigure the event horizon if all group members support reconfiguring the event horizon, and the new event horizon in the domain [EVENT_HORIZON_MIN, EVENT_HORIZON_MAX].

We use the group's latest common XCom protocol as a proxy to decide if all members support reconfiguring the event horizon.

If the common protocol is at least version 5 (x_1_4) then all members run compatible server instances.

Otherwise there are older instances, and it follows that the event horizon must be the default and cannot be reconfigured.

Enumerator
EVENT_HORIZON_ALLOWED
EVENT_HORIZON_INVALID
EVENT_HORIZON_UNCHANGEABLE

synode_allocation_type

enum class synode_allocation_type

strong

Enumerator
todo
local
remote
global

synode_reservation_status

enum class synode_reservation_status : int

strong

Enumerator
number_ok
no_nodes
delivery_timeout

xcom_send_app_wait_result

enum xcom_send_app_wait_result

Enumerator
SEND_REQUEST_FAILED
RECEIVE_REQUEST_FAILED
REQUEST_BOTCHED
RETRIES_EXCEEDED
REQUEST_OK_RECEIVED
REQUEST_FAIL_RECEIVED
REQUEST_OK_REDIRECT

Function Documentation

abort_processing()

static int abort_processing ( pax_msg *  p )

inlinestatic

accept_new_accept()

static int accept_new_accept ( pax_machine *  paxos, pax_msg *  mess  )

static

accept_new_prepare()

static int accept_new_prepare ( pax_machine *  paxos, pax_msg *  mess  )

static

accepted()

static int accepted ( pax_machine *  p )

inlinestatic

accepted_noop()

static int accepted_noop ( pax_machine *  p )

inlinestatic

acceptor_learner_task()

int acceptor_learner_task

(

task_arg

arg

)

action_ack_accept()

static void action_ack_accept ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_ack_prepare()

static void action_ack_prepare ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_ignorant()

static void action_ignorant ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_new_accept()

static void action_new_accept ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_new_prepare()

static void action_new_prepare ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_paxos_accept()

static void action_paxos_accept ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_paxos_learn()

static void action_paxos_learn ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_paxos_prepare()

static void action_paxos_prepare ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

action_paxos_start()

static void action_paxos_start ( pax_machine *  paxos, site_def const *  site, pax_msg *  mess  )

static

activate_sweeper()

static void activate_sweeper ( )

static

active_leaders()

static void active_leaders ( site_def *  site, leader_array *  leaders  )

static

add_event_horizon()

static synode_no add_event_horizon ( synode_no  s )

static

Add the event horizon to the given base synod s.

We are assuming right now that this function is used solely in the context of "we have received a reconfiguration command at synod s, when should it be scheduled to take effect?" The result of this function is when it should take effect.

Common case: there are no configurations pending, or if there are, none of them reconfigure the event horizon. The common case result is:

s + event_horizon(active_config) + 1

If an event horizon reconfiguration R is pending, it means that the command C proposed for synod s is concurrent with R, i.e., s falls in the interval ]proposed(R), start(R)[.

In this situation we apply the command C proposed for synod s after taking into account R's event horizon.

This means that the result is:

start(R) + event_horizon(R) + 1

add_node_adding_own_address()

static bool_t add_node_adding_own_address ( app_data_ptr  a )

static

This will test if we are receiving a boot request that contains ourselves.

This could happed in case of a misconfiguration of a local_address, that causes an add_node request to be erroneous delivered.

Parameters

a	app_data with an add node request

Returns

bool_t TRUE in case of error, meaning that my address is in the add_node list

add_node_test_connectivity_to_added_nodes()

static bool add_node_test_connectivity_to_added_nodes ( node_address *  nodes_to_change, u_int  number_of_nodes_to_change  )

static

add_node_unsafe_against_ipv4_old_nodes()

static bool_t add_node_unsafe_against_ipv4_old_nodes ( app_data_ptr  a )

static

add_proposer_synode()

static void add_proposer_synode ( int  i, synode_no *  syn_ptr  )

static

addone()

static int addone ( int  i )

inlinestatic

alive_node()

static bool alive_node ( site_def const *  site, u_int  i  )

inlinestatic

alive_task()

int alive_task

(

task_arg

arg

)

alloc_node_set()

node_set * alloc_node_set	(	node_set *	set,
		u_int	n
	)

allow_add_node()

static u_int allow_add_node ( app_data_ptr  a )

static

allow_channel_takeover()

static bool allow_channel_takeover ( site_def const *  site )

static

allow_event_horizon()

static allow_event_horizon_result allow_event_horizon ( xcom_event_horizon  event_horizon )

static

allow_remove_node()

static u_int allow_remove_node ( app_data_ptr  a )

static

analyze_leaders()

void analyze_leaders

(

site_def *

site

)

are_there_dead_nodes_in_new_config()

static bool_t are_there_dead_nodes_in_new_config ( app_data_ptr  a )

static

are_we_allowed_to_upgrade_to_v6()

int are_we_allowed_to_upgrade_to_v6

(

app_data_ptr

a

)

assign_lsn()

static uint64_t assign_lsn ( )

static

Assign the next log sequence number (lsn) for a message.

Initial propose sets lsn to msgno of the max message number as safe starting point, otherwise lsn shall be ever increasing. lsn ensures sender order known on receiver side, as messages may arrive "out of order" due to retransmission. We use max_synode instead of current_message to avoid any conflict with lsn allocated by a previous instance of the node.

backwards_compatible()

static bool_t backwards_compatible ( xcom_event_horizon  event_horizon )

static

better_snapshot()

static int better_snapshot ( gcs_snapshot *  gcs )

static

brand_app_data()

static void brand_app_data ( pax_msg *  p )

static

brand_client_msg()

static void brand_client_msg ( pax_msg *  msg, synode_no  msgno  )

static

broadcast_noop()

static void broadcast_noop ( synode_no  find, pax_machine *  p  )

static

bury_site()

static void bury_site ( uint32_t  id )

static

cache_manager_task()

int cache_manager_task

(

task_arg

arg

)

can_execute_cfgchange()

static client_reply_code can_execute_cfgchange ( pax_msg *  p )

static

Validates if a configuration command can be executed.

Checks whether the configuration command is aimed at the correct group. Checks whether the configuration command pertains to a node reincarnation.

Parameters

p	a pointer to the pax_msg of the configuration command

Return values

REQUEST_OK	if the reconfiguration command can be executed
REQUEST_RETRY	if XCom is still booting
REQUEST_FAIL	if the configuration command cannot be executed

can_send_snapshot()

static int can_send_snapshot ( )

static

check_if_add_node_is_unsafe()

static bool check_if_add_node_is_unsafe ( app_data_ptr  a, unsafe_node_check  unsafe  )

static

check_if_add_node_is_unsafe_against_event_horizon()

static bool check_if_add_node_is_unsafe_against_event_horizon ( app_data_ptr  a )

static

check_learn()

static pax_msg * check_learn ( site_def const *  site, pax_machine *  p  )

static

check_propose()

bool_t check_propose	(	site_def const *	site,
		pax_machine *	p
	)

clone_dispatch_table()

static msg_handler * clone_dispatch_table ( msg_handler const *  proto )

static

compute_delay()

static synode_no compute_delay ( synode_no  start, xcom_event_horizon  event_horizon  )

static

Returns the message number where it is safe for nodes in previous configuration to exit.

Parameters

start	start synod of the next configuration
event_horizon	event horizon of the next configuration

create_ack_accept_msg()

static pax_msg * create_ack_accept_msg ( pax_msg *  m, synode_no  synode  )

static

create_ack_prepare_msg()

static pax_msg * create_ack_prepare_msg ( pax_machine *  p, pax_msg *  pm, synode_no  synode  )

static

create_learn_msg_for_ignorant_node()

static pax_msg * create_learn_msg_for_ignorant_node ( pax_machine *  p, pax_msg *  pm, synode_no  synode  )

static

create_noop()

pax_msg * create_noop

(

pax_msg *

p

)

Initializes the message p as a Prepare message for a no-op, as in the message for Phase 1 (a) of the Paxos protocol.

Executed by Proposers.

Parameters

p	The no-op message to send

Return values

created

paxos message of type no_op

create_proposers()

static void create_proposers ( )

static

create_read()

static pax_msg * create_read ( site_def const *  site, pax_msg *  p  )

static

create_site_def_with_start()

static site_def * create_site_def_with_start ( app_data_ptr  a, synode_no  start  )

static

create_snapshot()

static gcs_snapshot * create_snapshot ( )

static

create_tiny_learn_msg()

static pax_msg * create_tiny_learn_msg ( pax_machine *  pm, pax_msg *  p  )

static

debug_loser()

static void debug_loser ( synode_no  x )

static

detector_task()

int detector_task

(

task_arg

arg

)

dispatch_get_event_horizon()

void dispatch_get_event_horizon	(	site_def const *	site,
		pax_msg *	p,
		linkage *	reply_queue
	)

dispatch_get_leaders()

void dispatch_get_leaders	(	site_def *	site,
		pax_msg *	p,
		linkage *	reply_queue
	)

dispatch_get_synode_app_data()

void dispatch_get_synode_app_data	(	site_def const *	site,
		pax_msg *	p,
		linkage *	reply_queue
	)

dispatch_op()

pax_msg * dispatch_op	(	site_def const *	site,
		pax_msg *	p,
		linkage *	reply_queue
	)

dispatch_p_event()

static void dispatch_p_event ( paxos_state_action *  vtbl, pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

inlinestatic

do_learn()

static void do_learn ( site_def const *  site, pax_machine *  p, pax_msg *  m  )

static

dump_debug_exec_state()

static void dump_debug_exec_state ( )

static

dump_exec_state()

static void dump_exec_state ( execute_context *  xc, long  dbg  )

static

empty_prop_input_queue()

static void empty_prop_input_queue ( )

static

empty_synode_number_pool()

static void empty_synode_number_pool ( )

static

execute_msg()

void execute_msg	(	site_def *	site,
		pax_machine *	pma,
		pax_msg *	p
	)

executor_task()

static int executor_task ( task_arg  arg )

static

fifo_empty()

static int fifo_empty ( )

inlinestatic

fifo_extract()

static synode_no fifo_extract ( )

inlinestatic

fifo_front()

static synode_no fifo_front ( )

inlinestatic

fifo_full()

static int fifo_full ( )

inlinestatic

fifo_insert()

static void fifo_insert ( synode_no  s )

inlinestatic

find_value()

static void find_value ( site_def const *  site, unsigned int *  wait, int  n  )

static

finished()

static int finished ( pax_machine *  p )

inlinestatic

first_event_horizon_reconfig()

static site_def const * first_event_horizon_reconfig ( )

static

Get the first pending configuration that reconfigures the event horizon.

Retrieve the first pending site_def, i.e. with the smallest start synod that is greater than executed_msg, that reconfigures the event horizon.

first_free_synode_local()

static synode_no first_free_synode_local ( synode_no  msgno )

static

fnv_hash()

static uint32_t fnv_hash ( unsigned char *  buf, size_t  length, uint32_t  sum  )

static

force_interval()

static void force_interval ( synode_no  start, synode_no  end, int  enforcer  )

static

force_pax_machine()

static void force_pax_machine ( pax_machine *  p, int  enforcer  )

static

found_active_leaders()

node_no found_active_leaders

(

site_def *

site

)

free_forced_config_site_def()

static void free_forced_config_site_def ( )

static

get_add_node_address()

static char * get_add_node_address ( app_data_ptr  a, unsigned int *  member  )

static

Retrieves the address that was used in the add_node request.

Parameters

a	app data containing the node to add
member	address we used to present ourselves to other nodes

Returns

char* a pointer to the address being added.

get_current_message()

synode_no get_current_message

(

)

get_delivered_msg()

synode_no get_delivered_msg

(

)

get_executor_site()

site_def const * get_executor_site

(

)

get_executor_site_rw()

site_def * get_executor_site_rw

(

)

get_last_delivered_msg()

synode_no get_last_delivered_msg

(

)

get_leader()

static node_no get_leader ( site_def const *  s )

static

get_max_synode()

synode_no get_max_synode

(

)

get_my_xcom_id()

uint32_t get_my_xcom_id

(

)

get_network_management_interface()

std::unique_ptr< Network_provider_management_interface > get_network_management_interface

(

)

get_network_operations_interface()

std::unique_ptr< Network_provider_operations_interface > get_network_operations_interface

(

)

get_proposer_site()

site_def const * get_proposer_site

(

)

get_proposer_synode()

static synode_no get_proposer_synode ( int  i )

static

get_sweep_start()

static synode_no get_sweep_start ( )

static

get_xcom_message()

int get_xcom_message	(	pax_machine **	p,
		synode_no	msgno,
		int	n
	)

getstart()

static synode_no getstart ( app_data_ptr  a )

static

global_synode_allocator()

static synode_no global_synode_allocator ( site_def *  site, synode_no  synode  )

static

got_all_snapshots()

static int got_all_snapshots ( )

static

handle_accept()

static void handle_accept ( site_def const *  site, pax_machine *  p, linkage *  reply_queue, pax_msg *  m  )

static

handle_ack_accept()

static void handle_ack_accept ( site_def const *  site, pax_machine *  p, pax_msg *  m  )

static

handle_ack_prepare()

static void handle_ack_prepare ( site_def const *  site, pax_machine *  p, pax_msg *  m  )

static

handle_add_node()

site_def * handle_add_node

(

app_data_ptr

a

)

Reconfigure the group membership: add new member(s).

It is possible that concurrent reconfigurations take effect between the time this reconfiguration was proposed and now.

Particularly, it is possible that any of the concurrent reconfigurations modified the event horizon and that the new member(s) do not support event horizon reconfigurations.

We account for these situations by validating if adding the new members is still possible under the current state.

If it is not, this reconfiguration does not produce any effect, i.e. no new configuration is installed.

handle_alive()

static void handle_alive ( site_def const *  site, linkage *  reply_queue, pax_msg *  pm  )

inlinestatic

handle_boot()

static void handle_boot ( site_def const *  site, linkage *  reply_queue, pax_msg *  p  )

inlinestatic

handle_client_msg()

static void handle_client_msg ( pax_msg *  p )

static

handle_config()

bool_t handle_config	(	app_data_ptr	a,
		bool const	forced
	)

handle_event_horizon()

bool_t handle_event_horizon

(

app_data_ptr

a

)

Reconfigure the event horizon.

It is possible that concurrent reconfigurations take effect between the time this reconfiguration was proposed and now.

Particularly, it is possible that any of the concurrent reconfigurations added a new member which does not support reconfiguring the event horizon.

We account for these situations by validating if the event horizon reconfiguration is still possible under the current state.

If it is not, this reconfiguration does not produce any effect, i.e. no new configuration is installed.

handle_fsm_exit()

static void handle_fsm_exit ( )

static

handle_fsm_force_config()

static void handle_fsm_force_config ( task_arg  fsmargs )

static

handle_fsm_net_boot()

static int handle_fsm_net_boot ( task_arg  fsmargs, xcom_fsm_state *  ctxt, int  cont  )

static

handle_fsm_snapshot()

static int handle_fsm_snapshot ( task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

handle_fsm_snapshot_wait()

static int handle_fsm_snapshot_wait ( xcom_fsm_state *  ctxt )

static

handle_fsm_terminate()

static int handle_fsm_terminate ( task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

handle_leaders()

bool_t handle_leaders

(

app_data_ptr

a

)

handle_learn()

void handle_learn	(	site_def const *	site,
		pax_machine *	p,
		pax_msg *	m
	)

Process the incoming Learn message.

Executed by Learners.

Parameters

site	XCom configuration
p	Paxos instance
m	Incoming message

handle_local_snapshot()

static int handle_local_snapshot ( task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

handle_max_leaders() [1/2]

bool_t handle_max_leaders

(

app_data_ptr

a

)

handle_max_leaders() [2/2]

static bool_t handle_max_leaders ( site_def *  new_config, app_data_ptr  a  )

static

handle_need_snapshot()

static void handle_need_snapshot ( linkage *  reply_queue, pax_msg *  pm  )

static

handle_prepare()

static void handle_prepare ( site_def const *  site, pax_machine *  p, linkage *  reply_queue, pax_msg *  pm  )

static

handle_read()

static void handle_read ( site_def const *  site, pax_machine *  p, linkage *  reply_queue, pax_msg *  pm  )

static

handle_remove_node()

site_def * handle_remove_node

(

app_data_ptr

a

)

handle_set_leaders() [1/2]

bool_t handle_set_leaders

(

app_data_ptr

a

)

handle_set_leaders() [2/2]

static bool_t handle_set_leaders ( site_def *  new_config, app_data_ptr  a  )

static

handle_simple_accept()

pax_msg * handle_simple_accept	(	pax_machine *	p,
		pax_msg *	m,
		synode_no	synode
	)

Process the incoming Accept from a Proposer, as in the message for Phase 2 (a) of the Paxos protocol.

Executed by Acceptors.

Parameters

p	Paxos instance
m	Incoming Accept message
synode	Synode of the Paxos instance/Accept message

Return values

pax_msg*	the reply to send to the Proposer (as in the Phase 2 (b) message of the Paxos protocol) if the Acceptor accepts the Accept
NULL	otherwise

handle_simple_ack_accept()

pax_msg * handle_simple_ack_accept	(	site_def const *	site,
		pax_machine *	p,
		pax_msg *	m
	)

Process the incoming acknowledge from an Acceptor to a sent Accept, as in the message for Phase 2 (b) of the Paxos protocol.

Executed by Proposers.

Parameters

site	XCom configuration
p	Paxos instance
m	Incoming message

Return values

pax_msg*	the Learn message to send to Leaners if a majority of Acceptors replied to the Proposer's Accept
NULL	otherwise

handle_simple_ack_prepare()

bool_t handle_simple_ack_prepare	(	site_def const *	site,
		pax_machine *	p,
		pax_msg *	m
	)

Process the incoming acknowledge from an Acceptor to a sent Prepare, as in the message for Phase 1 (b) of the Paxos protocol.

Executed by Proposers.

Parameters

site	XCom configuration
p	Paxos instance
m	Incoming message

Return values

TRUE	if a majority of Acceptors replied to the Proposer's Prepare
FALSE	otherwise

handle_simple_prepare()

pax_msg * handle_simple_prepare	(	pax_machine *	p,
		pax_msg *	pm,
		synode_no	synode
	)

Process the incoming Prepare message from a Proposer, as in the message for Phase 1 (a) of the Paxos protocol.

Executed by Acceptors.

Parameters

p	Paxos instance
pm	Incoming Prepare message
synode	Synode of the Paxos instance/Accept message

Return values

pax_msg*	the reply to send to the Proposer (as in the Phase 1 (b) message of the Paxos protocol) if the Acceptor accepts the Prepare
NULL	otherwise

handle_skip()

static void handle_skip ( site_def const *  site, pax_machine *  p, pax_msg *  m  )

static

handle_snapshot()

static int handle_snapshot ( task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

handle_tiny_learn()

void handle_tiny_learn	(	site_def const *	site,
		pax_machine *	pm,
		pax_msg *	p
	)

Process the incoming tiny, i.e.

without the learned value, Learn message. Executed by Learners.

Parameters

site	XCom configuration
pm	Paxos instance
p	Incoming message

handle_x_snapshot()

static void handle_x_snapshot ( gcs_snapshot *  gcs )

static

harmless()

static int harmless ( pax_msg const *  p )

static

iamthegreatest()

int iamthegreatest

(

site_def const *

s

)

ignore_message()

static int ignore_message ( synode_no  x, site_def *  site, char const *  dbg  )

inlinestatic

ignoresig()

static int ignoresig ( int  signum )

static

incompatible_proto_and_leaders() [1/2]

static bool incompatible_proto_and_leaders ( node_address const *  node )

static

incompatible_proto_and_leaders() [2/2]

static bool incompatible_proto_and_leaders ( xcom_proto  x_proto )

static

incompatible_proto_and_max_leaders() [1/2]

static bool incompatible_proto_and_max_leaders ( node_address const *  node )

static

incompatible_proto_and_max_leaders() [2/2]

static bool incompatible_proto_and_max_leaders ( xcom_proto  x_proto, node_no  max_leaders  )

static

incr_msgno()

static synode_no incr_msgno ( synode_no  msgno )

static

incr_synode()

synode_no incr_synode

(

synode_no

synode

)

inform_removed()

static void inform_removed ( int  index, int  all  )

static

init_app_msg()

app_data_ptr init_app_msg	(	app_data *	a,
		char *	payload,
		u_int	payload_size
	)

init_base_vars()

void init_base_vars

(

)

init_config_with_group()

app_data_ptr init_config_with_group	(	app_data *	a,
		node_list *	nl,
		cargo_type	type,
		uint32_t	group_id
	)

init_convert_into_local_server_msg()

app_data_ptr init_convert_into_local_server_msg

(

app_data *

a

)

init_get_event_horizon_msg()

app_data_ptr init_get_event_horizon_msg	(	app_data *	a,
		uint32_t	group_id
	)

init_get_leaders_msg()

app_data_ptr init_get_leaders_msg	(	app_data *	a,
		uint32_t	group_id
	)

init_get_msg()

app_data_ptr init_get_msg	(	app_data *	a,
		uint32_t	group_id,
		cargo_type const	t
	)

init_get_synode_app_data_msg()

static app_data_ptr init_get_synode_app_data_msg ( app_data *  a, uint32_t  group_id, synode_no_array *const  synodes  )

static

init_learn_msg()

static void init_learn_msg ( pax_msg *  p )

static

init_need_boot_op()

void init_need_boot_op	(	pax_msg *	p,
		node_address *	identity
	)

Initializes the message p as a need_boot_op message.

Parameters

p	The message to send
identity	The unique incarnation identifier of this XCom instance

init_node_set()

node_set * init_node_set	(	node_set *	set,
		u_int	n
	)

init_noop()

static site_def const * init_noop ( synode_no  find, pax_machine *  p  )

static

init_prepare_msg()

void init_prepare_msg

(

pax_msg *

p

)

Initializes the message p as a Prepare message, as in the message for Phase 1 (a) of the Paxos protocol.

Executed by Proposers.

Parameters

p	The message to send

init_propose_msg()

void init_propose_msg

(

pax_msg *

p

)

Initializes the message p as an Accept message, as in the message for Phase 2 (a) of the Paxos protocol.

Executed by Proposers.

Parameters

p	The message to send

init_proposers()

static void init_proposers ( )

static

init_set_cache_size_msg()

app_data_ptr init_set_cache_size_msg	(	app_data *	a,
		uint64_t	cache_limit
	)

init_set_event_horizon_msg()

app_data_ptr init_set_event_horizon_msg	(	app_data *	a,
		uint32_t	group_id,
		xcom_event_horizon	event_horizon
	)

init_set_leaders() [1/4]

void init_set_leaders	(	uint32_t	group_id,
		app_data *	a,
		leader_array const	leaders
	)

init_set_leaders() [2/4]

void init_set_leaders	(	uint32_t	group_id,
		app_data *	a,
		u_int	n,
		char const *	names[]
	)

init_set_leaders() [3/4]

void init_set_leaders	(	uint32_t	group_id,
		app_data *	leader_app,
		leader_array const	leaders,
		app_data *	max_app,
		node_no	max_leaders
	)

init_set_leaders() [4/4]

void init_set_leaders	(	uint32_t	group_id,
		app_data *	leader_app,
		u_int	n,
		char const *	names[],
		app_data *	max_app,
		node_no	max_leaders
	)

init_set_max_leaders()

void init_set_max_leaders	(	uint32_t	group_id,
		app_data *	a,
		node_no	max_leaders
	)

init_tasks()

static void init_tasks ( )

static

init_time_queue()

static void init_time_queue ( )

static

init_xcom_base()

void init_xcom_base

(

)

Reset lsn

initialize_lsn()

void initialize_lsn

(

uint64_t

n

)

install_ng_with_start()

static site_def * install_ng_with_start ( app_data_ptr  a, synode_no  start  )

static

install_node_group()

site_def * install_node_group

(

app_data_ptr

a

)

is_active_leader()

int is_active_leader	(	node_no	x,
		site_def *	site
	)

is_busy()

static int is_busy ( synode_no  s )

static

is_cargo_type()

static int is_cargo_type ( app_data_ptr  a, cargo_type  t  )

inlinestatic

Checks if a given app_data is from a given cargo_type.

Parameters

a	the app_data
t	the cargo type

Returns

int TRUE (1) if app_data a is from cargo_type t

is_config()

static int is_config ( cargo_type  x )

inlinestatic

is_dead_site()

static bool_t is_dead_site ( uint32_t  id )

static

is_empty_site()

static int is_empty_site ( site_def const *  s )

inlinestatic

is_forcing_node()

static int is_forcing_node ( pax_machine const *  p )

static

is_latest_config()

static bool_t is_latest_config ( site_def const *const  config )

static

is_leader()

static bool is_leader ( site_def *  site )

inlinestatic

is_member()

static int is_member ( site_def const *  site )

inlinestatic

is_node_v4_reachable()

int is_node_v4_reachable

(

char *

node_address

)

is_node_v4_reachable_with_info()

int is_node_v4_reachable_with_info

(

struct addrinfo *

retrieved_addr_info

)

is_unsafe_event_horizon_reconfiguration()

static bool_t is_unsafe_event_horizon_reconfiguration ( app_data_ptr  a )

static

is_unsafe_leaders_reconfiguration()

static bool_t is_unsafe_leaders_reconfiguration ( app_data_ptr  a )

static

is_unsafe_max_leaders_reconfiguration()

static bool_t is_unsafe_max_leaders_reconfiguration ( app_data_ptr  a )

static

is_unsafe_set_leaders_reconfiguration()

static bool_t is_unsafe_set_leaders_reconfiguration ( app_data_ptr  a )

static

is_view()

static int is_view ( cargo_type  x )

inlinestatic

latest_event_horizon_reconfig()

static site_def const * latest_event_horizon_reconfig ( )

static

Get the latest pending configuration that reconfigures the event horizon.

Retrieve the last pending site_def, i.e. with the greatest start synod that is greater than executed_msg, that reconfigures the event horizon.

leaders_set_by_client()

static bool leaders_set_by_client ( site_def const *  site )

static

learn_ok()

static bool learn_ok ( site_def const *  site, pax_machine const *  p  )

static

local_server()

int local_server

(

task_arg

arg

)

If an error occurs or if the client connection for the local server is forcefully shutdown, we continue processing the queue until the end resorting to time-based waits.

local_server_is_setup()

static bool_t local_server_is_setup ( )

static

local_server_shutdown_ssl()

static int local_server_shutdown_ssl ( connection_descriptor *  con, void *  buf, int  n, int *  ret  )

static

local_synode_allocator()

static synode_no local_synode_allocator ( synode_no  synode )

static

log_cfgchange_wrong_group()

static void log_cfgchange_wrong_group ( app_data_ptr  a, const char *const  message_fmt  )

static

Logs the fact that an add/remove node request is aimed at another group.

Parameters

a	a pointer to the app_data of the configuration command
message_fmt	a formatted message to log, containing a single s that will be replaced by the node's address

log_event_horizon_reconfiguration_failure()

static void log_event_horizon_reconfiguration_failure ( allow_event_horizon_result  error_code, xcom_event_horizon  attempted_event_horizon  )

static

log_get_synode_app_data_failure()

static void log_get_synode_app_data_failure ( xcom_get_synode_app_data_result  error_code )

static

log_ignored_forced_config()

static void log_ignored_forced_config ( app_data_ptr  a, char const *const  caller_name  )

static

majority()

static int majority ( bit_set const *  nodeset, site_def const *  s, int  all, int  delay, int  force  )

inlinestatic

match_leader()

static int match_leader ( char const *  addr, leader_array const  leaders  )

inlinestatic

match_my_msg()

bool_t match_my_msg	(	pax_msg *	learned,
		pax_msg *	mine
	)

max_check()

static node_no max_check ( site_def const *  site )

inlinestatic

min_proposer_synode()

static synode_no min_proposer_synode ( )

static

move_leader_array()

static void move_leader_array ( leader_array *  target, leader_array *  source  )

static

my_unique_id()

static synode_no my_unique_id ( synode_no  synode )

static

new_id()

uint32_t new_id

(

)

Create a new (hopefully unique) ID.

The basic idea is to create a hash from the host ID and a timestamp.

new_leader_array()

leader_array new_leader_array	(	u_int	n,
		char const *	names[]
	)

new_leader_info()

static reply_data * new_leader_info ( site_def *  site )

static

noop_match()

static int noop_match ( pax_machine *  p, pax_msg *  pm  )

inlinestatic

note_snapshot()

static void note_snapshot ( node_no  node )

static

ok_to_propose()

static int ok_to_propose ( pax_machine *  p )

static

own_message()

static int own_message ( pax_msg *  mess, site_def const *  site  )

static

paxos_fsm()

static void paxos_fsm ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_finished()

static int paxos_fsm_finished ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_idle()

int paxos_fsm_idle	(	pax_machine *	paxos,
		site_def const *	site,
		paxos_event	event,
		pax_msg *	mess
	)

paxos_fsm_p1_master_enter()

static int paxos_fsm_p1_master_enter ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p1_master_wait()

static int paxos_fsm_p1_master_wait ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p2_master_enter()

static int paxos_fsm_p2_master_enter ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p2_master_wait()

static int paxos_fsm_p2_master_wait ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p2_slave_enter()

static int paxos_fsm_p2_slave_enter ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p2_slave_wait()

static int paxos_fsm_p2_slave_wait ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p3_master_wait()

static int paxos_fsm_p3_master_wait ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p3_slave_enter()

static int paxos_fsm_p3_slave_enter ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_fsm_p3_slave_wait()

static int paxos_fsm_p3_slave_wait ( pax_machine *  paxos, site_def const *  site, paxos_event  event, pax_msg *  mess  )

static

paxos_timer_advance()

static void paxos_timer_advance ( )

static

paxos_timer_task()

static int paxos_timer_task ( task_arg  arg )

static

paxos_twait()

static void paxos_twait ( pax_machine *  p, unsigned int  t  )

static

paxos_twait_cancel()

static void paxos_twait_cancel ( pax_machine *  p )

static

paxos_wakeup()

static void paxos_wakeup ( unsigned int  t )

static

pm_finished()

int pm_finished

(

pax_machine *

p

)

Return values

1	if the value for the Paxos instance *p has been learned
0	otherwise

pre_process_incoming_ping()

int pre_process_incoming_ping	(	site_def const *	site,
		pax_msg const *	pm,
		int	has_client_already_booted,
		double	current_time
	)

Process incoming are_you_alive (i.e.

: ping) messages and act accordingly

GCS/XCom has a full mesh of connections between all nodes. A connects to B and B connects back to A.

If we cut out B with, for instance, a firewall, we have the A->B connection silently dead, but we have the B->A connection alive. Since we only do monitoring on one half of the connection (the incoming messages), we will consider that B is alive, although we can't contact it. In the same way, B will consider that A is dead, since it does not receive any message from it.

We must be able to break the outgoing connection if we detect that something is wrong, in order to make the bi-directional connection state consistent and report that node as unreachable. That can be done if we start receiving pings from a node that we consider that it is alive. After some pings, we just kill the outgoing connection, thus creating a consistent state.

Breaking this connection should only occur if the node has already booted, meaning that the whole joining process is complete and the node is up and running. This is due to the fact that we receive pings as part of the process of joining a group.

Parameters

site	current site definitions
pm	a possible ping message:
has_client_already_booted	check if this node has already booted
current_time	current XCom time

Returns

int 1 if the node connection is closed. 0, otherwise.

prep_majority()

static int prep_majority ( site_def const *  site, pax_machine const *  p  )

static

prepare()

static void prepare ( pax_msg *  p, pax_op  op  )

static

prepare_msg()

static int prepare_msg ( pax_msg *  p )

static

prepare_push_2p()

void prepare_push_2p	(	site_def const *	site,
		pax_machine *	p
	)

Initializes the proposer's message to go through a 2-phase Paxos on the proposer's reserved ballot (0,_).

Executed by Proposers.

Parameters

site	XCom configuration
p	Paxos instance

prepare_push_3p()

void prepare_push_3p	(	site_def const *	site,
		pax_machine *	p,
		pax_msg *	msg,
		synode_no	msgno,
		pax_msg_type	msg_type
	)

Initializes the message msg to go through a 3-phase, regular Paxos.

Executed by Proposers.

Parameters

site	XCom configuration
p	Paxos instance
msg	Message to send
msgno	Synode where msg will be proposed
msg_type	The type of the message, e.g. normal or no_op

primary_dispatch_table()

static msg_handler * primary_dispatch_table ( )

static

process_accept_op()

static void process_accept_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_ack_accept_op()

static void process_ack_accept_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_ack_prepare_op()

static void process_ack_prepare_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_are_you_alive_op()

static void process_are_you_alive_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_client_msg()

static void process_client_msg ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_die_op()

static void process_die_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_gcs_snapshot_op()

static void process_gcs_snapshot_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_i_am_alive_op()

static void process_i_am_alive_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_learn_op()

static void process_learn_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_need_boot_op()

static void process_need_boot_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_prepare_op()

static void process_prepare_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_read_op()

static void process_read_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_recover_learn_op()

static void process_recover_learn_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_skip_op()

static void process_skip_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_synode_allocated()

static void process_synode_allocated ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_synode_request()

static void process_synode_request ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

process_tiny_learn_op()

static void process_tiny_learn_op ( site_def const *  site, pax_msg *  p, linkage *  reply_queue  )

static

prop_majority()

static int prop_majority ( site_def const *  site, pax_machine const *  p  )

static

propose_missing_values()

static void propose_missing_values ( int  n )

static

propose_msg()

static int propose_msg ( pax_msg *  p )

static

propose_noop()

static void propose_noop ( synode_no  find, pax_machine *  p  )

static

propose_noop_2p()

static void propose_noop_2p ( synode_no  find, pax_machine *  p  )

static

proposer_task()

static int proposer_task ( task_arg  arg )

static

push_msg_2p()

static void push_msg_2p ( site_def const *  site, pax_machine *  p  )

static

push_msg_3p()

static void push_msg_3p ( site_def const *  site, pax_machine *  p, pax_msg *  msg, synode_no  msgno, pax_msg_type  msg_type  )

static

read_missing_values()

static void read_missing_values ( int  n )

static

recently_active()

static int recently_active ( pax_machine *  p )

static

recompute_node_set()

void recompute_node_set	(	node_set const *	old_set,
		node_list const *	old_nodes,
		node_set *	new_set,
		node_list const *	new_nodes
	)

recompute_node_sets()

static void recompute_node_sets ( site_def const *  old_site, site_def *  new_site  )

static

recompute_timestamps()

void recompute_timestamps	(	detector_state const	old_timestamp,
		node_list const *	old_nodes,
		detector_state	new_timestamp,
		node_list const *	new_nodes
	)

reconfigurable_event_horizon()

static bool_t reconfigurable_event_horizon ( xcom_proto  protocol_version )

static

remote_synode_allocator()

static node_no remote_synode_allocator ( site_def *  site, app_data const &  a  )

static

remove_proposer_synode()

static void remove_proposer_synode ( int  i )

static

reply_handler_task()

int reply_handler_task

(

task_arg

arg

)

reply_push_log()

static void reply_push_log ( synode_no  push, linkage *  reply_queue  )

static

reserve_synode_number()

static int reserve_synode_number ( synode_allocation_type *  synode_allocation, site_def **  site, synode_no *  msgno, int *  remote_retry, app_data *  a, synode_reservation_status *  ret  )

static

reset_snapshot_mask()

static void reset_snapshot_mask ( )

static

safe_app_data_copy()

bool_t safe_app_data_copy	(	pax_msg **	target,
		app_data_ptr	source
	)

Copies app data source into target and checks if the copy succeeded.

Sets *target to NULL if the copy fails.

Parameters

[in,out]	target	The pax_msg to which the app_data will be copied.
	source	The app data that will be copied.

Return values

TRUE	if the copy was successful.
FALSE	if the copy failed, in which case *target is set to NULL; a failed copy means that there was an error allocating memory for the copy.

secondary_dispatch_table()

static msg_handler * secondary_dispatch_table ( )

static

send_learn_msg()

static int send_learn_msg ( site_def const *  site, pax_msg *  p  )

static

send_need_boot()

static void send_need_boot ( )

static

send_propose_msg()

static int send_propose_msg ( pax_msg *  p )

static

send_read()

static void send_read ( synode_no  find )

static

send_tiny_learn_msg()

static int send_tiny_learn_msg ( site_def const *  site, pax_msg *  p  )

static

send_value()

static void send_value ( site_def const *  site, node_no  to, synode_no  synode  )

static

send_x_fsm_complete()

void send_x_fsm_complete

(

)

server_handle_need_snapshot()

static void server_handle_need_snapshot ( server *  srv, site_def const *  s, node_no  node  )

static

server_push_log()

static void server_push_log ( server *  srv, synode_no  push, node_no  node  )

static

server_send_snapshot()

static void server_send_snapshot ( server *  srv, site_def const *  s, gcs_snapshot *  gcs_snap, node_no  node  )

static

set_app_snap_getter()

void set_app_snap_getter

(

app_snap_getter

x

)

set_app_snap_handler()

void set_app_snap_handler

(

app_snap_handler

x

)

set_current_message()

synode_no set_current_message

(

synode_no

msgno

)

set_executed_msg()

synode_no set_executed_msg

(

synode_no

msgno

)

set_group()

void set_group

(

uint32_t

id

)

Set node group.

set_last_received_config()

void set_last_received_config

(

synode_no

received_config_change

)

set_learn_type()

static void set_learn_type ( pax_msg *  p )

static

set_log_end()

void set_log_end

(

gcs_snapshot *

gcs

)

set_max_synode()

void set_max_synode

(

synode_no

synode

)

set_proposer_startpoint()

static void set_proposer_startpoint ( )

static

set_start_and_boot()

static void set_start_and_boot ( site_def *  new_config, app_data_ptr  a  )

static

set_unique_id()

static void set_unique_id ( pax_msg *  msg, synode_no  synode  )

static

set_xcom_comms_cb()

void set_xcom_comms_cb

(

xcom_state_change_cb

x

)

set_xcom_exit_cb()

void set_xcom_exit_cb

(

xcom_state_change_cb

x

)

set_xcom_expel_cb()

void set_xcom_expel_cb

(

xcom_state_change_cb

x

)

set_xcom_input_try_pop_cb()

void set_xcom_input_try_pop_cb

(

xcom_input_try_pop_cb

pop

)

set_xcom_run_cb()

void set_xcom_run_cb

(

xcom_state_change_cb

x

)

setup_exit_handling()

static void setup_exit_handling ( execute_context *  xc, site_def *  site  )

static

should_handle_need_boot()

bool_t should_handle_need_boot	(	site_def const *	site,
		pax_msg *	p
	)

Returns

true if we should process the incoming need_boot_op message passed in parameter p.

should_ignore_forced_config_or_view()

static constexpr bool should_ignore_forced_config_or_view ( xcom_proto  protocol_version )

staticconstexpr

should_poll_cache()

static bool_t should_poll_cache ( pax_op  op )

static

site_install_action()

void site_install_action	(	site_def *	site,
		cargo_type	operation
	)

skip_msg()

static int skip_msg ( pax_msg *  p )

static

skip_value()

static void skip_value ( pax_msg *  p )

static

socket_read()

static result socket_read ( connection_descriptor *  rfd, void *  buf, int  n  )

static

socket_read_bytes()

static int64_t socket_read_bytes ( connection_descriptor *  rfd, char *  p, uint32_t  n  )

static

socket_read_msg()

static pax_msg * socket_read_msg ( connection_descriptor *  rfd, pax_msg *  p  )

static

socket_write()

static int64_t socket_write ( connection_descriptor *  wfd, void *  _buf, uint32_t  n, connnection_write_method  write_function = con_write  )

static

start_force_config()

static void start_force_config ( site_def *  s, int  enforcer  )

static

start_x_timer()

static void start_x_timer ( double  t )

static

started()

static int started ( pax_machine *  p )

inlinestatic

stop_x_timer()

static void stop_x_timer ( )

static

sweeper_task()

static int sweeper_task ( task_arg  arg )

static

synthesize_leaders()

void synthesize_leaders

(

leader_array *

leaders

)

teach_ignorant_node()

static void teach_ignorant_node ( site_def const *  site, pax_machine *  p, pax_msg *  pm, synode_no  synode, linkage *  reply_queue  )

static

terminate_and_exit()

void terminate_and_exit

(

)

terminate_proposers()

static void terminate_proposers ( )

static

too_far()

static int too_far ( synode_no  s )

inlinestatic

too_far_threshold()

static uint64_t too_far_threshold ( xcom_event_horizon  active_event_horizon )

static

Checks if the given synod s is outside the event horizon.

Common case: there are no configurations pending, or if there are, none of them reconfigure the event horizon. The common case threshold is:

last_executed_synod + event_horizon(active_config)

If an event horizon reconfiguration R is pending, it is possible that it reduces the event horizon. In that case, it is possible that the threshold above falls outside the new event horizon.

For example, consider last_executed_synod = 42 and event_horizon(active_config) = 10. At this point this member participates in synods up to 52. Now consider an event horizon reconfiguration that takes effect at synod 45, which modifies the event horizon to 2. This means that when last_executed_synod = 45, event_horizon(active_config) = 2. At this point this member should only be able to participate in synods up to 47. The member may have previously started processing messages directed to synods between 47 and 52, but will now ignore messages directed to those same synods.

We do not want to start processing messages that will eventually fall out of the event horizon. More importantly, the threshold above may not be safe due to the exit logic of executor_task.

When a node removes itself from the group on configuration C starting at synod start(C), the exit logic relies on knowing when a majority has executed synod start(C) - 1, i.e. the last message of the last configuration to contain the leaving node.

With a constant event horizon, we know that when synod start(C) + event_horizon is learnt, it is because a majority already executed or is ready to execute (and thus learned) synod start(C). This implies that a majority already executed start(C) - 1.

With a dynamic event horizon, we cannot be sure that when synod start(C) + event_horizon(C) is learnt, a majority already executed or is ready to execute synod start(C). This is because it is possible for a new, smaller, event horizon to take effect between start(C) and start(C) + event_horizon(C). If that happens, the threshold above allows nodes to participate in synods which are possibly beyond start(C) + event_horizon(C), which can lead to the value of synod start(C) + event_horizon(C) being learnt without a majority already having executed or being ready to execute synod start(C).

In order to maintain the assumption made by the executor_task's exit logic, when an event horizon reconfiguration R is pending we set the threshold to the minimum between:

last_executed_synod + event_horizon(active_config)

and:

start(R) - 1 + event_horizon(R)

too_far_threshold_new_event_horizon_pending()

static uint64_t too_far_threshold_new_event_horizon_pending ( site_def const *  new_config )

static

unsafe_against_event_horizon()

static bool unsafe_against_event_horizon ( node_address const *  node )

static

Check if a node is compatible with the group's event horizon.

A node is compatible with the group's configuration if:

a) The node supports event horizon reconfigurations, or b) The group's event horizon is, or is scheduled to be, the default event horizon.

unsafe_leaders()

bool unsafe_leaders

(

app_data *

a

)

update_best_snapshot()

static void update_best_snapshot ( gcs_snapshot *  gcs )

static

update_max_synode()

static void update_max_synode ( pax_msg *  p )

static

update_site()

static site_def * update_site ( site_def *  site, node_set const *  ns, synode_no  boot_key, synode_no  start  )

static

update_srv()

static void update_srv ( server **  target, server *  srv  )

inlinestatic

wait_for_cache()

static int wait_for_cache ( pax_machine **  pm, synode_no  synode, double  timeout  )

static

wakeup_delay()

static double wakeup_delay ( double  old )

static

warn_protoversion_mismatch()

void warn_protoversion_mismatch

(

connection_descriptor *

rfd

)

x_check_execute_inform()

static int x_check_execute_inform ( execute_context *  xc )

static

x_check_exit()

static int x_check_exit ( execute_context *  xc )

static

x_check_increment_execute()

static void x_check_increment_execute ( execute_context *  xc )

static

x_check_increment_fetch()

static void x_check_increment_fetch ( execute_context *  xc )

static

x_execute()

static void x_execute ( execute_context *  xc )

static

x_fetch()

static void x_fetch ( execute_context *  xc )

static

x_fsm_completion_task()

static int x_fsm_completion_task ( task_arg  arg )

static

x_terminate()

static void x_terminate ( execute_context *  xc )

static

xcom_check_reply()

static int xcom_check_reply ( int const  res )

static

xcom_client_add_node()

int xcom_client_add_node	(	connection_descriptor *	fd,
		node_list *	nl,
		uint32_t	group_id
	)

xcom_client_convert_into_local_server()

int xcom_client_convert_into_local_server

(

connection_descriptor *const

fd

)

xcom_client_disable_arbitrator()

int xcom_client_disable_arbitrator

(

connection_descriptor *

fd

)

xcom_client_enable_arbitrator()

int xcom_client_enable_arbitrator

(

connection_descriptor *

fd

)

xcom_client_get_leaders()

int xcom_client_get_leaders	(	connection_descriptor *	fd,
		uint32_t	group_id,
		leader_info_data *	leaders
	)

xcom_client_get_synode_app_data()

int xcom_client_get_synode_app_data	(	connection_descriptor *const	fd,
		uint32_t	group_id,
		synode_no_array *const	synodes,
		synode_app_data_array *const	reply
	)

xcom_client_remove_node()

int xcom_client_remove_node	(	connection_descriptor *	fd,
		node_list *	nl,
		uint32_t	group_id
	)

xcom_client_send_die()

int64_t xcom_client_send_die

(

connection_descriptor *

fd

)

xcom_client_set_cache_limit()

int xcom_client_set_cache_limit	(	connection_descriptor *	fd,
		uint64_t	cache_limit
	)

xcom_client_set_leaders() [1/2]

int xcom_client_set_leaders	(	connection_descriptor *	fd,
		u_int	n,
		char const *	names[],
		node_no	max_leaders,
		uint32_t	group_id
	)

xcom_client_set_leaders() [2/2]

int xcom_client_set_leaders	(	connection_descriptor *	fd,
		u_int	n,
		char const *	names[],
		uint32_t	group_id
	)

xcom_client_set_max_leaders()

int xcom_client_set_max_leaders	(	connection_descriptor *	fd,
		node_no	max_leaders,
		uint32_t	group_id
	)

xcom_fsm()

char const * xcom_fsm	(	xcom_actions	action,
		task_arg	fsmargs
	)

xcom_fsm_impl()

xcom_fsm_state * xcom_fsm_impl	(	xcom_actions	action,
		task_arg	fsmargs
	)

xcom_fsm_init()

static int xcom_fsm_init ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_recover_wait()

static int xcom_fsm_recover_wait ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_recover_wait_enter()

static int xcom_fsm_recover_wait_enter ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_run()

static int xcom_fsm_run ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_run_enter()

static int xcom_fsm_run_enter ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_snapshot_wait()

static int xcom_fsm_snapshot_wait ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_snapshot_wait_enter()

static int xcom_fsm_snapshot_wait_enter ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_start()

static int xcom_fsm_start ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_fsm_start_enter()

static int xcom_fsm_start_enter ( xcom_actions  action, task_arg  fsmargs, xcom_fsm_state *  ctxt  )

static

xcom_get_event_horizon()

static client_reply_code xcom_get_event_horizon ( xcom_event_horizon *  event_horizon )

static

Retrieves the latest event horizon.

There is no specific reason for this method to return the latest event horizon instead of the current one. Both would be acceptable results of this function, but we had to make a decision of one over the other.

Parameters

[out]

event_horizon

the latest event horizon

Return values

REQUEST_FAIL	XCom is not initialized yet
REQUEST_OK	function was successful and event_horizon contains the latest event horizon

xcom_get_maximum_event_horizon()

xcom_event_horizon xcom_get_maximum_event_horizon

(

)

xcom_get_minimum_event_horizon()

xcom_event_horizon xcom_get_minimum_event_horizon

(

)

xcom_input_free_signal_connection()

void xcom_input_free_signal_connection

(

void

)

xcom_input_new_signal_connection()

bool_t xcom_input_new_signal_connection	(	char const *	address,
		xcom_port	port
	)

xcom_input_signal()

bool_t xcom_input_signal

(

void

)

xcom_input_signal_connection_shutdown_ssl()

static bool_t xcom_input_signal_connection_shutdown_ssl ( )

static

xcom_input_signal_connection_shutdown_ssl_wait_for_peer()

static bool_t xcom_input_signal_connection_shutdown_ssl_wait_for_peer ( )

static

xcom_open_client_connection()

connection_descriptor * xcom_open_client_connection	(	char const *	server,
		xcom_port	port
	)

xcom_recv_proto()

static int xcom_recv_proto ( connection_descriptor *  rfd, xcom_proto *  x_proto, x_msg_type *  x_type, unsigned int *  tag  )

static

xcom_send()

void xcom_send	(	app_data_ptr	a,
		pax_msg *	msg
	)

xcom_send_app_wait()

static int xcom_send_app_wait ( connection_descriptor *  fd, app_data *  a, int  force, leader_info_data *  leaders  )

static

xcom_send_app_wait_and_get()

static xcom_send_app_wait_result xcom_send_app_wait_and_get ( connection_descriptor *  fd, app_data *  a, int  force, pax_msg *  p, leader_info_data *  leaders  )

static

Send a message and wait for response.

The caller is responsible for freeing p after calling this function, i.e. xdr_free((xdrproc_t)xdr_pax_msg, (char *)p)

xcom_send_cfg_wait()

int xcom_send_cfg_wait	(	connection_descriptor *	fd,
		node_list *	nl,
		uint32_t	group_id,
		cargo_type	ct,
		int	force
	)

xcom_send_client_app_data()

int64_t xcom_send_client_app_data	(	connection_descriptor *	fd,
		app_data_ptr	a,
		int	force
	)

xcom_send_proto()

static int xcom_send_proto ( connection_descriptor *  con, xcom_proto  x_proto, x_msg_type  x_type, unsigned int  tag  )

static

xcom_sleep()

void xcom_sleep

(

unsigned int

seconds

)

xcom_taskmain2()

int xcom_taskmain2

(

xcom_port

listen_port

)

xcom_thread_deinit()

void xcom_thread_deinit

(

)

xcom_thread_init()

void xcom_thread_init

(

)

xcom_timer()

static int xcom_timer ( task_arg  arg )

static

xcom_unique_long()

long xcom_unique_long

(

void

)

zero_leader_array()

static void zero_leader_array ( leader_array *  l )

static

Variable Documentation

alive_t

task_env* alive_t = nullptr

static

ARBITRATOR_HACK

int ARBITRATOR_HACK = 0

boot

task_env* boot = nullptr

cache_task

task_env* cache_task = nullptr

static

clicnt

int clicnt = 0

static

client_boot_done

int client_boot_done

extern

current_message

synode_no current_message

static

current_tick

unsigned int current_tick = 0

static

struct { ... } dead_sites

debug_xc

execute_context* debug_xc

static

struct { ... } delay_fifo

delivered_msg

synode_no delivered_msg = NULL_SYNODE

static

detector

task_env* detector = nullptr

detector_wait

linkage detector_wait

Initial value:

= {0, &detector_wait,
                         &detector_wait}

dispatch_table

msg_handler dispatch_table[LAST_OP]

static

Initial value:

= {process_client_msg,
                                              nullptr,
                                              process_prepare_op,
                                              process_ack_prepare_op,
                                              process_ack_prepare_op,
                                              process_accept_op,
                                              process_ack_accept_op,
                                              process_learn_op,
                                              process_recover_learn_op,
                                              nullptr,
                                              nullptr,
                                              nullptr,
                                              nullptr,
                                              nullptr,
                                              process_skip_op,
                                              process_i_am_alive_op,
                                              process_are_you_alive_op,
                                              process_need_boot_op,
                                              nullptr,
                                              process_die_op,
                                              process_read_op,
                                              process_gcs_snapshot_op,
                                              nullptr,
                                              process_tiny_learn_op,
                                              process_synode_request,
                                              process_synode_allocated}

exec_wait

linkage exec_wait

static

Initial value:

= {
    0, &exec_wait, &exec_wait}

executed_msg

synode_no executed_msg

executor

task_env* executor = nullptr

static

executor_site

site_def* executor_site = nullptr

static

first_event_horizon_aware_protocol

xcom_proto const first_event_horizon_aware_protocol = x_1_4

static

first_protocol_that_ignores_intermediate_forced_configs_or_views

constexpr xcom_proto first_protocol_that_ignores_intermediate_forced_configs_or_views

staticconstexpr

Initial value:

=
    x_1_8

forced_config

site_def* forced_config = nullptr

static

front

int front

get_app_snap_cb

app_snap_getter get_app_snap_cb

static

handle_app_snap_cb

app_snap_handler handle_app_snap_cb

static

id

unsigned long id[MAX_DEAD]

input_signal_connection

connection_descriptor* input_signal_connection {nullptr}

static

These fields are used to signal XCom's request queue.

After a request is added, one will write 1 byte to warn local_server_task that it has work to do.

We use two types of signalling connection:

• An anonymous pipe, when possible, in POSIX compatible systems
• A regular socket connection, in Windows

input_signal_connection is the connection_descriptor returned when one opens a local signalling connection. It will contain either:

• The write side of a connection, in case of using a pipe OR;
• A bidirectional connection, when using a regular socket connection;

input_signal_connection_pipe is the connection_descriptor that holds the read side of a pipe connection. It is only allocated when we are able to have a pipe connection.

input_signal_connection_pipe

connection_descriptor* input_signal_connection_pipe {nullptr}

killer

task_env* killer = nullptr

last_config_modification_id

synode_no last_config_modification_id

static

last_delivered_msg

synode_no last_delivered_msg = NULL_SYNODE

static

log_end_max

synode_no log_end_max

static

log_start_max

synode_no log_start_max

static

lsn

uint64_t lsn = 0

static

max_synode

synode_no max_synode

my_id

uint32_t my_id = 0

static

n

int n

net_boot

task_env* net_boot = nullptr

net_recover

task_env* net_recover = nullptr

netboot_ok

int netboot_ok

extern

no_duplicate_payload

int const no_duplicate_payload = 1

static

oblist

struct fp_name oblist[]

static

Initial value:

= {
    NAME(x_fetch), NAME(x_execute), NAME(x_terminate), {nullptr, nullptr}}

oom_abort

int oom_abort = 0

p1_idle_vtbl

paxos_state_action p1_idle_vtbl[last_p_event]

Initial value:

= {action_paxos_prepare,
                                                 nullptr,
                                                 action_paxos_accept,
                                                 nullptr,
                                                 action_paxos_learn,
                                                 action_paxos_start,
                                                 nullptr}

p1_master_enter_vtbl

paxos_state_action p1_master_enter_vtbl[last_p_event]

Initial value:

= {action_new_prepare,
                                                         action_ack_prepare,
                                                         action_new_accept,
                                                         nullptr,
                                                         action_paxos_learn,
                                                         nullptr,
                                                         nullptr}

p1_master_wait_vtbl

paxos_state_action p1_master_wait_vtbl[last_p_event]

Initial value:

= {action_new_prepare,
                                                        action_ack_prepare,
                                                        action_new_accept,
                                                        nullptr,
                                                        action_paxos_learn,
                                                        nullptr,
                                                        nullptr}

p2_master_enter_vtbl

paxos_state_action p2_master_enter_vtbl[last_p_event]

Initial value:

= {action_new_accept,
                                                         nullptr,
                                                         action_new_accept,
                                                         action_ack_accept,
                                                         action_paxos_learn,
                                                         nullptr,
                                                         nullptr}

p2_master_wait_vtbl

paxos_state_action p2_master_wait_vtbl[last_p_event]

Initial value:

= {action_new_prepare,
                                                        nullptr,
                                                        action_new_accept,
                                                        action_ack_accept,
                                                        action_paxos_learn,
                                                        nullptr,
                                                        nullptr}

p2_slave_wait_vtbl

paxos_state_action p2_slave_wait_vtbl[last_p_event]

Initial value:

= {action_new_prepare,
                                                       nullptr,
                                                       action_new_accept,
                                                       nullptr,
                                                       action_paxos_learn,
                                                       nullptr,
                                                       nullptr}

p3_master_wait_vtbl

paxos_state_action p3_master_wait_vtbl[last_p_event]

Initial value:

= {action_new_prepare,
                                                        nullptr,
                                                        action_new_accept,
                                                        nullptr,
                                                        action_paxos_learn,
                                                        nullptr,
                                                        nullptr}

p3_slave_wait_vtbl

paxos_state_action p3_slave_wait_vtbl[last_p_event]

Initial value:

= {action_new_prepare,
                                                       nullptr,
                                                       action_new_accept,
                                                       nullptr,
                                                       action_paxos_learn,
                                                       nullptr,
                                                       nullptr}

p_finished_vtbl

paxos_state_action p_finished_vtbl[last_p_event]

Initial value:

= {
    action_ignorant, nullptr, action_ignorant, nullptr,
    nullptr,         nullptr, nullptr}

paxos_default_timeout

unsigned constexpr int const paxos_default_timeout = 100

constexpr

paxos_event_name

const char* paxos_event_name[] = {p_events}

pipe_signal_connections

int pipe_signal_connections[2] = {-1, -1}

prop_finished

int prop_finished = 0

static

prop_input_queue

channel prop_input_queue

static

prop_started

int prop_started = 0

static

proposer

task_env* proposer[PROPOSERS]

static

proposer_site

site_def* proposer_site = nullptr

static

proposer_synodes

synode_no* proposer_synodes[PROPOSERS]

static

protoversion_warning_time

double protoversion_warning_time

static

Initial value:

=
    0.0

q

synode_no q[FIFO_SIZE]

rear

int rear

recovery_begin_cb

xcom_recovery_cb recovery_begin_cb = nullptr

static

recovery_end_cb

xcom_recovery_cb recovery_end_cb = nullptr

static

recovery_init_cb

xcom_recovery_cb recovery_init_cb = nullptr

static

recovery_restart_cb

xcom_recovery_cb recovery_restart_cb = nullptr

static

retry

task_env* retry = nullptr

static

sent_alive

double sent_alive = 0.0

static

single_writer_support

constexpr xcom_proto single_writer_support = x_1_9

staticconstexpr

snapshots

int snapshots[NSERVERS]

static

start_config

synode_no start_config = NULL_SYNODE

static

sweeper

task_env* sweeper = nullptr

static

synode_number_pool

synode_pool synode_number_pool

threephase

int const threephase = 0

Timestamp of previous protoversion warning.

time_queue

linkage time_queue[paxos_timer_range]

static

use_buffered_read

int use_buffered_read = 1

static

wait_forced_config

int wait_forced_config = 0

static

x_timer

task_env* x_timer = nullptr

static

xcom_actions_name

const char* xcom_actions_name[] = {x_actions}

xcom_comms_cb

xcom_state_change_cb xcom_comms_cb = nullptr

static

xcom_dbg_stack

long xcom_dbg_stack[DBG_STACK_SIZE]

xcom_dbg_stack_top

int xcom_dbg_stack_top = 0

xcom_debug_mask

long xcom_debug_mask

Initial value:

=
     D_FSM  | D_TRANSPORT

xcom_exit_cb

xcom_state_change_cb xcom_exit_cb = nullptr

static

xcom_expel_cb

xcom_state_change_cb xcom_expel_cb = nullptr

static

xcom_run_cb

xcom_state_change_cb xcom_run_cb = nullptr

static

xcom_shutdown

int xcom_shutdown = 0

xcom_terminate_cb

xcom_state_change_cb xcom_terminate_cb = nullptr

static

xcom_thread_input

void* xcom_thread_input = nullptr

xcom_try_pop_from_input_cb

xcom_input_try_pop_cb xcom_try_pop_from_input_cb = nullptr

static