`Io.Reader` buffer and `peek` paradox

Help
1

I am probably overthinking something, but I’ve hit a bit of a quandary while trying to support peek for a base64 Io.Reader implementation. The issue is that peek expects stream to fill up to the full buffer for the Io.Reader. But the base64 decoding ratio is 4 to 3, which can result in needing to read more than one byte from the input reader to fill the next byte in the buffer. So on a peek that fills up the buffer with a partial chunk, the full chunk needs to stay in a buffer somewhere for the next time stream reads from the input reader. Further, there is no guarantee that the upstream reader’s buffer is large enough to hold a chunk, and there is no way to use the interface’s buffer since the entire buffer can be used by the interface.

The solution I have come up with is to wrap the input reader in an indirect reader that has a chunk length buffer:

pub const Reader = struct {
    pub const min_buffer_len = 3;
    pub const chunk_len = 4;

    in: *Io.Reader,
    decoder: base64.Base64Decoder,
    interface: Io.Reader,
    err: ?base64.Error,

    indirect: ?Indirect,
    decoded_offset: u2,

    const Indirect = ReaderIndirect(chunk_len);

    pub fn init(in: *Io.Reader, decoder: base64.Base64Decoder, buffer: []u8) Reader {
        assert(buffer.len >= min_buffer_len);

        return .{
            .in = in,
            .decoder = decoder,
            .interface = .{
                .vtable = &.{
                    .stream = stream,
                    .readVec = readVec,
                },
                .buffer = buffer,
                .end = 0,
                .seek = 0,
            },
            .err = null,
            .indirect = null,
            .decoded_offset = 0,
        };
    }

    fn stream(r: *Io.Reader, _: *Io.Writer, _: Io.Limit) Io.Reader.StreamError!usize {
        try readChunk(r);
        return 0;
    }

    fn readVec(r: *Io.Reader, _: [][]u8) Io.Reader.Error!usize {
        try readChunk(r);
        return 0;
    }

    fn readChunk(r: *Io.Reader) Io.Reader.Error!void {
        const self: *Reader = @fieldParentPtr("interface", r);
        var in = if (self.in.buffer.len >= chunk_len) self.in else in: {
            if (self.indirect == null) {
                Indirect.init(&self.indirect, self.in);
            }
            break :in &self.indirect.?.interface;
        };

        // peek a chunk, or at least one byte worth of decoded data (2 bytes).
        const encoded = enc: {
            const buf = try in.peekGreedy(2);
            break :enc buf[0..@min(buf.len, chunk_len)];
        };
        const eos_seen = (encoded.len < chunk_len);
        assert(encoded.len <= chunk_len);

        const decoded_len = self.decoder.calcSizeForSlice(encoded) catch |err| {
            if (eos_seen) return error.EndOfStream;
            self.err = err;
            return error.ReadFailed;
        };
        assert(decoded_len <= Writer.chunk_len);

        if (eos_seen and decoded_len == 0) return error.EndOfStream;
        try r.rebase(1);

        const decoded = dec: {
            var buf: [chunk_len]u8 = undefined;
            const chunk = buf[0..decoded_len];
            self.decoder.decode(chunk, encoded) catch |err| {
                if (eos_seen) return error.EndOfStream;
                self.err = err;
                return error.ReadFailed;
            };
            break :dec chunk;
        };

        var offset: usize = @intCast(self.decoded_offset);
        assert(offset < decoded_len);

        const remaining = decoded_len - offset;
        const n = @min(r.buffer.len - r.end, remaining);
        @memcpy(r.buffer[r.end..][0..n], decoded[offset..][0..n]);
        r.end += n;
        offset += n;

        const chunk_delivered = offset == decoded_len;
        if (chunk_delivered) {
            in.toss(encoded.len);
            self.decoded_offset = 0;
        } else {
            self.decoded_offset = @intCast(offset);
        }
    }
};

But I’m really unhappy with the ergonomics of ReaderIndirect, which can’t be safely copied and so it can’t initialize it’s (self-referential) interface field until it can access *Self:

fn ReaderIndirect(buffer_len: comptime_int) type {
    return struct {
        const Self = @This();

        in: *Io.Reader,
        interface: Io.Reader,
        buffer: [buffer_len]u8,

        pub fn init(self: *?Self, in: *Io.Reader) void {
            self.* = .{
                .in = in,
                .interface = undefined,
                .buffer = undefined,
            };
            self.*.?.interface = .{
                .vtable = &.{
                    .stream = Self.stream,
                    .readVec = Self.readVec,
                },
                .buffer = &self.*.?.buffer,
                .seek = 0,
                .end = 0,
            };
        }

        // the rest is copied from testing.ReaderIndirect

        fn readVec(r: *Io.Reader, _: [][]u8) Io.Reader.Error!usize {
            try streamInner(r);
            return 0;
        }

        fn stream(r: *Io.Reader, _: *Io.Writer, _: Io.Limit) Io.Reader.StreamError!usize {
            try streamInner(r);
            return 0;
        }

        fn streamInner(r: *Io.Reader) Io.Reader.Error!void {
            const r_indirect: *@This() = @alignCast(@fieldParentPtr("interface", r));

            // If there's no room remaining in the buffer at all, make room.
            if (r.buffer.len == r.end) {
                try r.rebase(r.buffer.len);
            }

            var writer: Io.Writer = .{
                .buffer = r.buffer,
                .end = r.end,
                .vtable = &.{
                    .drain = Io.Writer.unreachableDrain,
                    .rebase = Io.Writer.unreachableRebase,
                },
            };
            defer r.end = writer.end;

            r_indirect.in.streamExact(&writer, r.buffer.len - r.end) catch |err| switch (err) {
                // Only forward EndOfStream if no new bytes were written to the buffer
                error.EndOfStream => |e| if (r.end == writer.end) {
                    return e;
                },
                error.WriteFailed => unreachable,
                else => |e| return e,
            };
        }
    };
}

It seems like any reader implementation that needs to read past it’s buffers
length of bytes from it’s input reader has this problem, are there other
simpler solutions to this out there?

2

I didn’t understand anything you described, but it appears you’re reading data from a reader, processing, and then serving the processed data through the reader interface.

If this is the crux of your problem, you can simply require that the input reader have a large enough buffer to hold a chunk. In the reader interface docs, it is stated the implementers are expected to place minimum bounds on the sizes of buffers.

3

It already assert a minimum size of the buffer parameter to init (it technically can be relaxed to > 0), but to properly support peek there is a minimum buffer size requirement on the in parameter’s buffer. This means that the Reader won’t be compatible with all input Io.Readers, and AFAIK there’s nothing in the docs about how to deal with input reader’s buffer requirements.

Here’s a pseudo code example to hopefully explains the issue better:

// "QUFBQUEK" decodes to "AAAAA"
var stream: Io.Reader = .fixed("QUFBQUEK");

// setup an `in` reader with a 2 byte buffer that reads from `stream`
var short_buf: [2]byte = undefined;
var in: std.testing.ReaderIndirect = .init(&stream, &short_buf);

// setup the base64 reader with a 4 byte buffer that reads from `in`
var chunk_buf: [4]byte = undefined;
var b64: Base64Reader = .init(&in.interface, &chunk_buf);
var r: *Io.Reader = &b64.interface;

// peek 2 bytes from `b64`, it's buffer is now 'AAA' after reading the first chunk'
assert(std.mem.eql(u8, "AA", try r.peek(2)));
assert(std.mem.eql(u8, "AAA", r.buffered()));

// since 4 bytes were consumed from `in`, the next bytes read from `in` will be
// "QUEK", and it's buffer can only be "", "Q", or "QU"
assert(std.mem.eql(u8, "", in.interface.buffered()) or 
       std.mem.eql(u8, "Q", in.interface.buffered()) or 
       std.mem.eql(u8, "QU", in.interface.buffered()));

// taking 4 bytes requires b64 to read both chunks from `in`, but it can only
// hold part of the second chunk.
assert(std.mem.eql(u8, "AAAA", try r.take(4))); // AAA is from the first chunk, A is from the second
assert(std.mem.eql(u8, "", r.buffered()));      // but the rest of the 2nd chunk wouldn't fit in r's buffer (at the end of `stream`)

// since r had to read both chunks, and the second chunk was 4 bytes, it
// cannot all still be held `in`'s 2 byte buffer, the "QU" couldn't be `peek`d.
assert(std.mem.eql(u8, "", in.interface.buffered() or
       std.mem.eql(u8, "E", in.interface.buffered()) or
       std.mem.eql(u8, "EK", in.interface.buffered()));

// so the next read to `b64` can't recover the remaining parts of the second chunk,
// which means there's no way to read the last 'A'.
assert(std.mem.eql(u8, "A", try r.take(1)); // kaboom!

Changing the size of chunk_buf to something larger doesn’t fix the problem, at some point a peek’ing consumer will fill up the buffer with a partial chunk. The only solution I see is to enforce a minimum buffer size on the input reader, or layer in the necessary buffer myself.

1 Like
4

Yes, this would be the typical way of doing it, and what you so in the decompression APIs. I would strongly opt for the former here to avoid unnecessary copies. It is understandable to want to keep your input as generic as possible, but there are times where you need to impose restrictions beyond just accepting any 'ol std.Io.Reader, and this would be an example of one of those times.

I would do the following:

  • Document the minimum buffer size for the input reader’s buffer
  • Assert/error if it is too small
  • Create a public constant that consumers can use such (i.e. min_buffer_size) when initializing their buffer. This makes it apparent that the size is not arbitrary, as well as makes it easy for you to change in the future.

In practice, this is not an imposition to consumers if that is your concern.