redis/ring.go

844 lines
18 KiB
Go
Raw Normal View History

2015-05-25 16:22:27 +03:00
package redis
import (
"context"
2020-09-11 11:24:38 +03:00
"crypto/tls"
2015-05-25 16:22:27 +03:00
"errors"
"fmt"
2020-06-10 10:36:22 +03:00
"net"
"strconv"
2015-05-25 16:22:27 +03:00
"sync"
"sync/atomic"
2015-05-25 16:22:27 +03:00
"time"
2020-07-06 07:51:51 +03:00
"github.com/cespare/xxhash/v2"
2023-05-16 17:02:22 +03:00
"github.com/dgryski/go-rendezvous" //nolint
chore: sync master (#2051) * Upgrade redis-server version (#1833) * Upgrade redis-server version Signed-off-by: monkey <golang@88.com> * XAutoClaim changed the return value Signed-off-by: monkey <golang@88.com> * add cmd: geosearch, geosearchstore (#1836) * add cmd: geosearch, geosearchstore Signed-off-by: monkey92t <golang@88.com> * GeoSearchQuery and GeoSearchLocationQuery changed to pointer passing Signed-off-by: monkey92t <golang@88.com> * Added missing method XInfoStreamFull to Cmdable interface * Run go mod tidy in redisotel Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com> * Revert "ConnPool check fd for bad conns (#1824)" (#1849) This reverts commit 346bfafddd36dd52d51b064033048de5552ee91e. * Automate release process (#1852) * Bump github.com/onsi/gomega from 1.10.5 to 1.14.0 (#1832) * Bump github.com/onsi/gomega from 1.10.5 to 1.14.0 Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.10.5 to 1.14.0. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.10.5...v1.14.0) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * Upgrade gomega to v1.15.0 Signed-off-by: monkey92t <golang@88.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: monkey92t <golang@88.com> * Add version.go * Fix otel example * Fix package name in release script * More fixes for otel example * And more * Fix release.sh * Release v8.11.3 (release.sh) * Create an annotated tag to give release.yml chance to run * Tweak tag.sh * Add Cmd.Slice helper to cast to []interface{} (#1859) * after the connection pool is closed, no new connections should be added (#1863) * after the connection pool is closed, no new connections should be added Signed-off-by: monkey92t <golang@88.com> * remove runGoroutine Signed-off-by: monkey92t <golang@88.com> * pool.popIdle add p.closed check Signed-off-by: monkey92t <golang@88.com> * upgrade golangci-lint v1.42.0 Signed-off-by: monkey92t <golang@88.com> * Bump github.com/onsi/gomega from 1.15.0 to 1.16.0 (#1865) Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.15.0 to 1.16.0. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.15.0...v1.16.0) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Add go 1.17 to the build matrix * Remove go 1.15 from build matrix * Add scan struct example (#1870) * Replace release job * Bump github.com/cespare/xxhash/v2 from 2.1.1 to 2.1.2 (#1872) Bumps [github.com/cespare/xxhash/v2](https://github.com/cespare/xxhash) from 2.1.1 to 2.1.2. - [Release notes](https://github.com/cespare/xxhash/releases) - [Commits](https://github.com/cespare/xxhash/compare/v2.1.1...v2.1.2) --- updated-dependencies: - dependency-name: github.com/cespare/xxhash/v2 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Fix tag script to push tag by tag * Fix releasing.md * Fix/pubsub ping mutex (#1878) * Fix PubSub.Ping to hold the lock * Fix PubSub.Ping to hold the lock * add write cmd data-race test Signed-off-by: monkey92t <golang@88.com> Co-authored-by: monkey92t <golang@88.com> * chore: cleanup OpenTelemetry example * chore: gofmt all code * Refactor TestParseURL This is in preparation for supporting query parameters in ParseURL: - use an expected *Options instance to execute assertions on - extract assertions into helper function - enable parallel testing - condense test table * Add query parameter parsing to ParseURL() Before this change, ParseURL would only accept a very restricted set of URLs (it returned an error, if it encountered any parameter). This commit introduces the ability to process URLs like redis://localhost/1?dial_timeout=10s and similar. Go programs which were providing a configuration tunable (e.g. CLI flag, config entry or environment variable) to configure the Redis connection now don't need to perform this task themselves. * chore: add links to readme * chore: fix discussions link * empty hooks.withContext removed * chore: gofmt * chore: use conventional commits and auto-generate changelog * feat: add acl auth support for sentinels * chore: swap to acl auth at the test-level * Add support for BLMove command * chore: update dependencies * chore: update link * feat: add SetVal method for each command * feat: add Cmd.{String,Int,Float,Bool}Slice helpers and an example * chore: tweak GH actions to run all jobs * chore: add Lua scripting example * Fix Redis Cluster issue during roll outs of new nodes with same addr (#1914) * fix: recycle connections in some Redis Cluster scenarios This issue was surfaced in a Cloud Provider solution that used for rolling out new nodes using the same address (hostname) of the nodes that will be replaced in a Redis Cluster, while the former ones once depromoted as Slaves would continue in service during some mintues for redirecting traffic. The solution basically identifies when the connection could be stale since a MOVED response will be returned using the same address (hostname) that is being used by the connection. At that moment we consider the connection as no longer usable forcing to recycle the connection. * chore: lazy reload when moved or ask * chore: use conv commit message * chore: release v8.11.4 (release.sh) * fix: add whitespace for avoid unlikely colisions * fix: format * chore: fix links * chore: use ctx parameter in cmdInfo * Bump github.com/onsi/ginkgo from 1.16.4 to 1.16.5 (#1925) Bumps [github.com/onsi/ginkgo](https://github.com/onsi/ginkgo) from 1.16.4 to 1.16.5. - [Release notes](https://github.com/onsi/ginkgo/releases) - [Changelog](https://github.com/onsi/ginkgo/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/ginkgo/compare/v1.16.4...v1.16.5) --- updated-dependencies: - dependency-name: github.com/onsi/ginkgo dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * feat: add support for time.Duration write and scan * test: add test case for setting and scanning durations * chore: fix linter * fix(extra/redisotel): set span.kind attribute to client According to the opentelemetry specification this should always be set to client for database client libraries. I've also removed the SetAttributes call and instead set the attributes during creation of the span. This is what the library SHOULD be doing according to the opentelemetry api specification. * chore: update otel example * fix: update some argument counts in pre-allocs In some cases number of pre-allocated places in argument array is missing 1 or 2 elements, which results in re-allocation of twice as large array * chore: add example how to delete keys without a ttl * chore: don't enable all lints * chore(deps): bump github.com/onsi/gomega from 1.16.0 to 1.17.0 Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.16.0 to 1.17.0. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.16.0...v1.17.0) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * feat: Add redis v7's NX, XX, GT, LT expire variants * chore: add missing readme * chore: tweak feature links * chore: remove Discord * fix: set timeout for WAIT command. Fixes #1963 * build: update `go` directive in `go.mod` to 1.17 This commit enables support for module graph pruning and lazy module loading for projects that are at Go 1.17 or higher. Reference: https://go.dev/ref/mod#go-mod-file-go Reference: https://go.dev/ref/mod#graph-pruning Reference: https://go.dev/ref/mod#lazy-loading Signed-off-by: Eng Zer Jun <engzerjun@gmail.com> * chore: update link * chore: export cmder.SetFirstKeyPos to support build module commands * feat(redisotel): ability to override TracerProvider (#1998) * fix: add missing Expire methods to Cmdable This is a followup to https://github.com/go-redis/redis/pull/1928 * chore(deps): bump github.com/onsi/gomega from 1.17.0 to 1.18.1 Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.17.0 to 1.18.1. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.17.0...v1.18.1) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * Update README.md (#2011) chore: add fmt library in example code * chore: instrumentation name and version (#2012) * fix: invalid type assert in stringArg * chore: cleanup * fix: example/otel compile error (#2028) * fix: rename Golang to Go (#2030) https://go.dev/doc/faq#go_or_golang * feat: add support for passing extra attributes added to spans * feat: set net.peer.name and net.peer.port in otel example * chore: tweak Uptrace copy * feat: add support for COPY command (#2016) * feat: add support for acl sentinel auth in universal client * chore(deps): bump actions/checkout from 2 to 3 Bumps [actions/checkout](https://github.com/actions/checkout) from 2 to 3. - [Release notes](https://github.com/actions/checkout/releases) - [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md) - [Commits](https://github.com/actions/checkout/compare/v2...v3) --- updated-dependencies: - dependency-name: actions/checkout dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> * chore: add hll example * chore: tweak release script * chore: release v8.11.5 (release.sh) * chore: add discord back Co-authored-by: Eugene Ponizovsky <ponizovsky@gmail.com> Co-authored-by: Bogdan Drutu <bogdandrutu@gmail.com> Co-authored-by: Vladimir Mihailenco <vladimir.webdev@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Kishan B <kishancs46@gmail.com> Co-authored-by: Dominik Menke <dom@digineo.de> Co-authored-by: Gökhan Özeloğlu <gozeloglu@gmail.com> Co-authored-by: Justin Sievenpiper <justin@sievenpiper.co> Co-authored-by: Алексей Романовский <aromanovsky@epiphan.com> Co-authored-by: Stavros Panakakakis <stavrospanakakis@gmail.com> Co-authored-by: Pau Freixes <pfreixes@gmail.com> Co-authored-by: Ethan Hur <ethan0311@gmail.com> Co-authored-by: Jackie <18378976+Pyrodash@users.noreply.github.com> Co-authored-by: Kristinn Björgvin Árdal <kristinnardalsecondary@gmail.com> Co-authored-by: ffenix113 <razerer@bigmir.net> Co-authored-by: Bastien Penavayre <bastienPenava@gmail.com> Co-authored-by: James3 Li(李麒傑) <james3_li@asus.com> Co-authored-by: Eng Zer Jun <engzerjun@gmail.com> Co-authored-by: gzjiangtao2014 <gzjiangtao2014@corp.netease.com> Co-authored-by: Nelz <nelz9999@users.noreply.github.com> Co-authored-by: Daniel Richter <Nexyz9@gmail.com> Co-authored-by: Seyed Ali Ghaffari <ali.ghaffari@outlook.com> Co-authored-by: lintanghui <lintanghui@bilibili.com> Co-authored-by: hidu <duv123+github@gmail.com> Co-authored-by: Jonas Lergell <jonas.lergell@volvocars.com> Co-authored-by: Alex Kahn <alexanderkahn@gmail.com>
2022-03-19 07:40:31 +03:00
2023-01-23 09:48:54 +03:00
"github.com/redis/go-redis/v9/internal"
"github.com/redis/go-redis/v9/internal/hashtag"
"github.com/redis/go-redis/v9/internal/pool"
"github.com/redis/go-redis/v9/internal/rand"
2015-05-25 16:22:27 +03:00
)
2016-08-09 16:32:08 +03:00
var errRingShardsDown = errors.New("redis: all ring shards are down")
2015-05-25 16:22:27 +03:00
//------------------------------------------------------------------------------
type ConsistentHash interface {
Get(string) string
}
type rendezvousWrapper struct {
*rendezvous.Rendezvous
}
func (w rendezvousWrapper) Get(key string) string {
return w.Lookup(key)
}
func newRendezvous(shards []string) ConsistentHash {
return rendezvousWrapper{rendezvous.New(shards, xxhash.Sum64String)}
}
//------------------------------------------------------------------------------
2015-05-25 16:22:27 +03:00
// RingOptions are used to configure a ring client and should be
// passed to NewRing.
type RingOptions struct {
2016-08-09 16:32:08 +03:00
// Map of name => host:port addresses of ring shards.
2015-05-25 16:22:27 +03:00
Addrs map[string]string
2022-08-17 13:18:58 +03:00
// NewClient creates a shard client with provided options.
NewClient func(opt *Options) *Client
// ClientName will execute the `CLIENT SETNAME ClientName` command for each conn.
ClientName string
2016-08-09 16:32:08 +03:00
// Frequency of PING commands sent to check shards availability.
// Shard is considered down after 3 subsequent failed checks.
HeartbeatFrequency time.Duration
// NewConsistentHash returns a consistent hash that is used
// to distribute keys across the shards.
//
// See https://medium.com/@dgryski/consistent-hashing-algorithmic-tradeoffs-ef6b8e2fcae8
// for consistent hashing algorithmic tradeoffs.
NewConsistentHash func(shards []string) ConsistentHash
2015-05-25 16:22:27 +03:00
// Following options are copied from Options struct.
2020-06-10 10:36:22 +03:00
Dialer func(ctx context.Context, network, addr string) (net.Conn, error)
OnConnect func(ctx context.Context, cn *Conn) error
2017-05-25 14:16:39 +03:00
2023-05-16 17:02:22 +03:00
Protocol int
Username string
2015-05-25 16:22:27 +03:00
Password string
2020-06-12 09:28:14 +03:00
DB int
2015-05-25 16:22:27 +03:00
2017-08-31 15:22:47 +03:00
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
2015-06-04 11:50:24 +03:00
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
ContextTimeoutEnabled bool
2015-05-25 16:22:27 +03:00
ContextTimeoutEnabled bool
// PoolFIFO uses FIFO mode for each node connection pool GET/PUT (default LIFO).
PoolFIFO bool
PoolSize int
PoolTimeout time.Duration
MinIdleConns int
MaxIdleConns int
MaxActiveConns int
ConnMaxIdleTime time.Duration
ConnMaxLifetime time.Duration
2020-09-11 11:24:38 +03:00
TLSConfig *tls.Config
Limiter Limiter
DisableIndentity bool
2015-05-25 16:22:27 +03:00
}
2016-08-09 16:32:08 +03:00
func (opt *RingOptions) init() {
if opt.NewClient == nil {
2022-08-17 13:18:58 +03:00
opt.NewClient = func(opt *Options) *Client {
return NewClient(opt)
}
}
2016-08-09 16:32:08 +03:00
if opt.HeartbeatFrequency == 0 {
opt.HeartbeatFrequency = 500 * time.Millisecond
}
2017-08-31 15:22:47 +03:00
if opt.NewConsistentHash == nil {
opt.NewConsistentHash = newRendezvous
}
2020-09-11 11:24:38 +03:00
if opt.MaxRetries == -1 {
opt.MaxRetries = 0
} else if opt.MaxRetries == 0 {
opt.MaxRetries = 3
}
2017-08-31 15:22:47 +03:00
switch opt.MinRetryBackoff {
case -1:
opt.MinRetryBackoff = 0
case 0:
opt.MinRetryBackoff = 8 * time.Millisecond
}
switch opt.MaxRetryBackoff {
case -1:
opt.MaxRetryBackoff = 0
case 0:
opt.MaxRetryBackoff = 512 * time.Millisecond
}
2016-08-09 16:32:08 +03:00
}
2016-06-05 14:10:30 +03:00
2020-04-19 16:40:26 +03:00
func (opt *RingOptions) clientOptions() *Options {
2015-05-25 16:22:27 +03:00
return &Options{
ClientName: opt.ClientName,
Dialer: opt.Dialer,
OnConnect: opt.OnConnect,
2017-05-25 14:16:39 +03:00
2023-05-16 17:02:22 +03:00
Protocol: opt.Protocol,
2020-06-12 09:28:14 +03:00
Username: opt.Username,
Password: opt.Password,
2020-06-12 09:28:14 +03:00
DB: opt.DB,
2015-05-25 16:22:27 +03:00
2020-09-11 11:24:38 +03:00
MaxRetries: -1,
DialTimeout: opt.DialTimeout,
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
ContextTimeoutEnabled: opt.ContextTimeoutEnabled,
2015-05-25 16:22:27 +03:00
ContextTimeoutEnabled: opt.ContextTimeoutEnabled,
PoolFIFO: opt.PoolFIFO,
PoolSize: opt.PoolSize,
PoolTimeout: opt.PoolTimeout,
MinIdleConns: opt.MinIdleConns,
MaxIdleConns: opt.MaxIdleConns,
MaxActiveConns: opt.MaxActiveConns,
ConnMaxIdleTime: opt.ConnMaxIdleTime,
ConnMaxLifetime: opt.ConnMaxLifetime,
2020-09-11 11:24:38 +03:00
TLSConfig: opt.TLSConfig,
Limiter: opt.Limiter,
DisableIndentity: opt.DisableIndentity,
2015-05-25 16:22:27 +03:00
}
}
2018-03-07 15:08:40 +03:00
//------------------------------------------------------------------------------
2015-05-25 16:22:27 +03:00
type ringShard struct {
Client *Client
down int32
addr string
2015-05-25 16:22:27 +03:00
}
2022-08-17 13:18:58 +03:00
func newRingShard(opt *RingOptions, addr string) *ringShard {
clopt := opt.clientOptions()
clopt.Addr = addr
return &ringShard{
2022-08-17 13:18:58 +03:00
Client: opt.NewClient(clopt),
addr: addr,
}
}
2015-05-25 16:22:27 +03:00
func (shard *ringShard) String() string {
var state string
if shard.IsUp() {
state = "up"
} else {
state = "down"
}
return fmt.Sprintf("%s is %s", shard.Client, state)
}
func (shard *ringShard) IsDown() bool {
2016-08-09 16:32:08 +03:00
const threshold = 3
return atomic.LoadInt32(&shard.down) >= threshold
2015-05-25 16:22:27 +03:00
}
func (shard *ringShard) IsUp() bool {
return !shard.IsDown()
}
// Vote votes to set shard state and returns true if state was changed.
func (shard *ringShard) Vote(up bool) bool {
if up {
changed := shard.IsDown()
atomic.StoreInt32(&shard.down, 0)
2015-05-25 16:22:27 +03:00
return changed
}
if shard.IsDown() {
return false
}
atomic.AddInt32(&shard.down, 1)
2015-05-25 16:22:27 +03:00
return shard.IsDown()
}
2018-03-07 15:08:40 +03:00
//------------------------------------------------------------------------------
2022-08-17 13:18:58 +03:00
type ringSharding struct {
opt *RingOptions
mu sync.RWMutex
shards *ringShards
closed bool
hash ConsistentHash
numShard int
onNewNode []func(rdb *Client)
// ensures exclusive access to SetAddrs so there is no need
// to hold mu for the duration of potentially long shard creation
setAddrsMu sync.Mutex
2018-03-07 15:08:40 +03:00
}
2022-08-17 13:18:58 +03:00
type ringShards struct {
m map[string]*ringShard
list []*ringShard
}
func newRingSharding(opt *RingOptions) *ringSharding {
c := &ringSharding{
opt: opt,
}
c.SetAddrs(opt.Addrs)
return c
}
func (c *ringSharding) OnNewNode(fn func(rdb *Client)) {
c.mu.Lock()
c.onNewNode = append(c.onNewNode, fn)
c.mu.Unlock()
}
// SetAddrs replaces the shards in use, such that you can increase and
// decrease number of shards, that you use. It will reuse shards that
// existed before and close the ones that will not be used anymore.
2022-08-17 13:18:58 +03:00
func (c *ringSharding) SetAddrs(addrs map[string]string) {
c.setAddrsMu.Lock()
defer c.setAddrsMu.Unlock()
cleanup := func(shards map[string]*ringShard) {
for addr, shard := range shards {
if err := shard.Client.Close(); err != nil {
internal.Logger.Printf(context.Background(), "shard.Close %s failed: %s", addr, err)
}
}
}
2022-08-17 13:18:58 +03:00
c.mu.RLock()
if c.closed {
c.mu.RUnlock()
return
}
existing := c.shards
c.mu.RUnlock()
shards, created, unused := c.newRingShards(addrs, existing)
c.mu.Lock()
if c.closed {
cleanup(created)
c.mu.Unlock()
return
}
2022-08-17 13:18:58 +03:00
c.shards = shards
c.rebalanceLocked()
2022-08-17 13:18:58 +03:00
c.mu.Unlock()
2018-07-22 10:50:26 +03:00
cleanup(unused)
2022-08-17 13:18:58 +03:00
}
func (c *ringSharding) newRingShards(
addrs map[string]string, existing *ringShards,
) (shards *ringShards, created, unused map[string]*ringShard) {
shards = &ringShards{m: make(map[string]*ringShard, len(addrs))}
created = make(map[string]*ringShard) // indexed by addr
unused = make(map[string]*ringShard) // indexed by addr
if existing != nil {
for _, shard := range existing.list {
unused[shard.addr] = shard
}
}
2022-08-17 13:18:58 +03:00
for name, addr := range addrs {
if shard, ok := unused[addr]; ok {
2022-08-17 13:18:58 +03:00
shards.m[name] = shard
delete(unused, addr)
2022-08-17 13:18:58 +03:00
} else {
shard := newRingShard(c.opt, addr)
shards.m[name] = shard
created[addr] = shard
for _, fn := range c.onNewNode {
fn(shard.Client)
}
2022-08-17 13:18:58 +03:00
}
}
2022-08-17 13:18:58 +03:00
for _, shard := range shards.m {
shards.list = append(shards.list, shard)
}
return
2018-03-07 15:08:40 +03:00
}
2022-08-17 13:18:58 +03:00
func (c *ringSharding) List() []*ringShard {
var list []*ringShard
2018-03-07 15:08:40 +03:00
c.mu.RLock()
if !c.closed {
2022-08-17 13:18:58 +03:00
list = c.shards.list
}
2018-03-07 15:08:40 +03:00
c.mu.RUnlock()
2018-03-07 15:08:40 +03:00
return list
}
2022-08-17 13:18:58 +03:00
func (c *ringSharding) Hash(key string) string {
key = hashtag.Key(key)
var hash string
2018-03-07 15:08:40 +03:00
c.mu.RLock()
defer c.mu.RUnlock()
if c.numShard > 0 {
hash = c.hash.Get(key)
}
2018-03-07 15:08:40 +03:00
return hash
}
2022-08-17 13:18:58 +03:00
func (c *ringSharding) GetByKey(key string) (*ringShard, error) {
2018-03-07 15:08:40 +03:00
key = hashtag.Key(key)
c.mu.RLock()
defer c.mu.RUnlock()
2018-03-07 15:08:40 +03:00
if c.closed {
return nil, pool.ErrClosed
}
if c.numShard == 0 {
return nil, errRingShardsDown
}
2022-08-17 13:18:58 +03:00
shardName := c.hash.Get(key)
if shardName == "" {
2018-03-07 15:08:40 +03:00
return nil, errRingShardsDown
}
2022-08-17 13:18:58 +03:00
return c.shards.m[shardName], nil
2018-03-07 15:08:40 +03:00
}
2022-08-17 13:18:58 +03:00
func (c *ringSharding) GetByName(shardName string) (*ringShard, error) {
if shardName == "" {
2018-03-07 15:08:40 +03:00
return c.Random()
}
c.mu.RLock()
defer c.mu.RUnlock()
2022-08-17 13:18:58 +03:00
return c.shards.m[shardName], nil
2018-03-07 15:08:40 +03:00
}
2022-08-17 13:18:58 +03:00
func (c *ringSharding) Random() (*ringShard, error) {
2018-03-07 15:08:40 +03:00
return c.GetByKey(strconv.Itoa(rand.Int()))
}
// Heartbeat monitors state of each shard in the ring.
2022-08-17 13:18:58 +03:00
func (c *ringSharding) Heartbeat(ctx context.Context, frequency time.Duration) {
2018-03-07 15:08:40 +03:00
ticker := time.NewTicker(frequency)
defer ticker.Stop()
2020-03-11 17:26:42 +03:00
for {
select {
case <-ticker.C:
var rebalance bool
for _, shard := range c.List() {
err := shard.Client.Ping(ctx).Err()
isUp := err == nil || err == pool.ErrPoolTimeout
if shard.Vote(isUp) {
internal.Logger.Printf(ctx, "ring shard state changed: %s", shard)
rebalance = true
}
2018-03-07 15:08:40 +03:00
}
if rebalance {
c.mu.Lock()
c.rebalanceLocked()
c.mu.Unlock()
}
case <-ctx.Done():
return
2018-03-07 15:08:40 +03:00
}
}
}
// rebalanceLocked removes dead shards from the Ring.
// Requires c.mu locked.
func (c *ringSharding) rebalanceLocked() {
if c.closed {
return
}
if c.shards == nil {
2022-08-17 13:18:58 +03:00
return
2018-03-07 15:08:40 +03:00
}
liveShards := make([]string, 0, len(c.shards.m))
for name, shard := range c.shards.m {
if shard.IsUp() {
liveShards = append(liveShards, name)
}
}
c.hash = c.opt.NewConsistentHash(liveShards)
c.numShard = len(liveShards)
}
2022-08-17 13:18:58 +03:00
func (c *ringSharding) Len() int {
2018-07-23 15:55:13 +03:00
c.mu.RLock()
defer c.mu.RUnlock()
return c.numShard
2018-07-23 15:55:13 +03:00
}
2022-08-17 13:18:58 +03:00
func (c *ringSharding) Close() error {
2018-03-07 15:08:40 +03:00
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return nil
}
c.closed = true
var firstErr error
2022-08-17 13:18:58 +03:00
for _, shard := range c.shards.list {
2018-03-07 15:08:40 +03:00
if err := shard.Client.Close(); err != nil && firstErr == nil {
firstErr = err
}
}
2018-03-07 15:08:40 +03:00
c.hash = nil
c.shards = nil
c.numShard = 0
2018-03-07 15:08:40 +03:00
return firstErr
}
//------------------------------------------------------------------------------
2019-02-01 00:24:44 +03:00
// Ring is a Redis client that uses consistent hashing to distribute
2015-09-12 09:36:03 +03:00
// keys across multiple Redis servers (shards). It's safe for
// concurrent use by multiple goroutines.
2015-05-25 16:22:27 +03:00
//
2015-11-21 11:20:01 +03:00
// Ring monitors the state of each shard and removes dead shards from
2019-02-01 00:24:44 +03:00
// the ring. When a shard comes online it is added back to the ring. This
2015-05-25 16:22:27 +03:00
// gives you maximum availability and partition tolerance, but no
// consistency between different shards or even clients. Each client
// uses shards that are available to the client and does not do any
// coordination when shard state is changed.
//
2016-08-09 16:32:08 +03:00
// Ring should be used when you need multiple Redis servers for caching
2015-05-25 16:22:27 +03:00
// and can tolerate losing data when one of the servers dies.
// Otherwise you should use Redis Cluster.
type Ring struct {
cmdable
2023-01-21 11:30:02 +03:00
hooksMixin
2022-08-17 13:18:58 +03:00
opt *RingOptions
sharding *ringSharding
cmdsInfoCache *cmdsInfoCache
heartbeatCancelFn context.CancelFunc
2015-05-25 16:22:27 +03:00
}
func NewRing(opt *RingOptions) *Ring {
2016-06-05 14:10:30 +03:00
opt.init()
2018-01-20 13:26:33 +03:00
hbCtx, hbCancel := context.WithCancel(context.Background())
2019-05-31 17:03:20 +03:00
ring := Ring{
2022-08-17 13:18:58 +03:00
opt: opt,
sharding: newRingSharding(opt),
heartbeatCancelFn: hbCancel,
2015-05-25 16:22:27 +03:00
}
2019-05-31 17:03:20 +03:00
ring.cmdsInfoCache = newCmdsInfoCache(ring.cmdsInfo)
2019-08-24 12:22:52 +03:00
ring.cmdable = ring.Process
2019-05-31 17:03:20 +03:00
2023-01-21 11:30:02 +03:00
ring.initHooks(hooks{
process: ring.process,
pipeline: func(ctx context.Context, cmds []Cmder) error {
return ring.generalProcessPipeline(ctx, cmds, false)
},
txPipeline: func(ctx context.Context, cmds []Cmder) error {
return ring.generalProcessPipeline(ctx, cmds, true)
},
2022-10-12 15:00:06 +03:00
})
2022-08-17 13:18:58 +03:00
go ring.sharding.Heartbeat(hbCtx, opt.HeartbeatFrequency)
2018-01-20 13:26:33 +03:00
2019-05-31 17:03:20 +03:00
return &ring
2015-05-25 16:22:27 +03:00
}
2022-08-17 13:18:58 +03:00
func (c *Ring) SetAddrs(addrs map[string]string) {
c.sharding.SetAddrs(addrs)
}
// Do create a Cmd from the args and processes the cmd.
2020-03-11 17:26:42 +03:00
func (c *Ring) Do(ctx context.Context, args ...interface{}) *Cmd {
cmd := NewCmd(ctx, args...)
_ = c.Process(ctx, cmd)
2019-05-31 17:03:20 +03:00
return cmd
}
2020-03-11 17:26:42 +03:00
func (c *Ring) Process(ctx context.Context, cmd Cmder) error {
2023-01-21 11:30:02 +03:00
err := c.processHook(ctx, cmd)
cmd.SetErr(err)
return err
}
2017-03-20 13:15:21 +03:00
// Options returns read-only Options that were used to create the client.
func (c *Ring) Options() *RingOptions {
return c.opt
}
2017-08-31 15:22:47 +03:00
func (c *Ring) retryBackoff(attempt int) time.Duration {
return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
}
// PoolStats returns accumulated connection pool stats.
func (c *Ring) PoolStats() *PoolStats {
2022-08-17 13:18:58 +03:00
shards := c.sharding.List()
var acc PoolStats
for _, shard := range shards {
s := shard.Client.connPool.Stats()
acc.Hits += s.Hits
acc.Misses += s.Misses
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
2018-08-12 10:08:21 +03:00
acc.IdleConns += s.IdleConns
}
return &acc
}
2018-07-23 15:55:13 +03:00
// Len returns the current number of shards in the ring.
func (c *Ring) Len() int {
2022-08-17 13:18:58 +03:00
return c.sharding.Len()
2018-07-23 15:55:13 +03:00
}
2017-04-11 17:29:31 +03:00
// Subscribe subscribes the client to the specified channels.
2020-03-11 17:26:42 +03:00
func (c *Ring) Subscribe(ctx context.Context, channels ...string) *PubSub {
2017-04-11 17:29:31 +03:00
if len(channels) == 0 {
panic("at least one channel is required")
}
2022-08-17 13:18:58 +03:00
shard, err := c.sharding.GetByKey(channels[0])
2017-04-11 17:29:31 +03:00
if err != nil {
2020-07-16 09:52:07 +03:00
// TODO: return PubSub with sticky error
2017-04-11 17:29:31 +03:00
panic(err)
}
2020-03-11 17:26:42 +03:00
return shard.Client.Subscribe(ctx, channels...)
2017-04-11 17:29:31 +03:00
}
// PSubscribe subscribes the client to the given patterns.
2020-03-11 17:26:42 +03:00
func (c *Ring) PSubscribe(ctx context.Context, channels ...string) *PubSub {
2017-04-11 17:29:31 +03:00
if len(channels) == 0 {
panic("at least one channel is required")
}
2022-08-17 13:18:58 +03:00
shard, err := c.sharding.GetByKey(channels[0])
2017-04-11 17:29:31 +03:00
if err != nil {
2020-07-16 09:52:07 +03:00
// TODO: return PubSub with sticky error
2017-04-11 17:29:31 +03:00
panic(err)
}
2020-03-11 17:26:42 +03:00
return shard.Client.PSubscribe(ctx, channels...)
2017-04-11 17:29:31 +03:00
}
2022-08-03 18:10:03 +03:00
// SSubscribe Subscribes the client to the specified shard channels.
func (c *Ring) SSubscribe(ctx context.Context, channels ...string) *PubSub {
if len(channels) == 0 {
panic("at least one channel is required")
}
shard, err := c.sharding.GetByKey(channels[0])
2022-08-03 18:10:03 +03:00
if err != nil {
// TODO: return PubSub with sticky error
panic(err)
}
return shard.Client.SSubscribe(ctx, channels...)
}
func (c *Ring) OnNewNode(fn func(rdb *Client)) {
c.sharding.OnNewNode(fn)
}
// ForEachShard concurrently calls the fn on each live shard in the ring.
// It returns the first error if any.
2020-03-11 17:26:42 +03:00
func (c *Ring) ForEachShard(
ctx context.Context,
fn func(ctx context.Context, client *Client) error,
) error {
2022-08-17 13:18:58 +03:00
shards := c.sharding.List()
var wg sync.WaitGroup
errCh := make(chan error, 1)
for _, shard := range shards {
if shard.IsDown() {
continue
}
wg.Add(1)
go func(shard *ringShard) {
defer wg.Done()
2020-03-11 17:26:42 +03:00
err := fn(ctx, shard.Client)
if err != nil {
select {
case errCh <- err:
default:
}
}
}(shard)
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
}
func (c *Ring) cmdsInfo(ctx context.Context) (map[string]*CommandInfo, error) {
2022-08-17 13:18:58 +03:00
shards := c.sharding.List()
var firstErr error
2018-05-17 15:21:51 +03:00
for _, shard := range shards {
cmdsInfo, err := shard.Client.Command(ctx).Result()
2018-05-17 15:21:51 +03:00
if err == nil {
return cmdsInfo, nil
}
2018-05-17 15:21:51 +03:00
if firstErr == nil {
firstErr = err
}
}
if firstErr == nil {
return nil, errRingShardsDown
}
2018-05-17 15:21:51 +03:00
return nil, firstErr
}
2020-09-14 15:37:05 +03:00
func (c *Ring) cmdInfo(ctx context.Context, name string) *CommandInfo {
cmdsInfo, err := c.cmdsInfoCache.Get(ctx)
2017-06-17 12:34:39 +03:00
if err != nil {
return nil
}
2018-03-06 15:50:48 +03:00
info := cmdsInfo[name]
2017-08-31 15:22:47 +03:00
if info == nil {
chore: sync master (#2051) * Upgrade redis-server version (#1833) * Upgrade redis-server version Signed-off-by: monkey <golang@88.com> * XAutoClaim changed the return value Signed-off-by: monkey <golang@88.com> * add cmd: geosearch, geosearchstore (#1836) * add cmd: geosearch, geosearchstore Signed-off-by: monkey92t <golang@88.com> * GeoSearchQuery and GeoSearchLocationQuery changed to pointer passing Signed-off-by: monkey92t <golang@88.com> * Added missing method XInfoStreamFull to Cmdable interface * Run go mod tidy in redisotel Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com> * Revert "ConnPool check fd for bad conns (#1824)" (#1849) This reverts commit 346bfafddd36dd52d51b064033048de5552ee91e. * Automate release process (#1852) * Bump github.com/onsi/gomega from 1.10.5 to 1.14.0 (#1832) * Bump github.com/onsi/gomega from 1.10.5 to 1.14.0 Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.10.5 to 1.14.0. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.10.5...v1.14.0) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * Upgrade gomega to v1.15.0 Signed-off-by: monkey92t <golang@88.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: monkey92t <golang@88.com> * Add version.go * Fix otel example * Fix package name in release script * More fixes for otel example * And more * Fix release.sh * Release v8.11.3 (release.sh) * Create an annotated tag to give release.yml chance to run * Tweak tag.sh * Add Cmd.Slice helper to cast to []interface{} (#1859) * after the connection pool is closed, no new connections should be added (#1863) * after the connection pool is closed, no new connections should be added Signed-off-by: monkey92t <golang@88.com> * remove runGoroutine Signed-off-by: monkey92t <golang@88.com> * pool.popIdle add p.closed check Signed-off-by: monkey92t <golang@88.com> * upgrade golangci-lint v1.42.0 Signed-off-by: monkey92t <golang@88.com> * Bump github.com/onsi/gomega from 1.15.0 to 1.16.0 (#1865) Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.15.0 to 1.16.0. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.15.0...v1.16.0) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Add go 1.17 to the build matrix * Remove go 1.15 from build matrix * Add scan struct example (#1870) * Replace release job * Bump github.com/cespare/xxhash/v2 from 2.1.1 to 2.1.2 (#1872) Bumps [github.com/cespare/xxhash/v2](https://github.com/cespare/xxhash) from 2.1.1 to 2.1.2. - [Release notes](https://github.com/cespare/xxhash/releases) - [Commits](https://github.com/cespare/xxhash/compare/v2.1.1...v2.1.2) --- updated-dependencies: - dependency-name: github.com/cespare/xxhash/v2 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Fix tag script to push tag by tag * Fix releasing.md * Fix/pubsub ping mutex (#1878) * Fix PubSub.Ping to hold the lock * Fix PubSub.Ping to hold the lock * add write cmd data-race test Signed-off-by: monkey92t <golang@88.com> Co-authored-by: monkey92t <golang@88.com> * chore: cleanup OpenTelemetry example * chore: gofmt all code * Refactor TestParseURL This is in preparation for supporting query parameters in ParseURL: - use an expected *Options instance to execute assertions on - extract assertions into helper function - enable parallel testing - condense test table * Add query parameter parsing to ParseURL() Before this change, ParseURL would only accept a very restricted set of URLs (it returned an error, if it encountered any parameter). This commit introduces the ability to process URLs like redis://localhost/1?dial_timeout=10s and similar. Go programs which were providing a configuration tunable (e.g. CLI flag, config entry or environment variable) to configure the Redis connection now don't need to perform this task themselves. * chore: add links to readme * chore: fix discussions link * empty hooks.withContext removed * chore: gofmt * chore: use conventional commits and auto-generate changelog * feat: add acl auth support for sentinels * chore: swap to acl auth at the test-level * Add support for BLMove command * chore: update dependencies * chore: update link * feat: add SetVal method for each command * feat: add Cmd.{String,Int,Float,Bool}Slice helpers and an example * chore: tweak GH actions to run all jobs * chore: add Lua scripting example * Fix Redis Cluster issue during roll outs of new nodes with same addr (#1914) * fix: recycle connections in some Redis Cluster scenarios This issue was surfaced in a Cloud Provider solution that used for rolling out new nodes using the same address (hostname) of the nodes that will be replaced in a Redis Cluster, while the former ones once depromoted as Slaves would continue in service during some mintues for redirecting traffic. The solution basically identifies when the connection could be stale since a MOVED response will be returned using the same address (hostname) that is being used by the connection. At that moment we consider the connection as no longer usable forcing to recycle the connection. * chore: lazy reload when moved or ask * chore: use conv commit message * chore: release v8.11.4 (release.sh) * fix: add whitespace for avoid unlikely colisions * fix: format * chore: fix links * chore: use ctx parameter in cmdInfo * Bump github.com/onsi/ginkgo from 1.16.4 to 1.16.5 (#1925) Bumps [github.com/onsi/ginkgo](https://github.com/onsi/ginkgo) from 1.16.4 to 1.16.5. - [Release notes](https://github.com/onsi/ginkgo/releases) - [Changelog](https://github.com/onsi/ginkgo/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/ginkgo/compare/v1.16.4...v1.16.5) --- updated-dependencies: - dependency-name: github.com/onsi/ginkgo dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * feat: add support for time.Duration write and scan * test: add test case for setting and scanning durations * chore: fix linter * fix(extra/redisotel): set span.kind attribute to client According to the opentelemetry specification this should always be set to client for database client libraries. I've also removed the SetAttributes call and instead set the attributes during creation of the span. This is what the library SHOULD be doing according to the opentelemetry api specification. * chore: update otel example * fix: update some argument counts in pre-allocs In some cases number of pre-allocated places in argument array is missing 1 or 2 elements, which results in re-allocation of twice as large array * chore: add example how to delete keys without a ttl * chore: don't enable all lints * chore(deps): bump github.com/onsi/gomega from 1.16.0 to 1.17.0 Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.16.0 to 1.17.0. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.16.0...v1.17.0) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * feat: Add redis v7's NX, XX, GT, LT expire variants * chore: add missing readme * chore: tweak feature links * chore: remove Discord * fix: set timeout for WAIT command. Fixes #1963 * build: update `go` directive in `go.mod` to 1.17 This commit enables support for module graph pruning and lazy module loading for projects that are at Go 1.17 or higher. Reference: https://go.dev/ref/mod#go-mod-file-go Reference: https://go.dev/ref/mod#graph-pruning Reference: https://go.dev/ref/mod#lazy-loading Signed-off-by: Eng Zer Jun <engzerjun@gmail.com> * chore: update link * chore: export cmder.SetFirstKeyPos to support build module commands * feat(redisotel): ability to override TracerProvider (#1998) * fix: add missing Expire methods to Cmdable This is a followup to https://github.com/go-redis/redis/pull/1928 * chore(deps): bump github.com/onsi/gomega from 1.17.0 to 1.18.1 Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.17.0 to 1.18.1. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](https://github.com/onsi/gomega/compare/v1.17.0...v1.18.1) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * Update README.md (#2011) chore: add fmt library in example code * chore: instrumentation name and version (#2012) * fix: invalid type assert in stringArg * chore: cleanup * fix: example/otel compile error (#2028) * fix: rename Golang to Go (#2030) https://go.dev/doc/faq#go_or_golang * feat: add support for passing extra attributes added to spans * feat: set net.peer.name and net.peer.port in otel example * chore: tweak Uptrace copy * feat: add support for COPY command (#2016) * feat: add support for acl sentinel auth in universal client * chore(deps): bump actions/checkout from 2 to 3 Bumps [actions/checkout](https://github.com/actions/checkout) from 2 to 3. - [Release notes](https://github.com/actions/checkout/releases) - [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md) - [Commits](https://github.com/actions/checkout/compare/v2...v3) --- updated-dependencies: - dependency-name: actions/checkout dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> * chore: add hll example * chore: tweak release script * chore: release v8.11.5 (release.sh) * chore: add discord back Co-authored-by: Eugene Ponizovsky <ponizovsky@gmail.com> Co-authored-by: Bogdan Drutu <bogdandrutu@gmail.com> Co-authored-by: Vladimir Mihailenco <vladimir.webdev@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Kishan B <kishancs46@gmail.com> Co-authored-by: Dominik Menke <dom@digineo.de> Co-authored-by: Gökhan Özeloğlu <gozeloglu@gmail.com> Co-authored-by: Justin Sievenpiper <justin@sievenpiper.co> Co-authored-by: Алексей Романовский <aromanovsky@epiphan.com> Co-authored-by: Stavros Panakakakis <stavrospanakakis@gmail.com> Co-authored-by: Pau Freixes <pfreixes@gmail.com> Co-authored-by: Ethan Hur <ethan0311@gmail.com> Co-authored-by: Jackie <18378976+Pyrodash@users.noreply.github.com> Co-authored-by: Kristinn Björgvin Árdal <kristinnardalsecondary@gmail.com> Co-authored-by: ffenix113 <razerer@bigmir.net> Co-authored-by: Bastien Penavayre <bastienPenava@gmail.com> Co-authored-by: James3 Li(李麒傑) <james3_li@asus.com> Co-authored-by: Eng Zer Jun <engzerjun@gmail.com> Co-authored-by: gzjiangtao2014 <gzjiangtao2014@corp.netease.com> Co-authored-by: Nelz <nelz9999@users.noreply.github.com> Co-authored-by: Daniel Richter <Nexyz9@gmail.com> Co-authored-by: Seyed Ali Ghaffari <ali.ghaffari@outlook.com> Co-authored-by: lintanghui <lintanghui@bilibili.com> Co-authored-by: hidu <duv123+github@gmail.com> Co-authored-by: Jonas Lergell <jonas.lergell@volvocars.com> Co-authored-by: Alex Kahn <alexanderkahn@gmail.com>
2022-03-19 07:40:31 +03:00
internal.Logger.Printf(ctx, "info for cmd=%s not found", name)
2017-08-31 15:22:47 +03:00
}
return info
}
2020-09-14 15:37:05 +03:00
func (c *Ring) cmdShard(ctx context.Context, cmd Cmder) (*ringShard, error) {
cmdInfo := c.cmdInfo(ctx, cmd.Name())
pos := cmdFirstKeyPos(cmd, cmdInfo)
if pos == 0 {
2022-08-17 13:18:58 +03:00
return c.sharding.Random()
}
firstKey := cmd.stringArg(pos)
2022-08-17 13:18:58 +03:00
return c.sharding.GetByKey(firstKey)
}
2019-06-04 13:30:47 +03:00
func (c *Ring) process(ctx context.Context, cmd Cmder) error {
var lastErr error
2018-09-07 11:45:56 +03:00
for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
if attempt > 0 {
2019-07-30 12:13:00 +03:00
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
return err
}
2018-09-07 11:45:56 +03:00
}
2020-09-14 15:37:05 +03:00
shard, err := c.cmdShard(ctx, cmd)
2018-09-07 11:45:56 +03:00
if err != nil {
return err
}
2020-03-11 17:26:42 +03:00
lastErr = shard.Client.Process(ctx, cmd)
2020-07-24 14:57:12 +03:00
if lastErr == nil || !shouldRetry(lastErr, cmd.readTimeout() == nil) {
return lastErr
2018-09-07 11:45:56 +03:00
}
2015-05-25 16:22:27 +03:00
}
return lastErr
2015-05-25 16:22:27 +03:00
}
2020-03-11 17:26:42 +03:00
func (c *Ring) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().Pipelined(ctx, fn)
2019-05-31 17:03:20 +03:00
}
2017-05-02 18:00:53 +03:00
func (c *Ring) Pipeline() Pipeliner {
pipe := Pipeline{
2023-01-21 11:30:02 +03:00
exec: pipelineExecer(c.processPipelineHook),
2015-06-04 11:50:24 +03:00
}
2019-05-31 17:03:20 +03:00
pipe.init()
return &pipe
2015-06-04 11:50:24 +03:00
}
2020-03-11 17:26:42 +03:00
func (c *Ring) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.TxPipeline().Pipelined(ctx, fn)
2019-07-19 12:00:11 +03:00
}
func (c *Ring) TxPipeline() Pipeliner {
pipe := Pipeline{
exec: func(ctx context.Context, cmds []Cmder) error {
cmds = wrapMultiExec(ctx, cmds)
2023-01-21 11:30:02 +03:00
return c.processTxPipelineHook(ctx, cmds)
},
2019-07-19 12:00:11 +03:00
}
pipe.init()
return &pipe
}
func (c *Ring) generalProcessPipeline(
ctx context.Context, cmds []Cmder, tx bool,
) error {
if tx {
// Trim multi .. exec.
cmds = cmds[1 : len(cmds)-1]
}
2015-06-04 11:50:24 +03:00
cmdsMap := make(map[string][]Cmder)
2015-06-04 11:50:24 +03:00
for _, cmd := range cmds {
2020-09-14 15:37:05 +03:00
cmdInfo := c.cmdInfo(ctx, cmd.Name())
2018-03-07 15:08:40 +03:00
hash := cmd.stringArg(cmdFirstKeyPos(cmd, cmdInfo))
if hash != "" {
2022-08-17 13:18:58 +03:00
hash = c.sharding.Hash(hash)
}
2018-03-07 15:08:40 +03:00
cmdsMap[hash] = append(cmdsMap[hash], cmd)
2015-06-04 11:50:24 +03:00
}
var wg sync.WaitGroup
for hash, cmds := range cmdsMap {
wg.Add(1)
go func(hash string, cmds []Cmder) {
defer wg.Done()
2015-06-04 11:50:24 +03:00
// TODO: retry?
shard, err := c.sharding.GetByName(hash)
if err != nil {
setCmdsErr(cmds, err)
return
}
if tx {
cmds = wrapMultiExec(ctx, cmds)
2023-01-21 11:30:02 +03:00
_ = shard.Client.processTxPipelineHook(ctx, cmds)
} else {
2023-01-21 11:30:02 +03:00
_ = shard.Client.processPipelineHook(ctx, cmds)
}
}(hash, cmds)
}
2017-08-31 15:22:47 +03:00
wg.Wait()
return cmdsFirstErr(cmds)
}
2015-06-04 11:50:24 +03:00
2020-03-11 17:26:42 +03:00
func (c *Ring) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
2019-04-22 12:48:06 +03:00
if len(keys) == 0 {
return fmt.Errorf("redis: Watch requires at least one key")
}
var shards []*ringShard
2019-04-22 12:48:06 +03:00
for _, key := range keys {
if key != "" {
2022-08-17 13:18:58 +03:00
shard, err := c.sharding.GetByKey(hashtag.Key(key))
2019-04-22 12:48:06 +03:00
if err != nil {
return err
}
shards = append(shards, shard)
}
}
if len(shards) == 0 {
return fmt.Errorf("redis: Watch requires at least one shard")
}
if len(shards) > 1 {
for _, shard := range shards[1:] {
if shard.Client != shards[0].Client {
err := fmt.Errorf("redis: Watch requires all keys to be in the same shard")
return err
}
}
}
2020-03-11 17:26:42 +03:00
return shards[0].Client.Watch(ctx, fn, keys...)
2019-04-22 12:48:06 +03:00
}
2020-06-10 15:04:12 +03:00
// Close closes the ring client, releasing any open resources.
//
// It is rare to Close a Ring, as the Ring is meant to be long-lived
// and shared between many goroutines.
func (c *Ring) Close() error {
c.heartbeatCancelFn()
2022-08-17 13:18:58 +03:00
return c.sharding.Close()
2020-06-10 15:04:12 +03:00
}