1 files changed, 215 insertions, 4 deletions

diff --git a/src/platform/posix/loop.c b/src/platform/posix/loop.c
index 88454d8..3c3d730 100644
--- a/src/platform/posix/loop.c
+++ b/src/platform/posix/loop.c
@@ -1,10 +1,14 @@
 #define _GNU_SOURCE
+#define _DARWIN_C_SOURCE
 #define _POSIX_C_SOURCE 200809L
 #include "common/loop.h"
+#include "platform/sync.h"
 
 #include <errno.h>
 #include <fcntl.h>
 #include <poll.h>
+#include <signal.h>
+#include <stdatomic.h>
 #include <stdlib.h>
 #include <string.h>
 #include <time.h>
@@ -31,6 +35,21 @@ struct yasl_timer_source {
     bool deleted;
 };
 
+struct yasl_post {
+    void (*fn)(void *ud);
+    void *ud;
+    struct yasl_post *next;
+};
+
+/* Signals this loop watches; the previous actions are restored on unwatch. */
+struct yasl_signal_watch {
+    void (*cb)(void *ud, int signo);
+    void *ud;
+    int count;
+    int signals[NSIG];
+    struct sigaction old[NSIG];
+};
+
 struct yasl_loop {
     struct yasl_fd_source *fds;
     struct pollfd *pollfds;
@@ -47,8 +66,21 @@ struct yasl_loop {
     int timer_capacity;
     uint32_t next_timer_id;
     int timer_firing_depth;
+
+    yasl_mutex post_mutex;
+    struct yasl_post *post_head;
+    struct yasl_post *post_tail;
+
+    struct yasl_signal_watch *signals;
 };
 
+/* Handlers run on any thread, so they only touch lock-free atomics: they mark
+ * the signal pending and write to the owning loop's wake pipe. The pipe is
+ * stored as fd + 1 so that the zero initializer means no pipe. */
+static atomic_int signal_wake_fd[NSIG];
+static atomic_bool signal_pending[NSIG];
+static struct yasl_loop *signal_owner[NSIG];
+
 uint64_t yasl_loop_now_ns(void)
 {
     struct timespec ts;
@@ -56,13 +88,41 @@ uint64_t yasl_loop_now_ns(void)
     return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec;
 }
 
+static void run_posts(struct yasl_loop *l)
+{
+    yasl_mutex_lock(&l->post_mutex);
+    struct yasl_post *p = l->post_head;
+    l->post_head = l->post_tail = NULL;
+    yasl_mutex_unlock(&l->post_mutex);
+    while (p) {
+        struct yasl_post *next = p->next;
+        p->fn(p->ud);
+        free(p);
+        p = next;
+    }
+}
+
+static void run_signals(struct yasl_loop *l)
+{
+    /* A callback may unwatch, so re-check the watch before each delivery. */
+    for (int signo = 1; signo < NSIG; signo++) {
+        struct yasl_signal_watch *w = l->signals;
+        if (!w) return;
+        if (signal_owner[signo] != l || !atomic_exchange(&signal_pending[signo], false))
+            continue;
+        w->cb(w->ud, signo);
+    }
+}
+
 static int loop_wake_cb(int fd, short revents, void *ud)
 {
     (void)revents;
-    (void)ud;
     char buf[128];
     ssize_t n;
     while ((n = read(fd, buf, sizeof(buf))) > 0 || (n < 0 && errno == EINTR)) {}
+    struct yasl_loop *l = ud;
+    run_posts(l);
+    run_signals(l);
     return 0;
 }
 
@@ -81,6 +141,13 @@ struct yasl_loop *yasl_loop_create(void)
         return NULL;
     }
 
+    if (yasl_mutex_init(&l->post_mutex) != 0) {
+        free(l->fds);
+        free(l->pollfds);
+        free(l);
+        return NULL;
+    }
+
     l->next_id = 1;
     l->next_timer_id = 1;
     l->wake_rfd = -1;
@@ -114,6 +181,13 @@ struct yasl_loop *yasl_loop_create(void)
 void yasl_loop_destroy(struct yasl_loop *l)
 {
     if (!l) return;
+    yasl_loop_unwatch_signals(l);
+    for (struct yasl_post *p = l->post_head; p;) {
+        struct yasl_post *next = p->next;
+        free(p);
+        p = next;
+    }
+    yasl_mutex_destroy(&l->post_mutex);
     if (l->wake_watch > 0)
         yasl_loop_del_fd(l, l->wake_watch);
     if (l->wake_rfd >= 0)
@@ -184,14 +258,141 @@ int yasl_loop_set_fd_events(struct yasl_loop *l, int id, short events)
     return -1;
 }
 
