config

Documentation

Index

• Constants
• Variables
• func MustValidateErrorStrategy(strategy ErrorStrategy)
• func ValidateErrorStrategy(strategy ErrorStrategy) error
• type ErrorStrategy
  • func (s ErrorStrategy) String() string
• type Option
  • func WithCoordinator(coordinator *coordinatorinternal.Coordinator) Option
  • func WithDynamicWorkers(controller *control.Controller) Option
  • func WithErrorStrategy(strategy ErrorStrategy) Option
  • func WithForcedDynamicWorkers(controller *control.Controller) Option
  • func WithInterruptibleWorkers(controller *control.Controller) Option
  • func WithLinkMetricsObserver(observer link.MetricsObserver) Option
  • func WithStageBudget(budget coordinatorinternal.StageBudget) Option
  • func WithStageMetricsObserver(observer metrics.StageMetricsObserver) Option
  • func WithStageName(name string) Option
  • func WithUnlimitedWorkers() Option
  • func WithWorkers(workers int) Option
• type Options
  • func BuildOptions(opts ...Option) *Options
  • func NewOptions() *Options
  • func (o *Options) Controller() *control.Controller
  • func (o *Options) Coordinator() *coordinatorinternal.Coordinator
  • func (o *Options) ErrorStrategy() ErrorStrategy
  • func (o *Options) Interruptible() bool
  • func (o *Options) LinkMetricsObserver() link.MetricsObserver
  • func (o *Options) ResolveStageName() string
  • func (o *Options) StageBudget() coordinatorinternal.StageBudget
  • func (o *Options) StageMetricsObserver() metrics.StageMetricsObserver
  • func (o *Options) StageName() string
  • func (o *Options) Unlimited() bool
  • func (o *Options) Workers() int

Constants

const (
	DefaultWorkers        = 16
	MinWorkers            = 1
	UnlimitedWorkerBuffer = 4096
)

Variables

var (
	// ErrNilController reports that a dynamic-worker option received a nil
	// concurrency controller.
	ErrNilController = errors.New("flx: nil concurrency controller")
	// ErrInterruptibleWorkersRequireContextTransform reports that forced dynamic
	// workers were requested for a transform that does not accept a context.
	ErrInterruptibleWorkersRequireContextTransform = errors.New("flx: WithInterruptibleWorkers/WithForcedDynamicWorkers requires MapContext/FlatMapContext")
)

var ErrInvalidErrorStrategy = errors.New("flx: invalid error strategy")

ErrInvalidErrorStrategy reports that an operation received an unsupported worker error handling mode.

Functions

func MustValidateErrorStrategy

func MustValidateErrorStrategy(strategy ErrorStrategy)

MustValidateErrorStrategy panics when strategy is unsupported.

func ValidateErrorStrategy

func ValidateErrorStrategy(strategy ErrorStrategy) error

ValidateErrorStrategy returns an error when strategy is unsupported.

Types

type ErrorStrategy

type ErrorStrategy uint8

ErrorStrategy controls how concurrent operations react to worker failures.

const (
	// ErrorStrategyFailFast cancels the operation and makes non-Err terminals
	// panic once the failure is observed.
	ErrorStrategyFailFast ErrorStrategy = iota
	// ErrorStrategyCollect records worker errors and returns them from Err
	// terminals without canceling sibling workers.
	ErrorStrategyCollect
	// ErrorStrategyContinue ignores worker errors and allows the operation to
	// continue without recording them in stream state.
	ErrorStrategyContinue
	// ErrorStrategyLogAndContinue is a deprecated compatibility alias for
	// ErrorStrategyContinue.
	//
	// Deprecated: use ErrorStrategyContinue.
	ErrorStrategyLogAndContinue = ErrorStrategyContinue
)

func (ErrorStrategy) String

func (s ErrorStrategy) String() string

String returns a human-readable representation of s.

type Option

