domi/forgejo-domi
1
0
Fork 0
forked from forgejo/forgejo
Code Pull requests Activity

Add MaxWorker settings to queues

Browse source
This commit is contained in:
Andrew Thornton 2019-12-30 15:54:19 +00:00
parent a492b3071c
commit b1c9fa7f1a
No known key found for this signature in database
GPG key ID: 3CDE74631F13A748
15 changed files with 312 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

100
modules/queue/manager.go
View file

@ -28,17 +28,28 @@ type Manager struct {
// 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
mutex sync.Mutex
QID int64
Queue Queue
Type Type
Name string
Configuration interface{}
ExemplarType string
Pool PoolManager
counter int64
PoolWorkers map[int64]*PoolWorkers
}
// PoolManager is a simple interface to get certain details from a worker pool
type PoolManager interface {
AddWorkers(number int, timeout time.Duration) context.CancelFunc
NumberOfWorkers() int
MaxNumberOfWorkers() int
SetMaxNumberOfWorkers(int)
BoostTimeout() time.Duration
BlockTimeout() time.Duration
BoostWorkers() int
SetSettings(maxNumberOfWorkers, boostWorkers int, timeout time.Duration)
}
// DescriptionList implements the sort.Interface
@ -76,18 +87,16 @@ func (m *Manager) Add(queue Queue,
t Type,
configuration,
exemplar interface{},
addWorkers func(number int, timeout time.Duration) context.CancelFunc,
numberOfWorkers func() int) int64 {
pool PoolManager) 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,
Queue: queue,
Type: t,
Configuration: string(cfg),
ExemplarType: reflect.TypeOf(exemplar).String(),
PoolWorkers: make(map[int64]*PoolWorkers),
Pool: pool,
}
m.mutex.Lock()
m.counter++
@ -177,20 +186,61 @@ func (q *Description) RemoveWorkers(pid int64) {
}
// 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)
func (q *Description) AddWorkers(number int, timeout time.Duration) context.CancelFunc {
if q.Pool != nil {
// the cancel will be added to the pool workers description above
return q.Pool.AddWorkers(number, timeout)
}
return nil
}
// NumberOfWorkers returns the number of workers in the queue
func (q *Description) NumberOfWorkers() int {
if q.numberOfWorkers != nil {
return q.numberOfWorkers()
if q.Pool != nil {
return q.Pool.NumberOfWorkers()
}
return -1
}
// MaxNumberOfWorkers returns the maximum number of workers for the pool
func (q *Description) MaxNumberOfWorkers() int {
if q.Pool != nil {
return q.Pool.MaxNumberOfWorkers()
}
return 0
}
// BoostWorkers returns the number of workers for a boost
func (q *Description) BoostWorkers() int {
if q.Pool != nil {
return q.Pool.BoostWorkers()
}
return -1
}
// BoostTimeout returns the timeout of the next boost
func (q *Description) BoostTimeout() time.Duration {
if q.Pool != nil {
return q.Pool.BoostTimeout()
}
return 0
}
// BlockTimeout returns the timeout til the next boost
func (q *Description) BlockTimeout() time.Duration {
if q.Pool != nil {
return q.Pool.BlockTimeout()
}
return 0
}
// SetSettings sets the setable boost values
func (q *Description) SetSettings(maxNumberOfWorkers, boostWorkers int, timeout time.Duration) {
if q.Pool != nil {
q.Pool.SetSettings(maxNumberOfWorkers, boostWorkers, timeout)
}
}
func (l DescriptionList) Len() int {
return len(l)
}

20
modules/queue/queue_channel.go
View file

@ -21,6 +21,7 @@ type ChannelQueueConfiguration struct {
QueueLength int
BatchLength int
Workers int
MaxWorkers int
BlockTimeout time.Duration
BoostTimeout time.Duration
BoostWorkers int
@ -50,20 +51,21 @@ func NewChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, erro
ctx, cancel := context.WithCancel(context.Background())
queue := &ChannelQueue{
pool: &WorkerPool{
baseCtx: ctx,
cancel: cancel,
batchLength: config.BatchLength,
handle: handle,
dataChan: dataChan,
blockTimeout: config.BlockTimeout,
boostTimeout: config.BoostTimeout,
boostWorkers: config.BoostWorkers,
baseCtx: ctx,
cancel: cancel,
batchLength: config.BatchLength,
handle: handle,
dataChan: dataChan,
blockTimeout: config.BlockTimeout,
boostTimeout: config.BoostTimeout,
boostWorkers: config.BoostWorkers,
maxNumberOfWorkers: config.MaxWorkers,
},
exemplar: exemplar,
workers: config.Workers,
name: config.Name,
}
queue.pool.qid = GetManager().Add(queue, ChannelQueueType, config, exemplar, queue.pool.AddWorkers, queue.pool.NumberOfWorkers)
queue.pool.qid = GetManager().Add(queue, ChannelQueueType, config, exemplar, queue.pool)
return queue, nil
}

2
modules/queue/queue_channel_test.go
View file

@ -27,6 +27,7 @@ func TestChannelQueue(t *testing.T) {
ChannelQueueConfiguration{
QueueLength: 20,
Workers: 1,
MaxWorkers: 10,
BlockTimeout: 1 * time.Second,
BoostTimeout: 5 * time.Minute,
BoostWorkers: 5,
@ -62,6 +63,7 @@ func TestChannelQueue_Batch(t *testing.T) {
QueueLength: 20,
BatchLength: 2,
Workers: 1,
MaxWorkers: 10,
BlockTimeout: 1 * time.Second,
BoostTimeout: 5 * time.Minute,
BoostWorkers: 5,

20
modules/queue/queue_disk.go
View file

@ -25,6 +25,7 @@ type LevelQueueConfiguration struct {
QueueLength int
BatchLength int
Workers int
MaxWorkers int
BlockTimeout time.Duration
BoostTimeout time.Duration
BoostWorkers int
@ -60,14 +61,15 @@ func NewLevelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
queue := &LevelQueue{
pool: &WorkerPool{
baseCtx: ctx,
cancel: cancel,
batchLength: config.BatchLength,
handle: handle,
dataChan: dataChan,
blockTimeout: config.BlockTimeout,
boostTimeout: config.BoostTimeout,
boostWorkers: config.BoostWorkers,
baseCtx: ctx,
cancel: cancel,
batchLength: config.BatchLength,
handle: handle,
dataChan: dataChan,
blockTimeout: config.BlockTimeout,
boostTimeout: config.BoostTimeout,
boostWorkers: config.BoostWorkers,
maxNumberOfWorkers: config.MaxWorkers,
},
queue: internal,
exemplar: exemplar,
@ -76,7 +78,7 @@ func NewLevelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
workers: config.Workers,
name: config.Name,
}
queue.pool.qid = GetManager().Add(queue, LevelQueueType, config, exemplar, queue.pool.AddWorkers, queue.pool.NumberOfWorkers)
queue.pool.qid = GetManager().Add(queue, LevelQueueType, config, exemplar, queue.pool)
return queue, nil
}

5
modules/queue/queue_disk_channel.go
View file

@ -23,6 +23,7 @@ type PersistableChannelQueueConfiguration struct {
Timeout time.Duration
MaxAttempts int
Workers int
MaxWorkers int
BlockTimeout time.Duration
BoostTimeout time.Duration
BoostWorkers int
@ -48,6 +49,7 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
QueueLength: config.QueueLength,
BatchLength: config.BatchLength,
Workers: config.Workers,
MaxWorkers: config.MaxWorkers,
BlockTimeout: config.BlockTimeout,
BoostTimeout: config.BoostTimeout,
BoostWorkers: config.BoostWorkers,
@ -63,6 +65,7 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
QueueLength: config.QueueLength,
BatchLength: config.BatchLength,
Workers: 1,
MaxWorkers: 6,
BlockTimeout: 1 * time.Second,
BoostTimeout: 5 * time.Minute,
BoostWorkers: 5,
@ -96,7 +99,7 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
},
closed: make(chan struct{}),
}
_ = GetManager().Add(queue, PersistableChannelQueueType, config, exemplar, nil, nil)
_ = GetManager().Add(queue, PersistableChannelQueueType, config, exemplar, nil)
return queue, nil
}

2
modules/queue/queue_disk_channel_test.go
View file

@ -36,6 +36,7 @@ func TestPersistableChannelQueue(t *testing.T) {
BatchLength: 2,
QueueLength: 20,
Workers: 1,
MaxWorkers: 10,
}, &testData{})
assert.NoError(t, err)
@ -89,6 +90,7 @@ func TestPersistableChannelQueue(t *testing.T) {
BatchLength: 2,
QueueLength: 20,
Workers: 1,
MaxWorkers: 10,
}, &testData{})
assert.NoError(t, err)

