Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-18 00:55:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Fix #861: Prevent M3U endless downloading caused by Redis lock expiry

Browse source 
Large M3U downloads that exceed the 300s Redis lock TTL caused Celery Beat
to re-queue duplicate tasks, creating overlapping downloads that never complete.

Files changed:

- core/utils.py: Add TaskLockRenewer class — a daemon thread that periodically renews Redis lock TTL (every 120s) to prevent expiry during long-running tasks.

- apps/m3u/tasks.py: Apply TaskLockRenewer to refresh_single_m3u_account and refresh_m3u_groups; add HTTP timeout (30s connect, 60s read) to M3U download (the only download path missing one); stream M3U download to disk instead of accumulating in memory; add Celery task time limits (1 hour hard limit).

- apps/epg/tasks.py: Apply TaskLockRenewer to refresh_epg_data and parse_programs_for_tvg_id; add Celery task time limits (30 min / 1 hour).

Verified with a throttled test server: locks renewed at T+120s, T+240s, T+360s; 50,000 streams processed with no duplicate tasks.
This commit is contained in:
None 2026-02-23 00:34:28 -06:00
parent 8babb7d111
commit c841cfc8fc
3 changed files with 264 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

26
apps/epg/tasks.py
View file

@ -24,7 +24,7 @@ from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from .models import EPGSource, EPGData, ProgramData
from core.utils import acquire_task_lock, release_task_lock, send_websocket_update, cleanup_memory, log_system_event
from core.utils import acquire_task_lock, release_task_lock, TaskLockRenewer, send_websocket_update, cleanup_memory, log_system_event
logger = logging.getLogger(__name__)
@ -146,12 +146,15 @@ def refresh_all_epg_data():
return "EPG data refreshed."
@shared_task
@shared_task(time_limit=1800, soft_time_limit=1700)
def refresh_epg_data(source_id):
if not acquire_task_lock('refresh_epg_data', source_id):
logger.debug(f"EPG refresh for {source_id} already running")
return
lock_renewer = TaskLockRenewer('refresh_epg_data', source_id)
lock_renewer.start()
source = None
try:
# Try to get the EPG source
@ -168,6 +171,7 @@ def refresh_epg_data(source_id):
logger.info(f"No orphaned task found for EPG source {source_id}")
# Release the lock and exit
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
@ -176,6 +180,7 @@ def refresh_epg_data(source_id):
# The source exists but is not active, just skip processing
if not source.is_active:
logger.info(f"EPG source {source_id} is not active. Skipping.")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
@ -184,6 +189,7 @@ def refresh_epg_data(source_id):
# Skip refresh for dummy EPG sources - they don't need refreshing
if source.source_type == 'dummy':
logger.info(f"Skipping refresh for dummy EPG source {source.name} (ID: {source_id})")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
gc.collect()
return
@ -194,6 +200,7 @@ def refresh_epg_data(source_id):
fetch_success = fetch_xmltv(source)
if not fetch_success:
logger.error(f"Failed to fetch XMLTV for source {source.name}")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
@ -202,6 +209,7 @@ def refresh_epg_data(source_id):
parse_channels_success = parse_channels_only(source)
if not parse_channels_success:
logger.error(f"Failed to parse channels for source {source.name}")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
@ -234,6 +242,7 @@ def refresh_epg_data(source_id):
source = None
# Force garbage collection before releasing the lock
gc.collect()
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
@ -1126,12 +1135,15 @@ def parse_channels_only(source):
@shared_task
@shared_task(time_limit=3600, soft_time_limit=3500)
def parse_programs_for_tvg_id(epg_id):
if not acquire_task_lock('parse_epg_programs', epg_id):
logger.info(f"Program parse for {epg_id} already in progress, skipping duplicate task")
return "Task already running"
lock_renewer = TaskLockRenewer('parse_epg_programs', epg_id)
lock_renewer.start()
source_file = None
program_parser = None
programs_to_create = []
@ -1161,11 +1173,13 @@ def parse_programs_for_tvg_id(epg_id):
# Skip program parsing for dummy EPG sources - they don't have program data files
if epg_source.source_type == 'dummy':
logger.info(f"Skipping program parsing for dummy EPG source {epg_source.name} (ID: {epg_id})")
lock_renewer.stop()
release_task_lock('parse_epg_programs', epg_id)
return
if not Channel.objects.filter(epg_data=epg).exists():
logger.info(f"No channels matched to EPG {epg.tvg_id}")
lock_renewer.stop()
release_task_lock('parse_epg_programs', epg_id)
return
@ -1207,6 +1221,7 @@ def parse_programs_for_tvg_id(epg_id):
epg_source.last_message = f"Failed to download EPG data, cannot parse programs"
epg_source.save(update_fields=['status', 'last_message'])
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="Failed to download EPG file")
lock_renewer.stop()
release_task_lock('parse_epg_programs', epg_id)
return
@ -1217,6 +1232,7 @@ def parse_programs_for_tvg_id(epg_id):
epg_source.last_message = f"Failed to download EPG data, file missing after download"
epg_source.save(update_fields=['status', 'last_message'])
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="File not found after download")
lock_renewer.stop()
release_task_lock('parse_epg_programs', epg_id)
return
@ -1232,6 +1248,7 @@ def parse_programs_for_tvg_id(epg_id):
epg_source.last_message = f"No URL provided, cannot fetch EPG data"
epg_source.save(update_fields=['status', 'last_message'])
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="No URL provided")
lock_renewer.stop()
release_task_lock('parse_epg_programs', epg_id)
return
@ -1379,7 +1396,7 @@ def parse_programs_for_tvg_id(epg_id):
epg_source = None
# Add comprehensive cleanup before releasing lock
cleanup_memory(log_usage=should_log_memory, force_collection=True)
# Memory tracking after processing
# Memory tracking after processing
if process:
try:
mem_after = process.memory_info().rss / 1024 / 1024
@ -1389,6 +1406,7 @@ def parse_programs_for_tvg_id(epg_id):
process = None
epg = None
programs_processed = None
lock_renewer.stop()
release_task_lock('parse_epg_programs', epg_id)

