diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a598f33..6da2a06d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 #### Data plane +- A peer that stops reading can no longer stall the node. TCP, Tor, Nym and + BLE wrote to their links directly from the caller's task, and a write + blocks once the peer's receive window and this node's send buffer are both + full — which is what a peer that has gone away, or whose path has just + changed medium, produces. The callers are the rx loop's tick handlers, so a + single unresponsive peer held every other arm of the event loop behind it: + other peers' liveness, forwarding, and control RPCs. Each connection now has + its own writer task owning the write half, and sending is an enqueue onto a + bounded queue, so the only code that can await the wire is a task with + nothing else to do. A peer that stops draining fills its queue and its sends + then fail immediately, which is the signal the caller's retry and liveness + handling already expects. Frames are written whole, and a write error takes + the connection down with it, so a peer never sees a partial frame it cannot + resynchronise from. BLE was the worst of the four: it awaited the L2CAP + write while holding the connection-pool mutex, so one unresponsive peer + froze every other BLE operation as well — connects, evictions, and each + receive loop's teardown. + - A per-peer `connect()`-ed UDP socket is no longer left pinned to an interface the host has moved off. Established UDP peers get their own socket for the send fast path; `open_connected_fd` binds the wildcard and then calls @@ -34,9 +52,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 peer on a connectionless transport is also heartbeated at once, so the far side re-pins to the new source address rather than waiting out its own heartbeat interval. A peer on a connection-oriented transport keeps the - periodic heartbeat instead: that send awaits an unbounded `write_all` on a - stream the medium change has very likely just stranded, and this reaction - runs on the rx loop. Measured on a live + periodic heartbeat instead. That was because such a send awaited an unbounded + `write_all` on a stream the medium change had very likely just stranded, and + this reaction runs on the rx loop; the writer-task change below removes that + hazard, so widening the fan-out to those transports is now open work rather + than something the design forbids. Measured on a live node, a WLAN/LAN switch in either direction now costs no reconnection at all — the Noise session, tree position and routes survive it. Linux and macOS (the platforms with the connected-socket fast path); elsewhere the heartbeat alone diff --git a/src/node/handlers/netmon.rs b/src/node/handlers/netmon.rs index 452b46d2..89ae4e0e 100644 --- a/src/node/handlers/netmon.rs +++ b/src/node/handlers/netmon.rs @@ -41,11 +41,11 @@ //! side re-pins on receipt instead of waiting out its own //! `heartbeat_interval_secs`. Without it the forward direction is fixed but //! the reverse still points at the old address until the node next happens -//! to send. This runs on the rx loop, so it covers the connectionless +//! to send. This runs on the rx loop, and it covers the connectionless //! transports only — see -//! [`Node::heartbeat_all_peers_after_net_change`] for why awaiting a -//! connection-oriented write here would hold the loop, and what a peer on -//! one of those gets instead. +//! [`Node::heartbeat_all_peers_after_net_change`] for what a peer on a +//! connection-oriented transport gets instead, and for why that filter has +//! outlived the reason it was written for. //! //! Nothing here tears a peering down. On a live node both WLAN→LAN and //! LAN→WLAN now cost no reconnection at all — the Noise session, the tree @@ -112,21 +112,24 @@ impl Node { /// learns the node's new source address in one RTT rather than at the next /// due interval. Returns how many went out. /// - /// The filter is not an optimisation. A connectionless transport's send - /// completes without ever awaiting the wire: the UDP fast path hands the - /// frame to the encrypt workers and returns, and a raw datagram write does - /// not wait for a peer. A connection-oriented one awaits `write_all` on a - /// stream, unbounded — the connect above it is wrapped in a timeout, the - /// write is not — and a medium change is precisely the condition that - /// leaves a send window full against a path that has just gone away. This - /// runs on the rx loop, so that write would hold every other arm of the - /// select for as long as the stranded socket takes to fail. + /// The filter was written for a hazard that no longer exists, and it is + /// kept deliberately rather than by oversight. It was this: a + /// connectionless transport's send completes without ever awaiting the + /// wire, because the UDP fast path hands the frame to the encrypt workers + /// and returns and a raw datagram write does not wait for a peer, while a + /// connection-oriented one awaited `write_all` on a stream, unbounded. A + /// medium change is precisely the condition that leaves a send window full + /// against a path that has just gone away, and this runs on the rx loop, so + /// that write held every other arm of the select for as long as the + /// stranded socket took to fail. /// - /// Bounding it with a timeout is not the fix either: dropping a partial - /// `write_all` would leave a half-written frame on the stream, which the - /// peer cannot resynchronise from. Nor can the fan-out simply be spawned, - /// because the send needs `&mut self` for the session counter and the MMP - /// sender record. + /// Every connection-oriented send now enqueues onto its connection's + /// bounded queue and returns, so none of them can await the wire from here. + /// Widening the fan-out to those transports is therefore open work rather + /// than something the send path forbids; it is left out of the change that + /// removed the hazard so the two stay separable. Note that the fan-out + /// still cannot simply be spawned, because the send needs `&mut self` for + /// the session counter and the MMP sender record. /// /// So a peer on TCP, Tor, Nym or BLE keeps the periodic heartbeat it had /// before this detector existed. It is not stranded by the omission: those diff --git a/src/transport/ble/io_android.rs b/src/transport/ble/io_android.rs index 566bbbba..0d901a0d 100644 --- a/src/transport/ble/io_android.rs +++ b/src/transport/ble/io_android.rs @@ -700,6 +700,17 @@ impl BleStream for AndroidStream { // this out via `next_send`. The queue is shallow (`SEND_QUEUE_CAP`) and // this waits for a slot rather than dropping, so backpressure reaches // the layers above instead of the link bufferbloating. + // + // The wait is only safe because of who calls it, and there are exactly + // two callers. The connection's writer task is one: an embedder that + // stops draining parks that task alone, and the layer above it is the + // connection's own bounded queue, which fills and starts refusing + // sends without waiting on anything. `pubkey_exchange` is the other, + // and it is safe for a different reason: it wraps this call in + // `tokio::time::timeout`, so its wait is bounded whatever the embedder + // does. Adding a third caller on a caller's task would put an + // unbounded wait back on the rx loop, which is the defect the writer + // task exists to remove. let mut payload = data.to_vec(); loop { if self.closed.load(Ordering::Relaxed) { diff --git a/src/transport/ble/mod.rs b/src/transport/ble/mod.rs index ee81c34d..d2dcb450 100644 --- a/src/transport/ble/mod.rs +++ b/src/transport/ble/mod.rs @@ -373,12 +373,19 @@ impl BleTransport { addr: &TransportAddr, data: &[u8], ) -> Result { - let pool = self.pool.lock().await; - let conn = match pool.get(addr) { - Some(c) => c, + // Take the MTU and the connection's send queue, then release the pool + // lock. Everything after this point must be lock-free: the previous + // shape awaited the L2CAP write while still holding this guard, so a + // peer that stopped draining blocked not just its own sender but every + // other BLE operation — connect, eviction, the receive loops' teardown. + let found = { + let pool = self.pool.lock().await; + pool.get(addr) + .map(|c| (c.effective_mtu() as usize, c.send_tx.clone())) + }; + let (mtu, send_tx) = match found { + Some(pair) => pair, None => { - // Drop pool lock before triggering background connect - drop(pool); // Fire-and-forget: connect_async spawns a background task let _ = self.connect_async(addr).await; return Err(TransportError::SendFailed("not connected".into())); @@ -386,7 +393,6 @@ impl BleTransport { }; // MTU check - let mtu = conn.effective_mtu() as usize; if data.len() > mtu { self.stats.record_mtu_exceeded(); return Err(TransportError::MtuExceeded { @@ -395,19 +401,27 @@ impl BleTransport { }); } - match conn.stream.send(data).await { - Ok(()) => { - self.stats.record_send(data.len()); - Ok(data.len()) - } - Err(e) => { + // Queue the frame for the connection's writer task. `try_send`, not + // `send`: waiting for a slot is the same stall in a different shape, + // which is what the Android backend's own queue does one layer down. + // The byte count is what was queued; bytes on the link are recorded by + // the writer task. + match send_tx.try_send(data.to_vec()) { + Ok(()) => Ok(data.len()), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { self.stats.record_send_error(); - // Drop pool lock before removing to avoid deadlock - drop(pool); - let mut pool = self.pool.lock().await; - pool.remove(addr); - warn!(addr = %addr, error = %e, "BLE send failed, connection removed"); - Err(e) + debug!( + addr = %addr, + depth = pool::SEND_QUEUE_DEPTH, + "BLE outbound queue full; peer is not draining" + ); + Err(TransportError::SendFailed( + "outbound queue full: peer not draining".into(), + )) + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + self.stats.record_send_error(); + Err(TransportError::SendFailed("connection writer gone".into())) } } } @@ -509,8 +523,19 @@ impl BleTransport { recv_mtu, )); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(send_loop( + Arc::clone(&stream), + send_rx, + addr.clone(), + Arc::clone(&self.pool), + Arc::clone(&self.stats), + )); + let conn = BleConnection { stream, + send_tx, + send_task: Some(send_task), recv_task: Some(recv_task), send_mtu, recv_mtu, @@ -630,8 +655,19 @@ impl BleTransport { recv_mtu, )); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(send_loop( + Arc::clone(&stream), + send_rx, + addr_clone.clone(), + Arc::clone(&pool), + Arc::clone(&stats), + )); + let conn = BleConnection { stream, + send_tx, + send_task: Some(send_task), recv_task: Some(recv_task), send_mtu, recv_mtu, @@ -1093,8 +1129,19 @@ async fn admit_inbound( recv_mtu, )); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(send_loop( + Arc::clone(&stream), + send_rx, + ta.clone(), + Arc::clone(&pool), + Arc::clone(&stats), + )); + let conn = BleConnection { stream, + send_tx, + send_task: Some(send_task), recv_task: Some(recv_task), send_mtu, recv_mtu, @@ -1125,6 +1172,39 @@ async fn admit_inbound( stats.record_connection_accepted(); } +/// Per-connection writer task: the only place a BLE write is ever awaited. +/// +/// The BLE case was the worst of the connection-oriented transports. The write +/// was awaited by the caller *while holding the pool mutex*, so a peer that +/// stopped draining its L2CAP link blocked every other BLE operation as well +/// as the caller's task — connects, evictions and each receive loop's +/// teardown all queue behind that one guard. Moving the write here removes +/// both halves of that: the caller enqueues and returns, and the pool lock is +/// never held across the link. +/// +/// On a write error the connection is removed from the pool, which is where +/// the old inline path put it too. Dropping the pool entry aborts this task +/// and the receive task through `BleConnection`'s `Drop`. +async fn send_loop( + stream: Arc, + mut frames: tokio::sync::mpsc::Receiver>, + addr: TransportAddr, + pool: Arc>>>, + stats: Arc, +) { + while let Some(frame) = frames.recv().await { + match stream.send(&frame).await { + Ok(()) => stats.record_send(frame.len()), + Err(e) => { + stats.record_send_error(); + warn!(addr = %addr, error = %e, "BLE send failed, connection removed"); + pool.lock().await.remove(&addr); + return; + } + } + } +} + /// Receive loop: reads packets from a BLE stream and delivers to node. /// /// Takes the connection's `BleStreamRead` — already positioned past the @@ -1560,8 +1640,19 @@ async fn scan_probe_loop( recv_mtu, )); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(send_loop( + Arc::clone(&stream), + send_rx, + ta.clone(), + Arc::clone(&pool), + Arc::clone(&stats), + )); + let conn = BleConnection { stream, + send_tx, + send_task: Some(send_task), recv_task: Some(recv_task), send_mtu, recv_mtu, @@ -1628,6 +1719,7 @@ mod tests { use crate::transport::framing::build_established_frame; use io::{MockBleIo, MockBleStream}; use secp256k1::{Secp256k1, SecretKey}; + use std::time::Duration; // ------------------------------------------------------------------ // PendingProbes — the retry/backoff policy for discovered addresses @@ -1909,6 +2001,93 @@ mod tests { (transport, rx) } + /// **The property the writer task exists to guarantee, on the transport + /// where it mattered most.** + /// + /// BLE was the worst of the connection-oriented transports: `send_async` + /// awaited the L2CAP write *while holding the pool mutex*, so a peer that + /// stopped draining blocked not only its own sender but every other BLE + /// operation — connects, evictions, and each receive loop's teardown all + /// queue behind that guard. + /// + /// The mock stream's send half is a bounded channel, so a peer that never + /// reads is a peer whose link has stopped draining. This inserts such a + /// connection with a real writer task behind it, pushes far more than + /// either queue holds, and asserts that every call returns promptly and + /// that the pool stays lockable throughout. + #[tokio::test] + async fn a_ble_peer_that_stops_reading_cannot_block_the_sender_or_the_pool() { + let io = MockBleIo::new("hci0", test_addr(1)); + let (transport, _rx) = make_transport(io); + + // `_deaf` is the far end. Holding it without ever calling `recv` is + // what makes this a stalled link rather than a closed one. + let (near, _deaf) = MockBleStream::pair(test_addr(1), test_addr(2), 2048); + let stream = Arc::new(near); + let ta = TransportAddr::from_string("AA:BB:CC:DD:EE:02"); + + let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(send_loop( + Arc::clone(&stream), + send_rx, + ta.clone(), + Arc::clone(&transport.pool), + Arc::clone(&transport.stats), + )); + + transport + .pool + .lock() + .await + .insert( + ta.clone(), + BleConnection { + stream, + send_tx, + send_task: Some(send_task), + recv_task: None, + send_mtu: 2048, + recv_mtu: 2048, + established_at: tokio::time::Instant::now(), + is_static: false, + addr: test_addr(2), + node_addr: None, + }, + ) + .unwrap(); + + let frame = vec![0xAB; 512]; + let mut queued = 0usize; + let mut refused = 0usize; + for _ in 0..512 { + match tokio::time::timeout(Duration::from_secs(2), transport.send_async(&ta, &frame)) + .await + { + Ok(Ok(_)) => queued += 1, + Ok(Err(_)) => refused += 1, + Err(_) => panic!( + "send blocked on a BLE peer that stopped reading; the write is back \ + on the caller's task" + ), + } + + // The pool must stay available the whole time. Before the writer + // task this guard was held across the L2CAP write, so a stalled + // link froze every other BLE operation too. + let guard = tokio::time::timeout(Duration::from_millis(100), transport.pool.lock()) + .await + .expect("the pool lock must never be held across a BLE write"); + drop(guard); + } + + assert!(queued > 0, "the first sends must be accepted"); + assert!( + refused > 0, + "a BLE peer that never drains must eventually have sends refused rather \ + than queued without bound: queued={queued}" + ); + } + #[test] fn test_transport_type() { let io = MockBleIo::new("hci0", test_addr(1)); @@ -2137,6 +2316,8 @@ mod tests { ta.clone(), BleConnection { stream: Arc::new(parked), + send_tx: tokio::sync::mpsc::channel(1).0, + send_task: None, recv_task: None, send_mtu: 2048, recv_mtu: 2048, @@ -2808,6 +2989,8 @@ mod tests { ta.clone(), BleConnection { stream: Arc::new(parked), + send_tx: tokio::sync::mpsc::channel(1).0, + send_task: None, recv_task: None, send_mtu: 64, recv_mtu: 64, diff --git a/src/transport/ble/pool.rs b/src/transport/ble/pool.rs index 837b1c0c..13c7e789 100644 --- a/src/transport/ble/pool.rs +++ b/src/transport/ble/pool.rs @@ -13,10 +13,23 @@ use crate::transport::{TransportAddr, TransportError}; use super::addr::BleAddr; +/// How many frames may be queued for one BLE connection before sends to it +/// fail. Shallower than the IP transports': a BLE link carries a fraction of +/// their throughput, so a queue of the same depth would represent seconds of +/// backlog rather than a burst. +pub const SEND_QUEUE_DEPTH: usize = 16; + /// A single BLE connection in the pool. pub struct BleConnection { /// The L2CAP stream for this connection. pub stream: S, + /// Frames queued for the writer task. Sending is an enqueue, never a + /// write: the write is awaited only by `send_task`, so no caller can be + /// held by a peer that has stopped draining — and, on this transport in + /// particular, no caller holds the pool lock while it happens. + pub send_tx: tokio::sync::mpsc::Sender>, + /// Writer task for this connection. + pub send_task: Option>, /// Background receive task handle. pub recv_task: Option>, /// Negotiated L2CAP send MTU. @@ -53,6 +66,9 @@ impl Drop for BleConnection { if let Some(task) = self.recv_task.take() { task.abort(); } + if let Some(task) = self.send_task.take() { + task.abort(); + } } } @@ -239,6 +255,8 @@ mod tests { fn test_conn(n: u8, is_static: bool) -> BleConnection<()> { BleConnection { stream: (), + send_tx: tokio::sync::mpsc::channel(1).0, + send_task: None, recv_task: None, send_mtu: 2048, recv_mtu: 2048, diff --git a/src/transport/nym/mod.rs b/src/transport/nym/mod.rs index 243d09a9..124d6ab6 100644 --- a/src/transport/nym/mod.rs +++ b/src/transport/nym/mod.rs @@ -20,8 +20,9 @@ use super::{ }; use crate::config::NymConfig; use crate::transport::socks5::{ - ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, Socks5Auth, - Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, + ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, SEND_QUEUE_DEPTH, + Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, + proxied_send_loop, }; use stats::NymStats; @@ -29,12 +30,10 @@ use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; -use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; -use tokio::net::tcp::OwnedWriteHalf; use tokio::sync::Mutex; use tokio::time::Instant; -use tracing::{debug, info, trace, warn}; +use tracing::{debug, info, warn}; // ============================================================================ // Nym Transport @@ -214,6 +213,7 @@ impl NymTransport { let mut pool = self.pool.lock().await; for (addr, conn) in pool.drain() { conn.recv_task.abort(); + conn.send_task.abort(); let _ = conn.recv_task.await; debug!( transport_id = %self.transport_id, @@ -256,42 +256,45 @@ impl NymTransport { }); } - // Get or create connection - let writer = { + // Get or create the connection's send queue. Never the write half: + // this function must not be able to await the wire (see + // `proxied_send_loop`). + let send_tx = { let pool = self.pool.lock().await; - pool.get(addr).map(|c| c.writer.clone()) + pool.get(addr).map(|c| c.send_tx.clone()) }; - let writer = match writer { - Some(w) => w, + let send_tx = match send_tx { + Some(tx) => tx, None => { // Connect-on-send self.connect(addr).await? } }; - // Write packet - let mut w = writer.lock().await; - match w.write_all(data).await { - Ok(()) => { - self.stats.record_send(data.len()); - trace!( + // Queue the frame. `try_send`, not `send`: awaiting a full queue would + // reinstate one level up exactly the block this removes. The byte + // count is what was queued; bytes on the wire are recorded by the + // writer task. + match send_tx.try_send(data.to_vec()) { + Ok(()) => Ok(data.len()), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + self.stats.record_send_error(); + debug!( transport_id = %self.transport_id, remote_addr = %addr, - bytes = data.len(), - "Nym packet sent" + depth = SEND_QUEUE_DEPTH, + "Nym outbound queue full; peer is not draining" ); - Ok(data.len()) + Err(TransportError::SendFailed( + "outbound queue full: peer not draining".to_string(), + )) } - Err(e) => { + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { self.stats.record_send_error(); - drop(w); - // Remove failed connection from pool - let mut pool = self.pool.lock().await; - if let Some(conn) = pool.remove(addr) { - conn.recv_task.abort(); - } - Err(TransportError::SendFailed(format!("{}", e))) + Err(TransportError::SendFailed( + "connection writer gone".to_string(), + )) } } } @@ -300,7 +303,7 @@ impl NymTransport { async fn connect( &self, addr: &TransportAddr, - ) -> Result>, TransportError> { + ) -> Result>, TransportError> { let target_addr = parse_target_addr(addr)?; let proxy_addr = self.config.socks5_addr(); let timeout_ms = self.config.connect_timeout_ms(); @@ -348,7 +351,6 @@ impl NymTransport { // Split and spawn receive task let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let transport_id = self.transport_id; let packet_tx = self.packet_tx.clone(); @@ -370,8 +372,21 @@ impl NymTransport { .await; }); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(proxied_send_loop( + write_half, + send_rx, + transport_id, + addr.clone(), + self.pool.clone(), + self.stats.clone(), + "Nym", + |_stats: &NymStats, _meta: &()| {}, + )); + let conn = ProxiedConnection { - writer: writer.clone(), + send_tx: send_tx.clone(), + send_task, recv_task, mtu, established_at: Instant::now(), @@ -390,7 +405,7 @@ impl NymTransport { "Nym mixnet connection established via SOCKS5" ); - Ok(writer) + Ok(send_tx) } /// Initiate a non-blocking connection to a remote address. @@ -500,7 +515,6 @@ impl NymTransport { /// Promote a completed background connection to the established pool. fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mtu: u16) { let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let transport_id = self.transport_id; let packet_tx = self.packet_tx.clone(); @@ -521,8 +535,21 @@ impl NymTransport { .await; }); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(proxied_send_loop( + write_half, + send_rx, + transport_id, + addr.clone(), + self.pool.clone(), + self.stats.clone(), + "Nym", + |_stats: &NymStats, _meta: &()| {}, + )); + let conn = ProxiedConnection { - writer, + send_tx, + send_task, recv_task, mtu, established_at: Instant::now(), @@ -539,6 +566,7 @@ impl NymTransport { ); } else { conn.recv_task.abort(); + conn.send_task.abort(); warn!( transport_id = %self.transport_id, remote_addr = %addr, @@ -552,6 +580,7 @@ impl NymTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); + conn.send_task.abort(); debug!( transport_id = %self.transport_id, remote_addr = %addr, diff --git a/src/transport/nym/stats.rs b/src/transport/nym/stats.rs index f349c93d..4199161d 100644 --- a/src/transport/nym/stats.rs +++ b/src/transport/nym/stats.rs @@ -94,6 +94,14 @@ impl ProxiedStats for NymStats { fn record_recv_error(&self) { self.base.record_recv_error(); } + + fn record_send(&self, bytes: usize) { + NymStats::record_send(self, bytes); + } + + fn record_send_error(&self) { + NymStats::record_send_error(self); + } } /// Point-in-time snapshot of Nym stats (non-atomic, copyable). diff --git a/src/transport/socks5/mod.rs b/src/transport/socks5/mod.rs index eb628830..4a0911ab 100644 --- a/src/transport/socks5/mod.rs +++ b/src/transport/socks5/mod.rs @@ -11,8 +11,8 @@ mod stats; pub use dialer::{DialError, Socks5Auth, Socks5Dialer, SocksTarget}; pub(crate) use pool::{ - ConnectingEntry, ConnectingPool, ProxiedConnection, ProxiedPool, ProxiedStats, poll_connecting, - proxied_receive_loop, + ConnectingEntry, ConnectingPool, ProxiedConnection, ProxiedPool, ProxiedStats, + SEND_QUEUE_DEPTH, poll_connecting, proxied_receive_loop, proxied_send_loop, }; pub(crate) use stats::ProxiedStatsBase; diff --git a/src/transport/socks5/pool.rs b/src/transport/socks5/pool.rs index 7e2ab7b3..1d3dab42 100644 --- a/src/transport/socks5/pool.rs +++ b/src/transport/socks5/pool.rs @@ -13,11 +13,13 @@ use std::time::Duration; use futures::FutureExt; use tokio::net::TcpStream; use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, mpsc}; use tokio::task::JoinHandle; use tokio::time::Instant; use tracing::{debug, trace}; +use tokio::io::AsyncWriteExt; + use crate::transport::framing::read_fmp_packet; use crate::transport::{ ConnectionState, PacketTx, ReceivedPacket, TransportAddr, TransportError, TransportId, @@ -28,8 +30,12 @@ use crate::transport::{ /// `M` is per-transport metadata: `Direction` for tor (drives /// inbound/outbound pool accounting), `()` for nym. pub(crate) struct ProxiedConnection { - /// Write half of the split stream. - pub writer: Arc>, + /// Frames queued for the writer task. Sending is an enqueue, never a + /// write: the write half belongs to `send_task`, so no caller can block on + /// the wire. A full queue is a peer that has stopped draining. + pub send_tx: mpsc::Sender>, + /// Writer task for this connection. + pub send_task: JoinHandle<()>, /// Receive task for this connection. pub recv_task: JoinHandle<()>, /// MTU for this connection. @@ -124,6 +130,79 @@ pub(crate) trait ProxiedStats: Send + Sync + 'static { fn record_recv(&self, bytes: usize); /// Record a receive error. fn record_recv_error(&self); + /// Record `bytes` actually written to the wire. + fn record_send(&self, bytes: usize); + /// Record a send error. + fn record_send_error(&self); +} + +/// How many frames may be queued for one connection before sends to it fail. +/// See `crate::transport::tcp::pool::SEND_QUEUE_DEPTH`, which this mirrors. +pub(crate) const SEND_QUEUE_DEPTH: usize = 64; + +/// Per-connection writer task: the only place a write to a proxied stream is +/// ever awaited. +/// +/// The reasoning is the TCP transport's, and the shape is deliberately the +/// same. `write_all` blocks once the local socket to the proxy stops draining, +/// and the callers are the rx loop's tick handlers, where that holds every +/// other arm of the select. The loop owns the write half, so nothing else can +/// block on it. +/// +/// Teardown mirrors [`proxied_receive_loop`]: the pool entry is removed and +/// `on_remove` fires only when the removal returned `Some`, taking the +/// metadata from the removed entry, so a concurrent `close`/`stop` of the same +/// address cannot double-count. +#[allow(clippy::too_many_arguments)] +pub(crate) async fn proxied_send_loop( + mut writer: OwnedWriteHalf, + mut frames: mpsc::Receiver>, + transport_id: TransportId, + remote_addr: TransportAddr, + pool: ProxiedPool, + stats: Arc, + label: &'static str, + on_remove: impl Fn(&S, &M) + Send + 'static, +) { + while let Some(frame) = frames.recv().await { + match writer.write_all(&frame).await { + Ok(()) => { + stats.record_send(frame.len()); + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = frame.len(), + "{} packet sent", + label + ); + } + Err(e) => { + stats.record_send_error(); + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + error = %e, + "{} write failed; dropping connection", + label + ); + let removed = { + let mut guard = pool.lock().await; + guard.remove(&remote_addr) + }; + if let Some(conn) = removed { + conn.recv_task.abort(); + on_remove(&stats, &conn.meta); + } + return; + } + } + } + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + "{} writer task exiting", + label + ); } /// Shared per-connection receive loop for the proxied transports. diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs index 25b29acc..c07e4689 100644 --- a/src/transport/tcp/mod.rs +++ b/src/transport/tcp/mod.rs @@ -42,9 +42,8 @@ use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use tokio::io::AsyncWriteExt; -use tokio::net::tcp::OwnedWriteHalf; use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, mpsc}; use tokio::task::JoinHandle; use tokio::time::Instant; use tracing::{debug, info, trace, warn}; @@ -271,6 +270,7 @@ impl TcpTransport { let mut pool = self.pool.lock().await; for (addr, conn) in pool.drain() { conn.recv_task.abort(); + conn.send_task.abort(); let _ = conn.recv_task.await; match conn.direction { Direction::Inbound => self.stats.record_pool_inbound_removed(), @@ -323,46 +323,56 @@ impl TcpTransport { }); } - // Get or create connection - let writer = { + // Get or create connection. What comes back is the queue into the + // connection's writer task, never the write half itself: this function + // must not be able to await the wire (see `tcp_send_loop`). + let send_tx = { let pool = self.pool.lock().await; - pool.get(addr).map(|c| c.writer.clone()) + pool.get(addr).map(|c| c.send_tx.clone()) }; - let writer = match writer { - Some(w) => w, + let send_tx = match send_tx { + Some(tx) => tx, None => { // Connect-on-send self.connect(addr).await? } }; - // Write packet directly (no framing transformation needed) - let mut w = writer.lock().await; - match w.write_all(data).await { - Ok(()) => { - self.stats.record_send(data.len()); - trace!( + // Hand the frame to the writer task. The copy buys the caller its + // freedom from the wire: `write_all` borrows, a queue must own. One + // memcpy of at most an MTU is a good trade for not stalling the rx + // loop on a peer that has stopped reading. + // + // `try_send` rather than `send`: awaiting a full queue would reinstate + // exactly the block this removes, one level up. A full queue means the + // writer task has not drained a frame in the time it took to fill 64 of + // them, which is a peer that is not receiving, so the send fails and + // the caller's own retry policy takes over. + // + // The byte count is what was queued, not what reached the wire — the + // same prediction the UDP fast path reports when it dispatches to the + // encrypt workers. Bytes actually written are recorded by the writer + // task as they go. + match send_tx.try_send(data.to_vec()) { + Ok(()) => Ok(data.len()), + Err(mpsc::error::TrySendError::Full(_)) => { + self.stats.record_send_error(); + debug!( transport_id = %self.transport_id, remote_addr = %addr, - bytes = data.len(), - "TCP packet sent" + depth = crate::transport::tcp::pool::SEND_QUEUE_DEPTH, + "TCP outbound queue full; peer is not draining" ); - Ok(data.len()) + Err(TransportError::SendFailed( + "outbound queue full: peer not draining".to_string(), + )) } - Err(e) => { + Err(mpsc::error::TrySendError::Closed(_)) => { self.stats.record_send_error(); - drop(w); - // Remove failed connection from pool - let mut pool = self.pool.lock().await; - if let Some(conn) = pool.remove(addr) { - conn.recv_task.abort(); - match conn.direction { - Direction::Inbound => self.stats.record_pool_inbound_removed(), - Direction::Outbound => self.stats.record_pool_outbound_removed(), - } - } - Err(TransportError::SendFailed(format!("{}", e))) + Err(TransportError::SendFailed( + "connection writer gone".to_string(), + )) } } } @@ -371,10 +381,7 @@ impl TcpTransport { /// /// Configures socket options, reads TCP_MAXSEG for MTU, splits the /// stream, spawns a receive task, and stores in the pool. - async fn connect( - &self, - addr: &TransportAddr, - ) -> Result>, TransportError> { + async fn connect(&self, addr: &TransportAddr) -> Result>, TransportError> { let socket_addr = resolve_socket_addr(addr).await?; let timeout_ms = self.config.connect_timeout_ms(); @@ -411,7 +418,6 @@ impl TcpTransport { // Split and spawn receive task let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let transport_id = self.transport_id; let packet_tx = self.packet_tx.clone(); @@ -438,8 +444,19 @@ impl TcpTransport { .await; }); + let (send_tx, send_rx) = mpsc::channel(crate::transport::tcp::pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(tcp_send_loop( + write_half, + send_rx, + transport_id, + addr.clone(), + self.pool.clone(), + self.stats.clone(), + )); + let conn = TcpConnection { - writer: writer.clone(), + send_tx: send_tx.clone(), + send_task, recv_task, mtu: mss_mtu, established_at: Instant::now(), @@ -459,7 +476,7 @@ impl TcpTransport { "TCP connection established (connect-on-send)" ); - Ok(writer) + Ok(send_tx) } /// Close a specific connection asynchronously. @@ -470,6 +487,7 @@ impl TcpTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); + conn.send_task.abort(); match conn.direction { Direction::Inbound => self.stats.record_pool_inbound_removed(), Direction::Outbound => self.stats.record_pool_outbound_removed(), @@ -665,7 +683,6 @@ impl TcpTransport { /// Called from `connection_state_sync()` when a background task completes. fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mss_mtu: u16) { let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let transport_id = self.transport_id; let packet_tx = self.packet_tx.clone(); @@ -691,8 +708,19 @@ impl TcpTransport { .await; }); + let (send_tx, send_rx) = mpsc::channel(crate::transport::tcp::pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(tcp_send_loop( + write_half, + send_rx, + transport_id, + addr.clone(), + self.pool.clone(), + self.stats.clone(), + )); + let conn = TcpConnection { - writer, + send_tx, + send_task, recv_task, mtu: mss_mtu, established_at: Instant::now(), @@ -714,6 +742,7 @@ impl TcpTransport { } else { // Pool locked — abort the recv task, connection will be retried conn.recv_task.abort(); + conn.send_task.abort(); warn!( transport_id = %self.transport_id, remote_addr = %addr, @@ -892,7 +921,6 @@ async fn accept_loop( // Split and spawn receive task let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let recv_pool = pool.clone(); let recv_packet_tx = packet_tx.clone(); @@ -921,8 +949,20 @@ async fn accept_loop( .await; }); + let (send_tx, send_rx) = + mpsc::channel(crate::transport::tcp::pool::SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(tcp_send_loop( + write_half, + send_rx, + transport_id, + remote_addr.clone(), + pool.clone(), + stats.clone(), + )); + let conn = TcpConnection { - writer, + send_tx, + send_task, recv_task, mtu: conn_mtu, established_at: Instant::now(), @@ -959,9 +999,80 @@ async fn accept_loop( } // ============================================================================ -// Receive Loop (per-connection) +// Per-connection Loops (writer and receiver) // ============================================================================ +/// Per-connection writer task: the only place a TCP write is ever awaited. +/// +/// This exists so the caller does not await the wire. `write_all` on a stream +/// blocks once the kernel send buffer fills, which is precisely what a peer +/// that has stopped draining causes — and the callers are the rx loop's tick +/// handlers, where blocking holds every other arm of the select behind it. The +/// loop owns the write half outright, so no other task can hold it and no +/// other task can be held by it. +/// +/// On a write error the connection is removed from the pool, mirroring +/// `tcp_receive_loop`'s teardown contract: the pool entry is removed and the +/// direction counter decremented only when the removal returned `Some`, so a +/// concurrent `close`/`stop` of the same address cannot double-count. The +/// receive task is aborted here rather than left to notice on its own, because +/// a half-closed connection is not something either side should keep. +/// +/// Frames are written whole. A partial write followed by an error takes the +/// connection down with it, so the peer never sees a frame it cannot +/// resynchronise from. +async fn tcp_send_loop( + mut writer: tokio::net::tcp::OwnedWriteHalf, + mut frames: mpsc::Receiver>, + transport_id: TransportId, + remote_addr: TransportAddr, + pool: ConnectionPool, + stats: Arc, +) { + while let Some(frame) = frames.recv().await { + match writer.write_all(&frame).await { + Ok(()) => { + stats.record_send(frame.len()); + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = frame.len(), + "TCP packet sent" + ); + } + Err(e) => { + stats.record_send_error(); + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + error = %e, + "TCP write failed; dropping connection" + ); + let removed = { + let mut pool = pool.lock().await; + pool.remove(&remote_addr) + }; + if let Some(conn) = removed { + conn.recv_task.abort(); + conn.send_task.abort(); + match conn.direction { + Direction::Inbound => stats.record_pool_inbound_removed(), + Direction::Outbound => stats.record_pool_outbound_removed(), + } + } + return; + } + } + } + // The sender side is gone: the pool entry was dropped, so the connection + // is already being torn down and there is nothing to clean up here. + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + "TCP writer task exiting" + ); +} + /// Per-connection TCP receive loop. /// /// Reads complete FMP packets using the stream reader, delivers them to @@ -1549,6 +1660,68 @@ mod tests { assert!(!transport.accept_connections()); } + /// **The property this transport's send path exists to guarantee.** + /// + /// A peer that stops reading fills its receive window, then this node's + /// kernel send buffer, and from that moment `write_all` blocks until the + /// peer drains or the connection dies. The callers are the rx loop's tick + /// handlers — the heartbeat sweep among them — so a blocking send holds + /// every other arm of the select behind it: control RPCs, forwarding, + /// every other peer's liveness. A medium change is precisely the condition + /// that produces such a peer, which is how this was found. + /// + /// The writer task owns the write half, so `send_async` can only ever + /// enqueue. This drives a peer that accepts the connection and then never + /// reads, pushes far more than any socket buffer will hold, and asserts + /// every call returns promptly — failing once the queue fills, rather than + /// blocking on a peer that is not listening. + #[tokio::test] + async fn a_peer_that_stops_reading_cannot_block_the_sender() { + let (tx1, _rx1) = packet_channel(100); + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + t1.start_async().await.unwrap(); + + // A listener that accepts and then never reads a byte. Holding the + // stream is the point: dropping it would close the connection and turn + // the writes into fast errors, which is not the case under test. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen_addr = listener.local_addr().unwrap(); + let deaf = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + tokio::time::sleep(Duration::from_secs(30)).await; + drop(stream); + }); + let remote = TransportAddr::from_string(&listen_addr.to_string()); + + // Enough to overrun any plausible socket buffer plus the queue behind + // it, so the blocking case cannot be missed by sending too little. + let frame = vec![0xAB; 1400]; + let mut queued = 0usize; + let mut refused = 0usize; + for _ in 0..8000 { + // The budget is per call and generous: a healthy enqueue is + // microseconds, while the old unbounded write parked here until + // the peer drained, which it never does. + match timeout(Duration::from_secs(2), t1.send_async(&remote, &frame)).await { + Ok(Ok(_)) => queued += 1, + Ok(Err(_)) => refused += 1, + Err(_) => panic!( + "send blocked on a peer that stopped reading; \ + the write is back on the caller's task" + ), + } + } + + assert!(queued > 0, "the first sends must be accepted"); + assert!( + refused > 0, + "a peer that never drains must eventually have sends refused rather than \ + queued without bound: queued={queued}" + ); + + deaf.abort(); + } + #[tokio::test] async fn test_connection_drop_and_reconnect() { let (tx1, _rx1) = packet_channel(100); @@ -2050,14 +2223,15 @@ mod tests { let client = TcpStream::connect(listen).await.unwrap(); let (server, peer_addr) = listener.accept().await.unwrap(); let remote = TransportAddr::from_string(&peer_addr.to_string()); - let (read_half, write_half) = server.into_split(); + let (read_half, _write_half) = server.into_split(); let pool: ConnectionPool = Arc::new(Mutex::new(HashMap::new())); let stats = Arc::new(TcpStats::new()); pool.lock().await.insert( remote.clone(), TcpConnection { - writer: Arc::new(Mutex::new(write_half)), + send_tx: mpsc::channel(1).0, + send_task: tokio::spawn(async {}), recv_task: tokio::spawn(async {}), mtu: 1400, established_at: Instant::now(), diff --git a/src/transport/tcp/pool.rs b/src/transport/tcp/pool.rs index 8777c376..e8e678ca 100644 --- a/src/transport/tcp/pool.rs +++ b/src/transport/tcp/pool.rs @@ -6,8 +6,7 @@ use std::collections::HashMap; use std::sync::Arc; use tokio::net::TcpStream; -use tokio::net::tcp::OwnedWriteHalf; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, mpsc}; use tokio::task::JoinHandle; use tokio::time::Instant; @@ -24,10 +23,26 @@ pub(crate) enum Direction { Outbound, } +/// How many frames may be queued for one connection before sends to it start +/// failing. +/// +/// The queue exists so the caller never awaits the wire; the bound exists so a +/// peer that has stopped draining cannot turn that into unbounded memory. Deep +/// enough to absorb a burst — a heartbeat sweep plus the forwarding this node +/// does for one peer — and shallow enough that a stranded peer is recognised +/// within a tick or two rather than after megabytes have piled up behind it. +pub(crate) const SEND_QUEUE_DEPTH: usize = 64; + /// State for a single TCP connection to a peer. pub(crate) struct TcpConnection { - /// Write half of the split stream. - pub(crate) writer: Arc>, + /// Frames queued for the writer task. Sending is an enqueue, never a + /// write: the write half belongs to `send_task` and nothing else can + /// block on it. A full queue is a peer that has stopped draining, and the + /// send fails rather than waiting. + pub(crate) send_tx: mpsc::Sender>, + /// Writer task for this connection. Owns the write half of the split + /// stream, so the only code that can ever await `write_all` is this task. + pub(crate) send_task: JoinHandle<()>, /// Receive task for this connection. pub(crate) recv_task: JoinHandle<()>, /// MSS-derived MTU for this connection (used for dynamic MTU re-reading). diff --git a/src/transport/tor/mod.rs b/src/transport/tor/mod.rs index 53fa65ae..96becca9 100644 --- a/src/transport/tor/mod.rs +++ b/src/transport/tor/mod.rs @@ -30,8 +30,9 @@ use super::{ }; use crate::config::TorConfig; use crate::transport::socks5::{ - ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, Socks5Auth, - Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, + ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, SEND_QUEUE_DEPTH, + Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, + proxied_send_loop, }; use crate::transport::tcp::INBOUND_FIRST_FRAME_TIMEOUT; use control::{ControlAuth, TorControlClient, TorMonitoringInfo}; @@ -42,13 +43,11 @@ use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; -use tokio::io::AsyncWriteExt; -use tokio::net::tcp::OwnedWriteHalf; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::Mutex; use tokio::task::JoinHandle; use tokio::time::Instant; -use tracing::{debug, info, trace, warn}; +use tracing::{debug, info, warn}; // ============================================================================ // Address Parsing @@ -497,6 +496,7 @@ impl TorTransport { let mut pool = self.pool.lock().await; for (addr, conn) in pool.drain() { conn.recv_task.abort(); + conn.send_task.abort(); let _ = conn.recv_task.await; match conn.meta { Direction::Inbound => self.stats.record_pool_inbound_removed(), @@ -639,45 +639,42 @@ impl TorTransport { } // Get or create connection - let writer = { + let send_tx = { let pool = self.pool.lock().await; - pool.get(addr).map(|c| c.writer.clone()) + pool.get(addr).map(|c| c.send_tx.clone()) }; - let writer = match writer { - Some(w) => w, + let send_tx = match send_tx { + Some(tx) => tx, None => { // Connect-on-send self.connect(addr).await? } }; - // Write packet directly (no framing transformation needed) - let mut w = writer.lock().await; - match w.write_all(data).await { - Ok(()) => { - self.stats.record_send(data.len()); - trace!( + // Queue the frame for the connection's writer task. `try_send`, not + // `send`: awaiting a full queue would reinstate the block this removes + // one level up. The byte count is what was queued; bytes on the wire + // are recorded by the writer task. + match send_tx.try_send(data.to_vec()) { + Ok(()) => Ok(data.len()), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + self.stats.record_send_error(); + debug!( transport_id = %self.transport_id, remote_addr = %addr, - bytes = data.len(), - "Tor packet sent" + depth = SEND_QUEUE_DEPTH, + "Tor outbound queue full; peer is not draining" ); - Ok(data.len()) + Err(TransportError::SendFailed( + "outbound queue full: peer not draining".to_string(), + )) } - Err(e) => { + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { self.stats.record_send_error(); - drop(w); - // Remove failed connection from pool - let mut pool = self.pool.lock().await; - if let Some(conn) = pool.remove(addr) { - conn.recv_task.abort(); - match conn.meta { - Direction::Inbound => self.stats.record_pool_inbound_removed(), - Direction::Outbound => self.stats.record_pool_outbound_removed(), - } - } - Err(TransportError::SendFailed(format!("{}", e))) + Err(TransportError::SendFailed( + "connection writer gone".to_string(), + )) } } } @@ -690,7 +687,7 @@ impl TorTransport { async fn connect( &self, addr: &TransportAddr, - ) -> Result>, TransportError> { + ) -> Result>, TransportError> { let tor_addr = parse_tor_addr(addr)?; let proxy_addr = self.config.socks5_addr(); let timeout_ms = self.config.connect_timeout_ms(); @@ -752,7 +749,6 @@ impl TorTransport { // Split and spawn receive task let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let transport_id = self.transport_id; let packet_tx = self.packet_tx.clone(); @@ -779,8 +775,24 @@ impl TorTransport { .await; }); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(proxied_send_loop( + write_half, + send_rx, + transport_id, + addr.clone(), + self.pool.clone(), + self.stats.clone(), + "Tor", + |stats: &TorStats, meta: &Direction| match meta { + Direction::Inbound => stats.record_pool_inbound_removed(), + Direction::Outbound => stats.record_pool_outbound_removed(), + }, + )); + let conn = ProxiedConnection { - writer: writer.clone(), + send_tx: send_tx.clone(), + send_task, recv_task, mtu, established_at: Instant::now(), @@ -800,7 +812,7 @@ impl TorTransport { "Tor circuit established via SOCKS5" ); - Ok(writer) + Ok(send_tx) } /// Initiate a non-blocking connection to a remote address. @@ -922,7 +934,6 @@ impl TorTransport { /// Called from `connection_state_sync()` when a background task completes. fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mtu: u16) { let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let transport_id = self.transport_id; let packet_tx = self.packet_tx.clone(); @@ -948,8 +959,24 @@ impl TorTransport { .await; }); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(proxied_send_loop( + write_half, + send_rx, + transport_id, + addr.clone(), + self.pool.clone(), + self.stats.clone(), + "Tor", + |stats: &TorStats, meta: &Direction| match meta { + Direction::Inbound => stats.record_pool_inbound_removed(), + Direction::Outbound => stats.record_pool_outbound_removed(), + }, + )); + let conn = ProxiedConnection { - writer, + send_tx, + send_task, recv_task, mtu, established_at: Instant::now(), @@ -970,6 +997,7 @@ impl TorTransport { } else { // Pool locked — abort the recv task, connection will be retried conn.recv_task.abort(); + conn.send_task.abort(); warn!( transport_id = %self.transport_id, remote_addr = %addr, @@ -983,6 +1011,7 @@ impl TorTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); + conn.send_task.abort(); match conn.meta { Direction::Inbound => self.stats.record_pool_inbound_removed(), Direction::Outbound => self.stats.record_pool_outbound_removed(), @@ -1196,7 +1225,6 @@ async fn tor_accept_loop( // Split stream and spawn receive task let (read_half, write_half) = stream.into_split(); - let writer = Arc::new(Mutex::new(write_half)); let recv_pool = pool.clone(); let recv_stats = stats.clone(); @@ -1225,8 +1253,24 @@ async fn tor_accept_loop( .await; }); + let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH); + let send_task = tokio::spawn(proxied_send_loop( + write_half, + send_rx, + transport_id, + remote_addr.clone(), + pool.clone(), + stats.clone(), + "Tor", + |stats: &TorStats, meta: &Direction| match meta { + Direction::Inbound => stats.record_pool_inbound_removed(), + Direction::Outbound => stats.record_pool_outbound_removed(), + }, + )); + let conn = ProxiedConnection { - writer, + send_tx, + send_task, recv_task, mtu, established_at: Instant::now(), @@ -1294,6 +1338,7 @@ fn validate_host_port(addr: &str, field_name: &str) -> Result<(), TransportError mod tests { use super::*; use crate::transport::packet_channel; + use tokio::io::AsyncWriteExt; fn make_config() -> TorConfig { TorConfig { @@ -2090,16 +2135,6 @@ mod tests { // Accept-loop readiness barrier // ======================================================================== - /// Build a throwaway `OwnedWriteHalf` for a hand-planted pool entry. - async fn spare_write_half() -> OwnedWriteHalf { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - let client = TcpStream::connect(addr).await.unwrap(); - let (_server, _) = listener.accept().await.unwrap(); - let (_read, write) = client.into_split(); - write - } - /// The accept loop can be torn down between the pool insert and the /// `ready_tx.send()`: the sender is dropped, so `ready_rx.await` returns /// `Err`. The receive loop must still fall through to its cleanup, or the @@ -2114,14 +2149,15 @@ mod tests { let client = TcpStream::connect(listen).await.unwrap(); let (server, peer_addr) = listener.accept().await.unwrap(); let remote = TransportAddr::from_string(&peer_addr.to_string()); - let (read_half, write_half) = server.into_split(); + let (read_half, _write_half) = server.into_split(); let pool: ProxiedPool = Arc::new(Mutex::new(HashMap::new())); let stats = Arc::new(TorStats::new()); pool.lock().await.insert( remote.clone(), ProxiedConnection { - writer: Arc::new(Mutex::new(write_half)), + send_tx: tokio::sync::mpsc::channel(1).0, + send_task: tokio::spawn(async {}), recv_task: tokio::spawn(async {}), mtu: 1400, established_at: Instant::now(), @@ -2178,7 +2214,7 @@ mod tests { let (server, peer_addr) = listener.accept().await.unwrap(); let remote = TransportAddr::from_string(&peer_addr.to_string()); drop(client); - let (read_half, write_half) = server.into_split(); + let (read_half, _write_half) = server.into_split(); let pool: ProxiedPool = Arc::new(Mutex::new(HashMap::new())); let stats = Arc::new(TorStats::new()); @@ -2214,7 +2250,8 @@ mod tests { pool.lock().await.insert( remote.clone(), ProxiedConnection { - writer: Arc::new(Mutex::new(write_half)), + send_tx: tokio::sync::mpsc::channel(1).0, + send_task: tokio::spawn(async {}), recv_task: tokio::spawn(async {}), mtu: 1400, established_at: Instant::now(), @@ -2270,7 +2307,8 @@ mod tests { pool.lock().await.insert( remote.clone(), ProxiedConnection { - writer: Arc::new(Mutex::new(spare_write_half().await)), + send_tx: tokio::sync::mpsc::channel(1).0, + send_task: tokio::spawn(async {}), recv_task: tokio::spawn(std::future::pending::<()>()), mtu: 1400, established_at: Instant::now(), diff --git a/src/transport/tor/stats.rs b/src/transport/tor/stats.rs index cf0d38cd..de90863c 100644 --- a/src/transport/tor/stats.rs +++ b/src/transport/tor/stats.rs @@ -159,6 +159,14 @@ impl ProxiedStats for TorStats { fn record_recv_error(&self) { self.base.record_recv_error(); } + + fn record_send(&self, bytes: usize) { + TorStats::record_send(self, bytes); + } + + fn record_send_error(&self) { + TorStats::record_send_error(self); + } } /// Point-in-time snapshot of Tor stats (non-atomic, copyable).