2
modules/queue/queue_disk_test.go
View file

@ -35,6 +35,7 @@ func TestLevelQueue(t *testing.T) {
DataDir: tmpDir,
BatchLength: 2,
Workers: 1,
MaxWorkers: 10,
QueueLength: 20,
BlockTimeout: 1 * time.Second,
BoostTimeout: 5 * time.Minute,
@ -94,6 +95,7 @@ func TestLevelQueue(t *testing.T) {
DataDir: tmpDir,
BatchLength: 2,
Workers: 1,
MaxWorkers: 10,
QueueLength: 20,
BlockTimeout: 1 * time.Second,
BoostTimeout: 5 * time.Minute,

20
modules/queue/queue_redis.go
View file

@ -49,6 +49,7 @@ type RedisQueueConfiguration struct {
QueueLength int
QueueName string
Workers int
MaxWorkers int
BlockTimeout time.Duration
BoostTimeout time.Duration
BoostWorkers int
@ -70,14 +71,15 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
var queue = &RedisQueue{
pool: &WorkerPool{
baseCtx: ctx,
cancel: cancel,
batchLength: config.BatchLength,
handle: handle,
dataChan: dataChan,
blockTimeout: config.BlockTimeout,
boostTimeout: config.BoostTimeout,
boostWorkers: config.BoostWorkers,
baseCtx: ctx,
cancel: cancel,
batchLength: config.BatchLength,
handle: handle,
dataChan: dataChan,
blockTimeout: config.BlockTimeout,
boostTimeout: config.BoostTimeout,
boostWorkers: config.BoostWorkers,
maxNumberOfWorkers: config.MaxWorkers,
},
queueName: config.QueueName,
exemplar: exemplar,
@ -102,7 +104,7 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
if err := queue.client.Ping().Err(); err != nil {
return nil, err
}
queue.pool.qid = GetManager().Add(queue, RedisQueueType, config, exemplar, queue.pool.AddWorkers, queue.pool.NumberOfWorkers)
queue.pool.qid = GetManager().Add(queue, RedisQueueType, config, exemplar, queue.pool)
return queue, nil
}

2
modules/queue/queue_wrapped.go
View file

@ -123,7 +123,7 @@ func NewWrappedQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, erro
name: config.Name,
},
}
_ = GetManager().Add(queue, WrappedQueueType, config, exemplar, nil, nil)
_ = GetManager().Add(queue, WrappedQueueType, config, exemplar, nil)
return queue, nil
}

