diff options
author | 6543 <6543@obermui.de> | 2021-02-10 22:28:32 +0100 |
---|---|---|
committer | GitHub <noreply@github.com> | 2021-02-10 21:28:32 +0000 |
commit | ac97ea573c1b10d03e72775e8f74b9fe5453bfc8 (patch) | |
tree | 99fa7488782a2a6c4362c49e4cdf04594c662ca3 /modules | |
parent | 4cffc46f651205b9d7eb0b1df46dd6117c6d95e9 (diff) | |
download | gitea-ac97ea573c1b10d03e72775e8f74b9fe5453bfc8.tar.gz gitea-ac97ea573c1b10d03e72775e8f74b9fe5453bfc8.zip |
[Vendor] Update go-redis to v8.5.0 (#13749)
* Update go-redis to v8.4.0
* github.com/go-redis/redis/v8 v8.4.0 -> v8.5.0
* Apply suggestions from code review
Co-authored-by: zeripath <art27@cantab.net>
* TODO
* Use the Queue termination channel as the default context for pushes
Signed-off-by: Andrew Thornton <art27@cantab.net>
* missed one
Signed-off-by: Andrew Thornton <art27@cantab.net>
Co-authored-by: zeripath <art27@cantab.net>
Diffstat (limited to 'modules')
-rw-r--r-- | modules/cache/cache_redis.go | 33 | ||||
-rw-r--r-- | modules/nosql/manager.go | 2 | ||||
-rw-r--r-- | modules/nosql/manager_redis.go | 2 | ||||
-rw-r--r-- | modules/queue/queue_bytefifo.go | 10 | ||||
-rw-r--r-- | modules/queue/queue_redis.go | 32 | ||||
-rw-r--r-- | modules/queue/unique_queue_redis.go | 19 | ||||
-rw-r--r-- | modules/session/redis.go | 27 |
7 files changed, 78 insertions, 47 deletions
diff --git a/modules/cache/cache_redis.go b/modules/cache/cache_redis.go index 3cb0292e21..148725ae66 100644 --- a/modules/cache/cache_redis.go +++ b/modules/cache/cache_redis.go @@ -8,10 +8,11 @@ import ( "fmt" "time" + "code.gitea.io/gitea/modules/graceful" "code.gitea.io/gitea/modules/nosql" "gitea.com/go-chi/cache" - "github.com/go-redis/redis/v7" + "github.com/go-redis/redis/v8" "github.com/unknwon/com" ) @@ -28,7 +29,7 @@ type RedisCacher struct { func (c *RedisCacher) Put(key string, val interface{}, expire int64) error { key = c.prefix + key if expire == 0 { - if err := c.c.Set(key, com.ToStr(val), 0).Err(); err != nil { + if err := c.c.Set(graceful.GetManager().HammerContext(), key, com.ToStr(val), 0).Err(); err != nil { return err } } else { @@ -36,7 +37,7 @@ func (c *RedisCacher) Put(key string, val interface{}, expire int64) error { if err != nil { return err } - if err = c.c.Set(key, com.ToStr(val), dur).Err(); err != nil { + if err = c.c.Set(graceful.GetManager().HammerContext(), key, com.ToStr(val), dur).Err(); err != nil { return err } } @@ -44,12 +45,12 @@ func (c *RedisCacher) Put(key string, val interface{}, expire int64) error { if c.occupyMode { return nil } - return c.c.HSet(c.hsetName, key, "0").Err() + return c.c.HSet(graceful.GetManager().HammerContext(), c.hsetName, key, "0").Err() } // Get gets cached value by given key. func (c *RedisCacher) Get(key string) interface{} { - val, err := c.c.Get(c.prefix + key).Result() + val, err := c.c.Get(graceful.GetManager().HammerContext(), c.prefix+key).Result() if err != nil { return nil } @@ -59,14 +60,14 @@ func (c *RedisCacher) Get(key string) interface{} { // Delete deletes cached value by given key. func (c *RedisCacher) Delete(key string) error { key = c.prefix + key - if err := c.c.Del(key).Err(); err != nil { + if err := c.c.Del(graceful.GetManager().HammerContext(), key).Err(); err != nil { return err } if c.occupyMode { return nil } - return c.c.HDel(c.hsetName, key).Err() + return c.c.HDel(graceful.GetManager().HammerContext(), c.hsetName, key).Err() } // Incr increases cached int-type value by given key as a counter. @@ -74,7 +75,7 @@ func (c *RedisCacher) Incr(key string) error { if !c.IsExist(key) { return fmt.Errorf("key '%s' not exist", key) } - return c.c.Incr(c.prefix + key).Err() + return c.c.Incr(graceful.GetManager().HammerContext(), c.prefix+key).Err() } // Decr decreases cached int-type value by given key as a counter. @@ -82,17 +83,17 @@ func (c *RedisCacher) Decr(key string) error { if !c.IsExist(key) { return fmt.Errorf("key '%s' not exist", key) } - return c.c.Decr(c.prefix + key).Err() + return c.c.Decr(graceful.GetManager().HammerContext(), c.prefix+key).Err() } // IsExist returns true if cached value exists. func (c *RedisCacher) IsExist(key string) bool { - if c.c.Exists(c.prefix+key).Val() == 1 { + if c.c.Exists(graceful.GetManager().HammerContext(), c.prefix+key).Val() == 1 { return true } if !c.occupyMode { - c.c.HDel(c.hsetName, c.prefix+key) + c.c.HDel(graceful.GetManager().HammerContext(), c.hsetName, c.prefix+key) } return false } @@ -100,17 +101,17 @@ func (c *RedisCacher) IsExist(key string) bool { // Flush deletes all cached data. func (c *RedisCacher) Flush() error { if c.occupyMode { - return c.c.FlushDB().Err() + return c.c.FlushDB(graceful.GetManager().HammerContext()).Err() } - keys, err := c.c.HKeys(c.hsetName).Result() + keys, err := c.c.HKeys(graceful.GetManager().HammerContext(), c.hsetName).Result() if err != nil { return err } - if err = c.c.Del(keys...).Err(); err != nil { + if err = c.c.Del(graceful.GetManager().HammerContext(), keys...).Err(); err != nil { return err } - return c.c.Del(c.hsetName).Err() + return c.c.Del(graceful.GetManager().HammerContext(), c.hsetName).Err() } // StartAndGC starts GC routine based on config string settings. @@ -132,7 +133,7 @@ func (c *RedisCacher) StartAndGC(opts cache.Options) error { } } - return c.c.Ping().Err() + return c.c.Ping(graceful.GetManager().HammerContext()).Err() } func init() { diff --git a/modules/nosql/manager.go b/modules/nosql/manager.go index ad61d6d18c..a89b5bb633 100644 --- a/modules/nosql/manager.go +++ b/modules/nosql/manager.go @@ -9,7 +9,7 @@ import ( "sync" "time" - "github.com/go-redis/redis/v7" + "github.com/go-redis/redis/v8" "github.com/syndtr/goleveldb/leveldb" ) diff --git a/modules/nosql/manager_redis.go b/modules/nosql/manager_redis.go index 7792a90112..d754a0e07d 100644 --- a/modules/nosql/manager_redis.go +++ b/modules/nosql/manager_redis.go @@ -10,7 +10,7 @@ import ( "strconv" "strings" - "github.com/go-redis/redis/v7" + "github.com/go-redis/redis/v8" ) var replacer = strings.NewReplacer("_", "", "-", "") diff --git a/modules/queue/queue_bytefifo.go b/modules/queue/queue_bytefifo.go index cad258bda8..4bb7bbf2b1 100644 --- a/modules/queue/queue_bytefifo.go +++ b/modules/queue/queue_bytefifo.go @@ -163,6 +163,11 @@ func (q *ByteFIFOQueue) Shutdown() { log.Debug("%s: %s Shutdown", q.typ, q.name) } +// IsShutdown returns a channel which is closed when this Queue is shutdown +func (q *ByteFIFOQueue) IsShutdown() <-chan struct{} { + return q.closed +} + // Terminate this queue and close the queue func (q *ByteFIFOQueue) Terminate() { log.Trace("%s: %s Terminating", q.typ, q.name) @@ -185,6 +190,11 @@ func (q *ByteFIFOQueue) Terminate() { log.Debug("%s: %s Terminated", q.typ, q.name) } +// IsTerminated returns a channel which is closed when this Queue is terminated +func (q *ByteFIFOQueue) IsTerminated() <-chan struct{} { + return q.terminated +} + var _ (UniqueQueue) = &ByteFIFOUniqueQueue{} // ByteFIFOUniqueQueue represents a UniqueQueue formed from a UniqueByteFifo diff --git a/modules/queue/queue_redis.go b/modules/queue/queue_redis.go index 04e7b5d252..2b1d36f0ad 100644 --- a/modules/queue/queue_redis.go +++ b/modules/queue/queue_redis.go @@ -5,10 +5,14 @@ package queue import ( + "context" + "fmt" + + "code.gitea.io/gitea/modules/graceful" "code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/modules/nosql" - "github.com/go-redis/redis/v7" + "github.com/go-redis/redis/v8" ) // RedisQueueType is the type for redis queue @@ -43,6 +47,8 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error) return nil, err } + byteFIFO.ctx = graceful.NewChannelContext(byteFIFOQueue.IsTerminated(), fmt.Errorf("queue has been terminated")) + queue := &RedisQueue{ ByteFIFOQueue: byteFIFOQueue, } @@ -53,13 +59,13 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error) } type redisClient interface { - RPush(key string, args ...interface{}) *redis.IntCmd - LPop(key string) *redis.StringCmd - LLen(key string) *redis.IntCmd - SAdd(key string, members ...interface{}) *redis.IntCmd - SRem(key string, members ...interface{}) *redis.IntCmd - SIsMember(key string, member interface{}) *redis.BoolCmd - Ping() *redis.StatusCmd + RPush(ctx context.Context, key string, args ...interface{}) *redis.IntCmd + LPop(ctx context.Context, key string) *redis.StringCmd + LLen(ctx context.Context, key string) *redis.IntCmd + SAdd(ctx context.Context, key string, members ...interface{}) *redis.IntCmd + SRem(ctx context.Context, key string, members ...interface{}) *redis.IntCmd + SIsMember(ctx context.Context, key string, member interface{}) *redis.BoolCmd + Ping(ctx context.Context) *redis.StatusCmd Close() error } @@ -67,6 +73,7 @@ var _ (ByteFIFO) = &RedisByteFIFO{} // RedisByteFIFO represents a ByteFIFO formed from a redisClient type RedisByteFIFO struct { + ctx context.Context client redisClient queueName string } @@ -82,8 +89,9 @@ func NewRedisByteFIFO(config RedisByteFIFOConfiguration) (*RedisByteFIFO, error) fifo := &RedisByteFIFO{ queueName: config.QueueName, } + fifo.ctx = graceful.GetManager().TerminateContext() fifo.client = nosql.GetManager().GetRedisClient(config.ConnectionString) - if err := fifo.client.Ping().Err(); err != nil { + if err := fifo.client.Ping(graceful.GetManager().ShutdownContext()).Err(); err != nil { return nil, err } return fifo, nil @@ -96,12 +104,12 @@ func (fifo *RedisByteFIFO) PushFunc(data []byte, fn func() error) error { return err } } - return fifo.client.RPush(fifo.queueName, data).Err() + return fifo.client.RPush(fifo.ctx, fifo.queueName, data).Err() } // Pop pops data from the start of the fifo func (fifo *RedisByteFIFO) Pop() ([]byte, error) { - data, err := fifo.client.LPop(fifo.queueName).Bytes() + data, err := fifo.client.LPop(fifo.ctx, fifo.queueName).Bytes() if err == nil || err == redis.Nil { return data, nil } @@ -115,7 +123,7 @@ func (fifo *RedisByteFIFO) Close() error { // Len returns the length of the fifo func (fifo *RedisByteFIFO) Len() int64 { - val, err := fifo.client.LLen(fifo.queueName).Result() + val, err := fifo.client.LLen(fifo.ctx, fifo.queueName).Result() if err != nil { log.Error("Error whilst getting length of redis queue %s: Error: %v", fifo.queueName, err) return -1 diff --git a/modules/queue/unique_queue_redis.go b/modules/queue/unique_queue_redis.go index 67efc66bc9..96fcad1a83 100644 --- a/modules/queue/unique_queue_redis.go +++ b/modules/queue/unique_queue_redis.go @@ -4,7 +4,12 @@ package queue -import "github.com/go-redis/redis/v7" +import ( + "fmt" + + "code.gitea.io/gitea/modules/graceful" + "github.com/go-redis/redis/v8" +) // RedisUniqueQueueType is the type for redis queue const RedisUniqueQueueType Type = "unique-redis" @@ -46,6 +51,8 @@ func NewRedisUniqueQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, return nil, err } + byteFIFO.ctx = graceful.NewChannelContext(byteFIFOQueue.IsTerminated(), fmt.Errorf("queue has been terminated")) + queue := &RedisUniqueQueue{ ByteFIFOUniqueQueue: byteFIFOQueue, } @@ -86,7 +93,7 @@ func NewRedisUniqueByteFIFO(config RedisUniqueByteFIFOConfiguration) (*RedisUniq // PushFunc pushes data to the end of the fifo and calls the callback if it is added func (fifo *RedisUniqueByteFIFO) PushFunc(data []byte, fn func() error) error { - added, err := fifo.client.SAdd(fifo.setName, data).Result() + added, err := fifo.client.SAdd(fifo.ctx, fifo.setName, data).Result() if err != nil { return err } @@ -98,12 +105,12 @@ func (fifo *RedisUniqueByteFIFO) PushFunc(data []byte, fn func() error) error { return err } } - return fifo.client.RPush(fifo.queueName, data).Err() + return fifo.client.RPush(fifo.ctx, fifo.queueName, data).Err() } // Pop pops data from the start of the fifo func (fifo *RedisUniqueByteFIFO) Pop() ([]byte, error) { - data, err := fifo.client.LPop(fifo.queueName).Bytes() + data, err := fifo.client.LPop(fifo.ctx, fifo.queueName).Bytes() if err != nil && err != redis.Nil { return data, err } @@ -112,13 +119,13 @@ func (fifo *RedisUniqueByteFIFO) Pop() ([]byte, error) { return data, nil } - err = fifo.client.SRem(fifo.setName, data).Err() + err = fifo.client.SRem(fifo.ctx, fifo.setName, data).Err() return data, err } // Has returns whether the fifo contains this data func (fifo *RedisUniqueByteFIFO) Has(data []byte) (bool, error) { - return fifo.client.SIsMember(fifo.setName, data).Result() + return fifo.client.SIsMember(fifo.ctx, fifo.setName, data).Result() } func init() { diff --git a/modules/session/redis.go b/modules/session/redis.go index 55e7a85168..334418bd7e 100644 --- a/modules/session/redis.go +++ b/modules/session/redis.go @@ -21,10 +21,11 @@ import ( "sync" "time" + "code.gitea.io/gitea/modules/graceful" "code.gitea.io/gitea/modules/nosql" "gitea.com/go-chi/session" - "github.com/go-redis/redis/v7" + "github.com/go-redis/redis/v8" ) // RedisStore represents a redis session store implementation. @@ -90,7 +91,7 @@ func (s *RedisStore) Release() error { return err } - return s.c.Set(s.prefix+s.sid, string(data), s.duration).Err() + return s.c.Set(graceful.GetManager().HammerContext(), s.prefix+s.sid, string(data), s.duration).Err() } // Flush deletes all session data. @@ -127,20 +128,20 @@ func (p *RedisProvider) Init(maxlifetime int64, configs string) (err error) { } p.c = nosql.GetManager().GetRedisClient(uri.String()) - return p.c.Ping().Err() + return p.c.Ping(graceful.GetManager().ShutdownContext()).Err() } // Read returns raw session store by session ID. func (p *RedisProvider) Read(sid string) (session.RawStore, error) { psid := p.prefix + sid if !p.Exist(sid) { - if err := p.c.Set(psid, "", p.duration).Err(); err != nil { + if err := p.c.Set(graceful.GetManager().HammerContext(), psid, "", p.duration).Err(); err != nil { return nil, err } } var kv map[interface{}]interface{} - kvs, err := p.c.Get(psid).Result() + kvs, err := p.c.Get(graceful.GetManager().HammerContext(), psid).Result() if err != nil { return nil, err } @@ -158,13 +159,13 @@ func (p *RedisProvider) Read(sid string) (session.RawStore, error) { // Exist returns true if session with given ID exists. func (p *RedisProvider) Exist(sid string) bool { - v, err := p.c.Exists(p.prefix + sid).Result() + v, err := p.c.Exists(graceful.GetManager().HammerContext(), p.prefix+sid).Result() return err == nil && v == 1 } // Destroy deletes a session by session ID. func (p *RedisProvider) Destroy(sid string) error { - return p.c.Del(p.prefix + sid).Err() + return p.c.Del(graceful.GetManager().HammerContext(), p.prefix+sid).Err() } // Regenerate regenerates a session store from old session ID to new one. @@ -176,17 +177,17 @@ func (p *RedisProvider) Regenerate(oldsid, sid string) (_ session.RawStore, err return nil, fmt.Errorf("new sid '%s' already exists", sid) } else if !p.Exist(oldsid) { // Make a fake old session. - if err = p.c.Set(poldsid, "", p.duration).Err(); err != nil { + if err = p.c.Set(graceful.GetManager().HammerContext(), poldsid, "", p.duration).Err(); err != nil { return nil, err } } - if err = p.c.Rename(poldsid, psid).Err(); err != nil { + if err = p.c.Rename(graceful.GetManager().HammerContext(), poldsid, psid).Err(); err != nil { return nil, err } var kv map[interface{}]interface{} - kvs, err := p.c.Get(psid).Result() + kvs, err := p.c.Get(graceful.GetManager().HammerContext(), psid).Result() if err != nil { return nil, err } @@ -205,7 +206,11 @@ func (p *RedisProvider) Regenerate(oldsid, sid string) (_ session.RawStore, err // Count counts and returns number of sessions. func (p *RedisProvider) Count() int { - return int(p.c.DBSize().Val()) + size, err := p.c.DBSize(graceful.GetManager().HammerContext()).Result() + if err != nil { + return 0 + } + return int(size) } // GC calls GC to clean expired sessions. |