Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
28 changes: 24 additions & 4 deletions ring.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (

"github.com/cespare/xxhash/v2"
"github.com/dgryski/go-rendezvous" //nolint

"github.com/redis/go-redis/v9/auth"

"github.com/redis/go-redis/v9/internal"
Expand Down Expand Up @@ -797,29 +798,48 @@ func (c *Ring) generalProcessPipeline(
cmdsMap[hash] = append(cmdsMap[hash], cmd)
}

var wg sync.WaitGroup
var (
wg sync.WaitGroup
mu sync.Mutex
firstErr error
)
for hash, cmds := range cmdsMap {
wg.Add(1)
go func(hash string, cmds []Cmder) {
defer wg.Done()

// TODO: retry?
shard, err := c.sharding.GetByName(hash)
if err != nil {
setCmdsErr(cmds, err)
mu.Lock()
if firstErr == nil {
firstErr = err
}
mu.Unlock()
return
}

if tx {
cmds = wrapMultiExec(ctx, cmds)
_ = shard.Client.processTxPipelineHook(ctx, cmds)
err = shard.Client.processTxPipelineHook(ctx, cmds)
} else {
_ = shard.Client.processPipelineHook(ctx, cmds)
err = shard.Client.processPipelineHook(ctx, cmds)
}
if err != nil {
setCmdsErr(cmds, err)
mu.Lock()
if firstErr == nil {
firstErr = err
}
mu.Unlock()
}
}(hash, cmds)
}

wg.Wait()
if firstErr != nil {
return firstErr
}
return cmdsFirstErr(cmds)
}

Expand Down
17 changes: 17 additions & 0 deletions ring_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -867,3 +867,20 @@ var _ = Describe("Ring GetShardClients and GetShardClientForKey", func() {
Expect(len(shardMap)).To(BeNumerically("<=", 2)) // At most 2 shards (our setup)
})
})

var _ = Describe("unreachable ring shard", func() {
It("pipeline returns dial error", func() {
ring := redis.NewRing(&redis.RingOptions{
Addrs: map[string]string{"shard1": "10.255.255.1:6379"},
DialTimeout: 100 * time.Millisecond,
HeartbeatFrequency: time.Hour,
})
defer ring.Close()

_, err := ring.Pipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Ping(ctx)
return nil
})
Expect(err).To(HaveOccurred())
})
})
Loading