Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

toc

package module

Go to main page

Versions in this module

v0

v0.11.0

Mar 30, 2026

v0.10.0

Mar 30, 2026

v0.9.0

Mar 29, 2026

v0.8.0

Mar 29, 2026

v0.7.0

Mar 29, 2026

Changes in this version

type IntervalStats

+ MeanSojournTime time.Duration

+ SojournTimeDelta time.Duration

type Stats

+ SojournTime time.Duration

v0.6.0

Mar 29, 2026

Changes in this version

type LimitSource

+ const LimitSourceMemoryLimiter

+ type MemoryLimiterHandle struct

+ func MemoryLimiter(pipeline *Pipeline, drum string, limits *LimitManager, budgetFraction float64, ...) *MemoryLimiterHandle

+ func (h *MemoryLimiterHandle) Callback() func(context.Context, memctl.MemInfo)

+ func (h *MemoryLimiterHandle) Stats() MemoryLimiterStats

+ type MemoryLimiterStats struct

+ Adjustments int64

+ Applied int64

+ Budget int64

+ Headroom int64

+ Weight int64

v0.5.1

Mar 29, 2026

v0.5.0

Mar 29, 2026

Changes in this version

type Pipeline

+ func (p *Pipeline) ResolveSegment(controlStage, drum string) ([]string, error)

+ type ResolvedTarget struct

+ ControlStage string

+ Drum string

+ Segment []string

+ type RopeDrumConfig struct

+ ControlStage string

+ Drum string

+ type RopeHandle struct

+ func (h *RopeHandle) Done() <-chan struct{}

+ func (h *RopeHandle) Drum() string

+ func (h *RopeHandle) Stop()

+ type RopeResolver struct

+ func NewRopeResolver(p RopeResolverParams) (*RopeResolver, error)

+ type RopeResolverParams struct

+ Drums []RopeDrumConfig

+ Interval time.Duration

+ Limits *LimitManager

+ Options []RopeOption

+ Pipeline *Pipeline

+ SourceID string

+ StageSnapshot func(string) IntervalStats

+ WeightMode bool

+ type RopeStarter interface

+ Start func(target ResolvedTarget) (*RopeHandle, error)

+ type RopeSupervisor struct

+ func NewRopeSupervisor(resolver *RopeResolver, starter RopeStarter, opts ...RopeSupervisorOption) *RopeSupervisor

+ func (rs *RopeSupervisor) ClearDesired()

+ func (rs *RopeSupervisor) SetIdentified(drum string)

+ func (rs *RopeSupervisor) Status() RopeSupervisorStatus

+ func (rs *RopeSupervisor) Stop()

+ type RopeSupervisorOption func(*RopeSupervisor)

+ func WithRopeSupervisorLogger(l *log.Logger) RopeSupervisorOption

+ type RopeSupervisorStatus struct

+ Active bool

+ Drum string

type WeightedBatcherStats

+ DroppedWeight int64

+ EmittedWeight int64

+ ReceivedWeight int64

v0.4.0

Mar 29, 2026

Changes in this version

type OversizePolicy

+ const OversizeAllow

type Stats

+ OversizeAdmissions int64

v0.3.0

Mar 28, 2026

Changes in this version

type IntervalStats

+ CompletedWeightDelta int64

type RopeOption

+ func WithControlStage(name string) RopeOption

type Stats

+ CompletedWeight int64

v0.2.2

Mar 28, 2026

v0.2.1

Mar 28, 2026

v0.2.0

Mar 27, 2026

Changes in this version

+ func PanicRecovery[T, R any](panicked *atomic.Int64) call.Decorator[T, R]

+ func ServiceTiming[T, R any](serviceNs *atomic.Int64, onDuration func(time.Duration)) call.Decorator[T, R]

type IntervalStats

+ StarvedTimeDelta time.Duration

type Stats

+ StarvedTime time.Duration

v0.1.0

Mar 25, 2026

Changes in this version

+ var ErrClosed = errors.New("toc: stage closed")

