Queue: Add monitoring

modules/queue/manager.go

@@ -0,0 +1,211 @@
+// Copyright 2019 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package queue
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"sort"
+	"sync"
+	"time"
+)
+
+var manager *Manager
+
+// Manager is a queue manager
+type Manager struct {
+	mutex sync.Mutex
+
+	counter int64
+	Queues  map[int64]*Description
+}
+
+// Description represents a working queue inheriting from Gitea.
+type Description struct {
+	mutex           sync.Mutex
+	QID             int64
+	Queue           Queue
+	Type            Type
+	Name            string
+	Configuration   interface{}
+	ExemplarType    string
+	addWorkers      func(number int, timeout time.Duration) context.CancelFunc
+	numberOfWorkers func() int
+	counter         int64
+	PoolWorkers     map[int64]*PoolWorkers
+}
+
+// DescriptionList implements the sort.Interface
+type DescriptionList []*Description
+
+// PoolWorkers represents a working queue inheriting from Gitea.
+type PoolWorkers struct {
+	PID        int64
+	Workers    int
+	Start      time.Time
+	Timeout    time.Time
+	HasTimeout bool
+	Cancel     context.CancelFunc
+}
+
+// PoolWorkersList implements the sort.Interface
+type PoolWorkersList []*PoolWorkers
+
+func init() {
+	_ = GetManager()
+}
+
+// GetManager returns a Manager and initializes one as singleton if there's none yet
+func GetManager() *Manager {
+	if manager == nil {
+		manager = &Manager{
+			Queues: make(map[int64]*Description),
+		}
+	}
+	return manager
+}
+
+// Add adds a queue to this manager
+func (m *Manager) Add(queue Queue,
+	t Type,
+	configuration,
+	exemplar interface{},
+	addWorkers func(number int, timeout time.Duration) context.CancelFunc,
+	numberOfWorkers func() int) int64 {
+
+	cfg, _ := json.Marshal(configuration)
+	desc := &Description{
+		Queue:           queue,
+		Type:            t,
+		Configuration:   string(cfg),
+		ExemplarType:    reflect.TypeOf(exemplar).String(),
+		PoolWorkers:     make(map[int64]*PoolWorkers),
+		addWorkers:      addWorkers,
+		numberOfWorkers: numberOfWorkers,
+	}
+	m.mutex.Lock()
+	m.counter++
+	desc.QID = m.counter
+	desc.Name = fmt.Sprintf("queue-%d", desc.QID)
+	if named, ok := queue.(Named); ok {
+		desc.Name = named.Name()
+	}
+	m.Queues[desc.QID] = desc
+	m.mutex.Unlock()
+	return desc.QID
+}
+
+// Remove a queue from the Manager
+func (m *Manager) Remove(qid int64) {
+	m.mutex.Lock()
+	delete(m.Queues, qid)
+	m.mutex.Unlock()
+}
+
+// GetDescription by qid
+func (m *Manager) GetDescription(qid int64) *Description {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	return m.Queues[qid]
+}
+
+// Descriptions returns the queue descriptions
+func (m *Manager) Descriptions() []*Description {
+	m.mutex.Lock()
+	descs := make([]*Description, 0, len(m.Queues))
+	for _, desc := range m.Queues {
+		descs = append(descs, desc)
+	}
+	m.mutex.Unlock()
+	sort.Sort(DescriptionList(descs))
+	return descs
+}
+
+// Workers returns the poolworkers
+func (q *Description) Workers() []*PoolWorkers {
+	q.mutex.Lock()
+	workers := make([]*PoolWorkers, 0, len(q.PoolWorkers))
+	for _, worker := range q.PoolWorkers {
+		workers = append(workers, worker)
+	}
+	q.mutex.Unlock()
+	sort.Sort(PoolWorkersList(workers))
+	return workers
+}
+
+// RegisterWorkers registers workers to this queue
+func (q *Description) RegisterWorkers(number int, start time.Time, hasTimeout bool, timeout time.Time, cancel context.CancelFunc) int64 {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	q.counter++
+	q.PoolWorkers[q.counter] = &PoolWorkers{
+		PID:        q.counter,
+		Workers:    number,
+		Start:      start,
+		Timeout:    timeout,
+		HasTimeout: hasTimeout,
+		Cancel:     cancel,
+	}
+	return q.counter
+}
+
+// CancelWorkers cancels pooled workers with pid
+func (q *Description) CancelWorkers(pid int64) {
+	q.mutex.Lock()
+	pw, ok := q.PoolWorkers[pid]
+	q.mutex.Unlock()
+	if !ok {
+		return
+	}
+	pw.Cancel()
+}
+
+// RemoveWorkers deletes pooled workers with pid
+func (q *Description) RemoveWorkers(pid int64) {
+	q.mutex.Lock()
+	delete(q.PoolWorkers, pid)
+	q.mutex.Unlock()
+}
+
+// AddWorkers adds workers to the queue if it has registered an add worker function
+func (q *Description) AddWorkers(number int, timeout time.Duration) {
+	if q.addWorkers != nil {
+		_ = q.addWorkers(number, timeout)
+	}
+}
+
+// NumberOfWorkers returns the number of workers in the queue
+func (q *Description) NumberOfWorkers() int {
+	if q.numberOfWorkers != nil {
+		return q.numberOfWorkers()
+	}
+	return -1
+}
+
+func (l DescriptionList) Len() int {
+	return len(l)
+}
+
+func (l DescriptionList) Less(i, j int) bool {
+	return l[i].Name < l[j].Name
+}
+
+func (l DescriptionList) Swap(i, j int) {
+	l[i], l[j] = l[j], l[i]
+}
+
+func (l PoolWorkersList) Len() int {
+	return len(l)
+}
+
+func (l PoolWorkersList) Less(i, j int) bool {
+	return l[i].Start.Before(l[j].Start)
+}
+
+func (l PoolWorkersList) Swap(i, j int) {
+	l[i], l[j] = l[j], l[i]
+}

