diff --git a/ring.go b/ring.go index 4ca431ef..a732b9bc 100644 --- a/ring.go +++ b/ring.go @@ -369,9 +369,10 @@ func (c *ringSharding) SetAddrs(addrs map[string]string) { return } existing := c.shards + onNewNode := c.onNewNode c.mu.RUnlock() - shards, created, unused := c.newRingShards(addrs, existing) + shards, created, unused := c.newRingShards(addrs, existing, onNewNode) c.mu.Lock() if c.closed { @@ -387,7 +388,7 @@ func (c *ringSharding) SetAddrs(addrs map[string]string) { } func (c *ringSharding) newRingShards( - addrs map[string]string, existing *ringShards, + addrs map[string]string, existing *ringShards, onNewNode []func(rdb *Client), ) (shards *ringShards, created, unused map[string]*ringShard) { shards = &ringShards{m: make(map[string]*ringShard, len(addrs))} created = make(map[string]*ringShard) // indexed by addr @@ -408,7 +409,7 @@ func (c *ringSharding) newRingShards( shards.m[name] = shard created[addr] = shard - for _, fn := range c.onNewNode { + for _, fn := range onNewNode { fn(shard.Client) } } diff --git a/ring_sharding_race_test.go b/ring_sharding_race_test.go new file mode 100644 index 00000000..3e393988 --- /dev/null +++ b/ring_sharding_race_test.go @@ -0,0 +1,55 @@ +package redis + +import ( + "sync" + "testing" +) + +// SetAddrs snapshots c.shards under c.mu before calling newRingShards, but +// newRingShards then ranges over c.onNewNode without holding c.mu, while +// OnNewNode appends to that slice under the lock. A goroutine resharding via +// SetAddrs and one registering a callback via OnNewNode therefore race on the +// onNewNode slice. clusterNodes.GetOrCreateWithNodeAddress reads the same kind +// of callback list under its lock; the ring path missed it. Run with -race +// against the pre-fix tree to see the report at ring.go. +func TestRingShardingOnNewNodeConcurrent(t *testing.T) { + opt := &RingOptions{Addrs: map[string]string{"shard0": ":6390"}} + opt.init() + + c := newRingSharding(opt) + defer c.Close() + + var wg sync.WaitGroup + wg.Add(2) + + // Register callbacks continuously for the whole duration of the resharding + // loop so an append is in flight while newRingShards iterates the slice. + done := make(chan struct{}) + go func() { + defer wg.Done() + for { + select { + case <-done: + return + default: + c.OnNewNode(func(*Client) {}) + } + } + }() + + go func() { + defer wg.Done() + defer close(done) + for i := 0; i < 50; i++ { + // Toggle the shard set so newRingShards creates a fresh shard on the + // even iterations, which is what makes it iterate onNewNode. + if i%2 == 0 { + c.SetAddrs(map[string]string{"shard0": ":6390", "shard1": ":6391"}) + } else { + c.SetAddrs(map[string]string{"shard0": ":6390"}) + } + } + }() + + wg.Wait() +}