memory

package
v0.16.0-correctness Latest Latest
Warning

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

 Go to latest Published: Aug 21, 2026 License: AGPL-3.0 Imports: 22 Imported by: 0

Documentation

Overview

Package memory provides memory tracking and budget enforcement for the query engine.

Index

Constants

View Source 
const HeapBackpressurePauseDuration = 50 * time.Millisecond

HeapBackpressurePauseDuration is how long PauseOnHeapBackpressure sleeps when HeapBackpressureActive fires. 50ms is one to two GC cycles at typical SF100 allocation rates — long enough for live heap to drop, short enough that downstream consumers don't time out.

Variables

View Source 
var ErrMemoryExceeded = fmt.Errorf("memory budget exceeded")

ErrMemoryExceeded is returned when a memory reservation exceeds the budget.

Functions

func DetectBudget 

func DetectBudget() int64

DetectBudget returns a recommended per-task memory budget (75% of the container cgroup limit). Returns 0 if no container limit is detected.

func DetectCgroupLimit 

func DetectCgroupLimit() int64

DetectCgroupLimit reads the raw container memory limit from cgroup files. Returns 0 if no container limit is detected.

Detection order:

  1. process's cgroup v2 scope (/sys/fs/cgroup/<scope>/memory.max), walking up the tree to find the tightest non-"max" limit. This catches systemd-run --scope MemoryMax=N for processes launched inside a transient unit on hosts whose root cgroup has unlimited memory.
  2. cgroups v2 root: /sys/fs/cgroup/memory.max
  3. cgroups v1: /sys/fs/cgroup/memory/memory.limit_in_bytes

func DetectMemoryLimit 

func DetectMemoryLimit() int64

DetectMemoryLimit returns the effective memory limit: cgroup limit if available, otherwise physical memory. Returns 0 if neither is detected.

func DetectPhysicalMemory 

func DetectPhysicalMemory() int64

DetectPhysicalMemory reads total physical memory from /proc/meminfo. Returns 0 if detection fails (non-Linux).

func HeapBackpressureActive 

func HeapBackpressureActive() bool

HeapBackpressureActive reports whether process heap usage is high enough that batch producers in fragment runners should pause briefly to let GC catch up and downstream operators drain. Cached for 100 ms across all callers so the per-batch overhead stays near zero.

The signal is intentionally a coarse tide gauge — a process-wide HeapAlloc check, NOT a per-operator tracker check. The motivating failure mode (Q17 SF100, 2026-05-07) was tasks heap-thrashing while the tracker reported only ~80MB used because the actual 20GB+ of per-task heap lived in transient parquet decode + hash routing allocations that no operator owns long enough to be Spillable.

Use this between batches in the consume loop, not in per-row hot paths. Returns false when GOMEMLIMIT is unset.

func HeapBackpressureThresholdBytes 

func HeapBackpressureThresholdBytes() int64

HeapBackpressureThresholdBytes returns the byte threshold at which the heap-backpressure valve fires (heapBackpressureRatio × GOMEMLIMIT), or 0 when no GOMEMLIMIT is set (valve disabled). Callers use it to judge whether an operator's own footprint is a material share of the pressure ceiling (#326).

func HeapPressureExceeded 

func HeapPressureExceeded() bool

HeapPressureExceeded is the exported view of the process heap-pressure circuit breaker: HeapAlloc (less registered reclaimable bytes) > heapPressureRatio × GOMEMLIMIT (100ms-cached). Used by the Phase-5 mmap-relief trigger to gate MADV_DONTNEED on genuine pressure rather than relieving page cache when there is headroom.

func NextInstanceID 

func NextInstanceID() uint64

NextInstanceID returns a fresh process-unique operator instance ID (>= 2).

func Offheap 

func Offheap[T any](r *OffheapRegistry, heapCap int) []T

Offheap returns an off-heap-backed slice of T (len 0, huge cap) whose appends grow in place within the reservation, or a heap slice of the given capacity when off-heap is unavailable (kill switch off, mmap failure, non-linux build). T MUST be pointer-free — the GC never scans these bytes.

func OffheapAvailable 

func OffheapAvailable() bool

OffheapAvailable reports whether the platform path is compiled in and the kill switch is on.

func OffheapExact 

func OffheapExact[T any](r *OffheapRegistry, n int) ([]T, bool)

OffheapExact returns a len=n off-heap slice over a reservation sized to exactly n elements, or ok=false when off-heap is unavailable.

Distinct from OffheapSized, which carves n elements out of a fixed 4GB reservation: that is right for the ONE table an operator owns, but the two-level group index owns 256 sub-tables (two_level_hash.go) and 256 × 4GB = 1TB of address space per aggregate — multiplied again by the partitioned-aggregation clone count. A right-sized reservation costs the same RSS (pages commit on touch either way) and the same one mapping, and keeps the virtual footprint equal to the real table size.

