From 901a52c61b85a03a27f9677bc274e0ea07b82d38 Mon Sep 17 00:00:00 2001 From: Shawn Smith Date: Thu, 31 Dec 2015 14:21:05 +0900 Subject: [PATCH 01/23] fix typo --- reader.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/reader.go b/reader.go index e0a7f53..ad64a5a 100644 --- a/reader.go +++ b/reader.go @@ -168,7 +168,7 @@ func (r *Reader) Peek(n int) ([]byte, error) { // Skip moves the reader forward 'n' bytes. // Returns the number of bytes skipped and any -// errors encountered. It is analagous to Seek(n, 1). +// errors encountered. It is analogous to Seek(n, 1). // If the underlying reader implements io.Seeker, then // that method will be used to skip forward. // From 50547e3e2fc4c22f61a2fdc0ed90deacb744f2b1 Mon Sep 17 00:00:00 2001 From: Shawn Smith Date: Thu, 31 Dec 2015 14:22:00 +0900 Subject: [PATCH 02/23] fix typos --- writer.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/writer.go b/writer.go index d6f44f5..2dc392a 100644 --- a/writer.go +++ b/writer.go @@ -98,7 +98,7 @@ func (w *Writer) Write(p []byte) (int, error) { return copy(w.buf[l:], p), nil } -// WriteString is analagous to Write, but it takes a string. +// WriteString is analogous to Write, but it takes a string. func (w *Writer) WriteString(s string) (int, error) { c, l, ln := cap(w.buf), len(w.buf), len(s) avail := c - l @@ -215,7 +215,7 @@ func (w *Writer) ReadFrom(r io.Reader) (int64, error) { // we only clear here // because we are sure // the writes have - // suceeded. otherwise, + // succeeded. otherwise, // we retain the data in case // future writes succeed. w.buf = w.buf[0:0] From d530c40f9c8ef5eca9a60c978574e15896dbce87 Mon Sep 17 00:00:00 2001 From: Shawn Smith Date: Thu, 21 Jan 2016 18:04:19 -0800 Subject: [PATCH 03/23] typo --- reader.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/reader.go b/reader.go index ad64a5a..6c04044 100644 --- a/reader.go +++ b/reader.go @@ -22,7 +22,7 @@ // stream implements it. `(*fwd.Reader).Next` returns a slice pointing // to the next `n` bytes in the read buffer (like `Peek`), but also // increments the read position. This allows users to process streams -// in aribtrary block sizes without having to manage appropriately-sized +// in arbitrary block sizes without having to manage appropriately-sized // slices. Additionally, obviating the need to copy the data from the // buffer to another location in memory can improve performance dramatically // in CPU-bound applications. From f603d16432b83107306797022be0222077fc53f8 Mon Sep 17 00:00:00 2001 From: Philip Hofer Date: Mon, 25 Jan 2016 15:22:51 -0800 Subject: [PATCH 04/23] maximize read buffer size Before, if r.n == len(r.data), then r.data wouldn't be re-sliced to r.data[:0], which would call Read() to be called with a smaller buffer than necessary. --- reader.go | 6 +++++- reader_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/reader.go b/reader.go index 6c04044..e591910 100644 --- a/reader.go +++ b/reader.go @@ -97,7 +97,11 @@ func (r *Reader) more() { // we can supply the maximum number of // bytes to the reader if r.n != 0 { - r.data = r.data[:copy(r.data[0:], r.data[r.n:])] + if r.n < len(r.data) { + r.data = r.data[:copy(r.data[0:], r.data[r.n:])] + } else { + r.data = r.data[:0] + } r.n = 0 } var a int diff --git a/reader_test.go b/reader_test.go index 5d40acf..e96303a 100644 --- a/reader_test.go +++ b/reader_test.go @@ -352,3 +352,47 @@ func TestReadFull(t *testing.T) { t.Fatalf("expected to read %d bytes; read %d", 1024, n) } } + +type readCounter struct { + r io.Reader + count int +} + +func (r *readCounter) Read(p []byte) (int, error) { + r.count++ + return r.r.Read(p) +} + +func TestReadFullPerf(t *testing.T) { + const size = 1 << 22 + data := randomBts(size) + + c := readCounter{ + r: &partialReader{ + r: bytes.NewReader(data), + }, + } + + r := NewReader(&c) + + const segments = 4 + out := make([]byte, size/segments) + + for i := 0; i < segments; i++ { + // force an unaligned read + _, err := r.Peek(5) + if err != nil { + t.Fatal(err) + } + + n, err := r.ReadFull(out) + if err != nil { + t.Fatal(err) + } + if n != size/segments { + t.Fatalf("read %d bytes, not %d", n, size/segments) + } + } + + t.Logf("called Read() on the underlying reader %d times to fill %d buffers", c.count, size/r.BufferSize()) +} From 222b992a5a35f619816f8380a44a3788b23e51f7 Mon Sep 17 00:00:00 2001 From: shawnps Date: Fri, 29 Jan 2016 12:39:26 +0900 Subject: [PATCH 05/23] rerun godoc2md --- README.md | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 8c7d25a..38349af 100644 --- a/README.md +++ b/README.md @@ -11,9 +11,9 @@ The `fwd.Writer` and `fwd.Reader` type provide similar functionality to their counterparts in `bufio`, plus a few extra utility methods that simplify read-ahead and write-ahead. I wrote this package to improve serialization -performance for http://github.com/philhofer/msgp, -where it provided about a 2x speedup over `bufio`. However, -care must be taken to understand the semantics of the +performance for http://github.com/tinylib/msgp, +where it provided about a 2x speedup over `bufio` for certain +workloads. However, care must be taken to understand the semantics of the extra methods provided by this package, as they allow the user to access and manipulate the buffer memory directly. @@ -26,7 +26,7 @@ in the stream, and uses the `io.Seeker` interface if the underlying stream implements it. `(*fwd.Reader).Next` returns a slice pointing to the next `n` bytes in the read buffer (like `Peek`), but also increments the read position. This allows users to process streams -in aribtrary block sizes without having to manage appropriately-sized +in arbitrary block sizes without having to manage appropriately-sized slices. Additionally, obviating the need to copy the data from the buffer to another location in memory can improve performance dramatically in CPU-bound applications. @@ -110,13 +110,14 @@ Buffered returns the number of bytes currently in the buffer func (r *Reader) Next(n int) ([]byte, error) ``` Next returns the next 'n' bytes in the stream. -If the returned slice has a length less than 'n', -an error will also be returned. Unlike Peek, Next advances the reader position. The returned bytes point to the same data as the buffer, so the slice is only valid until the next reader method call. An EOF is considered an unexpected error. +If an the returned slice is less than the +length asked for, an error will be returned, +and the reader position will not be incremented. @@ -156,6 +157,7 @@ func (r *Reader) ReadFull(b []byte) (int, error) ReadFull attempts to read len(b) bytes into 'b'. It returns the number of bytes read into 'b', and an error if it does not return len(b). +EOF is considered an unexpected error. @@ -174,7 +176,7 @@ func (r *Reader) Skip(n int) (int, error) ``` Skip moves the reader forward 'n' bytes. Returns the number of bytes skipped and any -errors encountered. It is analagous to Seek(n, 1). +errors encountered. It is analogous to Seek(n, 1). If the underlying reader implements io.Seeker, then that method will be used to skip forward. @@ -266,6 +268,8 @@ Next returns the next 'n' free bytes in the write buffer, flushing the writer as necessary. Next will return `io.ErrShortBuffer` if 'n' is greater than the size of the write buffer. +Calls to 'next' increment the write position by +the size of the returned buffer. @@ -297,7 +301,7 @@ WriteByte implements `io.ByteWriter` ``` go func (w *Writer) WriteString(s string) (int, error) ``` -WriteString is analagous to Write, but it takes a string. +WriteString is analogous to Write, but it takes a string. From 2eb7ce4c0c64e185cfdf44bb38d62f6ed26a5d2f Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Fri, 16 Jun 2017 14:59:17 -0500 Subject: [PATCH 06/23] Discard io.EOF when more than 0 bytes have been read Some `io.Reader` instances can return `io.EOF` with a non-zero number of bytes and will then return 0 bytes with the `io.EOF` error on the next read. This interferes with the expectation that `io.EOF` meant that no bytes had been read. Discard the error and buffer the read data so these types of `io.Reader` instances still work properly. Fixes #13. --- reader.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/reader.go b/reader.go index e591910..75be62a 100644 --- a/reader.go +++ b/reader.go @@ -109,6 +109,10 @@ func (r *Reader) more() { if a == 0 && r.state == nil { r.state = io.ErrNoProgress return + } else if a > 0 && r.state == io.EOF { + // discard the io.EOF if we read more than 0 bytes. + // the next call to Read should return io.EOF again. + r.state = nil } r.data = r.data[:len(r.data)+a] } From 7f183dbc6f36d1cb71b302cebfabe8bde48ebd8f Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Fri, 1 Sep 2017 12:51:31 +0100 Subject: [PATCH 07/23] correct spelling mistake --- writer_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/writer_test.go b/writer_test.go index 8bab07c..3dcf3a5 100644 --- a/writer_test.go +++ b/writer_test.go @@ -29,7 +29,7 @@ func (c chunkedWriter) Write(p []byte) (int, error) { return n, nil } -// analagous to Write(), but w/ str +// analogous to Write(), but w/ str func (c chunkedWriter) WriteString(s string) (int, error) { l := len(s) n := 0 From 414ae1bb9ed152f441b372db076776b3a937ebf5 Mon Sep 17 00:00:00 2001 From: Janusz Marcinkiewicz Date: Thu, 30 Jul 2020 09:22:05 +0200 Subject: [PATCH 08/23] Add Reader/Writer constructors with custom buffer --- reader.go | 17 +++++++++++++++-- reader_test.go | 25 +++++++++++++++++++++++++ writer.go | 24 ++++++++++++++++++------ writer_test.go | 26 +++++++++++++++++++++++++- 4 files changed, 83 insertions(+), 9 deletions(-) diff --git a/reader.go b/reader.go index 75be62a..6918d3e 100644 --- a/reader.go +++ b/reader.go @@ -50,11 +50,24 @@ func NewReader(r io.Reader) *Reader { } // NewReaderSize returns a new *Reader that -// reads from 'r' and has a buffer size 'n' +// reads from 'r' and has a buffer size 'n'. func NewReaderSize(r io.Reader, n int) *Reader { + buf := make([]byte, 0, max(n, minReaderSize)) + return NewReaderBuf(r, buf) +} + +// NewReaderBuf returns a new *Reader that +// reads from 'r' and uses 'buf' as a buffer. +// 'buf' is not used when has smaller capacity than 16, +// custom buffer is allocated instead. +func NewReaderBuf(r io.Reader, buf []byte) *Reader { + if cap(buf) < minReaderSize { + buf = make([]byte, 0, minReaderSize) + } + buf = buf[:0] rd := &Reader{ r: r, - data: make([]byte, 0, max(minReaderSize, n)), + data: buf, } if s, ok := r.(io.Seeker); ok { rd.rs = s diff --git a/reader_test.go b/reader_test.go index e96303a..c9a2d0a 100644 --- a/reader_test.go +++ b/reader_test.go @@ -396,3 +396,28 @@ func TestReadFullPerf(t *testing.T) { t.Logf("called Read() on the underlying reader %d times to fill %d buffers", c.count, size/r.BufferSize()) } + +func TestReaderBufCreation(t *testing.T) { + tests := []struct { + name string + buffer []byte + size int + }{ + {name: "nil", buffer: nil, size: minReaderSize}, + {name: "empty", buffer: []byte{}, size: minReaderSize}, + {name: "allocated", buffer: make([]byte, 0, 200), size: 200}, + {name: "filled", buffer: make([]byte, 200), size: 200}, + } + + for _, test := range tests { + var b bytes.Buffer + r := NewReaderBuf(&b, test.buffer) + + if r.BufferSize() != test.size { + t.Errorf("%s: unequal buffer size (got: %d, expected: %d)", test.name, r.BufferSize(), test.size) + } + if r.Buffered() != 0 { + t.Errorf("%s: unequal buffered bytes (got: %d, expected: 0)", test.name, r.Buffered()) + } + } +} diff --git a/writer.go b/writer.go index 2dc392a..4d6ea15 100644 --- a/writer.go +++ b/writer.go @@ -29,16 +29,28 @@ func NewWriter(w io.Writer) *Writer { } } -// NewWriterSize returns a new writer -// that writes to 'w' and has a buffer -// that is 'size' bytes. -func NewWriterSize(w io.Writer, size int) *Writer { - if wr, ok := w.(*Writer); ok && cap(wr.buf) >= size { +// NewWriterSize returns a new writer that +// writes to 'w' and has a buffer size 'n'. +func NewWriterSize(w io.Writer, n int) *Writer { + if wr, ok := w.(*Writer); ok && cap(wr.buf) >= n { return wr } + buf := make([]byte, 0, max(n, minWriterSize)) + return NewWriterBuf(w, buf) +} + +// NewWriterBuf returns a new writer +// that writes to 'w' and has 'buf' as a buffer. +// 'buf' is not used when has smaller capacity than 18, +// custom buffer is allocated instead. +func NewWriterBuf(w io.Writer, buf []byte) *Writer { + if cap(buf) < minWriterSize { + buf = make([]byte, 0, minWriterSize) + } + buf = buf[:0] return &Writer{ w: w, - buf: make([]byte, 0, max(size, minWriterSize)), + buf: buf, } } diff --git a/writer_test.go b/writer_test.go index 3dcf3a5..5868d29 100644 --- a/writer_test.go +++ b/writer_test.go @@ -235,5 +235,29 @@ func TestReadFrom(t *testing.T) { if !bytes.Equal(buf.Bytes(), bts) { t.Fatal("buf.Bytes() and input are not equal") } - +} + +func TestWriterBufCreation(t *testing.T) { + tests := []struct { + name string + buffer []byte + size int + }{ + {name: "nil", buffer: nil, size: minWriterSize}, + {name: "empty", buffer: []byte{}, size: minWriterSize}, + {name: "allocated", buffer: make([]byte, 0, 200), size: 200}, + {name: "filled", buffer: make([]byte, 200), size: 200}, + } + + for _, test := range tests { + var b bytes.Buffer + w := NewWriterBuf(&b, test.buffer) + + if w.BufferSize() != test.size { + t.Errorf("%s: unequal buffer size (got: %d, expected: %d)", test.name, w.BufferSize(), test.size) + } + if w.Buffered() != 0 { + t.Errorf("%s: unequal buffered bytes (got: %d, expected: 0)", test.name, w.Buffered()) + } + } } From 9bcb9cab1b0394d8516a132ec243ba944efdb644 Mon Sep 17 00:00:00 2001 From: Phil Hofer Date: Tue, 24 Nov 2020 00:35:43 +0000 Subject: [PATCH 09/23] reader: fix Skip(n) where n >= buffer size The (*Reader).Skip() code did not handle skip sizes above the buffer size correctly, because the inner reader loop only looped while 'r.buffered() < n', and that condition is obviously violated when the reader successfully buffers more bytes than are to be skipped. Refactor this method to be a little shorter and hopefully much clearer (and more correct) and add test coverage that exercises this code better. --- reader.go | 75 +++++++++++++++++++++----------------------------- reader_test.go | 42 ++++++++++++++++++++++++---- 2 files changed, 67 insertions(+), 50 deletions(-) diff --git a/reader.go b/reader.go index 6918d3e..72cc112 100644 --- a/reader.go +++ b/reader.go @@ -34,7 +34,10 @@ // package fwd -import "io" +import ( + "io" + "os" +) const ( // DefaultReaderSize is the default size of the read buffer @@ -187,6 +190,19 @@ func (r *Reader) Peek(n int) ([]byte, error) { return r.data[r.n : r.n+n], nil } +// discard(n) discards up to 'n' buffered bytes, and +// and returns the number of bytes discarded +func (r *Reader) discard(n int) int { + inbuf := r.buffered() + if inbuf <= n { + r.n = 0 + r.data = r.data[:0] + return inbuf + } + r.n += n + return n +} + // Skip moves the reader forward 'n' bytes. // Returns the number of bytes skipped and any // errors encountered. It is analogous to Seek(n, 1). @@ -201,33 +217,25 @@ func (r *Reader) Peek(n int) ([]byte, error) { // will not return `io.EOF` until the next call // to Read.) func (r *Reader) Skip(n int) (int, error) { - - // fast path - if r.buffered() >= n { - r.n += n - return n, nil + if n < 0 { + return 0, os.ErrInvalid } - // use seeker implementation - // if we can - if r.rs != nil { - return r.skipSeek(n) - } + // discard some or all of the current buffer + skipped := r.discard(n) - // loop on filling - // and then erasing - o := n - for r.buffered() < n && r.state == nil { + // if we can Seek() through the remaining bytes, do that + if n > skipped && r.rs != nil { + nn, err := r.rs.Seek(int64(n-skipped), 1) + return int(nn) + skipped, err + } + // otherwise, keep filling the buffer + // and discarding it up to 'n' + for skipped < n && r.state == nil { r.more() - // we can skip forward - // up to r.buffered() bytes - step := min(r.buffered(), n) - r.n += step - n -= step + skipped += r.discard(n - skipped) } - // at this point, n should be - // 0 if everything went smoothly - return o - n, r.noEOF() + return skipped, r.noEOF() } // Next returns the next 'n' bytes in the stream. @@ -262,20 +270,6 @@ func (r *Reader) Next(n int) ([]byte, error) { return out, nil } -// skipSeek uses the io.Seeker to seek forward. -// only call this function when n > r.buffered() -func (r *Reader) skipSeek(n int) (int, error) { - o := r.buffered() - // first, clear buffer - n -= o - r.n = 0 - r.data = r.data[:0] - - // then seek forward remaning bytes - i, err := r.rs.Seek(int64(n), 1) - return int(i) + o, err -} - // Read implements `io.Reader` func (r *Reader) Read(b []byte) (int, error) { // if we have data in the buffer, just @@ -381,13 +375,6 @@ func (r *Reader) WriteTo(w io.Writer) (int64, error) { return i, nil } -func min(a int, b int) int { - if a < b { - return a - } - return b -} - func max(a int, b int) int { if a < b { return b diff --git a/reader_test.go b/reader_test.go index c9a2d0a..0c394cc 100644 --- a/reader_test.go +++ b/reader_test.go @@ -133,6 +133,10 @@ func TestReadByte(t *testing.T) { } } +func remaining(r *Reader) int { + return r.Buffered() + r.r.(partialReader).r.(*bytes.Reader).Len() +} + func TestSkipNoSeek(t *testing.T) { bts := randomBts(1024) rd := NewReaderSize(partialReader{bytes.NewReader(bts)}, 200) @@ -145,6 +149,10 @@ func TestSkipNoSeek(t *testing.T) { t.Fatalf("Skip() returned a nil error, but skipped %d bytes instead of %d", n, 512) } + if remaining(rd) != 512 { + t.Errorf("expected 512 remaining; got %d", remaining(rd)) + } + var b byte b, err = rd.ReadByte() if err != nil { @@ -152,7 +160,7 @@ func TestSkipNoSeek(t *testing.T) { } if b != bts[512] { - t.Fatalf("at index %d: %d in; %d out", 512, bts[512], b) + t.Errorf("at index %d: %d in; %d out", 512, bts[512], b) } n, err = rd.Skip(10) @@ -162,16 +170,38 @@ func TestSkipNoSeek(t *testing.T) { if n != 10 { t.Fatalf("Skip() returned a nil error, but skipped %d bytes instead of %d", n, 10) } + // the number of bytes remaining in the buffer needs + // to comport with the number of bytes we expect to have skipped + if want := 1024 - 512 - 10 - 1; remaining(rd) != want { + t.Errorf("only %d bytes remaining (want %d)?", remaining(rd), want) + } + n, err = rd.Skip(10) + if err != nil { + t.Fatal(err) + } + if n != 10 { + t.Fatalf("Skip(10) a second time returned %d", n) + } + if want := 1024 - 512 - 10 - 10 - 1; remaining(rd) != want { + t.Errorf("only %d bytes remaining (want %d)?", remaining(rd), want) + } + b, err = rd.ReadByte() + if err != nil { + t.Fatalf("second ReadByte(): %s", err) + } + if b != bts[512+10+10+1] { + t.Errorf("expected %d but got %d", bts[512+10+10], b) + } - // now try to skip past the end - rd = NewReaderSize(partialReader{bytes.NewReader(bts)}, 200) - + // now try to skip past the end; we expect + // only to skip the number of bytes remaining + want := remaining(rd) n, err = rd.Skip(2000) if err != io.ErrUnexpectedEOF { t.Fatalf("expected error %q; got %q", io.EOF, err) } - if n != 1024 { - t.Fatalf("expected to skip only 1024 bytes; skipped %d", n) + if n != want { + t.Fatalf("expected to skip only %d bytes; skipped %d", want, n) } } From 5c56ac6d0bb94e41da1338caa092a0dec9317c3b Mon Sep 17 00:00:00 2001 From: Brad Peabody Date: Thu, 22 Jul 2021 10:25:04 -0700 Subject: [PATCH 10/23] tinygo support --- writer_tinygo.go | 18 ++++++++++++++++++ writer_unsafe.go | 2 +- 2 files changed, 19 insertions(+), 1 deletion(-) create mode 100644 writer_tinygo.go diff --git a/writer_tinygo.go b/writer_tinygo.go new file mode 100644 index 0000000..3b06e99 --- /dev/null +++ b/writer_tinygo.go @@ -0,0 +1,18 @@ +// +build tinygo + +package fwd + +import ( + "reflect" + "unsafe" +) + +// unsafe cast string as []byte +func unsafestr(b string) []byte { + l := uintptr(len(b)) + return *(*[]byte)(unsafe.Pointer(&reflect.SliceHeader{ + Len: l, + Cap: l, + Data: (*reflect.StringHeader)(unsafe.Pointer(&b)).Data, + })) +} diff --git a/writer_unsafe.go b/writer_unsafe.go index a0bf453..1701572 100644 --- a/writer_unsafe.go +++ b/writer_unsafe.go @@ -1,4 +1,4 @@ -// +build !appengine +// +build !appengine,!tinygo package fwd From 8be6da3b9e29bc5701fad75f3253a30805f28cf4 Mon Sep 17 00:00:00 2001 From: Cuong Manh Le Date: Mon, 16 Aug 2021 21:55:05 +0700 Subject: [PATCH 11/23] Fix invalid usage of reflect.SliceHeader Using plain reflect.SliceHeader struct violates unsafe pointer 6th rule. --- writer_unsafe.go | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/writer_unsafe.go b/writer_unsafe.go index 1701572..8ab5a4e 100644 --- a/writer_unsafe.go +++ b/writer_unsafe.go @@ -8,11 +8,12 @@ import ( ) // unsafe cast string as []byte -func unsafestr(b string) []byte { - l := len(b) - return *(*[]byte)(unsafe.Pointer(&reflect.SliceHeader{ - Len: l, - Cap: l, - Data: (*reflect.StringHeader)(unsafe.Pointer(&b)).Data, - })) +func unsafestr(s string) []byte { + var b []byte + sHdr := (*reflect.StringHeader)(unsafe.Pointer(&s)) + bHdr := (*reflect.SliceHeader)(unsafe.Pointer(&b)) + bHdr.Data = sHdr.Data + bHdr.Len = sHdr.Len + bHdr.Cap = sHdr.Len + return b } From b411808a24253af9235385bcbaf6896a873a18ec Mon Sep 17 00:00:00 2001 From: Mikkel Kamstrup Erlandsen Date: Fri, 15 Jul 2022 11:04:28 +0200 Subject: [PATCH 12/23] Fix panic in Reader.more() because of an unchecked error from the underlying reader. Fixes https://github.com/philhofer/fwd/issues/25 --- reader.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/reader.go b/reader.go index 72cc112..11307b0 100644 --- a/reader.go +++ b/reader.go @@ -129,6 +129,8 @@ func (r *Reader) more() { // discard the io.EOF if we read more than 0 bytes. // the next call to Read should return io.EOF again. r.state = nil + } else if r.state != nil { + return } r.data = r.data[:len(r.data)+a] } From e22748bc631ea4096937068f5f200daa92514f9c Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 11 Dec 2022 16:00:42 +0100 Subject: [PATCH 13/23] gofmt code for current go versions Signed-off-by: Sebastiaan van Stijn --- reader.go | 2 -- reader_test.go | 1 - writer_appengine.go | 1 + writer_tinygo.go | 1 + writer_unsafe.go | 1 + 5 files changed, 3 insertions(+), 3 deletions(-) diff --git a/reader.go b/reader.go index 11307b0..6fe8e90 100644 --- a/reader.go +++ b/reader.go @@ -31,7 +31,6 @@ // returns a slice pointing to the next `n` bytes of the writer, and increments // the write position by the length of the returned slice. This allows users // to write directly to the end of the buffer. -// package fwd import ( @@ -250,7 +249,6 @@ func (r *Reader) Skip(n int) (int, error) { // length asked for, an error will be returned, // and the reader position will not be incremented. func (r *Reader) Next(n int) ([]byte, error) { - // in case the buffer is too small if cap(r.data) < n { old := r.data[r.n:] diff --git a/reader_test.go b/reader_test.go index 0c394cc..dfee298 100644 --- a/reader_test.go +++ b/reader_test.go @@ -260,7 +260,6 @@ func TestSkipSeek(t *testing.T) { if n != 0 { t.Errorf("expected 0 bytes read; got %d", n) } - } func TestPeek(t *testing.T) { diff --git a/writer_appengine.go b/writer_appengine.go index e367f39..a978e3b 100644 --- a/writer_appengine.go +++ b/writer_appengine.go @@ -1,3 +1,4 @@ +//go:build appengine // +build appengine package fwd diff --git a/writer_tinygo.go b/writer_tinygo.go index 3b06e99..b060faf 100644 --- a/writer_tinygo.go +++ b/writer_tinygo.go @@ -1,3 +1,4 @@ +//go:build tinygo // +build tinygo package fwd diff --git a/writer_unsafe.go b/writer_unsafe.go index 8ab5a4e..e4cb4a8 100644 --- a/writer_unsafe.go +++ b/writer_unsafe.go @@ -1,3 +1,4 @@ +//go:build !appengine && !tinygo // +build !appengine,!tinygo package fwd From 7706de0536cbb7f874d69afeda0e58978b394400 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 11 Dec 2022 16:01:04 +0100 Subject: [PATCH 14/23] fix some godoc strings, and add doc links Signed-off-by: Sebastiaan van Stijn --- reader.go | 36 ++++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/reader.go b/reader.go index 6fe8e90..7c21f8f 100644 --- a/reader.go +++ b/reader.go @@ -1,10 +1,10 @@ -// The `fwd` package provides a buffered reader +// Package fwd provides a buffered reader // and writer. Each has methods that help improve // the encoding/decoding performance of some binary // protocols. // -// The `fwd.Writer` and `fwd.Reader` type provide similar -// functionality to their counterparts in `bufio`, plus +// The [Writer] and [Reader] type provide similar +// functionality to their counterparts in [bufio], plus // a few extra utility methods that simplify read-ahead // and write-ahead. I wrote this package to improve serialization // performance for http://github.com/tinylib/msgp, @@ -14,21 +14,21 @@ // the user to access and manipulate the buffer memory // directly. // -// The extra methods for `fwd.Reader` are `Peek`, `Skip` -// and `Next`. `(*fwd.Reader).Peek`, unlike `(*bufio.Reader).Peek`, +// The extra methods for [Reader] are [Reader.Peek], [Reader.Skip] +// and [Reader.Next]. (*fwd.Reader).Peek, unlike (*bufio.Reader).Peek, // will re-allocate the read buffer in order to accommodate arbitrarily -// large read-ahead. `(*fwd.Reader).Skip` skips the next `n` bytes -// in the stream, and uses the `io.Seeker` interface if the underlying -// stream implements it. `(*fwd.Reader).Next` returns a slice pointing -// to the next `n` bytes in the read buffer (like `Peek`), but also +// large read-ahead. (*fwd.Reader).Skip skips the next 'n' bytes +// in the stream, and uses the [io.Seeker] interface if the underlying +// stream implements it. (*fwd.Reader).Next returns a slice pointing +// to the next 'n' bytes in the read buffer (like Reader.Peek), but also // increments the read position. This allows users to process streams // in arbitrary block sizes without having to manage appropriately-sized // slices. Additionally, obviating the need to copy the data from the // buffer to another location in memory can improve performance dramatically // in CPU-bound applications. // -// `fwd.Writer` only has one extra method, which is `(*fwd.Writer).Next`, which -// returns a slice pointing to the next `n` bytes of the writer, and increments +// [Writer] only has one extra method, which is (*fwd.Writer).Next, which +// returns a slice pointing to the next 'n' bytes of the writer, and increments // the write position by the length of the returned slice. This allows users // to write directly to the end of the buffer. package fwd @@ -212,11 +212,11 @@ func (r *Reader) discard(n int) int { // // If the reader encounters // an EOF before skipping 'n' bytes, it -// returns io.ErrUnexpectedEOF. If the -// underlying reader implements io.Seeker, then +// returns [io.ErrUnexpectedEOF]. If the +// underlying reader implements [io.Seeker], then // those rules apply instead. (Many implementations -// will not return `io.EOF` until the next call -// to Read.) +// will not return [io.EOF] until the next call +// to Read). func (r *Reader) Skip(n int) (int, error) { if n < 0 { return 0, os.ErrInvalid @@ -270,7 +270,7 @@ func (r *Reader) Next(n int) ([]byte, error) { return out, nil } -// Read implements `io.Reader` +// Read implements [io.Reader]. func (r *Reader) Read(b []byte) (int, error) { // if we have data in the buffer, just // return that. @@ -325,7 +325,7 @@ func (r *Reader) ReadFull(b []byte) (int, error) { return n, nil } -// ReadByte implements `io.ByteReader` +// ReadByte implements [io.ByteReader]. func (r *Reader) ReadByte() (byte, error) { for r.buffered() < 1 && r.state == nil { r.more() @@ -338,7 +338,7 @@ func (r *Reader) ReadByte() (byte, error) { return b, nil } -// WriteTo implements `io.WriterTo` +// WriteTo implements [io.WriterTo]. func (r *Reader) WriteTo(w io.Writer) (int64, error) { var ( i int64 From ead0d75843d14b2835eecf0891f5e4ad07184fbe Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 11 Dec 2022 16:39:34 +0100 Subject: [PATCH 15/23] README: re-generate, some manual fixes, add doc reference badge The godoc2md tool has been deprecated, and is no longer maintained; it has some bugs which prevent it from generating correct links (so I removed those). Added a badge to the documentation on pkg.go.dev, which perhaps should be used as replacement for the docs in the README itself. Signed-off-by: Sebastiaan van Stijn --- README.md | 120 +++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 82 insertions(+), 38 deletions(-) diff --git a/README.md b/README.md index 38349af..62bd5c6 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,25 @@ # fwd - import "github.com/philhofer/fwd" -The `fwd` package provides a buffered reader +[![Go Reference](https://pkg.go.dev/badge/github.com/philhofer/fwd.svg)](https://pkg.go.dev/github.com/philhofer/fwd) + + +`import "github.com/philhofer/fwd"` + +* [Overview](#pkg-overview) +* [Index](#pkg-index) + +## Overview +Package fwd provides a buffered reader and writer. Each has methods that help improve the encoding/decoding performance of some binary protocols. -The `fwd.Writer` and `fwd.Reader` type provide similar +The `Writer` and `Reader` type provide similar functionality to their counterparts in `bufio`, plus a few extra utility methods that simplify read-ahead and write-ahead. I wrote this package to improve serialization -performance for http://github.com/tinylib/msgp, +performance for [github.com/tinylib/msgp](https://github.com/tinylib/msgp), where it provided about a 2x speedup over `bufio` for certain workloads. However, care must be taken to understand the semantics of the extra methods provided by this package, as they allow @@ -39,7 +47,37 @@ to write directly to the end of the buffer. -## Constants +## Index +* [Constants](#pkg-constants) +* [type Reader](#Reader) + * [func NewReader(r io.Reader) *Reader](#NewReader) + * [func NewReaderBuf(r io.Reader, buf []byte) *Reader](#NewReaderBuf) + * [func NewReaderSize(r io.Reader, n int) *Reader](#NewReaderSize) + * [func (r *Reader) BufferSize() int](#Reader.BufferSize) + * [func (r *Reader) Buffered() int](#Reader.Buffered) + * [func (r *Reader) Next(n int) ([]byte, error)](#Reader.Next) + * [func (r *Reader) Peek(n int) ([]byte, error)](#Reader.Peek) + * [func (r *Reader) Read(b []byte) (int, error)](#Reader.Read) + * [func (r *Reader) ReadByte() (byte, error)](#Reader.ReadByte) + * [func (r *Reader) ReadFull(b []byte) (int, error)](#Reader.ReadFull) + * [func (r *Reader) Reset(rd io.Reader)](#Reader.Reset) + * [func (r *Reader) Skip(n int) (int, error)](#Reader.Skip) + * [func (r *Reader) WriteTo(w io.Writer) (int64, error)](#Reader.WriteTo) +* [type Writer](#Writer) + * [func NewWriter(w io.Writer) *Writer](#NewWriter) + * [func NewWriterBuf(w io.Writer, buf []byte) *Writer](#NewWriterBuf) + * [func NewWriterSize(w io.Writer, n int) *Writer](#NewWriterSize) + * [func (w *Writer) BufferSize() int](#Writer.BufferSize) + * [func (w *Writer) Buffered() int](#Writer.Buffered) + * [func (w *Writer) Flush() error](#Writer.Flush) + * [func (w *Writer) Next(n int) ([]byte, error)](#Writer.Next) + * [func (w *Writer) ReadFrom(r io.Reader) (int64, error)](#Writer.ReadFrom) + * [func (w *Writer) Write(p []byte) (int, error)](#Writer.Write) + * [func (w *Writer) WriteByte(b byte) error](#Writer.WriteByte) + * [func (w *Writer) WriteString(s string) (int, error)](#Writer.WriteString) + + +## Constants ``` go const ( // DefaultReaderSize is the default size of the read buffer @@ -121,7 +159,7 @@ and the reader position will not be incremented. -### func (\*Reader) Peek +### func (\*Reader) Peek ``` go func (r *Reader) Peek(n int) ([]byte, error) ``` @@ -134,23 +172,23 @@ io.ErrUnexpectedEOF. -### func (\*Reader) Read +### func (\*Reader) Read ``` go func (r *Reader) Read(b []byte) (int, error) ``` -Read implements `io.Reader` +Read implements `io.Reader`. -### func (\*Reader) ReadByte +### func (\*Reader) ReadByte ``` go func (r *Reader) ReadByte() (byte, error) ``` -ReadByte implements `io.ByteReader` +ReadByte implements `io.ByteReader`. -### func (\*Reader) ReadFull +### func (\*Reader) ReadFull ``` go func (r *Reader) ReadFull(b []byte) (int, error) ``` @@ -161,7 +199,7 @@ EOF is considered an unexpected error. -### func (\*Reader) Reset +### func (\*Reader) Reset ``` go func (r *Reader) Reset(rd io.Reader) ``` @@ -170,7 +208,7 @@ and the read buffer. -### func (\*Reader) Skip +### func (\*Reader) Skip ``` go func (r *Reader) Skip(n int) (int, error) ``` @@ -182,27 +220,30 @@ that method will be used to skip forward. If the reader encounters an EOF before skipping 'n' bytes, it -returns io.ErrUnexpectedEOF. If the -underlying reader implements io.Seeker, then +returns `io.ErrUnexpectedEOF`. If the +underlying reader implements `io.Seeker`, then those rules apply instead. (Many implementations will not return `io.EOF` until the next call -to Read.) +to Read). -### func (\*Reader) WriteTo + +### func (\*Reader) WriteTo ``` go func (r *Reader) WriteTo(w io.Writer) (int64, error) ``` -WriteTo implements `io.WriterTo` +WriteTo implements `io.WriterTo`. -## type Writer + +## type Writer ``` go type Writer struct { // contains filtered or unexported fields } + ``` Writer is a buffered writer @@ -212,9 +253,7 @@ Writer is a buffered writer - - -### func NewWriter +### func NewWriter ``` go func NewWriter(w io.Writer) *Writer ``` @@ -223,18 +262,24 @@ that writes to 'w' and has a buffer that is `DefaultWriterSize` bytes. -### func NewWriterSize +### func NewWriterBuf ``` go -func NewWriterSize(w io.Writer, size int) *Writer +func NewWriterBuf(w io.Writer, buf []byte) *Writer ``` -NewWriterSize returns a new writer -that writes to 'w' and has a buffer -that is 'size' bytes. +NewWriterBuf returns a new writer +that writes to 'w' and has 'buf' as a buffer. +'buf' is not used when has smaller capacity than 18, +custom buffer is allocated instead. +### func NewWriterSize +``` go +func NewWriterSize(w io.Writer, n int) *Writer +``` +NewWriterSize returns a new writer that +writes to 'w' and has a buffer size 'n'. - -### func (\*Writer) BufferSize +### func (\*Writer) BufferSize ``` go func (w *Writer) BufferSize() int ``` @@ -242,7 +287,7 @@ BufferSize returns the maximum size of the buffer. -### func (\*Writer) Buffered +### func (\*Writer) Buffered ``` go func (w *Writer) Buffered() int ``` @@ -251,7 +296,7 @@ in the reader. -### func (\*Writer) Flush +### func (\*Writer) Flush ``` go func (w *Writer) Flush() error ``` @@ -260,7 +305,7 @@ to the underlying writer. -### func (\*Writer) Next +### func (\*Writer) Next ``` go func (w *Writer) Next(n int) ([]byte, error) ``` @@ -273,7 +318,7 @@ the size of the returned buffer. -### func (\*Writer) ReadFrom +### func (\*Writer) ReadFrom ``` go func (w *Writer) ReadFrom(r io.Reader) (int64, error) ``` @@ -281,7 +326,7 @@ ReadFrom implements `io.ReaderFrom` -### func (\*Writer) Write +### func (\*Writer) Write ``` go func (w *Writer) Write(p []byte) (int, error) ``` @@ -289,7 +334,7 @@ Write implements `io.Writer` -### func (\*Writer) WriteByte +### func (\*Writer) WriteByte ``` go func (w *Writer) WriteByte(b byte) error ``` @@ -297,7 +342,7 @@ WriteByte implements `io.ByteWriter` -### func (\*Writer) WriteString +### func (\*Writer) WriteString ``` go func (w *Writer) WriteString(s string) (int, error) ``` @@ -310,6 +355,5 @@ WriteString is analogous to Write, but it takes a string. - - - - -Generated by [godoc2md](http://godoc.org/github.com/davecheney/godoc2md) \ No newline at end of file +Generated by [godoc2md](https://github.com/davecheney/godoc2md) From 2ae61b800e4526d6b6538e26512b8ea1845550ea Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 11 Dec 2022 16:06:25 +0100 Subject: [PATCH 16/23] add go.mod I used go1.14 as version, to match other projects consuming this module (e.g. github.com/tinylib/msgp). Signed-off-by: Sebastiaan van Stijn --- go.mod | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 go.mod diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..9e6b657 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module github.com/philhofer/fwd + +go 1.15 From fbbf4953d98612dac320a9051da34e3e761d1df7 Mon Sep 17 00:00:00 2001 From: Dan Kegel Date: Tue, 11 Jun 2024 16:25:54 -0700 Subject: [PATCH 17/23] unsafestr: update to use unsafe.Slice() Adds support for tinygo 0.32, thereby fixing https://github.com/philhofer/fwd/issues/29 Passes tests on macos intel: - go 1.21.9: - go test - go 1.21.9 / tinygo 0.32 prerelease: - tinygo test - tinygo test -target wasi --- README.md | 9 +++++++++ go.mod | 2 +- writer_tinygo.go | 8 +------- 3 files changed, 11 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 62bd5c6..4e99523 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,15 @@ the write position by the length of the returned slice. This allows users to write directly to the end of the buffer. +## Portability + +Because it uses the unsafe package, there are theoretically +no promises about forward or backward portability. + +To stay compatible with tinygo 0.32, unsafestr() has been updated +to use unsafe.Slice() as suggested by +https://tinygo.org/docs/guides/compatibility, which also required +bumping go.mod to require at least go 1.20. ## Index diff --git a/go.mod b/go.mod index 9e6b657..77ccd8f 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ module github.com/philhofer/fwd -go 1.15 +go 1.20 // unsafe.StringData requires go1.20 or later diff --git a/writer_tinygo.go b/writer_tinygo.go index b060faf..c98cd57 100644 --- a/writer_tinygo.go +++ b/writer_tinygo.go @@ -4,16 +4,10 @@ package fwd import ( - "reflect" "unsafe" ) // unsafe cast string as []byte func unsafestr(b string) []byte { - l := uintptr(len(b)) - return *(*[]byte)(unsafe.Pointer(&reflect.SliceHeader{ - Len: l, - Cap: l, - Data: (*reflect.StringHeader)(unsafe.Pointer(&b)).Data, - })) + return unsafe.Slice(unsafe.StringData(b), len(b)) } From cdc6af6b87625c2806f04ee3f3e59ed8cb27115d Mon Sep 17 00:00:00 2001 From: Michal Hruby Date: Thu, 13 Jun 2024 05:03:59 +0100 Subject: [PATCH 18/23] Add Reader.InputOffset (#31) Add Reader.InputOffset (c.f. `json.Decoder.InputOffset`) --- reader.go | 25 +++++++++++++-- reader_test.go | 82 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 3 deletions(-) diff --git a/reader.go b/reader.go index 7c21f8f..88485d2 100644 --- a/reader.go +++ b/reader.go @@ -82,9 +82,10 @@ type Reader struct { r io.Reader // underlying reader // data[n:len(data)] is buffered data; data[len(data):cap(data)] is free buffer space - data []byte // data - n int // read offset - state error // last read error + data []byte // data + n int // read offset + inputOffset int64 // offset in the input stream + state error // last read error // if the reader past to NewReader was // also an io.Seeker, this is non-nil @@ -97,6 +98,7 @@ func (r *Reader) Reset(rd io.Reader) { r.r = rd r.data = r.data[0:0] r.n = 0 + r.inputOffset = 0 r.state = nil if s, ok := rd.(io.Seeker); ok { r.rs = s @@ -158,6 +160,9 @@ func (r *Reader) Buffered() int { return len(r.data) - r.n } // BufferSize returns the total size of the buffer func (r *Reader) BufferSize() int { return cap(r.data) } +// InputOffset returns the input stream byte offset of the current reader position +func (r *Reader) InputOffset() int64 { return r.inputOffset } + // Peek returns the next 'n' buffered bytes, // reading from the underlying reader if necessary. // It will only return a slice shorter than 'n' bytes @@ -197,10 +202,12 @@ func (r *Reader) discard(n int) int { inbuf := r.buffered() if inbuf <= n { r.n = 0 + r.inputOffset += int64(inbuf) r.data = r.data[:0] return inbuf } r.n += n + r.inputOffset += int64(n) return n } @@ -228,6 +235,7 @@ func (r *Reader) Skip(n int) (int, error) { // if we can Seek() through the remaining bytes, do that if n > skipped && r.rs != nil { nn, err := r.rs.Seek(int64(n-skipped), 1) + r.inputOffset += nn return int(nn) + skipped, err } // otherwise, keep filling the buffer @@ -267,6 +275,7 @@ func (r *Reader) Next(n int) ([]byte, error) { } out := r.data[r.n : r.n+n] r.n += n + r.inputOffset += int64(n) return out, nil } @@ -277,6 +286,7 @@ func (r *Reader) Read(b []byte) (int, error) { if r.buffered() != 0 { x := copy(b, r.data[r.n:]) r.n += x + r.inputOffset += int64(x) return x, nil } var n int @@ -293,6 +303,9 @@ func (r *Reader) Read(b []byte) (int, error) { if n == 0 { return 0, r.err() } + + r.inputOffset += int64(n) + return n, nil } @@ -312,9 +325,11 @@ func (r *Reader) ReadFull(b []byte) (int, error) { nn = copy(b[n:], r.data[r.n:]) n += nn r.n += nn + r.inputOffset += int64(nn) } else if l-n > cap(r.data) { nn, r.state = r.r.Read(b[n:]) n += nn + r.inputOffset += int64(nn) } else { r.more() } @@ -335,6 +350,8 @@ func (r *Reader) ReadByte() (byte, error) { } b := r.data[r.n] r.n++ + r.inputOffset++ + return b, nil } @@ -354,6 +371,7 @@ func (r *Reader) WriteTo(w io.Writer) (int64, error) { } r.data = r.data[0:0] r.n = 0 + r.inputOffset += int64(ii) } for r.state == nil { // here we just do @@ -367,6 +385,7 @@ func (r *Reader) WriteTo(w io.Writer) (int64, error) { } r.data = r.data[0:0] r.n = 0 + r.inputOffset += int64(ii) } } if r.state != io.EOF { diff --git a/reader_test.go b/reader_test.go index dfee298..21da9b9 100644 --- a/reader_test.go +++ b/reader_test.go @@ -382,6 +382,88 @@ func TestReadFull(t *testing.T) { } } +func TestInputOffset(t *testing.T) { + bts := randomBts(1024) + rd := NewReaderSize(partialReader{bytes.NewReader(bts)}, 128) + + if rd.InputOffset() != 0 { + t.Errorf("expected offset 0; got %d", rd.InputOffset()) + } + + // read a few bytes + rd.ReadFull(make([]byte, 10)) + + if rd.InputOffset() != 10 { + t.Errorf("expected offset 10; got %d", rd.InputOffset()) + } + + rd.Peek(384) + + // peeking doesn't advance the offset + if rd.InputOffset() != 10 { + t.Errorf("expected offset 10; got %d", rd.InputOffset()) + } + + rd.Next(246) + + if rd.InputOffset() != 256 { + t.Errorf("expected offset 256; got %d", rd.InputOffset()) + } + + rd.Skip(128) + + if rd.InputOffset() != 384 { + t.Errorf("expected offset 384; got %d", rd.InputOffset()) + } + + rd.ReadByte() + + if rd.InputOffset() != 385 { + t.Errorf("expected offset 385; got %d", rd.InputOffset()) + } + + n, _ := rd.Read(make([]byte, 128)) + + if rd.InputOffset() != int64(385+n) { + t.Errorf("expected offset %d; got %d", 385+n, rd.InputOffset()) + } + + rd.WriteTo(ioutil.Discard) + + if rd.InputOffset() != 1024 { + t.Errorf("expected offset 1024; got %d", rd.InputOffset()) + } + + // try to read more + _, err := rd.Read(make([]byte, 32)) + if err != io.EOF { + t.Fatalf("expected error %q; got %q", io.EOF, err) + } + + if rd.InputOffset() != 1024 { + t.Errorf("expected offset 1024; got %d", rd.InputOffset()) + } + + // reset the reader + rd.Reset(bytes.NewReader(bts)) + + if rd.InputOffset() != 0 { + t.Errorf("expected offset 0; got %d", rd.InputOffset()) + } + + rd.Skip(768 + 32) + + if rd.InputOffset() != 800 { + t.Errorf("expected offset 800; got %d", rd.InputOffset()) + } + + rd.WriteTo(ioutil.Discard) + + if rd.InputOffset() != 1024 { + t.Errorf("expected offset 1024; got %d", rd.InputOffset()) + } +} + type readCounter struct { r io.Reader count int From 20a13a1f6b7cb47a126dcb75152e21e1383bbaba Mon Sep 17 00:00:00 2001 From: Klaus Post Date: Mon, 16 Sep 2024 07:44:58 -0700 Subject: [PATCH 19/23] Add PeekByte (#32) Adds call to get a single byte and add Next happy path. --- reader.go | 45 ++++++++++++++++++++++++++++++++++++++++++++- reader_test.go | 29 +++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/reader.go b/reader.go index 88485d2..a24a896 100644 --- a/reader.go +++ b/reader.go @@ -196,6 +196,38 @@ func (r *Reader) Peek(n int) ([]byte, error) { return r.data[r.n : r.n+n], nil } +func (r *Reader) PeekByte() (b byte, err error) { + if len(r.data)-r.n >= 1 { + b = r.data[r.n] + } else { + b, err = r.peekByte() + } + return +} + +func (r *Reader) peekByte() (byte, error) { + const n = 1 + if cap(r.data) < n { + old := r.data[r.n:] + r.data = make([]byte, n+r.buffered()) + r.data = r.data[:copy(r.data, old)] + r.n = 0 + } + + // keep filling until + // we hit an error or + // read enough bytes + for r.buffered() < n && r.state == nil { + r.more() + } + + // we must have hit an error + if r.buffered() < n { + return 0, r.err() + } + return r.data[r.n], nil +} + // discard(n) discards up to 'n' buffered bytes, and // and returns the number of bytes discarded func (r *Reader) discard(n int) int { @@ -256,7 +288,18 @@ func (r *Reader) Skip(n int) (int, error) { // If an the returned slice is less than the // length asked for, an error will be returned, // and the reader position will not be incremented. -func (r *Reader) Next(n int) ([]byte, error) { +func (r *Reader) Next(n int) (b []byte, err error) { + if r.state == nil && len(r.data)-r.n >= n { + b = r.data[r.n : r.n+n] + r.n += n + r.inputOffset += int64(n) + } else { + b, err = r.next(n) + } + return +} + +func (r *Reader) next(n int) ([]byte, error) { // in case the buffer is too small if cap(r.data) < n { old := r.data[r.n:] diff --git a/reader_test.go b/reader_test.go index 21da9b9..da45b77 100644 --- a/reader_test.go +++ b/reader_test.go @@ -304,6 +304,35 @@ func TestPeek(t *testing.T) { } } +func TestPeekByte(t *testing.T) { + bts := randomBts(1024) + rd := NewReaderSize(partialReader{bytes.NewReader(bts)}, 200) + + // first, a peek < buffer size + var ( + peek byte + err error + ) + rd.Skip(100) + peek, err = rd.PeekByte() + if err != nil { + t.Fatal(err) + } + if peek != bts[100] { + t.Fatalf("peeked byte not equal: want %d got %d", bts[100], peek) + } + + // now, a peek > buffer size + rd.Skip(156) + peek, err = rd.PeekByte() + if err != nil { + t.Fatal(err) + } + if peek != bts[256] { + t.Fatalf("peeked byte not equal: want %d got %d", bts[256], peek) + } +} + func TestNext(t *testing.T) { size := 1024 bts := randomBts(size) From d287475b93ef642450d3c706d847e99abf63110e Mon Sep 17 00:00:00 2001 From: Sash Date: Sat, 17 Oct 2026 03:41:37 +0700 Subject: [PATCH 20/23] Keep bytes when a read returns them with an error. (#38) Reader.more dropped n > 0 when the error was not EOF. Those bytes are now buffered, and the error is returned after the caller has them. --- reader.go | 36 +++++++++++++++++++++++++----------- reader_test.go | 49 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 11 deletions(-) diff --git a/reader.go b/reader.go index a24a896..378b182 100644 --- a/reader.go +++ b/reader.go @@ -126,14 +126,17 @@ func (r *Reader) more() { if a == 0 && r.state == nil { r.state = io.ErrNoProgress return - } else if a > 0 && r.state == io.EOF { - // discard the io.EOF if we read more than 0 bytes. - // the next call to Read should return io.EOF again. - r.state = nil - } else if r.state != nil { - return } - r.data = r.data[:len(r.data)+a] + // io.Reader requires the caller to process n > 0 before the error. + // A non-EOF error used to return here and drop those bytes. + if a > 0 { + r.data = r.data[:len(r.data)+a] + if r.state == io.EOF { + // discard the io.EOF if we read more than 0 bytes. + // the next call to Read should return io.EOF again. + r.state = nil + } + } } // pop error @@ -332,6 +335,11 @@ func (r *Reader) Read(b []byte) (int, error) { r.inputOffset += int64(x) return x, nil } + // The buffer was already returned. Surface the error from that read + // before asking the underlying reader again. + if r.state != nil { + return 0, r.err() + } var n int // we have no buffered data; determine // whether or not to buffer or call @@ -363,19 +371,25 @@ func (r *Reader) ReadFull(b []byte) (int, error) { // either read buffered data, // or read directly for the underlying // buffer, or fetch more buffered data. - for n < l && r.state == nil { + // Buffered bytes are copied even when the fill also returned an error. + for n < l { if r.buffered() != 0 { nn = copy(b[n:], r.data[r.n:]) n += nn r.n += nn r.inputOffset += int64(nn) - } else if l-n > cap(r.data) { + continue + } + if r.state != nil { + break + } + if l-n > cap(r.data) { nn, r.state = r.r.Read(b[n:]) n += nn r.inputOffset += int64(nn) - } else { - r.more() + continue } + r.more() } if n < l { return n, r.noEOF() diff --git a/reader_test.go b/reader_test.go index da45b77..3a014a5 100644 --- a/reader_test.go +++ b/reader_test.go @@ -561,3 +561,52 @@ func TestReaderBufCreation(t *testing.T) { } } } + +// bytesWithError returns its bytes and a non-EOF error once, then io.EOF. +type bytesWithError struct { + b []byte + err error + did bool +} + +func (r *bytesWithError) Read(p []byte) (int, error) { + if r.did { + return 0, io.EOF + } + r.did = true + return copy(p, r.b), r.err +} + +func TestReadKeepsBytesReturnedWithError(t *testing.T) { + want := []byte{1, 2, 3, 4} + + rd := NewReaderSize(&bytesWithError{b: want, err: io.ErrClosedPipe}, 32) + got, err := rd.Next(len(want)) + if err != nil { + t.Fatalf("Next: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("Next got %x, want %x", got, want) + } + if _, err = rd.Next(1); err != io.ErrClosedPipe { + t.Fatalf("Next after bytes: got %v, want %v", err, io.ErrClosedPipe) + } + + rd = NewReaderSize(&bytesWithError{b: want, err: io.ErrClosedPipe}, 32) + buf := make([]byte, len(want)) + n, err := rd.Read(buf) + if err != nil || n != len(want) || !bytes.Equal(buf[:n], want) { + t.Fatalf("Read: n=%d err=%v buf=%x", n, err, buf[:n]) + } + n, err = rd.Read(buf) + if n != 0 || err != io.ErrClosedPipe { + t.Fatalf("Read after bytes: n=%d err=%v, want 0, %v", n, err, io.ErrClosedPipe) + } + + rd = NewReaderSize(&bytesWithError{b: want, err: io.ErrClosedPipe}, 32) + out := make([]byte, 8) + n, err = rd.ReadFull(out) + if n != len(want) || err != io.ErrClosedPipe || !bytes.Equal(out[:n], want) { + t.Fatalf("ReadFull: n=%d err=%v out=%x", n, err, out[:n]) + } +} From 716382dfadc2589d17af3f46045b31f65a9cb6ea Mon Sep 17 00:00:00 2001 From: Phil Hofer Date: Fri, 16 Oct 2026 13:49:11 -0700 Subject: [PATCH 21/23] Reader: Skip: use Seek return value correctly --- reader.go | 9 ++++++--- reader_test.go | 8 ++++++++ 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/reader.go b/reader.go index 378b182..c87d0ef 100644 --- a/reader.go +++ b/reader.go @@ -269,9 +269,12 @@ func (r *Reader) Skip(n int) (int, error) { // if we can Seek() through the remaining bytes, do that if n > skipped && r.rs != nil { - nn, err := r.rs.Seek(int64(n-skipped), 1) - r.inputOffset += nn - return int(nn) + skipped, err + offs, err := r.rs.Seek(int64(n-skipped), io.SeekCurrent) + if err == nil { + r.inputOffset = offs + return n, nil + } + return skipped, err } // otherwise, keep filling the buffer // and discarding it up to 'n' diff --git a/reader_test.go b/reader_test.go index 3a014a5..d3daf12 100644 --- a/reader_test.go +++ b/reader_test.go @@ -237,6 +237,10 @@ func TestSkipSeek(t *testing.T) { t.Fatalf("Skip() returned a nil error, but skipped %d bytes instead of %d", n, 10) } + if rd.InputOffset() != 512+1+10 { + t.Errorf("input offset = %d", rd.InputOffset()) + } + // now try to skip past the end rd.Reset(bytes.NewReader(bts)) @@ -251,6 +255,10 @@ func TestSkipSeek(t *testing.T) { t.Fatalf("should have returned %d bytes; returned %d", 2000, n) } + if rd.InputOffset() != 2000 { + t.Fatalf("input offset: %d", rd.InputOffset()) + } + // the next call to Read() // should return io.EOF n, err = rd.Read([]byte{0, 0, 0}) From 59b5012970af5470e7f7a49574a3629fee115584 Mon Sep 17 00:00:00 2001 From: Phil Hofer Date: Fri, 16 Oct 2026 13:56:36 -0700 Subject: [PATCH 22/23] Reader: handle ESPIPE when attempting to Seek If we get ESPIPE from io.Seeker.Seek, don't try seeking again and just swallow the error. Closes #17. --- reader.go | 10 +++++++++- reader_test.go | 23 +++++++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/reader.go b/reader.go index c87d0ef..4b31385 100644 --- a/reader.go +++ b/reader.go @@ -34,8 +34,10 @@ package fwd import ( + "errors" "io" "os" + "syscall" ) const ( @@ -274,7 +276,13 @@ func (r *Reader) Skip(n int) (int, error) { r.inputOffset = offs return n, nil } - return skipped, err + // if the io.Seeker is not actually usable, + // just swallow the error and don't try + // seeking again + if !errors.Is(err, syscall.ESPIPE) { + return skipped, err + } + r.rs = nil } // otherwise, keep filling the buffer // and discarding it up to 'n' diff --git a/reader_test.go b/reader_test.go index d3daf12..378fd26 100644 --- a/reader_test.go +++ b/reader_test.go @@ -5,6 +5,7 @@ import ( "io" "io/ioutil" "math/rand" + "os" "testing" "unsafe" ) @@ -618,3 +619,25 @@ func TestReadKeepsBytesReturnedWithError(t *testing.T) { t.Fatalf("ReadFull: n=%d err=%v out=%x", n, err, out[:n]) } } + +func TestPipeSeek(t *testing.T) { + r, w, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + defer r.Close() + body := "hello, world!\n" + fr := NewReader(r) + go func() { + defer w.Close() + io.WriteString(w, body) + }() + + n, err := fr.Skip(len(body)) + if err != nil { + t.Fatalf("Skip: %s", err) + } + if n != len(body) { + t.Fatalf("skipped %d instead of %d bytes?", n, len(body)) + } +} From 61bf48afaf8cb4db3ff637fc94175b18ff6e26c3 Mon Sep 17 00:00:00 2001 From: Phil Hofer Date: Fri, 16 Oct 2026 14:01:21 -0700 Subject: [PATCH 23/23] Writer: unsafe: modernize to use unsafe.Slice/unsafe.StringData --- writer_unsafe.go | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/writer_unsafe.go b/writer_unsafe.go index e4cb4a8..dcb7ebf 100644 --- a/writer_unsafe.go +++ b/writer_unsafe.go @@ -4,17 +4,10 @@ package fwd import ( - "reflect" "unsafe" ) // unsafe cast string as []byte func unsafestr(s string) []byte { - var b []byte - sHdr := (*reflect.StringHeader)(unsafe.Pointer(&s)) - bHdr := (*reflect.SliceHeader)(unsafe.Pointer(&b)) - bHdr.Data = sHdr.Data - bHdr.Len = sHdr.Len - bHdr.Cap = sHdr.Len - return b + return unsafe.Slice(unsafe.StringData(s), len(s)) }