Introduce support for running code in fibers
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:
Daan De Meyer
+11
-1
@@ -1017,6 +1017,14 @@ else
|
||||
endif
|
||||
conf.set10('HAVE_LIBCRYPT', have)
|
||||
|
||||
# musl declares the ucontext.h functions but does not implement them; the fiber bootstrap in
|
||||
# src/libsystemd/sd-future/fiber.c relies on them, so on musl we have to link to libucontext.
|
||||
if get_option('libc') == 'musl'
|
||||
libucontext = dependency('libucontext')
|
||||
else
|
||||
libucontext = []
|
||||
endif
|
||||
|
||||
bpf_framework = get_option('bpf-framework')
|
||||
bpf_compiler = get_option('bpf-compiler')
|
||||
libbpf = dependency('libbpf',
|
||||
@@ -1720,6 +1728,7 @@ libsystemd_includes = [basic_includes, include_directories(
|
||||
'src/libsystemd/sd-common',
|
||||
'src/libsystemd/sd-device',
|
||||
'src/libsystemd/sd-event',
|
||||
'src/libsystemd/sd-future',
|
||||
'src/libsystemd/sd-hwdb',
|
||||
'src/libsystemd/sd-id128',
|
||||
'src/libsystemd/sd-journal',
|
||||
@@ -1760,7 +1769,7 @@ libsystemd = shared_library(
|
||||
link_with : [libc_wrapper_static,
|
||||
libbasic_static],
|
||||
link_whole : [libsystemd_static],
|
||||
dependencies : [userspace],
|
||||
dependencies : [libucontext, userspace],
|
||||
link_depends : libsystemd_sym,
|
||||
install : true,
|
||||
install_tag: 'libsystemd',
|
||||
@@ -1782,6 +1791,7 @@ if static_libsystemd != 'false'
|
||||
dependencies : [libgcrypt_cflags,
|
||||
liblz4_cflags,
|
||||
libm,
|
||||
libucontext,
|
||||
libxz_cflags,
|
||||
libzstd_cflags,
|
||||
userspace],
|
||||
|
||||
@@ -242,4 +242,10 @@ Architecture uname_architecture(void);
|
||||
# error "Please register your architecture here!"
|
||||
#endif
|
||||
|
||||
#if defined(__hppa__) || defined(__hppa64__)
|
||||
# define STACK_GROWS_UP 1
|
||||
#else
|
||||
# define STACK_GROWS_UP 0
|
||||
#endif
|
||||
|
||||
DECLARE_STRING_TABLE_LOOKUP(architecture, Architecture);
|
||||
|
||||
@@ -111,10 +111,12 @@ typedef enum UnitNameMangle UnitNameMangle;
|
||||
typedef enum UnitType UnitType;
|
||||
typedef enum WaitFlags WaitFlags;
|
||||
|
||||
typedef struct Fiber Fiber;
|
||||
typedef struct Hashmap Hashmap;
|
||||
typedef struct HashmapBase HashmapBase;
|
||||
typedef struct IteratedCache IteratedCache;
|
||||
typedef struct Iterator Iterator;
|
||||
typedef struct LogContext LogContext;
|
||||
typedef struct OrderedHashmap OrderedHashmap;
|
||||
typedef struct OrderedSet OrderedSet;
|
||||
typedef struct Set Set;
|
||||
|
||||
@@ -177,6 +177,14 @@ size_t log_context_num_fields(void) {
|
||||
return _log_context_num_fields;
|
||||
}
|
||||
|
||||
void log_context_swap(LogContext **log_context, size_t *num_fields) {
|
||||
assert(log_context);
|
||||
assert(num_fields);
|
||||
|
||||
SWAP_TWO(_log_context, *log_context);
|
||||
SWAP_TWO(_log_context_num_fields, *num_fields);
|
||||
}
|
||||
|
||||
void _reset_log_level(int *saved_log_level) {
|
||||
assert(saved_log_level);
|
||||
|
||||
|
||||
@@ -66,6 +66,8 @@ size_t log_context_num_contexts(void);
|
||||
/* Returns the number of fields in all attached log contexts. */
|
||||
size_t log_context_num_fields(void);
|
||||
|
||||
void log_context_swap(LogContext **log_context, size_t *num_fields);
|
||||
|
||||
void _reset_log_level(int *saved_log_level);
|
||||
|
||||
#define _LOG_CONTEXT_SET_LOG_LEVEL(level, l) \
|
||||
|
||||
@@ -87,6 +87,12 @@ bool _log_message_dummy = false; /* Always false */
|
||||
} \
|
||||
} while (false)
|
||||
|
||||
void log_prefix_swap(const char **prefix) {
|
||||
assert(prefix);
|
||||
|
||||
SWAP_TWO(log_prefix, *prefix);
|
||||
}
|
||||
|
||||
static void log_close_console(void) {
|
||||
/* See comment in log_close_journal() */
|
||||
(void) safe_close_above_stdio(TAKE_FD(console_fd));
|
||||
|
||||
@@ -380,6 +380,8 @@ int log_syntax_parse_error_internal(
|
||||
void log_setup(void);
|
||||
|
||||
const char* _log_set_prefix(const char *prefix, bool force);
|
||||
|
||||
void log_prefix_swap(const char **prefix);
|
||||
static inline const char* _log_unset_prefixp(const char **p) {
|
||||
assert(p);
|
||||
_log_set_prefix(*p, true);
|
||||
|
||||
@@ -18,3 +18,10 @@ static_assert(MFD_NOEXEC_SEAL == 0x0008U, "");
|
||||
#else
|
||||
static_assert(MFD_EXEC == 0x0010U, "");
|
||||
#endif
|
||||
|
||||
/* since Linux 6.13 / glibc-2.42 */
|
||||
#ifndef MADV_GUARD_INSTALL
|
||||
# define MADV_GUARD_INSTALL 102
|
||||
#else
|
||||
static_assert(MADV_GUARD_INSTALL == 102, "");
|
||||
#endif
|
||||
|
||||
@@ -33,6 +33,7 @@ sd_daemon_sources = files('sd-daemon/sd-daemon.c')
|
||||
############################################################
|
||||
|
||||
sd_event_sources = files(
|
||||
'sd-event/event-future.c',
|
||||
'sd-event/event-util.c',
|
||||
'sd-event/sd-event.c',
|
||||
)
|
||||
@@ -75,6 +76,13 @@ sd_device_sources = files(
|
||||
|
||||
############################################################
|
||||
|
||||
sd_future_sources = files(
|
||||
'sd-future/fiber.c',
|
||||
'sd-future/sd-future.c',
|
||||
)
|
||||
|
||||
############################################################
|
||||
|
||||
sd_login_sources = files('sd-login/sd-login.c')
|
||||
|
||||
############################################################
|
||||
@@ -135,8 +143,9 @@ libsystemd_sources = files(
|
||||
'sd-resolve/sd-resolve.c',
|
||||
) + sd_journal_sources + sd_id128_sources + sd_daemon_sources \
|
||||
+ sd_event_sources + sd_bus_sources + sd_device_sources \
|
||||
+ sd_login_sources + sd_json_sources + sd_varlink_sources \
|
||||
+ sd_path_sources + sd_netlink_sources + sd_network_sources
|
||||
+ sd_future_sources + sd_login_sources + sd_json_sources \
|
||||
+ sd_varlink_sources + sd_path_sources + sd_netlink_sources \
|
||||
+ sd_network_sources
|
||||
|
||||
sources += libsystemd_sources
|
||||
|
||||
@@ -151,6 +160,7 @@ libsystemd_static = static_library(
|
||||
link_with : [libc_wrapper_static,
|
||||
libbasic_static],
|
||||
dependencies : [libm,
|
||||
libucontext,
|
||||
userspace],
|
||||
build_by_default : false)
|
||||
|
||||
@@ -179,6 +189,7 @@ simple_tests += files(
|
||||
'sd-bus/test-bus-vtable.c',
|
||||
'sd-device/test-device-util.c',
|
||||
'sd-device/test-sd-device-monitor.c',
|
||||
'sd-future/test-fiber.c',
|
||||
'sd-hwdb/test-sd-hwdb.c',
|
||||
'sd-id128/test-id128.c',
|
||||
'sd-journal/test-audit-type.c',
|
||||
|
||||
@@ -127,3 +127,9 @@ typedef struct sd_resolve sd_resolve;
|
||||
typedef struct sd_resolve_query sd_resolve_query;
|
||||
|
||||
typedef struct sd_hwdb sd_hwdb;
|
||||
|
||||
typedef struct sd_future sd_future;
|
||||
|
||||
typedef int (*sd_future_func_t)(sd_future *f);
|
||||
typedef int (*sd_fiber_func_t)(void *userdata);
|
||||
typedef _sd_destroy_t sd_fiber_destroy_t;
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
/* SPDX-License-Identifier: LGPL-2.1-or-later */
|
||||
|
||||
#include "sd-event.h"
|
||||
#include "sd-future.h"
|
||||
|
||||
#include "alloc-util.h"
|
||||
#include "errno-util.h"
|
||||
#include "event-future.h"
|
||||
|
||||
typedef struct TimeFuture {
|
||||
sd_event_source *source;
|
||||
|
||||
/* Result the future resolves with on natural expiry (vs. cancellation). 0 for normal sleep,
|
||||
* non-zero (e.g. -ETIMEDOUT) lets a fiber waiting on this future resume with that error. */
|
||||
int result;
|
||||
} TimeFuture;
|
||||
|
||||
static void* time_future_alloc(void) {
|
||||
return new0(TimeFuture, 1);
|
||||
}
|
||||
|
||||
static void time_future_free(sd_future *f) {
|
||||
TimeFuture *tf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
sd_event_source_unref(tf->source);
|
||||
free(tf);
|
||||
}
|
||||
|
||||
static int time_future_cancel(sd_future *f) {
|
||||
TimeFuture *tf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
int r;
|
||||
|
||||
r = sd_event_source_set_enabled(tf->source, SD_EVENT_OFF);
|
||||
RET_GATHER(r, sd_future_resolve(f, -ECANCELED));
|
||||
return r;
|
||||
}
|
||||
|
||||
static int time_future_set_priority(sd_future *f, int64_t priority) {
|
||||
TimeFuture *tf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
return sd_event_source_set_priority(tf->source, priority);
|
||||
}
|
||||
|
||||
static const sd_future_ops time_future_ops = {
|
||||
.size = sizeof(sd_future_ops),
|
||||
.alloc = time_future_alloc,
|
||||
.free = time_future_free,
|
||||
.cancel = time_future_cancel,
|
||||
.set_priority = time_future_set_priority,
|
||||
};
|
||||
|
||||
static int time_handler(sd_event_source *s, usec_t usec, void *userdata) {
|
||||
sd_future *f = ASSERT_PTR(userdata);
|
||||
TimeFuture *tf = ASSERT_PTR(sd_future_get_private(f));
|
||||
|
||||
return sd_future_resolve(f, tf->result);
|
||||
}
|
||||
|
||||
typedef int (*event_add_time_func)(
|
||||
sd_event *e,
|
||||
sd_event_source **ret,
|
||||
clockid_t clock,
|
||||
uint64_t usec,
|
||||
uint64_t accuracy,
|
||||
sd_event_time_handler_t callback,
|
||||
void *userdata);
|
||||
|
||||
static int future_new_time_internal(
|
||||
event_add_time_func add_time,
|
||||
sd_event *e,
|
||||
clockid_t clock,
|
||||
uint64_t usec,
|
||||
uint64_t accuracy,
|
||||
int result,
|
||||
sd_future **ret) {
|
||||
|
||||
int r;
|
||||
|
||||
assert(add_time);
|
||||
assert(e);
|
||||
assert(ret);
|
||||
|
||||
if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED))
|
||||
return -ECANCELED;
|
||||
|
||||
_cleanup_(sd_future_unrefp) sd_future *f = NULL;
|
||||
r = sd_future_new(&time_future_ops, &f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
TimeFuture *tf = sd_future_get_private(f);
|
||||
tf->result = result;
|
||||
|
||||
r = add_time(e, &tf->source, clock, usec, accuracy, time_handler, f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
if (sd_fiber_is_running()) {
|
||||
int64_t priority;
|
||||
|
||||
r = sd_fiber_get_priority(&priority);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
r = sd_event_source_set_priority(tf->source, priority);
|
||||
if (r < 0)
|
||||
return r;
|
||||
}
|
||||
|
||||
*ret = TAKE_PTR(f);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int future_new_time(sd_event *e, clockid_t clock, uint64_t usec, uint64_t accuracy, int result, sd_future **ret) {
|
||||
return future_new_time_internal(sd_event_add_time, e, clock, usec, accuracy, result, ret);
|
||||
}
|
||||
|
||||
int future_new_time_relative(sd_event *e, clockid_t clock, uint64_t usec, uint64_t accuracy, int result, sd_future **ret) {
|
||||
return future_new_time_internal(sd_event_add_time_relative, e, clock, usec, accuracy, result, ret);
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
/* SPDX-License-Identifier: LGPL-2.1-or-later */
|
||||
#pragma once
|
||||
|
||||
#include "sd-forward.h"
|
||||
|
||||
int future_new_time(sd_event *e, clockid_t clock, uint64_t usec, uint64_t accuracy, int result, sd_future **ret);
|
||||
int future_new_time_relative(sd_event *e, clockid_t clock, uint64_t usec, uint64_t accuracy, int result, sd_future **ret);
|
||||
@@ -0,0 +1,812 @@
|
||||
/* SPDX-License-Identifier: LGPL-2.1-or-later */
|
||||
|
||||
#include <pthread.h>
|
||||
#include <setjmp.h>
|
||||
#include <sys/mman.h>
|
||||
#include <sys/resource.h>
|
||||
#include <sys/uio.h>
|
||||
#include <threads.h>
|
||||
#include <ucontext.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#if HAVE_VALGRIND_VALGRIND_H
|
||||
#include <valgrind/valgrind.h>
|
||||
#endif
|
||||
|
||||
#include "sd-event.h"
|
||||
#include "sd-future.h"
|
||||
|
||||
#include "alloc-util.h"
|
||||
#include "architecture.h"
|
||||
#include "errno-util.h"
|
||||
#include "event-future.h"
|
||||
#include "log-context.h"
|
||||
#include "log.h"
|
||||
#include "memory-util.h"
|
||||
#include "pthread-util.h"
|
||||
#include "time-util.h"
|
||||
|
||||
#if HAS_FEATURE_ADDRESS_SANITIZER
|
||||
#include <sanitizer/common_interface_defs.h>
|
||||
#endif
|
||||
|
||||
/* glibc's _FORTIFY_SOURCE wraps siglongjmp() with __longjmp_chk, which asserts that the target SP is below
|
||||
* the current SP. That assumption is incompatible with fiber switching, where the target SP lives on a
|
||||
* separately-mmap'd stack and can be at any address relative to the caller. The fortify redirect happens
|
||||
* in <setjmp.h>'s declaration of siglongjmp; we sidestep it by declaring our own alias that links
|
||||
* directly to the unchecked "siglongjmp" symbol. musl doesn't fortify setjmp.h, so the alias is a plain
|
||||
* synonym there. */
|
||||
_noreturn_ extern void siglongjmp_unchecked(sigjmp_buf env, int val) __asm__("siglongjmp");
|
||||
|
||||
static thread_local Fiber *current_fiber = NULL;
|
||||
|
||||
typedef enum FiberState {
|
||||
FIBER_STATE_INITIAL,
|
||||
FIBER_STATE_READY,
|
||||
FIBER_STATE_SUSPENDED,
|
||||
FIBER_STATE_CANCELLED,
|
||||
FIBER_STATE_COMPLETED,
|
||||
_FIBER_STATE_MAX,
|
||||
_FIBER_STATE_INVALID = -EINVAL,
|
||||
} FiberState;
|
||||
|
||||
typedef struct Fiber {
|
||||
struct iovec stack;
|
||||
sigjmp_buf context; /* Where to jump to when entering or resuming the fiber. */
|
||||
sigjmp_buf resume_context; /* Where to jump back to when the fiber yields or completes. */
|
||||
|
||||
/* Caller's stack range, recorded by fiber_run() on each entry so the fiber's siglongjmp back
|
||||
* out (in fiber_swap() or the trampoline's terminate path) can hand AddressSanitizer the
|
||||
* destination stack info. With ucontext this comes for free via uc_link/uc_stack; sigjmp_buf
|
||||
* is opaque and doesn't carry it. */
|
||||
struct iovec resume_stack;
|
||||
|
||||
FiberState state;
|
||||
int result; /* Either resume error code or final return value */
|
||||
|
||||
sd_future *floating; /* Self-ref held while the fiber is floating; dropped on resolve. */
|
||||
|
||||
sd_event *event;
|
||||
sd_event_source *defer_event_source;
|
||||
sd_event_source *exit_event_source;
|
||||
|
||||
char *name;
|
||||
int64_t priority;
|
||||
sd_fiber_func_t func;
|
||||
void *userdata;
|
||||
sd_fiber_destroy_t destroy;
|
||||
|
||||
/* Storage for the swap performed in fiber_run(): while the fiber is suspended these hold the
|
||||
* fiber's own log state; while it is running they hold the caller's log state. The active state
|
||||
* always lives in the thread-locals in log.c / log-context.c. */
|
||||
LIST_HEAD(LogContext, log_context);
|
||||
size_t log_context_num_fields;
|
||||
const char *log_prefix;
|
||||
|
||||
#if HAVE_VALGRIND_VALGRIND_H
|
||||
unsigned stack_id;
|
||||
#endif
|
||||
} Fiber;
|
||||
|
||||
static Fiber* fiber_get_current(void) {
|
||||
return current_fiber;
|
||||
}
|
||||
|
||||
static void fiber_set_current(Fiber *f) {
|
||||
current_fiber = f;
|
||||
}
|
||||
|
||||
static int fiber_allocate_stack(size_t size, void **ret) {
|
||||
void *stack = NULL;
|
||||
int r;
|
||||
|
||||
/* The effective stack size is one page less than the given size, because we have to use
|
||||
* one page as the guard page for the stack. */
|
||||
|
||||
assert(size > 0 && size % page_size() == 0);
|
||||
assert(ret);
|
||||
|
||||
stack = mmap(/* addr= */ NULL, size,
|
||||
PROT_READ | PROT_WRITE,
|
||||
MAP_PRIVATE | MAP_ANONYMOUS | MAP_STACK,
|
||||
/* fd= */ -EBADF, /* offset= */ 0);
|
||||
if (stack == MAP_FAILED)
|
||||
return -errno;
|
||||
|
||||
/* Place the guard page where stack overflow will hit it: the high end on architectures
|
||||
* where the stack grows up (PA-RISC), the low end everywhere else. fiber_stack_usable()
|
||||
* mirrors this with the inverse offset. */
|
||||
void *guard = STACK_GROWS_UP ? (uint8_t*) stack + size - page_size() : stack;
|
||||
|
||||
/* Prefer MADV_GUARD_INSTALL (Linux 6.13+): unlike mprotect(PROT_NONE) it doesn't split
|
||||
* the VMA, so guard installation skips the mmap-lock contention and per-guard VMA cost.
|
||||
* Fall back to mprotect on older kernels, which return EINVAL for unknown advice.
|
||||
* FIXME: delete when baseline above 6.13. */
|
||||
r = RET_NERRNO(madvise(guard, page_size(), MADV_GUARD_INSTALL));
|
||||
if (r == -EINVAL)
|
||||
r = RET_NERRNO(mprotect(guard, page_size(), PROT_NONE));
|
||||
if (r < 0) {
|
||||
(void) munmap(stack, size);
|
||||
return r;
|
||||
}
|
||||
|
||||
*ret = TAKE_PTR(stack);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Usable stack range of a fiber: the full mmap region minus the guard page. Single source of
|
||||
* truth for the layout assumed by fiber_allocate_stack(); every consumer (ucontext ss_sp,
|
||||
* ASAN handoff iovecs, Valgrind stack registration) goes through here.
|
||||
*
|
||||
* iov_base is the lowest usable byte regardless of growth direction — that matches POSIX's
|
||||
* definition of stack_t.ss_sp, so libc's makecontext() handles the direction for us. Only the
|
||||
* guard page placement (and hence iov_base's offset within the mapping) varies. */
|
||||
static struct iovec fiber_stack_usable(const struct iovec *stack) {
|
||||
assert(stack);
|
||||
assert(stack->iov_len > page_size());
|
||||
return (struct iovec) {
|
||||
.iov_base = STACK_GROWS_UP ? stack->iov_base : (uint8_t*) stack->iov_base + page_size(),
|
||||
.iov_len = stack->iov_len - page_size(),
|
||||
};
|
||||
}
|
||||
|
||||
static inline void start_switch_stack(void **fake_stack_save, const struct iovec *dest) {
|
||||
#if HAS_FEATURE_ADDRESS_SANITIZER
|
||||
__sanitizer_start_switch_fiber(fake_stack_save,
|
||||
dest ? dest->iov_base : NULL,
|
||||
dest ? dest->iov_len : 0);
|
||||
#endif
|
||||
}
|
||||
|
||||
static inline void finish_switch_stack(void *fake_stack_save) {
|
||||
#if HAS_FEATURE_ADDRESS_SANITIZER
|
||||
__sanitizer_finish_switch_fiber(fake_stack_save, NULL, NULL);
|
||||
#endif
|
||||
}
|
||||
|
||||
/* Refresh f->resume_stack from whoever is currently the running fiber, so the next siglongjmp() out
|
||||
* of f (in the trampoline or fiber_swap()) can hand the right destination stack to ASAN. Must be
|
||||
* called before fiber_set_current(f) — relies on fiber_get_current() returning the caller. */
|
||||
static void fiber_set_resume_stack(Fiber *f, Fiber *resume) {
|
||||
assert(f);
|
||||
|
||||
if (resume)
|
||||
f->resume_stack = fiber_stack_usable(&resume->stack);
|
||||
else
|
||||
f->resume_stack = (struct iovec) {};
|
||||
}
|
||||
|
||||
_noreturn_ static void fiber_entry_point(void) {
|
||||
Fiber *f = ASSERT_PTR(fiber_get_current());
|
||||
void *fake_stack_save = NULL;
|
||||
|
||||
assert(f->func);
|
||||
assert(IN_SET(f->state, FIBER_STATE_INITIAL, FIBER_STATE_READY, FIBER_STATE_CANCELLED));
|
||||
|
||||
finish_switch_stack(NULL);
|
||||
|
||||
/* Capture our resumable point on the fiber's stack, then bounce back to whoever last set
|
||||
* f->resume_context. On bootstrap that's fiber_bootstrap(); on every subsequent yield it's
|
||||
* the most recent fiber_run(). sigsetjmp(buf, 0) skips the signal-mask save: switching is
|
||||
* thread-shared with respect to signal masks. */
|
||||
if (sigsetjmp(f->context, /* savemask= */ 0) == 0) {
|
||||
start_switch_stack(&fake_stack_save, &f->resume_stack);
|
||||
siglongjmp_unchecked(f->resume_context, /* val= */ 1);
|
||||
}
|
||||
|
||||
/* Re-entered for real via fiber_run()'s siglongjmp(f->context). */
|
||||
finish_switch_stack(fake_stack_save);
|
||||
|
||||
/* Block scope so the cleanups attached to LOG_SET_PREFIX / LOG_CONTEXT_PUSH_KEY_VALUE fire
|
||||
* before the siglongjmp below — siglongjmp skips _cleanup_ attributes, so we have to make
|
||||
* sure the scope ends via a normal control-flow path first. */
|
||||
{
|
||||
LOG_SET_PREFIX(f->name);
|
||||
LOG_CONTEXT_PUSH_KEY_VALUE("FIBER=", f->name);
|
||||
|
||||
f->result = f->state == FIBER_STATE_CANCELLED ? -ECANCELED : f->func(f->userdata);
|
||||
f->state = FIBER_STATE_COMPLETED;
|
||||
}
|
||||
|
||||
/* Pass NULL fake_stack_save to discard the fiber's fake stack since the fiber is done. */
|
||||
start_switch_stack(NULL, &f->resume_stack);
|
||||
|
||||
/* Bounce back to whichever fiber_run() call most recently entered us. resume_context is
|
||||
* per-fiber so nested fiber_run() — e.g. a bus method dispatched as a fiber handler while
|
||||
* sd_event_loop() itself runs in a fiber — is safe. */
|
||||
siglongjmp_unchecked(f->resume_context, 1);
|
||||
assert_not_reached();
|
||||
}
|
||||
|
||||
static int fiber_init(Fiber *f) {
|
||||
ucontext_t old_uc, uc;
|
||||
void *fake_stack_save = NULL;
|
||||
|
||||
assert(f);
|
||||
|
||||
if (getcontext(&uc) < 0)
|
||||
return -errno;
|
||||
|
||||
struct iovec fiber_stack = fiber_stack_usable(&f->stack);
|
||||
|
||||
uc.uc_link = NULL; /* Unused: trampoline siglongjmps out instead of returning. */
|
||||
uc.uc_stack.ss_sp = fiber_stack.iov_base;
|
||||
uc.uc_stack.ss_size = fiber_stack.iov_len;
|
||||
uc.uc_stack.ss_flags = 0;
|
||||
|
||||
Fiber *prev = fiber_get_current();
|
||||
fiber_set_current(f);
|
||||
|
||||
makecontext(&uc, fiber_entry_point, /* argc= */ 0);
|
||||
|
||||
fiber_set_resume_stack(f, prev);
|
||||
if (sigsetjmp(f->resume_context, /* savemask= */ 0) == 0) {
|
||||
start_switch_stack(&fake_stack_save, &fiber_stack);
|
||||
if (swapcontext(&old_uc, &uc) < 0) {
|
||||
finish_switch_stack(fake_stack_save);
|
||||
fiber_set_current(prev);
|
||||
return -errno;
|
||||
}
|
||||
assert_not_reached(); /* Trampoline siglongjmps back; swapcontext doesn't return. */
|
||||
}
|
||||
|
||||
finish_switch_stack(fake_stack_save);
|
||||
|
||||
fiber_set_current(prev);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Swap the thread-local log prefix and log context with the values stashed in f. While the fiber is
|
||||
* suspended, f holds the fiber's own log state; while it's running, f holds the caller's log state. The
|
||||
* swap is its own inverse, so the same call drives both directions. */
|
||||
static void fiber_swap_log_state(Fiber *f) {
|
||||
assert(f);
|
||||
log_prefix_swap(&f->log_prefix);
|
||||
log_context_swap(&f->log_context, &f->log_context_num_fields);
|
||||
}
|
||||
|
||||
static void reset_current_fiber(void) {
|
||||
/* Restore the caller's log state stashed in the running fiber (if any) before clearing
|
||||
* current_fiber. Without this, the child of a fork() that happened mid-fiber would inherit the
|
||||
* fiber's log prefix / context list in its thread-locals even though no fiber is running. */
|
||||
Fiber *f = fiber_get_current();
|
||||
if (f)
|
||||
fiber_swap_log_state(f);
|
||||
fiber_set_current(NULL);
|
||||
}
|
||||
|
||||
static sd_event_source* fiber_current_event_source(Fiber *f) {
|
||||
assert(f);
|
||||
assert(f->state != FIBER_STATE_COMPLETED);
|
||||
assert(f->event);
|
||||
|
||||
return sd_event_get_state(f->event) == SD_EVENT_EXITING ? f->exit_event_source : f->defer_event_source;
|
||||
}
|
||||
|
||||
static int atfork_ret;
|
||||
|
||||
static void install_atfork(void) {
|
||||
/* __register_atfork() either returns 0 or -ENOMEM, in its glibc implementation. Since it's
|
||||
* only half-documented (glibc doesn't document it but LSB does — though only superficially)
|
||||
* we'll check for errors only in the most generic fashion possible. */
|
||||
atfork_ret = pthread_atfork(/* prepare= */ NULL, /* parent= */ NULL, reset_current_fiber);
|
||||
if (atfork_ret != 0)
|
||||
log_debug_errno(atfork_ret, "pthread_atfork() failed: %m");
|
||||
}
|
||||
|
||||
static void fiber_resolve(sd_future *f) {
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
|
||||
fiber->defer_event_source = sd_event_source_disable_unref(fiber->defer_event_source);
|
||||
fiber->exit_event_source = sd_event_source_disable_unref(fiber->exit_event_source);
|
||||
/* The floating self-ref (if any) is potentially the last ref keeping the fiber alive — moving it
|
||||
* into a local _cleanup_ slot ensures sd_future_resolve() runs callbacks and waiters while f is
|
||||
* still valid; the local's cleanup drops the ref afterwards, at which point no further f->...
|
||||
* access can happen. */
|
||||
_unused_ _cleanup_(sd_future_unrefp) sd_future *floating = TAKE_PTR(fiber->floating);
|
||||
sd_future_resolve(f, fiber->result);
|
||||
}
|
||||
|
||||
static void fiber_enter(Fiber *fiber, Fiber *prev, void **fake_stack_save) {
|
||||
fiber_set_current(fiber);
|
||||
fiber_swap_log_state(fiber);
|
||||
|
||||
struct iovec fiber_stack = fiber_stack_usable(&fiber->stack);
|
||||
start_switch_stack(fake_stack_save, &fiber_stack);
|
||||
fiber_set_resume_stack(fiber, prev);
|
||||
}
|
||||
|
||||
static void fiber_leave(Fiber *fiber, Fiber *prev, void *fake_stack_save) {
|
||||
finish_switch_stack(fake_stack_save);
|
||||
fiber_swap_log_state(fiber);
|
||||
fiber_set_current(prev);
|
||||
}
|
||||
|
||||
static int fiber_run(sd_future *f) {
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
int r;
|
||||
|
||||
if (fiber->state == FIBER_STATE_COMPLETED)
|
||||
return -ESTALE;
|
||||
|
||||
assert(IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_READY, FIBER_STATE_CANCELLED));
|
||||
|
||||
static pthread_once_t atfork_once = PTHREAD_ONCE_INIT;
|
||||
r = pthread_once(&atfork_once, install_atfork);
|
||||
if (r != 0)
|
||||
return -r;
|
||||
if (atfork_ret != 0)
|
||||
return -atfork_ret;
|
||||
|
||||
LOG_SET_PREFIX(fiber->name);
|
||||
LOG_CONTEXT_PUSH_KEY_VALUE("FIBER=", fiber->name);
|
||||
|
||||
log_debug("Scheduling fiber");
|
||||
|
||||
/* Save the previously-current fiber (if any) so we can restore it when this fiber yields or
|
||||
* completes. This matters when fiber_run() is invoked from within another fiber (e.g. an
|
||||
* sd-event dispatch that happens to be running inside a fiber context itself): the
|
||||
* LOG_SET_PREFIX/LOG_CONTEXT_PUSH above attached to whichever fiber was current at that moment,
|
||||
* and their scope-level cleanup must see the same fiber_get_current() when it runs to detach
|
||||
* them from the correct list. */
|
||||
Fiber *prev = fiber_get_current();
|
||||
void *fake_stack_save = NULL;
|
||||
fiber_enter(fiber, prev, &fake_stack_save);
|
||||
|
||||
/* This is where we start executing the fiber. Once it yields, we continue here as if nothing
|
||||
* happened. resume_context captures this point; the fiber siglongjmps back to it. */
|
||||
if (sigsetjmp(fiber->resume_context, 0) == 0)
|
||||
siglongjmp_unchecked(fiber->context, 1);
|
||||
|
||||
fiber_leave(fiber, prev, fake_stack_save);
|
||||
|
||||
switch (fiber->state) {
|
||||
|
||||
case FIBER_STATE_COMPLETED:
|
||||
if (fiber->result < 0 && fiber->result != -ECANCELED)
|
||||
log_debug_errno(fiber->result, "Fiber failed with error: %m");
|
||||
else
|
||||
log_debug("Fiber finished executing");
|
||||
|
||||
fiber_resolve(f);
|
||||
break;
|
||||
|
||||
case FIBER_STATE_CANCELLED:
|
||||
case FIBER_STATE_READY:
|
||||
log_debug("Fiber yielded execution");
|
||||
|
||||
r = sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT);
|
||||
if (r < 0)
|
||||
return r;
|
||||
break;
|
||||
|
||||
case FIBER_STATE_SUSPENDED:
|
||||
log_debug("Fiber suspended execution");
|
||||
/* Fiber is waiting for something - don't re-queue it */
|
||||
break;
|
||||
|
||||
default:
|
||||
assert_not_reached();
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int fiber_cancel(sd_future *f) {
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
int r;
|
||||
|
||||
assert(fiber != fiber_get_current());
|
||||
|
||||
if (IN_SET(fiber->state, FIBER_STATE_COMPLETED, FIBER_STATE_CANCELLED))
|
||||
return 0;
|
||||
|
||||
if (fiber->state == FIBER_STATE_INITIAL) {
|
||||
/* The fiber's stack was allocated but never entered, so there are no scope-level cleanups
|
||||
* waiting to run. Skip the dispatch round-trip that would just have fiber_entry_point()
|
||||
* fall straight through with -ECANCELED, and settle the future right here — mirroring the
|
||||
* FIBER_STATE_COMPLETED branch of fiber_run(). */
|
||||
fiber->result = -ECANCELED;
|
||||
fiber->state = FIBER_STATE_COMPLETED;
|
||||
fiber_resolve(f);
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* Once we cancel a fiber, we want to immediately resume it with -ECANCELED. */
|
||||
r = sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
fiber->state = FIBER_STATE_CANCELLED;
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int fiber_on_defer(sd_event_source *s, void *userdata) {
|
||||
sd_future *f = ASSERT_PTR(userdata);
|
||||
return fiber_run(f);
|
||||
}
|
||||
|
||||
static int fiber_on_exit(sd_event_source *s, void *userdata) {
|
||||
sd_future *f = ASSERT_PTR(userdata);
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
|
||||
int r;
|
||||
|
||||
/* The fiber may already have completed via the regular defer path before sd_event_exit()
|
||||
* fires the exit source; in that case there's nothing left to drive and we'd otherwise
|
||||
* trip fiber_run()'s -ESTALE return, which sd_event would log spuriously and disable the
|
||||
* source for. */
|
||||
if (fiber->state == FIBER_STATE_COMPLETED)
|
||||
return 0;
|
||||
|
||||
/* If fiber_cancel() returned 1 the fiber was just marked cancelled and its deferred/exit event
|
||||
* source was re-armed; we let the event loop dispatch that source on the next iteration so it goes
|
||||
* through the normal fiber_on_defer/fiber_on_exit path rather than running it recursively here. */
|
||||
r = fiber_cancel(f);
|
||||
if (r != 0)
|
||||
return r;
|
||||
|
||||
return fiber_run(f);
|
||||
}
|
||||
|
||||
static void* fiber_alloc(void) {
|
||||
return new0(Fiber, 1);
|
||||
}
|
||||
|
||||
static void fiber_free(sd_future *f) {
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
|
||||
|
||||
/* To make sure all memory is deallocated, the fiber has to have completed by the time we free it to
|
||||
* make sure its stack has finished unwinding (which will invoke the registered cleanup functions).
|
||||
* As this function may get called when not running on a fiber ourselves, we can't guarantee here
|
||||
* that we can run the fiber to completion ourselves, so we insist that this happens before we get
|
||||
* here. To ensure fibers are cleaned up before exiting the event loop, exit handlers are added for
|
||||
* fibers created outside of existing fibers. For fibers created within running fibers, unwinding the
|
||||
* outer fiber should take care of cleaning up any created child fibers (for example using
|
||||
* sd_future_cancel_wait_unref()).
|
||||
*
|
||||
* FIBER_STATE_INITIAL is also accepted: the stack was allocated but never entered, so there are no
|
||||
* registered cleanups to run. This covers the partial-construction failure path in sd_fiber_new()
|
||||
* as well as fibers that are unrefed before the event loop ever dispatches them. */
|
||||
assert(IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_COMPLETED));
|
||||
|
||||
if (fiber->destroy)
|
||||
fiber->destroy(fiber->userdata);
|
||||
|
||||
#if HAVE_VALGRIND_VALGRIND_H
|
||||
if (fiber->stack.iov_base)
|
||||
VALGRIND_STACK_DEREGISTER(fiber->stack_id);
|
||||
#endif
|
||||
|
||||
if (fiber->stack.iov_base)
|
||||
(void) munmap(fiber->stack.iov_base, fiber->stack.iov_len);
|
||||
|
||||
sd_event_source_disable_unref(fiber->defer_event_source);
|
||||
sd_event_source_disable_unref(fiber->exit_event_source);
|
||||
sd_event_unref(fiber->event);
|
||||
|
||||
free(fiber->name);
|
||||
free(fiber);
|
||||
}
|
||||
|
||||
sd_future* sd_fiber_get_current(void) {
|
||||
Fiber *f = fiber_get_current();
|
||||
if (!f)
|
||||
return NULL;
|
||||
|
||||
return sd_event_source_get_userdata(fiber_current_event_source(f));
|
||||
}
|
||||
|
||||
int sd_fiber_is_running(void) {
|
||||
return !!fiber_get_current();
|
||||
}
|
||||
|
||||
sd_event* sd_fiber_get_event(void) {
|
||||
Fiber *f = fiber_get_current();
|
||||
assert_return(f, NULL);
|
||||
return f->event;
|
||||
}
|
||||
|
||||
int sd_fiber_get_priority(int64_t *ret) {
|
||||
Fiber *f = fiber_get_current();
|
||||
|
||||
assert_return(ret, -EINVAL);
|
||||
assert_return(f, -ESRCH);
|
||||
|
||||
*ret = f->priority;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int fiber_swap(FiberState state) {
|
||||
Fiber *f = ASSERT_PTR(fiber_get_current());
|
||||
|
||||
f->state = state;
|
||||
|
||||
void *fake_stack_save = NULL;
|
||||
|
||||
if (sigsetjmp(f->context, 0) == 0) {
|
||||
start_switch_stack(&fake_stack_save, &f->resume_stack);
|
||||
siglongjmp_unchecked(f->resume_context, 1);
|
||||
}
|
||||
|
||||
finish_switch_stack(fake_stack_save);
|
||||
|
||||
/* When we get here, we've been resumed. */
|
||||
|
||||
if (f->state == FIBER_STATE_CANCELLED)
|
||||
return -ECANCELED;
|
||||
|
||||
/* sd_fiber_resume() stashes the resumer's value (an async wakeup error from a deadline
|
||||
* timer, an io_uring CQE result, etc.) into f->result for us to surface here. Consume it
|
||||
* unconditionally so it doesn't pollute subsequent suspends or the fiber's eventual return
|
||||
* value — both negative errors and positive payloads (byte counts, accepted fds, revents
|
||||
* masks) are valid resume values. */
|
||||
return TAKE_GENERIC(f->result, int, 0);
|
||||
}
|
||||
|
||||
int sd_fiber_yield(void) {
|
||||
assert_return(fiber_get_current(), -ESRCH);
|
||||
return fiber_swap(FIBER_STATE_READY);
|
||||
}
|
||||
|
||||
int sd_fiber_suspend(void) {
|
||||
assert_return(fiber_get_current(), -ESRCH);
|
||||
return fiber_swap(FIBER_STATE_SUSPENDED);
|
||||
}
|
||||
|
||||
static int fiber_set_priority(sd_future *f, int64_t priority) {
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
int r = 0;
|
||||
|
||||
if (fiber->defer_event_source)
|
||||
RET_GATHER(r, sd_event_source_set_priority(fiber->defer_event_source, priority));
|
||||
|
||||
if (fiber->exit_event_source)
|
||||
RET_GATHER(r, sd_event_source_set_priority(fiber->exit_event_source, priority));
|
||||
|
||||
if (r >= 0)
|
||||
fiber->priority = priority;
|
||||
|
||||
return r;
|
||||
}
|
||||
|
||||
static const sd_future_ops fiber_future_ops;
|
||||
|
||||
int sd_fiber_resume(sd_future *f, int result) {
|
||||
assert_return(f, -EINVAL);
|
||||
assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL);
|
||||
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
|
||||
|
||||
if (fiber->state != FIBER_STATE_SUSPENDED)
|
||||
return 0;
|
||||
|
||||
/* Stash the result so fiber_swap() returns it from sd_fiber_suspend(). */
|
||||
fiber->result = result;
|
||||
fiber->state = FIBER_STATE_READY;
|
||||
return sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT);
|
||||
}
|
||||
|
||||
/* The fiber_future ops pass the Fiber pointer through as the future's private state. The fiber resolves
|
||||
* its own future once it finishes running, so fiber_cancel() intentionally does not resolve. */
|
||||
static const sd_future_ops fiber_future_ops = {
|
||||
.size = sizeof(sd_future_ops),
|
||||
.alloc = fiber_alloc,
|
||||
.free = fiber_free,
|
||||
.cancel = fiber_cancel,
|
||||
.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 r;
|
||||
|
||||
assert_return(e, -EINVAL);
|
||||
assert_return(name, -EINVAL);
|
||||
assert_return(func, -EINVAL);
|
||||
|
||||
if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED))
|
||||
return -ECANCELED;
|
||||
|
||||
_cleanup_(sd_future_unrefp) sd_future *f = NULL;
|
||||
r = sd_future_new(&fiber_future_ops, &f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
|
||||
|
||||
struct rlimit rl = { RLIM_INFINITY, RLIM_INFINITY };
|
||||
if (getrlimit(RLIMIT_STACK, &rl) < 0)
|
||||
log_debug_errno(errno, "Reading RLIMIT_STACK failed, ignoring: %m");
|
||||
if (rl.rlim_cur == RLIM_INFINITY)
|
||||
rl.rlim_cur = 8U * U64_MB; /* Same as the default thread stack size */
|
||||
|
||||
/* Reserve room for the guard page so the usable region stays above PTHREAD_STACK_MIN, which
|
||||
* is what libc/pthread routines (e.g. TLS setup on musl) assume. */
|
||||
size_t stack_len = ROUND_UP(rl.rlim_cur, page_size());
|
||||
if (stack_len < (size_t) PTHREAD_STACK_MIN + page_size())
|
||||
stack_len = ROUND_UP((size_t) PTHREAD_STACK_MIN + page_size(), page_size());
|
||||
|
||||
*fiber = (Fiber) {
|
||||
.stack.iov_len = stack_len,
|
||||
.state = FIBER_STATE_INITIAL,
|
||||
.name = strdup(name),
|
||||
.func = func,
|
||||
.userdata = userdata,
|
||||
.event = sd_event_ref(e),
|
||||
};
|
||||
if (!fiber->name)
|
||||
return -ENOMEM;
|
||||
|
||||
r = fiber_allocate_stack(fiber->stack.iov_len, &fiber->stack.iov_base);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
#if HAVE_VALGRIND_VALGRIND_H
|
||||
/* Register the usable stack range (above the guard page) before fiber_bootstrap() so the
|
||||
* trampoline's first sigsetjmp doesn't trip Valgrind's stack-tracking heuristics. */
|
||||
struct iovec usable = fiber_stack_usable(&fiber->stack);
|
||||
fiber->stack_id = VALGRIND_STACK_REGISTER(
|
||||
usable.iov_base,
|
||||
(uint8_t*) usable.iov_base + usable.iov_len);
|
||||
#endif
|
||||
|
||||
r = fiber_init(fiber);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
/* Execution of the fiber is driven by two event sources, one deferred, one exit. The exit event
|
||||
* source kicks in when sd_event_exit() is called, as from that point onwards only exit event
|
||||
* sources will be dispatched. */
|
||||
|
||||
r = sd_event_add_defer(e, &fiber->defer_event_source, fiber_on_defer, f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
r = sd_event_source_set_description(fiber->defer_event_source, fiber->name);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
r = sd_event_add_exit(e, &fiber->exit_event_source, fiber_on_exit, f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
r = sd_event_source_set_description(fiber->exit_event_source, fiber->name);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
/* If we're on a fiber, we'll rely on the parent fiber to cancel this fiber if the event loop is
|
||||
* exiting. Otherwise, we'll trigger cancellation of this fiber via the exit event source. Why cancel
|
||||
* via the exit event source? We can only run the fiber while the event loop is active, so we need to
|
||||
* make sure all fibers finish running before the event loop is finished, which an exit event source
|
||||
* allows us to do. */
|
||||
r = sd_event_source_set_enabled(fiber->exit_event_source, sd_fiber_is_running() ? SD_EVENT_OFF : SD_EVENT_ONESHOT);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
/* Stays in FIBER_STATE_INITIAL until the event loop first dispatches it via fiber_run(). */
|
||||
|
||||
if (ret)
|
||||
*ret = TAKE_PTR(f);
|
||||
else {
|
||||
/* Fire-and-forget: the fiber is guaranteed to resolve (via completion, cancellation, or
|
||||
* the event loop exit handler), so making the future floating cleans it up. */
|
||||
r = sd_fiber_set_floating(f, true);
|
||||
if (r < 0)
|
||||
return r;
|
||||
}
|
||||
|
||||
/* 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_floating(sd_future *f, int b) {
|
||||
assert_return(f, -EINVAL);
|
||||
assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL);
|
||||
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
|
||||
|
||||
if (!!fiber->floating == !!b)
|
||||
return 0;
|
||||
|
||||
/* The floating self-ref keeps the future alive until the fiber resolves; fiber_run() drops it
|
||||
* in the COMPLETED branch. Only valid for fiber futures because fibers uniquely guarantee
|
||||
* resolution (via completion, cancellation, or the event loop exit handler). */
|
||||
if (b)
|
||||
fiber->floating = sd_future_ref(f);
|
||||
else
|
||||
fiber->floating = sd_future_unref(fiber->floating);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int sd_fiber_get_floating(sd_future *f) {
|
||||
assert_return(f, -EINVAL);
|
||||
assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL);
|
||||
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
|
||||
return !!fiber->floating;
|
||||
}
|
||||
|
||||
int sd_fiber_sleep(uint64_t usec) {
|
||||
Fiber *f = fiber_get_current();
|
||||
int r;
|
||||
|
||||
if (!f)
|
||||
return usleep_safe(usec);
|
||||
|
||||
if (usec == 0)
|
||||
return sd_fiber_yield();
|
||||
|
||||
/* Match usleep_safe(USEC_INFINITY): suspend indefinitely. Passing USEC_INFINITY to
|
||||
* sd_event_add_time_relative() would overflow into -EOVERFLOW. */
|
||||
if (usec == USEC_INFINITY)
|
||||
return sd_fiber_suspend();
|
||||
|
||||
assert(f->event);
|
||||
|
||||
_cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL;
|
||||
r = future_new_time_relative(
|
||||
f->event,
|
||||
CLOCK_MONOTONIC,
|
||||
usec,
|
||||
/* accuracy= */ 1,
|
||||
/* result= */ 0,
|
||||
&timer);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
return sd_fiber_suspend();
|
||||
}
|
||||
|
||||
int sd_fiber_await(sd_future *target) {
|
||||
sd_future *f = sd_fiber_get_current();
|
||||
int r;
|
||||
|
||||
assert_return(f, -ESRCH);
|
||||
assert_return(target, -EINVAL);
|
||||
assert_return(target != f, -EDEADLK);
|
||||
|
||||
Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
|
||||
|
||||
if (sd_future_state(target) == SD_FUTURE_RESOLVED)
|
||||
return sd_future_result(target);
|
||||
|
||||
/* Note that we do allow waiting for other fibers when the event loop is exiting, since waiting for
|
||||
* other fibers does not require adding new event sources to the event loop. */
|
||||
if (sd_event_get_state(fiber->event) == SD_EVENT_FINISHED)
|
||||
return -ECANCELED;
|
||||
|
||||
_cleanup_(sd_future_cancel_wait_unrefp) sd_future *wait = NULL;
|
||||
r = sd_future_new_wait(target, &wait);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
return sd_fiber_suspend();
|
||||
}
|
||||
|
||||
sd_future* sd_fiber_timeout(uint64_t timeout) {
|
||||
Fiber *fiber = fiber_get_current();
|
||||
int r;
|
||||
|
||||
assert_return(fiber, NULL);
|
||||
|
||||
if (timeout == USEC_INFINITY)
|
||||
return NULL;
|
||||
|
||||
sd_future *timer;
|
||||
r = future_new_time_relative(
|
||||
fiber->event,
|
||||
CLOCK_MONOTONIC,
|
||||
timeout,
|
||||
/* accuracy= */ 1,
|
||||
/* result= */ -ETIME,
|
||||
&timer);
|
||||
if (r < 0)
|
||||
return NULL; /* On allocation failure no timer is armed and the scope becomes a no-op.
|
||||
* Errors here are rare; if the caller cares they can compare to NULL. */
|
||||
|
||||
return timer;
|
||||
}
|
||||
@@ -0,0 +1,263 @@
|
||||
/* SPDX-License-Identifier: LGPL-2.1-or-later */
|
||||
|
||||
#include "sd-future.h"
|
||||
|
||||
#include "alloc-util.h"
|
||||
#include "errno-util.h"
|
||||
#include "log.h"
|
||||
#include "macro.h"
|
||||
#include "set.h"
|
||||
|
||||
struct sd_future {
|
||||
unsigned n_ref;
|
||||
|
||||
int state;
|
||||
int result;
|
||||
|
||||
Set *waiters;
|
||||
|
||||
sd_future_func_t callback;
|
||||
void *userdata;
|
||||
|
||||
const sd_future_ops *ops;
|
||||
|
||||
/* Opaque per-future state owned by the future implementation (the code that called
|
||||
* sd_future_new()). The ops vtable above receives this pointer in its callbacks, and
|
||||
* external code can fetch it via sd_future_get_private(). */
|
||||
void *private;
|
||||
};
|
||||
|
||||
static int fiber_resume_trampoline(sd_future *f) {
|
||||
/* The future's result is what the fiber should resume with. Impls choose the value at
|
||||
* resolution time — e.g. a deadline timer resolves with -ETIME, a wait future resolves
|
||||
* with the target's result, a normal IO/sleep future resolves with 0 on success. */
|
||||
return sd_fiber_resume(sd_future_get_userdata(f), sd_future_result(f));
|
||||
}
|
||||
|
||||
int sd_future_resolve(sd_future *f, int result) {
|
||||
int r = 0;
|
||||
|
||||
assert_return(f, -EINVAL);
|
||||
|
||||
if (f->state != SD_FUTURE_PENDING)
|
||||
return 0;
|
||||
|
||||
/* Hold a self-ref across callback/waiter dispatch: callbacks (e.g. bus_fiber_resolved()
|
||||
* dropping the tracking-set's ref) may legitimately release what would otherwise be the
|
||||
* last reference, and we still access f->waiters below. The cleanup unrefs at scope exit,
|
||||
* which is when freeing is safe again. */
|
||||
_unused_ _cleanup_(sd_future_unrefp) sd_future *self = sd_future_ref(f);
|
||||
|
||||
f->state = SD_FUTURE_RESOLVED;
|
||||
f->result = result;
|
||||
|
||||
if (f->callback)
|
||||
RET_GATHER(r, f->callback(f));
|
||||
|
||||
/* We'd like the set to not be modified while iterating over it, hence take ownership over it in
|
||||
* a local variable. Otherwise code invoked via sd_future_resolve() could try to modify the set while
|
||||
* we're iterating over it (for example wait_future_free()). */
|
||||
Set *waiters = TAKE_PTR(f->waiters);
|
||||
sd_future *w;
|
||||
SET_FOREACH(w, waiters)
|
||||
RET_GATHER(r, sd_future_resolve(w, result));
|
||||
|
||||
set_free(waiters);
|
||||
|
||||
return r;
|
||||
}
|
||||
|
||||
static sd_future* sd_future_free(sd_future *f) {
|
||||
if (!f)
|
||||
return NULL;
|
||||
|
||||
if (f->state == SD_FUTURE_PENDING)
|
||||
sd_future_resolve(f, -ECANCELED);
|
||||
|
||||
set_free(f->waiters);
|
||||
|
||||
if (f->ops->free)
|
||||
f->ops->free(f);
|
||||
|
||||
return mfree(f);
|
||||
}
|
||||
|
||||
DEFINE_TRIVIAL_REF_UNREF_FUNC(sd_future, sd_future, sd_future_free);
|
||||
DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_unref);
|
||||
DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_unref);
|
||||
|
||||
sd_future* sd_future_cancel_wait_unref(sd_future *f) {
|
||||
int r;
|
||||
|
||||
if (!f)
|
||||
return NULL;
|
||||
|
||||
/* We have to be able to suspend until the fiber we're waiting for finishes, and that's only
|
||||
* possible if we're running on a fiber ourselves. */
|
||||
if (!sd_fiber_is_running())
|
||||
return sd_future_unref(f);
|
||||
|
||||
r = sd_future_cancel(f);
|
||||
if (r < 0)
|
||||
log_debug_errno(r, "Failed to cancel future, ignoring: %m");
|
||||
|
||||
if (f->state == SD_FUTURE_PENDING) {
|
||||
/* Fast path: when f's resolve callback already targets the current fiber (the default for
|
||||
* futures created on this fiber), we can suspend directly and let the existing trampoline
|
||||
* wake us up — no need to allocate a wait future just to learn about the resolution.
|
||||
* Otherwise fall back to sd_fiber_await() which sets up an explicit waiter. */
|
||||
if (f->callback == fiber_resume_trampoline && f->userdata == sd_fiber_get_current())
|
||||
r = sd_fiber_suspend();
|
||||
else
|
||||
r = sd_fiber_await(f);
|
||||
if (r < 0 && r != -ECANCELED)
|
||||
log_debug_errno(r, "Failed to wait for future to finish, ignoring: %m");
|
||||
}
|
||||
|
||||
return sd_future_unref(f);
|
||||
}
|
||||
|
||||
DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_cancel_wait_unref);
|
||||
DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_cancel_wait_unref);
|
||||
|
||||
int sd_future_new(const sd_future_ops *ops, sd_future **ret) {
|
||||
assert_return(ops, -EINVAL);
|
||||
assert_return(ops->size >= endoffsetof_field(sd_future_ops, set_priority), -EINVAL);
|
||||
assert_return(ops->alloc, -EINVAL);
|
||||
assert_return(ops->free, -EINVAL);
|
||||
assert_return(ret, -EINVAL);
|
||||
|
||||
sd_future *f = new(sd_future, 1);
|
||||
if (!f)
|
||||
return -ENOMEM;
|
||||
|
||||
*f = (sd_future) {
|
||||
.n_ref = 1,
|
||||
.state = SD_FUTURE_PENDING,
|
||||
.ops = ops,
|
||||
};
|
||||
|
||||
f->private = ops->alloc();
|
||||
if (!f->private) {
|
||||
free(f);
|
||||
return -ENOMEM;
|
||||
}
|
||||
|
||||
/* If we're being created on a fiber, default the callback to resuming that fiber on resolve —
|
||||
* this is almost always what you want, and it saves the usual set_callback boilerplate before
|
||||
* sd_fiber_suspend(). Callers that want different behavior can override with
|
||||
* sd_future_set_callback(). */
|
||||
sd_future *fiber = sd_fiber_get_current();
|
||||
if (fiber)
|
||||
(void) sd_future_set_callback(f, fiber_resume_trampoline, fiber);
|
||||
|
||||
*ret = f;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int sd_future_state(sd_future *f) {
|
||||
assert_return(f, -EINVAL);
|
||||
return f->state;
|
||||
}
|
||||
|
||||
int sd_future_result(sd_future *f) {
|
||||
assert_return(f, -EINVAL);
|
||||
assert_return(f->state == SD_FUTURE_RESOLVED, -EBUSY);
|
||||
return f->result;
|
||||
}
|
||||
|
||||
void* sd_future_get_userdata(sd_future *f) {
|
||||
assert_return(f, NULL);
|
||||
return f->userdata;
|
||||
}
|
||||
|
||||
void* sd_future_get_private(sd_future *f) {
|
||||
assert_return(f, NULL);
|
||||
return f->private;
|
||||
}
|
||||
|
||||
const sd_future_ops* sd_future_get_ops(sd_future *f) {
|
||||
assert_return(f, NULL);
|
||||
return f->ops;
|
||||
}
|
||||
|
||||
int sd_future_set_callback(sd_future *f, sd_future_func_t callback, void *userdata) {
|
||||
assert_return(f, -EINVAL);
|
||||
|
||||
f->callback = callback;
|
||||
f->userdata = userdata;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int sd_future_set_priority(sd_future *f, int64_t priority) {
|
||||
assert_return(f, -EINVAL);
|
||||
assert_return(f->state == SD_FUTURE_PENDING, -ESTALE);
|
||||
assert_return(f->ops->set_priority, -EOPNOTSUPP);
|
||||
|
||||
return f->ops->set_priority(f, priority);
|
||||
}
|
||||
|
||||
int sd_future_cancel(sd_future *f) {
|
||||
assert_return(f, -EINVAL);
|
||||
assert_return(f->ops->cancel, -EOPNOTSUPP);
|
||||
|
||||
if (f->state == SD_FUTURE_RESOLVED)
|
||||
return 0;
|
||||
|
||||
return f->ops->cancel(f);
|
||||
}
|
||||
|
||||
typedef struct WaitFuture {
|
||||
sd_future *target;
|
||||
} WaitFuture;
|
||||
|
||||
static void* wait_future_alloc(void) {
|
||||
return new0(WaitFuture, 1);
|
||||
}
|
||||
|
||||
static void wait_future_free(sd_future *f) {
|
||||
WaitFuture *wf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
|
||||
set_remove(wf->target->waiters, f);
|
||||
sd_future_unref(wf->target);
|
||||
free(wf);
|
||||
}
|
||||
|
||||
static int wait_future_cancel(sd_future *f) {
|
||||
WaitFuture *wf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
|
||||
|
||||
set_remove(wf->target->waiters, f);
|
||||
return sd_future_resolve(f, -ECANCELED);
|
||||
}
|
||||
|
||||
static const sd_future_ops wait_future_ops = {
|
||||
.size = sizeof(sd_future_ops),
|
||||
.alloc = wait_future_alloc,
|
||||
.free = wait_future_free,
|
||||
.cancel = wait_future_cancel,
|
||||
};
|
||||
|
||||
int sd_future_new_wait(sd_future *target, sd_future **ret) {
|
||||
int r;
|
||||
|
||||
assert_return(target, -EINVAL);
|
||||
assert_return(ret, -EINVAL);
|
||||
|
||||
_cleanup_(sd_future_unrefp) sd_future *f = NULL;
|
||||
r = sd_future_new(&wait_future_ops, &f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
WaitFuture *wf = sd_future_get_private(f);
|
||||
wf->target = sd_future_ref(target);
|
||||
|
||||
if (target->state == SD_FUTURE_RESOLVED)
|
||||
r = sd_future_resolve(f, target->result);
|
||||
else
|
||||
r = set_ensure_put(&target->waiters, &trivial_hash_ops, f);
|
||||
if (r < 0)
|
||||
return r;
|
||||
|
||||
*ret = TAKE_PTR(f);
|
||||
return 0;
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -69,6 +69,20 @@ typedef void (*_sd_destroy_t)(void *userdata);
|
||||
# define _SD_STRINGIFY(x) _SD_XSTRINGIFY(x)
|
||||
#endif
|
||||
|
||||
/* Mirror of CONCATENATE / UNIQ from macro-fundamental.h, available to public sd-* headers. */
|
||||
#ifndef _SD_CONCATENATE
|
||||
# define _SD_XCONCATENATE(x, y) x ## y
|
||||
# define _SD_CONCATENATE(x, y) _SD_XCONCATENATE(x, y)
|
||||
#endif
|
||||
|
||||
#ifndef _SD_UNIQ
|
||||
# ifdef __COUNTER__
|
||||
# define _SD_UNIQ __COUNTER__
|
||||
# else
|
||||
# define _SD_UNIQ __LINE__
|
||||
# endif
|
||||
#endif
|
||||
|
||||
#ifndef _SD_BEGIN_DECLARATIONS
|
||||
# ifdef __cplusplus
|
||||
# define _SD_BEGIN_DECLARATIONS \
|
||||
|
||||
@@ -36,6 +36,7 @@ _not_installed_headers = [
|
||||
'sd-dhcp6-option.h',
|
||||
'sd-dhcp6-protocol.h',
|
||||
'sd-dns-resolver.h',
|
||||
'sd-future.h',
|
||||
'sd-ipv4acd.h',
|
||||
'sd-ipv4ll.h',
|
||||
'sd-lldp-rx.h',
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
/* SPDX-License-Identifier: LGPL-2.1-or-later */
|
||||
#ifndef foosdfuturefoo
|
||||
#define foosdfuturefoo
|
||||
|
||||
/***
|
||||
systemd is free software; you can redistribute it and/or modify it
|
||||
under the terms of the GNU Lesser General Public License as published by
|
||||
the Free Software Foundation; either version 2.1 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
systemd is distributed in the hope that it will be useful, but
|
||||
WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
||||
Lesser General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU Lesser General Public License
|
||||
along with systemd; If not, see <https://www.gnu.org/licenses/>.
|
||||
***/
|
||||
|
||||
#include "_sd-common.h"
|
||||
|
||||
_SD_BEGIN_DECLARATIONS;
|
||||
|
||||
typedef struct sd_event sd_event;
|
||||
typedef struct sd_future sd_future;
|
||||
typedef struct sd_future_ops sd_future_ops;
|
||||
typedef int (*sd_future_func_t)(sd_future *f);
|
||||
typedef int (*sd_fiber_func_t)(void *userdata);
|
||||
typedef _sd_destroy_t sd_fiber_destroy_t;
|
||||
|
||||
struct sd_future_ops {
|
||||
size_t size;
|
||||
void* (*alloc)(void);
|
||||
void (*free)(sd_future *f);
|
||||
int (*cancel)(sd_future *f);
|
||||
int (*set_priority)(sd_future *f, int64_t priority);
|
||||
};
|
||||
|
||||
__extension__ typedef enum _SD_ENUM_TYPE_S64(sd_future_state_t) {
|
||||
SD_FUTURE_PENDING,
|
||||
SD_FUTURE_RESOLVED,
|
||||
_SD_ENUM_FORCE_S64(SD_FUTURE_STATE)
|
||||
} sd_future_state_t;
|
||||
|
||||
int sd_future_new(const sd_future_ops *ops, sd_future **ret);
|
||||
int sd_future_cancel(sd_future *f);
|
||||
int sd_future_resolve(sd_future *f, int result);
|
||||
|
||||
_SD_DECLARE_TRIVIAL_REF_UNREF_FUNC(sd_future);
|
||||
_SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_unref);
|
||||
void sd_future_unref_array_clear(sd_future *array[], size_t n);
|
||||
void sd_future_unref_array(sd_future *array[], size_t n);
|
||||
|
||||
sd_future* sd_future_cancel_wait_unref(sd_future *f);
|
||||
_SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_cancel_wait_unref);
|
||||
void sd_future_cancel_wait_unref_array_clear(sd_future *array[], size_t n);
|
||||
void sd_future_cancel_wait_unref_array(sd_future *array[], size_t n);
|
||||
|
||||
int sd_future_state(sd_future *f);
|
||||
int sd_future_result(sd_future *f);
|
||||
void* sd_future_get_userdata(sd_future *f);
|
||||
void* sd_future_get_private(sd_future *f);
|
||||
const sd_future_ops* sd_future_get_ops(sd_future *f);
|
||||
|
||||
int sd_future_set_callback(sd_future *f, sd_future_func_t callback, void *userdata);
|
||||
int sd_future_set_priority(sd_future *f, int64_t priority);
|
||||
|
||||
int sd_future_new_wait(sd_future *target, sd_future **ret);
|
||||
|
||||
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_set_floating(sd_future *f, int b);
|
||||
int sd_fiber_get_floating(sd_future *f);
|
||||
|
||||
int sd_fiber_is_running(void);
|
||||
sd_future* sd_fiber_get_current(void);
|
||||
int sd_fiber_get_priority(int64_t *ret);
|
||||
sd_event* sd_fiber_get_event(void);
|
||||
|
||||
int sd_fiber_yield(void);
|
||||
int sd_fiber_sleep(uint64_t usec);
|
||||
int sd_fiber_await(sd_future *target);
|
||||
int sd_fiber_suspend(void);
|
||||
int sd_fiber_resume(sd_future *f, int result);
|
||||
|
||||
sd_future* sd_fiber_timeout(uint64_t timeout);
|
||||
|
||||
#define SD_FIBER_TIMEOUT(timeout) _SD_FIBER_TIMEOUT(_SD_UNIQ, (timeout))
|
||||
#define _SD_FIBER_TIMEOUT(uniq, timeout) \
|
||||
sd_future *_SD_CONCATENATE(_sd_fto_, uniq) __attribute__((cleanup(sd_future_cancel_wait_unrefp), unused)) = sd_fiber_timeout(timeout)
|
||||
|
||||
#define SD_FIBER_WITH_TIMEOUT(timeout) _SD_FIBER_WITH_TIMEOUT(_SD_UNIQ, (timeout))
|
||||
#define _SD_FIBER_WITH_TIMEOUT(uniq, timeout) \
|
||||
for (sd_future *_SD_CONCATENATE(_sd_fto_, uniq) __attribute__((cleanup(sd_future_cancel_wait_unrefp), unused)) = sd_fiber_timeout(timeout), \
|
||||
*_SD_CONCATENATE(_sd_fto_b_, uniq) = (sd_future*) (uintptr_t) 1; \
|
||||
_SD_CONCATENATE(_sd_fto_b_, uniq); \
|
||||
_SD_CONCATENATE(_sd_fto_b_, uniq) = NULL)
|
||||
|
||||
_SD_END_DECLARATIONS;
|
||||
|
||||
#endif
|
||||
@@ -3,7 +3,7 @@
|
||||
integration_tests += [
|
||||
integration_test_template + {
|
||||
'name' : fs.name(meson.current_source_dir()),
|
||||
'coredump-exclude-regex' : '/(bash|python3.[0-9]+|systemd-executor)$',
|
||||
'coredump-exclude-regex' : '/(bash|python3.[0-9]+|systemd-executor|test-fiber)$',
|
||||
'cmdline' : integration_test_template['cmdline'] + [
|
||||
'''
|
||||
|
||||
|
||||
@@ -28,6 +28,7 @@ LIBC_LIB_PREFIXES: tuple[str, ...] = (
|
||||
'libm.so.',
|
||||
'libresolv.so.',
|
||||
'libc.musl-',
|
||||
'libucontext.',
|
||||
)
|
||||
|
||||
# GCC runtime support libraries (stack unwinding, soft-float helpers, C++ standard library). The
|
||||
|
||||
Reference in New Issue
Block a user