schedulingqueue

Documentation

Index

• Constants
• func AddListToCurrentTaskList(taskList []taskmodel.TaskIdType)
• func AddToCurrentTaskList(taskId taskmodel.TaskIdType)
• func GetSubtask(taskId taskmodel.TaskIdType, subtaskData *taskmodel.SubtaskBody, ...) error
• func MonitorCurrentTaskRoutine()
• func RemoveFromCurrentTaskList(taskId taskmodel.TaskIdType, pipeline redis.Pipeliner)
• func RemoveFromCurrentTaskListDirectly(taskId taskmodel.TaskIdType)
• func RemoveListFromCurrentTaskList(taskIdList []interface{}, pipeline redis.Pipeliner)
• type SchedulingQueue
  • func NewPriorityQueue(groupName string, queueName string, idx uint32) *SchedulingQueue
  • func NewRRQueue(groupName string, queueName string) *SchedulingQueue
  • func (queue *SchedulingQueue) AppendBoostTaskList(taskIdList *[]taskmodel.TaskIdType) error
  • func (queue *SchedulingQueue) AppendTask(taskId taskmodel.TaskIdType, taskType uint32, priority uint32) error
  • func (queue *SchedulingQueue) DecreaseTaskSliceCount(taskId taskmodel.TaskIdType) error
  • func (queue *SchedulingQueue) MoveTaskToTail(taskId taskmodel.TaskIdType, toDecrSlice bool) error
  • func (queue *SchedulingQueue) PopBoostTask(taskIdList *[]taskmodel.TaskIdType) error
  • func (queue *SchedulingQueue) RemoveTask(taskId taskmodel.TaskIdType) error
  • func (queue *SchedulingQueue) Schedule(retTaskId *taskmodel.TaskIdType, subtasks *[]taskmodel.SubtaskBody) (bool, error)
• type SchedulingTeam
  • func (queues *SchedulingTeam) AddTask(taskId taskmodel.TaskIdType, taskType uint32, priority uint32) error
  • func (queues *SchedulingTeam) GetTaskCount() (taskCount uint, err error)
  • func (queues *SchedulingTeam) Init(groupName string) error
  • func (queues *SchedulingTeam) Schedule(retTaskId *taskmodel.TaskIdType, subtasks *[]taskmodel.SubtaskBody) error

Constants

const (
	// 调度队列授予任务的基础时间片的大小, ms
	QueueBaseTimeSlice uint32 = 1000
	QueueTimeSliceStep uint32 = 1000
	RRQueueTimeSlice   uint32 = 5000

	// 优先级时间片bonus系数
	PriorityBonus_Low    = 1
	PriorityBonus_Medium = 8
	PriorityBonus_High   = 256

	// 任务调度静默期的上限
	QuietTaskMaxInterval = 600

	// 调度队列组中优先级队列的数量
	PrioirtyQueueCount uint32 = 2

	// 任务在优先级队列中分配的基础时间片的数量
	PriorityBaseQueueSliceCount uint32 = 40

	// 对优先级调度队列执行Priority Boost策略的间隔, 秒
	PriorityBoostInterval uint32 = 120
	PriorityBoostStep     uint32 = 60

	// 对低优先级调度队列执行Priority Boost策略的间隔, 秒
	RRPriorityBoostInterval uint32 = 240

	// 执行任务剩余时间加速策略的间隔, 秒
	RemainTaskAccelerationInteral uint32 = 300

	// 调度队列的Redis key前缀
	ScheduleQueueKeyPrefix = "Schedule"
)

const CheckCurrentTaskInterval = time.Duration(10) * time.Second

const CurrentTaskZSet = "current.task.list"

调度中任务队列

const DefaultCurrentTaskTimeout = time.Minute

const PriorityBoostMaxTaskCount = uint32(10)

const RRQueueIdx uint32 = 1000000

Functions

func AddListToCurrentTaskList

func AddListToCurrentTaskList(taskList []taskmodel.TaskIdType)

添加到调度中任务队列

func AddToCurrentTaskList

func AddToCurrentTaskList(taskId taskmodel.TaskIdType)

添加到调度中任务队列

func GetSubtask

func GetSubtask(
	taskId taskmodel.TaskIdType,
	subtaskData *taskmodel.SubtaskBody,
	finished *bool,
) error

get a subtask from the subtask queue

func MonitorCurrentTaskRoutine

