sd-future: rework cancellation, add slot callbacks and future groups

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',

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,

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

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;

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;
 

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;