queue

Documentation

Overview

Package queue implements a specialized concurrent queue system for Gitea.

Terminology:

1. Item: - An item can be a simple value, such as an integer, or a more complex structure that has multiple fields. Usually a item serves as a task or a message. Sets of items will be sent to a queue handler to be processed. - It's represented as a JSON-marshaled binary slice in the queue

2. Batch: - A collection of items that are grouped together for processing. Each worker receives a batch of items.

3. Worker: - Individual unit of execution designed to process items from the queue. It's a goroutine that calls the Handler. - Workers will get new items through a channel (WorkerPoolQueue is responsible for the distribution). - Workers operate in parallel. The default value of max workers is determined by the setting system.

4. Handler (represented by HandlerFuncT type): - It's the function responsible for processing items. Each active worker will call it. - If an item or some items are not psuccessfully rocessed, the handler could return them as "unhandled items". In such scenarios, the queue system ensures these unhandled items are returned to the base queue after a brief delay. This mechanism is particularly beneficial in cases where the processing entity (like a document indexer) is temporarily unavailable. It ensures that no item is skipped or lost due to transient failures in the processing mechanism.

5. Base queue: - Represents the underlying storage mechanism for the queue. There are several implementations: - Channel: Uses Go's native channel constructs to manage the queue, suitable for in-memory queuing. - LevelDB: Especially useful in persistent queues for single instances. - Redis: Suitable for clusters, where we may have multiple nodes. - Dummy: This is special, it's not a real queue, it's a immediate no-op queue, which is useful for tests. - They all have the same abstraction, the same interface, and they are tested by the same testing code.

6. WorkerPoolQueue:

• It's responsible to glue all together, using the "base queue" to provide "worker pool" functionality. It creates new workers if needed and can flush the queue, running all the items synchronously till it finishes.
• Its "Push" function doesn't block forever, it will return an error if the queue is full after the timeout.

7. Manager:

• The purpose of it is to serve as a centralized manager for multiple WorkerPoolQueue instances. Whenever we want to create a new queue, flush, or get a specific queue, we could use it.

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 the lack of proper transaction support. Simple queue's "Has" function always returns "has=false".

A WorkerPoolQueue is a generic struct; this means it will work with any type but just for that type. If you want another kind of items to run, you would have to call the manager to create a new WorkerPoolQueue for you with a different handler that works with this new type of item. As an example of this:

 func Init() error {
	 itemQueue = queue.CreateSimpleQueue(graceful.GetManager().ShutdownContext(), "queue-name", handler)
	 ...
 }
 func handler(items ...*mypkg.QueueItem) []*mypkg.QueueItem { ... }

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](ctx context.Context, name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T]
  • func CreateUniqueQueue[T any](ctx context.Context, name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T]
  • func NewWorkerPoolQueueWithContext[T any](ctx context.Context, name string, queueSetting setting.QueueSettings, ...) (*WorkerPoolQueue[T], error)
  • func (q *WorkerPoolQueue[T]) Cancel()
  • 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()
  • 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](ctx context.Context, 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](ctx context.Context, name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T]

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

func NewWorkerPoolQueueWithContext

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

func (*WorkerPoolQueue[T]) Cancel

func (q *WorkerPoolQueue[T]) Cancel()

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()

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