diff --git a/quinn-proto/src/connection.rs b/quinn-proto/src/connection.rs index 91e71d2c8..3a9841ad3 100644 --- a/quinn-proto/src/connection.rs +++ b/quinn-proto/src/connection.rs @@ -24,8 +24,8 @@ use crate::range_set::RangeSet; use crate::stream::{self, ReadError, Stream, WriteError}; use crate::transport_parameters::{self, TransportParameters}; use crate::{ - frame, Directionality, Frame, Side, StreamId, TransportError, MIN_INITIAL_SIZE, MIN_MTU, - RESET_TOKEN_SIZE, TIMER_GRANULARITY, VERSION, + frame, Directionality, Frame, Side, StreamId, Transmit, TransportError, MIN_INITIAL_SIZE, + MIN_MTU, RESET_TOKEN_SIZE, TIMER_GRANULARITY, VERSION, }; pub struct Connection { @@ -276,24 +276,12 @@ impl Connection { /// Connections should be polled for I/O after: /// - the application performed some I/O on the connection /// - an incoming packet is handled + /// - a packet is transmitted /// - any timer expires - pub fn poll_io(&mut self, now: u64) -> Option { - if let Some((destination, packet)) = self.next_packet(now) { - self.reset_idle_timeout(now); - return Some(Io::Transmit { - destination, - ecn: if self.sending_ecn { - Some(EcnCodepoint::ECT0) - } else { - None - }, - packet, - }); - } - + pub fn poll_io(&mut self) -> Option { for (&timer, update) in Timer::VALUES.iter().zip(self.io.timers.iter_mut()) { if let Some(update) = update.take() { - return Some(Io::TimerUpdate { timer, update }); + return Some(Io::TimerUpdate(TimerUpdate { timer, update })); } } @@ -2125,7 +2113,13 @@ impl Connection { (sent, acks) } - fn next_packet(&mut self, now: u64) -> Option<(SocketAddr, Box<[u8]>)> { + /// Returns packets to transmit + /// + /// Connections should be polled for transmit after: + /// - the application performed some I/O on the connection + /// - an incoming packet is handled + /// - the LossDetection timer expires + pub fn poll_transmit(&mut self, now: u64) -> Option { let (space_id, close) = match self.state { State::Draining | State::Drained => { return None; @@ -2333,7 +2327,16 @@ impl Connection { ); self.total_sent = self.total_sent.wrapping_add(buf.len() as u64); - Some((remote, buf.into())) + self.reset_idle_timeout(now); + Some(Transmit { + destination: remote, + packet: buf.into(), + ecn: if self.sending_ecn { + Some(EcnCodepoint::ECT0) + } else { + None + }, + }) } /// Close a connection immediately @@ -3198,14 +3201,8 @@ const MAX_ACK_BLOCKS: usize = 64; /// I/O operations to be immediately executed the backend. #[derive(Debug)] pub enum Io { - Transmit { - destination: SocketAddr, - /// Explicit congestion notification bits to set on the packet - ecn: Option, - packet: Box<[u8]>, - }, /// Stop or (re)start a timer - TimerUpdate { timer: Timer, update: TimerUpdate }, + TimerUpdate(TimerUpdate), /// Stop routing `connection_id` to this `Connection` RetireConnectionId { connection_id: ConnectionId }, } @@ -3221,7 +3218,7 @@ struct IoQueue { /// /// Note that this ordering exactly matches the values of the `Timer` enum for convenient /// indexing. - timers: [Option; 5], + timers: [Option; 5], retired_cids: Vec, } @@ -3237,24 +3234,31 @@ impl IoQueue { /// Start or reset a timer associated with this connection. fn timer_start(&mut self, timer: Timer, time: u64) { - self.timers[timer as usize] = Some(TimerUpdate::Start(time)); + self.timers[timer as usize] = Some(TimerSetting::Start(time)); } /// Start one of the timers associated with this connection. fn timer_stop(&mut self, timer: Timer) { - self.timers[timer as usize] = Some(TimerUpdate::Stop); + self.timers[timer as usize] = Some(TimerSetting::Stop); } } -/// Changes to a connection's timers +/// Change applicable to one of a connection's timers #[derive(Debug, Copy, Clone)] -pub enum TimerUpdate { +pub enum TimerSetting { /// Set the timer to expire at an a certain point in time, in absolute microseconds Start(u64), /// Cancel time timer if it's currently running Stop, } +/// Change to apply to a specific timer +#[derive(Debug, Copy, Clone)] +pub struct TimerUpdate { + pub timer: Timer, + pub update: TimerSetting, +} + struct PacketSpace { crypto: Option, dedup: Dedup, diff --git a/quinn-proto/src/endpoint.rs b/quinn-proto/src/endpoint.rs index 7596c7622..faf3411b5 100644 --- a/quinn-proto/src/endpoint.rs +++ b/quinn-proto/src/endpoint.rs @@ -25,8 +25,8 @@ use crate::packet::{ConnectionId, EcnCodepoint, Header, Packet, PacketDecodeErro use crate::stream::{ReadError, WriteError}; use crate::transport_parameters::TransportParameters; use crate::{ - Directionality, Side, StreamId, TransportError, MAX_CID_SIZE, MIN_CID_SIZE, MIN_INITIAL_SIZE, - RESET_TOKEN_SIZE, VERSION, + Directionality, Side, StreamId, Transmit, TransportError, MAX_CID_SIZE, MIN_CID_SIZE, + MIN_INITIAL_SIZE, RESET_TOKEN_SIZE, VERSION, }; /// The main entry point to the library @@ -37,7 +37,7 @@ use crate::{ pub struct Endpoint { log: Logger, rng: OsRng, - io: VecDeque, + transmits: VecDeque, incoming: VecDeque, connection_ids_initial: FnvHashMap, connection_ids: FnvHashMap, @@ -45,8 +45,10 @@ pub struct Endpoint { pub(crate) connections: Slab, config: Arc, server_config: Option, - /// Connections that might have I/O to perform - dirty_conns: FnvHashSet, + /// Connections that might have timer updates to apply perform + dirty_timers: FnvHashSet, + /// Connections that might have packets to send + needs_transmit: FnvHashSet, /// Connections that might have application-facing events to report eventful_conns: FnvHashSet, incoming_handshakes: usize, @@ -67,13 +69,14 @@ impl Endpoint { Ok(Self { log, rng, - io: VecDeque::new(), + transmits: VecDeque::new(), incoming: VecDeque::new(), connection_ids_initial: FnvHashMap::default(), connection_ids: FnvHashMap::default(), connection_remotes: FnvHashMap::default(), connections: Slab::new(), - dirty_conns: FnvHashSet::default(), + dirty_timers: FnvHashSet::default(), + needs_transmit: FnvHashSet::default(), eventful_conns: FnvHashSet::default(), incoming_handshakes: 0, config, @@ -99,40 +102,46 @@ impl Endpoint { None } - /// Get a pending IO operation - pub fn poll_io(&mut self, now: u64) -> Option { - if let Some(x) = self.io.pop_front() { + /// Get a pending timer update + pub fn poll_timers(&mut self) -> Option<(ConnectionHandle, TimerUpdate)> { + loop { + let &ch = self.dirty_timers.iter().next()?; + loop { + if let Some(io) = self.connections[ch].poll_io() { + return Some(( + ch, + match io { + connection::Io::TimerUpdate(x) => x, + connection::Io::RetireConnectionId { connection_id } => { + self.connection_ids.remove(&connection_id); + let new_cid = self.new_cid(); + self.connection_ids.insert(new_cid, ch); + self.connections[ch].issue_cid(new_cid); + continue; + } + }, + )); + } else { + self.dirty_timers.remove(&ch); + break; + } + } + } + } + + /// Get the next packet to transmit + pub fn poll_transmit(&mut self, now: u64) -> Option { + if let Some(x) = self.transmits.pop_front() { return Some(x); } loop { - let &ch = self.dirty_conns.iter().next()?; + let &ch = self.needs_transmit.iter().next()?; loop { - if let Some(io) = self.connections[ch].poll_io(now) { - return Some(match io { - connection::Io::Transmit { - destination, - ecn, - packet, - } => Io::Transmit { - destination, - ecn, - packet, - }, - connection::Io::TimerUpdate { timer, update } => Io::TimerUpdate { - connection: ch, - timer, - update, - }, - connection::Io::RetireConnectionId { connection_id } => { - self.connection_ids.remove(&connection_id); - let new_cid = self.new_cid(); - self.connection_ids.insert(new_cid, ch); - self.connections[ch].issue_cid(new_cid); - continue; - } - }); + if let Some(transmit) = self.connections[ch].poll_transmit(now) { + self.dirty_timers.insert(ch); + return Some(transmit); } else { - self.dirty_conns.remove(&ch); + self.needs_transmit.remove(&ch); break; } } @@ -169,7 +178,7 @@ impl Endpoint { .encode(&mut buf); buf.write::(0x0a1a_2a3a); // reserved version buf.write(VERSION); // supported version - self.io.push_back(Io::Transmit { + self.transmits.push_back(Transmit { destination: remote, ecn: None, packet: buf.into(), @@ -213,7 +222,8 @@ impl Endpoint { { self.conn_ready(ch); } - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); + self.dirty_timers.insert(ch); self.eventful_conns.insert(ch); return; } @@ -310,7 +320,7 @@ impl Endpoint { debug_assert!(buf.len() < inciting_dgram_len); - self.io.push_back(Io::Transmit { + self.transmits.push_back(Transmit { destination: remote, ecn: None, packet: buf.into(), @@ -335,7 +345,7 @@ impl Endpoint { server_name: server_name.into(), }), )?; - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); Ok(ch) } @@ -443,7 +453,7 @@ impl Endpoint { if self.incoming_handshakes == self.server_config.as_ref().unwrap().accept_buffer as usize { debug!(self.log, "rejecting connection due to full accept buffer"); - self.io.push_back(Io::Transmit { + self.transmits.push_back(Transmit { destination: remote, ecn: None, packet: initial_close( @@ -467,7 +477,7 @@ impl Endpoint { "rejecting connection due to invalid DCID length {len}", len = dst_cid.len() ); - self.io.push_back(Io::Transmit { + self.transmits.push_back(Transmit { destination: remote, ecn: None, packet: initial_close( @@ -519,7 +529,7 @@ impl Endpoint { encode.finish(&mut buf, header_crypto); buf.put_slice(&token); - self.io.push_back(Io::Transmit { + self.transmits.push_back(Transmit { destination: remote, ecn: None, packet: buf.into(), @@ -551,7 +561,7 @@ impl Endpoint { ) { Ok(()) => { self.incoming_handshakes += 1; - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); if self.connections[ch].has_1rtt() { self.conn_ready(ch); } @@ -559,7 +569,7 @@ impl Endpoint { Err(e) => { debug!(self.log, "handshake failed"; "reason" => %e); self.forget(ch); - self.io.push_back(Io::Transmit { + self.transmits.push_back(Transmit { destination: remote, ecn: None, packet: initial_close(crypto, header_crypto, &src_cid, &temp_loc_cid, 0, e), @@ -598,7 +608,7 @@ impl Endpoint { } self.connection_remotes .remove(&self.connections[ch].remote()); - self.dirty_conns.remove(&ch); + self.dirty_timers.remove(&ch); self.eventful_conns.remove(&ch); self.connections.remove(ch.0); } @@ -609,10 +619,16 @@ impl Endpoint { self.forget(ch); return; } - if let Timer::Idle = timer { - self.eventful_conns.insert(ch); + self.dirty_timers.insert(ch); + match timer { + Timer::LossDetection => { + self.needs_transmit.insert(ch); + } + Timer::Idle => { + self.eventful_conns.insert(ch); + } + Timer::PathValidation | Timer::Close | Timer::KeyDiscard => {} } - self.dirty_conns.insert(ch); } /// Transmit data on a stream @@ -628,7 +644,7 @@ impl Endpoint { data: &[u8], ) -> Result { let result = self.connections[ch].write(stream, data); - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); result } @@ -641,7 +657,7 @@ impl Endpoint { /// - when applied to a stream that does not have an active outgoing channel pub fn finish(&mut self, ch: ConnectionHandle, stream: StreamId) { self.connections[ch].finish(stream); - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); } /// Read data from a stream @@ -657,7 +673,7 @@ impl Endpoint { stream: StreamId, buf: &mut [u8], ) -> Result { - self.dirty_conns.insert(ch); // May need to send flow control frames after reading + self.needs_transmit.insert(ch); // May need to send flow control frames after reading match self.connections[ch].read(stream, buf) { x @ Err(ReadError::Finished) | x @ Err(ReadError::Reset { .. }) => { self.connections[ch].maybe_cleanup(stream); @@ -684,7 +700,7 @@ impl Endpoint { ch: ConnectionHandle, stream: StreamId, ) -> Result<(Bytes, u64), ReadError> { - self.dirty_conns.insert(ch); // May need to send flow control frames after reading + self.needs_transmit.insert(ch); // May need to send flow control frames after reading match self.connections[ch].read_unordered(stream) { x @ Err(ReadError::Finished) | x @ Err(ReadError::Reset { .. }) => { self.connections[ch].maybe_cleanup(stream); @@ -700,7 +716,7 @@ impl Endpoint { /// - when applied to a receive stream or an unopened send stream pub fn reset(&mut self, ch: ConnectionHandle, stream: StreamId, error_code: u16) { self.connections[ch].reset(stream, error_code); - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); } /// Instruct the peer to abandon transmitting data on a stream @@ -709,7 +725,7 @@ impl Endpoint { /// - when applied to a stream that has not begun receiving data pub fn stop_sending(&mut self, ch: ConnectionHandle, stream: StreamId, error_code: u16) { self.connections[ch].stop_sending(stream, error_code); - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); } /// Create a new stream @@ -725,7 +741,7 @@ impl Endpoint { /// Useful for preventing an otherwise idle connection from timing out. pub fn ping(&mut self, ch: ConnectionHandle) { self.connections[ch].ping(); - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); } /// Close a connection immediately @@ -738,7 +754,7 @@ impl Endpoint { return; } self.connections[ch].close(now, error_code, reason); - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); } /// Free a handshake slot for reuse @@ -752,7 +768,7 @@ impl Endpoint { pub fn accept_stream(&mut self, ch: ConnectionHandle) -> Option { let id = self.connections[ch].accept()?; - self.dirty_conns.insert(ch); + self.needs_transmit.insert(ch); Some(id) } @@ -982,23 +998,6 @@ impl From for Event { } } -/// I/O operations to be immediately executed the backend. -#[derive(Debug)] -pub enum Io { - Transmit { - destination: SocketAddr, - /// Explicit congestion notification bits to set on the packet - ecn: Option, - packet: Box<[u8]>, - }, - /// Start, stop, or reset a timer - TimerUpdate { - connection: ConnectionHandle, - timer: Timer, - update: TimerUpdate, - }, -} - #[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq)] pub enum Timer { LossDetection = 0, diff --git a/quinn-proto/src/lib.rs b/quinn-proto/src/lib.rs index 1f50b2206..9ea0dbfdf 100644 --- a/quinn-proto/src/lib.rs +++ b/quinn-proto/src/lib.rs @@ -11,6 +11,7 @@ extern crate lazy_static; extern crate slog; use std::fmt; +use std::net::SocketAddr; use std::ops; mod coding; @@ -22,7 +23,7 @@ mod transport_parameters; mod varint; mod connection; -pub use crate::connection::{ConnectionError, TimerUpdate}; +pub use crate::connection::{ConnectionError, TimerSetting, TimerUpdate}; mod crypto; pub use crate::crypto::{ClientConfig, ConnectError, TokenKey}; @@ -33,7 +34,7 @@ pub use crate::frame::{ApplicationClose, ConnectionClose}; mod endpoint; pub use crate::endpoint::{ - Config, ConnectionHandle, Endpoint, EndpointError, Event, Io, ServerConfig, Timer, + Config, ConnectionHandle, Endpoint, EndpointError, Event, ServerConfig, Timer, }; mod packet; @@ -193,6 +194,15 @@ impl coding::Codec for StreamId { } } +/// An outgoing packet +#[derive(Debug)] +pub struct Transmit { + pub destination: SocketAddr, + /// Explicit congestion notification bits to set on the packet + pub ecn: Option, + pub packet: Box<[u8]>, +} + // // Useful internal constants // diff --git a/quinn-proto/src/tests.rs b/quinn-proto/src/tests.rs index 192ee6483..55d9b4d12 100644 --- a/quinn-proto/src/tests.rs +++ b/quinn-proto/src/tests.rs @@ -182,19 +182,19 @@ impl Pair { fn drive_client(&mut self) { trace!(self.log, "client running"); self.client.drive(&self.log, self.time, self.server.addr); - for (destination, packet, ecn) in self.client.outbound.drain(..) { - if packet[0] & packet::LONG_HEADER_FORM == 0 { - let spin = packet[0] & packet::SPIN_BIT != 0; + for x in self.client.outbound.drain(..) { + if x.packet[0] & packet::LONG_HEADER_FORM == 0 { + let spin = x.packet[0] & packet::SPIN_BIT != 0; self.spins += (spin == self.last_spin) as u64; self.last_spin = spin; } if let Some(ref socket) = self.client.socket { - socket.send_to(&packet, destination).unwrap(); + socket.send_to(&x.packet, x.destination).unwrap(); } - if self.server.addr == destination { + if self.server.addr == x.destination { self.server .inbound - .push_back((self.time + self.latency, ecn, packet)); + .push_back((self.time + self.latency, x.ecn, x.packet)); } } } @@ -202,14 +202,14 @@ impl Pair { fn drive_server(&mut self) { trace!(self.log, "server running"); self.server.drive(&self.log, self.time, self.client.addr); - for (destination, packet, ecn) in self.server.outbound.drain(..) { + for x in self.server.outbound.drain(..) { if let Some(ref socket) = self.server.socket { - socket.send_to(&packet, destination).unwrap(); + socket.send_to(&x.packet, x.destination).unwrap(); } - if self.client.addr == destination { + if self.client.addr == x.destination { self.client .inbound - .push_back((self.time + self.latency, ecn, packet)); + .push_back((self.time + self.latency, x.ecn, x.packet)); } } } @@ -235,8 +235,8 @@ struct TestEndpoint { socket: Option, timers: [u64; 5], conn: Option, - outbound: VecDeque<(SocketAddr, Box<[u8]>, Option)>, - delayed: VecDeque<(SocketAddr, Box<[u8]>, Option)>, + outbound: VecDeque, + delayed: VecDeque, inbound: VecDeque<(u64, Option, Box<[u8]>)>, } @@ -292,45 +292,33 @@ impl TestEndpoint { self.endpoint .handle(now, remote, ecn, Vec::from(packet).into()); } - while let Some(x) = self.endpoint.poll_io(now) { - match x { - Io::Transmit { - destination, - packet, - ecn, - } => { - self.outbound.push_back((destination, packet, ecn)); + while let Some(x) = self.endpoint.poll_transmit(now) { + self.outbound.push_back(x); + } + while let Some((ch, x)) = self.endpoint.poll_timers() { + self.conn = Some(ch); + let time = match x.update { + TimerSetting::Stop => { + trace!( + log, + "{side:?} {timer:?} stop", + side = self.side, + timer = x.timer + ); + u64::max_value() } - Io::TimerUpdate { - timer, - update, - connection, - } => { - self.conn = Some(connection); - let time = match update { - TimerUpdate::Stop => { - trace!( - log, - "{side:?} {timer:?} stop", - side = self.side, - timer = timer - ); - u64::max_value() - } - TimerUpdate::Start(time) => { - trace!( - log, - "{side:?} {timer:?} set to expire at {:?}", - Duration::from_micros(time), - side = self.side, - timer = timer, - ); - time - } - }; - self.timers[timer as usize] = time; + TimerSetting::Start(time) => { + trace!( + log, + "{side:?} {timer:?} set to expire at {:?}", + Duration::from_micros(time), + side = self.side, + timer = x.timer, + ); + time } - } + }; + self.timers[x.timer as usize] = time; } } @@ -397,16 +385,16 @@ fn version_negotiate() { )[..] .into(), ); - let io = server.poll_io(0); - assert_matches!(io, Some(Io::Transmit { .. })); - if let Some(Io::Transmit { packet, .. }) = io { + let io = server.poll_transmit(0); + assert!(io.is_some()); + if let Some(Transmit { packet, .. }) = io { assert_ne!(packet[0] & 0x80, 0); assert_eq!(&packet[1..14], hex!("00000000 11 00000000 00000000")); assert!(packet[14..] .chunks(4) .any(|x| BigEndian::read_u32(x) == VERSION)); } - assert_matches!(server.poll_io(0), None); + assert_matches!(server.poll_transmit(0), None); assert_matches!(server.poll(), None); } diff --git a/quinn/src/builders.rs b/quinn/src/builders.rs index b547e6275..d9272fc57 100644 --- a/quinn/src/builders.rs +++ b/quinn/src/builders.rs @@ -66,7 +66,7 @@ impl<'a> EndpointBuilder<'a> { log: self.logger.clone(), socket, inner: quinn::Endpoint::new(self.logger, self.config, self.server_config)?, - outgoing: VecDeque::new(), + outgoing: None, epoch: Instant::now(), pending: FnvHashMap::default(), timers: FuturesUnordered::new(), diff --git a/quinn/src/lib.rs b/quinn/src/lib.rs index d3f0b9a66..02db88e63 100644 --- a/quinn/src/lib.rs +++ b/quinn/src/lib.rs @@ -76,7 +76,7 @@ use futures::task::{self, Task}; use futures::unsync::oneshot; use futures::Stream as FuturesStream; use futures::{Async, Future, Poll, Sink}; -use quinn_proto::{self as quinn, ConnectionHandle, Directionality, Side, StreamId}; +use quinn_proto::{self as quinn, ConnectionHandle, Directionality, Side, StreamId, TimerUpdate}; use slog::Logger; use tokio_io::{AsyncRead, AsyncWrite}; use tokio_timer::Delay; @@ -347,17 +347,32 @@ impl Future for Driver { } let _ = endpoint.incoming.poll_complete(); let mut blocked = false; - while !endpoint.outgoing.is_empty() { - { - let (destination, ecn, packet) = endpoint.outgoing.front().unwrap(); - match endpoint.socket.poll_send(destination, *ecn, packet) { + if let Some(ref x) = endpoint.outgoing { + match endpoint.socket.poll_send(&x.destination, x.ecn, &x.packet) { + Ok(Async::Ready(_)) => { + endpoint.outgoing = None; + } + Ok(Async::NotReady) => { + blocked = true; + } + Err(ref e) if e.kind() == io::ErrorKind::PermissionDenied => { + blocked = true; + } + Err(e) => { + return Err(e); + } + } + } + if !blocked { + while let Some(x) = endpoint.inner.poll_transmit(now) { + match endpoint.socket.poll_send(&x.destination, x.ecn, &x.packet) { Ok(Async::Ready(_)) => {} Ok(Async::NotReady) => { - blocked = true; + endpoint.outgoing = Some(x); break; } Err(ref e) if e.kind() == io::ErrorKind::PermissionDenied => { - blocked = true; + endpoint.outgoing = Some(x); break; } Err(e) => { @@ -365,38 +380,36 @@ impl Future for Driver { } } } - endpoint.outgoing.pop_front(); } - while let Some(io) = endpoint.inner.poll_io(now) { - use crate::quinn::Io::*; - match io { - Transmit { - destination, - packet, - ecn, - } => { - if !blocked { - match endpoint.socket.poll_send(&destination, ecn, &packet) { - Ok(Async::Ready(_)) => {} - Ok(Async::NotReady) => { - blocked = true; - } - Err(ref e) if e.kind() == io::ErrorKind::PermissionDenied => { - blocked = true; - } - Err(e) => { - return Err(e); + let mut timer_fired = false; + loop { + match endpoint.timers.poll() { + Ok(Async::Ready(Some(Some((ch, timer))))) => { + trace!(endpoint.log, "timeout"; "timer" => ?timer); + endpoint.inner.timeout(now, ch, timer); + if timer == quinn::Timer::Close { + // Connection drained + if let Some(p) = endpoint.pending.get_mut(&ch) { + p.drained = true; + if let Some(x) = p.closing.take() { + let _ = x.send(()); } } } - if blocked { - endpoint.outgoing.push_front((destination, ecn, packet)); - } + timer_fired = true; } + Ok(Async::Ready(Some(None))) => {} + Ok(Async::Ready(None)) | Ok(Async::NotReady) => { + break; + } + Err(()) => unreachable!(), + } + } + while let Some((ch, x)) = endpoint.inner.poll_timers() { + match x { TimerUpdate { - connection: ch, timer: timer @ quinn::Timer::Close, - update: quinn::TimerUpdate::Start(time), + update: quinn::TimerSetting::Start(time), } => { let instant = endpoint.epoch + duration_micros(time); endpoint.timers.push(Timer { @@ -407,9 +420,8 @@ impl Future for Driver { }); } TimerUpdate { - connection: ch, timer, - update: quinn::TimerUpdate::Start(time), + update: quinn::TimerSetting::Start(time), } => { let pending = endpoint.pending.get_mut(&ch).unwrap(); let cancel = &mut pending.cancel_timers[timer as usize]; @@ -428,9 +440,8 @@ impl Future for Driver { }); } TimerUpdate { - connection: ch, timer, - update: quinn::TimerUpdate::Stop, + update: quinn::TimerSetting::Stop, } => { trace!(endpoint.log, "timer stop"; "timer" => ?timer); // If a connection was lost, we already canceled its loss/idle timers. @@ -442,31 +453,7 @@ impl Future for Driver { } } } - let mut fired = false; - loop { - match endpoint.timers.poll() { - Ok(Async::Ready(Some(Some((ch, timer))))) => { - trace!(endpoint.log, "timeout"; "timer" => ?timer); - endpoint.inner.timeout(now, ch, timer); - if timer == quinn::Timer::Close { - // Connection drained - if let Some(p) = endpoint.pending.get_mut(&ch) { - p.drained = true; - if let Some(x) = p.closing.take() { - let _ = x.send(()); - } - } - } - fired = true; - } - Ok(Async::Ready(Some(None))) => {} - Ok(Async::Ready(None)) | Ok(Async::NotReady) => { - break; - } - Err(()) => unreachable!(), - } - } - if !fired { + if !timer_fired { break; } } @@ -489,7 +476,7 @@ struct EndpointInner { log: Logger, socket: UdpSocket, inner: quinn::Endpoint, - outgoing: VecDeque<(SocketAddr, Option, Box<[u8]>)>, + outgoing: Option, epoch: Instant, pending: FnvHashMap, // TODO: Replace this with something custom that avoids using oneshots to cancel