Struct Scheduler 

pub(crate) struct Scheduler {
    profile: ShapeProfile,
    unmatchable: Vec<AtomicU8>,
    burst_reported: Vec<AtomicBool>,
    class_bucket: Vec<usize>,
    state: Mutex<SchedState>,
    enabled: Arc<AtomicBool>,
}

One session’s shaper.

Held behind an Arc on ForwardCtx and cloned into every forwarding task, exactly as the two recorders are — and, unlike them, None when the session has no profile. There is nothing for an unshaped session to share, and an Option on the context is what makes “a session with shape: None adds nothing to this path” a compile-time fact rather than a runtime hope.

Fields

profile: ShapeProfile

unmatchable: Vec<AtomicU8>

One report-once mask per class, indexed the same way Class::Rule is. Bit MatcherField::bit is set the first time that (class, field) pair is reported. Relaxed atomics rather than a Mutex: the only operation is set this bit, tell me whether it was already set, the only consequence of losing a race is one duplicate event on a path that fires at most classes × 4 times per session, and taking a lock per non-matching rule per unit would put a contended mutex on the data path of every shaped session.

burst_reported: Vec<AtomicBool>

One report-once flag per class for this class’s burst cannot cover one of its own units, indexed the same way Class::Rule is.

A separate mask from Self::unmatchable rather than a fifth bit in it, because the two are answers to different questions and share no key: that one is per (class, field) and is set from classification, this one is per class and is set from release. Sharing the storage would make the first release refusal silence a later unmatchable field report on the same class, which is a diagnostic losing to an unrelated one. A relaxed swap for the same reason the other mask is relaxed: the only operation is claim this, tell me whether it was already claimed, the only cost of losing the race is one duplicate event, and the path fires at most once per class per session.

class_bucket: Vec<usize>

Which bucket each class charges, resolved once at construction.

ShapeProfile::try_new has already rejected a class naming a bucket that is not configured, so this is a total map and the lookup on the data path is an index rather than a string compare.

state: Mutex<SchedState>

Everything release mutates, behind one lock.

A Mutex and not a set of atomics, and the reason is the gate: a class parks on a gate created from the same read of the demand that decided to park it, and the withdrawal that releases it happens under the same lock. Split across atomics there is a window in which a withdrawal wakes nobody and the park that follows waits for a wake that has already happened. The lock is taken once per released unit, not once per byte, on a path that is already doing a write_all.

enabled: Arc<AtomicBool>

Whether this scheduler paces at all.

Shared with every other scheduler built from the same proxy, which is what makes turning pacing off a single act rather than a walk over the sessions. A scheduler built with no switch of its own owns one that is set and never cleared, so a session driven without a proxy behaves exactly as it did before this field existed.

Read in Self::acquire and nowhere else. Everything a shaper does besides pacing — classification, the per-stream queue depth, the statistics — keeps running while it is clear, so the profile is still there to resume with and the rows a caller was reading keep moving.

Implementations

impl Scheduler

pub(crate) fn new(profile: ShapeProfile) -> Self

Build the shaper for profile, with every bucket full as of now.

Full rather than empty: an empty burst would put a burst_bytes / rate_bps delay in front of the first object of every session and read as a broken proxy.

pub(crate) fn with_switch( profile: ShapeProfile, enabled: Arc<AtomicBool>, ) -> Self

Self::new, sharing enabled with every other scheduler this proxy builds.

The switch is handed in rather than owned because turning pacing off has to be one act for the whole proxy: a session builds its own scheduler, and a per-scheduler flag would have to be found and cleared once per session, which is a walk over a set that changes while it is being walked.

pub(crate) fn at(profile: ShapeProfile, now: Instant) -> Self

Self::new with the clock supplied, so the unit tests below fabricate their own timeline exactly as charge lets them.

fn build(profile: ShapeProfile, now: Instant, enabled: Arc<AtomicBool>) -> Self

The one constructor. Both public ones differ only in where the clock and the switch come from.

pub(crate) fn class_name(&self, index: usize) -> String

The name to label a report with, for a class index.

pub(crate) fn max_hold(&self) -> Option<Duration>

The deadline a queued unit is clamped to, or None to inherit EgressConfig::max_hold. Under the default Expiry::Deliver this is the instant a starved unit goes out anyway, which is why every starvation fixture is required to pin it: a 0-bps class delivers zero bytes is a statement about a sampling window, and the window only means something against a ceiling the fixture wrote down.

