libqb/tests/check_ipc.c

/*
 * Copyright (c) 2010 Red Hat, Inc.
 *
 * All rights reserved.
 *
 * Author: Angus Salkeld <asalkeld@redhat.com>
 *
 * This file is part of libqb.
 *
 * libqb 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.
 *
 * libqb 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 libqb.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "os_base.h"
#include <sys/wait.h>
#include <sys/un.h>
#include <signal.h>
#include <stdbool.h>
#include <fcntl.h>

#ifdef HAVE_GLIB
#include <glib.h>
#endif

#include "check_common.h"

#include <qb/qbdefs.h>
#include <qb/qblog.h>
#include <qb/qbipcc.h>
#include <qb/qbipcs.h>
#include <qb/qbloop.h>

#ifdef HAVE_FAILURE_INJECTION
#include "_failure_injection.h"
#endif

#define NUM_STRESS_CONNECTIONS 5000

static char ipc_name[256];

#define DEFAULT_MAX_MSG_SIZE (8192*16)
#ifndef __clang__
static int CALCULATED_DGRAM_MAX_MSG_SIZE = 0;

#define DGRAM_MAX_MSG_SIZE \
	(CALCULATED_DGRAM_MAX_MSG_SIZE == 0 ? \
	CALCULATED_DGRAM_MAX_MSG_SIZE = qb_ipcc_verify_dgram_max_msg_size(DEFAULT_MAX_MSG_SIZE) : \
	CALCULATED_DGRAM_MAX_MSG_SIZE)

#define MAX_MSG_SIZE (ipc_type == QB_IPC_SOCKET ? DGRAM_MAX_MSG_SIZE : DEFAULT_MAX_MSG_SIZE)

#else
/* because of clang's
   'variable length array in structure' extension will never be supported;
   assign default for SHM as we'll skip test that would use run-time
   established value (via qb_ipcc_verify_dgram_max_msg_size), anyway */
static const int MAX_MSG_SIZE = DEFAULT_MAX_MSG_SIZE;
#endif

/* The size the giant msg's data field needs to be to make
 * this the largests msg we can successfully send. */
#define GIANT_MSG_DATA_SIZE MAX_MSG_SIZE - sizeof(struct qb_ipc_response_header) - 8

static int enforce_server_buffer;
static qb_ipcc_connection_t *conn;
static enum qb_ipc_type ipc_type;
static enum qb_loop_priority global_loop_prio = QB_LOOP_MED;
static bool global_use_glib;
static int global_pipefd[2];

enum my_msg_ids {
	IPC_MSG_REQ_TX_RX,
	IPC_MSG_RES_TX_RX,
	IPC_MSG_REQ_DISPATCH,
	IPC_MSG_RES_DISPATCH,
	IPC_MSG_REQ_BULK_EVENTS,
	IPC_MSG_RES_BULK_EVENTS,
	IPC_MSG_REQ_STRESS_EVENT,
	IPC_MSG_RES_STRESS_EVENT,
	IPC_MSG_REQ_SELF_FEED,
	IPC_MSG_RES_SELF_FEED,
	IPC_MSG_REQ_SERVER_FAIL,
	IPC_MSG_RES_SERVER_FAIL,
	IPC_MSG_REQ_SERVER_DISCONNECT,
	IPC_MSG_RES_SERVER_DISCONNECT,
};


/* these 2 functions from pacemaker code */
static enum qb_ipcs_rate_limit
conv_libqb_prio2ratelimit(enum qb_loop_priority prio)
{
	/* this is an inversion of what libqb's qb_ipcs_request_rate_limit does */
	enum qb_ipcs_rate_limit ret = QB_IPCS_RATE_NORMAL;
	switch (prio) {
	case QB_LOOP_LOW:
		ret = QB_IPCS_RATE_SLOW;
		break;
	case QB_LOOP_HIGH:
		ret = QB_IPCS_RATE_FAST;
		break;
	default:
		qb_log(LOG_DEBUG, "Invalid libqb's loop priority %d,"
		       " assuming QB_LOOP_MED", prio);
		/* fall-through */
	case QB_LOOP_MED:
		break;
	}
	return ret;
}
#ifdef HAVE_GLIB
static gint
conv_prio_libqb2glib(enum qb_loop_priority prio)
{
	gint ret = G_PRIORITY_DEFAULT;
	switch (prio) {
	case QB_LOOP_LOW:
		ret = G_PRIORITY_LOW;
		break;
	case QB_LOOP_HIGH:
		ret = G_PRIORITY_HIGH;
		break;
	default:
		qb_log(LOG_DEBUG, "Invalid libqb's loop priority %d,"
		       " assuming QB_LOOP_MED", prio);
		/* fall-through */
	case QB_LOOP_MED:
		break;
	}
	return ret;
}

/* these 3 glue functions inspired from pacemaker, too */
static gboolean
gio_source_prepare(GSource *source, gint *timeout)
{
	qb_enter();
	*timeout = 500;
	return FALSE;
}
static gboolean
gio_source_check(GSource *source)
{
	qb_enter();
	return TRUE;
}
static gboolean
gio_source_dispatch(GSource *source, GSourceFunc callback, gpointer user_data)
{
	gboolean ret = G_SOURCE_CONTINUE;
	qb_enter();
	if (callback) {
		ret = callback(user_data);
	}
	return ret;
}
static GSourceFuncs gio_source_funcs = {
    .prepare = gio_source_prepare,
    .check = gio_source_check,
    .dispatch = gio_source_dispatch,
};

#endif


/* Test Cases
 *
 * 1) basic send & recv different message sizes
 *
 * 2) send message to start dispatch (confirm receipt)
 *
 * 3) flow control
 *
 * 4) authentication
 *
 * 5) thread safety
 *
 * 6) cleanup
 *
 * 7) service availability
 *
 * 8) multiple services
 *
 * 9) setting perms on the sockets
 */
static qb_loop_t *my_loop;
static qb_ipcs_service_t* s1;
static int32_t turn_on_fc = QB_FALSE;
static int32_t fc_enabled = 89;
static int32_t send_event_on_created = QB_FALSE;
static int32_t disconnect_after_created = QB_FALSE;
static int32_t num_bulk_events = 10;
static int32_t num_stress_events = 30000;
static int32_t reference_count_test = QB_FALSE;
static int32_t multiple_connections = QB_FALSE;
static int32_t set_perms_on_socket = QB_FALSE;


static int32_t
exit_handler(int32_t rsignal, void *data)
{
	qb_log(LOG_DEBUG, "caught signal %d", rsignal);
	qb_ipcs_destroy(s1);
	exit(0);
}

static void
set_ipc_name(const char *prefix)
{
	FILE *f;
	char process_name[256];

	/* The process-unique part of the IPC name has already been decided
	 * and stored in the file IPC_TEST_NAME_FILE
	 */
	f = fopen(IPC_TEST_NAME_FILE, "r");
	if (f) {
		fgets(process_name, sizeof(process_name), f);
		fclose(f);
		snprintf(ipc_name, sizeof(ipc_name), "%.44s%s", prefix, process_name);
	} else {
		/* This is the old code, use only as a fallback */
		static char t_sec[3] = "";
		if (t_sec[0] == '\0') {
			const char *const found = strrchr(__TIME__, ':');
			strncpy(t_sec, found ? found + 1 : "-", sizeof(t_sec) - 1);
			t_sec[sizeof(t_sec) - 1] = '\0';
		}

		snprintf(ipc_name, sizeof(ipc_name), "%.44s%s%lX%.4x", prefix, t_sec,
			 (unsigned long)getpid(), (unsigned) ((long) time(NULL) % (0x10000)));
	}
}

