jobsched

Documentation

Overview

Package jobsched provides a framework for job scheduling.

It enables the client to execute a lot of jobs on multiple goroutines and achieves load balance.

The client can customize the job scheduling algorithm by implementing a JobQueue.

Index

• func Run[Job, Properties, Feedback any](jobHandler JobHandler[Job, Properties, Feedback], ...) []framework.PanicRecord
• func RunWithoutFeedback[Job, Properties any](jobHandler JobHandler[Job, Properties, NoFeedback], ...) []framework.PanicRecord
• type Controller
  • func New[Job, Properties, Feedback any](jobHandler JobHandler[Job, Properties, Feedback], ...) Controller[Job, Properties, Feedback]
  • func NewWithoutFeedback[Job, Properties any](jobHandler JobHandler[Job, Properties, NoFeedback], ...) Controller[Job, Properties, NoFeedback]
• type FeedbackHandler
• type JobHandler
• type JobQueue
• type JobQueueMaker
• type Meta
• type MetaJob
• type NoFeedback
• type NoProperty
• type Options

Functions

func Run

func Run[Job, Properties, Feedback any](
	jobHandler JobHandler[Job, Properties, Feedback],
	feedbackHandler FeedbackHandler[Feedback],
	opts *Options[Job, Properties, Feedback],
	metaJob ...*MetaJob[Job, Properties],
) []framework.PanicRecord

Run creates a Controller with specified arguments, and then runs it. It returns the panic records of the Controller.

The parameters are the same as those of function New.

func RunWithoutFeedback

func RunWithoutFeedback[Job, Properties any](
	jobHandler JobHandler[Job, Properties, NoFeedback],
	opts *Options[Job, Properties, NoFeedback],
	metaJob ...*MetaJob[Job, Properties],
) []framework.PanicRecord

RunWithoutFeedback creates a Controller[Job, Properties, NoFeedback] with specified arguments, and then runs it. It returns the panic records of the Controller.

The parameters are the same as those of function NewWithoutFeedback.

It is equivalent to Run[Job, Properties, NoFeedback](jobHandler, nil, opts, metaJob...).

Types

type Controller

type Controller[Job, Properties, Feedback any] interface {
	framework.Controller

	// Input enables the client to input new jobs.
	//
	// If an item in metaJob is nil, it is treated as a zero-value job
	// (with the field Job to its zero value, Meta.Priority to 0,
	// Meta.CreationTime to time.Now(), and Meta.Custom to its zero value).
	// If an item in metaJob has the field Meta.CreationTime with a zero value,
	// this field is set to time.Now() by the framework.
	//
	// It returns the number of jobs input successfully.
	//
	// It is safe for concurrent use by multiple goroutines.
	//
	// The client can input new jobs before the first effective call to
	// the method Wait (i.e., the call after invoking the method Launch).
	// After calling the method Wait, Input does nothing and returns 0.
	// Note that the method Run calls Wait inside.
	Input(metaJob ...*MetaJob[Job, Properties]) int
}

Controller is a controller for this job scheduling framework.

It is used to launch, cancel, and wait for the job. Also, it is used to input new jobs.

The first type parameter Job is the type of jobs. The second type parameter Properties is the type of custom properties in the meta information of jobs. The third type parameter Feedback is the type of feedback on the jobs, which is collected and handled in a dedicated goroutine.

func New

func New[Job, Properties, Feedback any](
	jobHandler JobHandler[Job, Properties, Feedback],
	feedbackHandler FeedbackHandler[Feedback],
	opts *Options[Job, Properties, Feedback],
	metaJob ...*MetaJob[Job, Properties],
) Controller[Job, Properties, Feedback]

New creates a new Controller with options opts.

The first type parameter Job is the type of jobs. The second type parameter Properties is the type of custom properties in the meta information of jobs. The third type parameter Feedback is the type of feedback on the jobs, which is collected and handled in a dedicated goroutine.

jobHandler is the job handler. New panics if jobHandler is nil.

feedbackHandler is the feedback handler. If feedbackHandler is nil, all feedback is discarded silently. If the type of feedback is NoFeedback, the framework never calls feedbackHandler.

If opts are nil, a zero-value Options is used.

metaJob is initial jobs to process. If an item in metaJob is nil, it is treated as a zero-value job (with the field Job to its zero value, Meta.Priority to 0, Meta.CreationTime to time.Now(), and Meta.Custom to its zero value). If an item in metaJob has the field Meta.CreationTime with a zero value, this field is set to time.Now() by the framework.

func NewWithoutFeedback

func NewWithoutFeedback[Job, Properties any](
	jobHandler JobHandler[Job, Properties, NoFeedback],
	opts *Options[Job, Properties, NoFeedback],
	metaJob ...*MetaJob[Job, Properties],
) Controller[Job, Properties, NoFeedback]

NewWithoutFeedback is like New but sets the type of feedback to NoFeedback.

The parameters are similar to those of function New, except that NewWithoutFeedback does not require feedbackHandler.

It is equivalent to New[Job, Properties, NoFeedback](jobHandler, nil, opts, metaJob...).

type FeedbackHandler

type FeedbackHandler[Feedback any] func(
	canceler concurrency.Canceler,
	feedback Feedback,
)

FeedbackHandler is a function to process feedback.