pub(crate) fn on_expiry(&self) -> Expiry

What happens to a unit that outlives its clamp.

pub(crate) fn blocking_depth(&self) -> Option<QueueDepth>

The depth PendingQueue::accepts_more must carry, or None when this profile does not stop reading.

Some only under Overflow::Block. Under DropTail the read branch has to stay enabled, or nothing ever arrives to be dropped and objects_dropped could only ever be zero; under ResetStream the arriving unit is what triggers the reset. So the depth is installed on the queue for exactly one policy, and the other two evaluate it per unit through Self::admit.

Installing it in accepts_more rather than beside it is what gives Impairment{EgressQueueFull} a producer here at all: the once-per-transition latch is computed inside PendingQueue::push from accepts_more(), so a stream full by depth_objects but under EgressConfig::max_pending_bytes would otherwise never trip it.

pub(crate) fn classify( &self, side: ProxySide, meta: &ObjectMeta, unit_index: u64, report: impl FnMut(usize, MatcherField), ) -> Class

Classify one unit, reporting each unmatchable (class, field) once per session.

Rules are tried in configured order and the first match wins, so a profile’s order is a priority order for classification. A unit no rule claims is Class::Default.

report is called with the class index and the key that could not be carried, at most once per pair for the session’s whole life. It is a callback rather than a return value because the common answer is “nothing to report” and building a collection to say so would allocate on the data path.

unit_index is the per-stream count of hook-visible units, which is what Matcher::every_nth is defined against — never ObjectMeta::index_in_stream, which counts oversized objects the hook never sees and would silently shift the pattern.

The diagnostic sweep does not stop at the winner

The classification short-circuits — first match wins, and the loop below stops asking matches once it has an answer. The unmatchability check does not, and the asymmetry is what stops a dead rule going unreported: which class is this unit is about one unit, but can this rule ever fire is a question about the profile, and a rule’s answer to it does not depend on whether some earlier rule happened to claim this particular unit.

Returning at the winner made it depend on exactly that. A catch-all placed first — which is the value ClassRule::default() hands you, since Matcher::default() claims everything — matched every unit and so no later rule was ever examined, silencing every diagnostic behind it for the session’s whole life. Measured: same profile, same traffic, class order reversed, 1 report against 0.

The winner itself is skipped, and that is not the same thing: a rule that matched cannot have been defeated by an absent key, so reporting it would be a false positive on a working rule.

The cost is one Matcher::unmatchable_fields per rule per unit — three is_some tests over a fixed-size array, no allocation, and the relaxed fetch_or is reached only when a field is genuinely absent. A profile whose keys are all carried never touches an atomic.

pub(crate) fn classify_datagram( &self, side: ProxySide, draft: DraftVersion, meta: &AnyDatagramMeta, unit_index: u64, report: impl FnMut(usize, MatcherField), ) -> Class

Self::classify for a datagram.

The same two-part sweep — first match wins, every rule that did not match is asked whether it could have — over [Matcher::matches_datagram] and [Matcher::unmatchable_fields_datagram] instead of their framed siblings. The report-once mask is the one classify uses, so a session carrying both never reports a (class, field) pair twice and whichever carrier arrives first is the one that says it.

unit_index counts hook-visible datagrams per forwarding direction; see [Matcher::matches_datagram] for why that is the only scope a datagram has.

The draft is taken as an argument rather than read off the unit because an AnyDatagramMeta does not carry one — it is the resolved identity, not the header — where an ObjectMeta does.

fn report_once( &self, index: usize, field: MatcherField, report: &mut impl FnMut(usize, MatcherField), )

Set (index, field)’s bit and call report only if this is the first time anyone has.

One relaxed fetch_or; see Self::unmatchable for why the losing side of a race costs one duplicate event rather than a lock.

pub(crate) fn admit( &self, bytes: usize, queued_bytes: usize, queued_objects: usize, ) -> Admission

Whether a unit of bytes fits behind queued_objects units holding queued_bytes, and what to do if it does not.

The byte test counts the arriving unit (queued_bytes + bytes > depth_bytes) and the object test does not (queued_objects >= depth_objects), because they answer different questions: the byte depth is a budget the queue may not exceed, and the object depth is a count of slots, all of which are taken.

