core

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Dec 26, 2025 License: MIT Imports: 11 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func PostDelayedTaskAndReplyWithResult 

func PostDelayedTaskAndReplyWithResult[T any](
	targetRunner TaskRunner,
	task TaskWithResult[T],
	delay time.Duration,
	reply ReplyWithResult[T],
	replyRunner TaskRunner,
)

PostDelayedTaskAndReplyWithResult is similar to PostTaskAndReplyWithResult, but delays the execution of the task.

The reply is NOT delayed - it executes immediately after the task completes. Only the initial task execution is delayed by the specified duration.

Example: 

PostDelayedTaskAndReplyWithResult(
    runner,
    func(ctx context.Context) (string, error) {
        return "delayed result", nil
    },
    2*time.Second,  // Wait 2 seconds before starting task
    func(ctx context.Context, result string, err error) {
        fmt.Println(result)  // Executes immediately after task completes
    },
    replyRunner,
)

func PostDelayedTaskAndReplyWithResultAndTraits 

func PostDelayedTaskAndReplyWithResultAndTraits[T any](
	targetRunner TaskRunner,
	task TaskWithResult[T],
	delay time.Duration,
	taskTraits TaskTraits,
	reply ReplyWithResult[T],
	replyTraits TaskTraits,
	replyRunner TaskRunner,
)

PostDelayedTaskAndReplyWithResultAndTraits is the full-featured delayed version with separate traits for task and reply.

func PostTaskAndReplyWithResult 

func PostTaskAndReplyWithResult[T any](
	targetRunner TaskRunner,
	task TaskWithResult[T],
	reply ReplyWithResult[T],
	replyRunner TaskRunner,
)

PostTaskAndReplyWithResult executes a task that returns a result of type T and an error, then passes that result to a reply callback on the replyRunner.

This function uses closure capture to safely pass the result across goroutines. The captured variables (result and err) will escape to the heap, ensuring thread safety.

Execution guarantee (Happens-Before): - The task ALWAYS completes before the reply starts - The reply ALWAYS sees the final values written by the task - This is guaranteed by the sequential execution in wrappedTask

Example: 

PostTaskAndReplyWithResult(
    backgroundRunner,
    func(ctx context.Context) (int, error) {
        return len("Hello"), nil
    },
    func(ctx context.Context, length int, err error) {
        fmt.Printf("Length: %d\n", length)
    },
    uiRunner,
)

func PostTaskAndReplyWithResultAndTraits 

func PostTaskAndReplyWithResultAndTraits[T any](
	targetRunner TaskRunner,
	task TaskWithResult[T],
	taskTraits TaskTraits,
	reply ReplyWithResult[T],
	replyTraits TaskTraits,
	replyRunner TaskRunner,
)

PostTaskAndReplyWithResultAndTraits is the full-featured version that allows specifying different traits for the task and reply separately.

This is useful when: - Task is background work (BestEffort) but reply is UI update (UserVisible/UserBlocking) - Task has different priority requirements than the reply

Example: 

PostTaskAndReplyWithResultAndTraits(
    backgroundRunner,
    func(ctx context.Context) (*UserData, error) {
        return fetchUserFromDB(ctx)
    },
    TraitsBestEffort(),        // Background work, low priority
    func(ctx context.Context, user *UserData, err error) {
        updateUI(user)
    },
    TraitsUserVisible(),       // UI update, higher priority
    uiRunner,
)

func RegisterHandler added in v0.2.0 

func RegisterHandler[T any](m *JobManager, jobType string, handler TypedHandler[T]) error

RegisterHandler registers a type-safe handler for a job type. The handler will be called with deserialized arguments of type T.

Types

type DefaultLogger added in v0.2.0 

type DefaultLogger struct{}

DefaultLogger is a simple logger implementation using the standard log package

func NewDefaultLogger added in v0.2.0 

func NewDefaultLogger() *DefaultLogger

NewDefaultLogger creates a new DefaultLogger

func (*DefaultLogger) Debug added in v0.2.0 

func (l *DefaultLogger) Debug(msg string, fields ...Field)

Debug logs a debug message

func (*DefaultLogger) Error added in v0.2.0 

func (l *DefaultLogger) Error(msg string, fields ...Field)

Error logs an error message

func (*DefaultLogger) Info added in v0.2.0 

func (l *DefaultLogger) Info(msg string, fields ...Field)

Info logs an info message

func (*DefaultLogger) Warn added in v0.2.0 

func (l *DefaultLogger) Warn(msg string, fields ...Field)

Warn logs a warning message

type DelayManager 

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

func NewDelayManager 

func NewDelayManager() *DelayManager

func (*DelayManager) AddDelayedTask 

func (dm *DelayManager) AddDelayedTask(task Task, delay time.Duration, traits TaskTraits, target TaskRunner)

func (*DelayManager) Stop 

func (dm *DelayManager) Stop()

func (*DelayManager) TaskCount 

func (dm *DelayManager) TaskCount() int

type DelayedTask 

type DelayedTask struct {
	RunAt  time.Time
	Task   Task
	Traits TaskTraits
	Target TaskRunner
	// contains filtered or unexported fields
}

DelayedTask represents a task scheduled for the future

