An open's writes in a row to one place go in as one edit: a 10 MB body write is linear

Each body or data write copied and hashed the whole buffer, so a write the mount cut in 8 KB pieces was quadratic: 10 MB took 50 s and a 1 MB insert into 10 MB 11 s, holding the editor's turn. An open's appends to body, or inserts going on at data's address, are now held and put in as one splice (one copy, one undo step, one line-starts pass) before any other request, the close, or the editor's step once the writes pause 20 ms. fs.py-driven: 2 MB 2.19 -> 0.16 s, 10 MB 50.17 -> 0.78 s, 1 MB data into 10 MB 11.49 -> 0.26 s. A body-2m case (2 MB in 256 KB writes on one open) joins the perf gate: 84023 -> 15785 us; baselines re-recorded.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

2 files changed, 97 insertions, 2 deletions

diff --git a/src/ninep/pane.zig b/src/ninep/pane.zig
index d760bebd..ec6e8551 100644
--- a/src/ninep/pane.zig
+++ b/src/ninep/pane.zig
@@ -475,6 +475,55 @@ fn joinOnly(pane: *Pane, handle: u32) void {
     if (pane.fs.joined_by != 0 and pane.fs.joined_by != handle) pane.fs.joined = null;
 }
 
+/// How long writes in a row may pause before the editor's own step puts
+/// them in: long enough that a stream of 8 KB writes is one edit.
+pub const batch_pause_ns: u64 = 20 * std.time.ns_per_ms;
+
+/// The editor's step: the held writes go in once they have paused.
+pub fn flushPausedBatch(p: *Pardes) void {
+    const b = &p.fs.batch;
+    if (b.bytes.items.len == 0) return;
+    if (p.now_ns != 0 and p.now_ns -| b.last_ns < batch_pause_ns) return;
+    flushBatch(p);
+}
+
+/// Puts the held writes (`fs.batch`) in as one edit: before any other
+/// request, the open's release, or the editor's own step.
+pub fn flushBatch(p: *Pardes) void {
+    const b = &p.fs.batch;
+    if (b.bytes.items.len == 0) return;
+    defer b.bytes.clearRetainingCapacity();
+    const id = p.paneBySerial(b.serial) orelse return;
+    const pane = p.panes[id].?;
+    if (fileOf(pane) == null) return;
+    const before = dotOf(pane);
+    joinOnly(pane, b.handle);
+    const take = spliceBody(p, pane, b.at, b.at, b.bytes.items) orelse return;
+    pane.fs.joined = fileOf(pane).?.revision;
+    pane.fs.joined_by = b.handle;
+    setDot(pane, .{
+        .q0 = shiftOne(before.q0, clip(b.at), 0, clip(take)),
+        .q1 = shiftOne(before.q1, clip(b.at), 0, clip(take)),
+    });
+    if (!pane.fs.noscroll) showOffset(pane, b.at + take);
+}
+
+/// The held bytes a write at `at` on this open can go on from: its own
+/// batch, when it ends right at `at`; any other is put in first.
+fn batchFor(p: *Pardes, req: Req, pane: *Pane, at: usize) ?*@TypeOf(p.fs.batch) {
+    const b = &p.fs.batch;
+    const mine = b.serial == pane.serial and b.node == req.node and b.handle == req.handle;
+    if (b.bytes.items.len > 0 and !(mine and b.at + b.bytes.items.len == at)) flushBatch(p);
+    // Only an open's writes are held: one with no record (a test's, handle
+    // 0) is put in at once. A hold past 64 MiB is put in first.
+    if (req.handle == 0 or b.bytes.items.len > 64 << 20) {
+        flushBatch(p);
+        return null;
+    }
+    if (b.bytes.items.len == 0) b.* = .{ .serial = pane.serial, .node = req.node, .handle = req.handle, .at = at, .bytes = b.bytes };
+    return b;
+}
+
 fn writeBody(p: *Pardes, req: Req, id: usize, pane: *Pane) Reply {
     if (req.data.len == 0) return .{ .tag = req.tag, .written = 0 };
     if (pane.file == null) {
@@ -484,7 +533,14 @@ fn writeBody(p: *Pardes, req: Req, id: usize, pane: *Pane) Reply {
         p.emitWrite(id, req.data[0..take]);
         return .{ .tag = req.tag, .written = @intCast(take) };
     }
-    const at = bodyOf(pane).len;
+    const held = if (p.fs.batch.serial == pane.serial and p.fs.batch.node == req.node and p.fs.batch.handle == req.handle) p.fs.batch.bytes.items.len else 0;
+    const at = bodyOf(pane).len + held;
+    if (batchFor(p, req, pane, at)) |b| {
+        const take = wholeUtf8(req.data);
+        b.bytes.appendSlice(p.gpa, req.data[0..take]) catch return Reply.fail(req.tag, E.NOMEM);
+        b.last_ns = p.now_ns;
+        return .{ .tag = req.tag, .written = @intCast(take) };
+    }
     joinOnly(pane, req.handle);
     const take = spliceBody(p, pane, at, at, req.data) orelse
         return Reply.fail(req.tag, E.NOMEM);
@@ -538,9 +594,19 @@ fn writeData(p: *Pardes, req: Req, pane: *Pane) Reply {
     if (fileOf(pane) == null) return tree.failText(req.tag, E.INVAL, e_no_text);
     const pf = &pane.fs;
     if (pf.addr_failed) return tree.failText(req.tag, E.INVAL, e_addr_failed);
-    clampAddr(pf, bodyOf(pane).len);
+    const held = if (p.fs.batch.serial == pane.serial and p.fs.batch.node == req.node and p.fs.batch.handle == req.handle) p.fs.batch.bytes.items.len else 0;
+    clampAddr(pf, bodyOf(pane).len + held);
     const q0: usize = pf.addr.q0;
     const q1: usize = @max(q0, @as(usize, pf.addr.q1));
+    // An insert (an empty address) goes on from the open's held bytes.
+    if (q1 == q0) if (batchFor(p, req, pane, q0)) |b| {
+        const take = wholeUtf8(req.data);
+        b.bytes.appendSlice(p.gpa, req.data[0..take]) catch return Reply.fail(req.tag, E.NOMEM);
+        b.last_ns = p.now_ns;
+        pf.addr = .{ .q0 = clip(q0 + take), .q1 = clip(q0 + take) };
+        return .{ .tag = req.tag, .written = @intCast(take) };
+    };
+    flushBatch(p);
     const before = dotOf(pane);
     joinOnly(pane, req.handle);
     const take = spliceBody(p, pane, q0, q1, req.data) orelse
@@ -1097,6 +1163,32 @@ test "the writes of one open are one undo step, as bash's line-buffered printf m
     }
 }
 
+test "an open's writes in a row are held and go in as one edit, seen by the next request" {
+    const p = try withFile(testing.allocator, "one\n");
+    defer p.deinit();
+    const serial = serialOf(p);
+    const body = Node.of(serial, .body);
+    const h = call(p, .{ .tag = 1, .op = .open, .node = body, .omode = 1 }).reply.handle;
+    const revision = p.panes[0].?.file.?.revision;
+    for (0..100) |_| _ = call(p, .{ .tag = 2, .op = .write, .node = body, .handle = h, .data = "0123456789\n" });
+    // Held: nothing edited yet.
+    try testing.expectEqual(revision, p.panes[0].?.file.?.revision);
+    // Any other request puts them in, once.
+    try testing.expectEqual(@as(usize, 4 + 1100), rd(p, body, 0, 4096).bytes.len);
+    try testing.expectEqual(revision + 1, p.panes[0].?.file.?.revision);
+    _ = call(p, .{ .tag = 3, .op = .release, .node = body, .handle = h, .opened = true });
+    try testing.expectEqual(Status.ok, wr(p, Node.of(serial, .ctl), "Undo").reply.status);
+    try testing.expectEqualStrings("one\n", rd(p, body, 0, 64).bytes);
+    // data inserts at its address go on the same way.
+    const data = Node.of(serial, .data);
+    _ = wr(p, Node.of(serial, .addr), "#0");
+    const d = call(p, .{ .tag = 4, .op = .open, .node = data, .omode = 1 }).reply.handle;
+    _ = call(p, .{ .tag = 5, .op = .write, .node = data, .handle = d, .data = "a" });
+    _ = call(p, .{ .tag = 5, .op = .write, .node = data, .handle = d, .data = "b" });
+    _ = call(p, .{ .tag = 6, .op = .release, .node = data, .handle = d, .opened = true });
+    try testing.expectEqualStrings("abone\n", rd(p, body, 0, 64).bytes);
+}
+
 test "two opens writing one body are two undo steps, however their writes interleave" {
     const p = try withFile(testing.allocator, "one\n");
     defer p.deinit();
diff --git a/src/ninep/tree.zig b/src/ninep/tree.zig
index 0960f00b..6d09b9a6 100644
--- a/src/ninep/tree.zig
+++ b/src/ninep/tree.zig
@@ -454,6 +454,9 @@ pub fn stagedReply(p: *Pardes, req: Req) Reply {
 // ---- dispatch ----
 
 pub fn handle(p: *Pardes, req: Req) Reply {
+    // Held writes go in before anything but the next write of their open.
+    if (p.fs.batch.bytes.items.len > 0 and !(req.op == .write and req.node == p.fs.batch.node and req.handle == p.fs.batch.handle))
+        pane.flushBatch(p);
     if (req.op == .open and req.create) {
         if (Node.target(req.node)) |t| if (t == .pane and p.paneBySerial(t.pane.serial) == null)
             return Reply.fail(req.tag, E.NOENT);