301
apps/m3u/tasks.py
View file

@ -23,6 +23,7 @@ from core.utils import (
RedisClient,
acquire_task_lock,
release_task_lock,
TaskLockRenewer,
natural_sort_key,
log_system_event,
)
@ -66,7 +67,8 @@ def fetch_m3u_lines(account, use_cache=False):
account.save(update_fields=["status", "last_message"])
response = requests.get(
account.server_url, headers=headers, stream=True
account.server_url, headers=headers, stream=True,
timeout=(30, 60), # 30s connect, 60s read between chunks
)
# Log the actual response details for debugging
@ -126,119 +128,60 @@ def fetch_m3u_lines(account, use_cache=False):
start_time = time.time()
last_update_time = start_time
progress = 0
temp_content = b"" # Store content temporarily to validate before saving
has_content = False
# First, let's collect the content and validate it
send_m3u_update(account.id, "downloading", 0)
for chunk in response.iter_content(chunk_size=8192):
if chunk:
temp_content += chunk
has_content = True
# Stream directly to a temp file to avoid holding the entire
# M3U in memory (large files can be 100MB+, which would use
# ~3x that in RAM in certain approaches).
temp_path = file_path + ".tmp"
try:
send_m3u_update(account.id, "downloading", 0)
with open(temp_path, "wb") as tmp_file:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
tmp_file.write(chunk)
has_content = True
downloaded += len(chunk)
elapsed_time = time.time() - start_time
downloaded += len(chunk)
elapsed_time = time.time() - start_time
# Calculate download speed in KB/s
speed = downloaded / elapsed_time / 1024 # in KB/s
# Calculate download speed in KB/s
speed = downloaded / elapsed_time / 1024 # in KB/s
# Calculate progress percentage
if total_size and total_size > 0:
progress = (downloaded / total_size) * 100
# Calculate progress percentage
if total_size and total_size > 0:
progress = (downloaded / total_size) * 100
# Time remaining (in seconds)
time_remaining = (
(total_size - downloaded) / (speed * 1024)
if speed > 0
else 0
)
current_time = time.time()
if current_time - last_update_time >= 0.5:
last_update_time = current_time
if progress > 0:
# Update the account's last_message with detailed progress info
progress_msg = f"Downloading: {progress:.1f}% - {speed:.1f} KB/s - {time_remaining:.1f}s remaining"
account.last_message = progress_msg
account.save(update_fields=["last_message"])
send_m3u_update(
account.id,
"downloading",
progress,
speed=speed,
elapsed_time=elapsed_time,
time_remaining=time_remaining,
message=progress_msg,
# Time remaining (in seconds)
time_remaining = (
(total_size - downloaded) / (speed * 1024)
if speed > 0
else 0
)
# Check if we actually received any content
logger.info(f"Download completed. Has content: {has_content}, Content length: {len(temp_content)} bytes")
if not has_content or len(temp_content) == 0:
error_msg = f"Server responded successfully (HTTP {response.status_code}) but provided empty M3U file from URL: {account.server_url}"
logger.error(error_msg)
account.status = M3UAccount.Status.ERROR
account.last_message = error_msg
account.save(update_fields=["status", "last_message"])
send_m3u_update(
account.id,
"downloading",
100,
status="error",
error=error_msg,
)
return [], False
current_time = time.time()
if current_time - last_update_time >= 0.5:
last_update_time = current_time
if progress > 0:
# Update the account's last_message with detailed progress info
progress_msg = f"Downloading: {progress:.1f}% - {speed:.1f} KB/s - {time_remaining:.1f}s remaining"
account.last_message = progress_msg
account.save(update_fields=["last_message"])
# Basic validation: check if content looks like an M3U file
try:
content_str = temp_content.decode('utf-8', errors='ignore')
content_lines = content_str.strip().split('\n')
send_m3u_update(
account.id,
"downloading",
progress,
speed=speed,
elapsed_time=elapsed_time,
time_remaining=time_remaining,
message=progress_msg,
)
# Log first few lines for debugging (be careful not to log too much)
preview_lines = content_lines[:5]
logger.info(f"Content preview (first 5 lines): {preview_lines}")
logger.info(f"Total lines in content: {len(content_lines)}")
# Check if it's a valid M3U file (should start with #EXTM3U or contain M3U-like content)
is_valid_m3u = False
# First, check if this looks like an error response disguised as 200 OK
content_lower = content_str.lower()
if any(error_indicator in content_lower for error_indicator in [
'<html', '<!doctype html', 'error', 'not found', '404', '403', '500',
'access denied', 'unauthorized', 'forbidden', 'invalid', 'expired'
]):
logger.warning(f"Content appears to be an error response disguised as HTTP 200: {content_str[:200]!r}")
# Continue with M3U validation, but this gives us a clue
if content_lines and content_lines[0].strip().upper().startswith('#EXTM3U'):
is_valid_m3u = True
logger.info("Content validated as M3U: starts with #EXTM3U")
elif any(line.strip().startswith('#EXTINF:') for line in content_lines):
is_valid_m3u = True
logger.info("Content validated as M3U: contains #EXTINF entries")
elif any(line.strip().startswith('http') for line in content_lines):
# Has HTTP URLs, might be a simple M3U without headers
is_valid_m3u = True
logger.info("Content validated as M3U: contains HTTP URLs")
elif any(line.strip().startswith(('rtsp', 'rtp', 'udp')) for line in content_lines):
# Has RTSP/RTP/UDP URLs, might be a simple M3U without headers
is_valid_m3u = True
logger.info("Content validated as M3U: contains RTSP/RTP/UDP URLs")
if not is_valid_m3u:
# Log what we actually received for debugging
logger.error(f"Invalid M3U content received. First 200 characters: {content_str[:200]!r}")
# Try to provide more specific error messages based on content
if '<html' in content_lower or '<!doctype html' in content_lower:
error_msg = f"Server returned HTML page instead of M3U file from URL: {account.server_url}. This usually indicates an error or authentication issue."
elif 'error' in content_lower or 'not found' in content_lower:
error_msg = f"Server returned an error message instead of M3U file from URL: {account.server_url}. Content: {content_str[:100]}"
elif len(content_str.strip()) == 0:
error_msg = f"Server returned completely empty response from URL: {account.server_url}"
else:
error_msg = f"Server provided invalid M3U content from URL: {account.server_url}. Content does not appear to be a valid M3U file."
# Check if we actually received any content
logger.info(f"Download completed. Has content: {has_content}, Content length: {downloaded} bytes")
if not has_content or downloaded == 0:
error_msg = f"Server responded successfully (HTTP {response.status_code}) but provided empty M3U file from URL: {account.server_url}"
logger.error(error_msg)
account.status = M3UAccount.Status.ERROR
account.last_message = error_msg
@ -252,31 +195,113 @@ def fetch_m3u_lines(account, use_cache=False):
)
return [], False
except UnicodeDecodeError:
logger.error(f"Non-text content received. First 200 bytes: {temp_content[:200]!r}")
error_msg = f"Server provided non-text content from URL: {account.server_url}. Unable to process as M3U file."
logger.error(error_msg)
account.status = M3UAccount.Status.ERROR
account.last_message = error_msg
account.save(update_fields=["status", "last_message"])
send_m3u_update(
account.id,
"downloading",
100,
status="error",
error=error_msg,
)
return [], False
# Validate the file by reading only the first portion from
# disk — no need to load the entire file into memory just
# to check the header.
VALIDATION_READ_SIZE = 32768 # 32KB covers headers comfortably
try:
with open(temp_path, "rb") as vf:
head_bytes = vf.read(VALIDATION_READ_SIZE)
head_str = head_bytes.decode('utf-8', errors='ignore')
head_lines = head_str.strip().split('\n')
# Content is valid, save it to file
with open(file_path, "wb") as file:
file.write(temp_content)
# Count total lines efficiently without loading full file
with open(temp_path, "rb") as vf:
total_lines = sum(1 for _ in vf)
# Final update with 100% progress
final_msg = f"Download complete. Size: {total_size/1024/1024:.2f} MB, Time: {time.time() - start_time:.1f}s"
account.last_message = final_msg
account.save(update_fields=["last_message"])
send_m3u_update(account.id, "downloading", 100, message=final_msg)
# Log first few lines for debugging (be careful not to log too much)
preview_lines = head_lines[:5]
logger.info(f"Content preview (first 5 lines): {preview_lines}")
logger.info(f"Total lines in content: {total_lines}")
# Check if it's a valid M3U file (should start with #EXTM3U or contain M3U-like content)
is_valid_m3u = False
# First, check if this looks like an error response disguised as 200 OK
head_lower = head_str.lower()
if any(error_indicator in head_lower for error_indicator in [
'<html', '<!doctype html', 'error', 'not found', '404', '403', '500',
'access denied', 'unauthorized', 'forbidden', 'invalid', 'expired'
]):
logger.warning(f"Content appears to be an error response disguised as HTTP 200: {head_str[:200]!r}")
# Continue with M3U validation, but this gives us a clue
if head_lines and head_lines[0].strip().upper().startswith('#EXTM3U'):
is_valid_m3u = True
logger.info("Content validated as M3U: starts with #EXTM3U")
elif any(line.strip().startswith('#EXTINF:') for line in head_lines):
is_valid_m3u = True
logger.info("Content validated as M3U: contains #EXTINF entries")
elif any(line.strip().startswith('http') for line in head_lines):
# Has HTTP URLs, might be a simple M3U without headers
is_valid_m3u = True
logger.info("Content validated as M3U: contains HTTP URLs")
elif any(line.strip().startswith(('rtsp', 'rtp', 'udp')) for line in head_lines):
# Has RTSP/RTP/UDP URLs, might be a simple M3U without headers
is_valid_m3u = True
logger.info("Content validated as M3U: contains RTSP/RTP/UDP URLs")
if not is_valid_m3u:
# Log what we actually received for debugging
logger.error(f"Invalid M3U content received. First 200 characters: {head_str[:200]!r}")
# Try to provide more specific error messages based on content
if '<html' in head_lower or '<!doctype html' in head_lower:
error_msg = f"Server returned HTML page instead of M3U file from URL: {account.server_url}. This usually indicates an error or authentication issue."
elif 'error' in head_lower or 'not found' in head_lower:
error_msg = f"Server returned an error message instead of M3U file from URL: {account.server_url}. Content: {head_str[:100]}"
elif len(head_str.strip()) == 0:
error_msg = f"Server returned completely empty response from URL: {account.server_url}"
else:
error_msg = f"Server provided invalid M3U content from URL: {account.server_url}. Content does not appear to be a valid M3U file."
logger.error(error_msg)
account.status = M3UAccount.Status.ERROR
account.last_message = error_msg
account.save(update_fields=["status", "last_message"])
send_m3u_update(
account.id,
"downloading",
100,
status="error",
error=error_msg,
)
return [], False
except UnicodeDecodeError:
with open(temp_path, "rb") as vf:
first_bytes = vf.read(200)
logger.error(f"Non-text content received. First 200 bytes: {first_bytes!r}")
error_msg = f"Server provided non-text content from URL: {account.server_url}. Unable to process as M3U file."
logger.error(error_msg)
account.status = M3UAccount.Status.ERROR
account.last_message = error_msg
account.save(update_fields=["status", "last_message"])
send_m3u_update(
account.id,
"downloading",
100,
status="error",
error=error_msg,
)
return [], False
# Validation passed — promote temp file to final path
os.replace(temp_path, file_path)
# Final update with 100% progress
dl_size = downloaded / 1024 / 1024
final_msg = f"Download complete. Size: {dl_size:.2f} MB, Time: {time.time() - start_time:.1f}s"
account.last_message = final_msg
account.save(update_fields=["last_message"])
send_m3u_update(account.id, "downloading", 100, message=final_msg)
finally:
# Clean up temp file on any failure path
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except OSError:
pass
except requests.exceptions.HTTPError as e:
# Handle HTTP errors specifically with more context
status_code = e.response.status_code if e.response else "unknown"
@ -1210,9 +1235,13 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
if not acquire_task_lock("refresh_m3u_account_groups", account_id):
return f"Task already running for account_id={account_id}.", None
lock_renewer = TaskLockRenewer("refresh_m3u_account_groups", account_id)
lock_renewer.start()
try:
account = M3UAccount.objects.get(id=account_id, is_active=True)
except M3UAccount.DoesNotExist:
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return f"M3UAccount with ID={account_id} not found or inactive.", None
@ -1238,6 +1267,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
send_m3u_update(
account_id, "processing_groups", 100, status="error", error=error_msg
)
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return error_msg, None
@ -1250,6 +1280,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
send_m3u_update(
account_id, "processing_groups", 100, status="error", error=error_msg
)
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return error_msg, None
@ -1359,6 +1390,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
status="error",
error=error_msg,
)
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return error_msg, None
@ -1397,6 +1429,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
status="error",
error=error_msg,
)
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return error_msg, None
@ -1413,6 +1446,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
status="error",
error=error_msg,
)
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return error_msg, None
except Exception as e:
@ -1424,6 +1458,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
send_m3u_update(
account_id, "processing_groups", 100, status="error", error=error_msg
)
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return error_msg, None
else:
@ -1431,6 +1466,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
lines, success = fetch_m3u_lines(account, use_cache)
if not success:
# If fetch failed, don't continue processing
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
return f"Failed to fetch M3U data for account_id={account_id}.", None
@ -1525,6 +1561,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
process_groups(account, groups, scan_start_time)
lock_renewer.stop()
release_task_lock("refresh_m3u_account_groups", account_id)
if not full_refresh:
@ -2593,12 +2630,18 @@ def refresh_account_info(profile_id):
release_task_lock("refresh_account_info", profile_id)
return error_msg
@shared_task
@shared_task(time_limit=3600, soft_time_limit=3500)
def refresh_single_m3u_account(account_id):
"""Splits M3U processing into chunks and dispatches them as parallel tasks."""
if not acquire_task_lock("refresh_single_m3u_account", account_id):
return f"Task already running for account_id={account_id}."
# Keep the lock alive while this long-running task is working.
# Without renewal, the 300s lock TTL can expire during large
# downloads/parses, allowing duplicate tasks to start.
lock_renewer = TaskLockRenewer("refresh_single_m3u_account", account_id)
lock_renewer.start()
# Record start time
refresh_start_timestamp = timezone.now() # For the cleanup function
start_time = time.time() # For tracking elapsed time as float
@ -3055,6 +3098,8 @@ def refresh_single_m3u_account(account_id):
account.last_message = f"Error processing M3U: {str(e)}"
account.save(update_fields=["status", "last_message"])
raise # Re-raise the exception for Celery to handle
finally:
lock_renewer.stop()
release_task_lock("refresh_single_m3u_account", account_id)

