Struct Scheduler 

pub struct Scheduler {
    pub stats: Stats,
    /* private fields */
}

Fields

stats: Stats

Implementations

impl Scheduler

pub fn new(size: u64, sources: Vec<Source>, conns_per_source: &[usize]) -> Self

pub fn set_active_limit(&mut self, n: usize)

Cap how many connections may hold work at once, adjustable mid-transfer.

Why the concurrency search belongs here and not in a probe

Finding the useful connection count by probing — fetch a slab with one connection, then with two, then three, comparing goodput — is the standard approach and it is what this client did. HARP (Kim, Yildirim, Kosar, SC’16) names the cost directly: probing “may bring too much probing overhead”, because the samples are extra transfers whose price is paid before the real one starts. Measured here on a 3.15 MB object over a live path, the climbing probe made the transfer 1.96x slower than not probing at all (paired over 9 interleaved reps, p = 0.004) — the search cost more than the concurrency it found could save.

The probe is only necessary because concurrency is fixed when the transfer starts. Make it adjustable and the same search runs on the real transfer: start at one connection, measure aggregate goodput over a short window, admit another connection while the marginal gain justifies it, and stop. Every byte moved during the search is a byte of the object, so the search is free — the object had to be fetched anyway. What HARP buys with a historical corpus, this buys by putting the measurement in-band.

Connections above the limit stay dormant: they are not given work and open no socket. Raising the limit lets the next tick hand them work through the ordinary work-conserving path, so no new admission machinery is needed.

pub fn active_limit(&self) -> usize

The current concurrency cap.

pub fn set_limit_reason(&mut self, why: LimitReason)

Record why the cap is where it is. The transport owns the reasons: it is the one that sees refusals, starvation and the search’s verdict.

pub fn limit_reason(&self) -> LimitReason

Why active_limit sits below the budget, for a front end to explain a dormant connection. Meaningful only while active_limit() < n_conns().

pub fn conn_stalling(&self, j: usize, now: f64) -> bool

Would this tick reclaim connection j for silence?

The exact predicate tick applies, exposed so the transport can act on a stall in the SAME tick that reclaims it rather than one round later. A connection that was requested and has delivered nothing by the time this fires — while others on the same origin stream — is an origin that admits more connections than it serves. Lowering the concurrency cap after tick has already re-requested the reclaimed range hands it straight back to a connection the origin is going to starve again, and the transfer pays a whole stall timeout to learn nothing.

pub fn set_conn_ceiling(&mut self, n: usize)

Declare the largest concurrency this transfer can still reach.

Lowered when the origin refuses requests, raised when a refusal-free stretch earns a connection back. Assignment reserves work only for connections that can actually arrive, so telling the scheduler the real ceiling is what stops a throttled transfer from carving the object into budget-sized shares nobody will ever come for.

pub fn all_sources_suspended_until(&self, now: f64) -> Option<f64>

When every source is deliberately suspended, the earliest time one returns.

None means at least one source is usable now, so a lack of progress is a genuine stall. Some(t) means the scheduler has chosen to pause every source until t — nothing can move before then, and that silence is planned rather than pathological.

Why a caller must consult this

The transport’s no-progress watchdog exists to fail a transfer where nothing will ever happen again. A scheduled retry is the opposite of that, and conflating the two is not hypothetical: with one source (the common case — one URL, one CDN), stall_timeout 4.0s gives a watchdog of 4 * (4.0 + delta) = 16.2s, while five consecutive stalls suspend that sole source for min(4.0 * 2^3, 30) = 30s. The transfer is then killed at 16.2s for failing to make progress it had itself forbidden.

Measured consequence on a 121.7 MiB GitHub release asset: 4 of 8 runs at -x 8/-x 16 aborted with a digest mismatch, three of them having already received 126.9-127.0 MB of 127.6 MB — 99.6% complete, killed during a deliberate backoff over the last half-megabyte.

pub fn unassigned_is_empty(&self) -> bool

