authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-26 13:43:17-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-26 20:00:25-07:00
log04614d6ea17fff69ead42223c35a257da25462de
tree74ef210295b8cd105567c6a303aaacb984007a3a
parentde39c5f67ffec18b9a5aa51e84dbf5f04ce013bc

std.Io.Reader: add rebase to the vtable

This eliminates a footgun and special case handling with fixed buffers, as well as allowing decompression streams to keep a window in the output buffer.

3 files changed, 63 insertions(+), 42 deletions(-)

lib/std/Io.zig+1-1
...@@ -757,7 +757,7 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -757,7 +757,7 @@ pub fn Poller(comptime StreamEnum: type) type {
757 const unused = r.buffer[r.end..];757 const unused = r.buffer[r.end..];
758 if (unused.len >= min_len) return unused;758 if (unused.len >= min_len) return unused;
759 }759 }
760 if (r.seek > 0) r.rebase();760 if (r.seek > 0) r.rebase(r.buffer.len) catch unreachable;
761 {761 {
762 var list: std.ArrayListUnmanaged(u8) = .{762 var list: std.ArrayListUnmanaged(u8) = .{
763 .items = r.buffer[0..r.end],763 .items = r.buffer[0..r.end],
lib/std/Io/Reader.zig+36-37
...@@ -67,6 +67,18 @@ pub const VTable = struct {...@@ -67,6 +67,18 @@ pub const VTable = struct {
67 ///67 ///
68 /// This function is only called when `buffer` is empty.68 /// This function is only called when `buffer` is empty.
69 discard: *const fn (r: *Reader, limit: Limit) Error!usize = defaultDiscard,69 discard: *const fn (r: *Reader, limit: Limit) Error!usize = defaultDiscard,
70
71 /// Ensures `capacity` more data can be buffered without rebasing.
72 ///
73 /// Asserts `capacity` is within buffer capacity, or that the stream ends
74 /// within `capacity` bytes.
75 ///
76 /// Only called when `capacity` cannot fit into the unused capacity of
77 /// `buffer`.
78 ///
79 /// The default implementation moves buffered data to the start of
80 /// `buffer`, setting `seek` to zero, and cannot fail.
81 rebase: *const fn (r: *Reader, capacity: usize) RebaseError!void = defaultRebase,
70};82};
7183
72pub const StreamError = error{84pub const StreamError = error{
...@@ -97,6 +109,10 @@ pub const ShortError = error{...@@ -97,6 +109,10 @@ pub const ShortError = error{
97 ReadFailed,109 ReadFailed,
98};110};
99111
112pub const RebaseError = error{
113 EndOfStream,
114};
115
100pub const failing: Reader = .{116pub const failing: Reader = .{
101 .vtable = &.{117 .vtable = &.{
102 .stream = failingStream,118 .stream = failingStream,
...@@ -122,6 +138,7 @@ pub fn fixed(buffer: []const u8) Reader {...@@ -122,6 +138,7 @@ pub fn fixed(buffer: []const u8) Reader {
122 .vtable = &.{138 .vtable = &.{
123 .stream = endingStream,139 .stream = endingStream,
124 .discard = endingDiscard,140 .discard = endingDiscard,
141 .rebase = endingRebase,
125 },142 },
126 // This cast is safe because all potential writes to it will instead143 // This cast is safe because all potential writes to it will instead
127 // return `error.EndOfStream`.144 // return `error.EndOfStream`.
...@@ -780,11 +797,8 @@ pub fn peekDelimiterInclusive(r: *Reader, delimiter: u8) DelimiterError![]u8 {...@@ -780,11 +797,8 @@ pub fn peekDelimiterInclusive(r: *Reader, delimiter: u8) DelimiterError![]u8 {
780 @branchHint(.likely);797 @branchHint(.likely);
781 return buffer[seek .. end + 1];798 return buffer[seek .. end + 1];
782 }799 }
783 if (r.vtable.stream == &endingStream) {800 // TODO take a parameter for max search length rather than relying on buffer capacity
784 // Protect the `@constCast` of `fixed`.801 try rebase(r, r.buffer.len);
785 return error.EndOfStream;
786 }
787 r.rebase();
788 while (r.buffer.len - r.end != 0) {802 while (r.buffer.len - r.end != 0) {
789 const end_cap = r.buffer[r.end..];803 const end_cap = r.buffer[r.end..];
790 var writer: Writer = .fixed(end_cap);804 var writer: Writer = .fixed(end_cap);
...@@ -1050,11 +1064,7 @@ fn fillUnbuffered(r: *Reader, n: usize) Error!void {...@@ -1050,11 +1064,7 @@ fn fillUnbuffered(r: *Reader, n: usize) Error!void {
1050 };1064 };
1051 if (r.seek + n <= r.end) return;1065 if (r.seek + n <= r.end) return;
1052 };1066 };
1053 if (r.vtable.stream == &endingStream) {1067 try rebase(r, n);
1054 // Protect the `@constCast` of `fixed`.
1055 return error.EndOfStream;
1056 }
1057 rebaseCapacity(r, n);
1058 var writer: Writer = .{1068 var writer: Writer = .{
1059 .buffer = r.buffer,1069 .buffer = r.buffer,
1060 .vtable = &.{ .drain = Writer.fixedDrain },1070 .vtable = &.{ .drain = Writer.fixedDrain },
...@@ -1074,7 +1084,7 @@ fn fillUnbuffered(r: *Reader, n: usize) Error!void {...@@ -1074,7 +1084,7 @@ fn fillUnbuffered(r: *Reader, n: usize) Error!void {
1074///1084///
1075/// Asserts buffer capacity is at least 1.1085/// Asserts buffer capacity is at least 1.
1076pub fn fillMore(r: *Reader) Error!void {1086pub fn fillMore(r: *Reader) Error!void {
1077 rebaseCapacity(r, 1);1087 try rebase(r, 1);
1078 var writer: Writer = .{1088 var writer: Writer = .{
1079 .buffer = r.buffer,1089 .buffer = r.buffer,
1080 .end = r.end,1090 .end = r.end,
...@@ -1251,7 +1261,7 @@ pub fn takeLeb128(r: *Reader, comptime Result: type) TakeLeb128Error!Result {...@@ -1251,7 +1261,7 @@ pub fn takeLeb128(r: *Reader, comptime Result: type) TakeLeb128Error!Result {
12511261
1252pub fn expandTotalCapacity(r: *Reader, allocator: Allocator, n: usize) Allocator.Error!void {1262pub fn expandTotalCapacity(r: *Reader, allocator: Allocator, n: usize) Allocator.Error!void {
1253 if (n <= r.buffer.len) return;1263 if (n <= r.buffer.len) return;
1254 if (r.seek > 0) rebase(r);1264 if (r.seek > 0) rebase(r, r.buffer.len);
1255 var list: ArrayList(u8) = .{1265 var list: ArrayList(u8) = .{
1256 .items = r.buffer[0..r.end],1266 .items = r.buffer[0..r.end],
1257 .capacity = r.buffer.len,1267 .capacity = r.buffer.len,
...@@ -1297,37 +1307,20 @@ fn takeMultipleOf7Leb128(r: *Reader, comptime Result: type) TakeLeb128Error!Resu...@@ -1297,37 +1307,20 @@ fn takeMultipleOf7Leb128(r: *Reader, comptime Result: type) TakeLeb128Error!Resu
1297 }1307 }
1298}1308}
12991309
1300/// Left-aligns data such that `r.seek` becomes zero.1310/// Ensures `capacity` more data can be buffered without rebasing.
1301///1311pub fn rebase(r: *Reader, capacity: usize) RebaseError!void {
1302/// If `r.seek` is not already zero then `buffer` is mutated, making it illegal1312 if (r.end + capacity <= r.buffer.len) return;
1303/// to call this function with a const-casted `buffer`, such as in the case of1313 return r.vtable.rebase(r, capacity);
1304/// `fixed`. This issue can be avoided:1314}
1305/// * in implementations, by attempting a read before a rebase, in which1315
1306/// case the read will return `error.EndOfStream`, preventing the rebase.1316pub fn defaultRebase(r: *Reader, capacity: usize) RebaseError!void {
1307/// * in usage, by copying into a mutable buffer before initializing `fixed`.1317 if (r.end <= r.buffer.len - capacity) return;
1308pub fn rebase(r: *Reader) void {
1309 if (r.seek == 0) return;
1310 const data = r.buffer[r.seek..r.end];1318 const data = r.buffer[r.seek..r.end];
1311 @memmove(r.buffer[0..data.len], data);1319 @memmove(r.buffer[0..data.len], data);
1312 r.seek = 0;1320 r.seek = 0;
1313 r.end = data.len;1321 r.end = data.len;
1314}1322}
13151323
1316/// Ensures `capacity` more data can be buffered without rebasing, by rebasing
1317/// if necessary.
1318///
1319/// Asserts `capacity` is within the buffer capacity.
1320///
1321/// If the rebase occurs then `buffer` is mutated, making it illegal to call
1322/// this function with a const-casted `buffer`, such as in the case of `fixed`.
1323/// This issue can be avoided:
1324/// * in implementations, by attempting a read before a rebase, in which
1325/// case the read will return `error.EndOfStream`, preventing the rebase.
1326/// * in usage, by copying into a mutable buffer before initializing `fixed`.
1327pub fn rebaseCapacity(r: *Reader, capacity: usize) void {
1328 if (r.end > r.buffer.len - capacity) rebase(r);
1329}
1330
1331/// Advances the stream and decreases the size of the storage buffer by `n`,1324/// Advances the stream and decreases the size of the storage buffer by `n`,
1332/// returning the range of bytes no longer accessible by `r`.1325/// returning the range of bytes no longer accessible by `r`.
1333///1326///
...@@ -1683,6 +1676,12 @@ fn endingDiscard(r: *Reader, limit: Limit) Error!usize {...@@ -1683,6 +1676,12 @@ fn endingDiscard(r: *Reader, limit: Limit) Error!usize {
1683 return error.EndOfStream;1676 return error.EndOfStream;
1684}1677}
16851678
1679fn endingRebase(r: *Reader, capacity: usize) RebaseError!void {
1680 _ = r;
1681 _ = capacity;
1682 return error.EndOfStream;
1683}
1684
1686fn failingStream(r: *Reader, w: *Writer, limit: Limit) StreamError!usize {1685fn failingStream(r: *Reader, w: *Writer, limit: Limit) StreamError!usize {
1687 _ = r;1686 _ = r;
1688 _ = w;1687 _ = w;
lib/std/compress/zstd/Decompress.zig+26-4
...@@ -31,7 +31,12 @@ pub const Options = struct {...@@ -31,7 +31,12 @@ pub const Options = struct {
31 /// Verifying checksums is not implemented yet and will cause a panic if31 /// Verifying checksums is not implemented yet and will cause a panic if
32 /// you set this to true.32 /// you set this to true.
33 verify_checksum: bool = false,33 verify_checksum: bool = false,
34 /// Affects the minimum capacity of the provided buffer.34
35 /// The output buffer is asserted to have capacity for `window_len` plus
36 /// `zstd.block_size_max`.
37 ///
38 /// If `window_len` is too small, then some streams will fail to decompress
39 /// with `error.OutputBufferUndersize`.
35 window_len: u32 = zstd.default_window_len,40 window_len: u32 = zstd.default_window_len,
36};41};
3742
...@@ -69,8 +74,10 @@ pub const Error = error{...@@ -69,8 +74,10 @@ pub const Error = error{
69 WindowSizeUnknown,74 WindowSizeUnknown,
70};75};
7176
72/// If buffer that is written to is not big enough, some streams will fail with77/// When connecting `reader` to a `Writer`, `buffer` should be empty, and
73/// `error.OutputBufferUndersize`. A safe value is `zstd.default_window_len * 2`.78/// `Writer.buffer` capacity has requirements based on `Options.window_len`.
79///
80/// Otherwise, `buffer` has those requirements.
74pub fn init(input: *Reader, buffer: []u8, options: Options) Decompress {81pub fn init(input: *Reader, buffer: []u8, options: Options) Decompress {
75 return .{82 return .{
76 .input = input,83 .input = input,
...@@ -78,7 +85,10 @@ pub fn init(input: *Reader, buffer: []u8, options: Options) Decompress {...@@ -78,7 +85,10 @@ pub fn init(input: *Reader, buffer: []u8, options: Options) Decompress {
78 .verify_checksum = options.verify_checksum,85 .verify_checksum = options.verify_checksum,
79 .window_len = options.window_len,86 .window_len = options.window_len,
80 .reader = .{87 .reader = .{
81 .vtable = &.{ .stream = stream },88 .vtable = &.{
89 .stream = stream,
90 .rebase = rebase,
91 },
82 .buffer = buffer,92 .buffer = buffer,
83 .seek = 0,93 .seek = 0,
84 .end = 0,94 .end = 0,
...@@ -86,6 +96,18 @@ pub fn init(input: *Reader, buffer: []u8, options: Options) Decompress {...@@ -86,6 +96,18 @@ pub fn init(input: *Reader, buffer: []u8, options: Options) Decompress {
86 };96 };
87}97}
8898
99fn rebase(r: *Reader, capacity: usize) Reader.RebaseError!void {
100 const d: *Decompress = @alignCast(@fieldParentPtr("reader", r));
101 assert(capacity <= r.buffer.len - d.window_len);
102 assert(r.end + capacity > r.buffer.len);
103 const buffered = r.buffer[0..r.end];
104 const discard = buffered.len - d.window_len;
105 const keep = buffered[discard..];
106 @memmove(r.buffer[0..keep.len], keep);
107 r.end = keep.len;
108 r.seek -= discard;
109}
110
89fn stream(r: *Reader, w: *Writer, limit: Limit) Reader.StreamError!usize {111fn stream(r: *Reader, w: *Writer, limit: Limit) Reader.StreamError!usize {
90 const d: *Decompress = @alignCast(@fieldParentPtr("reader", r));112 const d: *Decompress = @alignCast(@fieldParentPtr("reader", r));
91 const in = d.input;113 const in = d.input;