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

DB optimizations (#572)

Browse Source 
DB optimizations
This commit is contained in:
euzu
2026-02-10 19:37:20 +01:00
committed by GitHub
parent 00c77eb2b9
commit 9bd44284d0
14 changed files with 1344 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/playlist.rs
+72 -36
View File
@@ -1,5 +1,5 @@
use crate::model::{AppConfig, ConfigFavourites, ConfigInput, ConfigRename, ReverseProxyDisabledHeaderConfig, TVGuide};
use crate::utils::m3u;
use crate::utils::{log_memory_snapshot, m3u};
use crate::utils::xtream;
use crate::utils::{epg, StepMeasureCallback};
use std::collections::{HashMap, HashSet};
@@ -11,7 +11,6 @@ use tokio::task::JoinSet;
use crate::api::model::{EventManager, EventMessage, PlaylistStorageState, UpdateGuard};
use crate::messaging::send_message;
use crate::model::messaging::MessageContent;
use crate::model::Epg;
use crate::model::FetchedPlaylist;
@@ -370,12 +369,13 @@ async fn playlist_download_from_input(client: &reqwest::Client, app_config: &Arc
}
async fn process_source(source_idx: usize, ctx: &PlaylistProcessingContext) -> (Vec<InputStats>, Vec<TargetStats>, Vec<TuliproxError>) {
log_memory_snapshot(format!("source[{source_idx}] start").as_str());
let sources = ctx.config.sources.load();
let mut errors = vec![];
let mut input_stats = HashMap::<Arc<str>, InputStats>::new();
let mut target_stats = Vec::<TargetStats>::new();
if let Some(source) = sources.get_source_at(source_idx) {
let mut source_playlists = Vec::with_capacity(128);
let mut source_playlists = Vec::with_capacity(source.inputs.len());
let broadcast_step = create_broadcast_callback(ctx.event_manager.as_ref());
// Download the sources
let mut source_downloaded = false;
@@ -386,6 +386,7 @@ async fn process_source(source_idx: usize, ctx: &PlaylistProcessingContext) -> (
};
if is_input_enabled(input, &ctx.user_targets) {
source_downloaded = true;
log_memory_snapshot(format!("source[{source_idx}] input '{}' before_download", input.name).as_str());
let start_time = Instant::now();
// Download the playlist for input
@@ -401,12 +402,14 @@ async fn process_source(source_idx: usize, ctx: &PlaylistProcessingContext) -> (
}
(playlist, download_err)
};
log_memory_snapshot(format!("source[{source_idx}] input '{}' after_download", input.name).as_str());
let tvguide = if input.input_type == InputType::Library {
None
} else {
download_input_epg(ctx, input, &mut error_list).await
};
log_memory_snapshot(format!("source[{source_idx}] input '{}' after_epg_download", input.name).as_str());
errors.append(&mut error_list);
let group_count = playlist_groups.get_group_count();
@@ -424,6 +427,7 @@ async fn process_source(source_idx: usize, ctx: &PlaylistProcessingContext) -> (
epg: tvguide,
}
);
log_memory_snapshot(format!("source[{source_idx}] input '{}' after_source_push", input.name).as_str());
}
let elapsed = start_time.elapsed().as_secs();
input_stats.insert(input_name.clone(), create_input_stat(group_count, channel_count, errors.len(),
@@ -436,23 +440,34 @@ async fn process_source(source_idx: usize, ctx: &PlaylistProcessingContext) -> (
errors.push(notify_err!("Source at index {source_idx} is empty: {}", source.inputs.iter().map(Clone::clone).collect::<Vec<Arc<str>>>().join(", ")));
} else {
debug_if_enabled!("Source has {} groups", source_playlists.iter_mut().map(FetchedPlaylist::get_channel_count).sum::<usize>());
for target in &source.targets {
if is_target_enabled(target, &ctx.user_targets) {
match process_playlist_for_target(ctx, &mut source_playlists, target,
&mut input_stats, &mut errors).await {
Ok(()) => {
target_stats.push(TargetStats::success(&target.name));
}
Err(mut err) => {
target_stats.push(TargetStats::failure(&target.name));
errors.append(&mut err);
}
let enabled_targets: Vec<_> = source.targets
.iter()
.filter(|target| is_target_enabled(target, &ctx.user_targets))
.collect();
for (idx, target) in enabled_targets.iter().enumerate() {
let consume_input_source = idx + 1 == enabled_targets.len();
debug!(
"Processing target '{}' (use_memory_cache={}, consume_input_source={})",
target.name, target.use_memory_cache, consume_input_source
);
log_memory_snapshot(format!("source[{source_idx}] target '{}' before_process", target.name).as_str());
match process_playlist_for_target(ctx, &mut source_playlists, target,
&mut input_stats, &mut errors, consume_input_source).await {
Ok(()) => {
target_stats.push(TargetStats::success(&target.name));
}
Err(mut err) => {
target_stats.push(TargetStats::failure(&target.name));
errors.append(&mut err);
}
}
log_memory_snapshot(format!("source[{source_idx}] target '{}' after_process", target.name).as_str());
}
}
}
}
log_memory_snapshot(format!("source[{source_idx}] end").as_str());
(input_stats.into_values().collect(), target_stats, errors)
}
@@ -655,11 +670,21 @@ fn get_processing_pipe(target: &ConfigTarget) -> ProcessingPipe {
}
}
fn execute_pipe<'a>(target: &ConfigTarget, pipe: &ProcessingPipe, fpl: &FetchedPlaylist<'a>,
duplicates: &mut HashSet<UUIDType>) -> FetchedPlaylist<'a> {
fn execute_pipe<'a>(target: &ConfigTarget, pipe: &ProcessingPipe, fpl: &mut FetchedPlaylist<'a>,
duplicates: &mut HashSet<UUIDType>, consume_source: bool) -> FetchedPlaylist<'a> {
let source = if consume_source {
if fpl.is_memory() {
MemoryPlaylistSource::new(fpl.source.take_groups()).boxed()
} else {
std::mem::replace(&mut fpl.source, MemoryPlaylistSource::default().boxed())
}
} else {
fpl.clone_source()
};
let mut new_fpl = FetchedPlaylist {
input: fpl.input,
source: fpl.clone_source(),
source,
epg: fpl.epg.clone(),
};
if target.options.as_ref().is_some_and(|opt| opt.remove_duplicates) {
@@ -709,11 +734,14 @@ async fn process_playlist_for_target(ctx: &PlaylistProcessingContext,
target: &ConfigTarget,
stats: &mut HashMap<Arc<str>, InputStats>,
errors: &mut Vec<TuliproxError>,
consume_input_source: bool,
) -> Result<(), Vec<TuliproxError>> {
debug_if_enabled!("Processing order is {}", &target.processing_order);
log_memory_snapshot(format!("target '{}' start", target.name).as_str());
let mut duplicates: HashSet<UUIDType> = HashSet::new();
let mut processed_fetched_playlists: Vec<FetchedPlaylist> = vec![];
let mut new_epg = vec![];
let mut new_playlist: Vec<PlaylistGroup> = vec![];
debug!("Executing processing pipes");
let broadcast_step = create_broadcast_callback(ctx.event_manager.as_ref());
@@ -721,11 +749,15 @@ async fn process_playlist_for_target(ctx: &PlaylistProcessingContext,
let pipe = get_processing_pipe(target);
let mut step = StepMeasure::new(&target.name, broadcast_step);
for provider_fpl in playlists.iter_mut() {
log_memory_snapshot(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);
let mut processed_fpl = execute_pipe(target, &pipe, provider_fpl, &mut duplicates, consume_input_source);
log_memory_snapshot(format!("target '{}' input '{}' after_pipe", target.name, provider_fpl.input.name).as_str());
processed_fpl.sort_by_provider_ordinal();
playlist_resolve_series(&ctx.config, &ctx.client, target, errors, &pipe, provider_fpl, &mut processed_fpl).await;
log_memory_snapshot(format!("target '{}' input '{}' after_series_resolve", target.name, provider_fpl.input.name).as_str());
playlist_resolve_vod(&ctx.config, &ctx.client, target, errors, provider_fpl, &mut processed_fpl).await;
log_memory_snapshot(format!("target '{}' input '{}' after_vod_resolve", target.name, provider_fpl.input.name).as_str());
// stats
let input_entry_name = processed_fpl.input.name.clone();
let group_count = processed_fpl.get_group_count();
@@ -734,15 +766,19 @@ async fn process_playlist_for_target(ctx: &PlaylistProcessingContext,
stat.processed_stats.group_count = group_count;
stat.processed_stats.channel_count = channel_count;
}
processed_fetched_playlists.push(processed_fpl);
process_playlist_epg(&mut processed_fpl, &mut new_epg).await;
log_memory_snapshot(format!("target '{}' input '{}' after_epg_apply", target.name, processed_fpl.input.name).as_str());
new_playlist.extend(processed_fpl.source.take_groups());
log_memory_snapshot(format!("target '{}' input '{}' after_take_groups", target.name, processed_fpl.input.name).as_str());
tokio::task::yield_now().await;
}
step.tick("filter rename map");
let (new_epg, mut new_playlist) = process_epg(&mut processed_fetched_playlists).await;
step.tick("epg");
step.tick("filter rename map + epg");
log_memory_snapshot(format!("target '{}' after_filter_rename_map_epg", target.name).as_str());
if target.favourites.is_some() {
step.broadcast("Processing favourites for '{}' playlist", &target.name);
process_favourites(&mut new_playlist, target.favourites.as_deref());
log_memory_snapshot(format!("target '{}' after_favourites", target.name).as_str());
}
if new_playlist.is_empty() {
@@ -753,24 +789,31 @@ async fn process_playlist_for_target(ctx: &PlaylistProcessingContext,
// Process Trakt categories
if trakt_playlist(&ctx.client, target, errors, &mut new_playlist).await {
step.tick("trakt categories");
log_memory_snapshot(format!("target '{}' after_trakt", target.name).as_str());
}
let mut flat_new_playlist = flatten_groups(new_playlist);
step.tick("playlist merge");
log_memory_snapshot(format!("target '{}' after_playlist_merge", target.name).as_str());
if sort_playlist(target, &mut flat_new_playlist) {
step.tick("playlist sort");
log_memory_snapshot(format!("target '{}' after_playlist_sort", target.name).as_str());
}
assign_channel_no_playlist(&mut flat_new_playlist);
step.tick("assigning channel numbers");
log_memory_snapshot(format!("target '{}' after_assign_channel_numbers", target.name).as_str());
map_playlist_counter(target, &mut flat_new_playlist);
step.tick("assigning channel counter");
log_memory_snapshot(format!("target '{}' after_assign_channel_counter", target.name).as_str());
if process_watch(&ctx.config, &ctx.client, target, &flat_new_playlist).await {
step.tick("group watches");
log_memory_snapshot(format!("target '{}' after_group_watches", target.name).as_str());
}
let result = persist_playlist(&ctx.config, &mut flat_new_playlist, flatten_tvguide(new_epg).as_ref(), target, ctx.playlist_state.as_ref()).await;
step.stop("Persisting playlists");
log_memory_snapshot(format!("target '{}' after_persist", target.name).as_str());
result
}
}
@@ -832,19 +875,6 @@ async fn trakt_playlist(client: &reqwest::Client, target: &ConfigTarget, errors:
true
}
async fn process_epg(processed_fetched_playlists: &mut Vec<FetchedPlaylist<'_>>) -> (Vec<Epg>, Vec<PlaylistGroup>) {
let mut new_playlist: Vec<PlaylistGroup> = vec![];
let mut new_epg = vec![];
// each fetched playlist can have its own epgl url.
// we need to process each input epg.
for fp in processed_fetched_playlists {
process_playlist_epg(fp, &mut new_epg).await;
new_playlist.extend(fp.source.take_groups());
}
(new_epg, new_playlist)
}
async fn process_watch(app_config: &Arc<AppConfig>, client: &reqwest::Client, target: &ConfigTarget, new_playlist: &[PlaylistGroup]) -> bool {
if let Some(watches) = &target.watch {
if default_as_default().eq_ignore_ascii_case(&target.name) {
@@ -883,6 +913,8 @@ pub async fn exec_processing(client: &reqwest::Client, app_config: Arc<AppConfig
None
};
log_memory_snapshot("exec_processing start");
// Initialize Context
let ctx = PlaylistProcessingContext {
client: client.clone(),
@@ -897,6 +929,7 @@ pub async fn exec_processing(client: &reqwest::Client, app_config: Arc<AppConfig
let start_time = Instant::now();
let (stats, errors) = process_sources(&ctx).await;
log_memory_snapshot("exec_processing after_process_sources");
// log errors
for err in &errors {
error!("{}", err.message);
@@ -927,7 +960,10 @@ pub async fn exec_processing(client: &reqwest::Client, app_config: Arc<AppConfig
if let Some(events) = &event_manager {
events.send_event(EventMessage::PlaylistUpdateProgress("Playlist Update".to_string(), update_finished_message.clone()));
}
log_memory_snapshot("exec_processing before_interner_gc");
debug!("StringInterner GC removed {} strings", interner_gc());
log_memory_snapshot("exec_processing after_interner_gc");
//trim_allocator_after_update();
info!("{update_finished_message}");
}
backend/src/processing/processor/xtream_series.rs
+99 -6
View File
@@ -8,8 +8,9 @@ use crate::repository::{get_input_storage_path, persist_input_series_info_batch,
use log::{error, info, log_enabled, Level};
use shared::error::TuliproxError;
use shared::model::{InputType, PlaylistEntry, SeriesStreamProperties, StreamProperties, XtreamSeriesInfo};
use shared::model::{PlaylistGroup, PlaylistItemType, XtreamCluster};
use shared::model::{PlaylistGroup, PlaylistItemType, UUIDType, XtreamCluster};
use indexmap::IndexMap;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::Instant;
@@ -138,7 +139,7 @@ async fn playlist_resolve_series_info(app_config: &Arc<AppConfig>, client: &reqw
group_series.into_values().collect()
}
#[allow(clippy::too_many_arguments)]
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub async fn playlist_resolve_series(cfg: &Arc<AppConfig>,
client: &reqwest::Client,
target: &ConfigTarget,
@@ -152,14 +153,106 @@ pub async fn playlist_resolve_series(cfg: &Arc<AppConfig>,
provider_fpl.source.release_resources(XtreamCluster::Series);
let series_playlist = playlist_resolve_series_info(cfg, client, errors, processed_fpl, resolve_series, resolve_delay).await;
provider_fpl.source.obtain_resources().await;
if series_playlist.is_empty() { return; }
// Keep input playlist as canonical source:
// - Store resolved series details back into provider_fpl items
// - Rebuild base (unmodified) episodes from provider_fpl metadata only
if provider_fpl.is_memory() {
// original content saved into original list
for plg in &series_playlist {
provider_fpl.update_playlist(plg).await;
let mut resolved_series_by_provider_id: HashMap<u32, SeriesStreamProperties> = HashMap::new();
for pli in processed_fpl.items() {
if pli.header.xtream_cluster != XtreamCluster::Series
|| pli.header.item_type != PlaylistItemType::SeriesInfo
{
continue;
}
let Some(provider_id) = pli.get_provider_id() else { continue; };
if provider_id == 0 {
continue;
}
if let Some(StreamProperties::Series(properties)) = pli.header.additional_properties.as_ref() {
resolved_series_by_provider_id
.entry(provider_id)
.or_insert_with(|| properties.as_ref().clone());
}
}
if !resolved_series_by_provider_id.is_empty() {
let mut existing_uuids: HashSet<UUIDType> = provider_fpl
.items()
.filter(|pli| {
pli.header.xtream_cluster == XtreamCluster::Series
&& pli.header.item_type == PlaylistItemType::Series
})
.map(|pli| pli.get_uuid())
.collect();
let mut base_series_groups: IndexMap<u32, PlaylistGroup> = IndexMap::new();
let input = provider_fpl.input;
for pli in provider_fpl.items_mut() {
if pli.header.xtream_cluster != XtreamCluster::Series
|| pli.header.item_type != PlaylistItemType::SeriesInfo
{
continue;
}
let Some(provider_id) = pli.get_provider_id() else { continue; };
if provider_id == 0 {
continue;
}
if let Some(resolved) = resolved_series_by_provider_id.get(&provider_id) {
pli.header.additional_properties = Some(StreamProperties::Series(Box::new(resolved.clone())));
}
let Some(StreamProperties::Series(series_props)) = pli.header.additional_properties.as_ref() else {
continue;
};
let (category_id, group_title, series_name) = {
let header = &pli.header;
(
header.category_id,
header.group.clone(),
if header.name.is_empty() {
header.title.clone()
} else {
header.name.clone()
},
)
};
if let Some(episodes) = parse_xtream_series_info(
&pli.get_uuid(),
series_props,
&group_title,
&series_name,
input,
) {
let group = base_series_groups.entry(category_id).or_insert_with(|| PlaylistGroup {
id: category_id,
title: group_title,
channels: Vec::new(),
xtream_cluster: XtreamCluster::Series,
});
for episode in episodes {
if existing_uuids.insert(episode.get_uuid()) {
group.channels.push(episode);
}
}
}
}
for group in base_series_groups.values().filter(|g| !g.channels.is_empty()) {
provider_fpl.update_playlist(group).await;
}
}
}
if series_playlist.is_empty() {
return;
}
// run the target processing pipe over newly derived episodes
// run the processing pipe over new items
let mut new_playlist = series_playlist;
for f in pipe {
backend/src/processing/processor/xtream_vod.rs
+25
View File
@@ -8,6 +8,7 @@ use log::{error, info, log_enabled, Level};
use shared::error::TuliproxError;
use shared::model::{InputType, PlaylistEntry, StreamProperties, VideoStreamProperties, XtreamVideoInfo};
use shared::model::{PlaylistItemType, XtreamCluster};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
@@ -41,6 +42,9 @@ pub async fn playlist_resolve_vod(app_config: &Arc<AppConfig>,
let mut last_log_time = Instant::now();
let mut processed_vod_info_count = 0;
let mut batch = Vec::with_capacity(BATCH_SIZE);
let sync_canonical_in_memory = provider_fpl.is_memory();
let mut resolved_vod_by_provider_id: Option<HashMap<u32, VideoStreamProperties>> =
sync_canonical_in_memory.then(HashMap::new);
provider_fpl.source.release_resources(XtreamCluster::Video);
@@ -72,6 +76,9 @@ pub async fn playlist_resolve_vod(app_config: &Arc<AppConfig>,
let video_stream_props = VideoStreamProperties::from_info(&info, pli);
batch.push((provider_id, video_stream_props.clone()));
if let Some(resolved) = resolved_vod_by_provider_id.as_mut() {
resolved.insert(provider_id, video_stream_props.clone());
}
if batch.len() >= BATCH_SIZE {
if let Err(err) = persist_input_vod_info_batch(app_config, &storage_path, XtreamCluster::Video, &input.name, std::mem::take(&mut batch)).await {
error!("Failed to persist batch VOD info: {err}");
@@ -99,6 +106,24 @@ pub async fn playlist_resolve_vod(app_config: &Arc<AppConfig>,
}
}
// Keep canonical input playlist updated for subsequent target processing in memory mode.
if let Some(resolved_vod_by_provider_id) = resolved_vod_by_provider_id {
for source_pli in provider_fpl.items_mut() {
if source_pli.header.xtream_cluster != XtreamCluster::Video
|| source_pli.header.item_type != PlaylistItemType::Video
{
continue;
}
let Some(provider_id) = source_pli.get_provider_id() else { continue; };
if provider_id == 0 {
continue;
}
if let Some(resolved) = resolved_vod_by_provider_id.get(&provider_id) {
source_pli.header.additional_properties = Some(StreamProperties::Video(Box::new(resolved.clone())));
}
}
}
provider_fpl.source.obtain_resources().await;
info!("resolved {processed_vod_info_count}/{vod_info_count} vod info");
}