only poll when necessary

Author: lherman-cs

pulsebeam/src/participant/actor.rs

@@ -1,11 +1,12 @@
+use std::pin::Pin;
 use std::time::Duration;
 use std::{collections::HashMap, sync::Arc};
 
 use pulsebeam_runtime::actor::ActorKind;
 use pulsebeam_runtime::prelude::*;
 use pulsebeam_runtime::{actor, mailbox, net};
 use str0m::{Rtc, RtcError, error::SdpError};
-use tokio::time::Instant;
+use tokio::time::{Instant, Sleep};
 use tokio_metrics::TaskMonitor;
 
 use crate::participant::core::{CoreEvent, ParticipantCore};
@@ -81,16 +82,19 @@ impl actor::Actor<ParticipantMessageSet> for ParticipantActor {
             .await;
         let mut stats_interval = tokio::time::interval(Duration::from_millis(200));
         let mut current_deadline = Instant::now() + Duration::from_secs(1);
+        let mut new_deadline = Some(current_deadline);
+
         let rtc_timer = tokio::time::sleep_until(current_deadline);
         tokio::pin!(rtc_timer);
 
         loop {
-            let Some(new_deadline) = self.core.poll_rtc() else {
-                break;
-            };
-            if new_deadline != current_deadline {
-                rtc_timer.as_mut().reset(new_deadline);
-                current_deadline = new_deadline;
+            match new_deadline {
+                Some(new_deadline) if new_deadline != current_deadline => {
+                    rtc_timer.as_mut().reset(new_deadline);
+                    current_deadline = new_deadline;
+                }
+                None => break,
+                _ => {}
             }
 
             let events: Vec<_> = self.core.drain_events().collect();
@@ -99,7 +103,6 @@ impl actor::Actor<ParticipantMessageSet> for ParticipantActor {
             }
 
             tokio::select! {
-                biased;
                 res = ctx.sys_rx.recv() => {
                     match res {
                         Some(msg) => match msg {
@@ -115,15 +118,22 @@ impl actor::Actor<ParticipantMessageSet> for ParticipantActor {
                 Ok(_) = self.egress.writable(), if !self.core.batcher.is_empty() => {
                     self.core.batcher.flush(&self.egress);
                 },
-                Some(batch) = gateway_rx.recv() => self.core.handle_udp_packet_batch(batch),
                 Some((meta, pkt)) = self.core.downstream.next() => {
                     self.core.handle_forward_rtp(meta, pkt);
+                    while let Some((mid, pkt)) = self.core.downstream.next().now_or_never().flatten() {
+                        self.core.handle_forward_rtp(mid, pkt);
+                    }
+
+                    new_deadline = self.core.poll_rtc();
+                },
+                Some(batch) = gateway_rx.recv() => {
+                    new_deadline = self.core.handle_udp_packet_batch(batch);
                 },
                 now = stats_interval.tick() => {
                     self.core.poll_stats(now);
                 }
                 _ = &mut rtc_timer => {
-                    self.core.handle_timeout();
+                    new_deadline = self.core.handle_timeout();
                 },
             }
         }

pulsebeam/src/participant/core.rs

@@ -67,7 +67,8 @@ impl ParticipantCore {
         self.events.drain(..)
     }
 
-    pub fn handle_udp_packet_batch(&mut self, batch: net::RecvPacketBatch) {
+    pub fn handle_udp_packet_batch(&mut self, batch: net::RecvPacketBatch) -> Option<Instant> {
+        let mut last_deadline = None;
         for pkt in batch.into_iter() {
             if let Ok(contents) = (*pkt).try_into() {
                 let recv = str0m::net::Receive {
@@ -79,15 +80,18 @@ impl ParticipantCore {
                 let _ = self
                     .rtc
                     .handle_input(Input::Receive(Instant::now().into(), recv));
-                self.poll_rtc();
+                last_deadline = self.poll_rtc();
             } else {
                 tracing::warn!(src = %batch.src, "Dropping malformed UDP packet");
             }
         }
+
+        last_deadline
     }
 
-    pub fn handle_timeout(&mut self) {
+    pub fn handle_timeout(&mut self) -> Option<Instant> {
         let _ = self.rtc.handle_input(Input::Timeout(Instant::now().into()));
+        self.poll_rtc()
     }
 
     pub fn handle_available_tracks(

pulsebeam/src/participant/downstream/mod.rs

@@ -65,18 +65,20 @@ impl DownstreamAllocator {
     pub fn handle_keyframe_request(&mut self, req: KeyframeRequest) {
         self.video.handle_keyframe_request(req);
     }
+
+    pub fn poll_next_unpin(&mut self, cx: &mut Context<'_>) -> Poll<Option<(Mid, RtpPacket)>> {
+        if let Poll::Ready(item) = self.audio.poll_next(cx) {
+            return Poll::Ready(item);
+        }
+
+        self.video.poll_next(cx)
+    }
 }
 
 impl Stream for DownstreamAllocator {
     type Item = (Mid, RtpPacket);
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let this = self.get_mut();
-
-        if let Poll::Ready(item) = this.audio.poll_next(cx) {
-            return Poll::Ready(item);
-        }
-
-        this.video.poll_next(cx)
+        self.get_mut().poll_next_unpin(cx)
     }
 }