func MonitorCurrentTaskRoutine()

协程 <<go_monitor_current_task>>

func RemoveFromCurrentTaskList

func RemoveFromCurrentTaskList(taskId taskmodel.TaskIdType, pipeline redis.Pipeliner)

从当前任务列表中删除任务

func RemoveFromCurrentTaskListDirectly

func RemoveFromCurrentTaskListDirectly(taskId taskmodel.TaskIdType)

从当前任务列表中删除任务

func RemoveListFromCurrentTaskList

func RemoveListFromCurrentTaskList(taskIdList []interface{}, pipeline redis.Pipeliner)

从当前任务列表中删除任务

Types

type SchedulingQueue

type SchedulingQueue struct {
	QuotaGroupName string                    // 队列所属的资源组的名称
	QueueIndex     uint32                    // 队列在资源组内的索引
	QueueKeyName   string                    // 队列的Key名
	TimeSlice      uint32                    // 任务队列的轮转时间片，单位为ms
	BaseQueueSlice uint32                    // 任务队列授予任务的基础时间片的数量, 单位为个
	Scheduler      scheduler.IQueueScheduler // 调度队列的调度接口
	NextQueue      *SchedulingQueue          // 下一个调度队列
	TaskCount      uint                      // 队列中的任务数
}

调度队列

func NewPriorityQueue

func NewPriorityQueue(groupName string, queueName string, idx uint32) *SchedulingQueue

创建一个优先级队列

func NewRRQueue

func NewRRQueue(groupName string, queueName string) *SchedulingQueue

创建一个低优先级队列

func (*SchedulingQueue) AppendBoostTaskList

func (queue *SchedulingQueue) AppendBoostTaskList(taskIdList *[]taskmodel.TaskIdType) error

向队列尾部添加PriorityBoost的任务列表

func (*SchedulingQueue) AppendTask

func (queue *SchedulingQueue) AppendTask(
	taskId taskmodel.TaskIdType,
	taskType uint32,
	priority uint32,
) error

各队列尾部添加任务

func (*SchedulingQueue) DecreaseTaskSliceCount

func (queue *SchedulingQueue) DecreaseTaskSliceCount(
	taskId taskmodel.TaskIdType,
) error

减少任务的时间片数量

func (*SchedulingQueue) MoveTaskToTail

func (queue *SchedulingQueue) MoveTaskToTail(taskId taskmodel.TaskIdType, toDecrSlice bool) error

将任务移到队尾

func (*SchedulingQueue) PopBoostTask

func (queue *SchedulingQueue) PopBoostTask(taskIdList *[]taskmodel.TaskIdType) error

从队列中取出前若干个任务

func (*SchedulingQueue) RemoveTask

func (queue *SchedulingQueue) RemoveTask(taskId taskmodel.TaskIdType) error

从队列中移除任务

func (*SchedulingQueue) Schedule

func (queue *SchedulingQueue) Schedule(
	retTaskId *taskmodel.TaskIdType,
	subtasks *[]taskmodel.SubtaskBody,
) (bool, error)

调度任务 从当前的调度队列中选出一个任务, 进行调度, 返回获取的子任务列表 处理调度队列为空的情况, retTaskId为0, subtasks返回的元素为空

type SchedulingTeam

type SchedulingTeam struct {
	TeamName       string             // 调度队列组名
	PriorityQueues []*SchedulingQueue // 优先级队列, 队列内的任务有优先级
	RRQueue        *SchedulingQueue   // 低优先级队列, 队列内的任务使用时间片轮转策略
}

调度队列组

func (*SchedulingTeam) AddTask

func (queues *SchedulingTeam) AddTask(
	taskId taskmodel.TaskIdType,
	taskType uint32,
	priority uint32,
) error

向调度队列组中添加任务

func (*SchedulingTeam) GetTaskCount

func (queues *SchedulingTeam) GetTaskCount() (taskCount uint, err error)

获取调度队列组中的任务数

func (*SchedulingTeam) Init

func (queues *SchedulingTeam) Init(groupName string) error

初始化

func (*SchedulingTeam) Schedule

func (queues *SchedulingTeam) Schedule(
	retTaskId *taskmodel.TaskIdType,
	subtasks *[]taskmodel.SubtaskBody,
) error

从调度队列组中选出一个任务来执行, 返回任务的子任务列表 处理调度队列为空的情况, retTaskId为0, subtasks返回的元素为空