modules/queue/queue.go

@@ -48,6 +48,11 @@ type Shutdownable interface {
 	Terminate()
 }
 
+// Named represents a queue with a name
+type Named interface {
+	Name() string
+}
+
 // Queue defines an interface to save an issue indexer queue
 type Queue interface {
 	Run(atShutdown, atTerminate func(context.Context, func()))

modules/queue/queue_channel.go

@@ -48,7 +48,7 @@ func NewChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, erro
 	dataChan := make(chan Data, config.QueueLength)
 
 	ctx, cancel := context.WithCancel(context.Background())
-	return &ChannelQueue{
+	queue := &ChannelQueue{
 		pool: &WorkerPool{
 			baseCtx:      ctx,
 			cancel:       cancel,
@@ -62,7 +62,9 @@ func NewChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, erro
 		exemplar: exemplar,
 		workers:  config.Workers,
 		name:     config.Name,
-	}, nil
+	}
+	queue.pool.qid = GetManager().Add(queue, ChannelQueueType, config, exemplar, queue.pool.AddWorkers, queue.pool.NumberOfWorkers)
+	return queue, nil
 }
 
 // Run starts to run the queue
@@ -73,7 +75,9 @@ func (c *ChannelQueue) Run(atShutdown, atTerminate func(context.Context, func())
 	atTerminate(context.Background(), func() {
 		log.Warn("ChannelQueue: %s is not terminatable!", c.name)
 	})
-	c.pool.addWorkers(c.pool.baseCtx, c.workers)
+	go func() {
+		_ = c.pool.AddWorkers(c.workers, 0)
+	}()
 }
 
 // Push will push the indexer data to queue
@@ -90,6 +94,11 @@ func (c *ChannelQueue) Push(data Data) error {
 	return nil
 }
 
+// Name returns the name of this queue
+func (c *ChannelQueue) Name() string {
+	return c.name
+}
+
 func init() {
 	queuesMap[ChannelQueueType] = NewChannelQueue
 }

modules/queue/queue_disk.go

@@ -50,7 +50,7 @@ func NewLevelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
 	}
 	config := configInterface.(LevelQueueConfiguration)
 
-	queue, err := levelqueue.Open(config.DataDir)
+	internal, err := levelqueue.Open(config.DataDir)
 	if err != nil {
 		return nil, err
 	}
