diff --git a/cluster.go b/cluster.go index bf160ee7..baecb8f5 100644 --- a/cluster.go +++ b/cluster.go @@ -316,6 +316,92 @@ func (c *ClusterClient) reaper(frequency time.Duration) { } } +func (c *ClusterClient) Pipeline() *Pipeline { + pipe := &Pipeline{ + exec: c.pipelineExec, + } + pipe.commandable.process = pipe.process + return pipe +} + +func (c *ClusterClient) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) { + return c.Pipeline().pipelined(fn) +} + +func (c *ClusterClient) pipelineExec(cmds []Cmder) error { + var retErr error + + cmdsMap := make(map[string][]Cmder) + for _, cmd := range cmds { + slot := hashtag.Slot(cmd.clusterKey()) + addr := c.slotMasterAddr(slot) + cmdsMap[addr] = append(cmdsMap[addr], cmd) + } + + for attempt := 0; attempt <= c.opt.getMaxRedirects(); attempt++ { + failedCmds := make(map[string][]Cmder) + + for addr, cmds := range cmdsMap { + client, err := c.getClient(addr) + if err != nil { + setCmdsErr(cmds, err) + retErr = err + continue + } + + cn, err := client.conn() + if err != nil { + setCmdsErr(cmds, err) + retErr = err + continue + } + + failedCmds, err = c.execClusterCmds(cn, cmds, failedCmds) + if err != nil { + retErr = err + } + client.putConn(cn, err, false) + } + + cmdsMap = failedCmds + } + + return retErr +} + +func (c *ClusterClient) execClusterCmds( + cn *pool.Conn, cmds []Cmder, failedCmds map[string][]Cmder, +) (map[string][]Cmder, error) { + if err := writeCmd(cn, cmds...); err != nil { + setCmdsErr(cmds, err) + return failedCmds, err + } + + var firstCmdErr error + for i, cmd := range cmds { + err := cmd.readReply(cn) + if err == nil { + continue + } + if isNetworkError(err) { + cmd.reset() + failedCmds[""] = append(failedCmds[""], cmds[i:]...) + break + } else if moved, ask, addr := isMovedError(err); moved { + c.lazyReloadSlots() + cmd.reset() + failedCmds[addr] = append(failedCmds[addr], cmd) + } else if ask { + cmd.reset() + failedCmds[addr] = append(failedCmds[addr], NewCmd("ASKING"), cmd) + } else if firstCmdErr == nil { + firstCmdErr = err + } + } + + return failedCmds, firstCmdErr +} + //------------------------------------------------------------------------------ // ClusterOptions are used to configure a cluster client and should be diff --git a/cluster_pipeline.go b/cluster_pipeline.go deleted file mode 100644 index 883c7708..00000000 --- a/cluster_pipeline.go +++ /dev/null @@ -1,140 +0,0 @@ -package redis - -import ( - "gopkg.in/redis.v3/internal/hashtag" - "gopkg.in/redis.v3/internal/pool" -) - -// ClusterPipeline is not thread-safe. -type ClusterPipeline struct { - commandable - - cluster *ClusterClient - - cmds []Cmder - closed bool -} - -// Pipeline creates a new pipeline which is able to execute commands -// against multiple shards. It's NOT safe for concurrent use by -// multiple goroutines. -func (c *ClusterClient) Pipeline() *ClusterPipeline { - pipe := &ClusterPipeline{ - cluster: c, - cmds: make([]Cmder, 0, 10), - } - pipe.commandable.process = pipe.process - return pipe -} - -func (c *ClusterClient) Pipelined(fn func(*ClusterPipeline) error) ([]Cmder, error) { - pipe := c.Pipeline() - if err := fn(pipe); err != nil { - return nil, err - } - cmds, err := pipe.Exec() - _ = pipe.Close() - return cmds, err -} - -func (pipe *ClusterPipeline) process(cmd Cmder) { - pipe.cmds = append(pipe.cmds, cmd) -} - -// Discard resets the pipeline and discards queued commands. -func (pipe *ClusterPipeline) Discard() error { - if pipe.closed { - return pool.ErrClosed - } - pipe.cmds = pipe.cmds[:0] - return nil -} - -func (pipe *ClusterPipeline) Exec() (cmds []Cmder, retErr error) { - if pipe.closed { - return nil, pool.ErrClosed - } - if len(pipe.cmds) == 0 { - return []Cmder{}, nil - } - - cmds = pipe.cmds - pipe.cmds = make([]Cmder, 0, 10) - - cmdsMap := make(map[string][]Cmder) - for _, cmd := range cmds { - slot := hashtag.Slot(cmd.clusterKey()) - addr := pipe.cluster.slotMasterAddr(slot) - cmdsMap[addr] = append(cmdsMap[addr], cmd) - } - - for attempt := 0; attempt <= pipe.cluster.opt.getMaxRedirects(); attempt++ { - failedCmds := make(map[string][]Cmder) - - for addr, cmds := range cmdsMap { - client, err := pipe.cluster.getClient(addr) - if err != nil { - setCmdsErr(cmds, err) - retErr = err - continue - } - - cn, err := client.conn() - if err != nil { - setCmdsErr(cmds, err) - retErr = err - continue - } - - failedCmds, err = pipe.execClusterCmds(cn, cmds, failedCmds) - if err != nil { - retErr = err - } - client.putConn(cn, err, false) - } - - cmdsMap = failedCmds - } - - return cmds, retErr -} - -// Close closes the pipeline, releasing any open resources. -func (pipe *ClusterPipeline) Close() error { - pipe.Discard() - pipe.closed = true - return nil -} - -func (pipe *ClusterPipeline) execClusterCmds( - cn *pool.Conn, cmds []Cmder, failedCmds map[string][]Cmder, -) (map[string][]Cmder, error) { - if err := writeCmd(cn, cmds...); err != nil { - setCmdsErr(cmds, err) - return failedCmds, err - } - - var firstCmdErr error - for i, cmd := range cmds { - err := cmd.readReply(cn) - if err == nil { - continue - } - if isNetworkError(err) { - cmd.reset() - failedCmds[""] = append(failedCmds[""], cmds[i:]...) - break - } else if moved, ask, addr := isMovedError(err); moved { - pipe.cluster.lazyReloadSlots() - cmd.reset() - failedCmds[addr] = append(failedCmds[addr], cmd) - } else if ask { - cmd.reset() - failedCmds[addr] = append(failedCmds[addr], NewCmd("ASKING"), cmd) - } else if firstCmdErr == nil { - firstCmdErr = err - } - } - - return failedCmds, firstCmdErr -} diff --git a/cluster_test.go b/cluster_test.go index ff091a2c..e7aa9b96 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -449,7 +449,7 @@ var _ = Describe("Cluster", func() { Expect(client.Set("C", "C_value", 0).Err()).NotTo(HaveOccurred()) var a, b, c *redis.StringCmd - cmds, err := client.Pipelined(func(pipe *redis.ClusterPipeline) error { + cmds, err := client.Pipelined(func(pipe *redis.Pipeline) error { a = pipe.Get("A") b = pipe.Get("B") c = pipe.Get("C") diff --git a/commands.go b/commands.go index c5235346..dc6dd08f 100644 --- a/commands.go +++ b/commands.go @@ -1543,7 +1543,7 @@ func (c *commandable) Time() *StringSliceCmd { //------------------------------------------------------------------------------ -func (c *commandable) Eval(script string, keys []string, args []string) *Cmd { +func (c *commandable) Eval(script string, keys []string, args ...interface{}) *Cmd { cmdArgs := make([]interface{}, 3+len(keys)+len(args)) cmdArgs[0] = "EVAL" cmdArgs[1] = script @@ -1563,7 +1563,7 @@ func (c *commandable) Eval(script string, keys []string, args []string) *Cmd { return cmd } -func (c *commandable) EvalSha(sha1 string, keys []string, args []string) *Cmd { +func (c *commandable) EvalSha(sha1 string, keys []string, args ...interface{}) *Cmd { cmdArgs := make([]interface{}, 3+len(keys)+len(args)) cmdArgs[0] = "EVALSHA" cmdArgs[1] = sha1 diff --git a/example_test.go b/example_test.go index 33800228..587ef78a 100644 --- a/example_test.go +++ b/example_test.go @@ -288,7 +288,7 @@ func ExampleScript() { return false `) - n, err := IncrByXX.Run(client, []string{"xx_counter"}, []string{"2"}).Result() + n, err := IncrByXX.Run(client, []string{"xx_counter"}, 2).Result() fmt.Println(n, err) err = client.Set("xx_counter", "40", 0).Err() @@ -296,7 +296,7 @@ func ExampleScript() { panic(err) } - n, err = IncrByXX.Run(client, []string{"xx_counter"}, []string{"2"}).Result() + n, err = IncrByXX.Run(client, []string{"xx_counter"}, 2).Result() fmt.Println(n, err) // Output: <nil> redis: nil diff --git a/pipeline.go b/pipeline.go index 888d8c40..95d389f2 100644 --- a/pipeline.go +++ b/pipeline.go @@ -13,7 +13,7 @@ import ( type Pipeline struct { commandable - client baseClient + exec func([]Cmder) error mu sync.Mutex // protects cmds cmds []Cmder @@ -21,25 +21,6 @@ type Pipeline struct { closed int32 } -func (c *Client) Pipeline() *Pipeline { - pipe := &Pipeline{ - client: c.baseClient, - cmds: make([]Cmder, 0, 10), - } - pipe.commandable.process = pipe.process - return pipe -} - -func (c *Client) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) { - pipe := c.Pipeline() - if err := fn(pipe); err != nil { - return nil, err - } - cmds, err := pipe.Exec() - _ = pipe.Close() - return cmds, err -} - func (pipe *Pipeline) process(cmd Cmder) { pipe.mu.Lock() pipe.cmds = append(pipe.cmds, cmd) @@ -73,7 +54,7 @@ func (pipe *Pipeline) Discard() error { // // Exec always returns list of commands and error of the first failed // command if any. -func (pipe *Pipeline) Exec() (cmds []Cmder, retErr error) { +func (pipe *Pipeline) Exec() ([]Cmder, error) { if pipe.isClosed() { return nil, pool.ErrClosed } @@ -85,31 +66,19 @@ func (pipe *Pipeline) Exec() (cmds []Cmder, retErr error) { return pipe.cmds, nil } - cmds = pipe.cmds - pipe.cmds = make([]Cmder, 0, 10) + cmds := pipe.cmds + pipe.cmds = nil - failedCmds := cmds - for i := 0; i <= pipe.client.opt.MaxRetries; i++ { - cn, err := pipe.client.conn() - if err != nil { - setCmdsErr(failedCmds, err) - return cmds, err - } + return cmds, pipe.exec(cmds) +} - if i > 0 { - resetCmds(failedCmds) - } - failedCmds, err = execCmds(cn, failedCmds) - pipe.client.putConn(cn, err, false) - if err != nil && retErr == nil { - retErr = err - } - if len(failedCmds) == 0 { - break - } +func (pipe *Pipeline) pipelined(fn func(*Pipeline) error) ([]Cmder, error) { + if err := fn(pipe); err != nil { + return nil, err } - - return cmds, retErr + cmds, err := pipe.Exec() + _ = pipe.Close() + return cmds, err } func execCmds(cn *pool.Conn, cmds []Cmder) ([]Cmder, error) { diff --git a/redis.go b/redis.go index aed713cf..ee4e69a9 100644 --- a/redis.go +++ b/redis.go @@ -174,3 +174,40 @@ func (c *Client) PoolStats() *PoolStats { FreeConns: s.FreeConns, } } + +func (c *Client) Pipeline() *Pipeline { + pipe := &Pipeline{ + exec: c.pipelineExec, + } + pipe.commandable.process = pipe.process + return pipe +} + +func (c *Client) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) { + return c.Pipeline().pipelined(fn) +} + +func (c *Client) pipelineExec(cmds []Cmder) error { + var retErr error + failedCmds := cmds + for i := 0; i <= c.opt.MaxRetries; i++ { + cn, err := c.conn() + if err != nil { + setCmdsErr(failedCmds, err) + return err + } + + if i > 0 { + resetCmds(failedCmds) + } + failedCmds, err = execCmds(cn, failedCmds) + c.putConn(cn, err, false) + if err != nil && retErr == nil { + retErr = err + } + if len(failedCmds) == 0 { + break + } + } + return retErr +} diff --git a/ring.go b/ring.go index fd74b3b6..02be83c6 100644 --- a/ring.go +++ b/ring.go @@ -241,66 +241,24 @@ func (ring *Ring) Close() (retErr error) { return retErr } -// RingPipeline creates a new pipeline which is able to execute commands -// against multiple shards. It's NOT safe for concurrent use by -// multiple goroutines. -type RingPipeline struct { - commandable - - ring *Ring - - cmds []Cmder - closed bool -} - -func (ring *Ring) Pipeline() *RingPipeline { - pipe := &RingPipeline{ - ring: ring, - cmds: make([]Cmder, 0, 10), +func (ring *Ring) Pipeline() *Pipeline { + pipe := &Pipeline{ + exec: ring.pipelineExec, } pipe.commandable.process = pipe.process return pipe } -func (ring *Ring) Pipelined(fn func(*RingPipeline) error) ([]Cmder, error) { - pipe := ring.Pipeline() - if err := fn(pipe); err != nil { - return nil, err - } - cmds, err := pipe.Exec() - pipe.Close() - return cmds, err +func (ring *Ring) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) { + return ring.Pipeline().pipelined(fn) } -func (pipe *RingPipeline) process(cmd Cmder) { - pipe.cmds = append(pipe.cmds, cmd) -} - -// Discard resets the pipeline and discards queued commands. -func (pipe *RingPipeline) Discard() error { - if pipe.closed { - return pool.ErrClosed - } - pipe.cmds = pipe.cmds[:0] - return nil -} - -// Exec always returns list of commands and error of the first failed -// command if any. -func (pipe *RingPipeline) Exec() (cmds []Cmder, retErr error) { - if pipe.closed { - return nil, pool.ErrClosed - } - if len(pipe.cmds) == 0 { - return pipe.cmds, nil - } - - cmds = pipe.cmds - pipe.cmds = make([]Cmder, 0, 10) +func (ring *Ring) pipelineExec(cmds []Cmder) error { + var retErr error cmdsMap := make(map[string][]Cmder) for _, cmd := range cmds { - name := pipe.ring.hash.Get(hashtag.Key(cmd.clusterKey())) + name := ring.hash.Get(hashtag.Key(cmd.clusterKey())) if name == "" { cmd.setErr(errRingShardsDown) if retErr == nil { @@ -311,11 +269,11 @@ func (pipe *RingPipeline) Exec() (cmds []Cmder, retErr error) { cmdsMap[name] = append(cmdsMap[name], cmd) } - for i := 0; i <= pipe.ring.opt.MaxRetries; i++ { + for i := 0; i <= ring.opt.MaxRetries; i++ { failedCmdsMap := make(map[string][]Cmder) for name, cmds := range cmdsMap { - client := pipe.ring.shards[name].Client + client := ring.shards[name].Client cn, err := client.conn() if err != nil { setCmdsErr(cmds, err) @@ -344,12 +302,5 @@ func (pipe *RingPipeline) Exec() (cmds []Cmder, retErr error) { cmdsMap = failedCmdsMap } - return cmds, retErr -} - -// Close closes the pipeline, releasing any open resources. -func (pipe *RingPipeline) Close() error { - pipe.Discard() - pipe.closed = true - return nil + return retErr } diff --git a/ring_test.go b/ring_test.go index 7d5cd91c..06e6c303 100644 --- a/ring_test.go +++ b/ring_test.go @@ -96,7 +96,7 @@ var _ = Describe("Redis ring", func() { Describe("pipelining", func() { It("returns an error when all shards are down", func() { ring := redis.NewRing(&redis.RingOptions{}) - _, err := ring.Pipelined(func(pipe *redis.RingPipeline) error { + _, err := ring.Pipelined(func(pipe *redis.Pipeline) error { pipe.Ping() return nil }) @@ -133,7 +133,7 @@ var _ = Describe("Redis ring", func() { keys = append(keys, string(key)) } - _, err := ring.Pipelined(func(pipe *redis.RingPipeline) error { + _, err := ring.Pipelined(func(pipe *redis.Pipeline) error { for _, key := range keys { pipe.Set(key, "value", 0).Err() } @@ -149,7 +149,7 @@ var _ = Describe("Redis ring", func() { }) It("supports hash tags", func() { - _, err := ring.Pipelined(func(pipe *redis.RingPipeline) error { + _, err := ring.Pipelined(func(pipe *redis.Pipeline) error { for i := 0; i < 100; i++ { pipe.Set(fmt.Sprintf("key%d{tag}", i), "value", 0).Err() } diff --git a/script.go b/script.go index 3f22f469..c5fc1d2b 100644 --- a/script.go +++ b/script.go @@ -8,8 +8,8 @@ import ( ) type scripter interface { - Eval(script string, keys []string, args []string) *Cmd - EvalSha(sha1 string, keys []string, args []string) *Cmd + Eval(script string, keys []string, args ...interface{}) *Cmd + EvalSha(sha1 string, keys []string, args ...interface{}) *Cmd ScriptExists(scripts ...string) *BoolSliceCmd ScriptLoad(script string) *StringCmd } @@ -35,18 +35,18 @@ func (s *Script) Exists(c scripter) *BoolSliceCmd { return c.ScriptExists(s.src) } -func (s *Script) Eval(c scripter, keys []string, args []string) *Cmd { - return c.Eval(s.src, keys, args) +func (s *Script) Eval(c scripter, keys []string, args ...interface{}) *Cmd { + return c.Eval(s.src, keys, args...) } -func (s *Script) EvalSha(c scripter, keys []string, args []string) *Cmd { - return c.EvalSha(s.hash, keys, args) +func (s *Script) EvalSha(c scripter, keys []string, args ...interface{}) *Cmd { + return c.EvalSha(s.hash, keys, args...) } -func (s *Script) Run(c scripter, keys []string, args []string) *Cmd { - r := s.EvalSha(c, keys, args) +func (s *Script) Run(c scripter, keys []string, args ...interface{}) *Cmd { + r := s.EvalSha(c, keys, args...) if err := r.Err(); err != nil && strings.HasPrefix(err.Error(), "NOSCRIPT ") { - return s.Eval(c, keys, args) + return s.Eval(c, keys, args...) } return r }