forked from mirror/redis
Ensure to use pointer methods where appropriate. Tidy up godoc.
This commit is contained in:
parent
1bf10a61e2
commit
1c4c05e970
404
commands.go
404
commands.go
File diff suppressed because it is too large
Load Diff
|
@ -3,7 +3,7 @@ package redis
|
||||||
import "sync"
|
import "sync"
|
||||||
|
|
||||||
type Scanner struct {
|
type Scanner struct {
|
||||||
client cmdable
|
client *cmdable
|
||||||
*ScanCmd
|
*ScanCmd
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
54
pipeline.go
54
pipeline.go
|
@ -22,32 +22,32 @@ type Pipeline struct {
|
||||||
closed int32
|
closed int32
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pipe *Pipeline) Process(cmd Cmder) error {
|
func (c *Pipeline) Process(cmd Cmder) error {
|
||||||
pipe.mu.Lock()
|
c.mu.Lock()
|
||||||
pipe.cmds = append(pipe.cmds, cmd)
|
c.cmds = append(c.cmds, cmd)
|
||||||
pipe.mu.Unlock()
|
c.mu.Unlock()
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Close closes the pipeline, releasing any open resources.
|
// Close closes the pipeline, releasing any open resources.
|
||||||
func (pipe *Pipeline) Close() error {
|
func (c *Pipeline) Close() error {
|
||||||
atomic.StoreInt32(&pipe.closed, 1)
|
atomic.StoreInt32(&c.closed, 1)
|
||||||
pipe.Discard()
|
c.Discard()
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pipe *Pipeline) isClosed() bool {
|
func (c *Pipeline) isClosed() bool {
|
||||||
return atomic.LoadInt32(&pipe.closed) == 1
|
return atomic.LoadInt32(&c.closed) == 1
|
||||||
}
|
}
|
||||||
|
|
||||||
// Discard resets the pipeline and discards queued commands.
|
// Discard resets the pipeline and discards queued commands.
|
||||||
func (pipe *Pipeline) Discard() error {
|
func (c *Pipeline) Discard() error {
|
||||||
defer pipe.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
pipe.mu.Lock()
|
c.mu.Lock()
|
||||||
if pipe.isClosed() {
|
if c.isClosed() {
|
||||||
return pool.ErrClosed
|
return pool.ErrClosed
|
||||||
}
|
}
|
||||||
pipe.cmds = pipe.cmds[:0]
|
c.cmds = c.cmds[:0]
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -56,30 +56,30 @@ func (pipe *Pipeline) Discard() error {
|
||||||
//
|
//
|
||||||
// Exec always returns list of commands and error of the first failed
|
// Exec always returns list of commands and error of the first failed
|
||||||
// command if any.
|
// command if any.
|
||||||
func (pipe *Pipeline) Exec() ([]Cmder, error) {
|
func (c *Pipeline) Exec() ([]Cmder, error) {
|
||||||
if pipe.isClosed() {
|
if c.isClosed() {
|
||||||
return nil, pool.ErrClosed
|
return nil, pool.ErrClosed
|
||||||
}
|
}
|
||||||
|
|
||||||
defer pipe.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
pipe.mu.Lock()
|
c.mu.Lock()
|
||||||
|
|
||||||
if len(pipe.cmds) == 0 {
|
if len(c.cmds) == 0 {
|
||||||
return pipe.cmds, nil
|
return c.cmds, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
cmds := pipe.cmds
|
cmds := c.cmds
|
||||||
pipe.cmds = nil
|
c.cmds = nil
|
||||||
|
|
||||||
return cmds, pipe.exec(cmds)
|
return cmds, c.exec(cmds)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pipe *Pipeline) pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
|
func (c *Pipeline) pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
|
||||||
if err := fn(pipe); err != nil {
|
if err := fn(c); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
cmds, err := pipe.Exec()
|
cmds, err := c.Exec()
|
||||||
_ = pipe.Close()
|
_ = c.Close()
|
||||||
return cmds, err
|
return cmds, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
110
ring.go
110
ring.go
|
@ -102,7 +102,7 @@ func (shard *ringShard) Vote(up bool) bool {
|
||||||
// concurrent use by multiple goroutines.
|
// concurrent use by multiple goroutines.
|
||||||
//
|
//
|
||||||
// Ring monitors the state of each shard and removes dead shards from
|
// Ring monitors the state of each shard and removes dead shards from
|
||||||
// the ring. When shard comes online it is added back to the ring. This
|
// the c. When shard comes online it is added back to the c. This
|
||||||
// gives you maximum availability and partition tolerance, but no
|
// gives you maximum availability and partition tolerance, but no
|
||||||
// consistency between different shards or even clients. Each client
|
// consistency between different shards or even clients. Each client
|
||||||
// uses shards that are available to the client and does not do any
|
// uses shards that are available to the client and does not do any
|
||||||
|
@ -149,58 +149,58 @@ func NewRing(opt *RingOptions) *Ring {
|
||||||
return ring
|
return ring
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) cmdInfo(name string) *CommandInfo {
|
func (c *Ring) cmdInfo(name string) *CommandInfo {
|
||||||
ring.cmdsInfoOnce.Do(func() {
|
c.cmdsInfoOnce.Do(func() {
|
||||||
for _, shard := range ring.shards {
|
for _, shard := range c.shards {
|
||||||
cmdsInfo, err := shard.Client.Command().Result()
|
cmdsInfo, err := shard.Client.Command().Result()
|
||||||
if err == nil {
|
if err == nil {
|
||||||
ring.cmdsInfo = cmdsInfo
|
c.cmdsInfo = cmdsInfo
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
ring.cmdsInfoOnce = &sync.Once{}
|
c.cmdsInfoOnce = &sync.Once{}
|
||||||
})
|
})
|
||||||
if ring.cmdsInfo == nil {
|
if c.cmdsInfo == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return ring.cmdsInfo[name]
|
return c.cmdsInfo[name]
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) cmdFirstKey(cmd Cmder) string {
|
func (c *Ring) cmdFirstKey(cmd Cmder) string {
|
||||||
cmdInfo := ring.cmdInfo(cmd.arg(0))
|
cmdInfo := c.cmdInfo(cmd.arg(0))
|
||||||
if cmdInfo == nil {
|
if cmdInfo == nil {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
return cmd.arg(int(cmdInfo.FirstKeyPos))
|
return cmd.arg(int(cmdInfo.FirstKeyPos))
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) addClient(name string, cl *Client) {
|
func (c *Ring) addClient(name string, cl *Client) {
|
||||||
ring.mu.Lock()
|
c.mu.Lock()
|
||||||
ring.hash.Add(name)
|
c.hash.Add(name)
|
||||||
ring.shards[name] = &ringShard{Client: cl}
|
c.shards[name] = &ringShard{Client: cl}
|
||||||
ring.mu.Unlock()
|
c.mu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) getClient(key string) (*Client, error) {
|
func (c *Ring) getClient(key string) (*Client, error) {
|
||||||
ring.mu.RLock()
|
c.mu.RLock()
|
||||||
|
|
||||||
if ring.closed {
|
if c.closed {
|
||||||
return nil, pool.ErrClosed
|
return nil, pool.ErrClosed
|
||||||
}
|
}
|
||||||
|
|
||||||
name := ring.hash.Get(hashtag.Key(key))
|
name := c.hash.Get(hashtag.Key(key))
|
||||||
if name == "" {
|
if name == "" {
|
||||||
ring.mu.RUnlock()
|
c.mu.RUnlock()
|
||||||
return nil, errRingShardsDown
|
return nil, errRingShardsDown
|
||||||
}
|
}
|
||||||
|
|
||||||
cl := ring.shards[name].Client
|
cl := c.shards[name].Client
|
||||||
ring.mu.RUnlock()
|
c.mu.RUnlock()
|
||||||
return cl, nil
|
return cl, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) Process(cmd Cmder) error {
|
func (c *Ring) Process(cmd Cmder) error {
|
||||||
cl, err := ring.getClient(ring.cmdFirstKey(cmd))
|
cl, err := c.getClient(c.cmdFirstKey(cmd))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cmd.setErr(err)
|
cmd.setErr(err)
|
||||||
return err
|
return err
|
||||||
|
@ -208,34 +208,34 @@ func (ring *Ring) Process(cmd Cmder) error {
|
||||||
return cl.baseClient.Process(cmd)
|
return cl.baseClient.Process(cmd)
|
||||||
}
|
}
|
||||||
|
|
||||||
// rebalance removes dead shards from the ring.
|
// rebalance removes dead shards from the c.
|
||||||
func (ring *Ring) rebalance() {
|
func (c *Ring) rebalance() {
|
||||||
defer ring.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
ring.mu.Lock()
|
c.mu.Lock()
|
||||||
|
|
||||||
ring.hash = consistenthash.New(ring.nreplicas, nil)
|
c.hash = consistenthash.New(c.nreplicas, nil)
|
||||||
for name, shard := range ring.shards {
|
for name, shard := range c.shards {
|
||||||
if shard.IsUp() {
|
if shard.IsUp() {
|
||||||
ring.hash.Add(name)
|
c.hash.Add(name)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// heartbeat monitors state of each shard in the ring.
|
// heartbeat monitors state of each shard in the c.
|
||||||
func (ring *Ring) heartbeat() {
|
func (c *Ring) heartbeat() {
|
||||||
ticker := time.NewTicker(100 * time.Millisecond)
|
ticker := time.NewTicker(100 * time.Millisecond)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
for _ = range ticker.C {
|
for _ = range ticker.C {
|
||||||
var rebalance bool
|
var rebalance bool
|
||||||
|
|
||||||
ring.mu.RLock()
|
c.mu.RLock()
|
||||||
|
|
||||||
if ring.closed {
|
if c.closed {
|
||||||
ring.mu.RUnlock()
|
c.mu.RUnlock()
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, shard := range ring.shards {
|
for _, shard := range c.shards {
|
||||||
err := shard.Client.Ping().Err()
|
err := shard.Client.Ping().Err()
|
||||||
if shard.Vote(err == nil || err == pool.ErrPoolTimeout) {
|
if shard.Vote(err == nil || err == pool.ErrPoolTimeout) {
|
||||||
internal.Logf("ring shard state changed: %s", shard)
|
internal.Logf("ring shard state changed: %s", shard)
|
||||||
|
@ -243,10 +243,10 @@ func (ring *Ring) heartbeat() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
ring.mu.RUnlock()
|
c.mu.RUnlock()
|
||||||
|
|
||||||
if rebalance {
|
if rebalance {
|
||||||
ring.rebalance()
|
c.rebalance()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -255,45 +255,45 @@ func (ring *Ring) heartbeat() {
|
||||||
//
|
//
|
||||||
// It is rare to Close a Ring, as the Ring is meant to be long-lived
|
// It is rare to Close a Ring, as the Ring is meant to be long-lived
|
||||||
// and shared between many goroutines.
|
// and shared between many goroutines.
|
||||||
func (ring *Ring) Close() (retErr error) {
|
func (c *Ring) Close() (retErr error) {
|
||||||
defer ring.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
ring.mu.Lock()
|
c.mu.Lock()
|
||||||
|
|
||||||
if ring.closed {
|
if c.closed {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
ring.closed = true
|
c.closed = true
|
||||||
|
|
||||||
for _, shard := range ring.shards {
|
for _, shard := range c.shards {
|
||||||
if err := shard.Client.Close(); err != nil {
|
if err := shard.Client.Close(); err != nil {
|
||||||
retErr = err
|
retErr = err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
ring.hash = nil
|
c.hash = nil
|
||||||
ring.shards = nil
|
c.shards = nil
|
||||||
|
|
||||||
return retErr
|
return retErr
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) Pipeline() *Pipeline {
|
func (c *Ring) Pipeline() *Pipeline {
|
||||||
pipe := Pipeline{
|
pipe := Pipeline{
|
||||||
exec: ring.pipelineExec,
|
exec: c.pipelineExec,
|
||||||
}
|
}
|
||||||
pipe.cmdable.process = pipe.Process
|
pipe.cmdable.process = pipe.Process
|
||||||
pipe.statefulCmdable.process = pipe.Process
|
pipe.statefulCmdable.process = pipe.Process
|
||||||
return &pipe
|
return &pipe
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
|
func (c *Ring) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
|
||||||
return ring.Pipeline().pipelined(fn)
|
return c.Pipeline().pipelined(fn)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ring *Ring) pipelineExec(cmds []Cmder) error {
|
func (c *Ring) pipelineExec(cmds []Cmder) error {
|
||||||
var retErr error
|
var retErr error
|
||||||
|
|
||||||
cmdsMap := make(map[string][]Cmder)
|
cmdsMap := make(map[string][]Cmder)
|
||||||
for _, cmd := range cmds {
|
for _, cmd := range cmds {
|
||||||
name := ring.hash.Get(hashtag.Key(ring.cmdFirstKey(cmd)))
|
name := c.hash.Get(hashtag.Key(c.cmdFirstKey(cmd)))
|
||||||
if name == "" {
|
if name == "" {
|
||||||
cmd.setErr(errRingShardsDown)
|
cmd.setErr(errRingShardsDown)
|
||||||
if retErr == nil {
|
if retErr == nil {
|
||||||
|
@ -304,11 +304,11 @@ func (ring *Ring) pipelineExec(cmds []Cmder) error {
|
||||||
cmdsMap[name] = append(cmdsMap[name], cmd)
|
cmdsMap[name] = append(cmdsMap[name], cmd)
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := 0; i <= ring.opt.MaxRetries; i++ {
|
for i := 0; i <= c.opt.MaxRetries; i++ {
|
||||||
failedCmdsMap := make(map[string][]Cmder)
|
failedCmdsMap := make(map[string][]Cmder)
|
||||||
|
|
||||||
for name, cmds := range cmdsMap {
|
for name, cmds := range cmdsMap {
|
||||||
client := ring.shards[name].Client
|
client := c.shards[name].Client
|
||||||
cn, err := client.conn()
|
cn, err := client.conn()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
setCmdsErr(cmds, err)
|
setCmdsErr(cmds, err)
|
||||||
|
|
54
tx.go
54
tx.go
|
@ -39,7 +39,7 @@ func (c *Client) Watch(fn func(*Tx) error, keys ...string) error {
|
||||||
tx := c.newTx()
|
tx := c.newTx()
|
||||||
if len(keys) > 0 {
|
if len(keys) > 0 {
|
||||||
if err := tx.Watch(keys...).Err(); err != nil {
|
if err := tx.Watch(keys...).Err(); err != nil {
|
||||||
tx.close()
|
_ = tx.close()
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -50,57 +50,57 @@ func (c *Client) Watch(fn func(*Tx) error, keys ...string) error {
|
||||||
return retErr
|
return retErr
|
||||||
}
|
}
|
||||||
|
|
||||||
func (tx *Tx) Process(cmd Cmder) error {
|
func (c *Tx) Process(cmd Cmder) error {
|
||||||
if tx.cmds == nil {
|
if c.cmds == nil {
|
||||||
return tx.baseClient.Process(cmd)
|
return c.baseClient.Process(cmd)
|
||||||
}
|
}
|
||||||
tx.cmds = append(tx.cmds, cmd)
|
c.cmds = append(c.cmds, cmd)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// close closes the transaction, releasing any open resources.
|
// close closes the transaction, releasing any open resources.
|
||||||
func (tx *Tx) close() error {
|
func (c *Tx) close() error {
|
||||||
if tx.closed {
|
if c.closed {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
tx.closed = true
|
c.closed = true
|
||||||
if err := tx.Unwatch().Err(); err != nil {
|
if err := c.Unwatch().Err(); err != nil {
|
||||||
internal.Logf("Unwatch failed: %s", err)
|
internal.Logf("Unwatch failed: %s", err)
|
||||||
}
|
}
|
||||||
return tx.baseClient.Close()
|
return c.baseClient.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Watch marks the keys to be watched for conditional execution
|
// Watch marks the keys to be watched for conditional execution
|
||||||
// of a transaction.
|
// of a transaction.
|
||||||
func (tx *Tx) Watch(keys ...string) *StatusCmd {
|
func (c *Tx) Watch(keys ...string) *StatusCmd {
|
||||||
args := make([]interface{}, 1+len(keys))
|
args := make([]interface{}, 1+len(keys))
|
||||||
args[0] = "WATCH"
|
args[0] = "WATCH"
|
||||||
for i, key := range keys {
|
for i, key := range keys {
|
||||||
args[1+i] = key
|
args[1+i] = key
|
||||||
}
|
}
|
||||||
cmd := NewStatusCmd(args...)
|
cmd := NewStatusCmd(args...)
|
||||||
tx.Process(cmd)
|
c.Process(cmd)
|
||||||
return cmd
|
return cmd
|
||||||
}
|
}
|
||||||
|
|
||||||
// Unwatch flushes all the previously watched keys for a transaction.
|
// Unwatch flushes all the previously watched keys for a transaction.
|
||||||
func (tx *Tx) Unwatch(keys ...string) *StatusCmd {
|
func (c *Tx) Unwatch(keys ...string) *StatusCmd {
|
||||||
args := make([]interface{}, 1+len(keys))
|
args := make([]interface{}, 1+len(keys))
|
||||||
args[0] = "UNWATCH"
|
args[0] = "UNWATCH"
|
||||||
for i, key := range keys {
|
for i, key := range keys {
|
||||||
args[1+i] = key
|
args[1+i] = key
|
||||||
}
|
}
|
||||||
cmd := NewStatusCmd(args...)
|
cmd := NewStatusCmd(args...)
|
||||||
tx.Process(cmd)
|
c.Process(cmd)
|
||||||
return cmd
|
return cmd
|
||||||
}
|
}
|
||||||
|
|
||||||
// Discard discards queued commands.
|
// Discard discards queued commands.
|
||||||
func (tx *Tx) Discard() error {
|
func (c *Tx) Discard() error {
|
||||||
if tx.cmds == nil {
|
if c.cmds == nil {
|
||||||
return errDiscard
|
return errDiscard
|
||||||
}
|
}
|
||||||
tx.cmds = tx.cmds[:1]
|
c.cmds = c.cmds[:1]
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -113,19 +113,19 @@ func (tx *Tx) Discard() error {
|
||||||
// Exec always returns list of commands. If transaction fails
|
// Exec always returns list of commands. If transaction fails
|
||||||
// TxFailedErr is returned. Otherwise Exec returns error of the first
|
// TxFailedErr is returned. Otherwise Exec returns error of the first
|
||||||
// failed command or nil.
|
// failed command or nil.
|
||||||
func (tx *Tx) MultiExec(fn func() error) ([]Cmder, error) {
|
func (c *Tx) MultiExec(fn func() error) ([]Cmder, error) {
|
||||||
if tx.closed {
|
if c.closed {
|
||||||
return nil, pool.ErrClosed
|
return nil, pool.ErrClosed
|
||||||
}
|
}
|
||||||
|
|
||||||
tx.cmds = []Cmder{NewStatusCmd("MULTI")}
|
c.cmds = []Cmder{NewStatusCmd("MULTI")}
|
||||||
if err := fn(); err != nil {
|
if err := fn(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
tx.cmds = append(tx.cmds, NewSliceCmd("EXEC"))
|
c.cmds = append(c.cmds, NewSliceCmd("EXEC"))
|
||||||
|
|
||||||
cmds := tx.cmds
|
cmds := c.cmds
|
||||||
tx.cmds = nil
|
c.cmds = nil
|
||||||
|
|
||||||
if len(cmds) == 2 {
|
if len(cmds) == 2 {
|
||||||
return []Cmder{}, nil
|
return []Cmder{}, nil
|
||||||
|
@ -134,18 +134,18 @@ func (tx *Tx) MultiExec(fn func() error) ([]Cmder, error) {
|
||||||
// Strip MULTI and EXEC commands.
|
// Strip MULTI and EXEC commands.
|
||||||
retCmds := cmds[1 : len(cmds)-1]
|
retCmds := cmds[1 : len(cmds)-1]
|
||||||
|
|
||||||
cn, err := tx.conn()
|
cn, err := c.conn()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
setCmdsErr(retCmds, err)
|
setCmdsErr(retCmds, err)
|
||||||
return retCmds, err
|
return retCmds, err
|
||||||
}
|
}
|
||||||
|
|
||||||
err = tx.execCmds(cn, cmds)
|
err = c.execCmds(cn, cmds)
|
||||||
tx.putConn(cn, err, false)
|
c.putConn(cn, err, false)
|
||||||
return retCmds, err
|
return retCmds, err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (tx *Tx) execCmds(cn *pool.Conn, cmds []Cmder) error {
|
func (c *Tx) execCmds(cn *pool.Conn, cmds []Cmder) error {
|
||||||
err := writeCmd(cn, cmds...)
|
err := writeCmd(cn, cmds...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
setCmdsErr(cmds[1:len(cmds)-1], err)
|
setCmdsErr(cmds[1:len(cmds)-1], err)
|
||||||
|
|
Loading…
Reference in New Issue