diff --git a/changedetectionio/__init__.py b/changedetectionio/__init__.py index fcdfaf221..c34ab04b7 100644 --- a/changedetectionio/__init__.py +++ b/changedetectionio/__init__.py @@ -11,11 +11,10 @@ import getopt import platform import signal -import eventlet -# Re-enable eventlet monkey patching now that Playwright is async -eventlet.monkey_patch() - import sys + +# Eventlet completely removed - using threading mode for SocketIO +# This provides better Python 3.12+ compatibility and eliminates eventlet/asyncio conflicts from changedetectionio import store from changedetectionio.flask_app import changedetection_app from loguru import logger @@ -30,22 +29,34 @@ def get_version(): # Parent wrapper or OS sends us a SIGTERM/SIGINT, do everything required for a clean shutdown def sigshutdown_handler(_signo, _stack_frame): name = signal.Signals(_signo).name - logger.critical(f'Shutdown: Got Signal - {name} ({_signo}), Saving DB to disk and calling shutdown') - datastore.sync_to_json() - logger.success('Sync JSON to disk complete.') + logger.critical(f'Shutdown: Got Signal - {name} ({_signo}), Fast shutdown initiated') - # Shutdown socketio server if available + # Set exit flag immediately to stop all loops + app.config.exit.set() + datastore.stop_thread = True + + # Shutdown workers immediately + try: + from changedetectionio import worker_handler + worker_handler.shutdown_workers() + except Exception as e: + logger.error(f"Error shutting down workers: {str(e)}") + + # Shutdown socketio server fast from changedetectionio.flask_app import socketio_server if socketio_server and hasattr(socketio_server, 'shutdown'): try: - logger.info("Shutting down Socket.IO server...") socketio_server.shutdown() except Exception as e: logger.error(f"Error shutting down Socket.IO server: {str(e)}") - # Set flags for clean shutdown - datastore.stop_thread = True - app.config.exit.set() + # Save data quickly + try: + datastore.sync_to_json() + logger.success('Fast sync to disk complete.') + except Exception as e: + logger.error(f"Error syncing to disk: {str(e)}") + sys.exit() def main(): @@ -212,13 +223,13 @@ def main(): # SocketIO instance is already initialized in flask_app.py - # Launch using eventlet SocketIO run method for proper integration (if enabled) + # Launch using SocketIO run method for proper integration (if enabled) if socketio_server: if ssl_mode: socketio.run(app, host=host, port=int(port), debug=False, - certfile='cert.pem', keyfile='privkey.pem') + certfile='cert.pem', keyfile='privkey.pem', allow_unsafe_werkzeug=True) else: - socketio.run(app, host=host, port=int(port), debug=False) + socketio.run(app, host=host, port=int(port), debug=False, allow_unsafe_werkzeug=True) else: # Run Flask app without Socket.IO if disabled logger.info("Starting Flask app without Socket.IO server") diff --git a/changedetectionio/realtime/socket_server.py b/changedetectionio/realtime/socket_server.py index 1546ca104..ef72130ba 100644 --- a/changedetectionio/realtime/socket_server.py +++ b/changedetectionio/realtime/socket_server.py @@ -26,42 +26,14 @@ class SignalHandler: queue_length_signal.connect(self.handle_queue_length, weak=False) # logger.info("SignalHandler: Connected to queue_length signal") - # Create and start the queue update thread - platform specific with manual override - import platform - system = platform.system().lower() - - # Check for manual override via environment variable - force_threading = strtobool(os.getenv('SOCKETIO_FORCE_THREADING', 'False')) - force_eventlet = strtobool(os.getenv('SOCKETIO_FORCE_EVENTLET', 'False')) - - use_threading = force_threading or (system == 'windows' and not force_eventlet) - - if use_threading: - # Use threading mode (Windows default or manual override) - import threading - self.polling_emitter_thread = threading.Thread( - target=self.polling_emit_running_or_queued_watches_threaded, - daemon=True - ) - self.polling_emitter_thread.start() - reason = "manual override" if force_threading else f"{system} default" - logger.info(f"Started polling thread using threading ({reason})") - else: - # Use eventlet on macOS/Linux (or manual override) - try: - import eventlet - self.polling_emitter_thread = eventlet.spawn(self.polling_emit_running_or_queued_watches) - reason = "manual override" if force_eventlet else f"{system} default" - logger.info(f"Started polling thread using eventlet ({reason})") - except ImportError: - # Fallback to threading - import threading - self.polling_emitter_thread = threading.Thread( - target=self.polling_emit_running_or_queued_watches_threaded, - daemon=True - ) - self.polling_emitter_thread.start() - logger.info("Eventlet not available: Started polling thread using threading") + # Create and start the queue update thread using standard threading + import threading + self.polling_emitter_thread = threading.Thread( + target=self.polling_emit_running_or_queued_watches_threaded, + daemon=True + ) + self.polling_emitter_thread.start() + logger.info("Started polling thread using threading (eventlet-free)") # Store the thread reference in socketio for clean shutdown self.socketio_instance.polling_emitter_thread = self.polling_emitter_thread @@ -104,43 +76,6 @@ class SignalHandler: except Exception as e: logger.error(f"Socket.IO error in handle_queue_length: {str(e)}") - def polling_emit_running_or_queued_watches(self): - """Greenlet that periodically updates the browser/frontend with current state of who is being checked or queued - This is because sometimes the browser page could reload (like on clicking on a link) but the data is old - """ - logger.info("Queue update eventlet greenlet started") - - # Import the watch_check_update signal, update_q, and worker_handler here to avoid circular imports - from changedetectionio.flask_app import app - from changedetectionio import worker_handler - watch_check_update = signal('watch_check_update') - - # Use eventlet sleep for non-blocking operation - from eventlet import sleep as eventlet_sleep - - # Get the stop event from the socketio instance - stop_event = self.socketio_instance.stop_event if hasattr(self.socketio_instance, 'stop_event') else None - - # Run until explicitly stopped - while stop_event is None or not stop_event.ready(): - try: - # Get current running UUIDs from async workers - running_uuids = set(worker_handler.get_running_uuids()) - - # Send updates for newly running UUIDs - with app.app_context(): - for uuid in running_uuids: - watch_check_update.send(app_context=app, watch_uuid=uuid) - eventlet_sleep(0.01) # Small yield - - except Exception as e: - logger.error(f"Error in queue update greenlet: {str(e)}") - # Sleep a bit to avoid flooding logs in case of persistent error - eventlet_sleep(0.5) - - eventlet_sleep(10) - - logger.info("Queue update eventlet greenlet stopped") def polling_emit_running_or_queued_watches_threaded(self): """Threading version of polling for Windows compatibility""" @@ -156,8 +91,10 @@ class SignalHandler: # Track previous state to avoid unnecessary emissions previous_running_uuids = set() - # Run until app shutdown - while not getattr(app.config, 'exit', threading.Event()).is_set(): + # Run until app shutdown - check exit flag more frequently for fast shutdown + exit_event = getattr(app.config, 'exit', threading.Event()) + + while not exit_event.is_set(): try: # Get current running UUIDs from async workers running_uuids = set(worker_handler.get_running_uuids()) @@ -166,29 +103,41 @@ class SignalHandler: newly_running = running_uuids - previous_running_uuids no_longer_running = previous_running_uuids - running_uuids - # Send updates for newly running UUIDs + # Send updates for newly running UUIDs (but exit fast if shutdown requested) for uuid in newly_running: + if exit_event.is_set(): + break logger.trace(f"Threading polling: UUID {uuid} started processing") with app.app_context(): watch_check_update.send(app_context=app, watch_uuid=uuid) time.sleep(0.01) # Small yield - # Send updates for UUIDs that finished processing - for uuid in no_longer_running: - logger.trace(f"Threading polling: UUID {uuid} finished processing") - with app.app_context(): - watch_check_update.send(app_context=app, watch_uuid=uuid) - time.sleep(0.01) # Small yield + # Send updates for UUIDs that finished processing (but exit fast if shutdown requested) + if not exit_event.is_set(): + for uuid in no_longer_running: + if exit_event.is_set(): + break + logger.trace(f"Threading polling: UUID {uuid} finished processing") + with app.app_context(): + watch_check_update.send(app_context=app, watch_uuid=uuid) + time.sleep(0.01) # Small yield # Update tracking for next iteration previous_running_uuids = running_uuids - # Sleep between polling cycles - time.sleep(10) # Check every 10 seconds for state changes + # Sleep between polling cycles, but check exit flag every 0.5 seconds for fast shutdown + for _ in range(20): # 20 * 0.5 = 10 seconds total + if exit_event.is_set(): + break + time.sleep(0.5) except Exception as e: logger.error(f"Error in threading polling: {str(e)}") - time.sleep(0.5) + # Even during error recovery, check for exit quickly + for _ in range(1): # 1 * 0.5 = 0.5 seconds + if exit_event.is_set(): + break + time.sleep(0.5) logger.info("Queue update thread stopped (threading mode)") @@ -273,39 +222,25 @@ def init_socketio(app, datastore): # Check for manual override via environment variable force_threading = strtobool(os.getenv('SOCKETIO_FORCE_THREADING', 'False')) - force_eventlet = strtobool(os.getenv('SOCKETIO_FORCE_EVENTLET', 'False')) + force_gevent = strtobool(os.getenv('SOCKETIO_FORCE_GEVENT', 'False')) if force_threading: - # Manual override to threading mode for testing + # Manual override to threading mode async_mode = 'threading' logger.info(f"SOCKETIO_FORCE_THREADING=True: Using {async_mode} mode for Socket.IO (manual override)") - elif force_eventlet: - # Manual override to eventlet mode for testing + elif force_gevent: + # Manual override to gevent mode for testing try: - import eventlet - async_mode = 'eventlet' - logger.info(f"SOCKETIO_FORCE_EVENTLET=True: Using {async_mode} mode for Socket.IO (manual override)") + import gevent + async_mode = 'gevent' + logger.info(f"SOCKETIO_FORCE_GEVENT=True: Using {async_mode} mode for Socket.IO (manual override)") except ImportError: async_mode = 'threading' - logger.warning(f"SOCKETIO_FORCE_EVENTLET=True but eventlet not available, falling back to {async_mode} mode") - elif system == 'windows': - # Windows: Use threading mode for better stability - # Eventlet can be problematic on Windows, especially with newer Python versions - async_mode = 'threading' - logger.info(f"Windows detected: Using {async_mode} mode for Socket.IO (more stable on Windows)") - elif system == 'darwin': # macOS - # macOS: Use eventlet but with fallback to threading - try: - import eventlet - async_mode = 'eventlet' - logger.info(f"macOS detected: Using {async_mode} mode for Socket.IO") - except ImportError: - async_mode = 'threading' - logger.warning(f"macOS: eventlet not available, falling back to {async_mode} mode") + logger.warning(f"SOCKETIO_FORCE_GEVENT=True but gevent not available, falling back to {async_mode} mode") else: - # Linux and others: Use eventlet (most stable) - async_mode = 'eventlet' - logger.info(f"Linux/Unix detected: Using {async_mode} mode for Socket.IO") + # Use threading mode for all platforms - simpler, more reliable, and future-proof + async_mode = 'threading' + logger.info(f"Platform: {system}, Python: {python_version.major}.{python_version.minor} - Using {async_mode} mode for Socket.IO") # Log platform info for debugging logger.info(f"Platform: {system}, Python: {python_version.major}.{python_version.minor}, Socket.IO mode: {async_mode}") @@ -370,38 +305,27 @@ def init_socketio(app, datastore): # Store the datastore reference on the socketio object for later use socketio.datastore = datastore - # Create a stop event for our queue update thread using eventlet Event - import eventlet.event - stop_event = eventlet.event.Event() - socketio.stop_event = stop_event + # No stop event needed for threading mode - threads check app.config.exit directly # Add a shutdown method to the socketio object def shutdown(): - """Shutdown the SocketIO server gracefully""" + """Shutdown the SocketIO server fast and aggressively""" try: - logger.info("Socket.IO: Shutting down server...") + logger.info("Socket.IO: Fast shutdown initiated...") - # Signal the queue update thread to stop - if hasattr(socketio, 'stop_event'): - socketio.stop_event.send() - logger.info("Socket.IO: Signaled queue update thread to stop") - - # Wait for the greenlet to exit (with timeout) + # For threading mode, give the thread a very short time to exit gracefully if hasattr(socketio, 'polling_emitter_thread'): - try: - # For eventlet greenlets - just kill it directly to avoid MAINLOOP issues - if not socketio.polling_emitter_thread.dead: - socketio.polling_emitter_thread.kill() - logger.info("Socket.IO: Queue update eventlet greenlet killed") + if socketio.polling_emitter_thread.is_alive(): + logger.info("Socket.IO: Waiting 1 second for polling thread to stop...") + socketio.polling_emitter_thread.join(timeout=1.0) # Only 1 second timeout + if socketio.polling_emitter_thread.is_alive(): + logger.info("Socket.IO: Polling thread still running after timeout - continuing with shutdown") else: - logger.info("Socket.IO: Queue update eventlet greenlet already dead") - except Exception as e: - logger.error(f"Error killing eventlet greenlet: {str(e)}") + logger.info("Socket.IO: Polling thread stopped quickly") + else: + logger.info("Socket.IO: Polling thread already stopped") - # Close any remaining client connections - # if hasattr(socketio, 'server'): - # socketio.server.disconnect() - logger.info("Socket.IO: Server shutdown complete") + logger.info("Socket.IO: Fast shutdown complete") except Exception as e: logger.error(f"Socket.IO error during shutdown: {str(e)}") diff --git a/changedetectionio/worker_handler.py b/changedetectionio/worker_handler.py index 9d517030c..2ebe5ac9c 100644 --- a/changedetectionio/worker_handler.py +++ b/changedetectionio/worker_handler.py @@ -28,14 +28,23 @@ def start_async_event_loop(): global async_loop logger.info("Starting async event loop for workers") - async_loop = asyncio.new_event_loop() - asyncio.set_event_loop(async_loop) - try: + # Create a new event loop for this thread + async_loop = asyncio.new_event_loop() + # Set it as the event loop for this thread + asyncio.set_event_loop(async_loop) + + logger.debug(f"Event loop created and set: {async_loop}") + + # Run the event loop forever async_loop.run_forever() except Exception as e: logger.error(f"Async event loop error: {e}") finally: + # Clean up + if async_loop and not async_loop.is_closed(): + async_loop.close() + async_loop = None logger.info("Async event loop stopped") @@ -50,16 +59,30 @@ def start_async_workers(n_workers, update_q, notification_q, app, datastore): async_loop_thread = threading.Thread(target=start_async_event_loop, daemon=True) async_loop_thread.start() - # Wait a moment for the loop to start - time.sleep(0.1) + # Wait for the loop to be available (with timeout for safety) + max_wait_time = 5.0 + wait_start = time.time() + while async_loop is None and (time.time() - wait_start) < max_wait_time: + time.sleep(0.1) + + if async_loop is None: + logger.error("Failed to start async event loop within timeout") + return + + # Additional brief wait to ensure loop is running + time.sleep(0.2) # Start async workers logger.info(f"Starting {n_workers} async workers") for i in range(n_workers): - task_future = asyncio.run_coroutine_threadsafe( - start_single_async_worker(i, update_q, notification_q, app, datastore), async_loop - ) - running_async_tasks.append(task_future) + try: + task_future = asyncio.run_coroutine_threadsafe( + start_single_async_worker(i, update_q, notification_q, app, datastore), async_loop + ) + running_async_tasks.append(task_future) + except RuntimeError as e: + logger.error(f"Failed to start async worker {i}: {e}") + continue async def start_single_async_worker(worker_id, update_q, notification_q, app, datastore): @@ -150,35 +173,44 @@ def is_watch_running(watch_uuid): def queue_item_async_safe(update_q, item): """Queue an item for async queue processing""" - if async_loop: - # For async queue, schedule the put operation - asyncio.run_coroutine_threadsafe(update_q.put(item), async_loop) + if async_loop and not async_loop.is_closed(): + try: + # For async queue, schedule the put operation + asyncio.run_coroutine_threadsafe(update_q.put(item), async_loop) + except RuntimeError as e: + logger.error(f"Failed to queue item: {e}") else: - logger.error("Async loop not available for queueing item") + logger.error("Async loop not available or closed for queueing item") def shutdown_workers(): - """Shutdown all async workers gracefully""" + """Shutdown all async workers fast and aggressively""" global async_loop, async_loop_thread, running_async_tasks - logger.info("Shutting down async workers...") + logger.info("Fast shutdown of async workers initiated...") - # Cancel all async tasks + # Cancel all async tasks immediately for task_future in running_async_tasks: task_future.cancel() running_async_tasks.clear() - # Stop the async event loop - if async_loop: - async_loop.call_soon_threadsafe(async_loop.stop) + # Stop the async event loop immediately + if async_loop and not async_loop.is_closed(): + try: + async_loop.call_soon_threadsafe(async_loop.stop) + except RuntimeError: + # Loop might already be stopped + pass async_loop = None - # Wait for the async thread to finish + # Give async thread minimal time to finish, then continue if async_loop_thread and async_loop_thread.is_alive(): - async_loop_thread.join(timeout=5) + async_loop_thread.join(timeout=1.0) # Only 1 second timeout + if async_loop_thread.is_alive(): + logger.info("Async thread still running after timeout - continuing with shutdown") async_loop_thread = None - logger.info("Async workers shutdown complete") + logger.info("Async workers fast shutdown complete") def adjust_async_worker_count(new_count, update_q=None, notification_q=None, app=None, datastore=None): diff --git a/requirements.txt b/requirements.txt index 7a6ea2cf5..ea69b1557 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -eventlet>=0.38.0 +# eventlet>=0.38.0 # Removed - replaced with threading mode for better Python 3.12+ compatibility feedgen~=0.9 flask-compress # 0.6.3 included compatibility fix for werkzeug 3.x (2.x had deprecation of url handlers) @@ -30,7 +30,9 @@ chardet>2.3.0 wtforms~=3.0 jsonpath-ng~=1.5.3 -dnspython==2.6.1 # related to eventlet fixes +# dnspython - Used by paho-mqtt for MQTT broker resolution +# Version pin removed since eventlet (which required the specific 2.6.1 pin) has been eliminated +# paho-mqtt will install compatible dnspython version automatically # jq not available on Windows so must be installed manually