Whether any work is still unclaimed by any connection.

Exposed so the ramp’s contract is testable: while concurrency is below the budget, work must remain here for connections admitted later to pick up.

pub fn unassigned_total(&self) -> u64

How many bytes are still unclaimed by any connection.

The same contract as Self::unassigned_is_empty, measured rather than merely asserted: a reserve that has been whittled down to one sliver is not empty and is not a reserve either.

pub fn busy_conns(&self) -> usize

How many connections currently hold a range.

pub fn with_active_limit(self, n: usize) -> Self

Start with only n connections active, ramping up from there.

pub fn with_theta_scale(self, s: f64) -> Self

pub fn with_health_ranking(self, on: bool) -> Self

Disable health-ranked victim selection (for A/B measurement only).

pub fn with_stall_timeout(self, t: f64) -> Self

pub fn mark_done(&mut self, lo: u64, hi: u64)

Mark [lo, hi) as already held, for resuming a partial transfer.

Must be called before the first tick: the initial split assigns all unassigned work, and bytes already on disk must not be part of it.

pub fn conn_health(&self, j: usize) -> Health

Health grade of a connection, for the progress UI and for tests.

pub fn conn_source(&self, j: usize) -> usize

Source index a connection belongs to, for the progress UI.

pub fn conn_rate(&self, j: usize) -> f64

Smoothed rate estimate of a connection (bytes/s), for the progress UI.

pub fn conn_range(&self, j: usize) -> Option<(u64, u64, u64)>

Active range of a connection, for the progress UI.

pub fn n_conns(&self) -> usize

pub fn is_complete(&self) -> bool

pub fn bytes_held(&self) -> u64

pub fn held_ranges(&self) -> Vec<(u64, u64)>

The ranges that are complete on disk, as (lo, hi) pairs.

This is the complement of the unassigned set minus what is still in flight, and it is what a resume record must contain. Reporting only a byte COUNT is not enough: positioned writes land ranges out of order, so “2 MB held” says nothing about which 2 MB, and a resume that assumed a contiguous prefix would skip holes and silently corrupt the file.

pub fn worst_delta(&self) -> f64

Coverage audit: held + outstanding + unassigned == size.

This is a SAFETY invariant and it does NOT imply liveness – the livelock this code is written to avoid (a fully-stolen range leaving a connection idle with a non-empty queue) satisfies it at every instant. liveness_holds is the property that matters. The largest measured request setup cost across sources, in seconds.

Exposed because a transport-layer watchdog must express its patience in units of what a request actually costs on this path rather than as a hardcoded constant: delta differs by an order of magnitude between a LAN mirror and a TLS connection through a proxy, and a fixed timeout is either trigger-happy on the slow path or useless on the fast one.

This is the same quantity the repair deadband is built from (theta = scale * sqrt(delta * T_rem / n)), so a client that widens delta widens both together, which is the intended coupling.

pub fn stall_timeout(&self) -> f64

The configured stall timeout, in seconds.

pub fn coverage_holds(&self) -> bool

pub fn liveness_holds(&self) -> bool

True when some enabled transition strictly decreases the unheld-byte count. False means the scheduler is stuck.

pub fn on_bytes(&mut self, conn: usize, n: u64, now: f64, dt: f64)

Record n bytes arriving on conn at time now over dt seconds.

Convenience wrapper that assumes the arrival is contiguous at the connection’s cursor. Real transports must use Scheduler::on_bytes_at: a response still draining from a range that was completed or stolen would otherwise be credited against whatever range the connection holds NOW, silently advancing a cursor over bytes that never arrived and leaving a hole of zeros in the output file.

pub fn on_bytes_at(&mut self, conn: usize, off: u64, n: u64, now: f64, dt: f64)

Record n bytes that landed at absolute offset off.

Arrivals that do not begin exactly at the connection’s cursor are stale (they belong to a superseded request) and are discarded: the bytes are still written to the file by the transport, but they are not credited, so the scheduler’s coverage accounting stays exact.

