From 5d424adbe809843410012b414b2651e3b528b251 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Jun 2026 14:45:24 -0500 Subject: [PATCH] perf(vod): optimize VOD batch processing and memory management - Introduced `lookup_by_name_year` function to limit database scans for movies and series without TMDB/IMDB IDs, enhancing performance by scoping lookups to current batch names. - Updated `process_movie_batch` and `process_series_batch` to utilize the new lookup function, reducing memory usage and improving efficiency. - Registered `refresh_vod_content` and `batch_refresh_series_episodes` for post-task memory cleanup in Celery, ensuring better resource management during VOD content refreshes. --- CHANGELOG.md | 1 + apps/vod/tasks.py | 57 ++++++++++++++++++++++++++++--------------- dispatcharr/celery.py | 4 ++- 3 files changed, 41 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d04219d2..55481706 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Performance - **XC live refresh releases bulk catalog data sooner during batch processing.** After filtering the single `get_all_live_streams` response, the full provider catalog list is dropped before batch DB work. `process_m3u_batch_direct` (the path XC refreshes use) now runs `gc.collect()` after each batch and clears batch slice references as thread futures complete. Large structures are still `del`'d in the refresh `finally` block before Celery's `task_postrun` runs `cleanup_memory()`. +- **VOD movie/series batch matching no longer scans the full no-ID catalog.** `process_movie_batch` and `process_series_batch` previously loaded every `Movie`/`Series` row without TMDB or IMDB IDs on each 1000-item chunk to resolve name+year duplicates. Lookup is now scoped to the names in the current batch via `lookup_by_name_year()`, which reduces memory and DB time per chunk. `refresh_vod_content` and `batch_refresh_series_episodes` are registered for Celery post-task memory cleanup (one GC pass at task end, not per chunk). - **EPG programme parse now streams through PostgreSQL staging instead of holding the full catalogue in Python.** `parse_programs_for_source` writes parsed rows into a session-scoped temp table in batches (`_EPG_PARSE_BATCH_SIZE=2500`), then atomically swaps them into `ProgramData` with batched `DELETE ... RETURNING` + `INSERT` (`_EPG_SWAP_BATCH_SIZE=5000`) so Postgres never materializes the entire guide in one statement. Peak Celery memory during large XMLTV refreshes drops sharply compared with building a monolithic in-memory list before bulk insert. The byte-offset programme index build is deferred until after the swap completes so index construction no longer competes with parse for memory and I/O. `refresh_epg_data` closes its DB connection in `finally`, and `build_programme_index_task` is registered as a memory-intensive Celery task. SQLite/dev installs keep an in-memory fallback path. - **Pooled PostgreSQL connections now rotate on a bounded lifetime.** psycopg3 client-side cache grows on handles kept open indefinitely by `django-db-geventpool`; recycling uWSGI workers would interrupt live streams. A thin custom backend (`dispatcharr.db.backends.postgresql_psycopg3`) closes and replaces pool connections after `DATABASE_POOL_CONN_MAX_LIFETIME` seconds (default 600, env-overridable; set `0` to disable) on checkout and return while preserving warm-pool reuse within the window. (Fixes #1343) - **Schedules Direct refresh only fetches guide data for mapped channels.** Schedule MD5 checks and schedule downloads now target mapped lineup stations instead of the entire lineup, and schedule MD5 cache for unmapped stations is pruned each refresh. Unmapped lineup entries no longer trigger wasted schedule API calls when their MD5 changes. diff --git a/apps/vod/tasks.py b/apps/vod/tasks.py index 367baff3..66566745 100644 --- a/apps/vod/tasks.py +++ b/apps/vod/tasks.py @@ -16,6 +16,27 @@ import re logger = logging.getLogger(__name__) +def lookup_by_name_year(model, name_year_pairs): + """Return {(name, year): row} for rows without TMDB/IMDB IDs. + + Scoped to the names in this batch instead of scanning the full table. + """ + if not name_year_pairs: + return {} + wanted = set(name_year_pairs) + names = {name for name, _ in wanted} + found = {} + for row in model.objects.filter( + tmdb_id__isnull=True, + imdb_id__isnull=True, + name__in=names, + ): + key = (row.name, row.year) + if key in wanted: + found[key] = row + return found + + @shared_task def refresh_vod_content(account_id): """Refresh VOD content for an M3U account with batch processing for improved performance""" @@ -176,6 +197,7 @@ def refresh_movies(client, account, categories_by_provider, relations, scan_star logger.info(f"Processing movie chunk {chunk_num}/{total_chunks} ({len(chunk)} movies)") process_movie_batch(account, chunk, categories_by_provider, relations, scan_start_time) + del all_movies_data logger.info(f"Completed processing all {total_movies} movies in {total_chunks} chunks") @@ -230,6 +252,7 @@ def refresh_series(client, account, categories_by_provider, relations, scan_star logger.info(f"Processing series chunk {chunk_num}/{total_chunks} ({len(chunk)} series)") process_series_batch(account, chunk, categories_by_provider, relations, scan_start_time) + del all_series_data logger.info(f"Completed processing all {total_series} series in {total_chunks} chunks") @@ -539,10 +562,12 @@ def process_movie_batch(account, batch, categories, relations, scan_start_time=N # Query by name+year for movies without external IDs name_year_keys = [k for k in movie_keys.keys() if k.startswith('name_')] if name_year_keys: - for movie in Movie.objects.filter(tmdb_id__isnull=True, imdb_id__isnull=True): - key = f"name_{movie.name}_{movie.year or 'None'}" - if key in name_year_keys: - existing_movies[key] = movie + name_year_pairs = [ + (movie_keys[k]['props']['name'], movie_keys[k]['props'].get('year')) + for k in name_year_keys + ] + for key_tuple, movie in lookup_by_name_year(Movie, name_year_pairs).items(): + existing_movies[f"name_{key_tuple[0]}_{key_tuple[1] or 'None'}"] = movie # Get existing relations stream_ids = [data['stream_id'] for data in movie_keys.values()] @@ -659,12 +684,7 @@ def process_movie_batch(account, batch, categories, relations, scan_start_time=N existing_by_tmdb = {m.tmdb_id: m for m in Movie.objects.filter(tmdb_id__in=tmdb_ids)} if tmdb_ids else {} existing_by_imdb = {m.imdb_id: m for m in Movie.objects.filter(imdb_id__in=imdb_ids)} if imdb_ids else {} - existing_by_name_year = {} - if name_year_pairs: - for movie in Movie.objects.filter(tmdb_id__isnull=True, imdb_id__isnull=True): - key = (movie.name, movie.year) - if key in name_year_pairs: - existing_by_name_year[key] = movie + existing_by_name_year = lookup_by_name_year(Movie, name_year_pairs) # Check each movie against the bulk query results movies_actually_created = [] @@ -911,10 +931,12 @@ def process_series_batch(account, batch, categories, relations, scan_start_time= # Query by name+year for series without external IDs name_year_keys = [k for k in series_keys.keys() if k.startswith('name_')] if name_year_keys: - for series in Series.objects.filter(tmdb_id__isnull=True, imdb_id__isnull=True): - key = f"name_{series.name}_{series.year or 'None'}" - if key in name_year_keys: - existing_series[key] = series + name_year_pairs = [ + (series_keys[k]['props']['name'], series_keys[k]['props'].get('year')) + for k in name_year_keys + ] + for key_tuple, series in lookup_by_name_year(Series, name_year_pairs).items(): + existing_series[f"name_{key_tuple[0]}_{key_tuple[1] or 'None'}"] = series # Get existing relations series_ids = [data['series_id'] for data in series_keys.values()] @@ -1023,12 +1045,7 @@ def process_series_batch(account, batch, categories, relations, scan_start_time= existing_by_tmdb = {s.tmdb_id: s for s in Series.objects.filter(tmdb_id__in=tmdb_ids)} if tmdb_ids else {} existing_by_imdb = {s.imdb_id: s for s in Series.objects.filter(imdb_id__in=imdb_ids)} if imdb_ids else {} - existing_by_name_year = {} - if name_year_pairs: - for series in Series.objects.filter(tmdb_id__isnull=True, imdb_id__isnull=True): - key = (series.name, series.year) - if key in name_year_pairs: - existing_by_name_year[key] = series + existing_by_name_year = lookup_by_name_year(Series, name_year_pairs) # Check each series against the bulk query results series_actually_created = [] diff --git a/dispatcharr/celery.py b/dispatcharr/celery.py index bdb5e48d..8154b52e 100644 --- a/dispatcharr/celery.py +++ b/dispatcharr/celery.py @@ -98,7 +98,9 @@ def cleanup_task_memory(**kwargs): 'apps.channels.tasks.match_epg_channels', 'apps.channels.tasks.match_selected_channels_epg', 'apps.channels.tasks.match_single_channel_epg', - 'core.tasks.rehash_streams' + 'core.tasks.rehash_streams', + 'apps.vod.tasks.refresh_vod_content', + 'apps.vod.tasks.batch_refresh_series_episodes', ] # Check if this is a memory-intensive task