pass down context

Signed-off-by: Andrew Thornton <[email protected]>

modules/queue/bytefifo.go

@@ -4,14 +4,16 @@
 
 package queue
 
+import "context"
+
 // ByteFIFO defines a FIFO that takes a byte array
 type ByteFIFO interface {
 	// Len returns the length of the fifo
-	Len() int64
+	Len(ctx context.Context) int64
 	// PushFunc pushes data to the end of the fifo and calls the callback if it is added
-	PushFunc(data []byte, fn func() error) error
+	PushFunc(ctx context.Context, data []byte, fn func() error) error
 	// Pop pops data from the start of the fifo
-	Pop() ([]byte, error)
+	Pop(ctx context.Context) ([]byte, error)
 	// Close this fifo
 	Close() error
 }
@@ -20,7 +22,7 @@ type ByteFIFO interface {
 type UniqueByteFIFO interface {
 	ByteFIFO
 	// Has returns whether the fifo contains this data
-	Has(data []byte) (bool, error)
+	Has(ctx context.Context, data []byte) (bool, error)
 }
 
 var _ ByteFIFO = &DummyByteFIFO{}
@@ -29,12 +31,12 @@ var _ ByteFIFO = &DummyByteFIFO{}
 type DummyByteFIFO struct{}
 
 // PushFunc returns nil
-func (*DummyByteFIFO) PushFunc(data []byte, fn func() error) error {
+func (*DummyByteFIFO) PushFunc(ctx context.Context, data []byte, fn func() error) error {
 	return nil
 }
 
 // Pop returns nil
-func (*DummyByteFIFO) Pop() ([]byte, error) {
+func (*DummyByteFIFO) Pop(ctx context.Context) ([]byte, error) {
 	return []byte{}, nil
 }
 
@@ -44,7 +46,7 @@ func (*DummyByteFIFO) Close() error {
 }
 
 // Len is always 0
-func (*DummyByteFIFO) Len() int64 {
+func (*DummyByteFIFO) Len(ctx context.Context) int64 {
 	return 0
 }
 
@@ -56,6 +58,6 @@ type DummyUniqueByteFIFO struct {
 }
 
 // Has always returns false
-func (*DummyUniqueByteFIFO) Has([]byte) (bool, error) {
+func (*DummyUniqueByteFIFO) Has(ctx context.Context, data []byte) (bool, error) {
 	return false, nil
 }

modules/queue/queue_bytefifo.go

@@ -96,7 +96,7 @@ func (q *ByteFIFOQueue) PushFunc(data Data, fn func() error) error {
 			}
 		}()
 	}
-	return q.byteFIFO.PushFunc(bs, fn)
+	return q.byteFIFO.PushFunc(q.terminateCtx, bs, fn)
 }
 
 // IsEmpty checks if the queue is empty
@@ -106,7 +106,7 @@ func (q *ByteFIFOQueue) IsEmpty() bool {
 	if !q.WorkerPool.IsEmpty() {
 		return false
 	}
-	return q.byteFIFO.Len() == 0
+	return q.byteFIFO.Len(q.terminateCtx) == 0
 }
 
 // Run runs the bytefifo queue
@@ -140,16 +140,20 @@ func (q *ByteFIFOQueue) readToChan() {
 			return
 		default:
 			q.lock.Lock()
-			bs, err := q.byteFIFO.Pop()
+			bs, err := q.byteFIFO.Pop(q.shutdownCtx)
 			if err != nil {
 				q.lock.Unlock()
+				if err == context.Canceled {
+					q.cancel()
+					return
+				}
 				log.Error("%s: %s Error on Pop: %v", q.typ, q.name, err)
 				time.Sleep(time.Millisecond * 100)
 				continue
 			}
 
 			if len(bs) == 0 {
-				if q.waitOnEmpty && q.byteFIFO.Len() == 0 {
+				if q.waitOnEmpty && q.byteFIFO.Len(q.shutdownCtx) == 0 {
 					q.lock.Unlock()
 					log.Trace("%s: %s Waiting on Empty", q.typ, q.name)
 					select {
@@ -205,10 +209,10 @@ func (q *ByteFIFOQueue) Terminate() {
 		return
 	default:
 	}
-	q.terminateCancel()
 	if log.IsDebug() {
-		log.Debug("%s: %s Closing with %d tasks left in queue", q.typ, q.name, q.byteFIFO.Len())
+		log.Debug("%s: %s Closing with %d tasks left in queue", q.typ, q.name, q.byteFIFO.Len(q.terminateCtx))
 	}
+	q.terminateCancel()
 	if err := q.byteFIFO.Close(); err != nil {
 		log.Error("Error whilst closing internal byte fifo in %s: %s: %v", q.typ, q.name, err)
 	}
@@ -263,5 +267,5 @@ func (q *ByteFIFOUniqueQueue) Has(data Data) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	return q.byteFIFO.(UniqueByteFIFO).Has(bs)
+	return q.byteFIFO.(UniqueByteFIFO).Has(q.terminateCtx, bs)
 }

modules/queue/queue_disk.go

@@ -5,6 +5,8 @@
 package queue
 
 import (
+	"context"
+
 	"code.gitea.io/gitea/modules/nosql"
 
 	"gitea.com/lunny/levelqueue"
@@ -83,7 +85,7 @@ func NewLevelQueueByteFIFO(connection, prefix string) (*LevelQueueByteFIFO, erro
 }
 
 // PushFunc will push data into the fifo
-func (fifo *LevelQueueByteFIFO) PushFunc(data []byte, fn func() error) error {
+func (fifo *LevelQueueByteFIFO) PushFunc(ctx context.Context, data []byte, fn func() error) error {
 	if fn != nil {
 		if err := fn(); err != nil {
 			return err
@@ -93,7 +95,7 @@ func (fifo *LevelQueueByteFIFO) PushFunc(data []byte, fn func() error) error {
 }
 
 // Pop pops data from the start of the fifo
-func (fifo *LevelQueueByteFIFO) Pop() ([]byte, error) {
+func (fifo *LevelQueueByteFIFO) Pop(ctx context.Context) ([]byte, error) {
 	data, err := fifo.internal.RPop()
 	if err != nil && err != levelqueue.ErrNotFound {
 		return nil, err
@@ -109,7 +111,7 @@ func (fifo *LevelQueueByteFIFO) Close() error {
 }
 
 // Len returns the length of the fifo
-func (fifo *LevelQueueByteFIFO) Len() int64 {
+func (fifo *LevelQueueByteFIFO) Len(ctx context.Context) int64 {
 	return fifo.internal.Len()
 }
 

modules/queue/queue_disk_channel.go

@@ -150,7 +150,7 @@ func (q *PersistableChannelQueue) Run(atShutdown, atTerminate func(func())) {
 	atShutdown(q.Shutdown)
 	atTerminate(q.Terminate)
 
-	if lq, ok := q.internal.(*LevelQueue); ok && lq.byteFIFO.Len() != 0 {
+	if lq, ok := q.internal.(*LevelQueue); ok && lq.byteFIFO.Len(lq.shutdownCtx) != 0 {
 		// Just run the level queue - we shut it down once it's flushed
 		go q.internal.Run(func(_ func()) {}, func(_ func()) {})
 		go func() {

modules/queue/queue_redis.go

@@ -6,7 +6,6 @@ package queue
 
 import (
 	"context"
-	"fmt"
 
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
@@ -47,8 +46,6 @@ 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,
 	}
@@ -73,8 +70,8 @@ var _ ByteFIFO = &RedisByteFIFO{}
 
 // RedisByteFIFO represents a ByteFIFO formed from a redisClient
 type RedisByteFIFO struct {
-	ctx       context.Context
-	client    redisClient
+	client redisClient
+
 	queueName string
 }
 
@@ -89,7 +86,6 @@ 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(graceful.GetManager().ShutdownContext()).Err(); err != nil {
 		return nil, err
@@ -98,18 +94,18 @@ func NewRedisByteFIFO(config RedisByteFIFOConfiguration) (*RedisByteFIFO, error)
 }
 
 // PushFunc pushes data to the end of the fifo and calls the callback if it is added
-func (fifo *RedisByteFIFO) PushFunc(data []byte, fn func() error) error {
+func (fifo *RedisByteFIFO) PushFunc(ctx context.Context, data []byte, fn func() error) error {
 	if fn != nil {
 		if err := fn(); err != nil {
 			return err
 		}
 	}
-	return fifo.client.RPush(fifo.ctx, fifo.queueName, data).Err()
+	return fifo.client.RPush(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.ctx, fifo.queueName).Bytes()
+func (fifo *RedisByteFIFO) Pop(ctx context.Context) ([]byte, error) {
+	data, err := fifo.client.LPop(ctx, fifo.queueName).Bytes()
 	if err == nil || err == redis.Nil {
 		return data, nil
 	}
@@ -122,8 +118,8 @@ func (fifo *RedisByteFIFO) Close() error {
 }
 
 // Len returns the length of the fifo
-func (fifo *RedisByteFIFO) Len() int64 {
-	val, err := fifo.client.LLen(fifo.ctx, fifo.queueName).Result()
+func (fifo *RedisByteFIFO) Len(ctx context.Context) int64 {
+	val, err := fifo.client.LLen(ctx, fifo.queueName).Result()
 	if err != nil {
 		log.Error("Error whilst getting length of redis queue %s: Error: %v", fifo.queueName, err)
 		return -1

modules/queue/unique_queue_disk.go

@@ -5,6 +5,8 @@
 package queue
 
 import (
+	"context"
+
 	"code.gitea.io/gitea/modules/nosql"
 
 	"gitea.com/lunny/levelqueue"
@@ -87,12 +89,12 @@ func NewLevelUniqueQueueByteFIFO(connection, prefix string) (*LevelUniqueQueueBy
 }
 
 // PushFunc pushes data to the end of the fifo and calls the callback if it is added
-func (fifo *LevelUniqueQueueByteFIFO) PushFunc(data []byte, fn func() error) error {
+func (fifo *LevelUniqueQueueByteFIFO) PushFunc(ctx context.Context, data []byte, fn func() error) error {
 	return fifo.internal.LPushFunc(data, fn)
 }
 
 // Pop pops data from the start of the fifo
-func (fifo *LevelUniqueQueueByteFIFO) Pop() ([]byte, error) {
+func (fifo *LevelUniqueQueueByteFIFO) Pop(ctx context.Context) ([]byte, error) {
 	data, err := fifo.internal.RPop()
 	if err != nil && err != levelqueue.ErrNotFound {
 		return nil, err
@@ -101,12 +103,12 @@ func (fifo *LevelUniqueQueueByteFIFO) Pop() ([]byte, error) {
 }
 
 // Len returns the length of the fifo
-func (fifo *LevelUniqueQueueByteFIFO) Len() int64 {
+func (fifo *LevelUniqueQueueByteFIFO) Len(ctx context.Context) int64 {
 	return fifo.internal.Len()
 }
 
 // Has returns whether the fifo contains this data
-func (fifo *LevelUniqueQueueByteFIFO) Has(data []byte) (bool, error) {
+func (fifo *LevelUniqueQueueByteFIFO) Has(ctx context.Context, data []byte) (bool, error) {
 	return fifo.internal.Has(data)
 }
 

modules/queue/unique_queue_disk_channel.go

@@ -182,7 +182,7 @@ func (q *PersistableChannelUniqueQueue) Run(atShutdown, atTerminate func(func())
 	atShutdown(q.Shutdown)
 	atTerminate(q.Terminate)
 
-	if luq, ok := q.internal.(*LevelUniqueQueue); ok && luq.ByteFIFOUniqueQueue.byteFIFO.Len() != 0 {
+	if luq, ok := q.internal.(*LevelUniqueQueue); ok && luq.ByteFIFOUniqueQueue.byteFIFO.Len(luq.shutdownCtx) != 0 {
 		// Just run the level queue - we shut it down once it's flushed
 		go q.internal.Run(func(_ func()) {}, func(_ func()) {})
 		go func() {

modules/queue/unique_queue_redis.go

@@ -5,9 +5,8 @@
 package queue
 
 import (
-	"fmt"
+	"context"
 
-	"code.gitea.io/gitea/modules/graceful"
 	"github.com/go-redis/redis/v8"
 )
 
@@ -51,8 +50,6 @@ 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,
 	}
@@ -92,8 +89,8 @@ 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.ctx, fifo.setName, data).Result()
+func (fifo *RedisUniqueByteFIFO) PushFunc(ctx context.Context, data []byte, fn func() error) error {
+	added, err := fifo.client.SAdd(ctx, fifo.setName, data).Result()
 	if err != nil {
 		return err
 	}
@@ -105,12 +102,12 @@ func (fifo *RedisUniqueByteFIFO) PushFunc(data []byte, fn func() error) error {
 			return err
 		}
 	}
-	return fifo.client.RPush(fifo.ctx, fifo.queueName, data).Err()
+	return fifo.client.RPush(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.ctx, fifo.queueName).Bytes()
+func (fifo *RedisUniqueByteFIFO) Pop(ctx context.Context) ([]byte, error) {
+	data, err := fifo.client.LPop(ctx, fifo.queueName).Bytes()
 	if err != nil && err != redis.Nil {
 		return data, err
 	}
@@ -119,13 +116,13 @@ func (fifo *RedisUniqueByteFIFO) Pop() ([]byte, error) {
 		return data, nil
 	}
 
-	err = fifo.client.SRem(fifo.ctx, fifo.setName, data).Err()
+	err = fifo.client.SRem(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.ctx, fifo.setName, data).Result()
+func (fifo *RedisUniqueByteFIFO) Has(ctx context.Context, data []byte) (bool, error) {
+	return fifo.client.SIsMember(ctx, fifo.setName, data).Result()
 }
 
 func init() {