The slice is NOT growable in place (cap == len); it is for fixed-size hash-entry arrays whose growth path allocates a new reservation and Releases the old one.

func OffheapSized 

func OffheapSized[T any](r *OffheapRegistry, n int) ([]T, bool)

OffheapSized returns a len=n off-heap slice over a single fresh reservation, or ok=false when off-heap is unavailable or n elements exceed one reservation. For fixed-size tables (hash entries) rather than append-grown arrays.

func OffheapToggleForTest 

func OffheapToggleForTest(v bool) bool

OffheapToggleForTest flips the kill switch and returns the previous value. Test-only seam for parity runs against the heap path.

func PageCachePressureActive 

func PageCachePressureActive() bool

PageCachePressureActive reports whether the kernel is currently refaulting recently-evicted file pages faster than the threshold — page-cache thrash the Go-heap hooks cannot see. Callers gate DISCRETIONARY memory use (decode-ahead width, speculative prefetch) on it; it must not gate correctness-bearing work.

func PageCachePressureActiveBounded 

func PageCachePressureActiveBounded(cap time.Duration) bool

PageCachePressureActiveBounded is PageCachePressureActive with the per-episode honor budget (see refaultSensor.ActiveBounded). cap <= 0 is unbounded.

func PageCachePressureBoundedIgnores 

func PageCachePressureBoundedIgnores() int64

PageCachePressureBoundedIgnores returns the count of episode-cap declines — the v3 rollout marker beside PageCachePressureStats.

func PageCachePressureDiscount 

func PageCachePressureDiscount() float64

PageCachePressureDiscount returns the designed-streaming pages/sec subtracted from the refault rate at the last sample — the streaming-discount rollout marker.

func PageCachePressureStats 

func PageCachePressureStats() (lastRate float64, activations int64)

PageCachePressureStats returns the sensor's last sampled refault rate (pages/sec) and its lifetime activation count, for rollout markers.

func PauseOnHeapBackpressure 

func PauseOnHeapBackpressure(ctx context.Context) error

PauseOnHeapBackpressure is a one-line helper that callers can invoke between batches: if heap pressure is high, sleep briefly so GC can catch up. Returns ctx.Err() if the context is cancelled during the pause, nil otherwise (including when no pressure is detected).

Cheap when no pressure: one cached-atomic check.

func PauseOnHeapBackpressureUnless 

func PauseOnHeapBackpressureUnless(ctx context.Context, exempt bool) error

PauseOnHeapBackpressureUnless is PauseOnHeapBackpressure with an exempt flag (drain-phase pipelines pass true and never pause).

func RawHeapBackpressureActive 

func RawHeapBackpressureActive() bool

RawHeapBackpressureActive is HeapBackpressureActive WITHOUT the reclaimable-bytes deduction: raw HeapAlloc against the same threshold. This is the right signal only for consumers that respond by freeing the reclaimable bytes themselves (the worker's cache-shed valve and the cache admission pause — eviction must engage while the adjusted gauge stays quiet) or that capture diagnostics of the high-heap state (heap profiler). Execution decisions must use HeapBackpressureActive.

func ReadRefaultCounterForDiagnostics 

func ReadRefaultCounterForDiagnostics() (int64, bool)

ReadRefaultCounterForDiagnostics reads the raw refault counter through the same source-selection logic the sensor uses. cmd/refault-probe prints it beside the sensor state to separate "counter not visible here" from "sensor not sampling" in the field.

func ReadSpilledRows 

func ReadSpilledRows(path string) ([]map[string]any, error)

ReadSpilledRows reads all rows from a binary spilled file.

func ReserveOrForce 

func ReserveOrForce(ctx context.Context, t *Tracker, sm *SpillManager, n int64, wait time.Duration, purpose string) bool

ReserveOrForce reserves n bytes on t before a large allocation, applying pre-emptive backpressure instead of reactive spill: a clean Reserve is tried first; on a budget miss the spill manager (when present) is asked to free the shortfall and the reservation is retried for up to wait. If the budget still doesn't admit it, the bytes are force-reserved so the ledger stays honest — the caller's allocation proceeds either way.

This is the non-deadlocking shape the accounting overhaul requires: per-allocation gating with a bounded wait and a forced fallback, never operator-entry gating (see project_admission_control_rejected_2026-05-18 — gating entry on reservations deadlocks under chained build/probe).

Returns true when the fallback fired and the reservation was forced.

func SetHeapBackpressureForTesting 

func SetHeapBackpressureForTesting(v int32)

SetHeapBackpressureForTesting forces the heap-backpressure gauges: v>0 forces active, v<0 forces inactive, v==0 restores real measurement. Test-only; callers must restore 0 via defer.

func SetPageCacheStreamingSource 

func SetPageCacheStreamingSource(read func() int64)

SetPageCacheStreamingSource registers the designed-streaming byte counter the refault sensor discounts before thresholding. Disabled by WADJET_REFAULT_STREAM_DISCOUNT=0. Call before or after sensor construction; the sensor consults the registration at each sample.

func SetReclaimableBytesFunc 

func SetReclaimableBytesFunc(f func() int64)

SetReclaimableBytesFunc registers the process's reclaimable-bytes reporter (worker startup: the decoded-chunk cache's Size). f must be cheap and non-blocking — it runs inside the gauges' 100ms refresh. nil unregisters.

