mirror of
https://github.com/redis/go-redis.git
synced 2026-10-24 07:59:20 +03:00
fix(ring): data race on onNewNode during SetAddrs (#3862)
Signed-off-by: Saddam <saddamr3e@gmail.com>
This commit is contained in:
7
ring.go
7
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)
|
||||
}
|
||||
}
|
||||
|
||||
55
ring_sharding_race_test.go
Normal file
55
ring_sharding_race_test.go
Normal file
@@ -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()
|
||||
}
|
||||
Reference in New Issue
Block a user