diff --git a/quicr/Cargo.toml b/quicr/Cargo.toml index b7a320166..907a3e5c8 100644 --- a/quicr/Cargo.toml +++ b/quicr/Cargo.toml @@ -14,6 +14,7 @@ futures = "0.1.21" fnv = "1.0.6" openssl = "0.10.6" failure = "0.1" +bytes = "0.4.6" [dev-dependencies] tokio = "0.1.5" diff --git a/quicr/examples/client.rs b/quicr/examples/client.rs index 5874594e0..abb3df313 100644 --- a/quicr/examples/client.rs +++ b/quicr/examples/client.rs @@ -59,11 +59,9 @@ fn run(log: Logger) -> Result<()> { endpoint.connect(&remote, url.host_str().map(|x| x.as_bytes())) .map_err(|e| format_err!("failed to connect: {}", e)) .and_then(|(conn, _)| { - eprintln!("connected!"); conn.open_bi().map_err(|e| format_err!("failed to open stream: {}", e)) }) .and_then(|(send, recv)| { - eprintln!("opened a stream"); tokio::io::write_all(send, request.as_bytes()).map_err(|e| format_err!("failed to send request: {}", e)) .map(move |(send, _)| (send, recv)) }) @@ -73,7 +71,6 @@ fn run(log: Logger) -> Result<()> { .map(|data| { io::stdout().write_all(&data).unwrap(); io::stdout().flush().unwrap(); - eprintln!("done") }) ).map_err(|e| e.into_inner().unwrap())?; diff --git a/quicr/examples/server.rs b/quicr/examples/server.rs index b5f59ec0c..a9feaabd9 100644 --- a/quicr/examples/server.rs +++ b/quicr/examples/server.rs @@ -14,6 +14,9 @@ use std::net::UdpSocket; use std::fs::File; use std::io::Read; use std::fmt; +use std::path::{self, Path, PathBuf}; +use std::str; +use std::rc::Rc; use futures::{Future, Stream}; use tokio::executor::current_thread::{self, CurrentThread}; @@ -62,6 +65,10 @@ fn main() { } fn run(log: Logger) -> Result<()> { + let root = ::std::env::args().nth(1).ok_or(format_err!("missing root argument"))?; + let root = Rc::new(Path::new(&root).to_owned()); + if !root.exists() { bail!("root path does not exist"); } + let socket = UdpSocket::bind("[::]:4433")?; let mut protocols = Vec::new(); const PROTO: &[u8] = b"hq-11"; @@ -96,25 +103,41 @@ fn run(log: Logger) -> Result<()> { let mut executor = CurrentThread::new_with_park(timer); executor.spawn(incoming.for_each(move |conn| { - info!(log, "got connection"); + let quicr::NewConnection { incoming, address, .. } = conn; + info!(log, "got connection"; "remote" => %address); + let root = root.clone(); + let log = log.clone(); + let log2 = log.clone(); current_thread::spawn( - conn.incoming.into_future() - .map_err(|_| unreachable!()) - .and_then(|(stream, _)| match stream { - Some(quicr::NewStream::Bi(send, recv)) => Ok((send, recv)), - Some(quicr::NewStream::Uni(_)) => unreachable!(), - None => Err(format_err!("no request submitted")), + incoming + .map_err(move |e| info!(log2, "connection terminated"; "remote" => %address, "reason" => %e)) + .and_then(|stream| { match stream { + quicr::NewStream::Bi(send, recv) => Ok((send, recv)), + quicr::NewStream::Uni(_) => unreachable!(), + }}) + .for_each(move |(send, recv)| { + let root = root.clone(); + let log = log.clone(); + let log2 = log.clone(); + let log3 = log.clone(); + current_thread::spawn( + recv.read_to_end(64 * 1024) + .map_err(|e| format_err!("failed reading request: {}", e)) + .map(move |data| (send, data)) + .and_then(move |(send, req)| { + info!(log, "got request"; "remote" => %address); + let resp = process_request(&root, &req).unwrap_or_else(move |e| { + error!(log, "failed to process request"; "reaosn" => %e.pretty()); + format!("failed to process request: {}\n", e.pretty()).into_bytes().into() + }); + tokio::io::write_all(send, resp).map_err(|e| format_err!("failed to send response: {}", e)) + }) + .and_then(|(send, _)| tokio::io::shutdown(send).map_err(|e| format_err!("failed to shutdown stream: {}", e))) + .map(move |_| info!(log3, "request complete"; "remote" => %address)) + .map_err(move |e| error!(log2, "request failed"; "reason" => %e.pretty())) + ); + Ok(()) }) - .and_then(|(send, recv)| recv.read_to_end(64 * 1024) - .map_err(|e| format_err!("failed reading request: {}", e)) - .map(move |data| (send, data))) - .and_then(move |(send, _)| { - eprintln!("processing request"); - tokio::io::write_all(send, b"hello\n").map_err(|e| format_err!("failed to send response: {}", e)) - }) - .and_then(|(send, _)| tokio::io::shutdown(send).map_err(|e| format_err!("failed to shutdown stream: {}", e))) - .map(|_| eprintln!("done")) - .map_err(|e| eprintln!("failed: {}", e)) ); Ok(()) })); @@ -123,3 +146,26 @@ fn run(log: Logger) -> Result<()> { Ok(()) } + +fn process_request(root: &Path, x: &[u8]) -> Result> { + if x.len() < 4 || &x[0..4] != b"GET " { bail!("missing GET"); } + if x[4..].len() < 2 || &x[x.len()-2..] != b"\r\n" { bail!("missing \\r\\n"); } + let path = str::from_utf8(&x[4..x.len()-2]).context("path is malformed UTF-8")?; + let path = Path::new(&path); + let mut real_path = PathBuf::from(root); + let mut components = path.components(); + match components.next() { + Some(path::Component::RootDir) => {} + _ => { bail!("path must be absolute"); } + } + for c in components { + match c { + path::Component::Normal(x) => { real_path.push(x); } + x => { bail!("illegal component in path: {:?}", x); } + } + } + let mut file = File::open(real_path)?; + let mut data = Vec::new(); + file.read_to_end(&mut data).context("failed reading file")?; + Ok(data.into()) +} diff --git a/quicr/src/lib.rs b/quicr/src/lib.rs index 1c087673b..30dfee9d0 100644 --- a/quicr/src/lib.rs +++ b/quicr/src/lib.rs @@ -10,6 +10,7 @@ extern crate fnv; extern crate openssl; #[macro_use] extern crate failure; +extern crate bytes; use std::{io, mem}; use std::net::{SocketAddr, SocketAddrV6}; @@ -29,10 +30,11 @@ use futures::task::{self, Task}; use futures::stream::FuturesUnordered; use fnv::{FnvHashMap, FnvHashSet}; use openssl::ssl; +use bytes::Bytes; use quicr::{Directionality, StreamId, ConnectionHandle}; -pub use quicr::{Config, ListenConfig, PersistentState, ConnectionError, TransportError, ReadError, WriteError}; +pub use quicr::{Config, ListenConfig, PersistentState, ConnectionError, TransportError}; #[derive(Debug, Fail)] pub enum Error { @@ -66,14 +68,15 @@ struct Pending { blocked_writers: FnvHashMap, blocked_readers: FnvHashMap, connecting: Option>>, - uni_opening: VecDeque>>, - bi_opening: VecDeque>>, + uni_opening: VecDeque>>, + bi_opening: VecDeque>>, cancel_loss_detect: Option>, cancel_idle: Option>, incoming_streams: VecDeque, incoming_streams_reader: Option, remote_recv_streams: FnvHashSet, - finishing: FnvHashMap>, + finishing: FnvHashMap>>, + error: Option, } impl Pending { @@ -89,7 +92,33 @@ impl Pending { incoming_streams_reader: None, remote_recv_streams: FnvHashSet::default(), finishing: FnvHashMap::default(), + error: None, }} + + pub fn fail(mut self, reason: ConnectionError) { + self.error = Some(reason.clone()); + for (_, writer) in self.blocked_writers.drain() { + writer.notify() + } + for (_, reader) in self.blocked_readers.drain() { + reader.notify() + } + if let Some(c) = self.connecting.take() { + let _ = c.send(Some(reason.clone())); + } + for x in self.uni_opening.drain(..) { + let _ = x.send(Err(reason.clone())); + } + for x in self.bi_opening.drain(..) { + let _ = x.send(Err(reason.clone())); + } + if let Some(x) = self.incoming_streams_reader.take() { + x.notify(); + } + for (_, x) in self.finishing.drain() { + let _ = x.send(Some(reason.clone())); + } + } } #[derive(Clone)] @@ -173,7 +202,6 @@ impl Future for Driver { match event { Connected { address, protocol } => { if let Some(c) = endpoint.pending.get_mut(&connection).unwrap().connecting.take() { - // Graceful close should be handled by drop impl let _ = c.send(None); } else { let conn = Rc::new(ConnectionInner { endpoint: Endpoint(self.0.clone()), conn: connection }); @@ -186,10 +214,7 @@ impl Future for Driver { } } ConnectionLost { reason } => { - if let Some(c) = endpoint.pending.get_mut(&connection).unwrap().connecting.take() { - // Graceful close should be handled by drop impl - let _ = c.send(Some(reason)); - } + endpoint.pending.remove(&connection).unwrap().fail(reason); } StreamWritable { stream } => { if let Some(writer) = endpoint.pending.get_mut(&connection).unwrap().blocked_writers.remove(&stream) { @@ -224,7 +249,7 @@ impl Future for Driver { } StreamFinished { stream } => { let _ = endpoint.pending.get_mut(&connection).unwrap() - .finishing.remove(&stream).unwrap().send(()); + .finishing.remove(&stream).unwrap().send(None); } } } @@ -257,37 +282,48 @@ impl Future for Driver { endpoint.outgoing.push_front((destination, packet)); } } + TimerStart { connection, timer: timer@quicr::Timer::Close, time } => { + let instant = endpoint.epoch + duration_micros(time); + endpoint.timers.push(Timer { + conn: connection, + ty: timer, + delay: endpoint.timer.delay(instant), + cancel: None, + }); + } TimerStart { connection, timer, time } => { // Loss detection and idle timers start before the connection is established let pending = endpoint.pending.entry(connection).or_insert_with(|| Pending::new(None)); use quicr::Timer::*; let mut cancel = match timer { - LossDetection => Some(&mut pending.cancel_loss_detect), - Idle => Some(&mut pending.cancel_idle), - Close => None + LossDetection => &mut pending.cancel_loss_detect, + Idle => &mut pending.cancel_idle, + Close => unreachable!() }; let instant = endpoint.epoch + duration_micros(time); - if let Some(cancel) = cancel.as_mut().and_then(|x| x.take()) { + if let Some(cancel) = cancel.take() { let _ = cancel.send(()); } let (send, recv) = oneshot::channel(); - if let Some(cancel) = cancel { *cancel = Some(send); } + *cancel = Some(send); trace!(endpoint.log, "timer start"; "timer" => ?timer, "time" => ?duration_micros(time)); endpoint.timers.push(Timer { conn: connection, ty: timer, delay: endpoint.timer.delay(instant), - cancel: recv, + cancel: Some(recv), }); } TimerStop { connection, timer } => { trace!(endpoint.log, "timer stop"; "timer" => ?timer); - let pending = endpoint.pending.get_mut(&connection).unwrap(); - use quicr::Timer::*; - match timer { - LossDetection => { pending.cancel_loss_detect.take().map(|x| x.send(()).unwrap()); } - Idle => { pending.cancel_idle.take().map(|x| x.send(())); } - Close => { unreachable!() } + // If a connection was lost, we already canceled its loss/idle timers. + if let Some(pending) = endpoint.pending.get_mut(&connection) { + use quicr::Timer::*; + match timer { + LossDetection => { pending.cancel_loss_detect.take().map(|x| x.send(()).unwrap()); } + Idle => { pending.cancel_idle.take().map(|x| x.send(())); } + Close => { unreachable!() } + } } } } @@ -328,12 +364,8 @@ struct ConnectionInner { pub struct Connection(Rc); -#[derive(Copy, Clone, Debug, Fail)] -#[fail(display = "connection lost")] -pub struct ConnectionLost; - impl Connection { - pub fn open_uni(&self) -> Box> { + pub fn open_uni(&self) -> Box> { let (send, recv) = oneshot::channel(); { let mut endpoint = self.0.endpoint.0.borrow_mut(); @@ -353,7 +385,7 @@ impl Connection { ) } - pub fn open_bi(&self) -> Box> { + pub fn open_bi(&self) -> Box> { let (send, recv) = oneshot::channel(); { let mut endpoint = self.0.endpoint.0.borrow_mut(); @@ -386,6 +418,7 @@ impl Drop for ConnectionInner { fn drop(&mut self) { let endpoint = &mut *self.endpoint.0.borrow_mut(); endpoint.inner.close(micros_from(endpoint.epoch.elapsed()), self.conn, 0, (&[][..]).into()); + endpoint.pending.remove(&self.conn); endpoint.driver.as_ref().map(|x| x.notify()); } } @@ -394,7 +427,7 @@ pub struct SendStream { endpoint: Endpoint, conn: Rc, stream: StreamId, - finishing: Option>, + finishing: Option>>, stop_reason: Option, finished: bool, } @@ -428,14 +461,17 @@ impl SendStream { pub fn poll_write(&mut self, buf: &[u8]) -> Poll { let mut endpoint = self.endpoint.0.borrow_mut(); + use quicr::WriteError::*; let n = match endpoint.inner.write(self.conn.conn, self.stream, buf.into()) { Ok(()) => buf.len(), - Err((ref unwritten, WriteError::Blocked)) if unwritten.len() < buf.len() => buf.len() - unwritten.len(), - Err((_, WriteError::Blocked)) => { - endpoint.pending.get_mut(&self.conn.conn).unwrap().blocked_writers.insert(self.stream, task::current()); + Err((ref unwritten, Blocked)) if unwritten.len() < buf.len() => buf.len() - unwritten.len(), + Err((_, Blocked)) => { + let pending = endpoint.pending.get_mut(&self.conn.conn).unwrap(); + if let Some(ref x) = pending.error { return Err(WriteError::ConnectionClosed(x.clone())); } + pending.blocked_writers.insert(self.stream, task::current()); return Ok(Async::NotReady); } - Err((_, WriteError::Stopped { error_code })) => { + Err((_, Stopped { error_code })) => { self.stop_reason = Some(error_code); return Err(WriteError::Stopped { error_code }); } @@ -444,7 +480,7 @@ impl SendStream { Ok(Async::Ready(n)) } - pub fn poll_finish(&mut self) -> Async<()> { + pub fn poll_finish(&mut self) -> Poll<(), ConnectionError> { let mut endpoint = self.endpoint.0.borrow_mut(); if self.finishing.is_none() { endpoint.inner.finish(self.conn.conn, self.stream); @@ -453,8 +489,14 @@ impl SendStream { endpoint.pending.get_mut(&self.conn.conn).unwrap().finishing.insert(self.stream, send); } let r = self.finishing.as_mut().unwrap().poll().unwrap(); - if let Async::Ready(()) = r { self.finished = true; } - r + match r { + Async::Ready(None) => { + self.finished = true; + Ok(Async::Ready(())) + } + Async::Ready(Some(e)) => Err(e), + Async::NotReady => Ok(Async::NotReady), + } } pub fn reset(&self, error_code: u16) { @@ -470,7 +512,7 @@ impl io::Write for SendStream { Ok(Async::Ready(n)) => Ok(n), Ok(Async::NotReady) => Err(io::Error::new(io::ErrorKind::WouldBlock, "stream blocked")), Err(WriteError::Stopped { .. }) => Err(io::Error::new(io::ErrorKind::ConnectionReset, "peer stopped this stream")), - Err(WriteError::Blocked) => unreachable!(), + Err(WriteError::ConnectionClosed(e)) => Err(io::Error::new(io::ErrorKind::ConnectionAborted, format!("connection closed: {}", e))), } } @@ -479,7 +521,7 @@ impl io::Write for SendStream { impl AsyncWrite for SendStream { fn shutdown(&mut self) -> Poll<(), io::Error> { - Ok(self.poll_finish()) + self.poll_finish().map_err(|e| io::Error::new(io::ErrorKind::ConnectionAborted, format!("connection closed: {}", e))) } } @@ -493,6 +535,16 @@ impl Drop for SendStream { } } +#[derive(Debug, Fail, Clone)] +pub enum WriteError { + /// The peer is no longer accepting data on this stream. + #[fail(display = "sending stopped by peer: error {}", error_code)] + Stopped { error_code: u16 }, + /// The connection was closed. + #[fail(display = "connection closed: {}", _0)] + ConnectionClosed(ConnectionError), +} + impl Drop for RecvStream { fn drop(&mut self) { let endpoint = &mut *self.endpoint.0.borrow_mut(); @@ -504,24 +556,25 @@ impl Drop for RecvStream { } impl RecvStream { - pub fn poll_read_unordered(&mut self) -> Poll<(Box<[u8]>, u64), ReadError> { + pub fn poll_read_unordered(&mut self) -> Poll<(Bytes, u64), ReadError> { let endpoint = &mut *self.endpoint.0.borrow_mut(); - use ReadError::*; + use quicr::ReadError::*; let pending = endpoint.pending.get_mut(&self.conn.conn).unwrap(); match endpoint.inner.read_unordered(self.conn.conn, self.stream) { - Ok((bytes, offset)) => Ok(Async::Ready((bytes.to_vec().into(), offset))), + Ok((bytes, offset)) => Ok(Async::Ready((bytes, offset))), Err(Blocked) => { + if let Some(ref x) = pending.error { return Err(ReadError::ConnectionClosed(x.clone())); } pending.blocked_readers.insert(self.stream, task::current()); Ok(Async::NotReady) } - Err(e@Reset { .. }) => { + Err(Reset { error_code }) => { pending.remote_recv_streams.remove(&self.stream); - Err(e) + Err(ReadError::Reset { error_code }) } - Err(e@Finished) => { + Err(Finished) => { pending.remote_recv_streams.remove(&self.stream); self.recvd = true; - Err(e) + Err(ReadError::Finished) } } } @@ -531,19 +584,33 @@ impl RecvStream { } } +#[derive(Debug, Fail, Clone)] +pub enum ReadError { + /// The peer abandoned transmitting data on this stream. + #[fail(display = "stream reset by peer: error {}", error_code)] + Reset { error_code: u16 }, + /// The data on this stream has been fully delivered and no more will be transmitted. + #[fail(display = "the stream has been completely received")] + Finished, + /// The connection was closed. + #[fail(display = "connection closed: {}", _0)] + ConnectionClosed(ConnectionError), +} + struct Timer { conn: ConnectionHandle, ty: quicr::Timer, delay: Delay, - cancel: oneshot::Receiver<()>, + cancel: Option>, } impl Future for Timer { type Item = Option<(ConnectionHandle, quicr::Timer)>; type Error = (); // FIXME fn poll(&mut self) -> Poll { - if self.ty != quicr::Timer::Close { - if let Async::Ready(()) = self.cancel.poll().unwrap() { + if let Some(ref mut cancel) = self.cancel { + if let Ok(Async::NotReady) = cancel.poll() {} + else { return Ok(Async::Ready(None)); } } @@ -567,10 +634,11 @@ pub enum NewStream { impl FuturesStream for IncomingStreams { type Item = NewStream; - type Error = (); // FIXME + type Error = ConnectionError; fn poll(&mut self) -> Poll, Self::Error> { let mut endpoint = self.endpoint.0.borrow_mut(); - let pending = endpoint.pending.get_mut(&self.conn.conn).unwrap(); + // TODO: Yield error + let pending = if let Some(x) = endpoint.pending.get_mut(&self.conn.conn) { x } else { return Ok(Async::Ready(None)); }; if let Some(x) = pending.incoming_streams.pop_front() { let recv = RecvStream::new(self.endpoint.clone(), self.conn.clone(), x); let stream = if x.directionality() == Directionality::Uni { @@ -580,8 +648,12 @@ impl FuturesStream for IncomingStreams { }; return Ok(Async::Ready(Some(stream))); } - pending.incoming_streams_reader = Some(task::current()); - return Ok(Async::NotReady); + if let Some(ref x) = pending.error { + Err(x.clone()) + } else { + pending.incoming_streams_reader = Some(task::current()); + Ok(Async::NotReady) + } } } @@ -596,30 +668,19 @@ impl Future for ReadToEnd { type Item = Box<[u8]>; type Error = ReadError; fn poll(&mut self) -> Poll { - let endpoint = &mut *self.stream.endpoint.0.borrow_mut(); - use ReadError::*; - let pending = endpoint.pending.get_mut(&self.stream.conn.conn).unwrap(); loop { - match endpoint.inner.read_unordered(self.stream.conn.conn, self.stream.stream) { - Ok((data, offset)) => { + match self.stream.poll_read_unordered() { + Ok(Async::Ready((data, offset))) => { let len = self.buffer.len().max(offset as usize + data.len()); - if len > self.size_limit { return Err(Finished); } + if len > self.size_limit { return Err(ReadError::Finished); } self.buffer.resize(len, 0); self.buffer[offset as usize..offset as usize+data.len()].copy_from_slice(&data); } - Err(Blocked) => { - pending.blocked_readers.insert(self.stream.stream, task::current()); - return Ok(Async::NotReady); - } - Err(e@Reset { .. }) => { - pending.remote_recv_streams.remove(&self.stream.stream); - return Err(e); - } - Err(Finished) => { - self.stream.recvd = true; - pending.remote_recv_streams.remove(&self.stream.stream); + Ok(Async::NotReady) => { return Ok(Async::NotReady); } + Err(ReadError::Finished) => { return Ok(Async::Ready(mem::replace(&mut self.buffer, Vec::new()).into())); } + Err(e) => { return Err(e); } } } } diff --git a/src/endpoint.rs b/src/endpoint.rs index 418bb4c23..21f0301f2 100644 --- a/src/endpoint.rs +++ b/src/endpoint.rs @@ -778,6 +778,7 @@ impl Endpoint { let rs = stream.recv_mut().unwrap(); if let Some(final_offset) = rs.final_offset() { if end > final_offset || (frame.fin && end != final_offset) { + debug!(self.log, "final offset error"; "frame end" => end, "final offset" => final_offset); self.events.push_back((conn, Event::ConnectionLost { reason: TransportError::FINAL_OFFSET_ERROR.into() })); return State::closed(TransportError::FINAL_OFFSET_ERROR); } @@ -785,6 +786,9 @@ impl Endpoint { let prev_end = rs.limit(); let new_bytes = end.saturating_sub(prev_end); if end > rs.max_data || data_recvd + new_bytes > max_data { + debug!(self.log, "flow control error"; + "stream" => frame.id.0, "recvd" => data_recvd, "new bytes" => new_bytes, + "max data" => max_data, "end" => end, "stream max data" => rs.max_data); self.events.push_back((conn, Event::ConnectionLost { reason: TransportError::FLOW_CONTROL_ERROR.into() })); return State::closed(TransportError::FLOW_CONTROL_ERROR); } @@ -868,6 +872,8 @@ impl Endpoint { if let Some(stream) = self.connections[conn.0].streams.get_mut(&id) { let ss = stream.send_mut().unwrap(); if offset > ss.max_data { + trace!(self.log, "stream limit increased"; "stream" => id.0, + "old" => ss.max_data, "new" => offset, "current offset" => ss.offset); if ss.offset == ss.max_data { self.events.push_back((conn, Event::StreamWritable { stream: id })); } @@ -1045,31 +1051,38 @@ impl Endpoint { fn reset_idle_timeout(&mut self, now: u64, conn: ConnectionHandle) { let dt = cmp::min(self.config.idle_timeout, self.connections[conn.0].params.idle_timeout) as u64 * 1000000; - self.io.push_back(Io::TimerStart { time: now + dt, connection: conn, timer: Timer::Idle }); + self.connections[conn.0].set_idle = Some(Some(now + dt)); } - /// Returns whether anything was sent fn flush_pending(&mut self, now: u64, conn: ConnectionHandle) { - let mut timer = None; let mut sent = false; - while let Some((packet, t)) = self.connections[conn.0].next_packet(&self.log, &self.config, now) { - timer = t.or(timer); + while let Some(packet) = self.connections[conn.0].next_packet(&self.log, &self.config, now) { self.io.push_back(Io::Transmit { destination: self.connections[conn.0].remote, packet: packet.into(), }); sent = true; } - if let Some(time) = timer { - self.io.push_back(Io::TimerStart { - connection: conn, - timer: Timer::LossDetection, - time - }); - } if sent { self.reset_idle_timeout(now, conn); } + { + let c = &mut self.connections[conn.0]; + if let Some(setting) = c.set_idle.take() { + if let Some(time) = setting { + self.io.push_back(Io::TimerStart { connection: conn, timer: Timer::Idle, time }); + } else { + self.io.push_back(Io::TimerStop { connection: conn, timer: Timer::Idle }); + } + } + if let Some(setting) = c.set_loss_detection.take() { + if let Some(time) = setting { + self.io.push_back(Io::TimerStart { connection: conn, timer: Timer::LossDetection, time }); + } else { + self.io.push_back(Io::TimerStop { connection: conn, timer: Timer::LossDetection }); + } + } + } } fn forget(&mut self, conn: ConnectionHandle) { @@ -1105,6 +1118,7 @@ impl Endpoint { self.events.push_back((conn, Event::ConnectionLost { reason: ConnectionError::TimedOut, })); + self.dirty_conns.insert(conn); // Ensure the loss detection timer cancellation goes through } Timer::LossDetection => { if self.connections[conn.0].handshake_sent != 0 { @@ -1146,14 +1160,7 @@ impl Endpoint { self.reset_idle_timeout(now, conn); self.connections[conn.0].rto_count += 1; } - let alarm = self.connections[conn.0].compute_loss_detection_alarm(&self.config); - if alarm != u64::max_value() { - self.io.push_back(Io::TimerStart { - connection: conn, - timer: Timer::LossDetection, - time: alarm, - }); - } + self.connections[conn.0].set_loss_detection_alarm(&self.config); self.dirty_conns.insert(conn); } } @@ -1173,19 +1180,7 @@ impl Endpoint { fn on_ack_received(&mut self, now: u64, conn: ConnectionHandle, ack: frame::Ack) { trace!(self.log, "got ack"; "ranges" => ?ack.iter().collect::>()); let was_blocked = self.connections[conn.0].blocked(); - let time = self.connections[conn.0].on_ack_received(&self.config, now, ack); - if time == u64::max_value() { - self.io.push_back(Io::TimerStop { - connection: conn, - timer: Timer::LossDetection, - }); - } else { - self.io.push_back(Io::TimerStart { - connection: conn, - timer: Timer::LossDetection, - time, - }); - } + self.connections[conn.0].on_ack_received(&self.config, now, ack); if was_blocked && !self.connections[conn.0].blocked() { for stream in self.connections[conn.0].blocked_streams.drain() { self.events.push_back((conn, Event::StreamWritable { stream })); @@ -1203,9 +1198,13 @@ impl Endpoint { pub fn write(&mut self, conn: ConnectionHandle, stream: StreamId, mut data: Bytes) -> Result<(), (Bytes, WriteError)> { if self.connections[conn.0].state.as_ref().unwrap().is_closed() { return Err((data, WriteError::Blocked)); } assert!(stream.directionality() == Directionality::Bi || stream.initiator() == self.connections[conn.0].side); - if self.connections[conn.0].blocked() { return Err((data, WriteError::Blocked)); } + if self.connections[conn.0].blocked() { + self.connections[conn.0].blocked_streams.insert(stream); + return Err((data, WriteError::Blocked)); + } let (stop_reason, stream_budget) = { let ss = self.connections[conn.0].streams.get_mut(&stream).expect("stream already closed").send_mut().unwrap(); + trace!(self.log, "writing"; "stream" => stream.0, "stream max data" => ss.max_data, "stream offset" => ss.offset); (match ss.state { stream::SendState::ResetSent { ref mut stop_reason } | stream::SendState::ResetRecvd { ref mut stop_reason } => stop_reason.take(), @@ -1219,15 +1218,19 @@ impl Endpoint { } let conn_budget = self.connections[conn.0].max_data - self.connections[conn.0].data_sent; - let result = if conn_budget < data.len() as u64 { + + let result = if conn_budget < stream_budget && conn_budget < data.len() as u64 { self.connections[conn.0].blocked_streams.insert(stream); Err((data.split_off(conn_budget as usize), WriteError::Blocked)) } else if stream_budget < data.len() as u64 { Err((data.split_off(stream_budget as usize), WriteError::Blocked)) } else { Ok(()) }; if !data.is_empty() { + trace!(self.log, "queuing"; "stream" => stream.0, "len" => data.len()); self.connections[conn.0].transmit(stream, data); self.dirty_conns.insert(conn); + } else { + trace!(self.log, "stream blocked"; "stream" => stream.0); } result } @@ -1266,6 +1269,7 @@ impl Endpoint { /// # Panics /// - when applied to a stream that does not have an active incoming channel pub fn read_unordered(&mut self, conn: ConnectionHandle, stream: StreamId) -> Result<(Bytes, u64), ReadError> { + self.dirty_conns.insert(conn); // May need to send flow control frames after reading match self.connections[conn.0].read(stream) { x@Err(ReadError::Finished) | x@Err(ReadError::Reset { .. }) => { self.connections[conn.0].maybe_cleanup(stream); @@ -1320,7 +1324,7 @@ impl Endpoint { fn close_common(&mut self, now: u64, conn: ConnectionHandle) { trace!(self.log, "connection closed"); - self.io.push_back(Io::TimerStop { connection: conn, timer: Timer::LossDetection }); + self.connections[conn.0].set_loss_detection = Some(None); self.io.push_back(Io::TimerStart { connection: conn, timer: Timer::Close, @@ -1347,6 +1351,7 @@ impl Endpoint { packet: self.connections[conn.0].make_close(&reason), }); self.reset_idle_timeout(now, conn); + self.dirty_conns.insert(conn); } self.connections[conn.0].state = Some(match self.connections[conn.0].state.take().unwrap() { State::Handshake(_) => State::HandshakeFailed(state::HandshakeFailed { reason, alert: None, app_closed: true }), @@ -1501,6 +1506,11 @@ struct Connection { /// Set iff we have received a non-ack frame since the last ack-only packet we sent permit_ack_only: bool, + // Timer updates: None if no change, Some(None) to stop, Some(Some(_)) to reset + + set_idle: Option>, + set_loss_detection: Option>, + // // Stream states // @@ -1664,6 +1674,9 @@ impl Connection { pending_acks: RangeSet::new(), permit_ack_only: false, + set_idle: None, + set_loss_detection: None, + streams, next_uni_stream: 0, next_bi_stream: match side { Side::Client => 1, Side::Server => 0 }, @@ -1683,7 +1696,7 @@ impl Connection { } /// Returns new loss detection alarm time, if applicable - fn on_packet_sent(&mut self, config: &Config, now: u64, packet_number: u64, packet: SentPacket) -> Option { + fn on_packet_sent(&mut self, config: &Config, now: u64, packet_number: u64, packet: SentPacket) { self.largest_sent_packet = packet_number; let bytes = packet.bytes; let handshake = packet.handshake; @@ -1697,14 +1710,12 @@ impl Connection { self.time_of_last_sent_handshake_packet = now; } self.bytes_in_flight += bytes as u64; - Some(self.compute_loss_detection_alarm(config)) - } else { - None + self.set_loss_detection_alarm(config); } } - /// Returns new loss detection alarm time - fn on_ack_received(&mut self, config: &Config, now: u64, ack: frame::Ack) -> u64 { + /// Updates set_loss_detection + fn on_ack_received(&mut self, config: &Config, now: u64, ack: frame::Ack) { self.largest_acked_packet = cmp::max(self.largest_acked_packet, ack.largest); // TODO: Validate if let Some(info) = self.sent_packets.get(&ack.largest).cloned() { self.latest_rtt = now - info.time; @@ -1721,7 +1732,7 @@ impl Connection { } } self.detect_lost_packets(config, now, ack.largest); - self.compute_loss_detection_alarm(config) + self.set_loss_detection_alarm(config); } fn update_rtt(&mut self, ack_delay: u64, ack_only: bool) { @@ -1848,9 +1859,10 @@ impl Connection { fn in_recovery(&self, packet: u64) -> bool { packet <= self.end_of_recovery } - fn compute_loss_detection_alarm(&self, config: &Config) -> u64 { + fn set_loss_detection_alarm(&mut self, config: &Config) { if self.bytes_in_flight == 0 { - return u64::max_value(); + self.set_loss_detection = Some(None); + return; } let mut alarm_duration: u64; @@ -1864,7 +1876,8 @@ impl Connection { alarm_duration = cmp::max(alarm_duration + self.max_ack_delay, config.min_tlp_timeout); alarm_duration = alarm_duration * 2u64.pow(self.handshake_count); - return self.time_of_last_sent_handshake_packet + alarm_duration; + self.set_loss_detection = Some(Some(self.time_of_last_sent_handshake_packet + alarm_duration)); + return; } if self.loss_time != 0 { @@ -1880,7 +1893,7 @@ impl Connection { alarm_duration = cmp::min(alarm_duration, tlp_duration); } } - self.time_of_last_sent_retransmittable_packet + alarm_duration + self.set_loss_detection = Some(Some(self.time_of_last_sent_retransmittable_packet + alarm_duration)); } /// Retransmit time-out @@ -1937,7 +1950,7 @@ impl Connection { }); } - fn next_packet(&mut self, log: &Logger, config: &Config, now: u64) -> Option<(Vec, Option)> { + fn next_packet(&mut self, log: &Logger, config: &Config, now: u64) -> Option> { let is_handshake; match *self.state.as_ref().unwrap() { ref x if x.is_closed() => { return None; } @@ -2074,6 +2087,7 @@ impl Connection { if pending.max_uni_stream_id && buf.len() + 9 < max_size { pending.max_uni_stream_id = false; sent.max_uni_stream_id = true; + trace!(log, "MAX_STREAM_ID (unidirectional)"); buf.write(frame::Type::MAX_STREAM_ID); buf.write(StreamId::new(!self.side, Directionality::Uni, self.max_remote_uni_stream)); } @@ -2082,6 +2096,7 @@ impl Connection { if pending.max_bi_stream_id && buf.len() + 9 < max_size { pending.max_bi_stream_id = false; sent.max_bi_stream_id = true; + trace!(log, "MAX_STREAM_ID (bidirectional)"); buf.write(frame::Type::MAX_STREAM_ID); buf.write(StreamId::new(!self.side, Directionality::Bi, self.max_remote_bi_stream)); } @@ -2093,12 +2108,13 @@ impl Connection { continue; } let len = cmp::min(stream.data.len(), max_size as usize - buf.len() - 25); - trace!(log, "STREAM"; "id" => stream.id.0, "off" => stream.offset, "len" => len, "fin" => stream.fin); let data = stream.data.split_to(len); + let fin = stream.fin && stream.data.is_empty(); + trace!(log, "STREAM"; "id" => stream.id.0, "off" => stream.offset, "len" => len, "fin" => fin); let frame = frame::Stream { id: stream.id, offset: stream.offset, - fin: stream.fin && stream.data.is_empty(), + fin: fin, data: data, }; frame.encode(true, &mut buf); @@ -2118,7 +2134,7 @@ impl Connection { } self.encrypt(is_handshake, number, &mut buf, header_len); - let timer = self.on_packet_sent(config, now, number, SentPacket { + self.on_packet_sent(config, now, number, SentPacket { acks, time: now, bytes: if ack_only { 0 } else { buf.len() as u16 }, handshake: is_handshake, @@ -2129,8 +2145,7 @@ impl Connection { // needlessly prevents us from ACKing the next packet if it's ACK-only, but saves the need for subtler logic to // avoid double-transmitting acks all the time. self.permit_ack_only = false; - - Some((buf, timer)) + Some(buf) } fn encrypt(&self, handshake: bool, number: u64, buf: &mut Vec, header_len: u16) { @@ -2183,7 +2198,7 @@ impl Connection { self.max_bi_streams = params.initial_max_streams_bidi as u64; self.max_uni_streams = params.initial_max_streams_uni as u64; self.max_data = params.initial_max_data as u64; - for i in match self.side { Side::Client => 0..config.max_remote_bi_streams, Side::Server => 1..(config.max_remote_bi_streams+1) } { + for i in match self.side { Side::Client => 0..config.max_remote_bi_streams, Side::Server => 0..(config.max_remote_bi_streams+1) } { let id = StreamId::new(!self.side, Directionality::Bi, i as u64); self.streams.get_mut(&id).unwrap().send_mut().unwrap().max_data = params.initial_max_stream_data as u64; } diff --git a/src/frame.rs b/src/frame.rs index 4bf25e057..8575efbbf 100644 --- a/src/frame.rs +++ b/src/frame.rs @@ -118,7 +118,7 @@ impl Frame { ConnectionClose(_) => Type::CONNECTION_CLOSE, ApplicationClose(_) => Type::APPLICATION_CLOSE, MaxData(_) => Type::MAX_DATA, - MaxStreamData { .. } => Type::MAX_STREAM_ID, + MaxStreamData { .. } => Type::MAX_STREAM_DATA, MaxStreamId(_) => Type::MAX_STREAM_ID, Ping => Type::PING, Blocked { .. } => Type::BLOCKED,