struct Reader [src]

Memoizes key information about a file handle such as: The size from calling stat, or the error that occurred therein. The current seek position. The error that occurred when trying to seek. Whether reading should be done positionally or streaming. Whether reading should be done via fd-to-fd syscalls (e.g. sendfile) versus plain variants (e.g. read). Fulfills the std.Io.Reader interface.
Fields

file: File
err: ?ReadError = null
mode: Reader.Mode = .positional
pos: u64 = 0Tracks the true seek position in the file. To obtain the logical position, use logicalPos.
size: ?u64 = null
size_err: ?SizeError = null
seek_err: ?Reader.SeekError = null
interface: std.Io.Reader
Members

atEnd (Function)
getSize (Function)
init (Function)
initInterface (Function)
initSize (Function)
initStreaming (Function)
logicalPos (Function)
Mode (enum)
read (Function)
readPositional (Function)
readStreaming (Function)
seekBy (Function)
SeekError (Error Set)
seekTo (Function)
SizeError (Error Set)
Source

  pub const Reader = struct {
    file: File,
    err: ?ReadError = null,
    mode: Reader.Mode = .positional,
    /// Tracks the true seek position in the file. To obtain the logical
    /// position, use `logicalPos`.
    pos: u64 = 0,
    size: ?u64 = null,
    size_err: ?SizeError = null,
    seek_err: ?Reader.SeekError = null,
    interface: std.Io.Reader,

    pub const SizeError = std.os.windows.GetFileSizeError || StatError || error{
        /// Occurs if, for example, the file handle is a network socket and therefore does not have a size.
        Streaming,
    };

    pub const SeekError = File.SeekError || error{
        /// Seeking fell back to reading, and reached the end before the requested seek position.
        /// `pos` remains at the end of the file.
        EndOfStream,
        /// Seeking fell back to reading, which failed.
        ReadFailed,
    };

    pub const Mode = enum {
        streaming,
        positional,
        /// Avoid syscalls other than `read` and `readv`.
        streaming_reading,
        /// Avoid syscalls other than `pread` and `preadv`.
        positional_reading,
        /// Indicates reading cannot continue because of a seek failure.
        failure,

        pub fn toStreaming(m: @This()) @This() {
            return switch (m) {
                .positional, .streaming => .streaming,
                .positional_reading, .streaming_reading => .streaming_reading,
                .failure => .failure,
            };
        }

        pub fn toReading(m: @This()) @This() {
            return switch (m) {
                .positional, .positional_reading => .positional_reading,
                .streaming, .streaming_reading => .streaming_reading,
                .failure => .failure,
            };
        }
    };

    pub fn initInterface(buffer: []u8) std.Io.Reader {
        return .{
            .vtable = &.{
                .stream = Reader.stream,
                .discard = Reader.discard,
                .readVec = Reader.readVec,
            },
            .buffer = buffer,
            .seek = 0,
            .end = 0,
        };
    }

    pub fn init(file: File, buffer: []u8) Reader {
        return .{
            .file = file,
            .interface = initInterface(buffer),
        };
    }

    pub fn initSize(file: File, buffer: []u8, size: ?u64) Reader {
        return .{
            .file = file,
            .interface = initInterface(buffer),
            .size = size,
        };
    }

    /// Positional is more threadsafe, since the global seek position is not
    /// affected, but when such syscalls are not available, preemptively
    /// initializing in streaming mode skips a failed syscall.
    pub fn initStreaming(file: File, buffer: []u8) Reader {
        return .{
            .file = file,
            .interface = Reader.initInterface(buffer),
            .mode = .streaming,
            .seek_err = error.Unseekable,
            .size_err = error.Streaming,
        };
    }

    pub fn getSize(r: *Reader) SizeError!u64 {
        return r.size orelse {
            if (r.size_err) |err| return err;
            if (is_windows) {
                if (windows.GetFileSizeEx(r.file.handle)) |size| {
                    r.size = size;
                    return size;
                } else |err| {
                    r.size_err = err;
                    return err;
                }
            }
            if (posix.Stat == void) {
                r.size_err = error.Streaming;
                return error.Streaming;
            }
            if (stat(r.file)) |st| {
                if (st.kind == .file) {
                    r.size = st.size;
                    return st.size;
                } else {
                    r.mode = r.mode.toStreaming();
                    r.size_err = error.Streaming;
                    return error.Streaming;
                }
            } else |err| {
                r.size_err = err;
                return err;
            }
        };
    }

    pub fn seekBy(r: *Reader, offset: i64) Reader.SeekError!void {
        switch (r.mode) {
            .positional, .positional_reading => {
                setPosAdjustingBuffer(r, @intCast(@as(i64, @intCast(r.pos)) + offset));
            },
            .streaming, .streaming_reading => {
                if (posix.SEEK == void) {
                    r.seek_err = error.Unseekable;
                    return error.Unseekable;
                }
                const seek_err = r.seek_err orelse e: {
                    if (posix.lseek_CUR(r.file.handle, offset)) |_| {
                        setPosAdjustingBuffer(r, @intCast(@as(i64, @intCast(r.pos)) + offset));
                        return;
                    } else |err| {
                        r.seek_err = err;
                        break :e err;
                    }
                };
                var remaining = std.math.cast(u64, offset) orelse return seek_err;
                while (remaining > 0) {
                    remaining -= discard(&r.interface, .limited64(remaining)) catch |err| {
                        r.seek_err = err;
                        return err;
                    };
                }
                r.interface.seek = 0;
                r.interface.end = 0;
            },
            .failure => return r.seek_err.?,
        }
    }

    pub fn seekTo(r: *Reader, offset: u64) Reader.SeekError!void {
        switch (r.mode) {
            .positional, .positional_reading => {
                setPosAdjustingBuffer(r, offset);
            },
            .streaming, .streaming_reading => {
                if (offset >= r.pos) return Reader.seekBy(r, @intCast(offset - r.pos));
                if (r.seek_err) |err| return err;
                posix.lseek_SET(r.file.handle, offset) catch |err| {
                    r.seek_err = err;
                    return err;
                };
                setPosAdjustingBuffer(r, offset);
            },
            .failure => return r.seek_err.?,
        }
    }

    pub fn logicalPos(r: *const Reader) u64 {
        return r.pos - r.interface.bufferedLen();
    }

    fn setPosAdjustingBuffer(r: *Reader, offset: u64) void {
        const logical_pos = logicalPos(r);
        if (offset < logical_pos or offset >= r.pos) {
            r.interface.seek = 0;
            r.interface.end = 0;
            r.pos = offset;
        } else {
            const logical_delta: usize = @intCast(offset - logical_pos);
            r.interface.seek += logical_delta;
        }
    }

    /// Number of slices to store on the stack, when trying to send as many byte
    /// vectors through the underlying read calls as possible.
    const max_buffers_len = 16;

    fn stream(io_reader: *std.Io.Reader, w: *std.Io.Writer, limit: std.Io.Limit) std.Io.Reader.StreamError!usize {
        const r: *Reader = @alignCast(@fieldParentPtr("interface", io_reader));
        switch (r.mode) {
            .positional, .streaming => return w.sendFile(r, limit) catch |write_err| switch (write_err) {
                error.Unimplemented => {
                    r.mode = r.mode.toReading();
                    return 0;
                },
                else => |e| return e,
            },
            .positional_reading => {
                const dest = limit.slice(try w.writableSliceGreedy(1));
                const n = try readPositional(r, dest);
                w.advance(n);
                return n;
            },
            .streaming_reading => {
                const dest = limit.slice(try w.writableSliceGreedy(1));
                const n = try readStreaming(r, dest);
                w.advance(n);
                return n;
            },
            .failure => return error.ReadFailed,
        }
    }

    fn readVec(io_reader: *std.Io.Reader, data: [][]u8) std.Io.Reader.Error!usize {
        const r: *Reader = @alignCast(@fieldParentPtr("interface", io_reader));
        switch (r.mode) {
            .positional, .positional_reading => {
                if (is_windows) {
                    // Unfortunately, `ReadFileScatter` cannot be used since it
                    // requires page alignment.
                    if (io_reader.seek == io_reader.end) {
                        io_reader.seek = 0;
                        io_reader.end = 0;
                    }
                    const first = data[0];
                    if (first.len >= io_reader.buffer.len - io_reader.end) {
                        return readPositional(r, first);
                    } else {
                        io_reader.end += try readPositional(r, io_reader.buffer[io_reader.end..]);
                        return 0;
                    }
                }
                var iovecs_buffer: [max_buffers_len]posix.iovec = undefined;
                const dest_n, const data_size = try io_reader.writableVectorPosix(&iovecs_buffer, data);
                const dest = iovecs_buffer[0..dest_n];
                assert(dest[0].len > 0);
                const n = posix.preadv(r.file.handle, dest, r.pos) catch |err| switch (err) {
                    error.Unseekable => {
                        r.mode = r.mode.toStreaming();
                        const pos = r.pos;
                        if (pos != 0) {
                            r.pos = 0;
                            r.seekBy(@intCast(pos)) catch {
                                r.mode = .failure;
                                return error.ReadFailed;
                            };
                        }
                        return 0;
                    },
                    else => |e| {
                        r.err = e;
                        return error.ReadFailed;
                    },
                };
                if (n == 0) {
                    r.size = r.pos;
                    return error.EndOfStream;
                }
                r.pos += n;
                if (n > data_size) {
                    io_reader.end += n - data_size;
                    return data_size;
                }
                return n;
            },
            .streaming, .streaming_reading => {
                if (is_windows) {
                    // Unfortunately, `ReadFileScatter` cannot be used since it
                    // requires page alignment.
                    if (io_reader.seek == io_reader.end) {
                        io_reader.seek = 0;
                        io_reader.end = 0;
                    }
                    const first = data[0];
                    if (first.len >= io_reader.buffer.len - io_reader.end) {
                        return readStreaming(r, first);
                    } else {
                        io_reader.end += try readStreaming(r, io_reader.buffer[io_reader.end..]);
                        return 0;
                    }
                }
                var iovecs_buffer: [max_buffers_len]posix.iovec = undefined;
                const dest_n, const data_size = try io_reader.writableVectorPosix(&iovecs_buffer, data);
                const dest = iovecs_buffer[0..dest_n];
                assert(dest[0].len > 0);
                const n = posix.readv(r.file.handle, dest) catch |err| {
                    r.err = err;
                    return error.ReadFailed;
                };
                if (n == 0) {
                    r.size = r.pos;
                    return error.EndOfStream;
                }
                r.pos += n;
                if (n > data_size) {
                    io_reader.end += n - data_size;
                    return data_size;
                }
                return n;
            },
            .failure => return error.ReadFailed,
        }
    }

    fn discard(io_reader: *std.Io.Reader, limit: std.Io.Limit) std.Io.Reader.Error!usize {
        const r: *Reader = @alignCast(@fieldParentPtr("interface", io_reader));
        const file = r.file;
        const pos = r.pos;
        switch (r.mode) {
            .positional, .positional_reading => {
                const size = r.getSize() catch {
                    r.mode = r.mode.toStreaming();
                    return 0;
                };
                const delta = @min(@intFromEnum(limit), size - pos);
                r.pos = pos + delta;
                return delta;
            },
            .streaming, .streaming_reading => {
                // Unfortunately we can't seek forward without knowing the
                // size because the seek syscalls provided to us will not
                // return the true end position if a seek would exceed the
                // end.
                fallback: {
                    if (r.size_err == null and r.seek_err == null) break :fallback;
                    var trash_buffer: [128]u8 = undefined;
                    if (is_windows) {
                        const n = windows.ReadFile(file.handle, limit.slice(&trash_buffer), null) catch |err| {
                            r.err = err;
                            return error.ReadFailed;
                        };
                        if (n == 0) {
                            r.size = pos;
                            return error.EndOfStream;
                        }
                        r.pos = pos + n;
                        return n;
                    }
                    var iovecs: [max_buffers_len]std.posix.iovec = undefined;
                    var iovecs_i: usize = 0;
                    var remaining = @intFromEnum(limit);
                    while (remaining > 0 and iovecs_i < iovecs.len) {
                        iovecs[iovecs_i] = .{ .base = &trash_buffer, .len = @min(trash_buffer.len, remaining) };
                        remaining -= iovecs[iovecs_i].len;
                        iovecs_i += 1;
                    }
                    const n = posix.readv(file.handle, iovecs[0..iovecs_i]) catch |err| {
                        r.err = err;
                        return error.ReadFailed;
                    };
                    if (n == 0) {
                        r.size = pos;
                        return error.EndOfStream;
                    }
                    r.pos = pos + n;
                    return n;
                }
                const size = r.getSize() catch return 0;
                const n = @min(size - pos, maxInt(i64), @intFromEnum(limit));
                file.seekBy(n) catch |err| {
                    r.seek_err = err;
                    return 0;
                };
                r.pos = pos + n;
                return n;
            },
            .failure => return error.ReadFailed,
        }
    }

    pub fn readPositional(r: *Reader, dest: []u8) std.Io.Reader.Error!usize {
        const n = r.file.pread(dest, r.pos) catch |err| switch (err) {
            error.Unseekable => {
                r.mode = r.mode.toStreaming();
                const pos = r.pos;
                if (pos != 0) {
                    r.pos = 0;
                    r.seekBy(@intCast(pos)) catch {
                        r.mode = .failure;
                        return error.ReadFailed;
                    };
                }
                return 0;
            },
            else => |e| {
                r.err = e;
                return error.ReadFailed;
            },
        };
        if (n == 0) {
            r.size = r.pos;
            return error.EndOfStream;
        }
        r.pos += n;
        return n;
    }

    pub fn readStreaming(r: *Reader, dest: []u8) std.Io.Reader.Error!usize {
        const n = r.file.read(dest) catch |err| {
            r.err = err;
            return error.ReadFailed;
        };
        if (n == 0) {
            r.size = r.pos;
            return error.EndOfStream;
        }
        r.pos += n;
        return n;
    }

    pub fn read(r: *Reader, dest: []u8) std.Io.Reader.Error!usize {
        switch (r.mode) {
            .positional, .positional_reading => return readPositional(r, dest),
            .streaming, .streaming_reading => return readStreaming(r, dest),
            .failure => return error.ReadFailed,
        }
    }

    pub fn atEnd(r: *Reader) bool {
        // Even if stat fails, size is set when end is encountered.
        const size = r.size orelse return false;
        return size - r.pos == 0;
    }
}  