correct gro receive

Author: kbalt

media/rtc/src/rtp_session/mod.rs

@@ -165,6 +165,19 @@ impl RtpSession {
         self.rx.remove(&ssrc);
     }
 
+    /// Returns an iterator over all outbound streams
+    pub fn tx_streams(&self) -> impl Iterator<Item = &RtpOutboundStream> {
+        self.tx.values()
+    }
+
+    /// Returns an iterator over all inbound streams (excluding rtx SSRCs as they're included in the original RtpInboundStream)
+    pub fn rx_streams(&self) -> impl Iterator<Item = &RtpInboundStream> {
+        self.rx.values().filter_map(|rx| match rx {
+            RxStream::Original(rx) => Some(rx),
+            RxStream::Rtx(..) => None,
+        })
+    }
+
     /// Hand of the RTCP packet to the RTP session
     #[must_use]
     pub fn receive_rtcp(

media/rtc/src/rtp_session/outbound/mod.rs

@@ -35,6 +35,8 @@ impl RtpOutboundStream {
             stats: RtpOutboundStats {
                 bytes_sent: 0,
                 packets_sent: 0,
+                rtx_packets_sent: 0,
+                rtx_bytes_sent: 0,
                 remote: None,
             },
             report_interval,
@@ -126,7 +128,10 @@ impl RtpOutboundStream {
     pub(crate) fn poll(&mut self, now: Instant) -> Option<RtpOutboundStreamEvent> {
         match self.queue.poll(now)? {
             RtpOutboundStreamEvent::SendRtpPacket { rtp_packet, is_rtx } => {
-                if !is_rtx {
+                if is_rtx {
+                    self.stats.rtx_packets_sent += 1;
+                    self.stats.rtx_bytes_sent += rtp_packet.payload.len() as u64;
+                } else {
                     self.stats.packets_sent += 1;
                     self.stats.bytes_sent += rtp_packet.payload.len() as u64;
                 }

media/rtc/src/rtp_session/outbound/queue.rs

@@ -52,7 +52,7 @@ struct Rtx {
     pt: u8,
 
     /// Queue of already sent RTP packets
-    sent_packets: VecDeque<(RtpPacket, Instant)>,
+    sent_packets: VecDeque<(RtpPacket, Instant, u32)>,
 
     /// Determines how long a RTP packet is stored for retransmission before being dropped
     sent_packets_max_size: Duration,
@@ -190,10 +190,10 @@ impl OutboundQueue {
 
         // Store packet in sent_packets if configured
         if let Some(rtx) = &mut self.rtx {
-            rtx.sent_packets.push_back((rtp_packet.clone(), now));
+            rtx.sent_packets.push_back((rtp_packet.clone(), now, 0));
 
             // Remove old packets
-            while let Some((_, sent_at)) = rtx.sent_packets.front()
+            while let Some((_, sent_at, _)) = rtx.sent_packets.front()
                 && now.saturating_duration_since(*sent_at) > rtx.sent_packets_max_size
             {
                 rtx.sent_packets.pop_front();
@@ -223,12 +223,16 @@ impl OutboundQueue {
         };
 
         for entry in entries {
-            if let Some((rtp_packet, _)) = rtx
+            if let Some((rtp_packet, _, sent_counter)) = rtx
                 .sent_packets
-                .iter()
+                .iter_mut()
                 .find(|x| x.0.sequence_number.0 == entry)
             {
-                rtx.retransmit_queue.push_back(rtp_packet.clone());
+                // Limit the same RTP packet to being sent 5 times
+                if *sent_counter < 5 {
+                    *sent_counter += 1;
+                    rtx.retransmit_queue.push_back(rtp_packet.clone());
+                }
             }
         }
     }

media/rtc/src/rtp_session/outbound/stats.rs

@@ -9,6 +9,12 @@ pub struct RtpOutboundStats {
     /// Amount of RTP packets sent
     pub packets_sent: u64,
 
+    /// Amount of RTX packets sent
+    pub rtx_packets_sent: u64,
+
+    /// Amount of RTX bytes sent
+    pub rtx_bytes_sent: u64,
+
     /// Stats that are dependent on the remote sending a receiver report block
     pub remote: Option<RtpOutboundRemoteStats>,
 }

media/rtc/src/sdp/mod.rs

@@ -2020,6 +2020,13 @@ impl SdpSession {
         Some(InboundMedia { media, stream })
     }
 
+    /// Returns an iterator over all internal RTP transports and their corresponding transport id
+    pub fn rtp_sessions(&self) -> impl Iterator<Item = (TransportId, &RtpSession)> {
+        self.transports
+            .values()
+            .map(|x| (x.public_id, &x.rtp_session))
+    }
+
     /// Returns the cumulative gathering state of all ice agents
     pub fn ice_gathering_state(&self) -> Option<IceGatheringState> {
         self.transports

media/rtc/src/tokio/mod.rs

@@ -159,17 +159,22 @@ impl TokioIoState {
 
                 for i in 0..num_msg {
                     let len = self.meta[i].len;
-
-                    let pkt = ReceivedPkt {
-                        data: self.bufs[i][..len].to_vec(),
-                        source: self.meta[i].addr,
-                        destination: self.meta[i].dst_ip.map_or(socket.local_addr(), |ip| {
-                            (ip, socket.local_addr().port()).into()
-                        }),
-                        component: *component,
-                    };
-
-                    session.receive(now, *transport_id, pkt);
+                    let stride = self.meta[i].stride;
+
+                    let packet = &self.bufs[i][..len];
+
+                    for packet in packet.chunks(stride) {
+                        let pkt = ReceivedPkt {
+                            data: packet.to_vec(),
+                            source: self.meta[i].addr,
+                            destination: self.meta[i].dst_ip.map_or(socket.local_addr(), |ip| {
+                                (ip, socket.local_addr().port()).into()
+                            }),
+                            component: *component,
+                        };
+
+                        session.receive(now, *transport_id, pkt);
+                    }
 
                     received = true;
                 }