From 4eae666c6bf58aa4bf7b90e15a9740454368b06b Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Sat, 19 Jan 2019 15:31:05 -0800 Subject: [PATCH] Isolate packet transmit and timer update This allows us to defer composing packets when UDP is blocked, reducing latency and memory requirements. UDP can become blocked when we're generating packets faster than they can be transmitted. --- quinn-proto/src/connection.rs | 66 ++++++++------- quinn-proto/src/endpoint.rs | 149 +++++++++++++++++----------------- quinn-proto/src/lib.rs | 14 +++- quinn-proto/src/tests.rs | 94 ++++++++++----------- quinn/src/builders.rs | 2 +- quinn/src/lib.rs | 111 +++++++++++-------------- 6 files changed, 212 insertions(+), 224 deletions(-) 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