offsetrange

Documentation

Overview

Package offsetrange defines a restriction and restriction tracker for offset ranges. An offset range is just a range, with a start and end, that can begin at an offset, and is commonly used to represent byte ranges for files or indices for iterable containers.

Index

• type GrowableTracker
  • func NewGrowableTracker(rest Restriction, rangeEndEstimator RangeEndEstimator) (*GrowableTracker, error)
  • func (tracker *GrowableTracker) End() int64
  • func (tracker *GrowableTracker) GetProgress() (done, remaining float64)
  • func (tracker *GrowableTracker) IsBounded() bool
  • func (tracker *GrowableTracker) Start() int64
  • func (tracker *GrowableTracker) TrySplit(fraction float64) (primary, residual interface{}, err error)
• type RangeEndEstimator
• type Restriction
  • func (r Restriction) EvenSplits(num int64) (splits []Restriction)
  • func (r Restriction) Size() float64
  • func (r Restriction) SizedSplits(size int64) (splits []Restriction)
• type Tracker
  • func NewTracker(rest Restriction) *Tracker
  • func (tracker *Tracker) GetError() error
  • func (tracker *Tracker) GetProgress() (done, remaining float64)
  • func (tracker *Tracker) GetRestriction() interface{}
  • func (tracker *Tracker) IsBounded() bool
  • func (tracker *Tracker) IsDone() bool
  • func (tracker *Tracker) TryClaim(rawPos interface{}) bool
  • func (tracker *Tracker) TrySplit(fraction float64) (primary, residual interface{}, err error)

Types

type GrowableTracker

type GrowableTracker struct {
	Tracker
	// contains filtered or unexported fields
}

GrowableTracker tracks a growable offset range restriction that can be represented as a range of integer values, for example for byte offsets in a file, or indices in an array. Note that this tracker makes no assumptions about the positions of blocks within the range, so users must handle validation of block positions if needed.

func NewGrowableTracker

func NewGrowableTracker(rest Restriction, rangeEndEstimator RangeEndEstimator) (*GrowableTracker, error)

NewGrowableTracker creates a GrowableTracker for handling a growable offset range. math.MaxInt64 is used as the end of the range to indicate infinity for an unbounded range.

An OffsetRange is considered growable when the end offset could grow (or change) during execution time (e.g. Kafka topic partition offset, appended file, ...).

The growable range is marked as done by claiming math.MaxInt64-1.

For bounded restrictions, this tracker works the same as offsetrange.Tracker. Use that directly if you have no need of estimating the end of a bound.

func (*GrowableTracker) End

func (tracker *GrowableTracker) End() int64

End returns the end range of the restriction tracked by a tracker.

func (*GrowableTracker) GetProgress

func (tracker *GrowableTracker) GetProgress() (done, remaining float64)

GetProgress reports progress based on the claimed size and unclaimed sizes of the restriction.

func (*GrowableTracker) IsBounded

func (tracker *GrowableTracker) IsBounded() bool

IsBounded checks if the current restriction is bounded or not.

func (*GrowableTracker) Start

func (tracker *GrowableTracker) Start() int64

Start returns the starting range of the restriction tracked by a tracker.

func (*GrowableTracker) TrySplit

func (tracker *GrowableTracker) TrySplit(fraction float64) (primary, residual interface{}, err error)

TrySplit splits at the nearest integer greater than the given fraction of the remainder. If the fraction given is outside of the [0, 1] range, it is clamped to 0 or 1.

type RangeEndEstimator

type RangeEndEstimator interface {
	// Estimate is called to get the end offset in TrySplit() functions.
	//
	// The end offset is exclusive for the range. The estimated end is not required to
	// monotonically increase as it will only be taken into consideration when the
	// estimated end offset is larger than the current position.
	// Returning math.MaxInt64 as the estimate implies the largest possible position for the range
	// is math.MaxInt64 - 1.
	//
	// Providing a good estimate is important for an accurate progress signal and will impact
	// splitting decisions by the runner.
	Estimate() int64
}

RangeEndEstimator provides the estimated end offset of the range. Users must implement this interface to use the offsetrange.GrowableTracker.

type Restriction

type Restriction struct {
	Start, End int64
}

Restriction is an offset range restriction, which represents a range of integers as a half-closed interval with boundaries [start, end).

func (Restriction) EvenSplits

func (r Restriction) EvenSplits(num int64) (splits []Restriction)

EvenSplits splits a restriction into a number of evenly sized restrictions in ascending order. Each split restriction is guaranteed to not be empty, and each unit from the original restriction is guaranteed to be contained in one split restriction.

Num should be greater than 0. Otherwise there is no way to split the restriction and this function will return the original restriction.

func (Restriction) Size

func (r Restriction) Size() float64

Size returns the restriction's size as the difference between Start and End.

func (Restriction) SizedSplits

func (r Restriction) SizedSplits(size int64) (splits []Restriction)

SizedSplits splits a restriction into multiple restrictions of the given size, in ascending order. If the restriction cannot be evenly split, the final restriction will be the remainder.

Example: (0, 24) split into size 10s -> {(0, 10), (10, 20), (20, 24)}

Size should be greater than 0. Otherwise there is no way to split the restriction and this function will return the original restriction.

type Tracker

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

Tracker tracks a restriction that can be represented as a range of integer values, for example for byte offsets in a file, or indices in an array. Note that this tracker makes no assumptions about the positions of blocks within the range, so users must handle validation of block positions if needed.

func NewTracker

func NewTracker(rest Restriction) *Tracker

NewTracker is a constructor for an Tracker given a start and end range.

func (*Tracker) GetError

func (tracker *Tracker) GetError() error

GetError returns the error that caused the tracker to stop, if there is one.

func (*Tracker) GetProgress

func (tracker *Tracker) GetProgress() (done, remaining float64)

GetProgress reports progress based on the claimed size and unclaimed sizes of the restriction.

func (*Tracker) GetRestriction

func (tracker *Tracker) GetRestriction() interface{}

GetRestriction returns a copy of the tracker's underlying offsetrange.Restriction.

func (*Tracker) IsBounded

func (tracker *Tracker) IsBounded() bool

IsBounded returns whether or not the restriction tracker is tracking a bounded restriction that has a set maximum value or an unbounded one which can grow indefinitely.

func (*Tracker) IsDone

func (tracker *Tracker) IsDone() bool

IsDone returns true if the most recent claimed element is at or past the end of the restriction

func (*Tracker) TryClaim

func (tracker *Tracker) TryClaim(rawPos interface{}) bool

TryClaim accepts an int64 position representing the starting position of a block of work. It successfully claims it if the position is greater than the previously claimed position and within the restriction. Claiming a position at or beyond the end of the restriction signals that the entire restriction has been processed and is now done, at which point this method signals to end processing.

The tracker stops with an error if a claim is attempted after the tracker has signalled to stop, if a position is claimed before the start of the restriction, or if a position is claimed before the latest successfully claimed.

func (*Tracker) TrySplit

func (tracker *Tracker) TrySplit(fraction float64) (primary, residual interface{}, err error)

TrySplit splits at the nearest integer greater than the given fraction of the remainder. If the fraction given is outside of the [0, 1] range, it is clamped to 0 or 1.