type DelayedTaskHeap 

type DelayedTaskHeap []*DelayedTask

DelayedTaskHeap implements heap.Interface

func (DelayedTaskHeap) Len 

func (h DelayedTaskHeap) Len() int

func (DelayedTaskHeap) Less 

func (h DelayedTaskHeap) Less(i, j int) bool

func (*DelayedTaskHeap) Peek 

func (h *DelayedTaskHeap) Peek() *DelayedTask

func (*DelayedTaskHeap) Pop 

func (h *DelayedTaskHeap) Pop() any

func (*DelayedTaskHeap) Push 

func (h *DelayedTaskHeap) Push(x any)

func (DelayedTaskHeap) Swap 

func (h DelayedTaskHeap) Swap(i, j int)

type ErrorHandler added in v0.2.0 

type ErrorHandler func(jobID string, operation string, err error)

ErrorHandler is called when an IO operation fails after all retries

type FIFOTaskQueue 

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

func NewFIFOTaskQueue 

func NewFIFOTaskQueue() *FIFOTaskQueue

func (*FIFOTaskQueue) Clear 

func (q *FIFOTaskQueue) Clear()

Clear removes all tasks from the queue and releases references

func (*FIFOTaskQueue) IsEmpty 

func (q *FIFOTaskQueue) IsEmpty() bool

func (*FIFOTaskQueue) Len 

func (q *FIFOTaskQueue) Len() int

func (*FIFOTaskQueue) MaybeCompact 

func (q *FIFOTaskQueue) MaybeCompact()

func (*FIFOTaskQueue) PeekTraits 

func (q *FIFOTaskQueue) PeekTraits() (TaskTraits, bool)

func (*FIFOTaskQueue) Pop 

func (q *FIFOTaskQueue) Pop() (TaskItem, bool)

func (*FIFOTaskQueue) PopUpTo 

func (q *FIFOTaskQueue) PopUpTo(max int) []TaskItem

func (*FIFOTaskQueue) Push 

func (q *FIFOTaskQueue) Push(t Task, traits TaskTraits)

type Field added in v0.2.0 

type Field struct {
	Key   string
	Value interface{}
}

Field represents a key-value pair for structured logging

func F added in v0.2.0 

func F(key string, value interface{}) Field

F creates a new Field with the given key and value

type JSONSerializer added in v0.2.0 

type JSONSerializer struct{}

JSONSerializer uses JSON encoding for serialization.

func NewJSONSerializer added in v0.2.0 

func NewJSONSerializer() *JSONSerializer

NewJSONSerializer creates a new JSON serializer

func (*JSONSerializer) Deserialize added in v0.2.0 

func (s *JSONSerializer) Deserialize(data []byte, target any) error

func (*JSONSerializer) Name added in v0.2.0 

func (s *JSONSerializer) Name() string

func (*JSONSerializer) Serialize added in v0.2.0 

func (s *JSONSerializer) Serialize(args any) ([]byte, error)

type JobEntity added in v0.2.0 

type JobEntity struct {
	ID        string
	Type      string
	ArgsData  []byte
	Status    JobStatus
	Result    string
	Priority  int
	CreatedAt time.Time
	UpdatedAt time.Time
}

type JobFilter added in v0.2.0 

type JobFilter struct {
	Status JobStatus // Empty means all
	Type   string    // Empty means all
	Limit  int       // 0 means no limit
	Offset int       // Default 0
}

type JobManager added in v0.2.0 

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

JobManager manages job lifecycle using a three-layer runner architecture: - Layer 1 (controlRunner): Fast control operations (<100μs, pure memory) - Layer 2 (ioRunner): Sequential IO operations (database, file, network) - Layer 3 (executionRunner): User job execution (may be slow/blocking)

func NewJobManager added in v0.2.0 

func NewJobManager(
	controlRunner TaskRunner,
	ioRunner TaskRunner,
	executionRunner TaskRunner,
	store JobStore,
	serializer JobSerializer,
) *JobManager

NewJobManager creates a new JobManager with the three-layer runner architecture

func (*JobManager) CancelJob added in v0.2.0 

func (m *JobManager) CancelJob(id string) error

CancelJob cancels an active job

func (*JobManager) GetActiveJobCount added in v0.2.0 

func (m *JobManager) GetActiveJobCount() int

GetActiveJobCount returns the number of active jobs

func (*JobManager) GetActiveJobs added in v0.2.0 

func (m *JobManager) GetActiveJobs() []*JobEntity

GetActiveJobs returns a snapshot of active jobs

func (*JobManager) GetJob added in v0.2.0 

func (m *JobManager) GetJob(ctx context.Context, id string) (*JobEntity, error)

GetJob retrieves a job by ID

func (*JobManager) GetRetryPolicy added in v0.2.0 

func (m *JobManager) GetRetryPolicy() RetryPolicy

GetRetryPolicy returns the current retry policy

func (*JobManager) ListJobs added in v0.2.0 

func (m *JobManager) ListJobs(ctx context.Context, filter JobFilter) ([]*JobEntity, error)

ListJobs returns jobs matching the filter (allows slight delay from recent writes)

func (*JobManager) SetErrorHandler added in v0.2.0 

