buffer/lib_test.go

package buffer_test

import (
	"bufio"
	"bytes"
	"code.squareroundforest.org/arpio/buffer"
	"errors"
	"io"
	"testing"
)

func TestLib(t *testing.T) {
	t.Run("default pool", func(t *testing.T) {
		t.Run("buffered reader", func(t *testing.T) {
			g := &gen{max: 1 << 18}
			r := buffer.BufferedReader(g, buffer.Options{})
			b, err := io.ReadAll(r)
			if err != nil {
				t.Fatal(err)
			}

			if !bytes.Equal(b, generate(1<<18)) {
				t.Fatal("output does not match", len(b))
			}
		})

		t.Run("buffered content", func(t *testing.T) {
			c := buffer.ContentFunc(func(w io.Writer) (int64, error) {
				g := &gen{max: 1 << 18}
				return io.Copy(w, g)
			})

			r := buffer.BufferedContent(c, buffer.Options{})
			b, err := io.ReadAll(r)
			if err != nil {
				t.Fatal(err)
			}

			if !bytes.Equal(b, generate(1<<18)) {
				t.Fatal("output does not match", len(b))
			}
		})
	})

	t.Run("zero reader", func(t *testing.T) {
		t.Run("buffered reader", func(t *testing.T) {
			r := buffer.BufferedReader(nil, buffer.Options{})
			b, err := io.ReadAll(r)
			if err != nil {
				t.Fatal(err)
			}

			if len(b) != 0 {
				t.Fatal("output does not match", len(b))
			}
		})

		t.Run("buffered content", func(t *testing.T) {
			r := buffer.BufferedContent(nil, buffer.Options{})
			b, err := io.ReadAll(r)
			if err != nil {
				t.Fatal(err)
			}

			if len(b) != 0 {
				t.Fatal("output does not match", len(b))
			}
		})
	})

	t.Run("uninitialized reader", func(t *testing.T) {
		t.Run("read", func(t *testing.T) {
			var r buffer.Reader
			p := make([]byte, 512)
			n, err := r.Read(p)
			if !errors.Is(err, io.EOF) {
				t.Fatal(err)
			}

			if n != 0 {
				t.Fatal(n)
			}
		})

		t.Run("read bytes", func(t *testing.T) {
			var r buffer.Reader
			b, ok, err := r.ReadBytes([]byte("123"), 512)
			if !errors.Is(err, io.EOF) {
				t.Fatal(err)
			}

			if ok {
				t.Fatal(ok)
			}

			if len(b) != 0 {
				t.Fatal(len(b))
			}
		})

		t.Run("read utf8", func(t *testing.T) {
			var r buffer.Reader
			runes, n, err := r.ReadUTF8(512)
			if !errors.Is(err, io.EOF) {
				t.Fatal(err)
			}

			if n != 0 {
				t.Fatal(n)
			}

			if len(runes) != 0 {
				t.Fatal(len(runes))
			}
		})

		t.Run("peek", func(t *testing.T) {
			var r buffer.Reader
			b, err := r.Peek(512)
			if !errors.Is(err, io.EOF) {
				t.Fatal(err)
			}

			if len(b) != 0 {
				t.Fatal(len(b))
			}
		})

		t.Run("buffered", func(t *testing.T) {
			var r buffer.Reader
			b := r.Buffered()
			if len(b) != 0 {
				t.Fatal(len(b))
			}
		})

		t.Run("write to", func(t *testing.T) {
			var (
				r buffer.Reader
				b bytes.Buffer
			)

			n, err := r.WriteTo(&b)
			if err != nil {
				t.Fatal(err)
			}

			if n != 0 {
				t.Fatal(n)
			}
		})
	})
}

// -- bench

type readerOnly struct {
	in io.Reader
}

type writerOnly struct {
	in io.Writer
}

func (r readerOnly) Read(p []byte) (int, error) {
	return r.in.Read(p)
}