+ var ErrStopping = errors.New("toc: stage is stopping")

+ var ErrWeightExceedsLimit = errors.New("toc: item weight exceeds MaxWIPWeight")

+ func Adapt(name string, prev, curr Stats, elapsed time.Duration) core.StageObservation

+ func BufferCapacity(throughput float64, protectionTime time.Duration) int

+ func FromChan[T any](ch <-chan T) <-chan rslt.Result[T]

+ func PublishOnSnapshot(ctx context.Context, pub ObservationPublisher, onError PublishErrorHandler) func(PipelineSnapshot)

+ type Batcher struct

+ func NewBatcher[T any](ctx context.Context, src <-chan rslt.Result[T], n int) *Batcher[T]

+ func (b *Batcher[T]) Out() <-chan rslt.Result[[]T]

+ func (b *Batcher[T]) Stats() BatcherStats

+ func (b *Batcher[T]) Wait() error

+ type BatcherStats struct

+ BatchCount int64

+ BufferedDepth int64

+ Dropped int64

+ Emitted int64

+ Forwarded int64

+ OutputBlockedTime time.Duration

+ Received int64

+ func (s BatcherStats) ToStats() Stats

+ type BudgetAllocatorHandle struct

+ func BudgetAllocator(stages []StageAllocation, drumName string, budgetFraction float64, ...) *BudgetAllocatorHandle

+ func (h *BudgetAllocatorHandle) Callback() func(context.Context, memctl.MemInfo)

+ type BufferZone int

+ const BufferGreen

+ const BufferRed

+ const BufferYellow

+ func (z BufferZone) String() string

+ type DiagnosisHandler func(ctx context.Context, msg DiagnosisMessage) error

+ type DiagnosisMessage struct

+ Diagnosis core.Diagnosis

+ PipelineID string

+ TimestampUnixNano int64

+ type DiagnosisPublisher interface

+ PublishDiagnosis func(ctx context.Context, msg DiagnosisMessage) error

+ type DiagnosisSubscriber interface

+ SubscribeAllDiagnosis func(ctx context.Context, fn DiagnosisHandler) (Subscription, error)

+ SubscribeDiagnosis func(ctx context.Context, pipelineID string, fn DiagnosisHandler) (Subscription, error)

+ type FocusingStep int

+ const StepElevate

+ const StepExploit

+ const StepIdentify

+ const StepPreventInertia

+ const StepSubordinate

+ func ClassifyStep(prevConstraint string, currConstraint string, ropeActive bool, ...) FocusingStep

+ func (s FocusingStep) String() string

+ type IncomingEdge struct

+ From string

+ Ratio int

+ type IntervalStats struct

+ ApproxUtilization float64

+ ArrivalRate float64

+ CurrActiveWorkers int

+ CurrBufferPenetration float64

+ CurrBufferedDepth int64

+ CurrQueueCapacity int

+ CurrTargetWorkers int

+ Duration time.Duration

+ ErrorRate float64

+ Goodput float64

+ IdleTimeDelta time.Duration

+ ItemsCanceled int64

+ ItemsCompleted int64

+ ItemsFailed int64

+ ItemsSubmitted int64

+ MeanServiceTime time.Duration

+ OutputBlockedDelta time.Duration

+ QueueGrowthRate float64

+ ResetDetected bool

+ ServiceTimeDelta time.Duration

+ Throughput float64

+ func Delta(prev, curr Stats, elapsed time.Duration) IntervalStats

+ func (s IntervalStats) BufferZone(yellowAt, redAt float64) BufferZone

+ type Join struct

+ func NewJoin[A, B, R any](ctx context.Context, srcA <-chan rslt.Result[A], srcB <-chan rslt.Result[B], ...) *Join[R]

+ func (j *Join[R]) Out() <-chan rslt.Result[R]

+ func (j *Join[R]) Stats() JoinStats

+ func (j *Join[R]) Wait() error

+ type JoinStats struct

+ Combined int64

+ DiscardedA int64

