task

package
v0.0.0-...-18b418f Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 8, 2024 License: MIT Imports: 15 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Id 

type Id = string

type Pool 

type Pool interface {
	ID() Id

	QueueTask(*Task) error

	MarkGlobal()
	IsGlobal() bool
	SignalAllQueued()

	CreatedInTask() *Task

	Wait(bool)
	Cancel()

	IsRoot() bool
	Status() PoolStatus
	AddCleanup(fn func(Pool))

	GetRootPool() *TaskPool
	GetWorkerPool() *WorkerPool

	LockExit()
	UnlockExit()

	RemoveTask(Id)

	Errors() []*Task
	AddError(t *Task)
}

type PoolStatus 

type PoolStatus struct {
	// The count of tasks that have completed on this task pool.
	// Complete *DOES* include failed tasks
	Complete int64

	// The count of failed tasks on this task pool
	Failed int

	// The count of all tasks that have been queued on this task pool
	Total int64

	// Percent to completion of all tasks
	Progress float64

	// How long the pool has been alive
	Runtime time.Duration
}

type QueueState 

type QueueState string
const (
	PreQueued QueueState = "pre-queued"
	InQueue   QueueState = "in-queue"
	Executing QueueState = "executing"
	Exited    QueueState = "exited"
)

type Task 

type Task struct {
	// contains filtered or unexported fields
}

func (*Task) Cancel 

func (t *Task) Cancel()

Cancel Unknowable if this is the last operation of a task, so t.success() will not have an effect after a task is cancelled. t.error() may override the exit status in special cases, such as a timeout, which is both an error and a reason for cancellation.

Cancellations are always external to the task. From within the body of the task, either error or success should be called. If a task finds itself not required to continue, success should be returned

func (*Task) CheckExit 

func (t *Task) CheckExit() bool

func (*Task) ClearAndRecompute 

func (t *Task) ClearAndRecompute()

func (*Task) ClearTimeout 

func (t *Task) ClearTimeout()

func (*Task) ErrorAndExit 

func (t *Task) ErrorAndExit(err error)

func (*Task) ExeTime 

func (t *Task) ExeTime() time.Duration

func (*Task) ExitIfSignaled 

func (t *Task) ExitIfSignaled()

ExitIfSignaled should be used intermittently to check if the task should exit. If the task should exit, it panics back to the top of safety work

func (*Task) GetChildTaskPool 

func (t *Task) GetChildTaskPool() *TaskPool

func (*Task) GetMeta 

func (t *Task) GetMeta() TaskMetadata

func (*Task) GetResult 

func (t *Task) GetResult(resultKey string) any

func (*Task) GetResults 

func (t *Task) GetResults() TaskResult

func (*Task) GetSignalChan 

func (t *Task) GetSignalChan() chan int

func (*Task) GetTaskPool 

func (t *Task) GetTaskPool() *TaskPool

func (*Task) JobName 

func (t *Task) JobName() string

func (*Task) Manipulate 

func (t *Task) Manipulate(fn func(meta TaskMetadata) error) error

Manipulate is used to change the metadata of a task while it is running. This can be useful to have a task be waiting for input from a client, and this function can be used to send that data to the task via a chan, for example.

func (*Task) Q 

func (t *Task) Q(tp *TaskPool) *Task

Q queues task on given taskPool tp, if tp is nil, will default to the global task pool. Essentially an alias for tp.QueueTask(t), so you can NewTask(...).Q(). Returns the given task to further support this

func (*Task) ReadError 

func (t *Task) ReadError() any

func (*Task) SetChildTaskPool 

func (t *Task) SetChildTaskPool(pool *TaskPool)

func (*Task) SetCleanup 

func (t *Task) SetCleanup(cleanup TaskHandler)

func (*Task) SetErrorCleanup 

func (t *Task) SetErrorCleanup(cleanup TaskHandler)

Pass a function to run if the task throws an error, in theory to cleanup any half-processed state that could litter if not finished

func (*Task) SetPostAction 

func (t *Task) SetPostAction(action func(TaskResult))

SetPostAction takes a function to be run after the task has successfully completed with the task results as the input of the function

func (*Task) SetResult 

func (t *Task) SetResult(results TaskResult)

func (*Task) SetTimeout 

func (t *Task) SetTimeout(timeout time.Time)

func (*Task) Status 

func (t *Task) Status() (bool, TaskExitStatus)

Status returns a boolean representing if a task has completed, and a string describing its exit type, if completed.

func (*Task) Success 

func (t *Task) Success(msg ...any)

func (*Task) SwLap 

func (t *Task) SwLap(label string)

Add a lap in the tasks stopwatch

func (*Task) TaskId 

func (t *Task) TaskId() Id

func (*Task) Wait 

func (t *Task) Wait() *Task

Wait Block until a task is finished. "Finished" can define success, failure, or cancel

type TaskExitStatus 

type TaskExitStatus = string
const (
	TaskSuccess  TaskExitStatus = "success"
	TaskCanceled TaskExitStatus = "cancelled"
	TaskError    TaskExitStatus = "error"
	TaskNoStatus TaskExitStatus = ""
)

type TaskHandler 

type TaskHandler func(*Task)

type TaskInterface 