func (w writerOnly) Write(p []byte) (int, error) {
	return w.in.Write(p)
}

func TestBenchmarkThroughput(t *testing.T) {
	p := buffer.NoPool(0)
	wo := writerOnly{io.Discard}
	src := &gen{max: 1 << 18}
	r := buffer.BufferedReader(src, buffer.Options{Pool: p})
	ro := readerOnly{r}
	n, err := io.Copy(wo, ro)
	if n != 1<<18 || err != nil {
		t.Fatal(n, err)
	}
}

func TestBenchmarkThroughputCompare(t *testing.T) {
	wo := writerOnly{io.Discard}
	src := &gen{max: 1 << 18}
	r := bufio.NewReader(src)
	ro := readerOnly{r}
	n, err := io.Copy(wo, ro)
	if n != 1<<18 || err != nil {
		t.Fatal(n, err)
	}
}

func BenchmarkThroughput(b *testing.B) {
	p := buffer.NoPool(0)
	wo := writerOnly{io.Discard}
	for i := 0; i < b.N; i++ {
		src := &gen{max: 1 << 18}
		r := buffer.BufferedReader(src, buffer.Options{Pool: p})
		ro := readerOnly{r}
		io.Copy(wo, ro)
	}
}

func BenchmarkThroughputCompare(b *testing.B) {
	wo := writerOnly{io.Discard}
	for i := 0; i < b.N; i++ {
		src := &gen{max: 1 << 18}
		r := bufio.NewReader(src)
		ro := readerOnly{r}
		io.Copy(wo, ro)
	}
}

func TestBenchmarkThroughputPooled(t *testing.T) {
	p := &foreverPool{allocSize: 1 << 12}
	wo := writerOnly{io.Discard}
	src := &gen{max: 1 << 18}
	r := buffer.BufferedReader(src, buffer.Options{Pool: p})
	ro := readerOnly{r}
	n, err := io.Copy(wo, ro)
	if n != 1<<18 || err != nil {
		t.Fatal(n, err)
	}
}

func TestBenchmarkThroughputPooledCompare(t *testing.T) {
	wo := writerOnly{io.Discard}
	r := bufio.NewReader(nil)
	src := &gen{max: 1 << 18}
	r.Reset(src)
	ro := readerOnly{r}
	n, err := io.Copy(wo, ro)
	if n != 1<<18 || err != nil {
		t.Fatal(n, err)
	}
}

func BenchmarkThroughputPooled(b *testing.B) {
	p := &foreverPool{allocSize: 1 << 12}
	wo := writerOnly{io.Discard}
	for i := 0; i < b.N; i++ {
		src := &gen{max: 1 << 18}
		r := buffer.BufferedReader(src, buffer.Options{Pool: p})
		ro := readerOnly{r}
		io.Copy(wo, ro)
	}
}

func BenchmarkThroughputPooledCompare(b *testing.B) {
	wo := writerOnly{io.Discard}
	r := bufio.NewReader(nil)
	for i := 0; i < b.N; i++ {
		src := &gen{max: 1 << 18}
		r.Reset(src)
		ro := readerOnly{r}
		io.Copy(wo, ro)
	}
}

func TestBenchmarkThroughputPooledParallel(t *testing.T) {
	p := newSyncedForeverPool(func() []byte {
		return make([]byte, 1<<12)
	})

	wo := writerOnly{io.Discard}
	src := &gen{max: 1 << 18}
	r := buffer.BufferedReader(src, buffer.Options{Pool: p})
	ro := readerOnly{r}
	n, err := io.Copy(wo, ro)
	if n != 1<<18 || err != nil {
		t.Fatal(n, err)
	}
}

func TestBenchmarkThroughputPooledParallelCompare(t *testing.T) {
	p := newSyncedForeverPool(func() *bufio.Reader {
		return bufio.NewReader(nil)
	})

	wo := writerOnly{io.Discard}
	r, err := p.Get()
	if err != nil {
		t.Fatal(err)
	}

	src := &gen{max: 1 << 18}
	r.Reset(src)
	ro := readerOnly{r}
	n, err := io.Copy(wo, ro)
	if n != 1<<18 || err != nil {
		t.Fatal(n, err)
	}
}