Overflow::Block answers Admission::Admit here and always will: under Block a full queue is one the read branch is not polling, so a unit reaching this function proves there was room for it. Deciding to drop it here as well would double-count the same limit and silently make Block destructive.

pub(crate) fn acquire(&self, class: Class, bytes: u64, now: Instant) -> Acquire

Ask whether a queued unit of class holding bytes may go at now, debiting its bucket when the answer is yes.

The release seam. Called from PendingQueue::pop_next_due and from nowhere else, exactly once per released unit. Class::Default and Class::Unshapeable answer Acquire::Now unconditionally and touch no lock: neither names a bucket, so there is nothing to charge and nothing to arbitrate. That is not a shortcut — it is what no rule claimed this means. A profile that wants a catch-all charges one by writing a class with an all-None Matcher.

The discipline is consulted before the bucket, and the order is load-bearing: charging first would let a class the discipline is holding back spend tokens the class ahead of it is entitled to, and the debit is not refundable.

While pacing is switched off

Every unit is granted here, before the bucket is read and before the discipline is consulted, so nothing is debited and no class is parked while the switch is clear.

It is read when a queue asks, and a queue asks when its head’s park expires. This function is the whole of the release decision, and a queue that has been refused here parks its head on the deadline the refusal named. Switching pacing off does not reach into that park; it changes the answer the next call gives. So the delay between the switch and a stream resuming is the park that was already running, and which park that is depends on why the head was refused:

• refused by a bucket that will refill — the refill instant, which is one unit’s worth of the configured rate;
• refused by a bucket that never refills, or one whose burst cannot cover a unit — no instant exists, so the park is the queue’s max_hold clamp, which on a stopped class is the whole of it;
• held back by the discipline — the class ahead withdrawing its demand, which is now immediate, because that class is granted here too.

The middle case is the one to know: switching pacing off does not promptly release a class configured at zero. Nothing here can, and nothing else in this crate can either — the park is a timer a queue armed, the queues are per stream, and no session-wide wake reaches them.

Nothing leaves a bucket half-charged whichever way the switch moves: the debit happens on the same call as the grant or not at all.

pub(crate) fn claim_burst_report(&self, class: Class) -> Option<String>

Claim the once-per-session right to report that class’s burst is smaller than one of its own units, and say what to call the class.

Some(name) for the first caller and None for every one after, so a class whose every unit is refused for the whole session reports once rather than once per release attempt. The name comes back with the claim because the caller needs both and asking twice would let a caller claim one class and label another.

Once per session per class, not once per stream. A burst that cannot cover an object is a property of the profile, so every stream carrying that class reproduces it and a per-stream report would say the same thing as many times as the session has streams. The rows that keep counting are the class’s own tokens_exhausted_episodes and the HoldClamped report on each clamped unit.

Answers None for the two rows that name no bucket: neither Class::Default nor Class::Unshapeable reaches charge at all, so neither can have produced the refusal this reports.

fn discipline_holds( &self, state: &mut SchedState, index: usize, bucket: usize, ) -> Option<Gate>

Whether the discipline is holding index back, and the gate to wait on if it is.

Scoped to the classes that share bucket: a discipline arbitrates between classes competing for one bucket, and two classes with their own buckets are not competing for anything.

fn wake_bucket(&self, state: &mut SchedState, bucket: usize)

Release every class parked on bucket.

Over-waking is safe and deliberate: a woken class re-asks and parks again on a fresh gate if it is still held back, which costs one loop iteration. Under-waking is a stall until max_hold, so the asymmetry decides the direction to err in.

pub(crate) fn declare_demand(&self, class: Class)

Record that one stream queue now holds an unreleased head of class.

Demand, not depth: one per queue, whatever it holds behind that head. A discipline arbitrates between classes that have something to send, and a class with ten streams waiting is not ten times more entitled than a class with one.

pub(crate) fn withdraw_demand(&self, class: Class)

Undo one Self::declare_demand, waking whatever it was holding back. Called from every path that can retire a head — a grant, a clear, the teardown drain, and PendingQueue’s Drop. The wake is unconditional rather than only when the count reached zero, because the classes it wakes re-ask and park again for free, while a missed wake is a max_hold stall on a stream that has nothing wrong with it.

Trait Implementations

impl Debug for Scheduler

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.