rows: the write worker charges a block in flight once, and a save that fits the staging permit waits instead of being skipped

CHANGELOG.md

@@ -4,6 +4,8 @@ Each release's page on GitHub has its notes and measurements. See [the 1.0.0 rel
 
 ## Unreleased
 
+- Conversation states saved to disk: a save that fits the staging permit (1 GiB and eight jobs) on its own now waits for a write in flight to finish instead of being skipped, and a block still being written is charged and copied once, however many saves name it (#1548, for #1538). A save larger than the permit is still skipped. So is a waiting save whose state leaves RAM, unless `--spill-gib` copies it then; that copy, like any made at an eviction, is skipped when the permit is full. Thanks to @wojo.
+
 - Qwen3.8-27B on CUDA: ModelOpt checkpoints (NVFP4 and FP8), such as `nvidia/Qwen3.8-27B-NVFP4`, now serve on cards from sm_80 to sm_89, such as a Jetson AGX Orin, where decode stopped at the first projection with `MissingLanePart`. The engine reserves the projections' partial sums for windows under 256 rows (about 21 MB for the 27B, and only on those cards) and reduces them without clusters, as Nemotron's ModelOpt path does; a card older than sm_80 is refused at load with the reason. It has not run on an sm_80 to sm_89 card yet. On sm_90 and newer, including the GB10, nothing changes (#1470, for #1452). Thanks to @javierbits and @valacroix.
 
 - More CUDA kernels take their `ldmatrix` loads and f16 `mma` from one home each: the prompt attention and Qwen 3.5 tensor-core attention take `qmm_frag`'s transposed load, the EXL3 grouped-expert, dense prompt and prompt kernels take its `ldmatrix4` and `ldmatrix2`, and the grouped EXL3 experts take `exl3_decode.cuh`'s f16 `mma16816`. All 1,935 kernel functions in the 14 CUDA images this reaches compile to the same code as before at sm_80, sm_86, sm_89, sm_120 and sm_121 (#1591). Thanks to @jschmied.

docs/api.md

@@ -604,9 +604,12 @@ cap below needs their room. After a restart, a crash or an eviction from RAM, th
 instead of reading the conversation again from its first token. Qwen3.8-27B, Nemotron, Flash Next and GLM-5.3-Flash on
 one Mac and Nemotron on CUDA keep them; other engines refuse the flag at startup.
 
-The engine stages an owned copy between rounds, and an I/O thread writes and fsyncs it. Staging has a 1 GiB byte
-permit and at most eight jobs; when either is full, the save is skipped. The copy waits for its GPU producers before
-RAM can be reused. A state becomes available for reuse only after all its files have been written and synced.
+The engine stages an owned copy between rounds, and an I/O thread writes and fsyncs it. Staging has a 1 GiB byte permit
+and at most eight jobs. A save that finds either full but would fit alone waits in the queue until a write finishes; one
+that cannot fit even then is skipped. So is a waiting save whose state leaves RAM, unless `--spill-gib` copies it then
+(below); that copy, like any made at an eviction, is skipped when the permit is full. A block still being written is
+charged and copied once, however many saves name it. The copy waits for its GPU producers before RAM can be reused. A
+state becomes available for reuse only after all its files have been written and synced.
 A planned stop (SIGTERM or SIGINT) drains writes for up to 20 seconds, then cancels the remainder; a blocked syscall
 can finish later using only its private temporary files and staged bytes. A killed server loses unfinished saves.
 Rows a stored state already holds for the same tokens are written once, so ten
@@ -656,11 +659,11 @@ publication.
 On CUDA, Nemotron, the Qwen3.5 hybrids and Flash Next take `--spill-gib` and `--checkpoint-slots` with the same store
 and cap as a Mac. At an eviction the engine thread stages a copy of the state into host memory, waiting for the GPU
 work that produced it, then releases its RAM entry and goes on; the I/O thread writes and fsyncs the copy while the
-engine decodes. Staging holds at most 1 GiB and eight jobs: a state whose new bytes (rows not already on disk) do not
-fit what is free is not saved, so one past 1 GiB never is, and a later request for it prefills. When the engine closes,
-at a planned stop (SIGTERM or SIGINT) or an idle unload, it saves the states still in RAM, newest first, within the same
-disk cap and 20-second stop budget. While other streams run, a queued request whose state is on disk has it read ahead
-on an I/O thread, as described above.
+engine decodes. Staging holds at most 1 GiB and eight jobs: a state whose new bytes (rows neither on disk nor being
+written) do not fit what is free is not saved, so one past 1 GiB never is, and a later request for it prefills. When the
+engine closes, at a planned stop (SIGTERM or SIGINT) or an idle unload, it saves the states still in RAM, newest first,
+within the same disk cap and 20-second stop budget. While other streams run, a queued request whose state is on disk has
+it read ahead on an I/O thread, as described above.
 
 Three routes let a client such as the Mac app ask about a conversation. Each takes the body of a chat completion
 (`messages`, `tools`, `chat_template_kwargs`), renders it as that request would be rendered, and answers on the engine's

zig/src/core/rows_async_fixture.zig

@@ -49,6 +49,15 @@ pub const Fixture = struct {
             rows.write.io.nap();
         }
     }
+
+    /// The engine's idle turns until `states` are on disk: a queued save is staged whenever `ready` says so.
+    pub fn idle(x: *Fixture, states: usize) !void {
+        const end = rows.write.nowNs() + 3 * std.time.ns_per_s;
+        while (x.r.states.items.len < states) {
+            if (rows.write.nowNs() >= end) return error.WriterTimedOut;
+            if (rows.write.ready(x.r)) _ = rows.write.step(x.r) else rows.write.io.nap();
+        }
+    }
 };
 
 pub fn prompt(salt: u32) [1100]u32 {
@@ -88,7 +97,7 @@ pub const Gate = struct {
     entered: std.atomic.Value(bool) = .init(false),
     release: std.atomic.Value(bool) = .init(false),
     left: std.atomic.Value(bool) = .init(false),
-    mode: enum { good, full, short, sync_failure } = .good,
+    mode: enum { good, full, short, sync_failure, first_fails } = .good,
     calls: usize = 0,
     sync_entered: std.atomic.Value(bool) = .init(false),
     sync_release: std.atomic.Value(bool) = .init(true),
@@ -103,7 +112,7 @@ pub const Gate = struct {
         while (!g.release.load(.acquire)) rows.write.io.nap();
         defer g.left.store(true, .release);
         g.calls += 1;
-        if (g.mode == .full) return error.NoSpaceLeft;
+        if (g.mode == .full or (g.mode == .first_fails and g.calls == 1)) return error.NoSpaceLeft;
         if (g.mode == .short and g.calls > 1) return 0;
         const n = platform.c.write(fd, bytes.ptr, if (g.mode == .short) @min(bytes.len, 7) else bytes.len);
         if (n < 0) return error.RowsWrite;

zig/src/core/rows_async_test.zig

@@ -347,7 +347,7 @@ test "spill I/O: killing a child inside a partial write leaves no published stat
     try std.testing.expectEqual(@as(usize, 0), fixture.files(reopened.dir, ".part"));
 }
 
-test "spill I/O: a full byte permit skips another save without waiting for the first job" {
+test "spill I/O: a full byte permit holds another save that fits it alone until the first job is written" {
     const x = try Fixture.open("full-permit", .finish, false, 1 << 20);
     defer x.close();
     var hold: Hold = .{};
@@ -362,13 +362,18 @@ test "spill I/O: a full byte permit skips another save without waiting for the f
     try x.keep(&second, 600);
     try std.testing.expectEqual(@as(usize, 1), x.r.worker.?.pending());
     try std.testing.expectEqual(used, x.r.worker.?.bytes());
-    try std.testing.expectEqual(error.StagingFull, x.r.failures.records[0].?.err);
+    try std.testing.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    try std.testing.expectEqual(@as(u64, 0), x.r.counts.skipped);
+    try std.testing.expectEqual(@as(u64, 0), x.r.failures.count);
+    try std.testing.expect(rows.write.pending(x.r) and !rows.write.ready(x.r) and !rows.write.step(x.r));
     hold.release.store(true, .release);
-    try x.settle();
-    try std.testing.expectEqual(@as(usize, 1), x.r.states.items.len);
+    try x.idle(2);
+    try std.testing.expectEqual(@as(usize, 0), x.r.queue.items.len);
+    try std.testing.expectEqual(@as(u64, 0), x.r.counts.skipped);
+    try std.testing.expectEqual(@as(u64, 2), x.r.counts.writes);
 }
 
-test "spill I/O: the job permit bounds metadata when many small saves queue behind a blocked write" {
+test "spill I/O: the job permit holds a ninth small save behind eight blocked writes until one is written" {
     const x = try Fixture.open("job-permit", .finish, false, 1 << 30);
     defer x.close();
     var hold: Hold = .{};
@@ -380,10 +385,11 @@ test "spill I/O: the job permit bounds metadata when many small saves queue behi
         try x.keep(p, 600);
     }
     try std.testing.expectEqual(@as(usize, 8), x.r.worker.?.pending());
-    try std.testing.expectEqual(@as(u64, 1), x.r.counts.skipped);
+    try std.testing.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    try std.testing.expectEqual(@as(u64, 0), x.r.counts.skipped);
     hold.release.store(true, .release);
-    try x.settle();
-    try std.testing.expectEqual(@as(usize, 8), x.r.states.items.len);
+    try x.idle(9);
+    try std.testing.expectEqual(@as(usize, 0), x.r.queue.items.len);
 }
 
 test "spill I/O: shutdown keeps a completed job's own filesystem error" {

zig/src/core/rows_fake.zig

@@ -33,6 +33,7 @@ pub const Fake = struct {
     slots: ?usize = null, // a family whose states each take one of so many slots (Nemotron CUDA's): past them, none
     gathers: ?usize = null, // copies out of kept states that succeed before every later one fails (null: all succeed)
     scatters: ?Fault = null, // copies into new states that succeed before every later one fails (null: all succeed)
+    gathered: usize = 0, // copies out of kept states made so far
 
     pub fn snapshots(f: *Fake) pc.Snapshots {
         return .{ .ptr = f, .vtable = &.{ .bytes = bytesFn, .save = saveFn, .restore = restoreFn, .drop = dropFn, .rows = &.{ .parts = partsFn, .gather = gatherFn, .blank = blankFn, .scatter = scatterFn } } };
@@ -134,6 +135,7 @@ pub const Fake = struct {
     }
     fn gatherFn(ptr: *anyopaque, saved: pc.Saved, ranges: []const image.Range, out: []u8) anyerror!void {
         const f = of(ptr);
+        f.gathered += 1;
         if (f.gathers) |*n| if (n.* == 0) return error.GatherFailed else {
             n.* -= 1;
         };

zig/src/core/rows_io.zig

@@ -36,13 +36,21 @@ pub const Job = struct {
     taken: bool = false,
     failure: ?anyerror = null,
 
+    /// The block files it writes: every file but the last, which is the state's own.
+    pub fn blocks(j: *const Job) []File {
+        return j.files[0..j.files.len -| 1];
+    }
+
     pub fn destroy(j: *Job) void {
         for (j.files) |*f| if (f.writer) |*w| w.abort();
         j.arena.deinit();
         allocator.destroy(j);
     }
 };
 
+/// The stems of the blocks jobs in flight are writing.
+pub const Promised = std.AutoHashMapUnmanaged(chain.Stem, void);
+
 pub const Worker = struct {
     dir: [:0]u8,
     io: file.IO,
@@ -78,6 +86,15 @@ pub const Worker = struct {
         return false;
     }
 
+    /// Whether the jobs in flight alone keep `amount` out: it fits an empty permit, and one is staged to be taken.
+    pub fn busy(w: *Worker, amount: u64) bool {
+        w.lock();
+        defer w.mutex.unlock();
+        if (w.stopped.load(.acquire) or amount > w.limit) return false;
+        for (w.slots) |s| if (s != null) return true;
+        return false;
+    }
+
     pub fn submit(w: *Worker, j: *Job) void {
         w.lock();
         defer w.mutex.unlock();
@@ -90,16 +107,32 @@ pub const Worker = struct {
         unreachable;
     }
 
-    /// Completed jobs stay owned here until the engine takes them, retaining their permits meanwhile.
+    /// Completed jobs stay owned here until the engine takes them, oldest first, retaining their permits meanwhile.
     pub fn take(w: *Worker) ?*Job {
         w.lock();
         defer w.mutex.unlock();
-        for (&w.slots) |*s| if (s.*) |j| if (j.done.load(.acquire)) {
-            s.* = null;
-            w.used -= j.staged;
-            return j;
+        const s = w.oldest(finished) orelse return null;
+        const j = s.*.?;
+        s.* = null;
+        w.used -= j.staged;
+        return j;
+    }
+
+    /// The slot of the earliest staged job `ok` accepts: ids rise as jobs are staged, whichever slot each took.
+    fn oldest(w: *Worker, comptime ok: fn (*const Job) bool) ?*?*Job {
+        var out: ?*?*Job = null;
+        for (&w.slots) |*s| if (s.*) |j| if (ok(j) and (out == null or j.id < out.?.*.?.id)) {
+            out = s;
         };
-        return null;
+        return out;
+    }
+
+    fn finished(j: *const Job) bool {
+        return j.done.load(.acquire);
+    }
+
+    fn untaken(j: *const Job) bool {
+        return !j.taken;
     }
 
     pub fn names(w: *Worker, name: chain.Name) bool {
@@ -112,9 +145,28 @@ pub const Worker = struct {
     pub fn wanted(w: *Worker, stem: chain.Stem) bool {
         w.lock();
         defer w.mutex.unlock();
-        for (w.slots) |s| if (s) |j| for (j.wants) |want| {
-            if (std.mem.eql(u8, want.name[0..16], &stem)) return true;
+        for (w.slots) |s| if (s) |j| if (lists(j, stem)) return true;
+        return false;
+    }
+
+    /// Replaces `set` with the blocks jobs that can still publish write; a job that skipped a block writes none of it.
+    pub fn promised(w: *Worker, gpa: std.mem.Allocator, set: *Promised) !void {
+        set.clearRetainingCapacity();
+        w.lock();
+        defer w.mutex.unlock();
+        const generation = w.generation.load(.acquire);
+        for (w.slots) |s| if (s) |j| if (j.generation == generation and !lost(j)) for (j.blocks()) |f| {
+            try set.put(gpa, chain.stem(f.trailer.name), {});
         };
+    }
+
+    /// Whether the job finished with an error (the worker sets `failure` before it publishes `done`).
+    fn lost(j: *const Job) bool {
+        return j.done.load(.acquire) and j.failure != null;
+    }
+
+    fn lists(j: *const Job, stem: chain.Stem) bool {
+        for (j.wants) |want| if (std.mem.eql(u8, want.name[0..16], &stem)) return true;
         return false;
     }
 
@@ -167,11 +219,9 @@ pub const Worker = struct {
     fn next(w: *Worker) ?*Job {
         w.lock();
         defer w.mutex.unlock();
-        for (w.slots) |s| if (s) |j| if (!j.taken) {
-            j.taken = true;
-            return j;
-        };
-        return null;
+        const j = (w.oldest(untaken) orelse return null).*.?;
+        j.taken = true;
+        return j;
     }
 
     fn run(w: *Worker) void {
@@ -274,3 +324,59 @@ fn compare(fd: c_int, bytes: []const u8, offset: *u64) !void {
         offset.* += got;
     }
 }
+
+const testing = std.testing;
+
+fn named(b: u8) chain.Name {
+    return @splat(b);
+}
+
+fn fileOf(b: u8, kind: file.Kind) File {
+    return .{ .trailer = .{ .kind = @backingInt(kind), .payload = 0, .payload_hash = 0, .name = named(b) }, .payload = &.{} };
+}
+
+/// A job as the worker's slots see it: an id, the blocks its state names and the files it writes (the last is its own).
+fn job(id: u64, wants: []Want, files: []File) Job {
+    return .{ .arena = undefined, .id = id, .generation = 1, .name = undefined, .at = 0, .shared = false, .tokens = &.{}, .spans = &.{}, .starts = &.{}, .wants = wants, .files = files, .bytes = 0, .staged = 1 };
+}
+
+fn threadless() Worker {
+    return .{ .dir = undefined, .io = .{}, .limit = 1 << 20, .thread = undefined };
+}
+
+test "rows io: jobs are run and taken in the order they were staged, whatever slot each holds" {
+    var w = threadless();
+    var jobs = [_]Job{ job(1, &.{}, &.{}), job(2, &.{}, &.{}), job(3, &.{}, &.{}), job(4, &.{}, &.{}) };
+    for (jobs[0..3]) |*j| w.submit(j);
+    try testing.expectEqual(&jobs[0], w.next().?);
+    jobs[0].done.store(true, .release);
+    try testing.expectEqual(&jobs[0], w.take().?);
+    w.submit(&jobs[3]); // the slot job 1 left: the lowest, though the newest
+    for (jobs[1..]) |*j| try testing.expectEqual(j, w.next().?);
+    try testing.expectEqual(@as(?*Job, null), w.next());
+    for (jobs[1..]) |*j| j.done.store(true, .release);
+    for (jobs[1..]) |*j| try testing.expectEqual(j, w.take().?);
+    try testing.expectEqual(@as(?*Job, null), w.take());
+}
+
+test "rows io: only the blocks a job that can still publish writes are promised" {
+    var w = threadless();
+    var wants = [_]Want{ .{ .name = named(1), .window = 0, .k = 0, .bytes = 1 }, .{ .name = named(2), .window = 0, .k = 1, .bytes = 1 } };
+    var writes = [_]File{ fileOf(1, .block), fileOf(2, .block), fileOf(8, .state) };
+    var skips = [_]File{ fileOf(3, .block), fileOf(9, .state) }; // it names blocks 1 and 2 and leaves them to the first job
+    var jobs = [_]Job{ job(1, &wants, &writes), job(2, &wants, &skips) };
+    for (&jobs) |*j| w.submit(j);
+    var set: Promised = .empty;
+    defer set.deinit(testing.allocator);
+    try w.promised(testing.allocator, &set);
+    try testing.expectEqual(@as(u32, 3), set.count());
+    for ([_]u8{ 1, 2, 3 }) |b| try testing.expect(set.contains(chain.stem(named(b))));
+    jobs[0].failure = error.NoSpaceLeft;
+    jobs[0].done.store(true, .release);
+    try w.promised(testing.allocator, &set);
+    try testing.expectEqual(@as(u32, 1), set.count());
+    try testing.expect(set.contains(chain.stem(named(3))));
+    w.fence();
+    try w.promised(testing.allocator, &set);
+    try testing.expectEqual(@as(u32, 0), set.count());
+}