func (m *JobManager) SetErrorHandler(handler ErrorHandler)

SetErrorHandler sets a custom error handler for failed IO operations

func (*JobManager) SetLogger added in v0.2.0 

func (m *JobManager) SetLogger(logger Logger)

SetLogger sets the logger for JobManager

func (*JobManager) SetRetryPolicy added in v0.2.0 

func (m *JobManager) SetRetryPolicy(policy RetryPolicy)

SetRetryPolicy sets the retry policy for IO operations

func (*JobManager) Shutdown added in v0.2.0 

func (m *JobManager) Shutdown(ctx context.Context) error

Shutdown gracefully shuts down the JobManager

func (*JobManager) Start added in v0.2.0 

func (m *JobManager) Start(ctx context.Context) error

Start initializes the JobManager and recovers unfinished jobs

func (*JobManager) SubmitDelayedJob added in v0.2.0 

func (m *JobManager) SubmitDelayedJob(
	ctx context.Context,
	id string,
	jobType string,
	args any,
	delay time.Duration,
	traits TaskTraits,
) error

SubmitDelayedJob submits a job with a delay before execution

func (*JobManager) SubmitJob added in v0.2.0 

func (m *JobManager) SubmitJob(ctx context.Context, id string, jobType string, args any, traits TaskTraits) error

SubmitJob submits a job for immediate execution

type JobSerializer added in v0.2.0 

type JobSerializer interface {
	// Serialize converts a Go value to bytes
	Serialize(args any) ([]byte, error)

	// Deserialize converts bytes back to a Go value
	Deserialize(data []byte, target any) error

	// Name returns the serializer name (for debugging/logging)
	Name() string
}

JobSerializer defines the interface for serializing and deserializing job arguments.

type JobStatus added in v0.2.0 

type JobStatus string
const (
	JobStatusPending   JobStatus = "PENDING"
	JobStatusRunning   JobStatus = "RUNNING"
	JobStatusCompleted JobStatus = "COMPLETED"
	JobStatusFailed    JobStatus = "FAILED"
	JobStatusCanceled  JobStatus = "CANCELED"
)

type JobStore added in v0.2.0 

type JobStore interface {
	// SaveJob saves a new job or updates an existing one
	SaveJob(ctx context.Context, job *JobEntity) error

	// UpdateStatus updates the status and result of a job
	UpdateStatus(ctx context.Context, id string, status JobStatus, result string) error

	// GetJob retrieves a job by ID
	GetJob(ctx context.Context, id string) (*JobEntity, error)

	// ListJobs returns jobs matching the filter
	ListJobs(ctx context.Context, filter JobFilter) ([]*JobEntity, error)

	// GetRecoverableJobs returns jobs that should be recovered on startup
	// (typically PENDING jobs)
	GetRecoverableJobs(ctx context.Context) ([]*JobEntity, error)

	// DeleteJob removes a job from storage
	DeleteJob(ctx context.Context, id string) error
}

JobStore defines the interface for job persistence. Implementations can use in-memory storage, databases, or other backends.

type Logger added in v0.2.0 

type Logger interface {
	// Debug logs a debug message with optional fields
	Debug(msg string, fields ...Field)

	// Info logs an info message with optional fields
	Info(msg string, fields ...Field)

	// Warn logs a warning message with optional fields
	Warn(msg string, fields ...Field)

	// Error logs an error message with optional fields
	Error(msg string, fields ...Field)
}

Logger interface for structured logging Implementations can provide custom logging behavior (e.g., integration with logrus, zap, etc.)

type MemoryJobStore added in v0.2.0 

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

MemoryJobStore is an in-memory implementation of JobStore. It uses sync.Map for concurrent-safe storage.

func NewMemoryJobStore added in v0.2.0 

func NewMemoryJobStore() *MemoryJobStore

NewMemoryJobStore creates a new in-memory job store

func (*MemoryJobStore) Clear added in v0.2.0 

func (s *MemoryJobStore) Clear()

Clear removes all jobs from the store (useful for testing)

func (*MemoryJobStore) Count added in v0.2.0 

func (s *MemoryJobStore) Count() int

Count returns the total number of jobs in the store

func (*MemoryJobStore) DeleteJob added in v0.2.0 

func (s *MemoryJobStore) DeleteJob(ctx context.Context, id string) error

func (*MemoryJobStore) GetJob added in v0.2.0 

func (s *MemoryJobStore) GetJob(ctx context.Context, id string) (*JobEntity, error)

func (*MemoryJobStore) GetRecoverableJobs added in v0.2.0 

func (s *MemoryJobStore) GetRecoverableJobs(ctx context.Context) ([]*JobEntity, error)

func (*MemoryJobStore) ListJobs added in v0.2.0 

func (s *MemoryJobStore) ListJobs(ctx context.Context, filter JobFilter) ([]*JobEntity, error)

func (*MemoryJobStore) SaveJob added in v0.2.0 

func (s *MemoryJobStore) SaveJob(ctx context.Context, job *JobEntity) error

func (*MemoryJobStore) UpdateStatus added in v0.2.0 

func (s *MemoryJobStore) UpdateStatus(ctx context.Context, id string, status JobStatus, result string) error

