From d992ac1f469714a2dd8f2e0c90914e407b5c3451 Mon Sep 17 00:00:00 2001 From: euzu Date: Tue, 10 Jun 2025 11:54:44 +0200 Subject: [PATCH 1/3] Refactoring shared streams for UHD streams --- Cargo.lock | 87 ++++++++++--------- Cargo.toml | 1 + src/api/api_utils.rs | 3 +- .../model/streams/shared_stream_manager.rs | 87 ++++++++++++------- 4 files changed, 102 insertions(+), 76 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index edc76eff9..b2a779353 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13,9 +13,9 @@ dependencies = [ [[package]] name = "adler2" -version = "2.0.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "512761e0bb2578dd7380c6baaa0f4ce03e84f95e960231d1dec8bf4d7d6e2627" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" [[package]] name = "aho-corasick" @@ -143,9 +143,9 @@ dependencies = [ [[package]] name = "async-compression" -version = "0.4.23" +version = "0.4.24" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b37fc50485c4f3f736a4fb14199f6d5f5ba008d7f28fe710306c92780f004c07" +checksum = "d615619615a650c571269c00dca41db04b9210037fa76ed8239f70404ab56985" dependencies = [ "brotli", "flate2", @@ -379,9 +379,9 @@ dependencies = [ [[package]] name = "cfg-if" -version = "1.0.0" +version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +checksum = "9555578bc9e57714c812a1f84e4fc5b4d21fcb063490c624de019f7464c91268" [[package]] name = "cfg_aliases" @@ -405,9 +405,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.39" +version = "4.5.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd60e63e9be68e5fb56422e397cf9baddded06dae1d2e523401542383bc72a9f" +checksum = "40b6887a1d8685cebccf115538db5c0efe625ccac9696ad45c409d96566e910f" dependencies = [ "clap_builder", "clap_derive", @@ -415,9 +415,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.5.39" +version = "4.5.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89cc6392a1f72bbeb820d71f32108f61fdaf18bc526e1d23954168a67759ef51" +checksum = "e0c66c08ce9f0c698cbce5c0279d0bb6ac936d8674174fe48f736533b964f59e" dependencies = [ "anstream", "anstyle", @@ -427,9 +427,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.5.32" +version = "4.5.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09176aae279615badda0765c0c0b3f6ed53f4709118af73cf4655d85d1530cd7" +checksum = "d2c7947ae4cc3d851207c1adb5b5e260ff0cca11446b1d6d1423788e442257ce" dependencies = [ "heck", "proc-macro2", @@ -439,9 +439,9 @@ dependencies = [ [[package]] name = "clap_lex" -version = "0.7.4" +version = "0.7.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f46ad14479a25103f283c0f10005961cf086d8dc42205bb44c46ac563475dca6" +checksum = "b94f61472cee1439c0b966b47e3aca9ae07e45d070759512cd390ea2bebc6675" [[package]] name = "colorchoice" @@ -546,9 +546,9 @@ dependencies = [ [[package]] name = "curl" -version = "0.4.47" +version = "0.4.48" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9fb4d13a1be2b58f14d60adba57c9834b78c62fd86c3e76a148f732686e9265" +checksum = "9e2d5c8f48d9c0c23250e52b55e82a6ab4fdba6650c931f5a0a57a43abda812b" dependencies = [ "curl-sys", "libc", @@ -556,14 +556,14 @@ dependencies = [ "openssl-sys", "schannel", "socket2", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] name = "curl-sys" -version = "0.4.80+curl-8.12.1" +version = "0.4.82+curl-8.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "55f7df2eac63200c3ab25bde3b2268ef2ee56af3d238e76d61f01c3c49bff734" +checksum = "c4d63638b5ec65f1a4ae945287b3fd035be4554bbaf211901159c9a2a74fb5be" dependencies = [ "cc", "libc", @@ -572,7 +572,7 @@ dependencies = [ "openssl-sys", "pkg-config", "vcpkg", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -809,9 +809,9 @@ dependencies = [ [[package]] name = "flate2" -version = "1.1.1" +version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ced92e76e966ca2fd84c8f7aa01a4aea65b0eb6648d72f7c8f3e2764a67fece" +checksum = "4a3d7db9596fecd151c5f638c0ee5d5bd487b6e0ea232e5dc96d5250f6f94b1d" dependencies = [ "crc32fast", "miniz_oxide", @@ -1069,9 +1069,9 @@ checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" [[package]] name = "hashbrown" -version = "0.15.3" +version = "0.15.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "84b26c544d002229e640969970a2e74021aadf6e2f96372b9c58eff97de08eb3" +checksum = "5971ac85611da7067dbfcabef3c70ebb5606018acd9e2a3903a0da507521e0d5" [[package]] name = "heck" @@ -1165,9 +1165,9 @@ dependencies = [ [[package]] name = "hyper-rustls" -version = "0.27.6" +version = "0.27.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "03a01595e11bdcec50946522c32dde3fc6914743000a68b93000965f2f02406d" +checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58" dependencies = [ "http 1.3.1", "hyper", @@ -1366,7 +1366,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cea70ddb795996207ad57735b50c5982d8844f38ba9ee5f1aedcfb708a2aa11e" dependencies = [ "equivalent", - "hashbrown 0.15.3", + "hashbrown 0.15.4", ] [[package]] @@ -1649,9 +1649,9 @@ dependencies = [ [[package]] name = "miniz_oxide" -version = "0.8.8" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3be647b768db090acb35d5ec5db2b0e1f1de11133ca123b9eacf5137868f892a" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" dependencies = [ "adler2", ] @@ -2013,9 +2013,9 @@ dependencies = [ [[package]] name = "portable-atomic" -version = "1.11.0" +version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "350e9b48cbc6b0e028b0473b114454c6316e57336ee184ceab6e53f72c178b3e" +checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483" [[package]] name = "portable-atomic-util" @@ -2362,9 +2362,9 @@ dependencies = [ [[package]] name = "rustc-demangle" -version = "0.1.24" +version = "0.1.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f" +checksum = "989e6739f80c4ad5b13e0fd7fe89531180375b18520cc8c82080e4dc4035b84f" [[package]] name = "rustc-hash" @@ -2619,9 +2619,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.15.0" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8917285742e9f3e1683f0a9c4e6b57960b7314d0b08d30d1ecd426713ee2eee9" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" [[package]] name = "socket2" @@ -2662,9 +2662,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" [[package]] name = "syn" -version = "2.0.101" +version = "2.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ce2b7fc941b3a24138a0a7cf8e858bfc6a992e7978a068a5c760deb0ed43caf" +checksum = "f6397daf94fa90f058bd0fd88429dd9e5738999cca8d701813c80723add80462" dependencies = [ "proc-macro2", "quote", @@ -2984,9 +2984,9 @@ dependencies = [ [[package]] name = "tracing-attributes" -version = "0.1.28" +version = "0.1.29" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "395ae124c09f9e6918a2310af6038fba074bcf474ac352496d5910dd59a2226d" +checksum = "1b1ffbcf9c6f6b99d386e7444eb608ba646ae452a36b39737deb9663b610f662" dependencies = [ "proc-macro2", "quote", @@ -2995,9 +2995,9 @@ dependencies = [ [[package]] name = "tracing-core" -version = "0.1.33" +version = "0.1.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c" +checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678" dependencies = [ "once_cell", ] @@ -3032,6 +3032,7 @@ dependencies = [ "chrono", "clap", "cron", + "dashmap", "deunicode", "enum-iterator", "env_logger", @@ -3078,9 +3079,9 @@ dependencies = [ [[package]] name = "twox-hash" -version = "2.1.0" +version = "2.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7b17f197b3050ba473acf9181f7b1d3b66d1cf7356c6cc57886662276e65908" +checksum = "8b907da542cbced5261bd3256de1b3a1bf340a3d37f93425a07362a1d687de56" [[package]] name = "typenum" diff --git a/Cargo.toml b/Cargo.toml index 014857eb0..20345b848 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,6 +63,7 @@ bitflags = "2" indexmap = "2" notify = "8" arc-swap = "1" +dashmap = "6" #[cfg(target_os = "macos")] libc = "0" #[cfg(target_os = "windows")] diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index 6c8f306c3..2573e7720 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -631,8 +631,7 @@ pub async fn stream_response(app_state: &AppState, debug_if_enabled!("Streaming shared stream request from {}", sanitize_sensitive_info(stream_url)); // Shared Stream response let shared_headers = provider_response.as_ref().map_or_else(Vec::new, |(h, _)| h.clone()); - SharedStreamManager::subscribe(app_state, stream_url, stream, shared_headers, stream_options.buffer_size).await; - if let Some(broadcast_stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url).await { + if let Some(broadcast_stream) = SharedStreamManager::subscribe(app_state, stream_url, stream, shared_headers, stream_options.buffer_size).await { let (status_code, header_map) = get_stream_response_with_headers(provider_response); let mut response = axum::response::Response::builder() .status(status_code); diff --git a/src/api/model/streams/shared_stream_manager.rs b/src/api/model/streams/shared_stream_manager.rs index ba3df51f4..7cc53f15d 100644 --- a/src/api/model/streams/shared_stream_manager.rs +++ b/src/api/model/streams/shared_stream_manager.rs @@ -4,7 +4,7 @@ use crate::api::model::stream_error::StreamError; use crate::utils::debug_if_enabled; use crate::utils::request::sanitize_sensitive_info; use bytes::Bytes; -use futures::stream::BoxStream; +use futures::stream::{BoxStream, FuturesUnordered}; use futures::{Stream, StreamExt}; use std::collections::HashMap; use std::sync::{Arc}; @@ -12,8 +12,9 @@ use tokio::sync::RwLock; use tokio::sync::mpsc::{Sender}; use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::task::{Context, Poll}; -use std::time::{Duration, Instant}; +use dashmap::DashMap; use log::{trace}; use tokio::sync::{mpsc}; use tokio::sync::mpsc::error::TrySendError; @@ -42,18 +43,11 @@ where } } - -// impl Drop for ReceiverStreamWrapper -// { -// fn drop(&mut self) { -// println!("receiver_dropped"); -// } -// } - fn convert_stream(stream: BoxStream) -> BoxStream> { Box::pin(ReceiverStreamWrapper { stream }.boxed()) } +type SubscriberId = usize; /// Represents the state of a shared provider URL. /// @@ -61,7 +55,8 @@ fn convert_stream(stream: BoxStream) -> BoxStream, buf_size: usize, - subscribers: Arc>>>, + subscribers: Arc>>, //Arc>>>, + next_subscriber_id: AtomicUsize, } impl SharedStreamState { @@ -70,13 +65,15 @@ impl SharedStreamState { Self { headers, buf_size, - subscribers: Arc::new(RwLock::new(Vec::new())), + subscribers: Arc::new(DashMap::new()), //Arc::new(RwLock::new(Vec::new())), + next_subscriber_id: AtomicUsize::new(1), } } - async fn subscribe(&self) -> BoxedProviderStream { + fn subscribe(&self) -> BoxedProviderStream { let (tx, rx) = mpsc::channel(self.buf_size); - self.subscribers.write().await.push(tx); + let id = self.next_subscriber_id.fetch_add(1, Ordering::AcqRel); + self.subscribers.insert(id, tx); convert_stream(ReceiverStream::new(rx).boxed()) } @@ -86,36 +83,64 @@ impl SharedStreamState { E: std::fmt::Debug + std::marker::Send { let mut source_stream = Box::pin(bytes_stream); - let subscriber = Arc::clone(&self.subscribers); + let subscriber = self.subscribers.clone(); let streaming_url = stream_url.to_string(); - let mut tick = tokio::time::interval(Duration::from_millis(5)); - //Spawn a task to forward items from the source stream to the broadcast channel tokio::spawn(async move { while let Some(item) = source_stream.next().await { if let Ok(data) = item { - if subscriber.read().await.is_empty() { + if subscriber.is_empty() { debug_if_enabled!("No active subscribers. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url)); // Cleanup for removing unused shared streams shared_streams.unregister(&streaming_url).await; break; } - let start_time = Instant::now(); - loop { - if subscriber.read().await.iter().any(|sender| sender.capacity() > 0) { + // backpressure + let mut all_full = true; + // Check if at least one sender still has capacity. + for sender in &*subscriber { + if sender.capacity() > 0 { + all_full = false; break; } - if start_time.elapsed().as_secs() > 5 { - break; - } - tick.tick().await; } - let mut subs = subscriber.write().await; - // TODO use drain_filter when stable - (*subs).retain(|sender| { + // If all are full, wait asynchronously until one has free capacity. + if all_full { + let mut futures = FuturesUnordered::new(); + for sender in &*subscriber { + futures.push(async move { + match sender.value().reserve().await { + Ok(permit) => { + drop(permit); + Ok(()) + }, + Err(_) => Err(()), + } + }); + } + + loop { + match futures.next().await { + Some(Ok(())) => { + break; + } + Some(Err(())) => { + // One closed – continue + } + None => { + // All futures failed + log::debug!("All senders are closed. Shutting down stream."); + shared_streams.unregister(&streaming_url).await; + break; + } + } + } + } + + subscriber.retain(|_id, sender| { match sender.try_send(data.clone()) { Ok(()) => true, Err(TrySendError::Closed(_)) => false, @@ -126,7 +151,6 @@ impl SharedStreamState { } }); } - tick.tick().await; } debug_if_enabled!("Shared stream exhausted. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url)); shared_streams.unregister(&streaming_url).await; @@ -156,7 +180,7 @@ impl SharedStreamManager { } async fn subscribe_stream(&self, stream_url: &str) -> Option { - let stream_data = self.shared_streams.read().await.get(stream_url)?.subscribe().await; + let stream_data = self.shared_streams.read().await.get(stream_url)?.subscribe(); Some(stream_data) } @@ -169,7 +193,7 @@ impl SharedStreamManager { stream_url: &str, bytes_stream: S, headers: Vec<(String, String)>, - buffer_size: usize,) + buffer_size: usize,) -> Option where S: Stream> + Unpin + 'static + std::marker::Send, E: std::fmt::Debug + std::marker::Send @@ -179,6 +203,7 @@ impl SharedStreamManager { shared_state.broadcast(stream_url, bytes_stream, Arc::clone(&app_state.shared_stream_manager)); app_state.shared_stream_manager.register(stream_url, shared_state).await; debug_if_enabled!("Created shared provider stream {}", sanitize_sensitive_info(stream_url)); + Self::subscribe_shared_stream(app_state, stream_url).await } /// Creates a broadcast notify stream for the given URL if a shared stream exists. From 6828e0f5d17e042b651df67aeafb105d4ea16101 Mon Sep 17 00:00:00 2001 From: euzu Date: Wed, 11 Jun 2025 15:50:15 +0200 Subject: [PATCH 2/3] refactored shared stream manager --- .../model/streams/shared_stream_manager.rs | 131 ++++++++++-------- 1 file changed, 71 insertions(+), 60 deletions(-) diff --git a/src/api/model/streams/shared_stream_manager.rs b/src/api/model/streams/shared_stream_manager.rs index 7cc53f15d..58be8c60e 100644 --- a/src/api/model/streams/shared_stream_manager.rs +++ b/src/api/model/streams/shared_stream_manager.rs @@ -77,81 +77,92 @@ impl SharedStreamState { convert_stream(ReceiverStream::new(rx).boxed()) } - fn broadcast(&self, stream_url: &str, bytes_stream: S, shared_streams: Arc) + fn broadcast( + &self, + stream_url: &str, + bytes_stream: S, + shared_streams: Arc, + ) where - S: Stream> + Unpin + 'static + std::marker::Send, - E: std::fmt::Debug + std::marker::Send + S: Stream> + Unpin + 'static + Send, + E: std::fmt::Debug + Send, { let mut source_stream = Box::pin(bytes_stream); - let subscriber = self.subscribers.clone(); + let subscribers = self.subscribers.clone(); let streaming_url = stream_url.to_string(); - //Spawn a task to forward items from the source stream to the broadcast channel tokio::spawn(async move { while let Some(item) = source_stream.next().await { - if let Ok(data) = item { - if subscriber.is_empty() { - debug_if_enabled!("No active subscribers. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url)); - // Cleanup for removing unused shared streams - shared_streams.unregister(&streaming_url).await; - break; - } + let Ok(data) = item else { + trace!("Shared stream received Err, skipping..."); + continue; + }; - // backpressure - let mut all_full = true; - // Check if at least one sender still has capacity. - for sender in &*subscriber { - if sender.capacity() > 0 { - all_full = false; - break; + if subscribers.is_empty() { + debug_if_enabled!("No active subscribers. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url)); + shared_streams.unregister(&streaming_url).await; + break; + } + + // Fast-path: at least one has capacity + if subscribers.iter().any(|sender| sender.value().capacity() > 0) { + // Try sending immediately + subscribers.retain(|_id, sender| match sender.try_send(data.clone()) { + Ok(()) => true, + Err(TrySendError::Closed(_)) => false, + Err(err) => { + trace!("broadcast try_send error: {:?}", err); + true } - } + }); + continue; + } - // If all are full, wait asynchronously until one has free capacity. - if all_full { - let mut futures = FuturesUnordered::new(); - for sender in &*subscriber { - futures.push(async move { - match sender.value().reserve().await { - Ok(permit) => { - drop(permit); - Ok(()) - }, - Err(_) => Err(()), - } - }); - } - - loop { - match futures.next().await { - Some(Ok(())) => { - break; - } - Some(Err(())) => { - // One closed – continue - } - None => { - // All futures failed - log::debug!("All senders are closed. Shutting down stream."); - shared_streams.unregister(&streaming_url).await; - break; - } - } - } - } - - subscriber.retain(|_id, sender| { - match sender.try_send(data.clone()) { - Ok(()) => true, - Err(TrySendError::Closed(_)) => false, - Err(err) => { - trace!("broadcast send error {err}"); - true + // All are full → wait until one becomes available + let mut futures = FuturesUnordered::new(); + for sender in subscribers.iter() { + let tx = sender.value().clone(); + futures.push(async move { + match tx.reserve().await { + Ok(permit) => { + drop(permit); + Ok(()) } + Err(_) => Err(()), } }); } + + let mut has_fillable_subscriber = false; + while let Some(result) = futures.next().await { + match result { + Ok(_) => { + has_fillable_subscriber = true; + break; + } + Err(_) => { + // ignore; continue waiting + } + } + } + + if !has_fillable_subscriber { + debug_if_enabled!("All subscribers closed. Shutting down shared provider stream {}",sanitize_sensitive_info(&streaming_url)); + shared_streams.unregister(&streaming_url).await; + break; + } + + // Re-validate subscribers and send again + subscribers.retain(|_id, sender| match sender.try_send(data.clone()) { + Ok(()) => true, + Err(TrySendError::Closed(_)) => false, + Err(err) => { + trace!("broadcast try_send error after reserve: {:?}", err); + true + } + }); } + debug_if_enabled!("Shared stream exhausted. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url)); shared_streams.unregister(&streaming_url).await; }); From d8f07c0eebff82e26300f6b263ca095ee96cad6c Mon Sep 17 00:00:00 2001 From: euzu Date: Wed, 11 Jun 2025 15:55:40 +0200 Subject: [PATCH 3/3] Active user manager --- src/api/model/active_user_manager.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/api/model/active_user_manager.rs b/src/api/model/active_user_manager.rs index 362dbf88c..7f9fcf5d9 100644 --- a/src/api/model/active_user_manager.rs +++ b/src/api/model/active_user_manager.rs @@ -195,10 +195,14 @@ impl ActiveUserManager { connection_data.connections -= 1; } - if connection_data.connections == 0 || connection_data.connections < connection_data.max_connections { - // Grace timeout expired, reset grace counters - connection_data.granted_grace = false; - connection_data.grace_ts = 0; + if connection_data.connections == 0 { + lock.remove(username); + } else { + if connection_data.connections < connection_data.max_connections { + // Grace timeout expired, reset grace counters + connection_data.granted_grace = false; + connection_data.grace_ts = 0; + } } } drop(lock);