Simplify, do not install examples or info manual, and change license3.1.7

Signed-off-by: Mattias Andrée <maandree@kth.se>

1 files changed, 1173 insertions, 0 deletions

diff --git a/libbus.c b/libbus.c
new file mode 100644
index 0000000..0a84353
--- /dev/null
+++ b/libbus.c
@@ -0,0 +1,1173 @@
+/* See LICENSE file for copyright and license details. */
+#include "bus.h"
+
+#include <sys/ipc.h>
+#include <sys/sem.h>
+#include <sys/shm.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <time.h>
+#include <unistd.h>
+
+
+#ifdef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_EVEN_HARDER
+# ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+#  define BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+# endif
+#endif
+#ifdef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+# ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+#  define BUS_SEMAPHORES_ARE_SYNCHRONOUS
+# endif
+#endif
+
+
+/**
+ * Semaphore used to signal `bus_write` that `bus_read` is ready
+ */
+#define S  0
+
+/**
+ * Semaphore for making `bus_write` wait while `bus_read` is reseting `S`
+ */
+#define W  1
+
+/**
+ * Binary semaphore for making `bus_write` exclusively locked
+ */
+#define X  2
+
+/**
+ * Semaphore used to cue `bus_read` that it may read the shared memory
+ */
+#define Q  3
+
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_EVEN_HARDER
+/**
+ * Semaphore used to notify `bus_read` that it may restore `S`
+ */
+# define N  4
+
+/**
+ * The number of semaphores in the semaphore array
+ */
+# define BUS_SEMAPHORES  5
+#else
+# define BUS_SEMAPHORES  4
+#endif
+
+/**
+ * The default permission mits of the bus
+ */
+#define DEFAULT_MODE  0600
+
+
+
+/**
+ * Decrease the value of a semaphore by 1
+ * 
+ * @param   bus:const bus_t *  The bus
+ * @param   semaphore:int      The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   flags:int          `SEM_UNDO` if the action should be undone when the program exits,
+ *                             `IPC_NOWAIT` if the action should fail if it would block
+ * @return  :int               0 on success, -1 on error
+ */
+#define acquire_semaphore(bus, semaphore, flags) \
+	semaphore_op(bus, semaphore, -1, flags)
+
+/**
+ * Increase the value of a semaphore by 1
+ * 
+ * @param   bus:const bus_t *  The bus
+ * @param   semaphore:int      The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   flags:int          `SEM_UNDO` if the action should be undone when the program exits
+ * @return  :int               0 on success, -1 on error
+ */
+#define release_semaphore(bus, semaphore, flags) \
+	semaphore_op(bus, semaphore, +1, flags)
+
+/**
+ * Wait for the value of a semaphore to become 0
+ * 
+ * @param   bus:const bus_t *  The bus
+ * @param   semaphore:int      The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   flags:int          `IPC_NOWAIT` if the action should fail if it would block
+ * @return  :int               0 on success, -1 on error
+ */
+#define zero_semaphore(bus, semaphore, flags) \
+	semaphore_op(bus, semaphore, 0, flags)
+
+/**
+ * Decrease the value of a semaphore by 1
+ * 
+ * @param   bus:const bus_t *                The bus
+ * @param   semaphore:int                    The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   flags:int                        `SEM_UNDO` if the action should be undone when the program exits,
+ *                                           `IPC_NOWAIT` if the action should fail if it would block
+ * @param   timeout:const struct timespec *  The amount of time to wait before failing
+ * @return  :int                             0 on success, -1 on error
+ */
+#define acquire_semaphore_timed(bus, semaphore, flags, timeout) \
+	semaphore_op_timed(bus, semaphore, -1, flags, timeout)
+
+/**
+ * Increase the value of a semaphore by 1
+ * 
+ * @param   bus:const bus_t *                The bus
+ * @param   semaphore:int                    The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   flags:int                        `SEM_UNDO` if the action should be undone when the program exits
+ * @param   timeout:const struct timespec *  The amount of time to wait before failing
+ * @return  :int                             0 on success, -1 on error
+ */
+#define release_semaphore_timed(bus, semaphore, flags, timeout) \
+	semaphore_op_timed(bus, semaphore, +1, flags, timeout)
+
+/**
+ * Wait for the value of a semaphore to become 0
+ * 
+ * @param   bus:const bus_t *                The bus
+ * @param   semaphore:int                    The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   flags:int                        `IPC_NOWAIT` if the action should fail if it would block
+ * @param   timeout:const struct timespec *  The amount of time to wait before failing
+ * @return  :int                             0 on success, -1 on error
+ */
+#define zero_semaphore_timed(bus, semaphore, flags, timeout) \
+	semaphore_op_timed(bus, semaphore, 0, flags, timeout)
+
+/**
+ * Open the semaphore array
+ * 
+ * @param   bus:const bus_t *  The bus
+ * @return  :int               0 on success, -1 on error
+ */
+#define open_semaphores(bus) \
+	(((bus)->sem_id = semget((bus)->key_sem, BUS_SEMAPHORES, 0)) == -1 ? -1 : 0)
+
+/**
+ * Write a message to the shared memory
+ * 
+ * @param   bus:const bus_t *  The bus
+ * @param   msg:const char *   The message
+ * @return  :int               0 on success, -1 on error
+ */
+#define write_shared_memory(bus, msg) \
+	(memcpy((bus)->message, msg, (strlen(msg) + 1) * sizeof(char)))
+
+
+/**
+ * Set `delta` to the convertion of `timeout` from absolute to relative time,
+ * measured in the clock whose ID is specified by `clockid`
+ * 
+ * @scope  timeout:struct timespec          Output variable for relative time
+ * @scope  timeout:const struct timespec *  The absolute time
+ * @scope  clockid:clockid_t                The clock time is measured
+ */
+#define DELTA \
+	do { \
+		if (absolute_time_to_delta_time(&delta, timeout, clockid) < 0) { \
+			goto fail; \
+		} else if ((delta.tv_sec < 0) || (delta.tv_nsec < 0)) { \
+			errno = EAGAIN; \
+			goto fail; \
+		}\
+	} while (0)
+
+
+/**
+ * If `flags & (bus_flag)`, this macro evalutes to `sys_flag`,
+ * otherwise this macro evalutes to 0.
+ */
+#define F(bus_flag, sys_flag) \
+	((flags & (bus_flag)) ? sys_flag : 0)
+
+
+
+/**
+ * Statement wrapper that goes to `fail` on failure
+ */
+#define t(inst) \
+	do { if ((inst) == -1) goto fail; } while (0)
+
+
+
+#ifdef _SEM_SEMUN_UNDEFINED
+union semun {
+	int val;
+	struct semid_ds *buf;
+	unsigned short *array;
+};
+#endif
+
+
+
+/**
+ * Create a semaphore array for the bus
+ * 
+ * @param   bus  Bus information to fill with the key of the created semaphore array
+ * @return       0 on success, -1 on error
+ */
+static int
+create_semaphores(bus_t *bus)
+{
+	int id = -1, rint, saved_errno;
+	double r;
+	union semun values;
+
+	values.array = NULL;
+
+	/* Create semaphore array. */
+	for (;;) {
+		rint = rand();
+		r = (double)rint;
+		r /= (double)RAND_MAX + 1;
+		r *= (1 << (8 * sizeof(key_t) - 2)) - 1;
+		bus->key_sem = (key_t)r + 1;
+		if (bus->key_sem == IPC_PRIVATE)
+			continue;
+		id = semget(bus->key_sem, BUS_SEMAPHORES, IPC_CREAT | IPC_EXCL | DEFAULT_MODE);
+		if (id != -1)
+			break;
+		if ((errno != EEXIST) && (errno != EINTR))
+			goto fail;
+	}
+
+	/* Initialise the array. */
+	values.array = calloc((size_t)BUS_SEMAPHORES, sizeof(unsigned short));
+	if (!values.array)
+		goto fail;
+	values.array[X] = 1;
+	if (semctl(id, 0, SETALL, values.array) == -1)
+		goto fail;
+	free(values.array);
+	values.array = NULL;
+
+	return 0;
+
+fail:
+	saved_errno = errno;
+	if (id != -1)
+		semctl(id, 0, IPC_RMID);
+	free(values.array);
+	errno = saved_errno;
+	return -1;
+}
+
+
+/**
+ * Create a shared memory for the bus
+ * 
+ * @param   bus  Bus information to fill with the key of the created shared memory
+ * @return       0 on success, -1 on error
+ */
+static int
+create_shared_memory(bus_t *bus)
+{
+	int id = -1, rint, saved_errno;
+	double r;
+	struct shmid_ds _info;
+
+	/* Create shared memory. */
+	for (;;) {
+		rint = rand();
+		r = (double)rint;
+		r /= (double)RAND_MAX + 1;
+		r *= (1 << (8 * sizeof(key_t) - 2)) - 1;
+		bus->key_shm = (key_t)r + 1;
+		if (bus->key_shm == IPC_PRIVATE)
+			continue;
+		id = shmget(bus->key_shm, (size_t)BUS_MEMORY_SIZE, IPC_CREAT | IPC_EXCL | DEFAULT_MODE);
+		if (id != -1)
+			break;
+		if ((errno != EEXIST) && (errno != EINTR))
+			goto fail;
+	}
+
+	return 0;
+
+fail:
+	saved_errno = errno;
+	if (id != -1)
+		shmctl(id, IPC_RMID, &_info);
+	errno = saved_errno;
+	return -1;
+}
+
+
+/**
+ * Remove the semaphore array for the bus
+ * 
+ * @param   bus  Bus information
+ * @return       0 on success, -1 on error
+ */
+static int
+remove_semaphores(const bus_t *bus)
+{
+	int id = semget(bus->key_sem, BUS_SEMAPHORES, 0);
+	return ((id == -1) || (semctl(id, 0, IPC_RMID) == -1)) ? -1 : 0;
+}
+
+
+/**
+ * Remove the shared memory for the bus
+ * 
+ * @param   bus  Bus information
+ * @return       0 on success, -1 on error
+ */
+static int
+remove_shared_memory(const bus_t *bus)
+{
+	struct shmid_ds _info;
+	int id = shmget(bus->key_shm, (size_t)BUS_MEMORY_SIZE, 0);
+	return ((id == -1) || (shmctl(id, IPC_RMID, &_info) == -1)) ? -1 : 0;
+}
+
+
+/**
+ * Increase or decrease the value of a semaphore, or wait the it to become 0
+ * 
+ * @param   bus        Bus information
+ * @param   semaphore  The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   delta      The adjustment to make to the semaphore's value, 0 to wait for it to become 0
+ * @param   flags      `SEM_UNDO` if the action should be undone when the program exits
+ * @return             0 on success, -1 on error
+ */
+static int
+semaphore_op(const bus_t *bus, int semaphore, int delta, int flags)
+{
+	struct sembuf op;
+	op.sem_num = (unsigned short)semaphore;
+	op.sem_op = (short)delta;
+	op.sem_flg = (short)flags;
+	return semop(bus->sem_id, &op, (size_t)1);
+}
+
+
+/**
+ * Increase or decrease the value of a semaphore, or wait the it to become 0
+ * 
+ * @param   bus        Bus information
+ * @param   semaphore  The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   delta      The adjustment to make to the semaphore's value, 0 to wait for it to become 0
+ * @param   flags      `SEM_UNDO` if the action should be undone when the program exits
+ * @param   timeout    The amount of time to wait before failing
+ * @return             0 on success, -1 on error
+ */
+static int
+semaphore_op_timed(const bus_t *bus, int semaphore, int delta, int flags, const struct timespec *timeout)
+{
+	struct sembuf op;
+	op.sem_num = (unsigned short)semaphore;
+	op.sem_op = (short)delta;
+	op.sem_flg = (short)flags;
+	return semtimedop(bus->sem_id, &op, (size_t)1, timeout);
+}
+
+
+/**
+ * Set the value of a semaphore
+ * 
+ * @param   bus        Bus information
+ * @param   semaphore  The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param   value      The new value of the semaphore
+ * @return             0 on success, -1 on error
+ */
+static int
+write_semaphore(const bus_t *bus, unsigned semaphore, int value)
+{
+	union semun semval;
+	semval.val = value;
+	return semctl(bus->sem_id, (unsigned short)semaphore, SETVAL, semval);
+}
+
+
+/**
+ * Open the shared memory for the bus
+ * 
+ * @param   bus    Bus information
+ * @param   flags  `BUS_RDONLY`, `BUS_WRONLY` or `BUS_RDWR`
+ * @return         0 on success, -1 on error
+ */
+static int
+open_shared_memory(bus_t *bus, int flags)
+{
+	int id;
+	void *address;
+	t(id = shmget(bus->key_shm, (size_t)BUS_MEMORY_SIZE, 0));
+	address = shmat(id, NULL, (flags & BUS_RDONLY) ? SHM_RDONLY : 0);
+	if ((address == (void *)-1) || !address)
+		goto fail;
+	bus->message = (char *)address;
+	return 0;
+fail:
+	return -1;
+}
+
+
+/**
+ * Close the shared memory for the bus
+ * 
+ * @param   bus  Bus information
+ * @return       0 on success, -1 on error
+ */
+static int
+close_shared_memory(bus_t *bus)
+{
+	t(shmdt(bus->message));
+	bus->message = NULL;
+	return 0;
+fail:
+	return -1;
+}
+
+
+/**
+ * Get a random ASCII letter or digit
+ * 
+ * @return  A random ASCII letter or digit
+ */
+static char
+randomchar(void)
+{
+	int rint = rand();
+	double r = (double)rint;
+	r /= (double)RAND_MAX + 1;
+	r *= 10 + 26 + 26;
+	return "0123456789qwertyuiopasdfghjklzxcvbnmQWERTYUIOPASDFGHJKLZXCVBNM"[(int)r];
+}
+
+
+/**
+ * Basically, this is `mkdir -p -m $mode $pathname`
+ * 
+ * @param   pathname  The pathname of the directory to create if missing
+ * @param   mode      The permission bits of any created directory
+ * @return            0 on sucess, -1 on error
+ */
+static int
+mkdirs(char *pathname, mode_t mode)
+{
+	size_t i, n = strlen(pathname);
+	char c;
+	for (i = 0; i < n; i++)
+		if (pathname[i] != '/')
+			break;
+	for (; i < n; i++) {
+		if (pathname[i] == '/') {
+			c = pathname[i];
+			if (access(pathname, F_OK))
+				if (mkdir(pathname, mode) < 0)
+					return -1;
+			pathname[i] = c;
+			break;
+		}
+	}
+	if (access(pathname, F_OK))
+		if (mkdir(pathname, mode) < 0)
+			return -1;
+	return 0;
+}
+
+
+/**
+ * Convert an absolute time to a relative time
+ * 
+ * @param   delta     Output parameter for the relative time
+ * @param   absolute  The absolute time
+ * @param   clockid   The ID of the clock the time is measured in
+ * @return            0 on success, -1 on error
+ */
+static int
+absolute_time_to_delta_time(struct timespec *delta, const struct timespec *absolute, clockid_t clockid)
+{
+	if (clock_gettime(clockid, delta) < 0)
+		return -1;
+
+	delta->tv_sec  = absolute->tv_sec  - delta->tv_sec;
+	delta->tv_nsec = absolute->tv_nsec - delta->tv_nsec;
+
+	if (delta->tv_nsec < 0L) {
+		delta->tv_nsec += 1000000000L;
+		delta->tv_sec -= 1;
+	}
+	if (delta->tv_nsec >= 1000000000L) {
+		delta->tv_nsec -= 1000000000L;
+		delta->tv_sec += 1;
+	}
+
+	return 0;
+}
+
+
+
+/**
+ * Create a new bus
+ * 
+ * @param   file      The pathname of the bus, `NULL` to create a random one
+ * @param   flags     `BUS_EXCL` (if `file` is not `NULL`) to fail if the file
+ *                    already exists, otherwise if the file exists, nothing
+ *                    will happen;
+ *                    `BUS_INTR` to fail if interrupted
+ * @param   out_file  Output parameter for the pathname of the bus
+ * @return            0 on success, -1 on error
+ */
+int
+bus_create(const char *restrict file, int flags, char **restrict out_file)
+{
+	int fd = -1, saved_errno;
+	bus_t bus;
+	char buf[1 + 2 * (3 * sizeof(ssize_t) + 2)];
+	size_t ptr, len;
+	ssize_t wrote;
+	char *genfile = NULL;
+	const char *env;
+
+	if (out_file)
+		*out_file = NULL;
+
+	bus.sem_id = -1;
+	bus.key_sem = -1;
+	bus.key_shm = -1;
+	bus.message = NULL;
+	bus.first_poll = 0;
+
+	srand((unsigned int)time(NULL) + (unsigned int)rand());
+
+	if (file) {
+		fd = open(file, O_WRONLY | O_CREAT | O_EXCL, DEFAULT_MODE);
+		if (fd == -1) {
+			if ((errno != EEXIST) || (flags & BUS_EXCL))
+				return -1;
+			goto done;
+		}
+	} else {
+		env = getenv("XDG_RUNTIME_DIR");
+		if (!env || !*env)
+			env = "/run";
+		genfile = malloc((strlen(env) + 6 + 7 + 30) * sizeof(char));
+		if (!genfile)
+			goto fail;
+		if (out_file)
+			*out_file = genfile;
+		sprintf(genfile, "%s/bus", env);
+		t(mkdirs(genfile, 0755));
+		sprintf(genfile, "%s/bus/random.", env);
+		len = strlen(genfile);
+		genfile[len + 30] = '\0';
+	retry:
+		for (ptr = 0; ptr < 30; ptr++)
+			genfile[len + ptr] = randomchar();
+		fd = open(genfile, O_WRONLY | O_CREAT | O_EXCL, DEFAULT_MODE);
+		if (fd == -1) {
+			if (errno == EEXIST)
+				goto retry;
+			return -1;
+		}
+	}
+
+	t(create_semaphores(&bus));
+	t(create_shared_memory(&bus));
+
+	sprintf(buf, "%zi\n%zi\n", (ssize_t)(bus.key_sem), (ssize_t)(bus.key_shm));
+	for (len = strlen(buf), ptr = 0; ptr < len;) {
+		wrote = write(fd, buf + ptr, len - ptr);
+		if (wrote < 0) {
+			if ((errno != EINTR) || (flags & BUS_INTR))
+				goto fail;
+		} else {
+			ptr += (size_t)wrote;
+		}
+	}
+	close(fd);
+
+done:
+	if (out_file && !*out_file) {
+		len = strlen(file) + 1;
+		*out_file = malloc(len * sizeof(char));
+		memcpy(*out_file, file, len * sizeof(char));
+	} else if (!out_file) {
+		free(genfile);
+	}
+	return 0;
+
+fail:
+	saved_errno = errno;
+	if (bus.key_sem)
+		remove_semaphores(&bus);
+	if (bus.key_shm)
+		remove_shared_memory(&bus);
+	if (fd == -1)
+		close(fd);
+	if (out_file)
+		*out_file = NULL;
+	free(genfile);
+	unlink(file);
+	errno = saved_errno;
+	return -1;
+}
+
+
+/**
+ * Remove a bus
+ * 
+ * @param   file  The pathname of the bus
+ * @return        0 on success, -1 on error
+ */
+int
+bus_unlink(const char *file)
+{
+	int r = 0, saved_errno = 0;
+	bus_t bus;
+	t(bus_open(&bus, file, -1));
+
+	r |= remove_semaphores(&bus);
+	if (r && !saved_errno)
+		saved_errno = errno;
+
+	r |= remove_shared_memory(&bus);
+	if (r && !saved_errno)
+		saved_errno = errno;
+
+	r |= unlink(file);
+	if (r && !saved_errno)
+		saved_errno = errno;
+
+	errno = saved_errno;
+	return r;
+fail:
+	return -1;
+}
+
+
+/**
+ * Open an existing bus
+ * 
+ * @param   bus    Bus information to fill
+ * @param   file   The filename of the bus
+ * @param   flags  `BUS_RDONLY`, `BUS_WRONLY` or `BUS_RDWR`
+ *                 any negative value is used internally
+ *                 for telling the function to not actually
+ *                 opening the bus, but just to parse the file
+ * @return         0 on success, -1 on error
+ */
+int
+bus_open(bus_t *restrict bus, const char *restrict file, int flags)
+{
+	int saved_errno;
+	char *line = NULL;
+	size_t len = 0;
+	FILE *f;
+
+	bus->sem_id = -1;
+	bus->key_sem = -1;
+	bus->key_shm = -1;
+	bus->message = NULL;
+
+	f = fopen(file, "r");
+
+	t(getline(&line, &len, f));
+	t(bus->key_sem = (key_t)atoll(line));
+	free(line), line = NULL, len = 0;
+
+	t(getline(&line, &len, f));
+	t(bus->key_shm = (key_t)atoll(line));
+	free(line), line = NULL;
+
+	fclose(f);
+
+	if (flags >= 0) {
+		t(open_semaphores(bus));
+		t(open_shared_memory(bus, flags));
+	}
+	return 0;
+fail:
+	saved_errno = errno;
+	free(line);
+	errno = saved_errno;
+	return -1;
+}
+
+
+/**
+ * Close a bus
+ * 
+ * @param   bus  Bus information
+ * @return       0 on success, -1 on error
+ */
+int
+bus_close(bus_t *bus)
+{
+	bus->sem_id = -1;
+	if (bus->message)
+		t(close_shared_memory(bus));
+	bus->message = NULL;
+	return 0;
+
+fail:
+	return -1;
+}
+
+
+/**
+ * Broadcast a message on a bus
+ * 
+ * @param   bus      Bus information
+ * @param   message  The message to write, may not be longer than
+ *                   `BUS_MEMORY_SIZE` including the NUL-termination
+ * @param   flags    `BUS_NOWAIT` if this function shall fail if
+ *                   another process is currently running this
+ *                   procedure
+ * @return           0 on success, -1 on error
+ */
+int
+bus_write(const bus_t *bus, const char *message, int flags)
+{
+	int saved_errno;
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	int state = 0;
+#endif
+	if (acquire_semaphore(bus, X, SEM_UNDO | F(BUS_NOWAIT, IPC_NOWAIT)) == -1)
+		return -1;
+	t(zero_semaphore(bus, W, 0));
+	write_shared_memory(bus, message);
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	t(release_semaphore(bus, N, SEM_UNDO));  state++;
+#endif
+	t(write_semaphore(bus, Q, 0));
+	t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	t(acquire_semaphore(bus, N, SEM_UNDO));  state--;
+#endif
+	t(release_semaphore(bus, X, SEM_UNDO));
+	return 0;
+
+fail:
+	saved_errno = errno;
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	if (state > 0)
+		acquire_semaphore(bus, N, SEM_UNDO);
+#endif
+	release_semaphore(bus, X, SEM_UNDO);
+	errno = saved_errno;
+	return -1;
+}
+
+
+/**
+ * Broadcast a message on a bus
+ * 
+ * @param   bus      Bus information
+ * @param   message  The message to write, may not be longer than
+ *                   `BUS_MEMORY_SIZE` including the NUL-termination
+ * @param   timeout  The time the operation shall fail with errno set
+ *                   to `EAGAIN` if not completed
+ * @param   clockid  The ID of the clock the `timeout` is measured with,
+ *                   it most be a predictable clock
+ * @return           0 on success, -1 on error
+ */
+int bus_write_timed(const bus_t *bus, const char *message,
+		    const struct timespec *timeout, clockid_t clockid)
+{
+	int saved_errno;
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	int state = 0;
+#endif
+	struct timespec delta;
+	if (!timeout)
+		return bus_write(bus, message, 0);
+
+	DELTA;
+	if (acquire_semaphore_timed(bus, X, SEM_UNDO, &delta) == -1)
+		return -1;
+	DELTA;
+	t(zero_semaphore_timed(bus, W, 0, &delta));
+	write_shared_memory(bus, message);
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	t(release_semaphore(bus, N, SEM_UNDO));  state++;
+#endif
+	t(write_semaphore(bus, Q, 0));
+	t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	t(acquire_semaphore(bus, N, SEM_UNDO));  state--;
+#endif
+	t(release_semaphore(bus, X, SEM_UNDO));
+	return 0;
+
+fail:
+	saved_errno = errno;
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+	if (state > 0)
+		acquire_semaphore(bus, N, SEM_UNDO);
+#endif
+	release_semaphore(bus, X, SEM_UNDO);
+	errno = saved_errno;
+	return -1;
+}
+
+
+/**
+ * Listen (in a loop, forever) for new message on a bus
+ * 
+ * @param   bus        Bus information
+ * @param   callback   Function to call when a message is received, the
+ *                     input parameters will be the read message and
+ *                     `user_data` from `bus_read`'s parameter with the
+ *                     same name. The message must have been parsed or
+ *                     copied when `callback` returns as it may be over
+ *                     overridden after that time. `callback` should
+ *                     return either of the the values:
+ *                       *  0:  stop listening
+ *                       *  1:  continue listening
+ *                       * -1:  an error has occurred
+ *                     However, the function [`bus_read`] will invoke
+ *                     `callback` with `message` set to `NULL`one time
+ *                     directly after it has started listening on the
+ *                     bus. This is to the the program now it can safely
+ *                     continue with any action that requires that the
+ *                     programs is listening on the bus.
+ * @param   user_data  Parameter passed to `callback`
+ * @return             0 on success, -1 on error
+ */
+int
+bus_read(const bus_t *restrict bus, int (*callback)(const char *message, void *user_data), void *user_data)
+{
+	int r, state = 0, saved_errno;
+	if (release_semaphore(bus, S, SEM_UNDO) == -1)
+		return -1;
+	t(r = callback(NULL, user_data));
+	if (!r)  goto done;
+	for (;;) {
+		t(release_semaphore(bus, Q, 0));
+		t(zero_semaphore(bus, Q, 0));
+		t(r = callback(bus->message, user_data));
+		if (!r)  goto done;
+		t(release_semaphore(bus, W, SEM_UNDO));  state++;
+		t(acquire_semaphore(bus, S, SEM_UNDO));  state++;
+		t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+		t(zero_semaphore(bus, N, 0));
+#endif
+		t(release_semaphore(bus, S, SEM_UNDO));  state--;
+		t(acquire_semaphore(bus, W, SEM_UNDO));  state--;
+	}
+
+fail:
+	saved_errno = errno;
+	if (state > 1)
+		release_semaphore(bus, S, SEM_UNDO);
+	if (state > 0)
+		acquire_semaphore(bus, W, SEM_UNDO);
+	acquire_semaphore(bus, S, SEM_UNDO);
+	errno = saved_errno;
+	return -1;
+
+done:
+	t(acquire_semaphore(bus, S, SEM_UNDO));
+	return 0;
+}
+
+
+/**
+ * Listen (in a loop, forever) for new message on a bus
+ * 
+ * @param   bus        Bus information
+ * @param   callback   Function to call when a message is received, the
+ *                     input parameters will be the read message and
+ *                     `user_data` from `bus_read`'s parameter with the
+ *                     same name. The message must have been parsed or
+ *                     copied when `callback` returns as it may be over
+ *                     overridden after that time. `callback` should
+ *                     return either of the the values:
+ *                       *  0:  stop listening
+ *                       *  1:  continue listening
+ *                       * -1:  an error has occurred
+ *                     However, the function [`bus_read`] will invoke
+ *                     `callback` with `message` set to `NULL`one time
+ *                     directly after it has started listening on the
+ *                     bus. This is to the the program now it can safely
+ *                     continue with any action that requires that the
+ *                     programs is listening on the bus.
+ * @param   user_data  Parameter passed to `callback`
+ * @param   timeout    The time the operation shall fail with errno set
+ *                     to `EAGAIN` if not completed, note that the callback
+ *                     function may or may not have been called
+ * @param   clockid    The ID of the clock the `timeout` is measured with,
+ *                     it most be a predictable clock
+ * @return             0 on success, -1 on error
+ */
+int bus_read_timed(const bus_t *restrict bus, int (*callback)(const char *message, void *user_data),
+                   void *user_data, const struct timespec *timeout, clockid_t clockid)
+{
+	int r, state = 0, saved_errno;
+	struct timespec delta;
+	if (!timeout)
+		return bus_read(bus, callback, user_data);
+
+	DELTA;
+	if (release_semaphore_timed(bus, S, SEM_UNDO, &delta) == -1)
+		return -1;
+	t(r = callback(NULL, user_data));
+	if (!r)  goto done;
+	for (;;) {
+		DELTA;
+		t(release_semaphore_timed(bus, Q, 0, &delta));
+		DELTA;
+		t(zero_semaphore_timed(bus, Q, 0, &delta));
+		t(r = callback(bus->message, user_data));
+		if (!r)  goto done;
+		t(release_semaphore(bus, W, SEM_UNDO));  state++;
+		t(acquire_semaphore(bus, S, SEM_UNDO));  state++;
+		t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+		t(zero_semaphore(bus, N, 0));
+#endif
+		t(release_semaphore(bus, S, SEM_UNDO));  state--;
+		t(acquire_semaphore(bus, W, SEM_UNDO));  state--;
+	}
+
+fail:
+	saved_errno = errno;
+	if (state > 1)
+		release_semaphore(bus, S, SEM_UNDO);
+	if (state > 0)
+		acquire_semaphore(bus, W, SEM_UNDO);
+	acquire_semaphore(bus, S, SEM_UNDO);
+	errno = saved_errno;
+	return -1;
+
+done:
+	t(acquire_semaphore(bus, S, SEM_UNDO));
+	return 0;
+}
+
+
+/**
+ * Announce that the thread is listening on the bus.
+ * This is required so the will does not miss any
+ * messages due to race conditions. Additionally,
+ * not calling this function will cause the bus the
+ * misbehave, is `bus_poll` is written to expect
+ * this function to have been called.
+ * 
+ * @param   bus  Bus information
+ * @return       0 on success, -1 on error
+ */
+int
+bus_poll_start(bus_t *bus)
+{
+	bus->first_poll = 1;
+	t(release_semaphore(bus, S, SEM_UNDO));
+	t(release_semaphore(bus, Q, 0));
+	return 0;
+
+fail:
+	return -1;
+}
+
+
+/**
+ * Announce that the thread has stopped listening on the bus.
+ * This is required so that the thread does not cause others
+ * to wait indefinitely.
+ * 
+ * @param   bus  Bus information
+ * @return       0 on success, -1 on error
+ */
+int
+bus_poll_stop(const bus_t *bus)
+{
+	return acquire_semaphore(bus, S, SEM_UNDO | IPC_NOWAIT);
+}
+
+
+/**
+ * Wait for a message to be broadcasted on the bus.
+ * The caller should make a copy of the received message,
+ * without freeing the original copy, and parse it in a
+ * separate thread. When the new thread has started be
+ * started, the caller of this function should then
+ * either call `bus_poll` again or `bus_poll_stop`.
+ * 
+ * @param   bus    Bus information
+ * @param   flags  `BUS_NOWAIT` if the bus should fail and set `errno` to
+ *                 `EAGAIN` if there isn't already a message available on the bus
+ * @return         The received message, `NULL` on error
+ */
+const char *
+bus_poll(bus_t *bus, int flags)
+{
+	int state = 0, saved_errno;
+	if (!bus->first_poll) {
+		t(release_semaphore(bus, W, SEM_UNDO));  state++;
+		t(acquire_semaphore(bus, S, SEM_UNDO));  state++;
+		t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+		t(zero_semaphore(bus, N, 0));
+#endif
+		t(release_semaphore(bus, S, SEM_UNDO));  state--;
+		t(acquire_semaphore(bus, W, SEM_UNDO));  state--;
+		t(release_semaphore(bus, Q, 0));
+	} else {
+		bus->first_poll = 0;
+	}
+	state--;
+	t(zero_semaphore(bus, Q, F(BUS_NOWAIT, IPC_NOWAIT)));
+	return bus->message;
+
+fail:
+	saved_errno = errno;
+	if (state > 1)
+		release_semaphore(bus, S, SEM_UNDO);
+	if (state > 0)
+		acquire_semaphore(bus, W, SEM_UNDO);
+	if (state < 0)
+		bus->first_poll = 1;
+	errno = saved_errno;
+	return NULL;
+}
+
+
+/**
+ * Wait for a message to be broadcasted on the bus.
+ * The caller should make a copy of the received message,
+ * without freeing the original copy, and parse it in a
+ * separate thread. When the new thread has started be
+ * started, the caller of this function should then
+ * either call `bus_poll_timed` again or `bus_poll_stop`.
+ * 
+ * @param   bus      Bus information
+ * @param   timeout  The time the operation shall fail with errno set
+ *                   to `EAGAIN` if not completed
+ * @param   clockid  The ID of the clock the `timeout` is measured with,
+ *                   it most be a predictable clock
+ * @return           The received message, `NULL` on error
+ */
+const char *bus_poll_timed(bus_t *bus, const struct timespec *timeout, clockid_t clockid)
+{
+	int state = 0, saved_errno;
+	struct timespec delta;
+	if (!timeout)
+		return bus_poll(bus, 0);
+
+	if (!bus->first_poll) {
+		t(release_semaphore(bus, W, SEM_UNDO));  state++;
+		t(acquire_semaphore(bus, S, SEM_UNDO));  state++;
+		t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+		t(zero_semaphore(bus, N, 0));
+#endif
+		t(release_semaphore(bus, S, SEM_UNDO));  state--;
+		t(acquire_semaphore(bus, W, SEM_UNDO));  state--;
+		t(release_semaphore(bus, Q, 0));
+	} else {
+		bus->first_poll = 0;
+	}
+	state--;
+	DELTA;
+	t(zero_semaphore_timed(bus, Q, 0, &delta));
+	return bus->message;
+
+fail:
+	saved_errno = errno;
+	if (state > 1)
+		release_semaphore(bus, S, SEM_UNDO);
+	if (state > 0)
+		acquire_semaphore(bus, W, SEM_UNDO);
+	if (state < 0)
+		bus->first_poll = 1;
+	errno = saved_errno;
+	return NULL;
+}
+
+
+/**
+ * Change the ownership of a bus
+ * 
+ * `stat(2)` can be used of the bus's associated file to get the bus's ownership
+ * 
+ * @param   file   The pathname of the bus
+ * @param   owner  The user ID of the bus's new owner
+ * @param   group  The group ID of the bus's new group
+ * @return         0 on success, -1 on error
+ */
+int
+bus_chown(const char *file, uid_t owner, gid_t group)
+{
+	bus_t bus;
+	struct semid_ds sem_stat;
+	struct shmid_ds shm_stat;
+	int shm_id;
+
+	t(bus_open(&bus, file, -1));
+	t(chown(file, owner, group));
+
+	/* chown sem */
+	t(open_semaphores(&bus));
+	t(semctl(bus.sem_id, 0, IPC_STAT, &sem_stat));
+	sem_stat.sem_perm.uid = owner;
+	sem_stat.sem_perm.gid = group;
+	t(semctl(bus.sem_id, 0, IPC_SET, &sem_stat));
+
+	/* chown shm */
+	t(shm_id = shmget(bus.key_shm, (size_t)BUS_MEMORY_SIZE, 0));
+	t(shmctl(shm_id, IPC_STAT, &shm_stat));
+	shm_stat.shm_perm.uid = owner;
+	shm_stat.shm_perm.gid = group;
+	t(shmctl(shm_id, IPC_SET, &shm_stat));
+
+	return 0;
+fail:
+	return -1;
+}
+
+
+/**
+ * Change the permissions for a bus
+ * 
+ * `stat(2)` can be used of the bus's associated file to get the bus's permissions
+ * 
+ * @param   file  The pathname of the bus
+ * @param   mode  The permissions of the bus, any permission for a user implies
+ *                full permissions for that user, except only the owner may
+ *                edit the bus's associated file
+ * @return        0 on success, -1 on error
+ */
+int
+bus_chmod(const char *file, mode_t mode)
+{
+	bus_t bus;
+	mode_t fmode;
+	struct semid_ds sem_stat;
+	struct shmid_ds shm_stat;
+	int shm_id;
+
+	mode = (mode & S_IRWXU) ? (mode | S_IRWXU) : (mode & (mode_t)~S_IRWXU);
+	mode = (mode & S_IRWXG) ? (mode | S_IRWXG) : (mode & (mode_t)~S_IRWXG);
+	mode = (mode & S_IRWXO) ? (mode | S_IRWXO) : (mode & (mode_t)~S_IRWXO);
+	mode &= (S_IWUSR | S_IWGRP | S_IWOTH | S_IRUSR | S_IRGRP | S_IROTH);
+	fmode = mode & (mode_t)~(S_IWGRP | S_IWOTH);
+
+	t(bus_open(&bus, file, -1));
+	t(chmod(file, fmode));
+
+	/* chmod sem */
+	t(open_semaphores(&bus));
+	t(semctl(bus.sem_id, 0, IPC_STAT, &sem_stat));
+	sem_stat.sem_perm.mode = (unsigned short)mode;
+	t(semctl(bus.sem_id, 0, IPC_SET, &sem_stat));
+
+	/* chmod shm */
+	t(shm_id = shmget(bus.key_shm, (size_t)BUS_MEMORY_SIZE, 0));
+	t(shmctl(shm_id, IPC_STAT, &shm_stat));
+	shm_stat.shm_perm.mode = (unsigned short)mode;
+	t(shmctl(shm_id, IPC_SET, &shm_stat));
+
+	return 0;
+fail:
+	return -1;
+}