feat: add DisableAutoPipelining to serve requests from the conn pool

[rueian]

Auto pipelining can cause high p99 latencies (ref) and we are still trying to find a solution. This PR allows users to always use the pooled connections instead of pipelining commands into the same connections, except for client-side caching and pubsub requests.

The benchmark on 2 GCP n2-highcpu-2 machines, one runs redis-server, and the other one runs the following program, shows slightly better latencies and throughput than go-redis across concurrency 2, 4, 8, 16, and 32.

package main

import (
	"context"
	"fmt"
	"math/rand"
	"net/http"
	_ "net/http/pprof"
	"os"
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/redis/go-redis/v9"
	"github.com/redis/rueidis"
)

func prepData(n int) []string {
	data := make([]string, n)
	for i := range data {
		data[i] = strconv.Itoa(i)
	}
	rand.Shuffle(len(data), func(i, j int) { data[i], data[j] = data[j], data[i] })
	return data
}

const (
	keyCount = 1000000
)

func main() {
	useGoRedis := os.Args[1] == "goredis"
	concurrency, err := strconv.Atoi(os.Args[2])
	if err != nil {
		panic(err)
	}

	rand.Seed(time.Now().UnixNano())
	bucket := []float64{50, 100, 150, 200, 250, 300, 350, 400, 450, 500, 750, 1000, 1500, 2000, 3000, 4000}

	rl := promauto.NewHistogram(prometheus.HistogramOpts{
		Name: "micro_read_latency", Buckets: bucket,
		ConstLabels: map[string]string{"client": os.Args[1]},
	})

	go func() {
		http.Handle("/metrics", promhttp.Handler())
		http.ListenAndServe(":2112", nil)
	}()

	rc, err := rueidis.NewClient(rueidis.ClientOption{
		InitAddress:           []string{os.Args[3]},
		DisableAutoPipelining: true,
	})
	if err != nil {
		panic(err)
	}

	gc := redis.NewUniversalClient(&redis.UniversalOptions{
		Addrs: []string{os.Args[3]},
	})

	ctx := context.Background()

	goredisRead := func(key string) error {
		return gc.Get(ctx, key).Err()
	}
	rueidisRead := func(key string) error {
		return rc.Do(ctx, rc.B().Get().Key(key).Build()).Error()
	}

	var rfn func(key string) error

	if useGoRedis {
		rfn = goredisRead
	} else {
		rfn = rueidisRead
	}

	read1Fn := func(keys []string) {
		for _, k := range keys {
			ts := time.Now()
			err := rfn(k)
			rl.Observe(float64(time.Since(ts).Microseconds()))
			if err != nil {
				panic(err)
			}
		}
	}

	{
		keys := prepData(keyCount)
		data := prepData(keyCount)
		commands := make(rueidis.Commands, len(keys))
		for i := range commands {
			commands[i] = rc.B().Set().Key(keys[i]).Value(data[i]).Build()
		}
		ts := time.Now()
		for _, resp := range rc.DoMulti(ctx, commands...) {
			if err := resp.Error(); err != nil {
				panic(err)
			}
		}
		fmt.Println("ready", time.Since(ts))
	}

	if useGoRedis {
		rc.Close()
	} else {
		gc.Close()
	}

	for i := 0; i < concurrency; i++ {
		go func() {
			keys := prepData(keyCount)
			for {
				read1Fn(keys)
			}
		}()
	}
	time.Sleep(time.Minute * 2)
}