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

@@ -83,6 +83,8 @@ pub struct Connection {
     prev_crypto: Option<PrevCrypto>,
     /// Latest PATH_CHALLENGE token issued to the peer along the current path
     path_challenge: Option<u64>,
+    /// 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<Event> {
-        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<StreamId> {
+        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<StreamId>,
+    incoming: VecDeque<StreamId>,
 }
 
 impl Streams {

quinn-proto/src/endpoint.rs

@@ -726,6 +726,12 @@ impl Endpoint {
         self.incoming.pop_front()
     }
 
+    pub fn accept_stream(&mut self, conn: ConnectionHandle) -> Option<StreamId> {
+        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<String> },
     /// 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

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,
 }
 

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),

quinn/src/lib.rs

@@ -148,7 +148,6 @@ struct Pending {
     uni_opening: VecDeque<oneshot::Sender<Result<StreamId, ConnectionError>>>,
     bi_opening: VecDeque<oneshot::Sender<Result<StreamId, ConnectionError>>>,
     cancel_timers: [Option<oneshot::Sender<()>>; 5],
-    incoming_streams: VecDeque<StreamId>,
     incoming_streams_reader: Option<Task>,
     finishing: FnvHashMap<StreamId, oneshot::Sender<Option<ConnectionError>>>,
     error: Option<ConnectionError>,
@@ -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<Option<Self::Item>, 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 {