static int
pipe_writer(int fd, int revents, void *data) {
	qb_enter();
	static const char buf[8] = { 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h' };

	ssize_t wbytes = 0, wbytes_sum = 0;

	//for (size_t i = 0; i < SIZE_MAX; i++) {
	for (size_t i = 0; i < 4096; i++) {
		wbytes_sum += wbytes;
		if ((wbytes = write(fd, buf, sizeof(buf))) == -1) {
			if (errno != EAGAIN) {
				perror("write");
				exit(-1);
			}
			break;
		}
	}
	if (wbytes_sum > 0) {
		qb_log(LOG_DEBUG, "written %zd bytes", wbytes_sum);
	}
	qb_leave();
	return 1;
}

static int
pipe_reader(int fd, int revents, void *data) {
	qb_enter();
	ssize_t rbytes, rbytes_sum = 0;
	size_t cnt = SIZE_MAX;
	char buf[4096] = { '\0' };
	while ((rbytes = read(fd, buf, sizeof(buf))) > 0 && rbytes < cnt) {
		cnt -= rbytes;
		rbytes_sum += rbytes;
	}
	if (rbytes_sum > 0) {
		ck_assert(buf[0] != '\0');  /* avoid dead store elimination */
		qb_log(LOG_DEBUG, "read %zd bytes", rbytes_sum);
		sleep(1);
	}
	qb_leave();
	return 1;
}

#if HAVE_GLIB
static gboolean
gio_pipe_reader(void *data) {
	return (pipe_reader(*((int *) data), 0, NULL) > 0);
}
static gboolean
gio_pipe_writer(void *data) {
	return (pipe_writer(*((int *) data), 0, NULL) > 0);
}
#endif

static int32_t
s1_msg_process_fn(qb_ipcs_connection_t *c,
		void *data, size_t size)
{
	struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data;
	struct qb_ipc_response_header response = { 0, };
	ssize_t res;

	if (req_pt->id == IPC_MSG_REQ_TX_RX) {
		response.size = sizeof(struct qb_ipc_response_header);
		response.id = IPC_MSG_RES_TX_RX;
		response.error = 0;
		res = qb_ipcs_response_send(c, &response, response.size);
		if (res < 0) {
			qb_perror(LOG_INFO, "qb_ipcs_response_send");
		} else if (res != response.size) {
			qb_log(LOG_DEBUG, "qb_ipcs_response_send %zd != %d",
			       res, response.size);
		}
		if (turn_on_fc) {
			qb_ipcs_request_rate_limit(s1, QB_IPCS_RATE_OFF);
		}
	} else if (req_pt->id == IPC_MSG_REQ_DISPATCH) {
		response.size = sizeof(struct qb_ipc_response_header);
		response.id = IPC_MSG_RES_DISPATCH;
		response.error = 0;
		res = qb_ipcs_event_send(c, &response,
					 sizeof(response));
		if (res < 0) {
			qb_perror(LOG_INFO, "qb_ipcs_event_send");
		}
	} else if (req_pt->id == IPC_MSG_REQ_BULK_EVENTS) {
		int32_t m;
		int32_t num;
		struct qb_ipcs_connection_stats_2 *stats;
		uint32_t max_size = MAX_MSG_SIZE;

		response.size = sizeof(struct qb_ipc_response_header);
		response.error = 0;

		stats = qb_ipcs_connection_stats_get_2(c, QB_FALSE);
		num = stats->event_q_length;
		free(stats);

		/* crazy large message */
		res = qb_ipcs_event_send(c, &response, max_size*10);
		ck_assert_int_eq(res, -EMSGSIZE);

		/* send one event before responding */
		res = qb_ipcs_event_send(c, &response, sizeof(response));
		ck_assert_int_eq(res, sizeof(response));
		response.id++;

		/* There should be one more item in the event queue now. */
		stats = qb_ipcs_connection_stats_get_2(c, QB_FALSE);
		ck_assert_int_eq(stats->event_q_length - num, 1);
		free(stats);

		/* send response */
		response.id = IPC_MSG_RES_BULK_EVENTS;
		res = qb_ipcs_response_send(c, &response, response.size);
		ck_assert_int_eq(res, sizeof(response));

		/* send the rest of the events after the response */
		for (m = 1; m < num_bulk_events; m++) {
			res = qb_ipcs_event_send(c, &response, sizeof(response));

			if (res == -EAGAIN || res == -ENOBUFS) {
				/* retry */
				usleep(1000);
				m--;
				continue;
			}
			ck_assert_int_eq(res, sizeof(response));
			response.id++;
		}

	} else if (req_pt->id == IPC_MSG_REQ_STRESS_EVENT) {
		struct {
			struct qb_ipc_response_header hdr __attribute__ ((aligned(8)));
			char data[GIANT_MSG_DATA_SIZE] __attribute__ ((aligned(8)));
			uint32_t sent_msgs __attribute__ ((aligned(8)));
		} __attribute__ ((aligned(8))) giant_event_send;
		int32_t m;

		response.size = sizeof(struct qb_ipc_response_header);
		response.error = 0;

		response.id = IPC_MSG_RES_STRESS_EVENT;
		res = qb_ipcs_response_send(c, &response, response.size);
		ck_assert_int_eq(res, sizeof(response));

		giant_event_send.hdr.error = 0;
		giant_event_send.hdr.id = IPC_MSG_RES_STRESS_EVENT;
		for (m = 0; m < num_stress_events; m++) {
			size_t sent_len = sizeof(struct qb_ipc_response_header);

			if (((m+1) % 1000) == 0) {
				sent_len = sizeof(giant_event_send);
				giant_event_send.sent_msgs = m + 1;
			}
			giant_event_send.hdr.size = sent_len;

			res = qb_ipcs_event_send(c, &giant_event_send, sent_len);
			if (res < 0) {
				if (res == -EAGAIN || res == -ENOBUFS) {
					/* yield to the receive process */
					usleep(1000);
					m--;
					continue;
				} else {
					qb_perror(LOG_DEBUG, "sending stress events");
					ck_assert_int_eq(res, sent_len);
				}
			} else if (((m+1) % 1000) == 0) {
				qb_log(LOG_DEBUG, "SENT: %d stress events sent", m+1);
			}
			giant_event_send.hdr.id++;
		}

	} else if (req_pt->id == IPC_MSG_REQ_SELF_FEED) {
		if (pipe(global_pipefd) != 0) {
			perror("pipefd");
			ck_assert(0);
		}
		fcntl(global_pipefd[0], F_SETFL, O_NONBLOCK);
		fcntl(global_pipefd[1], F_SETFL, O_NONBLOCK);
		if (global_use_glib) {
#ifdef HAVE_GLIB
			GSource *source_r, *source_w;
			source_r = g_source_new(&gio_source_funcs, sizeof(GSource));
			source_w = g_source_new(&gio_source_funcs, sizeof(GSource));
			ck_assert(source_r != NULL && source_w != NULL);
			g_source_set_priority(source_r, conv_prio_libqb2glib(QB_LOOP_HIGH));
			g_source_set_priority(source_w, conv_prio_libqb2glib(QB_LOOP_HIGH));
			g_source_set_can_recurse(source_r, FALSE);
			g_source_set_can_recurse(source_w, FALSE);
			g_source_set_callback(source_r, gio_pipe_reader, &global_pipefd[0], NULL);
			g_source_set_callback(source_w, gio_pipe_writer, &global_pipefd[1], NULL);
			g_source_add_unix_fd(source_r, global_pipefd[0], G_IO_IN);
			g_source_add_unix_fd(source_w, global_pipefd[1], G_IO_OUT);
			g_source_attach(source_r, NULL);
			g_source_attach(source_w, NULL);
#else
			ck_assert(0);
#endif
		} else {
			qb_loop_poll_add(my_loop, QB_LOOP_HIGH, global_pipefd[1],
			                 POLLOUT|POLLERR, NULL, pipe_writer);
			qb_loop_poll_add(my_loop, QB_LOOP_HIGH, global_pipefd[0],
			                 POLLIN|POLLERR, NULL, pipe_reader);
		}

	} else if (req_pt->id == IPC_MSG_REQ_SERVER_FAIL) {
		exit(0);
	} else if (req_pt->id == IPC_MSG_REQ_SERVER_DISCONNECT) {
		multiple_connections = QB_FALSE;
		qb_ipcs_disconnect(c);
	}
	return 0;
}

static int32_t
my_job_add(enum qb_loop_priority p,
			  void *data,
			  qb_loop_job_dispatch_fn fn)
{
	return qb_loop_job_add(my_loop, p, data, fn);
}

static int32_t
my_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
	void *data, qb_ipcs_dispatch_fn_t fn)
{
	return qb_loop_poll_add(my_loop, p, fd, events, data, fn);
}

static int32_t
my_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
	void *data, qb_ipcs_dispatch_fn_t fn)
{
	return qb_loop_poll_mod(my_loop, p, fd, events, data, fn);
}

