mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 00:55:50 +00:00
fix(m3u): fix Celery worker memory leak during M3U/XC refresh
Restructure cleanup in refresh_single_m3u_account so del statements run before gc.collect(), enabling cyclic garbage collection. Add worker_max_memory_per_child as a safety net against memory fragmentation. Closes #1012, Closes #1053
This commit is contained in:
None
parent
17a2ce9215
commit
ce9f7ccbf9
4 changed files with 156 additions and 38 deletions
|
|
@ -1170,14 +1170,12 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
|
|||
|
||||
retval = f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated."
|
||||
|
||||
# Aggressive garbage collection
|
||||
# del streams_to_create, streams_to_update, stream_hashes, existing_streams
|
||||
# from core.utils import cleanup_memory
|
||||
# cleanup_memory(log_usage=True, force_collection=True)
|
||||
|
||||
# Clean up database connections for threading
|
||||
connections.close_all()
|
||||
|
||||
# Free batch data structures (reference-counted deallocation)
|
||||
del streams_to_create, streams_to_update, stream_hashes, existing_streams
|
||||
|
||||
return retval
|
||||
|
||||
|
||||
|
|
@ -2642,6 +2640,16 @@ def refresh_single_m3u_account(account_id):
|
|||
lock_renewer = TaskLockRenewer("refresh_single_m3u_account", account_id)
|
||||
lock_renewer.start()
|
||||
|
||||
try:
|
||||
return _refresh_single_m3u_account_impl(account_id)
|
||||
finally:
|
||||
# Guaranteed cleanup on all exit paths (success, exception, early return)
|
||||
lock_renewer.stop()
|
||||
release_task_lock("refresh_single_m3u_account", account_id)
|
||||
|
||||
|
||||
def _refresh_single_m3u_account_impl(account_id):
|
||||
"""Implementation of M3U account refresh with guaranteed memory cleanup."""
|
||||
# Record start time
|
||||
refresh_start_timestamp = timezone.now() # For the cleanup function
|
||||
start_time = time.time() # For tracking elapsed time as float
|
||||
|
|
@ -2653,8 +2661,6 @@ def refresh_single_m3u_account(account_id):
|
|||
account = M3UAccount.objects.get(id=account_id, is_active=True)
|
||||
if not account.is_active:
|
||||
logger.debug(f"Account {account_id} is not active, skipping.")
|
||||
lock_renewer.stop()
|
||||
release_task_lock("refresh_single_m3u_account", account_id)
|
||||
return
|
||||
|
||||
# Set status to fetching
|
||||
|
|
@ -2683,8 +2689,6 @@ def refresh_single_m3u_account(account_id):
|
|||
else:
|
||||
logger.debug(f"No orphaned task found for M3U account {account_id}")
|
||||
|
||||
lock_renewer.stop()
|
||||
release_task_lock("refresh_single_m3u_account", account_id)
|
||||
return f"M3UAccount with ID={account_id} not found or inactive, task cleaned up"
|
||||
|
||||
# Fetch M3U lines and handle potential issues
|
||||
|
|
@ -2699,6 +2703,7 @@ def refresh_single_m3u_account(account_id):
|
|||
|
||||
extinf_data = data["extinf_data"]
|
||||
groups = data["groups"]
|
||||
del data # Free top-level dict; extinf_data/groups retain their references
|
||||
except json.JSONDecodeError as e:
|
||||
# Handle corrupted JSON file
|
||||
logger.error(
|
||||
|
|
@ -2734,8 +2739,6 @@ def refresh_single_m3u_account(account_id):
|
|||
logger.error(
|
||||
f"Failed to refresh M3U groups for account {account_id}: {result}"
|
||||
)
|
||||
lock_renewer.stop()
|
||||
release_task_lock("refresh_single_m3u_account", account_id)
|
||||
return "Failed to update m3u account - download failed or other error"
|
||||
|
||||
extinf_data, groups = result
|
||||
|
|
@ -2768,8 +2771,6 @@ def refresh_single_m3u_account(account_id):
|
|||
status="error",
|
||||
error=f"Error refreshing M3U groups: {str(e)}",
|
||||
)
|
||||
lock_renewer.stop()
|
||||
release_task_lock("refresh_single_m3u_account", account_id)
|
||||
return "Failed to update m3u account"
|
||||
|
||||
# Only proceed with parsing if we actually have data and no errors were encountered
|
||||
|
|
@ -2792,8 +2793,6 @@ def refresh_single_m3u_account(account_id):
|
|||
status="error",
|
||||
error="No data available for processing",
|
||||
)
|
||||
lock_renewer.stop()
|
||||
release_task_lock("refresh_single_m3u_account", account_id)
|
||||
return "Failed to update m3u account, no data available"
|
||||
|
||||
hash_keys = CoreSettings.get_m3u_hash_key().split(",")
|
||||
|
|
@ -2939,6 +2938,9 @@ def refresh_single_m3u_account(account_id):
|
|||
|
||||
logger.info(f"Processing {len(all_xc_streams)} XC streams in {len(batches)} batches")
|
||||
|
||||
# Free the original list; batches hold independent sliced copies
|
||||
del all_xc_streams
|
||||
|
||||
# Use threading for XC stream processing - now with consistent batch sizes
|
||||
max_workers = min(4, len(batches))
|
||||
logger.debug(f"Using {max_workers} threads for XC stream processing")
|
||||
|
|
@ -3099,32 +3101,38 @@ def refresh_single_m3u_account(account_id):
|
|||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing M3U for account {account_id}: {str(e)}")
|
||||
account.status = M3UAccount.Status.ERROR
|
||||
account.last_message = f"Error processing M3U: {str(e)}"
|
||||
account.save(update_fields=["status", "last_message"])
|
||||
try:
|
||||
account.status = M3UAccount.Status.ERROR
|
||||
account.last_message = f"Error processing M3U: {str(e)}"
|
||||
account.save(update_fields=["status", "last_message"])
|
||||
except Exception:
|
||||
logger.debug(f"Failed to update account {account_id} status during error handling")
|
||||
raise # Re-raise the exception for Celery to handle
|
||||
finally:
|
||||
lock_renewer.stop()
|
||||
release_task_lock("refresh_single_m3u_account", account_id)
|
||||
# Free large data structures regardless of success or failure
|
||||
if 'existing_groups' in locals():
|
||||
del existing_groups
|
||||
if 'extinf_data' in locals():
|
||||
del extinf_data
|
||||
if 'groups' in locals():
|
||||
del groups
|
||||
if 'batches' in locals():
|
||||
del batches
|
||||
if 'all_xc_streams' in locals():
|
||||
del all_xc_streams
|
||||
if 'data' in locals():
|
||||
del data
|
||||
if 'filtered_groups' in locals():
|
||||
del filtered_groups
|
||||
if 'channel_group_relationships' in locals():
|
||||
del channel_group_relationships
|
||||
|
||||
# Aggressive garbage collection
|
||||
# Only delete variables if they exist
|
||||
if 'existing_groups' in locals():
|
||||
del existing_groups
|
||||
if 'extinf_data' in locals():
|
||||
del extinf_data
|
||||
if 'groups' in locals():
|
||||
del groups
|
||||
if 'batches' in locals():
|
||||
del batches
|
||||
|
||||
from core.utils import cleanup_memory
|
||||
|
||||
cleanup_memory(log_usage=True, force_collection=True)
|
||||
|
||||
# Clean up cache file since we've fully processed it
|
||||
if os.path.exists(cache_path):
|
||||
os.remove(cache_path)
|
||||
# Remove cache file after processing (success or failure)
|
||||
cache_path = os.path.join(m3u_dir, f"{account_id}.json")
|
||||
try:
|
||||
os.remove(cache_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return f"Dispatched jobs complete."
|
||||
|
||||
|
|
|
|||
0
apps/m3u/tests/__init__.py
Normal file
0
apps/m3u/tests/__init__.py
Normal file
106
apps/m3u/tests/test_memory_cleanup.py
Normal file
106
apps/m3u/tests/test_memory_cleanup.py
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
"""
|
||||
Tests for memory cleanup behavior in M3U refresh tasks.
|
||||
|
||||
Verifies that database connections are properly closed, task locks are
|
||||
released on all exit paths, and garbage collection runs where expected.
|
||||
|
||||
"""
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from django.test import SimpleTestCase
|
||||
|
||||
from apps.m3u.models import M3UAccount
|
||||
|
||||
|
||||
class ProcessM3UBatchCleanupTests(SimpleTestCase):
|
||||
"""Verify process_m3u_batch_direct cleans up after processing."""
|
||||
|
||||
@patch("apps.m3u.tasks.Stream")
|
||||
@patch("apps.m3u.tasks.M3UAccount")
|
||||
def test_connections_closed_after_batch(self, mock_account_cls, mock_stream_cls):
|
||||
"""Database connections must be closed after batch processing (thread safety)."""
|
||||
from apps.m3u.tasks import process_m3u_batch_direct
|
||||
|
||||
mock_account = MagicMock()
|
||||
mock_account.filters.order_by.return_value = []
|
||||
mock_account_cls.objects.get.return_value = mock_account
|
||||
mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = (
|
||||
[]
|
||||
)
|
||||
mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123")
|
||||
|
||||
with patch("django.db.connections") as mock_connections:
|
||||
process_m3u_batch_direct(1, [], {}, ["name", "url"])
|
||||
mock_connections.close_all.assert_called()
|
||||
|
||||
|
||||
class LockReleaseTests(SimpleTestCase):
|
||||
"""Verify task lock is released on all exit paths."""
|
||||
|
||||
@patch("apps.m3u.tasks.delete_m3u_refresh_task_by_id", return_value=False)
|
||||
def test_lock_released_on_account_not_found(self, mock_delete):
|
||||
"""release_task_lock must be called when account does not exist."""
|
||||
with patch(
|
||||
"apps.m3u.tasks.acquire_task_lock", return_value=True
|
||||
), patch("apps.m3u.tasks.release_task_lock") as mock_release, patch(
|
||||
"apps.m3u.tasks.TaskLockRenewer"
|
||||
):
|
||||
with patch(
|
||||
"apps.m3u.tasks.M3UAccount.objects.get",
|
||||
side_effect=M3UAccount.DoesNotExist,
|
||||
):
|
||||
from apps.m3u.tasks import refresh_single_m3u_account
|
||||
|
||||
refresh_single_m3u_account(99999)
|
||||
|
||||
mock_release.assert_called_once_with(
|
||||
"refresh_single_m3u_account", 99999
|
||||
)
|
||||
|
||||
def test_lock_released_on_exception(self):
|
||||
"""release_task_lock must be called when an exception is raised."""
|
||||
mock_account = MagicMock()
|
||||
mock_account.is_active = True
|
||||
mock_account.account_type = "STD"
|
||||
mock_account.custom_properties = {}
|
||||
mock_account.filters.all.return_value = []
|
||||
mock_account.status = MagicMock()
|
||||
|
||||
with patch(
|
||||
"apps.m3u.tasks.acquire_task_lock", return_value=True
|
||||
), patch("apps.m3u.tasks.release_task_lock") as mock_release, patch(
|
||||
"apps.m3u.tasks.TaskLockRenewer"
|
||||
):
|
||||
with patch(
|
||||
"apps.m3u.tasks.M3UAccount.objects.get", return_value=mock_account
|
||||
):
|
||||
with patch("os.path.exists", return_value=False):
|
||||
with patch(
|
||||
"apps.m3u.tasks.refresh_m3u_groups",
|
||||
side_effect=RuntimeError("test"),
|
||||
):
|
||||
from apps.m3u.tasks import refresh_single_m3u_account
|
||||
|
||||
try:
|
||||
refresh_single_m3u_account(1)
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
mock_release.assert_called_once_with("refresh_single_m3u_account", 1)
|
||||
|
||||
|
||||
class XCCategoryCleanupTests(SimpleTestCase):
|
||||
"""Regression guard: process_xc_category_direct must continue to clean up."""
|
||||
|
||||
@patch("apps.m3u.tasks.XCClient")
|
||||
@patch("apps.m3u.tasks.M3UAccount")
|
||||
def test_xc_category_calls_gc_collect(self, mock_account_cls, mock_xc_client):
|
||||
"""gc.collect() must be called after XC category processing."""
|
||||
from apps.m3u.tasks import process_xc_category_direct
|
||||
|
||||
mock_account = MagicMock()
|
||||
mock_account_cls.objects.get.return_value = mock_account
|
||||
|
||||
with patch("gc.collect") as mock_gc, patch("django.db.connections"):
|
||||
process_xc_category_direct(1, {}, {}, ["name", "url"])
|
||||
mock_gc.assert_called()
|
||||
Loading…
Add table
Add a link
Reference in a new issue