From 4924a75a3ae1543e5f17be16657353caf05dd381 Mon Sep 17 00:00:00 2001 From: Daan De Meyer Date: Fri, 22 May 2026 06:36:52 +0000 Subject: [PATCH] sd-future: rework cancellation, add slot callbacks and future groups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cancellation propagation: - sd_fiber_resume() now stashes the resume value even when the target fiber isn't suspended yet (INITIAL or running/READY); the next fiber_swap() consumes it without yielding to the event loop. -ECANCELED is sticky once queued, so a concurrent async wakeup can't override a pending cancellation. - sd_future_cancel_wait_unref() remembers the last non-zero return from its internal sd_fiber_await() across loop iterations and, once the future has finally resolved, re-queues it via sd_fiber_resume(self, ...) so the calling fiber's next sd_fiber_suspend() / sd_fiber_yield() observes it. Previously those values were discarded. Together these let sd_future_cancel_wait_unref() forward interruptions (cancellations / outer SD_FIBER_TIMEOUT firings) instead of silently swallowing them. FIBER_STATE_CANCELLED is dropped in the process: fiber_cancel() now queues -ECANCELED through the same sd_fiber_resume() path (with an idempotency check so fiber_on_exit()'s two-pass arm-then-dispatch cycle still works), and fiber_swap() delivers it via the standard result_pending branch instead of a dedicated state-machine check. Net effect: one enum value gone, the cancel path goes through exactly the same plumbing as any other async wakeup. Decouple unreferencing a future from resolving it: futures must now be resolved by the time they're unreffed. To make this easier we introduce sd_future_cancel_unref() (plus _unrefp / array variants), which cancels and then unrefs. We use it everywhere where we can't assume we're on a fiber and thus can't wait for the future to resolve after cancelling. All future kinds we support today except fibers cancel synchronously, so this isn't a problem in practice. Even fibers cancel synchronously after creation until they're scheduled once by the event loop, after which they cancel asynchronously. Slot-based callbacks: Replaces the single-callback-per-future model (sd_future_set_callback, sd_future_get_userdata, and the wait_future indirection in sd_fiber_await) with explicit, awaiter-owned slots: - sd_future_add_callback() returns a refcounted sd_future_slot the caller owns via _cleanup_(sd_future_slot_unrefp). A future can have many slots; dropping a slot deregisters that one callback cleanly without resolving or otherwise touching the future. NULL ret_slot makes the slot 'floating' — its lifetime is bound to the future. - Callbacks never run inline. Each slot owns a one-shot defer event source and a one-shot exit event source on the future's event loop; the one applicable to the loop's current phase is armed when the future resolves — or at add time when the slot is attached to an already-RESOLVED future. Slots inherit the calling fiber's priority when added from inside one. Deferring unconditionally keeps the resolver free of reentrancy concerns (slot-set mutation during iteration, callbacks dropping the future's last ref) and frees callers from having to reason about whether a callback might fire before sd_future_add_callback() returns. - sd_fiber_await() registers a slot on the target directly and drops it on scope exit — no more wait_future allocation, no auto-set callback on sd_future_new(). sd_fiber_timeout() and sd_fiber_sleep() likewise install their resume callback explicitly via sd_future_add_callback(). - sd_future_new_wait() / sd_future_set_callback() / sd_future_get_userdata() are gone; existing callers in sd-bus, qmp-client, event-future, and fiber-io migrate to the slot API. sd_future_resume_callback() is exposed as a shared helper for the common "resume my fiber when this future settles" wiring. sd_future_new() now takes the sd_event the future belongs to, and sd_future_get_event() exposes it. This lets the rest of the API (slot dispatch, group child tracking, defer futures) avoid plumbing the event through separately. Future groups and defer futures: Adds sd_future_group, modelled on asyncio.TaskGroup: aggregate N child futures with a policy (default = wait-all-fail-fast; SD_FUTURE_GROUP_WAIT_ANY resolves on the first settled child; SD_FUTURE_GROUP_IGNORE_ERRORS collects all results without cancelling siblings on error). Policy is configured up-front via sd_future_group_set_policy() before any child is added — once children are in flight, the resolution mechanics are locked in. On group failure, the parent fiber that created the group is cancelled so it observes the failure even if it hasn't started awaiting yet — and not cancelled when it's already inside sd_future_group_await(), nor when the parent itself is the fiber driving the cancel. Finalize-time draining ensures the group only resolves once every child has actually settled, even if siblings need a round-trip through the event loop to honor a cancel. Also adds sd_future_new_defer() for one-shot 'resolve on next event loop iteration' futures. Tests cover the queued-while-running behavior of sd_fiber_resume(), the propagation guarantees of sd_future_cancel_wait_unref() for both timeouts and external cancellations (from main and from a peer fiber), and a new test-future-group.c with coverage for each policy, parent-cancel propagation (including the parent-drives-cancel skip path), child-drain-on-resolve, add-rejected-during-finalize, slot lifecycle, and the already-resolved add_callback path. --- src/libsystemd/meson.build | 2 + src/libsystemd/sd-bus/bus-future.c | 6 +- src/libsystemd/sd-bus/bus-objects.c | 12 +- src/libsystemd/sd-common/sd-forward.h | 3 +- src/libsystemd/sd-event/event-future.c | 115 +++- src/libsystemd/sd-future/fiber-io.c | 62 +- src/libsystemd/sd-future/fiber.c | 209 ++++--- src/libsystemd/sd-future/future-group.c | 319 +++++++++++ src/libsystemd/sd-future/sd-future.c | 352 ++++++++---- src/libsystemd/sd-future/test-fiber.c | 481 +++++++++++++++- src/libsystemd/sd-future/test-future-group.c | 567 +++++++++++++++++++ src/libsystemd/sd-varlink/sd-varlink.c | 4 +- src/shared/qmp-client.c | 6 +- src/systemd/sd-future.h | 42 +- 14 files changed, 1919 insertions(+), 261 deletions(-) create mode 100644 src/libsystemd/sd-future/future-group.c create mode 100644 src/libsystemd/sd-future/test-future-group.c diff --git a/src/libsystemd/meson.build b/src/libsystemd/meson.build index 24979e8882e6f..723ed2f057be2 100644 --- a/src/libsystemd/meson.build +++ b/src/libsystemd/meson.build @@ -80,6 +80,7 @@ sd_device_sources = files( sd_future_sources = files( 'sd-future/fiber-io.c', 'sd-future/fiber.c', + 'sd-future/future-group.c', 'sd-future/sd-future.c', ) @@ -196,6 +197,7 @@ simple_tests += files( 'sd-future/test-fiber.c', 'sd-future/test-fiber-io.c', 'sd-future/test-fiber-ops.c', + 'sd-future/test-future-group.c', 'sd-hwdb/test-sd-hwdb.c', 'sd-id128/test-id128.c', 'sd-journal/test-audit-type.c', diff --git a/src/libsystemd/sd-bus/bus-future.c b/src/libsystemd/sd-bus/bus-future.c index d7037b3f15bed..4676251b34a10 100644 --- a/src/libsystemd/sd-bus/bus-future.c +++ b/src/libsystemd/sd-bus/bus-future.c @@ -64,8 +64,8 @@ int bus_call_future(sd_bus *bus, sd_bus_message *m, uint64_t usec, sd_future **r assert(m); assert(ret); - _cleanup_(sd_future_unrefp) sd_future *f = NULL; - r = sd_future_new(&bus_future_ops, &f); + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + r = sd_future_new(sd_bus_get_event(bus), &bus_future_ops, &f); if (r < 0) return r; @@ -122,7 +122,7 @@ int bus_call_suspend( if (r < 0) return sd_bus_error_set_errno(reterr_error, r); - r = sd_fiber_suspend(); + r = sd_fiber_await(f); /* If the future isn't resolved, the suspend was interrupted before a reply arrived (fiber * cancelled, fiber-wide SD_FIBER_TIMEOUT scope expired, …). There's no reply to extract, diff --git a/src/libsystemd/sd-bus/bus-objects.c b/src/libsystemd/sd-bus/bus-objects.c index 795cc68837655..9cb8125540c69 100644 --- a/src/libsystemd/sd-bus/bus-objects.c +++ b/src/libsystemd/sd-bus/bus-objects.c @@ -373,8 +373,8 @@ DEFINE_PRIVATE_HASH_OPS_WITH_KEY_DESTRUCTOR( trivial_compare_func, bus_fiber_future_unref); -static int bus_fiber_resolved(sd_future *f) { - sd_bus *bus = ASSERT_PTR(sd_future_get_userdata(f)); +static int bus_fiber_resolved(sd_future *f, void *userdata) { + sd_bus *bus = ASSERT_PTR(userdata); assert_se(set_remove(bus->fiber_futures, f) == f); sd_future_unref(f); @@ -488,7 +488,7 @@ static int method_callbacks_run( .userdata = u, }; - _cleanup_(sd_future_unrefp) sd_future *f = NULL; + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; r = sd_fiber_new(bus->event, c->member, bus_fiber_entry, d, bus_fiber_data_destroy, &f); if (r < 0) return bus_maybe_reply_error(m, r, NULL); @@ -502,8 +502,10 @@ static int method_callbacks_run( return bus_maybe_reply_error(m, r, NULL); assert(r > 0); - /* Track the future on the bus so shutdown can cancel it and wait for it. */ - r = sd_future_set_callback(f, bus_fiber_resolved, bus); + /* Track the future on the bus so shutdown can cancel it and wait for it. + * Floating slot — tied to f's lifetime; bus_fiber_resolved is what drops + * the set's ref, which is the last ref and frees f (and the slot). */ + r = sd_future_add_callback(f, /* ret_slot= */ NULL, bus_fiber_resolved, bus); if (r < 0) { /* TAKE_PTR(f) hasn't run yet, so our cleanup attribute still owns the * ref; set_remove() returns the raw pointer without firing the hash_ops diff --git a/src/libsystemd/sd-common/sd-forward.h b/src/libsystemd/sd-common/sd-forward.h index aca0ed32095f7..5a0c189b3a19f 100644 --- a/src/libsystemd/sd-common/sd-forward.h +++ b/src/libsystemd/sd-common/sd-forward.h @@ -125,7 +125,8 @@ typedef struct sd_resolve_query sd_resolve_query; typedef struct sd_hwdb sd_hwdb; typedef struct sd_future sd_future; +typedef struct sd_future_slot sd_future_slot; -typedef int (*sd_future_func_t)(sd_future *f); +typedef int (*sd_future_func_t)(sd_future *f, void *userdata); typedef int (*sd_fiber_func_t)(void *userdata); typedef _sd_destroy_t sd_fiber_destroy_t; diff --git a/src/libsystemd/sd-event/event-future.c b/src/libsystemd/sd-event/event-future.c index 4e0a0a87fc204..d76b1ab4ad5d5 100644 --- a/src/libsystemd/sd-event/event-future.c +++ b/src/libsystemd/sd-event/event-future.c @@ -84,8 +84,8 @@ int future_new_io(sd_event *e, int fd, uint32_t events, sd_future **ret) { if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED)) return -ECANCELED; - _cleanup_(sd_future_unrefp) sd_future *f = NULL; - r = sd_future_new(&io_future_ops, &f); + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + r = sd_future_new(e, &io_future_ops, &f); if (r < 0) return r; @@ -126,6 +126,87 @@ int future_new_io(sd_event *e, int fd, uint32_t events, sd_future **ret) { return 0; } +typedef struct DeferFuture { + sd_event_source *source; + int result; +} DeferFuture; + +static void* defer_future_alloc(void) { + return new0(DeferFuture, 1); +} + +static void defer_future_free(sd_future *f) { + DeferFuture *df = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); + + sd_event_source_disable_unref(df->source); + free(df); +} + +static int defer_future_cancel(sd_future *f) { + DeferFuture *df = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); + int r; + + r = sd_event_source_set_enabled(df->source, SD_EVENT_OFF); + RET_GATHER(r, sd_future_resolve(f, -ECANCELED)); + return r; +} + +static int defer_future_set_priority(sd_future *f, int64_t priority) { + DeferFuture *df = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); + return sd_event_source_set_priority(df->source, priority); +} + +static const sd_future_ops defer_future_ops = { + .size = sizeof(sd_future_ops), + .alloc = defer_future_alloc, + .free = defer_future_free, + .cancel = defer_future_cancel, + .set_priority = defer_future_set_priority, +}; + +static int defer_handler(sd_event_source *s, void *userdata) { + sd_future *f = ASSERT_PTR(userdata); + DeferFuture *df = ASSERT_PTR(sd_future_get_private(f)); + return sd_future_resolve(f, df->result); +} + +int sd_future_new_defer(sd_event *e, int result, sd_future **ret) { + int r; + + assert_return(e, -EINVAL); + assert_return(ret, -EINVAL); + + if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED)) + return -ECANCELED; + + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + r = sd_future_new(e, &defer_future_ops, &f); + if (r < 0) + return r; + + DeferFuture *df = sd_future_get_private(f); + df->result = result; + + r = sd_event_add_defer(e, &df->source, defer_handler, f); + if (r < 0) + return r; + + if (sd_fiber_is_running()) { + int64_t priority; + + r = sd_fiber_get_priority(&priority); + if (r < 0) + return r; + + r = sd_event_source_set_priority(df->source, priority); + if (r < 0) + return r; + } + + *ret = TAKE_PTR(f); + return 0; +} + typedef struct TimeFuture { sd_event_source *source; @@ -200,8 +281,8 @@ static int future_new_time_internal( if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED)) return -ECANCELED; - _cleanup_(sd_future_unrefp) sd_future *f = NULL; - r = sd_future_new(&time_future_ops, &f); + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + r = sd_future_new(e, &time_future_ops, &f); if (r < 0) return r; @@ -271,13 +352,29 @@ int event_run_suspend(sd_event *e, uint64_t timeout) { if (fd < 0) return fd; + /* Wait for the inner-loop fd to become readable OR (optionally) the timeout to fire. */ + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *group = NULL; + r = sd_future_group_new(outer, &group); + if (r < 0) + return r; + + r = sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY); + if (r < 0) + return r; + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *io = NULL; r = future_new_io(outer, fd, EPOLLIN, &io); if (r < 0) return r; - _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL; + r = sd_future_group_add(group, io); + if (r < 0) + return r; + + io = sd_future_unref(io); + if (timeout != USEC_INFINITY) { + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL; r = future_new_time_relative( outer, CLOCK_MONOTONIC, @@ -287,9 +384,15 @@ int event_run_suspend(sd_event *e, uint64_t timeout) { &timer); if (r < 0) return r; + + r = sd_future_group_add(group, timer); + if (r < 0) + return r; + + timer = sd_future_unref(timer); } - r = sd_fiber_suspend(); + r = sd_future_group_await(group); if (r < 0) return r; diff --git a/src/libsystemd/sd-future/fiber-io.c b/src/libsystemd/sd-future/fiber-io.c index 9fe0acfccd5e8..abd49bd5db38b 100644 --- a/src/libsystemd/sd-future/fiber-io.c +++ b/src/libsystemd/sd-future/fiber-io.c @@ -10,7 +10,6 @@ #include "sd-event.h" #include "sd-future.h" -#include "alloc-util.h" #include "errno-util.h" #include "event-future.h" #include "fd-util.h" @@ -51,7 +50,7 @@ static ssize_t fiber_io_operation( if (r < 0) return r; - r = sd_fiber_suspend(); + r = sd_fiber_await(io); if (r < 0) return r; @@ -230,7 +229,7 @@ int sd_fiber_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) /* future_new_io resolves with the revents mask on success; translate any positive value * (e.g. POLLOUT) back to the connect(2) success status. */ - r = sd_fiber_suspend(); + r = sd_fiber_await(io); return r > 0 ? 0 : r; } @@ -395,17 +394,18 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time if (zero_timeout || r != 0) /* Either error or some fds are ready */ return r; - sd_future **futures = NULL; - CLEANUP_ARRAY(futures, n_fds, sd_future_cancel_wait_unref_array); + /* Use a WAIT_ANY group: the first child (an fd readiness or the timer) to settle resolves + * the group, which cancels its siblings on the spot. The user-supplied event mask is in + * poll() bits (struct pollfd), so translate to epoll bits. */ + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *group = NULL; + r = sd_future_group_new(e, &group); + if (r < 0) + return r; - futures = new0(sd_future*, n_fds); - if (!futures) - return -ENOMEM; + r = sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY); + if (r < 0) + return r; - /* Set up I/O event sources for all valid fds. POLL* and EPOLL* share their bit values (see - * EPOLL_POLL_COMMON_MASK in io-util.h), so we can pass the user-supplied event mask through - * to either backend without translation. */ - size_t n_io_futures = 0; for (size_t i = 0; i < n_fds; i++) { if (fds[i].fd < 0) continue; @@ -414,11 +414,16 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time if (events == 0) continue; - r = future_new_io(e, fds[i].fd, events, &futures[i]); + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *io = NULL; + r = future_new_io(e, fds[i].fd, events, &io); if (r < 0) return r; - n_io_futures++; + r = sd_future_group_add(group, io); + if (r < 0) + return r; + + io = sd_future_unref(io); } /* A timeout that overflows usec_t saturates to USEC_INFINITY in timespec_load(); treat that @@ -427,14 +432,19 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time * wait a very long time. */ usec_t usec = timeout ? timespec_load(timeout) : USEC_INFINITY; + size_t size; + r = sd_future_group_size(group, &size); + if (r < 0) + return r; + /* If every fd was skipped (negative or empty event mask) and we'd have no timer, there's * nothing that could ever wake the fiber up — same situation as n_fds == 0 && !timeout, * just not detectable upfront. Refuse rather than suspend forever. */ - if (n_io_futures == 0 && usec == USEC_INFINITY) + if (size == 0 && usec == USEC_INFINITY) return -EINVAL; - _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL; if (usec != USEC_INFINITY) { + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL; r = future_new_time_relative( e, CLOCK_MONOTONIC, @@ -444,9 +454,15 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time &timer); if (r < 0) return r; + + r = sd_future_group_add(group, timer); + if (r < 0) + return r; + + timer = sd_future_unref(timer); } - r = sd_fiber_suspend(); + r = sd_future_group_await(group); if (r < 0 && r != -ETIME) return r; @@ -457,14 +473,10 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time if (n != 0) return n; - /* No fds ready: distinguish our own timer from an external -ETIME. */ - if (timer && sd_future_state(timer) == SD_FUTURE_RESOLVED) - return 0; - - /* An IO future resolved with a revents mask (r > 0) but the readiness was already consumed - * by the time we swept — report 0 rather than leaking the bitmask as a (bogus) ppoll fd - * count to the caller. */ - if (r > 0) + /* No fds ready. The group's result is the winning child's result: 0 means the timer + * (created with result=0) fired; r > 0 means an IO future fired (revents mask) but the + * readiness was already drained when we swept. Both map to "0 fds ready". */ + if (r >= 0) return 0; return r; diff --git a/src/libsystemd/sd-future/fiber.c b/src/libsystemd/sd-future/fiber.c index 64dee8411df22..ab34164fcd5bf 100644 --- a/src/libsystemd/sd-future/fiber.c +++ b/src/libsystemd/sd-future/fiber.c @@ -39,13 +39,12 @@ * synonym there. */ _noreturn_ extern void siglongjmp_unchecked(sigjmp_buf env, int val) __asm__("siglongjmp"); -static thread_local Fiber *current_fiber = NULL; +static thread_local sd_future *current_fiber = NULL; typedef enum FiberState { FIBER_STATE_INITIAL, FIBER_STATE_READY, FIBER_STATE_SUSPENDED, - FIBER_STATE_CANCELLED, FIBER_STATE_COMPLETED, _FIBER_STATE_MAX, _FIBER_STATE_INVALID = -EINVAL, @@ -64,10 +63,10 @@ typedef struct Fiber { FiberState state; int result; /* Either resume error code or final return value */ + bool result_pending; /* sd_fiber_resume() stashed a value that fiber_swap() hasn't consumed yet */ sd_future *floating; /* Self-ref held while the fiber is floating; dropped on resolve. */ - sd_event *event; sd_event_source *defer_event_source; sd_event_source *exit_event_source; @@ -89,11 +88,7 @@ typedef struct Fiber { #endif } Fiber; -static Fiber* fiber_get_current(void) { - return current_fiber; -} - -static void fiber_set_current(Fiber *f) { +static void fiber_set_current(sd_future *f) { current_fiber = f; } @@ -167,7 +162,7 @@ static inline void finish_switch_stack(void *fake_stack_save) { /* Refresh f->resume_stack from whoever is currently the running fiber, so the next siglongjmp() out * of f (in the trampoline or fiber_swap()) can hand the right destination stack to ASAN. Must be - * called before fiber_set_current(f) — relies on fiber_get_current() returning the caller. */ + * called before fiber_set_current(f) — relies on sd_fiber_get_current() returning the caller. */ static void fiber_set_resume_stack(Fiber *f, Fiber *resume) { assert(f); @@ -178,11 +173,11 @@ static void fiber_set_resume_stack(Fiber *f, Fiber *resume) { } _noreturn_ static void fiber_entry_point(void) { - Fiber *f = ASSERT_PTR(fiber_get_current()); + Fiber *f = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(sd_fiber_get_current()))); void *fake_stack_save = NULL; assert(f->func); - assert(IN_SET(f->state, FIBER_STATE_INITIAL, FIBER_STATE_READY, FIBER_STATE_CANCELLED)); + assert(IN_SET(f->state, FIBER_STATE_INITIAL, FIBER_STATE_READY)); finish_switch_stack(NULL); @@ -205,7 +200,7 @@ _noreturn_ static void fiber_entry_point(void) { LOG_SET_PREFIX(f->name); LOG_CONTEXT_PUSH_KEY_VALUE("FIBER=", f->name); - f->result = f->state == FIBER_STATE_CANCELLED ? -ECANCELED : f->func(f->userdata); + f->result = f->func(f->userdata); f->state = FIBER_STATE_COMPLETED; } @@ -219,29 +214,28 @@ _noreturn_ static void fiber_entry_point(void) { assert_not_reached(); } -static int fiber_init(Fiber *f) { +static int fiber_init(sd_future *f) { + Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); ucontext_t old_uc, uc; void *fake_stack_save = NULL; - assert(f); - if (getcontext(&uc) < 0) return -errno; - struct iovec fiber_stack = fiber_stack_usable(&f->stack); + struct iovec fiber_stack = fiber_stack_usable(&fiber->stack); uc.uc_link = NULL; /* Unused: trampoline siglongjmps out instead of returning. */ uc.uc_stack.ss_sp = fiber_stack.iov_base; uc.uc_stack.ss_size = fiber_stack.iov_len; uc.uc_stack.ss_flags = 0; - Fiber *prev = fiber_get_current(); + sd_future *prev = sd_fiber_get_current(); fiber_set_current(f); makecontext(&uc, fiber_entry_point, /* argc= */ 0); - fiber_set_resume_stack(f, prev); - if (sigsetjmp(f->resume_context, /* savemask= */ 0) == 0) { + fiber_set_resume_stack(fiber, prev ? sd_future_get_private(prev) : NULL); + if (sigsetjmp(fiber->resume_context, /* savemask= */ 0) == 0) { start_switch_stack(&fake_stack_save, &fiber_stack); if (swapcontext(&old_uc, &uc) < 0) { finish_switch_stack(fake_stack_save); @@ -270,9 +264,10 @@ static void reset_current_fiber(void) { /* Restore the caller's log state stashed in the running fiber (if any) before clearing * current_fiber. Without this, the child of a fork() that happened mid-fiber would inherit the * fiber's log prefix / context list in its thread-locals even though no fiber is running. */ - Fiber *f = fiber_get_current(); + sd_future *f = sd_fiber_get_current(); if (f) { - fiber_swap_log_state(f); + Fiber *fiber = ASSERT_PTR(sd_future_get_private(f)); + fiber_swap_log_state(fiber); fiber_ops_set(NULL); } fiber_set_current(NULL); @@ -281,9 +276,9 @@ static void reset_current_fiber(void) { static sd_event_source* fiber_current_event_source(Fiber *f) { assert(f); assert(f->state != FIBER_STATE_COMPLETED); - assert(f->event); - return sd_event_get_state(f->event) == SD_EVENT_EXITING ? f->exit_event_source : f->defer_event_source; + sd_event *e = sd_event_source_get_event(ASSERT_PTR(f->defer_event_source)); + return sd_event_get_state(e) == SD_EVENT_EXITING ? f->exit_event_source : f->defer_event_source; } static int atfork_ret; @@ -318,18 +313,23 @@ static const FiberOps fiber_ops = { .cancel_wait_unref = sd_future_cancel_wait_unref, }; -static void fiber_enter(Fiber *fiber, Fiber *prev, void **fake_stack_save) { - fiber_set_current(fiber); +static void fiber_enter(sd_future *f, sd_future *prev, void **fake_stack_save) { + Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); + Fiber *prev_fiber = prev ? sd_future_get_private(prev) : NULL; + + fiber_set_current(f); fiber_swap_log_state(fiber); if (!prev) fiber_ops_set(&fiber_ops); struct iovec fiber_stack = fiber_stack_usable(&fiber->stack); start_switch_stack(fake_stack_save, &fiber_stack); - fiber_set_resume_stack(fiber, prev); + fiber_set_resume_stack(fiber, prev_fiber); } -static void fiber_leave(Fiber *fiber, Fiber *prev, void *fake_stack_save) { +static void fiber_leave(sd_future *f, sd_future *prev, void *fake_stack_save) { + Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); + finish_switch_stack(fake_stack_save); if (!prev) fiber_ops_set(NULL); @@ -344,7 +344,7 @@ static int fiber_run(sd_future *f) { if (fiber->state == FIBER_STATE_COMPLETED) return -ESTALE; - assert(IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_READY, FIBER_STATE_CANCELLED)); + assert(IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_READY)); static pthread_once_t atfork_once = PTHREAD_ONCE_INIT; r = pthread_once(&atfork_once, install_atfork); @@ -362,18 +362,18 @@ static int fiber_run(sd_future *f) { * completes. This matters when fiber_run() is invoked from within another fiber (e.g. an * sd-event dispatch that happens to be running inside a fiber context itself): the * LOG_SET_PREFIX/LOG_CONTEXT_PUSH above attached to whichever fiber was current at that moment, - * and their scope-level cleanup must see the same fiber_get_current() when it runs to detach + * and their scope-level cleanup must see the same sd_fiber_get_current() when it runs to detach * them from the correct list. */ - Fiber *prev = fiber_get_current(); + sd_future *prev = sd_fiber_get_current(); void *fake_stack_save = NULL; - fiber_enter(fiber, prev, &fake_stack_save); + fiber_enter(f, prev, &fake_stack_save); /* This is where we start executing the fiber. Once it yields, we continue here as if nothing * happened. resume_context captures this point; the fiber siglongjmps back to it. */ if (sigsetjmp(fiber->resume_context, 0) == 0) siglongjmp_unchecked(fiber->context, 1); - fiber_leave(fiber, prev, fake_stack_save); + fiber_leave(f, prev, fake_stack_save); switch (fiber->state) { @@ -386,7 +386,6 @@ static int fiber_run(sd_future *f) { fiber_resolve(f); break; - case FIBER_STATE_CANCELLED: case FIBER_STATE_READY: log_debug("Fiber yielded execution"); @@ -411,29 +410,33 @@ static int fiber_cancel(sd_future *f) { Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); int r; - assert(fiber != fiber_get_current()); + assert(f != sd_fiber_get_current()); - if (IN_SET(fiber->state, FIBER_STATE_COMPLETED, FIBER_STATE_CANCELLED)) + if (fiber->state == FIBER_STATE_COMPLETED) + return 0; + + /* Cancellation already queued — idempotent. Return 0 so fiber_on_exit() proceeds to + * fiber_run() instead of looping through the event source re-arm dance. */ + if (fiber->result_pending && fiber->result == -ECANCELED) return 0; if (fiber->state == FIBER_STATE_INITIAL) { /* The fiber's stack was allocated but never entered, so there are no scope-level cleanups - * waiting to run. Skip the dispatch round-trip that would just have fiber_entry_point() - * fall straight through with -ECANCELED, and settle the future right here — mirroring the - * FIBER_STATE_COMPLETED branch of fiber_run(). */ + * waiting to run. Skip the dispatch round-trip and settle the future right here — + * mirroring the FIBER_STATE_COMPLETED branch of fiber_run(). */ fiber->result = -ECANCELED; fiber->state = FIBER_STATE_COMPLETED; fiber_resolve(f); return 1; } - /* Once we cancel a fiber, we want to immediately resume it with -ECANCELED. */ - r = sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT); + /* Queue -ECANCELED as the resume value. sd_fiber_resume() also takes care of the + * SUSPENDED → READY transition and arming the event source. -ECANCELED is sticky once + * queued, so a concurrent async wakeup can't silently override the cancellation. */ + r = sd_fiber_resume(f, -ECANCELED); if (r < 0) return r; - fiber->state = FIBER_STATE_CANCELLED; - return 1; } @@ -444,7 +447,7 @@ static int fiber_on_defer(sd_event_source *s, void *userdata) { static int fiber_on_exit(sd_event_source *s, void *userdata) { sd_future *f = ASSERT_PTR(userdata); - Fiber *fiber = ASSERT_PTR(sd_future_get_private(f)); + Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); int r; /* The fiber may already have completed via the regular defer path before sd_event_exit() @@ -469,7 +472,7 @@ static void* fiber_alloc(void) { } static void fiber_free(sd_future *f) { - Fiber *fiber = ASSERT_PTR(sd_future_get_private(f)); + Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); /* To make sure all memory is deallocated, the fiber has to have completed by the time we free it to * make sure its stack has finished unwinding (which will invoke the registered cleanup functions). @@ -498,42 +501,46 @@ static void fiber_free(sd_future *f) { sd_event_source_disable_unref(fiber->defer_event_source); sd_event_source_disable_unref(fiber->exit_event_source); - sd_event_unref(fiber->event); free(fiber->name); free(fiber); } sd_future* sd_fiber_get_current(void) { - Fiber *f = fiber_get_current(); - if (!f) - return NULL; - - return sd_event_source_get_userdata(fiber_current_event_source(f)); + return current_fiber; } int sd_fiber_is_running(void) { - return !!fiber_get_current(); + return !!current_fiber; } sd_event* sd_fiber_get_event(void) { - Fiber *f = fiber_get_current(); + sd_future *f = sd_fiber_get_current(); assert_return(f, NULL); - return f->event; + return sd_future_get_event(f); } int sd_fiber_get_priority(int64_t *ret) { - Fiber *f = fiber_get_current(); + sd_future *f = sd_fiber_get_current(); assert_return(ret, -EINVAL); assert_return(f, -ESRCH); - *ret = f->priority; + Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); + *ret = fiber->priority; return 0; } static int fiber_swap(FiberState state) { - Fiber *f = ASSERT_PTR(fiber_get_current()); + Fiber *f = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(sd_fiber_get_current()))); + + /* A value queued by sd_fiber_resume() while the fiber was running short-circuits the swap: + * deliver it as if we had suspended and been resumed instantly, without round-tripping + * through the event loop. */ + if (f->result_pending) { + f->result_pending = false; + return TAKE_GENERIC(f->result, int, 0); + } f->state = state; @@ -546,26 +553,23 @@ static int fiber_swap(FiberState state) { finish_switch_stack(fake_stack_save); - /* When we get here, we've been resumed. */ - - if (f->state == FIBER_STATE_CANCELLED) - return -ECANCELED; - - /* sd_fiber_resume() stashes the resumer's value (an async wakeup error from a deadline - * timer, an io_uring CQE result, etc.) into f->result for us to surface here. Consume it - * unconditionally so it doesn't pollute subsequent suspends or the fiber's eventual return - * value — both negative errors and positive payloads (byte counts, accepted fds, revents - * masks) are valid resume values. */ + /* When we get here, we've been resumed. sd_fiber_resume() stashed the resumer's value + * (an async wakeup error from a deadline timer, an io_uring CQE result, a -ECANCELED + * from fiber_cancel(), etc.) into f->result for us to surface here. Consume it + * unconditionally so it doesn't pollute subsequent suspends or the fiber's eventual + * return value — both negative errors and positive payloads (byte counts, accepted fds, + * revents masks) are valid resume values. */ + f->result_pending = false; return TAKE_GENERIC(f->result, int, 0); } int sd_fiber_yield(void) { - assert_return(fiber_get_current(), -ESRCH); + assert_return(sd_fiber_get_current(), -ESRCH); return fiber_swap(FIBER_STATE_READY); } int sd_fiber_suspend(void) { - assert_return(fiber_get_current(), -ESRCH); + assert_return(sd_fiber_get_current(), -ESRCH); return fiber_swap(FIBER_STATE_SUSPENDED); } @@ -591,13 +595,30 @@ int sd_fiber_resume(sd_future *f, int result) { assert_return(f, -EINVAL); assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL); - Fiber *fiber = ASSERT_PTR(sd_future_get_private(f)); + Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); - if (fiber->state != FIBER_STATE_SUSPENDED) + /* Nothing to deliver to once the fiber has terminated. */ + if (fiber->state == FIBER_STATE_COMPLETED) + return 0; + + /* -ECANCELED is sticky once queued: a concurrent async wakeup (timer, io_uring CQE, …) + * mustn't silently override a pending cancellation. We only refuse the override if the + * new value is something *other* than -ECANCELED, so a second cancel-after-cancel is a + * harmless no-op via the equality. */ + if (fiber->result_pending && fiber->result == -ECANCELED && result != -ECANCELED) return 0; - /* Stash the result so fiber_swap() returns it from sd_fiber_suspend(). */ + /* Stash the result so fiber_swap() returns it from the next sd_fiber_suspend() (or + * sd_fiber_yield()). When the fiber has not yet suspended (state INITIAL or READY) the value + * is just queued — the next fiber_swap() consumes it without actually yielding to the event + * loop. This lets callers like sd_future_cancel_wait_unref() forward a cancellation/timeout + * they observed on the fiber's behalf instead of silently swallowing it. */ fiber->result = result; + fiber->result_pending = true; + + if (fiber->state != FIBER_STATE_SUSPENDED) + return 0; + fiber->state = FIBER_STATE_READY; return sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT); } @@ -622,8 +643,8 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED)) return -ECANCELED; - _cleanup_(sd_future_unrefp) sd_future *f = NULL; - r = sd_future_new(&fiber_future_ops, &f); + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + r = sd_future_new(e, &fiber_future_ops, &f); if (r < 0) return r; @@ -647,7 +668,6 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user .name = strdup(name), .func = func, .userdata = userdata, - .event = sd_event_ref(e), }; if (!fiber->name) return -ENOMEM; @@ -665,7 +685,7 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user (uint8_t*) usable.iov_base + usable.iov_len); #endif - r = fiber_init(fiber); + r = fiber_init(f); if (r < 0) return r; @@ -746,7 +766,7 @@ int sd_fiber_get_floating(sd_future *f) { } int sd_fiber_sleep(uint64_t usec) { - Fiber *f = fiber_get_current(); + sd_future *f = sd_fiber_get_current(); int r; if (!f) @@ -760,11 +780,9 @@ int sd_fiber_sleep(uint64_t usec) { if (usec == USEC_INFINITY) return sd_fiber_suspend(); - assert(f->event); - _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL; r = future_new_time_relative( - f->event, + sd_future_get_event(f), CLOCK_MONOTONIC, usec, /* accuracy= */ 1, @@ -773,7 +791,7 @@ int sd_fiber_sleep(uint64_t usec) { if (r < 0) return r; - return sd_fiber_suspend(); + return sd_fiber_await(timer); } int sd_fiber_await(sd_future *target) { @@ -784,36 +802,38 @@ int sd_fiber_await(sd_future *target) { assert_return(target, -EINVAL); assert_return(target != f, -EDEADLK); - Fiber *fiber = ASSERT_PTR(sd_future_get_private(f)); - if (sd_future_state(target) == SD_FUTURE_RESOLVED) return sd_future_result(target); /* Note that we do allow waiting for other fibers when the event loop is exiting, since waiting for * other fibers does not require adding new event sources to the event loop. */ - if (sd_event_get_state(fiber->event) == SD_EVENT_FINISHED) + if (sd_event_get_state(sd_future_get_event(f)) == SD_EVENT_FINISHED) return -ECANCELED; - _cleanup_(sd_future_cancel_wait_unrefp) sd_future *wait = NULL; - r = sd_future_new_wait(target, &wait); + _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot = NULL; + r = sd_future_add_callback(target, &slot, sd_future_resume_callback, f); if (r < 0) return r; - return sd_fiber_suspend(); + r = sd_fiber_suspend(); + if (r < 0) + return r; + + return sd_future_result(target); } sd_future* sd_fiber_timeout(uint64_t timeout) { - Fiber *fiber = fiber_get_current(); + sd_future *self = sd_fiber_get_current(); int r; - assert_return(fiber, NULL); + assert_return(self, NULL); if (timeout == USEC_INFINITY) return NULL; - sd_future *timer; + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL; r = future_new_time_relative( - fiber->event, + sd_future_get_event(self), CLOCK_MONOTONIC, timeout, /* accuracy= */ 1, @@ -823,5 +843,12 @@ sd_future* sd_fiber_timeout(uint64_t timeout) { return NULL; /* On allocation failure no timer is armed and the scope becomes a no-op. * Errors here are rare; if the caller cares they can compare to NULL. */ - return timer; + /* The whole point of SD_FIBER_TIMEOUT is to wake the calling fiber when the deadline + * fires (so a later sd_fiber_suspend / sd_fiber_await returns -ETIME from this timer's + * resolve). Install a floating resume callback bound to the timer's lifetime. */ + r = sd_future_add_callback(timer, /* ret_slot= */ NULL, sd_future_resume_callback, self); + if (r < 0) + return NULL; + + return TAKE_PTR(timer); } diff --git a/src/libsystemd/sd-future/future-group.c b/src/libsystemd/sd-future/future-group.c new file mode 100644 index 0000000000000..9be10df664112 --- /dev/null +++ b/src/libsystemd/sd-future/future-group.c @@ -0,0 +1,319 @@ +/* SPDX-License-Identifier: LGPL-2.1-or-later */ + +#include + +#include "sd-future.h" + +#include "alloc-util.h" +#include "errno-util.h" +#include "macro.h" + +typedef struct FutureGroup { + uint64_t policy; + + sd_future_slot **slots; + size_t n_slots; + + /* The fiber the group was created on (captured at sd_future_group_new()). When the + * group settles on an error and IGNORE_ERRORS is unset, this fiber is cancelled so it + * notices the failure even if it hasn't started awaiting the group — mirroring + * asyncio.TaskGroup's child-error-cancels-parent-task behaviour. parent_slot's + * callback NULLs `parent` if the parent resolves before the group does. + * + * parent_awaiting is set by sd_future_group_await(): it signals that someone is + * already going to observe the group's resolution via the await path, so cancelling + * the parent would just replace the group's real error with -ECANCELED. */ + sd_future *parent; + sd_future_slot *parent_slot; + bool parent_awaiting; + + /* Set once future_group_finalize() has been entered. The outcome is decided (stored in + * `result`) and the group is "draining" — waiting for any still-pending children to + * actually settle before we resolve. While set, the result cannot change and add + * rejects with -ESTALE. */ + bool finalizing; + int result; + + /* Reentrancy guard: set while finalize() is iterating slots so that cascading + * child_resolved callbacks (from synchronous cancellations) don't re-run + * future_group_check() and re-scan the slot vector mid-loop. */ + bool resolving; +} FutureGroup; + +static void* future_group_alloc(void) { + return new0(FutureGroup, 1); +} + +static void future_group_free(sd_future *f) { + FutureGroup *fg = ASSERT_PTR(sd_future_get_private(f)); + + sd_future_slot_unref(fg->parent_slot); + FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) + sd_future_slot_unref(*slot_p); + free(fg->slots); + free(fg); +} + +static int future_group_parent_resolved(sd_future *parent, void *userdata) { + FutureGroup *fg = ASSERT_PTR(userdata); + fg->parent = NULL; + return 0; +} + +static int future_group_check(sd_future *g); + +static int future_group_finalize(sd_future *g, int result) { + FutureGroup *fg = ASSERT_PTR(sd_future_get_private(g)); + int r = 0; + + if (fg->finalizing) + /* Outcome already locked: ignore subsequent attempts. Mirrors the old "group + * is already RESOLVED, so further cancels are no-ops" behaviour. */ + return 0; + + fg->finalizing = true; + fg->result = result; + + fg->resolving = true; + FOREACH_ARRAY(slot, fg->slots, fg->n_slots) { + sd_future *child = sd_future_slot_get_future(*slot); + if (sd_future_state(child) == SD_FUTURE_PENDING) + RET_GATHER(r, sd_future_cancel(child)); + } + fg->resolving = false; + + /* If we're settling because of an error (and the user hasn't opted into ignoring + * errors), cancel the parent fiber so it notices the failure even if it hasn't + * started awaiting the group yet — matches asyncio.TaskGroup's _on_task_done. Skip + * when parent is the currently-running fiber (e.g. the parent itself just called + * sd_future_cancel(group)): fiber_cancel asserts against self-cancellation. */ + if (result < 0 && + !(fg->policy & SD_FUTURE_GROUP_IGNORE_ERRORS) && + !fg->parent_awaiting && + fg->parent && + fg->parent != sd_fiber_get_current()) + RET_GATHER(r, sd_future_cancel(fg->parent)); + + /* Re-check: if every child settled synchronously during the cancel loop the group can + * resolve now; otherwise wait for the group_child_resolved callbacks to drive the + * drain branch of check(). */ + RET_GATHER(r, future_group_check(g)); + return r; +} + +static int future_group_check(sd_future *g) { + FutureGroup *fg = ASSERT_PTR(sd_future_get_private(g)); + + if (sd_future_state(g) == SD_FUTURE_RESOLVED) + return 0; + if (fg->resolving) + return 0; + + if (fg->finalizing) { + /* Outcome decided; resolve once every child has actually settled so callers + * observing the group's resolution see every child in RESOLVED state. An empty + * finalizing group resolves immediately (the FOREACH_ARRAY body never runs). */ + FOREACH_ARRAY(slot, fg->slots, fg->n_slots) + if (sd_future_state(sd_future_slot_get_future(*slot)) != SD_FUTURE_RESOLVED) + return 0; + return sd_future_resolve(g, fg->result); + } + + if (fg->n_slots == 0) + /* Empty group has nothing to wait for: leave it pending so the user can still + * add children (or cancel the group). Otherwise an early set_policy on a + * fresh group would settle it before any child got added. */ + return 0; + + bool wait_any = fg->policy & SD_FUTURE_GROUP_WAIT_ANY; + bool ignore_errors = fg->policy & SD_FUTURE_GROUP_IGNORE_ERRORS; + + size_t n_resolved = 0; + int first_error = 0, first_success = 0; + bool any_success = false; + + FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) { + sd_future *child = sd_future_slot_get_future(*slot_p); + if (sd_future_state(child) != SD_FUTURE_RESOLVED) + continue; + + n_resolved++; + int cr = sd_future_result(child); + if (cr < 0) { + if (first_error == 0) + first_error = cr; + } else if (!any_success) { + any_success = true; + first_success = cr; + } + } + + bool all_done = (n_resolved == fg->n_slots); + + int result; + if (wait_any && any_success) + result = first_success; /* wait_any short-circuits on first success */ + else if (!ignore_errors && first_error != 0) + result = first_error; /* fail-fast on error unless ignored */ + else if (all_done) + result = first_error; /* everyone settled: 0 if no errors */ + else + return 0; + + return future_group_finalize(g, result); +} + +static int future_group_cancel(sd_future *f) { + return future_group_finalize(f, -ECANCELED); +} + +static int future_group_set_priority(sd_future *f, int64_t priority) { + FutureGroup *fg = ASSERT_PTR(sd_future_get_private(f)); + int r = 0; + + FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) { + int q = sd_future_set_priority(sd_future_slot_get_future(*slot_p), priority); + /* -EOPNOTSUPP: impl doesn't support priorities. + * -ESTALE: child already resolved — expected during a group's lifetime. */ + if (q < 0 && !IN_SET(q, -EOPNOTSUPP, -ESTALE)) + RET_GATHER(r, q); + } + + return r; +} + +static const sd_future_ops future_group_ops = { + .size = sizeof(sd_future_ops), + .alloc = future_group_alloc, + .free = future_group_free, + .cancel = future_group_cancel, + .set_priority = future_group_set_priority, +}; + +int sd_future_group_new(sd_event *e, sd_future **ret) { + int r; + + assert_return(e, -EINVAL); + assert_return(ret, -EINVAL); + + _cleanup_(sd_future_cancel_unrefp) sd_future *g = NULL; + r = sd_future_new(e, &future_group_ops, &g); + if (r < 0) + return r; + + sd_future *parent = sd_fiber_get_current(); + if (parent) { + FutureGroup *fg = sd_future_get_private(g); + r = sd_future_add_callback(parent, &fg->parent_slot, future_group_parent_resolved, fg); + if (r < 0) + return r; + fg->parent = parent; + } + + *ret = TAKE_PTR(g); + return 0; +} + +int sd_future_group_size(sd_future *f, size_t *ret) { + assert_return(f, -EINVAL); + assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL); + assert_return(ret, -EINVAL); + + FutureGroup *fg = sd_future_get_private(f); + *ret = fg->n_slots; + return 0; +} + +int sd_future_group_set_policy(sd_future *f, uint64_t policy) { + assert_return(f, -EINVAL); + assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL); + assert_return(sd_future_state(f) == SD_FUTURE_PENDING, -ESTALE); + assert_return((policy & ~(uint64_t)(SD_FUTURE_GROUP_WAIT_ANY | SD_FUTURE_GROUP_IGNORE_ERRORS)) == 0, -EINVAL); + + /* Policy must be configured before any children are added — once a child is in flight, + * the resolution mechanics are locked in. This keeps the API friction-free: callers + * don't have to reason about mid-flight reshuffling of which children get cancelled. */ + FutureGroup *fg = sd_future_get_private(f); + if (fg->n_slots > 0) + return -ESTALE; + + fg->policy = policy; + return 0; +} + +static int group_child_resolved(sd_future *child, void *userdata) { + sd_future *g = ASSERT_PTR(userdata); + return future_group_check(g); +} + +int sd_future_group_add(sd_future *f, sd_future *child) { + int r; + + assert_return(f, -EINVAL); + assert_return(child, -EINVAL); + assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL); + assert_return(sd_future_state(f) == SD_FUTURE_PENDING, -ESTALE); + + FutureGroup *fg = sd_future_get_private(f); + if (fg->finalizing) + /* Group is draining: a freshly-added pending child would have missed the + * cancel loop and hang us forever waiting for it to settle. */ + return -ESTALE; + + if (!GREEDY_REALLOC(fg->slots, fg->n_slots + 1)) + return -ENOMEM; + + sd_future_slot *slot = NULL; + r = sd_future_add_callback(child, &slot, group_child_resolved, f); + if (r < 0) + return r; + + fg->slots[fg->n_slots++] = slot; + + return 0; +} + +int sd_future_group_add_many(sd_future *f, ...) { + assert_return(f, -EINVAL); + assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL); + + FutureGroup *fg = sd_future_get_private(f); + size_t before = fg->n_slots; + int r = 0; + + va_list ap; + va_start(ap, f); + for (;;) { + sd_future *child = va_arg(ap, sd_future*); + if (!child) + break; + + r = sd_future_group_add(f, child); + if (r < 0) + break; + } + va_end(ap); + + if (r < 0) + /* Roll back this call's additions. */ + while (fg->n_slots > before) { + sd_future_slot_unref(fg->slots[--fg->n_slots]); + fg->slots[fg->n_slots] = NULL; + } + + return r; +} + +int sd_future_group_await(sd_future *f) { + assert_return(f, -EINVAL); + assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL); + + /* Signal that someone is taking responsibility for the group's result via the await + * path, so finalize() won't also cancel the parent fiber (which would replace the + * group's real error with -ECANCELED). */ + FutureGroup *fg = sd_future_get_private(f); + fg->parent_awaiting = true; + + return sd_fiber_await(f); +} + diff --git a/src/libsystemd/sd-future/sd-future.c b/src/libsystemd/sd-future/sd-future.c index ba3a1c52d2fa6..8a4d2b02607d6 100644 --- a/src/libsystemd/sd-future/sd-future.c +++ b/src/libsystemd/sd-future/sd-future.c @@ -1,5 +1,8 @@ /* SPDX-License-Identifier: LGPL-2.1-or-later */ +#include + +#include "sd-event.h" #include "sd-future.h" #include "alloc-util.h" @@ -8,16 +11,33 @@ #include "macro.h" #include "set.h" +struct sd_future_slot { + unsigned n_ref; + + /* Back-pointer to the future the slot is attached to. + * + * Ref ownership is asymmetric (same trick as sd_bus_slot/bus->slots): when the slot + * is non-floating the SLOT owns a ref on the future; when floating, the FUTURE owns + * a ref on the slot. So `slots` is always a borrowed pointer collection regardless. */ + sd_future *future; + bool floating; + + sd_future_func_t callback; + void *userdata; + + sd_event_source *defer_source; + sd_event_source *exit_source; +}; + struct sd_future { unsigned n_ref; int state; int result; - Set *waiters; + sd_event *event; - sd_future_func_t callback; - void *userdata; + Set *slots; const sd_future_ops *ops; @@ -27,91 +47,159 @@ struct sd_future { void *private; }; -static int fiber_resume_trampoline(sd_future *f) { +static int dispatch_slot(sd_future_slot *s, sd_future *f) { + /* Invoked when the slot's chosen event source fires. Hold a self-ref on `f` + * across the callback: a floating slot lives in f->slots, so if the callback + * drops the last user-side ref to f, sd_future_free would iterate floating + * slots and unref us — pulling the rug out from under the `sd_future_slot_unref` + * below. Holding the ref keeps f alive until we've cleaned ourselves up; the + * cleanup at scope exit drops it, freeing f if no one else held a ref. */ + bool floating = s->floating; /* capture: non-floating s may be freed by callback */ + _unused_ _cleanup_(sd_future_unrefp) sd_future *self = sd_future_ref(f); + + int r = s->callback(f, s->userdata); + if (floating) + sd_future_slot_unref(s); + return r; +} + +static sd_event_source* slot_current_event_source(sd_future_slot *s) { + assert(s); + assert(s->future); + + return sd_event_get_state(s->future->event) == SD_EVENT_EXITING + ? s->exit_source + : s->defer_source; +} + +static int slot_arm(sd_future_slot *s) { + return sd_event_source_set_enabled(slot_current_event_source(s), SD_EVENT_ONESHOT); +} + +int sd_future_resume_callback(sd_future *f, void *userdata) { /* The future's result is what the fiber should resume with. Impls choose the value at * resolution time — e.g. a deadline timer resolves with -ETIME, a wait future resolves * with the target's result, a normal IO/sleep future resolves with 0 on success. */ - return sd_fiber_resume(sd_future_get_userdata(f), sd_future_result(f)); + return sd_fiber_resume(userdata, sd_future_result(f)); } int sd_future_resolve(sd_future *f, int result) { int r = 0; assert_return(f, -EINVAL); - - if (f->state != SD_FUTURE_PENDING) - return 0; - - /* Hold a self-ref across callback/waiter dispatch: callbacks (e.g. bus_fiber_resolved() - * dropping the tracking-set's ref) may legitimately release what would otherwise be the - * last reference, and we still access f->waiters below. The cleanup unrefs at scope exit, - * which is when freeing is safe again. */ - _unused_ _cleanup_(sd_future_unrefp) sd_future *self = sd_future_ref(f); + assert_return(f->state == SD_FUTURE_PENDING, -ESTALE); f->state = SD_FUTURE_RESOLVED; f->result = result; - if (f->callback) - RET_GATHER(r, f->callback(f)); - - /* We'd like the set to not be modified while iterating over it, hence take ownership over it in - * a local variable. Otherwise code invoked via sd_future_resolve() could try to modify the set while - * we're iterating over it (for example wait_future_free()). */ - Set *waiters = TAKE_PTR(f->waiters); - sd_future *w; - SET_FOREACH(w, waiters) - RET_GATHER(r, sd_future_resolve(w, result)); - - set_free(waiters); + /* Enable each slot's currently-applicable event source so the callback fires on + * the next loop iteration. The always-defer model frees the resolver from + * reentrancy concerns, slot-set mutation during iteration, and callbacks dropping + * the future's last ref. */ + sd_future_slot *s; + SET_FOREACH(s, f->slots) + RET_GATHER(r, slot_arm(s)); return r; } static sd_future* sd_future_free(sd_future *f) { - if (!f) - return NULL; - - if (f->state == SD_FUTURE_PENDING) - sd_future_resolve(f, -ECANCELED); - - set_free(f->waiters); + /* By the time we tear down, the future must have reached a terminal state. Callers + * abandoning a still-PENDING future must drive it to RESOLVED first — typically via + * sd_future_cancel_unref() (non-fiber, synchronous-cancel impls) or + * sd_future_cancel_wait_unref() (fiber, awaits actual resolution).. */ + assert(f->state == SD_FUTURE_RESOLVED); + + /* Any slot still in f->slots at this point must be floating: non-floating slots own + * a ref on f, so if any existed we wouldn't have reached free. (Slots can be added + * post-resolution via sd_future_add_callback — those are floating and may not have + * had their defer tick yet.) Tear them down by dropping the future's ref. */ + Set *slots = TAKE_PTR(f->slots); + sd_future_slot *s; + SET_FOREACH(s, slots) { + assert(s->floating); + sd_future_slot_unref(s); + } + set_free(slots); if (f->ops->free) f->ops->free(f); + sd_event_unref(f->event); return mfree(f); } +/* Unref is a pure refcount op: dropping a ref does not resolve or cancel. If a caller wants + * pending work to complete (and floating callbacks to observe the outcome) before release, + * they must drive resolution explicitly — typically via sd_future_cancel() or + * sd_future_cancel_wait_unref(). Reaching sd_future_free() with state PENDING is a programming + * error and trips an assert: callers must drive the future to RESOLVED before dropping the last + * ref. */ DEFINE_TRIVIAL_REF_UNREF_FUNC(sd_future, sd_future, sd_future_free); + DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_unref); DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_unref); -sd_future* sd_future_cancel_wait_unref(sd_future *f) { +sd_future* sd_future_cancel_unref(sd_future *f) { int r; if (!f) return NULL; - /* We have to be able to suspend until the fiber we're waiting for finishes, and that's only - * possible if we're running on a fiber ourselves. */ - if (!sd_fiber_is_running()) - return sd_future_unref(f); - + /* Synchronous-cancel teardown for non-fiber contexts: drive the future to RESOLVED via + * ops->cancel, then drop the ref. Safe for impls whose ops->cancel resolves synchronously. + * For impls whose cancel is asynchronous, the future stays PENDING after this call and + * sd_future_free's strict assert will trip. Callers in that situation must use + * sd_future_cancel_wait_unref() from a fiber to await the actual resolution. */ r = sd_future_cancel(f); if (r < 0) log_debug_errno(r, "Failed to cancel future, ignoring: %m"); - if (f->state == SD_FUTURE_PENDING) { - /* Fast path: when f's resolve callback already targets the current fiber (the default for - * futures created on this fiber), we can suspend directly and let the existing trampoline - * wake us up — no need to allocate a wait future just to learn about the resolution. - * Otherwise fall back to sd_fiber_await() which sets up an explicit waiter. */ - if (f->callback == fiber_resume_trampoline && f->userdata == sd_fiber_get_current()) - r = sd_fiber_suspend(); - else - r = sd_fiber_await(f); - if (r < 0 && r != -ECANCELED) - log_debug_errno(r, "Failed to wait for future to finish, ignoring: %m"); + return sd_future_unref(f); +} + +DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_cancel_unref); +DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_cancel_unref); + +sd_future* sd_future_cancel_wait_unref(sd_future *f) { + int r, q = 0; + + if (!f) + return NULL; + + /* Caller must be on a fiber: the wait step parks the calling fiber on the future until it + * actually resolves. Callers without a fiber should use sd_future_cancel_unref(), which + * works for impls whose cancel is synchronous. */ + assert(sd_fiber_is_running()); + + for (;;) { + r = sd_future_cancel(f); + if (r < 0) + log_debug_errno(r, "Failed to cancel future, ignoring: %m"); + + if (sd_future_state(f) != SD_FUTURE_PENDING) + break; + + r = sd_fiber_await(f); + if (r < 0) { + if (r != -ECANCELED) + log_debug_errno(r, "Failed to wait for future to finish, ignoring: %m"); + /* The await was interrupted by something targeting the calling fiber (a + * cancellation, an outer SD_FIBER_TIMEOUT firing, …). We have to keep looping + * until `f` actually resolves so unref is safe, so we can't honor it inline + * — but we mustn't drop it either. Remember the most recent one and re-queue + * it on the fiber once just before we return. */ + q = r; + } + + if (sd_future_state(f) != SD_FUTURE_PENDING) + break; + } + + if (q < 0) { + r = sd_fiber_resume(sd_fiber_get_current(), q); + if (r < 0) + log_debug_errno(r, "Failed to re-queue interruption (%i) on calling fiber, ignoring: %m", q); } return sd_future_unref(f); @@ -120,11 +208,13 @@ sd_future* sd_future_cancel_wait_unref(sd_future *f) { DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_cancel_wait_unref); DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_cancel_wait_unref); -int sd_future_new(const sd_future_ops *ops, sd_future **ret) { +int sd_future_new(sd_event *e, const sd_future_ops *ops, sd_future **ret) { + assert_return(e, -EINVAL); assert_return(ops, -EINVAL); assert_return(ops->size >= endoffsetof_field(sd_future_ops, set_priority), -EINVAL); assert_return(ops->alloc, -EINVAL); assert_return(ops->free, -EINVAL); + assert_return(ops->cancel, -EINVAL); assert_return(ret, -EINVAL); sd_future *f = new(sd_future, 1); @@ -135,26 +225,25 @@ int sd_future_new(const sd_future_ops *ops, sd_future **ret) { .n_ref = 1, .state = SD_FUTURE_PENDING, .ops = ops, + .event = sd_event_ref(e), }; f->private = ops->alloc(); if (!f->private) { + sd_event_unref(f->event); free(f); return -ENOMEM; } - /* If we're being created on a fiber, default the callback to resuming that fiber on resolve — - * this is almost always what you want, and it saves the usual set_callback boilerplate before - * sd_fiber_suspend(). Callers that want different behavior can override with - * sd_future_set_callback(). */ - sd_future *fiber = sd_fiber_get_current(); - if (fiber) - (void) sd_future_set_callback(f, fiber_resume_trampoline, fiber); - *ret = f; return 0; } +sd_event* sd_future_get_event(sd_future *f) { + assert_return(f, NULL); + return f->event; +} + int sd_future_state(sd_future *f) { assert_return(f, -EINVAL); return f->state; @@ -166,11 +255,6 @@ int sd_future_result(sd_future *f) { return f->result; } -void* sd_future_get_userdata(sd_future *f) { - assert_return(f, NULL); - return f->userdata; -} - void* sd_future_get_private(sd_future *f) { assert_return(f, NULL); return f->private; @@ -181,83 +265,115 @@ const sd_future_ops* sd_future_get_ops(sd_future *f) { return f->ops; } -int sd_future_set_callback(sd_future *f, sd_future_func_t callback, void *userdata) { - assert_return(f, -EINVAL); - - f->callback = callback; - f->userdata = userdata; - return 0; +sd_future* sd_future_slot_get_future(sd_future_slot *s) { + assert_return(s, NULL); + return s->future; } -int sd_future_set_priority(sd_future *f, int64_t priority) { - assert_return(f, -EINVAL); - assert_return(f->state == SD_FUTURE_PENDING, -ESTALE); - assert_return(f->ops->set_priority, -EOPNOTSUPP); - - return f->ops->set_priority(f, priority); -} +static sd_future_slot* sd_future_slot_free(sd_future_slot *s) { + if (!s) + return NULL; -int sd_future_cancel(sd_future *f) { - assert_return(f, -EINVAL); - assert_return(f->ops->cancel, -EOPNOTSUPP); + if (s->future) { + set_remove(s->future->slots, s); + if (!s->floating) + sd_future_unref(s->future); + } - if (f->state == SD_FUTURE_RESOLVED) - return 0; + sd_event_source_disable_unref(s->defer_source); + sd_event_source_disable_unref(s->exit_source); - return f->ops->cancel(f); + return mfree(s); } -typedef struct WaitFuture { - sd_future *target; -} WaitFuture; +DEFINE_TRIVIAL_REF_UNREF_FUNC(sd_future_slot, sd_future_slot, sd_future_slot_free); -static void* wait_future_alloc(void) { - return new0(WaitFuture, 1); +static int slot_dispatch_handler(sd_event_source *src, void *userdata) { + sd_future_slot *s = ASSERT_PTR(userdata); + return dispatch_slot(s, s->future); } -static void wait_future_free(sd_future *f) { - WaitFuture *wf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); +int sd_future_add_callback(sd_future *f, sd_future_slot **ret_slot, sd_future_func_t callback, void *userdata) { + int r; - set_remove(wf->target->waiters, f); - sd_future_unref(wf->target); - free(wf); -} + assert_return(f, -EINVAL); + assert_return(callback, -EINVAL); -static int wait_future_cancel(sd_future *f) { - WaitFuture *wf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f))); + _cleanup_(sd_future_slot_unrefp) sd_future_slot *s = new(sd_future_slot, 1); + if (!s) + return -ENOMEM; - set_remove(wf->target->waiters, f); - return sd_future_resolve(f, -ECANCELED); -} + *s = (sd_future_slot) { + .n_ref = 1, + .future = f, + .floating = ret_slot == NULL, + .callback = callback, + .userdata = userdata, + }; -static const sd_future_ops wait_future_ops = { - .size = sizeof(sd_future_ops), - .alloc = wait_future_alloc, - .free = wait_future_free, - .cancel = wait_future_cancel, -}; + /* Asymmetric ownership (same trick as sd_bus_slot): non-floating slot owns the + * future, floating slot is owned by it — avoids a cycle in either direction. */ + if (!s->floating) + sd_future_ref(f); -int sd_future_new_wait(sd_future *target, sd_future **ret) { - int r; + /* Never run the callback inline, always use a defer event source to schedule it, + * even if the future is already resolved. This simplifies callers which now don't + * have to worry about the callback being potentially called inline. */ - assert_return(target, -EINVAL); - assert_return(ret, -EINVAL); + r = sd_event_add_defer(f->event, &s->defer_source, slot_dispatch_handler, s); + if (r < 0) + return r; - _cleanup_(sd_future_unrefp) sd_future *f = NULL; - r = sd_future_new(&wait_future_ops, &f); + r = sd_event_source_set_enabled(s->defer_source, SD_EVENT_OFF); if (r < 0) return r; - WaitFuture *wf = sd_future_get_private(f); - wf->target = sd_future_ref(target); + r = sd_event_add_exit(f->event, &s->exit_source, slot_dispatch_handler, s); + if (r < 0) + return r; + + r = sd_event_source_set_enabled(s->exit_source, SD_EVENT_OFF); + if (r < 0) + return r; + + if (sd_fiber_is_running()) { + int64_t priority; + if (sd_fiber_get_priority(&priority) >= 0) { + (void) sd_event_source_set_priority(s->defer_source, priority); + (void) sd_event_source_set_priority(s->exit_source, priority); + } + } - if (target->state == SD_FUTURE_RESOLVED) - r = sd_future_resolve(f, target->result); - else - r = set_ensure_put(&target->waiters, &trivial_hash_ops, f); + if (f->state == SD_FUTURE_RESOLVED) { + r = slot_arm(s); + if (r < 0) + return r; + } + + r = set_ensure_put(&f->slots, &trivial_hash_ops, s); if (r < 0) return r; - *ret = TAKE_PTR(f); + if (!s->floating) + *ret_slot = s; + + TAKE_PTR(s); return 0; } + +int sd_future_set_priority(sd_future *f, int64_t priority) { + assert_return(f, -EINVAL); + assert_return(f->state == SD_FUTURE_PENDING, -ESTALE); + assert_return(f->ops->set_priority, -EOPNOTSUPP); + + return f->ops->set_priority(f, priority); +} + +int sd_future_cancel(sd_future *f) { + assert_return(f, -EINVAL); + + if (f->state == SD_FUTURE_RESOLVED) + return 0; + + return f->ops->cancel(f); +} diff --git a/src/libsystemd/sd-future/test-fiber.c b/src/libsystemd/sd-future/test-fiber.c index 2760b919a0045..63f54009f7655 100644 --- a/src/libsystemd/sd-future/test-fiber.c +++ b/src/libsystemd/sd-future/test-fiber.c @@ -817,7 +817,7 @@ TEST(fiber_floating) { ASSERT_EQ(counter, 2); } -static int drop_extra_ref(sd_future *f) { +static int drop_extra_ref(sd_future *f, void *userdata) { /* Drop an extra ref the test installed before the callback fires. After this returns, the * floating self-ref is the only thing keeping the future alive — exercising the path where * the floating unref in fiber_run() is the last unref. */ @@ -836,9 +836,10 @@ TEST(fiber_floating_callback_drops_ref) { ASSERT_OK(sd_fiber_set_floating(f, true)); - /* Bump the ref for the callback to drop, then install the callback. */ + /* Bump the ref for the callback to drop, then install the callback (floating slot, + * lifetime bound to f). */ sd_future_ref(f); - ASSERT_OK(sd_future_set_callback(f, drop_extra_ref, NULL)); + ASSERT_OK(sd_future_add_callback(f, NULL, drop_extra_ref, NULL)); /* Drop our handle. Refs remaining: floating self-ref + the extra ref the callback will drop. */ f = sd_future_unref(f); @@ -992,6 +993,232 @@ TEST(fiber_timeout_nested) { ASSERT_EQ(fired, 2); } +/* Test: sd_future_cancel_wait_unref() loops on cancel + await until the future actually + * resolves, even if an outer SD_FIBER_TIMEOUT interrupts the await early. The stubborn + * future below requires multiple cancels to resolve, so a single cancel + interrupted await + * leaves it pending — only the loop in sd_future_cancel_wait_unref() can drive it to + * resolution. */ +typedef struct StubbornFuture { + unsigned cancels_received; + unsigned cancels_needed; + unsigned *external_counter; +} StubbornFuture; + +static void* stubborn_alloc(void) { + return new0(StubbornFuture, 1); +} + +static void stubborn_free(sd_future *f) { + free(sd_future_get_private(f)); +} + +static int stubborn_cancel(sd_future *f) { + StubbornFuture *sf = ASSERT_PTR(sd_future_get_private(f)); + sf->cancels_received++; + if (sf->external_counter) + *sf->external_counter = sf->cancels_received; + if (sf->cancels_received >= sf->cancels_needed) + return sd_future_resolve(f, -ECANCELED); + return 0; +} + +static const sd_future_ops stubborn_future_ops = { + .size = sizeof(sd_future_ops), + .alloc = stubborn_alloc, + .free = stubborn_free, + .cancel = stubborn_cancel, +}; + +static int cancel_wait_loops_fiber(void *userdata) { + unsigned *cancel_count = ASSERT_PTR(userdata); + sd_future *f = NULL; + int r; + + r = sd_future_new(sd_fiber_get_event(), &stubborn_future_ops, &f); + if (r < 0) + return r; + + StubbornFuture *sf = sd_future_get_private(f); + sf->cancels_needed = 2; + sf->external_counter = cancel_count; + + /* Short timeout interrupts the first await before the future resolves. The loop in + * sd_future_cancel_wait_unref() must call cancel a second time to drive the + * stubborn future to resolution. */ + SD_FIBER_TIMEOUT(5 * USEC_PER_MSEC); + sd_future_cancel_wait_unref(f); + return 0; +} + +TEST(fiber_cancel_wait_unref_loops_until_resolved) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + unsigned cancel_count = 0; + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + ASSERT_OK(sd_fiber_new(e, "stubborn", cancel_wait_loops_fiber, &cancel_count, NULL, &f)); + + ASSERT_OK(sd_event_loop(e)); + /* Loop must have called cancel at least twice — once before the timeout interrupted + * the await, then again on the next iteration to actually resolve the future. */ + ASSERT_GE(cancel_count, 2u); +} + +/* Test: sd_fiber_resume() called on a running (not suspended) fiber queues the value rather than + * discarding it; the next fiber_swap() (here sd_fiber_suspend()) returns it without round-tripping + * through the event loop. If the swap actually yielded, this test would hang because nothing else + * is wired up to resume the fiber. */ +static int resume_queue_while_running_fiber(void *userdata) { + int r; + + r = sd_fiber_resume(sd_fiber_get_current(), 42); + if (r < 0) + return r; + + r = sd_fiber_suspend(); + if (r != 42) + return -EBADF; + + /* sd_fiber_yield() must also drain a queued value when it's set. */ + r = sd_fiber_resume(sd_fiber_get_current(), -EPIPE); + if (r < 0) + return r; + + r = sd_fiber_yield(); + if (r != -EPIPE) + return -EBADF; + + return 0; +} + +TEST(fiber_resume_queues_while_running) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + ASSERT_OK(sd_fiber_new(e, "resume-queue", resume_queue_while_running_fiber, NULL, NULL, &f)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK(sd_future_result(f)); +} + +/* Test: a timeout that fires during sd_future_cancel_wait_unref()'s internal await must not be + * swallowed — cancel_wait_unref re-queues it via sd_fiber_resume() so the calling fiber's next + * suspend observes -ETIME. */ +static int cancel_wait_propagates_timeout_fiber(void *userdata) { + unsigned *cancel_count = ASSERT_PTR(userdata); + sd_future *f = NULL; + int r; + + r = sd_future_new(sd_fiber_get_event(), &stubborn_future_ops, &f); + if (r < 0) + return r; + + StubbornFuture *sf = sd_future_get_private(f); + sf->cancels_needed = 2; + sf->external_counter = cancel_count; + + SD_FIBER_TIMEOUT(5 * USEC_PER_MSEC); + sd_future_cancel_wait_unref(f); + + /* The timeout that interrupted the await inside cancel_wait_unref was re-queued; this + * suspend consumes it. */ + return sd_fiber_suspend(); +} + +TEST(fiber_cancel_wait_unref_propagates_timeout) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + unsigned cancel_count = 0; + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + ASSERT_OK(sd_fiber_new(e, "propagate-timeout", cancel_wait_propagates_timeout_fiber, &cancel_count, NULL, &f)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(f), ETIME); + ASSERT_GE(cancel_count, 2u); +} + +/* Test: a cancellation of the calling fiber that lands while the fiber is suspended inside + * sd_future_cancel_wait_unref()'s internal await must not be swallowed — cancel_wait_unref + * re-queues it via sd_fiber_resume() so the next suspend on this fiber observes -ECANCELED. */ +static int cancel_wait_propagates_cancellation_fiber(void *userdata) { + unsigned *cancel_count = ASSERT_PTR(userdata); + sd_future *f = NULL; + int r; + + r = sd_future_new(sd_fiber_get_event(), &stubborn_future_ops, &f); + if (r < 0) + return r; + + StubbornFuture *sf = sd_future_get_private(f); + sf->cancels_needed = 2; + sf->external_counter = cancel_count; + + sd_future_cancel_wait_unref(f); + + /* If cancel_wait_unref had silently swallowed our cancellation, this suspend would + * actually park the fiber and the event loop would idle-exit without ever resolving the + * future — the assertion in the caller would then trip on a still-PENDING future. */ + return sd_fiber_suspend(); +} + +TEST(fiber_cancel_wait_unref_propagates_cancellation_from_main) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + unsigned cancel_count = 0; + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + ASSERT_OK(sd_fiber_new(e, "propagate-cancel", cancel_wait_propagates_cancellation_fiber, &cancel_count, NULL, &f)); + + /* One iteration drives the fiber through its first cancel + await inside + * cancel_wait_unref, leaving it suspended waiting for the stubborn future. */ + ASSERT_OK_POSITIVE(sd_event_run(e, 0)); + + /* Cancel from outside any fiber. The fiber is still suspended inside cancel_wait_unref's + * await — this queues -ECANCELED on it and re-arms its defer source. */ + ASSERT_OK_POSITIVE(sd_future_cancel(f)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(f), ECANCELED); + ASSERT_GE(cancel_count, 2u); +} + +typedef struct { + sd_future *target; +} PeerCancellerData; + +static int peer_canceller_fiber(void *userdata) { + PeerCancellerData *data = ASSERT_PTR(userdata); + return sd_future_cancel(data->target); +} + +TEST(fiber_cancel_wait_unref_propagates_cancellation_from_peer_fiber) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + unsigned cancel_count = 0; + _cleanup_(sd_future_unrefp) sd_future *target = NULL, *canceller = NULL; + ASSERT_OK(sd_fiber_new(e, "target", cancel_wait_propagates_cancellation_fiber, &cancel_count, NULL, &target)); + ASSERT_OK(sd_future_set_priority(target, 0)); + + /* Lower-priority canceller runs after target has reached cancel_wait_unref's await and + * suspended; it cancels target from within a fiber dispatch. */ + PeerCancellerData data = { .target = target }; + ASSERT_OK(sd_fiber_new(e, "canceller", peer_canceller_fiber, &data, NULL, &canceller)); + ASSERT_OK(sd_future_set_priority(canceller, 1)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK(sd_future_result(canceller)); + ASSERT_ERROR(sd_future_result(target), ECANCELED); + ASSERT_GE(cancel_count, 2u); +} + /* Test: signal mask is per-thread, not per-fiber. Changes one fiber makes via pthread_sigmask * must be visible to other fibers on the same thread, both while the modifying fiber is * suspended and after it resumes. The fiber switch (sigsetjmp/siglongjmp with savesigs=0) @@ -1168,4 +1395,252 @@ TEST(fiber_stack_guard) { ASSERT_TRUE(IN_SET(si.si_status, SIGSEGV, SIGBUS)); } +static int counting_callback(sd_future *f, void *userdata) { + int *counter = ASSERT_PTR(userdata); + (*counter)++; + return 0; +} + +/* Two callbacks on the same future both fire on resolution; a third whose slot is + * dropped before resolution never fires. */ +TEST(future_slot_lifecycle) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *target = NULL; + ASSERT_OK(sd_future_new_defer(e, 0, &target)); + + _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot_a = NULL; + _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot_b = NULL; + sd_future_slot *slot_c = NULL; + int a = 0, b = 0, c = 0; + + ASSERT_OK(sd_future_add_callback(target, &slot_a, counting_callback, &a)); + ASSERT_OK(sd_future_add_callback(target, &slot_b, counting_callback, &b)); + ASSERT_OK(sd_future_add_callback(target, &slot_c, counting_callback, &c)); + sd_future_slot_unref(slot_c); + + ASSERT_OK(sd_event_loop(e)); + + ASSERT_EQ(a, 1); + ASSERT_EQ(b, 1); + ASSERT_EQ(c, 0); +} + +TEST(future_floating_slot_fires_on_resolve) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *target = NULL; + ASSERT_OK(sd_future_new_defer(e, 0, &target)); + + int count = 0; + ASSERT_OK(sd_future_add_callback(target, NULL, counting_callback, &count)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_EQ(count, 1); +} + +static int defer_basic_fiber(void *userdata) { + int *counter = ASSERT_PTR(userdata); + _cleanup_(sd_future_unrefp) sd_future *defer = NULL; + _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot = NULL; + + ASSERT_OK(sd_future_new_defer(sd_fiber_get_event(), 0, &defer)); + ASSERT_OK(sd_future_add_callback(defer, &slot, counting_callback, counter)); + return sd_fiber_await(defer); +} + +TEST(future_new_defer_basic) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + int count = 0; + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "driver", defer_basic_fiber, &count, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + + ASSERT_EQ(count, 1); + ASSERT_OK_ZERO(sd_future_result(driver)); +} + +/* sd_future_add_callback on a RESOLVED future defers the callback to the next event-loop + * iteration; never fires inline. */ +TEST(add_callback_resolved_no_fiber_defers) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + + _cleanup_(sd_future_unrefp) sd_future *target = NULL; + ASSERT_OK(sd_future_new_defer(e, 0, &target)); + + /* Drive the loop manually — sd_event_loop would transition to FINISHED and block + * the subsequent sd_event_add_defer. */ + do + ASSERT_OK(sd_event_run(e, 0)); + while (sd_future_state(target) == SD_FUTURE_PENDING); + + int count = 0; + _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot = NULL; + ASSERT_OK(sd_future_add_callback(target, &slot, counting_callback, &count)); + ASSERT_EQ(count, 0); + + ASSERT_OK(sd_event_run(e, 0)); + ASSERT_EQ(count, 1); +} + +/* Same as above with a floating slot — also defers. */ +TEST(add_callback_resolved_no_fiber_floating_defers) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + + _cleanup_(sd_future_unrefp) sd_future *target = NULL; + ASSERT_OK(sd_future_new_defer(e, 0, &target)); + + do + ASSERT_OK(sd_event_run(e, 0)); + while (sd_future_state(target) == SD_FUTURE_PENDING); + + int count = 0; + ASSERT_OK(sd_future_add_callback(target, NULL, counting_callback, &count)); + ASSERT_EQ(count, 0); + + ASSERT_OK(sd_event_run(e, 0)); + ASSERT_EQ(count, 1); +} + +typedef struct FloatingFreedState { + sd_future *target; + int fired_count; +} FloatingFreedState; + +static int floating_freed_driver(void *userdata) { + FloatingFreedState *s = ASSERT_PTR(userdata); + + ASSERT_OK(sd_fiber_await(s->target)); + ASSERT_EQ(sd_future_state(s->target), SD_FUTURE_RESOLVED); + + ASSERT_OK(sd_future_add_callback(s->target, NULL, counting_callback, &s->fired_count)); + + /* Drop the last external ref before the defer tick. The future owns the floating + * slot; freeing it must tear the slot's defer source down rather than leave it + * firing with a stale userdata. */ + s->target = sd_future_unref(s->target); + + ASSERT_OK(sd_fiber_yield()); + return 0; +} + +TEST(add_callback_resolved_in_fiber_floating_future_freed) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + FloatingFreedState s = {}; + ASSERT_OK(sd_future_new_defer(e, 0, &s.target)); + + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "driver", floating_freed_driver, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + + ASSERT_EQ(s.fired_count, 0); +} + +typedef struct DeferCancelState { + sd_future *target; + int fired_count; +} DeferCancelState; + +static int defer_cancel_driver(void *userdata) { + DeferCancelState *s = ASSERT_PTR(userdata); + sd_future_slot *slot = NULL; + + (void) sd_fiber_await(s->target); + ASSERT_EQ(sd_future_state(s->target), SD_FUTURE_RESOLVED); + + ASSERT_OK(sd_future_add_callback(s->target, &slot, counting_callback, &s->fired_count)); + + /* Drop the slot before the defer fires; the wrapping slot owns the defer source. */ + sd_future_slot_unref(slot); + + ASSERT_OK(sd_fiber_yield()); + return 0; +} + +TEST(defer_slot_cancel_before_fire) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *target = NULL; + ASSERT_OK(sd_future_new_defer(e, 0, &target)); + + DeferCancelState s = { .target = target }; + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "driver", defer_cancel_driver, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + + ASSERT_EQ(s.fired_count, 0); +} + +/* Test: once -ECANCELED is queued on a fiber, a concurrent async wakeup with a different value + * (e.g. a timer firing, an io_uring CQE result) must not overwrite it. The fiber observes + * -ECANCELED on its next suspend, and the override value is dropped. */ +static int sticky_cancel_fiber(void *userdata) { + int r; + + /* Queue -ECANCELED on ourselves while running (state INITIAL/READY). */ + r = sd_fiber_resume(sd_fiber_get_current(), -ECANCELED); + if (r < 0) + return r; + + /* Try to override with a different value — must be a no-op. The return value of + * sd_fiber_resume is 0 either way; what we care about is the value actually + * observed by the next yield. */ + r = sd_fiber_resume(sd_fiber_get_current(), -EPIPE); + if (r < 0) + return r; + + /* Likewise: a positive override is also dropped. */ + r = sd_fiber_resume(sd_fiber_get_current(), 42); + if (r < 0) + return r; + + /* Next suspend consumes the stashed value: must be -ECANCELED, not -EPIPE or 42. */ + return sd_fiber_yield(); +} + +TEST(fiber_resume_cancellation_is_sticky) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + ASSERT_OK(sd_fiber_new(e, "sticky", sticky_cancel_fiber, NULL, NULL, &f)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(f), ECANCELED); +} + +/* Test: sd_future_new_defer() refuses to create a future when the event loop has already + * entered the EXITING/FINISHED phase — there's no future event-loop iteration left to fire + * the defer source on. Returns -ECANCELED. */ +TEST(future_new_defer_rejected_on_exiting_loop) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + + /* Drive the loop to FINISHED by asking it to exit immediately. */ + ASSERT_OK(sd_event_exit(e, 0)); + ASSERT_OK(sd_event_loop(e)); + + sd_future *f = NULL; + ASSERT_ERROR(sd_future_new_defer(e, 0, &f), ECANCELED); + ASSERT_NULL(f); +} + DEFINE_TEST_MAIN(LOG_DEBUG); diff --git a/src/libsystemd/sd-future/test-future-group.c b/src/libsystemd/sd-future/test-future-group.c new file mode 100644 index 0000000000000..4dab9cafc03be --- /dev/null +++ b/src/libsystemd/sd-future/test-future-group.c @@ -0,0 +1,567 @@ +/* SPDX-License-Identifier: LGPL-2.1-or-later */ + +#include "sd-event.h" +#include "sd-future.h" + +#include "tests.h" + +/* Body for "I exist to be cancelled" siblings: suspends until something cancels us. */ +static int suspend_fiber(void *userdata) { + return sd_fiber_suspend(); +} + +/* Body for "let other things make progress before I return": yields once (giving the event + * loop a chance to dispatch other pending sources) then returns the configured result. + * Useful for sequencing without a real timer — sd-event dispatches at the same priority by + * pending iteration, so a rearmed source goes to the back of the queue. */ +static int yield_then_return_fiber(void *userdata) { + int *result = ASSERT_PTR(userdata); + int r = sd_fiber_yield(); + if (r < 0) + return r; + return *result; +} + +/* WAIT_ALL happy path: three children all succeed. */ +TEST(future_group_wait_all_happy) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *c1 = NULL, *c2 = NULL, *c3 = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_new_defer(e, 0, &c1)); + ASSERT_OK(sd_future_new_defer(e, 0, &c2)); + ASSERT_OK(sd_future_new_defer(e, 0, &c3)); + ASSERT_OK(sd_future_group_add(group, c1)); + ASSERT_OK(sd_future_group_add(group, c2)); + ASSERT_OK(sd_future_group_add(group, c3)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(sd_future_result(group)); + ASSERT_OK_ZERO(sd_future_result(c1)); + ASSERT_OK_ZERO(sd_future_result(c2)); + ASSERT_OK_ZERO(sd_future_result(c3)); +} + +/* WAIT_ALL fail-fast (default): one child errors fast, others sleeping; group resolves with + * that error, sleepers observe -ECANCELED. */ +TEST(future_group_wait_all_fail_fast) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *errorer = NULL, *sleeper_a = NULL, *sleeper_b = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + + ASSERT_OK(sd_future_new_defer(e, -EINVAL, &errorer)); + ASSERT_OK(sd_fiber_new(e, "sleep-a", suspend_fiber, NULL, NULL, &sleeper_a)); + ASSERT_OK(sd_fiber_new(e, "sleep-b", suspend_fiber, NULL, NULL, &sleeper_b)); + + ASSERT_OK(sd_future_group_add(group, errorer)); + ASSERT_OK(sd_future_group_add(group, sleeper_a)); + ASSERT_OK(sd_future_group_add(group, sleeper_b)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(group), EINVAL); + ASSERT_ERROR(sd_future_result(errorer), EINVAL); + ASSERT_ERROR(sd_future_result(sleeper_a), ECANCELED); + ASSERT_ERROR(sd_future_result(sleeper_b), ECANCELED); +} + +/* WAIT_ALL with IGNORE_ERRORS: one child errors, others succeed; everyone runs to completion; + * group resolves with the first error. */ +TEST(future_group_wait_all_ignore_errors) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *errorer = NULL, *succeeder = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_IGNORE_ERRORS)); + + ASSERT_OK(sd_future_new_defer(e, -EINVAL, &errorer)); + ASSERT_OK(sd_future_new_defer(e, 0, &succeeder)); + ASSERT_OK(sd_future_group_add(group, errorer)); + ASSERT_OK(sd_future_group_add(group, succeeder)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(group), EINVAL); + ASSERT_ERROR(sd_future_result(errorer), EINVAL); + ASSERT_OK_ZERO(sd_future_result(succeeder)); +} + +/* WAIT_ANY: three sleepers with different durations; group resolves with shortest sleeper's + * result; siblings observe -ECANCELED. */ +TEST(future_group_wait_any) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *fast = NULL, *medium = NULL, *slow = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY)); + + ASSERT_OK(sd_future_new_defer(e, 0, &fast)); + ASSERT_OK(sd_fiber_new(e, "medium", suspend_fiber, NULL, NULL, &medium)); + ASSERT_OK(sd_fiber_new(e, "slow", suspend_fiber, NULL, NULL, &slow)); + ASSERT_OK(sd_future_group_add(group, fast)); + ASSERT_OK(sd_future_group_add(group, medium)); + ASSERT_OK(sd_future_group_add(group, slow)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(sd_future_result(group)); + ASSERT_OK_ZERO(sd_future_result(fast)); + ASSERT_ERROR(sd_future_result(medium), ECANCELED); + ASSERT_ERROR(sd_future_result(slow), ECANCELED); +} + +/* WAIT_ANY|IGNORE_ERRORS (FIRST_SUCCESS): fast errorer, slower success, slowest pending; + * group resolves with the success value; slowest is cancelled. */ +TEST(future_group_first_success) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *fast_err = NULL, *medium_ok = NULL, *slow_ok = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY|SD_FUTURE_GROUP_IGNORE_ERRORS)); + + ASSERT_OK(sd_future_new_defer(e, -EINVAL, &fast_err)); + ASSERT_OK(sd_future_new_defer(e, 0, &medium_ok)); + ASSERT_OK(sd_fiber_new(e, "slow-ok", suspend_fiber, NULL, NULL, &slow_ok)); + + ASSERT_OK(sd_future_group_add(group, fast_err)); + ASSERT_OK(sd_future_group_add(group, medium_ok)); + ASSERT_OK(sd_future_group_add(group, slow_ok)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(sd_future_result(group)); + ASSERT_ERROR(sd_future_result(fast_err), EINVAL); + ASSERT_OK_ZERO(sd_future_result(medium_ok)); + ASSERT_ERROR(sd_future_result(slow_ok), ECANCELED); +} + +/* WAIT_ANY|IGNORE_ERRORS, all fail: every child errors; group resolves with first error. */ +TEST(future_group_first_success_all_fail) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *err_a = NULL, *err_b = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY|SD_FUTURE_GROUP_IGNORE_ERRORS)); + + ASSERT_OK(sd_future_new_defer(e, -EINVAL, &err_a)); + ASSERT_OK(sd_future_new_defer(e, -EINVAL, &err_b)); + ASSERT_OK(sd_future_group_add(group, err_a)); + ASSERT_OK(sd_future_group_add(group, err_b)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(group), EINVAL); +} + +/* External cancellation: long-running children + a deferred event source that cancels the + * group; group and every child observe -ECANCELED. */ +static int cancel_trigger(sd_event_source *src, void *userdata) { + sd_future *group = ASSERT_PTR(userdata); + return sd_future_cancel(group); +} + +TEST(future_group_external_cancel) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *long_a = NULL, *long_b = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + + ASSERT_OK(sd_fiber_new(e, "long-a", suspend_fiber, NULL, NULL, &long_a)); + ASSERT_OK(sd_fiber_new(e, "long-b", suspend_fiber, NULL, NULL, &long_b)); + ASSERT_OK(sd_future_group_add(group, long_a)); + ASSERT_OK(sd_future_group_add(group, long_b)); + + /* Deferred event source runs after fibers have had a chance to suspend, then cancels + * the group from outside the fiber stack. */ + _cleanup_(sd_event_source_unrefp) sd_event_source *cancel_src = NULL; + ASSERT_OK(sd_event_add_defer(e, &cancel_src, cancel_trigger, group)); + ASSERT_OK(sd_event_source_set_priority(cancel_src, 100)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(group), ECANCELED); + ASSERT_ERROR(sd_future_result(long_a), ECANCELED); + ASSERT_ERROR(sd_future_result(long_b), ECANCELED); +} + +/* Drain invariant: a fail-fast group with a fast errorer + a still-sleeping sibling. The + * group's done callback must see *every* child in RESOLVED state — the group may not + * settle while any cancelled child is still draining. */ +typedef struct DrainCheckState { + sd_future *child_a; + sd_future *child_b; + int a_state_at_resolve; + int b_state_at_resolve; +} DrainCheckState; + +static int drain_check_cb(sd_future *f, void *userdata) { + DrainCheckState *s = ASSERT_PTR(userdata); + s->a_state_at_resolve = sd_future_state(s->child_a); + s->b_state_at_resolve = sd_future_state(s->child_b); + return 0; +} + +TEST(future_group_resolves_after_children_drain) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *errorer = NULL, *sleeper = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + + /* The errorer yields once so the sleeper's dispatch fires first and the sleeper + * actually enters its body before the errorer returns. That way the group's cancel + * of the sleeper goes through the async (FIBER_STATE_SUSPENDED) path, not the + * synchronous FIBER_STATE_INITIAL path — which is the case the drain invariant is + * about. */ + int err_result = -EINVAL; + ASSERT_OK(sd_fiber_new(e, "err", yield_then_return_fiber, &err_result, NULL, &errorer)); + ASSERT_OK(sd_fiber_new(e, "sleep", suspend_fiber, NULL, NULL, &sleeper)); + ASSERT_OK(sd_future_group_add(group, errorer)); + ASSERT_OK(sd_future_group_add(group, sleeper)); + + DrainCheckState s = { .child_a = errorer, .child_b = sleeper }; + _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot = NULL; + ASSERT_OK(sd_future_add_callback(group, &slot, drain_check_cb, &s)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_EQ(s.a_state_at_resolve, (int) SD_FUTURE_RESOLVED); + ASSERT_EQ(s.b_state_at_resolve, (int) SD_FUTURE_RESOLVED); + ASSERT_ERROR(sd_future_result(group), EINVAL); + ASSERT_ERROR(sd_future_result(sleeper), ECANCELED); +} + +/* Parent cancellation: a fiber creates a group with one errorer, then suspends without ever + * awaiting the group. When the errorer fails, the parent fiber's in-flight suspend must + * return -ECANCELED. With IGNORE_ERRORS the parent is left alone (a wake-up callback on + * the group lifts the suspend so we don't hang). */ +typedef struct ParentCancelState { + uint64_t policy; + int suspend_result; + int group_result; +} ParentCancelState; + +static int wake_parent_cb(sd_future *f, void *userdata) { + return sd_fiber_resume(userdata, 0); +} + +static int parent_cancel_driver(void *userdata) { + ParentCancelState *s = ASSERT_PTR(userdata); + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *errorer = NULL; + + ASSERT_OK(sd_future_group_new(sd_fiber_get_event(), &group)); + if (s->policy) + ASSERT_OK(sd_future_group_set_policy(group, s->policy)); + ASSERT_OK(sd_future_new_defer(sd_fiber_get_event(), -EINVAL, &errorer)); + ASSERT_OK(sd_future_group_add(group, errorer)); + + /* Wake-up slot so we don't hang in the IGNORE_ERRORS case (where the parent isn't + * cancelled). In the fail-fast case the parent's state goes CANCELLED before this + * fires, and sd_fiber_resume on a non-SUSPENDED fiber is a no-op. */ + _cleanup_(sd_future_slot_unrefp) sd_future_slot *wake_slot = NULL; + ASSERT_OK(sd_future_add_callback(group, &wake_slot, wake_parent_cb, sd_fiber_get_current())); + + s->suspend_result = sd_fiber_suspend(); + s->group_result = sd_future_result(group); + return 0; +} + +TEST(future_group_cancels_parent_on_child_error) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + ParentCancelState s = {}; + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "parent", parent_cancel_driver, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(s.suspend_result, ECANCELED); + ASSERT_ERROR(s.group_result, EINVAL); +} + +/* sd_future_group_await() suppresses parent-cancel: when the awaiter has explicitly + * opted in to receiving the group's resolution, the await returns the group's actual + * error (here -EINVAL) instead of -ECANCELED from the parent-cancel path. */ +typedef struct AwaitGetsErrorState { + int await_result; +} AwaitGetsErrorState; + +static int await_gets_error_driver(void *userdata) { + AwaitGetsErrorState *s = ASSERT_PTR(userdata); + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *errorer = NULL; + + ASSERT_OK(sd_future_group_new(sd_fiber_get_event(), &group)); + ASSERT_OK(sd_future_new_defer(sd_fiber_get_event(), -EINVAL, &errorer)); + ASSERT_OK(sd_future_group_add(group, errorer)); + + s->await_result = sd_future_group_await(group); + return 0; +} + +TEST(future_group_await_returns_real_error) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + AwaitGetsErrorState s = {}; + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "parent", await_gets_error_driver, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(s.await_result, EINVAL); +} + +TEST(future_group_does_not_cancel_parent_with_ignore_errors) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + ParentCancelState s = { .policy = SD_FUTURE_GROUP_IGNORE_ERRORS }; + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "parent", parent_cancel_driver, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + /* With IGNORE_ERRORS the parent isn't cancelled — the wake-up callback resumes the + * suspend with 0, and we read the group's error via sd_future_result. */ + ASSERT_OK_ZERO(s.suspend_result); + ASSERT_ERROR(s.group_result, EINVAL); +} + +/* Add already-resolved child to a default group: group stays PENDING until the next event-loop + * tick (via sd_future_add_callback's RESOLVED-on-fiber defer path), then resolves with the + * child's result. */ +typedef struct AddResolvedState { + sd_future *child; + sd_future *group; + int add_result; + int group_state_after_add; + int await_result; +} AddResolvedState; + +static int add_resolved_driver(void *userdata) { + AddResolvedState *s = ASSERT_PTR(userdata); + + /* Drive the pre-built child to completion. The defer resolves with -EINVAL, which + * makes sd_fiber_await return -EINVAL — we don't ASSERT_OK that. The child is now + * RESOLVED and we can add it to the group. */ + (void) sd_fiber_await(s->child); + ASSERT_EQ(sd_future_state(s->child), SD_FUTURE_RESOLVED); + + s->add_result = sd_future_group_add(s->group, s->child); + s->group_state_after_add = sd_future_state(s->group); + + /* Await drives the loop one more tick so the defer that wraps the RESOLVED child can + * fire group_child_resolved and settle the group. */ + s->await_result = sd_future_group_await(s->group); + return 0; +} + +TEST(future_group_add_resolved_child) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + AddResolvedState s = {}; + ASSERT_OK(sd_future_group_new(e, &s.group)); + ASSERT_OK(sd_future_new_defer(e, -EINVAL, &s.child)); + + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "driver", add_resolved_driver, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK(s.add_result); + ASSERT_EQ(s.group_state_after_add, SD_FUTURE_PENDING); + ASSERT_ERROR(s.await_result, EINVAL); + ASSERT_ERROR(sd_future_result(s.group), EINVAL); + s.child = sd_future_unref(s.child); + s.group = sd_future_unref(s.group); +} + +/* add_many with NULL sentinel: convenience adds multiple children, behaves like add. */ +typedef struct AddManyState { + sd_future *a; + sd_future *b; + sd_future *c; + sd_future *group; + int join_result; +} AddManyState; + +static int add_many_driver(void *userdata) { + AddManyState *s = ASSERT_PTR(userdata); + ASSERT_OK(sd_future_group_add_many(s->group, s->a, s->b, s->c, NULL)); + s->join_result = sd_future_group_await(s->group); + return 0; +} + +TEST(future_group_add_many) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + AddManyState s = {}; + ASSERT_OK(sd_future_group_new(e, &s.group)); + ASSERT_OK(sd_future_new_defer(e, 0, &s.a)); + ASSERT_OK(sd_future_new_defer(e, 0, &s.b)); + ASSERT_OK(sd_future_new_defer(e, 0, &s.c)); + + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "driver", add_many_driver, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(s.join_result); + + s.a = sd_future_unref(s.a); + s.b = sd_future_unref(s.b); + s.c = sd_future_unref(s.c); + s.group = sd_future_unref(s.group); +} + +/* Custom future ops whose cancel is a no-op on the first call and resolves on the second. + * Used to drive a group into the "finalizing but draining" state synchronously: one cancel + * of the group runs finalize, which calls our cancel once — but we don't resolve, so the + * group's state stays PENDING with finalizing=true. */ +typedef struct StubbornChild { + unsigned cancels; +} StubbornChild; + +static void* stubborn_child_alloc(void) { + return new0(StubbornChild, 1); +} + +static void stubborn_child_free(sd_future *f) { + free(sd_future_get_private(f)); +} + +static int stubborn_child_cancel(sd_future *f) { + StubbornChild *sc = ASSERT_PTR(sd_future_get_private(f)); + if (++sc->cancels >= 2) + return sd_future_resolve(f, -ECANCELED); + return 0; +} + +static const sd_future_ops stubborn_child_ops = { + .size = sizeof(sd_future_ops), + .alloc = stubborn_child_alloc, + .free = stubborn_child_free, + .cancel = stubborn_child_cancel, +}; + +/* Cancelling a group with a still-draining child puts the group into the finalizing-but-PENDING + * state. While in that state, sd_future_group_add() must reject with -ESTALE — a freshly-added + * child would have missed the cancel loop and hang us forever waiting for it to settle. */ +TEST(future_group_add_rejected_during_finalize) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *stubborn = NULL, *new_child = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_new(e, &stubborn_child_ops, &stubborn)); + ASSERT_OK(sd_future_group_add(group, stubborn)); + + /* First cancel: triggers finalize, calls stubborn's cancel (no-op #1, child still + * PENDING). Group is now finalizing=true with state PENDING. */ + ASSERT_OK(sd_future_cancel(group)); + ASSERT_EQ(sd_future_state(group), SD_FUTURE_PENDING); + ASSERT_EQ(sd_future_state(stubborn), SD_FUTURE_PENDING); + + ASSERT_OK(sd_future_new_defer(e, 0, &new_child)); + ASSERT_ERROR(sd_future_group_add(group, new_child), ESTALE); + + /* Second cancel of stubborn resolves it, which on the next loop iteration drives + * group_child_resolved → future_group_check → group resolves with the locked-in + * -ECANCELED. */ + ASSERT_OK(sd_future_cancel(stubborn)); + ASSERT_OK(sd_event_loop(e)); + ASSERT_ERROR(sd_future_result(group), ECANCELED); + ASSERT_ERROR(sd_future_result(stubborn), ECANCELED); + ASSERT_OK_ZERO(sd_future_result(new_child)); +} + +/* Policy must be configured before any children are added: once a child is in flight, the + * resolution mechanics are locked in. Multiple set_policy calls on a fresh group are fine. */ +TEST(future_group_set_policy_rejected_after_add) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + _cleanup_(sd_future_cancel_unrefp) sd_future *group = NULL; + _cleanup_(sd_future_unrefp) sd_future *child = NULL; + ASSERT_OK(sd_future_group_new(e, &group)); + + /* No children yet — set_policy can be called and re-called freely. */ + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY)); + ASSERT_OK(sd_future_group_set_policy(group, + SD_FUTURE_GROUP_WAIT_ANY | SD_FUTURE_GROUP_IGNORE_ERRORS)); + + ASSERT_OK(sd_future_new_defer(e, 0, &child)); + ASSERT_OK(sd_future_group_add(group, child)); + + /* Now that a child is registered, set_policy must reject. */ + ASSERT_ERROR(sd_future_group_set_policy(group, 0), ESTALE); + ASSERT_ERROR(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY), ESTALE); +} + +/* When the parent fiber itself drives the cancel of its own group, future_group_finalize() must + * skip the parent-cancel path — fiber_cancel asserts against self-cancellation, and even if that + * assertion were relaxed, queuing -ECANCELED on the parent would surface on a later suspend the + * caller didn't expect. */ +typedef struct SelfCancelState { + int cancel_return; + int group_result; + int yield_result; +} SelfCancelState; + +static int self_cancel_fiber(void *userdata) { + SelfCancelState *s = ASSERT_PTR(userdata); + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *child = NULL; + int r; + + r = sd_future_group_new(sd_fiber_get_event(), &group); + if (r < 0) + return r; + + r = sd_future_new_defer(sd_fiber_get_event(), 0, &child); + if (r < 0) + return r; + + r = sd_future_group_add(group, child); + if (r < 0) + return r; + + s->cancel_return = sd_future_cancel(group); + s->group_result = sd_future_result(group); + + /* If the parent-cancel guard didn't work, -ECANCELED would be queued on us by + * finalize() and surface here. Expect a clean yield (0). */ + s->yield_result = sd_fiber_yield(); + return 0; +} + +TEST(future_group_does_not_cancel_parent_when_parent_drives_cancel) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + + SelfCancelState s = {}; + _cleanup_(sd_future_unrefp) sd_future *driver = NULL; + ASSERT_OK(sd_fiber_new(e, "self-cancel", self_cancel_fiber, &s, NULL, &driver)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(sd_future_result(driver)); + ASSERT_OK(s.cancel_return); + ASSERT_ERROR(s.group_result, ECANCELED); + ASSERT_OK_ZERO(s.yield_result); +} + +DEFINE_TEST_MAIN(LOG_DEBUG); diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index 1b0efa3992037..c85b4a6780b2a 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -1105,7 +1105,7 @@ static int varlink_dispatch_fiber(sd_varlink *v, const char *method, sd_varlink_ .callback = callback, }; - _cleanup_(sd_future_unrefp) sd_future *f = NULL; + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; r = sd_fiber_new(v->server->event, method, varlink_fiber_entry, d, varlink_fiber_data_destroy, &f); if (r < 0) return r; @@ -1130,6 +1130,8 @@ static int varlink_dispatch_fiber(sd_varlink *v, const char *method, sd_varlink_ if (r < 0) return r; + /* Floating self-ref keeps the fiber alive; release our local ref without cancelling. */ + f = sd_future_unref(f); return 0; } diff --git a/src/shared/qmp-client.c b/src/shared/qmp-client.c index 8de903de1ad8c..fb3f9087ca310 100644 --- a/src/shared/qmp-client.c +++ b/src/shared/qmp-client.c @@ -905,8 +905,8 @@ int qmp_client_call_future( assert(command); assert(ret); - _cleanup_(sd_future_unrefp) sd_future *f = NULL; - r = sd_future_new(&qmp_call_future_ops, &f); + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + r = sd_future_new(qmp_client_get_event(c), &qmp_call_future_ops, &f); if (r < 0) return r; @@ -972,7 +972,7 @@ static int qmp_client_call_suspend( if (r < 0) return r; - r = sd_fiber_suspend(); + r = sd_fiber_await(call); /* If the future isn't resolved, the suspend was interrupted before a reply arrived (fiber * cancelled, fiber-wide SD_FIBER_TIMEOUT scope expired, …). There's no reply to extract, diff --git a/src/systemd/sd-future.h b/src/systemd/sd-future.h index 5e0fa22525615..6f8c6c87b6caa 100644 --- a/src/systemd/sd-future.h +++ b/src/systemd/sd-future.h @@ -32,7 +32,8 @@ struct timespec; typedef struct sd_event sd_event; typedef struct sd_future sd_future; typedef struct sd_future_ops sd_future_ops; -typedef int (*sd_future_func_t)(sd_future *f); +typedef struct sd_future_slot sd_future_slot; +typedef int (*sd_future_func_t)(sd_future *f, void *userdata); typedef int (*sd_fiber_func_t)(void *userdata); typedef _sd_destroy_t sd_fiber_destroy_t; @@ -50,7 +51,7 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_future_state_t) { _SD_ENUM_FORCE_S64(SD_FUTURE_STATE) } sd_future_state_t; -int sd_future_new(const sd_future_ops *ops, sd_future **ret); +int sd_future_new(sd_event *e, const sd_future_ops *ops, sd_future **ret); int sd_future_cancel(sd_future *f); int sd_future_resolve(sd_future *f, int result); @@ -59,6 +60,11 @@ _SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_unref); void sd_future_unref_array_clear(sd_future *array[], size_t n); void sd_future_unref_array(sd_future *array[], size_t n); +sd_future* sd_future_cancel_unref(sd_future *f); +_SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_cancel_unref); +void sd_future_cancel_unref_array_clear(sd_future *array[], size_t n); +void sd_future_cancel_unref_array(sd_future *array[], size_t n); + sd_future* sd_future_cancel_wait_unref(sd_future *f); _SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_cancel_wait_unref); void sd_future_cancel_wait_unref_array_clear(sd_future *array[], size_t n); @@ -66,14 +72,40 @@ void sd_future_cancel_wait_unref_array(sd_future *array[], size_t n); int sd_future_state(sd_future *f); int sd_future_result(sd_future *f); -void* sd_future_get_userdata(sd_future *f); void* sd_future_get_private(sd_future *f); const sd_future_ops* sd_future_get_ops(sd_future *f); +sd_event* sd_future_get_event(sd_future *f); + +int sd_future_add_callback(sd_future *f, sd_future_slot **ret_slot, sd_future_func_t callback, void *userdata); + +/* Create a future that resolves on the next iteration of `e`'s event loop with + * `result`. Use sd_future_add_callback() to attach callbacks; cancelling the + * future before it fires resolves it with -ECANCELED. */ +int sd_future_new_defer(sd_event *e, int result, sd_future **ret); + +int sd_future_resume_callback(sd_future *f, void *userdata); + +_SD_DECLARE_TRIVIAL_REF_UNREF_FUNC(sd_future_slot); +_SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future_slot, sd_future_slot_unref); + +sd_future* sd_future_slot_get_future(sd_future_slot *s); -int sd_future_set_callback(sd_future *f, sd_future_func_t callback, void *userdata); int sd_future_set_priority(sd_future *f, int64_t priority); -int sd_future_new_wait(sd_future *target, sd_future **ret); +__extension__ typedef enum _SD_ENUM_TYPE_S64(sd_future_group_policy_t) { + /* Default (0): wait for every child; cancel pending siblings on the + * first child error and resolve the group with that error. */ + SD_FUTURE_GROUP_WAIT_ANY = 1 << 0, /* resolve when one child settles */ + SD_FUTURE_GROUP_IGNORE_ERRORS = 1 << 1, /* don't cancel siblings on child error */ + _SD_ENUM_FORCE_S64(SD_FUTURE_GROUP_POLICY) +} sd_future_group_policy_t; + +int sd_future_group_new(sd_event *e, sd_future **ret); +int sd_future_group_set_policy(sd_future *f, uint64_t policy); +int sd_future_group_add(sd_future *f, sd_future *child); +int sd_future_group_add_many(sd_future *f, ...); +int sd_future_group_size(sd_future *f, size_t *ret); +int sd_future_group_await(sd_future *f); int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *userdata, sd_fiber_destroy_t destroy, sd_future **ret);