static int32_t
my_dispatch_del(int32_t fd)
{
	return qb_loop_poll_del(my_loop, fd);
}


/* taken from examples/ipcserver.c, with s/my_g/gio/ */
#ifdef HAVE_GLIB

#include <qb/qbarray.h>

static qb_array_t *gio_map;
static GMainLoop *glib_loop;

struct gio_to_qb_poll {
	int32_t is_used;
	int32_t events;
	int32_t source;
	int32_t fd;
	void *data;
	qb_ipcs_dispatch_fn_t fn;
	enum qb_loop_priority p;
};

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);

	qb_enter();

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

static void
gio_poll_destroy(gpointer data)
{
	struct gio_to_qb_poll *adaptor = (struct gio_to_qb_poll *)data;

	adaptor->is_used--;
	if (adaptor->is_used == 0) {
		qb_log(LOG_DEBUG, "fd %d adaptor destroyed\n", adaptor->fd);
		adaptor->fd = 0;
		adaptor->source = 0;
	}
}

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

	qb_enter();

	res = qb_array_index(gio_map, fd, (void **)&adaptor);
	if (res < 0) {
		return res;
	}
	if (adaptor->is_used && adaptor->source) {
		if (is_new) {
			return -EEXIST;
		}
		g_source_remove(adaptor->source);
		adaptor->source = 0;
	}

	channel = g_io_channel_unix_new(fd);
	if (!channel) {
		return -ENOMEM;
	}

	adaptor->fn = fn;
	adaptor->events = evts;
	adaptor->data = data;
	adaptor->p = p;
	adaptor->is_used++;
	adaptor->fd = fd;

	adaptor->source = g_io_add_watch_full(channel, conv_prio_libqb2glib(p),
	                                      evts, gio_read_socket, adaptor,
	                                      gio_poll_destroy);

	/* we are handing the channel off to be managed by mainloop now.
	 * remove our reference. */
	g_io_channel_unref(channel);

	return 0;
}

static int32_t
gio_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t evts,
                 void *data, qb_ipcs_dispatch_fn_t fn)
{
	return gio_dispatch_update(p, fd, evts, data, fn, TRUE);
}

static int32_t
gio_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t evts,
                 void *data, qb_ipcs_dispatch_fn_t fn)
{
	return gio_dispatch_update(p, fd, evts, data, fn, FALSE);
}

static int32_t
gio_dispatch_del(int32_t fd)
{
	struct gio_to_qb_poll *adaptor;
	if (qb_array_index(gio_map, fd, (void **)&adaptor) == 0) {
		g_source_remove(adaptor->source);
		adaptor->source = 0;
	}
	return 0;
}

#endif /* HAVE_GLIB */


static int32_t
s1_connection_closed(qb_ipcs_connection_t *c)
{
	if (multiple_connections) {
		return 0;
	}
	/* Stop the connection being freed when we call qb_ipcs_disconnect()
	   in the callback */
	if (disconnect_after_created == QB_TRUE) {
		disconnect_after_created = QB_FALSE;
		return 1;
	}

	qb_enter();
	qb_leave();
	return 0;
}

static void
outq_flush (void *data)
{
	static int i = 0;
	struct cs_ipcs_conn_context *cnx;
	cnx = qb_ipcs_context_get(data);

	qb_log(LOG_DEBUG,"iter %u\n", i);
	i++;
	if (i == 2) {
		qb_ipcs_destroy(s1);
		s1 = NULL;
	}
	/* if the reference counting is not working, this should fail
	 * for i > 1.
	 */
	qb_ipcs_event_send(data, "test", 4);
	assert(memcmp(cnx, "test", 4) == 0);
	if (i < 5) {
		qb_loop_job_add(my_loop, QB_LOOP_HIGH, data, outq_flush);
	} else {
		/* this single unref should clean everything up.
		 */
		qb_ipcs_connection_unref(data);
		qb_log(LOG_INFO, "end of test, stopping loop");
		qb_loop_stop(my_loop);
	}
}


static void
s1_connection_destroyed(qb_ipcs_connection_t *c)
{
	if (multiple_connections) {
		return;
	}

	qb_enter();
	if (reference_count_test) {
		struct cs_ipcs_conn_context *cnx;
		cnx = qb_ipcs_context_get(c);
		free(cnx);
	} else {
		qb_loop_stop(my_loop);
	}
	qb_leave();
}

static int32_t
s1_connection_accept(qb_ipcs_connection_t *c, uid_t uid, gid_t gid)
{
	if (set_perms_on_socket) {
		qb_ipcs_connection_auth_set(c, 555, 741, S_IRWXU|S_IRWXG|S_IROTH|S_IWOTH);
	}
	return 0;
}


static void
s1_connection_created(qb_ipcs_connection_t *c)
{
	uint32_t max = MAX_MSG_SIZE;
	if (multiple_connections) {
		return;
	}

	if (send_event_on_created) {
		struct qb_ipc_response_header response;
		int32_t res;

		response.size = sizeof(struct qb_ipc_response_header);
		response.id = IPC_MSG_RES_DISPATCH;
		response.error = 0;
		res = qb_ipcs_event_send(c, &response,
					 sizeof(response));
		ck_assert_int_eq(res, response.size);
	}
	if (reference_count_test) {
		struct cs_ipcs_conn_context *context;

		qb_ipcs_connection_ref(c);
		qb_loop_job_add(my_loop, QB_LOOP_HIGH, c, outq_flush);

		context = calloc(1, 20);
		memcpy(context, "test", 4);
		qb_ipcs_context_set(c, context);
	}


	ck_assert_int_le(max, qb_ipcs_connection_get_buffer_size(c));

}

static volatile sig_atomic_t usr1_bit;

static void usr1_bit_setter(int signal) {
    if (signal == SIGUSR1) {
        usr1_bit = 1;
    }
}

#define READY_SIGNALLER(name, data_arg)  void (name)(void *data_arg)
typedef READY_SIGNALLER(ready_signaller_fn, );

static
READY_SIGNALLER(usr1_signaller, parent_target)
{
	kill(*((pid_t *) parent_target), SIGUSR1);
}

#define NEW_PROCESS_RUNNER(name, ready_signaller_arg, signaller_data_arg, data_arg) \
	void (name)(ready_signaller_fn ready_signaller_arg, \
	      void *signaller_data_arg, void *data_arg)
typedef NEW_PROCESS_RUNNER(new_process_runner_fn, , , );

static
NEW_PROCESS_RUNNER(run_ipc_server, ready_signaller, signaller_data, data)
{
	int32_t res;
	qb_loop_signal_handle handle;

	struct qb_ipcs_service_handlers sh = {
		.connection_accept = s1_connection_accept,
		.connection_created = s1_connection_created,
		.msg_process = s1_msg_process_fn,
		.connection_destroyed = s1_connection_destroyed,
		.connection_closed = s1_connection_closed,
	};

	struct qb_ipcs_poll_handlers ph;
	uint32_t max_size = MAX_MSG_SIZE;

	my_loop = qb_loop_create();
	qb_loop_signal_add(my_loop, QB_LOOP_HIGH, SIGTERM,
	                   NULL, exit_handler, &handle);


	s1 = qb_ipcs_create(ipc_name, 4, ipc_type, &sh);
	ck_assert(s1 != 0);

	if (global_loop_prio != QB_LOOP_MED) {
		qb_ipcs_request_rate_limit(s1,
		                           conv_libqb_prio2ratelimit(global_loop_prio));
	}
	if (global_use_glib) {
#ifdef HAVE_GLIB
		ph = (struct qb_ipcs_poll_handlers) {
			.job_add = NULL,
			.dispatch_add = gio_dispatch_add,
			.dispatch_mod = gio_dispatch_mod,
			.dispatch_del = gio_dispatch_del,
		};
		glib_loop = g_main_loop_new(NULL, FALSE);
		gio_map = qb_array_create_2(16, sizeof(struct gio_to_qb_poll), 1);
		ck_assert(gio_map != NULL);
#else
		ck_assert(0);
#endif
	} else {
		ph = (struct qb_ipcs_poll_handlers) {
			.job_add = my_job_add,
			.dispatch_add = my_dispatch_add,
			.dispatch_mod = my_dispatch_mod,
			.dispatch_del = my_dispatch_del,
		};
	}

	if (enforce_server_buffer) {
		qb_ipcs_enforce_buffer_size(s1, max_size);
	}
	qb_ipcs_poll_handlers_set(s1, &ph);

	res = qb_ipcs_run(s1);
	ck_assert_int_eq(res, 0);

	if (ready_signaller != NULL) {
		ready_signaller(signaller_data);
	}

	if (global_use_glib) {
#ifdef HAVE_GLIB
		g_main_loop_run(glib_loop);
#endif
	} else {
		qb_loop_run(my_loop);
	}
	qb_log(LOG_DEBUG, "loop finished - done ...");
}