type NoOpLogger added in v0.2.0 

type NoOpLogger struct{}

NoOpLogger is a logger that discards all log messages Useful for tests or when logging is not desired

func NewNoOpLogger added in v0.2.0 

func NewNoOpLogger() *NoOpLogger

NewNoOpLogger creates a new NoOpLogger

func (*NoOpLogger) Debug added in v0.2.0 

func (l *NoOpLogger) Debug(msg string, fields ...Field)

func (*NoOpLogger) Error added in v0.2.0 

func (l *NoOpLogger) Error(msg string, fields ...Field)

func (*NoOpLogger) Info added in v0.2.0 

func (l *NoOpLogger) Info(msg string, fields ...Field)

func (*NoOpLogger) Warn added in v0.2.0 

func (l *NoOpLogger) Warn(msg string, fields ...Field)

type PriorityTaskQueue 

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

func NewPriorityTaskQueue 

func NewPriorityTaskQueue() *PriorityTaskQueue

func (*PriorityTaskQueue) Clear 

func (q *PriorityTaskQueue) Clear()

Clear removes all tasks from the queue and releases references

func (*PriorityTaskQueue) IsEmpty 

func (q *PriorityTaskQueue) IsEmpty() bool

func (*PriorityTaskQueue) Len 

func (q *PriorityTaskQueue) Len() int

func (*PriorityTaskQueue) MaybeCompact 

func (q *PriorityTaskQueue) MaybeCompact()

func (*PriorityTaskQueue) PeekTraits 

func (q *PriorityTaskQueue) PeekTraits() (TaskTraits, bool)

func (*PriorityTaskQueue) Pop 

func (q *PriorityTaskQueue) Pop() (TaskItem, bool)

func (*PriorityTaskQueue) PopUpTo 

func (q *PriorityTaskQueue) PopUpTo(max int) []TaskItem

func (*PriorityTaskQueue) Push 

func (q *PriorityTaskQueue) Push(t Task, traits TaskTraits)

type QueuePolicy added in v0.2.0 

type QueuePolicy int

QueuePolicy defines how to handle full queue situations 

const (
	// QueuePolicyDrop: Silently drop tasks when queue is full (current behavior)
	QueuePolicyDrop QueuePolicy = iota

	// QueuePolicyReject: Call rejection callback when queue is full
	QueuePolicyReject

	// QueuePolicyWait: Block until queue has space or context is done
	QueuePolicyWait
)

type RawJobHandler added in v0.2.0 

type RawJobHandler func(ctx context.Context, args []byte) error

RawJobHandler is the internal handler type that works with raw bytes

type RejectionCallback added in v0.2.0 

type RejectionCallback func(task Task, traits TaskTraits)

RejectionCallback is called when a task is rejected (QueuePolicyReject mode)

type RepeatingTaskHandle 

type RepeatingTaskHandle interface {
	// Stop stops the repeating task. It will not interrupt a currently executing task,
	// but will prevent future executions from being scheduled.
	Stop()

	// IsStopped returns true if the task has been stopped.
	IsStopped() bool
}

RepeatingTaskHandle controls the lifecycle of a repeating task.

type ReplyWithResult 

type ReplyWithResult[T any] func(ctx context.Context, result T, err error)

ReplyWithResult defines a reply callback that receives a result of type T and an error. This is the counterpart to TaskWithResult, receiving the values returned by the task.

type RetryPolicy added in v0.2.0 

type RetryPolicy struct {
	// MaxRetries is the maximum number of retry attempts (0 = no retry, 1 = one retry)
	MaxRetries int

	// InitialDelay is the delay before the first retry
	InitialDelay time.Duration

	// MaxDelay is the maximum delay between retries
	MaxDelay time.Duration

	// BackoffRatio is the multiplier for delay after each retry (e.g., 2.0 for exponential)
	// For example, with InitialDelay=100ms and BackoffRatio=2.0:
	// - Retry 1 delay: 100ms
	// - Retry 2 delay: 200ms
	// - Retry 3 delay: 400ms (capped by MaxDelay)
	BackoffRatio float64
}

RetryPolicy defines retry behavior for IO operations

func DefaultRetryPolicy added in v0.2.0 

func DefaultRetryPolicy() RetryPolicy

DefaultRetryPolicy returns a sensible default retry policy

func NoRetry added in v0.2.0 

func NoRetry() RetryPolicy

NoRetry returns a retry policy with no retries

type SequencedTaskRunner 

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

func NewSequencedTaskRunner 

func NewSequencedTaskRunner(threadPool ThreadPool) *SequencedTaskRunner

func (*SequencedTaskRunner) FlushAsync 

func (r *SequencedTaskRunner) FlushAsync(callback func())

FlushAsync posts a barrier task that executes the callback when all prior tasks complete. This is a non-blocking alternative to WaitIdle.

The callback will be executed on this runner's thread, after all tasks posted before FlushAsync have completed.

Example: 

runner.PostTask(task1)
runner.PostTask(task2)
runner.FlushAsync(func() {
    fmt.Println("task1 and task2 completed!")
})

func (*SequencedTaskRunner) GetThreadPool 

func (r *SequencedTaskRunner) GetThreadPool() ThreadPool