@@ -58,7 +58,7 @@ func NewLevelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
 	dataChan := make(chan Data, config.QueueLength)
 	ctx, cancel := context.WithCancel(context.Background())
 
-	return &LevelQueue{
+	queue := &LevelQueue{
 		pool: &WorkerPool{
 			baseCtx:      ctx,
 			cancel:       cancel,
@@ -69,13 +69,15 @@ func NewLevelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
 			boostTimeout: config.BoostTimeout,
 			boostWorkers: config.BoostWorkers,
 		},
-		queue:      queue,
+		queue:      internal,
 		exemplar:   exemplar,
 		closed:     make(chan struct{}),
 		terminated: make(chan struct{}),
 		workers:    config.Workers,
 		name:       config.Name,
-	}, nil
+	}
+	queue.pool.qid = GetManager().Add(queue, LevelQueueType, config, exemplar, queue.pool.AddWorkers, queue.pool.NumberOfWorkers)
+	return queue, nil
 }
 
 // Run starts to run the queue
@@ -83,7 +85,9 @@ func (l *LevelQueue) Run(atShutdown, atTerminate func(context.Context, func()))
 	atShutdown(context.Background(), l.Shutdown)
 	atTerminate(context.Background(), l.Terminate)
 
-	go l.pool.addWorkers(l.pool.baseCtx, l.workers)
+	go func() {
+		_ = l.pool.AddWorkers(l.workers, 0)
+	}()
 
 	go l.readToChan()
 
@@ -140,7 +144,7 @@ func (l *LevelQueue) readToChan() {
 
 			log.Trace("LevelQueue %s: task found: %#v", l.name, data)
 			l.pool.Push(data)
-			time.Sleep(time.Millisecond * 10)
+			time.Sleep(time.Millisecond * 100)
 
 		}
 	}
@@ -183,6 +187,11 @@ func (l *LevelQueue) Terminate() {
 	}
 }
 
+// Name returns the name of this queue
+func (l *LevelQueue) Name() string {
+	return l.name
+}
+
 func init() {
 	queuesMap[LevelQueueType] = NewLevelQueue
 }

modules/queue/queue_disk_channel.go

@@ -85,7 +85,7 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
 		return nil, ErrInvalidConfiguration{cfg: cfg}
 	}
 
-	return &PersistableChannelQueue{
+	queue := &PersistableChannelQueue{
 		ChannelQueue: channelQueue.(*ChannelQueue),
 		delayedStarter: delayedStarter{
 			cfg:         levelCfg,
@@ -95,7 +95,9 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
 			name:        config.Name,
 		},
 		closed: make(chan struct{}),
-	}, nil
+	}
+	_ = GetManager().Add(queue, PersistableChannelQueueType, config, exemplar, nil, nil)
+	return queue, nil
 }
 
 // Name returns the name of this queue
