1//! Memoizes key information about a file handle such as:
2//! * The size from calling stat, or the error that occurred therein.
3//! * The current seek position.
4//! * The error that occurred when trying to seek.
5//! * Whether reading should be done positionally or streaming.
6//! * Whether reading should be done via fd-to-fd syscalls (e.g. `sendfile`)
7//! versus plain variants (e.g. `read`).
8//!
9//! Fulfills the `Io.Reader` interface.
10const Reader = @This();
11
12const std = @import("../../std.zig");
13const Io = std.Io;
14const File = std.Io.File;
15const assert = std.debug.assert;
16
17io: Io,
18file: File,
19err: ?Error = null,
20mode: Mode = .positional,
21/// Tracks the true seek position in the file. To obtain the logical position,
22/// use `logicalPos`.
23pos: u64 = 0,
24size: ?u64 = null,
25size_err: ?SizeError = null,
26seek_err: ?SeekError = null,
27interface: Io.Reader,
28
29pub const Error = Io.Operation.FileReadStreaming.UnendingError || Io.Cancelable;
30
31pub const SizeError = File.StatError || error{
32 /// Occurs if, for example, the file handle is a network socket and therefore does not have a size.
33 Streaming,
34};
35
36pub const SeekError = File.SeekError || error{
37 /// Seeking fell back to reading, and reached the end before the requested seek position.
38 /// `pos` remains at the end of the file.
39 EndOfStream,
40 /// Seeking fell back to reading, which failed.
41 ReadFailed,
42};
43
44pub const Mode = enum {
45 streaming,
46 positional,
47 /// Avoid syscalls other than `read` and `readv`.
48 streaming_simple,
49 /// Avoid syscalls other than `pread` and `preadv`.
50 positional_simple,
51 /// Indicates reading cannot continue because of a seek failure.
52 failure,
53
54 pub fn toStreaming(m: @This()) @This() {
55 return switch (m) {
56 .positional, .streaming => .streaming,
57 .positional_simple, .streaming_simple => .streaming_simple,
58 .failure => .failure,
59 };
60 }
61
62 pub fn toSimple(m: @This()) @This() {
63 return switch (m) {
64 .positional, .positional_simple => .positional_simple,
65 .streaming, .streaming_simple => .streaming_simple,
66 .failure => .failure,
67 };
68 }
69};
70
71pub fn initInterface(buffer: []u8) Io.Reader {
72 return .{
73 .vtable = &.{
74 .stream = stream,
75 .discard = discard,
76 .readVec = readVec,
77 },
78 .buffer = buffer,
79 .seek = 0,
80 .end = 0,
81 };
82}
83
84pub fn init(file: File, io: Io, buffer: []u8) Reader {
85 return .{
86 .io = io,
87 .file = file,
88 .interface = initInterface(buffer),
89 };
90}
91
92pub fn initSize(file: File, io: Io, buffer: []u8, size: ?u64) Reader {
93 return .{
94 .io = io,
95 .file = file,
96 .interface = initInterface(buffer),
97 .size = size,
98 };
99}
100
101/// Positional is more threadsafe, since the global seek position is not
102/// affected, but when such syscalls are not available, preemptively
103/// initializing in streaming mode skips a failed syscall.
104pub fn initStreaming(file: File, io: Io, buffer: []u8) Reader {
105 return .{
106 .io = io,
107 .file = file,
108 .interface = Reader.initInterface(buffer),
109 .mode = .streaming,
110 .seek_err = error.Unseekable,
111 .size_err = error.Streaming,
112 };
113}
114
115pub fn getSize(r: *Reader) SizeError!u64 {
116 return r.size orelse {
117 if (r.size_err) |err| return err;
118 if (r.file.stat(r.io)) |st| {
119 if (st.kind == .file) {
120 r.size = st.size;
121 return st.size;
122 } else {
123 r.mode = r.mode.toStreaming();
124 r.size_err = error.Streaming;
125 return error.Streaming;
126 }
127 } else |err| {
128 r.size_err = err;
129 return err;
130 }
131 };
132}
133
134pub fn seekBy(r: *Reader, offset: i64) SeekError!void {
135 const io = r.io;
136 switch (r.mode) {
137 .positional, .positional_simple => {
138 setLogicalPos(r, @intCast(@as(i64, @intCast(logicalPos(r))) + offset));
139 },
140 .streaming, .streaming_simple => {
141 const seek_err = r.seek_err orelse e: {
142 if (io.vtable.fileSeekBy(io.userdata, r.file, offset)) |_| {
143 setLogicalPos(r, @intCast(@as(i64, @intCast(logicalPos(r))) + offset));
144 return;
145 } else |err| {
146 r.seek_err = err;
147 break :e err;
148 }
149 };
150 var remaining = std.math.cast(u64, offset) orelse return seek_err;
151 while (remaining > 0) {
152 remaining -= discard(&r.interface, .limited64(remaining)) catch |err| {
153 r.seek_err = err;
154 return err;
155 };
156 }
157 },
158 .failure => return r.seek_err.?,
159 }
160}
161
162/// Repositions logical read offset relative to the beginning of the file.
163pub fn seekTo(r: *Reader, offset: u64) SeekError!void {
164 const io = r.io;
165 switch (r.mode) {
166 .positional, .positional_simple => {
167 setLogicalPos(r, offset);
168 },
169 .streaming, .streaming_simple => {
170 const logical_pos = logicalPos(r);
171 if (offset >= logical_pos) return seekBy(r, @intCast(offset - logical_pos));
172 if (r.seek_err) |err| return err;
173 io.vtable.fileSeekTo(io.userdata, r.file, offset) catch |err| {
174 r.seek_err = err;
175 return err;
176 };
177 setLogicalPos(r, offset);
178 },
179 .failure => return r.seek_err.?,
180 }
181}
182
183pub fn logicalPos(r: *const Reader) u64 {
184 return r.pos - r.interface.bufferedLen();
185}
186
187fn setLogicalPos(r: *Reader, offset: u64) void {
188 const logical_pos = r.logicalPos();
189 if (offset < logical_pos or offset >= r.pos) {
190 r.interface.tossBuffered();
191 r.pos = offset;
192 } else r.interface.toss(@intCast(offset - logical_pos));
193}
194
195/// Number of slices to store on the stack, when trying to send as many byte
196/// vectors through the underlying read calls as possible.
197const max_buffers_len = 16;
198
199fn stream(io_reader: *Io.Reader, w: *Io.Writer, limit: Io.Limit) Io.Reader.StreamError!usize {
200 const r: *Reader = @alignCast(@fieldParentPtr("interface", io_reader));
201 return streamMode(r, w, limit, r.mode);
202}
203
204pub fn streamMode(r: *Reader, w: *Io.Writer, limit: Io.Limit, mode: Mode) Io.Reader.StreamError!usize {
205 switch (mode) {
206 .positional, .streaming => return w.sendFile(r, limit) catch |write_err| switch (write_err) {
207 error.Unimplemented => {
208 r.mode = r.mode.toSimple();
209 return 0;
210 },
211 else => |e| return e,
212 },
213 .positional_simple => {
214 const dest = limit.slice(try w.writableSliceGreedy(1));
215 var data: [1][]u8 = .{dest};
216 const n = try readVecPositional(r, &data);
217 w.advance(n);
218 return n;
219 },
220 .streaming_simple => {
221 const dest = limit.slice(try w.writableSliceGreedy(1));
222 var data: [1][]u8 = .{dest};
223 const n = try readVecStreaming(r, &data);
224 w.advance(n);
225 return n;
226 },
227 .failure => return error.ReadFailed,
228 }
229}
230
231fn readVec(io_reader: *Io.Reader, data: [][]u8) Io.Reader.Error!usize {
232 const r: *Reader = @alignCast(@fieldParentPtr("interface", io_reader));
233 switch (r.mode) {
234 .positional, .positional_simple => return readVecPositional(r, data),
235 .streaming, .streaming_simple => return readVecStreaming(r, data),
236 .failure => return error.ReadFailed,
237 }
238}
239
240fn readVecPositional(r: *Reader, data: [][]u8) Io.Reader.Error!usize {
241 const io = r.io;
242 var iovecs_buffer: [max_buffers_len][]u8 = undefined;
243 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, data);
244 const dest = iovecs_buffer[0..dest_n];
245 assert(dest[0].len > 0);
246 const n = io.vtable.fileReadPositional(io.userdata, r.file, dest, r.pos) catch |err| switch (err) {
247 error.Unseekable => {
248 r.mode = r.mode.toStreaming();
249 const pos = r.pos;
250 if (pos != 0) {
251 r.pos = 0;
252 r.seekBy(@intCast(pos)) catch {
253 r.mode = .failure;
254 return error.ReadFailed;
255 };
256 }
257 return 0;
258 },
259 else => |e| {
260 r.err = e;
261 return error.ReadFailed;
262 },
263 };
264 if (n == 0) {
265 r.size = r.pos;
266 return error.EndOfStream;
267 }
268 r.pos += n;
269 if (n > data_size) {
270 r.interface.end += n - data_size;
271 return data_size;
272 }
273 return n;
274}
275
276fn readVecStreaming(r: *Reader, data: [][]u8) Io.Reader.Error!usize {
277 const io = r.io;
278 var iovecs_buffer: [max_buffers_len][]u8 = undefined;
279 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, data);
280 const dest = iovecs_buffer[0..dest_n];
281 assert(dest[0].len > 0);
282 const n = r.file.readStreaming(io, dest) catch |err| switch (err) {
283 error.EndOfStream => {
284 r.size = r.pos;
285 return error.EndOfStream;
286 },
287 else => |e| {
288 r.err = e;
289 return error.ReadFailed;
290 },
291 };
292 r.pos += n;
293 if (n > data_size) {
294 r.interface.end += n - data_size;
295 return data_size;
296 }
297 return n;
298}
299
300fn discard(io_reader: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {
301 const r: *Reader = @alignCast(@fieldParentPtr("interface", io_reader));
302 const io = r.io;
303 const file = r.file;
304 switch (r.mode) {
305 .positional, .positional_simple => {
306 const size = r.getSize() catch {
307 r.mode = r.mode.toStreaming();
308 return 0;
309 };
310 const logical_pos = logicalPos(r);
311 const bytes_remaining = size - logical_pos;
312 if (bytes_remaining == 0) return error.EndOfStream;
313 const delta = @min(@backingInt(limit), bytes_remaining);
314 setLogicalPos(r, logical_pos + delta);
315 return delta;
316 },
317 .streaming, .streaming_simple => {
318 // Unfortunately we can't seek forward without knowing the
319 // size because the seek syscalls provided to us will not
320 // return the true end position if a seek would exceed the
321 // end.
322 fallback: {
323 if (r.size_err == null and r.seek_err == null) break :fallback;
324
325 const buffered_len = r.interface.bufferedLen();
326 var remaining = @backingInt(limit);
327 if (remaining <= buffered_len) {
328 r.interface.seek += remaining;
329 return remaining;
330 }
331 remaining -= buffered_len;
332 r.interface.seek = 0;
333 r.interface.end = 0;
334
335 var trash_buffer: [128]u8 = undefined;
336 var data: [1][]u8 = .{trash_buffer[0..@min(trash_buffer.len, remaining)]};
337 var iovecs_buffer: [max_buffers_len][]u8 = undefined;
338 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, &data);
339 const dest = iovecs_buffer[0..dest_n];
340 assert(dest[0].len > 0);
341 const n = file.readStreaming(io, dest) catch |err| switch (err) {
342 error.EndOfStream => {
343 r.size = r.pos;
344 return error.EndOfStream;
345 },
346 else => |e| {
347 r.err = e;
348 return error.ReadFailed;
349 },
350 };
351 r.pos += n;
352 if (n > data_size) {
353 r.interface.end += n - data_size;
354 remaining -= data_size;
355 } else {
356 remaining -= n;
357 }
358 return @backingInt(limit) - remaining;
359 }
360 const size = r.getSize() catch return 0;
361 const n = @min(size - r.pos, std.math.maxInt(i64), @backingInt(limit));
362 io.vtable.fileSeekBy(io.userdata, file, n) catch |err| {
363 r.seek_err = err;
364 return 0;
365 };
366 r.pos += n;
367 return n;
368 },
369 .failure => return error.ReadFailed,
370 }
371}
372
373/// Returns whether the stream is at the logical end.
374pub fn atEnd(r: *Reader) bool {
375 // Even if stat fails, size is set when end is encountered.
376 const size = r.size orelse return false;
377 return size - logicalPos(r) == 0;
378}