99 lines
2.2 KiB
Go
99 lines
2.2 KiB
Go
package gtq
|
|
|
|
import (
|
|
"sort"
|
|
"time"
|
|
)
|
|
|
|
// Scheduler is an internal goroutine which examines the Queues and
|
|
// choses which tasks to run next. Returns true if tasks were scheduled,
|
|
// or false if there's nothing left to schedule.
|
|
type Scheduler func(tq *TaskQueue) bool
|
|
|
|
// SimpleScheduler is the simplest possible implementation, which just takes
|
|
// tasks off the highest priority queue.
|
|
func SimpleScheduler(tq *TaskQueue) bool {
|
|
pc := tq.PriorityCounts()
|
|
// sort priorities
|
|
prios := make([]uint, 0, len(pc))
|
|
for k := range pc {
|
|
prios = append(prios, k)
|
|
}
|
|
sort.Sort(sort.Reverse(UIntSlice(prios)))
|
|
|
|
var queued uint
|
|
for _, prio := range prios {
|
|
q, ok := tq.queues.Load(prio)
|
|
if !ok {
|
|
continue
|
|
}
|
|
for q.(*Queue).Length() > 0 {
|
|
task := q.(*Queue).Remove()
|
|
tq.nextTask <- task
|
|
queued++
|
|
}
|
|
}
|
|
return (queued > 0)
|
|
}
|
|
|
|
// TimeSliceScheduler attempts to give tasks a fair share of the Stats()
|
|
// collection window, based on their priority level.
|
|
func TimeSliceScheduler(tq *TaskQueue) bool {
|
|
pc := tq.PriorityCounts()
|
|
// sort priorities
|
|
prios := make([]uint, 0, len(pc))
|
|
var prioritiesTotal uint
|
|
for k := range pc {
|
|
prios = append(prios, k)
|
|
prioritiesTotal += k
|
|
}
|
|
sort.Sort(sort.Reverse(UIntSlice(prios)))
|
|
|
|
stats := tq.Stats()
|
|
var window time.Duration
|
|
for _, stat := range stats {
|
|
window += stat.Duration
|
|
}
|
|
timeslice := window / time.Duration(prioritiesTotal)
|
|
|
|
var queued uint
|
|
for _, prio := range prios {
|
|
// calculate this priority level's fair share, and add outstanding items
|
|
// from it if it hasn't exceeded that share.
|
|
prioSlice := timeslice * time.Duration(prio)
|
|
if stats[prio].Duration < prioSlice {
|
|
q, ok := tq.queues.Load(prio)
|
|
if !ok {
|
|
continue
|
|
}
|
|
for q.(*Queue).Length() > 0 {
|
|
task := q.(*Queue).Remove()
|
|
tq.nextTask <- task
|
|
queued++
|
|
}
|
|
}
|
|
}
|
|
if queued > 0 {
|
|
return true
|
|
}
|
|
// second pass in case none of the priority levels ended up queueing
|
|
// anything, queue everything from the highest priority queue with items.
|
|
if tq.Length() > 0 {
|
|
for _, prio := range prios {
|
|
q, ok := tq.queues.Load(prio)
|
|
if !ok {
|
|
continue
|
|
}
|
|
for q.(*Queue).Length() > 0 {
|
|
task := q.(*Queue).Remove()
|
|
tq.nextTask <- task
|
|
queued++
|
|
}
|
|
if queued > 0 {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|