lee/changedetection.io
1
0
Fork 0
mirror of https://github.com/dgtlmoon/changedetection.io.git synced 2026-09-23 22:06:58 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Remove eventlet go with gevent!

Browse Source
This commit is contained in:
dgtlmoon
2025-06-02 18:25:40 +02:00
parent b4bfd23f98
commit 6866956e67
4 changed files with 142 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changedetectionio/__init__.py
+26 -15
View File
@@ -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")
changedetectionio/realtime/socket_server.py
+58 -134
View File
@@ -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)}")
changedetectionio/worker_handler.py
+54 -22
View File
@@ -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):
requirements.txt
+4 -2
View File
@@ -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