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);