README.md

@@ -460,7 +460,7 @@ module mymodule
 
 go 1.18
 
-require github.com/redis/rueidis v1.0.25-go1.18
+require github.com/redis/rueidis v1.0.26-go1.18
 ```
 
 ## Contributing

cluster.go

@@ -71,17 +71,18 @@ var retrycachep = util.NewPool(func(capacity int) *retrycache {
 })
 
 type clusterClient struct {
-	pslots [16384]conn
-	rslots []conn
-	opt    *ClientOption
-	conns  map[string]connrole
-	connFn connFn
-	sc     call
-	mu     sync.RWMutex
-	stop   uint32
-	cmd    Builder
-	retry  bool
-	aws    bool
+	pslots     [16384]conn
+	rslots     []conn
+	opt        *ClientOption
+	replicaOpt *ClientOption
+	conns      map[string]connrole
+	connFn     connFn
+	sc         call
+	mu         sync.RWMutex
+	stop       uint32
+	cmd        Builder
+	retry      bool
+	aws        bool
 }
 
 // NOTE: connrole and conn must be initialized at the same time
@@ -93,8 +94,8 @@ type connrole struct {
 func newClusterClient(opt *ClientOption, connFn connFn) (client *clusterClient, err error) {
 	client = &clusterClient{
 		cmd:    cmds.NewBuilder(cmds.InitSlot),
-		opt:    opt,
 		connFn: connFn,
+		opt:    opt,
 		conns:  make(map[string]connrole),
 		retry:  !opt.DisableRetry,
 		aws:    len(opt.InitAddress) == 1 && strings.Contains(opt.InitAddress[0], "amazonaws.com"),
@@ -104,6 +105,12 @@ func newClusterClient(opt *ClientOption, connFn connFn) (client *clusterClient,
 		return nil, ErrReplicaOnlyConflict
 	}
 
+	if opt.SendToReplicas != nil {
+		replicaOpt := *opt
+		replicaOpt.ReplicaOnly = true
+		client.replicaOpt = &replicaOpt
+	}
+
 	client.connFn = func(dst string, opt *ClientOption) conn {
 		cc := connFn(dst, opt)
 		cc.SetOnCloseHook(func(err error) {
@@ -225,7 +232,13 @@ func (c *clusterClient) _refresh() (err error) {
 	for master, g := range groups {
 		conns[master] = connrole{conn: c.connFn(master, c.opt), replica: false}
 		for _, addr := range g.nodes[1:] {
-			conns[addr] = connrole{conn: c.connFn(addr, c.opt), replica: true}
+			var cc conn
+			if c.opt.SendToReplicas != nil {
+				cc = c.connFn(addr, c.replicaOpt)
+			} else {
+				cc = c.connFn(addr, c.opt)
+			}
+			conns[addr] = connrole{conn: cc, replica: true}
 		}
 	}
 	// make sure InitAddress always be present
@@ -241,7 +254,8 @@ func (c *clusterClient) _refresh() (err error) {
 
 	c.mu.RLock()
 	for addr, cc := range c.conns {
-		if fresh, ok := conns[addr]; ok {
+		fresh, ok := conns[addr]
+		if ok && (cc.replica == fresh.replica || c.opt.SendToReplicas == nil) {
 			conns[addr] = connrole{
 				conn:    cc.conn,
 				replica: fresh.replica,

cluster_test.go

@@ -861,12 +861,21 @@ func TestClusterClientInit(t *testing.T) {
 	})
 
 	t.Run("Refresh cluster which has multi nodes per shard with SendToReplica option", func(t *testing.T) {
-		m := &mockConn{
+		primaryNodeConn := &mockConn{
 			DoFn: func(cmd Completed) RedisResult {
 				if strings.Join(cmd.Commands(), " ") == "CLUSTER SLOTS" {
 					return slotsMultiResp
 				}
-				return RedisResult{}
+				return RedisResult{
+					err: errors.New("unexpected call"),
+				}
+			},
+		}
+		replicaNodeConn := &mockConn{
+			DoFn: func(cmd Completed) RedisResult {
+				return RedisResult{
+					err: errors.New("unexpected call"),
+				}
 			},
 		}
 
@@ -878,8 +887,17 @@ func TestClusterClientInit(t *testing.T) {
 				},
 			},
 			func(dst string, opt *ClientOption) conn {
-				copiedM := *m
-				return &copiedM
+				if dst == "127.0.0.1:0" || dst == "127.0.2.1:0" {
+					if opt.ReplicaOnly {
+						t.Fatalf("unexpected replicaOnly option in primary node")
+					}
+					return primaryNodeConn
+				} else {
+					if !opt.ReplicaOnly {
+						t.Fatalf("unexpected replicaOnly option in replica node")
+					}
+					return replicaNodeConn
+				}
 			},
 		)
 		if err != nil {

internal/cmds/cmds.go

@@ -93,8 +93,8 @@ type Incomplete struct {
 // Completed represents a completed Redis command, should be created by the Build() of command builder.
 type Completed struct {
 	cs *CommandSlice
-	cf uint16
-	ks uint16
+	cf uint16 // cmd flag
+	ks uint16 // key slot
 }
 
 // Pin prevents a Completed to be recycled

pipe.go

@@ -20,7 +20,7 @@ import (
 )
 
 const LibName = "rueidis"
-const LibVer = "1.0.25"
+const LibVer = "1.0.26"
 
 var noHello = regexp.MustCompile("unknown command .?(HELLO|hello).?")
 
@@ -83,19 +83,19 @@ var _ wire = (*pipe)(nil)
 type pipe struct {
 	conn            net.Conn
 	error           atomic.Value
-	clhks           atomic.Value
-	pshks           atomic.Value
+	clhks           atomic.Value // closed hook, invoked after the conn is closed
+	pshks           atomic.Value // pubsub hook, registered by the SetPubSubHooks
 	queue           queue
 	cache           CacheStore
 	r               *bufio.Reader
 	w               *bufio.Writer
 	close           chan struct{}
 	onInvalidations func([]RedisMessage)
-	r2psFn          func() (p *pipe, err error)
-	r2pipe          *pipe
-	ssubs           *subs
-	nsubs           *subs
-	psubs           *subs
+	r2psFn          func() (p *pipe, err error) // func to build pipe for resp2 pubsub
+	r2pipe          *pipe                       // internal pipe for resp2 pubsub only
+	ssubs           *subs                       // pubsub smessage subscriptions
+	nsubs           *subs                       // pubsub  message subscriptions
+	psubs           *subs                       // pubsub pmessage subscriptions
 	info            map[string]RedisMessage
 	timeout         time.Duration
 	pinggap         time.Duration
@@ -108,7 +108,7 @@ type pipe struct {
 	state           int32
 	waits           int32
 	recvs           int32
-	r2ps            bool
+	r2ps            bool // identify this pipe is used for resp2 pubsub or not
 }
 
 func newPipe(connFn func() (net.Conn, error), option *ClientOption) (p *pipe, err error) {
@@ -224,7 +224,7 @@ func _newPipe(connFn func() (net.Conn, error), option *ClientOption, r2ps bool)
 			}
 			if err != nil {
 				if init[i][0] == "READONLY" {
-					// igore READONLY command error
+					// ignore READONLY command error
 					continue
 				}
 				if re, ok := err.(*RedisError); ok {
@@ -290,7 +290,7 @@ func _newPipe(connFn func() (net.Conn, error), option *ClientOption, r2ps bool)
 			defer resultsp.Put(resp)
 			for i, r := range resp.s[:len(resp.s)-2] { // skip error checking on the last CLIENT SETINFO
 				if init[i][0] == "READONLY" {
-					// igore READONLY command error
+					// ignore READONLY command error
 					continue
 				}
 				if err = r.Error(); err != nil {

rueidis.go

@@ -30,6 +30,8 @@ const (
 	DefaultReadBuffer = 1 << 19
 	// DefaultWriteBuffer is the default value of bufio.NewWriterSize for each connection, which is 0.5MiB
 	DefaultWriteBuffer = 1 << 19
+	// MaxPipelineMultiplex is the maximum meaningful value for ClientOption.PipelineMultiplex
+	MaxPipelineMultiplex = 8
 )
 
 var (
@@ -46,6 +48,8 @@ var (
 	// ErrReplicaOnlyNotSupported means ReplicaOnly flag is not supported by
 	// current client
 	ErrReplicaOnlyNotSupported = errors.New("ReplicaOnly is not supported for single client")
+	// ErrWrongPipelineMultiplex means wrong value for ClientOption.PipelineMultiplex
+	ErrWrongPipelineMultiplex = errors.New("ClientOption.PipelineMultiplex must not be bigger than MaxPipelineMultiplex")
 )
 
 // ClientOption should be passed to NewClient to construct a Client
@@ -314,6 +318,9 @@ func NewClient(option ClientOption) (client Client, err error) {
 			option.InitAddress[i], option.InitAddress[j] = option.InitAddress[j], option.InitAddress[i]
 		})
 	}
+	if option.PipelineMultiplex > MaxPipelineMultiplex {
+		return nil, ErrWrongPipelineMultiplex
+	}
 	if option.Sentinel.MasterSet != "" {
 		option.PipelineMultiplex = singleClientMultiplex(option.PipelineMultiplex)
 		return newSentinelClient(&option, makeConn)

rueidis_test.go

@@ -315,6 +315,16 @@ func TestTLSClient(t *testing.T) {
 	<-done
 }
 
+func TestNewClientMaxMultiplex(t *testing.T) {
+	_, err := NewClient(ClientOption{
+		InitAddress:       []string{"127.0.0.1:6379"},
+		PipelineMultiplex: MaxPipelineMultiplex + 1,
+	})
+	if err != ErrWrongPipelineMultiplex {
+		t.Fatalf("unexpected error %v", err)
+	}
+}
+
 func TestSingleClientMultiplex(t *testing.T) {
 	defer ShouldNotLeaked(SetupLeakDetection())
 	option := ClientOption{}

rueidiscompat/adapter.go

@@ -2028,9 +2028,7 @@ func (c *Compat) ZPopMax(ctx context.Context, key string, count ...int64) *ZSlic
 		resp = c.client.Do(ctx, c.client.B().Zpopmax().Key(key).Build())
 	case 1:
 		resp = c.client.Do(ctx, c.client.B().Zpopmax().Key(key).Count(count[0]).Build())
-		if count[0] > 1 {
-			return newZSliceCmd(resp)
-		}
+		return newZSliceCmd(resp)
 	default:
 		panic("too many arguments")
 	}
@@ -2044,9 +2042,7 @@ func (c *Compat) ZPopMin(ctx context.Context, key string, count ...int64) *ZSlic
 		resp = c.client.Do(ctx, c.client.B().Zpopmin().Key(key).Build())
 	case 1:
 		resp = c.client.Do(ctx, c.client.B().Zpopmin().Key(key).Count(count[0]).Build())
-		if count[0] > 1 {
-			return newZSliceCmd(resp)
-		}
+		return newZSliceCmd(resp)
 	default:
 		panic("too many arguments")
 	}

rueidisotel/README.md

@@ -1,7 +1,14 @@
-# OpenTelemetry Tracing
+# OpenTelemetry Tracing & Connection Metrics
 
 Use `rueidisotel.WithClient` to create a client with OpenTelemetry Tracing enabled.
 
+Use `rueidisotel.NewClient` to create a client with OpenTelemetry Tracing and Connection Metrics enabled.
+default metrics are:
+- `rueidis_dial_attempt`: number of dial attempts
+- `rueidis_dial_success`: number of successful dials
+- `rueidis_dial_conns`: number of connections
+- `rueidis_dial_latency`: dial latency in seconds
+
 ```golang
 package main
 