func BenchmarkThroughputPooledParallel(b *testing.B) {
	p := newSyncedForeverPool(func() []byte {
		return make([]byte, 1<<12)
	})

	b.ResetTimer()
	b.SetParallelism(128)
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			src := &gen{max: 1 << 18}
			r := buffer.BufferedReader(src, buffer.Options{Pool: p})
			ro := readerOnly{r}
			wo := writerOnly{io.Discard}
			io.Copy(wo, ro)
		}
	})
}

func BenchmarkThroughputPooledParallelCompare(b *testing.B) {
	p := newSyncedForeverPool(func() *bufio.Reader {
		return bufio.NewReader(nil)
	})

	b.ResetTimer()
	b.SetParallelism(128)
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			r, _ := p.Get()
			src := &gen{max: 1 << 18}
			r.Reset(src)
			ro := readerOnly{r}
			wo := writerOnly{io.Discard}
			io.Copy(wo, ro)
		}
	})
}

func TestBenchmarkScan(t *testing.T) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := buffer.NoPool(0)
	src := &gen{
		max:                1 << 18,
		customContentAfter: []int{delimiterPosition},
		customContent:      map[int][]byte{delimiterPosition: []byte("123")},
	}

	r := buffer.BufferedReader(src, buffer.Options{Pool: p})
	b, ok, err := r.ReadBytes([]byte{'1'}, 1<<18)
	if err != nil {
		t.Fatal(err)
	}

	if !ok {
		t.Fatal("delimiter")
	}

	if !bytes.Equal(b, append(generate(delimiterPosition), '1')) {
		t.Fatal("content")
	}
}

func TestBenchmarkScanCompare(t *testing.T) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	src := &gen{
		max:                1 << 18,
		customContentAfter: []int{delimiterPosition},
		customContent:      map[int][]byte{delimiterPosition: []byte("123")},
	}

	r := bufio.NewReader(src)
	b, err := r.ReadBytes('1')
	if err != nil {
		t.Fatal(err)
	}

	if !bytes.Equal(b, append(generate(delimiterPosition), '1')) {
		t.Fatal("content")
	}
}

func BenchmarkScan(b *testing.B) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := buffer.NoPool(0)
	for i := 0; i < b.N; i++ {
		src := &gen{
			max:                1 << 18,
			customContentAfter: []int{delimiterPosition},
			customContent:      map[int][]byte{delimiterPosition: []byte("123")},
		}

		r := buffer.BufferedReader(src, buffer.Options{Pool: p})
		r.ReadBytes([]byte{'1'}, 1<<18)
	}
}

func BenchmarkScanCompare(b *testing.B) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	for i := 0; i < b.N; i++ {
		src := &gen{
			max:                1 << 18,
			customContentAfter: []int{delimiterPosition},
			customContent:      map[int][]byte{delimiterPosition: []byte("123")},
		}

		r := bufio.NewReader(src)
		r.ReadBytes('1')
	}
}

func TestBenchmarkScanPooled(t *testing.T) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := &foreverPool{allocSize: 1 << 12}
	src := &gen{
		max:                1 << 18,
		customContentAfter: []int{delimiterPosition},
		customContent:      map[int][]byte{delimiterPosition: []byte("123")},
	}

	r := buffer.BufferedReader(src, buffer.Options{Pool: p})
	b, ok, err := r.ReadBytes([]byte{'1'}, 1<<18)
	if err != nil {
		t.Fatal(err)
	}

	if !ok {
		t.Fatal("delimiter")
	}

	if !bytes.Equal(b, append(generate(delimiterPosition), '1')) {
		t.Fatal("content")
	}
}

