queue

Documentation

Overview

Package queue implements a specialized queue system for Gitea.

There are two major kinds of concepts:

* The "base queue": channel, level, redis:

• They have the same abstraction, the same interface, and they are tested by the same testing code.
• The dummy(immediate) queue is special, it's not a real queue, it's only used as a no-op queue or a testing queue.

* The WorkerPoolQueue: it uses the "base queue" to provide "worker pool" function.

• It calls the "handler" to process the data in the base queue.
• Its "Push" function doesn't block forever, it will return an error if the queue is full after the timeout.

A queue can be "simple" or "unique". A unique queue will try to avoid duplicate items. Unique queue's "Has" function can be used to check whether an item is already in the queue, although it's not 100% reliable due to there is no proper transaction support. Simple queue's "Has" function always returns "has=false".

The HandlerFuncT function is called by the WorkerPoolQueue to process the data in the base queue. If the handler returns "unhandled" items, they will be re-queued to the base queue after a slight delay, in case the item processor (eg: document indexer) is not available.

Index

• Variables
• type BaseConfig
• type HandlerFuncT
• type ManagedWorkerPoolQueue
• type Manager
  • func GetManager() *Manager
  • func (m *Manager) AddManagedQueue(managed ManagedWorkerPoolQueue)
  • func (m *Manager) FlushAll(ctx context.Context, timeout time.Duration) error
  • func (m *Manager) GetManagedQueue(qid int64) ManagedWorkerPoolQueue
  • func (m *Manager) ManagedQueues() map[int64]ManagedWorkerPoolQueue
• type WorkerPoolQueue
  • func CreateSimpleQueue[T any](name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T]
  • func CreateUniqueQueue[T any](name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T]
  • func NewWorkerPoolQueueBySetting[T any](name string, queueSetting setting.QueueSettings, handler HandlerFuncT[T], ...) (*WorkerPoolQueue[T], error)
  • func (q *WorkerPoolQueue[T]) FlushWithContext(ctx context.Context, timeout time.Duration) (err error)
  • func (q *WorkerPoolQueue[T]) GetItemTypeName() string
  • func (q *WorkerPoolQueue[T]) GetName() string
  • func (q *WorkerPoolQueue[T]) GetQueueItemNumber() int
  • func (q *WorkerPoolQueue[T]) GetType() string
  • func (q *WorkerPoolQueue[T]) GetWorkerActiveNumber() int
  • func (q *WorkerPoolQueue[T]) GetWorkerMaxNumber() int
  • func (q *WorkerPoolQueue[T]) GetWorkerNumber() int
  • func (q *WorkerPoolQueue[T]) Has(data T) (bool, error)
  • func (q *WorkerPoolQueue[T]) Push(data T) error
  • func (q *WorkerPoolQueue[T]) RemoveAllItems(ctx context.Context) error
  • func (q *WorkerPoolQueue[T]) Run(atShutdown, atTerminate func(func()))
  • func (q *WorkerPoolQueue[T]) SetWorkerMaxNumber(num int)
  • func (q *WorkerPoolQueue[T]) ShutdownWait(timeout time.Duration)

Variables

var ErrAlreadyInQueue = util.NewAlreadyExistErrorf("already in queue")

Types

type BaseConfig

type BaseConfig struct {
	ManagedName string
	DataFullDir string // the caller must prepare an absolute path

	ConnStr string
	Length  int

	QueueFullName, SetFullName string
}

type HandlerFuncT

type HandlerFuncT[T any] func(...T) (unhandled []T)

type ManagedWorkerPoolQueue

type ManagedWorkerPoolQueue interface {
	GetName() string
	GetType() string
	GetItemTypeName() string
	GetWorkerNumber() int
	GetWorkerActiveNumber() int
	GetWorkerMaxNumber() int
	SetWorkerMaxNumber(num int)
	GetQueueItemNumber() int

	// FlushWithContext tries to make the handler process all items in the queue synchronously.
	// It is for testing purpose only. It's not designed to be used in a cluster.
	FlushWithContext(ctx context.Context, timeout time.Duration) error

	// RemoveAllItems removes all items in the base queue (on-the-fly items are not affected)
	RemoveAllItems(ctx context.Context) error
}

