std: rework zstd for new I/O API

[andrewrk]

Followup from #24329

It's basically the same logic, except reworked to use std.Io.Reader and std.Io.Writer and is therefore fully non-generic now.

Performance

master.zig

conststd=@import("std");
consttesting=std.testing;
constassert=std.debug.assert;
varwindow_buffer: [std.compress.zstd.DecompressorOptions.default_window_buffer_len*2]u8=undefined;
varfifo_buffer: [4096]u8=undefined;
pubfnmain() anyerror!void {
constinput_file=trystd.fs.openFileAbsolute("/home/andy/tmp/zig.zst", .{});
constoutput_file=trystd.fs.openFileAbsolute("/dev/null", .{ .mode=.write_only });
constin=input_file.deprecatedReader();
constout=output_file.deprecatedWriter();
vardecompress=std.compress.zstd.decompressor(in, .{
.window_buffer=&window_buffer,
.verify_checksum=false,
});
varfifo=std.fifo.LinearFifo(u8, .Slice).init(&fifo_buffer);
tryfifo.pump(decompress.reader(), out);
}

branch.zig

conststd=@import("std");
consttesting=std.testing;
constassert=std.debug.assert;
constzstd=std.compress.zstd;
varstdout_buffer: [zstd.default_window_len*2]u8=undefined;
varstdin_buffer: [4096]u8=undefined;
pubfnmain() anyerror!void {
constinput_file=trystd.fs.openFileAbsolute("/home/andy/tmp/zig.zst", .{});
constoutput_file=trystd.fs.openFileAbsolute("/dev/null", .{ .mode=.write_only });
varstdin_reader=input_file.reader(&stdin_buffer);
vardecompress: zstd.Decompress= .init(&stdin_reader.interface, &.{}, .{
.verify_checksum=false,
});
varstdout_writer=output_file.writer(&stdout_buffer);
_=trydecompress.reader.streamRemaining(&stdout_writer.interface);
trystdout_writer.interface.flush();
}

Benchmark 1 (8 runs): ./master
measurement mean ± σ min … max outliers delta
wall_time 1.94s ± 30.0ms 1.90s … 1.99s 0 ( 0%) 0%
peak_rss 16.4MB ± 271KB 16.0MB … 16.8MB 0 ( 0%) 0%
cpu_cycles 10.2G ± 132M 10.1G … 10.5G 1 (13%) 0%
instructions 33.6G ± 279 33.6G … 33.6G 0 ( 0%) 0%
cache_references 22.1M ± 8.17M 18.4M … 42.2M 1 (13%) 0%
cache_misses 536K ± 61.9K 471K … 644K 0 ( 0%) 0%
branch_misses 83.9M ± 112K 83.6M … 84.0M 1 (13%) 0%
Benchmark 2 (9 runs): ./branch
measurement mean ± σ min … max outliers delta
wall_time 1.83s ± 9.97ms 1.81s … 1.85s 0 ( 0%) ⚡- 5.3% ± 1.2%
peak_rss 16.7MB ± 255KB 16.3MB … 17.0MB 0 ( 0%) + 1.6% ± 1.7%
cpu_cycles 9.95G ± 52.8M 9.91G … 10.0G 0 ( 0%) ⚡- 2.3% ± 1.0%
instructions 33.7G ± 75.6 33.7G … 33.7G 0 ( 0%) + 0.4% ± 0.0%
cache_references 24.9M ± 1.22M 23.5M … 26.4M 0 ( 0%) + 12.8% ± 26.5%
cache_misses 639K ± 48.9K 581K … 722K 0 ( 0%) 💩+ 19.1% ± 10.7%
branch_misses 83.3M ± 27.8K 83.3M … 83.4M 0 ( 0%) - 0.7% ± 0.1%

[squeek502]

I think some clarification on what "buffer" refers to in this doc comment and that of Options.window_len would be helpful. It's not currently clear that the buffer parameter of init is used for the Reader, but the buffer in the doc comment (and Options.window_len) refers to the Writer.

[squeek502]

Seems to be a regression for many inputs where MalformedSequence is being erroneously hit.

Test code for master branch:

conststd=@import("std");
test"decompress" {
constwindow_buffer=trystd.testing.allocator.alloc(u8, std.compress.zstd.DecompressorOptions.default_window_buffer_len);
deferstd.testing.allocator.free(window_buffer);
constdata=@embedFile("input.zst");
varin_stream=std.io.fixedBufferStream(data);
varzstd_stream=std.compress.zstd.decompressor(in_stream.reader(), .{ .window_buffer=window_buffer });
constresult=tryzstd_stream.reader().readAllAlloc(std.testing.allocator, std.math.maxInt(usize));
deferstd.testing.allocator.free(result);
trystd.testing.expectEqualSlices(u8, @embedFile("input.orig"), result);
}

Test code for this branch:

