feat: streaming read with DoStream() and DoMultiStream()

[rueian]

This is a draft of #450.

Initial benchmark result with redjet v0.5.0:

▶ go test -bench . -benchtime=3s
goos: darwin
goarch: arm64
pkg: rueidis-benchmark/redjet
Benchmark/single/rueidis/payload1B-10         	  697377	      5117 ns/op	       0 B/op	       0 allocs/op
Benchmark/single/rueidis/payload1K-10         	  626092	      5241 ns/op	       0 B/op	       0 allocs/op
Benchmark/single/rueidis/payload1M-10         	   19602	    181655 ns/op	       1 B/op	       0 allocs/op
Benchmark/single/redjet/payload1B-10          	  657787	      5283 ns/op	     256 B/op	       4 allocs/op
Benchmark/single/redjet/payload1K-10          	  632977	      5408 ns/op	     256 B/op	       4 allocs/op
Benchmark/single/redjet/payload1M-10          	   17818	    201544 ns/op	     291 B/op	       5 allocs/op
Benchmark/pipe20/rueidis/payload1B-10         	  328735	     10990 ns/op	       0 B/op	       0 allocs/op
Benchmark/pipe20/rueidis/payload1K-10         	  276231	     12719 ns/op	       0 B/op	       0 allocs/op
Benchmark/pipe20/rueidis/payload1M-10         	     607	   5568295 ns/op	      46 B/op	       0 allocs/op
Benchmark/pipe20/redjet/payload1B-10          	  328198	     10975 ns/op	     560 B/op	      23 allocs/op
Benchmark/pipe20/redjet/payload1K-10          	  274506	     12821 ns/op	     560 B/op	      23 allocs/op
Benchmark/pipe20/redjet/payload1M-10          	     618	   5762557 ns/op	    1424 B/op	      43 allocs/op
PASS

Code:

package bench

import (
	"context"
	"io"
	"strings"
	"testing"

	"github.com/coder/redjet"
	"github.com/redis/rueidis"
)

func Benchmark(b *testing.B) {
	tests := [][2]string{
		{"payload1B", strings.Repeat("x", 1)},
		{"payload1K", strings.Repeat("x", 1024)},
		{"payload1M", strings.Repeat("x", 1024*1024)},
	}

	rj := redjet.New("127.0.0.1:6379")
	defer rj.Close()

	ru, err := rueidis.NewClient(rueidis.ClientOption{InitAddress: []string{"127.0.0.1:6379"}})
	if err != nil {
		panic(err)
	}
	defer ru.Close()

	for _, t := range tests {
		if err := ru.Do(context.Background(), ru.B().Set().Key(t[0]).Value(t[1]).Build()).Error(); err != nil {
			panic(err)
		}
	}

	b.Run("single", func(b *testing.B) {
		b.Run("rueidis", func(b *testing.B) {
			for _, t := range tests {
				b.Run(t[0], func(b *testing.B) {
					b.ResetTimer()
					b.ReportAllocs()
					b.RunParallel(func(pb *testing.PB) {
						for pb.Next() {
							s := ru.DoStream(context.Background(), ru.B().Get().Key(t[0]).Build())
							n, err := s.WriteTo(io.Discard)
							if err != nil {
								panic(err)
							}
							if n != int64(len(t[1])) {
								panic("wrong")
							}
						}
					})
				})
			}
		})

		b.Run("redjet", func(b *testing.B) {
			for _, t := range tests {
				b.Run(t[0], func(b *testing.B) {
					b.ResetTimer()
					b.ReportAllocs()
					b.RunParallel(func(pb *testing.PB) {
						for pb.Next() {
							n, err := rj.Command(context.Background(), "GET", t[0]).WriteTo(io.Discard)
							if err != nil {
								panic(err)
							}
							if n != int64(len(t[1])) {
								panic("wrong")
							}
						}
					})
				})
			}
		})
	})

	b.Run("pipe20", func(b *testing.B) {
		b.Run("rueidis", func(b *testing.B) {
			for _, t := range tests {
				b.Run(t[0], func(b *testing.B) {
					b.ResetTimer()
					b.ReportAllocs()
					b.RunParallel(func(pb *testing.PB) {
						cmds := make(rueidis.Commands, 20)
						for n := range cmds {
							cmds[n] = ru.B().Get().Key(t[0]).Build().Pin()
						}
						for pb.Next() {
							s := ru.DoMultiStream(context.Background(), cmds...)
							for range cmds {
								n, err := s.WriteTo(io.Discard)
								if err != nil {
									panic(err)
								}
								if n != int64(len(t[1])) {
									panic("wrong")
								}
							}
						}
					})
				})
			}
		})

		b.Run("redjet", func(b *testing.B) {
			for _, t := range tests {
				b.Run(t[0], func(b *testing.B) {
					b.ResetTimer()
					b.ReportAllocs()
					b.RunParallel(func(pb *testing.PB) {
						for pb.Next() {
							var p *redjet.Pipeline
							for n := 0; n < 20; n++ {
								p = rj.Pipeline(context.Background(), p, "GET", t[0])
							}
							for p.Next() {
								n, err := p.WriteTo(io.Discard)
								if err != nil {
									panic(err)
								}
								if n != int64(len(t[1])) {
									panic("wrong")
								}
							}
							p.Close()
						}
					})
				})
			}
		})
	})
}

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (a171bb7) 95.66% compared to head (7d97891) 95.71%.
Report is 8 commits behind head on main.

❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #460      +/-   ##
==========================================
+ Coverage   95.66%   95.71%   +0.04%     
==========================================
  Files          67       67              
  Lines       31821    32116     +295     
==========================================
+ Hits        30443    30741     +298     
+ Misses       1184     1182       -2     
+ Partials      194      193       -1     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.