+ DiscardedB int64

+ Errors int64

+ ExtraA int64

+ ExtraB int64

+ OutputBlockedTime time.Duration

+ ReceivedA int64

+ ReceivedB int64

+ type LimitHandle struct

+ func (h *LimitHandle) Close()

+ func (h *LimitHandle) ProposeCount(limit int)

+ func (h *LimitHandle) ProposeWeight(limit int64)

+ type LimitManager struct

+ func NewLimitManager(setCount func(int) int, setWeight func(int64) int64, defaultCount int, ...) *LimitManager

+ func (m *LimitManager) Effective() LimitSnapshot

+ func (m *LimitManager) ProposeCount(source string, limit int)

+ func (m *LimitManager) ProposeWeight(source string, limit int64)

+ func (m *LimitManager) Register(source LimitSource) *LimitHandle

+ func (m *LimitManager) WithdrawCount(source string)

+ func (m *LimitManager) WithdrawWeight(source string)

+ type LimitSnapshot struct

+ AppliedCount int

+ AppliedWeight int64

+ CountProposals map[string]int

+ CountSource string

+ CountSources int

+ EffectiveCount int

+ EffectiveWeight int64

+ WeightProposals map[string]int64

+ WeightSource string

+ WeightSources int

+ type LimitSource = string

+ const LimitSourceBudgetAllocator

+ const LimitSourceMemoryRope

+ const LimitSourceProcessingRope

+ const LimitSourceWeightRope

+ type MemoryFeverZone int

+ const MemoryGreen

+ const MemoryRed

+ const MemoryUnknown

+ const MemoryYellow

+ func MemoryFever(headroom, limit uint64, yellowAt, redAt float64) MemoryFeverZone

+ func (z MemoryFeverZone) String() string

+ type MemoryRopeHandle struct

+ func MemoryRope(pipeline *Pipeline, drum string, limits *LimitManager, budgetFraction float64, ...) *MemoryRopeHandle

+ func (h *MemoryRopeHandle) Callback() func(context.Context, memctl.MemInfo)

+ func (h *MemoryRopeHandle) Stats() MemoryRopeStats

+ type MemoryRopeStats struct

+ Adjustments int64

+ Applied int64

+ Budget int64

+ Headroom int64

+ Weight int64

+ type Merge struct

+ func NewMerge[T any](ctx context.Context, sources ...<-chan rslt.Result[T]) *Merge[T]

+ func (m *Merge[T]) Out() <-chan rslt.Result[T]

+ func (m *Merge[T]) Stats() MergeStats

+ func (m *Merge[T]) Wait() error

+ type MergeStats struct

+ Dropped int64

+ Forwarded int64

+ Received int64

+ SourceDropped []int64

+ SourceForwarded []int64

+ SourceReceived []int64

+ type MissingResultError struct

+ Source string

+ func (e *MissingResultError) Error() string

+ type ObservationBatch struct

+ Observations []core.StageObservation

+ PipelineID string

+ TimestampUnixNano int64

+ WindowDurationNano int64

+ type ObservationHandler func(ctx context.Context, batch ObservationBatch) error

+ type ObservationPublisher interface

+ PublishObservations func(ctx context.Context, batch ObservationBatch) error

+ type ObservationSubscriber interface

+ SubscribeAllObservations func(ctx context.Context, fn ObservationHandler) (Subscription, error)

+ SubscribeObservations func(ctx context.Context, pipelineID string, fn ObservationHandler) (Subscription, error)

+ type Observer struct

+ func NewObserver(pipelineID string) *Observer

+ func (o *Observer) AddStage(s ObserverStage)

+ func (o *Observer) Run(ctx context.Context, interval time.Duration, fn func(PipelineSnapshot))

+ func (o *Observer) RunWithTicker(ctx context.Context, ticks <-chan time.Time, fn func(PipelineSnapshot))

+ func (o *Observer) SetDiagnosis(fn func() *core.Diagnosis)

