No OneTemporary
Actions

Size

103 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/crmd/control.c b/crmd/control.c
	index f79db1ebb5..f6588f9bd0 100644
	--- a/crmd/control.c
	+++ b/crmd/control.c
	@@ -1,907 +1,913 @@
	/*
	* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
	*
	* This program is free software; you can redistribute it and/or
	* modify it under the terms of the GNU General Public
	* License as published by the Free Software Foundation; either
	* version 2 of the License, or (at your option) any later version.
	*
	* This software is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	* General Public License for more details.
	*
	* You should have received a copy of the GNU General Public
	* License along with this library; if not, write to the Free Software
	* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	*/

	#include <crm_internal.h>

	#include <sys/param.h>

	#include <crm/crm.h>

	#include <crm/msg_xml.h>

	#include <crm/pengine/rules.h>
	#include <crm/cluster/internal.h>
	#include <crm/common/ipcs.h>

	#include <crmd.h>
	#include <crmd_fsa.h>
	#include <fsa_proto.h>
	#include <crmd_messages.h>
	#include <crmd_callbacks.h>
	#include <crmd_lrm.h>
	#include <tengine.h>

	#include <sys/types.h>
	#include <sys/stat.h>

	qb_ipcs_service_t *ipcs = NULL;

	extern gboolean crm_connect_corosync(crm_cluster_t * cluster);
	extern void crmd_ha_connection_destroy(gpointer user_data);

	void crm_shutdown(int nsig);
	gboolean crm_read_options(gpointer user_data);

	gboolean fsa_has_quorum = FALSE;
	crm_trigger_t *fsa_source = NULL;
	crm_trigger_t *config_read = NULL;

	/* A_HA_CONNECT */
	void
	do_ha_control(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state,
	enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	gboolean registered = FALSE;
	static crm_cluster_t *cluster = NULL;

	if (cluster == NULL) {
	cluster = calloc(1, sizeof(crm_cluster_t));
	}

	if (action & A_HA_DISCONNECT) {
	crm_cluster_disconnect(cluster);
	crm_info("Disconnected from the cluster");

	set_bit(fsa_input_register, R_HA_DISCONNECTED);
	}

	if (action & A_HA_CONNECT) {
	crm_set_status_callback(&peer_update_callback);

	if (is_openais_cluster()) {
	#if SUPPORT_COROSYNC
	registered = crm_connect_corosync(cluster);
	#endif
	} else if (is_heartbeat_cluster()) {
	#if SUPPORT_HEARTBEAT
	cluster->destroy = crmd_ha_connection_destroy;
	cluster->hb_dispatch = crmd_ha_msg_callback;

	registered = crm_cluster_connect(cluster);
	fsa_cluster_conn = cluster->hb_conn;

	crm_trace("Be informed of Node Status changes");
	if (registered &&
	fsa_cluster_conn->llc_ops->set_nstatus_callback(fsa_cluster_conn,
	crmd_ha_status_callback,
	fsa_cluster_conn) != HA_OK) {

	crm_err("Cannot set nstatus callback: %s",
	fsa_cluster_conn->llc_ops->errmsg(fsa_cluster_conn));
	registered = FALSE;
	}

	crm_trace("Be informed of CRM Client Status changes");
	if (registered &&
	fsa_cluster_conn->llc_ops->set_cstatus_callback(fsa_cluster_conn,
	crmd_client_status_callback,
	fsa_cluster_conn) != HA_OK) {

	crm_err("Cannot set cstatus callback: %s",
	fsa_cluster_conn->llc_ops->errmsg(fsa_cluster_conn));
	registered = FALSE;
	}

	if (registered) {
	crm_trace("Requesting an initial dump of CRMD client_status");
	fsa_cluster_conn->llc_ops->client_status(fsa_cluster_conn, NULL, CRM_SYSTEM_CRMD,
	-1);
	}
	#endif
	}
	fsa_our_uname = cluster->uname;
	fsa_our_uuid = cluster->uuid;

	if (registered == FALSE) {
	set_bit(fsa_input_register, R_HA_DISCONNECTED);
	register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
	return;
	}

	populate_cib_nodes(node_update_none, __FUNCTION__);
	clear_bit(fsa_input_register, R_HA_DISCONNECTED);
	crm_info("Connected to the cluster");
	}

	if (action & ~(A_HA_CONNECT \| A_HA_DISCONNECT)) {
	crm_err("Unexpected action %s in %s", fsa_action2string(action), __FUNCTION__);
	}
	}

	/* A_SHUTDOWN */
	void
	do_shutdown(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	/* just in case */
	set_bit(fsa_input_register, R_SHUTDOWN);

	if (is_heartbeat_cluster()) {
	if (is_set(fsa_input_register, pe_subsystem->flag_connected)) {
	crm_info("Terminating the %s", pe_subsystem->name);
	if (stop_subsystem(pe_subsystem, TRUE) == FALSE) {
	/* its gone... */
	crm_err("Faking %s exit", pe_subsystem->name);
	clear_bit(fsa_input_register, pe_subsystem->flag_connected);
	} else {
	crm_info("Waiting for subsystems to exit");
	crmd_fsa_stall(FALSE);
	}
	}
	crm_info("All subsystems stopped, continuing");
	}

	if (stonith_api) {
	/* Prevent it from comming up again */
	clear_bit(fsa_input_register, R_ST_REQUIRED);

	crm_info("Disconnecting STONITH...");
	stonith_api->cmds->disconnect(stonith_api);
	}
	}

	/* A_SHUTDOWN_REQ */
	void
	do_shutdown_req(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state,
	enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	xmlNode *msg = NULL;

	crm_info("Sending shutdown request to %s", crm_str(fsa_our_dc));
	msg = create_request(CRM_OP_SHUTDOWN_REQ, NULL, NULL, CRM_SYSTEM_DC, CRM_SYSTEM_CRMD, NULL);

	/* set_bit(fsa_input_register, R_STAYDOWN); */
	if (send_cluster_message(NULL, crm_msg_crmd, msg, TRUE) == FALSE) {
	register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
	}
	free_xml(msg);
	}

	extern crm_ipc_t *attrd_ipc;
	extern char *max_generation_from;
	extern xmlNode *max_generation_xml;
	extern GHashTable *resource_history;
	extern GHashTable *voted;
	extern GHashTable *reload_hash;

	void log_connected_client(gpointer key, gpointer value, gpointer user_data);

	void
	log_connected_client(gpointer key, gpointer value, gpointer user_data)
	{
	crm_client_t *client = value;

	crm_err("%s is still connected at exit", crm_client_name(client));
	}

	int
	crmd_exit(int rc)
	{
	GListPtr gIter = NULL;

	crm_trace("Preparing to exit");

	if(ipcs) {
	crm_trace("Closing IPC server");
	mainloop_del_ipc_server(ipcs);
	}

	if (attrd_ipc) {
	crm_trace("Closing attrd connection");
	crm_ipc_close(attrd_ipc);
	crm_ipc_destroy(attrd_ipc);
	attrd_ipc = NULL;
	}
	#if SUPPORT_HEARTBEAT
	if (fsa_cluster_conn) {
	crm_trace("Disconnecting heartbeat");
	fsa_cluster_conn->llc_ops->delete(fsa_cluster_conn);
	fsa_cluster_conn = NULL;
	}
	#endif

	for (gIter = fsa_message_queue; gIter != NULL; gIter = gIter->next) {
	fsa_data_t *fsa_data = gIter->data;

	crm_info("Dropping %s: [ state=%s cause=%s origin=%s ]",
	fsa_input2string(fsa_data->fsa_input),
	fsa_state2string(fsa_state),
	fsa_cause2string(fsa_data->fsa_cause), fsa_data->origin);
	delete_fsa_input(fsa_data);
	}
	g_list_free(fsa_message_queue);
	fsa_message_queue = NULL;

	clear_bit(fsa_input_register, R_MEMBERSHIP);

	if (pe_subsystem->client && pe_subsystem->client->ipcs) {
	crm_trace("Disconnecting Policy Engine");
	qb_ipcs_disconnect(pe_subsystem->client->ipcs);
	}
	free(pe_subsystem);

	free(cib_subsystem);
	free(te_subsystem);

	if (reload_hash) {
	crm_trace("Destroying reload cache with %d members", g_hash_table_size(reload_hash));
	g_hash_table_destroy(reload_hash);
	}
	if (voted) {
	crm_trace("Destroying voted cache with %d members", g_hash_table_size(voted));
	g_hash_table_destroy(voted);
	}

	cib_delete(fsa_cib_conn);
	fsa_cib_conn = NULL;

	verify_stopped(fsa_state, LOG_WARNING);
	clear_bit(fsa_input_register, R_LRM_CONNECTED);
	lrm_state_destroy_all();

	+ /* This basically will not work, since mainloop has a reference to it */
	mainloop_destroy_trigger(fsa_source);
	+
	mainloop_destroy_trigger(config_read);
	mainloop_destroy_trigger(stonith_reconnect);
	+ mainloop_destroy_trigger(transition_trigger);

	if(stonith_api) {
	crm_trace("Disconnecting fencing API");
	stonith_api->cmds->free(stonith_api);
	}

	crm_client_cleanup();
	empty_uuid_cache();
	crm_peer_destroy();

	free(transition_timer);
	free(integration_timer);
	free(finalization_timer);
	free(election_trigger);
	free(election_timeout);
	free(shutdown_escalation_timer);
	free(wait_timer);
	free(recheck_timer);

	free(fsa_our_dc_version);
	free(fsa_our_uname);
	free(fsa_our_uuid);
	free(fsa_our_dc);
	free(te_uuid);

	free(fsa_pe_ref);

	free(max_generation_from);
	free_xml(max_generation_xml);

	mainloop_destroy_signal(SIGUSR1);
	mainloop_destroy_signal(SIGTERM);
	mainloop_destroy_signal(SIGTRAP);
	mainloop_destroy_signal(SIGCHLD);

	if (crmd_mainloop) {
	int lpc = 0;
	GMainContext *ctx = g_main_loop_get_context(crmd_mainloop);
	crm_trace("Draining mainloop %d %d", g_main_loop_is_running(crmd_mainloop), g_main_context_pending(ctx));
	while(g_main_context_pending(ctx) && lpc < 10) {
	lpc++;
	crm_trace("Iteration %d", lpc);
	g_main_context_dispatch(ctx);
	}

	crm_trace("Closing mainloop %d %d", g_main_loop_is_running(crmd_mainloop), g_main_context_pending(ctx));
	g_main_loop_quit(crmd_mainloop);
	+
	+ /* Won't strictly do anything since we're inside it now */
	g_main_loop_unref(crmd_mainloop);
	crmd_mainloop = NULL;
	}

	crm_trace("Done");
	return crm_exit(rc);
	}

	/* A_EXIT_0, A_EXIT_1 */
	void
	do_exit(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	int exit_code = 0;
	int log_level = LOG_INFO;
	const char *exit_type = "gracefully";

	if (action & A_EXIT_1) {
	exit_code = 1;
	log_level = LOG_ERR;
	exit_type = "forcefully";
	}

	verify_stopped(cur_state, LOG_ERR);
	do_crm_log(log_level, "Performing %s - %s exiting the CRMd",
	fsa_action2string(action), exit_type);

	if (is_set(fsa_input_register, R_IN_RECOVERY)) {
	crm_err("Could not recover from internal error");
	exit_code = 2;
	}
	if (is_set(fsa_input_register, R_STAYDOWN)) {
	crm_warn("Inhibiting respawn by Heartbeat");
	exit_code = 100;
	}

	crm_info("[%s] stopped (%d)", crm_system_name, exit_code);
	delete_fsa_input(msg_data);
	crmd_exit(exit_code);
	}

	/* A_STARTUP */
	void
	do_startup(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	int was_error = 0;
	int interval = 1; /* seconds between DC heartbeats */

	crm_debug("Registering Signal Handlers");
	mainloop_add_signal(SIGTERM, crm_shutdown);

	fsa_source = mainloop_add_trigger(G_PRIORITY_HIGH, crm_fsa_trigger, NULL);
	config_read = mainloop_add_trigger(G_PRIORITY_HIGH, crm_read_options, NULL);
	+ transition_trigger = mainloop_add_trigger(G_PRIORITY_LOW, te_graph_trigger, NULL);

	crm_debug("Creating CIB and LRM objects");
	fsa_cib_conn = cib_new();

	lrm_state_init_local();

	/* set up the timers */
	transition_timer = calloc(1, sizeof(fsa_timer_t));
	integration_timer = calloc(1, sizeof(fsa_timer_t));
	finalization_timer = calloc(1, sizeof(fsa_timer_t));
	election_trigger = calloc(1, sizeof(fsa_timer_t));
	election_timeout = calloc(1, sizeof(fsa_timer_t));
	shutdown_escalation_timer = calloc(1, sizeof(fsa_timer_t));
	wait_timer = calloc(1, sizeof(fsa_timer_t));
	recheck_timer = calloc(1, sizeof(fsa_timer_t));

	interval = interval * 1000;

	if (election_trigger != NULL) {
	election_trigger->source_id = 0;
	election_trigger->period_ms = -1;
	election_trigger->fsa_input = I_DC_TIMEOUT;
	election_trigger->callback = crm_timer_popped;
	election_trigger->repeat = FALSE;
	} else {
	was_error = TRUE;
	}

	if (election_timeout != NULL) {
	election_timeout->source_id = 0;
	election_timeout->period_ms = -1;
	election_timeout->fsa_input = I_ELECTION_DC;
	election_timeout->callback = crm_timer_popped;
	election_timeout->repeat = FALSE;
	} else {
	was_error = TRUE;
	}

	if (transition_timer != NULL) {
	transition_timer->source_id = 0;
	transition_timer->period_ms = -1;
	transition_timer->fsa_input = I_PE_CALC;
	transition_timer->callback = crm_timer_popped;
	transition_timer->repeat = FALSE;
	} else {
	was_error = TRUE;
	}

	if (integration_timer != NULL) {
	integration_timer->source_id = 0;
	integration_timer->period_ms = -1;
	integration_timer->fsa_input = I_INTEGRATED;
	integration_timer->callback = crm_timer_popped;
	integration_timer->repeat = FALSE;
	} else {
	was_error = TRUE;
	}

	if (finalization_timer != NULL) {
	finalization_timer->source_id = 0;
	finalization_timer->period_ms = -1;
	finalization_timer->fsa_input = I_FINALIZED;
	finalization_timer->callback = crm_timer_popped;
	finalization_timer->repeat = FALSE;
	/* for possible enabling... a bug in the join protocol left
	* a slave in S_PENDING while we think its in S_NOT_DC
	*
	* raising I_FINALIZED put us into a transition loop which is
	* never resolved.
	* in this loop we continually send probes which the node
	* NACK's because its in S_PENDING
	*
	* if we have nodes where heartbeat is active but the
	* CRM is not... then this will be handled in the
	* integration phase
	*/
	finalization_timer->fsa_input = I_ELECTION;

	} else {
	was_error = TRUE;
	}

	if (shutdown_escalation_timer != NULL) {
	shutdown_escalation_timer->source_id = 0;
	shutdown_escalation_timer->period_ms = -1;
	shutdown_escalation_timer->fsa_input = I_STOP;
	shutdown_escalation_timer->callback = crm_timer_popped;
	shutdown_escalation_timer->repeat = FALSE;
	} else {
	was_error = TRUE;
	}

	if (wait_timer != NULL) {
	wait_timer->source_id = 0;
	wait_timer->period_ms = 2000;
	wait_timer->fsa_input = I_NULL;
	wait_timer->callback = crm_timer_popped;
	wait_timer->repeat = FALSE;
	} else {
	was_error = TRUE;
	}

	if (recheck_timer != NULL) {
	recheck_timer->source_id = 0;
	recheck_timer->period_ms = -1;
	recheck_timer->fsa_input = I_PE_CALC;
	recheck_timer->callback = crm_timer_popped;
	recheck_timer->repeat = FALSE;
	} else {
	was_error = TRUE;
	}

	/* set up the sub systems */
	cib_subsystem = calloc(1, sizeof(struct crm_subsystem_s));
	te_subsystem = calloc(1, sizeof(struct crm_subsystem_s));
	pe_subsystem = calloc(1, sizeof(struct crm_subsystem_s));

	if (cib_subsystem != NULL) {
	cib_subsystem->pid = -1;
	cib_subsystem->name = CRM_SYSTEM_CIB;
	cib_subsystem->flag_connected = R_CIB_CONNECTED;
	cib_subsystem->flag_required = R_CIB_REQUIRED;

	} else {
	was_error = TRUE;
	}

	if (te_subsystem != NULL) {
	te_subsystem->pid = -1;
	te_subsystem->name = CRM_SYSTEM_TENGINE;
	te_subsystem->flag_connected = R_TE_CONNECTED;
	te_subsystem->flag_required = R_TE_REQUIRED;

	} else {
	was_error = TRUE;
	}

	if (pe_subsystem != NULL) {
	pe_subsystem->pid = -1;
	pe_subsystem->path = CRM_DAEMON_DIR;
	pe_subsystem->name = CRM_SYSTEM_PENGINE;
	pe_subsystem->command = CRM_DAEMON_DIR "/" CRM_SYSTEM_PENGINE;
	pe_subsystem->args = NULL;
	pe_subsystem->flag_connected = R_PE_CONNECTED;
	pe_subsystem->flag_required = R_PE_REQUIRED;

	} else {
	was_error = TRUE;
	}

	if (was_error == FALSE && is_heartbeat_cluster()) {
	if (start_subsystem(pe_subsystem) == FALSE) {
	was_error = TRUE;
	}
	}

	if (was_error) {
	register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
	}

	}

	static int32_t
	crmd_ipc_accept(qb_ipcs_connection_t * c, uid_t uid, gid_t gid)
	{
	crm_trace("Connection %p", c);
	if (crm_client_new(c, uid, gid) == NULL) {
	return -EIO;
	}
	return 0;
	}

	static void
	crmd_ipc_created(qb_ipcs_connection_t * c)
	{
	crm_trace("Connection %p", c);
	}

	static int32_t
	crmd_ipc_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
	{
	uint32_t id = 0;
	uint32_t flags = 0;
	crm_client_t *client = crm_client_get(c);

	xmlNode *msg = crm_ipcs_recv(client, data, size, &id, &flags);

	crm_trace("Invoked: %s", crm_client_name(client));

	if (flags & crm_ipc_client_response) {
	crm_ipcs_send_ack(client, id, "ack", __FUNCTION__, __LINE__);
	}

	if (msg == NULL) {
	return 0;
	}
	#if ENABLE_ACL
	determine_request_user(client->user, msg, F_CRM_USER);
	#endif

	crm_trace("Processing msg from %s", crm_client_name(client));
	crm_log_xml_trace(msg, "CRMd[inbound]");

	crm_xml_add(msg, F_CRM_SYS_FROM, client->id);
	if (crmd_authorize_message(msg, client, NULL)) {
	route_message(C_IPC_MESSAGE, msg);
	}

	trigger_fsa(fsa_source);
	free_xml(msg);
	return 0;
	}

	static int32_t
	crmd_ipc_closed(qb_ipcs_connection_t * c)
	{
	crm_client_t *client = crm_client_get(c);
	struct crm_subsystem_s *the_subsystem = NULL;

	crm_trace("Connection %p", c);

	if (client->userdata == NULL) {
	crm_trace("Client hadn't registered with us yet");

	} else if (strcasecmp(CRM_SYSTEM_PENGINE, client->userdata) == 0) {
	the_subsystem = pe_subsystem;

	} else if (strcasecmp(CRM_SYSTEM_TENGINE, client->userdata) == 0) {
	the_subsystem = te_subsystem;

	} else if (strcasecmp(CRM_SYSTEM_CIB, client->userdata) == 0) {
	the_subsystem = cib_subsystem;
	}

	if (the_subsystem != NULL) {
	the_subsystem->source = NULL;
	the_subsystem->client = NULL;
	crm_info("Received HUP from %s:[%d]", the_subsystem->name, the_subsystem->pid);

	} else {
	/* else that was a transient client */
	crm_trace("Received HUP from transient client");
	}

	crm_trace("Disconnecting client %s (%p)", crm_client_name(client), client);
	free(client->userdata);
	crm_client_destroy(client);

	trigger_fsa(fsa_source);
	return 0;
	}

	static void
	crmd_ipc_destroy(qb_ipcs_connection_t * c)
	{
	crm_trace("Connection %p", c);
	}

	/* A_STOP */
	void
	do_stop(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	if (is_heartbeat_cluster()) {
	stop_subsystem(pe_subsystem, FALSE);
	}

	crm_trace("Closing IPC server");
	mainloop_del_ipc_server(ipcs); ipcs = NULL;
	register_fsa_input(C_FSA_INTERNAL, I_TERMINATE, NULL);
	}

	/* A_STARTED */
	void
	do_started(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	static struct qb_ipcs_service_handlers crmd_callbacks = {
	.connection_accept = crmd_ipc_accept,
	.connection_created = crmd_ipc_created,
	.msg_process = crmd_ipc_dispatch,
	.connection_closed = crmd_ipc_closed,
	.connection_destroyed = crmd_ipc_destroy
	};

	if (cur_state != S_STARTING) {
	crm_err("Start cancelled... %s", fsa_state2string(cur_state));
	return;

	} else if (is_set(fsa_input_register, R_MEMBERSHIP) == FALSE) {
	crm_info("Delaying start, no membership data (%.16llx)", R_MEMBERSHIP);

	crmd_fsa_stall(TRUE);
	return;

	} else if (is_set(fsa_input_register, R_LRM_CONNECTED) == FALSE) {
	crm_info("Delaying start, LRM not connected (%.16llx)", R_LRM_CONNECTED);

	crmd_fsa_stall(TRUE);
	return;

	} else if (is_set(fsa_input_register, R_CIB_CONNECTED) == FALSE) {
	crm_info("Delaying start, CIB not connected (%.16llx)", R_CIB_CONNECTED);

	crmd_fsa_stall(TRUE);
	return;

	} else if (is_set(fsa_input_register, R_READ_CONFIG) == FALSE) {
	crm_info("Delaying start, Config not read (%.16llx)", R_READ_CONFIG);

	crmd_fsa_stall(TRUE);
	return;

	} else if (is_set(fsa_input_register, R_PEER_DATA) == FALSE) {

	/* try reading from HA */
	crm_info("Delaying start, No peer data (%.16llx)", R_PEER_DATA);

	#if SUPPORT_HEARTBEAT
	if (is_heartbeat_cluster()) {
	HA_Message *msg = NULL;

	crm_trace("Looking for a HA message");
	msg = fsa_cluster_conn->llc_ops->readmsg(fsa_cluster_conn, 0);
	if (msg != NULL) {
	crm_trace("There was a HA message");
	ha_msg_del(msg);
	}
	}
	#endif
	crmd_fsa_stall(TRUE);
	return;
	}

	crm_debug("Init server comms");
	ipcs = crmd_ipc_server_init(&crmd_callbacks);
	if (ipcs == NULL) {
	crm_err("Failed to create IPC server: shutting down and inhibiting respawn");
	register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
	}

	if (stonith_reconnect == NULL) {
	int dummy;

	stonith_reconnect = mainloop_add_trigger(G_PRIORITY_LOW, te_connect_stonith, &dummy);
	}
	set_bit(fsa_input_register, R_ST_REQUIRED);
	mainloop_set_trigger(stonith_reconnect);

	crm_notice("The local CRM is operational");
	clear_bit(fsa_input_register, R_STARTING);
	register_fsa_input(msg_data->fsa_cause, I_PENDING, NULL);
	}

	/* A_RECOVER */
	void
	do_recover(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	set_bit(fsa_input_register, R_IN_RECOVERY);
	crm_warn("Fast-tracking shutdown in response to errors");

	register_fsa_input(C_FSA_INTERNAL, I_TERMINATE, NULL);
	}

	/* INDENT-OFF */
	pe_cluster_option crmd_opts[] = {
	/* name, old-name, validate, default, description */
	{ "dc-version", NULL, "string", NULL, "none", NULL, "Version of Pacemaker on the cluster's DC.", "Includes the hash which identifies the exact Mercurial changeset it was built from. Used for diagnostic purposes." },
	{ "cluster-infrastructure", NULL, "string", NULL, "heartbeat", NULL, "The messaging stack on which Pacemaker is currently running.", "Used for informational and diagnostic purposes." },
	{ XML_CONFIG_ATTR_DC_DEADTIME, "dc_deadtime", "time", NULL, "20s", &check_time, "How long to wait for a response from other nodes during startup.", "The \"correct\" value will depend on the speed/load of your network and the type of switches used." },
	{ XML_CONFIG_ATTR_RECHECK, "cluster_recheck_interval", "time",
	"Zero disables polling. Positive values are an interval in seconds (unless other SI units are specified. eg. 5min)", "15min", &check_timer,
	"Polling interval for time based changes to options, resource parameters and constraints.",
	"The Cluster is primarily event driven, however the configuration can have elements that change based on time."
	" To ensure these changes take effect, we can optionally poll the cluster's status for changes." },
	{ XML_CONFIG_ATTR_ELECTION_FAIL, "election_timeout", "time", NULL, "2min", &check_timer, "* Advanced Use Only *.", "If need to adjust this value, it probably indicates the presence of a bug." },
	{ XML_CONFIG_ATTR_FORCE_QUIT, "shutdown_escalation", "time", NULL, "20min", &check_timer, "* Advanced Use Only *.", "If need to adjust this value, it probably indicates the presence of a bug." },
	{ "crmd-integration-timeout", NULL, "time", NULL, "3min", &check_timer, "* Advanced Use Only *.", "If need to adjust this value, it probably indicates the presence of a bug." },
	{ "crmd-finalization-timeout", NULL, "time", NULL, "30min", &check_timer, "* Advanced Use Only *.", "If you need to adjust this value, it probably indicates the presence of a bug." },
	{ "crmd-transition-delay", NULL, "time", NULL, "0s", &check_timer, "* Advanced Use Only *\nEnabling this option will slow down cluster recovery under all conditions", "Delay cluster recovery for the configured interval to allow for additional/related events to occur.\nUseful if your configuration is sensitive to the order in which ping updates arrive." },
	{ XML_ATTR_EXPECTED_VOTES, NULL, "integer", NULL, "2", &check_number, "The number of nodes expected to be in the cluster", "Used to calculate quorum in openais based clusters." },
	};
	/* INDENT-ON */

	void
	crmd_metadata(void)
	{
	config_metadata("CRM Daemon", "1.0",
	"CRM Daemon Options",
	"This is a fake resource that details the options that can be configured for the CRM Daemon.",
	crmd_opts, DIMOF(crmd_opts));
	}

	static void
	verify_crmd_options(GHashTable * options)
	{
	verify_all_options(options, crmd_opts, DIMOF(crmd_opts));
	}

	static const char *
	crmd_pref(GHashTable * options, const char *name)
	{
	return get_cluster_pref(options, crmd_opts, DIMOF(crmd_opts), name);
	}

	static void
	config_query_callback(xmlNode * msg, int call_id, int rc, xmlNode * output, void *user_data)
	{
	const char *value = NULL;
	GHashTable *config_hash = NULL;
	crm_time_t *now = crm_time_new(NULL);

	if (rc != pcmk_ok) {
	fsa_data_t *msg_data = NULL;

	crm_err("Local CIB query resulted in an error: %s", pcmk_strerror(rc));
	register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);

	if (rc == -EACCES \|\| rc == -pcmk_err_dtd_validation) {
	crm_err("The cluster is mis-configured - shutting down and staying down");
	set_bit(fsa_input_register, R_STAYDOWN);
	}
	goto bail;
	}

	crm_debug("Call %d : Parsing CIB options", call_id);
	config_hash =
	g_hash_table_new_full(crm_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str);

	unpack_instance_attributes(output, output, XML_CIB_TAG_PROPSET, NULL, config_hash,
	CIB_OPTIONS_FIRST, FALSE, now);

	verify_crmd_options(config_hash);

	value = crmd_pref(config_hash, XML_CONFIG_ATTR_DC_DEADTIME);
	election_trigger->period_ms = crm_get_msec(value);

	value = crmd_pref(config_hash, XML_CONFIG_ATTR_FORCE_QUIT);
	shutdown_escalation_timer->period_ms = crm_get_msec(value);
	crm_debug("Shutdown escalation occurs after: %dms", shutdown_escalation_timer->period_ms);

	value = crmd_pref(config_hash, XML_CONFIG_ATTR_ELECTION_FAIL);
	election_timeout->period_ms = crm_get_msec(value);

	value = crmd_pref(config_hash, XML_CONFIG_ATTR_RECHECK);
	recheck_timer->period_ms = crm_get_msec(value);
	crm_debug("Checking for expired actions every %dms", recheck_timer->period_ms);

	value = crmd_pref(config_hash, "crmd-transition-delay");
	transition_timer->period_ms = crm_get_msec(value);

	value = crmd_pref(config_hash, "crmd-integration-timeout");
	integration_timer->period_ms = crm_get_msec(value);

	value = crmd_pref(config_hash, "crmd-finalization-timeout");
	finalization_timer->period_ms = crm_get_msec(value);

	#if SUPPORT_COROSYNC
	if (is_classic_ais_cluster()) {
	value = crmd_pref(config_hash, XML_ATTR_EXPECTED_VOTES);
	crm_debug("Sending expected-votes=%s to corosync", value);
	send_ais_text(crm_class_quorum, value, TRUE, NULL, crm_msg_ais);
	}
	#endif

	set_bit(fsa_input_register, R_READ_CONFIG);
	crm_trace("Triggering FSA: %s", __FUNCTION__);
	mainloop_set_trigger(fsa_source);

	g_hash_table_destroy(config_hash);
	bail:
	crm_time_free(now);
	}

	gboolean
	crm_read_options(gpointer user_data)
	{
	int call_id =
	fsa_cib_conn->cmds->query(fsa_cib_conn, XML_CIB_TAG_CRMCONFIG, NULL, cib_scope_local);

	fsa_register_cib_callback(call_id, FALSE, NULL, config_query_callback);
	crm_trace("Querying the CIB... call %d", call_id);
	return TRUE;
	}

	/* A_READCONFIG */
	void
	do_read_config(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state,
	enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	mainloop_set_trigger(config_read);
	}

	void
	crm_shutdown(int nsig)
	{
	if (crmd_mainloop != NULL && g_main_is_running(crmd_mainloop)) {
	if (is_set(fsa_input_register, R_SHUTDOWN)) {
	crm_err("Escalating the shutdown");
	register_fsa_input_before(C_SHUTDOWN, I_ERROR, NULL);

	} else {
	set_bit(fsa_input_register, R_SHUTDOWN);
	register_fsa_input(C_SHUTDOWN, I_SHUTDOWN, NULL);

	if (shutdown_escalation_timer->period_ms < 1) {
	const char *value = crmd_pref(NULL, XML_CONFIG_ATTR_FORCE_QUIT);
	int msec = crm_get_msec(value);

	crm_debug("Using default shutdown escalation: %dms", msec);
	shutdown_escalation_timer->period_ms = msec;
	}

	/* cant rely on this... */
	crm_notice("Requesting shutdown, upper limit is %dms",
	shutdown_escalation_timer->period_ms);
	crm_timer_start(shutdown_escalation_timer);
	}

	} else {
	crm_info("exit from shutdown");
	crmd_exit(EX_OK);
	}
	}
	diff --git a/crmd/tengine.c b/crmd/tengine.c
	index 9ff458c734..8e236f1f4e 100644
	--- a/crmd/tengine.c
	+++ b/crmd/tengine.c
	@@ -1,233 +1,229 @@
	/*
	* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
	*
	* This program is free software; you can redistribute it and/or
	* modify it under the terms of the GNU General Public
	* License as published by the Free Software Foundation; either
	* version 2 of the License, or (at your option) any later version.
	*
	* This software is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	* General Public License for more details.
	*
	* You should have received a copy of the GNU General Public
	* License along with this library; if not, write to the Free Software
	* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	*/

	#include <crm_internal.h>

	#include <sys/param.h>
	#include <crm/crm.h>
	#include <crmd_fsa.h>

	#include <sys/types.h>
	#include <sys/wait.h>

	#include <unistd.h> /* for access */

	#include <sys/types.h> /* for calls to open */
	#include <sys/stat.h> /* for calls to open */
	#include <fcntl.h> /* for calls to open */
	#include <pwd.h> /* for getpwuid */
	#include <grp.h> /* for initgroups */

	#include <sys/time.h> /* for getrlimit */
	#include <sys/resource.h> /* for getrlimit */

	#include <errno.h>

	#include <crm/msg_xml.h>
	#include <crm/common/xml.h>
	#include <crmd_messages.h>
	#include <crmd_callbacks.h>

	#include <crmd.h>
	#include <tengine.h>
	#include <te_callbacks.h>

	extern crm_graph_functions_t te_graph_fns;
	struct crm_subsystem_s *te_subsystem = NULL;
	stonith_t *stonith_api = NULL;

	static void
	global_cib_callback(const xmlNode * msg, int callid, int rc, xmlNode * output)
	{
	}

	static crm_graph_t *
	create_blank_graph(void)
	{
	crm_graph_t *a_graph = unpack_graph(NULL, NULL);

	a_graph->complete = TRUE;
	a_graph->abort_reason = "DC Takeover";
	a_graph->completion_action = tg_restart;
	return a_graph;
	}

	/* A_TE_START, A_TE_STOP, A_TE_RESTART */
	void
	do_te_control(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state,
	enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{
	gboolean init_ok = TRUE;

	if (action & A_TE_STOP) {
	if (transition_graph) {
	destroy_graph(transition_graph);
	transition_graph = NULL;
	}

	if (fsa_cib_conn) {
	fsa_cib_conn->cmds->del_notify_callback(fsa_cib_conn, T_CIB_DIFF_NOTIFY,
	te_update_diff);
	}

	clear_bit(fsa_input_register, te_subsystem->flag_connected);
	crm_info("Transitioner is now inactive");
	}

	if ((action & A_TE_START) == 0) {
	return;

	} else if (is_set(fsa_input_register, te_subsystem->flag_connected)) {
	crm_debug("The transitioner is already active");
	return;

	} else if ((action & A_TE_START) && cur_state == S_STOPPING) {
	crm_info("Ignoring request to start %s while shutting down", te_subsystem->name);
	return;
	}

	te_uuid = crm_generate_uuid();
	crm_info("Registering TE UUID: %s", te_uuid);

	- if (transition_trigger == NULL) {
	- transition_trigger = mainloop_add_trigger(G_PRIORITY_LOW, te_graph_trigger, NULL);
	- }
	-
	if (pcmk_ok !=
	fsa_cib_conn->cmds->add_notify_callback(fsa_cib_conn, T_CIB_DIFF_NOTIFY, te_update_diff)) {
	crm_err("Could not set CIB notification callback");
	init_ok = FALSE;
	}

	if (pcmk_ok != fsa_cib_conn->cmds->set_op_callback(fsa_cib_conn, global_cib_callback)) {
	crm_err("Could not set CIB global callback");
	init_ok = FALSE;
	}

	if (init_ok) {
	set_graph_functions(&te_graph_fns);

	if (transition_graph) {
	destroy_graph(transition_graph);
	}

	/* create a blank one */
	crm_debug("Transitioner is now active");
	transition_graph = create_blank_graph();
	set_bit(fsa_input_register, te_subsystem->flag_connected);
	}
	}

	/* A_TE_INVOKE, A_TE_CANCEL */
	void
	do_te_invoke(long long action,
	enum crmd_fsa_cause cause,
	enum crmd_fsa_state cur_state,
	enum crmd_fsa_input current_input, fsa_data_t * msg_data)
	{

	if (AM_I_DC == FALSE \|\| (fsa_state != S_TRANSITION_ENGINE && (action & A_TE_INVOKE))) {
	crm_notice("No need to invoke the TE (%s) in state %s",
	fsa_action2string(action), fsa_state2string(fsa_state));
	return;
	}

	if (action & A_TE_CANCEL) {
	crm_debug("Cancelling the transition: %s",
	transition_graph->complete ? "inactive" : "active");
	abort_transition(INFINITY, tg_restart, "Peer Cancelled", NULL);
	if (transition_graph->complete == FALSE) {
	crmd_fsa_stall(FALSE);
	}

	} else if (action & A_TE_HALT) {
	crm_debug("Halting the transition: %s", transition_graph->complete ? "inactive" : "active");
	abort_transition(INFINITY, tg_stop, "Peer Halt", NULL);
	if (transition_graph->complete == FALSE) {
	crmd_fsa_stall(FALSE);
	}

	} else if (action & A_TE_INVOKE) {
	const char *value = NULL;
	xmlNode *graph_data = NULL;
	ha_msg_input_t *input = fsa_typed_data(fsa_dt_ha_msg);
	const char *ref = crm_element_value(input->msg, XML_ATTR_REFERENCE);
	const char *graph_file = crm_element_value(input->msg, F_CRM_TGRAPH);
	const char *graph_input = crm_element_value(input->msg, F_CRM_TGRAPH_INPUT);

	if (graph_file == NULL && input->xml == NULL) {
	crm_log_xml_err(input->msg, "Bad command");
	register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL);
	return;
	}

	if (transition_graph->complete == FALSE) {
	crm_info("Another transition is already active");
	abort_transition(INFINITY, tg_restart, "Transition Active", NULL);
	return;
	}

	if (fsa_pe_ref == NULL \|\| safe_str_neq(fsa_pe_ref, ref)) {
	crm_info("Transition is redundant: %s vs. %s", crm_str(fsa_pe_ref), crm_str(ref));
	abort_transition(INFINITY, tg_restart, "Transition Redundant", NULL);
	}

	graph_data = input->xml;

	if (graph_data == NULL && graph_file != NULL) {
	graph_data = filename2xml(graph_file);
	}

	if (is_timer_started(transition_timer)) {
	crm_debug("The transitioner wait for a transition timer");
	return;
	}

	CRM_CHECK(graph_data != NULL,
	crm_err("Input raised by %s is invalid", msg_data->origin);
	crm_log_xml_err(input->msg, "Bad command");
	return);

	destroy_graph(transition_graph);
	transition_graph = unpack_graph(graph_data, graph_input);
	CRM_CHECK(transition_graph != NULL, transition_graph = create_blank_graph(); return);
	crm_info("Processing graph %d (ref=%s) derived from %s", transition_graph->id, ref,
	graph_input);

	value = crm_element_value(graph_data, "failed-stop-offset");
	if (value) {
	free(failed_stop_offset);
	failed_stop_offset = strdup(value);
	}

	value = crm_element_value(graph_data, "failed-start-offset");
	if (value) {
	free(failed_start_offset);
	failed_start_offset = strdup(value);
	}

	trigger_graph();
	print_graph(LOG_DEBUG_2, transition_graph);

	if (graph_data != input->xml) {
	free_xml(graph_data);
	}
	}
	}
	diff --git a/lib/common/ipc.c b/lib/common/ipc.c
	index 5100503b61..d870b5af2e 100644
	--- a/lib/common/ipc.c
	+++ b/lib/common/ipc.c
	@@ -1,1145 +1,1145 @@
	/*
	* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
	*
	* This library is free software; you can redistribute it and/or
	* modify it under the terms of the GNU Lesser General Public
	* License as published by the Free Software Foundation; either
	* version 2.1 of the License, or (at your option) any later version.
	*
	* This library is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	* Lesser General Public License for more details.
	*
	* You should have received a copy of the GNU Lesser General Public
	* License along with this library; if not, write to the Free Software
	* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	*/

	#include <crm_internal.h>

	#include <sys/param.h>

	#include <stdio.h>
	#include <sys/types.h>
	#include <sys/stat.h>
	#include <unistd.h>
	#include <grp.h>

	#include <errno.h>
	#include <fcntl.h>
	#include <bzlib.h>

	#include <crm/crm.h>
	#include <crm/msg_xml.h>
	#include <crm/common/ipc.h>
	#include <crm/common/ipcs.h>

	struct crm_ipc_response_header {
	struct qb_ipc_response_header qb;
	uint32_t size_uncompressed;
	uint32_t size_compressed;
	uint32_t flags;
	};

	static int hdr_offset = 0;
	static int ipc_buffer_max = 0;
	static int pick_ipc_buffer(int max);

	static inline void
	crm_ipc_init(void)
	{
	if (hdr_offset == 0) {
	hdr_offset = sizeof(struct crm_ipc_response_header);
	}
	if (ipc_buffer_max == 0) {
	ipc_buffer_max = pick_ipc_buffer(0);
	}
	}

	static char *
	generateReference(const char custom1, const char custom2)
	{
	static uint ref_counter = 0;
	const char *local_cust1 = custom1;
	const char *local_cust2 = custom2;
	int reference_len = 4;
	char *since_epoch = NULL;

	reference_len += 20; /* too big */
	reference_len += 40; /* too big */

	if (local_cust1 == NULL) {
	local_cust1 = "_empty_";
	}
	reference_len += strlen(local_cust1);

	if (local_cust2 == NULL) {
	local_cust2 = "_empty_";
	}
	reference_len += strlen(local_cust2);

	since_epoch = calloc(1, reference_len);

	if (since_epoch != NULL) {
	sprintf(since_epoch, "%s-%s-%ld-%u",
	local_cust1, local_cust2, (unsigned long)time(NULL), ref_counter++);
	}

	return since_epoch;
	}

	xmlNode *
	create_request_adv(const char task, xmlNode msg_data,
	const char host_to, const char sys_to,
	const char sys_from, const char uuid_from, const char *origin)
	{
	char *true_from = NULL;
	xmlNode *request = NULL;
	char *reference = generateReference(task, sys_from);

	if (uuid_from != NULL) {
	true_from = generate_hash_key(sys_from, uuid_from);
	} else if (sys_from != NULL) {
	true_from = strdup(sys_from);
	} else {
	crm_err("No sys from specified");
	}

	/* host_from will get set for us if necessary by CRMd when routed */
	request = create_xml_node(NULL, __FUNCTION__);
	crm_xml_add(request, F_CRM_ORIGIN, origin);
	crm_xml_add(request, F_TYPE, T_CRM);
	crm_xml_add(request, F_CRM_VERSION, CRM_FEATURE_SET);
	crm_xml_add(request, F_CRM_MSG_TYPE, XML_ATTR_REQUEST);
	crm_xml_add(request, F_CRM_REFERENCE, reference);
	crm_xml_add(request, F_CRM_TASK, task);
	crm_xml_add(request, F_CRM_SYS_TO, sys_to);
	crm_xml_add(request, F_CRM_SYS_FROM, true_from);

	/* HOSTTO will be ignored if it is to the DC anyway. */
	if (host_to != NULL && strlen(host_to) > 0) {
	crm_xml_add(request, F_CRM_HOST_TO, host_to);
	}

	if (msg_data != NULL) {
	add_message_xml(request, F_CRM_DATA, msg_data);
	}
	free(reference);
	free(true_from);

	return request;
	}

	/*
	* This method adds a copy of xml_response_data
	*/
	xmlNode *
	create_reply_adv(xmlNode * original_request, xmlNode * xml_response_data, const char *origin)
	{
	xmlNode *reply = NULL;

	const char *host_from = crm_element_value(original_request, F_CRM_HOST_FROM);
	const char *sys_from = crm_element_value(original_request, F_CRM_SYS_FROM);
	const char *sys_to = crm_element_value(original_request, F_CRM_SYS_TO);
	const char *type = crm_element_value(original_request, F_CRM_MSG_TYPE);
	const char *operation = crm_element_value(original_request, F_CRM_TASK);
	const char *crm_msg_reference = crm_element_value(original_request, F_CRM_REFERENCE);

	if (type == NULL) {
	crm_err("Cannot create new_message, no message type in original message");
	CRM_ASSERT(type != NULL);
	return NULL;
	#if 0
	} else if (strcasecmp(XML_ATTR_REQUEST, type) != 0) {
	crm_err("Cannot create new_message, original message was not a request");
	return NULL;
	#endif
	}
	reply = create_xml_node(NULL, __FUNCTION__);
	if (reply == NULL) {
	crm_err("Cannot create new_message, malloc failed");
	return NULL;
	}

	crm_xml_add(reply, F_CRM_ORIGIN, origin);
	crm_xml_add(reply, F_TYPE, T_CRM);
	crm_xml_add(reply, F_CRM_VERSION, CRM_FEATURE_SET);
	crm_xml_add(reply, F_CRM_MSG_TYPE, XML_ATTR_RESPONSE);
	crm_xml_add(reply, F_CRM_REFERENCE, crm_msg_reference);
	crm_xml_add(reply, F_CRM_TASK, operation);

	/* since this is a reply, we reverse the from and to */
	crm_xml_add(reply, F_CRM_SYS_TO, sys_from);
	crm_xml_add(reply, F_CRM_SYS_FROM, sys_to);

	/* HOSTTO will be ignored if it is to the DC anyway. */
	if (host_from != NULL && strlen(host_from) > 0) {
	crm_xml_add(reply, F_CRM_HOST_TO, host_from);
	}

	if (xml_response_data != NULL) {
	add_message_xml(reply, F_CRM_DATA, xml_response_data);
	}

	return reply;
	}

	/* Libqb based IPC */

	/* Server... */

	GHashTable *client_connections = NULL;

	crm_client_t *
	crm_client_get(qb_ipcs_connection_t * c)
	{
	if (client_connections) {
	return g_hash_table_lookup(client_connections, c);
	}

	crm_trace("No client found for %p", c);
	return NULL;
	}

	crm_client_t *
	crm_client_get_by_id(const char *id)
	{
	gpointer key;
	crm_client_t *client;
	GHashTableIter iter;

	if (client_connections && id) {
	g_hash_table_iter_init(&iter, client_connections);
	while (g_hash_table_iter_next(&iter, &key, (gpointer *) & client)) {
	if (strcmp(client->id, id) == 0) {
	return client;
	}
	}
	}

	crm_trace("No client found with id=%s", id);
	return NULL;
	}

	const char *
	crm_client_name(crm_client_t * c)
	{
	if (c == NULL) {
	return "null";
	} else if (c->name == NULL && c->id == NULL) {
	return "unknown";
	} else if (c->name == NULL) {
	return c->id;
	} else {
	return c->name;
	}
	}

	void
	crm_client_init(void)
	{
	if (client_connections == NULL) {
	crm_trace("Creating client hash table");
	client_connections = g_hash_table_new(g_direct_hash, g_direct_equal);
	}
	}

	void
	crm_client_cleanup(void)
	{
	- if (client_connections == NULL) {
	+ if (client_connections != NULL) {
	int active = g_hash_table_size(client_connections);

	if (active) {
	crm_err("Exiting with %d active connections", active);
	}
	g_hash_table_destroy(client_connections);
	}
	}

	crm_client_t *
	crm_client_new(qb_ipcs_connection_t * c, uid_t uid_client, gid_t gid_client)
	{
	static uid_t uid_server = 0;
	static gid_t gid_cluster = 0;

	crm_client_t *client = NULL;

	CRM_LOG_ASSERT(c);
	if (c == NULL) {
	return NULL;
	}

	if (gid_cluster == 0) {
	uid_server = getuid();
	if(crm_user_lookup(CRM_DAEMON_USER, NULL, &gid_cluster) < 0) {
	static bool have_error = FALSE;
	if(have_error == FALSE) {
	crm_warn("Could not find group for user %s", CRM_DAEMON_USER);
	have_error = TRUE;
	}
	}
	}

	if(gid_cluster != 0 && gid_client != 0) {
	uid_t best_uid = -1; /* Passing -1 to chown(2) means don't change */

	if(uid_client == 0 \|\| uid_server == 0) { /* Someone is priveliged, but the other may not be */
	best_uid = QB_MAX(uid_client, uid_server);
	crm_trace("Allowing user %u to clean up after disconnect", best_uid);
	}

	crm_trace("Giving access to group %u", gid_cluster);
	qb_ipcs_connection_auth_set(c, best_uid, gid_cluster, S_IRUSR \| S_IWUSR \| S_IRGRP \| S_IWGRP);
	}

	crm_client_init();

	client = calloc(1, sizeof(crm_client_t));

	client->ipcs = c;
	client->kind = CRM_CLIENT_IPC;
	client->pid = crm_ipcs_client_pid(c);

	client->id = crm_generate_uuid();

	crm_info("Connecting %p for uid=%d gid=%d pid=%u id=%s", c, uid_client, gid_client, client->pid, client->id);

	#if ENABLE_ACL
	client->user = uid2username(uid_client);
	#endif

	g_hash_table_insert(client_connections, c, client);
	return client;
	}

	void
	crm_client_destroy(crm_client_t * c)
	{
	if (c == NULL) {
	return;
	}

	if (client_connections) {
	if (c->ipcs) {
	crm_trace("Destroying %p/%p (%d remaining)",
	c, c->ipcs, crm_hash_table_size(client_connections) - 1);
	g_hash_table_remove(client_connections, c->ipcs);

	} else {
	crm_trace("Destroying remote connection %p (%d remaining)",
	c, crm_hash_table_size(client_connections) - 1);
	g_hash_table_remove(client_connections, c->id);
	}
	}

	if (c->event_timer) {
	g_source_remove(c->event_timer);
	}

	crm_info("Destroying %d events", g_list_length(c->event_queue));
	while (c->event_queue) {
	struct iovec *event = c->event_queue->data;

	c->event_queue = g_list_remove(c->event_queue, event);
	free(event[0].iov_base);
	free(event[1].iov_base);
	free(event);
	}

	free(c->id);
	free(c->name);
	free(c->user);
	if (c->remote) {
	if (c->remote->auth_timeout) {
	g_source_remove(c->remote->auth_timeout);
	}
	free(c->remote->buffer);
	free(c->remote);
	}
	free(c);
	}

	int
	crm_ipcs_client_pid(qb_ipcs_connection_t * c)
	{
	struct qb_ipcs_connection_stats stats;

	stats.client_pid = 0;
	qb_ipcs_connection_stats_get(c, &stats, 0);
	return stats.client_pid;
	}

	xmlNode *
	crm_ipcs_recv(crm_client_t * c, void data, size_t size, uint32_t id, uint32_t * flags)
	{
	xmlNode *xml = NULL;
	char *uncompressed = NULL;
	char text = ((char )data) + sizeof(struct crm_ipc_response_header);
	struct crm_ipc_response_header *header = data;

	if (id) {
	id = ((struct qb_ipc_response_header )data)->id;
	}
	if (flags) {
	*flags = header->flags;
	}

	if (header->flags & crm_ipc_compressed) {
	int rc = 0;
	unsigned int size_u = 1 + header->size_uncompressed;
	uncompressed = calloc(1, hdr_offset + size_u);

	crm_trace("Decompressing message data %d bytes into %d bytes",
	header->size_compressed, size_u);

	rc = BZ2_bzBuffToBuffDecompress(uncompressed, &size_u, text, header->size_compressed, 1, 0);
	text = uncompressed;

	if (rc != BZ_OK) {
	crm_err("Decompression failed: %s (%d)", bz2_strerror(rc), rc);
	free(uncompressed);
	return NULL;
	}
	}

	CRM_ASSERT(text[header->size_uncompressed - 1] == 0);

	crm_trace("Received %.200s", text);
	xml = string2xml(text);

	free(uncompressed);
	return xml;
	}

	ssize_t crm_ipcs_flush_events(crm_client_t * c);

	static gboolean
	crm_ipcs_flush_events_cb(gpointer data)
	{
	crm_client_t *c = data;

	c->event_timer = 0;
	crm_ipcs_flush_events(c);
	return FALSE;
	}

	ssize_t
	crm_ipcs_flush_events(crm_client_t * c)
	{
	int sent = 0;
	ssize_t rc = 0;
	int queue_len = 0;

	if (c == NULL) {
	return pcmk_ok;

	} else if (c->event_timer) {
	/* There is already a timer, wait until it goes off */
	crm_trace("Timer active for %p - %d", c->ipcs, c->event_timer);
	return pcmk_ok;
	}

	queue_len = g_list_length(c->event_queue);
	while (c->event_queue && sent < 100) {
	struct crm_ipc_response_header *header = NULL;
	struct iovec *event = c->event_queue->data;

	rc = qb_ipcs_event_sendv(c->ipcs, event, 2);
	if (rc < 0) {
	break;
	}

	sent++;
	header = event[0].iov_base;
	if (header->flags & crm_ipc_compressed) {
	crm_trace("Event %d to %p[%d] (%d compressed bytes) sent",
	header->qb.id, c->ipcs, c->pid, rc);
	} else {
	crm_trace("Event %d to %p[%d] (%d bytes) sent: %.120s",
	header->qb.id, c->ipcs, c->pid, rc, event[1].iov_base);
	}

	c->event_queue = g_list_remove(c->event_queue, event);
	free(event[0].iov_base);
	free(event[1].iov_base);
	free(event);
	}

	queue_len -= sent;
	if (sent > 0 \|\| c->event_queue) {
	crm_trace("Sent %d events (%d remaining) for %p[%d]: %s",
	sent, queue_len, c->ipcs, c->pid, pcmk_strerror(rc < 0 ? rc : 0));
	}

	if (c->event_queue) {
	if (queue_len % 100 == 0 && queue_len > 99) {
	crm_warn("Event queue for %p[%d] has grown to %d", c->ipcs, c->pid, queue_len);

	} else if (queue_len > 500) {
	crm_err("Evicting slow client %p[%d]: event queue reached %d entries",
	c->ipcs, c->pid, queue_len);
	qb_ipcs_disconnect(c->ipcs);
	return rc;
	}

	c->event_timer = g_timeout_add(1000 + 100 * queue_len, crm_ipcs_flush_events_cb, c);
	}

	return rc;
	}

	ssize_t
	crm_ipc_prepare(uint32_t request, xmlNode * message, struct iovec ** result)
	{
	static int biggest = 0;

	struct iovec *iov;
	unsigned int total = 0;
	char *compressed = NULL;
	char *buffer = dump_xml_unformatted(message);
	struct crm_ipc_response_header *header = calloc(1, sizeof(struct crm_ipc_response_header));

	CRM_ASSERT(result != NULL);

	*result = NULL;
	iov = calloc(2, sizeof(struct iovec));

	crm_ipc_init();

	iov[0].iov_len = hdr_offset;
	iov[0].iov_base = header;

	header->size_uncompressed = 1 + strlen(buffer);
	total = hdr_offset + header->size_uncompressed;

	if (total < ipc_buffer_max) {
	iov[1].iov_base = buffer;
	iov[1].iov_len = header->size_uncompressed;

	} else {
	unsigned int new_size = 0;

	if (total > biggest) {
	biggest = 2 * QB_MAX(total, biggest);
	crm_notice("Message exceeds the configured ipc limit (%d bytes), "
	"consider configuring PCMK_ipc_buffer to %d or higher "
	"to avoid compression overheads", ipc_buffer_max, biggest);
	}

	if (crm_compress_string
	(buffer, header->size_uncompressed, ipc_buffer_max, &compressed, &new_size)) {

	header->flags \|= crm_ipc_compressed;
	header->size_compressed = new_size;

	iov[1].iov_len = header->size_compressed;
	iov[1].iov_base = compressed;

	free(buffer);

	} else {
	ssize_t rc = -EMSGSIZE;

	crm_log_xml_trace(message, "EMSGSIZE");

	crm_err
	("Could not compress the message into less than the configured ipc limit (%d bytes)."
	"Set PCMK_ipc_buffer to a higher value (%d bytes suggested)", ipc_buffer_max,
	biggest);

	free(compressed);
	free(buffer);
	free(header);
	free(iov);

	return rc;
	}
	}

	header->qb.size = iov[0].iov_len + iov[1].iov_len;
	header->qb.id = request; /* Replying to a specific request */

	*result = iov;
	return header->qb.size;
	}

	ssize_t
	crm_ipcs_sendv(crm_client_t * c, struct iovec * iov, enum crm_ipc_server_flags flags)
	{
	ssize_t rc;
	static uint32_t id = 1;
	struct crm_ipc_response_header *header = iov[0].iov_base;

	if (flags & crm_ipc_server_event) {
	header->qb.id = id++; /* We don't really use it, but doesn't hurt to set one */

	if (flags & crm_ipc_server_free) {
	crm_trace("Sending the original to %p[%d]", c->ipcs, c->pid);
	c->event_queue = g_list_append(c->event_queue, iov);

	} else {
	struct iovec *iov_copy = calloc(2, sizeof(struct iovec));

	crm_trace("Sending a copy to %p[%d]", c->ipcs, c->pid);
	iov_copy[0].iov_len = iov[0].iov_len;
	iov_copy[0].iov_base = malloc(iov[0].iov_len);
	memcpy(iov_copy[0].iov_base, iov[0].iov_base, iov[0].iov_len);

	iov_copy[1].iov_len = iov[1].iov_len;
	iov_copy[1].iov_base = malloc(iov[1].iov_len);
	memcpy(iov_copy[1].iov_base, iov[1].iov_base, iov[1].iov_len);

	c->event_queue = g_list_append(c->event_queue, iov_copy);
	}

	} else {
	CRM_LOG_ASSERT(header->qb.id != 0); /* Replying to a specific request */

	rc = qb_ipcs_response_sendv(c->ipcs, iov, 2);
	if (rc < header->qb.size) {
	crm_notice("Response %d to %p[%d] (%d bytes) failed: %s (%d)",
	header->qb.id, c->ipcs, c->pid, header->qb.size, pcmk_strerror(rc), rc);

	} else {
	crm_trace("Response %d sent, %d bytes to %p[%d]", header->qb.id, rc, c->ipcs, c->pid);
	}

	if (flags & crm_ipc_server_free) {
	free(iov[0].iov_base);
	free(iov[1].iov_base);
	free(iov);
	}
	}

	if (flags & crm_ipc_server_event) {
	rc = crm_ipcs_flush_events(c);
	} else {
	crm_ipcs_flush_events(c);
	}

	if (rc == -EPIPE \|\| rc == -ENOTCONN) {
	crm_trace("Client %p disconnected", c->ipcs);
	}

	return rc;
	}

	ssize_t
	crm_ipcs_send(crm_client_t * c, uint32_t request, xmlNode * message,
	enum crm_ipc_server_flags flags)
	{
	struct iovec *iov = NULL;
	ssize_t rc = 0;

	if(c == NULL) {
	return -EDESTADDRREQ;
	}

	rc = crm_ipc_prepare(request, message, &iov);
	if (rc > 0) {
	rc = crm_ipcs_sendv(c, iov, flags \| crm_ipc_server_free);

	} else {
	free(iov);
	crm_notice("Message to %p[%d] failed: %s (%d)",
	c->ipcs, c->pid, pcmk_strerror(rc), rc);
	}

	return rc;
	}

	void
	crm_ipcs_send_ack(crm_client_t * c, uint32_t request, const char tag, const char function,
	int line)
	{
	xmlNode *ack = create_xml_node(NULL, tag);

	crm_xml_add(ack, "function", function);
	crm_xml_add_int(ack, "line", line);
	crm_ipcs_send(c, request, ack, 0);
	free_xml(ack);
	}

	/* Client... */

	#define MIN_MSG_SIZE 12336 /* sizeof(struct qb_ipc_connection_response) */
	#define MAX_MSG_SIZE 501024 / 50k default */

	struct crm_ipc_s {
	struct pollfd pfd;

	int buf_size;
	int msg_size;
	int need_reply;
	char *buffer;
	char *name;

	qb_ipcc_connection_t *ipc;

	};

	static int
	pick_ipc_buffer(int max)
	{
	const char *env = getenv("PCMK_ipc_buffer");

	if (env) {
	max = crm_parse_int(env, "0");
	}

	if (max <= 0) {
	max = MAX_MSG_SIZE;
	}

	if (max < MIN_MSG_SIZE) {
	max = MIN_MSG_SIZE;
	}

	crm_trace("Using max message size of %d", max);
	return max;
	}

	crm_ipc_t *
	crm_ipc_new(const char *name, size_t max_size)
	{
	crm_ipc_t *client = NULL;

	client = calloc(1, sizeof(crm_ipc_t));

	client->name = strdup(name);
	client->buf_size = pick_ipc_buffer(max_size);
	client->buffer = malloc(client->buf_size);

	client->pfd.fd = -1;
	client->pfd.events = POLLIN;
	client->pfd.revents = 0;

	return client;
	}

	bool
	crm_ipc_connect(crm_ipc_t * client)
	{
	client->need_reply = FALSE;
	client->ipc = qb_ipcc_connect(client->name, client->buf_size);

	if (client->ipc == NULL) {
	crm_perror(LOG_INFO, "Could not establish %s connection", client->name);
	return FALSE;
	}

	client->pfd.fd = crm_ipc_get_fd(client);
	if (client->pfd.fd < 0) {
	crm_perror(LOG_INFO, "Could not obtain file descriptor for %s connection", client->name);
	return FALSE;
	}

	qb_ipcc_context_set(client->ipc, client);

	return TRUE;
	}

	void
	crm_ipc_close(crm_ipc_t * client)
	{
	if (client) {
	crm_trace("Disconnecting %s IPC connection %p (%p.%p)", client->name, client, client->ipc);

	if (client->ipc) {
	qb_ipcc_connection_t *ipc = client->ipc;

	client->ipc = NULL;
	qb_ipcc_disconnect(ipc);
	}
	}
	}

	void
	crm_ipc_destroy(crm_ipc_t * client)
	{
	if (client) {
	if (client->ipc && qb_ipcc_is_connected(client->ipc)) {
	crm_notice("Destroying an active IPC connection to %s", client->name);
	/* The next line is basically unsafe
	*
	* If this connection was attached to mainloop and mainloop is active,
	* the 'disconnected' callback will end up back here and we'll end
	* up free'ing the memory twice - something that can still happen
	* even without this if we destroy a connection and it closes before
	* we call exit
	*/
	/* crm_ipc_close(client); */
	}
	crm_trace("Destroying IPC connection to %s: %p", client->name, client);
	free(client->buffer);
	free(client->name);
	free(client);
	}
	}

	int
	crm_ipc_get_fd(crm_ipc_t * client)
	{
	int fd = 0;

	CRM_ASSERT(client != NULL);
	if (client->ipc && qb_ipcc_fd_get(client->ipc, &fd) == 0) {
	return fd;
	}

	crm_perror(LOG_ERR, "Could not obtain file IPC descriptor for %s", client->name);
	return -EINVAL;
	}

	bool
	crm_ipc_connected(crm_ipc_t * client)
	{
	bool rc = FALSE;

	if (client == NULL) {
	crm_trace("No client");
	return FALSE;

	} else if (client->ipc == NULL) {
	crm_trace("No connection");
	return FALSE;

	} else if (client->pfd.fd < 0) {
	crm_trace("Bad descriptor");
	return FALSE;
	}

	rc = qb_ipcc_is_connected(client->ipc);
	if (rc == FALSE) {
	client->pfd.fd = -EINVAL;
	}
	return rc;
	}

	int
	crm_ipc_ready(crm_ipc_t * client)
	{
	CRM_ASSERT(client != NULL);

	if (crm_ipc_connected(client) == FALSE) {
	return -ENOTCONN;
	}

	client->pfd.revents = 0;
	return poll(&(client->pfd), 1, 0);
	}

	static int
	crm_ipc_decompress(crm_ipc_t * client)
	{
	struct crm_ipc_response_header header = (struct crm_ipc_response_header )client->buffer;

	if (header->flags & crm_ipc_compressed) {
	int rc = 0;
	unsigned int size_u = 1 + header->size_uncompressed;
	char *uncompressed = calloc(1, hdr_offset + size_u);

	crm_trace("Decompressing message data %d bytes into %d bytes",
	header->size_compressed, size_u);

	rc = BZ2_bzBuffToBuffDecompress(uncompressed + hdr_offset, &size_u,
	client->buffer + hdr_offset, header->size_compressed, 1, 0);

	if (rc != BZ_OK) {
	crm_err("Decompression failed: %s (%d)", bz2_strerror(rc), rc);
	free(uncompressed);
	return -EILSEQ;
	}

	CRM_ASSERT((header->size_uncompressed + hdr_offset) >= ipc_buffer_max);
	CRM_ASSERT(size_u == header->size_uncompressed);

	memcpy(uncompressed, client->buffer, hdr_offset); /* Preserve the header */
	header = (struct crm_ipc_response_header *)uncompressed;

	free(client->buffer);
	client->buf_size = hdr_offset + size_u;
	client->buffer = uncompressed;
	}

	CRM_ASSERT(client->buffer[hdr_offset + header->size_uncompressed - 1] == 0);
	return pcmk_ok;
	}

	long
	crm_ipc_read(crm_ipc_t * client)
	{
	struct crm_ipc_response_header *header = NULL;

	CRM_ASSERT(client != NULL);
	CRM_ASSERT(client->ipc != NULL);
	CRM_ASSERT(client->buffer != NULL);

	crm_ipc_init();

	client->buffer[0] = 0;
	client->msg_size = qb_ipcc_event_recv(client->ipc, client->buffer, client->buf_size - 1, 0);
	if (client->msg_size >= 0) {
	int rc = crm_ipc_decompress(client);

	if (rc != pcmk_ok) {
	return rc;
	}

	header = (struct crm_ipc_response_header *)client->buffer;
	crm_trace("Recieved %s event %d, size=%d, rc=%d, text: %.100s",
	client->name, header->qb.id, header->qb.size, client->msg_size,
	client->buffer + hdr_offset);

	} else {
	crm_trace("No message from %s recieved: %s", client->name, pcmk_strerror(client->msg_size));
	}

	if (crm_ipc_connected(client) == FALSE \|\| client->msg_size == -ENOTCONN) {
	crm_err("Connection to %s failed", client->name);
	}

	if (header) {
	/* Data excluding the header */
	return header->size_uncompressed;
	}
	return -ENOMSG;
	}

	const char *
	crm_ipc_buffer(crm_ipc_t * client)
	{
	CRM_ASSERT(client != NULL);
	return client->buffer + sizeof(struct crm_ipc_response_header);
	}

	const char *
	crm_ipc_name(crm_ipc_t * client)
	{
	CRM_ASSERT(client != NULL);
	return client->name;
	}

	static int
	internal_ipc_send_recv(crm_ipc_t * client, const void *iov)
	{
	int rc = 0;

	do {
	rc = qb_ipcc_sendv_recv(client->ipc, iov, 2, client->buffer, client->buf_size, -1);
	} while (rc == -EAGAIN && crm_ipc_connected(client));

	return rc;
	}

	static int
	internal_ipc_send_request(crm_ipc_t * client, const void *iov, int ms_timeout)
	{
	int rc = 0;
	time_t timeout = time(NULL) + 1 + (ms_timeout / 1000);

	do {
	rc = qb_ipcc_sendv(client->ipc, iov, 2);
	} while (rc == -EAGAIN && time(NULL) < timeout && crm_ipc_connected(client));

	return rc;
	}

	static int
	internal_ipc_get_reply(crm_ipc_t * client, int request_id, int ms_timeout)
	{
	time_t timeout = time(NULL) + 1 + (ms_timeout / 1000);
	int rc = 0;

	crm_ipc_init();

	/* get the reply */
	crm_trace("client %s waiting on reply to msg id %d", client->name, request_id);
	do {

	rc = qb_ipcc_recv(client->ipc, client->buffer, client->buf_size, 1000);
	if (rc > 0) {
	struct crm_ipc_response_header *hdr = NULL;

	int rc = crm_ipc_decompress(client);

	if (rc != pcmk_ok) {
	return rc;
	}

	hdr = (struct crm_ipc_response_header *)client->buffer;
	if (hdr->qb.id == request_id) {
	/* Got it */
	break;
	} else if (hdr->qb.id < request_id) {
	xmlNode *bad = string2xml(crm_ipc_buffer(client));

	crm_err("Discarding old reply %d (need %d)", hdr->qb.id, request_id);
	crm_log_xml_notice(bad, "OldIpcReply");

	} else {
	xmlNode *bad = string2xml(crm_ipc_buffer(client));

	crm_err("Discarding newer reply %d (need %d)", hdr->qb.id, request_id);
	crm_log_xml_notice(bad, "ImpossibleReply");
	CRM_ASSERT(hdr->qb.id <= request_id);
	}
	} else if (crm_ipc_connected(client) == FALSE) {
	crm_err("Server disconnected client %s while waiting for msg id %d", client->name,
	request_id);
	break;
	}

	} while (time(NULL) < timeout);

	return rc;
	}

	int
	crm_ipc_send(crm_ipc_t * client, xmlNode * message, enum crm_ipc_flags flags, int32_t ms_timeout,
	xmlNode ** reply)
	{
	long rc = 0;
	struct iovec *iov;
	static uint32_t id = 0;
	struct crm_ipc_response_header *header;

	crm_ipc_init();

	if (client == NULL) {
	crm_notice("Invalid connection");
	return -ENOTCONN;

	} else if (crm_ipc_connected(client) == FALSE) {
	/* Don't even bother */
	crm_notice("Connection to %s closed", client->name);
	return -ENOTCONN;
	}

	if (client->need_reply) {
	crm_trace("Trying again to obtain pending reply from %s", client->name);
	rc = qb_ipcc_recv(client->ipc, client->buffer, client->buf_size, 300);
	if (rc < 0) {
	crm_warn("Sending to %s (%p) is disabled until pending reply is recieved", client->name,
	client->ipc);
	return -EALREADY;

	} else {
	crm_notice("Lost reply from %s (%p) finally arrived, sending re-enabled", client->name,
	client->ipc);
	client->need_reply = FALSE;
	}
	}

	rc = crm_ipc_prepare(++id, message, &iov);
	if(rc < 0) {
	return rc;
	}

	header = iov[0].iov_base;
	header->flags \|= flags;

	if (ms_timeout == 0) {
	ms_timeout = 5000;
	}

	crm_trace("Sending from client: %s request id: %d bytes: %u timeout:%d msg...",
	client->name, header->qb.id, header->qb.size, ms_timeout);

	if (ms_timeout > 0) {

	rc = internal_ipc_send_request(client, iov, ms_timeout);

	if (rc <= 0) {
	crm_trace("Failed to send from client %s request %d with %u bytes...",
	client->name, header->qb.id, header->qb.size);
	goto send_cleanup;

	} else if (is_not_set(flags, crm_ipc_client_response)) {
	crm_trace("Message sent, not waiting for reply to %d from %s to %u bytes...",
	header->qb.id, client->name, header->qb.size);

	goto send_cleanup;
	}

	rc = internal_ipc_get_reply(client, header->qb.id, ms_timeout);
	if (rc < 0) {
	/* No reply, for now, disable sending
	*
	* The alternative is to close the connection since we don't know
	* how to detect and discard out-of-sequence replies
	*
	* TODO - implement the above
	*/
	client->need_reply = TRUE;
	}

	} else {
	rc = internal_ipc_send_recv(client, iov);
	}

	if (rc > 0) {
	struct crm_ipc_response_header hdr = (struct crm_ipc_response_header )client->buffer;

	crm_trace("Recieved response %d, size=%d, rc=%ld, text: %.200s", hdr->qb.id, hdr->qb.size,
	rc, crm_ipc_buffer(client));

	if (reply) {
	*reply = string2xml(crm_ipc_buffer(client));
	}

	} else {
	crm_trace("Response not recieved: rc=%ld, errno=%d", rc, errno);
	}

	send_cleanup:
	if (crm_ipc_connected(client) == FALSE) {
	crm_notice("Connection to %s closed: %s (%ld)", client->name, pcmk_strerror(rc), rc);

	} else if (rc == -ETIMEDOUT) {
	crm_warn("Request %d to %s (%p) failed: %s (%ld) after %dms",
	header->qb.id, client->name, client->ipc, pcmk_strerror(rc), rc, ms_timeout);
	crm_write_blackbox(0, NULL);

	} else if (rc <= 0) {
	crm_warn("Request %d to %s (%p) failed: %s (%ld)",
	header->qb.id, client->name, client->ipc, pcmk_strerror(rc), rc);
	}

	free(header);
	free(iov[1].iov_base);
	free(iov);
	return rc;
	}

	/* Utils */

	xmlNode *
	create_hello_message(const char *uuid,
	const char client_name, const char major_version, const char *minor_version)
	{
	xmlNode *hello_node = NULL;
	xmlNode *hello = NULL;

	if (uuid == NULL \|\| strlen(uuid) == 0
	\|\| client_name == NULL \|\| strlen(client_name) == 0
	\|\| major_version == NULL \|\| strlen(major_version) == 0
	\|\| minor_version == NULL \|\| strlen(minor_version) == 0) {
	crm_err("Missing fields, Hello message will not be valid.");
	return NULL;
	}

	hello_node = create_xml_node(NULL, XML_TAG_OPTIONS);
	crm_xml_add(hello_node, "major_version", major_version);
	crm_xml_add(hello_node, "minor_version", minor_version);
	crm_xml_add(hello_node, "client_name", client_name);
	crm_xml_add(hello_node, "client_uuid", uuid);

	crm_trace("creating hello message");
	hello = create_request(CRM_OP_HELLO, hello_node, NULL, NULL, client_name, uuid);
	free_xml(hello_node);

	return hello;
	}
	diff --git a/lib/common/mainloop.c b/lib/common/mainloop.c
	index d1ebbb9dc0..e422e7a4f1 100644
	--- a/lib/common/mainloop.c
	+++ b/lib/common/mainloop.c
	@@ -1,932 +1,989 @@
	/*
	* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
	*
	* This library is free software; you can redistribute it and/or
	* modify it under the terms of the GNU Lesser General Public
	* License as published by the Free Software Foundation; either
	* version 2.1 of the License, or (at your option) any later version.
	*
	* This library is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	* Lesser General Public License for more details.
	*
	* You should have received a copy of the GNU Lesser General Public
	* License along with this library; if not, write to the Free Software
	* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	*/

	#include <crm_internal.h>

	#ifndef _GNU_SOURCE
	# define _GNU_SOURCE
	#endif

	#include <stdlib.h>
	#include <signal.h>
	#include <errno.h>

	#include <sys/wait.h>

	#include <crm/crm.h>
	#include <crm/common/xml.h>
	#include <crm/common/mainloop.h>
	#include <crm/common/ipcs.h>

	struct mainloop_child_s {
	pid_t pid;
	char *desc;
	unsigned timerid;
	unsigned watchid;
	gboolean timeout;
	void *privatedata;

	/* Called when a process dies */
	void (callback) (mainloop_child_t p, pid_t pid, int core, int signo, int exitcode);
	};

	struct trigger_s {
	GSource source;
	gboolean running;
	gboolean trigger;
	void *user_data;
	guint id;

	};

	static gboolean
	crm_trigger_prepare(GSource * source, gint * timeout)
	{
	crm_trigger_t trig = (crm_trigger_t ) source;

	/* cluster-glue's FD and IPC related sources make use of
	* g_source_add_poll() but do not set a timeout in their prepare
	* functions
	*
	* This means mainloop's poll() will block until an event for one
	* of these sources occurs - any /other/ type of source, such as
	* this one or g_idle_*, that doesn't use g_source_add_poll() is
	* S-O-L and wont be processed until there is something fd-based
	* happens.
	*
	* Luckily the timeout we can set here affects all sources and
	* puts an upper limit on how long poll() can take.
	*
	* So unconditionally set a small-ish timeout, not too small that
	* we're in constant motion, which will act as an upper bound on
	* how long the signal handling might be delayed for.
	*/
	timeout = 500; / Timeout in ms */

	return trig->trigger;
	}

	static gboolean
	crm_trigger_check(GSource * source)
	{
	crm_trigger_t trig = (crm_trigger_t ) source;

	return trig->trigger;
	}

	static gboolean
	crm_trigger_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
	{
	int rc = TRUE;
	crm_trigger_t trig = (crm_trigger_t ) source;

	if (trig->running) {
	/* Wait until the existing job is complete before starting the next one */
	return TRUE;
	}
	trig->trigger = FALSE;

	if (callback) {
	rc = callback(trig->user_data);
	if (rc < 0) {
	crm_trace("Trigger handler %p not yet complete", trig);
	trig->running = TRUE;
	rc = TRUE;
	}
	}
	return rc;
	}

	+static void
	+crm_trigger_finalize(GSource * source)
	+{
	+ crm_trace("Trigger %p destroyed", source);
	+}
	+
	+#if 0
	+struct _GSourceCopy
	+{
	+ gpointer callback_data;
	+ GSourceCallbackFuncs *callback_funcs;
	+
	+ const GSourceFuncs *source_funcs;
	+ guint ref_count;
	+
	+ GMainContext *context;
	+
	+ gint priority;
	+ guint flags;
	+ guint source_id;
	+
	+ GSList *poll_fds;
	+
	+ GSource *prev;
	+ GSource *next;
	+
	+ char *name;
	+
	+ void *priv;
	+};
	+
	+static int
	+g_source_refcount(GSource * source)
	+{
	+ /* Duplicating the contents of private header files is a necessary evil */
	+ if (source) {
	+ struct _GSourceCopy evil = (struct _GSourceCopy)source;
	+ return evil->ref_count;
	+ }
	+ return 0;
	+}
	+#else
	+static int g_source_refcount(GSource * source)
	+{
	+ return 0;
	+}
	+#endif
	+
	static GSourceFuncs crm_trigger_funcs = {
	crm_trigger_prepare,
	crm_trigger_check,
	crm_trigger_dispatch,
	- NULL
	+ crm_trigger_finalize,
	};

	static crm_trigger_t *
	mainloop_setup_trigger(GSource * source, int priority, int (*dispatch) (gpointer user_data),
	gpointer userdata)
	{
	crm_trigger_t *trigger = NULL;

	trigger = (crm_trigger_t *) source;

	trigger->id = 0;
	trigger->trigger = FALSE;
	trigger->user_data = userdata;

	if (dispatch) {
	g_source_set_callback(source, dispatch, trigger, NULL);
	}

	g_source_set_priority(source, priority);
	g_source_set_can_recurse(source, FALSE);

	+ crm_trace("Setup %p with ref-count=%u", source, g_source_refcount(source));
	trigger->id = g_source_attach(source, NULL);
	+ crm_trace("Attached %p with ref-count=%u", source, g_source_refcount(source));
	+
	return trigger;
	}

	void
	mainloop_trigger_complete(crm_trigger_t * trig)
	{
	crm_trace("Trigger handler %p complete", trig);
	trig->running = FALSE;
	}

	/* If dispatch returns:
	* -1: Job running but not complete
	* 0: Remove the trigger from mainloop
	* 1: Leave the trigger in mainloop
	*/
	crm_trigger_t *
	mainloop_add_trigger(int priority, int (*dispatch) (gpointer user_data), gpointer userdata)
	{
	GSource *source = NULL;

	CRM_ASSERT(sizeof(crm_trigger_t) > sizeof(GSource));
	source = g_source_new(&crm_trigger_funcs, sizeof(crm_trigger_t));
	CRM_ASSERT(source != NULL);

	return mainloop_setup_trigger(source, priority, dispatch, userdata);
	}

	void
	mainloop_set_trigger(crm_trigger_t * source)
	{
	source->trigger = TRUE;
	}

	gboolean
	mainloop_destroy_trigger(crm_trigger_t * source)
	{
	GSource gs = (GSource )source;
	- g_source_destroy(gs);
	+ crm_trace("Destroying %p with ref-count=%u", source, g_source_refcount(gs));
	+
	+ g_source_unref(gs); /* The caller no longer carries a reference to source */
	+ if(g_source_refcount(gs) > 1) {
	+ crm_info("Trigger %p is still referenced %u times", source, g_source_refcount(gs));
	+ }
	+ g_source_destroy(gs); /* Remove from mainloop and free() now that ref-count is 0 */
	return TRUE;
	}

	typedef struct signal_s {
	crm_trigger_t trigger; /* must be first */
	void (*handler) (int sig);
	int signal;

	} crm_signal_t;

	static crm_signal_t *crm_signals[NSIG];

	static gboolean
	crm_signal_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
	{
	crm_signal_t sig = (crm_signal_t ) source;

	if(sig->signal != SIGCHLD) {
	crm_info("Invoking handler for signal %d: %s", sig->signal, strsignal(sig->signal));
	}

	sig->trigger.trigger = FALSE;
	if (sig->handler) {
	sig->handler(sig->signal);
	}
	return TRUE;
	}

	static void
	mainloop_signal_handler(int sig)
	{
	if (sig > 0 && sig < NSIG && crm_signals[sig] != NULL) {
	mainloop_set_trigger((crm_trigger_t *) crm_signals[sig]);
	}
	}

	static GSourceFuncs crm_signal_funcs = {
	crm_trigger_prepare,
	crm_trigger_check,
	crm_signal_dispatch,
	- NULL
	+ crm_trigger_finalize,
	};

	gboolean
	crm_signal(int sig, void (*dispatch) (int sig))
	{
	sigset_t mask;
	struct sigaction sa;
	struct sigaction old;

	if (sigemptyset(&mask) < 0) {
	crm_perror(LOG_ERR, "Call to sigemptyset failed");
	return FALSE;
	}

	memset(&sa, 0, sizeof(struct sigaction));
	sa.sa_handler = dispatch;
	sa.sa_flags = SA_RESTART;
	sa.sa_mask = mask;

	if (sigaction(sig, &sa, &old) < 0) {
	crm_perror(LOG_ERR, "Could not install signal handler for signal %d", sig);
	return FALSE;
	}

	return TRUE;
	}

	gboolean
	mainloop_add_signal(int sig, void (*dispatch) (int sig))
	{
	GSource *source = NULL;
	int priority = G_PRIORITY_HIGH - 1;

	if (sig == SIGTERM) {
	/* TERM is higher priority than other signals,
	* signals are higher priority than other ipc.
	* Yes, minus: smaller is "higher"
	*/
	priority--;
	}

	if (sig >= NSIG \|\| sig < 0) {
	crm_err("Signal %d is out of range", sig);
	return FALSE;

	} else if (crm_signals[sig] != NULL && crm_signals[sig]->handler == dispatch) {
	crm_trace("Signal handler for %d is already installed", sig);
	return TRUE;

	} else if (crm_signals[sig] != NULL) {
	crm_err("Different signal handler for %d is already installed", sig);
	return FALSE;
	}

	CRM_ASSERT(sizeof(crm_signal_t) > sizeof(GSource));
	source = g_source_new(&crm_signal_funcs, sizeof(crm_signal_t));

	crm_signals[sig] = (crm_signal_t *) mainloop_setup_trigger(source, priority, NULL, NULL);
	CRM_ASSERT(crm_signals[sig] != NULL);

	crm_signals[sig]->handler = dispatch;
	crm_signals[sig]->signal = sig;

	if (crm_signal(sig, mainloop_signal_handler) == FALSE) {
	crm_signal_t *tmp = crm_signals[sig];

	crm_signals[sig] = NULL;

	mainloop_destroy_trigger((crm_trigger_t *) tmp);
	return FALSE;
	}
	#if 0
	/* If we want signals to interrupt mainloop's poll(), instead of waiting for
	* the timeout, then we should call siginterrupt() below
	*
	* For now, just enforce a low timeout
	*/
	if (siginterrupt(sig, 1) < 0) {
	crm_perror(LOG_INFO, "Could not enable system call interruptions for signal %d", sig);
	}
	#endif

	return TRUE;
	}

	gboolean
	mainloop_destroy_signal(int sig)
	{
	crm_signal_t *tmp = NULL;

	if (sig >= NSIG \|\| sig < 0) {
	crm_err("Signal %d is out of range", sig);
	return FALSE;

	} else if (crm_signal(sig, NULL) == FALSE) {
	crm_perror(LOG_ERR, "Could not uninstall signal handler for signal %d", sig);
	return FALSE;

	} else if (crm_signals[sig] == NULL) {
	return TRUE;
	}

	crm_trace("Destroying signal %d", sig);
	tmp = crm_signals[sig];
	crm_signals[sig] = NULL;
	mainloop_destroy_trigger((crm_trigger_t *) tmp);
	return TRUE;
	}

	static qb_array_t *gio_map = NULL;

	/*
	* libqb...
	*/
	struct gio_to_qb_poll {
	int32_t is_used;
	GIOChannel *channel;
	guint source;
	int32_t events;
	void *data;
	qb_ipcs_dispatch_fn_t fn;
	enum qb_loop_priority p;
	};

	static int
	gio_adapter_refcount(struct gio_to_qb_poll *adaptor)
	{
	/* This is evil
	* Looking at the giochannel header file, ref_count is the first member of channel
	* So cheat...
	*/
	if (adaptor && adaptor->channel) {
	int ref = (void )adaptor->channel;

	return *ref;
	}
	return 0;
	}

	static gboolean
	gio_read_socket(GIOChannel * gio, GIOCondition condition, gpointer data)
	{
	struct gio_to_qb_poll adaptor = (struct gio_to_qb_poll )data;
	gint fd = g_io_channel_unix_get_fd(gio);

	crm_trace("%p.%d %d (ref=%d)", data, fd, condition, gio_adapter_refcount(adaptor));

	if (condition & G_IO_NVAL) {
	crm_trace("Marking failed adaptor %p unused", adaptor);
	adaptor->is_used = QB_FALSE;
	}

	return (adaptor->fn(fd, condition, adaptor->data) == 0);
	}

	static void
	gio_poll_destroy(gpointer data)
	{
	/* adaptor->source is valid but about to be destroyed (ref_count == 0) in gmain.c
	* adaptor->channel will still have ref_count > 0... should be == 1
	*/
	struct gio_to_qb_poll adaptor = (struct gio_to_qb_poll )data;

	crm_trace("Destroying adaptor %p channel %p (ref=%d)", adaptor, adaptor->channel,
	gio_adapter_refcount(adaptor));
	adaptor->is_used = QB_FALSE;
	adaptor->channel = NULL;
	adaptor->source = 0;
	}

	static int32_t
	gio_poll_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t evts,
	void *data, qb_ipcs_dispatch_fn_t fn)
	{
	struct gio_to_qb_poll *adaptor;
	GIOChannel *channel;
	int32_t res = 0;

	res = qb_array_index(gio_map, fd, (void **)&adaptor);
	if (res < 0) {
	crm_err("Array lookup failed for fd=%d: %d", fd, res);
	return res;
	}

	crm_trace("Adding fd=%d to mainloop as adapater %p", fd, adaptor);
	if (adaptor->is_used) {
	crm_err("Adapter for descriptor %d is still in-use", fd);
	return -EEXIST;
	}

	/* channel is created with ref_count = 1 */
	channel = g_io_channel_unix_new(fd);
	if (!channel) {
	crm_err("No memory left to add fd=%d", fd);
	return -ENOMEM;
	}

	/* Because unlike the poll() API, glib doesn't tell us about HUPs by default */
	evts \|= (G_IO_HUP \| G_IO_NVAL \| G_IO_ERR);

	adaptor->channel = channel;
	adaptor->fn = fn;
	adaptor->events = evts;
	adaptor->data = data;
	adaptor->p = p;
	adaptor->is_used = QB_TRUE;
	adaptor->source =
	g_io_add_watch_full(channel, G_PRIORITY_DEFAULT, evts, gio_read_socket, adaptor,
	gio_poll_destroy);

	/* Now that mainloop now holds a reference to adaptor->channel,
	* thanks to g_io_add_watch_full(), drop ours from g_io_channel_unix_new().
	*
	* This means that adaptor->channel will be free'd by:
	* g_main_context_dispatch()
	* -> g_source_destroy_internal()
	* -> g_source_callback_unref()
	* shortly after gio_poll_destroy() completes
	*/
	g_io_channel_unref(adaptor->channel);

	crm_trace("Added to mainloop with gsource id=%d, ref=%d", adaptor->source,
	gio_adapter_refcount(adaptor));
	if (adaptor->source > 0) {
	return 0;
	}

	return -EINVAL;
	}

	static int32_t
	gio_poll_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t evts,
	void *data, qb_ipcs_dispatch_fn_t fn)
	{
	return 0;
	}

	static int32_t
	gio_poll_dispatch_del(int32_t fd)
	{
	struct gio_to_qb_poll *adaptor;

	crm_trace("Looking for fd=%d", fd);
	if (qb_array_index(gio_map, fd, (void **)&adaptor) == 0) {
	crm_trace("Marking adaptor %p unused (ref=%d)", adaptor, gio_adapter_refcount(adaptor));
	adaptor->is_used = QB_FALSE;
	}
	return 0;
	}

	struct qb_ipcs_poll_handlers gio_poll_funcs = {
	.job_add = NULL,
	.dispatch_add = gio_poll_dispatch_add,
	.dispatch_mod = gio_poll_dispatch_mod,
	.dispatch_del = gio_poll_dispatch_del,
	};

	static enum qb_ipc_type
	pick_ipc_type(enum qb_ipc_type requested)
	{
	const char *env = getenv("PCMK_ipc_type");

	if (env && strcmp("shared-mem", env) == 0) {
	return QB_IPC_SHM;
	} else if (env && strcmp("socket", env) == 0) {
	return QB_IPC_SOCKET;
	} else if (env && strcmp("posix", env) == 0) {
	return QB_IPC_POSIX_MQ;
	} else if (env && strcmp("sysv", env) == 0) {
	return QB_IPC_SYSV_MQ;
	} else if (requested == QB_IPC_NATIVE) {
	/* We prefer shared memory because the server never blocks on
	* send. If part of a message fits into the socket, libqb
	* needs to block until the remainder can be sent also.
	* Otherwise the client will wait forever for the remaining
	* bytes.
	*/
	return QB_IPC_SHM;
	}
	return requested;
	}

	qb_ipcs_service_t *
	mainloop_add_ipc_server(const char *name, enum qb_ipc_type type,
	struct qb_ipcs_service_handlers * callbacks)
	{
	int rc = 0;
	qb_ipcs_service_t *server = NULL;

	if (gio_map == NULL) {
	gio_map = qb_array_create_2(64, sizeof(struct gio_to_qb_poll), 1);
	}

	crm_client_init();
	server = qb_ipcs_create(name, 0, pick_ipc_type(type), callbacks);
	qb_ipcs_poll_handlers_set(server, &gio_poll_funcs);

	rc = qb_ipcs_run(server);
	if (rc < 0) {
	crm_err("Could not start %s IPC server: %s (%d)", name, pcmk_strerror(rc), rc);
	return NULL;
	}

	return server;
	}

	void
	mainloop_del_ipc_server(qb_ipcs_service_t * server)
	{
	if (server) {
	qb_ipcs_destroy(server);
	}
	}

	struct mainloop_io_s {
	char *name;
	void *userdata;

	guint source;
	crm_ipc_t *ipc;
	GIOChannel *channel;

	int (dispatch_fn_ipc) (const char buffer, ssize_t length, gpointer userdata);
	int (*dispatch_fn_io) (gpointer userdata);
	void (*destroy_fn) (gpointer userdata);

	};

	static int
	mainloop_gio_refcount(mainloop_io_t * client)
	{
	/* This is evil
	* Looking at the giochannel header file, ref_count is the first member of channel
	* So cheat...
	*/
	if (client && client->channel) {
	int ref = (void )client->channel;

	return *ref;
	}
	return 0;
	}

	static gboolean
	mainloop_gio_callback(GIOChannel * gio, GIOCondition condition, gpointer data)
	{
	gboolean keep = TRUE;
	mainloop_io_t *client = data;

	if (condition & G_IO_IN) {
	if (client->ipc) {
	long rc = 0;
	int max = 10;

	do {
	rc = crm_ipc_read(client->ipc);
	if (rc <= 0) {
	crm_trace("Message acquisition from %s[%p] failed: %s (%ld)",
	client->name, client, pcmk_strerror(rc), rc);

	} else if (client->dispatch_fn_ipc) {
	const char *buffer = crm_ipc_buffer(client->ipc);

	crm_trace("New message from %s[%p] = %d", client->name, client, rc, condition);
	if (client->dispatch_fn_ipc(buffer, rc, client->userdata) < 0) {
	crm_trace("Connection to %s no longer required", client->name);
	keep = FALSE;
	}
	}

	} while (keep && rc > 0 && --max > 0);

	} else {
	crm_trace("New message from %s[%p]", client->name, client);
	if (client->dispatch_fn_io) {
	if (client->dispatch_fn_io(client->userdata) < 0) {
	crm_trace("Connection to %s no longer required", client->name);
	keep = FALSE;
	}
	}
	}
	}

	if (client->ipc && crm_ipc_connected(client->ipc) == FALSE) {
	crm_err("Connection to %s[%p] closed (I/O condition=%d)", client->name, client, condition);
	keep = FALSE;

	} else if (condition & (G_IO_HUP \| G_IO_NVAL \| G_IO_ERR)) {
	crm_trace("The connection %s[%p] has been closed (I/O condition=%d, refcount=%d)",
	client->name, client, condition, mainloop_gio_refcount(client));
	keep = FALSE;

	} else if ((condition & G_IO_IN) == 0) {
	/*
	#define GLIB_SYSDEF_POLLIN =1
	#define GLIB_SYSDEF_POLLPRI =2
	#define GLIB_SYSDEF_POLLOUT =4
	#define GLIB_SYSDEF_POLLERR =8
	#define GLIB_SYSDEF_POLLHUP =16
	#define GLIB_SYSDEF_POLLNVAL =32

	typedef enum
	{
	G_IO_IN GLIB_SYSDEF_POLLIN,
	G_IO_OUT GLIB_SYSDEF_POLLOUT,
	G_IO_PRI GLIB_SYSDEF_POLLPRI,
	G_IO_ERR GLIB_SYSDEF_POLLERR,
	G_IO_HUP GLIB_SYSDEF_POLLHUP,
	G_IO_NVAL GLIB_SYSDEF_POLLNVAL
	} GIOCondition;

	A bitwise combination representing a condition to watch for on an event source.

	G_IO_IN There is data to read.
	G_IO_OUT Data can be written (without blocking).
	G_IO_PRI There is urgent data to read.
	G_IO_ERR Error condition.
	G_IO_HUP Hung up (the connection has been broken, usually for pipes and sockets).
	G_IO_NVAL Invalid request. The file descriptor is not open.
	*/
	crm_err("Strange condition: %d", condition);
	}

	/* keep == FALSE results in mainloop_gio_destroy() being called
	* just before the source is removed from mainloop
	*/
	return keep;
	}

	static void
	mainloop_gio_destroy(gpointer c)
	{
	mainloop_io_t *client = c;

	/* client->source is valid but about to be destroyed (ref_count == 0) in gmain.c
	* client->channel will still have ref_count > 0... should be == 1
	*/
	crm_trace("Destroying client %s[%p] %d", client->name, c, mainloop_gio_refcount(client));

	if (client->ipc) {
	crm_ipc_close(client->ipc);
	}

	if (client->destroy_fn) {
	client->destroy_fn(client->userdata);
	}

	if (client->ipc) {
	crm_ipc_destroy(client->ipc);
	}

	crm_trace("Destroyed client %s[%p] %d", client->name, c, mainloop_gio_refcount(client));
	free(client->name);

	memset(client, 0, sizeof(mainloop_io_t)); /* A bit of pointless paranoia */
	free(client);
	}

	mainloop_io_t *
	mainloop_add_ipc_client(const char name, int priority, size_t max_size, void userdata,
	struct ipc_client_callbacks *callbacks)
	{
	mainloop_io_t *client = NULL;
	crm_ipc_t *conn = crm_ipc_new(name, max_size);

	if (conn && crm_ipc_connect(conn)) {
	int32_t fd = crm_ipc_get_fd(conn);

	client = mainloop_add_fd(name, priority, fd, userdata, NULL);
	client->ipc = conn;
	client->destroy_fn = callbacks->destroy;
	client->dispatch_fn_ipc = callbacks->dispatch;
	}

	if (conn && client == NULL) {
	crm_trace("Connection to %s failed", name);
	crm_ipc_close(conn);
	crm_ipc_destroy(conn);
	}

	return client;
	}

	void
	mainloop_del_ipc_client(mainloop_io_t * client)
	{
	mainloop_del_fd(client);
	}

	crm_ipc_t *
	mainloop_get_ipc_client(mainloop_io_t * client)
	{
	if (client) {
	return client->ipc;
	}
	return NULL;
	}

	mainloop_io_t *
	mainloop_add_fd(const char name, int priority, int fd, void userdata,
	struct mainloop_fd_callbacks * callbacks)
	{
	mainloop_io_t *client = NULL;

	if (fd > 0) {
	client = calloc(1, sizeof(mainloop_io_t));
	client->name = strdup(name);
	client->userdata = userdata;

	if (callbacks) {
	client->destroy_fn = callbacks->destroy;
	client->dispatch_fn_io = callbacks->dispatch;
	}

	client->channel = g_io_channel_unix_new(fd);
	client->source =
	g_io_add_watch_full(client->channel, priority,
	(G_IO_IN \| G_IO_HUP \| G_IO_NVAL \| G_IO_ERR), mainloop_gio_callback,
	client, mainloop_gio_destroy);

	/* Now that mainloop now holds a reference to adaptor->channel,
	* thanks to g_io_add_watch_full(), drop ours from g_io_channel_unix_new().
	*
	* This means that adaptor->channel will be free'd by:
	* g_main_context_dispatch() or g_source_remove()
	* -> g_source_destroy_internal()
	* -> g_source_callback_unref()
	* shortly after mainloop_gio_destroy() completes
	*/
	g_io_channel_unref(client->channel);
	crm_trace("Added connection %d for %s[%p].%d %d", client->source, client->name, client, fd,
	mainloop_gio_refcount(client));
	}

	return client;
	}

	void
	mainloop_del_fd(mainloop_io_t * client)
	{
	if (client != NULL) {
	crm_trace("Removing client %s[%p] %d", client->name, client, mainloop_gio_refcount(client));
	if (client->source) {
	/* Results in mainloop_gio_destroy() being called just
	* before the source is removed from mainloop
	*/
	g_source_remove(client->source);
	}
	}
	}

	pid_t
	mainloop_child_pid(mainloop_child_t * child)
	{
	return child->pid;
	}

	const char *
	mainloop_child_name(mainloop_child_t * child)
	{
	return child->desc;
	}

	int
	mainloop_child_timeout(mainloop_child_t * child)
	{
	return child->timeout;
	}

	void *
	mainloop_child_userdata(mainloop_child_t * child)
	{
	return child->privatedata;
	}

	void
	mainloop_clear_child_userdata(mainloop_child_t * child)
	{
	child->privatedata = NULL;
	}

	static gboolean
	child_timeout_callback(gpointer p)
	{
	mainloop_child_t *child = p;

	child->timerid = 0;
	if (child->timeout) {
	crm_crit("%s process (PID %d) will not die!", child->desc, (int)child->pid);
	return FALSE;
	}

	child->timeout = TRUE;
	crm_warn("%s process (PID %d) timed out", child->desc, (int)child->pid);

	if (kill(child->pid, SIGKILL) < 0) {
	if (errno == ESRCH) {
	/* Nothing left to do */
	return FALSE;
	}
	crm_perror(LOG_ERR, "kill(%d, KILL) failed", child->pid);
	}

	child->timerid = g_timeout_add(5000, child_timeout_callback, child);
	return FALSE;
	}

	static GListPtr child_list = NULL;

	static void
	child_death_dispatch(int signal)
	{
	GListPtr iter = child_list;

	while(iter) {
	int rc = 0;
	int core = 0;
	int signo = 0;
	int status = 0;
	int exitcode = 0;

	GListPtr saved = NULL;
	mainloop_child_t *child = iter->data;

	rc = waitpid(child->pid, &status, WNOHANG);
	if(rc == 0) {
	iter = iter->next;
	continue;

	} else if(rc != child->pid) {
	signo = signal;
	exitcode = 1;
	status = 1;
	crm_perror(LOG_ERR, "Call to waitpid(%d) failed", child->pid);

	} else {
	crm_trace("Managed process %d exited: %p", child->pid, child);

	if (WIFEXITED(status)) {
	exitcode = WEXITSTATUS(status);
	crm_trace("Managed process %d (%s) exited with rc=%d", child->pid, child->desc, exitcode);

	} else if (WIFSIGNALED(status)) {
	signo = WTERMSIG(status);
	crm_trace("Managed process %d (%s) exited with signal=%d", child->pid, child->desc, signo);
	}
	#ifdef WCOREDUMP
	if (WCOREDUMP(status)) {
	core = 1;
	crm_err("Managed process %d (%s) dumped core", child->pid, child->desc);
	}
	#endif
	}

	if (child->callback) {
	child->callback(child, child->pid, core, signo, exitcode);
	}

	crm_trace("Removing process entry %p for %d", child, child->pid);

	saved = iter;
	iter = iter->next;

	child_list = g_list_remove_link(child_list, saved);
	g_list_free(saved);

	if (child->timerid != 0) {
	crm_trace("Removing timer %d", child->timerid);
	g_source_remove(child->timerid);
	child->timerid = 0;
	}
	free(child->desc);
	free(child);
	}
	}

	/* Create/Log a new tracked process
	* To track a process group, use -pid
	*/
	void
	mainloop_child_add(pid_t pid, int timeout, const char desc, void privatedata,
	void (callback) (mainloop_child_t p, pid_t pid, int core, int signo, int exitcode))
	{
	static bool need_init = TRUE;
	mainloop_child_t *child = g_new(mainloop_child_t, 1);

	child->pid = pid;
	child->timerid = 0;
	child->timeout = FALSE;
	child->privatedata = privatedata;
	child->callback = callback;

	if(desc) {
	child->desc = strdup(desc);
	}

	if (timeout) {
	child->timerid = g_timeout_add(timeout, child_timeout_callback, child);
	}

	child_list = g_list_append(child_list, child);

	if(need_init) {
	need_init = FALSE;

	/* Do NOT use g_child_watch_add() and friends, they rely on pthreads */
	mainloop_add_signal(SIGCHLD, child_death_dispatch);

	/* In case they terminated before the signal handler was installed */
	child_death_dispatch(SIGCHLD);
	}
	}
	diff --git a/valgrind-pcmk.suppressions b/valgrind-pcmk.suppressions
	index 811c2ea0cc..1dccbf52a5 100644
	--- a/valgrind-pcmk.suppressions
	+++ b/valgrind-pcmk.suppressions
	@@ -1,153 +1,263 @@
	# Valgrind suppressions for PE testing
	{
	Valgrind bug
	Memcheck:Addr8
	fun:__strspn_sse42
	fun:crm_get_msec
	}

	{
	Ignore crm_system_name
	Memcheck:Leak
	fun:malloc
	fun:crm_strdup_fn
	fun:crm_log_init_worker
	fun:crm_log_init
	fun:main
	}

	{
	libqb fixed upstream 1
	Memcheck:Leak
	fun:realloc
	fun:_grow_bin_array
	fun:_qb_array_grow
	}

	{
	libqb fixed upstream 2
	Memcheck:Leak
	fun:calloc
	fun:qb_log_dcs_get
	fun:_qb_log_callsite_get
	}

	{
	Cman - Points to uninitialized bytes
	Memcheck:Param
	socketcall.sendmsg(msg)
	fun:__sendmsg_nocancel
	obj:*/libcman.so.3.0
	obj:*/libcman.so.3.0
	}

	{
	Cman - Jump or move depends on uninitialized values
	Memcheck:Cond
	obj:*/libcman.so.3.0
	obj:*/libcman.so.3.0
	}

	{
	quarks - hashtable
	Memcheck:Leak
	fun:calloc
	fun:g_malloc0
	obj:/libglib-
	fun:g_slice_alloc
	fun:g_hash_table_new_full
	fun:g_quark_from_static_string
	}

	+{
	+ quarks - hashtable 2
	+ Memcheck:Leak
	+ fun:malloc
	+ fun:g_malloc
	+ fun:g_slice_alloc
	+ fun:g_hash_table_new_full
	+ fun:g_quark_from_static_string
	+}
	+
	+{
	+ quarks - hashtable 3
	+ Memcheck:Leak
	+ fun:calloc
	+ fun:g_malloc0
	+ fun:g_hash_table_new_full
	+ fun:g_quark_from_static_string
	+}
	+
	+{
	+ quarks - hashtable 4
	+ Memcheck:Leak
	+ fun:malloc
	+ fun:realloc
	+ fun:g_realloc
	+ fun:g_quark_from_static_string
	+}
	+
	+{
	+ g_mainloop - create
	+ Memcheck:Leak
	+ fun:calloc
	+ fun:g_malloc0
	+ fun:g_main_loop_new
	+ fun:crmd_init
	+}
	+
	+{
	+ g_mainloop - run 1
	+ Memcheck:Leak
	+ fun:malloc
	+ fun:g_malloc
	+ fun:g_slice_alloc
	+ fun:g_slice_alloc0
	+ obj:/libglib-
	+ fun:g_main_context_dispatch
	+ obj:/libglib-
	+ fun:g_main_loop_run
	+}
	+
	+{
	+ g_mainloop - run 2
	+ Memcheck:Leak
	+ fun:malloc
	+ fun:realloc
	+ fun:g_realloc
	+ obj:/libglib-
	+ fun:g_array_set_size
	+ fun:g_static_private_set
	+ obj:/libglib-
	+ fun:g_main_context_dispatch
	+ obj:/libglib-
	+ fun:g_main_loop_run
	+}
	+
	+{
	+ g_mainloop - run 3
	+ Memcheck:Leak
	+ fun:malloc
	+ fun:g_malloc
	+ fun:g_slice_alloc
	+ fun:g_array_sized_new
	+ fun:g_static_private_set
	+ obj:/libglib-
	+ fun:g_main_context_dispatch
	+ obj:/libglib-
	+ fun:g_main_loop_run
	+}
	+
	+{
	+ g_mainloop - run 4
	+ Memcheck:Leak
	+ fun:calloc
	+ fun:g_malloc0
	+ fun:g_thread_self
	+ fun:g_main_loop_run
	+}
	+
	+{
	+ g_mainloop - run 5
	+ Memcheck:Leak
	+ fun:malloc
	+ fun:g_malloc
	+ obj:/libglib-
	+ fun:g_main_loop_run
	+}
	+
	+{
	+ g_mainloop - run 5
	+ Memcheck:Leak
	+ fun:malloc
	+ fun:realloc
	+ fun:g_realloc
	+ obj:/libglib-
	+ fun:g_ptr_array_add
	+ fun:g_main_context_check
	+ obj:/libglib-
	+ fun:g_main_loop_run
	+}
	+
	{
	glib types
	Memcheck:Leak
	fun:malloc
	fun:realloc
	fun:g_realloc
	obj:/libgobject-
	fun:g_type_register_static
	}

	{
	glib types 2
	Memcheck:Leak
	fun:realloc
	fun:g_realloc
	obj:/libgobject-
	fun:g_type_register_static
	fun:g_param_type_register_static
	}

	{
	glib types 3
	Memcheck:Leak
	fun:calloc
	fun:g_malloc0
	obj:/libgobject-
	fun:g_type_register_fundamental
	}

	{
	glib types 4
	Memcheck:Leak
	fun:calloc
	fun:g_malloc0
	obj:/libgobject-
	obj:/libgobject-
	fun:g_type_register_fundamental
	}

	{
	glib - the return
	Memcheck:Leak
	fun:calloc
	fun:g_malloc0
	obj:/libgobject-
	obj:/libgobject-
	fun:_dl_init
	}

	{
	glib - seriously?
	Memcheck:Leak
	fun:calloc
	fun:g_malloc0
	obj:/libgobject-
	obj:/libgobject-
	obj:/libgobject-
	fun:_dl_init
	}

	{
	glib - this isnt funny anymore
	Memcheck:Leak
	fun:malloc
	fun:realloc
	fun:g_realloc
	obj:/libgobject-
	fun:g_type_register_fundamental
	}

	{
	glib - why do you hate me?
	Memcheck:Leak
	fun:calloc
	fun:g_malloc0
	obj:/libgobject-
	obj:/libgobject-
	fun:call_init.part.0
	fun:_dl_init
	}

	{
	dear glib - you suck at memory management
	Memcheck:Leak
	fun:calloc
	fun:g_malloc0
	obj:/libgobject-
	obj:/libgobject-
	obj:/libgobject-
	fun:call_init.part.0
	fun:_dl_init
	}
	\ No newline at end of file

File Metadata

Mime Type: text/x-diff
Expires: Mon, Apr 21, 6:14 PM (1 d, 4 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1662693
Default Alt Text: (103 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions