forked from mirror/redis
Merge pull request #936 from go-redis/fix/retry-cluster-pipeline-read-cmds
Retry cluster pipeline read commands
This commit is contained in:
commit
7f89fbac80
14
cluster.go
14
cluster.go
|
@ -1314,14 +1314,15 @@ func (c *ClusterClient) pipelineProcessCmds(
|
|||
}
|
||||
|
||||
err = cn.WithReader(c.opt.ReadTimeout, func(rd *proto.Reader) error {
|
||||
return c.pipelineReadCmds(rd, cmds, failedCmds)
|
||||
return c.pipelineReadCmds(node, rd, cmds, failedCmds)
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *ClusterClient) pipelineReadCmds(
|
||||
rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
|
||||
node *clusterNode, rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
|
||||
) error {
|
||||
var firstErr error
|
||||
for _, cmd := range cmds {
|
||||
err := cmd.readReply(rd)
|
||||
if err == nil {
|
||||
|
@ -1336,9 +1337,14 @@ func (c *ClusterClient) pipelineReadCmds(
|
|||
continue
|
||||
}
|
||||
|
||||
return err
|
||||
failedCmds.mu.Lock()
|
||||
failedCmds.m[node] = append(failedCmds.m[node], cmd)
|
||||
failedCmds.mu.Unlock()
|
||||
if firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return firstErr
|
||||
}
|
||||
|
||||
func (c *ClusterClient) checkMovedErr(
|
||||
|
|
Loading…
Reference in New Issue