+ func (o *Observer) Snapshot() PipelineSnapshot

+ type ObserverStage struct

+ Name string

+ Stats func() Stats

+ UnitLabel string

+ type Options struct

+ Capacity int

+ ContinueOnError bool

+ MaxWIP int

+ MaxWIPWeight int64

+ OversizePolicy OversizePolicy

+ TrackAllocations bool

+ TrackServiceTimeDist bool

+ Weight func(T) int64

+ Workers int

+ type OversizePolicy int

+ const OversizeReject

+ const OversizeWait

+ type Pipeline struct

+ func NewPipeline() *Pipeline

+ func (p *Pipeline) AddEdge(from, to string)

+ func (p *Pipeline) AddEdgeWithRatio(from, to string, ratio int)

+ func (p *Pipeline) AddStage(name string, stats func() Stats)

+ func (p *Pipeline) AncestorsOf(target string) []string

+ func (p *Pipeline) DirectPredecessors(name string) []string

+ func (p *Pipeline) EdgeRatio(from, to string) int

+ func (p *Pipeline) Freeze()

+ func (p *Pipeline) HasPath(from, to string) bool

+ func (p *Pipeline) Heads() []string

+ func (p *Pipeline) HeadsTo(target string) []string

+ func (p *Pipeline) Incoming(name string) []IncomingEdge

+ func (p *Pipeline) StageStats(name string) func() Stats

+ func (p *Pipeline) Stages() []string

+ type PipelineSnapshot struct

+ At time.Time

+ Diagnosis *core.Diagnosis

+ GoHeap uint64

+ PipelineID string

+ RSS uint64

+ RSSOK bool

+ Stages []StageSnapshotEntry

+ type PublishErrorHandler func(err error, batch ObservationBatch)

+ type Publisher interface

+ type Reporter struct

+ func NewReporter(interval time.Duration, opts ...ReporterOption) *Reporter

+ func (r *Reporter) AddStage(name string, fn func() Stats)

+ func (r *Reporter) Run(ctx context.Context)

+ type ReporterOption func(*Reporter)

+ func WithLogger(l *log.Logger) ReporterOption

+ type RopeController struct

+ func NewRopeController(pipeline *Pipeline, drum string, limits *LimitManager, ...) *RopeController

+ func NewWeightRopeController(pipeline *Pipeline, drum string, limits *LimitManager, ...) *RopeController

+ func (rc *RopeController) Run(ctx context.Context)

+ func (rc *RopeController) RunWithTicker(ctx context.Context, ticks <-chan time.Time)

+ func (rc *RopeController) Stats() RopeStats

+ type RopeOption func(*RopeController)

+ func WithInitialRopeLength(n int) RopeOption

+ func WithRopeLogger(l *log.Logger) RopeOption

+ func WithRopeSafetyFactor(factor float64) RopeOption

+ type RopeStats struct

+ AdjustmentCount int64

+ DrumErrorRate float64

+ DrumGoodput float64

+ HeadAppliedWIP int

+ RopeLength int

+ RopeUtilization float64

+ RopeWIP int

+ type ScalingHistory struct

+ func NewScalingHistory() *ScalingHistory

+ func (h *ScalingHistory) DiminishingReturns(workers int, threshold float64) bool

+ func (h *ScalingHistory) Len() int

+ func (h *ScalingHistory) Record(workers int, throughput float64)

+ func (h *ScalingHistory) Reset()

+ func (h *ScalingHistory) ScalingGain(workers int) (float64, bool)

+ type ServiceTimeSummary struct

+ Count int64

+ Max time.Duration

+ Mean time.Duration

+ Min time.Duration

+ Overflow int64

+ P50 time.Duration

+ P95 time.Duration

+ P99 time.Duration

+ StdDev time.Duration

+ Underflow int64

+ type Stage struct

+ func Pipe[T, R any](ctx context.Context, src <-chan rslt.Result[T], ...) *Stage[T, R]

+ func Start[T, R any](ctx context.Context, fn func(context.Context, T) (R, error), opts Options[T]) *Stage[T, R]