rueidisotel/metrics.go

@@ -14,15 +14,9 @@ import (
 )
 
 var (
-	DefaultHistogramBuckets = []float64{
+	defaultHistogramBuckets = []float64{
 		.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10,
 	}
-	DefaultDialFn = func(dst string, dialer *net.Dialer, cfg *tls.Config) (conn net.Conn, err error) {
-		if cfg != nil {
-			return tls.DialWithDialer(dialer, "tcp", dst, cfg)
-		}
-		return dialer.Dial("tcp", dst)
-	}
 )
 
 type HistogramOption struct {
@@ -44,10 +38,13 @@ func WithHistogramOption(histogramOption HistogramOption) Option {
 // - rueidis_dial_conns: number of active connections
 // - rueidis_dial_latency: dial latency in seconds
 func NewClient(clientOption rueidis.ClientOption, opts ...Option) (rueidis.Client, error) {
-	oclient := newClient(opts...)
+	oclient, err := newClient(opts...)
+	if err != nil {
+		return nil, err
+	}
 
 	if clientOption.DialFn == nil {
-		clientOption.DialFn = DefaultDialFn
+		clientOption.DialFn = defaultDialFn
 	}
 
 	attempt, err := oclient.meter.Int64Counter("rueidis_dial_attempt")
@@ -90,13 +87,13 @@ func NewClient(clientOption rueidis.ClientOption, opts ...Option) (rueidis.Clien
 	return oclient, nil
 }
 
-func newClient(opts ...Option) *otelclient {
+func newClient(opts ...Option) (*otelclient, error) {
 	cli := &otelclient{}
 	for _, opt := range opts {
 		opt(cli)
 	}
 	if cli.histogramOption.Buckets == nil {
-		cli.histogramOption.Buckets = DefaultHistogramBuckets
+		cli.histogramOption.Buckets = defaultHistogramBuckets
 	}
 	if cli.meterProvider == nil {
 		cli.meterProvider = otel.GetMeterProvider() // Default to global MeterProvider
@@ -109,9 +106,16 @@ func newClient(opts ...Option) *otelclient {
 	cli.meter = cli.meterProvider.Meter(name)
 	cli.tracer = cli.tracerProvider.Tracer(name)
 	// Now create the counters using the meter
-	cli.cscMiss, _ = cli.meter.Int64Counter("rueidis_do_cache_miss")
-	cli.cscHits, _ = cli.meter.Int64Counter("rueidis_do_cache_hits")
-	return cli
+	var err error
+	cli.cscMiss, err = cli.meter.Int64Counter("rueidis_do_cache_miss")
+	if err != nil {
+		return nil, err
+	}
+	cli.cscHits, err = cli.meter.Int64Counter("rueidis_do_cache_hits")
+	if err != nil {
+		return nil, err
+	}
+	return cli, nil
 }
 
 func trackDialing(
@@ -132,7 +136,8 @@ func trackDialing(
 			return nil, err
 		}
 
-		dialLatency.Record(ctx, time.Since(start).Seconds())
+		// Use floating point division for higher precision (instead of Seconds method).
+		dialLatency.Record(ctx, float64(time.Since(start))/float64(time.Second))
 		success.Add(ctx, 1)
 		conns.Add(ctx, 1)
 
@@ -157,3 +162,10 @@ func (t *connTracker) Close() error {
 
 	return t.Conn.Close()
 }
+
+func defaultDialFn(dst string, dialer *net.Dialer, cfg *tls.Config) (conn net.Conn, err error) {
+	if cfg != nil {
+		return tls.DialWithDialer(dialer, "tcp", dst, cfg)
+	}
+	return dialer.Dial("tcp", dst)
+}

rueidisotel/trace.go

@@ -22,8 +22,13 @@ var (
 var _ rueidis.Client = (*otelclient)(nil)
 
 // WithClient creates a new rueidis.Client with OpenTelemetry tracing enabled.
+//
+// Deprecated: use NewClient() instead.
 func WithClient(client rueidis.Client, opts ...Option) rueidis.Client {
-	cli := newClient(opts...)
+	cli, err := newClient(opts...)
+	if err != nil {
+		panic(err)
+	}
 	cli.client = client
 	return cli
 }