@@ -219,6 +219,10 @@ type ringSharding struct {
219219 hash ConsistentHash
220220 numShard int
221221 onNewNode []func (rdb * Client )
222+
223+ // ensures exclusive access to SetAddrs so there is no need
224+ // to hold mu for the duration of potentially long shard creation
225+ setAddrsMu sync.Mutex
222226}
223227
224228type ringShards struct {
@@ -245,46 +249,62 @@ func (c *ringSharding) OnNewNode(fn func(rdb *Client)) {
245249// decrease number of shards, that you use. It will reuse shards that
246250// existed before and close the ones that will not be used anymore.
247251func (c * ringSharding ) SetAddrs (addrs map [string ]string ) {
248- c .mu .Lock ()
252+ c .setAddrsMu .Lock ()
253+ defer c .setAddrsMu .Unlock ()
249254
255+ cleanup := func (shards map [string ]* ringShard ) {
256+ for addr , shard := range shards {
257+ if err := shard .Client .Close (); err != nil {
258+ internal .Logger .Printf (context .Background (), "shard.Close %s failed: %s" , addr , err )
259+ }
260+ }
261+ }
262+
263+ c .mu .RLock ()
250264 if c .closed {
251- c .mu .Unlock ()
265+ c .mu .RUnlock ()
252266 return
253267 }
268+ existing := c .shards
269+ c .mu .RUnlock ()
270+
271+ shards , created , unused := c .newRingShards (addrs , existing )
254272
255- shards , cleanup := c .newRingShards (addrs , c .shards )
273+ c .mu .Lock ()
274+ if c .closed {
275+ cleanup (created )
276+ c .mu .Unlock ()
277+ return
278+ }
256279 c .shards = shards
257280 c .rebalanceLocked ()
258281 c .mu .Unlock ()
259282
260- cleanup ()
283+ cleanup (unused )
261284}
262285
263286func (c * ringSharding ) newRingShards (
264- addrs map [string ]string , existingShards * ringShards ,
265- ) (* ringShards , func ()) {
266- shardMap := make (map [string ]* ringShard ) // indexed by addr
267- unusedShards := make (map [string ]* ringShard ) // indexed by addr
268-
269- if existingShards != nil {
270- for _ , shard := range existingShards .list {
271- addr := shard .Client .opt .Addr
272- shardMap [addr ] = shard
273- unusedShards [addr ] = shard
274- }
275- }
287+ addrs map [string ]string , existing * ringShards ,
288+ ) (shards * ringShards , created , unused map [string ]* ringShard ) {
289+
290+ shards = & ringShards {m : make (map [string ]* ringShard , len (addrs ))}
291+ created = make (map [string ]* ringShard ) // indexed by addr
292+ unused = make (map [string ]* ringShard ) // indexed by addr
276293
277- shards := & ringShards {
278- m : make (map [string ]* ringShard ),
294+ if existing != nil {
295+ for _ , shard := range existing .list {
296+ unused [shard .addr ] = shard
297+ }
279298 }
280299
281300 for name , addr := range addrs {
282- if shard , ok := shardMap [addr ]; ok {
301+ if shard , ok := unused [addr ]; ok {
283302 shards .m [name ] = shard
284- delete (unusedShards , addr )
303+ delete (unused , addr )
285304 } else {
286305 shard := newRingShard (c .opt , addr )
287306 shards .m [name ] = shard
307+ created [addr ] = shard
288308
289309 for _ , fn := range c .onNewNode {
290310 fn (shard .Client )
@@ -296,13 +316,7 @@ func (c *ringSharding) newRingShards(
296316 shards .list = append (shards .list , shard )
297317 }
298318
299- return shards , func () {
300- for addr , shard := range unusedShards {
301- if err := shard .Client .Close (); err != nil {
302- internal .Logger .Printf (context .Background (), "shard.Close %s failed: %s" , addr , err )
303- }
304- }
305- }
319+ return
306320}
307321
308322func (c * ringSharding ) List () []* ringShard {
0 commit comments