@@ -127,7 +129,9 @@ func (p *PersistableChannelQueue) Run(atShutdown, atTerminate func(context.Conte
 	// Just run the level queue - we shut it down later
 	go p.internal.Run(func(_ context.Context, _ func()) {}, func(_ context.Context, _ func()) {})
 
-	go p.ChannelQueue.pool.addWorkers(p.ChannelQueue.pool.baseCtx, p.workers)
+	go func() {
+		_ = p.ChannelQueue.pool.AddWorkers(p.workers, 0)
+	}()
 
 	<-p.closed
 	p.ChannelQueue.pool.cancel()

modules/queue/queue_redis.go

@@ -67,7 +67,7 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
 	dataChan := make(chan Data, config.QueueLength)
 	ctx, cancel := context.WithCancel(context.Background())
 
-	var queue = RedisQueue{
+	var queue = &RedisQueue{
 		pool: &WorkerPool{
 			baseCtx:      ctx,
 			cancel:       cancel,
@@ -100,7 +100,9 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
 	if err := queue.client.Ping().Err(); err != nil {
 		return nil, err
 	}
-	return &queue, nil
+	queue.pool.qid = GetManager().Add(queue, RedisQueueType, config, exemplar, queue.pool.AddWorkers, queue.pool.NumberOfWorkers)
+
+	return queue, nil
 }
 
 // Run runs the redis queue
@@ -108,7 +110,9 @@ func (r *RedisQueue) Run(atShutdown, atTerminate func(context.Context, func()))
 	atShutdown(context.Background(), r.Shutdown)
 	atTerminate(context.Background(), r.Terminate)
 
-	go r.pool.addWorkers(r.pool.baseCtx, r.workers)
+	go func() {
+		_ = r.pool.AddWorkers(r.workers, 0)
+	}()
 
 	go r.readToChan()
 
@@ -198,6 +202,11 @@ func (r *RedisQueue) Terminate() {
 	}
 }
 
+// Name returns the name of this queue
+func (r *RedisQueue) Name() string {
+	return r.name
+}
+
 func init() {
 	queuesMap[RedisQueueType] = NewRedisQueue
 }

modules/queue/queue_wrapped.go

@@ -111,7 +111,7 @@ func NewWrappedQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, erro
 		return nil, ErrInvalidConfiguration{cfg: cfg}
 	}
 
-	return &WrappedQueue{
+	queue = &WrappedQueue{
 		handle:   handle,
 		channel:  make(chan Data, config.QueueLength),
 		exemplar: exemplar,
@@ -122,7 +122,14 @@ func NewWrappedQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, erro
 			maxAttempts: config.MaxAttempts,
 			name:        config.Name,
 		},
-	}, nil
+	}
+	_ = GetManager().Add(queue, WrappedQueueType, config, exemplar, nil, nil)
+	return queue, nil
+}
+
+// Name returns the name of the queue
+func (q *WrappedQueue) Name() string {
+	return q.name + "-wrapper"
 }
 
 // Push will push the data to the internal channel checking it against the exemplar

modules/queue/workerpool.go

@@ -18,6 +18,7 @@ type WorkerPool struct {
 	baseCtx         context.Context
 	cancel          context.CancelFunc
 	cond            *sync.Cond
+	qid             int64
 	numberOfWorkers int
 	batchLength     int
 	handle          HandlerFunc
@@ -68,8 +69,21 @@ func (p *WorkerPool) pushBoost(data Data) {
 				return
 			}
 			p.blockTimeout *= 2
-			log.Warn("Worker Channel blocked for %v - adding %d temporary workers for %s, block timeout now %v", ourTimeout, p.boostWorkers, p.boostTimeout, p.blockTimeout)
 			ctx, cancel := context.WithCancel(p.baseCtx)
+			desc := GetManager().GetDescription(p.qid)
+			if desc != nil {
+				log.Warn("Worker Channel for %v blocked for %v - adding %d temporary workers for %s, block timeout now %v", desc.Name, ourTimeout, p.boostWorkers, p.boostTimeout, p.blockTimeout)
+
+				start := time.Now()
+				pid := desc.RegisterWorkers(p.boostWorkers, start, false, start, cancel)
+				go func() {
+					<-ctx.Done()
+					desc.RemoveWorkers(pid)
+					cancel()
+				}()
+			} else {
+				log.Warn("Worker Channel blocked for %v - adding %d temporary workers for %s, block timeout now %v", ourTimeout, p.boostWorkers, p.boostTimeout, p.blockTimeout)
+			}
 			go func() {
 				<-time.After(p.boostTimeout)
 				cancel()
@@ -95,12 +109,26 @@ func (p *WorkerPool) NumberOfWorkers() int {
 func (p *WorkerPool) AddWorkers(number int, timeout time.Duration) context.CancelFunc {
 	var ctx context.Context
 	var cancel context.CancelFunc
+	start := time.Now()
+	end := start
+	hasTimeout := false
 	if timeout > 0 {
 		ctx, cancel = context.WithTimeout(p.baseCtx, timeout)
+		end = start.Add(timeout)
+		hasTimeout = true
 	} else {
 		ctx, cancel = context.WithCancel(p.baseCtx)
 	}
 
+	desc := GetManager().GetDescription(p.qid)
+	if desc != nil {
+		pid := desc.RegisterWorkers(number, start, hasTimeout, end, cancel)
+		go func() {
+			<-ctx.Done()
+			desc.RemoveWorkers(pid)
+			cancel()
+		}()
+	}
 	p.addWorkers(ctx, number)
 	return cancel
 }