feat: more aggresive quality upgrade (#46)

* more aggressive initial bwe

* fix: incorrect keyframe request on transition

* fix: miss switching opportunity

* init depacketizer

* update keyframe filter

* save things

* add keyframe request retry

* increase keyframe debounce to 500ms

* remove unused

* rename push to forward

* ref: make it consistent with poll_slow and poll_fast

* fix: warnings

Author: lherman-cs

pulsebeam/src/controller.rs

@@ -123,7 +123,7 @@ impl ControllerActor {
             .set_rtp_mode(true)
             // .set_stats_interval(Some(Duration::from_millis(200)))
             // TODO: enable bwe
-            .enable_bwe(Some(str0m::bwe::Bitrate::kbps(300)))
+            .enable_bwe(Some(str0m::bwe::Bitrate::mbps(1)))
             // Uncomment this to see statistics
             // .set_stats_interval(Some(Duration::from_secs(1)))
             // enable for compatibility, some clients don't support remote ice-lite

pulsebeam/src/node.rs

@@ -1,5 +1,5 @@
 use crate::shard::ShardMessageSet;
-use crate::{api, controller, gateway, shard};
+use crate::{api, controller, gateway};
 use pulsebeam_runtime::actor::RunnerConfig;
 use pulsebeam_runtime::prelude::*;
 use pulsebeam_runtime::{actor, net, rand};

pulsebeam/src/participant/actor.rs

@@ -142,7 +142,7 @@ impl actor::Actor<ParticipantMessageSet> for ParticipantActor {
 
                 // Priority 4: Background tasks
                 now = stats_interval.tick() => {
-                    self.core.poll_stats(now);
+                    self.core.poll_slow(now);
                 }
                 _ = &mut rtc_timer => {
                     new_deadline = self.core.handle_timeout();

pulsebeam/src/participant/core.rs

@@ -126,9 +126,10 @@ impl ParticipantCore {
         self.update_desired_bitrate();
     }
 
-    pub fn poll_stats(&mut self, now: Instant) {
+    pub fn poll_slow(&mut self, now: Instant) {
         self.update_desired_bitrate();
-        self.upstream.poll_stats(now);
+        self.downstream.poll_slow(now);
+        self.upstream.poll_slow(now);
     }
 
     pub fn poll_rtc(&mut self) -> Option<Instant> {

pulsebeam/src/participant/downstream/audio.rs

@@ -1,23 +1,19 @@
 use futures_lite::StreamExt;
 use pulsebeam_runtime::sync::spmc;
 
-use std::pin::Pin;
+use crate::entity::TrackId;
+use crate::rtp;
+use crate::rtp::RtpPacket;
+use crate::rtp::timeline::Timeline;
+use crate::track::TrackReceiver;
 use std::sync::Arc;
 use std::task::Waker;
 use std::task::ready;
 use std::task::{Context, Poll};
 use std::time::Duration;
 use str0m::media::Mid;
 use tokio::time::Instant;
-use tokio_stream::{Stream, StreamMap};
-
-use crate::entity::TrackId;
-use crate::rtp;
-use crate::rtp::RtpPacket;
-use crate::rtp::timeline::Timeline;
-use crate::track::SimulcastReceiver;
-use crate::track::TrackReceiver;
-use pulsebeam_runtime::sync::spmc::RecvError;
+use tokio_stream::StreamMap;
 
 pub struct AudioAllocator {
     inputs: StreamMap<Arc<TrackId>, spmc::Receiver<RtpPacket>>,

pulsebeam/src/participant/downstream/mod.rs

@@ -1,7 +1,6 @@
 mod audio;
 mod video;
 
-use crate::participant::bitrate::BitrateController;
 use crate::participant::downstream::audio::AudioAllocator;
 use crate::participant::downstream::video::VideoAllocator;
 use crate::rtp::RtpPacket;
@@ -14,7 +13,7 @@ use str0m::media::{KeyframeRequest, MediaKind, Mid};
 use tokio::time::Instant;
 
 pub struct DownstreamAllocator {
-    available_bandwidth: BitrateController,
+    available_bandwidth: Bitrate,
 
     audio: AudioAllocator,
     video: VideoAllocator,
@@ -23,7 +22,7 @@ pub struct DownstreamAllocator {
 impl DownstreamAllocator {
     pub fn new() -> Self {
         Self {
-            available_bandwidth: BitrateController::default(),
+            available_bandwidth: Bitrate::mbps(1),
             audio: AudioAllocator::new(),
             video: VideoAllocator::default(),
         }
@@ -52,33 +51,35 @@ impl DownstreamAllocator {
 
     /// Handle BWE and compute both current and desired bitrate in one pass.
     pub fn update_bitrate(&mut self, available_bandwidth: Bitrate) -> (Bitrate, Bitrate) {
-        self.available_bandwidth
-            .update(available_bandwidth, Instant::now());
+        self.available_bandwidth = available_bandwidth;
         self.update_allocations()
     }
 
     pub fn update_allocations(&mut self) -> (Bitrate, Bitrate) {
-        self.video
-            .update_allocations(self.available_bandwidth.current())
+        self.video.update_allocations(self.available_bandwidth)
     }
 
     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)>> {
+    pub fn poll_slow(&mut self, now: Instant) {
+        self.video.poll_slow(now);
+    }
+
+    pub fn poll_fast(&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)
+        self.video.poll_fast(cx)
     }
 }
 
 impl Stream for DownstreamAllocator {
     type Item = (Mid, RtpPacket);
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.get_mut().poll_next_unpin(cx)
+        self.get_mut().poll_fast(cx)
     }
 }