starred/liberated-systemd
1
0 Fork 0
mirror of https://github.com/Jeffrey-Sardina/liberated-systemd.git synced 2026-10-19 01:29:53 -05:00
Code Issues Packages Projects Releases Wiki Activity

sd-future: Various cleanups and API improvements (#44053)

Browse Source
This commit is contained in:
Daan De Meyer authored and GitHub committed 2026-10-06 14:54:45 +02:00
2 parents 4f8ca5032f dc0f758a14
commit ef4e8a3db2
22 files changed
+725 -393
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/libsystemd/sd-bus/bus-objects.c
+5 -1
View File
@@ -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);
src/libsystemd/sd-bus/test-bus-chat.c
+3 -3
View File
@@ -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));
src/libsystemd/sd-bus/test-bus-fiber.c
+7 -10
View File
@@ -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));
}
src/libsystemd/sd-bus/test-bus-objects.c
+2 -2
View File
@@ -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));
src/libsystemd/sd-bus/test-bus-peersockaddr.c
+2 -2
View File
@@ -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));
src/libsystemd/sd-bus/test-bus-server.c
+2 -2
View File
@@ -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));
src/libsystemd/sd-bus/test-bus-watch-bind.c
+4 -4
View File
@@ -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));
src/libsystemd/sd-event/event-future.c
+4
View File
@@ -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;
src/libsystemd/sd-event/test-event-future.c
+13 -12
View File
@@ -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, &registrar));
&d, &registrar));
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)
src/libsystemd/sd-future/fiber.c
+20 -5
View File
@@ -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);
src/libsystemd/sd-future/future-group.c
+136 -115
View File
@@ -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);
}
src/libsystemd/sd-future/test-channel.c
+9 -9
View File
@@ -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));
src/libsystemd/sd-future/test-fiber-io.c
+37 -39
View File
@@ -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));
src/libsystemd/sd-future/test-fiber-ops.c
+15 -15
View File
@@ -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));
src/libsystemd/sd-future/test-fiber.c
+123 -101
View File
@@ -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, &registrar));
ASSERT_OK(sd_fiber_new(e, "register-slot", slot_priority_register, &s, &registrar));
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));
src/libsystemd/sd-future/test-future-group.c
+284 -38
View File
@@ -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);
src/libsystemd/sd-varlink/sd-varlink.c
+5 -1
View File
@@ -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;
src/libsystemd/sd-varlink/test-varlink.c
+6 -6
View File
@@ -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));
src/shared/main-func.c
+1 -2
View File
@@ -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");
src/systemd/sd-future.h
+38 -21
View File
@@ -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);
src/test/test-qmp-client.c
+4 -4
View File
@@ -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));
}
src/update-helper/update-helper.c
+5 -1
View File
@@ -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);