Types

type AccountedOperator 

type AccountedOperator interface {
	// Inspect returns a coherent point-in-time footprint snapshot. It must be
	// safe to call concurrently with the operator's own pipeline goroutine and
	// with SpillSome (the contract test exercises this under -race), and must
	// return all-zero byte fields with State==OpClosed after Close/Finalize.
	Inspect() OperatorFootprint

	// EstimateRelief reports, without side effects, how many bytes a
	// SpillSome(target) call would free right now. RequestRelief uses it to
	// rank candidates and to detect a delivered<claimed shortfall afterward. It
	// must be <= Inspect().SpillableBytes and must be a pure read.
	EstimateRelief(target int64) int64

	// SpillSome attempts to free at least target bytes by writing reclaimable
	// state to disk, returning bytes actually released back to the shared
	// tracker. RequestRelief serializes calls per instance via OpSpilling, so
	// re-entrancy is not required; concurrency with Inspect IS required. May
	// return < target.
	SpillSome(target int64) (int64, error)
}

AccountedOperator is the single interface that replaces Spillable + Inspectable. Every pipeline-breaker that detains reclaimable bytes implements it and registers with the SpillManager.

type OffheapRegistry 

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

OffheapRegistry tracks the reservations owned by one operator so Reset and Close can return them. Not safe for concurrent mutation — owners are single-goroutine (each morsel clone owns its own registry), and adoption hands whole registries over (AdoptFrom).

func NewOffheapRegistry 

func NewOffheapRegistry() *OffheapRegistry

NewOffheapRegistry returns an empty registry.

func (*OffheapRegistry) AdoptFrom 

func (r *OffheapRegistry) AdoptFrom(o *OffheapRegistry)