GetThreadPool returns the underlying ThreadPool used by this runner

func (*SequencedTaskRunner) IsClosed 

func (r *SequencedTaskRunner) IsClosed() bool

IsClosed returns true if the runner has been shut down.

func (*SequencedTaskRunner) Metadata 

func (r *SequencedTaskRunner) Metadata() map[string]any

Metadata returns the metadata associated with the task runner

func (*SequencedTaskRunner) Name 

func (r *SequencedTaskRunner) Name() string

Name returns the name of the task runner

func (*SequencedTaskRunner) PostDelayedTask 

func (r *SequencedTaskRunner) PostDelayedTask(task Task, delay time.Duration)

PostDelayedTask submits a task to execute after a delay

func (*SequencedTaskRunner) PostDelayedTaskWithTraits 

func (r *SequencedTaskRunner) PostDelayedTaskWithTraits(task Task, delay time.Duration, traits TaskTraits)

PostDelayedTaskWithTraits submits a delayed task with specified traits

func (*SequencedTaskRunner) PostRepeatingTask 

func (r *SequencedTaskRunner) PostRepeatingTask(task Task, interval time.Duration) RepeatingTaskHandle

PostRepeatingTask submits a task that repeats at a fixed interval

func (*SequencedTaskRunner) PostRepeatingTaskWithInitialDelay 

func (r *SequencedTaskRunner) PostRepeatingTaskWithInitialDelay(
	task Task,
	initialDelay, interval time.Duration,
	traits TaskTraits,
) RepeatingTaskHandle

PostRepeatingTaskWithInitialDelay submits a repeating task with an initial delay The task will first execute after initialDelay, then repeat every interval.

func (*SequencedTaskRunner) PostRepeatingTaskWithTraits 

func (r *SequencedTaskRunner) PostRepeatingTaskWithTraits(
	task Task,
	interval time.Duration,
	traits TaskTraits,
) RepeatingTaskHandle

PostRepeatingTaskWithTraits submits a repeating task with specific traits

func (*SequencedTaskRunner) PostTask 

func (r *SequencedTaskRunner) PostTask(task Task)

PostTask submits a task with default traits

func (*SequencedTaskRunner) PostTaskAndReply 

func (r *SequencedTaskRunner) PostTaskAndReply(task Task, reply Task, replyRunner TaskRunner)

PostTaskAndReply executes task on this runner, then posts reply to replyRunner. If task panics, reply will not be executed.

func (*SequencedTaskRunner) PostTaskAndReplyWithTraits 

func (r *SequencedTaskRunner) PostTaskAndReplyWithTraits(
	task Task,
	taskTraits TaskTraits,
	reply Task,
	replyTraits TaskTraits,
	replyRunner TaskRunner,
)

PostTaskAndReplyWithTraits allows specifying different traits for task and reply. This is useful when task is background work (BestEffort) but reply is UI update (UserVisible).

func (*SequencedTaskRunner) PostTaskWithTraits 

func (r *SequencedTaskRunner) PostTaskWithTraits(task Task, traits TaskTraits)

PostTaskWithTraits submits a task with specified traits

func (*SequencedTaskRunner) SetMetadata 

func (r *SequencedTaskRunner) SetMetadata(key string, value interface{})

SetMetadata sets a metadata key-value pair

func (*SequencedTaskRunner) SetName 

func (r *SequencedTaskRunner) SetName(name string)

SetName sets the name of the task runner

func (*SequencedTaskRunner) Shutdown 

func (r *SequencedTaskRunner) Shutdown()

Shutdown gracefully stops the runner by: 1. Marking it as closed (stops accepting new tasks) 2. Clearing all pending tasks in the queue 3. All repeating tasks will automatically stop on their next execution 4. Signaling all WaitShutdown() waiters

Note: This method is non-blocking and can be safely called from within a task. Note: This will not interrupt currently executing tasks.

func (*SequencedTaskRunner) WaitIdle 

func (r *SequencedTaskRunner) WaitIdle(ctx context.Context) error

WaitIdle blocks until all currently queued tasks have completed execution. This is implemented by posting a barrier task and waiting for it to execute.

Due to the sequential nature of SequencedTaskRunner, when the barrier task executes, all tasks posted before WaitIdle are guaranteed to have completed.

Returns error if: - Context is cancelled or deadline exceeded - Runner is closed when WaitIdle is called

Note: Tasks posted after WaitIdle is called are not waited for. Note: Repeating tasks will continue to repeat and are not waited for.

func (*SequencedTaskRunner) WaitShutdown 

func (r *SequencedTaskRunner) WaitShutdown(ctx context.Context) error

WaitShutdown blocks until Shutdown() is called on this runner.

This is useful for waiting for the runner to be shut down, either by an external caller or by a task running on the runner itself.

Returns error if context is cancelled or deadline exceeded.

Example: 

// Task shuts down the runner when condition is met
runner.PostTask(func(ctx context.Context) {
    if conditionMet() {
        me := GetCurrentTaskRunner(ctx)
        me.Shutdown()
    }
})

// Main thread waits for shutdown
runner.WaitShutdown(context.Background())

type SingleThreadTaskRunner 

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

