// Copyright 2014 beego Author. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package task
import (
"context"
"log"
"math"
"sort"
"strconv"
"strings"
"sync"
"time"
)
// bounds provides a range of acceptable values (plus a map of name to value).
type bounds struct {
min, max uint
names map[string]uint
}
type taskManager struct {
adminTaskList map[string]Tasker
taskLock sync.RWMutex
stop chan bool
changed chan bool
started bool
}
func newTaskManager() *taskManager {
return &taskManager{
adminTaskList: make(map[string]Tasker),
taskLock: sync.RWMutex{},
stop: make(chan bool),
changed: make(chan bool),
started: false,
}
}
// The bounds for each field.
var (
globalTaskManager *taskManager
seconds = bounds{0, 59, nil}
minutes = bounds{0, 59, nil}
hours = bounds{0, 23, nil}
days = bounds{1, 31, nil}
months = bounds{1, 12, map[string]uint{
"jan": 1,
"feb": 2,
"mar": 3,
"apr": 4,
"may": 5,
"jun": 6,
"jul": 7,
"aug": 8,
"sep": 9,
"oct": 10,
"nov": 11,
"dec": 12,
}}
weeks = bounds{0, 6, map[string]uint{
"sun": 0,
"mon": 1,
"tue": 2,
"wed": 3,
"thu": 4,
"fri": 5,
"sat": 6,
}}
)
const (
// Set the top bit if a star was included in the expression.
starBit = 1 << 63
)
// Schedule time taks schedule
type Schedule struct {
Second uint64
Minute uint64
Hour uint64
Day uint64
Month uint64
Week uint64
}
// TaskFunc task func type
type TaskFunc func(ctx context.Context) error
// Tasker task interface
type Tasker interface {
GetSpec(ctx context.Context) string
GetStatus(ctx context.Context) string
Run(ctx context.Context) error
SetNext(context.Context, time.Time)
GetNext(ctx context.Context) time.Time
SetPrev(context.Context, time.Time)
GetPrev(ctx context.Context) time.Time
}
// task error
type taskerr struct {
t time.Time
errinfo string
}
// Task task struct
// It's not a thread-safe structure.
// Only nearest errors will be saved in ErrList
type Task struct {
Taskname string
Spec *Schedule
SpecStr string
DoFunc TaskFunc
Prev time.Time
Next time.Time
Errlist []*taskerr // like errtime:errinfo
ErrLimit int // max length for the errlist, 0 stand for no limit
errCnt int // records the error count during the execution
}
// NewTask add new task with name, time and func
func NewTask(tname string, spec string, f TaskFunc) *Task {
task := &Task{
Taskname: tname,
DoFunc: f,
// Make configurable
ErrLimit: 100,
SpecStr: spec,
// we only store the pointer, so it won't use too many space
Errlist: make([]*taskerr, 100, 100),
}
task.SetCron(spec)
return task
}
// GetSpec get spec string
func (t *Task) GetSpec(context.Context) string {
return t.SpecStr
}
// GetStatus get current task status
func (t *Task) GetStatus(context.Context) string {
var str string
for _, v := range t.Errlist {
str += v.t.String() + ":" + v.errinfo + "
"
}
return str
}
// Run run all tasks
func (t *Task) Run(ctx context.Context) error {
err := t.DoFunc(ctx)
if err != nil {
index := t.errCnt % t.ErrLimit
t.Errlist[index] = &taskerr{t: t.Next, errinfo: err.Error()}
t.errCnt++
}
return err
}
// SetNext set next time for this task
func (t *Task) SetNext(ctx context.Context, now time.Time) {
t.Next = t.Spec.Next(now)
}
// GetNext get the next call time of this task
func (t *Task) GetNext(context.Context) time.Time {
return t.Next
}
// SetPrev set prev time of this task
func (t *Task) SetPrev(ctx context.Context, now time.Time) {
t.Prev = now
}
// GetPrev get prev time of this task
func (t *Task) GetPrev(context.Context) time.Time {
return t.Prev
}
// six columns mean:
// second:0-59
// minute:0-59
// hour:1-23
// day:1-31
// month:1-12
// week:0-6(0 means Sunday)
// SetCron some signals:
// *: any time
// ,: separate signal
// -:duration
// /n : do as n times of time duration
// ///////////////////////////////////////////////////////
// 0/30 * * * * * every 30s
// 0 43 21 * * * 21:43
// 0 15 05 * * * 05:15
// 0 0 17 * * * 17:00
// 0 0 17 * * 1 17:00 in every Monday
// 0 0,10 17 * * 0,2,3 17:00 and 17:10 in every Sunday, Tuesday and Wednesday
// 0 0-10 17 1 * * 17:00 to 17:10 in 1 min duration each time on the first day of month
// 0 0 0 1,15 * 1 0:00 on the 1st day and 15th day of month
// 0 42 4 1 * * 4:42 on the 1st day of month
// 0 0 21 * * 1-6 21:00 from Monday to Saturday
// 0 0,10,20,30,40,50 * * * * every 10 min duration
// 0 */10 * * * * every 10 min duration
// 0 * 1 * * * 1:00 to 1:59 in 1 min duration each time
// 0 0 1 * * * 1:00
// 0 0 */1 * * * 0 min of hour in 1 hour duration
// 0 0 * * * * 0 min of hour in 1 hour duration
// 0 2 8-20/3 * * * 8:02, 11:02, 14:02, 17:02, 20:02
// 0 30 5 1,15 * * 5:30 on the 1st day and 15th day of month
func (t *Task) SetCron(spec string) {
t.Spec = t.parse(spec)
}
func (t *Task) parse(spec string) *Schedule {
if len(spec) > 0 && spec[0] == '@' {
return t.parseSpec(spec)
}
// Split on whitespace. We require 5 or 6 fields.
// (second) (minute) (hour) (day of month) (month) (day of week, optional)
fields := strings.Fields(spec)
if len(fields) != 5 && len(fields) != 6 {
log.Panicf("Expected 5 or 6 fields, found %d: %s", len(fields), spec)
}
// If a sixth field is not provided (DayOfWeek), then it is equivalent to star.
if len(fields) == 5 {
fields = append(fields, "*")
}
schedule := &Schedule{
Second: getField(fields[0], seconds),
Minute: getField(fields[1], minutes),
Hour: getField(fields[2], hours),
Day: getField(fields[3], days),
Month: getField(fields[4], months),
Week: getField(fields[5], weeks),
}
return schedule
}
func (t *Task) parseSpec(spec string) *Schedule {
switch spec {
case "@yearly", "@annually":
return &Schedule{
Second: 1 << seconds.min,
Minute: 1 << minutes.min,
Hour: 1 << hours.min,
Day: 1 << days.min,
Month: 1 << months.min,
Week: all(weeks),
}
case "@monthly":
return &Schedule{
Second: 1 << seconds.min,
Minute: 1 << minutes.min,
Hour: 1 << hours.min,
Day: 1 << days.min,
Month: all(months),
Week: all(weeks),
}
case "@weekly":
return &Schedule{
Second: 1 << seconds.min,
Minute: 1 << minutes.min,
Hour: 1 << hours.min,
Day: all(days),
Month: all(months),
Week: 1 << weeks.min,
}
case "@daily", "@midnight":
return &Schedule{
Second: 1 << seconds.min,
Minute: 1 << minutes.min,
Hour: 1 << hours.min,
Day: all(days),
Month: all(months),
Week: all(weeks),
}
case "@hourly":
return &Schedule{
Second: 1 << seconds.min,
Minute: 1 << minutes.min,
Hour: all(hours),
Day: all(days),
Month: all(months),
Week: all(weeks),
}
}
log.Panicf("Unrecognized descriptor: %s", spec)
return nil
}
// Next set schedule to next time
func (s *Schedule) Next(t time.Time) time.Time {
// Start at the earliest possible time (the upcoming second).
t = t.Add(1*time.Second - time.Duration(t.Nanosecond())*time.Nanosecond)
// This flag indicates whether a field has been incremented.
added := false
// If no time is found within five years, return zero.
yearLimit := t.Year() + 5
WRAP:
if t.Year() > yearLimit {
return time.Time{}
}
// Find the first applicable month.
// If it's this month, then do nothing.
for 1< 0
dowMatch = 1< 0
)
if s.Day&starBit > 0 || s.Week&starBit > 0 {
return domMatch && dowMatch
}
return domMatch || dowMatch
}
// StartTask start all tasks
func StartTask() {
globalTaskManager.StartTask()
}
// StopTask stop all tasks
func StopTask() {
globalTaskManager.StopTask()
}
// AddTask add task with name
func AddTask(taskName string, t Tasker) {
globalTaskManager.AddTask(taskName, t)
}
// DeleteTask delete task with name
func DeleteTask(taskName string) {
globalTaskManager.DeleteTask(taskName)
}
// ClearTask clear all tasks
func ClearTask() {
globalTaskManager.ClearTask()
}
// StartTask start all tasks
func (m *taskManager) StartTask() {
m.taskLock.Lock()
defer m.taskLock.Unlock()
if m.started {
// If already started, no need to start another goroutine.
return
}
m.started = true
registerCommands()
go m.run()
}
func (m *taskManager) run() {
now := time.Now().Local()
m.taskLock.Lock()
for _, t := range m.adminTaskList {
t.SetNext(nil, now)
}
m.taskLock.Unlock()
for {
// we only use RLock here because NewMapSorter copy the reference, do not change any thing
m.taskLock.RLock()
sortList := NewMapSorter(m.adminTaskList)
m.taskLock.RUnlock()
sortList.Sort()
var effective time.Time
if len(m.adminTaskList) == 0 || sortList.Vals[0].GetNext(context.Background()).IsZero() {
// If there are no entries yet, just sleep - it still handles new entries
// and stop requests.
effective = now.AddDate(10, 0, 0)
} else {
effective = sortList.Vals[0].GetNext(context.Background())
}
select {
case now = <-time.After(effective.Sub(now)):
// Run every entry whose next time was this effective time.
for _, e := range sortList.Vals {
if e.GetNext(context.Background()) != effective {
break
}
go e.Run(nil)
e.SetPrev(context.Background(), e.GetNext(context.Background()))
e.SetNext(nil, effective)
}
continue
case <-m.changed:
now = time.Now().Local()
m.taskLock.Lock()
for _, t := range m.adminTaskList {
t.SetNext(nil, now)
}
m.taskLock.Unlock()
continue
case <-m.stop:
m.taskLock.Lock()
if m.started {
m.started = false
}
m.taskLock.Unlock()
return
}
}
}
// StopTask stop all tasks
func (m *taskManager) StopTask() {
go func() {
m.stop <- true
}()
}
// AddTask add task with name
func (m *taskManager) AddTask(taskname string, t Tasker) {
isChanged := false
m.taskLock.Lock()
t.SetNext(nil, time.Now().Local())
m.adminTaskList[taskname] = t
if m.started {
isChanged = true
}
m.taskLock.Unlock()
if isChanged {
go func() {
m.changed <- true
}()
}
}
// DeleteTask delete task with name
func (m *taskManager) DeleteTask(taskname string) {
isChanged := false
m.taskLock.Lock()
delete(m.adminTaskList, taskname)
if m.started {
isChanged = true
}
m.taskLock.Unlock()
if isChanged {
go func() {
m.changed <- true
}()
}
}
// ClearTask clear all tasks
func (m *taskManager) ClearTask() {
isChanged := false
m.taskLock.Lock()
m.adminTaskList = make(map[string]Tasker)
if m.started {
isChanged = true
}
m.taskLock.Unlock()
if isChanged {
go func() {
m.changed <- true
}()
}
}
// MapSorter sort map for tasker
type MapSorter struct {
Keys []string
Vals []Tasker
}
// NewMapSorter create new tasker map
func NewMapSorter(m map[string]Tasker) *MapSorter {
ms := &MapSorter{
Keys: make([]string, 0, len(m)),
Vals: make([]Tasker, 0, len(m)),
}
for k, v := range m {
ms.Keys = append(ms.Keys, k)
ms.Vals = append(ms.Vals, v)
}
return ms
}
// Sort sort tasker map
func (ms *MapSorter) Sort() {
sort.Sort(ms)
}
func (ms *MapSorter) Len() int { return len(ms.Keys) }
func (ms *MapSorter) Less(i, j int) bool {
if ms.Vals[i].GetNext(context.Background()).IsZero() {
return false
}
if ms.Vals[j].GetNext(context.Background()).IsZero() {
return true
}
return ms.Vals[i].GetNext(context.Background()).Before(ms.Vals[j].GetNext(context.Background()))
}
func (ms *MapSorter) Swap(i, j int) {
ms.Vals[i], ms.Vals[j] = ms.Vals[j], ms.Vals[i]
ms.Keys[i], ms.Keys[j] = ms.Keys[j], ms.Keys[i]
}
func getField(field string, r bounds) uint64 {
// list = range {"," range}
var bits uint64
ranges := strings.FieldsFunc(field, func(r rune) bool { return r == ',' })
for _, expr := range ranges {
bits |= getRange(expr, r)
}
return bits
}
// getRange returns the bits indicated by the given expression:
// number | number "-" number [ "/" number ]
func getRange(expr string, r bounds) uint64 {
var (
start, end, step uint
rangeAndStep = strings.Split(expr, "/")
lowAndHigh = strings.Split(rangeAndStep[0], "-")
singleDigit = len(lowAndHigh) == 1
)
var extrastar uint64
if lowAndHigh[0] == "*" || lowAndHigh[0] == "?" {
start = r.min
end = r.max
extrastar = starBit
} else {
start = parseIntOrName(lowAndHigh[0], r.names)
switch len(lowAndHigh) {
case 1:
end = start
case 2:
end = parseIntOrName(lowAndHigh[1], r.names)
default:
log.Panicf("Too many hyphens: %s", expr)
}
}
switch len(rangeAndStep) {
case 1:
step = 1
case 2:
step = mustParseInt(rangeAndStep[1])
// Special handling: "N/step" means "N-max/step".
if singleDigit {
end = r.max
}
default:
log.Panicf("Too many slashes: %s", expr)
}
if start < r.min {
log.Panicf("Beginning of range (%d) below minimum (%d): %s", start, r.min, expr)
}
if end > r.max {
log.Panicf("End of range (%d) above maximum (%d): %s", end, r.max, expr)
}
if start > end {
log.Panicf("Beginning of range (%d) beyond end of range (%d): %s", start, end, expr)
}
return getBits(start, end, step) | extrastar
}
// parseIntOrName returns the (possibly-named) integer contained in expr.
func parseIntOrName(expr string, names map[string]uint) uint {
if names != nil {
if namedInt, ok := names[strings.ToLower(expr)]; ok {
return namedInt
}
}
return mustParseInt(expr)
}
// mustParseInt parses the given expression as an int or panics.
func mustParseInt(expr string) uint {
num, err := strconv.Atoi(expr)
if err != nil {
log.Panicf("Failed to parse int from %s: %s", expr, err)
}
if num < 0 {
log.Panicf("Negative number (%d) not allowed: %s", num, expr)
}
return uint(num)
}
// getBits sets all bits in the range [min, max], modulo the given step size.
func getBits(min, max, step uint) uint64 {
var bits uint64
// If step is 1, use shifts.
if step == 1 {
return ^(math.MaxUint64 << (max + 1)) & (math.MaxUint64 << min)
}
// Else, use a simple loop.
for i := min; i <= max; i += step {
bits |= 1 << i
}
return bits
}
// all returns all bits within the given bounds. (plus the star bit)
func all(r bounds) uint64 {
return getBits(r.min, r.max, 1) | starBit
}
func init() {
globalTaskManager = newTaskManager()
}