Documentation
¶
Overview ¶
Package iothrottle implements adaptive per-worker concurrency throttling based on continuously-sampled iowait%.
The model in one sentence: IO cannot block a worker; it acts as a condition to increase concurrency. Every eligible worker runs at least one task; higher concurrency unlocks only when the storage subsystem tolerates it.
Semantics of the three canonical task shapes (see the design thread in the CHANGELOG / MANTRA discussion for the reasoning):
No-IO task: iowait% stays low → growth fires at the minimum spread interval → reaches configured ceiling as fast as the physical fit allows.
Warmup task (bioinformatics norm — load big ref, then compute): a new task launch spikes iowait during its warmup, growth waits for iowait to fall back below YELLOW, launches next task; warmups naturally stagger across the pool.
Continuous-heavy task: iowait never calms → growth never fires → concurrency stays at 1 (or wherever we started). Zero configuration needed — the algorithm produces the correct behaviour on its own.
The controller is intentionally stateless across restarts: a worker that reboots starts at effective=1 and re-converges. This keeps the design simple; memoization ("this flavor stabilized at N for this step") is deferred to a later iteration.
Index ¶
Constants ¶
const ( // RED: iowait% at which we start pushing back. RedThresholdPct = 70.0 // YELLOW: iowait% above which we stop growing (hold current // concurrency). Below YELLOW is GREEN. YellowThresholdPct = 40.0 // Number of consecutive smoothed-RED samples before we throttle. // The smoothing window already averages over SmoothingWindow // seconds, so a single smoothed-RED reading already implies // sustained pressure across roughly that whole window. Requiring // additional streak samples on top would just add latency to the // throttle without meaningful noise rejection. RedStreakToThrottle = 1 // Minimum time between growth events. Even a completely IO-free // task shouldn't ramp faster than one +1 every 2s — that keeps the // probe/observe cycle honest on very fast machines. MinGrowSpread = 2 * time.Second // Ceiling on the doubling growSpread. Beyond ~60s of quiet the // signal is well-established; there's no benefit to waiting // longer. MaxGrowSpread = 60 * time.Second // Length of the rolling window (in samples) used to smooth iowait // for decision-making. With a 1Hz sampler this is a 5-second // window. SmoothingWindow = 5 )
Tunables. Deliberately not exposed through config in this first iteration — we want to observe defaults across real workloads before giving operators knobs they'd have to guess at. The UI badge exposes the live effect so miscalibration is visible.
const SampleInterval = 1 * time.Second
SampleInterval is the tick period for the sampler goroutine. Kept at 1 Hz because that matches the SmoothingWindow constant: 5 samples = 5 seconds of averaging. Faster sampling would need a proportionally larger window to stay stable.
Variables ¶
This section is empty.
Functions ¶
func StartSampler ¶
func StartSampler(ctx context.Context, t *Throttle, peaks *PeakTracker)
StartSampler runs a goroutine that reads iowait% every SampleInterval and feeds it into the given Throttle. It maintains its own state for the cumulative-jiffies delta, so it does not contend with workerstats.CollectWorkerStats (which is called on a separate cadence, from the ping loop).
When a non-nil PeakTracker is passed, the sampler additionally folds per-tick cpu%, mem%, and iowait% into it so the ping loop can report sub-ping-interval peaks to the server. Pass nil to keep only the throttle feed (older behaviour). A cpu% or mem% read error skips that dimension for the tick but does not stop the sampler.
Returns immediately. The goroutine stops when ctx is done. Failures to read cpu.Times are logged but non-fatal — a missed sample just means the throttle sees no new data for that tick.
Types ¶
type PeakTracker ¶ added in v0.9.3
type PeakTracker struct {
// contains filtered or unexported fields
}
PeakTracker records the maximum cpu%, mem%, and iowait% observed since the last Drain(). The sampler goroutine feeds one sample per second; the ping loop calls Drain when it builds a WorkerStats, so each ping's peak_* fields reflect the true worst-case value between this ping and the previous one — not just whatever happened to be true at ping o'clock (which is what the raw cpu_usage_percent / mem_usage_percent already report).
Zero values on Drain mean "no samples arrived since the last drain", which is what the server persists — the summary aggregation then treats them as NULL.
Concurrency: Observe() and Drain() may run from different goroutines (sampler tick vs. ping loop). The mutex is fine-grained and non-contended in practice — the sampler at 1 Hz and the ping at ~0.2 Hz — so we don't bother with atomics.
func NewPeakTracker ¶ added in v0.9.3
func NewPeakTracker() *PeakTracker
NewPeakTracker returns a zeroed tracker.
func (*PeakTracker) Drain ¶ added in v0.9.3
func (p *PeakTracker) Drain() (cpu, mem, iowait, disk float32, hasData bool)
Drain returns the current maxes and resets the tracker. `hasData` is false when no Observe has landed since the last Drain — the caller then leaves the ping's peak_* fields unset rather than writing a misleading 0.
func (*PeakTracker) Observe ¶ added in v0.9.3
func (p *PeakTracker) Observe(cpu, mem, iowait, disk float32)
Observe folds one sample into the running maxes. Negative values are clamped to 0 so a gopsutil hiccup can't poison the peak. NaN is rejected because it would poison the max via comparison (>NaN is always false so NaN would stick as the max forever); NaN never arrives from gopsutil in normal operation but we guard anyway.
`disk` is the MAX across every disk the worker sees at this tick — aggregating here rather than tracking per-disk keeps the ping payload flat while still catching "any disk approaching full".
type Throttle ¶
type Throttle struct {
// contains filtered or unexported fields
}
Throttle is the per-worker adaptive concurrency controller. It is safe for concurrent access.
func New ¶
New builds a fresh Throttle. The initial ceiling is what the server (or the operator) has configured for this worker; the initial effective is min(1, configuredMax) — we always start conservative and let the sampler ramp up.
func (*Throttle) Sample ¶
Sample feeds one iowait% sample into the controller. Intended to be called at ~1Hz by an external ticker goroutine. Returns the effective concurrency that results from this sample.
func (*Throttle) SetCeiling ¶
SetCeiling updates the configured maximum. If effective exceeds the new ceiling it is clamped down immediately.
Source Files