force min ping replica under option (#29)

Author: Artem Klyukvin

rediscluster/cluster.go

@@ -108,6 +108,8 @@ type Opts struct {
 	RoundRobinSeed RoundRobinSeed
 	// LatencyOrientedRR - when MasterAndSlaves is used, prefer hosts with lower latency (has lower priority than WeightProvider)
 	LatencyOrientedRR bool
+	// ForceMinLatencyReplica - when LatencyOrientedRR is used forces min latency replica instead of using `weight[i] = sum(ping_latency[i]) / ping_latency[i]` algorithm
+	ForceMinLatencyReplica bool
 	// WeightProvider - enables to explicitly set weights of replicas (has higher priority than LatencyOrientedRR)
 	WeightProvider WeightProvider
 	// Enable connection with TLS
@@ -132,7 +134,8 @@ type Cluster struct {
 
 	opts Opts
 
-	latencyAwareness uint32
+	latencyAwareness       uint32
+	forceMinLatencyReplica uint32
 
 	m sync.Mutex
 
@@ -242,6 +245,10 @@ func NewCluster(ctx context.Context, initAddrs []string, opts Opts) (*Cluster, e
 	if cluster.opts.LatencyOrientedRR {
 		cluster.latencyAwareness = enabled
 	}
+	cluster.forceMinLatencyReplica = disabled
+	if cluster.opts.ForceMinLatencyReplica {
+		cluster.forceMinLatencyReplica = enabled
+	}
 	cluster.opts.HostOpts.TLSEnabled = opts.TLSEnabled
 	cluster.opts.HostOpts.TLSConfig = opts.TLSConfig
 
@@ -311,6 +318,15 @@ func (c *Cluster) SetLatencyOrientedRR(v bool) {
 	}
 }
 
+// SetForceMinLatencyReplica changes "min latency replica forcing" on the fly
+func (c *Cluster) SetForceMinLatencyReplica(v bool) {
+	if v {
+		atomic.StoreUint32(&c.forceMinLatencyReplica, enabled)
+	} else {
+		atomic.StoreUint32(&c.forceMinLatencyReplica, disabled)
+	}
+}
+
 func (c *Cluster) control() {
 	t := time.NewTicker(c.opts.CheckInterval)
 	defer t.Stop()

rediscluster/slotrange.go

@@ -2,6 +2,7 @@ package rediscluster
 
 import (
 	"bytes"
+	"math"
 	"sync/atomic"
 	"time"
 
@@ -173,13 +174,28 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) {
 		}
 		for _, shard := range newConfig.shards {
 			sumLatency := uint32(0)
-			for _, addr := range shard.addr {
+			minLatencyID := 0
+			minLatency := uint32(math.MaxUint32)
+
+			for i, addr := range shard.addr {
 				node := newConfig.nodes[addr]
-				sumLatency += atomic.LoadUint32(&node.ping)
+				pingLatency := atomic.LoadUint32(&node.ping)
+				if pingLatency < minLatency {
+					minLatency = pingLatency
+					minLatencyID = i
+				}
+
+				sumLatency += pingLatency
 			}
 			for i, addr := range shard.addr {
 				node := newConfig.nodes[addr]
+
 				weight := sumLatency / atomic.LoadUint32(&node.ping)
+				if atomic.LoadUint32(&c.forceMinLatencyReplica) == enabled && i == minLatencyID {
+					const alwaysPrefer = 1_000_000
+					weight = alwaysPrefer
+				}
+
 				atomic.StoreUint32(&shard.pingWeights[i], weight)
 			}
 		}