libxtc 0.4.0
Async concurrency for C: Tokio + Seastar + BEAM, in one library
Loading...
Searching...
No Matches
xtc_sync.h
1/*-
2 * Copyright (c) 2026, The XTC Project
3 * @stability stable
4 * Use of this source code is governed by the ISC License.
5 *
6 * src/inc/xtc_sync.h
7 * L3 synchronization primitives. M9 ships:
8 *
9 * notify Tokio-style one-shot wake-of-any-waiter
10 * semaphore counting; backpressure currency
11 * abort_source Seastar-style structured cancellation
12 *
13 * The full M9 surface (mutex, rwlock, barrier, gate) lands in
14 * M9.5 alongside the lock-manager work in M13. These three are
15 * the minimum needed for the M10 supervisor.
16 */
17
18#ifndef XTC_SYNC_H
19#define XTC_SYNC_H
20
21#include "xtc_export.h"
22
23#include <stdint.h>
24
25#include "xtc.h"
26#include "xtc_loop.h"
27
28/*
29 * Every blocking call below that parks a fiber (notify_wait,
30 * sem_acquire, amutex_lock, arwlock_rdlock/wrlock, barrier_wait,
31 * gate_drain) is kill-safe: a proc killed by xtc_exit_pid while parked
32 * leaves the wait queue, passes on anything already handed to it and
33 * cancels its deadline timer before it unwinds. A timed wait always
34 * cancels its deadline timer on return. See xtc_sync(3).
35 */
36
37/* ----- notify ----------------------------------------------------- */
38
39typedef struct xtc_notify xtc_notify_t;
40
41/*
42 * PUBLIC: int xtc_notify_create __P((xtc_notify_t **));
43 * PUBLIC: void xtc_notify_destroy __P((xtc_notify_t *));
44 * PUBLIC: int xtc_notify_signal __P((xtc_notify_t *));
45 * PUBLIC: int xtc_notify_wait __P((xtc_notify_t *, int64_t));
46 */
47XTC_API int xtc_notify_create(xtc_notify_t **out);
48XTC_API void xtc_notify_destroy(xtc_notify_t *n);
49
50/* Wake one waiter (the oldest parked fiber). If no fiber is waiting,
51 * the signal is "stored" and the next wait returns immediately.
52 * Subsequent signals before a wait collapse into one. */
53XTC_API int xtc_notify_signal(xtc_notify_t *n);
54
55/* Block (yield) the calling task until a signal arrives. timeout_ns
56 * < 0 = forever; 0 = non-blocking; > 0 = bounded. Returns XTC_E_AGAIN
57 * on timeout, XTC_OK on signal received. */
58XTC_API int xtc_notify_wait(xtc_notify_t *n, int64_t timeout_ns);
59
60/* ----- semaphore -------------------------------------------------- */
61
62typedef struct xtc_sem xtc_sem_t;
63
64/*
65 * PUBLIC: int xtc_sem_create __P((unsigned, xtc_sem_t **));
66 * PUBLIC: void xtc_sem_destroy __P((xtc_sem_t *));
67 * PUBLIC: int xtc_sem_post __P((xtc_sem_t *, unsigned));
68 * PUBLIC: int xtc_sem_acquire __P((xtc_sem_t *, unsigned, int64_t));
69 * PUBLIC: int xtc_sem_try_acquire __P((xtc_sem_t *, unsigned));
70 * PUBLIC: int xtc_sem_count __P((const xtc_sem_t *));
71 */
72XTC_API int xtc_sem_create(unsigned initial, xtc_sem_t **out);
73XTC_API void xtc_sem_destroy(xtc_sem_t *s);
74
75/* Add `n` units. Wakes the parked fibers the new count can satisfy,
76 * oldest first. */
77XTC_API int xtc_sem_post(xtc_sem_t *s, unsigned n);
78
79/* Take `n` units, blocking up to timeout_ns. A parked fiber stays
80 * queued until it takes its units, times out or is killed. */
81XTC_API int xtc_sem_acquire(xtc_sem_t *s, unsigned n, int64_t timeout_ns);
82
83/* Take `n` units, returning XTC_E_AGAIN immediately if not enough. */
84XTC_API int xtc_sem_try_acquire(xtc_sem_t *s, unsigned n);
85
86XTC_API int xtc_sem_count(const xtc_sem_t *s);
87
88/* ----- abort_source ----------------------------------------------- */
89
90typedef struct xtc_abort_source xtc_abort_source_t;
91typedef struct xtc_abort_token xtc_abort_token_t;
92/*
93 * Structured cancellation. An abort_source is owned by some parent
94 * (e.g. a supervisor) and produces tokens that children check.
95 * When the source is fired, every token answers true to is_aborted.
96 *
97 * PUBLIC: int xtc_abort_source_create __P((xtc_abort_source_t **));
98 * PUBLIC: void xtc_abort_source_destroy __P((xtc_abort_source_t *));
99 * PUBLIC: int xtc_abort_source_fire __P((xtc_abort_source_t *, int));
100 * PUBLIC: int xtc_abort_source_token __P((xtc_abort_source_t *, xtc_abort_token_t *));
101 *
102 * PUBLIC: int xtc_abort_token_is_aborted __P((const xtc_abort_token_t *));
103 * PUBLIC: int xtc_abort_token_reason __P((const xtc_abort_token_t *));
104 */
105XTC_API int xtc_abort_source_create(xtc_abort_source_t **out);
106XTC_API void xtc_abort_source_destroy(xtc_abort_source_t *s);
107
108/* Atomically fire the source with a reason code. All current and
109 * future tokens see is_aborted=true. */
110XTC_API int xtc_abort_source_fire(xtc_abort_source_t *s, int reason);
111
112/* Mint a token bound to the source. */
113XTC_API int xtc_abort_source_token(xtc_abort_source_t *s, xtc_abort_token_t *out);
114
115/*
116 * Public token shape (so callers can keep one on the stack). The
117 * implementation only reads fields documented here.
118 */
119struct xtc_abort_token {
120 xtc_abort_source_t *src;
121};
122
123XTC_API int xtc_abort_token_is_aborted(const xtc_abort_token_t *t);
124XTC_API int xtc_abort_token_reason(const xtc_abort_token_t *t);
125
126/* ----- mutex ------------------------------------------------------ */
127
128typedef struct xtc_amutex xtc_amutex_t;
129
130/*
131 * Async parking mutex. When free, the lock is a fast uncontended
132 * flag. When contended, a caller running inside a process /
133 * coroutine parks the fiber (yields to its loop) rather than blocking
134 * the OS thread, so a process can hold the lock across its own park
135 * (e.g. a blocking-pool offload) without wedging the loop when
136 * another process on that loop contends. Fiber waiters form a FIFO
137 * queue with direct hand-off (fair, no thundering herd); a caller
138 * that is not on a loop blocks on a condvar as a fallback.
139 *
140 * PUBLIC: int xtc_amutex_create __P((xtc_amutex_t **));
141 * PUBLIC: int xtc_amutex_create_ex __P((xtc_amutex_t **, unsigned));
142 * PUBLIC: xtc_amutex_t *xtc_amutex_static __P((unsigned));
143 * PUBLIC: void xtc_amutex_destroy __P((xtc_amutex_t *));
144 * PUBLIC: int xtc_amutex_lock __P((xtc_amutex_t *, int64_t));
145 * PUBLIC: int xtc_amutex_try_lock __P((xtc_amutex_t *));
146 * PUBLIC: int xtc_amutex_unlock __P((xtc_amutex_t *));
147 */
148XTC_API int xtc_amutex_create(xtc_amutex_t **out);
149
150/*
151 * Recursive variant: with XTC_AMUTEX_RECURSIVE the same owner may
152 * re-lock without deadlocking; the lock is released only when the
153 * matching number of unlocks have run. Ownership is tracked by FIBER
154 * identity on a loop (so two fibers sharing one OS thread are distinct
155 * owners) and by OS thread off a loop. Owner/count are maintained
156 * under the mutex's own lock, so the recursion accounting is race-free
157 * across loops.
158 */
159#define XTC_AMUTEX_RECURSIVE 0x1u
160XTC_API int xtc_amutex_create_ex(xtc_amutex_t **out, unsigned flags);
161
162/*
163 * Process-global static mutexes. Returns a stable, lazily-created
164 * recursive amutex for `slot` (0 .. XTC_AMUTEX_STATIC_MAX-1); repeated
165 * calls with the same slot return the same object. Never destroyed by
166 * the caller. Intended for adapters (e.g. a SQLite mutex vtable) that
167 * need named, never-freed mutexes.
168 */
169#define XTC_AMUTEX_STATIC_MAX 32u
170XTC_API xtc_amutex_t *xtc_amutex_static(unsigned slot);
171
172XTC_API void xtc_amutex_destroy(xtc_amutex_t *m);
173XTC_API int xtc_amutex_lock(xtc_amutex_t *m, int64_t timeout_ns);
174XTC_API int xtc_amutex_try_lock(xtc_amutex_t *m);
175XTC_API int xtc_amutex_unlock(xtc_amutex_t *m);
176
177/* ----- arwlock (parking reader/writer latch) --------------------- */
178
179typedef struct xtc_arwlock xtc_arwlock_t;
180
181/*
182 * Shared/exclusive latch whose contended waiters PARK the fiber
183 * (yield to the loop) rather than blocking the OS thread -- the
184 * reader/writer analogue of xtc_amutex. A holder may park on I/O
185 * while latched, and lock coupling may hold a parent latch across a
186 * child fix, without wedging a cooperative loop. Off a loop, waiters
187 * block on a condvar. FIFO fairness: an acquirer that finds a waiter
188 * queued ahead queues too, so a read stream cannot starve a writer.
189 * timeout_ns: <0 wait forever, 0 try, >0 deadline (XTC_E_AGAIN on
190 * timeout). Release with xtc_arwlock_unlock regardless of mode.
191 *
192 * PUBLIC: int xtc_arwlock_create __P((xtc_arwlock_t **));
193 * PUBLIC: void xtc_arwlock_destroy __P((xtc_arwlock_t *));
194 * PUBLIC: int xtc_arwlock_rdlock __P((xtc_arwlock_t *, int64_t));
195 * PUBLIC: int xtc_arwlock_wrlock __P((xtc_arwlock_t *, int64_t));
196 * PUBLIC: int xtc_arwlock_unlock __P((xtc_arwlock_t *));
197 */
198XTC_API int xtc_arwlock_create(xtc_arwlock_t **out);
199XTC_API void xtc_arwlock_destroy(xtc_arwlock_t *r);
200XTC_API int xtc_arwlock_rdlock(xtc_arwlock_t *r, int64_t timeout_ns);
201XTC_API int xtc_arwlock_wrlock(xtc_arwlock_t *r, int64_t timeout_ns);
202XTC_API int xtc_arwlock_unlock(xtc_arwlock_t *r);
203
204/* ----- rwlock ----------------------------------------------------- */
205
206typedef struct xtc_rwlock xtc_rwlock_t;
207
208/*
209 * Reader/writer lock with writer priority (writers don't starve).
210 *
211 * PUBLIC: int xtc_rwlock_create __P((xtc_rwlock_t **));
212 * PUBLIC: void xtc_rwlock_destroy __P((xtc_rwlock_t *));
213 * PUBLIC: int xtc_rwlock_rdlock __P((xtc_rwlock_t *, int64_t));
214 * PUBLIC: int xtc_rwlock_wrlock __P((xtc_rwlock_t *, int64_t));
215 * PUBLIC: int xtc_rwlock_unlock __P((xtc_rwlock_t *));
216 */
217XTC_API int xtc_rwlock_create(xtc_rwlock_t **out);
218XTC_API void xtc_rwlock_destroy(xtc_rwlock_t *r);
219XTC_API int xtc_rwlock_rdlock(xtc_rwlock_t *r, int64_t timeout_ns);
220XTC_API int xtc_rwlock_wrlock(xtc_rwlock_t *r, int64_t timeout_ns);
221XTC_API int xtc_rwlock_unlock(xtc_rwlock_t *r);
222
223/* ----- barrier ---------------------------------------------------- */
224
225typedef struct xtc_barrier xtc_barrier_t;
226
227/*
228 * N-task rendezvous. Reusable: after the Nth waiter arrives, all
229 * are released and the barrier resets to wait for another N.
230 *
231 * PUBLIC: int xtc_barrier_create __P((unsigned, xtc_barrier_t **));
232 * PUBLIC: void xtc_barrier_destroy __P((xtc_barrier_t *));
233 * PUBLIC: int xtc_barrier_wait __P((xtc_barrier_t *));
234 */
235XTC_API int xtc_barrier_create(unsigned n, xtc_barrier_t **out);
236XTC_API void xtc_barrier_destroy(xtc_barrier_t *b);
237XTC_API int xtc_barrier_wait(xtc_barrier_t *b);
238
239/* ----- gate ------------------------------------------------------- */
240
241typedef struct xtc_gate xtc_gate_t;
242
243/*
244 * Seastar-style gate: counts outstanding operations so callers can
245 * drain. enter/leave around each protected operation; close stops
246 * accepting new entries; drain blocks until count reaches zero.
247 *
248 * Pattern:
249 * xtc_gate_enter(g);
250 * do_work();
251 * xtc_gate_leave(g);
252 * ...
253 * xtc_gate_close(g);
254 * xtc_gate_drain(g, timeout_ns); // wait for in-flight ops to finish
255 *
256 * PUBLIC: int xtc_gate_create __P((xtc_gate_t **));
257 * PUBLIC: void xtc_gate_destroy __P((xtc_gate_t *));
258 * PUBLIC: int xtc_gate_enter __P((xtc_gate_t *));
259 * PUBLIC: int xtc_gate_leave __P((xtc_gate_t *));
260 * PUBLIC: int xtc_gate_close __P((xtc_gate_t *));
261 * PUBLIC: int xtc_gate_drain __P((xtc_gate_t *, int64_t));
262 * PUBLIC: int xtc_gate_count __P((const xtc_gate_t *));
263 */
264XTC_API int xtc_gate_create(xtc_gate_t **out);
265XTC_API void xtc_gate_destroy(xtc_gate_t *g);
266XTC_API int xtc_gate_enter(xtc_gate_t *g);
267XTC_API int xtc_gate_leave(xtc_gate_t *g);
268XTC_API int xtc_gate_close(xtc_gate_t *g);
269XTC_API int xtc_gate_drain(xtc_gate_t *g, int64_t timeout_ns);
270XTC_API int xtc_gate_count(const xtc_gate_t *g);
271
272#endif /* XTC_SYNC_H */
xtc_abort_token
Definition xtc_sync.h:119