1 files changed, 394 insertions, 0 deletions
diff --git a/src/shared/barrier.c b/src/shared/barrier.c
new file mode 100644
index 0000000..bd5bdd7
--- /dev/null
+++ b/src/shared/barrier.c
@@ -0,0 +1,394 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include <errno.h>
+#include <fcntl.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <sys/eventfd.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include "barrier.h"
+#include "errno-util.h"
+#include "fd-util.h"
+#include "io-util.h"
+#include "macro.h"
+
+/**
+ * Barriers
+ * This barrier implementation provides a simple synchronization method based
+ * on file-descriptors that can safely be used between threads and processes. A
+ * barrier object contains 2 shared counters based on eventfd. Both processes
+ * can now place barriers and wait for the other end to reach a random or
+ * specific barrier.
+ * Barriers are numbered, so you can either wait for the other end to reach any
+ * barrier or the last barrier that you placed. This way, you can use barriers
+ * for one-way *and* full synchronization. Note that even-though barriers are
+ * numbered, these numbers are internal and recycled once both sides reached the
+ * same barrier (implemented as a simple signed counter). It is thus not
+ * possible to address barriers by their ID.
+ *
+ * Barrier-API: Both ends can place as many barriers via barrier_place() as
+ * they want and each pair of barriers on both sides will be implicitly linked.
+ * Each side can use the barrier_wait/sync_*() family of calls to wait for the
+ * other side to place a specific barrier. barrier_wait_next() waits until the
+ * other side calls barrier_place(). No links between the barriers are
+ * considered and this simply serves as most basic asynchronous barrier.
+ * barrier_sync_next() is like barrier_wait_next() and waits for the other side
+ * to place their next barrier via barrier_place(). However, it only waits for
+ * barriers that are linked to a barrier we already placed. If the other side
+ * already placed more barriers than we did, barrier_sync_next() returns
+ * immediately.
+ * barrier_sync() extends barrier_sync_next() and waits until the other end
+ * placed as many barriers via barrier_place() as we did. If they already placed
+ * as many as we did (or more), it returns immediately.
+ *
+ * Additionally to basic barriers, an abortion event is available.
+ * barrier_abort() places an abortion event that cannot be undone. An abortion
+ * immediately cancels all placed barriers and replaces them. Any running and
+ * following wait/sync call besides barrier_wait_abortion() will immediately
+ * return false on both sides (otherwise, they always return true).
+ * barrier_abort() can be called multiple times on both ends and will be a
+ * no-op if already called on this side.
+ * barrier_wait_abortion() can be used to wait for the other side to call
+ * barrier_abort() and is the only wait/sync call that does not return
+ * immediately if we aborted outself. It only returns once the other side
+ * called barrier_abort().
+ *
+ * Barriers can be used for in-process and inter-process synchronization.
+ * However, for in-process synchronization you could just use mutexes.
+ * Therefore, main target is IPC and we require both sides to *not* share the FD
+ * table. If that's given, barriers provide target tracking: If the remote side
+ * exit()s, an abortion event is implicitly queued on the other side. This way,
+ * a sync/wait call will be woken up if the remote side crashed or exited
+ * unexpectedly. However, note that these abortion events are only queued if the
+ * barrier-queue has been drained. Therefore, it is safe to place a barrier and
+ * exit. The other side can safely wait on the barrier even though the exit
+ * queued an abortion event. Usually, the abortion event would overwrite the
+ * barrier, however, that's not true for exit-abortion events. Those are only
+ * queued if the barrier-queue is drained (thus, the receiving side has placed
+ * more barriers than the remote side).
+ */
+
+/**
+ * barrier_create() - Initialize a barrier object
+ * @obj: barrier to initialize
+ *
+ * This initializes a barrier object. The caller is responsible of allocating
+ * the memory and keeping it valid. The memory does not have to be zeroed
+ * beforehand.
+ * Two eventfd objects are allocated for each barrier. If allocation fails, an
+ * error is returned.
+ *
+ * If this function fails, the barrier is reset to an invalid state so it is
+ * safe to call barrier_destroy() on the object regardless whether the
+ * initialization succeeded or not.
+ *
+ * The caller is responsible to destroy the object via barrier_destroy() before
+ * releasing the underlying memory.
+ *
+ * Returns: 0 on success, negative error code on failure.
+ */
+int barrier_create(Barrier *b) {
+        _unused_ _cleanup_(barrier_destroyp) Barrier *staging = b;
+
+        assert(b);
+
+        b->me = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
+        if (b->me < 0)
+                return -errno;
+
+        b->them = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
+        if (b->them < 0)
+                return -errno;
+
+        if (pipe2(b->pipe, O_CLOEXEC | O_NONBLOCK) < 0)
+                return -errno;
+
+        staging = NULL;
+        return 0;
+}
+
+/**
+ * barrier_destroy() - Destroy a barrier object
+ * @b: barrier to destroy or NULL
+ *
+ * This destroys a barrier object that has previously been passed to
+ * barrier_create(). The object is released and reset to invalid
+ * state. Therefore, it is safe to call barrier_destroy() multiple
+ * times or even if barrier_create() failed. However, barrier must be
+ * always initialized with BARRIER_NULL.
+ *
+ * If @b is NULL, this is a no-op.
+ */
+Barrier* barrier_destroy(Barrier *b) {
+        if (!b)
+                return NULL;
+
+        b->me = safe_close(b->me);
+        b->them = safe_close(b->them);
+        safe_close_pair(b->pipe);
+        b->barriers = 0;
+        return NULL;
+}
+
+/**
+ * barrier_set_role() - Set the local role of the barrier
+ * @b: barrier to operate on
+ * @role: role to set on the barrier
+ *
+ * This sets the roles on a barrier object. This is needed to know
+ * which side of the barrier you're on. Usually, the parent creates
+ * the barrier via barrier_create() and then calls fork() or clone().
+ * Therefore, the FDs are duplicated and the child retains the same
+ * barrier object.
+ *
+ * Both sides need to call barrier_set_role() after fork() or clone()
+ * are done. If this is not done, barriers will not work correctly.
+ *
+ * Note that barriers could be supported without fork() or clone(). However,
+ * this is currently not needed so it hasn't been implemented.
+ */
+void barrier_set_role(Barrier *b, unsigned role) {
+        assert(b);
+        assert(IN_SET(role, BARRIER_PARENT, BARRIER_CHILD));
+        /* make sure this is only called once */
+        assert(b->pipe[0] >= 0 && b->pipe[1] >= 0);
+
+        if (role == BARRIER_PARENT)
+                b->pipe[1] = safe_close(b->pipe[1]);
+        else {
+                b->pipe[0] = safe_close(b->pipe[0]);
+
+                /* swap me/them for children */
+                SWAP_TWO(b->me, b->them);
+        }
+}
+
+/* places barrier; returns false if we aborted, otherwise true */
+static bool barrier_write(Barrier *b, uint64_t buf) {
+        ssize_t len;
+
+        /* prevent new sync-points if we already aborted */
+        if (barrier_i_aborted(b))
+                return false;
+
+        assert(b->me >= 0);
+        do
+                len = write(b->me, &buf, sizeof(buf));
+        while (len < 0 && ERRNO_IS_TRANSIENT(errno));
+
+        if (len != sizeof(buf))
+                goto error;
+
+        /* lock if we aborted */
+        if (buf >= (uint64_t)BARRIER_ABORTION) {
+                if (barrier_they_aborted(b))
+                        b->barriers = BARRIER_WE_ABORTED;
+                else
+                        b->barriers = BARRIER_I_ABORTED;
+        } else if (!barrier_is_aborted(b))
+                b->barriers += buf;
+
+        return !barrier_i_aborted(b);
+
+error:
+        /* If there is an unexpected error, we have to make this fatal. There
+         * is no way we can recover from sync-errors. Therefore, we close the
+         * pipe-ends and treat this as abortion. The other end will notice the
+         * pipe-close and treat it as abortion, too. */
+
+        safe_close_pair(b->pipe);
+        b->barriers = BARRIER_WE_ABORTED;
+        return false;
+}
+
+/* waits for barriers; returns false if they aborted, otherwise true */
+static bool barrier_read(Barrier *b, int64_t comp) {
+        if (barrier_they_aborted(b))
+                return false;
+
+        while (b->barriers > comp) {
+                struct pollfd pfd[2] = {
+                        { .fd = b->pipe[0] >= 0 ? b->pipe[0] : b->pipe[1],
+                          .events = POLLHUP },
+                        { .fd = b->them,
+                          .events = POLLIN }};
+                uint64_t buf;
+                int r;
+
+                r = ppoll_usec(pfd, ELEMENTSOF(pfd), USEC_INFINITY);
+                if (r == -EINTR)
+                        continue;
+                if (r < 0)
+                        goto error;
+
+                if (pfd[1].revents) {
+                        ssize_t len;
+
+                        /* events on @them signal new data for us */
+                        len = read(b->them, &buf, sizeof(buf));
+                        if (len < 0 && ERRNO_IS_TRANSIENT(errno))
+                                continue;
+
+                        if (len != sizeof(buf))
+                                goto error;
+                } else if (pfd[0].revents & (POLLHUP | POLLERR | POLLNVAL))
+                        /* POLLHUP on the pipe tells us the other side exited.
+                         * We treat this as implicit abortion. But we only
+                         * handle it if there's no event on the eventfd. This
+                         * guarantees that exit-abortions do not overwrite real
+                         * barriers. */
+                        buf = BARRIER_ABORTION;
+                else
+                        continue;
+
+                /* lock if they aborted */
+                if (buf >= (uint64_t)BARRIER_ABORTION) {
+                        if (barrier_i_aborted(b))
+                                b->barriers = BARRIER_WE_ABORTED;
+                        else
+                                b->barriers = BARRIER_THEY_ABORTED;
+                } else if (!barrier_is_aborted(b))
+                        b->barriers -= buf;
+        }
+
+        return !barrier_they_aborted(b);
+
+error:
+        /* If there is an unexpected error, we have to make this fatal. There
+         * is no way we can recover from sync-errors. Therefore, we close the
+         * pipe-ends and treat this as abortion. The other end will notice the
+         * pipe-close and treat it as abortion, too. */
+
+        safe_close_pair(b->pipe);
+        b->barriers = BARRIER_WE_ABORTED;
+        return false;
+}
+
+/**
+ * barrier_place() - Place a new barrier
+ * @b: barrier object
+ *
+ * This places a new barrier on the barrier object. If either side already
+ * aborted, this is a no-op and returns "false". Otherwise, the barrier is
+ * placed and this returns "true".
+ *
+ * Returns: true if barrier was placed, false if either side aborted.
+ */
+bool barrier_place(Barrier *b) {
+        assert(b);
+
+        if (barrier_is_aborted(b))
+                return false;
+
+        barrier_write(b, BARRIER_SINGLE);
+        return true;
+}
+
+/**
+ * barrier_abort() - Abort the synchronization
+ * @b: barrier object to abort
+ *
+ * This aborts the barrier-synchronization. If barrier_abort() was already
+ * called on this side, this is a no-op. Otherwise, the barrier is put into the
+ * ABORT-state and will stay there. The other side is notified about the
+ * abortion. Any following attempt to place normal barriers or to wait on normal
+ * barriers will return immediately as "false".
+ *
+ * You can wait for the other side to call barrier_abort(), too. Use
+ * barrier_wait_abortion() for that.
+ *
+ * Returns: false if the other side already aborted, true otherwise.
+ */
+bool barrier_abort(Barrier *b) {
+        assert(b);
+
+        barrier_write(b, BARRIER_ABORTION);
+        return !barrier_they_aborted(b);
+}
+
+/**
+ * barrier_wait_next() - Wait for the next barrier of the other side
+ * @b: barrier to operate on
+ *
+ * This waits until the other side places its next barrier. This is independent
+ * of any barrier-links and just waits for any next barrier of the other side.
+ *
+ * If either side aborted, this returns false.
+ *
+ * Returns: false if either side aborted, true otherwise.
+ */
+bool barrier_wait_next(Barrier *b) {
+        assert(b);
+
+        if (barrier_is_aborted(b))
+                return false;
+
+        barrier_read(b, b->barriers - 1);
+        return !barrier_is_aborted(b);
+}
+
+/**
+ * barrier_wait_abortion() - Wait for the other side to abort
+ * @b: barrier to operate on
+ *
+ * This waits until the other side called barrier_abort(). This can be called
+ * regardless whether the local side already called barrier_abort() or not.
+ *
+ * If the other side has already aborted, this returns immediately.
+ *
+ * Returns: false if the local side aborted, true otherwise.
+ */
+bool barrier_wait_abortion(Barrier *b) {
+        assert(b);
+
+        barrier_read(b, BARRIER_THEY_ABORTED);
+        return !barrier_i_aborted(b);
+}
+
+/**
+ * barrier_sync_next() - Wait for the other side to place a next linked barrier
+ * @b: barrier to operate on
+ *
+ * This is like barrier_wait_next() and waits for the other side to call
+ * barrier_place(). However, this only waits for linked barriers. That means, if
+ * the other side already placed more barriers than (or as much as) we did, this
+ * returns immediately instead of waiting.
+ *
+ * If either side aborted, this returns false.
+ *
+ * Returns: false if either side aborted, true otherwise.
+ */
+bool barrier_sync_next(Barrier *b) {
+        assert(b);
+
+        if (barrier_is_aborted(b))
+                return false;
+
+        barrier_read(b, MAX((int64_t)0, b->barriers - 1));
+        return !barrier_is_aborted(b);
+}
+
+/**
+ * barrier_sync() - Wait for the other side to place as many barriers as we did
+ * @b: barrier to operate on
+ *
+ * This is like barrier_sync_next() but waits for the other side to call
+ * barrier_place() as often as we did (in total). If they already placed as much
+ * as we did (or more), this returns immediately instead of waiting.
+ *
+ * If either side aborted, this returns false.
+ *
+ * Returns: false if either side aborted, true otherwise.
+ */
+bool barrier_sync(Barrier *b) {
+        assert(b);
+
+        if (barrier_is_aborted(b))
+                return false;
+
+        barrier_read(b, 0);
+        return !barrier_is_aborted(b);
+}