Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Only check latencies once every 10 seconds with routeByLatency #2795

Merged
merged 5 commits into from
Nov 20, 2024
+24 −1
Merged

Only check latencies once every 10 seconds with routeByLatency #2795

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7545393
Only check latencies once every 10 seconds with `routeByLatency`
justinmir Nov 9, 2023
97cee82
use UnixNano instead of Unix for better precision
justinmir Feb 29, 2024
5dcad41
Merge branch 'master' into only-update-latency-in-gc-if-stale
justinmir Apr 11, 2024
2b4bdfc
Merge branch 'master' into only-update-latency-in-gc-if-stale
justinmir Oct 18, 2024
0cb00f7
Merge branch 'master' into only-update-latency-in-gc-if-stale
ofekshenawa Nov 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 24 additions & 1 deletion osscluster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,10 @@ import (
"github.com/redis/go-redis/v9/internal/rand"
)

const (
minLatencyMeasurementInterval = 10 * time.Second
)

var errClusterNoNodes = fmt.Errorf("redis: cluster has no nodes")

// ClusterOptions are used to configure a cluster client and should be
Expand Down Expand Up @@ -316,6 +320,10 @@ type clusterNode struct {
latency uint32 // atomic
generation uint32 // atomic
failing uint32 // atomic

// last time the latency measurement was performed for the node, stored in nanoseconds
// from epoch
lastLatencyMeasurement int64 // atomic
}

func newClusterNode(clOpt *ClusterOptions, addr string) *clusterNode {
Expand Down Expand Up @@ -368,6 +376,7 @@ func (n *clusterNode) updateLatency() {
latency = float64(dur) / float64(successes)
}
atomic.StoreUint32(&n.latency, uint32(latency+0.5))
n.SetLastLatencyMeasurement(time.Now())
}

func (n *clusterNode) Latency() time.Duration {
Expand Down Expand Up @@ -397,6 +406,10 @@ func (n *clusterNode) Generation() uint32 {
return atomic.LoadUint32(&n.generation)
}

func (n *clusterNode) LastLatencyMeasurement() int64 {
return atomic.LoadInt64(&n.lastLatencyMeasurement)
}

func (n *clusterNode) SetGeneration(gen uint32) {
for {
v := atomic.LoadUint32(&n.generation)
Expand All @@ -406,6 +419,15 @@ func (n *clusterNode) SetGeneration(gen uint32) {
}
}

func (n *clusterNode) SetLastLatencyMeasurement(t time.Time) {
for {
v := atomic.LoadInt64(&n.lastLatencyMeasurement)
if t.UnixNano() < v || atomic.CompareAndSwapInt64(&n.lastLatencyMeasurement, v, t.UnixNano()) {
break
}
}
}

//------------------------------------------------------------------------------

type clusterNodes struct {
Expand Down Expand Up @@ -493,10 +515,11 @@ func (c *clusterNodes) GC(generation uint32) {
c.mu.Lock()

c.activeAddrs = c.activeAddrs[:0]
now := time.Now()
for addr, node := range c.nodes {
if node.Generation() >= generation {
c.activeAddrs = append(c.activeAddrs, addr)
if c.opt.RouteByLatency {
if c.opt.RouteByLatency && node.LastLatencyMeasurement() < now.Add(-minLatencyMeasurementInterval).UnixNano() {
go node.updateLatency()
}
continue
Expand Down
Loading