euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-23 00:03:33 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

Reduce heap allocation (#764)

Browse Source 
- Refactored PlaylistSource to avoid Box<dyn> whre possible
- Some other small heap allocation fixes
This commit is contained in:
euzu
2026-05-28 17:58:41 +02:00
committed by GitHub
parent 081e3fd94c
commit 4c5d966248
39 changed files with 1200 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/filtered_playlist_source.rs
-173
View File
@@ -1,173 +0,0 @@
use crate::repository::{MemoryPlaylistSource, PlaylistSource};
use futures::future::BoxFuture;
use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemType, UUIDType, XtreamCluster};
use std::collections::HashSet;
use std::sync::Arc;
pub(crate) struct FilteredPlaylistSource {
inner: Box<dyn PlaylistSource>,
skip_set: Arc<HashSet<XtreamCluster>>,
}
impl FilteredPlaylistSource {
pub(crate) fn new(inner: Box<dyn PlaylistSource>, skip_set: HashSet<XtreamCluster>) -> Self {
Self {
inner,
skip_set: Arc::new(skip_set),
}
}
fn filter_group(&self, mut group: PlaylistGroup) -> Option<PlaylistGroup> {
if self.skip_set.contains(&group.xtream_cluster) {
return None;
}
group.channels.retain(|item| !self.skip_set.contains(&item.header.xtream_cluster));
if group.channels.is_empty() {
None
} else {
Some(group)
}
}
}
impl PlaylistSource for FilteredPlaylistSource {
fn is_memory(&self) -> bool {
self.inner.is_memory()
}
fn get_channel_count(&mut self) -> usize {
let skip_set = Arc::clone(&self.skip_set);
self.inner
.items()
.filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster))
.count()
}
fn get_group_count(&mut self) -> usize {
let skip_set = Arc::clone(&self.skip_set);
let mut groups = HashSet::<(XtreamCluster, Arc<str>)>::new();
for item in self
.inner
.items()
.filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster))
{
let pli = item.as_ref();
groups.insert((pli.header.xtream_cluster, Arc::clone(&pli.header.group)));
}
groups.len()
}
fn is_empty(&mut self) -> bool {
self.get_channel_count() == 0
}
#[allow(clippy::wrong_self_convention)]
fn into_items(&mut self) -> Box<dyn Iterator<Item = PlaylistItem> + Send + '_> {
let skip_set = Arc::clone(&self.skip_set);
Box::new(
self.inner
.into_items()
.filter(move |item| !skip_set.contains(&item.header.xtream_cluster)),
)
}
fn items_mut(&mut self) -> Box<dyn Iterator<Item = &mut PlaylistItem> + Send + '_> {
let skip_set = Arc::clone(&self.skip_set);
Box::new(self.inner.items_mut().filter_map(move |item| {
if skip_set.contains(&item.header.xtream_cluster) {
None
} else {
Some(item)
}
}))
}
fn items<'a>(&'a mut self) -> Box<dyn Iterator<Item = std::borrow::Cow<'a, PlaylistItem>> + Send + 'a> {
let skip_set = Arc::clone(&self.skip_set);
Box::new(
self.inner
.items()
.filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster)),
)
}
fn update_playlist<'a>(&'a mut self, plg: &'a PlaylistGroup) -> BoxFuture<'a, ()> {
if self.skip_set.contains(&plg.xtream_cluster) {
return Box::pin(async move {});
}
self.inner.update_playlist(plg)
}
fn get_missing_vod_info_count(&mut self) -> usize {
let skip_set = Arc::clone(&self.skip_set);
self.inner
.items()
.filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster))
.filter(|item| {
let pli = item.as_ref();
pli.header.xtream_cluster == XtreamCluster::Video
&& pli.header.item_type == PlaylistItemType::Video
&& !pli.has_details()
})
.count()
}
fn get_missing_series_info_count(&mut self) -> usize {
let skip_set = Arc::clone(&self.skip_set);
self.inner
.items()
.filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster))
.filter(|item| {
let pli = item.as_ref();
pli.header.xtream_cluster == XtreamCluster::Series
&& pli.header.item_type == PlaylistItemType::SeriesInfo
&& pli.header.id.parse::<u32>().is_ok_and(|id| id > 0)
&& !pli.has_details()
})
.count()
}
fn deduplicate(&mut self, duplicates: &mut HashSet<UUIDType>) {
if self.inner.is_memory() {
let filtered_groups = self
.inner
.take_groups()
.into_iter()
.filter_map(|group| self.filter_group(group))
.collect::<Vec<_>>();
let mut memory = MemoryPlaylistSource::new(filtered_groups).boxed();
memory.deduplicate(duplicates);
self.inner = memory;
return;
}
self.inner.deduplicate(duplicates);
}
fn take_groups(&mut self) -> Vec<PlaylistGroup> {
self.inner
.take_groups()
.into_iter()
.filter_map(|group| self.filter_group(group))
.collect()
}
fn clone_box(&self) -> Box<dyn PlaylistSource> {
Box::new(Self {
inner: self.inner.clone_box(),
skip_set: Arc::clone(&self.skip_set),
})
}
fn release_resources(&mut self, cluster: XtreamCluster) {
self.inner.release_resources(cluster);
}
fn obtain_resources(&mut self) -> BoxFuture<'_, ()> {
self.inner.obtain_resources()
}
fn sort_by_provider_ordinal(&mut self) {
self.inner.sort_by_provider_ordinal();
}
}
backend/src/processing/processor/library.rs
+25 -6
View File
@@ -1,4 +1,4 @@
use crate::library::{EpisodeMetadata, MediaMetadata, MetadataAsyncIter, MetadataCacheEntry, TechnicalMetadata};
use crate::library::{Actor, EpisodeMetadata, MediaMetadata, MetadataAsyncIter, MetadataCacheEntry, TechnicalMetadata, VideoClipMetadata};
use crate::library::resolve_metadata_storage_path;
use crate::model::{AppConfig, ConfigInput};
use shared::concat_string;
@@ -41,6 +41,24 @@ fn technical_bitrate(technical: Option<&TechnicalMetadata>) -> u32 {
.unwrap_or_default()
}
fn join_actor_names(actors: &[Actor]) -> Arc<str> {
let mut names = String::new();
for (idx, actor) in actors.iter().enumerate() {
if idx > 0 {
names.push_str(", ");
}
names.push_str(&actor.name);
}
names.intern()
}
fn youtube_trailer_key(videos: Option<&[VideoClipMetadata]>) -> Option<&str> {
videos?
.iter()
.find(|video| video.site.eq_ignore_ascii_case("youtube"))
.map(|video| video.key.as_str())
}
pub async fn download_library_playlist(_client: &reqwest::Client, app_config: &Arc<AppConfig>, input: &ConfigInput) -> (Vec<PlaylistGroup>, Vec<TuliproxError>) {
let config = &*app_config.config.load();
let Some(library_config) = config.library.as_ref() else { return (vec![], vec![]) };
@@ -280,10 +298,11 @@ pub fn metadata_cache_entry_to_xtream_movie_info(
.and_then(|s| s.to_str())
.map(ToString::to_string).unwrap_or_default();
let actor_names = movie.actors.as_ref().map(|a| a.iter().map(|a| a.name.clone()).collect::<Vec<_>>().join(", ").intern());
let actor_names = movie.actors.as_deref().map(join_actor_names);
let technical = movie.technical.as_ref();
let duration_secs = technical_duration_secs(technical).or_else(|| movie.runtime.map(|runtime| runtime * 60));
let duration = duration_secs.map(duration_secs_to_xtream_duration);
let youtube_trailer = youtube_trailer_key(movie.videos.as_deref());
let properties = VideoStreamProperties {
name: movie.title.clone().into(),
@@ -297,7 +316,7 @@ pub fn metadata_cache_entry_to_xtream_movie_info(
rating: movie.rating,
rating_5based: None,
stream_type: Some("movie".intern()),
trailer: movie.videos.as_ref().and_then(|v| v.iter().find(|video| video.site.eq_ignore_ascii_case("youtube")).map(|video| video.key.clone().into())),
trailer: youtube_trailer.map(Into::into),
tmdb: movie.tmdb_id,
is_adult: 0,
details: Some(VideoStreamDetailProperties {
@@ -308,7 +327,7 @@ pub fn metadata_cache_entry_to_xtream_movie_info(
release_date: movie.year.map(|y| format!("{y}-01-01").into()),
episode_run_time: movie.runtime,
director: movie.directors.as_ref().map(|d| d.join(", ").into()),
youtube_trailer: movie.videos.as_ref().and_then(|v| v.iter().find(|video| video.site.eq_ignore_ascii_case("youtube")).map(|video| video.key.clone().into())),
youtube_trailer: youtube_trailer.map(Into::into),
actors: actor_names.clone(),
cast: actor_names.clone(),
genre: movie.genres.as_ref().map(|g| g.join(", ").into()),
@@ -347,9 +366,9 @@ pub fn metadata_cache_entry_to_xtream_series_info(
MediaMetadata::Series(m) => m,
};
let actor_names: Arc<str> = series.actors.as_ref().map(|a| a.iter().map(|a| a.name.clone()).collect::<Vec<_>>().join(", ")).unwrap_or_default().into();
let actor_names: Arc<str> = series.actors.as_deref().map(join_actor_names).unwrap_or_default();
let release_date = series.year.map(|y| format!("{y}-01-01"));
let youtube_trailer = series.videos.as_ref().and_then(|v| v.iter().find(|video| video.site.eq_ignore_ascii_case("youtube")).map(|video| video.key.clone())).unwrap_or_default();
let youtube_trailer = youtube_trailer_key(series.videos.as_deref()).unwrap_or_default();
let series_thumbnail = thumbnail_url(entry, api_base_path);
let series_art = series.poster.clone().or(series_thumbnail.clone());
backend/src/processing/processor/mod.rs
-1
View File
@@ -8,7 +8,6 @@ mod sort;
mod trakt;
mod library;
mod stream_probe;
mod filtered_playlist_source;
mod probe_handle_guard;
mod resolve_options;
pub use self::playlist::*;
backend/src/processing/processor/playlist.rs
+38 -31
View File
@@ -1,4 +1,3 @@
use super::filtered_playlist_source::FilteredPlaylistSource;
use crate::{
api::{
model::{
@@ -66,12 +65,23 @@ use tokio::{
const PLAYLIST_UPDATE_MAX_DURATION_SECS: u64 = 3600;
fn join_arc_strs(values: &[Arc<str>], separator: &str) -> String {
let mut result = String::new();
for value in values {
if !result.is_empty() {
result.push_str(separator);
}
result.push_str(value.as_ref());
}
result
}
fn is_valid(pli: &PlaylistItem, filter: &Filter, match_as_ascii: bool) -> bool {
let provider = ValueProvider { pli, match_as_ascii };
filter.filter(&provider)
}
pub fn apply_filter_to_source(source: &mut dyn PlaylistSource, filter: &Filter) -> Option<Vec<PlaylistGroup>> {
pub fn apply_filter_to_source(source: &mut PlaylistSource, filter: &Filter) -> Option<Vec<PlaylistGroup>> {
let mut groups: IndexMap<CategoryKey, PlaylistGroup> = IndexMap::new();
for pli in source.into_items() {
if is_valid(&pli, filter, false) {
@@ -100,7 +110,7 @@ pub fn apply_filter_to_source(source: &mut dyn PlaylistSource, filter: &Filter)
}
}
fn filter_playlist(source: &mut dyn PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
fn filter_playlist(source: &mut PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
apply_filter_to_source(source, &target.filter)
}
@@ -152,14 +162,13 @@ fn exec_rename(pli: &mut PlaylistItem, rename: Option<&Vec<ConfigRename>>) {
if log_enabled!(log::Level::Debug) && *value != *cap {
trace_if_enabled!("Renamed {}={value} to {cap}", &r.field);
}
let value = cap.into_owned();
set_field_value(result, r.field, value);
set_field_value(result, r.field, cap.as_ref());
}
}
}
}
fn rename_playlist(source: &mut dyn PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
fn rename_playlist(source: &mut PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
match &target.rename {
Some(renames) if !renames.is_empty() => {
let mut groups: IndexMap<(XtreamCluster, Arc<str>), PlaylistGroup> = IndexMap::new();
@@ -233,7 +242,7 @@ fn map_channel_and_flatten(channel: PlaylistItem, mapping: &Mapping) -> Vec<Play
result
}
fn map_playlist(source: &mut dyn PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
fn map_playlist(source: &mut PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
let mapping_binding = target.mapping.load();
let mappings = mapping_binding.as_ref()?;
let valid_mappings = mappings.iter().filter(|m| m.mapper.as_ref().is_some_and(|v| !v.is_empty()));
@@ -396,17 +405,14 @@ async fn download_plex_media_server_playlist(
}
}
fn filter_skipped_clusters_from_source(
source: Box<dyn PlaylistSource>,
input: &ConfigInput,
) -> Box<dyn PlaylistSource> {
fn filter_skipped_clusters_from_source(source: PlaylistSource, input: &ConfigInput) -> PlaylistSource {
let skip_clusters = collect_effective_skip_clusters(input);
if skip_clusters.is_empty() {
return source;
}
let skip_set: HashSet<XtreamCluster> = skip_clusters.into_iter().collect();
Box::new(FilteredPlaylistSource::new(source, skip_set))
PlaylistSource::filtered(source, skip_set)
}
#[allow(clippy::too_many_lines)]
@@ -679,7 +685,7 @@ async fn process_source(
if !disabled_inputs.is_empty() && !source_downloaded {
warn!(
"Source at index {source_idx} has no enabled inputs for the given targets. Disabled: {}",
disabled_inputs.iter().map(std::convert::AsRef::as_ref).collect::<Vec<&str>>().join(", ")
join_arc_strs(&disabled_inputs, ", ")
);
}
if source_downloaded {
@@ -687,7 +693,7 @@ async fn process_source(
debug!("Source at index {source_idx} is empty");
errors.push(TuliproxError::RepositoryPlaylist(format!(
"Source at index {source_idx} is empty: {}",
source.inputs.iter().map(Clone::clone).collect::<Vec<Arc<str>>>().join(", ")
join_arc_strs(&source.inputs, ", ")
)));
} else {
debug_if_enabled!(
@@ -768,17 +774,17 @@ async fn invalidate_input_cache_status(ctx: &PlaylistProcessingContext, input: &
async fn load_cached_input_playlist(
ctx: &PlaylistProcessingContext,
input: &Arc<ConfigInput>,
) -> (Box<dyn PlaylistSource>, Option<TuliproxError>) {
) -> (PlaylistSource, Option<TuliproxError>) {
match load_input_playlist(ctx, input, None).await {
Ok(pl_source) => (pl_source, None),
Err(err) => (MemoryPlaylistSource::default().boxed(), Some(err)),
Err(err) => (MemoryPlaylistSource::default().into_source(), Some(err)),
}
}
async fn download_input(
ctx: &PlaylistProcessingContext,
input: &Arc<ConfigInput>,
) -> (Vec<TuliproxError>, Box<dyn PlaylistSource>, Option<TuliproxError>) {
) -> (Vec<TuliproxError>, PlaylistSource, Option<TuliproxError>) {
// Coordination Logic
let need_download = !ctx.is_input_downloaded(&input.name).await;
// Keep this lock for the whole critical section (download + persist/load + mark processed)
@@ -806,7 +812,7 @@ async fn download_input(
PlaylistDownloadResult::new(vec![], vec![], true, false)
};
let mut preloaded_playlist: Option<(Box<dyn PlaylistSource>, Option<TuliproxError>)> = None;
let mut preloaded_playlist: Option<(PlaylistSource, Option<TuliproxError>)> = None;
if playlist_download_result.was_cached {
let (cached_playlist, cached_error) = load_cached_input_playlist(ctx, input).await;
// Defensive fallback: if cache metadata says "valid" but persisted data is unreadable,
@@ -852,12 +858,12 @@ async fn download_input(
} else if playlist_download_result.was_cached || playlist_download_result.persisted {
match load_input_playlist(ctx, input, None).await {
Ok(pl_source) => (pl_source, None),
Err(e) => (MemoryPlaylistSource::default().boxed(), Some(e)),
Err(e) => (MemoryPlaylistSource::default().into_source(), Some(e)),
}
} else {
debug!("Persisting input '{}' playlist", input.name);
let (pl, err) = persist_input_playlist(&ctx.config, input, playlist_download_result.downloaded_playlist).await;
(MemoryPlaylistSource::new(pl).boxed(), err)
(MemoryPlaylistSource::new(pl).into_source(), err)
};
let playlist = filter_skipped_clusters_from_source(playlist, input);
@@ -1018,7 +1024,7 @@ async fn process_sources(processing_ctx: &PlaylistProcessingContext) -> (Vec<Sou
}
}
pub type ProcessingPipe = Vec<fn(source: &mut dyn PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>>>;
pub type ProcessingPipe = Vec<fn(source: &mut PlaylistSource, target: &ConfigTarget) -> Option<Vec<PlaylistGroup>>>;
fn get_processing_pipe(target: &ConfigTarget) -> ProcessingPipe {
match &target.processing_order {
@@ -1037,15 +1043,15 @@ fn execute_pipe<'a>(
fpl: &mut FetchedPlaylist<'a>,
duplicates: &mut HashSet<UUIDType>,
consume_source: bool,
) -> FetchedPlaylist<'a> {
) -> Result<FetchedPlaylist<'a>, TuliproxError> {
let source = if consume_source {
if fpl.is_memory() {
MemoryPlaylistSource::new(fpl.source.take_groups()).boxed()
MemoryPlaylistSource::new(fpl.source.take_groups()).into_source()
} else {
std::mem::replace(&mut fpl.source, MemoryPlaylistSource::default().boxed())
std::mem::replace(&mut fpl.source, MemoryPlaylistSource::default().into_source())
}
} else {
fpl.clone_source()
fpl.clone_source()?
};
let mut new_fpl = FetchedPlaylist { input: fpl.input, source, epg: fpl.epg.clone() };
@@ -1054,15 +1060,15 @@ fn execute_pipe<'a>(
}
for f in pipe {
if let Some(groups) = f(new_fpl.source.as_mut(), target) {
new_fpl.source = MemoryPlaylistSource::new(groups).boxed();
if let Some(groups) = f(&mut new_fpl.source, target) {
new_fpl.source = MemoryPlaylistSource::new(groups).into_source();
}
}
// Ensure source is memory-based for downstream mutable processing (VOD/series resolution)
if !new_fpl.is_memory() {
new_fpl.source = MemoryPlaylistSource::new(new_fpl.source.take_groups()).boxed();
new_fpl.source = MemoryPlaylistSource::new(new_fpl.source.take_groups()).into_source();
}
new_fpl
Ok(new_fpl)
}
// This method is needed, because of duplicate group names in different inputs.
@@ -1116,7 +1122,8 @@ async fn process_playlist_for_target(
format!("target '{}' input '{}' before_pipe", target.name, provider_fpl.input.name).as_str(),
);
step.broadcast("Executing transformations on '{}' playlist", &target.name);
let mut processed_fpl = execute_pipe(target, &pipe, provider_fpl, &mut duplicates, consume_input_source);
let mut processed_fpl =
execute_pipe(target, &pipe, provider_fpl, &mut duplicates, consume_input_source).map_err(|err| vec![err])?;
log_memory_snapshot(
format!("target '{}' input '{}' after_pipe", target.name, provider_fpl.input.name).as_str(),
);
@@ -1830,7 +1837,7 @@ mod tests {
},
];
let source = MemoryPlaylistSource::new(groups).boxed();
let source = MemoryPlaylistSource::new(groups).into_source();
let input = ConfigInput {
name: "skip_live".intern(),
input_type: InputType::Xtream,
backend/src/processing/processor/xtream_series.rs
+2 -4
View File
@@ -18,9 +18,7 @@ use crate::processing::processor::{
FOREGROUND_RETRY_BATCH_MAX_SIZE as RETRY_BATCH_MAX_SIZE,
};
use crate::repository::persists_input_series_info;
use crate::repository::{
get_input_storage_path, persist_input_series_info_batch, MemoryPlaylistSource, PlaylistSource,
};
use crate::repository::{get_input_storage_path, persist_input_series_info_batch, MemoryPlaylistSource};
use crate::repository::{xtream_get_file_path, BPlusTreeQuery};
use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome};
use crate::utils::{debug_if_enabled, xtream};
@@ -124,7 +122,7 @@ async fn playlist_resolve_series_info(
// Apply pipe transformations to new groups
let mut new_playlist = groups_to_add;
for f in pipe {
let mut source = MemoryPlaylistSource::new(new_playlist);
let mut source = MemoryPlaylistSource::new(new_playlist).into_source();
if let Some(v) = f(&mut source, target) {
new_playlist = v;
} else {