quinn_proto/connection/streams/
recv.rs

1use std::collections::hash_map::Entry;
2use std::mem;
3
4use thiserror::Error;
5use tracing::debug;
6
7use super::state::get_or_insert_recv;
8use super::{ClosedStream, Retransmits, ShouldTransmit, StreamId, StreamsState};
9use crate::connection::assembler::{Assembler, Chunk, IllegalOrderedRead};
10use crate::connection::streams::state::StreamRecv;
11use crate::{TransportError, VarInt, frame};
12
13#[derive(Debug, Default)]
14pub(super) struct Recv {
15    // NB: when adding or removing fields, remember to update `reinit`.
16    state: RecvState,
17    pub(super) assembler: Assembler,
18    sent_max_stream_data: u64,
19    pub(super) end: u64,
20    pub(super) stopped: bool,
21}
22
23impl Recv {
24    pub(super) fn new(initial_max_data: u64) -> Box<Self> {
25        Box::new(Self {
26            state: RecvState::default(),
27            assembler: Assembler::new(),
28            sent_max_stream_data: initial_max_data,
29            end: 0,
30            stopped: false,
31        })
32    }
33
34    /// Reset to the initial state
35    pub(super) fn reinit(&mut self, initial_max_data: u64) {
36        self.state = RecvState::default();
37        self.assembler.reinit();
38        self.sent_max_stream_data = initial_max_data;
39        self.end = 0;
40        self.stopped = false;
41    }
42
43    /// Process a STREAM frame
44    ///
45    /// Return value is `(number_of_new_bytes_ingested, stream_is_closed)`
46    pub(super) fn ingest(
47        &mut self,
48        frame: frame::Stream,
49        payload_len: usize,
50        received: u64,
51        max_data: u64,
52    ) -> Result<(u64, bool), TransportError> {
53        let end = frame.offset + frame.data.len() as u64;
54        if end >= 2u64.pow(62) {
55            return Err(TransportError::FLOW_CONTROL_ERROR(
56                "maximum stream offset too large",
57            ));
58        }
59
60        if let Some(final_offset) = self.final_offset() {
61            if end > final_offset || (frame.fin && end != final_offset) {
62                debug!(end, final_offset, "final size error");
63                return Err(TransportError::FINAL_SIZE_ERROR(""));
64            }
65        }
66
67        let new_bytes = self.credit_consumed_by(end, received, max_data)?;
68
69        // Stopped streams don't need to wait for the actual data, they just need to know
70        // how much there was.
71        if frame.fin && !self.stopped {
72            if let RecvState::Recv { ref mut size } = self.state {
73                *size = Some(end);
74            }
75        }
76
77        self.end = self.end.max(end);
78        // Don't bother storing data or releasing stream-level flow control credit if the stream's
79        // already stopped
80        if !self.stopped {
81            self.assembler
82                .insert(frame.offset, frame.data, payload_len)
83                .map_err(|_| TransportError::INTERNAL_ERROR("too many gaps in stream buffer"))?;
84        }
85
86        Ok((new_bytes, frame.fin && self.stopped))
87    }
88
89    pub(super) fn stop(&mut self) -> Result<(u64, ShouldTransmit), ClosedStream> {
90        if self.stopped {
91            return Err(ClosedStream { _private: () });
92        }
93
94        self.stopped = true;
95        self.assembler.clear();
96        // Issue flow control credit for unread data
97        let read_credits = self.end - self.assembler.bytes_read();
98        // This may send a spurious STOP_SENDING if we've already received all data, but it's a bit
99        // fiddly to distinguish that from the case where we've received a FIN but are missing some
100        // data that the peer might still be trying to retransmit, in which case a STOP_SENDING is
101        // still useful.
102        Ok((read_credits, ShouldTransmit(self.is_receiving())))
103    }
104
105    /// Returns the window that should be advertised in a `MAX_STREAM_DATA` frame
106    ///
107    /// The method returns a tuple which consists of the window that should be
108    /// announced, as well as a boolean parameter which indicates if a new
109    /// transmission of the value is recommended. If the boolean value is
110    /// `false` the new window should only be transmitted if a previous transmission
111    /// had failed.
112    pub(super) fn max_stream_data(&mut self, stream_receive_window: u64) -> (u64, ShouldTransmit) {
113        let max_stream_data = self.assembler.bytes_read() + stream_receive_window;
114
115        // Only announce a window update if it's significant enough
116        // to make it worthwhile sending a MAX_STREAM_DATA frame.
117        // We use here a fraction of the configured stream receive window to make
118        // the decision, and accommodate for streams using bigger windows requiring
119        // less updates. A fixed size would also work - but it would need to be
120        // smaller than `stream_receive_window` in order to make sure the stream
121        // does not get stuck.
122        let diff = max_stream_data - self.sent_max_stream_data;
123        let transmit = self.can_send_flow_control() && diff >= (stream_receive_window / 8);
124        (max_stream_data, ShouldTransmit(transmit))
125    }
126
127    /// Records that a `MAX_STREAM_DATA` announcing a certain window was sent
128    ///
129    /// This will suppress enqueuing further `MAX_STREAM_DATA` frames unless
130    /// either the previous transmission was not acknowledged or the window
131    /// further increased.
132    pub(super) fn record_sent_max_stream_data(&mut self, sent_value: u64) {
133        if sent_value > self.sent_max_stream_data {
134            self.sent_max_stream_data = sent_value;
135        }
136    }
137
138    /// Whether the total amount of data that the peer will send on this stream is unknown
139    ///
140    /// True until we've received either a reset or the final frame.
141    ///
142    /// Implies that the sender might benefit from stream-level flow control updates, and we might
143    /// need to issue connection-level flow control updates due to flow control budget use by this
144    /// stream in the future, even if it's been stopped.
145    pub(super) fn final_offset_unknown(&self) -> bool {
146        matches!(self.state, RecvState::Recv { size: None })
147    }
148
149    /// Whether stream-level flow control updates should be sent for this stream
150    pub(super) fn can_send_flow_control(&self) -> bool {
151        // Stream-level flow control is redundant if the sender has already sent the whole stream,
152        // and moot if we no longer want data on this stream.
153        self.final_offset_unknown() && !self.stopped
154    }
155
156    /// Whether data is still being accepted from the peer
157    pub(super) fn is_receiving(&self) -> bool {
158        matches!(self.state, RecvState::Recv { .. })
159    }
160
161    fn final_offset(&self) -> Option<u64> {
162        match self.state {
163            RecvState::Recv { size } => size,
164            RecvState::ResetRecvd { size, .. } => Some(size),
165        }
166    }
167
168    /// Returns `false` iff the reset was redundant
169    pub(super) fn reset(
170        &mut self,
171        error_code: VarInt,
172        final_offset: VarInt,
173        received: u64,
174        max_data: u64,
175    ) -> Result<bool, TransportError> {
176        // Validate final_offset
177        if let Some(offset) = self.final_offset() {
178            if offset != final_offset.into_inner() {
179                return Err(TransportError::FINAL_SIZE_ERROR("inconsistent value"));
180            }
181        } else if self.end > u64::from(final_offset) {
182            return Err(TransportError::FINAL_SIZE_ERROR(
183                "lower than high water mark",
184            ));
185        }
186        self.credit_consumed_by(final_offset.into(), received, max_data)?;
187
188        if matches!(self.state, RecvState::ResetRecvd { .. }) {
189            return Ok(false);
190        }
191        self.state = RecvState::ResetRecvd {
192            size: final_offset.into(),
193            error_code,
194        };
195        // Nuke buffers so that future reads fail immediately, which ensures future reads don't
196        // issue flow control credit redundant to that already issued. We could instead special-case
197        // reset streams during read, but it's unclear if there's any benefit to retaining data for
198        // reset streams.
199        self.assembler.clear();
200        Ok(true)
201    }
202
203    pub(super) fn reset_code(&self) -> Option<VarInt> {
204        match self.state {
205            RecvState::ResetRecvd { error_code, .. } => Some(error_code),
206            _ => None,
207        }
208    }
209
210    /// Compute the amount of flow control credit consumed, or return an error if more was consumed
211    /// than issued
212    fn credit_consumed_by(
213        &self,
214        offset: u64,
215        received: u64,
216        max_data: u64,
217    ) -> Result<u64, TransportError> {
218        let prev_end = self.end;
219        let new_bytes = offset.saturating_sub(prev_end);
220        if offset > self.sent_max_stream_data || received + new_bytes > max_data {
221            debug!(
222                received,
223                new_bytes,
224                max_data,
225                offset,
226                stream_max_data = self.sent_max_stream_data,
227                "flow control error"
228            );
229            return Err(TransportError::FLOW_CONTROL_ERROR(""));
230        }
231
232        Ok(new_bytes)
233    }
234}
235
236/// Chunks returned from [`RecvStream::read()`][crate::RecvStream::read].
237///
238/// ### Note: Finalization Needed
239/// Bytes read from the stream are not released from the congestion window until
240/// either [`Self::finalize()`] is called, or this type is dropped.
241///
242/// It is recommended that you call [`Self::finalize()`] because it returns a flag
243/// telling you whether reading from the stream has resulted in the need to transmit a packet.
244///
245/// If this type is leaked, the stream will remain blocked on the remote peer until
246/// another read from the stream is done.
247pub struct Chunks<'a> {
248    id: StreamId,
249    ordered: bool,
250    streams: &'a mut StreamsState,
251    pending: &'a mut Retransmits,
252    state: ChunksState,
253    read: u64,
254}
255
256impl<'a> Chunks<'a> {
257    pub(super) fn new(
258        id: StreamId,
259        ordered: bool,
260        streams: &'a mut StreamsState,
261        pending: &'a mut Retransmits,
262    ) -> Result<Self, ReadableError> {
263        let mut entry = match streams.recv.entry(id) {
264            Entry::Occupied(entry) => entry,
265            Entry::Vacant(_) => return Err(ReadableError::ClosedStream),
266        };
267
268        let mut recv =
269            match get_or_insert_recv(streams.stream_receive_window)(entry.get_mut()).stopped {
270                true => return Err(ReadableError::ClosedStream),
271                false => entry.remove().unwrap().into_inner(), // this can't fail due to the previous get_or_insert_with
272            };
273
274        recv.assembler.ensure_ordering(ordered)?;
275        Ok(Self {
276            id,
277            ordered,
278            streams,
279            pending,
280            state: ChunksState::Readable(recv),
281            read: 0,
282        })
283    }
284
285    /// Next
286    ///
287    /// Should call finalize() when done calling this.
288    pub fn next(&mut self, max_length: usize) -> Result<Option<Chunk>, ReadError> {
289        let rs = match self.state {
290            ChunksState::Readable(ref mut rs) => rs,
291            ChunksState::Reset(error_code) => {
292                return Err(ReadError::Reset(error_code));
293            }
294            ChunksState::Finished => {
295                return Ok(None);
296            }
297            ChunksState::Finalized => panic!("must not call next() after finalize()"),
298        };
299
300        if let Some(chunk) = rs.assembler.read(max_length, self.ordered) {
301            self.read += chunk.bytes.len() as u64;
302            return Ok(Some(chunk));
303        }
304
305        match rs.state {
306            RecvState::ResetRecvd { error_code, .. } => {
307                debug_assert_eq!(self.read, 0, "reset streams have empty buffers");
308                let state = mem::replace(&mut self.state, ChunksState::Reset(error_code));
309                // At this point if we have `rs` self.state must be `ChunksState::Readable`
310                let recv = match state {
311                    ChunksState::Readable(recv) => StreamRecv::Open(recv),
312                    _ => unreachable!("state must be ChunkState::Readable"),
313                };
314                self.streams.stream_recv_freed(self.id, recv);
315                Err(ReadError::Reset(error_code))
316            }
317            RecvState::Recv { size } => {
318                if size == Some(rs.end) && rs.assembler.bytes_read() == rs.end {
319                    let state = mem::replace(&mut self.state, ChunksState::Finished);
320                    // At this point if we have `rs` self.state must be `ChunksState::Readable`
321                    let recv = match state {
322                        ChunksState::Readable(recv) => StreamRecv::Open(recv),
323                        _ => unreachable!("state must be ChunkState::Readable"),
324                    };
325                    self.streams.stream_recv_freed(self.id, recv);
326                    Ok(None)
327                } else {
328                    // We don't need a distinct `ChunksState` variant for a blocked stream because
329                    // retrying a read harmlessly re-traces our steps back to returning
330                    // `Err(Blocked)` again. The buffers can't refill and the stream's own state
331                    // can't change so long as this `Chunks` exists.
332                    Err(ReadError::Blocked)
333                }
334            }
335        }
336    }
337
338    /// Mark the read data as consumed from the stream.
339    ///
340    /// The number of read bytes will be released from the congestion window,
341    /// allowing the remote peer to send more data if it was previously blocked.
342    ///
343    /// If [`ShouldTransmit::should_transmit()`] returns `true`,
344    /// a packet needs to be sent to the peer informing them that the stream is unblocked.
345    /// This means that you should call [`Connection::poll_transmit()`][crate::Connection::poll_transmit]
346    /// and send the returned packet as soon as is reasonable, to unblock the remote peer.
347    pub fn finalize(mut self) -> ShouldTransmit {
348        self.finalize_inner()
349    }
350
351    fn finalize_inner(&mut self) -> ShouldTransmit {
352        let state = mem::replace(&mut self.state, ChunksState::Finalized);
353        if let ChunksState::Finalized = state {
354            // Noop on repeated calls
355            return ShouldTransmit(false);
356        }
357
358        // We issue additional stream ID credit after the application is notified that a previously
359        // open stream has finished or been reset and we've therefore disposed of its state, as
360        // recorded by `stream_freed` calls in `next`.
361        let mut should_transmit = self.streams.queue_max_stream_id(self.pending);
362
363        // If the stream hasn't finished, we may need to issue stream-level flow control credit
364        if let ChunksState::Readable(mut rs) = state {
365            let (_, max_stream_data) = rs.max_stream_data(self.streams.stream_receive_window);
366            should_transmit |= max_stream_data.0;
367            if max_stream_data.0 {
368                self.pending.max_stream_data.insert(self.id);
369            }
370            // Return the stream to storage for future use
371            self.streams
372                .recv
373                .insert(self.id, Some(StreamRecv::Open(rs)));
374        }
375
376        // Issue connection-level flow control credit for any data we read regardless of state
377        let max_data = self.streams.add_read_credits(self.read);
378        self.pending.max_data |= max_data.0;
379        should_transmit |= max_data.0;
380        ShouldTransmit(should_transmit)
381    }
382}
383
384impl Drop for Chunks<'_> {
385    fn drop(&mut self) {
386        let _ = self.finalize_inner();
387    }
388}
389
390enum ChunksState {
391    Readable(Box<Recv>),
392    Reset(VarInt),
393    Finished,
394    Finalized,
395}
396
397/// Errors triggered when reading from a recv stream
398#[derive(Debug, Error, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
399pub enum ReadError {
400    /// No more data is currently available on this stream.
401    ///
402    /// If more data on this stream is received from the peer, an `Event::StreamReadable` will be
403    /// generated for this stream, indicating that retrying the read might succeed.
404    #[error("blocked")]
405    Blocked,
406    /// The peer abandoned transmitting data on this stream.
407    ///
408    /// Carries an application-defined error code.
409    #[error("reset by peer: code {0}")]
410    Reset(VarInt),
411}
412
413/// Errors triggered when opening a recv stream for reading
414#[derive(Debug, Error, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
415pub enum ReadableError {
416    /// The stream has not been opened or was already stopped, finished, or reset
417    #[error("closed stream")]
418    ClosedStream,
419    /// Attempted an ordered read following an unordered read
420    ///
421    /// Performing an unordered read allows discontinuities to arise in the receive buffer of a
422    /// stream which cannot be recovered, making further ordered reads impossible.
423    #[error("ordered read after unordered read")]
424    IllegalOrderedRead,
425}
426
427impl From<IllegalOrderedRead> for ReadableError {
428    fn from(_: IllegalOrderedRead) -> Self {
429        Self::IllegalOrderedRead
430    }
431}
432
433#[derive(Debug, Copy, Clone, Eq, PartialEq)]
434enum RecvState {
435    Recv { size: Option<u64> },
436    ResetRecvd { size: u64, error_code: VarInt },
437}
438
439impl Default for RecvState {
440    fn default() -> Self {
441        Self::Recv { size: None }
442    }
443}
444
445#[cfg(test)]
446mod tests {
447    use bytes::Bytes;
448
449    use crate::{Dir, Side};
450
451    use super::*;
452
453    #[test]
454    fn reordered_frames_while_stopped() {
455        const INITIAL_BYTES: u64 = 3;
456        const INITIAL_OFFSET: u64 = 3;
457        const RECV_WINDOW: u64 = 8;
458        let mut s = Recv::new(RECV_WINDOW);
459        let mut data_recvd = 0;
460        // Receive bytes 3..6
461        let (new_bytes, is_closed) = s
462            .ingest(
463                frame::Stream {
464                    id: StreamId::new(Side::Client, Dir::Uni, 0),
465                    offset: INITIAL_OFFSET,
466                    fin: false,
467                    data: Bytes::from_static(&[0; INITIAL_BYTES as usize]),
468                },
469                123,
470                data_recvd,
471                data_recvd + 1024,
472            )
473            .unwrap();
474        data_recvd += new_bytes;
475        assert_eq!(new_bytes, INITIAL_OFFSET + INITIAL_BYTES);
476        assert!(!is_closed);
477
478        let (credits, transmit) = s.stop().unwrap();
479        assert!(transmit.should_transmit());
480        assert_eq!(
481            credits,
482            INITIAL_OFFSET + INITIAL_BYTES,
483            "full connection flow control credit is issued by stop"
484        );
485
486        let (max_stream_data, transmit) = s.max_stream_data(RECV_WINDOW);
487        assert!(!transmit.should_transmit());
488        assert_eq!(
489            max_stream_data, RECV_WINDOW,
490            "stream flow control credit isn't issued by stop"
491        );
492
493        // Receive byte 7
494        let (new_bytes, is_closed) = s
495            .ingest(
496                frame::Stream {
497                    id: StreamId::new(Side::Client, Dir::Uni, 0),
498                    offset: RECV_WINDOW - 1,
499                    fin: false,
500                    data: Bytes::from_static(&[0; 1]),
501                },
502                123,
503                data_recvd,
504                data_recvd + 1024,
505            )
506            .unwrap();
507        data_recvd += new_bytes;
508        assert_eq!(new_bytes, RECV_WINDOW - (INITIAL_OFFSET + INITIAL_BYTES));
509        assert!(!is_closed);
510
511        let (max_stream_data, transmit) = s.max_stream_data(RECV_WINDOW);
512        assert!(!transmit.should_transmit());
513        assert_eq!(
514            max_stream_data, RECV_WINDOW,
515            "stream flow control credit isn't issued after stop"
516        );
517
518        // Receive bytes 0..3
519        let (new_bytes, is_closed) = s
520            .ingest(
521                frame::Stream {
522                    id: StreamId::new(Side::Client, Dir::Uni, 0),
523                    offset: 0,
524                    fin: false,
525                    data: Bytes::from_static(&[0; INITIAL_OFFSET as usize]),
526                },
527                123,
528                data_recvd,
529                data_recvd + 1024,
530            )
531            .unwrap();
532        assert_eq!(
533            new_bytes, 0,
534            "reordered frames don't issue connection-level flow control for stopped streams"
535        );
536        assert!(!is_closed);
537
538        let (max_stream_data, transmit) = s.max_stream_data(RECV_WINDOW);
539        assert!(!transmit.should_transmit());
540        assert_eq!(
541            max_stream_data, RECV_WINDOW,
542            "stream flow control credit isn't issued after stop"
543        );
544    }
545}
quinn_proto/connection/streams/recv.rs

quinn_proto/connection/streams/
recv.rs
