From e0896142dbbbadd87a6339111a406307edca1b97 Mon Sep 17 00:00:00 2001
From: Romain Vimont <rom@rom1v.com>
Date: Fri, 12 Nov 2021 18:50:50 +0100
Subject: [PATCH] Introduce interruptor tool

An interruptor instance will help to wake up a blocking call from
another thread (typically to terminate immediately on Ctrl+C).
---
 app/meson.build     |  1 +
 app/src/util/intr.c | 83 +++++++++++++++++++++++++++++++++++++++++++++
 app/src/util/intr.h | 78 ++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 162 insertions(+)
 create mode 100644 app/src/util/intr.c
 create mode 100644 app/src/util/intr.h

diff --git a/app/meson.build b/app/meson.build
index 94b4994f..8240dd16 100644
--- a/app/meson.build
+++ b/app/meson.build
@@ -25,6 +25,7 @@ src = [
     'src/stream.c',
     'src/video_buffer.c',
     'src/util/file.c',
+    'src/util/intr.c',
     'src/util/log.c',
     'src/util/net.c',
     'src/util/process.c',
diff --git a/app/src/util/intr.c b/app/src/util/intr.c
new file mode 100644
index 00000000..d4fce55c
--- /dev/null
+++ b/app/src/util/intr.c
@@ -0,0 +1,83 @@
+#include "intr.h"
+
+#include "util/log.h"
+
+#include <assert.h>
+
+bool
+sc_intr_init(struct sc_intr *intr) {
+    bool ok = sc_mutex_init(&intr->mutex);
+    if (!ok) {
+        LOGE("Could not init intr mutex");
+        return false;
+    }
+
+    intr->socket = SC_INVALID_SOCKET;
+    intr->process = SC_PROCESS_NONE;
+
+    atomic_store_explicit(&intr->interrupted, false, memory_order_relaxed);
+
+    return true;
+}
+
+bool
+sc_intr_set_socket(struct sc_intr *intr, sc_socket socket) {
+    assert(intr->process == SC_PROCESS_NONE);
+
+    sc_mutex_lock(&intr->mutex);
+    bool interrupted =
+        atomic_load_explicit(&intr->interrupted, memory_order_relaxed);
+    if (!interrupted) {
+        intr->socket = socket;
+    }
+    sc_mutex_unlock(&intr->mutex);
+
+    return !interrupted;
+}
+
+bool
+sc_intr_set_process(struct sc_intr *intr, sc_pid pid) {
+    assert(intr->socket == SC_INVALID_SOCKET);
+
+    sc_mutex_lock(&intr->mutex);
+    bool interrupted =
+        atomic_load_explicit(&intr->interrupted, memory_order_relaxed);
+    if (!interrupted) {
+        intr->process = pid;
+    }
+    sc_mutex_unlock(&intr->mutex);
+
+    return !interrupted;
+}
+
+void
+sc_intr_interrupt(struct sc_intr *intr) {
+    sc_mutex_lock(&intr->mutex);
+
+    atomic_store_explicit(&intr->interrupted, true, memory_order_relaxed);
+
+    // No more than one component to interrupt
+    assert(intr->socket == SC_INVALID_SOCKET ||
+           intr->process == SC_PROCESS_NONE);
+
+    if (intr->socket != SC_INVALID_SOCKET) {
+        LOGD("Interrupting socket");
+        net_interrupt(intr->socket);
+        intr->socket = SC_INVALID_SOCKET;
+    }
+    if (intr->process != SC_PROCESS_NONE) {
+        LOGD("Interrupting process");
+        sc_process_terminate(intr->process);
+        intr->process = SC_PROCESS_NONE;
+    }
+
+    sc_mutex_unlock(&intr->mutex);
+}
+
+void
+sc_intr_destroy(struct sc_intr *intr) {
+    assert(intr->socket == SC_INVALID_SOCKET);
+    assert(intr->process == SC_PROCESS_NONE);
+
+    sc_mutex_destroy(&intr->mutex);
+}
diff --git a/app/src/util/intr.h b/app/src/util/intr.h
new file mode 100644
index 00000000..7f0fb9b7
--- /dev/null
+++ b/app/src/util/intr.h
@@ -0,0 +1,78 @@
+#ifndef SC_INTR_H
+#define SC_INTR_H
+
+#include "common.h"
+
+#include <stdatomic.h>
+#include <stdbool.h>
+
+#include "net.h"
+#include "process.h"
+#include "thread.h"
+
+/**
+ * Interruptor to wake up a blocking call from another thread
+ *
+ * It allows to register a socket or a process before a blocking call, and
+ * interrupt/close from another thread to wake up the blocking call.
+ */
+struct sc_intr {
+    sc_mutex mutex;
+
+    sc_socket socket;
+    sc_pid process;
+
+    // Written protected by the mutex to avoid race conditions against
+    // sc_intr_set_socket() and sc_intr_set_process(), but can be read
+    // (atomically) without mutex
+    atomic_bool interrupted;
+};
+
+/**
+ * Initialize an interruptor
+ */
+bool
+sc_intr_init(struct sc_intr *intr);
+
+/**
+ * Set a socket as the interruptible component
+ *
+ * Call with SC_INVALID_SOCKET to unset.
+ */
+bool
+sc_intr_set_socket(struct sc_intr *intr, sc_socket socket);
+
+/**
+ * Set a process as the interruptible component
+ *
+ * Call with SC_PROCESS_NONE to unset.
+ */
+bool
+sc_intr_set_process(struct sc_intr *intr, sc_pid socket);
+
+/**
+ * Interrupt the current interruptible component
+ *
+ * Must be called from a different thread.
+ */
+void
+sc_intr_interrupt(struct sc_intr *intr);
+
+/**
+ * Read the interrupted state
+ *
+ * It is exposed as a static inline function because it just loads from an
+ * atomic.
+ */
+static inline bool
+sc_intr_is_interrupted(struct sc_intr *intr) {
+    return atomic_load_explicit(&intr->interrupted, memory_order_relaxed);
+}
+
+/**
+ * Destroy the interruptor
+ */
+void
+sc_intr_destroy(struct sc_intr *intr);
+
+#endif