static pid_t
run_function_in_new_process(const char *role,
                            new_process_runner_fn new_process_runner,
                            void *data)
{
	char formatbuf[1024];
	pid_t parent_target, pid1, pid2;
	struct sigaction orig_sa, purpose_sa;
	sigset_t orig_mask, purpose_mask, purpose_clear_mask;

	sigemptyset(&purpose_mask);
	sigaddset(&purpose_mask, SIGUSR1);

	sigprocmask(SIG_BLOCK, &purpose_mask, &orig_mask);
	purpose_clear_mask = orig_mask;
	sigdelset(&purpose_clear_mask, SIGUSR1);

	purpose_sa.sa_handler = usr1_bit_setter;
	purpose_sa.sa_mask = purpose_mask;
	purpose_sa.sa_flags = SA_RESTART;

	/* Double-fork so the servers can be reaped in a timely manner */
	parent_target = getpid();
	pid1 = fork();
	if (pid1 == 0) {
		pid2 = fork();
		if (pid2 == -1) {
			fprintf (stderr, "Can't fork twice\n");
			exit(0);
		}
		if (pid2 == 0) {
			sigprocmask(SIG_SETMASK, &orig_mask, NULL);

			if (role == NULL) {
				qb_log_format_set(QB_LOG_STDERR, "lib/%f|%l[%P] %b");
			} else {
				snprintf(formatbuf, sizeof(formatbuf),
				         "lib/%%f|%%l|%s[%%P] %%b", role);
				qb_log_format_set(QB_LOG_STDERR, formatbuf);
			}

			new_process_runner(usr1_signaller, &parent_target, data);
			exit(0);
		} else {
			waitpid(pid2, NULL, 0);
			exit(0);
		}
	}

	usr1_bit = 0;
	/* XXX assume never fails */
	sigaction(SIGUSR1, &purpose_sa, &orig_sa);

	do {
		/* XXX assume never fails with EFAULT */
		sigsuspend(&purpose_clear_mask);
	} while (usr1_bit != 1);
	usr1_bit = 0;
	sigprocmask(SIG_SETMASK, &orig_mask, NULL);
	/* give children a slight/non-strict scheduling advantage */
	sched_yield();

	return pid1;
}

static void
request_server_exit(void)
{
	struct qb_ipc_request_header req_header;
	struct qb_ipc_response_header res_header;
	struct iovec iov[1];
	int32_t res;

	/*
	 * tell the server to exit
	 */
	req_header.id = IPC_MSG_REQ_SERVER_FAIL;
	req_header.size = sizeof(struct qb_ipc_request_header);

	iov[0].iov_len = req_header.size;
	iov[0].iov_base = &req_header;

	ck_assert_int_eq(QB_TRUE, qb_ipcc_is_connected(conn));

	res = qb_ipcc_sendv_recv(conn, iov, 1,
				 &res_header,
				 sizeof(struct qb_ipc_response_header), -1);
	/*
	 * confirm we get -ENOTCONN or ECONNRESET
	 */
	if (res != -ECONNRESET && res != -ENOTCONN) {
		qb_log(LOG_ERR, "id:%d size:%d", res_header.id, res_header.size);
		ck_assert_int_eq(res, -ENOTCONN);
	}
}

static void
kill_server(pid_t pid)
{
	kill(pid, SIGTERM);
	waitpid(pid, NULL, 0);
}

static int32_t
verify_graceful_stop(pid_t pid)
{
	int wait_rc = 0;
	int status = 0;
	int rc = 0;
	int tries;

	/* We need the server to be able to exit by itself */
	for (tries = 10;  tries >= 0; tries--) {
		sleep(1);
		wait_rc = waitpid(pid, &status, WNOHANG);
		if (wait_rc > 0) {
			break;
		}
	}

	ck_assert_int_eq(wait_rc, pid);
	rc = WIFEXITED(status);
	if (rc) {
		rc = WEXITSTATUS(status);
		ck_assert_int_eq(rc, 0);
	} else {
		ck_assert(rc != 0);
	}

	return 0;
}

struct my_req {
	struct qb_ipc_request_header hdr;
	char message[1024 * 1024];
};

static struct my_req request;
static int32_t
send_and_check(int32_t req_id, uint32_t size,
	       int32_t ms_timeout, int32_t expect_perfection)
{
	struct qb_ipc_response_header res_header;
	int32_t res;
	int32_t try_times = 0;
	uint32_t max_size = MAX_MSG_SIZE;

	request.hdr.id = req_id;
	request.hdr.size = sizeof(struct qb_ipc_request_header) + size;

	/* check that we can't send a message that is too big
	 * and we get the right return code.
	 */
	res = qb_ipcc_send(conn, &request, max_size*2);
	ck_assert_int_eq(res, -EMSGSIZE);

repeat_send:
	res = qb_ipcc_send(conn, &request, request.hdr.size);
	try_times++;
	if (res < 0) {
		if (res == -EAGAIN && try_times < 10) {
			goto repeat_send;
		} else {
			if (res == -EAGAIN && try_times >= 10) {
				fc_enabled = QB_TRUE;
			}
			errno = -res;
			qb_perror(LOG_INFO, "qb_ipcc_send");
			return res;
		}
	}

	if (req_id == IPC_MSG_REQ_DISPATCH) {
		res = qb_ipcc_event_recv(conn, &res_header,
					 sizeof(struct qb_ipc_response_header),
					 ms_timeout);
	} else {
		res = qb_ipcc_recv(conn, &res_header,
				   sizeof(struct qb_ipc_response_header),
				   ms_timeout);
	}
	if (res == -EINTR) {
		return -1;
	}
	if (res == -EAGAIN || res == -ETIMEDOUT) {
		fc_enabled = QB_TRUE;
		qb_perror(LOG_DEBUG, "qb_ipcc_recv");
		return res;
	}
	if (expect_perfection) {
		ck_assert_int_eq(res, sizeof(struct qb_ipc_response_header));
		ck_assert_int_eq(res_header.id, req_id + 1);
		ck_assert_int_eq(res_header.size, sizeof(struct qb_ipc_response_header));
	}
	return res;
}


static int32_t
process_async_connect(int32_t fd, int32_t revents, void *data)
{
	qb_loop_t *cl = (qb_loop_t *)data;
	int res;

	res = qb_ipcc_connect_continue(conn);
	ck_assert_int_eq(res, 0);
	qb_loop_stop(cl);
	return 0;
}
static void test_ipc_connect_async(void)
{
	struct qb_ipc_request_header req_header;
	struct qb_ipc_response_header res_header;
	int32_t res;
	pid_t pid;
	uint32_t max_size = MAX_MSG_SIZE;
	int connect_fd;
	struct iovec iov[1];
	static qb_loop_t *cl;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	conn = qb_ipcc_connect_async(ipc_name, max_size, &connect_fd);
	ck_assert(conn != NULL);

	cl = qb_loop_create();
	res = qb_loop_poll_add(cl, QB_LOOP_MED,
			 connect_fd, POLLIN,
			 cl, process_async_connect);
	ck_assert_int_eq(res, 0);
	qb_loop_run(cl);

	/* Send some data */
	req_header.id = IPC_MSG_REQ_TX_RX;
	req_header.size = sizeof(struct qb_ipc_request_header);

	iov[0].iov_len = req_header.size;
	iov[0].iov_base = &req_header;

	res = qb_ipcc_sendv_recv(conn, iov, 1,
				 &res_header,
				 sizeof(struct qb_ipc_response_header), 5000);

	ck_assert_int_ge(res, 0);

	request_server_exit();
	verify_graceful_stop(pid);


	qb_ipcc_disconnect(conn);
}

