redis/pipeline.go

138 lines
2.6 KiB
Go
Raw Normal View History

package redis
2015-11-04 15:25:48 +03:00
import (
"sync"
"sync/atomic"
"gopkg.in/redis.v3/internal/pool"
2015-11-04 15:25:48 +03:00
)
2015-06-04 11:50:24 +03:00
// Pipeline implements pipelining as described in
2015-11-04 15:25:48 +03:00
// http://redis.io/topics/pipelining. It's safe for concurrent use
2015-09-12 09:36:03 +03:00
// by multiple goroutines.
2014-05-11 11:42:40 +04:00
type Pipeline struct {
2015-01-24 15:12:48 +03:00
commandable
2014-05-11 11:42:40 +04:00
2016-03-09 12:49:05 +03:00
client baseClient
2015-06-04 11:50:24 +03:00
2015-11-04 15:25:48 +03:00
mu sync.Mutex // protects cmds
cmds []Cmder
closed int32
}
2014-05-11 11:42:40 +04:00
func (c *Client) Pipeline() *Pipeline {
2015-01-24 15:12:48 +03:00
pipe := &Pipeline{
2015-06-04 11:50:24 +03:00
client: c.baseClient,
cmds: make([]Cmder, 0, 10),
2012-08-17 22:36:48 +04:00
}
2015-01-24 15:12:48 +03:00
pipe.commandable.process = pipe.process
return pipe
2014-05-11 11:42:40 +04:00
}
2012-08-25 23:44:53 +04:00
2015-06-04 11:50:24 +03:00
func (c *Client) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
pipe := c.Pipeline()
if err := fn(pipe); err != nil {
2014-07-02 17:18:19 +04:00
return nil, err
}
2015-06-04 11:50:24 +03:00
cmds, err := pipe.Exec()
2015-11-04 15:25:48 +03:00
_ = pipe.Close()
2014-05-11 11:42:40 +04:00
return cmds, err
2012-08-17 22:36:48 +04:00
}
2015-06-04 11:50:24 +03:00
func (pipe *Pipeline) process(cmd Cmder) {
2015-11-04 15:25:48 +03:00
pipe.mu.Lock()
2015-06-04 11:50:24 +03:00
pipe.cmds = append(pipe.cmds, cmd)
2015-11-04 15:25:48 +03:00
pipe.mu.Unlock()
2015-01-24 15:12:48 +03:00
}
2015-09-12 09:36:03 +03:00
// Close closes the pipeline, releasing any open resources.
2015-06-04 11:50:24 +03:00
func (pipe *Pipeline) Close() error {
2015-11-04 15:25:48 +03:00
atomic.StoreInt32(&pipe.closed, 1)
2015-06-04 11:50:24 +03:00
pipe.Discard()
return nil
}
2015-11-04 15:25:48 +03:00
func (pipe *Pipeline) isClosed() bool {
return atomic.LoadInt32(&pipe.closed) == 1
}
2015-06-04 11:50:24 +03:00
// Discard resets the pipeline and discards queued commands.
func (pipe *Pipeline) Discard() error {
2015-11-04 15:25:48 +03:00
defer pipe.mu.Unlock()
pipe.mu.Lock()
if pipe.isClosed() {
return pool.ErrClosed
2014-05-11 11:42:40 +04:00
}
2015-06-04 11:50:24 +03:00
pipe.cmds = pipe.cmds[:0]
2014-05-11 11:42:40 +04:00
return nil
2012-08-20 00:19:00 +04:00
}
2015-11-04 15:25:48 +03:00
// Exec executes all previously queued commands using one
// client-server roundtrip.
//
2014-05-11 11:42:40 +04:00
// Exec always returns list of commands and error of the first failed
// command if any.
2015-06-04 11:50:24 +03:00
func (pipe *Pipeline) Exec() (cmds []Cmder, retErr error) {
2015-11-04 15:25:48 +03:00
if pipe.isClosed() {
return nil, pool.ErrClosed
2014-05-11 11:42:40 +04:00
}
2015-11-04 15:25:48 +03:00
defer pipe.mu.Unlock()
pipe.mu.Lock()
2015-06-04 11:50:24 +03:00
if len(pipe.cmds) == 0 {
return pipe.cmds, nil
2015-03-18 13:41:24 +03:00
}
2014-05-11 11:42:40 +04:00
2015-06-04 11:50:24 +03:00
cmds = pipe.cmds
pipe.cmds = make([]Cmder, 0, 10)
2015-06-04 11:50:24 +03:00
failedCmds := cmds
for i := 0; i <= pipe.client.opt.MaxRetries; i++ {
2016-03-15 15:04:35 +03:00
cn, err := pipe.client.conn()
if err != nil {
2015-06-04 11:50:24 +03:00
setCmdsErr(failedCmds, err)
return cmds, err
}
2015-06-04 11:50:24 +03:00
if i > 0 {
resetCmds(failedCmds)
}
failedCmds, err = execCmds(cn, failedCmds)
2016-03-08 18:18:52 +03:00
pipe.client.putConn(cn, err, false)
2015-06-04 11:50:24 +03:00
if err != nil && retErr == nil {
retErr = err
}
if len(failedCmds) == 0 {
break
}
}
return cmds, retErr
}
func execCmds(cn *pool.Conn, cmds []Cmder) ([]Cmder, error) {
if err := writeCmd(cn, cmds...); err != nil {
2014-05-11 11:42:40 +04:00
setCmdsErr(cmds, err)
2015-06-04 11:50:24 +03:00
return cmds, err
}
2014-05-11 11:42:40 +04:00
var firstCmdErr error
2015-06-04 11:50:24 +03:00
var failedCmds []Cmder
for _, cmd := range cmds {
2015-10-07 17:09:20 +03:00
err := cmd.readReply(cn)
2015-06-04 11:50:24 +03:00
if err == nil {
continue
}
if firstCmdErr == nil {
2015-03-18 13:41:24 +03:00
firstCmdErr = err
}
2015-06-04 11:50:24 +03:00
if shouldRetry(err) {
failedCmds = append(failedCmds, cmd)
}
}
2015-06-04 11:50:24 +03:00
return failedCmds, firstCmdErr
}