readyreader: add buffered reader (#9055)

This PR introduces a buffered `io.Reader` that automatically releases
its read buffer when empty. To optimize memory usage, the reader defers
buffer reallocation until data is available in the underlying
`readyreader.Reader` (achieved by calling `ReadOnReady`).

In a follow-up PR (#9032), the HTTP/2 framer will be updated to utilize
this new buffered reader whenever the underlying reader implements the
`readyreader.Reader` interface.

The implementation and associated tests are based on the [standard
library's](https://cs.opensource.google/go/go/+/refs/tags/go1.26.2:src/bufio/bufio.go;l=35).

RELEASE NOTES: N/A

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: arjan-bal

internal/transport/readyreader/ready_reader.go

@@ -63,9 +63,9 @@ type readState struct {
 	buf       *[]byte
 }
 
-// newNonBlockingReader returns a ReadyReader if the passed reader supports
+// newNonBlocking returns a ReadyReader if the passed reader supports
 // non-memory-pinning reads, else nil.
-func newNonBlockingReader(r io.Reader) Reader {
+func newNonBlocking(r io.Reader) Reader {
 	if rr, ok := r.(Reader); ok {
 		return rr
 	}
@@ -153,10 +153,101 @@ func (c *blockingReader) ReadOnReady(bufSize int, pool mem.BufferPool) (*[]byte,
 
 // New detects if [syscall.RawConn] is available for non-memory-pinning reads.
 // If [syscall.RawConn] is unavailable, it falls back to using the simpler
-// [net.Conn] interface for reads.
+// [io.Reader] interface for reads.
 func New(r io.Reader) Reader {
-	if r := newNonBlockingReader(r); r != nil {
+	if r := newNonBlocking(r); r != nil {
 		return r
 	}
 	return &blockingReader{reader: r}
 }
+
+// bufReadyReader implements buffering for a ReadyReader object.
+// A new bufReadyReader is created by calling [NewBuffered].
+type bufReadyReader struct {
+	buf       *[]byte
+	pool      mem.BufferPool
+	bufSize   int
+	rd        Reader // reader provided by the caller
+	r, w      int    // buf read and write positions
+	err       error
+	constPool constBufferPool // stored as a field to avoid heap allocations.
+}
+
+// NewBuffered returns a new [io.Reader] with a buffer of the specified size
+// which is allocated from the provided pool.
+func NewBuffered(rd Reader, size int, pool mem.BufferPool) io.Reader {
+	return &bufReadyReader{
+		rd:      rd,
+		pool:    pool,
+		bufSize: size,
+	}
+}
+
+func (b *bufReadyReader) readErr() error {
+	err := b.err
+	b.err = nil
+	return err
+}
+
+func (b *bufReadyReader) buffered() int { return b.w - b.r }
+
+// Read reads data into p. It returns the number of bytes read into p. The
+// bytes are taken from at most one Read on the underlying [ReadyReader],
+// hence n may be less than len(p). If the underlying [ReadyReader] can return
+// a non-zero count with io.EOF, then this Read method can do so as well; see
+// the [io.Reader] docs.
+func (b *bufReadyReader) Read(p []byte) (n int, err error) {
+	n = len(p)
+	if n == 0 {
+		if b.buffered() > 0 {
+			return 0, nil
+		}
+		return 0, b.readErr()
+	}
+	if b.r == b.w {
+		if b.err != nil {
+			return 0, b.readErr()
+		}
+		if len(p) >= b.bufSize {
+			// Large read, empty buffer.
+			// Read directly into p to avoid copy.
+			b.constPool.buffer = p
+			_, n, b.err = b.rd.ReadOnReady(len(p), &b.constPool)
+			return n, b.readErr()
+		}
+		// One read.
+		b.r = 0
+		b.w = 0
+		b.buf, n, b.err = b.rd.ReadOnReady(b.bufSize, b.pool)
+		if n == 0 {
+			if b.buf != nil {
+				b.pool.Put(b.buf)
+				b.buf = nil
+			}
+			return 0, b.readErr()
+		}
+		b.w += n
+	}
+
+	// copy as much as we can
+	// b.buf must be non-nil since b.r != b.w.
+	buf := *b.buf
+	n = copy(p, buf[b.r:b.w])
+	b.r += n
+	if b.r == b.w {
+		// Consumed entire buffer, release it.
+		b.pool.Put(b.buf)
+		b.buf = nil
+	}
+	return n, nil
+}
+
+type constBufferPool struct {
+	buffer []byte
+}
+
+func (p *constBufferPool) Get(int) *[]byte {
+	return &p.buffer
+}
+
+func (p *constBufferPool) Put(*[]byte) {}