static void
test_ipc_txrx_timeout(void)
{
	struct qb_ipc_request_header req_header;
	struct qb_ipc_response_header res_header;
	struct iovec iov[1];
	int32_t res;
	int32_t c = 0;
	int32_t j = 0;
	pid_t pid;
	uint32_t max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	/* The dispatch response will only come over
	 * the event channel, we want to verify the receive times
	 * out when an event is returned with no response */
	req_header.id = IPC_MSG_REQ_DISPATCH;
	req_header.size = sizeof(struct qb_ipc_request_header);

	iov[0].iov_len = req_header.size;
	iov[0].iov_base = &req_header;

	res = qb_ipcc_sendv_recv(conn, iov, 1,
				 &res_header,
				 sizeof(struct qb_ipc_response_header), 5000);

	ck_assert_int_eq(res, -ETIMEDOUT);

	request_server_exit();
	verify_graceful_stop(pid);

	/*
	 * this needs to free up the shared mem
	 */
	qb_ipcc_disconnect(conn);
}

static int32_t recv_timeout = -1;
static void
test_ipc_txrx(void)
{
	int32_t j;
	int32_t c = 0;
	size_t size;
	pid_t pid;
	uint32_t max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	size = QB_MIN(sizeof(struct qb_ipc_request_header), 64);
	for (j = 1; j < 19; j++) {
		size *= 2;
		if (size >= max_size)
			break;
		if (send_and_check(IPC_MSG_REQ_TX_RX, size,
				   recv_timeout, QB_TRUE) < 0) {
			break;
		}
	}
	if (turn_on_fc) {
		/* can't signal server to shutdown if flow control is on */
		ck_assert_int_eq(fc_enabled, QB_TRUE);
		qb_ipcc_disconnect(conn);
		/* TODO - figure out why this sleep is necessary */
		sleep(1);
		kill_server(pid);
	} else {
		request_server_exit();
		qb_ipcc_disconnect(conn);
		verify_graceful_stop(pid);
	}
}

static void
test_ipc_getauth(void)
{
	int32_t j;
	int32_t c = 0;
	pid_t pid;
	pid_t spid;
	uid_t suid;
	gid_t sgid;
	int res;
	uint32_t max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	res = qb_ipcc_auth_get(NULL, NULL, NULL, NULL);
	ck_assert(res == -EINVAL);

	res = qb_ipcc_auth_get(conn, &spid, &suid, &sgid);
	ck_assert(res == 0);
#ifndef HAVE_GETPEEREID
	/* GETPEEREID doesn't return a PID */
	ck_assert(spid != 0);
#endif
	ck_assert(suid == getuid());
	ck_assert(sgid == getgid());

	request_server_exit();
	qb_ipcc_disconnect(conn);
	verify_graceful_stop(pid);
}

static void
test_ipc_exit(void)
{
	struct qb_ipc_request_header req_header;
	struct qb_ipc_response_header res_header;
	struct iovec iov[1];
	int32_t res;
	int32_t c = 0;
	int32_t j = 0;
	pid_t pid;
	uint32_t max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	req_header.id = IPC_MSG_REQ_TX_RX;
	req_header.size = sizeof(struct qb_ipc_request_header);

	iov[0].iov_len = req_header.size;
	iov[0].iov_base = &req_header;

	res = qb_ipcc_sendv_recv(conn, iov, 1,
				 &res_header,
				 sizeof(struct qb_ipc_response_header), -1);
	ck_assert_int_eq(res, sizeof(struct qb_ipc_response_header));

	request_server_exit();
	verify_graceful_stop(pid);

	/*
	 * this needs to free up the shared mem
	 */
	qb_ipcc_disconnect(conn);
}

START_TEST(test_ipc_exit_us)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	recv_timeout = 5000;
	test_ipc_exit();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_exit_shm)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	recv_timeout = 1000;
	test_ipc_exit();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_txrx_shm_timeout)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_txrx_timeout();
	qb_leave();
}
END_TEST


START_TEST(test_ipc_txrx_us_timeout)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_txrx_timeout();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_shm_connect_async)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_connect_async();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_us_connect_async)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_connect_async();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_txrx_shm_getauth)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_getauth();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_txrx_us_getauth)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_getauth();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_txrx_shm_tmo)
{
	qb_enter();
	turn_on_fc = QB_FALSE;
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	recv_timeout = 1000;
	test_ipc_txrx();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_txrx_shm_block)
{
	qb_enter();
	turn_on_fc = QB_FALSE;
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	recv_timeout = -1;
	test_ipc_txrx();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_fc_shm)
{
	qb_enter();
	turn_on_fc = QB_TRUE;
	ipc_type = QB_IPC_SHM;
	recv_timeout = 500;
	set_ipc_name(__func__);
	test_ipc_txrx();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_txrx_us_block)
{
	qb_enter();
	turn_on_fc = QB_FALSE;
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	recv_timeout = -1;
	test_ipc_txrx();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_txrx_us_tmo)
{
	qb_enter();
	turn_on_fc = QB_FALSE;
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	recv_timeout = 1000;
	test_ipc_txrx();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_fc_us)
{
	qb_enter();
	turn_on_fc = QB_TRUE;
	ipc_type = QB_IPC_SOCKET;
	recv_timeout = 500;
	set_ipc_name(__func__);
	test_ipc_txrx();
	qb_leave();
}
END_TEST

struct my_res {
	struct qb_ipc_response_header hdr;
	char message[1024 * 1024];
};

struct dispatch_data {
	pid_t server_pid;
	enum my_msg_ids msg_type;
	uint32_t repetitions;
};

static inline
NEW_PROCESS_RUNNER(client_dispatch, ready_signaller, signaller_data, data)
{
	uint32_t max_size = MAX_MSG_SIZE;
	int32_t size;
	int32_t c = 0;
	int32_t j;
	pid_t server_pid = ((struct dispatch_data *) data)->server_pid;
	enum my_msg_ids msg_type = ((struct dispatch_data *) data)->msg_type;

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(server_pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	if (ready_signaller != NULL) {
		ready_signaller(signaller_data);
	}

	size = QB_MIN(sizeof(struct qb_ipc_request_header), 64);

	for (uint32_t r = ((struct dispatch_data *) data)->repetitions;
			r > 0; r--) {
		for (j = 1; j < 19; j++) {
			size *= 2;
			if (size >= max_size)
				break;
			if (send_and_check(msg_type, size,
					   recv_timeout, QB_TRUE) < 0) {
				break;
			}
		}
	}
}

static void
test_ipc_dispatch(void)
{
	pid_t pid;
	struct dispatch_data data;

	pid = run_function_in_new_process(NULL, run_ipc_server, NULL);
	ck_assert(pid != -1);
	data = (struct dispatch_data){.server_pid = pid,
	                              .msg_type = IPC_MSG_REQ_DISPATCH,
	                              .repetitions = 1};

	client_dispatch(NULL, NULL, (void *) &data);

	request_server_exit();
	qb_ipcc_disconnect(conn);
	verify_graceful_stop(pid);
}

START_TEST(test_ipc_dispatch_us)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_dispatch();
	qb_leave();
}
END_TEST

static int32_t events_received;

static int32_t
count_stress_events(int32_t fd, int32_t revents, void *data)
{
	struct {
		struct qb_ipc_response_header hdr __attribute__ ((aligned(8)));
		char data[GIANT_MSG_DATA_SIZE] __attribute__ ((aligned(8)));
		uint32_t sent_msgs __attribute__ ((aligned(8)));
	} __attribute__ ((aligned(8))) giant_event_recv;
	qb_loop_t *cl = (qb_loop_t*)data;
	int32_t res;

	res = qb_ipcc_event_recv(conn, &giant_event_recv,
				 sizeof(giant_event_recv),
				 -1);
	if (res > 0) {
		events_received++;

		if ((events_received % 1000) == 0) {
			qb_log(LOG_DEBUG, "RECV: %d stress events processed", events_received);
			if (res != sizeof(giant_event_recv)) {
				qb_log(LOG_DEBUG, "Unexpected recv size, expected %d got %d",
				       sizeof(giant_event_recv), res);

				ck_assert_int_eq(res, sizeof(giant_event_recv));
			} else if (giant_event_recv.sent_msgs != events_received) {
				qb_log(LOG_DEBUG, "Server event mismatch. Server thinks we got %d msgs, but we only received %d",
					giant_event_recv.sent_msgs, events_received);
				/* This indicates that data corruption is occurring. Since the events
				 * received is placed at the end of the giant msg, it is possible
				 * that buffers were not allocated correctly resulting in us
				 * reading/writing to uninitialized memeory at some point. */
				ck_assert_int_eq(giant_event_recv.sent_msgs, events_received);
			}
		}
	} else if (res != -EAGAIN) {
		qb_perror(LOG_DEBUG, "count_stress_events");
		qb_loop_stop(cl);
		return -1;
	}

	if (events_received >= num_stress_events) {
		qb_loop_stop(cl);
		return -1;
	}
	return 0;
}