pub fn suspend_source(&mut self, src: usize, until: f64)

Suspend a source (429/503 with Retry-After) and reclaim its ranges.

pub fn retire_source(&mut self, src: usize) -> Vec<usize>

Take a source out of the transfer permanently and reclaim its ranges.

Why permanent removal is a different operation from suspension

suspend_source is a pause: the source comes back when the clock says so, which is right for a 429 or a transient stall. Some failures are not pauses. A host that no longer resolves, a mirror serving a different build, a TLS certificate that will not validate — none of those get better on a timer, and retrying them costs a full setup every backoff for the rest of the download.

Retiring reclaims whatever the source’s connections were holding, so those bytes go back to the unassigned set and the surviving connections pick them up through the ordinary work-conserving path. Nothing is stranded.

Returns the connection indices that were freed, so the caller can tear down their sockets. Retiring the LAST live source is permitted — a transfer with nowhere left to fetch from has to be able to say so, and live_sources returning zero is how the caller learns it — but it is never something the scheduler does on its own.

pub fn replace_source(&mut self, src: usize, replacement: Source) -> Vec<usize>

Point a retired source’s connections at a replacement mirror.

The reserve bench

This is what a mirror list is worth. A Metalink for a distribution image commonly names fifteen to twenty mirrors, while politeness and physics together justify perhaps four connections — so most of the list is not a source, it is a reserve. Without a way to substitute, those reserves are decoration: the transfer either survives on the mirrors it opened with or it does not.

Substituting in place, rather than growing the connection set, is what keeps the aggregate socket count equal to what politeness authorised. The replaced source’s connections are reused as they are — same indices, same detector state reset — so the caller’s own per-connection bookkeeping (hostnames for the progress view, per-connection targets) stays index- aligned with the scheduler’s.

The replacement starts with the incoming Source’s estimates and a clean stall count. Carrying the dead mirror’s gamma_est across would price the new host by the old one’s failure and bias the very next repair against it.

Returns the connection indices now belonging to the replacement, empty if src does not exist.

pub fn live_sources(&self) -> usize

Sources that have not been retired.

pub fn source_is_live(&self, src: usize) -> bool

Is this source still in the transfer?

pub fn source_stalls(&self, src: usize) -> u32

Consecutive stalls charged against a source, for a caller deciding whether to substitute a reserve mirror for it.

pub fn source_priority(&self, src: usize) -> u32

The publisher-stated preference a source was created with.

pub fn n_sources(&self) -> usize

How many sources this transfer was built with, live or retired.

pub fn on_conn_error(&mut self, conn: usize, now: f64, retry_after: f64)

A connection’s transport failed: reclaim its range NOW, and hold that connection back for retry_after seconds.

Why silence is not the right signal for a failure

The stall timeout exists to grade a connection that is delivering nothing, and it has to be patient — several seconds at least, scaled to the measured setup cost, because a slow path is not a broken one. A fetch that has already returned an error needs none of that patience: the question the timeout is there to answer has been answered, by the transport, definitively.

Without this the two are conflated, and the cost is paid in whole stall timeouts. A connection whose socket was closed by the peer, whose body was truncated, or whose request was refused looks exactly like a slow one, so the range is not re-requested for 4-45 s (the range stall_timeout covers on real paths). Early in a transfer the other connections cover for it and nothing is visible; at the end, when the remaining work has concentrated onto one or two connections, the whole transfer freezes for it — the reported “downloads stall past 90%, transfer rate falls to zero, every connection shows disconnected” failure.

retry_after is the caller’s backoff for THIS connection only. The range goes back to the unassigned set immediately either way, so an idle connection can pick it up on the next tick without waiting for it.

pub fn tick(&mut self, now: f64) -> Vec<Action>

Advance the scheduler. Returns the actions the caller must perform.

pub fn theta_now(&self, now: f64) -> f64

The current repair deadband, in seconds. Exposed for measurement.