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