Introduce support for running code in fibers (#39771)
Traditionally, asynchronous programming in systemd has been achieved using sd-event along with the asynchronous interfaces of sd-bus and sd-varlink. This works well when the system is reacting to events and all code triggered by those events can run without blocking. In these scenarios, the global Manager object is passed as userdata to the callback, and the callback can use the stack as usual, declaring local state and ensuring proper cleanup via _cleanup_. Control flow structures, such as loops, work as expected, and everything runs smoothly. However, challenges arise when the code needs to perform long-running operations within these callbacks. Since the system cannot block execution within the callback, we can't directly invoke a long-running operation and wait for its result without introducing complexities. Instead, we need to initiate the long-running task, register for completion with sd-event, sd-bus, or sd-varlink, and provide a callback to be invoked when the operation completes. This callback, however, only receives a single userdata pointer, which forces us to bundle all local variables into a struct and pass it along as part of the callback. On top of that, after queuing the asynchronous operation, the caller continues executing. As the caller's stack unwinds when the function exits, the resources and state within the local scope may be prematurely cleaned up. Therefore, the struct must store copies of the local variables or ensure proper reference counting to prevent premature resource cleanup. When multiple long-running operations need to be initiated within a loop, the complexity grows further. We must introduce additional shared state to track the completion of all operations before we can run any code that depends on their results. Furthermore, since the daemon may be shut down at any time, we must track the lifecycle of each long-running operation in the global Manager struct, ensuring proper cleanup even when stack unwinding can no longer manage the resources for us. Fibers, or green threads, provide a more natural way of handling asynchronous operations. By enabling cooperative multitasking within a single thread, fibers allow us to write code that looks like it’s running synchronously, but with the ability to yield control at predefined points, such as when waiting for long-running tasks to complete. With fibers, we can simplify the control flow by running asynchronous operations within a fiber, allowing us to "pause" execution while waiting for the long-running operation to finish and then "resume" the operation once it's complete. This eliminates the need for multiple callback chains, extensive state tracking, and the potential pitfalls of stack unwinding. This commit introduces the ability to execute long-running operations in a non-blocking manner while maintaining the simplicity and readability of synchronous code. The fiber-based approach will significantly improve the handling of complex workflows, making the code easier to write and maintain. The implementation is based on ucontext.h's makecontext() (with a fallback to the venerable sigaltstack() approach on musl), sigsetjmp()/siglongjmp() and sd-event. ucontext.h provides us with alternate stacks that we can switch between. We use sigsetjmp()/siglongjmp() instead of swapcontext() because the latter forcibly saves/restores a per context signal mask every time it is called. Using sigsetjmp()/siglongjmp(), we can avoid the unnecessary syscall and maintain a per thread signal mask, which makes much more sense than having a per fiber signal mask. The default stack size is the same as a regular thread. Because we use mmap() to allocate the stack, the memory won't actually be used until it is paged in by the kernel, so we don't actually use 8MB per fiber. To integrate fibers with the event loop, each fiber is assigned a deferred event source which resumes the fiber when enabled. The deferred event source is oneshot by default so the fiber will run immediately until it yields or suspends. If it yields, the deferred event source is enabled again (oneshot) immediately. If it suspends, before it suspends, one or more event sources are registered with sd-event that will enable the deferred event source (oneshot) to resume the fiber once the operation it is waiting for completes. Yielding or suspending the fiber is done by calling sd_fiber_yield() or sd_fiber_suspend() respectively. Both of these return zero on success or any error value from the async operation that caused the fiber to resume. This is also how fiber cancellation is implemented. When a fiber is cancelled, sd_fiber_yield() and sd_fiber_suspend() will return ECANCELED when the fiber is resumed, allowing the fiber to unwind its stack (which allows cleanup to happen automatically) and finish. Instead of having applications work directly with fibers, we hide them behind a generic futures interface to represent long-running operations, regardless of whether those operations are running on a fiber or not. Aside from fibers, the futures library (sd-future) will for example allow waiting for sd-event sources and doing sd-bus calls in the background as well. Fibers can suspend until a future is ready with sd_fiber_await() or by having the future wake up the fiber explicitly in its callback. A future always defaults to waking up the current fiber. Each future kind plugs into the library by providing an sd_future_ops vtable (alloc, free, cancel, set_priority). The library treats the impl pointer returned by alloc() as a black box. Future Implementations retrieve it via sd_future_get_private(). A future starts in SD_FUTURE_PENDING and transitions exactly once to SD_FUTURE_RESOLVED, carrying an integer result. Consumers can react to that transition either by installing a one-shot callback with sd_future_set_callback() (callback-style code) or by waiting on it from a fiber via sd_fiber_await() (synchronous-looking fiber code). sd_fiber_await() is itself built on a "wait future" that resolves when its target resolves; sd_future_new_wait() exposes the same primitive directly so non-fiber callers can chain futures without involving a fiber. Cancellation is cooperative: sd_future_cancel() invokes the future impl's cancel callback, which is responsible for tearing down its work and ultimately resolving the promise with -ECANCELED. For fiber futures this is what surfaces as the ECANCELED return from sd_fiber_yield()/sd_fiber_suspend() mentioned above. Fire-and-forget fibers — created by passing a NULL ret to sd_fiber_new() — take a self-reference on their future so they outlive the caller's scope. The self-ref is dropped when the fiber resolves. This floating mechanism (sd_fiber_set_floating()) is restricted to fiber futures because they uniquely guarantee resolution; allowing it for arbitrary future kinds would risk silent leaks for kinds that may never resolve. Note that fiber cleanup depends on the runtime operating normally. Each fiber's _cleanup_-style cleanups live on the fiber's own stack and run only when the fiber is resumed and allowed to unwind, which requires a working event loop to drive it to completion. The exit event source registered for top-level fibers ensures unwind on a normal sd_event_exit(), but if the event loop itself terminates abnormally (e.g. an unrecoverable allocation failure mid-dispatch) before all fibers have resolved, their stacks never unwind and any resources they own leak. The code lives in libsystemd as sd-future (not exported) for the following reasons: - We may want to make this a public libsystemd API in the future - The code can't live in src/basic as it makes heavy use of sd-event - The code can't live in src/shared as sd-bus and sd-event make use of it The log and log-context headers are updated with functions to allow fibers to have their own log prefix and log context.
This commit is contained in:
+210
-23
@@ -1,6 +1,7 @@
|
||||
/* SPDX-License-Identifier: LGPL-2.1-or-later */
|
||||
|
||||
#include "sd-event.h"
|
||||
#include "sd-future.h"
|
||||
#include "sd-json.h"
|
||||
|
||||
#include "alloc-util.h"
|
||||
@@ -226,19 +227,23 @@ static int qmp_extract_response_id(sd_json_variant *v, uint64_t *ret) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* Returns 0 on success (ret_result = "return" value), -EIO on QMP error (reterr_desc set). */
|
||||
static int qmp_parse_response(sd_json_variant *v, sd_json_variant **ret_result, const char **reterr_desc) {
|
||||
/* Returns 0 on success (ret_result = freshly reffed "return" value), -EIO on QMP error
|
||||
* (ret_error_desc set to a freshly allocated string). Caller owns both outputs. */
|
||||
static int qmp_parse_response(sd_json_variant *v, sd_json_variant **ret_result, char **reterr_error_desc) {
|
||||
const char *desc;
|
||||
|
||||
desc = qmp_extract_error_description(v);
|
||||
if (desc) {
|
||||
if (reterr_desc)
|
||||
*reterr_desc = desc;
|
||||
if (reterr_error_desc) {
|
||||
*reterr_error_desc = strdup(desc);
|
||||
if (!*reterr_error_desc)
|
||||
return -ENOMEM;
|
||||
}
|
||||
return -EIO;
|
||||
}
|
||||
|
||||
if (ret_result)
|
||||
*ret_result = sd_json_variant_by_key(v, "return");
|
||||
*ret_result = sd_json_variant_ref(sd_json_variant_by_key(v, "return"));
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -273,8 +278,8 @@ static int qmp_client_build_command(
|
||||
|
||||
/* Route c->current to event callback or matching async slot. Returns 1 on dispatch. */
|
||||
static int qmp_client_dispatch(QmpClient *c) {
|
||||
sd_json_variant *result = NULL;
|
||||
const char *desc = NULL;
|
||||
_cleanup_(sd_json_variant_unrefp) sd_json_variant *result = NULL;
|
||||
_cleanup_free_ char *desc = NULL;
|
||||
uint64_t id;
|
||||
int error, r;
|
||||
|
||||
@@ -318,8 +323,8 @@ static int qmp_client_dispatch(QmpClient *c) {
|
||||
}
|
||||
|
||||
/* Synchronous slot (no callback): leave c->current pinned so qmp_client_call() can
|
||||
* pick up the reply and hand out borrowed pointers into it. The sync caller owns a
|
||||
* ref on the slot and detects completion by observing slot->client turning NULL. */
|
||||
* pick the reply up after its pump loop. The sync caller owns a ref on the slot and
|
||||
* detects completion by observing slot->client turning NULL. */
|
||||
if (!slot->callback) {
|
||||
qmp_slot_disconnect(slot, /* unref= */ true);
|
||||
return 1;
|
||||
@@ -574,6 +579,10 @@ static void qmp_client_clear(QmpClient *c) {
|
||||
qmp_client_detach_event(c);
|
||||
qmp_client_clear_current(c);
|
||||
json_stream_done(&c->stream);
|
||||
/* qmp_client_handle_disconnect() above drained every entry via qmp_client_fail_pending();
|
||||
* the set is borrow-only for non-floating slots, so set_free() can't safely run a
|
||||
* destructor over leftovers — enforce the drain invariant instead. */
|
||||
assert(set_isempty(c->slots));
|
||||
c->slots = set_free(c->slots);
|
||||
}
|
||||
|
||||
@@ -745,7 +754,7 @@ DEFINE_TRIVIAL_CLEANUP_FUNC(QmpClientArgs*, qmp_client_args_close_fds);
|
||||
|
||||
/* Shared send path for qmp_client_invoke() and qmp_client_call(). A NULL callback registers
|
||||
* a "synchronous" slot: dispatch_reply leaves c->current pinned on match instead of invoking
|
||||
* a callback, so qmp_client_call() can hand out borrowed pointers into the reply. If ret_slot
|
||||
* a callback, so qmp_client_call() can pick the reply up after its pump loop. If ret_slot
|
||||
* is NULL the slot is allocated as floating (owned by c->slots); otherwise a reference is
|
||||
* handed back to the caller. */
|
||||
static int qmp_client_send(
|
||||
@@ -810,21 +819,193 @@ int qmp_client_invoke(
|
||||
return qmp_client_send(c, command, args, callback, userdata, ret_slot);
|
||||
}
|
||||
|
||||
typedef struct QmpFuture {
|
||||
QmpSlot *slot; /* owned, non-floating; NULL once disconnected */
|
||||
sd_json_variant *result;
|
||||
char *error_desc;
|
||||
} QmpFuture;
|
||||
|
||||
static void* qmp_future_alloc(void) {
|
||||
return new0(QmpFuture, 1);
|
||||
}
|
||||
|
||||
static void qmp_future_free(sd_future *f) {
|
||||
QmpFuture *qf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
qmp_slot_unref(qf->slot);
|
||||
sd_json_variant_unref(qf->result);
|
||||
free(qf->error_desc);
|
||||
free(qf);
|
||||
}
|
||||
|
||||
static int qmp_future_cancel(sd_future *f) {
|
||||
QmpFuture *qf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
|
||||
/* Drop the pending slot so dispatch_reply won't try to fire our callback (and touch
|
||||
* freed memory) when the reply eventually arrives. */
|
||||
qf->slot = qmp_slot_unref(qf->slot);
|
||||
return sd_future_resolve(f, -ECANCELED);
|
||||
}
|
||||
|
||||
static const sd_future_ops qmp_call_future_ops = {
|
||||
.size = sizeof(sd_future_ops),
|
||||
.alloc = qmp_future_alloc,
|
||||
.free = qmp_future_free,
|
||||
.cancel = qmp_future_cancel,
|
||||
};
|
||||
|
||||
static int qmp_future_callback(
|
||||
QmpClient *c,
|
||||
sd_json_variant *result,
|
||||
const char *desc,
|
||||
int error,
|
||||
void *userdata) {
|
||||
|
||||
sd_future *f = ASSERT_PTR(userdata);
|
||||
QmpFuture *qf = ASSERT_PTR(sd_future_get_private(f));
|
||||
int r;
|
||||
|
||||
assert(result || desc || error);
|
||||
|
||||
if (result)
|
||||
qf->result = sd_json_variant_ref(result);
|
||||
if (desc) {
|
||||
qf->error_desc = strdup(desc);
|
||||
if (!qf->error_desc)
|
||||
/* No usable reply payload to surface — propagate as transport-style
|
||||
* failure so suspend() / sd_future_result() see the OOM. */
|
||||
return sd_future_resolve(f, -ENOMEM);
|
||||
}
|
||||
|
||||
/* Resolve with 0 on a success reply and -EIO on a QMP-level error (matching the synchronous
|
||||
* path's errno for the desc-without-ret_error_desc case), so a caller awaiting the future
|
||||
* learns about call failures from the resolution value alone. The reply payload itself
|
||||
* (result or error_desc) is always stashed on the QmpFuture so future_get_qmp_reply() can
|
||||
* hand the description string back on top of the bare -EIO. With no reply at all (transport
|
||||
* failure, disconnect), resolve with the propagated transport errno; cancellation surfaces
|
||||
* as -ECANCELED via qmp_future_cancel(). */
|
||||
if (result)
|
||||
r = 0;
|
||||
else if (desc)
|
||||
r = -EIO;
|
||||
else
|
||||
r = error;
|
||||
|
||||
return sd_future_resolve(f, r);
|
||||
}
|
||||
|
||||
int qmp_client_call_future(
|
||||
QmpClient *c,
|
||||
const char *command,
|
||||
QmpClientArgs *args,
|
||||
sd_future **ret) {
|
||||
|
||||
int r;
|
||||
|
||||
assert(c);
|
||||
assert(command);
|
||||
assert(ret);
|
||||
|
||||
_cleanup_(sd_future_unrefp) sd_future *f = NULL;
|
||||
r = sd_future_new(&qmp_call_future_ops, &f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
QmpFuture *qf = sd_future_get_private(f);
|
||||
|
||||
r = qmp_client_send(c, command, args, qmp_future_callback, f, &qf->slot);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
*ret = TAKE_PTR(f);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Extract the reply from a resolved qmp_client_call_future(). Returns 1 on success (with
|
||||
* *ret_result a fresh reference the caller unrefs), -EIO on a QMP-level error (with the detail
|
||||
* description copied into *ret_error_desc when the caller passed one to receive it), and the
|
||||
* future's negative resume errno when no reply landed at all (transport failure / cancellation).
|
||||
*/
|
||||
int future_get_qmp_reply(sd_future *f, sd_json_variant **ret_result, char **reterr_error_desc) {
|
||||
assert(f);
|
||||
assert(sd_future_get_ops(f) == &qmp_call_future_ops);
|
||||
assert(sd_future_state(f) == SD_FUTURE_RESOLVED);
|
||||
|
||||
QmpFuture *qf = ASSERT_PTR(sd_future_get_private(f));
|
||||
|
||||
/* No reply at all: transport failure or cancellation — surface the future result. */
|
||||
if (!qf->result && !qf->error_desc)
|
||||
return sd_future_result(f);
|
||||
|
||||
if (qf->error_desc) {
|
||||
if (reterr_error_desc) {
|
||||
char *desc = strdup(qf->error_desc);
|
||||
if (!desc)
|
||||
return -ENOMEM;
|
||||
*reterr_error_desc = desc;
|
||||
}
|
||||
return -EIO;
|
||||
}
|
||||
|
||||
if (reterr_error_desc)
|
||||
*reterr_error_desc = NULL;
|
||||
if (ret_result)
|
||||
*ret_result = sd_json_variant_ref(qf->result);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int qmp_client_call_suspend(
|
||||
QmpClient *c,
|
||||
const char *command,
|
||||
QmpClientArgs *args,
|
||||
sd_json_variant **ret_result,
|
||||
char **ret_error_desc) {
|
||||
|
||||
int r;
|
||||
|
||||
assert(c);
|
||||
assert(command);
|
||||
assert(sd_fiber_is_running());
|
||||
|
||||
_cleanup_(sd_future_cancel_wait_unrefp) sd_future *call = NULL;
|
||||
r = qmp_client_call_future(c, command, args, &call);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
r = sd_fiber_suspend();
|
||||
|
||||
/* If the future isn't resolved, the suspend was interrupted before a reply arrived (fiber
|
||||
* cancelled, fiber-wide SD_FIBER_TIMEOUT scope expired, …). There's no reply to extract,
|
||||
* so surface the resume error directly. When the future is resolved, future_get_qmp_reply()
|
||||
* already encodes success (1), QMP-level error (-EIO with the desc captured if asked for),
|
||||
* and no-reply (negative future result) — pass it through. */
|
||||
if (sd_future_state(call) != SD_FUTURE_RESOLVED)
|
||||
return r;
|
||||
|
||||
return future_get_qmp_reply(call, ret_result, ret_error_desc);
|
||||
}
|
||||
|
||||
int qmp_client_call(
|
||||
QmpClient *c,
|
||||
const char *command,
|
||||
QmpClientArgs *args,
|
||||
sd_json_variant **ret_result,
|
||||
const char **ret_error_desc) {
|
||||
char **reterr_error_desc) {
|
||||
|
||||
_cleanup_(qmp_slot_unrefp) QmpSlot *slot = NULL;
|
||||
_cleanup_(sd_json_variant_unrefp) sd_json_variant *result = NULL;
|
||||
_cleanup_free_ char *desc = NULL;
|
||||
int r;
|
||||
|
||||
assert_return(c, -EINVAL);
|
||||
assert_return(command, -EINVAL);
|
||||
|
||||
/* Drop any reply pinned by a previous qmp_client_call() before we pin a new one. */
|
||||
qmp_client_clear_current(c);
|
||||
/* If we're on a fiber sharing the QMP client's event loop, use the async + suspend path so
|
||||
* multiple concurrent qmp_client_call() invocations across fibers don't deadlock each other
|
||||
* on the process+wait pump. */
|
||||
if (sd_fiber_is_running() && qmp_client_get_event(c) == sd_fiber_get_event())
|
||||
return qmp_client_call_suspend(c, command, args, ret_result, reterr_error_desc);
|
||||
|
||||
_cleanup_(qmp_slot_unrefp) QmpSlot *slot = NULL;
|
||||
|
||||
/* NULL callback marks this as a synchronous slot: dispatch_reply matches on id like
|
||||
* any other slot (so stray unknown-id replies still get logged and dropped), but
|
||||
@@ -855,18 +1036,24 @@ int qmp_client_call(
|
||||
return r;
|
||||
}
|
||||
|
||||
sd_json_variant *result = NULL;
|
||||
const char *desc = NULL;
|
||||
int error = qmp_parse_response(c->current, &result, &desc);
|
||||
_cleanup_(sd_json_variant_unrefp) sd_json_variant *current = TAKE_PTR(c->current);
|
||||
r = qmp_parse_response(current, &result, &desc);
|
||||
if (r < 0 && r != -EIO)
|
||||
return r;
|
||||
|
||||
/* If caller doesn't ask for the error string, surface the error as the return code. */
|
||||
if (!ret_error_desc && error < 0)
|
||||
return error;
|
||||
/* QMP-level error: copy the description into *ret_error_desc when the caller asked for it,
|
||||
* and surface the failure via the return value (matching qmp_client_call_suspend() /
|
||||
* sd_bus_call()'s "negative on error" convention). */
|
||||
if (desc) {
|
||||
if (reterr_error_desc)
|
||||
*reterr_error_desc = TAKE_PTR(desc);
|
||||
return -EIO;
|
||||
}
|
||||
|
||||
if (ret_result)
|
||||
*ret_result = result;
|
||||
if (ret_error_desc)
|
||||
*ret_error_desc = desc;
|
||||
*ret_result = TAKE_PTR(result);
|
||||
if (reterr_error_desc)
|
||||
*reterr_error_desc = NULL;
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
+12
-4
@@ -68,15 +68,23 @@ int qmp_client_invoke(
|
||||
qmp_command_callback_t callback,
|
||||
void *userdata);
|
||||
|
||||
/* Synchronous send + receive. Pumps the event loop until the reply arrives. *ret_result and
|
||||
* *ret_error_desc are borrowed pointers into the last reply, valid until the next
|
||||
* qmp_client_call(). Same contract as sd_varlink_call(). */
|
||||
int qmp_client_call(
|
||||
QmpClient *client,
|
||||
const char *command,
|
||||
QmpClientArgs *args,
|
||||
sd_json_variant **ret_result,
|
||||
const char **ret_error_desc);
|
||||
char **reterr_error_desc);
|
||||
|
||||
int qmp_client_call_future(
|
||||
QmpClient *client,
|
||||
const char *command,
|
||||
QmpClientArgs *args,
|
||||
sd_future **ret);
|
||||
|
||||
int future_get_qmp_reply(
|
||||
sd_future *f,
|
||||
sd_json_variant **ret_result,
|
||||
char **reterr_error_desc);
|
||||
|
||||
void qmp_client_bind_event(QmpClient *c, qmp_event_callback_t callback, void *userdata);
|
||||
void qmp_client_bind_disconnect(QmpClient *c, qmp_disconnect_callback_t callback, void *userdata);
|
||||
|
||||
Reference in New Issue
Block a user