mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-17 16:50:53 +00:00
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.
This commit is contained in:
SergeantPanda
parent
56199aef81
commit
5d424adbe8
3 changed files with 41 additions and 21 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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 = []
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue