From b189d7230e321e19f2b94e9d2e9e3d5164bc5494 Mon Sep 17 00:00:00 2001 From: Broque Thomas Date: Mon, 16 Feb 2026 21:51:23 -0800 Subject: [PATCH] Fix file descriptor leak from per-call event loop creation --- utils/async_helpers.py | 35 +++++++++++++++++ web_server.py | 88 ++++++++++++++++++++---------------------- 2 files changed, 77 insertions(+), 46 deletions(-) create mode 100644 utils/async_helpers.py diff --git a/utils/async_helpers.py b/utils/async_helpers.py new file mode 100644 index 00000000..5acb1b7c --- /dev/null +++ b/utils/async_helpers.py @@ -0,0 +1,35 @@ +import asyncio +import threading + +_loop = None +_thread = None +_lock = threading.Lock() + + +def _run_loop(loop): + asyncio.set_event_loop(loop) + loop.run_forever() + + +def _get_loop(): + global _loop, _thread + if _loop is None or _loop.is_closed(): + with _lock: + if _loop is None or _loop.is_closed(): + _loop = asyncio.new_event_loop() + _thread = threading.Thread(target=_run_loop, args=(_loop,), daemon=True) + _thread.start() + return _loop + + +def run_async(coro): + """Drop-in replacement for asyncio.run() that uses a single shared event loop. + + A dedicated daemon thread runs one event loop for the entire process. + Callers submit coroutines and block until the result is ready. + This avoids creating/destroying event loops per call (FD leak) + and works correctly with both long-lived and short-lived threads. + """ + loop = _get_loop() + future = asyncio.run_coroutine_threadsafe(coro, loop) + return future.result() diff --git a/web_server.py b/web_server.py index de9c084f..3a7fcb7d 100644 --- a/web_server.py +++ b/web_server.py @@ -19,6 +19,7 @@ from urllib.parse import urljoin from concurrent.futures import ThreadPoolExecutor, as_completed from flask import Flask, render_template, request, jsonify, redirect, send_file, Response from utils.logging_config import get_logger +from utils.async_helpers import run_async # --- Core Application Imports --- # Import the same core clients and config manager used by the GUI app @@ -337,7 +338,7 @@ def get_cached_transfer_data(): live_transfers_lookup = {} try: # First, get Soulseek downloads from API - transfers_data = asyncio.run(soulseek_client._make_request('GET', 'transfers/downloads')) + transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) if transfers_data: all_transfers = [] for user_data in transfers_data: @@ -354,7 +355,7 @@ def get_cached_transfer_data(): # Also add YouTube downloads (through orchestrator) try: - all_downloads = asyncio.run(soulseek_client.get_all_downloads()) + all_downloads = run_async(soulseek_client.get_all_downloads()) for download in all_downloads: # Only add YouTube downloads (Soulseek ones are already in the lookup) if download.username == 'youtube': @@ -601,7 +602,7 @@ class WebUIDownloadMonitor: live_transfers = {} # Get Soulseek downloads from API - transfers_data = asyncio.run(soulseek_client._make_request('GET', 'transfers/downloads')) + transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) if transfers_data: for user_data in transfers_data: username = user_data.get('username', 'Unknown') @@ -614,7 +615,7 @@ class WebUIDownloadMonitor: # Also get YouTube downloads (through orchestrator) try: - all_downloads = asyncio.run(soulseek_client.get_all_downloads()) + all_downloads = run_async(soulseek_client.get_all_downloads()) for download in all_downloads: # Only add YouTube downloads (Soulseek ones are already in the lookup) if download.username == 'youtube': @@ -687,7 +688,7 @@ class WebUIDownloadMonitor: if username and download_id: try: print(f"๐Ÿšซ Cancelling errored download: {download_id} from {username}") - asyncio.run(soulseek_client.cancel_download(download_id, username, remove=True)) + run_async(soulseek_client.cancel_download(download_id, username, remove=True)) print(f"โœ… Successfully cancelled errored download {download_id}") except Exception as cancel_error: print(f"โš ๏ธ Warning: Failed to cancel errored download {download_id}: {cancel_error}") @@ -784,7 +785,7 @@ class WebUIDownloadMonitor: if username and download_id: try: print(f"๐Ÿšซ Cancelling stuck queued download: {download_id} from {username}") - asyncio.run(soulseek_client.cancel_download(download_id, username, remove=True)) + run_async(soulseek_client.cancel_download(download_id, username, remove=True)) print(f"โœ… Successfully cancelled stuck download {download_id}") except Exception as cancel_error: print(f"โš ๏ธ Warning: Failed to cancel stuck download {download_id}: {cancel_error}") @@ -882,7 +883,7 @@ class WebUIDownloadMonitor: if username and download_id: try: print(f"๐Ÿšซ Cancelling stuck 0% download: {download_id} from {username}") - asyncio.run(soulseek_client.cancel_download(download_id, username, remove=True)) + run_async(soulseek_client.cancel_download(download_id, username, remove=True)) print(f"โœ… Successfully cancelled stuck 0% download {download_id}") except Exception as cancel_error: print(f"โš ๏ธ Warning: Failed to cancel stuck 0% download {download_id}: {cancel_error}") @@ -1681,10 +1682,7 @@ def run_service_test(service, test_config): # Test the orchestrator's configured download source (not just Soulseek) download_mode = config_manager.get('download_source.mode', 'soulseek') - async def check(): - return await soulseek_client.check_connection() - - if asyncio.run(check()): + if run_async(soulseek_client.check_connection()): # Success message based on active mode mode_messages = { 'soulseek': "Successfully connected to slskd.", @@ -2099,9 +2097,7 @@ def get_status(): if current_time - _status_cache_timestamps['soulseek'] > STATUS_CACHE_TTL: soulseek_start = time.time() try: - async def _check_soulseek(): - return await soulseek_client.check_connection() - soulseek_status = asyncio.run(_check_soulseek()) + soulseek_status = run_async(soulseek_client.check_connection()) except Exception: soulseek_status = False soulseek_response_time = (time.time() - soulseek_start) * 1000 @@ -2235,7 +2231,7 @@ def get_system_stats(): # Calculate total download speed from active soulseek transfers total_download_speed = 0.0 try: - transfers_data = asyncio.run(soulseek_client._make_request('GET', 'transfers/downloads')) + transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) if transfers_data: for user_data in transfers_data: if 'directories' in user_data: @@ -3713,7 +3709,7 @@ def search_music(): add_activity_item("๐Ÿ”", "Search Started", f"'{query}'", "Now") try: - tracks, albums = asyncio.run(soulseek_client.search(query)) + tracks, albums = run_async(soulseek_client.search(query)) # Convert to dictionaries for JSON response processed_albums = [] @@ -3917,7 +3913,7 @@ def stream_enhanced_search_track(): try: # Search slskd with timeout - tracks_result, _ = asyncio.run(soulseek_client.search(query, timeout=15)) + tracks_result, _ = run_async(soulseek_client.search(query, timeout=15)) if tracks_result: logger.info(f"โœ… Found {len(tracks_result)} results for query: '{query}'") @@ -3991,7 +3987,7 @@ def start_download(): filename = track_data.get('filename') file_size = track_data.get('size', 0) - download_id = asyncio.run(soulseek_client.download( + download_id = run_async(soulseek_client.download( username, filename, file_size @@ -4040,7 +4036,7 @@ def start_download(): if not username or not filename: return jsonify({"error": "Missing username or filename."}), 400 - download_id = asyncio.run(soulseek_client.download(username, filename, file_size)) + download_id = run_async(soulseek_client.download(username, filename, file_size)) logger.info(f"๐Ÿ“ฅ Download ID returned: {download_id}") if download_id: @@ -4194,7 +4190,7 @@ def get_download_status(): try: global _processed_download_ids - transfers_data = asyncio.run(soulseek_client._make_request('GET', 'transfers/downloads')) + transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) # Don't return early if no Soulseek transfers - YouTube downloads need to be checked too! all_transfers = [] @@ -4254,7 +4250,7 @@ def get_download_status(): transfer_id = file_info.get('id') if transfer_id: try: - asyncio.run(soulseek_client.cancel_download(str(transfer_id), username, remove=True)) + run_async(soulseek_client.cancel_download(str(transfer_id), username, remove=True)) except Exception: pass _orphaned_download_keys.discard(context_key) @@ -4343,7 +4339,7 @@ def get_download_status(): # Also include YouTube downloads in the response try: - all_youtube_downloads = asyncio.run(soulseek_client.get_all_downloads()) + all_youtube_downloads = run_async(soulseek_client.get_all_downloads()) for download in all_youtube_downloads: if download.username == 'youtube': @@ -4426,7 +4422,7 @@ def cancel_download(): try: # Call the same client method the GUI uses - success = asyncio.run(soulseek_client.cancel_download(download_id, username, remove=True)) + success = run_async(soulseek_client.cancel_download(download_id, username, remove=True)) if success: return jsonify({"success": True, "message": "Download cancelled."}) else: @@ -4442,7 +4438,7 @@ def clear_finished_downloads(): """ try: # This single client call handles clearing everything that is no longer active - success = asyncio.run(soulseek_client.clear_all_completed_downloads()) + success = run_async(soulseek_client.clear_all_completed_downloads()) if success: return jsonify({"success": True, "message": "Finished downloads cleared."}) else: @@ -4694,7 +4690,7 @@ def clear_all_searches(): Clear all searches from slskd search history. """ try: - success = asyncio.run(soulseek_client.clear_all_searches()) + success = run_async(soulseek_client.clear_all_searches()) if success: add_activity_item("๐Ÿงน", "Search Cleanup", "All search history cleared manually", "Now") return jsonify({"success": True, "message": "All searches cleared."}) @@ -4714,7 +4710,7 @@ def maintain_search_history(): keep_searches = data.get('keep_searches', 50) trigger_threshold = data.get('trigger_threshold', 200) - success = asyncio.run(soulseek_client.maintain_search_history_with_buffer( + success = run_async(soulseek_client.maintain_search_history_with_buffer( keep_searches=keep_searches, trigger_threshold=trigger_threshold )) if success: @@ -6576,7 +6572,7 @@ def _start_enhanced_album_download(enhanced_tracks, unmatched_tracks, spotify_ar continue # Start download - download_id = asyncio.run(soulseek_client.download(username, filename, size)) + download_id = run_async(soulseek_client.download(username, filename, size)) if download_id: context_key = f"{username}::{extract_filename(filename)}" @@ -6620,7 +6616,7 @@ def _start_enhanced_album_download(enhanced_tracks, unmatched_tracks, spotify_ar if not username or not filename: continue - download_id = asyncio.run(soulseek_client.download(username, filename, size)) + download_id = run_async(soulseek_client.download(username, filename, size)) if download_id: context_key = f"{username}::{extract_filename(filename)}" @@ -6705,7 +6701,7 @@ def _start_album_download_tasks(album_result, spotify_artist, spotify_album): 'disc_number': corrected_meta.get('disc_number', 1) } - download_id = asyncio.run(soulseek_client.download(username, filename, size)) + download_id = run_async(soulseek_client.download(username, filename, size)) if download_id: context_key = f"{username}::{extract_filename(filename)}" @@ -6766,7 +6762,7 @@ def start_matched_download(): if not username or not filename: return jsonify({"success": False, "error": "Missing username or filename"}), 400 - download_id = asyncio.run(soulseek_client.download(username, filename, size)) + download_id = run_async(soulseek_client.download(username, filename, size)) if download_id: context_key = f"{username}::{extract_filename(filename)}" @@ -6826,7 +6822,7 @@ def start_matched_download(): download_payload['title'] = parsed_meta.get('title') or download_payload.get('title') download_payload['artist'] = parsed_meta.get('artist') or download_payload.get('artist') - download_id = asyncio.run(soulseek_client.download(username, filename, size)) + download_id = run_async(soulseek_client.download(username, filename, size)) if download_id: context_key = f"{username}::{extract_filename(filename)}" @@ -9835,7 +9831,7 @@ def _simple_monitor_task(): else: print("๐Ÿ” [Auto Cleanup] Starting scheduled search cleanup...") - success = asyncio.run(soulseek_client.maintain_search_history_with_buffer( + success = run_async(soulseek_client.maintain_search_history_with_buffer( keep_searches=50, trigger_threshold=200 )) if success: @@ -9862,7 +9858,7 @@ def _simple_monitor_task(): break if not has_active_batches: - asyncio.run(soulseek_client.clear_all_completed_downloads()) + run_async(soulseek_client.clear_all_completed_downloads()) print("โœ… [Auto Cleanup] Periodic download cleanup completed") last_download_cleanup = current_time @@ -12512,7 +12508,7 @@ def _process_failed_tracks_to_wishlist_exact(batch_id): # Auto-cleanup: Clear completed downloads from slskd try: logger.info(f"๐Ÿงน [Auto-Cleanup] Clearing completed downloads from slskd after batch {batch_id}") - asyncio.run(soulseek_client.clear_all_completed_downloads()) + run_async(soulseek_client.clear_all_completed_downloads()) logger.info(f"โœ… [Auto-Cleanup] Completed downloads cleared from slskd") except Exception as cleanup_error: logger.warning(f"โš ๏ธ [Auto-Cleanup] Failed to clear completed downloads: {cleanup_error}") @@ -13197,7 +13193,7 @@ def _run_post_processing_worker(task_id, batch_id): # Query the download orchestrator for the status which contains the real file path # CRITICAL FIX: Use the actual download_id designated by the client, not the internal task_id actual_download_id = task.get('download_id') or task_id - status = asyncio.run(soulseek_client.get_download_status(actual_download_id)) + status = run_async(soulseek_client.get_download_status(actual_download_id)) if status and status.file_path: real_path = status.file_path if os.path.exists(real_path): @@ -13590,7 +13586,7 @@ def _download_track_worker(task_id, batch_id=None): try: # Perform search with timeout - tracks_result, _ = asyncio.run(soulseek_client.search(query, timeout=30)) + tracks_result, _ = run_async(soulseek_client.search(query, timeout=30)) print(f"๐Ÿ” [DEBUG] Search completed for task {task_id}, got {len(tracks_result) if tracks_result else 0} results") # CRITICAL: Check cancellation immediately after search returns @@ -13801,7 +13797,7 @@ def _attempt_download_with_candidates(task_id, candidates, track, batch_id=None) # Initiate download print(f"๐Ÿš€ [Modal Worker] Starting download: {username} / {os.path.basename(filename)}") - download_id = asyncio.run(soulseek_client.download(username, filename, size)) + download_id = run_async(soulseek_client.download(username, filename, size)) if download_id: # Store context for post-processing with complete Spotify metadata (GUI PARITY) @@ -13889,7 +13885,7 @@ def _attempt_download_with_candidates(task_id, candidates, track, batch_id=None) print(f"๐Ÿšซ [Modal Worker] Task {task_id} cancelled after download {download_id} started - attempting to cancel download") # Try to cancel the download immediately try: - asyncio.run(soulseek_client.cancel_download(download_id, username, remove=True)) + run_async(soulseek_client.cancel_download(download_id, username, remove=True)) print(f"โœ… Successfully cancelled active download {download_id}") except Exception as cancel_error: print(f"โš ๏ธ Warning: Failed to cancel active download {download_id}: {cancel_error}") @@ -14097,7 +14093,7 @@ def _store_batch_source(batch_id, username, filename): # Access SoulseekClient directly (soulseek_client is DownloadOrchestrator) slsk = soulseek_client.soulseek if hasattr(soulseek_client, 'soulseek') else soulseek_client _sr.info(f"Browsing {username}:{folder_path}...") - files = asyncio.run(slsk.browse_user_directory(username, folder_path)) + files = run_async(slsk.browse_user_directory(username, folder_path)) if not files: _sr.info(f"Browse returned no files for {username}:{folder_path}") return @@ -14557,7 +14553,7 @@ def cancel_download_task(): if download_id and username: try: # This is an async call, so we run it and wait - asyncio.run(soulseek_client.cancel_download(download_id, username, remove=True)) + run_async(soulseek_client.cancel_download(download_id, username, remove=True)) print(f"โœ… Successfully cancelled Soulseek download {download_id} for task {task_id}") except Exception as e: print(f"โš ๏ธ Warning: Failed to cancel download on slskd, but worker already moved on. Error: {e}") @@ -14803,7 +14799,7 @@ def cancel_task_v2(): # Step 1: Always search for real download ID first print(f"๐Ÿ” [Atomic Cancel] Searching slskd transfers for real download ID") try: - all_transfers = asyncio.run(soulseek_client._make_request('GET', 'transfers/downloads')) + all_transfers = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) if all_transfers: # Look through transfers to find matching download for user_data in all_transfers: @@ -14832,7 +14828,7 @@ def cancel_task_v2(): endpoint = f'transfers/downloads/{username}/{real_download_id}?remove=true' print(f"๐ŸŒ [Atomic Cancel] Using slskd web UI format: {endpoint}") - response = asyncio.run(soulseek_client._make_request('DELETE', endpoint)) + response = run_async(soulseek_client._make_request('DELETE', endpoint)) if response is not None: print(f"โœ… [Atomic Cancel] Successfully cancelled with slskd web UI format: {real_download_id}") success = True @@ -14841,14 +14837,14 @@ def cancel_task_v2(): # Fallback: Try without remove parameter endpoint2 = f'transfers/downloads/{username}/{real_download_id}' - response2 = asyncio.run(soulseek_client._make_request('DELETE', endpoint2)) + response2 = run_async(soulseek_client._make_request('DELETE', endpoint2)) if response2 is not None: print(f"โœ… [Atomic Cancel] Successfully cancelled without remove param: {real_download_id}") success = True else: # Final fallback: Try simple format (sync.py style) endpoint3 = f'transfers/downloads/{real_download_id}' - response3 = asyncio.run(soulseek_client._make_request('DELETE', endpoint3)) + response3 = run_async(soulseek_client._make_request('DELETE', endpoint3)) if response3 is not None: print(f"โœ… [Atomic Cancel] Successfully cancelled with simple format: {real_download_id}") success = True @@ -18160,13 +18156,13 @@ def _run_sync_task(playlist_id, playlist_name, tracks_json): sync_start_time = time.time() setup_duration = (sync_start_time - task_start_time) * 1000 print(f"โฑ๏ธ [TIMING] Setup completed at {time.strftime('%H:%M:%S')} (took {setup_duration:.1f}ms)") - print(f"๐Ÿš€ Starting actual sync process with asyncio.run()...") + print(f"๐Ÿš€ Starting actual sync process with run_async()...") # Attach original tracks map to sync_service for wishlist with album images sync_service._original_tracks_map = original_tracks_map # Run the sync (this is a blocking call within this thread) - result = asyncio.run(sync_service.sync_playlist(playlist, download_missing=False)) + result = run_async(sync_service.sync_playlist(playlist, download_missing=False)) sync_duration = (time.time() - sync_start_time) * 1000 total_duration = (time.time() - task_start_time) * 1000