summaryrefslogtreecommitdiff
path: root/src/platform/posix
diff options
context:
space:
mode:
Diffstat (limited to 'src/platform/posix')
-rw-r--r--src/platform/posix/loop.c646
1 files changed, 308 insertions, 338 deletions
diff --git a/src/platform/posix/loop.c b/src/platform/posix/loop.c
index 49132f8..6624575 100644
--- a/src/platform/posix/loop.c
+++ b/src/platform/posix/loop.c
@@ -6,7 +6,9 @@
#include <errno.h>
#include <fcntl.h>
+#include <limits.h>
#include <poll.h>
+#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdlib.h>
@@ -16,33 +18,33 @@
#define YASL_PRIVATE_LOOP_INIT_CAP 32
-typedef int (*yasl_private_loop_fd_fn)(int fd, short revents, void *ud);
+typedef void (*fd_fn)(void *ud, int fd, short revents);
-struct yasl_private_fd_source {
- yasl_private_loop_fd_fn fn;
+struct fd_source {
+ fd_fn fn;
void *ud;
int id;
int fd;
short events;
};
-struct yasl_private_timer_source {
+struct timer_source {
yasl_private_timer_cb fn;
void *ud;
uint64_t interval_ns;
uint64_t deadline_ns;
- yasl_private_timer_id id;
+ uint64_t id;
bool deleted;
};
-struct yasl_private_post {
+struct post {
void (*fn)(void *ud);
void *ud;
- struct yasl_private_post *next;
+ struct post *next;
};
/* Signals this loop watches; the previous actions are restored on unwatch. */
-struct yasl_private_signal_watch {
+struct signal_watch {
void (*cb)(void *ud, int signo);
void *ud;
int count;
@@ -50,28 +52,34 @@ struct yasl_private_signal_watch {
struct sigaction old[NSIG];
};
-struct yasl_private_loop {
- struct yasl_private_fd_source *fds;
+struct yasl_loop {
+ struct fd_source *fds;
struct pollfd *pollfds;
int count;
int capacity;
int next_id;
- bool running;
+ 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 yasl_private_timer_source *timers;
+ struct timer_source *timers;
int timer_count;
int timer_capacity;
- uint32_t next_timer_id;
+ uint64_t next_timer_id;
int timer_firing_depth;
yasl_private_mutex post_mutex;
- struct yasl_private_post *post_head;
- struct yasl_private_post *post_tail;
+ struct post *post_head;
+ struct post *post_tail;
- struct yasl_private_signal_watch *signals;
+ 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
@@ -79,7 +87,7 @@ struct yasl_private_loop {
* 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_private_loop *signal_owner[NSIG];
+static struct yasl_loop *signal_owner[NSIG];
uint64_t yasl_private_loop_now_ns(void)
{
@@ -88,25 +96,25 @@ uint64_t yasl_private_loop_now_ns(void)
return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec;
}
-static void run_posts(struct yasl_private_loop *l)
+static void run_posts(struct yasl_loop *l)
{
yasl_private_mutex_lock(&l->post_mutex);
- struct yasl_private_post *p = l->post_head;
+ struct post *p = l->post_head;
l->post_head = l->post_tail = NULL;
yasl_private_mutex_unlock(&l->post_mutex);
while (p) {
- struct yasl_private_post *next = p->next;
+ struct post *next = p->next;
p->fn(p->ud);
free(p);
p = next;
}
}
-static void run_signals(struct yasl_private_loop *l)
+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_private_signal_watch *w = l->signals;
+ struct signal_watch *w = l->signals;
if (!w) return;
if (signal_owner[signo] != l || !atomic_exchange(&signal_pending[signo], false))
continue;
@@ -114,148 +122,143 @@ static void run_signals(struct yasl_private_loop *l)
}
}
-static int loop_wake_cb(int fd, short revents, void *ud)
+static void loop_wake_cb(void *ud, int fd, short revents)
{
(void)revents;
char buf[128];
ssize_t n;
while ((n = read(fd, buf, sizeof(buf))) > 0 || (n < 0 && errno == EINTR)) {}
- struct yasl_private_loop *l = ud;
+ struct yasl_loop *l = ud;
run_posts(l);
run_signals(l);
- return 0;
}
-struct yasl_private_loop *yasl_private_loop_create(void)
+static bool open_wake_pipe(int p[2])
{
- struct yasl_private_loop *l = calloc(1, sizeof(*l));
- if (!l) return NULL;
+#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));
- if (!l->fds || !l->pollfds) {
- free(l->fds);
- free(l->pollfds);
- free(l);
- return NULL;
+ 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 (yasl_private_mutex_init(&l->post_mutex) != 0) {
+ 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 = -1;
- l->wake_wfd = -1;
-
- int p[2];
-#if defined(__linux__) && defined(O_CLOEXEC)
- if (pipe2(p, O_CLOEXEC | O_NONBLOCK) == 0) {
- l->wake_rfd = p[0];
- l->wake_wfd = p[1];
- }
-#else
- if (pipe(p) == 0) {
- int f0 = fcntl(p[0], F_GETFL, 0);
- int f1 = fcntl(p[1], F_GETFL, 0);
- if (f0 >= 0) fcntl(p[0], F_SETFL, f0 | O_NONBLOCK);
- if (f1 >= 0) fcntl(p[1], F_SETFL, f1 | O_NONBLOCK);
- fcntl(p[0], F_SETFD, FD_CLOEXEC);
- fcntl(p[1], F_SETFD, FD_CLOEXEC);
- l->wake_rfd = p[0];
- l->wake_wfd = p[1];
- }
-#endif
-
- if (l->wake_rfd >= 0) {
- l->wake_watch = yasl_private_loop_add_fd(l, l->wake_rfd, POLLIN, loop_wake_cb, l);
- }
+ 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;
}
-void yasl_private_loop_destroy(struct yasl_private_loop *l)
+int yasl_loop_destroy(yasl_loop *l)
{
- if (!l) return;
- yasl_private_loop_unwatch_signals(l);
- for (struct yasl_private_post *p = l->post_head; p;) {
- struct yasl_private_post *next = p->next;
+ 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);
- if (l->wake_watch > 0)
- yasl_private_loop_del_fd(l, l->wake_watch);
- if (l->wake_rfd >= 0)
- close(l->wake_rfd);
- if (l->wake_wfd >= 0)
- close(l->wake_wfd);
+ close(l->wake_rfd);
+ close(l->wake_wfd);
free(l->fds);
free(l->pollfds);
free(l->timers);
free(l);
+ return 0;
}
-int yasl_private_loop_add_fd(struct yasl_private_loop *l, int fd, short events,
- int (*fn)(int fd, short revents, void *ud), void *ud)
+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)
- return -1;
-
+ if (!l || fd < 0 || !fn) return -EINVAL;
if (l->count >= l->capacity) {
- int new_cap = l->capacity ? l->capacity * 2 : YASL_PRIVATE_LOOP_INIT_CAP;
- struct yasl_private_fd_source *new_fds = realloc(l->fds, (size_t)new_cap * sizeof(*new_fds));
- if (!new_fds) return -1;
+ 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 -1;
+ 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 yasl_private_fd_source){
- .fn = fn, .ud = ud, .id = id, .fd = fd, .events = events,
- };
- l->pollfds[l->count] = (struct pollfd){
- .fd = fd, .events = events, .revents = 0,
- };
+ 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;
}
-int yasl_private_loop_del_fd(struct yasl_private_loop *l, int id)
+static int fd_index(const struct yasl_loop *l, int id)
{
- if (!l || id <= 0) return -1;
- for (int i = 0; i < l->count; i++) {
- if (l->fds[i].id == id) {
- l->count--;
- l->fds[i] = l->fds[l->count];
- l->pollfds[i] = l->pollfds[l->count];
- return 0;
- }
- }
+ for (int i = 0; l && id > 0 && i < l->count; i++)
+ if (l->fds[i].id == id)
+ return i;
return -1;
}
-int yasl_private_loop_set_fd_events(struct yasl_private_loop *l, int id, short events)
+int yasl_loop_del_fd(yasl_loop *l, int id)
{
- if (!l || id <= 0) return -1;
- for (int i = 0; i < l->count; i++) {
- if (l->fds[i].id == id) {
- l->fds[i].events = events;
- l->pollfds[i].events = events;
- return 0;
- }
- }
- return -1;
+ 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)
@@ -270,18 +273,23 @@ static void wake_write(int fd)
errno = saved;
}
-void yasl_private_loop_kick(struct yasl_private_loop *l)
+void yasl_private_loop_kick(yasl_loop *l)
{
- if (!l || l->wake_wfd < 0) return;
- wake_write(l->wake_wfd);
+ if (l) wake_write(l->wake_wfd);
}
-bool yasl_private_loop_post(struct yasl_private_loop *l, void (*fn)(void *ud), void *ud)
+void yasl_private_loop_wake(yasl_loop *l)
{
- if (!l || !fn || l->wake_wfd < 0) return false;
- struct yasl_private_post *p = malloc(sizeof(*p));
+ 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 yasl_private_post){.fn = fn, .ud = ud};
+ *p = (struct post){.fn = fn, .ud = ud};
yasl_private_mutex_lock(&l->post_mutex);
if (l->post_tail)
l->post_tail->next = p;
@@ -305,85 +313,49 @@ static bool signal_valid(int signo)
return signo > 0 && signo < NSIG && signo != SIGKILL && signo != SIGSTOP;
}
-bool yasl_private_loop_watch_signals(struct yasl_private_loop *l, const int *signals, int n,
- void (*cb)(void *ud, int signo), void *ud)
+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 || l->wake_wfd < 0) {
- errno = EINVAL;
- return false;
- }
+ 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])) {
- errno = EINVAL;
- return false;
- }
- if (signal_owner[signals[i]] && signal_owner[signals[i]] != l) {
- errno = EBUSY;
- return false;
- }
+ if (!signal_valid(signals[i])) return -EINVAL;
+ if (signal_owner[signals[i]]) return -EBUSY;
wanted[signals[i]] = true;
}
- struct yasl_private_signal_watch *w = calloc(1, sizeof(*w));
- if (!w) return false;
+ struct signal_watch *w = calloc(1, sizeof(*w));
+ if (!w) return -ENOMEM;
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_private_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;
+ 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;
}
-
- /* 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;
+ return 0;
}
-void yasl_private_loop_unwatch_signals(struct yasl_private_loop *l)
+void yasl_loop_unwatch_signals(yasl_loop *l)
{
if (!l || !l->signals) return;
- struct yasl_private_signal_watch *w = l->signals;
+ struct signal_watch *w = l->signals;
l->signals = NULL;
for (int i = 0; i < w->count; i++) {
int signo = w->signals[i];
@@ -395,19 +367,19 @@ void yasl_private_loop_unwatch_signals(struct yasl_private_loop *l)
free(w);
}
-static void timer_cleanup_deleted(struct yasl_private_loop *l)
+static void timer_cleanup_deleted(struct yasl_loop *l)
{
- if (!l || l->timer_firing_depth > 0) return;
+ if (l->timer_firing_depth > 0) return;
int dest = 0;
for (int src = 0; src < l->timer_count; src++) {
- struct yasl_private_timer_source t = l->timers[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 struct yasl_private_loop *l, yasl_private_timer_id id)
+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++)
@@ -416,7 +388,7 @@ bool yasl_private_loop_timer_armed(const struct yasl_private_loop *l, yasl_priva
return false;
}
-int yasl_private_loop_poll_set(const struct yasl_private_loop *l, const struct pollfd **fds)
+int yasl_private_loop_poll_set(const yasl_loop *l, const struct pollfd **fds)
{
if (!l) {
*fds = NULL;
@@ -426,215 +398,213 @@ int yasl_private_loop_poll_set(const struct yasl_private_loop *l, const struct p
return l->count;
}
-int yasl_private_loop_next_timeout_ms(const struct yasl_private_loop *l)
+int yasl_private_loop_next_timeout_ms(const yasl_loop *l)
{
- if (!l || l->timer_count <= 0) return -1;
+ if (!l) return -1;
uint64_t now = yasl_private_loop_now_ns();
uint64_t min_deadline = UINT64_MAX;
- bool has_active = false;
- for (int i = 0; i < l->timer_count; i++) {
- if (!l->timers[i].deleted) {
- has_active = true;
- if (l->timers[i].deadline_ns < min_deadline)
- min_deadline = l->timers[i].deadline_ns;
- }
- }
- if (!has_active) return -1;
+ 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 diff = min_deadline - now;
- uint64_t ms = (diff + 999999ull) / 1000000ull;
- return (ms > (uint64_t)INT_MAX) ? INT_MAX : (int)ms;
+ uint64_t ms = (min_deadline - now + 999999ull) / 1000000ull;
+ return ms > (uint64_t)INT_MAX ? INT_MAX : (int)ms;
}
-yasl_private_timer_id yasl_private_loop_add_timer(struct yasl_private_loop *l, uint64_t interval_ns,
- yasl_private_timer_cb cb, void *ud)
+uint64_t yasl_loop_add_timer(yasl_loop *l, uint64_t interval_ns,
+ yasl_private_timer_cb cb, void *ud)
{
- if (!l || !cb || interval_ns == 0) return 0;
-
+ 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 yasl_private_timer_source *new_timers = realloc(l->timers, (size_t)new_cap * sizeof(*new_timers));
- if (!new_timers) return 0;
+ 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;
}
-
- uint32_t id = l->next_timer_id++;
- if (l->next_timer_id == 0) l->next_timer_id = 1;
-
- uint64_t now = yasl_private_loop_now_ns();
- l->timers[l->timer_count++] = (struct yasl_private_timer_source){
+ /* 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 = now + interval_ns,
- .deleted = false,
+ .deadline_ns = yasl_private_loop_now_ns() + interval_ns,
};
-
- yasl_private_loop_kick(l);
+ /* Owner-thread only, and the next wait computes its timeout from the
+ * timers, so no wake is needed. */
return id;
}
-bool yasl_private_loop_del_timer(struct yasl_private_loop *l, yasl_private_timer_id 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;
- if (l->timer_firing_depth == 0) {
- if (i == l->timer_count - 1) {
- while (l->timer_count > 0 && l->timers[l->timer_count - 1].deleted)
- l->timer_count--;
- } else {
- timer_cleanup_deleted(l);
- }
- }
+ timer_cleanup_deleted(l);
return true;
}
}
return false;
}
-int yasl_private_loop_step(struct yasl_private_loop *l, int timeout_ms)
+bool yasl_private_loop_set_driver(yasl_loop *l, const struct yasl_private_loop_driver *driver)
{
- if (!l) return -1;
-
- int timer_to = yasl_private_loop_next_timeout_ms(l);
- int effective_timeout = timeout_ms;
- if (timer_to >= 0) {
- if (effective_timeout < 0 || timer_to < effective_timeout)
- effective_timeout = timer_to;
- }
-
- if (l->count <= 0) {
- if (effective_timeout > 0) {
- struct timespec req = {
- .tv_sec = effective_timeout / 1000,
- .tv_nsec = (long)(effective_timeout % 1000) * 1000000L,
- };
- nanosleep(&req, NULL);
- }
- } else {
- int n;
- do {
- n = poll(l->pollfds, (nfds_t)l->count, effective_timeout);
- } while (n < 0 && errno == EINTR);
-
- if (n < 0) return -1;
+ 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;
+}
- int handled = 0;
+bool yasl_private_loop_owned(const yasl_loop *l)
+{
+ return l && pthread_equal(l->owner, pthread_self());
+}
- if (l->count > 0) {
- int count = l->count;
- struct yasl_private_fd_source sstack[32];
- struct pollfd pstack[32];
- struct yasl_private_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 && scopy != sstack) free(scopy);
- if (pcopy && pcopy != pstack) free(pcopy);
- return -1;
- }
+bool yasl_private_loop_stepping(const yasl_loop *l)
+{
+ return l && l->stepping;
+}
- 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) {
- short revents = pcopy[i].revents;
- // Fast-path: check if source is still at index i before linear scan
- bool valid = false;
- if (i < l->count && l->fds[i].id == scopy[i].id) {
- l->pollfds[i].revents = 0;
- valid = true;
- } else {
- for (int j = 0; j < l->count; j++) {
- if (l->fds[j].id == scopy[i].id) {
- l->pollfds[j].revents = 0;
- valid = true;
- break;
- }
- }
- }
- if (valid && scopy[i].fn) {
- scopy[i].fn(scopy[i].fd, revents, scopy[i].ud);
- handled++;
- }
- }
+/* 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;
+}
- if (l->timer_count > 0) {
- uint64_t now = yasl_private_loop_now_ns();
- l->timer_firing_depth++;
- int initial_count = l->timer_count;
- for (int i = 0; i < initial_count; i++) {
- if (i >= l->timer_count) break;
- if (l->timers[i].deleted || l->timers[i].deadline_ns > now)
- continue;
-
- yasl_private_timer_id tid = l->timers[i].id;
- uint64_t interval = l->timers[i].interval_ns;
- void *ud = l->timers[i].ud;
- struct yasl_private_timer_source timer_src = l->timers[i];
-
- uint64_t next_interval = timer_src.fn(ud, tid, interval);
- handled++;
-
- // Fast-path: timer is usually still at index i unless an earlier timer was deleted
- if (i < l->timer_count && l->timers[i].id == tid) {
- if (next_interval == 0 || l->timers[i].deleted) {
- l->timers[i].deleted = true;
- } else {
- l->timers[i].interval_ns = next_interval;
- if (l->timers[i].deadline_ns + next_interval > now)
- l->timers[i].deadline_ns += next_interval;
- else
- l->timers[i].deadline_ns = now + next_interval;
- }
- } else {
- for (int j = 0; j < l->timer_count; j++) {
- if (l->timers[j].id == tid) {
- if (next_interval == 0 || l->timers[j].deleted) {
- l->timers[j].deleted = true;
- } else {
- l->timers[j].interval_ns = next_interval;
- if (l->timers[j].deadline_ns + next_interval > now)
- l->timers[j].deadline_ns += next_interval;
- else
- l->timers[j].deadline_ns = now + next_interval;
- }
- break;
- }
- }
- }
- }
- l->timer_firing_depth--;
- if (l->timer_firing_depth == 0) {
- timer_cleanup_deleted(l);
- }
+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;
}
-int yasl_private_loop_run(struct yasl_private_loop *l)
+static int step(struct yasl_loop *l, int timeout_ms)
{
- if (!l) return -1;
- l->running = true;
- while (l->running) {
- if (yasl_private_loop_step(l, -1) < 0)
- return -1;
+ /* 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);
}
- return 0;
+ /* 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;
+ return fds + fire_timers(l);
+}
+
+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_private_loop_quit(struct yasl_private_loop *l)
+void yasl_loop_quit(yasl_loop *l)
{
- if (l) l->running = false;
+ if (l) l->quit = true;
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-25 02:31:48 +0000