69
core/utils.py
View file

@ -222,6 +222,75 @@ def release_task_lock(task_name, id):
# Remove the lock
redis_client.delete(lock_id)
class TaskLockRenewer:
"""Periodically renews a Redis task lock to prevent expiry during long-running tasks.
Use as a context manager after acquiring a lock:
if acquire_task_lock("my_task", task_id):
with TaskLockRenewer("my_task", task_id):
# ... long-running work ...
release_task_lock("my_task", task_id)
A daemon thread extends the lock TTL at regular intervals so that
slow downloads or large parsing jobs don't lose their lock mid-operation.
"""
def __init__(self, task_name, id, ttl=300, renewal_interval=120):
self.task_name = task_name
self.id = id
self.ttl = ttl
self.renewal_interval = renewal_interval
self.lock_id = f"task_lock_{task_name}_{id}"
self._stop_event = threading.Event()
self._thread = None
def _renew_loop(self):
"""Background loop that extends the lock TTL until stopped."""
while not self._stop_event.wait(self.renewal_interval):
try:
redis_client = RedisClient.get_client()
if redis_client.exists(self.lock_id):
redis_client.expire(self.lock_id, self.ttl)
logger.debug(
f"Renewed lock {self.lock_id} TTL to {self.ttl}s"
)
else:
# Lock was deleted externally (e.g. manual release) — stop renewing
logger.warning(
f"Lock {self.lock_id} no longer exists, stopping renewal"
)
break
except Exception as e:
logger.error(f"Error renewing lock {self.lock_id}: {e}")
def start(self):
"""Start the background renewal thread."""
self._stop_event.clear()
self._thread = threading.Thread(
target=self._renew_loop, daemon=True,
name=f"lock-renew-{self.task_name}-{self.id}"
)
self._thread.start()
return self
def stop(self):
"""Stop the renewal thread."""
self._stop_event.set()
if self._thread and self._thread.is_alive():
self._thread.join(timeout=5)
self._thread = None
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop()
return False
def send_websocket_update(group_name, event_type, data, collect_garbage=False):
"""
Standardized function to send WebSocket updates with proper memory management.