func TestBenchmarkScanPooledCompare(t *testing.T) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	src := &gen{
		max:                1 << 18,
		customContentAfter: []int{delimiterPosition},
		customContent:      map[int][]byte{delimiterPosition: []byte("123")},
	}

	r := bufio.NewReader(nil)
	r.Reset(src)
	b, err := r.ReadBytes('1')
	if err != nil {
		t.Fatal(err)
	}

	if !bytes.Equal(b, append(generate(delimiterPosition), '1')) {
		t.Fatal("content")
	}
}

func BenchmarkScanPooled(b *testing.B) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := &foreverPool{allocSize: 1 << 12}
	for i := 0; i < b.N; i++ {
		src := &gen{
			max:                1 << 18,
			customContentAfter: []int{delimiterPosition},
			customContent:      map[int][]byte{delimiterPosition: []byte("123")},
		}

		r := buffer.BufferedReader(src, buffer.Options{Pool: p})
		r.ReadBytes([]byte{'1'}, 1<<18)
	}
}

func BenchmarkScanPooledCompare(b *testing.B) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	r := bufio.NewReader(nil)
	for i := 0; i < b.N; i++ {
		src := &gen{
			max:                1 << 18,
			customContentAfter: []int{delimiterPosition},
			customContent:      map[int][]byte{delimiterPosition: []byte("123")},
		}

		r.Reset(src)
		r.ReadBytes('1')
	}
}

func TestBenchmarkScanPooledParallel(t *testing.T) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := newSyncedForeverPool(func() []byte {
		return make([]byte, 1<<12)
	})

	src := &gen{
		max:                1 << 18,
		customContentAfter: []int{delimiterPosition},
		customContent:      map[int][]byte{delimiterPosition: []byte("123")},
	}

	r := buffer.BufferedReader(src, buffer.Options{Pool: p})
	b, ok, err := r.ReadBytes([]byte{'1'}, 1<<18)
	if err != nil {
		t.Fatal(err)
	}

	if !ok {
		t.Fatal("delimiter")
	}

	if !bytes.Equal(b, append(generate(delimiterPosition), '1')) {
		t.Fatal("content")
	}
}

func TestBenchmarkScanPooledParallelCompare(t *testing.T) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := newSyncedForeverPool(func() *bufio.Reader {
		return bufio.NewReader(nil)
	})

	src := &gen{
		max:                1 << 18,
		customContentAfter: []int{delimiterPosition},
		customContent:      map[int][]byte{delimiterPosition: []byte("123")},
	}

	r, err := p.Get()
	if err != nil {
		t.Fatal(err)
	}

	r.Reset(src)
	b, err := r.ReadBytes('1')
	if err != nil {
		t.Fatal(err)
	}

	if !bytes.Equal(b, append(generate(delimiterPosition), '1')) {
		t.Fatal("content")
	}
}

func BenchmarkScanPooledParallel(b *testing.B) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := newSyncedForeverPool(func() []byte {
		return make([]byte, 1<<12)
	})

	b.ResetTimer()
	b.SetParallelism(128)
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			src := &gen{
				max:                1 << 18,
				customContentAfter: []int{delimiterPosition},
				customContent:      map[int][]byte{delimiterPosition: []byte("123")},
			}

			r := buffer.BufferedReader(src, buffer.Options{Pool: p})
			r.ReadBytes([]byte{'1'}, 1<<18)
		}
	})
}

func BenchmarkScanPooledParallelCompare(b *testing.B) {
	const delimiterPosition = 1<<17 + 1<<16 - 3
	p := newSyncedForeverPool(func() *bufio.Reader {
		return bufio.NewReader(nil)
	})

	b.ResetTimer()
	b.SetParallelism(128)
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			src := &gen{
				max:                1 << 18,
				customContentAfter: []int{delimiterPosition},
				customContent:      map[int][]byte{delimiterPosition: []byte("123")},
			}

			r, _ := p.Get()
			r.Reset(src)
			r.ReadBytes('1')
		}
	})
}