Skip to content

Commit 7b3f6f4

Browse files
committed
fix(gc): worker registration parks under a pending collect (ADR-0175)
Smell-audited: 3: ADR-0175 landed in the preceding commit (DA fork embedded); the DA's finished-form pick adopted — TooManyThreads run-unregistered fallthrough closed in future/thread workers + shared workerContext builder; alloc-side assert deferred as D-566 (survey recorded); registeredCountRelaxed renamed (it is seq_cst now) The spawn-to-register window: stopWorld computed its park target from registered_count, so a drainer/future/Thread worker spawned moments earlier ran unregistered while the collector walked thread roots. registerThread now publishes (atomic release slots), then parks if gc_requested is set; the flag-store/count-read vs count-add/flag-load pairs are seq_cst (Dekker — do not relax); the walk reads slots with acquire loads. New deterministic unit test (red pre-fix). Post-fix the e2e drain case runs 600/600 clean on a 4-vCPU-pinned x86_64 host (pre-fix 3/221, 4/600); D-548 (a) future/promise SIGABRT measured unchanged — distinct race, stays open.
1 parent f94a88e commit 7b3f6f4

5 files changed

Lines changed: 184 additions & 64 deletions

File tree

‎src/runtime/agent.zig‎

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -365,15 +365,9 @@ fn enqueueDirect(rt: *Runtime, agent_val: Value, action: Action) !void {
365365
/// its operand-stack roots (the in-flight action), like `future.worker`.
366366
fn drainer(a: *Agent) void {
367367
const agent_val = Value.encodeHeapPtr(.agent, a);
368-
var ctx: root_set.ThreadGcContext = .{
369-
.frame_slot = &env_mod.current_frame,
370-
.analysis_frame_slot = &root_set.analysis_frame_head,
371-
.eval_frame_slot = &root_set.eval_frame_head,
372-
.self_guard_slot = &root_set.gc_self_guard,
373-
// Publish this drainer's STM transaction (an action may run a `dosync`)
374-
// so it is GC-rooted during a collect (#4a' in-txn-map rooting).
375-
.tx_slot = @ptrCast(&lock_tx.current_tx),
376-
};
368+
// The tx_slot publishes this drainer's STM transaction (an action may run
369+
// a `dosync`) so it is GC-rooted during a collect (#4a' in-txn-map rooting).
370+
var ctx = root_set.workerContext(@ptrCast(&lock_tx.current_tx));
377371
// Must NOT drain while unregistered: an unregistered worker's operand stack
378372
// is invisible to the mark phase, so a concurrent collect (when auto-collect
379373
// turns ON) would sweep objects live only on this thread → use-after-free. On

‎src/runtime/concurrency/safepoint.zig‎

Lines changed: 60 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,10 @@ var resume_cond: std.Io.Condition = .init;
5858

5959
/// Worker safe point: register as parked, wake the collector, then block until
6060
/// the pending collection finishes. Called when `gc_requested` is observed set
61-
/// at the `alloc` boundary or the VM back-edge poll. Signalling `all_parked` on
61+
/// at the `alloc` boundary, the VM back-edge poll, or the registration
62+
/// safepoint at the tail of `root_set.registerThread` (ADR-0175 — a worker
63+
/// born while a collect is arming/running parks before its first mutator
64+
/// step). Signalling `all_parked` on
6265
/// every entry (not only the last) is robust against the collector checking the
6366
/// count before this thread parks. The resume guard tests the FLAG, not the
6467
/// count, so a spurious `Condition` wakeup re-parks correctly.
@@ -96,7 +99,11 @@ pub fn stopWorld(self_registered: bool) void {
9699
if (waited > max_stopworld_wait_ns.load(.monotonic))
97100
max_stopworld_wait_ns.store(waited, .monotonic);
98101
}
99-
gc_requested.store(true, .release);
102+
// Seq_cst, pairing with `registerThread`'s seq_cst count-add + flag-load
103+
// (ADR-0175): a worker mid-registration is either visible in the count read
104+
// below (we wait for its park) or sees this flag and parks at its
105+
// registration safepoint — never both-miss (the spawn-to-register window).
106+
gc_requested.store(true, .seq_cst);
100107
io_default.lockMutex(&sp_mutex);
101108
defer io_default.unlockMutex(&sp_mutex);
102109
// Recompute the target on every wake: a worker that finishes a tiny action
@@ -107,7 +114,7 @@ pub fn stopWorld(self_registered: bool) void {
107114
// no `registry_mutex`-under-`sp_mutex` inversion. A late-arriving worker
108115
// raises the count and parks at its first poll; an overshoot relaxes `<`.
109116
while (true) {
110-
const registered = root_set.registeredCountRelaxed();
117+
const registered = root_set.registeredCountLockFree();
111118
const target: u32 = if (self_registered and registered > 0) registered - 1 else registered;
112119
if (parked_count >= target) return;
113120
io_default.condWait(&all_parked, &sp_mutex);
@@ -359,6 +366,56 @@ test "collectStopTheWorld with no other registered worker is a fenced collect (D
359366
try testing.expect(!gc_requested.load(.acquire));
360367
}
361368

369+
test "registerThread during an active STW collect parks the newborn until resumeWorld (spawn-to-register window)" {
370+
var tio: ThreadedIo = .{};
371+
tio.start();
372+
defer tio.deinit();
373+
374+
const Shared = struct {
375+
/// True between stopWorld-return and resumeWorld on the main thread —
376+
/// the span where NO registered worker may run mutator code.
377+
var in_stw: std.atomic.Value(bool) = .init(false);
378+
var ran_during_collect: std.atomic.Value(bool) = .init(false);
379+
var registered_ret: std.atomic.Value(bool) = .init(false);
380+
381+
fn worker() void {
382+
var ctx: root_set.ThreadGcContext = .{
383+
.frame_slot = &env_mod.current_frame,
384+
.analysis_frame_slot = &root_set.analysis_frame_head,
385+
.eval_frame_slot = &root_set.eval_frame_head,
386+
.self_guard_slot = &root_set.gc_self_guard,
387+
};
388+
root_set.registerThread(&ctx) catch return;
389+
// First mutator step after registration: if the collector is still
390+
// between stopWorld and resumeWorld here, the newborn raced the
391+
// collect (the spawn-to-register window).
392+
if (in_stw.load(.seq_cst)) ran_during_collect.store(true, .seq_cst);
393+
registered_ret.store(true, .release);
394+
root_set.unregisterThread(&ctx);
395+
}
396+
};
397+
Shared.in_stw.store(false, .monotonic);
398+
Shared.ran_during_collect.store(false, .monotonic);
399+
Shared.registered_ret.store(false, .monotonic);
400+
401+
// Arm + complete the rendezvous with ZERO registered workers (target 0 →
402+
// returns at once) — the collector is now "collecting".
403+
stopWorld(false);
404+
Shared.in_stw.store(true, .seq_cst);
405+
406+
// A worker born mid-collect must park inside registerThread until
407+
// resumeWorld, never run its first mutator step during the collect.
408+
var t = try std.Thread.spawn(.{}, Shared.worker, .{});
409+
while (parkedCountForTest() == 0 and !Shared.registered_ret.load(.acquire)) std.atomic.spinLoopHint();
410+
411+
Shared.in_stw.store(false, .seq_cst);
412+
resumeWorld();
413+
t.join();
414+
415+
try testing.expect(!Shared.ran_during_collect.load(.seq_cst));
416+
try testing.expectEqual(@as(u32, 0), parkedCountForTest());
417+
}
418+
362419
test "collectStopTheWorld parks real workers allocating through gc.alloc, then resumes them (D-244 #4)" {
363420
var tio: ThreadedIo = .{};
364421
tio.start();

‎src/runtime/future.zig‎

Lines changed: 27 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@ const Value = value_mod.Value;
3131
const HeapHeader = value_mod.HeapHeader;
3232
const Runtime = @import("runtime.zig").Runtime;
3333
const Env = @import("env.zig").Env;
34-
const env_mod = @import("env.zig");
3534
const root_set = @import("gc/root_set.zig");
3635
const io_default = @import("concurrency/io_default.zig");
3736
const lock_tx = @import("concurrency/lock_tx.zig");
@@ -136,15 +135,9 @@ pub fn alloc(rt: *Runtime, env: *Env, thunk: Value, loc: SourceLocation) !Value
136135
/// slots (no conveyed dynamic bindings yet — binding conveyance is a follow-up).
137136
fn worker(f: *Future) void {
138137
const fut_val = Value.encodeHeapPtr(.future, f);
139-
var ctx: root_set.ThreadGcContext = .{
140-
.frame_slot = &env_mod.current_frame,
141-
.analysis_frame_slot = &root_set.analysis_frame_head,
142-
.eval_frame_slot = &root_set.eval_frame_head,
143-
.self_guard_slot = &root_set.gc_self_guard,
144-
// Publish this worker's STM transaction so a `dosync` in the thunk is
145-
// GC-rooted during a collect (#4a' in-txn-map rooting).
146-
.tx_slot = @ptrCast(&lock_tx.current_tx),
147-
};
138+
// The tx_slot publishes this worker's STM transaction so a `dosync` in the
139+
// thunk is GC-rooted during a collect (#4a' in-txn-map rooting).
140+
var ctx = root_set.workerContext(@ptrCast(&lock_tx.current_tx));
148141
const registered = if (root_set.registerThread(&ctx)) |_| true else |_| false;
149142
defer if (registered) root_set.unregisterThread(&ctx);
150143

@@ -155,16 +148,24 @@ fn worker(f: *Future) void {
155148

156149
var result_state: FutureState = .realised_error;
157150
var result_value: Value = .nil_val;
158-
if (f.rt.vtable) |vt| {
159-
if (vt.callFn(f.rt, f.env, f.thunk, &.{}, .{})) |result| {
160-
result_state = .realised_value;
161-
result_value = result;
162-
} else |_| {
163-
// ADR-0120: marshal the worker's error into a GC-heap exception
164-
// Value (survives this thread) so `deref` re-raises the REAL error
165-
// (kind/message/location), not a generic `future_thunk_failed`.
166-
result_state = .realised_error;
167-
result_value = worker_error.capture(f.rt);
151+
// ADR-0175: NEVER run the thunk unregistered (registry cap hit) — an
152+
// unregistered mutator is invisible to the STW rendezvous + root walk,
153+
// the exact corruption the registration safepoint closes. The future
154+
// realises as an error WITHOUT allocating (this thread may not touch the
155+
// GC heap): `cached` stays nil, so `deref` raises `future_thunk_failed`
156+
// (errorValue's nil-guard routes it past the reraise).
157+
if (registered) {
158+
if (f.rt.vtable) |vt| {
159+
if (vt.callFn(f.rt, f.env, f.thunk, &.{}, .{})) |result| {
160+
result_state = .realised_value;
161+
result_value = result;
162+
} else |_| {
163+
// ADR-0120: marshal the worker's error into a GC-heap exception
164+
// Value (survives this thread) so `deref` re-raises the REAL error
165+
// (kind/message/location), not a generic `future_thunk_failed`.
166+
result_state = .realised_error;
167+
result_value = worker_error.capture(f.rt);
168+
}
168169
}
169170
}
170171

@@ -241,7 +242,12 @@ pub fn errorValue(v: Value) ?Value {
241242
const f = v.decodePtr(*Future);
242243
io_default.lockMutex(&f.cell.mutex);
243244
defer io_default.unlockMutex(&f.cell.mutex);
244-
return if (f.state == .realised_error) f.cached else null;
245+
// Nil-guard: a future that failed WITHOUT a marshalled exception (worker
246+
// registration refused at the registry cap — that path may not allocate
247+
// one) has `cached == nil`; report "no exception value" so the deref
248+
// consumer raises the generic `future_thunk_failed` instead of
249+
// re-raising nil (ADR-0175).
250+
return if (f.state == .realised_error and !f.cached.isNil()) f.cached else null;
245251
}
246252

247253
/// Wait up to `timeout_ms` for the worker (the 3-arity `deref` support).

‎src/runtime/gc/root_set.zig‎

Lines changed: 81 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -331,7 +331,7 @@ var registry_mutex: std.Io.Mutex = .init;
331331

332332
/// Live registered-worker count, mirroring the non-null slots in
333333
/// `thread_registry`. Maintained under `registry_mutex` (register/unregister)
334-
/// but exposed lock-free via `registeredCountRelaxed` so `safepoint.stopWorld`
334+
/// but exposed lock-free via `registeredCountLockFree` so `safepoint.stopWorld`
335335
/// can re-read the rendezvous target while holding `sp_mutex` — taking
336336
/// `registry_mutex` under `sp_mutex` would invert the lock order against the
337337
/// unregister path (registry_mutex → sp_mutex wake). The array scan in
@@ -351,20 +351,69 @@ var registered_count: std.atomic.Value(u32) = .init(0);
351351
/// dormant D-244 #4 path, validated separately under user awareness.
352352
pub threadlocal var is_registered_worker: bool = false;
353353

354+
/// Build the CALLING thread's standard worker context: all four TLS root
355+
/// slots plus the caller-supplied opaque `tx_slot` (opaque because root_set
356+
/// must not import lock_tx — see `ThreadGcContext.tx_slot`). Must run ON the
357+
/// worker thread (the addresses taken are its threadlocals). One shared
358+
/// builder so a future worker family cannot silently drop a slot (the
359+
/// slot-drift hazard the ADR-0175 review named).
360+
pub fn workerContext(tx_slot: *const ?*anyopaque) ThreadGcContext {
361+
return .{
362+
.frame_slot = &env_mod.current_frame,
363+
.analysis_frame_slot = &analysis_frame_head,
364+
.eval_frame_slot = &eval_frame_head,
365+
.self_guard_slot = &gc_self_guard,
366+
.tx_slot = tx_slot,
367+
};
368+
}
369+
354370
/// Register a worker thread's published roots. Locked (workers register
355371
/// at `Thread.spawn`). Returns `error.TooManyThreads` past the cap.
372+
///
373+
/// Registration is itself a GC SAFEPOINT (ADR-0175). A worker spawned while a
374+
/// stop-the-world collect was arming is invisible to that collect's rendezvous
375+
/// target (it was not yet in the registry), so `stopWorld` can return with
376+
/// target 0 and start walking roots while this thread comes to life. Without
377+
/// the post-publish park below, the newborn would run mutator code (push
378+
/// binding/eval frames, pop an agent action) CONCURRENTLY with the collector's
379+
/// `nextThreadRoots` walk — the spawn-to-register window (the
380+
/// phase16_gc_torture agent-drain GP fault on the 4-vCPU Linux runner).
381+
///
382+
/// The seq_cst pairing makes the handshake airtight (Dekker shape): the
383+
/// collector stores `gc_requested` then reads `registered_count`; this thread
384+
/// adds `registered_count` then loads `gc_requested`. Under seq_cst it is
385+
/// impossible that BOTH the collector misses this thread's count AND this
386+
/// thread misses the collector's flag — so either the collector waits for our
387+
/// park, or we park here before touching the heap. The park runs AFTER
388+
/// `registry_mutex` is released (a parked thread must never hold it — peers
389+
/// registering and the next walk would block behind a thread the collector
390+
/// will not wake). A freshly registered ctx publishes only empty TLS roots, so
391+
/// a collector already mid-walk reading our slot sees an empty contribution.
392+
///
393+
/// CONTRACT for callers: (a) no lock may be held at the call site (the park
394+
/// must not strand a lock the collector's world could need); (b) code run
395+
/// BEFORE this call must touch only spawner-pinned objects and pure bit-ops
396+
/// (an unregistered thread's roots are invisible to the collector); (c) on
397+
/// `TooManyThreads` the worker must NOT run its mutator body — it reports its
398+
/// domain failure without GC allocation and exits (an unregistered mutator is
399+
/// exactly the corruption this safepoint closes).
356400
pub fn registerThread(ctx: *ThreadGcContext) error{TooManyThreads}!void {
357-
io_default.lockMutex(&registry_mutex);
358-
defer io_default.unlockMutex(&registry_mutex);
359-
for (&thread_registry) |*slot| {
360-
if (slot.* == null) {
361-
slot.* = ctx;
362-
is_registered_worker = true;
363-
_ = registered_count.fetchAdd(1, .release);
364-
return;
365-
}
401+
{
402+
io_default.lockMutex(&registry_mutex);
403+
defer io_default.unlockMutex(&registry_mutex);
404+
const idx: usize = blk: {
405+
for (&thread_registry, 0..) |*slot, i| {
406+
if (slot.* == null) break :blk i;
407+
}
408+
return error.TooManyThreads;
409+
};
410+
// Release-store so a lock-free walker (`threadContextAt`) that sees the
411+
// slot also sees the fully-initialised ctx it points at.
412+
@atomicStore(?*ThreadGcContext, &thread_registry[idx], ctx, .release);
413+
is_registered_worker = true;
414+
_ = registered_count.fetchAdd(1, .seq_cst);
366415
}
367-
return error.TooManyThreads;
416+
if (safepoint.gc_requested.load(.seq_cst)) safepoint.park();
368417
}
369418

370419
/// Deregister a worker thread's context (at `Thread.join`). No-op if absent.
@@ -376,8 +425,8 @@ pub fn unregisterThread(ctx: *ThreadGcContext) void {
376425
defer io_default.unlockMutex(&registry_mutex);
377426
for (&thread_registry) |*slot| {
378427
if (slot.* == ctx) {
379-
slot.* = null;
380-
_ = registered_count.fetchSub(1, .release);
428+
@atomicStore(?*ThreadGcContext, slot, null, .release);
429+
_ = registered_count.fetchSub(1, .seq_cst);
381430
removed = true;
382431
break;
383432
}
@@ -394,9 +443,13 @@ pub fn unregisterThread(ctx: *ThreadGcContext) void {
394443
}
395444

396445
/// Lock-free snapshot of the registered-worker count for `safepoint.stopWorld`'s
397-
/// rendezvous-target recompute (see `registered_count`).
398-
pub fn registeredCountRelaxed() u32 {
399-
return registered_count.load(.acquire);
446+
/// rendezvous-target recompute (see `registered_count` — taking `registry_mutex`
447+
/// under `sp_mutex` would invert the unregister path's lock order). Seq_cst: the
448+
/// collector reads this AFTER its seq_cst `gc_requested` store, pairing with
449+
/// `registerThread`'s seq_cst count-add + flag-load so a registering newborn is
450+
/// either counted here or parks there (ADR-0175 spawn-window handshake).
451+
pub fn registeredCountLockFree() u32 {
452+
return registered_count.load(.seq_cst);
400453
}
401454

402455
/// Count of currently-registered worker contexts (test/introspection).
@@ -417,8 +470,11 @@ pub fn registeredThreadCount() usize {
417470
/// single-thread collect sees an empty registry → no-op. `markTxFn` receives the
418471
/// opaque `current_tx` value (the worker's `?*LockingTransaction`, reinterpreted).
419472
pub fn markRegisteredTxs(context: *anyopaque, markTxFn: *const fn (*anyopaque, ?*anyopaque) void) void {
420-
for (thread_registry) |slot| {
421-
if (slot) |ctx| markTxFn(context, ctx.tx_slot.*);
473+
for (&thread_registry) |*slot| {
474+
// Acquire-load: a newborn may register mid-collect (it then parks at
475+
// its registration safepoint with all-empty roots — ADR-0175), so the
476+
// slot read must be atomic against that concurrent release-store.
477+
if (@atomicLoad(?*ThreadGcContext, slot, .acquire)) |ctx| markTxFn(context, ctx.tx_slot.*);
422478
}
423479
}
424480

@@ -428,9 +484,12 @@ pub fn markRegisteredTxs(context: *anyopaque, markTxFn: *const fn (*anyopaque, ?
428484
// worker), read through its published TLS pointers. `.end` terminates the walk
429485
// past the registry cap. A sparse (null) registry slot yields an all-empty
430486
// contribution and the cursor advances. Reading the registry array during the
431-
// walk is safe because (a) it is empty until Phase-B real threads (#3a/#3b are
432-
// runtime-inert), and (b) the #3b-step2 safepoint guarantees no concurrent
433-
// register/unregister during collect.
487+
// walk is safe because (a) slot reads are atomic acquire-loads paired with the
488+
// register/unregister release-stores, (b) a newborn registering mid-collect
489+
// parks at its registration safepoint BEFORE running any mutator code, so a
490+
// mid-walk registration contributes only its empty TLS roots (ADR-0175), and
491+
// (c) unregistration cannot happen mid-collect — the rendezvous completed, so
492+
// every already-registered worker is parked (or blocked-counted).
434493
const ThreadRoots = struct {
435494
frame_head: ?*env_mod.BindingFrame,
436495
eval_head: ?*EvalFrame,
@@ -447,7 +506,7 @@ fn threadContextAt(idx: usize) ThreadSource {
447506
} };
448507
const ri = idx - 1;
449508
if (ri >= MAX_GC_THREADS) return .end;
450-
if (thread_registry[ri]) |ctx| return .{ .roots = .{
509+
if (@atomicLoad(?*ThreadGcContext, &thread_registry[ri], .acquire)) |ctx| return .{ .roots = .{
451510
.frame_head = ctx.frame_slot.*,
452511
.eval_head = ctx.eval_frame_slot.*,
453512
.self_guard = ctx.self_guard_slot.*,

0 commit comments

Comments
 (0)