Struct UdpManager 

pub struct UdpManager<E: FlowKeyExtractor = SourceTupleExtractor> { /* private fields */ }

The pure UDP flow manager. Generic over the flow-key extractor so the seam stays type-checked; the shell instantiates it with SourceTupleExtractor.

Implementations

impl UdpManager<SourceTupleExtractor>

pub fn new( cluster: ClusterConfig, max_flows: usize, max_rx_datagram_size: usize, hash_seed: u64, ) -> Self

Construct a manager with the default client source-address extractor.

impl<E: FlowKeyExtractor> UdpManager<E>

pub fn with_extractor( cluster: ClusterConfig, max_flows: usize, max_rx_datagram_size: usize, hash_seed: u64, extractor: E, ) -> Self

Construct a manager with a custom flow-key extractor.

pub fn flow_count(&self) -> usize

Number of currently admitted flows. Mirrors udp.active_flows.

pub fn max_flows(&self) -> usize

The configured flow-table cap.

pub fn cluster_config(&self) -> &ClusterConfig

The cluster routing + per-cluster knobs new flows are admitted under.

pub fn is_draining(&self) -> bool

Whether the listener is draining.

pub fn flow(&self, flow: FlowId) -> Option<&UdpFlow>

Borrow a flow by id (for the shell’s access log on close).

pub fn handle_input(&mut self, input: ManagerInput<'_>, now: Instant)

Feed one input into the manager. Pure: now is injected.

pub fn abort_flow(&mut self, flow: FlowId, _now: Instant, reason: CloseReason)

Tear down a single flow on demand, emitting the same outputs a normal idle close produces — Output::Metric(MetricEvent::FlowEvicted) then Output::CloseFlow(flow) — so the shell draining poll_output() decrements udp.active_flows and frees the upstream socket exactly once.

The shell calls this when it cannot establish a flow it just admitted: the upstream connect() failed (EMFILE / connection refused) or no backend resolved. Without it the flow would sit AwaitingBackend / Established pinning a max_flows slot for the full idle timeout.

Idempotent: a missing flow or one already Closing is a no-op — no double-evict, no gauge underflow. Works in either AwaitingBackend or Established. now is accepted for signature symmetry with the other time-driven entry points (the teardown itself is time-independent).

pub fn close_all(&mut self, now: Instant)

Tear down EVERY live flow, emitting — for each — the same outputs a normal idle close produces (Output::Metric(MetricEvent::FlowEvicted) then Output::CloseFlow(flow)), so the shell draining poll_output() decrements udp.active_flows and frees each upstream socket exactly once per flow.

The shell calls this on listener remove / deactivate / soft-stop before dropping the manager, so the active-flows gauge does not leak. Draining the flow table + slab to zero; a flow already Closing is skipped (idempotent, no double-evict, no underflow). After this returns, flow_count() == 0 and no timer is armed.

pub fn handle_timeout(&mut self, now: Instant)

Fire all flows whose idle deadline has elapsed at now. Deadline revalidation: a flow is selected, and re-checked before its close, by idle_deadline <= now alone. A flow that saw traffic since the expiry was armed has a later idle_deadline, so the stale expiry leaves it open; UdpFlow::timer_gen is not consulted.

Called ONLY from a wheel expiry: the shell’s single timer entry has just been delivered and consumed. now is therefore the wheel’s tick date, not the deadline — crate::timer rounds a delay to the nearest tick, so with a 100 ms tick the entry arrives up to 50 ms EARLY on the tick grid (up to 99 ms off it; see duration_to_tick) and no flow need be due at all. Either way the shell now holds nothing, so armed_deadline is cleared on entry and reschedule re-emits ArmTimer even when the minimum deadline has not moved. Without that, an early expiry is a LOST WAKEUP: nothing is closed, nothing is re-armed, and the flow is never reaped until some other flow’s deadline happens to change the minimum. This is the consume-then-reschedule rule, and its cost is one extra wheel wakeup per early-fired expiry.

pub fn poll_timeout(&self) -> Option<Instant>

The next manager-wide deadline, or None if no flow is armed. After a Self::handle_timeout at deadline d, the value returned here is guaranteed > d (or None) — the strict-advance invariant handle_timeout asserts in debug builds, which is what stops the shell busy-looping.

pub fn poll_output(&mut self) -> Option<Output>

Drain the next queued output, or None when the queue is empty.