gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-10-03 21:03:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Isolate packet transmit and timer update

Browse Source 
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.
This commit is contained in:
Benjamin Saunders
2019-01-19 15:31:05 -08:00
committed by Dirkjan Ochtman
parent 82059b57fd
commit 4eae666c6b
6 changed files with 212 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quinn-proto/src/connection.rs
+35 -31
View File
@@ -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<Io> {
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<Io> {
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<Transmit> {
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<EcnCodepoint>,
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<TimerUpdate>; 5],
timers: [Option<TimerSetting>; 5],
retired_cids: Vec<ConnectionId>,
}
@@ -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<CryptoSpace>,
dedup: Dedup,
quinn-proto/src/endpoint.rs
+74 -75
View File
@@ -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<Io>,
transmits: VecDeque<Transmit>,
incoming: VecDeque<ConnectionHandle>,
connection_ids_initial: FnvHashMap<ConnectionId, ConnectionHandle>,
connection_ids: FnvHashMap<ConnectionId, ConnectionHandle>,
@@ -45,8 +45,10 @@ pub struct Endpoint {
pub(crate) connections: Slab<Connection>,
config: Arc<Config>,
server_config: Option<ServerConfig>,
/// Connections that might have I/O to perform
dirty_conns: FnvHashSet<ConnectionHandle>,
/// Connections that might have timer updates to apply perform
dirty_timers: FnvHashSet<ConnectionHandle>,
/// Connections that might have packets to send
needs_transmit: FnvHashSet<ConnectionHandle>,
/// Connections that might have application-facing events to report
eventful_conns: FnvHashSet<ConnectionHandle>,
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<Io> {
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<Transmit> {
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::<u32>(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<usize, WriteError> {
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<usize, 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(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<StreamId> {
let id = self.connections[ch].accept()?;
self.dirty_conns.insert(ch);
self.needs_transmit.insert(ch);
Some(id)
}
@@ -982,23 +998,6 @@ impl From<ConnectionError> 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<EcnCodepoint>,
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,
quinn-proto/src/lib.rs
+12 -2
View File
@@ -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<EcnCodepoint>,
pub packet: Box<[u8]>,
}
//
// Useful internal constants
//
quinn-proto/src/tests.rs
+41 -53
View File
@@ -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<UdpSocket>,
timers: [u64; 5],
conn: Option<ConnectionHandle>,
outbound: VecDeque<(SocketAddr, Box<[u8]>, Option<EcnCodepoint>)>,
delayed: VecDeque<(SocketAddr, Box<[u8]>, Option<EcnCodepoint>)>,
outbound: VecDeque<Transmit>,
delayed: VecDeque<Transmit>,
inbound: VecDeque<(u64, Option<EcnCodepoint>, 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);
}
quinn/src/builders.rs
+1 -1
View File
@@ -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(),
quinn/src/lib.rs
+49 -62
View File
@@ -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<quinn::EcnCodepoint>, Box<[u8]>)>,
outgoing: Option<quinn::Transmit>,
epoch: Instant,
pending: FnvHashMap<ConnectionHandle, Pending>,
// TODO: Replace this with something custom that avoids using oneshots to cancel