Documentation
¶
Overview ¶
Package limits is the one definition of Pulse's instance resource limits: the Limits struct the root facade aliases as pulse.Limits, the built-in defaults, the per-field precedence resolver, the construction-time validation and the shared constructors every PULSE_LIMIT_* raise site uses.
It is a leaf over the standard library and Pulse's errors package so that predict (internal/descriptor, no-execute) and the runtime (internal/service, internal/processing) can both import it and agree on every rule.
Encoding, identical on every input layer (Options, feature profile, --limit): 0 means "use the default", Unlimited (-1) means no limit, and any other negative value is refused. An EFFECTIVE (resolved) Limits never carries 0: every field is either a positive bound or Unlimited.
Index ¶
- Constants
- Variables
- func Check(l Limits, n Name, observed int64) *errors.CodedError
- func CheckChainStages(l Limits, stages int) *errors.CodedError
- func CheckComposeSlots(l Limits, slots int) *errors.CodedError
- func CheckDerivedColumns(l Limits, derived int) *errors.CodedError
- func CheckJoinBuildRows(l Limits, rows int64) *errors.CodedError
- func CheckMatrixDim(l Limits, p int) *errors.CodedError
- func Default(n Name) int64
- func DictionaryDecodeBytes(d *encoding.Dictionary) int64
- func Digest(l Limits) string
- func EstimateMemory(in MemoryInputs) int64
- func Exceeded(n Name, configured, observed int64) *errors.CodedError
- func FirstCertain(fs []Finding) *errors.CodedError
- func Invalid(n Name, value int64) *errors.CodedError
- func IsUnlimited(v int64) bool
- func KMeansBufferBytes(records, p, k int64) int64
- func MatrixRowBufferBytes(records, rowBytes int64) int64
- func MatrixRowWorkBytes(rows, rowBytes int64) int64
- func MatrixStateBytes(blocks, buckets, accumulatorBytes int64) int64
- func MergeBlocks(shardRecords ...int64) int64
- func MergeBufferedRows(shardRecords ...int64) int64
- func Option(n Name) string
- func ParseValue(n Name, s string) (int64, error)
- func RecordBytes(schema *encoding.Schema) int64
- func SchemaDecodeBytes(schema *encoding.Schema) int64
- func Set(l *Limits, n Name, v int64) bool
- func Spell(n Name, v int64) string
- func Unit(n Name) string
- func Validate(l Limits) error
- func Value(l Limits, n Name) int64
- type Finding
- type Grade
- type JoinBound
- type Limits
- type MemoryArm
- type MemoryInputs
- type Name
Constants ¶
const ( JoinBoundDictionary = "right_key_dictionary" JoinBoundLeftXRight = "left_x_right" )
Join bound bases (JoinBound.Basis, LimitFinding.Details.join_bound).
const ( DefaultRequestTimeout time.Duration = Unlimited DefaultMaxGroups int64 = 10_000_000 DefaultMaxCrosstabCells int64 = 10_000_000 DefaultMaxEstimatedMemory int64 = Unlimited DefaultMaxMatrixDim int64 = 2_048 DefaultMaxComposeSlots int64 = 1_000 DefaultMaxChainStages int64 = 1_000 DefaultMaxJoinBuildRows int64 = 100_000_000 DefaultMaxDerivedColumns int64 = 1_000 )
Built-in defaults. RequestTimeout and MaxEstimatedMemory are opt-in (Unlimited by default); the others are set high enough that no legitimate request trips them.
const ( UnitNanoseconds = "nanoseconds" UnitBytes = "bytes" UnitCount = "count" )
Units a limit value is expressed in, as the manifest reports them.
const DigestPrefix = "lim1:"
DigestPrefix versions the limits_digest algorithm. A change to the hashed form (not to a limit value) bumps it.
const MergeBlockSize = 4096
MergeBlockSize mirrors linalg.MergeBlockSize (the blocked-merge block length in records; a run holds one CoMoment per populated block per bucket). The leaf may not import linalg, so the value is restated here and pinned equal by TestMergeBlockSizeMatchesLinalg.
const Unlimited = -1
Unlimited disables a limit. Untyped, so it is assignable to both the time.Duration and the int64 fields.
Variables ¶
var ErrUnknownName = stderrors.New("unknown limit")
ErrUnknownName is returned by ParseName for a key that names no limit.
Functions ¶
func Check ¶
func Check(l Limits, n Name, observed int64) *errors.CodedError
Check is the runtime rule for an exact count: PULSE_LIMIT_EXCEEDED when observed exceeds the effective limit n in l, nil otherwise (including when the limit is Unlimited).
func CheckChainStages ¶
func CheckChainStages(l Limits, stages int) *errors.CodedError
CheckChainStages refuses a ProcessChain call carrying more than MaxChainStages stages.
func CheckComposeSlots ¶
func CheckComposeSlots(l Limits, slots int) *errors.CodedError
CheckComposeSlots refuses a Compose call carrying more than MaxComposeSlots requests.
func CheckDerivedColumns ¶
func CheckDerivedColumns(l Limits, derived int) *errors.CodedError
CheckDerivedColumns refuses a request carrying more than MaxDerivedColumns window, feature and attribute items once every for_each is expanded.
func CheckJoinBuildRows ¶
func CheckJoinBuildRows(l Limits, rows int64) *errors.CodedError
CheckJoinBuildRows refuses a join whose build (right) side holds more than MaxJoinBuildRows records. The runtime calls it twice: on the header-only right-side count before the build decodes a record, and on the running decode count as a backstop for a miscount.
func CheckMatrixDim ¶
func CheckMatrixDim(l Limits, p int) *errors.CodedError
CheckMatrixDim refuses a matrix whose dimension p (its resolved member count) exceeds MaxMatrixDim.
func Default ¶
Default returns the limit's built-in default (nanoseconds for RequestTimeout); 0 for an unknown name.
func DictionaryDecodeBytes ¶
func DictionaryDecodeBytes(d *encoding.Dictionary) int64
DictionaryDecodeBytes is what encoding.Dictionary.ReadFrom allocates decoding d: per entry dictEntryDecodeBytes plus its label's string copy (rounded up a size class: +1/8 + 8), and the shared label buffer — dictBufferBytesPerEntry a entry (min 256), doubled until every label fits, each generation allocated. 0 for nil.
func Digest ¶
Digest returns the limits_digest of the effective limits l: DigestPrefix + sha256hex over "name=value\n" lines in declaration order. Two instances share a digest iff every effective limit is equal; it is independent of the feature-set digest.
func EstimateMemory ¶
func EstimateMemory(in MemoryInputs) int64
EstimateMemory is the upper-bound resident footprint of one run, in bytes, per arm, where rows is JoinedRows when set (a fan-out join), else records:
every arm base + 3 x max(records x stride, archive bytes)
(the payload + GC headroom; an archive's every byte)
+ schema decodes x schema bytes, on an archive
+ join right rows x (3 x stride + 2 x record bytes)
+ matrix merge-block state + vector-grouper point store
+ partition buckets x (512 + 256 x aggs), per-shard arm
+ test instances x 1024 + test levels x 256
+ buffered-test values x 32
buffered + rows x (record bytes + 16 x aggs + 128 x derived)
+ rows x (fan-out - 1) x 24
+ buckets x (512 + 256 x aggs) + cells x 512
streaming + buckets x (512 + 256 x aggs)
fused + cells x 512
plus (rows + partition records) x fan-out x 96 per value-state aggregation on every arm. Saturates at math.MaxInt64 rather than wrapping.
func Exceeded ¶
func Exceeded(n Name, configured, observed int64) *errors.CodedError
Exceeded builds the PULSE_LIMIT_EXCEEDED error every raise site returns, so the details are identical everywhere and the message carries the configured value. Details: {limit, configured, observed, option}; RequestTimeout values are nanoseconds.
func FirstCertain ¶
func FirstCertain(fs []Finding) *errors.CodedError
FirstCertain returns the error of the first Certain finding in fs — the pre-flight refusal — or nil when none is certain.
func Invalid ¶
func Invalid(n Name, value int64) *errors.CodedError
Invalid builds the PULSE_LIMIT_INVALID error for a refused value. Details: {limit, value}.
func IsUnlimited ¶
IsUnlimited reports whether an effective limit value disables the limit (Unlimited, or a non-positive value that never resolves).
func KMeansBufferBytes ¶
KMeansBufferBytes is GROUP_KMEANS's resident point store: records x p x 8 (every row's member tuple, records the upper bound on clustered rows) plus k x p x 8 (the centroids). Saturating.
func MatrixRowBufferBytes ¶
MatrixRowBufferBytes is a buffered matrix's row store: records x rowBytes (vectors.Matrix.RowBytes) — every admitted row kept until finalize, records being the upper bound on admitted rows. Saturating; 0 for a streamable matrix (rowBytes 0).
func MatrixRowWorkBytes ¶
MatrixRowWorkBytes is what a buffered matrix's kept rows cost a run for the memory estimate: rows x rowBytes x the row-work factor — the row store (MatrixRowBufferBytes is its bare payload) plus the rank finalize's working set over it. rows is every kept row copy (records x the fan-out multiplicity). Saturating; 0 for a streamable matrix.
func MatrixStateBytes ¶
MatrixStateBytes is one matrix's merge-block state: blocks x buckets x accumulatorBytes — one linalg.CoMoment per populated merge block per bucket, held until finalize. Saturating.
func MergeBlocks ¶
MergeBlocks is the number of merge fragments a run over the given per-shard record counts (archive order) populates. Blocks key by the GLOBAL record ordinal, so a shard holds one fragment per global block its ordinal range touches: ceil(n / MergeBlockSize) on a single file, one more for a shard whose edges fall inside blocks. An upper bound on the per-fragment states held until finalize (a non-head fragment keeps rows instead: MergeBufferedRows).
func MergeBufferedRows ¶
MergeBufferedRows is the rows a run over the given per-shard record counts (archive order) BUFFERS in non-head merge fragments: a shard that starts inside a block keeps that block's rows (up to the block end) for the finalize to replay on top of the head fragment. 0 on a single file.
func Option ¶
Option returns the Go spelling of the limit's option, "Options.Limits.<Field>"; empty for an unknown name.
func ParseValue ¶
ParseValue parses one input-layer value of limit n as written in a text form (`pulse mcp --limit`, the feature profile's request_timeout string): "unlimited" or "-1" is Unlimited, "0" is "use the default"; request_timeout takes a Go duration ("30s", "2m"), every other limit a base-10 integer (underscores allowed: "10_000_000"). A malformed value, or any negative other than Unlimited, is refused.
func RecordBytes ¶
RecordBytes is the resident size, in bytes, of one buffered record decoded over schema at full width (a projected decode holds fewer slots, so this bounds it): the record header, one slot per field and per set-mask / parent-group word, the bit planes, and the aux block for decimal128 fields.
func SchemaDecodeBytes ¶
SchemaDecodeBytes is what decoding schema's block allocates, in bytes: per field descriptor fieldDecodeBytes, per dictionary DictionaryDecodeBytes, per parent group its entry table. An upper bound calibrated on measured decodes (TestSchemaDecodeBytes_BoundsDecode).
func Set ¶
Set stores v in the limit's field of l. It reports false (and leaves l untouched) for an unknown name.
func Spell ¶
Spell renders an effective limit value in the text form ParseValue accepts: "unlimited" for a disabled limit, a Go duration for RequestTimeout, a plain integer otherwise — so a reported value can be pasted back into `--limit name=value`.
func Unit ¶
Unit returns the unit a limit's value is expressed in: nanoseconds for RequestTimeout, bytes for MaxEstimatedMemory, count otherwise; empty for an unknown name.
Types ¶
type Finding ¶
Finding is one limit an estimate exceeds. Estimated is the figure predict (or the pre-flight) computed, in the limit's unit. Join is the basis of a join fan-out memory finding (zero on every other finding); it rides LimitFinding.Details beside a located finding's location key.
func Evaluate ¶
Evaluate grades one estimate against the effective limit n in l: a finding (and true) when the limit is bounded and estimated exceeds it, else false. A limit equal to its bound is not exceeded.
func (Finding) DetailsMap ¶
DetailsMap is the finding's Details payload (nil when it carries none): a join fan-out finding's {join: 0, join_rows_bound, join_bound}.
func (Finding) Err ¶
func (f Finding) Err() *errors.CodedError
Err is the PULSE_LIMIT_EXCEEDED error a breach of f raises, with the estimate as the observed figure — what the runtime reports for a certain finding.
type Grade ¶
type Grade string
Grade is how sure a predict-time limit finding is that the run would breach the limit.
const ( // Certain: the estimate is the exact figure the runtime will see // (a matrix dimension, a Compose slot count, a chain stage count). // A certain finding refuses the request — predict marks it invalid // and the process pre-flight raises PULSE_LIMIT_EXCEEDED before any // record is decoded. Certain Grade = "certain" // Possible: the estimate is an upper bound (a Cartesian product of // axis sizes, a dictionary size clamped by the record count). A // possible finding is a warning only: the request may still run. Possible Grade = "possible" )
type JoinBound ¶
JoinBound is the joined-row bound a join fan-out memory estimate assumed: Rows joined rows (left x the per-key multiplicity), on Basis — JoinBoundDictionary (the right key's dictionary: left x (right - dictionary + 1), sound only when every dictionary entry occurs) or JoinBoundLeftXRight (a key with no dictionary: every right row matching every left row). The zero value means "not a join bound".
type Limits ¶
type Limits struct {
// RequestTimeout bounds the wall-clock time of one top-level call
// (Process, ProcessStream / ProcessStreamResult, Facet /
// FacetSchema, the whole Compose / ComposeParallel / ProcessChain
// call). A trip is PULSE_LIMIT_EXCEEDED; the caller's own ctx
// deadline or cancel passes through unchanged. Default: no timeout.
RequestTimeout time.Duration
// MaxGroups bounds the number of distinct group buckets one grouped
// run may create. Default 10,000,000.
MaxGroups int64
// MaxCrosstabCells bounds rows x columns of one crosstab. Default
// 10,000,000.
MaxCrosstabCells int64
// MaxEstimatedMemory bounds the predicted in-memory footprint of a
// run, in bytes. The estimate is an upper bound. Default: no limit.
MaxEstimatedMemory int64
// MaxMatrixDim bounds the dimension p of one requested matrix.
// Default 2,048.
MaxMatrixDim int64
// MaxComposeSlots bounds the number of requests in one Compose call.
// Default 1,000.
MaxComposeSlots int64
// MaxChainStages bounds the number of stages in one ProcessChain
// call. Default 1,000.
MaxChainStages int64
// MaxJoinBuildRows bounds the record count of a join's build (right)
// side, which is held in memory. Default 100,000,000.
MaxJoinBuildRows int64
// MaxDerivedColumns bounds the number of window, feature and
// attribute items one request carries once every `for_each` is
// expanded (one per plain item, one per combination). Checked before
// any for_each item renders. Default 1,000.
MaxDerivedColumns int64
}
Limits holds the instance resource limits. On an input layer a zero field means "use the default" and Unlimited (-1) means no limit; any other negative value is refused at pulse.New with PULSE_LIMIT_INVALID. Requests can never override an instance limit.
func Resolve ¶
Resolve returns the effective limits, field by field: a non-zero opts value wins, else a non-zero profile value, else the built-in default. profile may be nil (no feature profile, or a profile without a `limits` section). Pure: neither input is validated or modified — Validate the inputs first.
type MemoryArm ¶
type MemoryArm string
MemoryArm is the execution arm a request runs on, which decides the shape of its resident state.
const ( // ArmBuffered materialises every record before processing // (non-streamable Process, the buffered crosstab): O(records). ArmBuffered MemoryArm = "buffered" // ArmStreaming folds records into per-bucket state as they decode: // O(buckets). ArmStreaming MemoryArm = "streaming" // ArmFusedCrosstab folds records into the crosstab grid in the // decode loop: O(cells). ArmFusedCrosstab MemoryArm = "fused_crosstab" )
type MemoryInputs ¶
type MemoryInputs struct {
// Arm is the execution arm the request runs on.
Arm MemoryArm
// Records is the record count the arm reads (a join's left side).
Records int64
// Stride is the schema's physical record stride
// (encoding.Schema.RecordByteSize) — the payload bytes per record.
Stride int64
// RecordBytes is the resident size of one buffered record over the
// schema (RecordBytes).
RecordBytes int64
// Buckets is the group-bucket bound (1 when ungrouped).
Buckets int64
// Aggregations is the number of aggregator slots (the crosstab
// cell counts as one).
Aggregations int64
// ValueStateAggregations is how many of them keep per-value state
// (distinct / mode / frequency families).
ValueStateAggregations int64
// DerivedColumns is the number of attribute, feature and window
// outputs added to every record.
DerivedColumns int64
// Cells is the crosstab grid bound (0 without a crosstab).
Cells int64
// JoinRightRows is a join build side's record count (0 without a
// join); each is a materialised record of RecordBytes plus its
// payload.
JoinRightRows int64
// MatrixBytes is the matrices' merge-block state (MatrixStateBytes,
// summed over every matrix) plus a buffered matrix's row store
// (MatrixRowBufferBytes).
MatrixBytes int64
// GrouperBytes is a buffered vector grouper's point store
// (KMeansBufferBytes, GROUP_KMEANS) — 0 without one.
GrouperBytes int64
// ArchiveBytes is a shard archive's byte length — every byte Open
// and the scan read onto the heap (afero.ReadFile, once each),
// whatever the filesystem; 0 for a single file. It replaces records
// x stride in the payload term when larger: every shard's header,
// schema block and dictionaries, the canonical _schema.pulse entry
// and the zip directory ride with the payload.
ArchiveBytes int64
// SchemaDecodes is how many schema blocks the run decodes, with
// their dictionaries, beyond the cohort's own (an archive: 1, the
// canonical entry — Open's record-count peek and the scan read
// every shard's layout-only and allocate no dictionary); 0 for a
// single file or an anchor-opened shard. Each costs SchemaBytes.
SchemaDecodes int64
// SchemaBytes is what one decode of the cohort's schema allocates
// (SchemaDecodeBytes).
SchemaBytes int64
// PartitionBuckets is the bucket states the per-shard parallel
// reducer's partials hold until the merge — the sum over shards of
// each shard's bucket bound (1 per shard ungrouped); 0 off that arm.
// Each costs what a merged bucket costs; the merge folds into one
// partial while the others are still live.
PartitionBuckets int64
// PartitionRecords is the records whose per-value state the
// per-shard partials hold beside the merged state (the run's record
// count on the per-shard parallel arm, else 0): charged per
// value-state aggregation like Records.
PartitionRecords int64
// FanOut is the most group buckets one record lands in
// (FanOutMultiplicity: the product of every fan-out grouper's key
// bound, GROUP_SET_PER_ELEMENT's set dictionary size); 0 or 1 when
// no grouper fans out. It scales the per-value state (distinct
// (bucket, value) pairs can reach records x FanOut) and, on the
// buffered arm, the partition slots of the extra buckets.
FanOut int64
// TestInstances is the tier-1 test instances the run holds until
// finalize: tests x group buckets (every test runs once per group),
// tests ungrouped.
TestInstances int64
// TestLevels is the split levels / contingency cells the test
// instances key, summed over tests — per test the smaller of its
// instances x schema-known levels and the rows it admits (records x
// FanOut).
TestLevels int64
// TestValues is the values buffered (non-streamable) tests keep:
// per such test records x FanOut x its value columns.
TestValues int64
// JoinedRows is the rows a join's probe emits into the processor
// when they can outnumber the left side (a right key that repeats:
// left x the right key's multiplicity); 0 means Records. It replaces
// Records in every per-row processing term — per-value state, the
// buffered arm's materialised records and fan-out slots — while the
// payload term stays on Records (the left scan).
JoinedRows int64
}
MemoryInputs are the figures EstimateMemory reads. Counts are non-negative; a figure the caller cannot know makes the estimate unknowable, and the caller reports no finding instead of guessing.
type Name ¶
type Name string
Name is a limit's snake_case key — the spelling the feature profile `limits` section, `pulse mcp --limit` and every PULSE_LIMIT_* error's `limit` detail use.
const ( RequestTimeout Name = "request_timeout" MaxGroups Name = "max_groups" MaxCrosstabCells Name = "max_crosstab_cells" MaxEstimatedMemory Name = "max_estimated_memory" MaxMatrixDim Name = "max_matrix_dim" MaxComposeSlots Name = "max_compose_slots" MaxChainStages Name = "max_chain_stages" MaxJoinBuildRows Name = "max_join_build_rows" MaxDerivedColumns Name = "max_derived_columns" )
The limit names, in declaration order.
Source Files