-void yasl_loop_kick(struct yasl_loop *l)
+static void wake_write(int fd)
 {
-    if (!l || l->wake_wfd < 0) return;
+    /* A full pipe already wakes the loop, so EAGAIN needs no retry. */
+    int saved = errno;
     char b = 1;
     ssize_t n;
     do {
-        n = write(l->wake_wfd, &b, 1);
+        n = write(fd, &b, 1);
     } while (n < 0 && errno == EINTR);
+    errno = saved;
+}
+
+void yasl_loop_kick(struct yasl_loop *l)
+{
+    if (!l || l->wake_wfd < 0) return;
+    wake_write(l->wake_wfd);
+}
+
+bool yasl_loop_post(struct yasl_loop *l, void (*fn)(void *ud), void *ud)
+{
+    if (!l || !fn || l->wake_wfd < 0) return false;
+    struct yasl_post *p = malloc(sizeof(*p));
+    if (!p) return false;
+    *p = (struct yasl_post){.fn = fn, .ud = ud};
+    yasl_mutex_lock(&l->post_mutex);
+    if (l->post_tail)
+        l->post_tail->next = p;
+    else
+        l->post_head = p;
+    l->post_tail = p;
+    yasl_mutex_unlock(&l->post_mutex);
+    wake_write(l->wake_wfd);
+    return true;
+}
+
+static void signal_handler(int signo)
+{
+    atomic_store(&signal_pending[signo], true);
+    int fd = atomic_load(&signal_wake_fd[signo]) - 1;
+    if (fd >= 0) wake_write(fd);
+}
+
+static bool signal_valid(int signo)
+{
+    return signo > 0 && signo < NSIG && signo != SIGKILL && signo != SIGSTOP;
+}
+
+bool yasl_loop_watch_signals(struct yasl_loop *l, const int *signals, int n,
+                             void (*cb)(void *ud, int signo), void *ud)
+{
+    if (!l || !signals || n <= 0 || !cb || l->wake_wfd < 0) {
+        errno = EINVAL;
+        return false;
+    }
+    bool wanted[NSIG] = {0};
+    for (int i = 0; i < n; i++) {
+        if (!signal_valid(signals[i])) {
+            errno = EINVAL;
+            return false;
+        }
+        if (signal_owner[signals[i]] && signal_owner[signals[i]] != l) {
+            errno = EBUSY;
+            return false;
+        }
+        wanted[signals[i]] = true;
+    }
+    struct yasl_signal_watch *w = calloc(1, sizeof(*w));
+    if (!w) return false;
+    w->cb = cb;
+    w->ud = ud;
+
+    /* Keep the old watch's saved actions for signals that stay watched, so
+     * that unwatch restores what the application had before any watch. */
+    struct yasl_signal_watch *old = l->signals;
+    struct sigaction sa = {.sa_handler = signal_handler, .sa_flags = SA_RESTART};
+    sigemptyset(&sa.sa_mask);
+    for (int signo = 1; signo < NSIG; signo++) {
+        if (!wanted[signo]) continue;
+        int prev = -1;
+        for (int i = 0; old && i < old->count; i++)
+            if (old->signals[i] == signo) prev = i;
+        if (prev >= 0) {
+            w->old[w->count] = old->old[prev];
+        } else {
+            atomic_store(&signal_pending[signo], false);
+            atomic_store(&signal_wake_fd[signo], l->wake_wfd + 1);
+            if (sigaction(signo, &sa, &w->old[w->count]) != 0) {
+                int err = errno;
+                atomic_store(&signal_wake_fd[signo], 0);
+                for (int i = 0; i < w->count; i++) {
+                    int done = w->signals[i];
+                    bool kept = false;
+                    for (int j = 0; old && j < old->count; j++)
+                        if (old->signals[j] == done) kept = true;
+                    if (kept) continue;
+                    sigaction(done, &w->old[i], NULL);
+                    atomic_store(&signal_wake_fd[done], 0);
+                }
+                free(w);
+                errno = err;
+                return false;
+            }
+        }
+        w->signals[w->count++] = signo;
+    }
+
+    /* Restore signals the old watch had that the new one drops. */
+    for (int i = 0; old && i < old->count; i++) {
+        int signo = old->signals[i];
+        if (wanted[signo]) continue;
+        sigaction(signo, &old->old[i], NULL);
+        atomic_store(&signal_wake_fd[signo], 0);
+        atomic_store(&signal_pending[signo], false);
+        signal_owner[signo] = NULL;
+    }
+    for (int i = 0; i < w->count; i++)
+        signal_owner[w->signals[i]] = l;
+    free(old);
+    l->signals = w;
+    return true;
+}
+
+void yasl_loop_unwatch_signals(struct yasl_loop *l)
+{
+    if (!l || !l->signals) return;
+    struct yasl_signal_watch *w = l->signals;
+    l->signals = NULL;
+    for (int i = 0; i < w->count; i++) {
+        int signo = w->signals[i];
+        sigaction(signo, &w->old[i], NULL);
+        atomic_store(&signal_wake_fd[signo], 0);
+        atomic_store(&signal_pending[signo], false);
+        signal_owner[signo] = NULL;
+    }
+    free(w);
 }
 
 static void timer_cleanup_deleted(struct yasl_loop *l)
@@ -215,6 +416,16 @@ bool yasl_loop_timer_armed(const struct yasl_loop *l, yasl_timer_id id)
     return false;
 }
 
+int yasl_loop_poll_set(const struct yasl_loop *l, const struct pollfd **fds)
+{
+    if (!l) {
+        *fds = NULL;
+        return 0;
+    }
+    *fds = l->pollfds;
+    return l->count;
+}
+
 int yasl_loop_next_timeout_ms(const struct yasl_loop *l)
 {
     if (!l || l->timer_count <= 0) return -1;