static int32_t
count_bulk_events(int32_t fd, int32_t revents, void *data)
{
	qb_loop_t *cl = (qb_loop_t*)data;
	struct qb_ipc_response_header res_header;
	int32_t res;

	res = qb_ipcc_event_recv(conn, &res_header,
				 sizeof(struct qb_ipc_response_header),
				 -1);
	if (res > 0) {
		events_received++;
	}

	if (events_received >= num_bulk_events) {
		qb_loop_stop(cl);
		return -1;
	}
	return 0;
}

static void
test_ipc_stress_connections(void)
{
	int32_t c = 0;
	int32_t j = 0;
	uint32_t max_size = MAX_MSG_SIZE;
	int32_t connections = 0;
	pid_t pid;

	multiple_connections = QB_TRUE;

	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_CLEAR_ALL,
			  QB_LOG_FILTER_FILE, "*", LOG_TRACE);
	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD,
			  QB_LOG_FILTER_FILE, "*", LOG_INFO);
	qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE);

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	for (connections = 1; connections < NUM_STRESS_CONNECTIONS; connections++) {
		if (conn) {
			qb_ipcc_disconnect(conn);
			conn = NULL;
		}
		do {
			conn = qb_ipcc_connect(ipc_name, max_size);
			if (conn == NULL) {
				j = waitpid(pid, NULL, WNOHANG);
				ck_assert_int_eq(j, 0);
				sleep(1);
				c++;
			}
		} while (conn == NULL && c < 5);
		ck_assert(conn != NULL);

		if (((connections+1) % 1000) == 0) {
			qb_log(LOG_INFO, "%d ipc connections made", connections+1);
		}
	}
	multiple_connections = QB_FALSE;

	/* Re-enable logging here so we get the "Free'ing" message which allows
	   for resources.test to clear up after us if needed */
	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_CLEAR_ALL,
			  QB_LOG_FILTER_FILE, "*", LOG_TRACE);
	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD,
			  QB_LOG_FILTER_FILE, "*", LOG_TRACE);
	qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE);

	request_server_exit();
	qb_ipcc_disconnect(conn);
	verify_graceful_stop(pid);

}

static void
test_ipc_bulk_events(void)
{
	int32_t c = 0;
	int32_t j = 0;
	pid_t pid;
	int32_t res;
	qb_loop_t *cl;
	int32_t fd;
	uint32_t max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	events_received = 0;
	cl = qb_loop_create();
	res = qb_ipcc_fd_get(conn, &fd);
	ck_assert_int_eq(res, 0);
	res = qb_loop_poll_add(cl, QB_LOOP_MED,
			 fd, POLLIN,
			 cl, count_bulk_events);
	ck_assert_int_eq(res, 0);

	res = send_and_check(IPC_MSG_REQ_BULK_EVENTS,
			     0,
			     recv_timeout, QB_TRUE);
	ck_assert_int_eq(res, sizeof(struct qb_ipc_response_header));

	qb_loop_run(cl);
	ck_assert_int_eq(events_received, num_bulk_events);

	request_server_exit();
	qb_ipcc_disconnect(conn);
	verify_graceful_stop(pid);
}

static void
test_ipc_stress_test(void)
{
	struct {
		struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
		char data[GIANT_MSG_DATA_SIZE] __attribute__ ((aligned(8)));
		uint32_t sent_msgs __attribute__ ((aligned(8)));
	} __attribute__ ((aligned(8))) giant_req;

	struct qb_ipc_response_header res_header;
	struct iovec iov[1];
	int32_t c = 0;
	int32_t j = 0;
	pid_t pid;
	int32_t res;
	qb_loop_t *cl;
	int32_t fd;
	uint32_t max_size = MAX_MSG_SIZE;
	/* This looks strange, but it serves an important purpose.
	 * This test forces the server to enforce the MAX_MSG_SIZE
	 * limit from the server side, which overrides the client's
	 * buffer limit.  To verify this functionality is working
	 * we set the client limit lower than what the server
	 * is enforcing. */
	int32_t client_buf_size = max_size - 1024;
	int32_t real_buf_size;

	enforce_server_buffer = 1;
	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	enforce_server_buffer = 0;
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, client_buf_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	real_buf_size = qb_ipcc_get_buffer_size(conn);
	ck_assert_int_ge(real_buf_size, max_size);

	qb_log(LOG_DEBUG, "Testing %d iterations of EVENT msg passing.", num_stress_events);

	events_received = 0;
	cl = qb_loop_create();
	res = qb_ipcc_fd_get(conn, &fd);
	ck_assert_int_eq(res, 0);
	res = qb_loop_poll_add(cl, QB_LOOP_MED,
			 fd, POLLIN,
			 cl, count_stress_events);
	ck_assert_int_eq(res, 0);

	res = send_and_check(IPC_MSG_REQ_STRESS_EVENT, 0, recv_timeout, QB_TRUE);

	qb_loop_run(cl);
	ck_assert_int_eq(events_received, num_stress_events);

	giant_req.hdr.id = IPC_MSG_REQ_SERVER_FAIL;
	giant_req.hdr.size = sizeof(giant_req);

	if (giant_req.hdr.size <= client_buf_size) {
		ck_assert_int_eq(1, 0);
	}

	iov[0].iov_len = giant_req.hdr.size;
	iov[0].iov_base = &giant_req;
	res = qb_ipcc_sendv_recv(conn, iov, 1,
				 &res_header,
				 sizeof(struct qb_ipc_response_header), -1);
	if (res != -ECONNRESET && res != -ENOTCONN) {
		qb_log(LOG_ERR, "id:%d size:%d", res_header.id, res_header.size);
		ck_assert_int_eq(res, -ENOTCONN);
	}

	qb_ipcc_disconnect(conn);
	verify_graceful_stop(pid);
}

#ifndef __clang__ /* see variable length array in structure' at the top */
START_TEST(test_ipc_stress_test_us)
{
	qb_enter();
	send_event_on_created = QB_FALSE;
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_stress_test();
	qb_leave();
}
END_TEST
#endif

START_TEST(test_ipc_stress_connections_us)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_stress_connections();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_bulk_events_us)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_bulk_events();
	qb_leave();
}
END_TEST

static
READY_SIGNALLER(connected_signaller, _)
{
	request_server_exit();
}

START_TEST(test_ipc_us_native_prio_dlock)
{
	pid_t server_pid, alphaclient_pid;
	struct dispatch_data data;

	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);

	/* this is to demonstrate that native event loop can deal even
	   with "extreme" priority disproportions */
	global_loop_prio = QB_LOOP_LOW;
	multiple_connections = QB_TRUE;
	recv_timeout = -1;

	server_pid = run_function_in_new_process("server", run_ipc_server,
	                                         NULL);
	ck_assert(server_pid != -1);
	data = (struct dispatch_data){.server_pid = server_pid,
	                              .msg_type = IPC_MSG_REQ_SELF_FEED,
	                              .repetitions = 1};
	alphaclient_pid = run_function_in_new_process("alphaclient",
	                                              client_dispatch,
	                                              (void *) &data);
	ck_assert(alphaclient_pid != -1);

	//sleep(1);
	sched_yield();

	data.repetitions = 0;
	client_dispatch(connected_signaller, NULL, (void *) &data);
	verify_graceful_stop(server_pid);

	multiple_connections = QB_FALSE;
	qb_leave();
}
END_TEST