SingleThreadTaskRunner binds a dedicated Goroutine to execute tasks sequentially. It guarantees that all tasks submitted to it run on the same Goroutine (Thread Affinity).

Use cases: 1. Blocking IO operations (e.g., NetworkReceiver) 2. CGO calls that require Thread Local Storage 3. Simulating Main Thread / UI Thread behavior

Key differences from SequencedTaskRunner: - SequencedTaskRunner: Tasks execute sequentially but may run on different worker goroutines - SingleThreadTaskRunner: Tasks execute sequentially AND always on the same dedicated goroutine

func NewSingleThreadTaskRunner 

func NewSingleThreadTaskRunner() *SingleThreadTaskRunner

NewSingleThreadTaskRunner creates and starts a new SingleThreadTaskRunner. It immediately spawns a dedicated goroutine for task execution.

func (*SingleThreadTaskRunner) FlushAsync 

func (r *SingleThreadTaskRunner) FlushAsync(callback func())

FlushAsync posts a barrier task that executes the callback when all prior tasks complete. This is a non-blocking alternative to WaitIdle.

The callback will be executed on this runner's dedicated goroutine, after all tasks posted before FlushAsync have completed.

Example: 

runner.PostTask(task1)
runner.PostTask(task2)
runner.FlushAsync(func() {
    fmt.Println("task1 and task2 completed!")
})

func (*SingleThreadTaskRunner) GetQueuePolicy added in v0.2.0 

func (r *SingleThreadTaskRunner) GetQueuePolicy() QueuePolicy

GetQueuePolicy returns the current queue policy

func (*SingleThreadTaskRunner) GetThreadPool 

func (r *SingleThreadTaskRunner) GetThreadPool() ThreadPool

GetThreadPool returns nil because SingleThreadTaskRunner doesn't use a thread pool

func (*SingleThreadTaskRunner) IsClosed 

func (r *SingleThreadTaskRunner) IsClosed() bool

IsClosed returns true if the runner has been stopped

func (*SingleThreadTaskRunner) Metadata 

func (r *SingleThreadTaskRunner) Metadata() map[string]interface{}

Metadata returns the metadata associated with the task runner

func (*SingleThreadTaskRunner) Name 

func (r *SingleThreadTaskRunner) Name() string

Name returns the name of the task runner

func (*SingleThreadTaskRunner) PostDelayedTask 

func (r *SingleThreadTaskRunner) PostDelayedTask(task Task, delay time.Duration)

PostDelayedTask submits a delayed task

func (*SingleThreadTaskRunner) PostDelayedTaskWithTraits 

func (r *SingleThreadTaskRunner) PostDelayedTaskWithTraits(task Task, delay time.Duration, traits TaskTraits)

PostDelayedTaskWithTraits submits a delayed task with traits. Uses time.AfterFunc which is independent of the global TaskScheduler, ensuring IO-related timers are not affected by scheduler load.

func (*SingleThreadTaskRunner) PostRepeatingTask 

func (r *SingleThreadTaskRunner) PostRepeatingTask(task Task, interval time.Duration) RepeatingTaskHandle

PostRepeatingTask submits a task that repeats at a fixed interval

func (*SingleThreadTaskRunner) PostRepeatingTaskWithInitialDelay 

func (r *SingleThreadTaskRunner) PostRepeatingTaskWithInitialDelay(
	task Task,
	initialDelay, interval time.Duration,
	traits TaskTraits,
) RepeatingTaskHandle

PostRepeatingTaskWithInitialDelay submits a repeating task with an initial delay

func (*SingleThreadTaskRunner) PostRepeatingTaskWithTraits 

func (r *SingleThreadTaskRunner) PostRepeatingTaskWithTraits(
	task Task,
	interval time.Duration,
	traits TaskTraits,
) RepeatingTaskHandle

PostRepeatingTaskWithTraits submits a repeating task with traits

func (*SingleThreadTaskRunner) PostTask 

func (r *SingleThreadTaskRunner) PostTask(task Task)

PostTask submits a task for execution

func (*SingleThreadTaskRunner) PostTaskAndReply 

func (r *SingleThreadTaskRunner) PostTaskAndReply(task Task, reply Task, replyRunner TaskRunner)

PostTaskAndReply executes task on this runner, then posts reply to replyRunner. If task panics, reply will not be executed. Both task and reply will execute on the same dedicated goroutine if replyRunner is this runner.

func (*SingleThreadTaskRunner) PostTaskAndReplyWithTraits 

func (r *SingleThreadTaskRunner) PostTaskAndReplyWithTraits(
	task Task,
	taskTraits TaskTraits,
	reply Task,
	replyTraits TaskTraits,
	replyRunner TaskRunner,
)

PostTaskAndReplyWithTraits allows specifying different traits for task and reply. This is useful when task is background work (BestEffort) but reply is UI update (UserVisible). Note: For SingleThreadTaskRunner, traits don't affect execution order since all tasks run sequentially on the same goroutine, but they may be used for logging or metrics.

func (*SingleThreadTaskRunner) PostTaskWithTraits 

func (r *SingleThreadTaskRunner) PostTaskWithTraits(task Task, traits TaskTraits)