AdoptFrom moves every reservation owned by o into r (merge/adoption paths transfer array ownership between aggregates; the arrays' new owner must also own their unmapping).

func (*OffheapRegistry) Close 

func (r *OffheapRegistry) Close()

Close unmaps every owned reservation. Idempotent. The owner must have dropped every slice referencing them first (aggregate Close/reset order guarantees this; a stale read after Close faults loudly rather than corrupting).

func (*OffheapRegistry) Mappings 

func (r *OffheapRegistry) Mappings() int

Mappings reports how many reservations are live (tests/diagnostics).

func (*OffheapRegistry) Release 

func (r *OffheapRegistry) Release(p unsafe.Pointer) bool

Release unmaps the single reservation whose base address is p and forgets it, returning whether a reservation matched. Lets a grow-and- rehash owner (hash table doubling) return the OLD table's pages immediately instead of holding dead RSS until registry Close. The caller must have dropped every slice into the reservation first.

type OpState 

type OpState int32

OpState is the lifecycle phase of an AccountedOperator instance. The SpillManager uses it to skip operators that cannot release bytes (Closed) or that are mid-spill (Spilling) when ranking relief candidates. 

const (
	OpRegistered OpState = iota // registered, not yet holding reclaimable bytes
	OpActive                    // holding bytes; SpillableBytes may be > 0
	OpSpilling                  // a SpillSome is in flight on this instance
	OpClosed                    // finalized/closed; Inspect must report zeros
)

func (OpState) String 

func (s OpState) String() string

type OperatorFootprint 

type OperatorFootprint struct {
	// OwnedBytes is every byte this instance is accountable for: column data
	// plus operator overhead (hash arena/index, scratch). It is what the
	// drift-backstop ranks on and what PublishOwned reports. It corresponds to
	// the old PeakFootprint's unit (incl. overhead) but as a live, not peak,
	// value.
	OwnedBytes int64

	// RetainedBytes is the subset of OwnedBytes held in batches detained past
	// their producer's lifetime (the build/sort/window retain sites).
	// Observability only.
	RetainedBytes int64

	// SpillableBytes is the bytes RequestRelief may target on this instance
	// right now WITHOUT triggering a rebuild loop or re-reading
	// about-to-be-merged state. It is the floating relief signal, not a raw
	// footprint. For Sort post-finalize and HashAggregate post-merger it is 0.
	SpillableBytes int64

	// SpillReadBytes is bytes currently being read back from disk during a
	// merge/finalize (about to re-enter the heap). Reported so the advisor
	// never counts them as reclaimable; never summed into any tier.
	SpillReadBytes int64

	State      OpState
	InstanceID uint64
	Name       string

	// Departed is only set on entries returned by SpillManager.Inspect for
	// closed operators: the number of same-Name instances coalesced into this
	// entry (the footprint itself is the max-peak instance's). Zero on live
	// entries and on snapshots returned by AccountedOperator.Inspect.
	Departed int

	// SharedViews lists zero-byte references this instance holds onto a build
	// owned by another instance (broadcast probes). Empty for owners.
	SharedViews []ViewRef
}

OperatorFootprint is the single accounting snapshot an AccountedOperator exposes. It subsumes the old SpillableSnapshot (Name/Current/Peak) and adds the byte-class split the floating-budget SpillManager ranks on.

Byte-class invariants (asserted by the contract test): 

OwnedBytes     >= RetainedBytes
OwnedBytes     >= SpillableBytes
SpillableBytes >= 0
when State == OpClosed: all byte fields == 0

type Reservoir 

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

Reservoir is a named, capped pool of bytes carved out of the worker's GOMEMLIMIT envelope (e.g. the LRU file cache, the result store, codec pools). It is a TierSystemReservoir in the accounting model: observed and bounded at admission, never spilled reactively. Each reservoir wraps a Tracker so callers Reserve/Release against it and so its live usage feeds the effective-budget accessor; the recorded cap feeds the boot-time invariant.

Phase 1 introduces Reservoir alongside the existing flat sharedPoolBudget Tracker — it does not replace it. The multi-reservoir carve-up and the hard refuse-to-start gate land in later phases once every reservoir is registered and the invariant constants are validated against the constrained-memory deploy envelopes.

func NewReservoir 

func NewReservoir(name string, cap int64) *Reservoir

NewReservoir creates a HARD-capped reservoir backed by a Tracker whose budget equals the cap. Callers Reserve/Release against Tracker() to record usage.

func NewReservoirFunc 

func NewReservoirFunc(name string, cap int64, actual func() int64) *Reservoir

NewReservoirFunc creates a HARD-capped reservoir whose Actual() reads a live accessor (e.g. LRUCache.Size, ResultStore.UsedBytes) rather than its own tracker — for objects that already maintain their own byte count.

func NewSoftReservoir 

func NewSoftReservoir(name string, cap int64, actual func() int64) *Reservoir

NewSoftReservoir creates an accounting-only reservoir: Actual() feeds Available() but cap is EXCLUDED from the boot-invariant Σcaps. For pools whose worst-case cap is not boot-knowable (batchpool) or whose cap is a rough static estimate (scan/file-read-buf, codec/s2).

func (*Reservoir) Actual 

func (r *Reservoir) Actual() int64

Actual returns the bytes currently accounted in this reservoir — the live accessor when set, otherwise the internal tracker's usage.

func (*Reservoir) Cap 

func (r *Reservoir) Cap() int64

Cap returns the reservoir's hard byte cap.

func (*Reservoir) Name 

func (r *Reservoir) Name() string

Name returns the reservoir name.

func (*Reservoir) Soft 

func (r *Reservoir) Soft() bool

Soft reports whether this reservoir's cap is excluded from the boot invariant.

func (*Reservoir) Tracker 

func (r *Reservoir) Tracker() *Tracker

Tracker exposes the backing tracker so callers Reserve/Release against it.

type ReservoirRegistry 

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

ReservoirRegistry holds all reservoirs for a worker and validates the boot-time GOMEMLIMIT invariant. One instance lives per process.

func NewReservoirRegistry 

func NewReservoirRegistry() *ReservoirRegistry

NewReservoirRegistry creates an empty registry.

func (*ReservoirRegistry) Available 

func (rr *ReservoirRegistry) Available() int64

Available returns the effective budget still spendable by operators: 

GOMEMLIMIT − Σ(reservoir actual usage) − GC headroom

It uses the live runtime GOMEMLIMIT read, not a cached config value, so the budget floats with real reservoir occupancy. It returns math.MaxInt64 when GOMEMLIMIT is unset (unlimited) and clamps to 0 rather than going negative.

func (*ReservoirRegistry) LogInvariant 

func (rr *ReservoirRegistry) LogInvariant(logger *slog.Logger)

LogInvariant validates and logs the boot-time invariant without refusing to start. Phase 1 wiring: surfaces the numbers on every worker boot so the constrained-memory deploy envelopes can be confirmed before a later phase makes the invariant fatal. A violation logs at WARN; satisfaction at INFO.

func (*ReservoirRegistry) Register 

func (rr *ReservoirRegistry) Register(r *Reservoir)

Register adds a reservoir to the registry.

func (*ReservoirRegistry) TotalActual 

func (rr *ReservoirRegistry) TotalActual() int64

TotalActual returns Σ(reservoir actual occupancy) across all registered reservoirs (hard and soft). Used by the Phase-4 drift reconciler to subtract the system-reservoir accounting from HeapInuse.

func (*ReservoirRegistry) Validate 

func (rr *ReservoirRegistry) Validate() error

Validate enforces the boot-time invariant: the sum of reservoir caps plus a minimum operator footprint plus GC headroom must fit inside GOMEMLIMIT. It returns a non-nil error describing the shortfall otherwise. It is a no-op (returns nil) when GOMEMLIMIT is unset (<=0 or math.MaxInt64), matching the spill.go heap-pressure guards.

Validate reports the verdict; the caller decides whether a violation is fatal. Phase 1 callers log it; the hard refuse-to-start lands once every reservoir is registered (later phases).

type SpillManager 

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

SpillManager handles spilling data to disk when memory budget is exceeded.

func NewSpillManager 

func NewSpillManager(dir string, tracker *Tracker) (*SpillManager, error)

NewSpillManager creates a spill manager that writes temp files to the given directory.

func (*SpillManager) Cleanup 

func (sm *SpillManager) Cleanup() error

Cleanup removes all spill files.

func (*SpillManager) DegradedBudget 

func (sm *SpillManager) DegradedBudget() int64

DegradedBudget returns the reduced budget this view thresholds against, or 0 for a normal manager.

func (*SpillManager) DegradedView 

func (sm *SpillManager) DegradedView(budgetCap int64) *SpillManager

DegradedView returns a SpillManager view for a poison-suspect retry (#318): it shares sm's spill directory and tracker (charges land on the shared pool exactly as before), but its spill DECISIONS threshold against budgetCap instead of the full pool budget. With a deliberately small cap, every registered operator spills early and keeps its in-memory state bounded — the ADR-0006 machinery engages far below the heap ceiling that killed the previous attempt, instead of at 40% of a budget the task already proved it can blow through.

Like TrackingOnlyView, the view keeps its own AccountedOperator registry: the degraded task's operators are invisible to the parent manager's relief targeting. That is acceptable here by construction — the low thresholds keep their footprints too small to be useful relief victims.

func (*SpillManager) HeapDrift 

func (sm *SpillManager) HeapDrift(heapInuse int64) int64

HeapDrift returns the Phase-4 accounting drift in bytes: 

HeapInuse − (Tracker.OwnedTotal + Σreservoir.Actual)

the unaccounted Go-heap residual (parquet decode arenas, kernel scratch, shuffle decode buffers) that no operator or reservoir owns. The caller supplies heapInuse (from runtime.MemStats) so this method never pays the STW ReadMemStats cost — the worker stats loop already samples it. mmap'd files and spill-readback page cache are absent from HeapInuse AND from the accounting terms, so they're symmetrically excluded (no category error). May be negative (sample skew / eviction); callers report it as-is.

NOTE: this is HeapInuse-shaped and deliberately DISTINCT from driftExceeds, which measures OwnedTotal−Used (a within-ledger reserve-vs-published gap that drives the relief backstop). They share only OwnedTotal(); do not unify them.

func (*SpillManager) Inspect 

func (sm *SpillManager) Inspect() []OperatorFootprint

Inspect returns the per-operator footprints for the AccountedOperator registry — live operators plus one aggregated entry per departed operator Name (max-peak instance, Departed = closed-instance count). Bounded by distinct names, not by how many instances have ever closed.

func (*SpillManager) IsTrackingOnly 

func (sm *SpillManager) IsTrackingOnly() bool

IsTrackingOnly reports whether this manager is a TrackingOnlyView — a clone-partial accounting view whose operators must never spill (there is no concurrent spill format). The #326 drain-instead-of-sleep valve checks it so a clone under heap pressure sleeps rather than drains.

func (*SpillManager) RegisterAccounted 

func (sm *SpillManager) RegisterAccounted(op AccountedOperator) func()

RegisterAccounted adds an AccountedOperator to the Phase-2 relief registry and returns an unregister closure. The closure folds the operator's final footprint (peak OwnedBytes) into the per-Name departed aggregate so a closed operator's peak is still surfaced by Inspect.

func (*SpillManager) ReleaseTracking 

func (sm *SpillManager) ReleaseTracking(bytes int64)

ReleaseTracking releases the given amount from the memory tracker after spilling frees memory. Callers must track their own reserved amount and pass the delta. Do NOT use Reset() on shared trackers — it wipes other concurrent operators' accounting.

func (*SpillManager) RemoveSpilled 

func (sm *SpillManager) RemoveSpilled(path string)

RemoveSpilled unlinks one SpillRows file and drops it from the manager's file list. Operators that consumed a spilled-rows file call this instead of a bare os.Remove so the bookkeeping shrinks with the directory.

This matters on the SHARED (worker-injected) manager path (#324): there plan.Cleanup never calls sm.Cleanup() — doing so would unlink concurrent queries' files — so a SpillRows file nobody removes individually outlives the query for the worker's lifetime, and sm.files would accumulate stale paths at one entry per spill forever.

func (*SpillManager) RequestRelief 

func (sm *SpillManager) RequestRelief(target int64) (int64, error)

RequestRelief asks registered AccountedOperators to free at least target bytes by spilling. It ranks candidates by SpillableBytes descending and ACCUMULATES relief across all willing operators — it never skips an operator for being individually too small (that skip reanimated the chained build/probe deadlock, see project_admission_control_rejected_2026-05-18). It never gates entry, never sleeps on the streaming path, and never refuses an operator the chance to spill.

After pass 1, if pressure persists, a rotate pass advances past the most-recently-tried victims so we don't re-hammer the same one. Finally, if nothing was freed AND the tracker is drifting (OwnedTotal materially exceeds tracker.Used, i.e. there is an accounting gap), the drift-backstop force-spills the largest-OwnedBytes operator regardless of its SpillableBytes claim — the reactive floor that honest accounting must not delete.

func (*SpillManager) SetCheapFraction 

func (sm *SpillManager) SetCheapFraction(f float64)

SetCheapFraction overrides the SpillCheap threshold fraction (default 0.90). Ignored unless 0 < f < 1.

func (*SpillManager) SetFloatingBudgetActive 

func (sm *SpillManager) SetFloatingBudgetActive(active bool)

SetFloatingBudgetActive enables (true) or disables (false, default) the floating-budget spill threshold in ShouldSpillFor. When false, ShouldSpillFor uses the tuned static 40%/90% thresholds even if a reservoir registry is wired. When true, SpillCheap triggers at cheapFrac of the live floating budget (GOMEMLIMIT − Σreservoir actual − GC headroom) and SpillExpensive at 0.98. Deploy-gated: safe only once the system reservoirs account for the untracked transient heap (mmap working set via Phase-4 RSS-sampling).

func (*SpillManager) SetReservoirs 

func (sm *SpillManager) SetReservoirs(rr *ReservoirRegistry)

SetReservoirs wires the reservoir registry for ACCOUNTING — it makes Available()/Inspect reflect live reservoir occupancy. It does NOT activate the floating spill threshold; ShouldSpillFor stays on the tuned static 40%/90% path until SetFloatingBudgetActive(true) is called separately. Safe to call once at construction; nil leaves the static fallback.

func (*SpillManager) ShouldSpill 

func (sm *SpillManager) ShouldSpill() bool

ShouldSpill returns true when the operator should spill to disk.

It checks two independent signals:

  1. **Per-tracker budget** — the original cooperative signal: each operator reports its tracked allocations and spills when its share of the budget is exhausted. Cheap (atomic load) and accurate when every allocation paths through the tracker.

  2. **Process-wide heap pressure** — checks runtime.MemStats.HeapAlloc against GOMEMLIMIT and triggers spill when the heap approaches the soft limit. This catches allocations that bypass the tracker (probe pipeline batches, gather buffers, scan source channel buffers, every non-build operator that doesn't currently report memory). Without this signal, the SF100 deploy would hit 31 GB anon-rss with a 1.4 GB tracker budget — the tracker's view of memory was 22× smaller than reality, and the per-tracker spill check stayed under threshold while the process climbed past physical RAM.

runtime.ReadMemStats is moderately expensive (sub-millisecond), so the reading is rate-limited to once per 100 ms across all callers.

func (*SpillManager) ShouldSpillFor 

func (sm *SpillManager) ShouldSpillFor(urgency SpillUrgency) bool

ShouldSpillFor returns true when an operator with the given spill cost class should spill. SpillCheap operators trigger at 40% of the per-tracker budget; SpillExpensive operators trigger at 90%. Either class also triggers if the global heap-pressure circuit breaker fires.

The 40% SpillCheap threshold (was 60% pre-2026-05-19) is sized so that 3 concurrent fragment tasks at SF100 mc=3 — each holding ~5.7 GB peak HashJoin/build=lineitem state on a 14.8 GB GOMEMLIMIT worker — stay cumulatively under the process heap limit. With per-task share = budget / 3 ≈ 5 GB, a 60% threshold lets each task ramp to 3 GB before spilling, and the 1.1× untracked overhead pushes 3 × 3 = 9 GB to ~10 GB heap. A 40% threshold caps the per-task pre-spill peak at 2 GB; cumulative 3 × 2 × 1.1 ≈ 6.6 GB heap. The architectural mechanism is documented in project_q17_sf100_instrumented_2026-05-17.md and project_bufio_fix_sf100_2026-05-18.md (worker-w0 hitting 19.7 GB during shuffle-stage-7 from concurrent fragment-task heap stacking).

Threshold sweep on the local Q17 probe (TestHashJoin_Q17ShapeRepro): 

threshold   peak heap (vs tracker budget)
30%         0.63×
40% (now)   0.74×
50%         0.92×
60% (was)   1.10×
70%         1.27×

Wall time was flat across thresholds — earlier spill is essentially free on local NVMe. Spill bytes 67-80 MB across the sweep (lower threshold → more bytes spilled, as expected).

func (*SpillManager) SpillBudget 

func (sm *SpillManager) SpillBudget() int64

SpillBudget is the budget operators should size their own spill decisions against (drain floors, relief targets): the tracker's budget, unless this is a DegradedView whose reduced cap wins. Exported so operators do not reach for Tracker().Budget() and miss the degradation.

func (*SpillManager) SpillDir 

func (sm *SpillManager) SpillDir() string

SpillDir returns the directory used for spill files.

func (*SpillManager) SpillRows 

func (sm *SpillManager) SpillRows(rows []map[string]any) (string, error)

SpillRows writes rows to a temporary binary file on disk and returns the file path. Format: [column names header] [row marker + typed values per column]... [end marker]

func (*SpillManager) SpilledFiles 

func (sm *SpillManager) SpilledFiles() []string

SpilledFiles returns the list of current spill files.

func (*SpillManager) TrackBatch 

func (sm *SpillManager) TrackBatch(bytes int64)

TrackBatch adds an estimated batch size to the memory tracker. Unlike Reserve(), this always succeeds — it accumulates usage past the budget so ShouldSpill() can detect the threshold crossing.

func (*SpillManager) Tracker 

func (sm *SpillManager) Tracker() *Tracker

Tracker returns the underlying memory tracker. May return nil if the SpillManager was constructed without one. Used by call sites that need to report tracker accounting outside of the operator-level spill API.

func (*SpillManager) TrackingOnlyView 

func (sm *SpillManager) TrackingOnlyView() *SpillManager

TrackingOnlyView returns a SpillManager that charges the SAME tracker as sm (TrackBatch/ReleaseTracking/PublishOwned all flow to the shared pool) but never asks its operators to spill: ShouldSpillFor and ShouldSpill answer false unconditionally, including the heap-pressure backstop.

Built for morsel-parallel clone partials (morsel-execution.md §4.3): clone accumulation must be visible to admission and to the PRIMARY operator's spill trigger — clone reservations push tracker.Used up, so the primary trips at the same cumulative point the serial pipeline would — but clones themselves never spill. There is no concurrent spill format; under pressure the fragment runner collapses parallelism and merges clones into the spill-armed primary instead. Operators registered on the view (RegisterAccounted) are invisible to the real manager's relief targeting, which walks only its own registry — by design, since a clone cannot honor SpillSome.

type SpillUrgency 

type SpillUrgency int

SpillUrgency describes how much pressure is needed before this operator should spill. Operators self-classify based on the cost of their spill path.

SpillCheap is for spill paths that are bounded and recoverable: build-side hash tables, hash-aggregate hash tables. Triggering slightly early costs little.

SpillExpensive is for spill paths that stream large data to disk just to read it back: probe-side bridge collectors. Triggering this unnecessarily destroys wall-clock proportional to the probe table size. 

const (
	SpillCheap     SpillUrgency = iota // spill when budget is 60% used
	SpillExpensive                     // spill when budget is 90% used
)

type Tracker 

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

Tracker tracks memory usage with a configurable budget. Hierarchical: a query tracker can have child operator trackers.

func NewTracker 

func NewTracker(name string, budget int64) *Tracker

NewTracker creates a root memory tracker with the given budget in bytes.

func (*Tracker) Budget 

func (t *Tracker) Budget() int64

Budget returns the configured budget in bytes.

func (*Tracker) Child 

func (t *Tracker) Child(name string) *Tracker

Child creates a child tracker that shares the parent's budget.

func (*Tracker) ForceReserve 

func (t *Tracker) ForceReserve(n int64)

ForceReserve adds n bytes without checking or rolling back on over-budget. Used by operators that need to track usage for spill detection without failing.

func (*Tracker) Name 

func (t *Tracker) Name() string

Name returns the tracker name.

func (*Tracker) OwnedFor 

func (t *Tracker) OwnedFor(instanceID uint64) int64

OwnedFor returns the published owned bytes for one instance, wait-free. Returns 0 if the instance has never published.

func (*Tracker) OwnedSnapshot 

func (t *Tracker) OwnedSnapshot() map[uint64]int64

OwnedSnapshot returns a copy of instanceID -> published owned bytes. Wait-free per entry (atomic.Load); the walk itself uses sync.Map's lock-free Range.

func (*Tracker) OwnedTotal 

func (t *Tracker) OwnedTotal() int64

OwnedTotal returns the sum of all published owned-byte counters across instances. Wait-free per entry.

func (*Tracker) Peak 

func (t *Tracker) Peak() int64

Peak returns the highest value ever observed by used. Used by per-task observability hooks to surface peak per-tracker footprint at task end.

func (*Tracker) PublishOwned 

func (t *Tracker) PublishOwned(instanceID uint64, total int64)

PublishOwned records the externally-reported owned-byte total for the given instance. It is an idempotent overwrite (Store of the latest total), not additive — callers publish their current owned footprint and readers see the latest. Lock-free on the common path: the *atomic.Int64 is created once per instance via LoadOrStore, then updated in place.

func (*Tracker) Release 

func (t *Tracker) Release(n int64)

Release frees n bytes of previously reserved memory.

func (*Tracker) Reserve 

func (t *Tracker) Reserve(n int64) error

Reserve attempts to allocate n bytes. Returns ErrMemoryExceeded if the budget would be exceeded.

func (*Tracker) ReserveBlocking 

func (t *Tracker) ReserveBlocking(ctx context.Context, n int64, pollInterval time.Duration) error

ReserveBlocking attempts to reserve n bytes, retrying every pollInterval until success or ctx cancellation. Used as an admission gate so callers can wait for the budget to free instead of failing immediately on a transient over-budget condition.

The polling design is intentional: the existing tracker is lock-free (atomic counters), and adding a sync.Cond would require holding a mutex in every Release path on the hot batch loop. Polling at 100ms keeps admission overhead near zero (one Reserve attempt + one timer goroutine per blocked task) while bounding wait latency to one poll interval.

Returns ctx.Err() on cancellation or any non-budget error from Reserve.

func (*Tracker) Reset 

func (t *Tracker) Reset()

Reset resets the usage counter. Does not reset peak — peak is a high-water mark for the tracker's lifetime; callers that want a fresh peak should construct a new tracker.

func (*Tracker) Transfer 

func (t *Tracker) Transfer(from, to *Tracker, n int64)

Transfer moves n bytes of accounting from one tracker to another. When `from` and `to` share a parent, the bubbled release and reserve cancel at the common ancestor, so the parent's total is conserved; when they live in different subtrees the bytes move between subtrees, which is the intended semantics.

Concurrency: like the rest of Tracker, this is lock-free. It is two independent atomic ops on `used` (not transactional across the pair), matching Reserve's Add-then-rollback non-atomicity. A concurrent reader may briefly observe the in-between state, but both endpoints converge to a consistent sum, which is what the conserved-sum property test asserts under -race.

If the subtraction would drive `from.used` below zero — an accounting bug, bytes released that were never reserved — the source is clamped back to 0 and a WARN is emitted, rather than letting `used` go negative as bare Release does. The destination still receives the full n.

The receiver is unused; Transfer is a method only so callers can write tracker.Transfer(from, to, n) on any convenient tracker handle.

func (*Tracker) UnpublishOwned 

func (t *Tracker) UnpublishOwned(instanceID uint64)

UnpublishOwned removes an instance's published owned-byte counter. It MUST be called when an operator instance is done (deregistration / Close), otherwise OwnedTotal accumulates the last-published value of every operator instance for the life of the worker — across a full SF100 run that balloons to hundreds of GB of phantom "owned" bytes (observed on SF100 Run 1, 2026-06-01: accounting_drift went to -146 GB because thousands of closed instances leaked their counters here).

func (*Tracker) Used 

func (t *Tracker) Used() int64

Used returns the current memory usage in bytes.

type ViewRef 

type ViewRef struct {
	OwnerInstanceID uint64 // InstanceID of the AccountedOperator that owns the bytes
	WouldFreeBytes  int64  // invariant: always 0 (closing a view frees nothing)
}

ViewRef is a zero-byte handle a shared-build consumer (e.g. a broadcast-join probe) reports so attribution does not double-count the shared build. A ViewRef NEVER owns bytes and NEVER frees bytes: WouldFreeBytes is always 0.

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.