#if HAVE_GLIB
START_TEST(test_ipc_us_glib_prio_dlock)
{
	pid_t server_pid, alphaclient_pid;
	struct dispatch_data data;

	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);

	global_use_glib = QB_TRUE;
	/* this is to make the test pass at all, since GLib is strict
	   on priorities -- QB_LOOP_MED or lower would fail for sure */
	global_loop_prio = QB_LOOP_HIGH;
	multiple_connections = QB_TRUE;
	recv_timeout = -1;

	server_pid = run_function_in_new_process("server", run_ipc_server,
	                                         NULL);
	ck_assert(server_pid != -1);
	data = (struct dispatch_data){.server_pid = server_pid,
	                              .msg_type = IPC_MSG_REQ_SELF_FEED,
	                              .repetitions = 1};
	alphaclient_pid = run_function_in_new_process("alphaclient",
	                                              client_dispatch,
	                                              (void *) &data);
	ck_assert(alphaclient_pid != -1);

	//sleep(1);
	sched_yield();

	data.repetitions = 0;
	client_dispatch(connected_signaller, NULL, (void *) &data);
	verify_graceful_stop(server_pid);

	multiple_connections = QB_FALSE;
	global_loop_prio = QB_LOOP_MED;
	global_use_glib = QB_FALSE;
	qb_leave();
}
END_TEST
#endif

static void
test_ipc_event_on_created(void)
{
	int32_t c = 0;
	int32_t j = 0;
	pid_t pid;
	int32_t res;
	qb_loop_t *cl;
	int32_t fd;
	uint32_t max_size = MAX_MSG_SIZE;

	num_bulk_events = 1;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	events_received = 0;
	cl = qb_loop_create();
	res = qb_ipcc_fd_get(conn, &fd);
	ck_assert_int_eq(res, 0);
	res = qb_loop_poll_add(cl, QB_LOOP_MED,
			 fd, POLLIN,
			 cl, count_bulk_events);
	ck_assert_int_eq(res, 0);

	qb_loop_run(cl);
	ck_assert_int_eq(events_received, num_bulk_events);

	request_server_exit();
	qb_ipcc_disconnect(conn);
	verify_graceful_stop(pid);
}

START_TEST(test_ipc_event_on_created_us)
{
	qb_enter();
	send_event_on_created = QB_TRUE;
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_event_on_created();
	qb_leave();
}
END_TEST

static void
test_ipc_disconnect_after_created(void)
{
	struct qb_ipc_request_header req_header;
	struct qb_ipc_response_header res_header;
	struct iovec iov[1];
	int32_t c = 0;
	int32_t j = 0;
	pid_t pid;
	int32_t res;
	uint32_t max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	ck_assert_int_eq(QB_TRUE, qb_ipcc_is_connected(conn));

	req_header.id = IPC_MSG_REQ_SERVER_DISCONNECT;
	req_header.size = sizeof(struct qb_ipc_request_header);

	iov[0].iov_len = req_header.size;
	iov[0].iov_base = &req_header;

	res = qb_ipcc_sendv_recv(conn, iov, 1,
				 &res_header,
				 sizeof(struct qb_ipc_response_header), -1);
	/*
	 * confirm we get -ENOTCONN or -ECONNRESET
	 */
	if (res != -ECONNRESET && res != -ENOTCONN) {
		qb_log(LOG_ERR, "id:%d size:%d", res_header.id, res_header.size);
		ck_assert_int_eq(res, -ENOTCONN);
	}
	ck_assert_int_eq(QB_FALSE, qb_ipcc_is_connected(conn));
	qb_ipcc_disconnect(conn);
	sleep(1); /* Give it time to stop */
	kill_server(pid);
}

START_TEST(test_ipc_disconnect_after_created_us)
{
	qb_enter();
	disconnect_after_created = QB_TRUE;
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_disconnect_after_created();
	qb_leave();
}
END_TEST

static void
test_ipc_server_fail(void)
{
	int32_t j;
	int32_t c = 0;
	pid_t pid;
	uint32_t max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	request_server_exit();
	if (_fi_unlink_inject_failure == QB_TRUE) {
		_fi_truncate_called = _fi_openat_called = 0;
	}
	ck_assert_int_eq(QB_FALSE, qb_ipcc_is_connected(conn));
	qb_ipcc_disconnect(conn);
	if (_fi_unlink_inject_failure == QB_TRUE) {
		ck_assert_int_ne(_fi_truncate_called + _fi_openat_called, 0);
	}
	verify_graceful_stop(pid);
}

START_TEST(test_ipc_server_fail_soc)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_server_fail();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_dispatch_shm)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_dispatch();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_stress_test_shm)
{
	qb_enter();
	send_event_on_created = QB_FALSE;
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_stress_test();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_stress_connections_shm)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_stress_connections();
	qb_leave();
}
END_TEST

// Check perms uses illegal access to libqb internals
// DO NOT try this at home.
#include "../lib/ipc_int.h"
#include "../lib/ringbuffer_int.h"
START_TEST(test_ipc_server_perms)
{
	pid_t pid;
	struct stat st;
	int j;
	uint32_t max_size;
	int res;
	int c = 0;

	// Can only test this if we are root
	if (getuid() != 0) {
		return;
	}

	ipc_type = QB_IPC_SHM;
	set_perms_on_socket = QB_TRUE;
	max_size = MAX_MSG_SIZE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	/* Check perms - uses illegal access to libqb internals */

	/* BSD uses /var/run for sockets so we can't alter the perms on the
	   directory */
#ifdef __linux__
	char sockdir[PATH_MAX];
	strcpy(sockdir, conn->request.u.shm.rb->shared_hdr->hdr_path);
	*strrchr(sockdir, '/') = 0;

	res = stat(sockdir, &st);

	ck_assert_int_eq(res, 0);
	ck_assert(st.st_mode & S_IRWXG);
	ck_assert_int_eq(st.st_uid, 555);
	ck_assert_int_eq(st.st_gid, 741);
#endif

	res = stat(conn->request.u.shm.rb->shared_hdr->hdr_path, &st);
	ck_assert_int_eq(res, 0);
	ck_assert_int_eq(st.st_uid, 555);
	ck_assert_int_eq(st.st_gid, 741);

	qb_ipcc_disconnect(conn);
	verify_graceful_stop(pid);
}
END_TEST

START_TEST(test_ipc_disp_shm_native_prio_dlock)
{
	pid_t server_pid, alphaclient_pid;
	struct dispatch_data data;

	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);

	/* this is to demonstrate that native event loop can deal even
	   with "extreme" priority disproportions */
	global_loop_prio = QB_LOOP_LOW;
	multiple_connections = QB_TRUE;
	recv_timeout = -1;

	server_pid = run_function_in_new_process("server", run_ipc_server,
	                                         NULL);
	ck_assert(server_pid != -1);
	data = (struct dispatch_data){.server_pid = server_pid,
	                              .msg_type = IPC_MSG_REQ_SELF_FEED,
	                              .repetitions = 1};
	alphaclient_pid = run_function_in_new_process("alphaclient",
	                                              client_dispatch,
	                                              (void *) &data);
	ck_assert(alphaclient_pid != -1);

	//sleep(1);
	sched_yield();

	data.repetitions = 0;
	client_dispatch(connected_signaller, NULL, (void *) &data);
	verify_graceful_stop(server_pid);

	multiple_connections = QB_FALSE;
	qb_leave();
}
END_TEST

#if HAVE_GLIB
START_TEST(test_ipc_disp_shm_glib_prio_dlock)
{
	pid_t server_pid, alphaclient_pid;
	struct dispatch_data data;

	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);

	global_use_glib = QB_TRUE;
	/* this is to make the test pass at all, since GLib is strict
	   on priorities -- QB_LOOP_MED or lower would fail for sure */
	global_loop_prio = QB_LOOP_HIGH;
	multiple_connections = QB_TRUE;
	recv_timeout = -1;

	server_pid = run_function_in_new_process("server", run_ipc_server,
	                                         NULL);
	ck_assert(server_pid != -1);
	data = (struct dispatch_data){.server_pid = server_pid,
	                              .msg_type = IPC_MSG_REQ_SELF_FEED,
	                              .repetitions = 1};
	alphaclient_pid = run_function_in_new_process("alphaclient",
	                                              client_dispatch,
	                                              (void *) &data);
	ck_assert(alphaclient_pid != -1);

	//sleep(1);
	sched_yield();

	data.repetitions = 0;
	client_dispatch(connected_signaller, NULL, (void *) &data);
	verify_graceful_stop(server_pid);

	multiple_connections = QB_FALSE;
	global_loop_prio = QB_LOOP_MED;
	global_use_glib = QB_FALSE;
	qb_leave();
}
END_TEST
#endif