+ func (s *Stage[T, R]) ActiveWorkers() int

+ func (s *Stage[T, R]) Cause() error

+ func (s *Stage[T, R]) CloseInput()

+ func (s *Stage[T, R]) DisableMaxWIPWeight()

+ func (s *Stage[T, R]) DiscardAndCause() error

+ func (s *Stage[T, R]) DiscardAndWait() error

+ func (s *Stage[T, R]) Limits() *LimitManager

+ func (s *Stage[T, R]) MaxWIP() int

+ func (s *Stage[T, R]) MaxWIPWeight() int64

+ func (s *Stage[T, R]) Out() <-chan rslt.Result[R]

+ func (s *Stage[T, R]) PauseAdmission()

+ func (s *Stage[T, R]) Paused() bool

+ func (s *Stage[T, R]) ResumeAdmission()

+ func (s *Stage[T, R]) SetMaxWIP(n int) int

+ func (s *Stage[T, R]) SetMaxWIPWeight(n int64) int64

+ func (s *Stage[T, R]) SetWorkers(n int) (int, error)

+ func (s *Stage[T, R]) Stats() Stats

+ func (s *Stage[T, R]) Submit(ctx context.Context, item T) error

+ func (s *Stage[T, R]) TargetWorkers() int

+ func (s *Stage[T, R]) Wait() error

+ type StageAllocation struct

+ Limits *LimitManager

+ Name string

+ Share float64

+ type StageSnapshotEntry struct

+ Name string

+ Order int

+ QueueCapacity int

+ QueueDepth int64

+ State core.StageState

+ Stats Stats

+ UnitLabel string

+ Workers int

+ type Stats struct

+ ActiveWorkers int

+ Admitted int64

+ AdmittedWeight int64

+ AllocTrackingActive bool

+ BufferedDepth int64

+ Canceled int64

+ Completed int64

+ Dropped int64

+ Failed int64

+ Forwarded int64

+ IdleTime time.Duration

+ InFlightWeight int64

+ MaxWIP int

+ MaxWIPWeight int64

+ MaxWIPWeightEnabled bool

+ MaxWaiterCount int

+ ObservedAllocBytes uint64

+ ObservedAllocObjects uint64

+ OutputBlockedTime time.Duration

+ Panicked int64

+ Paused bool

+ QueueCapacity int

+ Received int64

+ ServiceTime time.Duration

+ ServiceTimeDist ServiceTimeSummary

+ Submitted int64

+ TargetWorkers int

+ WIPWaitCount int64

+ WIPWaitNs int64

+ WaiterCount int

+ type Subscriber interface

+ type Subscription interface

+ Close func() error

+ type Tee struct

+ func NewTee[T any](ctx context.Context, src <-chan rslt.Result[T], n int) *Tee[T]

+ func (t *Tee[T]) Branch(i int) <-chan rslt.Result[T]

+ func (t *Tee[T]) Stats() TeeStats

+ func (t *Tee[T]) Wait() error

+ type TeeStats struct

+ BranchBlockedTime []time.Duration

+ BranchDelivered []int64

+ FullyDelivered int64

+ PartiallyDelivered int64

+ Received int64

+ Undelivered int64

+ type WeightedBatcher struct

+ func NewWeightedBatcher[T any](ctx context.Context, src <-chan rslt.Result[T], threshold int, ...) *WeightedBatcher[T]

+ func (b *WeightedBatcher[T]) Out() <-chan rslt.Result[[]T]

+ func (b *WeightedBatcher[T]) Stats() WeightedBatcherStats

+ func (b *WeightedBatcher[T]) Wait() error

+ type WeightedBatcherStats struct

+ BatchCount int64

+ BufferedDepth int64

+ BufferedWeight int64

+ Dropped int64

+ Emitted int64

+ Forwarded int64

+ OutputBlockedTime time.Duration

+ Received int64

+ func (s WeightedBatcherStats) ToStats() Stats