#define _GNU_SOURCE #define _DARWIN_C_SOURCE #define _POSIX_C_SOURCE 200809L #include "common/loop.h" #include "platform/sync.h" #include #include #include #include #include #include #include #include #include #include #include #define YASL_PRIVATE_LOOP_INIT_CAP 32 typedef void (*fd_fn)(void *ud, int fd, short revents); struct fd_source { fd_fn fn; void *ud; int id; int fd; short events; }; struct timer_source { yasl_private_timer_cb fn; void *ud; uint64_t interval_ns; uint64_t deadline_ns; uint64_t id; bool deleted; }; struct post { void (*fn)(void *ud); void *ud; struct post *next; }; /* Signals this loop watches; the previous actions are restored on unwatch. */ struct signal_watch { void (*cb)(void *ud, int signo); void *ud; int count; int signals[NSIG]; struct sigaction old[NSIG]; }; struct yasl_loop { struct fd_source *fds; struct pollfd *pollfds; int count; int capacity; int next_id; pthread_t owner; /* Set by yasl_loop_quit; run clears it when it returns. */ bool quit; bool stepping; int wake_rfd; int wake_wfd; int wake_watch; struct timer_source *timers; int timer_count; int timer_capacity; uint64_t next_timer_id; int timer_firing_depth; yasl_private_mutex post_mutex; struct post *post_head; struct post *post_tail; struct signal_watch *signals; struct yasl_private_loop_driver driver; bool has_driver; }; /* 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_private_loop_now_ns(void) { struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; } static void run_posts(struct yasl_loop *l) { yasl_private_mutex_lock(&l->post_mutex); struct post *p = l->post_head; l->post_head = l->post_tail = NULL; yasl_private_mutex_unlock(&l->post_mutex); while (p) { struct 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 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); } } /* Only wakes the wait; the step runs posts and signals last. */ static void loop_wake_cb(void *ud, int fd, short revents) { (void)ud; (void)revents; char buf[128]; ssize_t n; while ((n = read(fd, buf, sizeof(buf))) > 0 || (n < 0 && errno == EINTR)) {} } static bool open_wake_pipe(int p[2]) { #if defined(__linux__) && defined(O_CLOEXEC) return pipe2(p, O_CLOEXEC | O_NONBLOCK) == 0; #else if (pipe(p) != 0) return false; for (int i = 0; i < 2; i++) { int flags = fcntl(p[i], F_GETFL, 0); if (flags < 0 || fcntl(p[i], F_SETFL, flags | O_NONBLOCK) < 0 || fcntl(p[i], F_SETFD, FD_CLOEXEC) < 0) { int err = errno; close(p[0]); close(p[1]); errno = err; return false; } } return true; #endif } yasl_loop *yasl_loop_create(void) { struct yasl_loop *l = calloc(1, sizeof(*l)); if (!l) return NULL; l->capacity = YASL_PRIVATE_LOOP_INIT_CAP; l->fds = malloc((size_t)l->capacity * sizeof(*l->fds)); l->pollfds = malloc((size_t)l->capacity * sizeof(*l->pollfds)); int p[2]; int err = 0; if (!l->fds || !l->pollfds) err = ENOMEM; else if ((err = yasl_private_mutex_init(&l->post_mutex)) == 0 && !open_wake_pipe(p)) { err = errno; yasl_private_mutex_destroy(&l->post_mutex); } if (err) { free(l->fds); free(l->pollfds); free(l); errno = err; return NULL; } l->owner = pthread_self(); l->next_id = 1; l->next_timer_id = 1; l->wake_rfd = p[0]; l->wake_wfd = p[1]; /* The first watch fits the initial capacity, so it cannot fail. */ l->wake_watch = yasl_loop_add_fd(l, l->wake_rfd, POLLIN, loop_wake_cb, l); return l; } int yasl_loop_destroy(yasl_loop *l) { if (!l) return 0; if (l->stepping || l->has_driver) return -EBUSY; yasl_loop_unwatch_signals(l); for (struct post *p = l->post_head; p;) { struct post *next = p->next; free(p); p = next; } yasl_private_mutex_destroy(&l->post_mutex); close(l->wake_rfd); close(l->wake_wfd); free(l->fds); free(l->pollfds); free(l->timers); free(l); return 0; } int yasl_loop_add_fd(yasl_loop *l, int fd, short events, void (*fn)(void *ud, int fd, short revents), void *ud) { if (!l || fd < 0 || !fn) return -EINVAL; if (l->count >= l->capacity) { int new_cap = l->capacity * 2; struct fd_source *new_fds = realloc(l->fds, (size_t)new_cap * sizeof(*new_fds)); if (!new_fds) return -ENOMEM; l->fds = new_fds; struct pollfd *new_pollfds = realloc(l->pollfds, (size_t)new_cap * sizeof(*new_pollfds)); if (!new_pollfds) return -ENOMEM; l->pollfds = new_pollfds; l->capacity = new_cap; } int id = l->next_id++; if (l->next_id <= 0) l->next_id = 1; l->fds[l->count] = (struct fd_source){.fn = fn, .ud = ud, .id = id, .fd = fd, .events = events}; l->pollfds[l->count] = (struct pollfd){.fd = fd, .events = events}; l->count++; return id; } static int fd_index(const struct yasl_loop *l, int id) { for (int i = 0; l && id > 0 && i < l->count; i++) if (l->fds[i].id == id) return i; return -1; } int yasl_loop_del_fd(yasl_loop *l, int id) { int i = fd_index(l, id); if (i < 0) return -ENOENT; l->count--; l->fds[i] = l->fds[l->count]; l->pollfds[i] = l->pollfds[l->count]; return 0; } int yasl_loop_set_fd_events(yasl_loop *l, int id, short events) { int i = fd_index(l, id); if (i < 0) return -ENOENT; l->fds[i].events = events; l->pollfds[i].events = events; return 0; } short yasl_private_loop_revents(const yasl_loop *l, int id) { int i = fd_index(l, id); return i < 0 ? 0 : l->pollfds[i].revents; } static void wake_write(int fd) { /* A full pipe already wakes the loop, so EAGAIN needs no retry. */ int saved = errno; char b = 1; ssize_t n; do { n = write(fd, &b, 1); } while (n < 0 && errno == EINTR); errno = saved; } void yasl_private_loop_kick(yasl_loop *l) { if (l) wake_write(l->wake_wfd); } void yasl_private_loop_wake(yasl_loop *l) { if (l && !pthread_equal(l->owner, pthread_self())) wake_write(l->wake_wfd); } bool yasl_private_loop_post(yasl_loop *l, void (*fn)(void *ud), void *ud) { if (!l || !fn) return false; struct post *p = malloc(sizeof(*p)); if (!p) return false; *p = (struct post){.fn = fn, .ud = ud}; yasl_private_mutex_lock(&l->post_mutex); if (l->post_tail) l->post_tail->next = p; else l->post_head = p; l->post_tail = p; yasl_private_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; } int yasl_loop_watch_signals(yasl_loop *l, const int *signals, int n, void (*cb)(void *ud, int signo), void *ud) { if (!l || !signals || n <= 0 || !cb) return -EINVAL; if (l->signals) return -EBUSY; bool wanted[NSIG] = {0}; for (int i = 0; i < n; i++) { if (!signal_valid(signals[i])) return -EINVAL; if (signal_owner[signals[i]]) return -EBUSY; wanted[signals[i]] = true; } struct signal_watch *w = calloc(1, sizeof(*w)); if (!w) return -ENOMEM; w->cb = cb; w->ud = ud; 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; 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++) { sigaction(w->signals[i], &w->old[i], NULL); atomic_store(&signal_wake_fd[w->signals[i]], 0); } free(w); return -err; } w->signals[w->count++] = signo; } for (int i = 0; i < w->count; i++) signal_owner[w->signals[i]] = l; l->signals = w; return 0; } void yasl_loop_unwatch_signals(yasl_loop *l) { if (!l || !l->signals) return; struct 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) { if (l->timer_firing_depth > 0) return; int dest = 0; for (int src = 0; src < l->timer_count; src++) { struct timer_source t = l->timers[src]; l->timers[dest] = t; dest += !t.deleted; } l->timer_count = dest; } bool yasl_private_loop_timer_armed(const yasl_loop *l, yasl_private_timer_id id) { if (!l || id == 0) return false; for (int i = 0; i < l->timer_count; i++) if (l->timers[i].id == id) return !l->timers[i].deleted; return false; } int yasl_private_loop_poll_set(const yasl_loop *l, const struct pollfd **fds) { if (!l) { *fds = NULL; return 0; } *fds = l->pollfds; return l->count; } int yasl_private_loop_next_timeout_ms(const yasl_loop *l) { if (!l) return -1; uint64_t now = yasl_private_loop_now_ns(); uint64_t min_deadline = UINT64_MAX; for (int i = 0; i < l->timer_count; i++) if (!l->timers[i].deleted && l->timers[i].deadline_ns < min_deadline) min_deadline = l->timers[i].deadline_ns; if (min_deadline == UINT64_MAX) return -1; if (min_deadline <= now) return 0; uint64_t ms = (min_deadline - now + 999999ull) / 1000000ull; return ms > (uint64_t)INT_MAX ? INT_MAX : (int)ms; } uint64_t yasl_loop_add_timer(yasl_loop *l, uint64_t interval_ns, yasl_private_timer_cb cb, void *ud) { if (!l || !cb) { errno = EINVAL; return 0; } if (l->timer_count >= l->timer_capacity) { int new_cap = l->timer_capacity ? l->timer_capacity * 2 : 16; struct timer_source *new_timers = realloc(l->timers, (size_t)new_cap * sizeof(*new_timers)); if (!new_timers) { errno = ENOMEM; return 0; } l->timers = new_timers; l->timer_capacity = new_cap; } /* 64 bits never wrap in practice, so an ID never names a newer timer. */ uint64_t id = l->next_timer_id++; l->timers[l->timer_count++] = (struct timer_source){ .fn = cb, .ud = ud, .id = id, .interval_ns = interval_ns, .deadline_ns = yasl_private_loop_now_ns() + interval_ns, }; /* Owner-thread only, and the next wait computes its timeout from the * timers, so no wake is needed. */ return id; } bool yasl_loop_del_timer(yasl_loop *l, uint64_t id) { if (!l || id == 0) return false; for (int i = l->timer_count - 1; i >= 0; i--) { if (l->timers[i].id == id && !l->timers[i].deleted) { l->timers[i].deleted = true; timer_cleanup_deleted(l); return true; } } return false; } bool yasl_private_loop_set_driver(yasl_loop *l, const struct yasl_private_loop_driver *driver) { if (!l) return false; if (driver && l->has_driver) { errno = EBUSY; return false; } l->driver = driver ? *driver : (struct yasl_private_loop_driver){0}; l->has_driver = driver != NULL; return true; } bool yasl_private_loop_owned(const yasl_loop *l) { return l && pthread_equal(l->owner, pthread_self()); } bool yasl_private_loop_stepping(const yasl_loop *l) { return l && l->stepping; } /* poll with an EINTR retry that keeps the deadline. */ static int poll_all(struct yasl_loop *l, int timeout_ms) { uint64_t deadline = timeout_ms > 0 ? yasl_private_loop_now_ns() + YASL_PRIVATE_MS_TO_NS(timeout_ms) : 0; for (;;) { if (poll(l->pollfds, (nfds_t)l->count, timeout_ms) >= 0) return 0; if (errno != EINTR) return -errno; if (timeout_ms > 0) { uint64_t now = yasl_private_loop_now_ns(); timeout_ms = now >= deadline ? 0 : (int)((deadline - now + 999999ull) / 1000000ull); } } } static int dispatch_fds(struct yasl_loop *l) { int handled = 0; int count = l->count; struct fd_source sstack[32]; struct pollfd pstack[32]; struct fd_source *scopy = count <= 32 ? sstack : malloc((size_t)count * sizeof(*scopy)); struct pollfd *pcopy = count <= 32 ? pstack : malloc((size_t)count * sizeof(*pcopy)); if (!scopy || !pcopy) { if (scopy != sstack) free(scopy); if (pcopy != pstack) free(pcopy); return -ENOMEM; } memcpy(scopy, l->fds, (size_t)count * sizeof(*scopy)); memcpy(pcopy, l->pollfds, (size_t)count * sizeof(*pcopy)); for (int i = 0; i < count; i++) { if (!pcopy[i].revents) continue; /* A callback may have removed or moved this watch. */ int j = i < l->count && l->fds[i].id == scopy[i].id ? i : fd_index(l, scopy[i].id); if (j < 0) continue; l->pollfds[j].revents = 0; scopy[i].fn(scopy[i].ud, scopy[i].fd, pcopy[i].revents); handled++; } if (scopy != sstack) free(scopy); if (pcopy != pstack) free(pcopy); return handled; } static void reschedule(struct timer_source *t, uint64_t next, uint64_t now) { if (next == 0 || t->deleted) { t->deleted = true; return; } t->interval_ns = next; t->deadline_ns = t->deadline_ns + next > now ? t->deadline_ns + next : now + next; } static int fire_timers(struct yasl_loop *l) { int handled = 0; uint64_t now = yasl_private_loop_now_ns(); l->timer_firing_depth++; /* Timers added by a callback wait for the next step. */ int initial_count = l->timer_count; for (int i = 0; i < initial_count && i < l->timer_count; i++) { if (l->timers[i].deleted || l->timers[i].deadline_ns > now) continue; struct timer_source t = l->timers[i]; uint64_t next = t.fn(t.ud, t.id, t.interval_ns); handled++; /* Compaction waits for depth zero, so index i still holds this timer. */ reschedule(&l->timers[i], next, now); } l->timer_firing_depth--; timer_cleanup_deleted(l); return handled; } static int step(struct yasl_loop *l, int timeout_ms) { /* A driver cannot change during a step: its display cannot close then. */ const struct yasl_private_loop_driver *d = l->has_driver ? &l->driver : NULL; if (d && d->prepare(d->ud)) timeout_ms = 0; int timer_to = yasl_private_loop_next_timeout_ms(l); if (timer_to >= 0 && (timeout_ms < 0 || timer_to < timeout_ms)) timeout_ms = timer_to; int r; if (d && d->wait) { r = d->wait(d->ud, timeout_ms); /* The platform wait does not fill revents; read them now. */ if (r >= 0) r = poll_all(l, 0); } else { r = poll_all(l, timeout_ms); } /* check closes what prepare opened, whatever the wait returned. */ if (d) d->check(d->ud); if (r < 0) return r; if (d && d->deliver) d->deliver(d->ud); int fds = dispatch_fds(l); if (fds < 0) return fds; int timers = fire_timers(l); /* Last, so a signal sees what came with it: a child's final output is * read before its exit is handled. */ run_posts(l); run_signals(l); return fds + timers; } int yasl_loop_step(yasl_loop *l, int timeout_ms) { if (!l) return -EINVAL; if (!pthread_equal(l->owner, pthread_self())) return -EPERM; if (l->stepping) return -EBUSY; l->stepping = true; int r = step(l, timeout_ms); l->stepping = false; return r; } int yasl_loop_run(yasl_loop *l) { if (!l) return -EINVAL; if (!pthread_equal(l->owner, pthread_self())) return -EPERM; /* A nested run must not consume the outer run's quit. */ if (l->stepping) return -EBUSY; int r = 0; while (!l->quit) { r = yasl_loop_step(l, -1); if (r < 0) break; } l->quit = false; return r < 0 ? r : 0; } void yasl_loop_quit(yasl_loop *l) { if (l) l->quit = true; }