START_TEST(test_ipc_bulk_events_shm)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_bulk_events();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_event_on_created_shm)
{
	qb_enter();
	send_event_on_created = QB_TRUE;
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_event_on_created();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_server_fail_shm)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_server_fail();
	qb_leave();
}
END_TEST

#ifdef HAVE_FAILURE_INJECTION
START_TEST(test_ipcc_truncate_when_unlink_fails_shm)
{
	char sock_file[PATH_MAX];
	struct sockaddr_un socka;

	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);

	sprintf(sock_file, "%s/%s", SOCKETDIR, ipc_name);
	sock_file[sizeof(socka.sun_path)] = '\0';

	/* If there's an old socket left from a previous run this test will fail
	   unexpectedly, so try to remove it first */
	unlink(sock_file);

	_fi_unlink_inject_failure = QB_TRUE;
	test_ipc_server_fail();
	_fi_unlink_inject_failure = QB_FALSE;
	unlink(sock_file);
	qb_leave();
}
END_TEST
#endif

static void
test_ipc_service_ref_count(void)
{
	int32_t c = 0;
	int32_t j = 0;
	pid_t pid;
	uint32_t max_size = MAX_MSG_SIZE;

	reference_count_test = QB_TRUE;

	pid = run_function_in_new_process("server", run_ipc_server, NULL);
	ck_assert(pid != -1);

	do {
		conn = qb_ipcc_connect(ipc_name, max_size);
		if (conn == NULL) {
			j = waitpid(pid, NULL, WNOHANG);
			ck_assert_int_eq(j, 0);
			(void)poll(NULL, 0, 400);
			c++;
		}
	} while (conn == NULL && c < 5);
	ck_assert(conn != NULL);

	sleep(5);

	kill_server(pid);
}


START_TEST(test_ipc_service_ref_count_shm)
{
	qb_enter();
	ipc_type = QB_IPC_SHM;
	set_ipc_name(__func__);
	test_ipc_service_ref_count();
	qb_leave();
}
END_TEST

START_TEST(test_ipc_service_ref_count_us)
{
	qb_enter();
	ipc_type = QB_IPC_SOCKET;
	set_ipc_name(__func__);
	test_ipc_service_ref_count();
	qb_leave();
}
END_TEST

#if 0
static void test_max_dgram_size(void)
{
	/* most implementations will not let you set a dgram buffer
	 * of 1 million bytes. This test verifies that the we can detect
	 * the max dgram buffersize regardless, and that the value we detect
	 * is consistent. */
	int32_t init;
	int32_t i;

	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_REMOVE,
			  QB_LOG_FILTER_FILE, "*", LOG_TRACE);

	init = qb_ipcc_verify_dgram_max_msg_size(1000000);
	ck_assert(init > 0);
	for (i = 0; i < 100; i++) {
		int try = qb_ipcc_verify_dgram_max_msg_size(1000000);
#if 0
		ck_assert_int_eq(init, try);
#else
		/* extra troubleshooting, report also on i and errno variables;
		   related: https://github.com/ClusterLabs/libqb/issues/234 */
		if (init != try) {
#ifdef ci_dump_shm_usage
			system("df -h | grep -e /shm >/tmp/_shm_usage");
#endif
			ck_abort_msg("Assertion 'init==try' failed:"
				     " init==%#x, try==%#x, i=%d, errno=%d",
				     init, try, i, errno);
		}
#endif
	}

	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD,
			  QB_LOG_FILTER_FILE, "*", LOG_TRACE);
}

START_TEST(test_ipc_max_dgram_size)
{
	qb_enter();
	test_max_dgram_size();
	qb_leave();
}
END_TEST
#endif

static Suite *
make_shm_suite(void)
{
	TCase *tc;
	Suite *s = suite_create("shm");

	add_tcase(s, tc, test_ipc_shm_connect_async, 7);

	add_tcase(s, tc, test_ipc_txrx_shm_getauth, 7);
	add_tcase(s, tc, test_ipc_txrx_shm_timeout, 28);
	add_tcase(s, tc, test_ipc_server_fail_shm, 7);
	add_tcase(s, tc, test_ipc_txrx_shm_block, 7);
	add_tcase(s, tc, test_ipc_txrx_shm_tmo, 7);
	add_tcase(s, tc, test_ipc_fc_shm, 7);
	add_tcase(s, tc, test_ipc_dispatch_shm, 15);
	add_tcase(s, tc, test_ipc_stress_test_shm, 15);
	add_tcase(s, tc, test_ipc_bulk_events_shm, 15);
	add_tcase(s, tc, test_ipc_exit_shm, 6);
	add_tcase(s, tc, test_ipc_event_on_created_shm, 9);
	add_tcase(s, tc, test_ipc_service_ref_count_shm, 9);
	add_tcase(s, tc, test_ipc_server_perms, 7);
	add_tcase(s, tc, test_ipc_stress_connections_shm, 3600 /* ? */);
	add_tcase(s, tc, test_ipc_disp_shm_native_prio_dlock, 15);
#if HAVE_GLIB
	add_tcase(s, tc, test_ipc_disp_shm_glib_prio_dlock, 15);
#endif
#ifdef HAVE_FAILURE_INJECTION
	add_tcase(s, tc, test_ipcc_truncate_when_unlink_fails_shm, 8);
#endif

	return s;
}

static Suite *
make_soc_suite(void)
{
	Suite *s = suite_create("socket");
	TCase *tc;

	add_tcase(s, tc, test_ipc_us_connect_async, 7);

	add_tcase(s, tc, test_ipc_txrx_us_getauth, 7);
	add_tcase(s, tc, test_ipc_txrx_us_timeout, 28);
/* Commented out for the moment as space in /dev/shm on the CI machines
   causes random failures */
/*	add_tcase(s, tc, test_ipc_max_dgram_size, 30); */
	add_tcase(s, tc, test_ipc_server_fail_soc, 7);
	add_tcase(s, tc, test_ipc_txrx_us_block, 7);
	add_tcase(s, tc, test_ipc_txrx_us_tmo, 7);
	add_tcase(s, tc, test_ipc_fc_us, 7);
	add_tcase(s, tc, test_ipc_exit_us, 6);
	add_tcase(s, tc, test_ipc_dispatch_us, 15);
#ifndef __clang__ /* see variable length array in structure' at the top */
	add_tcase(s, tc, test_ipc_stress_test_us, 58);
#endif
	add_tcase(s, tc, test_ipc_bulk_events_us, 15);
	add_tcase(s, tc, test_ipc_event_on_created_us, 9);
	add_tcase(s, tc, test_ipc_disconnect_after_created_us, 9);
	add_tcase(s, tc, test_ipc_service_ref_count_us, 9);
	add_tcase(s, tc, test_ipc_stress_connections_us, 3600 /* ? */);
	add_tcase(s, tc, test_ipc_us_native_prio_dlock, 15);
#if HAVE_GLIB
	add_tcase(s, tc, test_ipc_us_glib_prio_dlock, 15);
#endif

	return s;
}

int32_t
main(void)
{
	int32_t number_failed;
	SRunner *sr;
	Suite *s;
	int32_t do_shm_tests = QB_TRUE;

	set_ipc_name("ipc_test");
#ifdef DISABLE_IPC_SHM
	do_shm_tests = QB_FALSE;
#endif /* DISABLE_IPC_SHM */

	s = make_soc_suite();
	sr = srunner_create(s);

	if (do_shm_tests) {
		srunner_add_suite(sr, make_shm_suite());
	}

	qb_log_init("check", LOG_USER, LOG_EMERG);
	atexit(qb_log_fini);
	qb_log_ctl(QB_LOG_SYSLOG, QB_LOG_CONF_ENABLED, QB_FALSE);
	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD,
			  QB_LOG_FILTER_FILE, "*", LOG_TRACE);
	qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE);
	qb_log_format_set(QB_LOG_STDERR, "lib/%f|%l| %b");

	srunner_run_all(sr, CK_VERBOSE);
	number_failed = srunner_ntests_failed(sr);
	srunner_free(sr);
	return (number_failed == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
}