Adding upstream version 3.1.8.upstream/3.1.8upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1489 insertions, 0 deletions

diff --git a/lib/sam.c b/lib/sam.c
new file mode 100644
index 0000000..94dbf2a
--- /dev/null
+++ b/lib/sam.c
@@ -0,0 +1,1489 @@
+/*
+ * Copyright (c) 2009-2011 Red Hat, Inc.
+ *
+ * All rights reserved.
+ *
+ * Author: Jan Friesse (jfriesse@redhat.com)
+ *
+ * This software licensed under BSD license, the text of which follows:
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ *   this list of conditions and the following disclaimer.
+ * - Redistributions in binary form must reproduce the above copyright notice,
+ *   this list of conditions and the following disclaimer in the documentation
+ *   and/or other materials provided with the distribution.
+ * - Neither the name of the Red Hat, Inc. nor the names of its
+ *   contributors may be used to endorse or promote products derived from this
+ *   software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
+ * THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+/*
+ * Provides a SAM API
+ */
+
+#include <config.h>
+
+#include <limits.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <errno.h>
+#include <poll.h>
+
+#include <corosync/corotypes.h>
+#include <qb/qbipcc.h>
+#include <corosync/corodefs.h>
+#include <corosync/cmap.h>
+#include <corosync/hdb.h>
+#include <corosync/quorum.h>
+
+#include <corosync/sam.h>
+
+#include "util.h"
+
+#include <stdio.h>
+#include <sys/wait.h>
+#include <signal.h>
+
+#define SAM_CMAP_S_FAILED		"failed"
+#define SAM_CMAP_S_REGISTERED		"stopped"
+#define SAM_CMAP_S_STARTED		"running"
+#define SAM_CMAP_S_Q_WAIT		"waiting for quorum"
+
+#define SAM_RP_MASK_Q(pol)	(pol & (~SAM_RECOVERY_POLICY_QUORUM))
+#define SAM_RP_MASK_C(pol)	(pol & (~SAM_RECOVERY_POLICY_CMAP))
+#define SAM_RP_MASK(pol)	(pol & (~(SAM_RECOVERY_POLICY_QUORUM | SAM_RECOVERY_POLICY_CMAP)))
+
+enum sam_internal_status_t {
+	SAM_INTERNAL_STATUS_NOT_INITIALIZED = 0,
+	SAM_INTERNAL_STATUS_INITIALIZED,
+	SAM_INTERNAL_STATUS_REGISTERED,
+	SAM_INTERNAL_STATUS_STARTED,
+	SAM_INTERNAL_STATUS_FINALIZED
+};
+
+enum sam_command_t {
+	SAM_COMMAND_START,
+	SAM_COMMAND_STOP,
+	SAM_COMMAND_HB,
+	SAM_COMMAND_DATA_STORE,
+	SAM_COMMAND_WARN_SIGNAL_SET,
+	SAM_COMMAND_MARK_FAILED,
+};
+
+enum sam_reply_t {
+	SAM_REPLY_OK,
+	SAM_REPLY_ERROR,
+};
+
+enum sam_parent_action_t {
+	SAM_PARENT_ACTION_ERROR,
+	SAM_PARENT_ACTION_RECOVERY,
+	SAM_PARENT_ACTION_QUIT,
+	SAM_PARENT_ACTION_CONTINUE
+};
+
+enum sam_cmap_key_t {
+	SAM_CMAP_KEY_RECOVERY,
+	SAM_CMAP_KEY_HC_PERIOD,
+	SAM_CMAP_KEY_LAST_HC,
+	SAM_CMAP_KEY_STATE,
+};
+
+static struct {
+	int time_interval;
+	sam_recovery_policy_t recovery_policy;
+	enum sam_internal_status_t internal_status;
+	unsigned int instance_id;
+	int child_fd_out;
+	int child_fd_in;
+	int term_send;
+	int warn_signal;
+	int am_i_child;
+
+	sam_hc_callback_t hc_callback;
+	pthread_t cb_thread;
+	int cb_rpipe_fd, cb_wpipe_fd;
+	int cb_registered;
+
+	void *user_data;
+	size_t user_data_size;
+	size_t user_data_allocated;
+
+	pthread_mutex_t lock;
+
+	quorum_handle_t quorum_handle;
+	uint32_t quorate;
+	int quorum_fd;
+
+	cmap_handle_t cmap_handle;
+	char cmap_pid_path[CMAP_KEYNAME_MAXLEN];
+} sam_internal_data;
+
+extern const char *__progname;
+
+static cs_error_t sam_cmap_update_key (enum sam_cmap_key_t key, const char *value)
+{
+	cs_error_t err;
+	const char *svalue;
+	uint64_t hc_period, last_hc;
+
+	const char *ssvalue[] = { [SAM_RECOVERY_POLICY_QUIT] = "quit", [SAM_RECOVERY_POLICY_RESTART] = "restart" };
+	char key_name[CMAP_KEYNAME_MAXLEN];
+
+	switch (key) {
+	case SAM_CMAP_KEY_RECOVERY:
+		svalue = ssvalue[SAM_RP_MASK (sam_internal_data.recovery_policy)];
+
+		if (snprintf(key_name, CMAP_KEYNAME_MAXLEN, "%s%s", sam_internal_data.cmap_pid_path,
+		    "recovery") >= CMAP_KEYNAME_MAXLEN) {
+
+			err = CS_ERR_NAME_TOO_LONG;
+			goto exit_error;
+		}
+
+		if ((err = cmap_set_string(sam_internal_data.cmap_handle, key_name, svalue)) != CS_OK) {
+			goto exit_error;
+		}
+		break;
+	case SAM_CMAP_KEY_HC_PERIOD:
+		hc_period = sam_internal_data.time_interval;
+
+		if (snprintf(key_name, CMAP_KEYNAME_MAXLEN, "%s%s", sam_internal_data.cmap_pid_path,
+		    "poll_period") >= CMAP_KEYNAME_MAXLEN) {
+
+			err = CS_ERR_NAME_TOO_LONG;
+			goto exit_error;
+		}
+
+		if ((err = cmap_set_uint64(sam_internal_data.cmap_handle, key_name, hc_period)) != CS_OK) {
+			goto exit_error;
+		}
+		break;
+	case SAM_CMAP_KEY_LAST_HC:
+		last_hc = cs_timestamp_get();
+
+		if (snprintf(key_name, CMAP_KEYNAME_MAXLEN, "%s%s", sam_internal_data.cmap_pid_path,
+		    "last_updated") >= CMAP_KEYNAME_MAXLEN) {
+
+			err = CS_ERR_NAME_TOO_LONG;
+			goto exit_error;
+		}
+		if ((err = cmap_set_uint64(sam_internal_data.cmap_handle, key_name, last_hc)) != CS_OK) {
+			goto exit_error;
+		}
+		break;
+	case SAM_CMAP_KEY_STATE:
+		svalue = value;
+		if (snprintf(key_name, CMAP_KEYNAME_MAXLEN, "%s%s", sam_internal_data.cmap_pid_path,
+		    "state") >= CMAP_KEYNAME_MAXLEN) {
+
+			err = CS_ERR_NAME_TOO_LONG;
+			goto exit_error;
+		}
+
+		if ((err = cmap_set_string(sam_internal_data.cmap_handle, key_name, svalue)) != CS_OK) {
+			goto exit_error;
+		}
+		break;
+	}
+
+	return (CS_OK);
+
+exit_error:
+	return (err);
+}
+
+static cs_error_t sam_cmap_destroy_pid_path (void)
+{
+	cmap_iter_handle_t iter;
+	cs_error_t err;
+	char key_name[CMAP_KEYNAME_MAXLEN + 1];
+
+	err = cmap_iter_init(sam_internal_data.cmap_handle, sam_internal_data.cmap_pid_path, &iter);
+	if (err != CS_OK) {
+		goto error_exit;
+	}
+
+	while ((err = cmap_iter_next(sam_internal_data.cmap_handle, iter, key_name, NULL, NULL)) == CS_OK) {
+		cmap_delete(sam_internal_data.cmap_handle, key_name);
+	}
+
+	err = cmap_iter_finalize(sam_internal_data.cmap_handle, iter);
+
+error_exit:
+	return (err);
+}
+
+static cs_error_t sam_cmap_register (void)
+{
+	cs_error_t err;
+	cmap_handle_t cmap_handle;
+
+	if ((err = cmap_initialize (&cmap_handle)) != CS_OK) {
+		return (err);
+	}
+
+	snprintf(sam_internal_data.cmap_pid_path, CMAP_KEYNAME_MAXLEN, "resources.process.%d.", getpid());
+
+	sam_internal_data.cmap_handle = cmap_handle;
+
+	if ((err = sam_cmap_update_key (SAM_CMAP_KEY_RECOVERY, NULL)) != CS_OK) {
+		goto destroy_finalize_error;
+	}
+
+	if ((err = sam_cmap_update_key (SAM_CMAP_KEY_HC_PERIOD, NULL)) != CS_OK) {
+		goto destroy_finalize_error;
+	}
+
+	return (CS_OK);
+
+destroy_finalize_error:
+	sam_cmap_destroy_pid_path ();
+	cmap_finalize (cmap_handle);
+	return (err);
+}
+
+static void quorum_notification_fn (
+        quorum_handle_t handle,
+        uint32_t quorate,
+        uint64_t ring_id,
+        uint32_t view_list_entries,
+        uint32_t *view_list)
+{
+	sam_internal_data.quorate = quorate;
+}
+
+cs_error_t sam_initialize (
+	int time_interval,
+	sam_recovery_policy_t recovery_policy)
+{
+	quorum_callbacks_t quorum_callbacks;
+	uint32_t quorum_type;
+	cs_error_t err;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_NOT_INITIALIZED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	if (SAM_RP_MASK (recovery_policy) != SAM_RECOVERY_POLICY_QUIT &&
+	    SAM_RP_MASK (recovery_policy) != SAM_RECOVERY_POLICY_RESTART) {
+		return (CS_ERR_INVALID_PARAM);
+	}
+
+	if (recovery_policy & SAM_RECOVERY_POLICY_QUORUM) {
+		/*
+		 * Initialize quorum
+		 */
+		quorum_callbacks.quorum_notify_fn = quorum_notification_fn;
+		if ((err = quorum_initialize (&sam_internal_data.quorum_handle, &quorum_callbacks, &quorum_type)) != CS_OK) {
+			goto exit_error;
+		}
+
+		if ((err = quorum_trackstart (sam_internal_data.quorum_handle, CS_TRACK_CHANGES)) != CS_OK) {
+			goto exit_error_quorum;
+		}
+
+		if ((err = quorum_fd_get (sam_internal_data.quorum_handle, &sam_internal_data.quorum_fd)) != CS_OK) {
+			goto exit_error_quorum;
+		}
+
+		/*
+		 * Dispatch initial quorate state
+		 */
+		if ((err = quorum_dispatch (sam_internal_data.quorum_handle, CS_DISPATCH_ONE)) != CS_OK) {
+			goto exit_error_quorum;
+		}
+	}
+	sam_internal_data.recovery_policy = recovery_policy;
+
+	sam_internal_data.time_interval = time_interval;
+
+	sam_internal_data.internal_status = SAM_INTERNAL_STATUS_INITIALIZED;
+
+	sam_internal_data.warn_signal = SIGTERM;
+
+	sam_internal_data.am_i_child = 0;
+
+	sam_internal_data.user_data = NULL;
+	sam_internal_data.user_data_size = 0;
+	sam_internal_data.user_data_allocated = 0;
+
+	pthread_mutex_init (&sam_internal_data.lock, NULL);
+
+	return (CS_OK);
+
+exit_error_quorum:
+	quorum_finalize (sam_internal_data.quorum_handle);
+exit_error:
+	return (err);
+}
+
+/*
+ * Wrapper on top of write(2) function. It handles EAGAIN and EINTR states and sends whole buffer if possible.
+ */
+static size_t sam_safe_write (
+	int d,
+	const void *buf,
+	size_t nbyte)
+{
+	ssize_t bytes_write;
+	ssize_t tmp_bytes_write;
+
+	bytes_write = 0;
+
+	do {
+		tmp_bytes_write = write (d, (const char *)buf + bytes_write,
+			(nbyte - bytes_write > SSIZE_MAX) ? SSIZE_MAX : nbyte - bytes_write);
+
+		if (tmp_bytes_write == -1) {
+			if (!(errno == EAGAIN || errno == EINTR))
+				return -1;
+		} else {
+			bytes_write += tmp_bytes_write;
+		}
+	} while (bytes_write != nbyte);
+
+	return (bytes_write);
+}
+
+/*
+ * Wrapper on top of read(2) function. It handles EAGAIN and EINTR states and reads whole buffer if possible.
+ */
+static size_t sam_safe_read (
+	int d,
+	void *buf,
+	size_t nbyte)
+{
+	ssize_t bytes_read;
+	ssize_t tmp_bytes_read;
+
+	bytes_read = 0;
+
+	do {
+		tmp_bytes_read = read (d, (char *)buf + bytes_read,
+			(nbyte - bytes_read > SSIZE_MAX) ? SSIZE_MAX : nbyte - bytes_read);
+
+		if (tmp_bytes_read == -1) {
+			if (!(errno == EAGAIN || errno == EINTR))
+				return -1;
+		} else {
+			bytes_read += tmp_bytes_read;
+		}
+
+	} while (bytes_read != nbyte && tmp_bytes_read != 0);
+
+	return (bytes_read);
+}
+
+static cs_error_t sam_read_reply (
+	int child_fd_in)
+{
+	char reply;
+	cs_error_t err;
+
+	if (sam_safe_read (sam_internal_data.child_fd_in, &reply, sizeof (reply)) != sizeof (reply)) {
+		return (CS_ERR_LIBRARY);
+	}
+
+	switch (reply) {
+	case SAM_REPLY_ERROR:
+		/*
+		 * Read error and return that
+		 */
+		if (sam_safe_read (sam_internal_data.child_fd_in, &err, sizeof (err)) != sizeof (err)) {
+			return (CS_ERR_LIBRARY);
+		}
+
+		return (err);
+		break;
+	case SAM_REPLY_OK:
+		/*
+		 * Everything correct
+		 */
+		break;
+	default:
+		return (CS_ERR_LIBRARY);
+		break;
+	}
+
+	return (CS_OK);
+}
+
+cs_error_t sam_data_getsize (size_t *size)
+{
+	if (size == NULL) {
+		return (CS_ERR_INVALID_PARAM);
+	}
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_INITIALIZED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED) {
+
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	pthread_mutex_lock (&sam_internal_data.lock);
+
+	*size = sam_internal_data.user_data_size;
+
+	pthread_mutex_unlock (&sam_internal_data.lock);
+
+	return (CS_OK);
+}
+
+cs_error_t sam_data_restore (
+	void *data,
+	size_t size)
+{
+	cs_error_t err;
+
+	err = CS_OK;
+
+	if (data == NULL) {
+		return (CS_ERR_INVALID_PARAM);
+	}
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_INITIALIZED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED) {
+
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	pthread_mutex_lock (&sam_internal_data.lock);
+
+	if (sam_internal_data.user_data_size == 0) {
+		err = CS_OK;
+
+		goto error_unlock;
+	}
+
+	if (size < sam_internal_data.user_data_size) {
+		err = CS_ERR_INVALID_PARAM;
+
+		goto error_unlock;
+	}
+
+	memcpy (data, sam_internal_data.user_data, sam_internal_data.user_data_size);
+
+	pthread_mutex_unlock (&sam_internal_data.lock);
+
+	return (CS_OK);
+
+error_unlock:
+	pthread_mutex_unlock (&sam_internal_data.lock);
+
+	return (err);
+}
+
+cs_error_t sam_data_store (
+	const void *data,
+	size_t size)
+{
+	cs_error_t err;
+	char command;
+	char *new_data;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_INITIALIZED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED) {
+
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+
+	if (data == NULL) {
+		size = 0;
+	}
+
+	pthread_mutex_lock (&sam_internal_data.lock);
+
+	if (sam_internal_data.am_i_child) {
+		/*
+		 * We are child so we must send data to parent
+		 */
+		command = SAM_COMMAND_DATA_STORE;
+		if (sam_safe_write (sam_internal_data.child_fd_out, &command, sizeof (command)) != sizeof (command)) {
+			err = CS_ERR_LIBRARY;
+
+			goto error_unlock;
+		}
+
+		if (sam_safe_write (sam_internal_data.child_fd_out, &size, sizeof (size)) != sizeof (size)) {
+			err = CS_ERR_LIBRARY;
+
+			goto error_unlock;
+		}
+
+		if (data != NULL && sam_safe_write (sam_internal_data.child_fd_out, data, size) != size) {
+			err = CS_ERR_LIBRARY;
+
+			goto error_unlock;
+		}
+
+		/*
+		 * And wait for reply
+		 */
+		if ((err = sam_read_reply (sam_internal_data.child_fd_in)) != CS_OK) {
+			goto error_unlock;
+		}
+	}
+
+	/*
+	 * We are parent or we received OK reply from parent -> do required action
+	 */
+	if (data == NULL) {
+		free (sam_internal_data.user_data);
+		sam_internal_data.user_data = NULL;
+		sam_internal_data.user_data_allocated = 0;
+		sam_internal_data.user_data_size = 0;
+	} else {
+		if (sam_internal_data.user_data_allocated < size) {
+			if ((new_data = realloc (sam_internal_data.user_data, size)) == NULL) {
+				err = CS_ERR_NO_MEMORY;
+
+				goto error_unlock;
+			}
+
+			sam_internal_data.user_data_allocated = size;
+		} else {
+			new_data = sam_internal_data.user_data;
+		}
+		sam_internal_data.user_data = new_data;
+		sam_internal_data.user_data_size = size;
+
+		memcpy (sam_internal_data.user_data, data, size);
+	}
+
+	pthread_mutex_unlock (&sam_internal_data.lock);
+
+	return (CS_OK);
+
+error_unlock:
+	pthread_mutex_unlock (&sam_internal_data.lock);
+
+	return (err);
+}
+
+cs_error_t sam_start (void)
+{
+	char command;
+	cs_error_t err;
+	sam_recovery_policy_t recpol;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	recpol = sam_internal_data.recovery_policy;
+
+	if (recpol & SAM_RECOVERY_POLICY_QUORUM || recpol & SAM_RECOVERY_POLICY_CMAP) {
+		pthread_mutex_lock (&sam_internal_data.lock);
+	}
+
+	command = SAM_COMMAND_START;
+
+	if (sam_safe_write (sam_internal_data.child_fd_out, &command, sizeof (command)) != sizeof (command)) {
+		if (recpol & SAM_RECOVERY_POLICY_QUORUM || recpol & SAM_RECOVERY_POLICY_CMAP) {
+			pthread_mutex_unlock (&sam_internal_data.lock);
+		}
+
+		return (CS_ERR_LIBRARY);
+	}
+
+	if (recpol & SAM_RECOVERY_POLICY_QUORUM || recpol & SAM_RECOVERY_POLICY_CMAP) {
+		/*
+		 * Wait for parent reply
+		 */
+		if ((err = sam_read_reply (sam_internal_data.child_fd_in)) != CS_OK) {
+			pthread_mutex_unlock (&sam_internal_data.lock);
+
+			return (err);
+		}
+
+		pthread_mutex_unlock (&sam_internal_data.lock);
+	}
+
+	if (sam_internal_data.hc_callback)
+		if (sam_safe_write (sam_internal_data.cb_wpipe_fd, &command, sizeof (command)) != sizeof (command))
+			return (CS_ERR_LIBRARY);
+
+	sam_internal_data.internal_status = SAM_INTERNAL_STATUS_STARTED;
+
+	return (CS_OK);
+}
+
+cs_error_t sam_stop (void)
+{
+	char command;
+	cs_error_t err;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	command = SAM_COMMAND_STOP;
+
+	if (sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_CMAP) {
+		pthread_mutex_lock (&sam_internal_data.lock);
+	}
+
+	if (sam_safe_write (sam_internal_data.child_fd_out, &command, sizeof (command)) != sizeof (command)) {
+		if (sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_CMAP) {
+			pthread_mutex_unlock (&sam_internal_data.lock);
+		}
+
+		return (CS_ERR_LIBRARY);
+	}
+
+	if (sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_CMAP) {
+		/*
+		 * Wait for parent reply
+		 */
+		if ((err = sam_read_reply (sam_internal_data.child_fd_in)) != CS_OK) {
+			pthread_mutex_unlock (&sam_internal_data.lock);
+
+			return (err);
+		}
+
+		pthread_mutex_unlock (&sam_internal_data.lock);
+	}
+
+	if (sam_internal_data.hc_callback)
+		if (sam_safe_write (sam_internal_data.cb_wpipe_fd, &command, sizeof (command)) != sizeof (command))
+			return (CS_ERR_LIBRARY);
+
+	sam_internal_data.internal_status = SAM_INTERNAL_STATUS_REGISTERED;
+
+	return (CS_OK);
+}
+
+cs_error_t sam_hc_send (void)
+{
+	char command;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	command = SAM_COMMAND_HB;
+
+	if (sam_safe_write (sam_internal_data.child_fd_out, &command, sizeof (command)) != sizeof (command))
+		return (CS_ERR_LIBRARY);
+
+	return (CS_OK);
+}
+
+cs_error_t sam_finalize (void)
+{
+	cs_error_t error;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_INITIALIZED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	if (sam_internal_data.internal_status == SAM_INTERNAL_STATUS_STARTED) {
+		error = sam_stop ();
+		if (error != CS_OK)
+			goto exit_error;
+	}
+
+	sam_internal_data.internal_status = SAM_INTERNAL_STATUS_FINALIZED;
+
+	free (sam_internal_data.user_data);
+
+exit_error:
+	return (CS_OK);
+}
+
+cs_error_t sam_mark_failed (void)
+{
+	char command;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED &&
+	    sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	if (!(sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_CMAP)) {
+		return (CS_ERR_INVALID_PARAM);
+	}
+
+	command = SAM_COMMAND_MARK_FAILED;
+
+	if (sam_safe_write (sam_internal_data.child_fd_out, &command, sizeof (command)) != sizeof (command))
+		return (CS_ERR_LIBRARY);
+
+	return (CS_OK);
+}
+
+cs_error_t sam_warn_signal_set (int warn_signal)
+{
+	char command;
+	cs_error_t err;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_INITIALIZED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED &&
+		sam_internal_data.internal_status != SAM_INTERNAL_STATUS_STARTED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	pthread_mutex_lock (&sam_internal_data.lock);
+
+	if (sam_internal_data.am_i_child) {
+		/*
+		 * We are child so we must send data to parent
+		 */
+		command = SAM_COMMAND_WARN_SIGNAL_SET;
+		if (sam_safe_write (sam_internal_data.child_fd_out, &command, sizeof (command)) != sizeof (command)) {
+			err = CS_ERR_LIBRARY;
+
+			goto error_unlock;
+		}
+
+		if (sam_safe_write (sam_internal_data.child_fd_out, &warn_signal, sizeof (warn_signal)) !=
+		   sizeof (warn_signal)) {
+			err = CS_ERR_LIBRARY;
+
+			goto error_unlock;
+		}
+
+		/*
+		 * And wait for reply
+		 */
+		if ((err = sam_read_reply (sam_internal_data.child_fd_in)) != CS_OK) {
+			goto error_unlock;
+		}
+	}
+
+	/*
+	 * We are parent or we received OK reply from parent -> do required action
+	 */
+	sam_internal_data.warn_signal = warn_signal;
+
+	pthread_mutex_unlock (&sam_internal_data.lock);
+
+	return (CS_OK);
+
+error_unlock:
+	pthread_mutex_unlock (&sam_internal_data.lock);
+
+	return (err);
+}
+
+static cs_error_t sam_parent_reply_send (
+	cs_error_t err,
+	int parent_fd_in,
+	int parent_fd_out)
+{
+	char reply;
+
+	if (err == CS_OK) {
+		reply = SAM_REPLY_OK;
+
+		if (sam_safe_write (parent_fd_out, &reply, sizeof (reply)) != sizeof (reply)) {
+			err = CS_ERR_LIBRARY;
+			goto error_reply;
+		}
+
+		return (CS_OK);
+	}
+
+error_reply:
+	reply = SAM_REPLY_ERROR;
+	if (sam_safe_write (parent_fd_out, &reply, sizeof (reply)) != sizeof (reply)) {
+		return (CS_ERR_LIBRARY);
+	}
+	if (sam_safe_write (parent_fd_out, &err, sizeof (err)) != sizeof (err)) {
+		return (CS_ERR_LIBRARY);
+	}
+
+	return (err);
+}
+
+
+static cs_error_t sam_parent_warn_signal_set (
+	int parent_fd_in,
+	int parent_fd_out)
+{
+	int warn_signal;
+	cs_error_t err;
+
+	err = CS_OK;
+
+	if (sam_safe_read (parent_fd_in, &warn_signal, sizeof (warn_signal)) != sizeof (warn_signal)) {
+		err = CS_ERR_LIBRARY;
+		goto error_reply;
+	}
+
+	err = sam_warn_signal_set (warn_signal);
+	if (err != CS_OK) {
+		goto error_reply;
+	}
+
+
+	return (sam_parent_reply_send (CS_OK, parent_fd_in, parent_fd_out));
+
+error_reply:
+	return (sam_parent_reply_send (err, parent_fd_in, parent_fd_out));
+}
+
+static cs_error_t sam_parent_wait_for_quorum (
+	int parent_fd_in,
+	int parent_fd_out)
+{
+	cs_error_t err;
+	struct pollfd pfds[2];
+	int poll_err;
+
+	if (sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_CMAP) {
+		if ((err = sam_cmap_update_key (SAM_CMAP_KEY_STATE, SAM_CMAP_S_Q_WAIT)) != CS_OK) {
+			goto error_reply;
+		}
+	}
+
+	/*
+	 * Update current quorum
+	 */
+	if ((err = quorum_dispatch (sam_internal_data.quorum_handle, CS_DISPATCH_ALL)) != CS_OK) {
+		goto error_reply;
+	}
+
+	/*
+	 * Wait for quorum
+	 */
+	while (!sam_internal_data.quorate) {
+		pfds[0].fd = parent_fd_in;
+		pfds[0].events = 0;
+		pfds[0].revents = 0;
+
+		pfds[1].fd = sam_internal_data.quorum_fd;
+		pfds[1].events = POLLIN;
+		pfds[1].revents = 0;
+
+		poll_err = poll (pfds, 2, -1);
+
+		if (poll_err == -1) {
+			/*
+			 *  Error in poll
+			 *  If it is EINTR, continue, otherwise QUIT
+			 */
+			if (errno != EINTR) {
+				err = CS_ERR_LIBRARY;
+				goto error_reply;
+			}
+		}
+
+		if (pfds[0].revents != 0) {
+			if (pfds[0].revents == POLLERR || pfds[0].revents == POLLHUP ||pfds[0].revents == POLLNVAL) {
+				/*
+				 * Child has exited
+				 */
+				return (CS_OK);
+			}
+		}
+
+		if (pfds[1].revents != 0) {
+			if ((err = quorum_dispatch (sam_internal_data.quorum_handle, CS_DISPATCH_ONE)) != CS_OK) {
+				goto error_reply;
+			}
+		}
+	}
+
+	if (sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_CMAP) {
+		if ((err = sam_cmap_update_key (SAM_CMAP_KEY_STATE, SAM_CMAP_S_STARTED)) != CS_OK) {
+			goto error_reply;
+		}
+	}
+
+	return (sam_parent_reply_send (CS_OK, parent_fd_in, parent_fd_out));
+
+error_reply:
+	if (sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_CMAP) {
+		sam_cmap_update_key (SAM_CMAP_KEY_STATE, SAM_CMAP_S_REGISTERED);
+	}
+
+	return (sam_parent_reply_send (err, parent_fd_in, parent_fd_out));
+}
+
+static cs_error_t sam_parent_cmap_state_set (
+	int parent_fd_in,
+	int parent_fd_out,
+	int state)
+{
+	cs_error_t err;
+	const char *state_s;
+
+	if (state == 1) {
+		state_s = SAM_CMAP_S_STARTED;
+	} else {
+		state_s = SAM_CMAP_S_REGISTERED;
+	}
+
+	if ((err = sam_cmap_update_key (SAM_CMAP_KEY_STATE, state_s)) != CS_OK) {
+		goto error_reply;
+	}
+
+	return (sam_parent_reply_send (CS_OK, parent_fd_in, parent_fd_out));
+
+error_reply:
+	return (sam_parent_reply_send (err, parent_fd_in, parent_fd_out));
+}
+
+static cs_error_t sam_parent_kill_child (
+	int *action,
+	pid_t child_pid)
+{
+	/*
+	 *  Kill child process
+	 */
+	if (!sam_internal_data.term_send) {
+		/*
+		 * We didn't send warn_signal yet.
+		 */
+		kill (child_pid, sam_internal_data.warn_signal);
+
+		sam_internal_data.term_send = 1;
+	} else {
+		/*
+		 * We sent child warning. Now, we will not be so nice
+		 */
+		kill (child_pid, SIGKILL);
+		*action = SAM_PARENT_ACTION_RECOVERY;
+	}
+
+	return (CS_OK);
+}
+
+static cs_error_t sam_parent_mark_child_failed (
+	int *action,
+	pid_t child_pid)
+{
+	sam_recovery_policy_t recpol;
+
+	recpol = sam_internal_data.recovery_policy;
+
+	sam_internal_data.term_send = 1;
+	sam_internal_data.recovery_policy = SAM_RECOVERY_POLICY_QUIT |
+	    (SAM_RP_MASK_C (recpol) ? SAM_RECOVERY_POLICY_CMAP : 0) |
+	    (SAM_RP_MASK_Q (recpol) ? SAM_RECOVERY_POLICY_QUORUM : 0);
+
+	return (sam_parent_kill_child (action, child_pid));
+}
+
+static cs_error_t sam_parent_data_store (
+	int parent_fd_in,
+	int parent_fd_out)
+{
+	char *user_data;
+	ssize_t size;
+	cs_error_t err;
+
+	err = CS_OK;
+	user_data = NULL;
+
+	if (sam_safe_read (parent_fd_in, &size, sizeof (size)) != sizeof (size)) {
+		err = CS_ERR_LIBRARY;
+		goto error_reply;
+	}
+
+	if (size > 0) {
+		user_data = malloc (size);
+		if (user_data == NULL) {
+			err = CS_ERR_NO_MEMORY;
+			goto error_reply;
+		}
+
+		if (sam_safe_read (parent_fd_in, user_data, size) != size) {
+			err = CS_ERR_LIBRARY;
+			goto free_error_reply;
+		}
+	}
+
+	err = sam_data_store (user_data, size);
+	if (err != CS_OK) {
+		goto free_error_reply;
+	}
+
+	free (user_data);
+
+	return (sam_parent_reply_send (CS_OK, parent_fd_in, parent_fd_out));
+
+free_error_reply:
+	free (user_data);
+error_reply:
+	return (sam_parent_reply_send (err, parent_fd_in, parent_fd_out));
+}
+
+static enum sam_parent_action_t sam_parent_handler (
+	int parent_fd_in,
+	int parent_fd_out,
+	pid_t child_pid)
+{
+	int poll_error;
+	int action;
+	int status;
+	ssize_t bytes_read;
+	char command;
+	int time_interval;
+	struct pollfd pfds[2];
+	nfds_t nfds;
+	cs_error_t err;
+	sam_recovery_policy_t recpol;
+
+	status = 0;
+
+	action = SAM_PARENT_ACTION_CONTINUE;
+	recpol = sam_internal_data.recovery_policy;
+
+	while (action == SAM_PARENT_ACTION_CONTINUE) {
+		pfds[0].fd = parent_fd_in;
+		pfds[0].events = POLLIN;
+		pfds[0].revents = 0;
+		nfds = 1;
+
+		if (status == 1 && sam_internal_data.time_interval != 0) {
+			time_interval = sam_internal_data.time_interval;
+		} else {
+			time_interval = -1;
+		}
+
+		if (recpol & SAM_RECOVERY_POLICY_QUORUM) {
+			pfds[nfds].fd = sam_internal_data.quorum_fd;
+			pfds[nfds].events = POLLIN;
+			pfds[nfds].revents = 0;
+			nfds++;
+		}
+
+		poll_error = poll (pfds, nfds, time_interval);
+
+		if (poll_error == -1) {
+			/*
+			 *  Error in poll
+			 *  If it is EINTR, continue, otherwise QUIT
+			 */
+			if (errno != EINTR) {
+				action = SAM_PARENT_ACTION_ERROR;
+			}
+		}
+
+		if (poll_error == 0) {
+			/*
+			 *  Time limit expires
+			 */
+			if (status == 0) {
+				action = SAM_PARENT_ACTION_QUIT;
+			} else {
+				sam_parent_kill_child (&action, child_pid);
+			}
+		}
+
+		if (poll_error > 0) {
+			if (pfds[0].revents != 0) {
+				/*
+				 *  We have EOF or command in pipe
+				 */
+				bytes_read = sam_safe_read (parent_fd_in, &command, 1);
+
+				if (bytes_read == 0) {
+					/*
+					 *  Handle EOF -> Take recovery action or quit if sam_start wasn't called
+					 */
+					if (status == 0)
+						action = SAM_PARENT_ACTION_QUIT;
+					else
+						action = SAM_PARENT_ACTION_RECOVERY;
+
+					continue;
+				}
+
+				if (bytes_read == -1) {
+					action = SAM_PARENT_ACTION_ERROR;
+					goto action_exit;
+				}
+
+				if (recpol & SAM_RECOVERY_POLICY_CMAP) {
+					sam_cmap_update_key (SAM_CMAP_KEY_LAST_HC, NULL);
+				}
+
+				/*
+				 * We have read command
+				 */
+				switch (command) {
+				case SAM_COMMAND_START:
+					if (status == 0) {
+						/*
+						 *  Not started yet
+						 */
+						if (recpol & SAM_RECOVERY_POLICY_QUORUM) {
+							if (sam_parent_wait_for_quorum (parent_fd_in,
+							    parent_fd_out) != CS_OK) {
+								continue;
+							}
+						}
+
+						if (recpol & SAM_RECOVERY_POLICY_CMAP) {
+							if (sam_parent_cmap_state_set (parent_fd_in,
+							    parent_fd_out, 1) != CS_OK) {
+								continue;
+							    }
+						}
+
+						status = 1;
+					}
+					break;
+				case SAM_COMMAND_STOP:
+					if (status == 1) {
+						/*
+						 *  Started
+						 */
+						if (recpol & SAM_RECOVERY_POLICY_CMAP) {
+							if (sam_parent_cmap_state_set (parent_fd_in,
+							    parent_fd_out, 0) != CS_OK) {
+								continue;
+							    }
+						}
+
+						status = 0;
+					}
+					break;
+				case SAM_COMMAND_DATA_STORE:
+					sam_parent_data_store (parent_fd_in, parent_fd_out);
+					break;
+				case SAM_COMMAND_WARN_SIGNAL_SET:
+					sam_parent_warn_signal_set (parent_fd_in, parent_fd_out);
+					break;
+				case SAM_COMMAND_MARK_FAILED:
+					status = 1;
+					sam_parent_mark_child_failed (&action, child_pid);
+					break;
+				}
+			} /* if (pfds[0].revents != 0) */
+
+			if ((sam_internal_data.recovery_policy & SAM_RECOVERY_POLICY_QUORUM) &&
+			    pfds[1].revents != 0) {
+				/*
+				 * Handle quorum change
+				 */
+				err = quorum_dispatch (sam_internal_data.quorum_handle, CS_DISPATCH_ALL);
+
+				if (status == 1 &&
+				    (!sam_internal_data.quorate || (err != CS_ERR_TRY_AGAIN && err != CS_OK))) {
+					sam_parent_kill_child (&action, child_pid);
+				}
+			}
+		} /* select_error > 0 */
+	} /* action == SAM_PARENT_ACTION_CONTINUE */
+
+action_exit:
+	return action;
+}
+
+cs_error_t sam_register (
+	unsigned int *instance_id)
+{
+	cs_error_t error;
+	pid_t pid;
+	int pipe_error;
+	int pipe_fd_out[2], pipe_fd_in[2];
+	enum sam_parent_action_t action, old_action;
+	int child_status;
+	sam_recovery_policy_t recpol;
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_INITIALIZED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	recpol = sam_internal_data.recovery_policy;
+
+	if (recpol & SAM_RECOVERY_POLICY_CMAP) {
+		/*
+		 * Register to cmap
+		 */
+		if ((error = sam_cmap_register ()) != CS_OK) {
+			goto error_exit;
+		}
+	}
+
+	error = CS_OK;
+
+	while (1) {
+		if ((pipe_error = pipe (pipe_fd_out)) != 0) {
+			error = CS_ERR_LIBRARY;
+			goto error_exit;
+		}
+
+		if ((pipe_error = pipe (pipe_fd_in)) != 0) {
+			close (pipe_fd_out[0]);
+			close (pipe_fd_out[1]);
+
+			error = CS_ERR_LIBRARY;
+			goto error_exit;
+		}
+
+		if (recpol & SAM_RECOVERY_POLICY_CMAP) {
+			if ((error = sam_cmap_update_key (SAM_CMAP_KEY_STATE, SAM_CMAP_S_REGISTERED)) != CS_OK) {
+				goto error_exit;
+			}
+		}
+
+		sam_internal_data.instance_id++;
+
+		sam_internal_data.term_send = 0;
+
+		pid = fork ();
+
+		if (pid == -1) {
+			/*
+			 *  Fork error
+			 */
+			sam_internal_data.instance_id--;
+
+			error = CS_ERR_LIBRARY;
+			goto error_exit;
+		}
+
+		if (pid == 0) {
+			/*
+			 *  Child process
+			 */
+			close (pipe_fd_out[0]);
+			close (pipe_fd_in[1]);
+
+			sam_internal_data.child_fd_out = pipe_fd_out[1];
+			sam_internal_data.child_fd_in = pipe_fd_in[0];
+
+			if (instance_id)
+				*instance_id = sam_internal_data.instance_id;
+
+			sam_internal_data.am_i_child = 1;
+			sam_internal_data.internal_status = SAM_INTERNAL_STATUS_REGISTERED;
+
+			pthread_mutex_init (&sam_internal_data.lock, NULL);
+
+			goto error_exit;
+		} else {
+			/*
+			 *  Parent process
+			 */
+			close (pipe_fd_out[1]);
+			close (pipe_fd_in[0]);
+
+			action = sam_parent_handler (pipe_fd_out[0], pipe_fd_in[1], pid);
+
+			close (pipe_fd_out[0]);
+			close (pipe_fd_in[1]);
+
+			if (action == SAM_PARENT_ACTION_ERROR) {
+				error = CS_ERR_LIBRARY;
+				goto error_exit;
+			}
+
+			/*
+			 * We really don't like zombies
+			 */
+			while (waitpid (pid, &child_status, 0) == -1 && errno == EINTR)
+				;
+
+			old_action = action;
+
+			if (action == SAM_PARENT_ACTION_RECOVERY) {
+				if (SAM_RP_MASK (sam_internal_data.recovery_policy) == SAM_RECOVERY_POLICY_QUIT)
+					action = SAM_PARENT_ACTION_QUIT;
+			}
+
+
+			if (action == SAM_PARENT_ACTION_QUIT) {
+				if (recpol & SAM_RECOVERY_POLICY_QUORUM) {
+					quorum_finalize (sam_internal_data.quorum_handle);
+				}
+
+				if (recpol & SAM_RECOVERY_POLICY_CMAP) {
+					if (old_action == SAM_PARENT_ACTION_RECOVERY) {
+						/*
+						 * Mark as failed
+						 */
+						sam_cmap_update_key (SAM_CMAP_KEY_STATE, SAM_CMAP_S_FAILED);
+					} else {
+						sam_cmap_destroy_pid_path ();
+					}
+				}
+
+				exit (WEXITSTATUS (child_status));
+			}
+
+
+		}
+	}
+
+error_exit:
+	return (error);
+}
+
+static void *hc_callback_thread (void *unused_param)
+{
+	int poll_error;
+	int status;
+	ssize_t bytes_readed;
+	char command;
+	int time_interval, tmp_time_interval;
+	int counter;
+	struct pollfd pfds;
+
+	status = 0;
+	counter = 0;
+
+	time_interval = sam_internal_data.time_interval >> 2;
+
+	while (1) {
+		pfds.fd = sam_internal_data.cb_rpipe_fd;
+		pfds.events = POLLIN;
+		pfds.revents = 0;
+
+		if (status == 1) {
+			tmp_time_interval = time_interval;
+		} else {
+			tmp_time_interval = -1;
+		}
+
+		poll_error = poll (&pfds, 1, tmp_time_interval);
+
+		if (poll_error == 0) {
+			if (sam_hc_send () == CS_OK) {
+				counter++;
+			}
+
+			if (counter >= 4) {
+				if (sam_internal_data.hc_callback () != 0) {
+					status = 3;
+				}
+
+				counter = 0;
+			}
+		}
+
+		if (poll_error > 0) {
+			bytes_readed = sam_safe_read (sam_internal_data.cb_rpipe_fd, &command, 1);
+
+			if (bytes_readed > 0) {
+				if (status == 0 && command == SAM_COMMAND_START)
+					status = 1;
+
+				if (status == 1 && command == SAM_COMMAND_STOP)
+					status = 0;
+
+			}
+		}
+	}
+
+	/*
+	 * This makes compiler happy, it's same as return (NULL);
+	 */
+	return (unused_param);
+}
+
+cs_error_t sam_hc_callback_register (sam_hc_callback_t cb)
+{
+	cs_error_t error = CS_OK;
+	pthread_attr_t thread_attr;
+	int pipe_error;
+	int pipe_fd[2];
+
+	if (sam_internal_data.internal_status != SAM_INTERNAL_STATUS_REGISTERED) {
+		return (CS_ERR_BAD_HANDLE);
+	}
+
+	if (sam_internal_data.time_interval == 0) {
+		return (CS_ERR_INVALID_PARAM);
+	}
+
+	if (sam_internal_data.cb_registered) {
+		sam_internal_data.hc_callback = cb;
+
+		return (CS_OK);
+	}
+
+	/*
+	 * We know, this is first registration
+	 */
+
+	if (cb == NULL) {
+		return (CS_ERR_INVALID_PARAM);
+	}
+
+	pipe_error = pipe (pipe_fd);
+
+	if (pipe_error != 0) {
+		/*
+		 *  Pipe creation error
+		 */
+		error = CS_ERR_LIBRARY;
+		goto error_exit;
+	}
+
+	sam_internal_data.cb_rpipe_fd = pipe_fd[0];
+	sam_internal_data.cb_wpipe_fd = pipe_fd[1];
+
+	/*
+	 * Create thread attributes
+	 */
+	error = pthread_attr_init (&thread_attr);
+	if (error != 0) {
+		error = CS_ERR_LIBRARY;
+		goto error_close_fd_exit;
+	}
+
+
+	pthread_attr_setdetachstate (&thread_attr, PTHREAD_CREATE_DETACHED);
+	pthread_attr_setstacksize (&thread_attr, 32768);
+
+	/*
+	 * Create thread
+	 */
+	error = pthread_create (&sam_internal_data.cb_thread, &thread_attr, hc_callback_thread, NULL);
+
+	if (error != 0) {
+		error = CS_ERR_LIBRARY;
+		goto error_attr_destroy_exit;
+	}
+
+	/*
+	 * Cleanup
+	 */
+	pthread_attr_destroy(&thread_attr);
+
+	sam_internal_data.cb_registered = 1;
+	sam_internal_data.hc_callback = cb;
+
+	return (CS_OK);
+
+error_attr_destroy_exit:
+	pthread_attr_destroy(&thread_attr);
+error_close_fd_exit:
+	sam_internal_data.cb_rpipe_fd = sam_internal_data.cb_wpipe_fd = 0;
+	close (pipe_fd[0]);
+	close (pipe_fd[1]);
+error_exit:
+	return (error);
+}