diff --git a/src/libsystemd/sd-bus/bus-objects.c b/src/libsystemd/sd-bus/bus-objects.c index 164e1dc059..2f8a40854c 100644 --- a/src/libsystemd/sd-bus/bus-objects.c +++ b/src/libsystemd/sd-bus/bus-objects.c @@ -489,7 +489,11 @@ static int method_callbacks_run( }; _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); + r = sd_fiber_new(bus->event, c->member, bus_fiber_entry, d, &f); + if (r < 0) + return bus_maybe_reply_error(m, r, NULL); + + r = sd_fiber_set_destroy_callback(f, bus_fiber_data_destroy); if (r < 0) return bus_maybe_reply_error(m, r, NULL); diff --git a/src/libsystemd/sd-bus/test-bus-chat.c b/src/libsystemd/sd-bus/test-bus-chat.c index d6f4860c41..38c5341894 100644 --- a/src/libsystemd/sd-bus/test-bus-chat.c +++ b/src/libsystemd/sd-bus/test-bus-chat.c @@ -645,9 +645,9 @@ TEST(chat) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "client-1", client1, NULL, /* destroy= */ NULL, &f_client1)); - ASSERT_OK(sd_fiber_new(e, "client-2", client2, NULL, /* destroy= */ NULL, &f_client2)); - ASSERT_OK(sd_fiber_new(e, "server", server, bus, /* destroy= */ NULL, &f_server)); + ASSERT_OK(sd_fiber_new(e, "client-1", client1, NULL, &f_client1)); + ASSERT_OK(sd_fiber_new(e, "client-2", client2, NULL, &f_client2)); + ASSERT_OK(sd_fiber_new(e, "server", server, bus, &f_server)); ASSERT_OK(sd_event_loop(e)); diff --git a/src/libsystemd/sd-bus/test-bus-fiber.c b/src/libsystemd/sd-bus/test-bus-fiber.c index 28d8819f2b..c68b871448 100644 --- a/src/libsystemd/sd-bus/test-bus-fiber.c +++ b/src/libsystemd/sd-bus/test-bus-fiber.c @@ -153,10 +153,8 @@ static int concurrency_fiber(void *userdata) { /* Two concurrent calls on the shared client bus. Each lands in method_concurrent which * blocks on the peer; if fiber dispatch works the second is entered while the first is * suspended, so max_in_flight on the context reaches 2. */ - ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "call-a", call_concurrent_fiber, client, - /* destroy= */ NULL, &f_a)); - ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "call-b", call_concurrent_fiber, client, - /* destroy= */ NULL, &f_b)); + ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "call-a", call_concurrent_fiber, client, &f_a)); + ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "call-b", call_concurrent_fiber, client, &f_b)); ASSERT_OK(sd_fiber_await(f_a)); ASSERT_OK(sd_fiber_await(f_b)); @@ -177,7 +175,7 @@ TEST(fiber_method_concurrency) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "concurrency", concurrency_fiber, &s, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "concurrency", concurrency_fiber, &s, &f)); ASSERT_OK(sd_event_loop(e)); @@ -252,7 +250,7 @@ TEST(fiber_method_errors) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "errors", errors_fiber, &s, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "errors", errors_fiber, &s, &f)); ASSERT_OK(sd_event_loop(e)); @@ -307,7 +305,7 @@ static int interrupted_calls_fiber(void *userdata) { s->c->waiter = error == ETIME ? NULL : sd_fiber_get_current(); ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "interrupted-call", interrupted_call_fiber, - &c, /* destroy= */ NULL, &caller)); + &c, &caller)); if (error != ETIME) { /* Interrupt only once the server has the call in hand and withheld its reply. */ @@ -337,8 +335,7 @@ TEST(fiber_method_interrupted) { ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, s.fds)); ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "interrupted-calls", interrupted_calls_fiber, &s, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "interrupted-calls", interrupted_calls_fiber, &s, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); } @@ -418,7 +415,7 @@ TEST(signal_channel) { ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, s.fds)); ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "signal-channel", signal_channel_fiber, &s, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "signal-channel", signal_channel_fiber, &s, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); } diff --git a/src/libsystemd/sd-bus/test-bus-objects.c b/src/libsystemd/sd-bus/test-bus-objects.c index 5d8e772cb4..fc21278ca9 100644 --- a/src/libsystemd/sd-bus/test-bus-objects.c +++ b/src/libsystemd/sd-bus/test-bus-objects.c @@ -603,8 +603,8 @@ int main(int argc, char *argv[]) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "server", server, &c, /* destroy= */ NULL, &f_server)); - ASSERT_OK(sd_fiber_new(e, "client", client, &c, /* destroy= */ NULL, &f_client)); + ASSERT_OK(sd_fiber_new(e, "server", server, &c, &f_server)); + ASSERT_OK(sd_fiber_new(e, "client", client, &c, &f_client)); ASSERT_OK(sd_event_loop(e)); diff --git a/src/libsystemd/sd-bus/test-bus-peersockaddr.c b/src/libsystemd/sd-bus/test-bus-peersockaddr.c index bee76c9b10..cff473636a 100644 --- a/src/libsystemd/sd-bus/test-bus-peersockaddr.c +++ b/src/libsystemd/sd-bus/test-bus-peersockaddr.c @@ -160,10 +160,10 @@ TEST(description) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "server", server, INT_TO_PTR(fd), /* destroy= */ NULL, &f_server)); + ASSERT_OK(sd_fiber_new(e, "server", server, INT_TO_PTR(fd), &f_server)); TAKE_FD(fd); - ASSERT_OK(sd_fiber_new(e, "client", client, a, /* destroy= */ NULL, &f_client)); + ASSERT_OK(sd_fiber_new(e, "client", client, a, &f_client)); ASSERT_OK(sd_event_loop(e)); diff --git a/src/libsystemd/sd-bus/test-bus-server.c b/src/libsystemd/sd-bus/test-bus-server.c index 1edcec858f..f67ad2ed21 100644 --- a/src/libsystemd/sd-bus/test-bus-server.c +++ b/src/libsystemd/sd-bus/test-bus-server.c @@ -123,8 +123,8 @@ static int test_one(bool client_negotiate_unix_fds, bool server_negotiate_unix_f ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "server", server, &c, /* destroy= */ NULL, &f_server)); - ASSERT_OK(sd_fiber_new(e, "client", client, &c, /* destroy= */ NULL, &f_client)); + ASSERT_OK(sd_fiber_new(e, "server", server, &c, &f_server)); + ASSERT_OK(sd_fiber_new(e, "client", client, &c, &f_client)); ASSERT_OK(sd_event_loop(e)); diff --git a/src/libsystemd/sd-bus/test-bus-watch-bind.c b/src/libsystemd/sd-bus/test-bus-watch-bind.c index bbcd70d1dd..9e5e51363d 100644 --- a/src/libsystemd/sd-bus/test-bus-watch-bind.c +++ b/src/libsystemd/sd-bus/test-bus-watch-bind.c @@ -222,17 +222,17 @@ int main(int argc, char *argv[]) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "server", server, path, /* destroy= */ NULL, &f_server)); + ASSERT_OK(sd_fiber_new(e, "server", server, path, &f_server)); - ASSERT_OK(sd_fiber_new(e, "client-1", client1, path, /* destroy= */ NULL, &f_client1)); - ASSERT_OK(sd_fiber_new(e, "client-2", client2, path, /* destroy= */ NULL, &f_client2)); + ASSERT_OK(sd_fiber_new(e, "client-1", client1, path, &f_client1)); + ASSERT_OK(sd_fiber_new(e, "client-2", client2, path, &f_client2)); RequestExitArgs args = { .path = path, .client1 = f_client1, .client2 = f_client2, }; - ASSERT_OK(sd_fiber_new(e, "request-exit", request_exit, &args, /* destroy= */ NULL, &f_exit)); + ASSERT_OK(sd_fiber_new(e, "request-exit", request_exit, &args, &f_exit)); ASSERT_OK(sd_event_loop(e)); diff --git a/src/libsystemd/sd-event/event-future.c b/src/libsystemd/sd-event/event-future.c index 63b488e57d..40ae758b08 100644 --- a/src/libsystemd/sd-event/event-future.c +++ b/src/libsystemd/sd-event/event-future.c @@ -608,6 +608,10 @@ int event_run_suspend(sd_event *e, uint64_t timeout) { return r; } + r = sd_future_group_seal(group); + if (r < 0) + return r; + r = sd_fiber_await(group); if (r < 0) return r; diff --git a/src/libsystemd/sd-event/test-event-future.c b/src/libsystemd/sd-event/test-event-future.c index 6f89ee58e9..e25ee133c9 100644 --- a/src/libsystemd/sd-event/test-event-future.c +++ b/src/libsystemd/sd-event/test-event-future.c @@ -228,7 +228,7 @@ TEST(future_child_process_own_cancel_wait) { ASSERT_OK(fork_stubborn_child(&pidref)); ASSERT_OK(future_new_child(e, &pidref, WEXITED, &f)); ASSERT_OK(future_child_set_process_own(f, true)); - ASSERT_OK(sd_fiber_new(e, "child-cancel-wait", child_cancel_wait_fiber, f, /* destroy= */ NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "child-cancel-wait", child_cancel_wait_fiber, f, &fiber)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), ECANCELED); @@ -308,7 +308,7 @@ TEST(future_child_await) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); ASSERT_OK(fork_child(7, &pidref)); - ASSERT_OK(sd_fiber_new(e, "child-await", child_await_fiber, &pidref, /* destroy= */ NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "child-await", child_await_fiber, &pidref, &fiber)); ASSERT_OK(sd_event_loop(e)); ASSERT_EQ(sd_future_result(fiber), 7); @@ -328,6 +328,8 @@ TEST(future_group_add_child) { ASSERT_OK(future_group_add_child(group, &b, WEXITED)); ASSERT_EQ(sd_future_group_size(group), 2U); + ASSERT_OK(sd_future_group_seal(group)); + ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(group)); ASSERT_ERROR(pidref_kill(&a, 0), ESRCH); @@ -367,7 +369,7 @@ TEST(event_future_priority) { EventFuturePriorityData d = { .type = type, .fd = fds[0], .ret = &f }; ASSERT_OK(sd_fiber_new(e, "new-event-future", event_future_priority_fiber, - &d, /* destroy= */ NULL, ®istrar)); + &d, ®istrar)); ASSERT_OK(sd_future_set_priority(registrar, priority)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK_ZERO(sd_future_result(registrar)); @@ -525,7 +527,7 @@ TEST(sd_event_loop_fiber) { /* Create a fiber that will create and run the inner event loop */ _cleanup_(sd_future_unrefp) sd_future *f = NULL; int inner_timer_count = 0; - ASSERT_OK(sd_fiber_new(outer, "event-runner", event_run_fiber_func, &inner_timer_count, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "event-runner", event_run_fiber_func, &inner_timer_count, &f)); /* Run the outer event loop */ ASSERT_OK(sd_event_loop(outer)); @@ -559,7 +561,7 @@ TEST(sd_event_run_fiber_timeout) { /* Create a fiber that will run sd_event_run() with timeout */ _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(outer, "event-timeout", event_run_fiber_timeout_func, NULL, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "event-timeout", event_run_fiber_timeout_func, NULL, &f)); /* Run the outer event loop */ ASSERT_OK(sd_event_loop(outer)); @@ -591,7 +593,7 @@ TEST(sd_event_run_zero_timeout) { ASSERT_OK(sd_event_set_exit_on_idle(outer, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(outer, "run-suspend-zero", sd_event_run_zero_timeout_fiber, NULL, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "run-suspend-zero", sd_event_run_zero_timeout_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(outer)); ASSERT_OK_ZERO(sd_future_result(f)); @@ -650,7 +652,7 @@ TEST(sd_event_run_immediate) { ASSERT_OK_EQ_ERRNO(write(pipefd[1], "X", 1), 1); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(outer, "run-suspend-immediate", sd_event_run_immediate_fiber, pipefd, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "run-suspend-immediate", sd_event_run_immediate_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(outer)); ASSERT_OK_ZERO(sd_future_result(f)); @@ -691,7 +693,7 @@ TEST(sd_event_run_io) { ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(outer, "run-suspend-io", sd_event_run_io_fiber, pipefd, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "run-suspend-io", sd_event_run_io_fiber, pipefd, &f)); /* First iteration: fiber runs, adds IO source, suspends because no data */ ASSERT_OK_POSITIVE(sd_event_run(outer, 0)); @@ -755,7 +757,7 @@ TEST(sd_event_run_loop) { ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(outer, "run-suspend-loop", sd_event_run_loop_fiber, pipefd, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "run-suspend-loop", sd_event_run_loop_fiber, pipefd, &f)); /* Let the fiber run through a few timeout iterations */ for (int i = 0; i < 10; i++) @@ -808,7 +810,7 @@ TEST(sd_event_run_timer) { ASSERT_OK(sd_event_set_exit_on_idle(outer, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(outer, "run-suspend-timer", sd_event_run_timer_fiber, NULL, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "run-suspend-timer", sd_event_run_timer_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(outer)); ASSERT_OK_ZERO(sd_future_result(f)); @@ -844,8 +846,7 @@ TEST(sd_event_run_interrupted) { ASSERT_OK(sd_event_add_io(inner, &source, pipefd[0], EPOLLIN, io_callback, &count)); EventRunInterruptedState s = { .inner = inner, .timeout = timeout, .finite = finite }; - ASSERT_OK(sd_fiber_new(outer, "event-interrupted", event_run_interrupted_fiber, &s, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(outer, "event-interrupted", event_run_interrupted_fiber, &s, &f)); ASSERT_OK_POSITIVE(sd_event_run(outer, 0)); ASSERT_EQ(sd_future_state(f), SD_FUTURE_PENDING); if (!timeout) diff --git a/src/libsystemd/sd-future/fiber.c b/src/libsystemd/sd-future/fiber.c index 85bdcac4ee..ee03691cc0 100644 --- a/src/libsystemd/sd-future/fiber.c +++ b/src/libsystemd/sd-future/fiber.c @@ -685,7 +685,7 @@ static const sd_future_ops fiber_future_ops = { .set_priority = fiber_set_priority, }; -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) { +int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *userdata, sd_future **ret) { int r; assert_return(e, -EINVAL); @@ -786,13 +786,28 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user f = sd_future_unref(f); } - /* We only take ownership of the given userdata pointer on success so assign the destroy callback - * at the very end so we don't clean up the userdata pointer on failure. */ - fiber->destroy = destroy; - return 0; } +int sd_fiber_set_destroy_callback(sd_future *f, sd_fiber_destroy_t callback) { + assert_return(f, -EINVAL); + assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL); + + fiber_get(f)->destroy = callback; + return 0; +} + +int sd_fiber_get_destroy_callback(sd_future *f, sd_fiber_destroy_t *ret) { + assert_return(f, -EINVAL); + assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL); + + sd_fiber_destroy_t callback = fiber_get(f)->destroy; + if (ret) + *ret = callback; + + return !!callback; +} + int sd_fiber_set_floating(sd_future *f, int b) { assert_return(f, -EINVAL); assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL); diff --git a/src/libsystemd/sd-future/future-group.c b/src/libsystemd/sd-future/future-group.c index 34b0351773..e77c2d8c24 100644 --- a/src/libsystemd/sd-future/future-group.c +++ b/src/libsystemd/sd-future/future-group.c @@ -8,36 +8,32 @@ #include "errno-util.h" #include "macro.h" +typedef enum FutureGroupFlags { + FUTURE_GROUP_PRIORITY_SET = 1 << 0, + FUTURE_GROUP_FINALIZING = 1 << 1, + FUTURE_GROUP_SEALED = 1 << 2, + /* FUTURE_GROUP_CANCELLING is set while future_group_cancel_children() runs. The cancel callback of a + * child can cancel the group again. Without the flag, future_group_cancel_children() would then call + * itself until the stack overflows. */ + FUTURE_GROUP_CANCELLING = 1 << 3, + FUTURE_GROUP_CHILDREN_ADDED = 1 << 4, +} FutureGroupFlags; + typedef struct FutureGroup { uint64_t policy; + FutureGroupFlags flags; - /* An explicit priority applies to existing and future children. Otherwise preserve theirs. */ int64_t priority; - bool priority_set; sd_future_slot **slots; size_t n_slots; - /* The fiber the group was created on, if it uses the same event loop. 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 (a child error cancels - * the parent). parent_slot's callback NULLs `parent` if the parent resolves before the - * group does. */ sd_future *parent; sd_future_slot *parent_slot; - /* 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; - bool sealed; - - /* Set while cancelling children: a cancellation that comes back around to the group would - * otherwise recurse until the stack overflows. */ - bool cancelling; + int first_error; } FutureGroup; static void future_group_free(sd_future *f) { @@ -62,19 +58,19 @@ static int future_group_cancel_children(sd_future *g) { FutureGroup *fg = ASSERT_PTR(sd_future_get_private(g)); int r = 0; - if (fg->cancelling) + if (FLAGS_SET(fg->flags, FUTURE_GROUP_CANCELLING)) return 0; - /* A cancellation error does not make a pending child safe to release. Keep waiting for its - * resolution even when cancellation is unsupported or fails; child implementations must still - * arrange completion before the group can finish draining. */ - fg->cancelling = true; + /* A child stays in the group if cancelling it fails or is not supported, and the group keeps + * waiting for the child to resolve. The child itself has to resolve eventually. Otherwise the + * group never resolves. */ + fg->flags |= FUTURE_GROUP_CANCELLING; 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_PENDING) RET_GATHER(r, sd_future_cancel(child)); } - fg->cancelling = false; + fg->flags &= ~FUTURE_GROUP_CANCELLING; return r; } @@ -83,23 +79,20 @@ static int future_group_finalize(sd_future *g, int result, bool propagate_error) 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; + /* The first outcome is final. A later call, such as a repeated sd_future_cancel(), cancels the + * children again. Some children only resolve after repeated cancellation. Without the repeated + * cancellation, sd_future_cancel_wait_unref() on the group would wait forever for such a child. */ + if (!FLAGS_SET(fg->flags, FUTURE_GROUP_FINALIZING)) { + fg->flags |= FUTURE_GROUP_FINALIZING; + fg->result = result; + } RET_GATHER(r, future_group_cancel_children(g)); - /* If we're settling because of a child 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. An active await means the parent will receive the - * group's actual error, so cancelling it would only hide that error behind -ECANCELED. - * The suppression ends automatically when an interrupted wait returns; a peer's wait - * cannot suppress cancellation of the parent. - * Explicit group cancellation never propagates upward. */ + /* If a child error selects the outcome and IGNORE_ERRORS is not set, cancel the parent fiber. The + * parent then sees the failure even if it does not await the group yet. A parent that awaits the + * group reads the error from the group after the wait. Cancelling that parent would make the wait + * return -ECANCELED instead. Explicit cancellation of the group never cancels the parent. */ if (propagate_error && result < 0 && !FLAGS_SET(fg->policy, SD_FUTURE_GROUP_IGNORE_ERRORS) && fg->parent && @@ -107,9 +100,6 @@ static int future_group_finalize(sd_future *g, int result, bool propagate_error) sd_fiber_get_awaiting(fg->parent) != g) 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; } @@ -120,10 +110,9 @@ static int future_group_check(sd_future *g) { if (sd_future_state(g) == SD_FUTURE_RESOLVED) 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). */ + if (FLAGS_SET(fg->flags, FUTURE_GROUP_FINALIZING)) { + /* The group only resolves after every child has resolved. Code that runs when the group + * resolves then sees every child in the RESOLVED state. */ FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) if (sd_future_state(sd_future_slot_get_future(*slot_p)) != SD_FUTURE_RESOLVED) return 0; @@ -133,64 +122,42 @@ static int future_group_check(sd_future *g) { bool wait_any = FLAGS_SET(fg->policy, SD_FUTURE_GROUP_WAIT_ANY); bool ignore_errors = FLAGS_SET(fg->policy, SD_FUTURE_GROUP_IGNORE_ERRORS); - size_t n_resolved = 0; - int first_error = 0, first_success = 0; - bool any_success = false; + /* A child can resolve before the group handles the error of another child. Under WAIT_ANY, the + * success of that child selects the outcome. */ + if (wait_any) + FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) { + sd_future *child = sd_future_slot_get_future(*slot_p); - 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; + if (sd_future_state(child) == SD_FUTURE_RESOLVED && sd_future_result(child) >= 0) + return future_group_finalize(g, sd_future_result(child), /* propagate_error= */ true); } - } - bool all_done = (n_resolved == fg->n_slots); + if (!ignore_errors && fg->first_error != 0) + return future_group_finalize(g, fg->first_error, /* propagate_error= */ true); - 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 + /* The owner can add more children until it seals the group. */ + if (!FLAGS_SET(fg->flags, FUTURE_GROUP_SEALED) || fg->n_slots > 0) return 0; - return future_group_finalize(g, result, /* propagate_error= */ true); + /* A WAIT_ANY group that never had a child resolves with -ECHILD, like waitid() does when there + * are no children. -ECHILD must not cancel the parent fiber. A child error can only reach this + * point with IGNORE_ERRORS, and IGNORE_ERRORS never cancels the parent either. */ + return future_group_finalize( + g, + fg->first_error ?: (wait_any ? -ECHILD : 0), + /* propagate_error= */ false); } static int future_group_cancel(sd_future *f) { - FutureGroup *fg = ASSERT_PTR(sd_future_get_private(f)); - int r; - - /* Explicit group cancellation affects its children, not the fiber that created the group. */ - if (!fg->finalizing) - return future_group_finalize(f, -ECANCELED, /* propagate_error= */ false); - - /* The outcome is locked, but children that escalate on repeated cancellation still need to - * see every attempt, or sd_future_cancel_wait_unref() on the group could never drive them. */ - r = future_group_cancel_children(f); - - /* The last pending child may have settled synchronously, in which case the group can resolve - * now rather than after the next dispatch. */ - RET_GATHER(r, future_group_check(f)); - return r; + return future_group_finalize(f, -ECANCELED, /* propagate_error= */ false); } static int future_group_set_child_priority(sd_future *child, int64_t priority) { int r; r = sd_future_set_priority(child, priority); - /* Some children do not support priorities or have already resolved. */ + /* sd_future_set_priority() fails with -EOPNOTSUPP for a child without priority support, and + * with -ESTALE for a child that has already resolved. */ if (r < 0 && !IN_SET(r, -EOPNOTSUPP, -ESTALE)) return r; @@ -207,7 +174,7 @@ static int future_group_set_priority(sd_future *f, int64_t priority) { return r; fg->priority = priority; - fg->priority_set = true; + fg->flags |= FUTURE_GROUP_PRIORITY_SET; return 0; } @@ -259,18 +226,52 @@ int sd_future_group_set_policy(sd_future *f, uint64_t policy) { assert_return(sd_future_state(f) == SD_FUTURE_PENDING, -ESTALE); assert_return((policy & ~(uint64_t) _SD_FUTURE_GROUP_POLICY_MASK) == 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. */ + /* The group handles the result of each child according to the policy. A new policy would not + * apply to the results that the group already handled. Only a group that never had a child + * therefore accepts a new policy. */ FutureGroup *fg = sd_future_get_private(f); - assert_return(fg->n_slots == 0, -ESTALE); + assert_return(!FLAGS_SET(fg->flags, FUTURE_GROUP_CHILDREN_ADDED), -ESTALE); fg->policy = policy; return 0; } +static void future_group_remove_child(FutureGroup *fg, sd_future *child) { + size_t n = 0; + + assert(fg); + assert(child); + + /* A child that was added more than once has one slot per add. Remove every slot of the child, so + * that the group handles the child once. slot_dispatch_handler() allows a callback to free its own + * slot. It also holds a reference on the child until the callback returns, so freeing the slots + * never frees the child. */ + FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) { + if (sd_future_slot_get_future(*slot_p) == child) + sd_future_slot_unref(*slot_p); + else + fg->slots[n++] = *slot_p; + } + + fg->n_slots = n; +} + static int group_child_resolved(sd_future *child, void *userdata) { sd_future *g = ASSERT_PTR(userdata); + FutureGroup *fg = ASSERT_PTR(sd_future_get_private(g)); + int cr = sd_future_result(child); + + future_group_remove_child(fg, child); + + if (FLAGS_SET(fg->flags, FUTURE_GROUP_FINALIZING)) + return future_group_check(g); + + if (cr >= 0 && FLAGS_SET(fg->policy, SD_FUTURE_GROUP_WAIT_ANY)) + return future_group_finalize(g, cr, /* propagate_error= */ true); + + if (cr < 0 && fg->first_error == 0) + fg->first_error = cr; + return future_group_check(g); } @@ -281,17 +282,18 @@ int sd_future_group_add(sd_future *f, sd_future *child) { 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); - /* A group waiting for itself can never settle. */ + /* A group that contains itself would wait for itself and never resolve. */ assert_return(child != f, -EINVAL); - /* Child notifications are dispatched on the child's event loop. */ + /* The callbacks of a child run on the event loop of the child, so the group has to use the same + * event loop. */ assert_return(sd_future_get_event(child) == sd_future_get_event(f), -EINVAL); FutureGroup *fg = sd_future_get_private(f); - /* Group is draining: a freshly-added pending child would have missed the cancel loop - * and hang us forever waiting for it to settle. */ - assert_return(!fg->finalizing, -ESTALE); - assert_return(!fg->sealed, -ESTALE); + /* A finalizing group has already cancelled its children. A child added now would not be + * cancelled, and the group could wait for it forever. */ + assert_return(!FLAGS_SET(fg->flags, FUTURE_GROUP_FINALIZING), -ESTALE); + assert_return(!FLAGS_SET(fg->flags, FUTURE_GROUP_SEALED), -ESTALE); if (!GREEDY_REALLOC(fg->slots, fg->n_slots + 1)) return -ENOMEM; @@ -301,13 +303,39 @@ int sd_future_group_add(sd_future *f, sd_future *child) { if (r < 0) return r; - if (fg->priority_set) { + if (FLAGS_SET(fg->flags, FUTURE_GROUP_PRIORITY_SET)) { r = future_group_set_child_priority(child, fg->priority); if (r < 0) return r; } fg->slots[fg->n_slots++] = TAKE_PTR(slot); + fg->flags |= FUTURE_GROUP_CHILDREN_ADDED; + + return 0; +} + +int sd_future_group_spawn(sd_future *g, const char *name, sd_fiber_func_t func, void *userdata, sd_future **ret) { + int r; + + assert_return(g, -EINVAL); + assert_return(sd_future_get_ops(g) == &future_group_ops, -EINVAL); + + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + r = sd_fiber_new(sd_future_get_event(g), name, func, userdata, &f); + if (r < 0) + return r; + + r = sd_future_group_add(g, f); + if (r < 0) + return r; + + /* sd_future_group_add() took its own reference. The cleanup of f would cancel the fiber, so hand + * our reference to the caller or drop it with sd_future_unref(). */ + if (ret) + *ret = TAKE_PTR(f); + else + f = sd_future_unref(f); return 0; } @@ -318,6 +346,7 @@ int sd_future_group_add_many_internal(sd_future *f, ...) { FutureGroup *fg = sd_future_get_private(f); size_t before = fg->n_slots; + bool children_added = FLAGS_SET(fg->flags, FUTURE_GROUP_CHILDREN_ADDED); int r = 0; va_list ap; @@ -333,14 +362,17 @@ int sd_future_group_add_many_internal(sd_future *f, ...) { } va_end(ap); - if (r < 0) - /* No callbacks run inline while adding children, so the group cannot start finalizing - * during this call. Roll back only this call's additions. */ + if (r < 0) { + /* sd_future_group_add() never runs callbacks, so the group cannot start finalizing during + * this call. Remove only the children that this call added. */ while (fg->n_slots > before) { sd_future_slot_unref(fg->slots[--fg->n_slots]); fg->slots[fg->n_slots] = NULL; } + SET_FLAG(fg->flags, FUTURE_GROUP_CHILDREN_ADDED, children_added); + } + return r; } @@ -349,18 +381,7 @@ int sd_future_group_seal(sd_future *f) { assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL); FutureGroup *fg = sd_future_get_private(f); - fg->sealed = true; + fg->flags |= FUTURE_GROUP_SEALED; - /* A group with children resolves from the callbacks of its children. A finalizing group - * resolves once its pending children settle. */ - if (fg->n_slots > 0 || fg->finalizing) - return 0; - - /* No child can be added to a sealed group. An empty sealed group stays pending forever - * unless it resolves here. WAIT_ANY waits for a successful child, and an empty group has - * none. The group resolves with -ECHILD, like waitid() does when there are no children. */ - return future_group_finalize( - f, - FLAGS_SET(fg->policy, SD_FUTURE_GROUP_WAIT_ANY) ? -ECHILD : 0, - /* propagate_error= */ false); + return future_group_check(f); } diff --git a/src/libsystemd/sd-future/test-channel.c b/src/libsystemd/sd-future/test-channel.c index bad7e055b0..33cec657f2 100644 --- a/src/libsystemd/sd-future/test-channel.c +++ b/src/libsystemd/sd-future/test-channel.c @@ -390,8 +390,8 @@ TEST(channel_fiber_push_pop_fifo) { }; _cleanup_(sd_future_unrefp) sd_future *prod = NULL, *cons = NULL; - ASSERT_OK(sd_fiber_new(e, "producer", producer_fiber, &s, NULL, &prod)); - ASSERT_OK(sd_fiber_new(e, "consumer", consumer_fiber, &s, NULL, &cons)); + ASSERT_OK(sd_fiber_new(e, "producer", producer_fiber, &s, &prod)); + ASSERT_OK(sd_fiber_new(e, "consumer", consumer_fiber, &s, &cons)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(prod)); @@ -537,7 +537,7 @@ TEST(channel_fiber_pop_close_wakeup) { ASSERT_OK(sd_channel_new(e, 2, SD_CHANNEL_OVERFLOW_WAIT, int_destroy, &c)); _cleanup_(sd_future_unrefp) sd_future *fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "blocked-pop", blocked_pop_fiber, c, NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "blocked-pop", blocked_pop_fiber, c, &fiber)); /* Priority 100 makes the close run after the fiber has suspended in sd_channel_pop(). */ _cleanup_(sd_event_source_unrefp) sd_event_source *src = NULL; @@ -738,7 +738,7 @@ TEST(channel_push_closed) { FiberOp op = { .channel = c, .item = INT_TO_PTR(1) }; _cleanup_(sd_future_unrefp) sd_future *fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, &fiber)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(op.result, EPIPE); @@ -758,7 +758,7 @@ TEST(channel_push_closed_while_waiting) { FiberOp op = { .channel = c, .item = INT_TO_PTR(2) }; _cleanup_(sd_future_unrefp) sd_future *fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, &fiber)); run_until_awaiting(e, fiber); ASSERT_OK(sd_channel_close(c)); @@ -780,7 +780,7 @@ TEST(channel_push_cancelled_while_waiting) { FiberOp op = { .channel = c, .item = INT_TO_PTR(2) }; _cleanup_(sd_future_unrefp) sd_future *fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, &fiber)); run_until_awaiting(e, fiber); ASSERT_OK(sd_future_cancel(fiber)); @@ -802,7 +802,7 @@ TEST(channel_push_cancelled_after_send) { FiberOp op = { .channel = c, .item = INT_TO_PTR(2) }; _cleanup_(sd_future_unrefp) sd_future *fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "push", push_fiber, &op, &fiber)); run_until_awaiting(e, fiber); /* The pop moves item 2 into the buffer before the fiber wakes up from the cancellation. */ @@ -831,7 +831,7 @@ TEST(channel_pop_cancelled_after_delivery) { FiberOp op = { .channel = c }; _cleanup_(sd_future_unrefp) sd_future *fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "pop", pop_fiber, &op, NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "pop", pop_fiber, &op, &fiber)); run_until_awaiting(e, fiber); /* The push hands item 5 to the fiber before the fiber wakes up from the cancellation. */ @@ -872,7 +872,7 @@ TEST(channel_push_pop_interrupted) { ASSERT_OK_POSITIVE(sd_channel_try_push(c, INT_TO_PTR(1))); _cleanup_(sd_future_unrefp) sd_future *fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "interrupted", interrupted_push_pop_fiber, c, NULL, &fiber)); + ASSERT_OK(sd_fiber_new(e, "interrupted", interrupted_push_pop_fiber, c, &fiber)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(fiber)); diff --git a/src/libsystemd/sd-future/test-fiber-io.c b/src/libsystemd/sd-future/test-fiber-io.c index c78ee62ca3..4bb8c69bb5 100644 --- a/src/libsystemd/sd-future/test-fiber-io.c +++ b/src/libsystemd/sd-future/test-fiber-io.c @@ -47,7 +47,7 @@ TEST(fiber_io_basic) { PipeIOContext ctx = { .pipefd = pipefd }; _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "pipe-read", pipe_read_fiber, &ctx, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "pipe-read", pipe_read_fiber, &ctx, &f)); /* Write data to the pipe */ ASSERT_OK_EQ_ERRNO(write(pipefd[1], "hello", 5), 5); @@ -106,9 +106,9 @@ TEST(fiber_io_read_write) { * available. The write fiber will run second, write data to the pipe, causing the read fiber to get * resumed. */ _cleanup_(sd_future_unrefp) sd_future *fr = NULL, *fw = NULL; - ASSERT_OK(sd_fiber_new(e, "pipe-read", pipe_read_order_fiber, &ctx, NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "pipe-read", pipe_read_order_fiber, &ctx, &fr)); ASSERT_OK(sd_future_set_priority(fr, 0)); - ASSERT_OK(sd_fiber_new(e, "pipe-write", pipe_write_order_fiber, &ctx, NULL, &fw)); + ASSERT_OK(sd_fiber_new(e, "pipe-write", pipe_write_order_fiber, &ctx, &fw)); ASSERT_OK(sd_future_set_priority(fw, 1)); /* Run the scheduler - should process the I/O */ @@ -152,7 +152,7 @@ TEST(fiber_io_concurrent) { ASSERT_OK_ERRNO(pipe2(pipes[i], O_CLOEXEC | O_NONBLOCK)); args[i][0] = pipes[i][0]; args[i][1] = 'A' + i; - ASSERT_OK(sd_fiber_new(e, "concurrent-read", concurrent_read_fiber, args[i], NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, "concurrent-read", concurrent_read_fiber, args[i], &fibers[i])); } /* Write data in reverse order */ @@ -189,7 +189,7 @@ TEST(fiber_io_cancel) { ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "blocking-read", blocking_read_fiber, INT_TO_PTR(pipefd[0]), NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "blocking-read", blocking_read_fiber, INT_TO_PTR(pipefd[0]), &f)); /* Run once - fiber will suspend on read */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -230,7 +230,7 @@ TEST(fiber_io_interrupted) { ASSERT_OK_EQ_ERRNO(write(pipefd[1], "x", 1), 1); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "interrupted-read", interrupted_read_fiber, INT_TO_PTR(pipefd[0]), NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "interrupted-read", interrupted_read_fiber, INT_TO_PTR(pipefd[0]), &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); } @@ -301,9 +301,9 @@ TEST(fiber_io_readv_writev) { * available. The write fiber will run second, write data to the pipe, causing the read fiber to get * resumed. */ _cleanup_(sd_future_unrefp) sd_future *fr = NULL, *fw = NULL; - ASSERT_OK(sd_fiber_new(e, "pipe-readv", pipe_readv_order_fiber, &ctx, NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "pipe-readv", pipe_readv_order_fiber, &ctx, &fr)); ASSERT_OK(sd_future_set_priority(fr, 0)); - ASSERT_OK(sd_fiber_new(e, "pipe-writev", pipe_writev_order_fiber, &ctx, NULL, &fw)); + ASSERT_OK(sd_fiber_new(e, "pipe-writev", pipe_writev_order_fiber, &ctx, &fw)); ASSERT_OK(sd_future_set_priority(fw, 1)); /* Run the scheduler - should process the I/O */ @@ -352,7 +352,7 @@ TEST(fiber_io_readv_concurrent) { args[i][0] = pipes[i][0]; args[i][1] = 'A' + i; args[i][2] = 'a' + i; - ASSERT_OK(sd_fiber_new(e, "concurrent-readv", concurrent_readv_fiber, args[i], NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, "concurrent-readv", concurrent_readv_fiber, args[i], &fibers[i])); } /* Write data in reverse order */ @@ -420,9 +420,9 @@ TEST(fiber_io_recv_send) { /* Higher priority for the recv fiber, which will run first and suspend */ _cleanup_(sd_future_unrefp) sd_future *fs = NULL, *fr = NULL; - ASSERT_OK(sd_fiber_new(e, "socket-recv", socket_recv_order_fiber, &ctx, NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "socket-recv", socket_recv_order_fiber, &ctx, &fr)); ASSERT_OK(sd_future_set_priority(fr, 0)); - ASSERT_OK(sd_fiber_new(e, "socket-send", socket_send_order_fiber, &ctx, NULL, &fs)); + ASSERT_OK(sd_fiber_new(e, "socket-send", socket_send_order_fiber, &ctx, &fs)); ASSERT_OK(sd_future_set_priority(fs, 1)); ASSERT_OK(sd_event_loop(e)); @@ -466,7 +466,7 @@ TEST(fiber_io_recv_peek) { ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, sockfd)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "socket-recv-peek", socket_recv_peek_fiber, INT_TO_PTR(sockfd[1]), NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "socket-recv-peek", socket_recv_peek_fiber, INT_TO_PTR(sockfd[1]), &f)); /* Write data to the socket */ ASSERT_OK_EQ_ERRNO(write(sockfd[0], "peek", 4), 4); @@ -507,7 +507,7 @@ TEST(fiber_io_connect) { /* Create fiber to connect */ _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "socket-connect", socket_connect_fiber, &addr, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "socket-connect", socket_connect_fiber, &addr, &f)); /* Run the event loop - connection should complete */ ASSERT_OK(sd_event_loop(e)); @@ -560,9 +560,9 @@ TEST(fiber_io_recvmsg_sendmsg) { ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, sockfd)); _cleanup_(sd_future_unrefp) sd_future *fs = NULL, *fr = NULL; - ASSERT_OK(sd_fiber_new(e, "socket-recvmsg", socket_recvmsg_fiber, INT_TO_PTR(sockfd[1]), NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "socket-recvmsg", socket_recvmsg_fiber, INT_TO_PTR(sockfd[1]), &fr)); ASSERT_OK(sd_future_set_priority(fr, 1)); - ASSERT_OK(sd_fiber_new(e, "socket-sendmsg", socket_sendmsg_fiber, INT_TO_PTR(sockfd[0]), NULL, &fs)); + ASSERT_OK(sd_fiber_new(e, "socket-sendmsg", socket_sendmsg_fiber, INT_TO_PTR(sockfd[0]), &fs)); ASSERT_OK(sd_future_set_priority(fs, 0)); ASSERT_OK(sd_event_loop(e)); @@ -605,9 +605,9 @@ TEST(fiber_io_recvfrom_sendto) { ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, sockfd)); _cleanup_(sd_future_unrefp) sd_future *fs = NULL, *fr = NULL; - ASSERT_OK(sd_fiber_new(e, "socket-recvfrom", socket_recvfrom_fiber, INT_TO_PTR(sockfd[1]), NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "socket-recvfrom", socket_recvfrom_fiber, INT_TO_PTR(sockfd[1]), &fr)); ASSERT_OK(sd_future_set_priority(fr, 1)); - ASSERT_OK(sd_fiber_new(e, "socket-sendto", socket_sendto_fiber, INT_TO_PTR(sockfd[0]), NULL, &fs)); + ASSERT_OK(sd_fiber_new(e, "socket-sendto", socket_sendto_fiber, INT_TO_PTR(sockfd[0]), &fs)); ASSERT_OK(sd_future_set_priority(fs, 0)); ASSERT_OK(sd_event_loop(e)); @@ -702,9 +702,9 @@ TEST(fiber_io_sendmsg_recvmsg_fd) { _cleanup_(sd_future_unrefp) sd_future *fs = NULL, *fr = NULL; int args[2] = { sockfd[0], test_fd }; - ASSERT_OK(sd_fiber_new(e, "socket-recvmsg-fd", socket_recvmsg_fd_fiber, INT_TO_PTR(sockfd[1]), NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "socket-recvmsg-fd", socket_recvmsg_fd_fiber, INT_TO_PTR(sockfd[1]), &fr)); ASSERT_OK(sd_future_set_priority(fr, 1)); - ASSERT_OK(sd_fiber_new(e, "socket-sendmsg-fd", socket_sendmsg_fd_fiber, args, NULL, &fs)); + ASSERT_OK(sd_fiber_new(e, "socket-sendmsg-fd", socket_sendmsg_fd_fiber, args, &fs)); ASSERT_OK(sd_future_set_priority(fs, 0)); ASSERT_OK(sd_event_loop(e)); @@ -744,7 +744,7 @@ TEST(fiber_io_socket_cancel) { ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, sockfd)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "blocking-recv", blocking_recv_fiber, INT_TO_PTR(sockfd[0]), NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "blocking-recv", blocking_recv_fiber, INT_TO_PTR(sockfd[0]), &f)); /* Run once - fiber will suspend on recv */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -794,7 +794,7 @@ TEST(fiber_io_accept_basic) { /* Create fiber to accept connection */ _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "accept", accept_fiber, INT_TO_PTR(listen_fd), NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "accept", accept_fiber, INT_TO_PTR(listen_fd), &f)); /* Connect from outside fiber context */ _cleanup_close_ int connect_fd = -EBADF; @@ -847,7 +847,7 @@ TEST(fiber_io_accept_multiple) { /* Create fiber to accept multiple connections */ _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "accept-multiple", accept_multiple_fiber, INT_TO_PTR(listen_fd), NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "accept-multiple", accept_multiple_fiber, INT_TO_PTR(listen_fd), &f)); /* Connect multiple times */ int connect_fds[3] = { -EBADF, -EBADF, -EBADF }; @@ -907,7 +907,7 @@ TEST(fiber_io_accept_and_read) { /* Create fiber to accept and read */ _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "accept-and-read", accept_and_read_fiber, INT_TO_PTR(listen_fd), NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "accept-and-read", accept_and_read_fiber, INT_TO_PTR(listen_fd), &f)); /* Connect and send data */ _cleanup_close_ int connect_fd = -EBADF; @@ -954,7 +954,7 @@ TEST(fiber_poll_immediate) { ASSERT_OK_EQ_ERRNO(write(pipefd[1], "X", 1), 1); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-immediate", poll_immediate_fiber, pipefd, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-immediate", poll_immediate_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -992,7 +992,7 @@ TEST(fiber_poll) { ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-suspend", poll_fiber, pipefd, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-suspend", poll_fiber, pipefd, &f)); /* Run once - fiber will suspend on poll */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -1046,7 +1046,7 @@ TEST(fiber_poll_multiple) { ASSERT_OK_ERRNO(pipe2(pipes[i], O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-multiple", poll_multiple_fiber, pipes, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-multiple", poll_multiple_fiber, pipes, &f)); /* Run once - fiber will suspend waiting for data */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -1094,7 +1094,7 @@ TEST(fiber_poll_pollout) { ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-pollout", poll_pollout_fiber, pipefd, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-pollout", poll_pollout_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -1134,7 +1134,7 @@ TEST(fiber_poll_timeout) { ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-timeout", poll_timeout_fiber, pipefd, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-timeout", poll_timeout_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -1172,8 +1172,7 @@ TEST(fiber_poll_interrupted) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); PollInterruptState s = { .fd = pipefd[0], .timeout = timeout }; - ASSERT_OK(sd_fiber_new(e, "poll-interrupt", poll_interrupt_fiber, &s, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-interrupt", poll_interrupt_fiber, &s, &f)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); if (!timeout) ASSERT_OK(sd_future_cancel(f)); @@ -1223,7 +1222,7 @@ TEST(fiber_poll_zero_timeout) { ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-zero-timeout", poll_zero_timeout_fiber, pipefd, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-zero-timeout", poll_zero_timeout_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -1240,7 +1239,7 @@ TEST(fiber_poll_zero_fds) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-zero-fds", poll_zero_fds_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-zero-fds", poll_zero_fds_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_EQ(sd_future_result(f), 0); @@ -1257,7 +1256,7 @@ TEST(fiber_poll_zero_fds_no_timeout) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-zero-fds-no-timeout", poll_zero_fds_no_timeout_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-zero-fds-no-timeout", poll_zero_fds_no_timeout_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), EINVAL); @@ -1281,8 +1280,7 @@ TEST(fiber_poll_all_entries_skipped) { ASSERT_OK_ERRNO(pipe2(pipefd, O_NONBLOCK | O_CLOEXEC)); ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "poll-all-skipped", poll_all_entries_skipped_fiber, pipefd, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-all-skipped", poll_all_entries_skipped_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); } @@ -1323,7 +1321,7 @@ TEST(fiber_poll_negative_fd) { ASSERT_OK_EQ_ERRNO(write(pipefd[1], "N", 1), 1); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "poll-negative-fd", poll_negative_fd_fiber, pipefd, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "poll-negative-fd", poll_negative_fd_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -1370,7 +1368,7 @@ TEST(fiber_io_same_fd_multiple_fibers) { for (size_t i = 0; i < ELEMENTSOF(fibers); i++) { args[i].pipefd = pipefd[0]; args[i].counter = &counter; - ASSERT_OK(sd_fiber_new(e, "shared-fd-read", shared_fd_read_fiber, &args[i], NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, "shared-fd-read", shared_fd_read_fiber, &args[i], &fibers[i])); } /* All fibers should suspend waiting for data */ @@ -1421,7 +1419,7 @@ TEST(fiber_io_blocking_fd_preserved) { ASSERT_OK_EQ_ERRNO(write(pipefd[1], "blocking", 8), 8); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "blocking-fd-preserve", blocking_fd_preserve_fiber, pipefd, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "blocking-fd-preserve", blocking_fd_preserve_fiber, pipefd, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -1474,7 +1472,7 @@ TEST(fiber_io_connect_blocking) { ASSERT_OK(listen(listen_fd, 1)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "connect-blocking", socket_connect_blocking_fiber, &addr, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "connect-blocking", socket_connect_blocking_fiber, &addr, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); diff --git a/src/libsystemd/sd-future/test-fiber-ops.c b/src/libsystemd/sd-future/test-fiber-ops.c index 35a9952710..5918767f13 100644 --- a/src/libsystemd/sd-future/test-fiber-ops.c +++ b/src/libsystemd/sd-future/test-fiber-ops.c @@ -50,7 +50,7 @@ TEST(wait_for_terminate_fiber_basic) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "wait-simple", wait_simple_fiber, NULL, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "wait-simple", wait_simple_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -102,7 +102,7 @@ TEST(wait_for_terminate_fiber_multiple) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "wait-multiple", wait_multiple_fiber, NULL, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "wait-multiple", wait_multiple_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -143,7 +143,7 @@ TEST(wait_for_terminate_fiber_concurrent) { /* Create 3 fibers, each waiting for a different child */ for (size_t i = 0; i < ELEMENTSOF(fibers); i++) - ASSERT_OK(sd_fiber_new(e, "concurrent-wait", concurrent_wait_fiber, INT_TO_PTR(20 + i), /* destroy= */ NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, "concurrent-wait", concurrent_wait_fiber, INT_TO_PTR(20 + i), &fibers[i])); ASSERT_OK(sd_event_loop(e)); @@ -212,9 +212,9 @@ TEST(loop_read_write_suspend) { }; _cleanup_(sd_future_unrefp) sd_future *fr = NULL, *fw = NULL; - ASSERT_OK(sd_fiber_new(e, "loop-read", loop_read_suspend_fiber, &ctx, /* destroy= */ NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "loop-read", loop_read_suspend_fiber, &ctx, &fr)); ASSERT_OK(sd_future_set_priority(fr, 0)); - ASSERT_OK(sd_fiber_new(e, "loop-write", loop_write_suspend_fiber, &ctx, /* destroy= */ NULL, &fw)); + ASSERT_OK(sd_fiber_new(e, "loop-write", loop_write_suspend_fiber, &ctx, &fw)); ASSERT_OK(sd_future_set_priority(fw, 1)); ASSERT_OK(sd_event_loop(e)); @@ -245,7 +245,7 @@ TEST(loop_read_exact_short) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; ASSERT_OK(sd_fiber_new(e, "loop-read-exact", loop_read_exact_short_fiber, - INT_TO_PTR(pipefd[0]), /* destroy= */ NULL, &f)); + INT_TO_PTR(pipefd[0]), &f)); /* Write a few bytes and close the write end — less than the fiber asked for. */ ASSERT_OK_EQ_ERRNO(write(pipefd[1], "abc", 3), (ssize_t) 3); @@ -288,7 +288,7 @@ TEST(loop_write_full_timeout) { LoopWriteTimeoutContext ctx = { .fd = pipefd[1], .result = 0 }; _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "loop-write-timeout", loop_write_timeout_fiber, &ctx, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "loop-write-timeout", loop_write_timeout_fiber, &ctx, &f)); ASSERT_OK(sd_event_loop(e)); @@ -351,9 +351,9 @@ TEST(ppoll_usec_dispatch) { PpollDispatchContext ctx = { .pipefd = pipefd }; _cleanup_(sd_future_unrefp) sd_future *fr = NULL, *fw = NULL; - ASSERT_OK(sd_fiber_new(e, "ppoll-read", ppoll_dispatch_read_fiber, &ctx, /* destroy= */ NULL, &fr)); + ASSERT_OK(sd_fiber_new(e, "ppoll-read", ppoll_dispatch_read_fiber, &ctx, &fr)); ASSERT_OK(sd_future_set_priority(fr, 0)); - ASSERT_OK(sd_fiber_new(e, "ppoll-write", ppoll_dispatch_write_fiber, &ctx, /* destroy= */ NULL, &fw)); + ASSERT_OK(sd_fiber_new(e, "ppoll-write", ppoll_dispatch_write_fiber, &ctx, &fw)); ASSERT_OK(sd_future_set_priority(fw, 1)); ASSERT_OK(sd_event_loop(e)); @@ -384,7 +384,7 @@ TEST(loop_write_zero_timeout_nonblock) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; ASSERT_OK(sd_fiber_new(e, "loop-write-zt-nb", loop_write_zero_timeout_nonblock_fiber, - INT_TO_PTR(pipefd[1]), /* destroy= */ NULL, &f)); + INT_TO_PTR(pipefd[1]), &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), EAGAIN); @@ -445,10 +445,10 @@ TEST(loop_write_zero_timeout_blocking) { _cleanup_(sd_future_unrefp) sd_future *fw = NULL, *fr = NULL; ASSERT_OK(sd_fiber_new(e, "loop-write-zt-blk", loop_write_zero_blocking_writer_fiber, - &ctx, /* destroy= */ NULL, &fw)); + &ctx, &fw)); ASSERT_OK(sd_future_set_priority(fw, 0)); ASSERT_OK(sd_fiber_new(e, "loop-read-zt-blk", loop_write_zero_blocking_reader_fiber, - &ctx, /* destroy= */ NULL, &fr)); + &ctx, &fr)); ASSERT_OK(sd_future_set_priority(fr, 1)); ASSERT_OK(sd_event_loop(e)); @@ -477,7 +477,7 @@ TEST(loop_read_no_poll_nonblock) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; ASSERT_OK(sd_fiber_new(e, "loop-read-np-nb", loop_read_no_poll_nonblock_fiber, - INT_TO_PTR(pipefd[0]), /* destroy= */ NULL, &f)); + INT_TO_PTR(pipefd[0]), &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), EAGAIN); @@ -544,10 +544,10 @@ TEST(loop_read_no_poll_blocking) { _cleanup_(sd_future_unrefp) sd_future *fr = NULL, *fw = NULL; ASSERT_OK(sd_fiber_new(e, "loop-read-np-blk", loop_read_no_poll_blocking_reader_fiber, - &ctx, /* destroy= */ NULL, &fr)); + &ctx, &fr)); ASSERT_OK(sd_future_set_priority(fr, 0)); ASSERT_OK(sd_fiber_new(e, "loop-write-np-blk", loop_read_no_poll_blocking_writer_fiber, - &ctx, /* destroy= */ NULL, &fw)); + &ctx, &fw)); ASSERT_OK(sd_future_set_priority(fw, 1)); ASSERT_OK(sd_event_loop(e)); diff --git a/src/libsystemd/sd-future/test-fiber.c b/src/libsystemd/sd-future/test-fiber.c index ac699a99c1..8b4aef3fbc 100644 --- a/src/libsystemd/sd-future/test-fiber.c +++ b/src/libsystemd/sd-future/test-fiber.c @@ -29,7 +29,7 @@ TEST(fiber_simple) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; int value = 5; - ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_EQ(sd_future_result(f), 5); } @@ -53,7 +53,7 @@ TEST(fiber_single_yield) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "yielding", yielding_fiber, &counter, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "yielding", yielding_fiber, &counter, &f)); /* First iteration: fiber runs until first yield */ ASSERT_EQ(counter, 0); @@ -91,7 +91,7 @@ TEST(fiber_multiple_yield) { for (size_t i = 0; i < ELEMENTSOF(fibers); i++) { _cleanup_free_ char *name = NULL; ASSERT_OK(asprintf(&name, "counting-%zu", i)); - ASSERT_OK(sd_fiber_new(e, name, counting_fiber, NULL, NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, name, counting_fiber, NULL, &fibers[i])); } ASSERT_OK(sd_event_loop(e)); @@ -122,7 +122,7 @@ TEST(fiber_priority_ascending) { for (size_t i = 0; i < ELEMENTSOF(fibers); i++) { _cleanup_free_ char *name = NULL; ASSERT_OK(asprintf(&name, "priority-%zu", i)); - ASSERT_OK(sd_fiber_new(e, name, priority_fiber, &counter, NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, name, priority_fiber, &counter, &fibers[i])); ASSERT_OK(sd_future_set_priority(fibers[i], i)); } @@ -147,7 +147,7 @@ TEST(fiber_priority_identical) { for (size_t i = 0; i < ELEMENTSOF(fibers); i++) { _cleanup_free_ char *name = NULL; ASSERT_OK(asprintf(&name, "priority-%zu", i)); - ASSERT_OK(sd_fiber_new(e, name, priority_fiber, &counter, NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, name, priority_fiber, &counter, &fibers[i])); } ASSERT_OK(sd_event_loop(e)); @@ -169,7 +169,7 @@ TEST(fiber_error_return) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "error", error_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "error", error_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_EQ(sd_future_result(f), -ENOENT); @@ -186,7 +186,7 @@ TEST(fiber_cancel_basic) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; int value = 42; - ASSERT_OK(sd_fiber_new(e, "cancel", cancel_fiber, &value, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "cancel", cancel_fiber, &value, &f)); ASSERT_OK(sd_future_cancel(f)); ASSERT_OK(sd_event_loop(e)); @@ -215,7 +215,7 @@ TEST(fiber_cancel_propagation_via_yield) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; int yield_count = 0; - ASSERT_OK(sd_fiber_new(e, "yielding", fiber_that_yields, &yield_count, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "yielding", fiber_that_yields, &yield_count, &f)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_EQ(yield_count, 1); @@ -239,7 +239,7 @@ TEST(fiber_cancel_completed) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; int value = 42; - ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, &f)); /* Run the fiber to completion */ ASSERT_OK(sd_event_loop(e)); @@ -273,7 +273,7 @@ TEST(fiber_cancel_one_of_many) { CLEANUP_ELEMENTS(fibers, sd_future_unref_array_clear); int counters[3] = {0, 0, 0}; for (size_t i = 0; i < ELEMENTSOF(fibers); i++) - ASSERT_OK(sd_fiber_new(e, "multiple-yield", multiple_yield_fiber, &counters[i], NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, "multiple-yield", multiple_yield_fiber, &counters[i], &fibers[i])); /* Run one iteration - all fibers yield after incrementing once */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -332,11 +332,11 @@ TEST(fiber_wait_for_basic) { /* Create target fiber with lower priority (runs second) */ _cleanup_(sd_future_unrefp) sd_future *target = NULL, *waiter = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "slow", slow_fiber, &counter, NULL, &target)); + ASSERT_OK(sd_fiber_new(e, "slow", slow_fiber, &counter, &target)); ASSERT_OK(sd_future_set_priority(target, 1)); /* Create waiter fiber with higher priority (runs first) */ - ASSERT_OK(sd_fiber_new(e, "waiting", waiting_fiber, target, NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "waiting", waiting_fiber, target, &waiter)); ASSERT_OK(sd_future_set_priority(waiter, 0)); ASSERT_OK(sd_event_loop(e)); @@ -367,10 +367,10 @@ TEST(fiber_wait_for_completed) { int value = 100; /* Create target fiber with higher priority (runs first) */ - ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, NULL, &target)); + ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, &target)); ASSERT_OK(sd_future_set_priority(target, 0)); /* Create waiter fiber with lower priority (runs second, after target completes) */ - ASSERT_OK(sd_fiber_new(e, "wait-for-completed", wait_for_completed_fiber, target, NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "wait-for-completed", wait_for_completed_fiber, target, &waiter)); ASSERT_OK(sd_future_set_priority(waiter, 1)); ASSERT_OK(sd_event_loop(e)); @@ -398,9 +398,9 @@ TEST(fiber_await_resolved) { int value = 77; /* Higher-priority target runs to completion before the waiter starts. */ - ASSERT_OK(sd_fiber_new(e, "target", simple_fiber, &value, NULL, &target)); + ASSERT_OK(sd_fiber_new(e, "target", simple_fiber, &value, &target)); ASSERT_OK(sd_future_set_priority(target, 0)); - ASSERT_OK(sd_fiber_new(e, "await-resolved", await_resolved_fiber, target, NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "await-resolved", await_resolved_fiber, target, &waiter)); ASSERT_OK(sd_future_set_priority(waiter, 1)); ASSERT_OK(sd_event_loop(e)); @@ -428,8 +428,8 @@ TEST(fiber_wait_for_cancelled) { _cleanup_(sd_future_unrefp) sd_future *target = NULL, *waiter = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "yielding", fiber_that_yields, &counter, NULL, &target)); - ASSERT_OK(sd_fiber_new(e, "wait-for-cancelled", wait_for_cancelled_fiber, target, NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "yielding", fiber_that_yields, &counter, &target)); + ASSERT_OK(sd_fiber_new(e, "wait-for-cancelled", wait_for_cancelled_fiber, target, &waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -461,12 +461,12 @@ TEST(fiber_wait_for_multiple_waiters) { _cleanup_(sd_future_unrefp) sd_future *target = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "slow", slow_fiber, &counter, NULL, &target)); + ASSERT_OK(sd_fiber_new(e, "slow", slow_fiber, &counter, &target)); sd_future *waiters[3] = {}; CLEANUP_ELEMENTS(waiters, sd_future_unref_array_clear); for (size_t i = 0; i < ELEMENTSOF(waiters); i++) - ASSERT_OK(sd_fiber_new(e, "multi-waiter", multi_waiter_fiber, target, NULL, &waiters[i])); + ASSERT_OK(sd_fiber_new(e, "multi-waiter", multi_waiter_fiber, target, &waiters[i])); ASSERT_OK(sd_event_loop(e)); @@ -499,11 +499,11 @@ TEST(fiber_wait_for_chain) { CLEANUP_ELEMENTS(fibers, sd_future_unref_array_clear); int value = 10; - ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, NULL, &fibers[0])); + ASSERT_OK(sd_fiber_new(e, "simple", simple_fiber, &value, &fibers[0])); /* Each subsequent fiber waits for the previous and adds 1 */ for (size_t i = 1; i < ELEMENTSOF(fibers); i++) - ASSERT_OK(sd_fiber_new(e, "chain-waiter", chain_waiter_fiber, fibers[i - 1], NULL, &fibers[i])); + ASSERT_OK(sd_fiber_new(e, "chain-waiter", chain_waiter_fiber, fibers[i - 1], &fibers[i])); ASSERT_OK(sd_event_loop(e)); @@ -548,7 +548,7 @@ static int nested_run_outer_fiber(void *userdata) { /* Spawn a fiber on the inner event loop. Driving it via sd_event_loop(inner) causes fiber_run() to * be invoked while we are already executing inside fiber_run() for the outer fiber. */ - r = sd_fiber_new(inner, "inner", nested_run_inner_fiber, counter, NULL, &nested); + r = sd_fiber_new(inner, "inner", nested_run_inner_fiber, counter, &nested); if (r < 0) return r; @@ -573,7 +573,7 @@ TEST(fiber_nested_run) { _cleanup_(sd_future_unrefp) sd_future *outer = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "outer", nested_run_outer_fiber, &counter, NULL, &outer)); + ASSERT_OK(sd_fiber_new(e, "outer", nested_run_outer_fiber, &counter, &outer)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(outer)); @@ -613,7 +613,7 @@ static int nested_current_check_outer_fiber(void *userdata) { if (r < 0) return r; - r = sd_fiber_new(inner, "inner", nested_current_check_inner_fiber, slots, NULL, &nested); + r = sd_fiber_new(inner, "inner", nested_current_check_inner_fiber, slots, &nested); if (r < 0) return r; @@ -640,7 +640,7 @@ TEST(fiber_nested_run_current_restored) { sd_future *slots[2] = {}; _cleanup_(sd_future_unrefp) sd_future *outer = NULL; - ASSERT_OK(sd_fiber_new(e, "outer", nested_current_check_outer_fiber, slots, NULL, &outer)); + ASSERT_OK(sd_fiber_new(e, "outer", nested_current_check_outer_fiber, slots, &outer)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(outer)); @@ -673,7 +673,7 @@ static int running_ancestor_fiber(void *userdata) { ASSERT_OK(sd_event_new(&inner)); ASSERT_OK(sd_event_set_exit_on_idle(inner, true)); ASSERT_OK(sd_fiber_new(inner, "cancel-ancestor", cancel_running_ancestor_fiber, - sd_fiber_get_current(), /* destroy= */ NULL, &child)); + sd_fiber_get_current(), &child)); /* The ready child is dispatched inline by the nested loop, without suspending the ancestor. */ ASSERT_OK(sd_event_loop(inner)); @@ -692,8 +692,7 @@ TEST(fiber_cancel_running_ancestor) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "running-ancestor", running_ancestor_fiber, &yield_first, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "running-ancestor", running_ancestor_fiber, &yield_first, &f)); if (yield_first) ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -722,7 +721,7 @@ static int nested_cancellation_fiber(void *userdata) { return -ENOMEM; /* Create a nested fiber within this fiber */ - r = sd_fiber_new(sd_fiber_get_event(), name, nested_cancellation_fiber, counter, NULL, &nested); + r = sd_fiber_new(sd_fiber_get_event(), name, nested_cancellation_fiber, counter, &nested); if (r < 0) return r; @@ -752,7 +751,7 @@ TEST(fiber_exit_ready) { bool finished = false; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "exit-ready", exit_ready_fiber, &finished, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "exit-ready", exit_ready_fiber, &finished, &f)); /* Yield leaves the fiber READY without a queued result. The exit handler must deliver * cancellation and run it in the same dispatch, without needing another iteration. */ @@ -772,7 +771,7 @@ static int exit_ready_parent_fiber(void *userdata) { _cleanup_(sd_future_cancel_wait_unrefp) sd_future *child = NULL; ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "exit-ready-child", cancel_fiber, - /* userdata= */ NULL, /* destroy= */ NULL, &child)); + /* userdata= */ NULL, &child)); *ret_child = sd_future_ref(child); return sd_fiber_suspend(); } @@ -787,8 +786,7 @@ TEST(fiber_exit_ready_child) { _cleanup_(sd_future_unrefp) sd_future *parent = NULL, *child = NULL; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "exit-ready-parent", exit_ready_parent_fiber, &child, - /* destroy= */ NULL, &parent)); + ASSERT_OK(sd_fiber_new(e, "exit-ready-parent", exit_ready_parent_fiber, &child, &parent)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); /* Parent creates child and suspends. */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); /* Child yields, leaving only its defer armed. */ @@ -819,11 +817,11 @@ TEST(fiber_nested_cancellation) { /* Create outer fiber with higher priority (runs first) */ _cleanup_(sd_future_unrefp) sd_future *outer = NULL; - ASSERT_OK(sd_fiber_new(e, "outer", nested_cancellation_fiber, &counter, NULL, &outer)); + ASSERT_OK(sd_fiber_new(e, "outer", nested_cancellation_fiber, &counter, &outer)); /* Create exit fiber with lower priority (runs after all nested fibers have suspended) */ _cleanup_(sd_future_unrefp) sd_future *exit_fiber = NULL; - ASSERT_OK(sd_fiber_new(e, "exit-loop", exit_loop_fiber, NULL, NULL, &exit_fiber)); + ASSERT_OK(sd_fiber_new(e, "exit-loop", exit_loop_fiber, NULL, &exit_fiber)); ASSERT_OK(sd_future_set_priority(exit_fiber, 1)); /* Run the event loop - the exit fiber should cause it to exit, @@ -858,7 +856,7 @@ static int nested_fiber_cleanup_fiber(void *userdata) { int r; /* Create a nested fiber within this fiber. */ - r = sd_fiber_new(sd_fiber_get_event(), "nested", nested_fiber_cleanup_nested_fiber, userdata, NULL, &nested); + r = sd_fiber_new(sd_fiber_get_event(), "nested", nested_fiber_cleanup_nested_fiber, userdata, &nested); if (r < 0) return r; @@ -873,7 +871,7 @@ TEST(nested_fiber_cleanup) { _cleanup_(sd_future_unrefp) sd_future *outer = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "outer", nested_fiber_cleanup_fiber, &counter, NULL, &outer)); + ASSERT_OK(sd_fiber_new(e, "outer", nested_fiber_cleanup_fiber, &counter, &outer)); ASSERT_OK(sd_event_loop(e)); @@ -902,7 +900,7 @@ TEST(fiber_priority_get) { int64_t got_priority = 0; _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "priority-check", priority_check_fiber, &got_priority, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "priority-check", priority_check_fiber, &got_priority, &f)); ASSERT_OK(sd_future_set_priority(f, 10)); ASSERT_OK(sd_event_loop(e)); @@ -931,7 +929,7 @@ TEST(fiber_floating) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "floating", floating_fiber, &counter, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "floating", floating_fiber, &counter, &f)); ASSERT_OK_ZERO(sd_fiber_get_floating(f)); ASSERT_OK(sd_fiber_set_floating(f, true)); @@ -955,17 +953,46 @@ static void fire_and_forget_destroy(void *userdata) { TEST(fiber_fire_and_forget) { _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_unrefp) sd_future *f = NULL; int counter = 0; ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "fire-and-forget", floating_fiber, &counter, - fire_and_forget_destroy, /* ret= */ NULL)); + ASSERT_OK(sd_fiber_new(e, "fire-and-forget", floating_fiber, &counter, &f)); + ASSERT_OK(sd_fiber_set_destroy_callback(f, fire_and_forget_destroy)); + ASSERT_OK(sd_fiber_set_floating(f, true)); + f = sd_future_unref(f); ASSERT_EQ(counter, 0); ASSERT_OK(sd_event_loop(e)); ASSERT_EQ(counter, 3); } +TEST(fiber_destroy_callback) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL, *defer = NULL; + sd_fiber_destroy_t callback; + int counter = 0; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_fiber_new(e, "destroy-callback", floating_fiber, &counter, &f)); + + ASSERT_OK_ZERO(sd_fiber_get_destroy_callback(f, &callback)); + ASSERT_NULL(callback); + + ASSERT_OK(sd_fiber_set_destroy_callback(f, fire_and_forget_destroy)); + ASSERT_OK_POSITIVE(sd_fiber_get_destroy_callback(f, &callback)); + ASSERT_PTR_EQ(callback, fire_and_forget_destroy); + ASSERT_OK_POSITIVE(sd_fiber_get_destroy_callback(f, /* ret= */ NULL)); + + ASSERT_OK(sd_fiber_set_destroy_callback(f, /* callback= */ NULL)); + ASSERT_OK_ZERO(sd_fiber_get_destroy_callback(f, &callback)); + ASSERT_NULL(callback); + + ASSERT_OK(sd_future_new_defer(e, 0, &defer)); + ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_fiber_set_destroy_callback(defer, fire_and_forget_destroy)), EINVAL); + ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_fiber_get_destroy_callback(defer, &callback)), EINVAL); +} + static void unstarted_fiber_destroy(void *userdata) { int *counter = ASSERT_PTR(userdata); @@ -974,23 +1001,22 @@ static void unstarted_fiber_destroy(void *userdata) { } TEST(fiber_floating_exit_before_start) { - bool return_handle; + bool queue_cancellation; - FOREACH_ARGUMENT(return_handle, false, true) { + FOREACH_ARGUMENT(queue_cancellation, false, true) { _cleanup_(sd_event_unrefp) sd_event *e = NULL; _cleanup_(sd_future_unrefp) sd_future *f = NULL; int counter = 0; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "unstarted-floating", yielding_fiber, &counter, - unstarted_fiber_destroy, return_handle ? &f : NULL)); - if (f) { - ASSERT_OK(sd_fiber_set_floating(f, true)); - /* An already-queued cancellation must not bypass synchronous cancellation of - * a fiber whose stack has never been entered. */ + ASSERT_OK(sd_fiber_new(e, "unstarted-floating", yielding_fiber, &counter, &f)); + ASSERT_OK(sd_fiber_set_destroy_callback(f, unstarted_fiber_destroy)); + ASSERT_OK(sd_fiber_set_floating(f, true)); + /* An already-queued cancellation must not bypass synchronous cancellation of a fiber + * whose stack has never been entered. */ + if (queue_cancellation) ASSERT_OK(sd_fiber_resume(f, -ECANCELED)); - f = sd_future_unref(f); - } + f = sd_future_unref(f); /* Cancellation drops the floating self-reference without ever entering the fiber. * The exit handler must return without accessing the freed fiber afterwards. */ @@ -1015,7 +1041,7 @@ TEST(fiber_floating_callback_drops_ref) { sd_future *f = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "floating-cb", floating_fiber, &counter, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "floating-cb", floating_fiber, &counter, &f)); ASSERT_OK(sd_fiber_set_floating(f, true)); @@ -1038,7 +1064,7 @@ TEST(fiber_floating_toggle) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; int counter = 0; - ASSERT_OK(sd_fiber_new(e, "floating-toggle", floating_fiber, &counter, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "floating-toggle", floating_fiber, &counter, &f)); /* Toggling floating on and off again should leave the refcount unchanged: set_floating(true) * takes a ref and set_floating(false) drops it. If the accounting were off, the subsequent @@ -1074,7 +1100,7 @@ TEST(fiber_timeout_suspend_expires) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "timeout-suspend", timeout_suspend_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "timeout-suspend", timeout_suspend_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), ETIME); @@ -1093,7 +1119,7 @@ TEST(fiber_timeout_sleep_in_time) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "in-time", timeout_in_time_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "in-time", timeout_in_time_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); @@ -1112,7 +1138,7 @@ TEST(fiber_timeout_infinite_no_op) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "infinite", timeout_infinite_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "infinite", timeout_infinite_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); @@ -1132,7 +1158,7 @@ TEST(fiber_with_timeout_block) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "with-timeout", with_timeout_block_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "with-timeout", with_timeout_block_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), ETIME); @@ -1169,7 +1195,7 @@ TEST(fiber_timeout_nested) { int fired = 0; _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "nested-timeout", nested_timeout_fiber, &fired, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "nested-timeout", nested_timeout_fiber, &fired, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); @@ -1231,7 +1257,7 @@ TEST(fiber_cancel_wait_unref_loops_until_resolved) { 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_fiber_new(e, "stubborn", cancel_wait_loops_fiber, &cancel_count, &f)); ASSERT_OK(sd_event_loop(e)); /* Loop must have called cancel at least twice — once before the timeout interrupted @@ -1272,7 +1298,7 @@ TEST(fiber_resume_queues_while_running) { 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_fiber_new(e, "resume-queue", resume_queue_while_running_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -1293,7 +1319,7 @@ TEST(fiber_self_resume_does_not_arm_source) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_fiber_new(e, "self-resume", self_resume_then_suspend_fiber, - /* userdata= */ NULL, /* destroy= */ NULL, &f)); + /* userdata= */ NULL, &f)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); /* Yield. */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); /* Consume queued results, then really suspend. */ @@ -1325,7 +1351,7 @@ TEST(fiber_self_resume_during_exit_does_not_arm_source) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_fiber_new(e, "self-resume-exit", self_resume_during_exit_fiber, - /* userdata= */ NULL, /* destroy= */ NULL, &f)); + /* userdata= */ NULL, &f)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); /* Cleanup must really suspend before this lower-priority source wakes it. A stray dispatch @@ -1368,7 +1394,7 @@ TEST(fiber_cancel_wait_unref_propagates_timeout) { 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_fiber_new(e, "propagate-timeout", cancel_wait_propagates_timeout_fiber, &cancel_count, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), ETIME); @@ -1406,7 +1432,7 @@ TEST(fiber_cancel_wait_unref_propagates_cancellation_from_main) { 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)); + ASSERT_OK(sd_fiber_new(e, "propagate-cancel", cancel_wait_propagates_cancellation_fiber, &cancel_count, &f)); /* One iteration drives the fiber through its first cancel + await inside * cancel_wait_unref, leaving it suspended waiting for the stubborn future. */ @@ -1437,13 +1463,13 @@ TEST(fiber_cancel_wait_unref_propagates_cancellation_from_peer_fiber) { 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_fiber_new(e, "target", cancel_wait_propagates_cancellation_fiber, &cancel_count, &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_fiber_new(e, "canceller", peer_canceller_fiber, &data, &canceller)); ASSERT_OK(sd_future_set_priority(canceller, 1)); ASSERT_OK(sd_event_loop(e)); @@ -1510,7 +1536,7 @@ TEST(fiber_cancel_wait_unref_no_phantom_cancellation_on_async_resolve) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "async-resolve", cancel_wait_async_resolve_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "async-resolve", cancel_wait_async_resolve_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(sd_future_result(f)); @@ -1554,8 +1580,7 @@ TEST(fiber_cancel_wait_unref_chain) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "cleanup-chain", cancel_wait_chain_fiber, &mode, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "cleanup-chain", cancel_wait_chain_fiber, &mode, &f)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); if (mode != 1) ASSERT_OK(sd_future_cancel(f)); @@ -1649,8 +1674,7 @@ TEST(future_default_event_constructors) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; ASSERT_OK(sd_event_default(&e)); - ASSERT_OK(sd_fiber_new(SD_EVENT_DEFAULT, "default-event", default_event_fiber, e, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(SD_EVENT_DEFAULT, "default-event", default_event_fiber, e, &f)); while (ASSERT_OK(sd_event_run(e, 0)) > 0) ; @@ -1685,8 +1709,7 @@ TEST(fiber_await_result) { _cleanup_(sd_future_unrefp) sd_future *f = NULL; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "await-result", await_result_fiber, /* userdata= */ NULL, - /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "await-result", await_result_fiber, /* userdata= */ NULL, &f)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); @@ -1713,8 +1736,7 @@ TEST(fiber_await_last_reference) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); - ASSERT_OK(sd_fiber_new(e, "await-borrowed", await_borrowed_fiber, target, - /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "await-borrowed", await_borrowed_fiber, target, &waiter)); /* Suspend in await, so the waiter's own reference and the wait itself keep target alive. */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -1740,7 +1762,7 @@ TEST(fiber_await_unowned) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); - ASSERT_OK(sd_fiber_new(e, "await-unowned", await_unowned_fiber, target, /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "await-unowned", await_unowned_fiber, target, &waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_EQ(sd_future_state(waiter), SD_FUTURE_PENDING); @@ -1800,7 +1822,8 @@ TEST(fiber_await_invalid) { mf->freed = &target_freed; s.other = target; ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_fiber_await(target)), ESRCH); - ASSERT_OK(sd_fiber_new(e, "await-invalid", await_invalid_fiber, &s, await_invalid_destroy, &waiter)); + ASSERT_OK(sd_fiber_new(e, "await-invalid", await_invalid_fiber, &s, &waiter)); + ASSERT_OK(sd_fiber_set_destroy_callback(waiter, await_invalid_destroy)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(waiter)); @@ -1821,7 +1844,7 @@ TEST(fiber_get_awaiting) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); ASSERT_NULL(ASSERT_RETURN_EXPECTED(sd_fiber_get_awaiting(target))); - ASSERT_OK(sd_fiber_new(e, "get-awaiting", await_borrowed_fiber, target, /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "get-awaiting", await_borrowed_fiber, target, &waiter)); ASSERT_NULL(sd_fiber_get_awaiting(waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -1842,7 +1865,7 @@ TEST(fiber_await_resumed_before_completion) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); - ASSERT_OK(sd_fiber_new(e, "await-resumed", await_borrowed_fiber, target, /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "await-resumed", await_borrowed_fiber, target, &waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_EQ(sd_future_state(waiter), SD_FUTURE_PENDING); @@ -1881,7 +1904,7 @@ TEST(fiber_cancel_wait_unref_propagates_cancellation_colliding_with_resolve) { ASSERT_OK(sd_future_new(e, &manual_future_ops, &mf)); _cleanup_(sd_future_unrefp) sd_future *fib = NULL; - ASSERT_OK(sd_fiber_new(e, "collision", cancel_wait_collision_fiber, mf, NULL, &fib)); + ASSERT_OK(sd_fiber_new(e, "collision", cancel_wait_collision_fiber, mf, &fib)); /* Drive the fiber to suspend inside cancel_wait_unref's await on mf (mf's no-op cancel leaves * it PENDING). */ @@ -1922,7 +1945,7 @@ TEST(fiber_cancel_wait_unref_keeps_interrupt_value_over_later_resolution) { ASSERT_OK(sd_future_new(e, &manual_future_ops, &mf)); _cleanup_(sd_future_unrefp) sd_future *fib = NULL; - ASSERT_OK(sd_fiber_new(e, "keep-interrupt", cancel_wait_keeps_interrupt_value_fiber, mf, NULL, &fib)); + ASSERT_OK(sd_fiber_new(e, "keep-interrupt", cancel_wait_keeps_interrupt_value_fiber, mf, &fib)); /* Reach the first await on mf (no-op cancel leaves mf PENDING). */ ASSERT_OK_POSITIVE(sd_event_run(e, 0)); @@ -1991,9 +2014,9 @@ TEST(fiber_signal_mask_is_per_thread) { ASSERT_OK_ZERO(-pthread_sigmask(SIG_SETMASK, NULL, &saved)); _cleanup_(sd_future_unrefp) sd_future *waiter = NULL, *peer = NULL; - ASSERT_OK(sd_fiber_new(e, "sigmask-peer", sigmask_peer_fiber, NULL, NULL, &peer)); + ASSERT_OK(sd_fiber_new(e, "sigmask-peer", sigmask_peer_fiber, NULL, &peer)); ASSERT_OK(sd_future_set_priority(peer, 1)); - ASSERT_OK(sd_fiber_new(e, "sigmask-waiter", sigmask_waiter_fiber, peer, NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "sigmask-waiter", sigmask_waiter_fiber, peer, &waiter)); ASSERT_OK(sd_future_set_priority(waiter, 0)); ASSERT_OK(sd_event_loop(e)); @@ -2046,9 +2069,9 @@ TEST(fiber_log_context_per_fiber) { size_t peer_observed = 0; _cleanup_(sd_future_unrefp) sd_future *waiter = NULL, *peer = NULL; - ASSERT_OK(sd_fiber_new(e, "log-peer", log_context_peer_fiber, &peer_observed, NULL, &peer)); + ASSERT_OK(sd_fiber_new(e, "log-peer", log_context_peer_fiber, &peer_observed, &peer)); ASSERT_OK(sd_future_set_priority(peer, 1)); - ASSERT_OK(sd_fiber_new(e, "log-waiter", log_context_waiter_fiber, peer, NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "log-waiter", log_context_waiter_fiber, peer, &waiter)); ASSERT_OK(sd_future_set_priority(waiter, 0)); ASSERT_OK(sd_event_loop(e)); @@ -2105,7 +2128,7 @@ TEST(fiber_stack_guard) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *f = NULL; - ASSERT_OK(sd_fiber_new(e, "overflow", stack_overflow_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "overflow", stack_overflow_fiber, NULL, &f)); (void) sd_event_loop(e); _exit(EXIT_SUCCESS); /* unreachable if the guard fires */ } @@ -2169,8 +2192,7 @@ TEST(future_slot_inherits_fiber_priority) { else ASSERT_OK(sd_event_add_defer(e, &probe, slot_priority_probe, &s)); - ASSERT_OK(sd_fiber_new(e, "register-slot", slot_priority_register, &s, - /* destroy= */ NULL, ®istrar)); + ASSERT_OK(sd_fiber_new(e, "register-slot", slot_priority_register, &s, ®istrar)); ASSERT_OK(sd_future_set_priority(registrar, -10)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK_ZERO(sd_future_result(registrar)); @@ -2309,7 +2331,7 @@ TEST(future_resolve_after_fork) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); - ASSERT_OK(sd_fiber_new(e, "await-fork", await_borrowed_fiber, target, /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "await-fork", await_borrowed_fiber, target, &waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); /* In the child, the event loop belongs to the parent. sd_future_resolve() and sd_fiber_resume() @@ -2415,7 +2437,7 @@ TEST(future_new_defer_basic) { 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_fiber_new(e, "driver", defer_basic_fiber, &count, &driver)); ASSERT_OK(sd_event_loop(e)); @@ -2498,7 +2520,7 @@ TEST(add_callback_resolved_in_fiber_floating_future_freed) { 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_fiber_new(e, "driver", floating_freed_driver, &s, &driver)); ASSERT_OK(sd_event_loop(e)); @@ -2536,7 +2558,7 @@ TEST(defer_slot_cancel_before_fire) { 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_fiber_new(e, "driver", defer_cancel_driver, &s, &driver)); ASSERT_OK(sd_event_loop(e)); @@ -2576,7 +2598,7 @@ TEST(fiber_resume_cancellation_is_sticky) { 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_fiber_new(e, "sticky", sticky_cancel_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), ECANCELED); @@ -2611,7 +2633,7 @@ TEST(fiber_resume_timeout_is_sticky) { 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-timeout", sticky_timeout_fiber, NULL, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "sticky-timeout", sticky_timeout_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(f), ETIME); @@ -2655,8 +2677,8 @@ TEST(fiber_resume_cancel_outranks_timeout) { ASSERT_OK(sd_event_set_exit_on_idle(e, true)); _cleanup_(sd_future_unrefp) sd_future *escalate = NULL, *keep = NULL; - ASSERT_OK(sd_fiber_new(e, "escalate", timeout_then_cancel_fiber, NULL, NULL, &escalate)); - ASSERT_OK(sd_fiber_new(e, "keep", cancel_then_timeout_fiber, NULL, NULL, &keep)); + ASSERT_OK(sd_fiber_new(e, "escalate", timeout_then_cancel_fiber, NULL, &escalate)); + ASSERT_OK(sd_fiber_new(e, "keep", cancel_then_timeout_fiber, NULL, &keep)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(escalate), ECANCELED); @@ -2705,7 +2727,7 @@ TEST(fiber_await_completion_wins) { ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); AwaitInterrupted a = { .target = target }; - ASSERT_OK(sd_fiber_new(e, "completion-wins", await_then_yield_fiber, &a, /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "completion-wins", await_then_yield_fiber, &a, &waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_PTR_EQ(sd_fiber_get_awaiting(waiter), target); @@ -2746,7 +2768,7 @@ TEST(fiber_interrupted) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "interrupted", interrupted_fiber, NULL, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "interrupted", interrupted_fiber, NULL, &f)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(f)); } @@ -2789,7 +2811,7 @@ TEST(fiber_timeout_unref_drops_queued_timeout) { ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); TimeoutScope t = { .target = target, .nested = *nested }; - ASSERT_OK(sd_fiber_new(e, "timeout-scope", timeout_scope_fiber, &t, /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "timeout-scope", timeout_scope_fiber, &t, &waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_PTR_EQ(sd_fiber_get_awaiting(waiter), target); @@ -2826,7 +2848,7 @@ TEST(fiber_return_drops_queued_interruption) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); ASSERT_OK(sd_future_new(e, &manual_future_ops, &target)); - ASSERT_OK(sd_fiber_new(e, "return-queued", await_then_return_fiber, target, /* destroy= */ NULL, &waiter)); + ASSERT_OK(sd_fiber_new(e, "return-queued", await_then_return_fiber, target, &waiter)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK(sd_future_resolve(target, 0)); diff --git a/src/libsystemd/sd-future/test-future-group.c b/src/libsystemd/sd-future/test-future-group.c index 0de9d0c75e..a7862ec8ee 100644 --- a/src/libsystemd/sd-future/test-future-group.c +++ b/src/libsystemd/sd-future/test-future-group.c @@ -126,13 +126,16 @@ TEST(future_group_size) { ASSERT_OK(sd_future_group_add(group, second)); ASSERT_EQ(sd_future_group_size(group), 2U); + /* The group releases its children once they resolve, but stays pending until it is sealed. */ ASSERT_OK(sd_event_loop(e)); + ASSERT_EQ(sd_future_group_size(group), 0U); + ASSERT_EQ(sd_future_state(group), SD_FUTURE_PENDING); + + ASSERT_OK(sd_future_group_seal(group)); ASSERT_OK_ZERO(sd_future_result(group)); - ASSERT_EQ(sd_future_group_size(group), 2U); } -/* Distinct errors pin down selection by insertion order when multiple children have settled - * before the group's callbacks run, including when completion order is reversed. */ +/* The child that resolves first selects the error, even though it was added last. */ TEST(future_group_first_error) { _cleanup_(sd_event_unrefp) sd_event *e = NULL; uint64_t policy; @@ -150,12 +153,20 @@ TEST(future_group_first_error) { ASSERT_OK(sd_future_group_new(e, &b)); ASSERT_OK(sd_future_group_add_many(group, a, b)); ASSERT_OK(sd_future_resolve(b, -EIO)); - ASSERT_OK(sd_future_resolve(a, -EINVAL)); + + while (sd_future_group_size(group) > 1) + ASSERT_OK_POSITIVE(sd_event_run(e, 0)); + + /* Without IGNORE_ERRORS, the error of b already resolved the group and cancelled a. */ + if (sd_future_state(a) == SD_FUTURE_PENDING) + ASSERT_OK(sd_future_resolve(a, -EINVAL)); + + ASSERT_OK(sd_future_group_seal(group)); while (sd_future_state(group) == SD_FUTURE_PENDING) ASSERT_OK_POSITIVE(sd_event_run(e, 0)); - ASSERT_ERROR(sd_future_result(group), EINVAL); + ASSERT_ERROR(sd_future_result(group), EIO); } } @@ -174,6 +185,8 @@ TEST(future_group_wait_all_happy) { ASSERT_OK(sd_future_group_add(group, c2)); ASSERT_OK(sd_future_group_add(group, c3)); + ASSERT_OK(sd_future_group_seal(group)); + ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(group)); ASSERT_OK_ZERO(sd_future_result(c1)); @@ -192,8 +205,8 @@ TEST(future_group_wait_all_fail_fast) { 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_fiber_new(e, "sleep-a", suspend_fiber, NULL, &sleeper_a)); + ASSERT_OK(sd_fiber_new(e, "sleep-b", suspend_fiber, NULL, &sleeper_b)); ASSERT_OK(sd_future_group_add(group, errorer)); ASSERT_OK(sd_future_group_add(group, sleeper_a)); @@ -222,6 +235,8 @@ TEST(future_group_wait_all_ignore_errors) { ASSERT_OK(sd_future_group_add(group, errorer)); ASSERT_OK(sd_future_group_add(group, succeeder)); + ASSERT_OK(sd_future_group_seal(group)); + ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(group), EINVAL); ASSERT_ERROR(sd_future_result(errorer), EINVAL); @@ -240,8 +255,8 @@ TEST(future_group_wait_any) { ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY)); ASSERT_OK(sd_future_new_defer(e, 42, &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_fiber_new(e, "medium", suspend_fiber, NULL, &medium)); + ASSERT_OK(sd_fiber_new(e, "slow", suspend_fiber, 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)); @@ -301,7 +316,7 @@ TEST(future_group_first_success) { ASSERT_OK(sd_future_new_defer(e, -EINVAL, &fast_err)); ASSERT_OK(sd_future_new_defer(e, 77, &medium_ok)); - ASSERT_OK(sd_fiber_new(e, "slow-ok", suspend_fiber, NULL, NULL, &slow_ok)); + ASSERT_OK(sd_fiber_new(e, "slow-ok", suspend_fiber, NULL, &slow_ok)); ASSERT_OK(sd_future_group_add(group, fast_err)); ASSERT_OK(sd_future_group_add(group, medium_ok)); @@ -329,6 +344,8 @@ TEST(future_group_first_success_all_fail) { ASSERT_OK(sd_future_group_add(group, err_a)); ASSERT_OK(sd_future_group_add(group, err_b)); + ASSERT_OK(sd_future_group_seal(group)); + ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(sd_future_result(group), EINVAL); ASSERT_ERROR(sd_future_result(err_a), EINVAL); @@ -350,8 +367,8 @@ TEST(future_group_external_cancel) { _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_fiber_new(e, "long-a", suspend_fiber, NULL, &long_a)); + ASSERT_OK(sd_fiber_new(e, "long-b", suspend_fiber, NULL, &long_b)); ASSERT_OK(sd_future_group_add(group, long_a)); ASSERT_OK(sd_future_group_add(group, long_b)); @@ -398,8 +415,8 @@ TEST(future_group_resolves_after_children_drain) { * 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_fiber_new(e, "err", yield_then_return_fiber, &err_result, &errorer)); + ASSERT_OK(sd_fiber_new(e, "sleep", suspend_fiber, NULL, &sleeper)); ASSERT_OK(sd_future_group_add(group, errorer)); ASSERT_OK(sd_future_group_add(group, sleeper)); @@ -437,6 +454,7 @@ static int parent_cancel_driver(void *userdata) { 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)); + ASSERT_OK(sd_future_group_seal(group)); /* 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 queued -ECANCELED takes precedence over this @@ -456,7 +474,7 @@ TEST(future_group_cancels_parent_on_child_error) { 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_fiber_new(e, "parent", parent_cancel_driver, &s, &driver)); ASSERT_OK(sd_event_loop(e)); ASSERT_ERROR(s.suspend_result, ECANCELED); @@ -483,8 +501,7 @@ static int seal_parent_driver(void *userdata) { ASSERT_OK(sd_future_group_new(sd_fiber_get_event(), &group)); ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY)); ASSERT_OK(sd_future_add_callback(group, &wake_slot, wake_parent_cb, sd_fiber_get_current())); - ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "seal", seal_from_peer, group, - /* destroy= */ NULL, &peer)); + ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "seal", seal_from_peer, group, &peer)); s->suspend_result = sd_fiber_suspend(); s->group_result = sd_future_result(group); @@ -501,7 +518,7 @@ TEST(future_group_seal_does_not_cancel_parent) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "parent", seal_parent_driver, &s, /* destroy= */ NULL, &driver)); + ASSERT_OK(sd_fiber_new(e, "parent", seal_parent_driver, &s, &driver)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(driver)); @@ -531,7 +548,7 @@ TEST(future_group_await_returns_real_error) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "parent", await_gets_error_driver, &await_result, /* destroy= */ NULL, &driver)); + ASSERT_OK(sd_fiber_new(e, "parent", await_gets_error_driver, &await_result, &driver)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(driver)); @@ -545,7 +562,7 @@ TEST(future_group_does_not_cancel_parent_with_ignore_errors) { 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_fiber_new(e, "parent", parent_cancel_driver, &s, &driver)); ASSERT_OK(sd_event_loop(e)); /* With IGNORE_ERRORS the parent isn't cancelled — the wake-up callback resumes the @@ -593,7 +610,7 @@ TEST(future_group_add_resolved_child) { 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_fiber_new(e, "driver", add_resolved_driver, &s, &driver)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK(s.add_result); @@ -616,6 +633,7 @@ typedef struct 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)); + ASSERT_OK(sd_future_group_seal(s->group)); ASSERT_OK_ZERO(sd_fiber_await(s->group)); s->join_result = sd_future_result(s->group); return 0; @@ -633,7 +651,7 @@ TEST(future_group_add_many) { 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_fiber_new(e, "driver", add_many_driver, &s, &driver)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(s.join_result); @@ -780,7 +798,7 @@ TEST(future_group_cancel_wait_unref_drives_stubborn_child) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "cancel-wait-group", cancel_wait_group_fiber, &cancels, NULL, &driver)); + ASSERT_OK(sd_fiber_new(e, "cancel-wait-group", cancel_wait_group_fiber, &cancels, &driver)); /* One iteration runs the fiber up to the await inside sd_future_cancel_wait_unref(): the group * is finalizing, but the stubborn child ignored the first cancel. */ @@ -885,6 +903,47 @@ TEST(future_group_set_policy_rejected_after_add) { ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY)), ESTALE); } +TEST(future_group_set_policy_after_failed_add_many) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL, *other = NULL; + _cleanup_(sd_future_cancel_unrefp) sd_future *group = NULL; + _cleanup_(sd_future_cancel_unrefp) sd_future *child = NULL, *foreign = NULL; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_new(&other)); + ASSERT_OK(sd_future_group_new(e, &group)); + + ASSERT_OK(sd_future_new_defer(e, 0, &child)); + ASSERT_OK(sd_future_new_defer(other, 0, &foreign)); + ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_future_group_add_many(group, child, foreign)), EINVAL); + ASSERT_EQ(sd_future_group_size(group), 0U); + + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY)); +} + +TEST(future_group_set_policy_rejected_after_child_resolved) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_cancel_unrefp) sd_future *group = NULL; + _cleanup_(sd_future_unrefp) sd_future *child = NULL; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_future_group_new(e, &group)); + + ASSERT_OK(sd_future_new_defer(e, 0, &child)); + ASSERT_OK(sd_future_group_add(group, child)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_EQ(sd_future_group_size(group), 0U); + ASSERT_EQ(sd_future_state(group), SD_FUTURE_PENDING); + + /* If sd_future_group_set_policy() accepted WAIT_ANY here, sd_future_group_seal() would resolve + * the group with -ECHILD, although the child succeeded. */ + ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY)), ESTALE); + + ASSERT_OK(sd_future_group_seal(group)); + ASSERT_OK_ZERO(sd_future_result(group)); +} + /* 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 @@ -928,7 +987,7 @@ TEST(future_group_does_not_cancel_parent_when_parent_drives_cancel) { 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_fiber_new(e, "self-cancel", self_cancel_fiber, &s, &driver)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(driver)); @@ -980,8 +1039,7 @@ TEST(future_group_parent_await_scope) { ParentAwaitState s = { .await = await, .result = -ECANCELED }; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "parent-await", parent_await_driver, &s, - /* destroy= */ NULL, &parent)); + ASSERT_OK(sd_fiber_new(e, "parent-await", parent_await_driver, &s, &parent)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); if (s.await) ASSERT_OK(sd_future_cancel(parent)); @@ -1022,8 +1080,7 @@ TEST(future_group_external_cancel_leaves_parent_running) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); - ASSERT_OK(sd_fiber_new(e, "parent", parent_await_driver, &s, - /* destroy= */ NULL, &parent)); + ASSERT_OK(sd_fiber_new(e, "parent", parent_await_driver, &s, &parent)); s.parent = parent; ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK(sd_fiber_resume(parent, 42)); @@ -1031,8 +1088,7 @@ TEST(future_group_external_cancel_leaves_parent_running) { ASSERT_OK_POSITIVE(sd_event_run(e, 0)); if (peer) - ASSERT_OK(sd_fiber_new(e, "cancel-group", cancel_group_from_peer, &s, - /* destroy= */ NULL, &canceller)); + ASSERT_OK(sd_fiber_new(e, "cancel-group", cancel_group_from_peer, &s, &canceller)); else ASSERT_OK(cancel_group_from_peer(&s)); @@ -1063,15 +1119,13 @@ TEST(future_group_peer_await_does_not_suppress_parent_cancel) { ParentAwaitState s = { .result = -ECANCELED }; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "parent", parent_await_driver, &s, - /* destroy= */ NULL, &parent)); + ASSERT_OK(sd_fiber_new(e, "parent", parent_await_driver, &s, &parent)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK(sd_fiber_resume(parent, 42)); while (!s.ready) ASSERT_OK_POSITIVE(sd_event_run(e, 0)); - ASSERT_OK(sd_fiber_new(e, "peer-await", peer_await_driver, s.group, - /* destroy= */ NULL, &peer)); + ASSERT_OK(sd_fiber_new(e, "peer-await", peer_await_driver, s.group, &peer)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK(sd_future_resolve(s.child, -EIO)); while (ASSERT_OK(sd_event_run(e, 0)) > 0) @@ -1112,6 +1166,7 @@ TEST(future_group_priority) { ASSERT_EQ(sd_future_state(probe), SD_FUTURE_PENDING); ASSERT_OK(sd_future_resolve(unsupported, 0)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_future_group_seal(group)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(group)); ASSERT_OK_ZERO(sd_future_result(probe)); @@ -1141,6 +1196,7 @@ TEST(future_group_priority_unset) { ASSERT_EQ(sd_future_state(pending), explicit_priority ? SD_FUTURE_PENDING : SD_FUTURE_RESOLVED); ASSERT_EQ(sd_future_state(probe), explicit_priority ? SD_FUTURE_RESOLVED : SD_FUTURE_PENDING); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_future_group_seal(group)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(group)); ASSERT_OK_ZERO(sd_future_result(probe)); @@ -1209,6 +1265,7 @@ TEST(future_group_priority_failure) { ASSERT_OK(sd_future_resolve(failing, 0)); ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_future_group_seal(group)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(group)); ASSERT_OK_ZERO(sd_future_result(later)); @@ -1312,8 +1369,7 @@ TEST(future_group_does_not_capture_cross_event_parent) { ASSERT_OK(sd_event_new(&other)); s.event = other; ASSERT_OK(sd_future_group_new(other, &child)); - ASSERT_OK(sd_fiber_new(e, "cross-event-parent", cross_event_parent_driver, &s, - /* destroy= */ NULL, &parent)); + ASSERT_OK(sd_fiber_new(e, "cross-event-parent", cross_event_parent_driver, &s, &parent)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK(sd_future_group_add(s.group, child)); ASSERT_OK(sd_future_resolve(child, -EIO)); @@ -1357,8 +1413,8 @@ TEST(future_group_outlives_parent) { ParentLifetimeState s = {}; ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_fiber_new(e, "short-lived-parent", parent_returns_before_child, &s, - parent_lifetime_destroy, &parent)); + ASSERT_OK(sd_fiber_new(e, "short-lived-parent", parent_returns_before_child, &s, &parent)); + ASSERT_OK(sd_fiber_set_destroy_callback(parent, parent_lifetime_destroy)); ASSERT_OK_POSITIVE(sd_event_run(e, 0)); ASSERT_OK_ZERO(sd_future_result(parent)); parent = sd_future_unref(parent); @@ -1376,4 +1432,194 @@ TEST(future_group_outlives_parent) { sd_future_unref(s.child); } +TEST(future_group_add_child_twice) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *child = NULL; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_new_defer(e, 0, &child)); + ASSERT_OK(sd_future_group_add(group, child)); + ASSERT_OK(sd_future_group_add(group, child)); + ASSERT_EQ(sd_future_group_size(group), 2U); + ASSERT_OK(sd_future_group_seal(group)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(sd_future_result(group)); + ASSERT_EQ(sd_future_group_size(group), 0U); +} + +typedef struct ReenterState { + sd_future *group; + unsigned n_destroyed; +} ReenterState; + +static int return_zero_fiber(void *userdata) { + return 0; +} + +static void reenter_destroy(void *userdata) { + ReenterState *s = ASSERT_PTR(userdata); + + s->n_destroyed++; + ASSERT_OK(sd_future_group_seal(s->group)); +} + +TEST(future_group_destroy_callback_reenters_group) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_unrefp) sd_future *group = NULL; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_future_group_new(e, &group)); + + ReenterState s = { .group = group }; + + /* The group holds the only reference to each fiber, so releasing a fiber frees it and runs its + * destroy callback. The callback seals the group while the group still has other children. */ + for (unsigned i = 0; i < 3; i++) { + _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL; + + ASSERT_OK(sd_fiber_new(e, "child", return_zero_fiber, &s, &f)); + ASSERT_OK(sd_fiber_set_destroy_callback(f, reenter_destroy)); + ASSERT_OK(sd_future_group_add(group, f)); + f = sd_future_unref(f); + } + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(sd_future_result(group)); + ASSERT_EQ(s.n_destroyed, 3U); +} + +typedef struct SpawnState { + unsigned n_destroyed; +} SpawnState; + +static int failing_fiber(void *userdata) { + return -EIO; +} + +static void spawn_destroy(void *userdata) { + SpawnState *s = ASSERT_PTR(userdata); + + s->n_destroyed++; +} + +static void spawn_with_destroy(sd_future *group, const char *name, sd_fiber_func_t func, SpawnState *s) { + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + + ASSERT_OK(sd_future_group_spawn(group, name, func, s, &f)); + ASSERT_OK(sd_fiber_set_destroy_callback(f, spawn_destroy)); +} + +TEST(future_group_spawn_cancel_drains) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_unrefp) sd_future *group = NULL; + SpawnState s = {}; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_IGNORE_ERRORS)); + for (unsigned i = 0; i < 3; i++) + spawn_with_destroy(group, "sleep", suspend_fiber, &s); + + /* Let every fiber start and suspend, so that cancellation has to wait for the fibers. */ + while (ASSERT_OK(sd_event_run(e, 0)) > 0) + ; + ASSERT_EQ(sd_future_group_size(group), 3U); + + ASSERT_OK(sd_future_cancel(group)); + ASSERT_EQ(sd_future_state(group), SD_FUTURE_PENDING); + + while (ASSERT_OK(sd_event_run(e, 0)) > 0) + ; + ASSERT_ERROR(sd_future_result(group), ECANCELED); + ASSERT_EQ(sd_future_group_size(group), 0U); + ASSERT_EQ(s.n_destroyed, 3U); +} + +static int set_flag_fiber(void *userdata) { + bool *ran = ASSERT_PTR(userdata); + + *ran = true; + return 0; +} + +TEST(future_group_spawn_without_ret) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_unrefp) sd_future *group = NULL; + bool ran = false; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_future_group_new(e, &group)); + + ASSERT_OK(sd_future_group_spawn(group, "flag", set_flag_fiber, &ran, /* ret= */ NULL)); + ASSERT_EQ(sd_future_group_size(group), 1U); + ASSERT_OK(sd_future_group_seal(group)); + ASSERT_EQ(sd_future_state(group), SD_FUTURE_PENDING); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_TRUE(ran); + ASSERT_OK_ZERO(sd_future_result(group)); + ASSERT_EQ(sd_future_group_size(group), 0U); +} + +TEST(future_group_spawn_sealed) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_unrefp) sd_future *group = NULL, *f = NULL; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_future_group_new(e, &group)); + ASSERT_OK(sd_future_group_seal(group)); + + ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_future_group_spawn(group, "late", failing_fiber, NULL, &f)), ESTALE); + ASSERT_NULL(f); + ASSERT_OK_ZERO(sd_future_result(group)); +} + +static int spawner_fiber(void *userdata) { + SpawnState *s = ASSERT_PTR(userdata); + _cleanup_(sd_future_cancel_wait_unrefp) sd_future *group = NULL; + + ASSERT_OK(sd_future_group_new(sd_fiber_get_event(), &group)); + ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_IGNORE_ERRORS)); + + /* Spawn many short-lived children one after another into a group that is never sealed. The + * group releases each failing child, stays pending, and doesn't cancel the spawning fiber. */ + for (unsigned i = 0; i < 100; i++) { + spawn_with_destroy(group, "fail", failing_fiber, s); + ASSERT_EQ(sd_future_group_size(group), 1U); + + for (unsigned n = 0; sd_future_group_size(group) > 0; n++) { + ASSERT_LT(n, 10U); + ASSERT_OK(sd_fiber_yield()); + } + + ASSERT_EQ(sd_future_state(group), SD_FUTURE_PENDING); + } + + /* The cleanup of the group cancels the suspended child and waits until it has finished. */ + spawn_with_destroy(group, "sleep", suspend_fiber, s); + ASSERT_OK(sd_fiber_yield()); + ASSERT_EQ(sd_future_group_size(group), 1U); + + return 0; +} + +TEST(future_group_spawn_from_fiber) { + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + _cleanup_(sd_future_unrefp) sd_future *spawner = NULL; + SpawnState s = {}; + + ASSERT_OK(sd_event_new(&e)); + ASSERT_OK(sd_event_set_exit_on_idle(e, true)); + ASSERT_OK(sd_fiber_new(e, "spawner", spawner_fiber, &s, &spawner)); + + ASSERT_OK(sd_event_loop(e)); + ASSERT_OK_ZERO(sd_future_result(spawner)); + ASSERT_EQ(s.n_destroyed, 101U); +} + DEFINE_TEST_MAIN(LOG_DEBUG); diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index 584b4c85fd..9ee41be142 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -1165,7 +1165,11 @@ static int varlink_dispatch_fiber(sd_varlink *v, const char *method, sd_varlink_ }; _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); + r = sd_fiber_new(v->server->event, method, varlink_fiber_entry, d, &f); + if (r < 0) + return r; + + r = sd_fiber_set_destroy_callback(f, varlink_fiber_data_destroy); if (r < 0) return r; diff --git a/src/libsystemd/sd-varlink/test-varlink.c b/src/libsystemd/sd-varlink/test-varlink.c index 169a5f372b..6b11fc80b7 100644 --- a/src/libsystemd/sd-varlink/test-varlink.c +++ b/src/libsystemd/sd-varlink/test-varlink.c @@ -395,7 +395,7 @@ TEST(chat) { ASSERT_OK(sd_varlink_attach_event(c, e, 0)); - ASSERT_OK(sd_fiber_new(e, "client", client_fiber, (void*) sp, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "client", client_fiber, (void*) sp, &f)); ASSERT_OK(sd_event_loop(e)); @@ -1272,7 +1272,7 @@ TEST(upgrade) { ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600)); ASSERT_OK(sd_varlink_server_attach_event(s, e, 0)); - ASSERT_OK(sd_fiber_new(e, "upgrade-client", upgrade_client_fiber, (void*) sp, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "upgrade-client", upgrade_client_fiber, (void*) sp, &f)); /* Run the event loop. Exits on idle once the client fiber completes and all server connections * have been torn down. */ @@ -1347,7 +1347,7 @@ TEST(upgrade_pipelining) { ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600)); ASSERT_OK(sd_varlink_server_attach_event(s, e, 0)); - ASSERT_OK(sd_fiber_new(e, "upgrade-pipelining-client", upgrade_pipelining_client_fiber, (void*) sp, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "upgrade-pipelining-client", upgrade_pipelining_client_fiber, (void*) sp, &f)); ASSERT_OK(sd_event_loop(e)); @@ -1387,7 +1387,7 @@ static int respond_upgrade_callback(sd_varlink *link, int input_fd, int output_f respond_upgrade_input_fd = input_fd; respond_upgrade_output_fd = output_fd; - ASSERT_OK(sd_fiber_new(e, "respond-upgrade-io", respond_upgrade_io_fiber, /* userdata= */ NULL, /* destroy= */ NULL, /* ret= */ NULL)); + ASSERT_OK(sd_fiber_new(e, "respond-upgrade-io", respond_upgrade_io_fiber, /* userdata= */ NULL, /* ret= */ NULL)); return 0; } @@ -1468,7 +1468,7 @@ TEST(respond_upgrade) { ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600)); ASSERT_OK(sd_varlink_server_attach_event(s, e, 0)); - ASSERT_OK(sd_fiber_new(e, "respond-upgrade-client", respond_upgrade_client_fiber, (void*) sp, /* destroy= */ NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "respond-upgrade-client", respond_upgrade_client_fiber, (void*) sp, &f)); ASSERT_OK(sd_event_loop(e)); @@ -1577,7 +1577,7 @@ TEST(execute_directory) { .n_servers = ELEMENTSOF(names), .reply_count = &reply_count, }; - ASSERT_OK(sd_fiber_new(e, "execute-dir-client", execute_dir_client_fiber, &args, NULL, &f)); + ASSERT_OK(sd_fiber_new(e, "execute-dir-client", execute_dir_client_fiber, &args, &f)); ASSERT_OK(sd_event_loop(e)); diff --git a/src/shared/main-func.c b/src/shared/main-func.c index a90fb458bd..6d749fb184 100644 --- a/src/shared/main-func.c +++ b/src/shared/main-func.c @@ -85,8 +85,7 @@ int run_main_fiber(int argc, char *argv[], main_fiber_func_t func) { /* Fire-and-forget: the loop unwinds the fiber through its exit event source, and a suspended * fiber is one we couldn't drop a reference to anyway. */ - r = sd_fiber_new(event, program_invocation_short_name, main_fiber_trampoline, &ctx, - /* destroy= */ NULL, /* ret= */ NULL); + r = sd_fiber_new(event, program_invocation_short_name, main_fiber_trampoline, &ctx, /* ret= */ NULL); if (r < 0) return log_error_errno(r, "Failed to spawn main fiber: %m"); diff --git a/src/systemd/sd-future.h b/src/systemd/sd-future.h index 340165a8fb..d6b3058ef2 100644 --- a/src/systemd/sd-future.h +++ b/src/systemd/sd-future.h @@ -100,17 +100,24 @@ sd_future* sd_future_slot_get_future(sd_future_slot *s); int sd_future_set_priority(sd_future *f, int64_t priority); -/* Group policies select an outcome as follows: - * WAIT_ALL: first child error, or 0 once every child succeeds. - * WAIT_ALL | IGNORE_ERRORS: wait for every child, then return the first error or 0. - * WAIT_ANY: a successful child's result if any have succeeded, otherwise the first child error. - * WAIT_ANY | IGNORE_ERRORS: wait for a success, or for every child to fail; return that success or - * the first error, respectively. - * "First" means insertion order among the children already resolved when the outcome is selected. - * Once selected, the outcome is final: remaining children are cancelled, and the group resolves only - * after every child has finished. IGNORE_ERRORS also suppresses cancellation of the parent fiber; - * it does not turn an unsuccessful group result into success. An empty group stays pending until it - * is sealed or cancelled. */ +/* The policy of a group selects the outcome of the group from the results of its children. + * With WAIT_ALL, the first child error is the outcome. If every child succeeds, the outcome is 0. + * With WAIT_ALL | IGNORE_ERRORS, the group waits for every child. The outcome is the first child + * error, or 0 if every child succeeded. + * With WAIT_ANY, the result of the first successful child is the outcome. If a child fails first, its + * error is the outcome. + * With WAIT_ANY | IGNORE_ERRORS, the result of the first successful child is the outcome. If every + * child fails, the first child error is the outcome. + * Waiting for every child also waits for sd_future_group_seal(). The owner can add more children until + * it seals the group. A group that is not sealed therefore stays pending when it has no pending + * children. + * The group releases its reference to a child once it has handled the resolution of the child. + * "First" refers to the order in which the group handles resolutions. The order is not specified for + * children that resolve in the same event loop iteration. Under WAIT_ANY, a success is still the + * outcome if the successful child resolved before the group handled an error. + * The first outcome that the policy selects is final. The group then cancels its remaining children + * and resolves after every child has finished. IGNORE_ERRORS also prevents a child error from + * cancelling the parent fiber. */ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_future_group_policy_t) { SD_FUTURE_GROUP_WAIT_ALL = 0, SD_FUTURE_GROUP_WAIT_ANY = 1 << 0, @@ -119,24 +126,34 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_future_group_policy_t) { _SD_ENUM_FORCE_S64(SD_FUTURE_GROUP_POLICY) } sd_future_group_policy_t; -/* The calling fiber becomes the parent only if it belongs to e. Unless IGNORE_ERRORS is set, a child - * error cancels the parent if it is not awaiting the group with sd_fiber_await(). Explicit group - * cancellation does not cancel the parent. */ +/* If the calling fiber runs on e, it becomes the parent of the group. Unless IGNORE_ERRORS is set, a + * child error cancels the parent. The parent is not cancelled while it awaits the group with + * sd_fiber_await(). Cancelling the group does not cancel the parent. */ int sd_future_group_new(sd_event *e, sd_future **ret); int sd_future_group_set_policy(sd_future *f, uint64_t policy); -/* The group and all its children must belong to the same event loop. Adding a child takes a new - * reference; it does not consume the caller's reference. To hand ownership to the group, release the - * caller's reference with sd_future_unref(), not a cancellation cleanup helper. */ +/* The group and all its children must belong to the same event loop. sd_future_group_add() takes a + * new reference to the child and does not drop the reference of the caller. To give the group the only + * reference, drop the reference of the caller with sd_future_unref(). A cleanup helper that cancels the + * child would stop the child instead. The group can drop the last reference to a child before other + * callbacks on the child have run. A floating callback on the child then never runs. Keep a reference + * to the child to read its result or to rely on its other callbacks. */ int sd_future_group_add(sd_future *f, sd_future *child); int sd_future_group_add_many_internal(sd_future *f, ...) _sd_sentinel_; #define sd_future_group_add_many(f, ...) sd_future_group_add_many_internal(f, __VA_ARGS__, NULL) -/* After sealing, adding a child fails with -ESTALE. Sealing an empty group resolves it with 0 under - * WAIT_ALL and with -ECHILD under WAIT_ANY. An -ECHILD result does not cancel the parent fiber. */ +/* After sealing, adding a child fails with -ESTALE. A sealed group without pending children resolves + * right away. If the group never had a child, it resolves with 0 under WAIT_ALL and with -ECHILD + * under WAIT_ANY. An -ECHILD result does not cancel the parent fiber. */ int sd_future_group_seal(sd_future *f); -/* NULL is treated as an empty group. */ +/* NULL is treated as an empty group. The size does not include children that the group released. */ size_t sd_future_group_size(sd_future *f); +/* sd_future_group_spawn() creates a fiber on the event loop of the group and adds it to the group. If + * ret is not NULL, it receives a new reference to the fiber, for example to set a destroy callback with + * sd_fiber_set_destroy_callback(). Drop that reference with sd_future_unref(). */ +int sd_future_group_spawn(sd_future *g, const char *name, sd_fiber_func_t func, void *userdata, sd_future **ret); -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); +int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *userdata, sd_future **ret); +int sd_fiber_set_destroy_callback(sd_future *f, sd_fiber_destroy_t callback); +int sd_fiber_get_destroy_callback(sd_future *f, sd_fiber_destroy_t *ret); int sd_fiber_set_floating(sd_future *f, int b); int sd_fiber_get_floating(sd_future *f); diff --git a/src/test/test-qmp-client.c b/src/test/test-qmp-client.c index 4b361a2e8b..af88b6896c 100644 --- a/src/test/test-qmp-client.c +++ b/src/test/test-qmp-client.c @@ -146,8 +146,8 @@ static void run_qmp_test(sd_fiber_func_t mock_fn, sd_fiber_func_t client_fn) { ASSERT_OK(qmp_client_connect_fd(&client, TAKE_FD(qmp_fds[0]))); ASSERT_OK(qmp_client_attach_event(client, event, SD_EVENT_PRIORITY_NORMAL)); - ASSERT_OK(sd_fiber_new(event, "mock", mock_fn, FD_TO_PTR(TAKE_FD(qmp_fds[1])), NULL, &mock_f)); - ASSERT_OK(sd_fiber_new(event, "client", client_fn, client, NULL, &client_f)); + ASSERT_OK(sd_fiber_new(event, "mock", mock_fn, FD_TO_PTR(TAKE_FD(qmp_fds[1])), &mock_f)); + ASSERT_OK(sd_fiber_new(event, "client", client_fn, client, &client_f)); ASSERT_OK(sd_event_loop(event)); ASSERT_OK(sd_future_result(client_f)); @@ -536,7 +536,7 @@ static int qmp_client_call_interrupted_fiber(void *userdata) { ASSERT_OK(qmp_client_attach_event(client, sd_fiber_get_event(), SD_EVENT_PRIORITY_NORMAL)); QmpInterruptedCall c = { .fd = TAKE_FD(fds[1]), .caller = sd_fiber_get_current(), .error = error }; ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "interrupted-mock", mock_qmp_call_interrupted_fiber, - &c, /* destroy= */ NULL, &mock)); + &c, &mock)); /* Finish the handshake before testing interruption of a command awaiting its reply. */ ASSERT_OK_POSITIVE(qmp_client_call(client, "query-status", /* args= */ NULL, @@ -567,7 +567,7 @@ TEST(qmp_call_interrupted) { ASSERT_OK(sd_event_new(&e)); ASSERT_OK(sd_fiber_new(e, "interrupted-client", qmp_client_call_interrupted_fiber, - &error, /* destroy= */ NULL, &caller)); + &error, &caller)); ASSERT_OK(sd_event_loop(e)); ASSERT_OK_ZERO(sd_future_result(caller)); } diff --git a/src/update-helper/update-helper.c b/src/update-helper/update-helper.c index 8f1a4071b8..961d35e476 100644 --- a/src/update-helper/update-helper.c +++ b/src/update-helper/update-helper.c @@ -420,10 +420,14 @@ static int user_units_operation(char **users, UserUnitOperationFunc func, const }; _cleanup_(sd_future_cancel_wait_unrefp) sd_future *f = NULL; - r = sd_fiber_new(sd_fiber_get_event(), *user, user_unit_operation_fiber, o, free, &f); + r = sd_fiber_new(sd_fiber_get_event(), *user, user_unit_operation_fiber, o, &f); if (r < 0) return log_error_errno(r, "Failed to create new fiber for '%s': %m", *user); + r = sd_fiber_set_destroy_callback(f, free); + if (r < 0) + return log_error_errno(r, "Failed to set destroy callback of fiber for '%s': %m", *user); + TAKE_PTR(o); r = sd_future_group_add(g, f);