mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-23 08:12:34 +02:00
Feature/optimizations (#777)
* perf: binary-search EPG programme lookup Replaces the linear position() scan with a binary search via partition_point for EPG programmes. Reduces complexity to O(log p). * perf: cache interned PlaylistItemType label Avoids repetitive hash-map lookups during sorting and filtering. The five fixed labels are now interned once via OnceLock and returned as a cheap Arc clone. * perf: avoid per-event String allocation in web-UI XMLTV parser Uses borrowed Cow types instead of allocating a new String for every element name. Tracks the active text state using a small TextTag enum. * perf: throttle per-chunk download control poll Throttles the control_signal lock polling to a 200ms interval on multi-GB downloads to prevent CPU thrashing. Immediate actions (pause/cancel) remain guaranteed via control_notify. * perf: pre-size filtered channel buffers in apply_filter_to_playlist Pre-allocates the filtered channel vector based on the source group length. Eliminates step-by-step reallocations on large playlists. * fix: surface playlist cache load failures instead of silent empty fallback Replaces silent error swallowing with real error logging (distinguishing missing files from corruption) while preserving the empty fallback behavior. * perf: remove redundant first drain loop in get_remote_content_as_file Removes a dead, redundant loop. The stream body is now consumed only once, allowing the idle timeout to protect the transfer correctly. * perf: back off instead of busy-spinning on empty client-stream chunks Introduces a short Sleep backoff (Poll::Pending) when a provider sends continuous empty keep-alive chunks. Prevents the CPU from spinning hot. * perf: make library orphan cleanup linear instead of O(n^2) Reduces the complexity of cleaning up orphaned entries from O(n^2) to linear O(n). Loads the metadata library only once and pre-computes partitions. * perf: indexed batched merge for series expansion instead of O(n^2) scans Replaces expensive O(n^2) scans during series expansion with index-based batch processing. Builds temporary lookup indexes and merges groups linearly. * perf: pre-size group flatten buffers to reduce reallocation churn Pre-allocates target merge vectors and category indexes to eliminate repeated memory reallocations when flattening large target playlists. * perf: stream custom provider playlist endpoint instead of materializing it Switches from full in-memory materialization to lazy conversion and streaming of JSON/binary responses. Massively reduces the memory footprint. * refactor: centralize InputType batch/family checks behind enum helpers Moves scattered matches! checks for M3u/Xtream types into centralized enum helpers (is_batch(), etc.) to prevent logic drift. * refactor: centralize XMLTV tag/attribute literals into named constants Replaces hardcoded inline strings in the XMLTV parser with named constants (EPG_TAG_TITLE, etc.) as a single source of truth. * refactor: derive GeoIpUpdateError Display/Error via thiserror Replaces manual Display and Error implementations with the thiserror macro, allowing new error variants to self-document their formatting. * refactor: attach per-cluster facts to XtreamCluster instead of scattered matches Centralizes redundant API actions and collection name mappings directly onto the XtreamCluster struct instead of using scattered match blocks. * refactor: centralize PlaylistItemType video/series classification Introduces is_video() / is_series() on the enum to unify duplicate matches! checks across filters, probes, and tracking modules. * feat: add per-stage summary logs to VOD/series resolve queueing Adds a concise debug summary per processing stage (total counters and elapsed time) to make hot-path diagnostics easier without raising the log level. * refactor: generate ItemField get/set from a single direct-field list Uses a callback macro to synchronously generate both getter and setter blocks for directly-bound fields (Name, Url, etc.) from a single master list. * refactor: derive input-type routing from a single capability descriptor Bundles InputType behaviors (backends, connection rules) into a single capabilities() descriptor to prevent bugs when adding new variants. * refactor: derive output-format routing from a single capability descriptor Centralizes output format features (filter, EPG, and cache support) into a single TargetCapabilities descriptor instead of spreading logic across no-op match arms.
This commit is contained in:
@@ -115,10 +115,15 @@ fn filter_playlist(source: &mut PlaylistSource, target: &ConfigTarget) -> Option
|
||||
}
|
||||
|
||||
pub fn apply_filter_to_playlist(playlist: &mut [PlaylistGroup], filter: &Filter) -> Option<Vec<PlaylistGroup>> {
|
||||
let mut new_playlist = Vec::with_capacity(128);
|
||||
for pg in playlist.iter_mut() {
|
||||
let channels =
|
||||
pg.channels.iter().filter(|&pli| is_valid(pli, filter, false)).cloned().collect::<Vec<PlaylistItem>>();
|
||||
// NOTE: the source `playlist` is intentionally cloned (not drained) here because
|
||||
// the caller reuses the same slice for every target output and for the no-filter
|
||||
// fallback path, so the survivors cannot be moved out of it. Cap the initial
|
||||
// allocation so selective filters do not retain capacity for every source item.
|
||||
const INITIAL_FILTERED_GROUP_CAPACITY: usize = 256;
|
||||
let mut new_playlist = Vec::with_capacity(playlist.len());
|
||||
for pg in playlist.iter() {
|
||||
let mut channels = Vec::with_capacity(pg.channels.len().min(INITIAL_FILTERED_GROUP_CAPACITY));
|
||||
channels.extend(pg.channels.iter().filter(|&pli| is_valid(pli, filter, false)).cloned());
|
||||
if !channels.is_empty() {
|
||||
new_playlist.push(PlaylistGroup {
|
||||
id: pg.id,
|
||||
@@ -1074,9 +1079,10 @@ fn execute_pipe<'a>(
|
||||
// This method is needed, because of duplicate group names in different inputs.
|
||||
// We merge the same group names considering cluster together.
|
||||
fn flatten_groups(playlistgroups: Vec<PlaylistGroup>) -> Vec<PlaylistGroup> {
|
||||
let mut sort_order: Vec<PlaylistGroup> = vec![];
|
||||
let upper_bound = playlistgroups.len();
|
||||
let mut sort_order: Vec<PlaylistGroup> = Vec::with_capacity(upper_bound);
|
||||
let mut idx: usize = 0;
|
||||
let mut group_map: HashMap<CategoryKey, usize> = HashMap::new();
|
||||
let mut group_map: HashMap<CategoryKey, usize> = HashMap::with_capacity(upper_bound);
|
||||
for group in playlistgroups {
|
||||
let normalized_title: Arc<str> = shared::utils::deunicode_string(&group.title).to_lowercase().intern();
|
||||
let key = (group.xtream_cluster, normalized_title);
|
||||
|
||||
@@ -58,7 +58,7 @@ pub enum GenericProbeMetadataOutcome {
|
||||
}
|
||||
|
||||
fn requires_provider_connection_for_generic_probe(input_type: InputType) -> bool {
|
||||
!(matches!(input_type, InputType::Library) || input_type.is_media_server())
|
||||
input_type.capabilities().requires_provider_connection_for_probe
|
||||
}
|
||||
|
||||
fn uses_seekable_remote_probe(item_type: PlaylistItemType, is_remote_probe: bool) -> bool {
|
||||
@@ -177,9 +177,9 @@ async fn prepare_generic_stream_metadata(
|
||||
InputType::Xtream | InputType::XtreamBatch => {
|
||||
let cluster = if item_type.is_live() {
|
||||
XtreamCluster::Live
|
||||
} else if matches!(item_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo) {
|
||||
} else if item_type.is_video() {
|
||||
XtreamCluster::Video
|
||||
} else if matches!(item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) {
|
||||
} else if item_type.is_series() {
|
||||
XtreamCluster::Series
|
||||
} else {
|
||||
// Generic probing currently supports live/video/series payload shapes.
|
||||
@@ -223,9 +223,7 @@ async fn prepare_generic_stream_metadata(
|
||||
debug!("Skipping unsupported generic stream probe for {unique_id}: {safe_probe_url}");
|
||||
return Ok(PreparedGenericProbeOutcome::Noop);
|
||||
}
|
||||
let is_remote_probe = reqwest::Url::parse(&probe_url)
|
||||
.ok()
|
||||
.is_some_and(|url| matches!(url.scheme(), "http" | "https"));
|
||||
let is_remote_probe = reqwest::Url::parse(&probe_url).is_ok_and(|url| matches!(url.scheme(), "http" | "https"));
|
||||
let config = app_config.config.load();
|
||||
let metadata_update = config.metadata_update.clone().unwrap_or_default();
|
||||
let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60);
|
||||
@@ -422,7 +420,7 @@ pub fn update_properties(
|
||||
raw_audio: Option<Arc<str>>,
|
||||
stats: ProbeStreamStats,
|
||||
) {
|
||||
if matches!(item_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo) {
|
||||
if item_type.is_video() {
|
||||
let mut props = if let Some(StreamProperties::Video(p)) = props_opt {
|
||||
*p.clone()
|
||||
} else {
|
||||
@@ -453,9 +451,8 @@ pub fn update_properties(
|
||||
}
|
||||
*props_opt = Some(StreamProperties::Video(Box::new(props)));
|
||||
}
|
||||
else if matches!(item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) {
|
||||
let mut props = if let Some(StreamProperties::Episode(p)) = props_opt {
|
||||
*p.clone()
|
||||
else if item_type.is_series() {
|
||||
let mut props = if let Some(StreamProperties::Episode(p)) = props_opt { *p.clone()
|
||||
} else {
|
||||
EpisodeStreamProperties {
|
||||
episode_id: virtual_id,
|
||||
|
||||
@@ -42,7 +42,7 @@ fn is_compatible_content_type(cluster: XtreamCluster, content_type: TraktContent
|
||||
|
||||
fn is_matchable_playlist_item(item_type: PlaylistItemType, content_type: TraktContentType) -> bool {
|
||||
match content_type {
|
||||
TraktContentType::Vod => matches!(item_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo),
|
||||
TraktContentType::Vod => item_type.is_video(),
|
||||
TraktContentType::Series => matches!(item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo),
|
||||
TraktContentType::Both => matches!(
|
||||
item_type,
|
||||
@@ -246,7 +246,7 @@ fn series_children_by_parent_code(playlist: &[PlaylistGroup]) -> HashMap<Arc<str
|
||||
let mut children = HashMap::<Arc<str>, Vec<&PlaylistItem>>::new();
|
||||
for playlist_group in playlist {
|
||||
for channel in &playlist_group.channels {
|
||||
if matches!(channel.header.item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries)
|
||||
if channel.header.item_type.is_series()
|
||||
&& !channel.header.parent_code.is_empty()
|
||||
{
|
||||
children.entry(channel.header.parent_code.clone()).or_default().push(channel);
|
||||
|
||||
@@ -143,9 +143,8 @@ pub async fn update_live_stream_metadata(
|
||||
|
||||
let mut success = false;
|
||||
let mut not_found = false;
|
||||
let is_remote_probe = reqwest::Url::parse(probe_url_cow.as_ref())
|
||||
.ok()
|
||||
.is_some_and(|u| matches!(u.scheme(), "http" | "https"));
|
||||
let is_remote_probe =
|
||||
reqwest::Url::parse(probe_url_cow.as_ref()).is_ok_and(|u| matches!(u.scheme(), "http" | "https"));
|
||||
let probe_params = crate::utils::ffmpeg::ProbeParams {
|
||||
url: probe_url_cow.as_ref(),
|
||||
user_agent: user_agent.as_deref(),
|
||||
|
||||
@@ -131,9 +131,7 @@ async fn playlist_resolve_series_info(
|
||||
}
|
||||
|
||||
// Apply resolved episodes to playlist
|
||||
for group in new_playlist {
|
||||
fpl.update_playlist(&group).await;
|
||||
}
|
||||
fpl.extend_playlist(new_playlist);
|
||||
}
|
||||
|
||||
fn sync_resolved_series_properties(provider_fpl: &mut FetchedPlaylist<'_>, processed_fpl: &mut FetchedPlaylist<'_>) {
|
||||
@@ -198,8 +196,16 @@ fn queue_background_series_info(
|
||||
// Extract filter before iterating to avoid borrow conflict
|
||||
let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());
|
||||
|
||||
let stage_started = Instant::now();
|
||||
let mut total = 0usize;
|
||||
let mut skipped_filter = 0usize;
|
||||
let mut skipped_resolve_filter = 0usize;
|
||||
let mut queued = 0usize;
|
||||
let mut expanded = 0usize;
|
||||
for pli in fpl.items_mut() {
|
||||
total += 1;
|
||||
if !filter(pli) {
|
||||
skipped_filter += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -207,6 +213,7 @@ fn queue_background_series_info(
|
||||
if let Some(r_filter) = resolve_filter {
|
||||
let provider = ValueProvider { pli, match_as_ascii: false };
|
||||
if !r_filter.filter(&provider) {
|
||||
skipped_resolve_filter += 1;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
@@ -245,13 +252,22 @@ fn queue_background_series_info(
|
||||
);
|
||||
}
|
||||
mgr.queue_task_background(input_name_arc.clone(), task);
|
||||
queued += 1;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(group_obj) = expand_series_item(pli, input) {
|
||||
groups_to_add.push(group_obj);
|
||||
expanded += 1;
|
||||
}
|
||||
}
|
||||
|
||||
if log_enabled!(Level::Debug) {
|
||||
debug!(
|
||||
"[Task] Series resolve queueing summary for input {input_name_arc}: total={total}, queued={queued}, expanded={expanded}, skipped_filter={skipped_filter}, skipped_resolve_filter={skipped_resolve_filter}, elapsed_ms={}",
|
||||
stage_started.elapsed().as_millis()
|
||||
);
|
||||
}
|
||||
groups_to_add
|
||||
}
|
||||
|
||||
@@ -997,8 +1013,7 @@ pub async fn update_series_metadata(
|
||||
active_handle,
|
||||
);
|
||||
let is_remote_probe = reqwest::Url::parse(probe_url.as_ref())
|
||||
.ok()
|
||||
.is_some_and(|u| matches!(u.scheme(), "http" | "https"));
|
||||
.is_ok_and(|u| matches!(u.scheme(), "http" | "https"));
|
||||
let probe_params = crate::utils::ffmpeg::ProbeParams {
|
||||
url: probe_url.as_ref(),
|
||||
user_agent: user_agent.as_deref(),
|
||||
|
||||
@@ -447,8 +447,15 @@ fn queue_background_vod_info(
|
||||
let input = fpl.input;
|
||||
// Extract filter before iterating to avoid borrow conflict
|
||||
let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());
|
||||
let stage_started = std::time::Instant::now();
|
||||
let mut total = 0usize;
|
||||
let mut skipped_filter = 0usize;
|
||||
let mut skipped_resolve_filter = 0usize;
|
||||
let mut queued = 0usize;
|
||||
for pli in fpl.items_mut() {
|
||||
total += 1;
|
||||
if !filter(pli) {
|
||||
skipped_filter += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -456,6 +463,7 @@ fn queue_background_vod_info(
|
||||
if let Some(r_filter) = resolve_filter {
|
||||
let provider = ValueProvider { pli, match_as_ascii: false };
|
||||
if !r_filter.filter(&provider) {
|
||||
skipped_resolve_filter += 1;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
@@ -500,8 +508,17 @@ fn queue_background_vod_info(
|
||||
);
|
||||
}
|
||||
mgr.queue_task_background(input.name.clone(), task);
|
||||
queued += 1;
|
||||
}
|
||||
}
|
||||
|
||||
if log_enabled!(Level::Debug) {
|
||||
debug!(
|
||||
"[Task] VOD resolve queueing summary for input {}: total={total}, queued={queued}, skipped_filter={skipped_filter}, skipped_resolve_filter={skipped_resolve_filter}, elapsed_ms={}",
|
||||
input.name,
|
||||
stage_started.elapsed().as_millis()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async fn update_vod_info_immediate(
|
||||
@@ -859,9 +876,8 @@ pub async fn update_vod_metadata(
|
||||
temp_handle.as_ref().and_then(ProbeHandleGuard::handle),
|
||||
active_handle,
|
||||
);
|
||||
let is_remote_probe = reqwest::Url::parse(probe_url.as_ref())
|
||||
.ok()
|
||||
.is_some_and(|u| matches!(u.scheme(), "http" | "https"));
|
||||
let is_remote_probe =
|
||||
reqwest::Url::parse(probe_url.as_ref()).is_ok_and(|u| matches!(u.scheme(), "http" | "https"));
|
||||
let probe_params = crate::utils::ffmpeg::ProbeParams {
|
||||
url: probe_url.as_ref(),
|
||||
user_agent: user_agent.as_deref(),
|
||||
|
||||
Reference in New Issue
Block a user