queue

Documentation

Index

• Constants
• type Item
  • func NewItem(p int, val interface{}) *Item
  • func (i *Item) Value() interface{}
• type PriorityQueue
  • func (pq PriorityQueue) Len() int
  • func (pq PriorityQueue) Less(i, j int) bool
  • func (pq *PriorityQueue) Pop() interface{}
  • func (pq *PriorityQueue) Push(x interface{})
  • func (pq PriorityQueue) Swap(i, j int)
• type TaskQueue
  • func NewTaskQueue(db types.DB, lg types.Logger, t *entity.Tenant, ls types.Listener) *TaskQueue
  • func (c *TaskQueue) EnqueueUserTask(task *entity.UserTask, delta int)
  • func (c *TaskQueue) FetchTasks()
  • func (c *TaskQueue) PopAllUserTasks() (tasks entity.UserTasks, err error)
  • func (c *TaskQueue) PopUserTasks(n int) (tasks entity.UserTasks, popped int, err error)

Constants

const FetchFromDBWatermarkRatio = 2

Types

type Item

type Item struct {
	// contains filtered or unexported fields
}

An Item is something we manage in a priority queue.

func NewItem

func NewItem(p int, val interface{}) *Item

func (*Item) Value

func (i *Item) Value() interface{}

type PriorityQueue

type PriorityQueue []*Item

A PriorityQueue implements heap.Interface and holds Items.

func (PriorityQueue) Len

func (pq PriorityQueue) Len() int

func (PriorityQueue) Less

func (pq PriorityQueue) Less(i, j int) bool

func (*PriorityQueue) Pop

func (pq *PriorityQueue) Pop() interface{}

func (*PriorityQueue) Push

func (pq *PriorityQueue) Push(x interface{})

func (PriorityQueue) Swap

func (pq PriorityQueue) Swap(i, j int)

type TaskQueue

type TaskQueue struct {
	Tenant     *entity.Tenant
	DelayTasks interface{} // TODO: time wheel
	CronTasks  interface{} // TODO: time wheel
	UserTasks  *PriorityQueue
	// contains filtered or unexported fields
}

TaskQueue is in-memory cache for tenant tasks and resource quota, automatically handling fetching tasks from database when necessary.

func NewTaskQueue

func NewTaskQueue(db types.DB, lg types.Logger, t *entity.Tenant, ls types.Listener) *TaskQueue

func (*TaskQueue) EnqueueUserTask

func (c *TaskQueue) EnqueueUserTask(task *entity.UserTask, delta int)

EnqueueUserTask put task back to the priority queue, adding delta to task's priority. This is useful when task has to be retried as soon as possible before processing other tasks.

func (*TaskQueue) FetchTasks

func (c *TaskQueue) FetchTasks()

FetchTasks read tasks from database and populate in-memory queue

func (*TaskQueue) PopAllUserTasks

func (c *TaskQueue) PopAllUserTasks() (tasks entity.UserTasks, err error)

func (*TaskQueue) PopUserTasks

func (c *TaskQueue) PopUserTasks(n int) (tasks entity.UserTasks, popped int, err error)

PopUserTasks pops at most n entity.UserTasks from cache, returns tasks and the actual number successfully popped. When the number of cached entity.UserTasks is zero or less than FetchFromDBWatermarkRatio * n, it will populate tasks in background.