type Manager

type Manager struct {
	Queues map[int64]ManagedWorkerPoolQueue
	// contains filtered or unexported fields
}

Manager is a manager for the queues created by "CreateXxxQueue" functions, these queues are called "managed queues".

func GetManager

func GetManager() *Manager

func (*Manager) AddManagedQueue

func (m *Manager) AddManagedQueue(managed ManagedWorkerPoolQueue)

func (*Manager) FlushAll

func (m *Manager) FlushAll(ctx context.Context, timeout time.Duration) error

FlushAll tries to make all managed queues process all items synchronously, until timeout or the queue is empty. It is for testing purpose only. It's not designed to be used in a cluster.

func (*Manager) GetManagedQueue

func (m *Manager) GetManagedQueue(qid int64) ManagedWorkerPoolQueue

func (*Manager) ManagedQueues

func (m *Manager) ManagedQueues() map[int64]ManagedWorkerPoolQueue

type WorkerPoolQueue

type WorkerPoolQueue[T any] struct {
	// contains filtered or unexported fields
}

WorkerPoolQueue is a queue that uses a pool of workers to process items It can use different underlying (base) queue types

func CreateSimpleQueue

func CreateSimpleQueue[T any](name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T]

CreateSimpleQueue creates a simple queue from global setting config provider by name

func CreateUniqueQueue

func CreateUniqueQueue[T any](name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T]

CreateUniqueQueue creates a unique queue from global setting config provider by name

func NewWorkerPoolQueueBySetting

func NewWorkerPoolQueueBySetting[T any](name string, queueSetting setting.QueueSettings, handler HandlerFuncT[T], unique bool) (*WorkerPoolQueue[T], error)

func (*WorkerPoolQueue[T]) FlushWithContext

func (q *WorkerPoolQueue[T]) FlushWithContext(ctx context.Context, timeout time.Duration) (err error)

func (*WorkerPoolQueue[T]) GetItemTypeName

func (q *WorkerPoolQueue[T]) GetItemTypeName() string

func (*WorkerPoolQueue[T]) GetName

func (q *WorkerPoolQueue[T]) GetName() string

func (*WorkerPoolQueue[T]) GetQueueItemNumber

func (q *WorkerPoolQueue[T]) GetQueueItemNumber() int

func (*WorkerPoolQueue[T]) GetType

func (q *WorkerPoolQueue[T]) GetType() string

func (*WorkerPoolQueue[T]) GetWorkerActiveNumber

func (q *WorkerPoolQueue[T]) GetWorkerActiveNumber() int

func (*WorkerPoolQueue[T]) GetWorkerMaxNumber

func (q *WorkerPoolQueue[T]) GetWorkerMaxNumber() int

func (*WorkerPoolQueue[T]) GetWorkerNumber

func (q *WorkerPoolQueue[T]) GetWorkerNumber() int

func (*WorkerPoolQueue[T]) Has

func (q *WorkerPoolQueue[T]) Has(data T) (bool, error)

Has only works for unique queues. Keep in mind that this check may not be reliable (due to lacking of proper transaction support) There could be a small chance that duplicate items appear in the queue

func (*WorkerPoolQueue[T]) Push

func (q *WorkerPoolQueue[T]) Push(data T) error

Push adds an item to the queue, it may block for a while and then returns an error if the queue is full

func (*WorkerPoolQueue[T]) RemoveAllItems

func (q *WorkerPoolQueue[T]) RemoveAllItems(ctx context.Context) error

RemoveAllItems removes all items in the baes queue

func (*WorkerPoolQueue[T]) Run

func (q *WorkerPoolQueue[T]) Run(atShutdown, atTerminate func(func()))

func (*WorkerPoolQueue[T]) SetWorkerMaxNumber

func (q *WorkerPoolQueue[T]) SetWorkerMaxNumber(num int)

func (*WorkerPoolQueue[T]) ShutdownWait

func (q *WorkerPoolQueue[T]) ShutdownWait(timeout time.Duration)

ShutdownWait shuts down the queue, waits for all workers to finish their jobs, and pushes the unhandled items back to the base queue It waits for all workers (handlers) to finish their jobs, in case some buggy handlers would hang forever, a reasonable timeout is needed