sched

README

sched

sched is a high performance task scheduling library with future support.

Usage

// Init sched, with tasks should recovered when reboot
futures, err := sched.Init(
    "redis://127.0.0.1:6379/1"， 
    &ArbitraryTask1{}, 
    &ArbitraryTask2{},
)
if err != nil {
    panic(err)
}
// Retrieve task's future
for i := range futures {
    fmt.Printf("%v", futures[i].Get())
}

// Setup tasks, use future.Get() to retrieve the future of task
future, err := sched.Submit(&ArbitraryTask{...})
if err != nil {
    panic(err)
}
fmt.Printf("%v", future.Get())

// Launch a task, use future.Get() to retrieve the future of task
future, err := sched.Trigger(&ArbitraryTask{...})
if err != nil {
    panic(err)
}
fmt.Printf("%v", future.Get())

// Pause sched
sched.Pause()

// Resume sched
sched.Resume()

// Wait sched schedule all tasks
sched.Wait()

// Stop sched gracefully
sched.Stop()

Task Design

Learn more regarding task design, see test examples.

License

MIT © Changkun Ou

Documentation

Overview

Package sched provides a consistently reliable task scheduler with future support.

Introduction

sched is a consistently reliable embedded task scheduler library for Go. It applies to be a microkernel of an internal application service, and pluggable tasks must implements sched Task interface.

sched not only schedules a task at a specific time or reschedules a planned task immediately, but also flexible to support periodically tasks, which differ from traditional non-consistently unreliable cron task scheduling.

Furthermore, sched manage tasks, like goroutine runtime scheduler, uses greedy scheduling schedules all tasks and a distributed lock mechanism that ensures tasks can only be executed once across multiple replica instances.

Usage

Callers must initialize sched database when using sched. sched schedules different tasks in a priority queue and schedules task with minimum goroutines when tasks with same execution time arrival:

// Init sched, with tasks should recovered when reboot
futures, err := sched.Init(
	"redis://127.0.0.1:6379/1"，
	&ArbitraryTask1{},
	&ArbitraryTask2{},
)
if err != nil {
	panic(err)
}
// Retrieve task's future
for i := range futures {
	fmt.Printf("%v", futures[i].Get())
}

// Setup tasks, use future.Get() to retrieve the future of task
future, err := sched.Submit(&ArbitraryTask{...})
if err != nil {
	panic(err)
}
fmt.Printf("%v", future.Get())

// Launch a task, use future.Get() to retrieve the future of task
future, err := sched.Trigger(&ArbitraryTask{...})
if err != nil {
	panic(err)
}
fmt.Printf("%v", future.Get())

// Pause sched
sched.Pause()

// Resume sched
sched.Resume()

// Stop sched gracefully
sched.Stop()

Task interface

A Task that can be scheduled by sched must implements the following task interface:

// Task interface for sched
type Task interface {
	GetID() (id string)
	SetID(id string)
	IsValidID() bool
	GetExecution() (execute time.Time)
	SetExecution(new time.Time) (old time.Time)
	GetTimeout() (lockTimeout time.Duration)
	GetRetryTime() (execute time.Time)
	Execute() (result interface{}, retry bool, fail error)
}

Note that your task must be a serilizable struct by `json.Marshal()`, otherwise it cannot be persist by goshceudler (e.g. `type Func func()` cannot be scheduled)

Task Future

A Task can have a future of its final execution result. `sched.Submit` and `sched.Trigger` both returns future object of `Execute()`'s `result interface{}`. To get the future:

future.Get().(YourResultType)

Index

• func Pause()
• func Resume()
• func Stop()
• func Wait()
• type Task
• type TaskFuture
  • func Init(db string, all ...Task) ([]TaskFuture, error)
  • func Submit(t Task) (TaskFuture, error)
  • func Trigger(t Task) (TaskFuture, error)

Functions

func Pause

func Pause()

Pause stops the sched from running, this is a pair call with Resume(), Pause() must be called first

Pause() is the only way that completely pause sched from running. the internal sched0.pause() is only used for internal scheduling, which is not a real pause.

func Resume

func Resume()

Resume resumes sched and start executing tasks this is a pair call with Pause(), Resume() must be called second

func Stop

func Stop()

Stop stops runtime scheduler gracefully. Note that the call should only be called then application terminates

func Wait

func Wait()

Wait waits all tasks to be scheduled.

Types

type Task

type Task interface {
	// GetID must returns a unique ID for all of the scheduled task.
	GetID() (id string)
	// SetID will set id as the unique ID for the scheduled task.
	SetID(id string)
	// IsValidID verifies that an ID is an valid ID for the task.
	IsValidID() bool
	// GetExecution returns the time for task execution.
	GetExecution() (execute time.Time)
	// SetExecution sets a new time for the task execution
	SetExecution(new time.Time) (old time.Time)
	// GetTimeout returns the locking time for a giving task.
	// Users should aware that this time should *not* longer than the task execution time.
	// For instance, if your task consumes 1 second for execution,
	// then the locking time must shorter than 1 second.
	GetTimeout() (lockTimeout time.Duration)
	// GetRetryTime returns the retry time if a task was failed.
	GetRetryTime() (execute time.Time)
	// Execute executes the actual task, it can return a result,
	// or if the task need a retry, or it was failed in this execution.
	Execute() (result interface{}, retry bool, fail error)
}

Task interface for sched

type TaskFuture

type TaskFuture interface {
	Get() interface{}
}

TaskFuture is the future of Task execution

func Init

func Init(db string, all ...Task) ([]TaskFuture, error)

Init initialize task scheduler

func Submit

func Submit(t Task) (TaskFuture, error)

Submit given tasks

func Trigger

func Trigger(t Task) (TaskFuture, error)

Trigger given tasks immediately