conststd=@import("std");
test"decompress" {
constallocator=std.testing.allocator;
varout: std.ArrayListUnmanaged(u8) =.empty;
deferout.deinit(allocator);
tryout.ensureUnusedCapacity(allocator, std.compress.zstd.default_window_len);
constdata=@embedFile("input.zst");
varin: std.io.Reader= .fixed(data);
varzstd_stream: std.compress.zstd.Decompress= .init(&in, &.{}, .{});
tryzstd_stream.reader.appendRemaining(allocator, null, &out, .unlimited);
trystd.testing.expectEqualSlices(u8, @embedFile("input.orig"), out.items);
}

Files that reproduce the error:

zstd-repro-20250725.zip

The error:

$ zig test writergate.zig
1/1 writergate.test.decompress...FAIL (ReadFailed)
/home/ryan/Programming/zig/zig/lib/std/math.zig:578:21: 0x106cb17 in sub__anon_7902 (std.zig)
if (ov[1] != 0) return error.Overflow;
^
/home/ryan/Programming/zig/zig/lib/std/compress/zstd/Decompress.zig:649:21: 0x10507b1 in decodeSequence (std.zig)
return error.MalformedSequence;
^
/home/ryan/Programming/zig/zig/lib/std/compress/zstd/Decompress.zig:197:42: 0x1052012 in readInFrame (std.zig)
bytes_written += try decode.decodeSequence(w.buffer, write_pos + bytes_written, &bit_stream);
^
/home/ryan/Programming/zig/zig/lib/std/compress/zstd/Decompress.zig:116:21: 0x1053d90 in stream (std.zig)
return error.ReadFailed;
^
/home/ryan/Programming/zig/zig/lib/std/Io/Reader.zig:310:37: 0x102e44e in appendRemaining__anon_2084 (std.zig)
error.ReadFailed => return error.ReadFailed,
^
/home/ryan/Programming/zig/tmp/zstd/writergate.zig:12:5: 0x1029c8e in test.decompress (writergate.zig)
try zstd_stream.reader.appendRemaining(allocator, null, &out, .unlimited);
^
0 passed; 0 skipped; 1 failed.

This seems to affect a lot of the generated files from zstd's decodecorpus tool which should all be able to be decoded without error.

[andrewrk]

Sorry, I need to make the buffer size capacity requirements more clear. The output buffer is too small. Minimum is zstd.default_window_len + zstd.block_size_max.

[squeek502]

Unless I'm misunderstanding, upping the output buffer size doesn't fix the MalformedSequence error.

[andrewrk]

appendRemaining has:

trylist.ensureUnusedCapacity(gpa, 1);

The 1 here does not guarantee enough buffer capacity. appendRemaining needs to take a parameter that tells how many bytes to increase unused capacity by in order to be viable.

So then your test code needs:

tryzstd_stream.reader.appendRemaining(allocator, null, &out, .unlimited, zstd.block_size_max);

[squeek502]

Okay, switched to Writer.Allocating and this test does pass:

conststd=@import("std");
test"decompress" {
constallocator=std.testing.allocator;
constdata=@embedFile("input.zst");
varin: std.io.Reader= .fixed(data);
varzstd_stream: std.compress.zstd.Decompress= .init(&in, &.{}, .{
.window_len=std.compress.zstd.default_window_len,
});
varout: std.io.Writer.Allocating= .init(allocator);
deferout.deinit();
tryout.ensureUnusedCapacity(std.compress.zstd.default_window_len);
constlen=tryzstd_stream.reader.streamRemaining(&out.writer);
trystd.testing.expectEqual(@embedFile("input.orig").len, len);
trystd.testing.expectEqualSlices(u8, @embedFile("input.orig"), out.getWritten());
}

(but note that the out.ensureUnusedCapacity(std.compress.zstd.default_window_len) is required to avoid error.WindowOversize with this setup)

With this setup, generated inputs from zstd's decodecorpus tool are all passing as expected on this branch, so confirmed not a regression but more like a potential footgun.

[andrewrk]

With this setup, generated inputs from zstd's decodecorpus tool are all passing as expected on this branch, so confirmed not a regression but more like a potential footgun.

Yep I hear you loud and clear.

[andrewrk]

Going to tack on documentation improvements and appendRemaining improvements in a follow-up branch.

[L3P3]

The cache_ benchmarks got way worse. Why and what to do?

[rohlem]

@L3P3 The numbers all describe aspects of the same single benchmark.
The new branch is faster overall, so the individual numbers aren't a reason to worry.

Details:

The way I understand it, wall_time tells us how long the benchmarked code ran.
Here the new code is better (was faster), so this branch is an improvement compared to before.

The other numbers tell us aspects of how the code ran, what the CPU/system did in this time.
If the wall_time was worse (slower), the numbers could tell us why it was worse - i.e. which aspects of the code we could try to fix/change.
However, since the time improved, the results tell us
"even though we have more cache references, and more cache misses, the program was faster than before".
So there's no reason to worry about the changed internal numbers.

There's always room to try to make any code faster - in this case, we could investigate these changed values as a start.
However, the code structure changed significantly in this branch - the program now does things differently - so it's totally expected that the internal numbers changed.
The merged branch is still an improvement in overall time, so the new approach is currently more efficient than the previous one.

[squeek502]

In my experience, cache usage metrics are also much more variable than the others, and much more affected by other things happening on the machine.