From 2d8f87d1fa3e9de9f32cf0bc0ca2a131c84ea222 Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Mon, 7 Jan 2019 21:31:59 -0800 Subject: [PATCH] Implement dynamic stream ID flow control This allows applications to easily handle arbitrary numbers of streams concurrently with predictable resource use. --- quinn-proto/src/connection.rs | 53 ++++++++++++++++++++++------------- quinn-proto/src/endpoint.rs | 17 ++++++----- quinn-proto/src/stream.rs | 2 +- quinn-proto/src/tests.rs | 29 +++++++++++-------- quinn/src/lib.rs | 20 ++++++------- 5 files changed, 72 insertions(+), 49 deletions(-) diff --git a/quinn-proto/src/connection.rs b/quinn-proto/src/connection.rs index 173ee728b..2ba98c903 100644 --- a/quinn-proto/src/connection.rs +++ b/quinn-proto/src/connection.rs @@ -83,6 +83,8 @@ pub struct Connection { prev_crypto: Option, /// Latest PATH_CHALLENGE token issued to the peer along the current path path_challenge: Option, + /// Whether the remote endpoint has opened any streams the application doesn't know about yet + stream_opened: bool, // // Queued non-retransmittable 1-RTT data @@ -216,6 +218,7 @@ impl Connection { highest_space: SpaceId::Initial, prev_crypto: None, path_challenge: None, + stream_opened: false, path_challenge_pending: false, ping_pending: false, @@ -249,6 +252,7 @@ impl Connection { max_remote_uni: config.stream_window_uni, max_remote_bi: config.stream_window_bidi, finished: Vec::new(), + incoming: VecDeque::new(), }, config, rem_cids: Vec::new(), @@ -299,7 +303,15 @@ impl Connection { /// - an incoming packet is handled, or /// - the idle timer expires pub fn poll(&mut self) -> Option { - self.events.pop_front() + if mem::replace(&mut self.stream_opened, false) { + return Some(Event::StreamOpened); + } + + if let Some(x) = self.events.pop_front() { + return Some(x); + } + + None } fn on_packet_sent(&mut self, now: u64, space: SpaceId, packet_number: u64, packet: SentPacket) { @@ -1422,11 +1434,13 @@ impl Connection { } } - let fresh = mem::replace(&mut rs.fresh, false); - self.events.push_back(Event::StreamReadable { - stream: frame.id, - fresh, - }); + if mem::replace(&mut rs.fresh, false) { + self.stream_opened = self.streams.incoming.is_empty(); + self.streams.incoming.push_back(frame.id); + } else { + self.events + .push_back(Event::StreamReadable { stream: frame.id }); + } self.data_recvd += new_bytes; } Frame::Ack(ack) => { @@ -1554,8 +1568,12 @@ impl Connection { } }; self.data_recvd += final_offset.saturating_sub(offset); - self.events - .push_back(Event::StreamReadable { stream: id, fresh }); + if fresh { + self.stream_opened = self.streams.incoming.is_empty(); + self.streams.incoming.push_back(id); + } else { + self.events.push_back(Event::StreamReadable { stream: id }); + } } Frame::DataBlocked { offset } => { debug!(self.log, "peer claims to be blocked at connection level"; "offset" => offset); @@ -2239,23 +2257,13 @@ impl Connection { /// /// Called when one side of a stream transitions to a closed state pub fn maybe_cleanup(&mut self, id: StreamId) { - let new = match self.streams.streams.entry(id) { + match self.streams.streams.entry(id) { hash_map::Entry::Vacant(_) => unreachable!(), hash_map::Entry::Occupied(e) => { if e.get().is_closed() { e.remove_entry(); - if id.initiator() != self.side { - Some(id.directionality()) - } else { - None - } - } else { - None } } - }; - if let Some(ty) = new { - self.alloc_remote_stream(ty); } } @@ -2291,6 +2299,12 @@ impl Connection { self.streams.streams.insert(id, stream); } + pub fn accept(&mut self) -> Option { + let id = self.streams.incoming.pop_front()?; + self.alloc_remote_stream(id.directionality()); + Some(id) + } + pub fn finish(&mut self, id: StreamId) { let ss = self .streams @@ -2668,6 +2682,7 @@ struct Streams { max_remote_bi: u64, finished: Vec, + incoming: VecDeque, } impl Streams { diff --git a/quinn-proto/src/endpoint.rs b/quinn-proto/src/endpoint.rs index 505b033a0..6db207332 100644 --- a/quinn-proto/src/endpoint.rs +++ b/quinn-proto/src/endpoint.rs @@ -726,6 +726,12 @@ impl Endpoint { self.incoming.pop_front() } + pub fn accept_stream(&mut self, conn: ConnectionHandle) -> Option { + let id = self.connections[conn.0].accept()?; + self.dirty_conns.insert(conn); + Some(id) + } + #[doc(hidden)] pub fn force_key_update(&mut self, conn: ConnectionHandle) { self.connections[conn.0].force_key_update(); @@ -919,13 +925,10 @@ pub enum Event { Connected { protocol: Option }, /// A connection was lost. ConnectionLost { reason: ConnectionError }, - /// A stream has data or errors waiting to be read - StreamReadable { - /// The affected stream - stream: StreamId, - /// Whether this is the first event on the stream - fresh: bool, - }, + /// One or more new streams has been opened and is readable + StreamOpened, + /// An existing stream has data or errors waiting to be read + StreamReadable { stream: StreamId }, /// A formerly write-blocked stream might now accept a write StreamWritable { stream: StreamId }, /// All data sent on `stream` has been received by the peer diff --git a/quinn-proto/src/stream.rs b/quinn-proto/src/stream.rs index 910627721..ee5f6ac81 100644 --- a/quinn-proto/src/stream.rs +++ b/quinn-proto/src/stream.rs @@ -138,7 +138,7 @@ pub struct Recv { /// reads pub unordered: bool, pub assembler: Assembler, - /// Whether the application is aware of this stream yet + /// Whether the application has been notified of this stream yet pub fresh: bool, } diff --git a/quinn-proto/src/tests.rs b/quinn-proto/src/tests.rs index 427e4d2da..2a32180e4 100644 --- a/quinn-proto/src/tests.rs +++ b/quinn-proto/src/tests.rs @@ -529,7 +529,8 @@ fn finish_stream() { assert_matches!(pair.client.poll(), Some((conn, Event::StreamFinished { stream })) if conn == client_conn && stream == s); assert_matches!(pair.client.poll(), None); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: true })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn); + assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s); assert_matches!(pair.server.poll(), None); assert_matches!(pair.server.read_unordered(server_conn, s), Ok((ref data, 0)) if data == MSG); assert_matches!( @@ -554,8 +555,9 @@ fn reset_stream() { pair.client.reset(client_conn, s, ERROR); pair.drive(); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: true })) if conn == server_conn && stream == s); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: false })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn); + assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream })) if conn == server_conn && stream == s); assert_matches!(pair.server.read_unordered(server_conn, s), Ok((ref data, 0)) if data == MSG); assert_matches!( pair.server.read_unordered(server_conn, s), @@ -579,8 +581,9 @@ fn stop_stream() { pair.server.stop_sending(server_conn, s, ERROR); pair.drive(); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: true })) if conn == server_conn && stream == s); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: false })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn); + assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream })) if conn == server_conn && stream == s); assert_matches!(pair.server.read_unordered(server_conn, s), Ok((ref data, 0)) if data == MSG); assert_matches!( pair.server.read_unordered(server_conn, s), @@ -717,7 +720,8 @@ fn stream_id_backpressure() { pair.drive(); assert_matches!(pair.client.poll(), Some((conn, Event::StreamFinished { stream })) if conn == client_conn && stream == s); assert_matches!(pair.client.poll(), None); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: true })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn); + assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s); assert_matches!( pair.server.read_unordered(server_conn, s), Err(ReadError::Finished) @@ -735,7 +739,8 @@ fn stream_id_backpressure() { pair.client.finish(client_conn, s); pair.drive(); // Make sure the server actually processes data on the newly-available stream - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: true })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn); + assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s); assert_matches!(pair.server.poll(), None); assert_matches!( pair.server.read_unordered(server_conn, s), @@ -756,7 +761,8 @@ fn key_update() { pair.client.write(client_conn, s, MSG1).unwrap(); pair.drive(); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: true })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn); + assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s); assert_matches!(pair.server.poll(), None); assert_matches!( pair.server.read_unordered(server_conn, s), @@ -769,7 +775,7 @@ fn key_update() { pair.client.write(client_conn, s, MSG2).unwrap(); pair.drive(); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: false })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream })) if conn == server_conn && stream == s); assert_matches!(pair.server.poll(), None); assert_matches!( pair.server.read_unordered(server_conn, s), @@ -801,8 +807,9 @@ fn key_update_reordered() { pair.client.finish_delay(); pair.drive(); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: true })) if conn == server_conn && stream == s); - assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream, fresh: false })) if conn == server_conn && stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn); + assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s); + assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream })) if conn == server_conn && stream == s); assert_matches!(pair.server.poll(), None); assert_matches!( pair.server.read_unordered(server_conn, s), diff --git a/quinn/src/lib.rs b/quinn/src/lib.rs index a86598eb7..f42d27526 100644 --- a/quinn/src/lib.rs +++ b/quinn/src/lib.rs @@ -148,7 +148,6 @@ struct Pending { uni_opening: VecDeque>>, bi_opening: VecDeque>>, cancel_timers: [Option>; 5], - incoming_streams: VecDeque, incoming_streams_reader: Option, finishing: FnvHashMap>>, error: Option, @@ -165,7 +164,6 @@ impl Pending { uni_opening: VecDeque::new(), bi_opening: VecDeque::new(), cancel_timers: [None, None, None, None, None], - incoming_streams: VecDeque::new(), incoming_streams_reader: None, finishing: FnvHashMap::default(), error: None, @@ -702,17 +700,17 @@ impl Future for Driver { writer.notify(); } } - StreamReadable { stream, fresh } => { + StreamOpened => { + let pending = endpoint.pending.get_mut(&connection).unwrap(); + if let Some(x) = pending.incoming_streams_reader.take() { + x.notify(); + } + } + StreamReadable { stream } => { let pending = endpoint.pending.get_mut(&connection).unwrap(); if let Some(reader) = pending.blocked_readers.remove(&stream) { reader.notify(); } - if fresh { - pending.incoming_streams.push_back(stream); - if let Some(x) = pending.incoming_streams_reader.take() { - x.notify(); - } - } } StreamAvailable { directionality } => { let pending = endpoint.pending.get_mut(&connection).unwrap(); @@ -1487,8 +1485,7 @@ impl FuturesStream for IncomingStreams { type Error = ConnectionError; fn poll(&mut self) -> Poll, Self::Error> { let mut endpoint = self.0.endpoint.borrow_mut(); - let pending = endpoint.pending.get_mut(&self.0.conn).unwrap(); - if let Some(x) = pending.incoming_streams.pop_front() { + if let Some(x) = endpoint.inner.accept_stream(self.0.conn) { let stream = BiStream::new(self.0.clone(), x); let stream = if x.directionality() == Directionality::Uni { NewStream::Uni(RecvStream(stream)) @@ -1497,6 +1494,7 @@ impl FuturesStream for IncomingStreams { }; return Ok(Async::Ready(Some(stream))); } + let pending = endpoint.pending.get_mut(&self.0.conn).unwrap(); if let Some(ref x) = pending.error { Err(x.clone()) } else {