PostTaskWithTraits submits a task with traits (traits are ignored for single-threaded execution) The behavior when the queue is full depends on the configured QueuePolicy: - QueuePolicyDrop: Silently drops the task (default) - QueuePolicyReject: Calls the rejection callback if set - QueuePolicyWait: Blocks until queue has space or context is done

func (*SingleThreadTaskRunner) RejectedCount added in v0.2.0 

func (r *SingleThreadTaskRunner) RejectedCount() int64

RejectedCount returns the number of tasks that have been rejected due to full queue Only incremented when QueuePolicy is QueuePolicyReject

func (*SingleThreadTaskRunner) SetMetadata 

func (r *SingleThreadTaskRunner) SetMetadata(key string, value interface{})

SetMetadata sets a metadata key-value pair

func (*SingleThreadTaskRunner) SetName 

func (r *SingleThreadTaskRunner) SetName(name string)

SetName sets the name of the task runner

func (*SingleThreadTaskRunner) SetQueuePolicy added in v0.2.0 

func (r *SingleThreadTaskRunner) SetQueuePolicy(policy QueuePolicy)

SetQueuePolicy sets the policy for handling full queue situations

func (*SingleThreadTaskRunner) SetRejectionCallback added in v0.2.0 

func (r *SingleThreadTaskRunner) SetRejectionCallback(callback RejectionCallback)

SetRejectionCallback sets the callback to be called when a task is rejected Only used when QueuePolicy is set to QueuePolicyReject

func (*SingleThreadTaskRunner) Shutdown 

func (r *SingleThreadTaskRunner) Shutdown()

Shutdown marks the runner as closed and signals shutdown waiters. Unlike Stop(), this method does NOT immediately terminate the runLoop. This allows tasks to call Shutdown() from within themselves.

After calling Shutdown(): - WaitShutdown() will return - IsClosed() will return true - New tasks posted will be ignored - Existing queued tasks will still execute - Call Stop() to actually terminate the runLoop

func (*SingleThreadTaskRunner) Stop 

func (r *SingleThreadTaskRunner) Stop()

Stop stops the runner and releases resources

func (*SingleThreadTaskRunner) WaitIdle 

func (r *SingleThreadTaskRunner) WaitIdle(ctx context.Context) error

WaitIdle blocks until all currently queued tasks have completed execution. This is implemented by posting a barrier task and waiting for it to execute.

Since SingleThreadTaskRunner executes tasks sequentially on a dedicated goroutine, when the barrier task executes, all tasks posted before WaitIdle are guaranteed to have completed.

Returns error if: - Context is cancelled or deadline exceeded - Runner is closed when WaitIdle is called

Note: Tasks posted after WaitIdle is called are not waited for. Note: Repeating tasks will continue to repeat and are not waited for.

func (*SingleThreadTaskRunner) WaitShutdown 

func (r *SingleThreadTaskRunner) WaitShutdown(ctx context.Context) error

WaitShutdown blocks until Shutdown() is called on this runner.

This is useful for waiting for the runner to be shut down, either by an external caller or by a task running on the runner itself.

Returns error if context is cancelled or deadline exceeded.

Example: 

// IO thread: receives messages and posts shutdown when condition met
ioRunner.PostTask(func(ctx context.Context) {
    for {
        msg := receiveMessage()
        mainRunner.PostTask(func(ctx context.Context) {
            if shouldShutdown(msg) {
                me := GetCurrentTaskRunner(ctx)
                me.Shutdown()
            }
        })
    }
})

// Main thread waits for shutdown
mainRunner.WaitShutdown(context.Background())

type Task 

type Task func(ctx context.Context)

Task is the unit of work (Closure)

type TaskItem 

type TaskItem struct {
	Task   Task
	Traits TaskTraits
}

type TaskPriority 

type TaskPriority int
const (
	// TaskPriorityBestEffort: Lowest priority
	TaskPriorityBestEffort TaskPriority = iota

	// TaskPriorityUserVisible: Default priority
	TaskPriorityUserVisible

	// TaskPriorityUserBlocking: Highest priority
	// `UserBlocking` means the task may block the main thread.
	// If main thread is blocked, the UI will be unresponsive.
	// The user experience will be affected if the task blocks the main thread.
	TaskPriorityUserBlocking
)

type TaskQueue 

type TaskQueue interface {
	Push(t Task, traits TaskTraits)
	Pop() (TaskItem, bool)
	PopUpTo(max int) []TaskItem
	PeekTraits() (TaskTraits, bool)
	Len() int
	IsEmpty() bool
	MaybeCompact()
	Clear() // Clear all tasks from the queue
}

TaskQueue defines the interface for different queue implementations

type TaskRunner 