type Option func(*Options)

Option mutates the execution settings for one transform or terminal call.

func WithCoordinator

func WithCoordinator(coordinator *coordinatorinternal.Coordinator) Option

WithCoordinator registers one pipeline coordinator to receive runtime snapshots for this operation.

func WithDynamicWorkers

func WithDynamicWorkers(controller *control.Controller) Option

WithDynamicWorkers enables graceful dynamic resizing for the current operation. Shrinking does not interrupt workers that already hold a slot.

func WithErrorStrategy

func WithErrorStrategy(strategy ErrorStrategy) Option

WithErrorStrategy configures how worker panics and errors are handled for the current operation.

func WithForcedDynamicWorkers

func WithForcedDynamicWorkers(controller *control.Controller) Option

WithForcedDynamicWorkers enables forced dynamic resizing for the current operation. Shrinking cancels excess workers via context and only works with MapContext or FlatMapContext variants.

func WithInterruptibleWorkers

func WithInterruptibleWorkers(controller *control.Controller) Option

WithInterruptibleWorkers is a compatibility alias for WithForcedDynamicWorkers.

func WithLinkMetricsObserver

func WithLinkMetricsObserver(observer link.MetricsObserver) Option

WithLinkMetricsObserver registers the callback that receives link metrics.

func WithStageBudget

func WithStageBudget(budget coordinatorinternal.StageBudget) Option

WithStageBudget constrains one coordinator-managed dynamic stage to a worker budget window.

func WithStageMetricsObserver

func WithStageMetricsObserver(observer metrics.StageMetricsObserver) Option

WithStageMetricsObserver registers the callback that receives stage metrics.

func WithStageName

func WithStageName(name string) Option

WithStageName labels the current transform or terminal operation as one named stage.

func WithUnlimitedWorkers

func WithUnlimitedWorkers() Option

WithUnlimitedWorkers spawns one worker per item for the current operation.

func WithWorkers

func WithWorkers(workers int) Option

WithWorkers limits the current operation to a fixed number of workers.

type Options

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

Options stores execution settings for one transform or terminal operation.

func BuildOptions

func BuildOptions(opts ...Option) *Options

BuildOptions applies opts in order on top of the default settings.

func NewOptions

func NewOptions() *Options

NewOptions returns the default operation settings.

func (*Options) Controller

func (o *Options) Controller() *control.Controller

Controller returns the dynamic concurrency controller, if configured.

func (*Options) Coordinator

func (o *Options) Coordinator() *coordinatorinternal.Coordinator

Coordinator returns the configured pipeline coordinator, if any.

func (*Options) ErrorStrategy

func (o *Options) ErrorStrategy() ErrorStrategy

ErrorStrategy returns the worker failure strategy for the operation.

func (*Options) Interruptible

func (o *Options) Interruptible() bool

Interruptible reports whether dynamic worker shrink should cancel workers.

func (*Options) LinkMetricsObserver

func (o *Options) LinkMetricsObserver() link.MetricsObserver

LinkMetricsObserver returns the configured observer for link metrics.

func (*Options) ResolveStageName

func (o *Options) ResolveStageName() string

ResolveStageName returns the stable stage identity for this operation, generating one when a coordinator is attached and no explicit name was set.

func (*Options) StageBudget

func (o *Options) StageBudget() coordinatorinternal.StageBudget

StageBudget returns the configured worker budget for one stage.

func (*Options) StageMetricsObserver

func (o *Options) StageMetricsObserver() metrics.StageMetricsObserver

StageMetricsObserver returns the configured observer for stage metrics.

func (*Options) StageName

func (o *Options) StageName() string

StageName returns the configured stage identity.

func (*Options) Unlimited

func (o *Options) Unlimited() bool

Unlimited reports whether the operation should spawn one worker per item.

func (*Options) Workers

func (o *Options) Workers() int

Workers returns the configured fixed worker count.