mirror of
https://github.com/go-gitea/gitea.git
synced 2024-08-28 22:14:20 -04:00
* Upgrade levelqueue to version 0.2.0 This adds functionality for Unique Queues * Add UniqueQueue interface and functions to create them * Add UniqueQueue implementations * Move TestPullRequests over to use UniqueQueue * Reduce code duplication * Add bytefifos * Ensure invalid types are logged * Fix close race in PersistableChannelQueue Shutdown
173 lines
4.7 KiB
Go
173 lines
4.7 KiB
Go
// Copyright 2020 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 (
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// WrappedUniqueQueueType is the type for a wrapped delayed starting queue
|
|
const WrappedUniqueQueueType Type = "unique-wrapped"
|
|
|
|
// WrappedUniqueQueueConfiguration is the configuration for a WrappedUniqueQueue
|
|
type WrappedUniqueQueueConfiguration struct {
|
|
Underlying Type
|
|
Timeout time.Duration
|
|
MaxAttempts int
|
|
Config interface{}
|
|
QueueLength int
|
|
Name string
|
|
}
|
|
|
|
// WrappedUniqueQueue wraps a delayed starting unique queue
|
|
type WrappedUniqueQueue struct {
|
|
*WrappedQueue
|
|
table map[Data]bool
|
|
tlock sync.Mutex
|
|
ready bool
|
|
}
|
|
|
|
// NewWrappedUniqueQueue will attempt to create a unique queue of the provided type,
|
|
// but if there is a problem creating this queue it will instead create
|
|
// a WrappedUniqueQueue with delayed startup of the queue instead and a
|
|
// channel which will be redirected to the queue
|
|
//
|
|
// Please note that this Queue does not guarantee that a particular
|
|
// task cannot be processed twice or more at the same time. Uniqueness is
|
|
// only guaranteed whilst the task is waiting in the queue.
|
|
func NewWrappedUniqueQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error) {
|
|
configInterface, err := toConfig(WrappedUniqueQueueConfiguration{}, cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
config := configInterface.(WrappedUniqueQueueConfiguration)
|
|
|
|
queue, err := NewQueue(config.Underlying, handle, config.Config, exemplar)
|
|
if err == nil {
|
|
// Just return the queue there is no need to wrap
|
|
return queue, nil
|
|
}
|
|
if IsErrInvalidConfiguration(err) {
|
|
// Retrying ain't gonna make this any better...
|
|
return nil, ErrInvalidConfiguration{cfg: cfg}
|
|
}
|
|
|
|
wrapped := &WrappedUniqueQueue{
|
|
WrappedQueue: &WrappedQueue{
|
|
channel: make(chan Data, config.QueueLength),
|
|
exemplar: exemplar,
|
|
delayedStarter: delayedStarter{
|
|
cfg: config.Config,
|
|
underlying: config.Underlying,
|
|
timeout: config.Timeout,
|
|
maxAttempts: config.MaxAttempts,
|
|
name: config.Name,
|
|
},
|
|
},
|
|
table: map[Data]bool{},
|
|
}
|
|
|
|
// wrapped.handle is passed to the delayedStarting internal queue and is run to handle
|
|
// data passed to
|
|
wrapped.handle = func(data ...Data) {
|
|
for _, datum := range data {
|
|
wrapped.tlock.Lock()
|
|
if !wrapped.ready {
|
|
delete(wrapped.table, data)
|
|
// If our table is empty all of the requests we have buffered between the
|
|
// wrapper queue starting and the internal queue starting have been handled.
|
|
// We can stop buffering requests in our local table and just pass Push
|
|
// direct to the internal queue
|
|
if len(wrapped.table) == 0 {
|
|
wrapped.ready = true
|
|
}
|
|
}
|
|
wrapped.tlock.Unlock()
|
|
handle(datum)
|
|
}
|
|
}
|
|
_ = GetManager().Add(queue, WrappedUniqueQueueType, config, exemplar)
|
|
return wrapped, nil
|
|
}
|
|
|
|
// Push will push the data to the internal channel checking it against the exemplar
|
|
func (q *WrappedUniqueQueue) Push(data Data) error {
|
|
return q.PushFunc(data, nil)
|
|
}
|
|
|
|
// PushFunc will push the data to the internal channel checking it against the exemplar
|
|
func (q *WrappedUniqueQueue) PushFunc(data Data, fn func() error) error {
|
|
if !assignableTo(data, q.exemplar) {
|
|
return fmt.Errorf("Unable to assign data: %v to same type as exemplar: %v in %s", data, q.exemplar, q.name)
|
|
}
|
|
|
|
q.tlock.Lock()
|
|
if q.ready {
|
|
// ready means our table is empty and all of the requests we have buffered between the
|
|
// wrapper queue starting and the internal queue starting have been handled.
|
|
// We can stop buffering requests in our local table and just pass Push
|
|
// direct to the internal queue
|
|
q.tlock.Unlock()
|
|
return q.internal.(UniqueQueue).PushFunc(data, fn)
|
|
}
|
|
|
|
locked := true
|
|
defer func() {
|
|
if locked {
|
|
q.tlock.Unlock()
|
|
}
|
|
}()
|
|
if _, ok := q.table[data]; ok {
|
|
return ErrAlreadyInQueue
|
|
}
|
|
// FIXME: We probably need to implement some sort of limit here
|
|
// If the downstream queue blocks this table will grow without limit
|
|
q.table[data] = true
|
|
if fn != nil {
|
|
err := fn()
|
|
if err != nil {
|
|
delete(q.table, data)
|
|
return err
|
|
}
|
|
}
|
|
locked = false
|
|
q.tlock.Unlock()
|
|
|
|
q.channel <- data
|
|
return nil
|
|
}
|
|
|
|
// Has checks if the data is in the queue
|
|
func (q *WrappedUniqueQueue) Has(data Data) (bool, error) {
|
|
q.tlock.Lock()
|
|
defer q.tlock.Unlock()
|
|
if q.ready {
|
|
return q.internal.(UniqueQueue).Has(data)
|
|
}
|
|
_, has := q.table[data]
|
|
return has, nil
|
|
}
|
|
|
|
// IsEmpty checks whether the queue is empty
|
|
func (q *WrappedUniqueQueue) IsEmpty() bool {
|
|
q.tlock.Lock()
|
|
if len(q.table) > 0 {
|
|
q.tlock.Unlock()
|
|
return false
|
|
}
|
|
if q.ready {
|
|
q.tlock.Unlock()
|
|
return q.internal.IsEmpty()
|
|
}
|
|
q.tlock.Unlock()
|
|
return false
|
|
}
|
|
|
|
func init() {
|
|
queuesMap[WrappedUniqueQueueType] = NewWrappedUniqueQueue
|
|
}
|