Wake and post to loops from any thread, and watch signals

wsap_term runs its own eventfds to wake the loop from its pty worker and
accessibility threads, and its own signalfd or self-pipe per platform.

yasl_loop_kick() is now documented as safe from any thread and from a
signal handler. yasl_loop_post() queues a function to run on the owning
thread during a later step. yasl_loop_watch_signals() installs a
process-wide SA_RESTART handler per signal that records the signal and
kicks the loop, so callbacks run on the loop thread on every platform
without touching thread masks.

Cocoa dispatch waited only for AppKit events, so kicks, posts, and fd
watches could not end an indefinite wait, and yasl_post_event() did not
wake yasl_run(). While waiting, Cocoa now mirrors the loop's descriptors
in a kqueue whose run loop source posts a wake event. If kqueue rejects a
descriptor, the wait is bounded to 10 ms.

The Cocoa change is not built or run here; it needs a macOS check.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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;