87
modules/queue/workerpool.go
View file

@ -14,24 +14,25 @@ import (
// WorkerPool takes
type WorkerPool struct {
lock sync.Mutex
baseCtx context.Context
cancel context.CancelFunc
cond *sync.Cond
qid int64
numberOfWorkers int
batchLength int
handle HandlerFunc
dataChan chan Data
blockTimeout time.Duration
boostTimeout time.Duration
boostWorkers int
lock sync.Mutex
baseCtx context.Context
cancel context.CancelFunc
cond *sync.Cond
qid int64
maxNumberOfWorkers int
numberOfWorkers int
batchLength int
handle HandlerFunc
dataChan chan Data
blockTimeout time.Duration
boostTimeout time.Duration
boostWorkers int
}
// Push pushes the data to the internal channel
func (p *WorkerPool) Push(data Data) {
p.lock.Lock()
if p.blockTimeout > 0 && p.boostTimeout > 0 {
if p.blockTimeout > 0 && p.boostTimeout > 0 && (p.numberOfWorkers <= p.maxNumberOfWorkers || p.maxNumberOfWorkers < 0) {
p.lock.Unlock()
p.pushBoost(data)
} else {
@ -63,7 +64,7 @@ func (p *WorkerPool) pushBoost(data Data) {
}
case <-timer.C:
p.lock.Lock()
if p.blockTimeout > ourTimeout {
if p.blockTimeout > ourTimeout || (p.numberOfWorkers > p.maxNumberOfWorkers && p.maxNumberOfWorkers >= 0) {
p.lock.Unlock()
p.dataChan <- data
return
@ -71,11 +72,15 @@ func (p *WorkerPool) pushBoost(data Data) {
p.blockTimeout *= 2
ctx, cancel := context.WithCancel(p.baseCtx)
desc := GetManager().GetDescription(p.qid)
boost := p.boostWorkers
if (boost+p.numberOfWorkers) > p.maxNumberOfWorkers && p.maxNumberOfWorkers >= 0 {
boost = p.maxNumberOfWorkers - p.numberOfWorkers
}
if desc != nil {
log.Warn("WorkerPool: %d (for %s) Channel blocked for %v - adding %d temporary workers for %s, block timeout now %v", p.qid, desc.Name, ourTimeout, p.boostWorkers, p.boostTimeout, p.blockTimeout)
log.Warn("WorkerPool: %d (for %s) Channel blocked for %v - adding %d temporary workers for %s, block timeout now %v", p.qid, desc.Name, ourTimeout, boost, p.boostTimeout, p.blockTimeout)
start := time.Now()
pid := desc.RegisterWorkers(p.boostWorkers, start, false, start, cancel)
pid := desc.RegisterWorkers(boost, start, false, start, cancel)
go func() {
<-ctx.Done()
desc.RemoveWorkers(pid)
@ -91,7 +96,7 @@ func (p *WorkerPool) pushBoost(data Data) {
p.blockTimeout /= 2
p.lock.Unlock()
}()
p.addWorkers(ctx, p.boostWorkers)
p.addWorkers(ctx, boost)
p.lock.Unlock()
p.dataChan <- data
}
@ -105,7 +110,53 @@ func (p *WorkerPool) NumberOfWorkers() int {
return p.numberOfWorkers
}
// AddWorkers adds workers to the pool
// MaxNumberOfWorkers returns the maximum number of workers automatically added to the pool
func (p *WorkerPool) MaxNumberOfWorkers() int {
p.lock.Lock()
defer p.lock.Unlock()
return p.maxNumberOfWorkers
}
// BoostWorkers returns the number of workers for a boost
func (p *WorkerPool) BoostWorkers() int {
p.lock.Lock()
defer p.lock.Unlock()
return p.boostWorkers
}
// BoostTimeout returns the timeout of the next boost
func (p *WorkerPool) BoostTimeout() time.Duration {
p.lock.Lock()
defer p.lock.Unlock()
return p.boostTimeout
}
// BlockTimeout returns the timeout til the next boost
func (p *WorkerPool) BlockTimeout() time.Duration {
p.lock.Lock()
defer p.lock.Unlock()
return p.blockTimeout
}
// SetSettings sets the setable boost values
func (p *WorkerPool) SetSettings(maxNumberOfWorkers, boostWorkers int, timeout time.Duration) {
p.lock.Lock()
defer p.lock.Unlock()
p.maxNumberOfWorkers = maxNumberOfWorkers
p.boostWorkers = boostWorkers
p.boostTimeout = timeout
}
// SetMaxNumberOfWorkers sets the maximum number of workers automatically added to the pool
// Changing this number will not change the number of current workers but will change the limit
// for future additions
func (p *WorkerPool) SetMaxNumberOfWorkers(newMax int) {
p.lock.Lock()
defer p.lock.Unlock()
p.maxNumberOfWorkers = newMax
}
// AddWorkers adds workers to the pool - this allows the number of workers to go above the limit
func (p *WorkerPool) AddWorkers(number int, timeout time.Duration) context.CancelFunc {
var ctx context.Context
var cancel context.CancelFunc