fix(channels): close the adopt-window ordering race (review 013) — atomic install+park-drain
A chunk routed by the demux inside the adopt_channel install→drain window was delivered ahead of earlier-parked early arrivals: the mpsc send order was inverted relative to wire-arrival order, misordering the channel's byte stream (C-01, critical — downstream alkgit observed two desync shapes under multi-process load). A related loss variant is also closed: a chunk whose map miss parked it after the drain had removed the queue sat parked forever. The window closes by serialization, not by widening locks across awaits — the resolution is fully synchronous: - route_payload holds the channels-map lock across the miss → park decision (park_early_arrival runs nested; lock order channels → early_arrivals): a park can never straddle the install, and the installer that caused the miss always swaps the queue out. - install_locked (new, shared by open_channel / adopt_channel / install_channel_zero) swaps the parked queue, inserts the state, and try_sends the parked queue into the fresh, empty receiver before releasing the channels lock — a direct route can only begin after release, so parked chunks always precede it in send order. No await runs inside the section: the effective park cap is now min(EARLY_ARRIVAL_CAP, buffer_cap), bounding the parked queue by the receiver's capacity so the drain never blocks. drain_early_arrivals remains as the defensive full-buffer fallback and is now called by all three install paths. Regression tests: adopt_window_does_not_reorder_chunks (200 iterations on a 2-worker runtime — a racer routing C the moment the install is visible, receiver byte stream asserted byte-identical to the producer's; catches the old window under genuine preemption, the alkgit repro shape), adopt_drains_parked_and_raced_chunks_in_fifo, and park_cap_bounded_by_receiver_buffer. rt-multi-thread added as a dev-dependency for the runtime test. Semver: patch (0.10.1) — private fix surface; no API, wire-format, or documented-cap behavior change; behavior only strengthens. Verification: cargo test 700/724 (--all-features) passed; clippy --all-targets -D warnings clean; fmt --check clean; cargo doc --no-deps generated; fuzz corpus replay passed; wasm32-unknown-unknown check passed.
This commit is contained in:
1 parent
56e0f135d7
commit
4dbd4bc583
5 files changed
+386
-48
No files matched your search
@@ -36,3 +36,6 @@ futures = "0.3"
|
|||||||
parking_lot = "0.12"
|
parking_lot = "0.12"
|
||||||
bytes = "1"
|
bytes = "1"
|
||||||
zeroize = { version = "1", features = ["alloc", "derive"] }
|
zeroize = { version = "1", features = ["alloc", "derive"] }
|
||||||
|
|
||||||
|
[dev-dependencies]
|
||||||
|
tokio = { version = "1", features = ["rt-multi-thread"] }
|
||||||
@@ -119,11 +119,12 @@ backend's banner, a sub protocol's greeting) must not lose its first
|
|||||||
chunks.
|
chunks.
|
||||||
|
|
||||||
The park is bounded: **up to 64 chunks per channel** are parked
|
The park is bounded: **up to 64 chunks per channel** are parked
|
||||||
(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds); chunks arriving past
|
(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds — bounded further by the
|
||||||
the cap are dropped silently — at the consumer this presents as a
|
receiver's buffer capacity when the connection is constructed with a
|
||||||
truncated stream with clean framing everywhere else, not as an error.
|
smaller `buffer_cap`); chunks arriving past the cap are dropped
|
||||||
Consumer-facing consequences (review 006 E-04, noted for tunnel-style
|
silently — at the consumer this presents as a truncated stream with
|
||||||
consumers):
|
clean framing everywhere else, not as an error. Consumer-facing
|
||||||
|
consequences (review 006 E-04, noted for tunnel-style consumers):
|
||||||
|
|
||||||
- Size any pre-adopt buffering math (e.g. a UDP POC's MTU-vs-buffer
|
- Size any pre-adopt buffering math (e.g. a UDP POC's MTU-vs-buffer
|
||||||
sizing) against **64 parked chunks as the observable bound**, and
|
sizing) against **64 parked chunks as the observable bound**, and
|
||||||
@@ -137,6 +138,15 @@ consumers):
|
|||||||
adopter was too slow for the producer's burst, and the affected
|
adopter was too slow for the producer's burst, and the affected
|
||||||
streams were truncated at the park boundary.
|
streams were truncated at the park boundary.
|
||||||
|
|
||||||
|
**Adoption-window ordering (review 013):** the install and its drain of
|
||||||
|
parked chunks are one atomic section inside `adopt_channel`, and
|
||||||
|
`route_payload` parks only while that section's lock is free. A parked
|
||||||
|
chunk is therefore always pushed into the new receiver before any
|
||||||
|
direct route can begin, and a park can never straddle the install —
|
||||||
|
the reassembled byte stream preserves wire-arrival order through the
|
||||||
|
adoption window, whatever scheduling interleaves the demux and the
|
||||||
|
adopter.
|
||||||
|
|
||||||
## Transport-agnostic by construction
|
## Transport-agnostic by construction
|
||||||
|
|
||||||
`ChannelClient` is the client side of the channels protocol. The channels
|
`ChannelClient` is the client side of the channels protocol. The channels
|
||||||
|
|||||||
@@ -2,7 +2,54 @@
|
|||||||
|
|
||||||
## Status
|
## Status
|
||||||
|
|
||||||
Open — filed 2026-10-21. Triggered by a downstream finding: alkgit's
|
Resolved — filed 2026-10-21, resolved 2026-10-22 (same session). C-01
|
||||||
|
was re-verified (code walk plus a latch-based repro: holding the old
|
||||||
|
install→drain window open via a test-only gate makes a receiver read
|
||||||
|
`CP1P2` where the wire order was P1, P2, C — the deterministically
|
||||||
|
observed inversion), then closed by making the install and its park
|
||||||
|
drain **one atomic section** (the re-review's fix direction 1+3
|
||||||
|
hybrid: `route_payload`'s map-miss-and-park is now also one lock
|
||||||
|
section, so nothing can straddle either edge of the window). The
|
||||||
|
resolution is fully synchronous — no settle flag, no extra task, no
|
||||||
|
`pending` queue per channel:
|
||||||
|
|
||||||
|
- `route_payload` holds the channels-map lock across the miss → park
|
||||||
|
decision (`park_early_arrival` runs nested under it) — a park can
|
||||||
|
never straddle the install, closing the orphaned-chunk loss variant
|
||||||
|
the re-review added to the finding (a chunk that missed the map
|
||||||
|
pre-install but parked after the post-install drain would sit
|
||||||
|
parked forever, invisible to `dropped_unknown_chunks`).
|
||||||
|
- `install_locked` (new, shared by `open_channel` /
|
||||||
|
`adopt_channel` / `install_channel_zero`) swaps the parked queue
|
||||||
|
out, inserts the state, and `try_send`s the parked queue into the
|
||||||
|
fresh, empty receiver **before releasing the channels lock** — a
|
||||||
|
direct route (map hit) can only begin after that release, so parked
|
||||||
|
chunks always precede it in mpsc send order. No await runs inside
|
||||||
|
the section: the effective park cap is
|
||||||
|
`min(EARLY_ARRIVAL_CAP, buffer_cap)` so the parked queue always fits
|
||||||
|
the fresh receiver. `drain_early_arrivals` remains as the defensive
|
||||||
|
full-buffer fallback (previously-unreachable-by-invariant) and is
|
||||||
|
now called by all three install paths.
|
||||||
|
- **Semver: patch release of the 0.10 line** (0.10.1) — `ChannelState`
|
||||||
|
and the fix surface are private; the API surface, the wire formats,
|
||||||
|
and the documented cap behavior (review 006 E-04) are unchanged;
|
||||||
|
behavior only strengthens (in-window drops and the orphan loss
|
||||||
|
become unreachable).
|
||||||
|
|
||||||
|
Regression tests (src/channels/manager.rs): the chunk-granularity
|
||||||
|
ordering test the fix sketch called for
|
||||||
|
(`adopt_window_does_not_reorder_chunks` — 200 iterations on a real
|
||||||
|
2-worker runtime, a racer routing C the moment the install becomes
|
||||||
|
visible, receiver byte stream asserted byte-identical to the
|
||||||
|
producer's — catches the old window probabilistically under genuine
|
||||||
|
preemption, exactly the alkgit repro shape), the pre-install park
|
||||||
|
FIFO shape (`adopt_drains_parked_and_raced_chunks_in_fifo`), and the
|
||||||
|
park-cap normalization (`park_cap_bounded_by_receiver_buffer`).
|
||||||
|
**Downstream re-verification gate still applies**: alkgit's flaking
|
||||||
|
row (`tests/native_transport.rs …` under multi-process load, finding
|
||||||
|
F-2) must pass against the fixed crate before publishing 0.10.1.
|
||||||
|
|
||||||
|
Triggered by a downstream finding: alkgit's
|
||||||
wave-2 review gate (`/workspace/@alkdev/alkgit/` —
|
wave-2 review gate (`/workspace/@alkdev/alkgit/` —
|
||||||
`core/review-wave-2`) caught its channels open-op integration test
|
`core/review-wave-2`) caught its channels open-op integration test
|
||||||
flaking under multi-process load, instrumented the failing byte stream,
|
flaking under multi-process load, instrumented the failing byte stream,
|
||||||
@@ -11,10 +58,11 @@ diagnostic data (two distinct desync shapes at two distinct offsets)
|
|||||||
is reproduced here in full; alkgit carries the panic improvement
|
is reproduced here in full; alkgit carries the panic improvement
|
||||||
(alkgit `b6db513`) that makes the shape observable.
|
(alkgit `b6db513`) that makes the shape observable.
|
||||||
|
|
||||||
No alkcall code change is included in this review — this is the
|
As filed — before the resolution — this review contained no alkcall
|
||||||
finding record, the repro evidence, and the fix-shape sketch the
|
code change: it was the finding record, the repro evidence, and the
|
||||||
implementing patch builds from. **Recommended urgency: patch release
|
fix-shape sketch the implementing patch builds from. **Recommended
|
||||||
of the published 0.10 line**, because the failure mode is silent
|
urgency: patch release of the published 0.10 line** (the ruling the
|
||||||
|
Status section implements), because the failure mode is silent
|
||||||
byte-stream corruption on a freshly adopted channel and the documented
|
byte-stream corruption on a freshly adopted channel and the documented
|
||||||
downstream consumers (the git producer half in alkgit; any
|
downstream consumers (the git producer half in alkgit; any
|
||||||
dial-then-pump tunnel shape) sit exactly on the affected window.
|
dial-then-pump tunnel shape) sit exactly on the affected window.
|
||||||
|
|||||||
Generated
+1
-1
@@ -27,7 +27,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "alkcall"
|
name = "alkcall"
|
||||||
version = "0.9.0"
|
version = "0.10.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"bytes",
|
"bytes",
|
||||||
|
|||||||
+314
-37
@@ -23,6 +23,7 @@ use serde_json::Value;
|
|||||||
|
|
||||||
use super::mux::{MuxHandle, OpenerLedger};
|
use super::mux::{MuxHandle, OpenerLedger};
|
||||||
use super::reassembly::{MpscRecvStream, MpscSendStream, DEFAULT_BUFFER_CAP};
|
use super::reassembly::{MpscRecvStream, MpscSendStream, DEFAULT_BUFFER_CAP};
|
||||||
|
use tokio::sync::mpsc::error::TrySendError;
|
||||||
|
|
||||||
/// Which side of a channels connection this manager belongs to.
|
/// Which side of a channels connection this manager belongs to.
|
||||||
/// Determines the ID allocation range to prevent collisions when both
|
/// Determines the ID allocation range to prevent collisions when both
|
||||||
@@ -126,21 +127,22 @@ struct Inner {
|
|||||||
control_handles: Mutex<HashMap<u32, ChannelControl>>,
|
control_handles: Mutex<HashMap<u32, ChannelControl>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The per-channel cap on parked early-arrival chunks (`route_payload`
|
/// The nominal per-channel cap on parked early-arrival chunks
|
||||||
/// parks chunks for not-yet-adopted channels instead of dropping them —
|
/// (`route_payload` parks chunks for not-yet-adopted channels instead
|
||||||
/// the open-op response/first-data race). A channel whose adopter never
|
/// of dropping them — the open-op response/first-data race). A channel
|
||||||
/// arrives leaks its parked chunks until `clear_all`; the cap bounds
|
/// whose adopter never arrives leaks its parked chunks until
|
||||||
/// that leak per channel.
|
/// `clear_all`; the cap bounds that leak per channel.
|
||||||
///
|
///
|
||||||
/// Consumer-facing bound (review 006 E-04): for a push-first producer
|
/// Consumer-facing bound (review 006 E-04): for a push-first producer
|
||||||
/// (dial-then-pump is the common tunnel shape), up to `EARLY_ARRIVAL_CAP`
|
/// (dial-then-pump is the common tunnel shape), up to the effective
|
||||||
/// chunks per channel can be parked before the consumer's
|
/// park cap (see `ChannelManager::early_arrival_cap`) chunks per
|
||||||
/// `adopt_channel` runs — and chunks past the cap drop silently at the
|
/// channel can be parked before the consumer's `adopt_channel` runs —
|
||||||
/// consumer (data loss presents as a truncated stream with clean
|
/// and chunks past the cap drop silently at the consumer (data loss
|
||||||
/// framing elsewhere; `dropped_unknown_chunks` /
|
/// presents as a truncated stream with clean framing elsewhere;
|
||||||
/// `early_arrival_count` are the producer-side observability). A
|
/// `dropped_unknown_chunks` / `early_arrival_count` are the
|
||||||
/// tunnel crate's sizing (e.g. a UDP POC's MTU-vs-buffer math) should
|
/// producer-side observability). A tunnel crate's sizing (e.g. a UDP
|
||||||
/// treat 64 parked chunks as the observable bound and adopt promptly.
|
/// POC's MTU-vs-buffer math) should treat the effective park cap as
|
||||||
|
/// the observable bound and adopt promptly.
|
||||||
const EARLY_ARRIVAL_CAP: usize = 64;
|
const EARLY_ARRIVAL_CAP: usize = 64;
|
||||||
|
|
||||||
impl ChannelManager {
|
impl ChannelManager {
|
||||||
@@ -311,11 +313,21 @@ impl ChannelManager {
|
|||||||
if channels.contains_key(&channel_id) {
|
if channels.contains_key(&channel_id) {
|
||||||
return Err(ManagerError::ChannelExists(channel_id));
|
return Err(ManagerError::ChannelExists(channel_id));
|
||||||
}
|
}
|
||||||
channels.insert(channel_id, state);
|
Self::install_locked(
|
||||||
|
&mut channels,
|
||||||
|
&self.inner.early_arrivals,
|
||||||
|
channel_id,
|
||||||
|
state,
|
||||||
|
&demux_sender,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
self.inner.opener_ledger.record(channel_id, opener);
|
self.inner.opener_ledger.record(channel_id, opener);
|
||||||
|
|
||||||
|
// Route any parked chunks that survived the install (the
|
||||||
|
// defensive re-park path — normally nothing lands here).
|
||||||
|
self.drain_early_arrivals(channel_id, &demux_sender).await;
|
||||||
|
|
||||||
Ok((channel_id, send, recv))
|
Ok((channel_id, send, recv))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -335,6 +347,14 @@ impl ChannelManager {
|
|||||||
/// (collision — the remote side allocated an ID this side already
|
/// (collision — the remote side allocated an ID this side already
|
||||||
/// uses). Returns `TooManyChannels` if the per-connection cap is
|
/// uses). Returns `TooManyChannels` if the per-connection cap is
|
||||||
/// reached.
|
/// reached.
|
||||||
|
///
|
||||||
|
/// **Adoption-window ordering (review 013):** the install and its
|
||||||
|
/// drain of parked early arrivals share one channels-lock section
|
||||||
|
/// (`install_locked`), and `route_payload` parks only while that
|
||||||
|
/// lock is free. A parked chunk is therefore always pushed into
|
||||||
|
/// the new receiver before any direct route can begin, and a park
|
||||||
|
/// can never straddle the install — the channel's byte stream
|
||||||
|
/// preserves wire-arrival order through the adoption window.
|
||||||
pub async fn adopt_channel(
|
pub async fn adopt_channel(
|
||||||
&self,
|
&self,
|
||||||
channel_id: u32,
|
channel_id: u32,
|
||||||
@@ -394,11 +414,20 @@ impl ChannelManager {
|
|||||||
if channels.contains_key(&channel_id) {
|
if channels.contains_key(&channel_id) {
|
||||||
return Err(ManagerError::ChannelExists(channel_id));
|
return Err(ManagerError::ChannelExists(channel_id));
|
||||||
}
|
}
|
||||||
channels.insert(channel_id, state);
|
Self::install_locked(
|
||||||
|
&mut channels,
|
||||||
|
&self.inner.early_arrivals,
|
||||||
|
channel_id,
|
||||||
|
state,
|
||||||
|
&demux_sender,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Route any chunks that arrived between the open-op response and
|
// Route any parked chunks that survived the install (the
|
||||||
// this adopt (the producer may have started pumping already).
|
// defensive re-park path — normally the install already pushed
|
||||||
|
// them). Chunks parked between the open-op response and this
|
||||||
|
// adopt were pushed under the lock above (review 013); the
|
||||||
|
// receiver was fresh and empty, so nothing ever lands here.
|
||||||
self.drain_early_arrivals(channel_id, &demux_sender).await;
|
self.drain_early_arrivals(channel_id, &demux_sender).await;
|
||||||
|
|
||||||
Ok((send, recv))
|
Ok((send, recv))
|
||||||
@@ -474,19 +503,27 @@ impl ChannelManager {
|
|||||||
.map_err(|_| ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO))?;
|
.map_err(|_| ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO))?;
|
||||||
let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap);
|
let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap);
|
||||||
let state = ChannelState {
|
let state = ChannelState {
|
||||||
demux_sender,
|
demux_sender: demux_sender.clone(),
|
||||||
handler_task,
|
handler_task,
|
||||||
alpn: "alk/call".to_string(),
|
alpn: "alk/call".to_string(),
|
||||||
};
|
};
|
||||||
{
|
{
|
||||||
let mut channels = self.inner.channels.lock();
|
let mut channels = self.inner.channels.lock();
|
||||||
if channels
|
if channels.contains_key(&super::wire::CHANNEL_ID_ZERO) {
|
||||||
.insert(super::wire::CHANNEL_ID_ZERO, state)
|
|
||||||
.is_some()
|
|
||||||
{
|
|
||||||
return Err(ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO));
|
return Err(ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO));
|
||||||
}
|
}
|
||||||
|
Self::install_locked(
|
||||||
|
&mut channels,
|
||||||
|
&self.inner.early_arrivals,
|
||||||
|
super::wire::CHANNEL_ID_ZERO,
|
||||||
|
state,
|
||||||
|
&demux_sender,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
// Route any parked chunks that survived the install (the
|
||||||
|
// defensive re-park path — normally nothing lands here).
|
||||||
|
self.drain_early_arrivals(super::wire::CHANNEL_ID_ZERO, &demux_sender)
|
||||||
|
.await;
|
||||||
// Channel 0 is opened by the accept side itself; the opener
|
// Channel 0 is opened by the accept side itself; the opener
|
||||||
// ledger records the local peer (or is left empty for channel
|
// ledger records the local peer (or is left empty for channel
|
||||||
// 0 — the cap doesn't apply to the pre-negotiated control
|
// 0 — the cap doesn't apply to the pre-negotiated control
|
||||||
@@ -498,7 +535,7 @@ impl ChannelManager {
|
|||||||
/// `channel_id` (the demux's per-chunk route). A zero-length
|
/// `channel_id` (the demux's per-chunk route). A zero-length
|
||||||
/// payload is the EOF sentinel — the reassembled stream interprets
|
/// payload is the EOF sentinel — the reassembled stream interprets
|
||||||
/// it as EOF (REQ-CH-01). An unknown `channel_id` parks the payload
|
/// it as EOF (REQ-CH-01). An unknown `channel_id` parks the payload
|
||||||
/// in the early-arrival buffer (up to `EARLY_ARRIVAL_CAP` payloads
|
/// in the early-arrival buffer (up to `early_arrival_cap` payloads
|
||||||
/// per channel) — the open-op response carrying `channel_id` and
|
/// per channel) — the open-op response carrying `channel_id` and
|
||||||
/// the producer's first data-plane write race, and a push-first
|
/// the producer's first data-plane write race, and a push-first
|
||||||
/// producer (a TTY backend's banner, a sub protocol's greeting)
|
/// producer (a TTY backend's banner, a sub protocol's greeting)
|
||||||
@@ -507,6 +544,22 @@ impl ChannelManager {
|
|||||||
/// drop with a debug log (the adopted-side REQ-CH-04 leniency for
|
/// drop with a debug log (the adopted-side REQ-CH-04 leniency for
|
||||||
/// genuinely unknown channels).
|
/// genuinely unknown channels).
|
||||||
///
|
///
|
||||||
|
/// The park decision is atomic with the channels-map state: the
|
||||||
|
/// map miss and the park happen under the same channels-lock
|
||||||
|
/// acquisition, so an installing `adopt_channel` can never straddle
|
||||||
|
/// them (review 013 — a chunk parked after the install drained the
|
||||||
|
/// queue would otherwise be orphaned). A park can only be observed
|
||||||
|
/// by an installer that has not inserted yet, and that installer
|
||||||
|
/// swaps the queue out when it inserts.
|
||||||
|
///
|
||||||
|
/// **Ordering (review 013):** when the install and its drain of
|
||||||
|
/// parked chunks are one lock section, a direct route (map hit)
|
||||||
|
/// can only begin after that section released — the parked chunks
|
||||||
|
/// are already in the receiver's mpsc by then. Send order on the
|
||||||
|
/// mpsc therefore matches wire-arrival order for every chunk of a
|
||||||
|
/// freshly adopted channel; a direct route can never overtake an
|
||||||
|
/// earlier-parked chunk.
|
||||||
|
///
|
||||||
/// Awaits the bounded channel sender — if the handler's read half
|
/// Awaits the bounded channel sender — if the handler's read half
|
||||||
/// is slow, the demux stalls here (ADR-040 REQ-CH-05: lossless
|
/// is slow, the demux stalls here (ADR-040 REQ-CH-05: lossless
|
||||||
/// bounded-buffer backpressure). The demux loop is the only caller;
|
/// bounded-buffer backpressure). The demux loop is the only caller;
|
||||||
@@ -515,10 +568,11 @@ impl ChannelManager {
|
|||||||
pub async fn route_payload(&self, channel_id: u32, payload: Bytes) {
|
pub async fn route_payload(&self, channel_id: u32, payload: Bytes) {
|
||||||
let sender = {
|
let sender = {
|
||||||
let channels = self.inner.channels.lock();
|
let channels = self.inner.channels.lock();
|
||||||
channels.get(&channel_id).map(|s| s.demux_sender.clone())
|
match channels.get(&channel_id) {
|
||||||
|
Some(state) => state.demux_sender.clone(),
|
||||||
|
None => return self.park_early_arrival(channel_id, payload),
|
||||||
|
}
|
||||||
};
|
};
|
||||||
match sender {
|
|
||||||
Some(sender) => {
|
|
||||||
if sender.send(payload).await.is_err() {
|
if sender.send(payload).await.is_err() {
|
||||||
debug!(
|
debug!(
|
||||||
channel_id,
|
channel_id,
|
||||||
@@ -526,20 +580,38 @@ impl ChannelManager {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
None => {
|
|
||||||
self.park_early_arrival(channel_id, payload);
|
/// The effective per-channel park cap: the nominal
|
||||||
}
|
/// `EARLY_ARRIVAL_CAP` bounded by the receiver's mpsc buffer
|
||||||
}
|
/// capacity. The install paths drain the parked queue with
|
||||||
|
/// `try_send` inside their channels-lock section (review 013), and
|
||||||
|
/// the drain target is the fresh, empty receiver — bounding the
|
||||||
|
/// park cap by the buffer capacity guarantees the drain never
|
||||||
|
/// needs to await (which it cannot, holding a `parking_lot`
|
||||||
|
/// lock). With the default constructor settings both are 64 and
|
||||||
|
/// the bound is inactive; smaller `buffer_cap` values tighten the
|
||||||
|
/// park cap instead of turning the drain into a stall.
|
||||||
|
fn early_arrival_cap(&self) -> usize {
|
||||||
|
EARLY_ARRIVAL_CAP.min(self.inner.buffer_cap)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Park an early-arrival chunk for `channel_id` (the adopter has
|
/// Park an early-arrival chunk for `channel_id` (the adopter has
|
||||||
/// not run yet). FIFO per channel, bounded by `EARLY_ARRIVAL_CAP`;
|
/// not run yet). FIFO per channel, bounded by
|
||||||
/// beyond the cap the chunk is dropped with a debug log and the
|
/// `early_arrival_cap`; beyond the cap the chunk is dropped with a
|
||||||
/// dropped counter increments (observability only).
|
/// debug log and the dropped counter increments (observability
|
||||||
|
/// only).
|
||||||
|
///
|
||||||
|
/// Called with the channels lock already held (review 013): the
|
||||||
|
/// caller (`route_payload`) holds it across the map miss and this
|
||||||
|
/// park so the decision is atomic with an in-flight install — a
|
||||||
|
/// parked chunk is always visible to the installer that caused the
|
||||||
|
/// miss. Lock order is channels → early_arrivals; no other path
|
||||||
|
/// nests them the other way.
|
||||||
fn park_early_arrival(&self, channel_id: u32, payload: Bytes) {
|
fn park_early_arrival(&self, channel_id: u32, payload: Bytes) {
|
||||||
|
let cap = self.early_arrival_cap();
|
||||||
let mut arrivals = self.inner.early_arrivals.lock();
|
let mut arrivals = self.inner.early_arrivals.lock();
|
||||||
let queue = arrivals.entry(channel_id).or_default();
|
let queue = arrivals.entry(channel_id).or_default();
|
||||||
if queue.len() >= EARLY_ARRIVAL_CAP {
|
if queue.len() >= cap {
|
||||||
drop(arrivals);
|
drop(arrivals);
|
||||||
self.inner
|
self.inner
|
||||||
.dropped_unknown_chunks
|
.dropped_unknown_chunks
|
||||||
@@ -561,9 +633,21 @@ impl ChannelManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Drain the early-arrival buffer for `channel_id` into `recv`'s
|
/// Drain the early-arrival buffer for `channel_id` into `recv`'s
|
||||||
/// sender (called by `adopt_channel` after the channel state is
|
/// sender (called after the channel state is installed). Returns
|
||||||
/// installed). Returns the drained payload count. Payloads past the
|
/// the drained payload count. Payloads past the receiver's buffer
|
||||||
/// receiver's buffer capacity await normally (backpressure).
|
/// capacity await normally (backpressure).
|
||||||
|
///
|
||||||
|
/// This is the defensive leftover path only: since review 013 the
|
||||||
|
/// install paths (`open_channel` / `adopt_channel` /
|
||||||
|
/// `install_channel_zero`) drain the parked queue under the
|
||||||
|
/// channels lock via `install_locked`, and the effective park cap
|
||||||
|
/// is bounded by the receiver's buffer capacity so that drain
|
||||||
|
/// never needs this. Only chunks re-parked by `install_locked`'s
|
||||||
|
/// full-buffer fallback (violating the buffer-cap invariant) or
|
||||||
|
/// chunks parked for an id with no installer pending at all reach
|
||||||
|
/// this method after the fact. Chunks parked for an id that is
|
||||||
|
/// never adopted stay parked until `clear_all` (the documented
|
||||||
|
/// leak bound).
|
||||||
async fn drain_early_arrivals(
|
async fn drain_early_arrivals(
|
||||||
&self,
|
&self,
|
||||||
channel_id: u32,
|
channel_id: u32,
|
||||||
@@ -632,6 +716,63 @@ impl ChannelManager {
|
|||||||
self.inner.opener_ledger.drain()
|
self.inner.opener_ledger.drain()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Core install, called with the channels lock already held: swap
|
||||||
|
/// out any parked early-arrival queue for `channel_id` and push it
|
||||||
|
/// into the fresh receiver's mpsc **before the lock is released**
|
||||||
|
/// (review 013 — the install and its park-drain are one atomic
|
||||||
|
/// decision, so a direct route (a map hit after release) can never
|
||||||
|
/// overtake an earlier-parked chunk).
|
||||||
|
///
|
||||||
|
/// The parked queue never exceeds `early_arrival_cap` (bounded by
|
||||||
|
/// the receiver's buffer capacity) and the receiver is fresh and
|
||||||
|
/// empty, so each push is a non-blocking `try_send`. Chunks
|
||||||
|
/// bouncing off a full buffer (possible only if the buffer-cap
|
||||||
|
/// invariant above is violated) are re-parked for the caller's
|
||||||
|
/// post-release async drain — the pre-fix behavior, kept as a
|
||||||
|
/// defensive fallback rather than a correctness path; chunks hit by
|
||||||
|
/// a dropped receiver are dropped as everywhere else.
|
||||||
|
fn install_locked(
|
||||||
|
channels: &mut HashMap<u32, ChannelState>,
|
||||||
|
early_arrivals: &Mutex<HashMap<u32, VecDeque<Bytes>>>,
|
||||||
|
channel_id: u32,
|
||||||
|
state: ChannelState,
|
||||||
|
demux_sender: &mpsc::Sender<Bytes>,
|
||||||
|
) {
|
||||||
|
let parked = early_arrivals
|
||||||
|
.lock()
|
||||||
|
.remove(&channel_id)
|
||||||
|
.unwrap_or_default();
|
||||||
|
channels.insert(channel_id, state);
|
||||||
|
let mut leftovers = VecDeque::new();
|
||||||
|
for payload in parked {
|
||||||
|
if let Err(error) = demux_sender.try_send(payload) {
|
||||||
|
match error {
|
||||||
|
TrySendError::Full(bounced) => {
|
||||||
|
debug!(
|
||||||
|
channel_id,
|
||||||
|
"install: receiver buffer full draining early arrivals"
|
||||||
|
);
|
||||||
|
leftovers.push_back(bounced);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
TrySendError::Closed(_) => {
|
||||||
|
debug!(
|
||||||
|
channel_id,
|
||||||
|
"install: receiver dropped, dropping early arrival"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !leftovers.is_empty() {
|
||||||
|
let mut arrivals = early_arrivals.lock();
|
||||||
|
let queue = arrivals.entry(channel_id).or_default();
|
||||||
|
for payload in leftovers {
|
||||||
|
queue.push_back(payload);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// `true` if `channel_id` is currently open.
|
/// `true` if `channel_id` is currently open.
|
||||||
pub fn has_channel(&self, channel_id: u32) -> bool {
|
pub fn has_channel(&self, channel_id: u32) -> bool {
|
||||||
self.inner.channels.lock().contains_key(&channel_id)
|
self.inner.channels.lock().contains_key(&channel_id)
|
||||||
@@ -1134,4 +1275,140 @@ mod tests {
|
|||||||
Ok(_) => panic!("expected TooManyChannels, got Ok"),
|
Ok(_) => panic!("expected TooManyChannels, got Ok"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Review 013 regression: a chunk routed through the adoption
|
||||||
|
/// window must not overtake earlier-parked chunks on the receiver's
|
||||||
|
/// byte stream. The racer routes `C` as soon as the install is
|
||||||
|
/// visible — under the pre-fix code and real preemption that could
|
||||||
|
/// land in the install→drain window and be delivered first (with
|
||||||
|
/// P1/P2 parked); the install and its park-drain being one lock
|
||||||
|
/// section makes P1, P2, C the only reachable order, whichever
|
||||||
|
/// side of the install the racer's route lands on (pre-install →
|
||||||
|
/// parked and drained FIFO; post-install → sent after the park
|
||||||
|
/// queue already went out). 200 iterations to give a real
|
||||||
|
/// scheduler the chance to catch the old window; the receiver's
|
||||||
|
/// byte stream must be byte-identical to the producer's every time.
|
||||||
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||||
|
async fn adopt_window_does_not_reorder_chunks() {
|
||||||
|
use std::time::Duration;
|
||||||
|
use tokio::io::AsyncReadExt;
|
||||||
|
|
||||||
|
for iteration in 0..200u32 {
|
||||||
|
let manager = make_manager_with_runner().await;
|
||||||
|
let channel_id = 7 + iteration * 2;
|
||||||
|
|
||||||
|
manager
|
||||||
|
.route_payload(channel_id, Bytes::from_static(b"P1"))
|
||||||
|
.await;
|
||||||
|
manager
|
||||||
|
.route_payload(channel_id, Bytes::from_static(b"P2"))
|
||||||
|
.await;
|
||||||
|
|
||||||
|
let adopt_manager = manager.clone();
|
||||||
|
let adopter = tokio::spawn(async move {
|
||||||
|
adopt_manager
|
||||||
|
.adopt_channel(channel_id, "alk/tty", None)
|
||||||
|
.await
|
||||||
|
.expect("adopt")
|
||||||
|
});
|
||||||
|
let race_manager = manager.clone();
|
||||||
|
let racer = tokio::spawn(async move {
|
||||||
|
loop {
|
||||||
|
if race_manager.has_channel(channel_id) {
|
||||||
|
race_manager
|
||||||
|
.route_payload(channel_id, Bytes::from_static(b"C"))
|
||||||
|
.await;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
tokio::task::yield_now().await;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
let (_send, mut recv) = adopter.await.expect("adopt task join");
|
||||||
|
racer.await.expect("race task join");
|
||||||
|
let mut buf = [0u8; 5];
|
||||||
|
tokio::time::timeout(Duration::from_secs(10), recv.read_exact(&mut buf))
|
||||||
|
.await
|
||||||
|
.expect("read within timeout")
|
||||||
|
.expect("read all");
|
||||||
|
assert_eq!(
|
||||||
|
&buf, b"P1P2C",
|
||||||
|
"iteration {iteration}: the receiver's byte stream diverged from
|
||||||
|
the producer's wire order (C overtook parked chunks)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Parked chunks routed while the install is in flight (map still
|
||||||
|
/// empty) are drained in FIFO order after the parked prefix — the
|
||||||
|
/// park decision is atomic with the map state, so a chunk can only
|
||||||
|
/// miss the map while the adopter has not inserted yet, and that
|
||||||
|
/// adopter swaps the whole queue out when it does.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn adopt_drains_parked_and_raced_chunks_in_fifo() {
|
||||||
|
let manager = make_manager_with_runner().await;
|
||||||
|
|
||||||
|
manager.route_payload(7, Bytes::from_static(b"P1")).await;
|
||||||
|
manager.route_payload(7, Bytes::from_static(b"P2")).await;
|
||||||
|
|
||||||
|
let adopt_manager = manager.clone();
|
||||||
|
let adopter = tokio::spawn(async move {
|
||||||
|
adopt_manager
|
||||||
|
.adopt_channel(7, "alk/tty", None)
|
||||||
|
.await
|
||||||
|
.expect("adopt")
|
||||||
|
});
|
||||||
|
|
||||||
|
while !manager.has_channel(7) {
|
||||||
|
tokio::task::yield_now().await;
|
||||||
|
}
|
||||||
|
|
||||||
|
let (_send, mut recv) = adopter.await.expect("adopt task join");
|
||||||
|
manager.route_payload(7, Bytes::from_static(b"C")).await;
|
||||||
|
|
||||||
|
use tokio::io::AsyncReadExt;
|
||||||
|
let mut buf = [0u8; 5];
|
||||||
|
recv.read_exact(&mut buf).await.expect("read all");
|
||||||
|
assert_eq!(&buf, b"P1P2C", "raced chunk must follow the parked prefix");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The effective park cap is the nominal EARLY_ARRIVAL_CAP bounded
|
||||||
|
/// by the receiver's buffer capacity (review 013 — the install
|
||||||
|
/// drains the parked queue with non-blocking sends into the fresh
|
||||||
|
/// receiver, which must always fit). With buffer_cap 4 the fifth
|
||||||
|
/// parked chunk drops at park time (the documented E-04 cap
|
||||||
|
/// behavior, tightened) and the first four arrive on adopt.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn park_cap_bounded_by_receiver_buffer() {
|
||||||
|
let (_client, server) = duplex(1024);
|
||||||
|
let (_reader, writer) = tokio::io::split(server);
|
||||||
|
let (handle, runner) = MuxRunner::new(Box::new(writer));
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let _ = runner.run().await;
|
||||||
|
});
|
||||||
|
let manager = ChannelManager::new(handle, 256, 4, None, ChannelSide::Accept);
|
||||||
|
|
||||||
|
for chunk in b"abcde" {
|
||||||
|
manager
|
||||||
|
.route_payload(7, Bytes::copy_from_slice(&[*chunk]))
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
assert_eq!(
|
||||||
|
manager.dropped_unknown_chunks(),
|
||||||
|
1,
|
||||||
|
"the fifth parked chunk dropped (park cap = min(64, buffer_cap=4))"
|
||||||
|
);
|
||||||
|
|
||||||
|
let (_send, mut recv) = manager
|
||||||
|
.adopt_channel(7, "alk/tty", None)
|
||||||
|
.await
|
||||||
|
.expect("adopt");
|
||||||
|
|
||||||
|
use tokio::io::AsyncReadExt;
|
||||||
|
let mut buf = [0u8; 4];
|
||||||
|
recv.read_exact(&mut buf)
|
||||||
|
.await
|
||||||
|
.expect("read parked quartet");
|
||||||
|
assert_eq!(&buf, b"abcd", "parked chunks arrive in FIFO order");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user