type TaskRunner interface {
	PostTask(task Task)
	PostTaskWithTraits(task Task, traits TaskTraits)
	PostDelayedTask(task Task, delay time.Duration)

	// [v2.1 New] Support delayed tasks with specific traits
	PostDelayedTaskWithTraits(task Task, delay time.Duration, traits TaskTraits)

	// [v2.2 New] Support repeating tasks
	PostRepeatingTask(task Task, interval time.Duration) RepeatingTaskHandle
	PostRepeatingTaskWithTraits(task Task, interval time.Duration, traits TaskTraits) RepeatingTaskHandle
	PostRepeatingTaskWithInitialDelay(task Task, initialDelay, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

	// [v2.3 New] Support task and reply pattern
	// PostTaskAndReply executes task on this runner, then posts reply to replyRunner
	PostTaskAndReply(task Task, reply Task, replyRunner TaskRunner)
	// PostTaskAndReplyWithTraits allows specifying traits for both task and reply
	PostTaskAndReplyWithTraits(task Task, taskTraits TaskTraits, reply Task, replyTraits TaskTraits, replyRunner TaskRunner)

	// [v2.4 New] Synchronization and lifecycle management
	// WaitIdle blocks until all currently queued tasks have completed execution
	// Tasks posted after WaitIdle is called are not waited for
	// Returns error if context is cancelled or runner is closed
	WaitIdle(ctx context.Context) error

	// FlushAsync posts a barrier task that executes callback when all prior tasks complete
	// This is a non-blocking alternative to WaitIdle
	FlushAsync(callback func())

	// WaitShutdown blocks until Shutdown() is called on this runner
	// Returns error if context is cancelled
	WaitShutdown(ctx context.Context) error

	// Shutdown marks the runner as closed and clears all pending tasks
	// This method is non-blocking and can be safely called from within a task
	Shutdown()

	// IsClosed returns true if the runner has been shut down
	IsClosed() bool

	// [v2.5 New] Identification and Metadata
	// Name returns the name of the task runner
	Name() string
	// Metadata returns the metadata associated with the task runner
	Metadata() map[string]any

	// [v2.6 New] Thread Pool Access
	// GetThreadPool returns the underlying ThreadPool used by this runner
	// Returns nil for runners that don't use a thread pool (e.g., SingleThreadTaskRunner)
	GetThreadPool() ThreadPool
}

func GetCurrentTaskRunner 

func GetCurrentTaskRunner(ctx context.Context) TaskRunner

type TaskScheduler 

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

func NewFIFOTaskScheduler 

func NewFIFOTaskScheduler(workerCount int) *TaskScheduler

func NewPriorityTaskScheduler 

func NewPriorityTaskScheduler(workerCount int) *TaskScheduler

func (*TaskScheduler) ActiveTaskCount 

func (s *TaskScheduler) ActiveTaskCount() int

func (*TaskScheduler) DelayedTaskCount 

func (s *TaskScheduler) DelayedTaskCount() int

func (*TaskScheduler) GetWork 

func (s *TaskScheduler) GetWork(stopCh <-chan struct{}) (Task, bool)

GetWork (Called by Worker)

func (*TaskScheduler) OnTaskEnd 

func (s *TaskScheduler) OnTaskEnd()

func (*TaskScheduler) OnTaskStart 

func (s *TaskScheduler) OnTaskStart()

func (*TaskScheduler) PostDelayedInternal 

func (s *TaskScheduler) PostDelayedInternal(task Task, delay time.Duration, traits TaskTraits, target TaskRunner)

PostDelayedInternal

func (*TaskScheduler) PostInternal 

func (s *TaskScheduler) PostInternal(task Task, traits TaskTraits)

PostInternal

func (*TaskScheduler) QueuedTaskCount 

func (s *TaskScheduler) QueuedTaskCount() int

func (*TaskScheduler) Shutdown 

func (s *TaskScheduler) Shutdown()

func (*TaskScheduler) ShutdownGraceful added in v0.2.0 

func (s *TaskScheduler) ShutdownGraceful(timeout time.Duration) error

ShutdownGraceful waits for all queued and active tasks to complete Returns error if timeout is exceeded before tasks complete

func (*TaskScheduler) WorkerCount 

func (s *TaskScheduler) WorkerCount() int

Metrics

type TaskTraits 

type TaskTraits struct {
	Priority TaskPriority
}

func DefaultTaskTraits 

func DefaultTaskTraits() TaskTraits

func TraitsBestEffort 

func TraitsBestEffort() TaskTraits

func TraitsUserBlocking 

func TraitsUserBlocking() TaskTraits

func TraitsUserVisible 

func TraitsUserVisible() TaskTraits

type TaskWithResult 

type TaskWithResult[T any] func(ctx context.Context) (T, error)

TaskWithResult defines a task that returns a result of type T and an error. This is used with PostTaskAndReplyWithResult to pass data from task to reply.

type ThreadPool 

type ThreadPool interface {
	PostInternal(task Task, traits TaskTraits)
	PostDelayedInternal(task Task, delay time.Duration, traits TaskTraits, target TaskRunner)

	Start(ctx context.Context)
	Stop()

	ID() string
	IsRunning() bool

	WorkerCount() int
	QueuedTaskCount() int  // In queue
	ActiveTaskCount() int  // Executing
	DelayedTaskCount() int // Delayed
}

============================================================================= ThreadPool: Define task execution interface =============================================================================

type TypedHandler added in v0.2.0 

type TypedHandler[T any] func(ctx context.Context, args T) error

TypedHandler is a generic handler type for type-safe job handlers

type WorkSource 

type WorkSource interface {
	GetWork(stopCh <-chan struct{}) (Task, bool)
}

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.