type TaskInterface interface {
	TaskId() Id
	JobName() string
	GetTaskPool() *TaskPool
	GetChildTaskPool() *TaskPool
	Status() (bool, TaskExitStatus)
	GetMeta() TaskMetadata
	GetResult(string) any
	GetResults() TaskResult

	Q(pool *TaskPool) *Task

	Wait() *Task
	Cancel()

	SwLap(string)
	ClearTimeout()

	ReadError() any
	ClearAndRecompute()
	SetPostAction(action func(TaskResult))
	SetCleanup(cleanup func())
	SetErrorCleanup(cleanup func())

	ExeTime() time.Duration
}

type TaskMetadata 

type TaskMetadata interface {
	JobName() string
	MetaString() string
	FormatToResult() TaskResult
	Verify() error
}

type TaskPool 

type TaskPool struct {
	// contains filtered or unexported fields
}

func (*TaskPool) AddCleanup 

func (tp *TaskPool) AddCleanup(fn func(pool Pool))

func (*TaskPool) AddError 

func (tp *TaskPool) AddError(t *Task)

func (*TaskPool) Cancel 

func (tp *TaskPool) Cancel()

func (*TaskPool) CreatedInTask 

func (tp *TaskPool) CreatedInTask() *Task

func (*TaskPool) Errors 

func (tp *TaskPool) Errors() []*Task

func (*TaskPool) GetRootPool 

func (tp *TaskPool) GetRootPool() *TaskPool

func (*TaskPool) GetWorkerPool 

func (tp *TaskPool) GetWorkerPool() *WorkerPool

func (*TaskPool) ID 

func (tp *TaskPool) ID() Id

func (*TaskPool) IsGlobal 

func (tp *TaskPool) IsGlobal() bool

func (*TaskPool) IsRoot 

func (tp *TaskPool) IsRoot() bool

func (*TaskPool) LockExit 

func (tp *TaskPool) LockExit()

func (*TaskPool) MarkGlobal 

func (tp *TaskPool) MarkGlobal()

MarkGlobal specifies the work queue as being a "global" one

func (*TaskPool) QueueTask 

func (tp *TaskPool) QueueTask(t *Task) (err error)

func (*TaskPool) RemoveTask 

func (tp *TaskPool) RemoveTask(taskId Id)

func (*TaskPool) SignalAllQueued 

func (tp *TaskPool) SignalAllQueued()

func (*TaskPool) Status 

func (tp *TaskPool) Status() PoolStatus

func (*TaskPool) UnlockExit 

func (tp *TaskPool) UnlockExit()

func (*TaskPool) Wait 

func (tp *TaskPool) Wait(supplementWorker bool)

Wait Parks the thread on the work queue until all the tasks have been queued and finish. **If you never call tp.SignalAllQueued(), the waiters will never wake up** Make sure that you SignalAllQueued before parking here if it is the only thread loading tasks

type TaskResult 

type TaskResult = map[string]any

type TaskService 

type TaskService interface {
	RegisterJob(jobName string, fn TaskHandler)
	NewTaskPool(replace bool, createdBy *Task) *TaskPool

	GetTask(taskId Id) *Task
	GetTaskPool(Id) *TaskPool

	DispatchJob(jobName string, meta TaskMetadata, pool *TaskPool) (*Task, error)
}

type WorkerPool 

type WorkerPool struct {
	// contains filtered or unexported fields
}

func NewWorkerPool 

func NewWorkerPool(initWorkers int, logLevel int) *WorkerPool

func (*WorkerPool) AddHit 

func (wp *WorkerPool) AddHit(time time.Time, target *Task)

func (*WorkerPool) DispatchJob 

func (wp *WorkerPool) DispatchJob(jobName string, meta TaskMetadata, pool *TaskPool) (*Task, error)

func (*WorkerPool) GetTask 

func (wp *WorkerPool) GetTask(taskId Id) *Task

func (*WorkerPool) GetTaskPool 

func (wp *WorkerPool) GetTaskPool(tpId Id) *TaskPool

func (*WorkerPool) GetTaskPoolByJobName 

func (wp *WorkerPool) GetTaskPoolByJobName(jobName string) *TaskPool

func (*WorkerPool) NewTaskPool 

func (wp *WorkerPool) NewTaskPool(replace bool, createdBy *Task) *TaskPool

NewTaskPool `replace` spawns a temporary replacement thread on the parent worker pool. This prevents a deadlock when the queue fills up while adding many tasks, and none are being executed

`parent` allows chaining of task pools for floating updates to the top. This makes it possible for clients to subscribe to a single task, and get notified about all of the sub-updates of that task See taskPool.go

func (*WorkerPool) RegisterJob 

func (wp *WorkerPool) RegisterJob(jobName string, fn TaskHandler)

RegisterJob adds a template for a repeatable job that can be called upon later in the program

func (*WorkerPool) Run 

func (wp *WorkerPool) Run()

Run launches the standard threads for this worker pool

func (*WorkerPool) Status 

func (wp *WorkerPool) Status() (int, int, int, int, int)

Status returns the count of tasks in the queue, the total number of tasks accepted, number of busy workers, and the total number of live workers in the worker pool

func (*WorkerPool) Stop 

func (wp *WorkerPool) Stop()

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.