From d0f092ff06484a6120a073f833f9f7b98c193ac2 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Sat, 10 Oct 2026 12:22:06 +0200 Subject: [PATCH] session: write to the peer from a task of its own A session read from and wrote to its socket in one loop: while it was writing, it was not reading. Writing waits when the peer's receive window is closed, and the peer's window closes when the peer is not reading, which it is not while it is writing to us and waiting on our window in turn. Two routers sending each other every path at once, as they do with ADD-PATH both ways after a restart, each stopped reading within seconds and waited on the other until the send hold timer ended both sessions eight minutes later. The status page's TCP row showed it: both windows zero, the send queue full, the receive queue unread. The batching of writes in the previous change made a session spend far longer writing and so made this certain; it was possible before. Now each session has a writer task. The session encodes its messages into batches and hands them to the task through a bounded queue without ever waiting: a batch the queue has no room for is kept, and while any is kept the session takes no more messages from the rest of the server (which then waits, as before, in its own task) but goes on reading the peer. The writer task's failure, or a write that makes no progress for the send hold time, comes back to the session as a local failure and ends it. When the session ends, what is still to go out, a NOTIFICATION among it, is handed over and the writer shuts our side once it has written it. Testing: a new integration test has one peer's table in the server and a second peer send a table of its own while reading nothing of what the server sends it, and checks that the server takes the second table all the same; against the previous commit the test hangs until its timeout. The send hold timer test now reads the writer's failure. The session tests, the integration tests, the kernel tests and every lab phase pass. Co-Authored-By: Claude Fable 5.1 --- crates/integration-tests/tests/deadlock.rs | 92 +++++++++++ crates/server/src/connection.rs | 171 ++++++++++++++++----- 2 files changed, 226 insertions(+), 37 deletions(-) create mode 100644 crates/integration-tests/tests/deadlock.rs diff --git a/crates/integration-tests/tests/deadlock.rs b/crates/integration-tests/tests/deadlock.rs new file mode 100644 index 0000000..f34d11c --- /dev/null +++ b/crates/integration-tests/tests/deadlock.rs @@ -0,0 +1,92 @@ +//! A peer that sends us a table while not reading what we send it must +//! not stop us reading it: two routers sending each other their tables at +//! once would otherwise each wait on the other for ever. + +use std::net::Ipv4Addr; +use std::sync::Arc; +use std::time::Duration; + +use bgp_integration_tests::announcement::Announcement; +use bgp_integration_tests::prefix; +use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::TestServer; +use bgp_server::policy::ExportPolicy; +use eyre::Result; + +const LOCAL_ASN: u32 = 64601; +const ROUTES: u32 = 100_000; + +fn nth(n: u32) -> bgp_primitives::ip_prefix::IpPrefix { + let [a, b, c, _] = (0x0a00_0000u32 + n * 256).to_be_bytes(); + prefix(&format!("{a}.{b}.{c}.0/24")) +} + +#[tokio::test] +async fn a_peer_that_does_not_read_is_still_read() -> Result<()> { + let server = TestServer::start_with(LOCAL_ASN, |c| { + c.export = Arc::new(ExportPolicy { + learned_routes: true, + all_paths: false, + ..ExportPolicy::default() + }); + }) + .await?; + // One peer's table, for the server to pass on. + let mut first = server + .established(64701, Ipv4Addr::new(10, 0, 0, 1)) + .await?; + // A route to an UPDATE, each with attributes of its own, as a real + // table has them: passing it on is then megabytes, not a few hundred + // UPDATEs that the socket buffers would take whole. + for i in 0..ROUTES { + first + .announce( + Announcement::of([nth(i).into()], &[64701, 64510]) + .with_communities(&[((i >> 16) as u16 + 1, i as u16)]), + ) + .await?; + } + server.wait_for_routes(ROUTES as usize).await?; + + // A second peer, which the server starts sending that table to as + // soon as it is up, sends a table of its own and reads nothing back: + // the server's writes to it fill its socket and stop. The server has + // to go on reading it all the same. The peer's receive buffer is + // small, as a real link's window would be, so that this happens with + // the first few hundred kilobytes and not after the loopback's + // megabytes. + let socket = tokio::net::TcpSocket::new_v4()?; + socket.set_recv_buffer_size(16 * 1024)?; + let stream = socket.connect(server.addr).await?; + let mut second = TestPeer::from_stream(stream); + second + .handshake(64702, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; + server.wait_for_sessions(2).await?; + // Sent over a couple of seconds, so that the server's export to the + // peer is well under way while the peer is still sending. + let sending = async { + for start in (0..ROUTES).step_by(100) { + second + .announce(Announcement::of( + (start..start + 100).map(|i| nth(i).into()), + &[64702, 64510], + )) + .await?; + if start % 2000 == 0 { + tokio::time::sleep(Duration::from_millis(40)).await; + } + } + Ok::<_, eyre::Report>(()) + }; + tokio::time::timeout(Duration::from_secs(30), sending) + .await + .expect("the server reads the peer's table though the peer reads nothing")?; + tokio::time::timeout( + Duration::from_secs(30), + server.wait_for_routes(2 * ROUTES as usize), + ) + .await + .expect("every route is in the table")?; + Ok(()) +} diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index 914a419..bf0135e 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -31,7 +31,9 @@ use prometheus::IntGauge; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::sync::mpsc::error::TrySendError; use tokio::sync::{mpsc, watch}; +use tokio::task::JoinHandle; use tokio::time::{Instant, Interval, MissedTickBehavior, Sleep, interval_at, sleep, timeout}; use tracing::{debug, error, info, warn}; @@ -67,6 +69,10 @@ const OUTBOUND_QUEUE_CAPACITY: usize = 1024; /// round trip. Gathered, they go as full segments in one write. const WRITE_BATCH_BYTES: usize = 64 * 1024; +/// How many batches may wait for the writer task at a time: a megabyte +/// or so a session, beyond what the socket's own buffer takes. +const WRITER_QUEUE: usize = 16; + /// The least time a write to the peer may make no progress before the /// session is given up: the Send Hold Timer of RFC 9687, which is this or /// twice the hold time, whichever is longer. A peer that has stopped @@ -602,6 +608,8 @@ struct Validation { enum Wake { /// Something for the state machine. Event(Event), + /// The writer has room for a batch that was waiting. + WriterRoom(mpsc::OwnedPermit>), /// Messages from the rest of the server to send, as many as were /// waiting, to be written together. Outbound(Vec), @@ -638,8 +646,17 @@ struct Session { said: Said, fsm: Fsm, frames: FrameReader, - writer: OwnedWriteHalf, - /// Reused for every message written. + /// Batches of messages on their way to the writer task, which writes + /// them to the peer; see [`write_to_peer`]. + writer: mpsc::Sender>, + writer_task: JoinHandle<()>, + /// Batches the writer had no room for yet, oldest first. While there + /// are any, no more messages are taken from the rest of the server; + /// the peer is still read. + pending: VecDeque>, + /// What the writer task says if it fails. + write_failed: mpsc::Receiver, + /// Messages encoded and not yet made a batch of. write_buf: BytesMut, /// How messages from the peer are decoded: the width of AS numbers and /// where prefixes carry path identifiers are known once the OPENs are @@ -652,7 +669,8 @@ struct Session { hold_timer: Option>>, keepalive_timer: Option, /// How long a write may make no progress; see [`MIN_SEND_HOLD_TIME`]. - send_hold_time: Duration, + /// The writer task watches it. + send_hold: watch::Sender, /// Whether a NOTIFICATION of ours is on its way to the peer. notified: bool, /// This connection's slot in the collision table, once admitted. @@ -686,6 +704,10 @@ impl Session { // wait for the acknowledgement of the one before it. let _ = stream.set_nodelay(true); let (reader, writer) = stream.into_split(); + let (batches, queue) = mpsc::channel(WRITER_QUEUE); + let (failed, write_failed) = mpsc::channel(1); + let (send_hold, hold) = watch::channel(MIN_SEND_HOLD_TIME); + let writer_task = tokio::spawn(write_to_peer(writer, queue, hold, failed)); let (fsm, first_actions) = Fsm::new(FsmConfig { router_id: server.router_id, local_asn: server.local_asn, @@ -710,13 +732,16 @@ impl Session { rejections: server.rejections.of_peer(peer_addr.into(), &label), fsm, frames: FrameReader::new(reader), - writer, + writer: batches, + writer_task, + pending: VecDeque::new(), + write_failed, write_buf: BytesMut::new(), ctx: ParserContext::default(), send_ctx: ParserContext::default(), hold_timer: None, keepalive_timer: None, - send_hold_time: MIN_SEND_HOLD_TIME, + send_hold, notified: false, collision: None, established: None, @@ -746,13 +771,14 @@ impl Session { // not to go; the NOTIFICATION is, now. Contained as well: // whatever panicked may do so again. self.write_buf.clear(); - let sent = async { - self.send(¬ification).await?; - self.flush().await - }; - if !matches!(contained(sent).await, Ok(Ok(()))) { + self.pending.clear(); + if !matches!( + contained(async { self.send(¬ification) }).await, + Ok(Ok(())) + ) { warn!(peer_addr = %socket_text(self.peer_addr), "Could not send NOTIFICATION after a panic"); } + self.flush(); } error } @@ -762,12 +788,25 @@ impl Session { async fn close(mut self) { // A new connection from the peer must not find this one in its way. self.collision = None; + // Whatever is still to go out, a NOTIFICATION among it, is given to + // the writer, which closes our side once it has written it. + self.flush(); + let hand_over = async { + for batch in std::mem::take(&mut self.pending) { + if self.writer.send(batch).await.is_err() { + break; + } + } + }; + let _ = timeout(CLOSE_LINGER, hand_over).await; + let (closed, _) = mpsc::channel(1); + drop(std::mem::replace(&mut self.writer, closed)); + let _ = timeout(CLOSE_LINGER, &mut self.writer_task).await; if !self.notified { return; } // Our side is done sending; what the peer still sends is read and // thrown away until it closes its side too. - let _ = self.writer.shutdown().await; let _ = timeout(CLOSE_LINGER, self.frames.discard_until_closed()).await; } @@ -783,6 +822,13 @@ impl Session { .into_iter() .flat_map(|msg| self.fsm.handle(Event::Outbound(msg))) .collect(), + Wake::WriterRoom(permit) => { + if let Some(batch) = self.pending.pop_front() { + permit.send(batch); + } + self.push_pending(); + vec![] + } Wake::Revalidate(suspects) => self.revalidate(&suspects), Wake::Swept(outcome) => self.swept(&outcome), Wake::StaleTimeout => { @@ -819,7 +865,15 @@ impl Session { }, () = expired(&mut self.hold_timer) => Event::HoldTimerExpired, () = ticked(&mut self.keepalive_timer) => Event::KeepaliveTimerExpired, - wake = from_server(&mut self.established) => return Ok(wake), + Some(error) = self.write_failed.recv() => Event::LocalFailure(error), + permit = self.writer.clone().reserve_owned(), if !self.pending.is_empty() => { + return Ok(match permit { + Ok(permit) => Wake::WriterRoom(permit), + // The writer task is gone; its word is on its way. + Err(_) => return Ok(Wake::Event(Event::LocalFailure(SessionError::SendHoldTimerExpired(*self.send_hold.borrow())))), + }); + } + wake = from_server(&mut self.established, self.pending.is_empty()) => return Ok(wake), () = killed(&mut self.collision) => Event::CollisionLost, leave = leaving(&mut self.leave) => Event::Leave(leave), }; @@ -855,19 +909,16 @@ impl Session { /// or as each batch of `WRITE_BATCH_BYTES` fills. async fn execute(&mut self, actions: Vec) -> Option> { let result = self.execute_all(actions).await; - // Written whatever came of the actions: a NOTIFICATION that ends - // the session is the last thing the peer is to get. - match self.flush().await { - Ok(()) => result, - Err(e) => Some(result.unwrap_or(Err(e))), - } + // Whatever came of the actions goes to the writer as one batch. + self.flush(); + result } async fn execute_all(&mut self, actions: Vec) -> Option> { let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { - Action::Send(msg) => match self.send(&msg).await { + Action::Send(msg) => match self.send(&msg) { Ok(()) => {} // The session is ending anyway and the Close that // follows says why; a NOTIFICATION is best effort. @@ -977,7 +1028,8 @@ impl Session { ..self.ctx.clone() }; let hold_time = Duration::from_secs(negotiated.hold_time_secs.into()); - self.send_hold_time = MIN_SEND_HOLD_TIME.max(2 * hold_time); + self.send_hold + .send_replace(MIN_SEND_HOLD_TIME.max(2 * hold_time)); self.collision = self.server.collisions.admit( self.server.router_id, @@ -1321,9 +1373,9 @@ impl Session { vec![] } - /// Puts `msg` in the write buffer, and writes the buffer out if it - /// has a batch's worth; the rest goes with [`Session::flush`]. - async fn send(&mut self, msg: &Message) -> Result<(), SessionError> { + /// Puts `msg` in the write buffer, and makes a batch of the buffer if + /// it has a batch's worth; the rest goes with [`Session::flush`]. + fn send(&mut self, msg: &Message) -> Result<(), SessionError> { if let (Message::Update(update), Some(established)) = (msg, &self.established) { let (withdrawn, announced) = prefix_counts(update); debug!(peer = %established.peer_id, withdrawn, announced, "Sending UPDATE"); @@ -1341,21 +1393,35 @@ impl Session { .with_label_values(&[message_type_name(msg)]) .inc(); if self.write_buf.len() >= WRITE_BATCH_BYTES { - self.flush().await?; + self.flush(); } Ok(()) } - /// Writes what is in the write buffer to the peer. - async fn flush(&mut self) -> Result<(), SessionError> { + /// Makes a batch of the write buffer and gives it to the writer, or + /// keeps it until the writer has room. Never waits: a session that + /// waited to write would not be reading, and a peer doing the same + /// would wait on it for ever. + fn flush(&mut self) { if self.write_buf.is_empty() { - return Ok(()); + return; } - let written = timeout(self.send_hold_time, self.writer.write_all(&self.write_buf)).await; - self.write_buf.clear(); - match written { - Ok(written) => Ok(written?), - Err(_) => Err(SessionError::SendHoldTimerExpired(self.send_hold_time)), + self.pending.push_back(self.write_buf.split().to_vec()); + self.push_pending(); + } + + /// Gives the writer what it has room for, oldest first. + fn push_pending(&mut self) { + while let Some(batch) = self.pending.pop_front() { + match self.writer.try_send(batch) { + Ok(()) => {} + Err(TrySendError::Full(batch)) => { + self.pending.push_front(batch); + return; + } + // The writer task is gone; its word is on its way. + Err(TrySendError::Closed(_)) => return, + } } } } @@ -1406,7 +1472,7 @@ fn sweep( /// to send to the peer, a request to end the session, word that the /// validation data has changed, the outcome of a sweep of stale routes, /// or the time being up for the routes kept from the peer's last session. -async fn from_server(established: &mut Option) -> Wake { +async fn from_server(established: &mut Option, may_take_outbound: bool) -> Wake { let Some(Established { outbound, ending, @@ -1432,7 +1498,7 @@ async fn from_server(established: &mut Option) -> Wake { }; tokio::select! { Some(error) = ending.recv() => Wake::Event(Event::LocalFailure(error)), - Some(msg) = outbound.recv() => { + Some(msg) = outbound.recv(), if may_take_outbound => { // And whatever else is waiting, to go out together. let mut messages = vec![msg]; while messages.len() < OUTBOUND_QUEUE_CAPACITY @@ -1452,6 +1518,33 @@ async fn from_server(established: &mut Option) -> Wake { } } +/// Writes a session's batches to the peer, as its own task, so that the +/// session's reading never waits on its writing. Two routers sending each +/// other their tables at once would otherwise each stop reading while +/// held up writing to the other's closed window, and neither would read +/// again until the send hold timer ended both sessions. A write that +/// makes no progress for the send hold time, or fails, ends the session +/// through `failed`; the remaining batches are dropped with it. Once the +/// session has nothing more to send, our side of the connection is shut. +async fn write_to_peer( + mut writer: OwnedWriteHalf, + mut queue: mpsc::Receiver>, + send_hold: watch::Receiver, + failed: mpsc::Sender, +) { + while let Some(batch) = queue.recv().await { + let hold_time = *send_hold.borrow(); + let error = match timeout(hold_time, writer.write_all(&batch)).await { + Ok(Ok(())) => continue, + Ok(Err(e)) => e.into(), + Err(_) => SessionError::SendHoldTimerExpired(hold_time), + }; + let _ = failed.try_send(error); + return; + } + let _ = writer.shutdown().await; +} + /// Resolves with how the daemon is leaving, once it is. Cancel-safe. async fn leaving(leave: &mut watch::Receiver>) -> Leave { let leave = match leave.wait_for(Option::is_some).await { @@ -1734,18 +1827,22 @@ mod tests { PeerSettings::unconfigured(90, Arc::default(), None, GracefulRestartConfig::default()), ) .unwrap(); - session.send_hold_time = Duration::from_millis(100); + session.send_hold.send_replace(Duration::from_millis(100)); - // Until the socket's buffers are full, writes succeed. + // Until the socket's buffers are full, writes go through; then + // the writer gives up, and says so. let filler = Message::Notification(NotificationMessage { error: BgpError::Cease(CeaseSubcode::AdminShutdown), data: vec![0; 4000], }); let stuck = tokio::time::timeout(Duration::from_secs(30), async { loop { - if let Err(e) = session.send(&filler).await { + session.send(&filler).unwrap(); + session.flush(); + if let Ok(e) = session.write_failed.try_recv() { return e; } + tokio::time::sleep(Duration::from_millis(1)).await; } }) .await