Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/sctp-proto/src/association/stream.rs

rust627 lines
1use crate::association::Association;
2use crate::association::state::AssociationState;
3use crate::chunk::chunk_payload_data::{ChunkPayloadData, PayloadProtocolIdentifier};
4use crate::error::{Error, Result};
5use crate::queue::reassembly_queue::{Chunks, ReassemblyQueue};
6use crate::{ErrorCauseCode, Side};
7 
8use crate::util::{ByteSlice, BytesArray, BytesSource};
9use alloc::vec;
10use alloc::vec::Vec;
11use bytes::Bytes;
12use core::fmt;
13use log::{debug, error, trace};
14 
15/// Identifier for a stream within a particular association
16pub type StreamId = u16;
17 
18/// Why a stream reset did not complete successfully.
19#[non_exhaustive]
20#[derive(Debug, Copy, Clone, PartialEq, Eq)]
21pub enum StreamResetError {
22    /// The peer explicitly denied the reset request.
23    Denied,
24    /// The reset failed because of a protocol error or exhausted retransmissions.
25    Failed,
26}
27 
28/// Application events about streams
29#[non_exhaustive]
30#[derive(Debug, PartialEq, Eq)]
31pub enum StreamEvent {
32    /// One or more new streams has been opened
33    Opened {
34        /// Which stream was opened
35        id: StreamId,
36    },
37    /// A currently open stream has data or errors waiting to be read
38    Readable {
39        /// Which stream is now readable
40        id: StreamId,
41    },
42    /// A formerly write-blocked stream might be ready for a write or have been stopped
43    ///
44    /// Only generated for streams that are currently open.
45    Writable {
46        /// Which stream is now writable
47        id: StreamId,
48    },
49    /// The stream was torn down by a reset, an inbound reset
50    /// request from the peer naming the id was processed or
51    /// its reciprocal to a locally-initiated one.
52    ///
53    /// No more data can be read from or written to it.
54    ///
55    /// Once this fires, either [`StreamEvent::ResetComplete`] or
56    /// [`StreamEvent::ResetFailed`] for the same id is guaranteed to eventually
57    /// follow, unless the association closes first.
58    Finished {
59        /// Which stream has been finished
60        id: StreamId,
61    },
62    /// A reset handshake involving this stream id completed successfully.
63    ///
64    /// The id can be reused unless a newer reset for the same id has since
65    /// started or failed.
66    ResetComplete {
67        /// Which stream id completed its reset.
68        id: StreamId,
69    },
70    /// A reset handshake involving this stream id did not complete.
71    ///
72    /// The id remains unavailable for reuse until a later reset succeeds or
73    /// the association closes.
74    ResetFailed {
75        /// Which stream id failed to reset.
76        id: StreamId,
77        /// Why the reset did not complete.
78        reason: StreamResetError,
79    },
80    /// The peer asked us to stop sending on an outgoing stream
81    Stopped {
82        /// Which stream has been stopped
83        id: StreamId,
84        /// Error code supplied by the peer
85        error_code: ErrorCauseCode,
86    },
87    /// At least one new stream of a certain directionality may be opened
88    Available,
89    /// The number of bytes of outgoing data buffered is lower than the threshold.
90    BufferedAmountLow {
91        /// Which stream is now readable
92        id: StreamId,
93    },
94    /// The number of bytes of outgoing data buffered is higher than the threshold.
95    BufferedAmountHigh {
96        /// Which stream has crossed the high threshold
97        id: StreamId,
98    },
99}
100 
101/// Reliability type for stream
102#[derive(Debug, Copy, Clone, PartialEq, Default)]
103pub enum ReliabilityType {
104    /// ReliabilityTypeReliable is used for reliable transmission
105    #[default]
106    Reliable = 0,
107    /// ReliabilityTypeRexmit is used for partial reliability by retransmission count
108    Rexmit = 1,
109    /// ReliabilityTypeTimed is used for partial reliability by retransmission duration
110    Timed = 2,
111}
112 
113impl fmt::Display for ReliabilityType {
114    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
115        let s = match *self {
116            ReliabilityType::Reliable => "Reliable",
117            ReliabilityType::Rexmit => "Rexmit",
118            ReliabilityType::Timed => "Timed",
119        };
120        write!(f, "{}", s)
121    }
122}
123 
124impl From<u8> for ReliabilityType {
125    fn from(v: u8) -> ReliabilityType {
126        match v {
127            1 => ReliabilityType::Rexmit,
128            2 => ReliabilityType::Timed,
129            _ => ReliabilityType::Reliable,
130        }
131    }
132}
133 
134/// Stream represents an SCTP stream
135pub struct Stream<'a> {
136    pub(crate) stream_identifier: StreamId,
137    pub(crate) association: &'a mut Association,
138}
139 
140impl<'a> Stream<'a> {
141    /// read reads a packet of len(p) bytes, dropping the Payload Protocol Identifier.
142    /// Returns EOF when the stream is reset or an error if the stream is closed
143    /// otherwise.
144    pub fn read(&mut self) -> Result<Option<Chunks>> {
145        self.read_sctp()
146    }
147 
148    /// read_sctp reads a packet of len(p) bytes and returns the associated Payload
149    /// Protocol Identifier.
150    /// Returns EOF when the stream is reset or an error if the stream is closed
151    /// otherwise.
152    pub fn read_sctp(&mut self) -> Result<Option<Chunks>> {
153        let (message, drained) =
154            if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
155                if s.state != RecvSendState::ReadWritable && s.state != RecvSendState::Readable {
156                    return Err(Error::ErrStreamClosed);
157                }
158 
159                let message = s.reassembly_queue.read();
160                let drained = message.is_some() && s.reassembly_queue.get_num_bytes() == 0;
161                (message, drained)
162            } else {
163                return Err(Error::ErrStreamClosed);
164            };
165 
166        if drained {
167            self.association
168                .finish_retiring_stream(self.stream_identifier)?;
169        }
170 
171        Ok(message)
172    }
173 
174    /// write_sctp writes len(p) bytes from p to the DTLS connection
175    pub fn write_sctp(&mut self, p: &Bytes, ppi: PayloadProtocolIdentifier) -> Result<usize> {
176        self.write_source(&mut ByteSlice::from_slice(p), ppi)
177    }
178 
179    /// Send data on the given stream.
180    ///
181    /// Uses the deafult payload protocol (PPI).
182    ///
183    /// Returns the number of bytes successfully written.
184    pub fn write(&mut self, data: &[u8]) -> Result<usize> {
185        self.write_with_ppi(data, self.get_default_payload_type()?)
186    }
187 
188    /// Send data on the given stream, with a specific payload protocol.
189    ///
190    /// Returns the number of bytes successfully written.
191    pub fn write_with_ppi(&mut self, data: &[u8], ppi: PayloadProtocolIdentifier) -> Result<usize> {
192        self.write_source(&mut ByteSlice::from_slice(data), ppi)
193    }
194 
195    /// write writes len(p) bytes from p with the default Payload Protocol Identifier
196    pub fn write_chunk(&mut self, p: &Bytes) -> Result<usize> {
197        self.write_source(
198            &mut ByteSlice::from_slice(p),
199            self.get_default_payload_type()?,
200        )
201    }
202 
203    /// Send data on the given stream
204    ///
205    /// Returns the number of bytes and chunks successfully written.
206    /// Note that this method might also write a partial chunk. In this case
207    /// it will not count this chunk as fully written. However
208    /// the chunk will be advanced and contain only non-written data after the call.
209    pub fn write_chunks(&mut self, data: &mut [Bytes]) -> Result<usize> {
210        self.write_source(
211            &mut BytesArray::from_chunks(data),
212            self.get_default_payload_type()?,
213        )
214    }
215 
216    /// write_source writes BytesSource to the DTLS connection
217    fn write_source<B: BytesSource>(
218        &mut self,
219        source: &mut B,
220        ppi: PayloadProtocolIdentifier,
221    ) -> Result<usize> {
222        if !self.is_writable() {
223            return Err(Error::ErrStreamClosed);
224        }
225 
226        if source.remaining() > self.association.max_send_message_size() as usize {
227            return Err(Error::ErrOutboundPacketTooLarge);
228        }
229 
230        let state: AssociationState = self.association.state();
231        match state {
232            AssociationState::ShutdownSent
233            | AssociationState::ShutdownAckSent
234            | AssociationState::ShutdownPending
235            | AssociationState::ShutdownReceived => return Err(Error::ErrStreamClosed),
236            _ => {}
237        };
238 
239        let (p, _) = source.pop_chunk(self.association.max_send_message_size() as usize);
240 
241        if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
242            let (is_buffered_amount_high, chunks) = s.packetize(&p, ppi);
243            self.association.send_payload_data(chunks)?;
244 
245            if is_buffered_amount_high {
246                trace!("StreamEvent::BufferedAmountHigh");
247                self.association
248                    .events
249                    .push_back(crate::association::Event::Stream(
250                        StreamEvent::BufferedAmountHigh {
251                            id: self.stream_identifier,
252                        },
253                    ));
254            }
255 
256            Ok(p.len())
257        } else {
258            Err(Error::ErrStreamClosed)
259        }
260    }
261 
262    pub fn is_readable(&self) -> bool {
263        if let Some(s) = self.association.streams.get(&self.stream_identifier) {
264            s.state == RecvSendState::Readable || s.state == RecvSendState::ReadWritable
265        } else {
266            false
267        }
268    }
269 
270    pub fn is_writable(&self) -> bool {
271        // RFC 6525 section 5.1.2 A1 forbids assigning new SSNs while an
272        // Outgoing SSN Reset Request for this stream is pending. A failed
273        // reset also quarantines the outgoing direction because the peer may
274        // already have reset its corresponding incoming stream.
275        if self
276            .association
277            .stream_reset_blocked(self.stream_identifier)
278        {
279            return false;
280        }
281        if let Some(s) = self.association.streams.get(&self.stream_identifier) {
282            s.state == RecvSendState::Writable || s.state == RecvSendState::ReadWritable
283        } else {
284            false
285        }
286    }
287 
288    /// stop closes the read-direction of the stream.
289    /// Future calls to read are not permitted after calling stop.
290    ///    
291    /// NOTE: a stream closed without a queued reset never produces
292    /// `StreamEvent::Finished` followed by `StreamEvent::ResetComplete` or
293    /// `StreamEvent::ResetFailed`.
294    pub fn stop(&mut self) -> Result<()> {
295        let retiring = self
296            .association
297            .retiring_streams
298            .contains_key(&self.stream_identifier);
299        let reset = self
300            .association
301            .streams
302            .get(&self.stream_identifier)
303            .is_some_and(|s| {
304                s.state == RecvSendState::Readable || s.state == RecvSendState::ReadWritable
305            });
306 
307        if reset
308            && !retiring
309            && !self
310                .association
311                .stream_reset_in_progress(self.stream_identifier)
312        {
313            // Reset the outgoing stream
314            // https://tools.ietf.org/html/rfc6525
315            //
316            // Queued before clearing the read bit below, send_reset_request
317            // is the only fallible step, and failing after the state
318            // mutation would leave the stream half-closed with the reset
319            // silently dropped and unretryable.
320            self.association
321                .send_reset_request(self.stream_identifier)?;
322        }
323 
324        if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
325            s.state = ((s.state as u8) & 0x2).into();
326        }
327 
328        // stop() promises that future reads are not permitted. If a reset was
329        // preserving unread boundary DATA, discard that old generation now so
330        // its Finished event and unrelated association events cannot stall.
331        if retiring {
332            self.association
333                .discard_retiring_streams(self.stream_identifier);
334        }
335 
336        Ok(())
337    }
338 
339    /// finish closes the write-direction of the stream.
340    /// Future calls to write are not permitted after calling Close.
341    pub fn finish(&mut self) -> Result<()> {
342        if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
343            s.state = ((s.state as u8) & 0x1).into();
344        }
345        Ok(())
346    }
347 
348    /// close shuts down both the read and write halves of this stream.
349    ///
350    /// This is a convenience method that calls `finish()` followed by `stop()`.
351    /// Resets the stream when both halves are shutdown.
352    ///
353    /// The single failure mode is `stop()`'s, if the association is not
354    /// established, no reset was queued, retrying `close()` will
355    /// attempt the reset again.
356    pub fn close(&mut self) -> Result<()> {
357        self.finish()?;
358        self.stop()
359    }
360 
361    /// stream_identifier returns the Stream identifier associated to the stream.
362    pub fn stream_identifier(&self) -> StreamId {
363        self.stream_identifier
364    }
365 
366    /// set_default_payload_type sets the default payload type used by write.
367    pub fn set_default_payload_type(
368        &mut self,
369        default_payload_type: PayloadProtocolIdentifier,
370    ) -> Result<()> {
371        if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
372            s.default_payload_type = default_payload_type;
373            Ok(())
374        } else {
375            Err(Error::ErrStreamClosed)
376        }
377    }
378 
379    /// get_default_payload_type returns the payload type associated to the stream.
380    pub fn get_default_payload_type(&self) -> Result<PayloadProtocolIdentifier> {
381        if let Some(s) = self.association.streams.get(&self.stream_identifier) {
382            Ok(s.default_payload_type)
383        } else {
384            Err(Error::ErrStreamClosed)
385        }
386    }
387 
388    /// set_reliability_params sets reliability parameters for this stream.
389    pub fn set_reliability_params(
390        &mut self,
391        unordered: bool,
392        rel_type: ReliabilityType,
393        rel_val: u32,
394    ) -> Result<()> {
395        if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
396            debug!(
397                "[{}] reliability params: ordered={} type={} value={}",
398                s.side, !unordered, rel_type, rel_val
399            );
400            s.unordered = unordered;
401            s.reliability_type = rel_type;
402            s.reliability_value = rel_val;
403            Ok(())
404        } else {
405            Err(Error::ErrStreamClosed)
406        }
407    }
408 
409    /// buffered_amount returns the number of bytes of data currently queued to be sent over this stream.
410    pub fn buffered_amount(&self) -> Result<usize> {
411        if let Some(s) = self.association.streams.get(&self.stream_identifier) {
412            Ok(s.buffered_amount)
413        } else {
414            Err(Error::ErrStreamClosed)
415        }
416    }
417 
418    /// buffered_amount_low_threshold returns the number of bytes of buffered outgoing data that is
419    /// considered "low." Defaults to 0.
420    pub fn buffered_amount_low_threshold(&self) -> Result<usize> {
421        if let Some(s) = self.association.streams.get(&self.stream_identifier) {
422            Ok(s.buffered_amount_low)
423        } else {
424            Err(Error::ErrStreamClosed)
425        }
426    }
427 
428    /// set_buffered_amount_low_threshold is used to update the threshold.
429    /// See buffered_amount_low_threshold().
430    pub fn set_buffered_amount_low_threshold(&mut self, th: usize) -> Result<()> {
431        if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
432            s.buffered_amount_low = th;
433            Ok(())
434        } else {
435            Err(Error::ErrStreamClosed)
436        }
437    }
438 
439    /// buffered_amount_high_threshold returns the number of bytes of buffered outgoing data that is
440    /// considered "high." Defaults to usize::MAX (effectively disabled).
441    pub fn buffered_amount_high_threshold(&self) -> Result<usize> {
442        if let Some(s) = self.association.streams.get(&self.stream_identifier) {
443            Ok(s.buffered_amount_high)
444        } else {
445            Err(Error::ErrStreamClosed)
446        }
447    }
448 
449    /// set_buffered_amount_high_threshold is used to update the threshold.
450    /// See buffered_amount_high_threshold().
451    pub fn set_buffered_amount_high_threshold(&mut self, th: usize) -> Result<()> {
452        if let Some(s) = self.association.streams.get_mut(&self.stream_identifier) {
453            s.buffered_amount_high = th;
454            Ok(())
455        } else {
456            Err(Error::ErrStreamClosed)
457        }
458    }
459}
460 
461#[derive(Debug, Copy, Clone, Eq, PartialEq, Default)]
462pub enum RecvSendState {
463    #[default]
464    Closed = 0,
465    Readable = 1,
466    Writable = 2,
467    ReadWritable = 3,
468}
469 
470impl From<u8> for RecvSendState {
471    fn from(v: u8) -> Self {
472        match v {
473            1 => RecvSendState::Readable,
474            2 => RecvSendState::Writable,
475            3 => RecvSendState::ReadWritable,
476            _ => RecvSendState::Closed,
477        }
478    }
479}
480 
481/// StreamState represents the state of an SCTP stream
482#[derive(Default, Debug)]
483pub struct StreamState {
484    pub(crate) side: Side,
485    pub(crate) max_payload_size: u32,
486    pub(crate) stream_identifier: StreamId,
487    pub(crate) default_payload_type: PayloadProtocolIdentifier,
488    pub(crate) reassembly_queue: ReassemblyQueue,
489    pub(crate) sequence_number: u16,
490    pub(crate) state: RecvSendState,
491    pub(crate) unordered: bool,
492    pub(crate) reliability_type: ReliabilityType,
493    pub(crate) reliability_value: u32,
494    pub(crate) buffered_amount: usize,
495    pub(crate) buffered_amount_low: usize,
496    pub(crate) buffered_amount_high: usize,
497}
498impl StreamState {
499    pub(crate) fn new(
500        side: Side,
501        stream_identifier: StreamId,
502        max_payload_size: u32,
503        max_receive_message_size: u32,
504        default_payload_type: PayloadProtocolIdentifier,
505    ) -> Self {
506        StreamState {
507            side,
508            stream_identifier,
509            max_payload_size,
510            default_payload_type,
511            reassembly_queue: ReassemblyQueue::new(stream_identifier, max_receive_message_size),
512            sequence_number: 0,
513            state: RecvSendState::ReadWritable,
514            unordered: false,
515            reliability_type: ReliabilityType::Reliable,
516            reliability_value: 0,
517            buffered_amount: 0,
518            buffered_amount_low: 0,
519            buffered_amount_high: usize::MAX,
520        }
521    }
522 
523    pub(crate) fn handle_data(&mut self, pd: &ChunkPayloadData) -> Result<bool> {
524        self.reassembly_queue.push(pd.clone())
525    }
526 
527    fn packetize(
528        &mut self,
529        raw: &Bytes,
530        ppi: PayloadProtocolIdentifier,
531    ) -> (bool, Vec<ChunkPayloadData>) {
532        let mut i = 0;
533        let mut remaining = raw.len();
534 
535        // From draft-ietf-rtcweb-data-protocol-09, section 6:
536        //   All Data Channel Establishment Protocol messages MUST be sent using
537        //   ordered delivery and reliable transmission.
538        let unordered = ppi != PayloadProtocolIdentifier::Dcep && self.unordered;
539 
540        let mut chunks = vec![];
541 
542        let head_abandoned = false;
543        let head_all_inflight = false;
544        while remaining != 0 {
545            // self.association.max_payload_size
546            let fragment_size = core::cmp::min(self.max_payload_size as usize, remaining);
547 
548            // Copy the userdata since we'll have to store it until acked
549            // and the caller may re-use the buffer in the mean time
550            let user_data = raw.slice(i..i + fragment_size);
551 
552            let chunk = ChunkPayloadData {
553                stream_identifier: self.stream_identifier,
554                user_data,
555                unordered,
556                beginning_fragment: i == 0,
557                ending_fragment: remaining - fragment_size == 0,
558                immediate_sack: false,
559                payload_type: ppi,
560                stream_sequence_number: self.sequence_number,
561                abandoned: head_abandoned, // all fragmented chunks use the same abandoned
562                all_inflight: head_all_inflight, // all fragmented chunks use the same all_inflight
563                ..Default::default()
564            };
565 
566            chunks.push(chunk);
567 
568            remaining -= fragment_size;
569            i += fragment_size;
570        }
571 
572        // RFC 4960 Sec 6.6
573        // Note: When transmitting ordered and unordered data, an endpoint does
574        // not increment its Stream Sequence Number when transmitting a DATA
575        // chunk with U flag set to 1.
576        if !unordered {
577            self.sequence_number = self.sequence_number.wrapping_add(1);
578        }
579 
580        let old_amount = self.buffered_amount;
581        self.buffered_amount += raw.len();
582        let new_amount = self.buffered_amount;
583 
584        // Check if we crossed the high threshold
585        let is_buffered_amount_high =
586            old_amount < self.buffered_amount_high && new_amount >= self.buffered_amount_high;
587 
588        (is_buffered_amount_high, chunks)
589    }
590 
591    /// This method is called by association's read_loop (go-)routine to notify this stream
592    /// of the specified amount of outgoing data has been delivered to the peer.
593    pub(crate) fn on_buffer_released(&mut self, n_bytes_released: i64) -> bool {
594        if n_bytes_released <= 0 {
595            return false;
596        }
597 
598        let from_amount = self.buffered_amount;
599        let new_amount = if from_amount < n_bytes_released as usize {
600            self.buffered_amount = 0;
601            error!(
602                "[{}] released buffer size {} should be <= {}",
603                self.side, n_bytes_released, 0,
604            );
605            0
606        } else {
607            self.buffered_amount -= n_bytes_released as usize;
608 
609            from_amount - n_bytes_released as usize
610        };
611 
612        let buffered_amount_low = self.buffered_amount_low;
613 
614        trace!(
615            "[{}] bufferedAmount = {}, from_amount = {}, buffered_amount_low = {}",
616            self.side, new_amount, from_amount, buffered_amount_low,
617        );
618 
619        from_amount > buffered_amount_low && new_amount <= buffered_amount_low
620    }
621 
622    pub(crate) fn get_num_bytes_in_reassembly_queue(&self) -> usize {
623        // No lock is required as it reads the size with atomic load function.
624        self.reassembly_queue.get_num_bytes()
625    }
626}