The first parameter is a canceler to interrupt all job processors. The second parameter is the feedback to be processed.

type JobHandler

type JobHandler[Job, Properties, Feedback any] func(
	canceler concurrency.Canceler,
	rank int,
	job Job,
) (newJobs []*MetaJob[Job, Properties], feedback Feedback)

JobHandler is a function to process a job.

The first type parameter Job is the type of jobs. The second type parameter Properties is the type of custom properties in the meta information of jobs. The third type parameter Feedback is the type of feedback on the jobs, which is collected and handled in a dedicated goroutine.

The first parameter is a canceler to interrupt all job processors. The second parameter is the rank of the worker goroutine (from 0 to ctrl.NumGoroutine()-1) to identify the goroutine uniquely. The third parameter is the job to be processed.

It returns new jobs generated during or after processing the current job, called newJobs. If an item in newJobs is nil, it is treated as a zero-value job (with the field Job to its zero value, Meta.Priority to 0, Meta.CreationTime to time.Now(), and Meta.Custom to its zero value). If an item in newJobs has the field Meta.CreationTime with a zero value, this field is set to time.Now() by the framework.

In addition, it also returns feedback on this job, which is collected and handled in a dedicated goroutine. In particular, if the type of feedback is NoFeedback, the framework skips the feedback handler.

type JobQueue

type JobQueue[Job, Properties any] interface {
	// Len returns the number of jobs in the queue.
	Len() int

	// Enqueue adds metaJob into the queue.
	//
	// The framework guarantees that all items in metaJob are never nil
	// and have a non-zero creation time in their meta information.
	Enqueue(metaJob ...*MetaJob[Job, Properties])

	// Dequeue removes and returns a job in the queue.
	//
	// It panics if the queue is nil or empty.
	Dequeue() Job
}

JobQueue is a queue for job scheduling. The client can customize a job scheduling algorithm by implementing this interface.

The first type parameter Job is the type of jobs. The second type parameter Properties is the type of custom properties in the meta information of jobs.

A JobQueue is used in only one goroutine. Its implementation does not need to consider concurrency issues.

type JobQueueMaker

type JobQueueMaker[Job, Properties any] interface {
	// New creates a new job queue.
	New() JobQueue[Job, Properties]
}

JobQueueMaker is a maker for creating job queues.

The first type parameter Job is the type of jobs. The second type parameter Properties is the type of custom properties in the meta information of jobs.

It has a method New, with no parameter. The client should set any parameters required for creating job queues in the instance of this interface.

type Meta

type Meta[Properties any] struct {
	// The priority of the job, a non-negative integer.
	//
	// The greater the value, the higher the priority.
	// The default value 0 corresponds to the lowest priority.
	Priority uint

	// The creation time of the job.
	//
	// A zero-value CreationTime is set to time.Now() by the framework
	// before adding to a job queue.
	CreationTime time.Time

	// Custom properties, used to customize job scheduling algorithm.
	//
	// If no custom property is required, set the type parameter to NoProperty.
	Custom Properties
}

Meta contains the meta information of a job, including its priority, creation time, and any custom properties.

type MetaJob

type MetaJob[Job, Properties any] struct {
	Meta Meta[Properties]
	Job  Job
}

MetaJob combines the job and its meta information.

type NoFeedback

type NoFeedback struct{}

NoFeedback is a special case of feedback type to indicate that the job handler has no feedback. In this case, the framework skips the feedback handler.

type NoProperty

type NoProperty = struct{}

NoProperty is an alias for struct{}.

It is used to instantiate Meta without custom properties:

Meta[NoProperty]

type Options

type Options[Job, Properties, Feedback any] struct {
	// The number of worker goroutines to process jobs.
	// Non-positive values for using max(1, runtime.NumCPU()-2).
	NumWorker int

	// The maker to create a new job queue.
	// It enables the client to make custom JobQueue.
	// If it is nil, an FCFS (first come, first served) job queue is used.
	JobQueueMaker JobQueueMaker[Job, Properties]

	// The buffer size of the feedback channel.
	// Non-positive values for no buffer.
	// Only take effect when the type of feedback is not NoFeedback.
	FeedbackChanBufSize int

	// The setup function called by each worker goroutine that processes jobs.
	//
	// If it is not nil, each worker goroutine calls it
	// when the goroutine starts.
	//
	// Its first parameter is the controller.
	// Its second parameter is the rank of the worker goroutine
	// (from 0 to ctrl.NumGoroutine()-1) to identify the goroutine uniquely.
	//
	// The client is responsible for guaranteeing that
	// this function is safe for concurrency.
	Setup func(ctrl Controller[Job, Properties, Feedback], rank int)

	// The cleanup function called by each worker goroutine that processes jobs.
	//
	// If this cleanup function is not nil,
	// and the goroutine has successfully executed the setup function
	// (if the setup function is not nil),
	// then each worker goroutine calls this cleanup function
	// before the goroutine ends (even if the goroutine panics).
	//
	// Its first parameter is the controller.
	// Its second parameter is the rank of the worker goroutine
	// (from 0 to ctrl.NumGoroutine()-1) to identify the goroutine uniquely.
	//
	// The client is responsible for guaranteeing that
	// this function is safe for concurrency.
	Cleanup func(ctrl Controller[Job, Properties, Feedback], rank int)
}

Options are options for creating Controller.