mirror of
https://github.com/dgtlmoon/changedetection.io.git
synced 2026-09-29 08:46:45 +00:00
Realtime UI - Socketio tweaks and refactor (#3220)
This commit is contained in:
dgtlmoon
@@ -0,0 +1,395 @@
|
||||
"""
|
||||
Worker management module for changedetection.io
|
||||
|
||||
Handles asynchronous workers for dynamic worker scaling.
|
||||
Sync worker support has been removed in favor of async-only architecture.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
from loguru import logger
|
||||
|
||||
# Global worker state
|
||||
running_async_tasks = []
|
||||
async_loop = None
|
||||
async_loop_thread = None
|
||||
|
||||
# Track currently processing UUIDs for async workers
|
||||
currently_processing_uuids = set()
|
||||
|
||||
# Configuration - async workers only
|
||||
USE_ASYNC_WORKERS = True
|
||||
|
||||
|
||||
def start_async_event_loop():
|
||||
"""Start a dedicated event loop for async workers in a separate thread"""
|
||||
global async_loop
|
||||
logger.info("Starting async event loop for workers")
|
||||
|
||||
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")
|
||||
|
||||
|
||||
def start_async_workers(n_workers, update_q, notification_q, app, datastore):
|
||||
"""Start the async worker management system"""
|
||||
global async_loop_thread, async_loop, running_async_tasks, currently_processing_uuids
|
||||
|
||||
# Clear any stale UUID tracking state
|
||||
currently_processing_uuids.clear()
|
||||
|
||||
# Start the event loop in a separate thread
|
||||
async_loop_thread = threading.Thread(target=start_async_event_loop, daemon=True)
|
||||
async_loop_thread.start()
|
||||
|
||||
# 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):
|
||||
try:
|
||||
# Use a factory function to create named worker coroutines
|
||||
def create_named_worker(worker_id):
|
||||
async def named_worker():
|
||||
task = asyncio.current_task()
|
||||
if task:
|
||||
task.set_name(f"async-worker-{worker_id}")
|
||||
return await start_single_async_worker(worker_id, update_q, notification_q, app, datastore)
|
||||
return named_worker()
|
||||
|
||||
task_future = asyncio.run_coroutine_threadsafe(create_named_worker(i), 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):
|
||||
"""Start a single async worker with auto-restart capability"""
|
||||
from changedetectionio.async_update_worker import async_update_worker
|
||||
|
||||
# Check if we're in pytest environment - if so, be more gentle with logging
|
||||
import os
|
||||
in_pytest = "pytest" in os.sys.modules or "PYTEST_CURRENT_TEST" in os.environ
|
||||
|
||||
while not app.config.exit.is_set():
|
||||
try:
|
||||
if not in_pytest:
|
||||
logger.info(f"Starting async worker {worker_id}")
|
||||
await async_update_worker(worker_id, update_q, notification_q, app, datastore)
|
||||
# If we reach here, worker exited cleanly
|
||||
if not in_pytest:
|
||||
logger.info(f"Async worker {worker_id} exited cleanly")
|
||||
break
|
||||
except asyncio.CancelledError:
|
||||
# Task was cancelled (normal shutdown)
|
||||
if not in_pytest:
|
||||
logger.info(f"Async worker {worker_id} cancelled")
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"Async worker {worker_id} crashed: {e}")
|
||||
if not in_pytest:
|
||||
logger.info(f"Restarting async worker {worker_id} in 5 seconds...")
|
||||
await asyncio.sleep(5)
|
||||
|
||||
if not in_pytest:
|
||||
logger.info(f"Async worker {worker_id} shutdown complete")
|
||||
|
||||
|
||||
def start_workers(n_workers, update_q, notification_q, app, datastore):
|
||||
"""Start async workers - sync workers are deprecated"""
|
||||
start_async_workers(n_workers, update_q, notification_q, app, datastore)
|
||||
|
||||
|
||||
def add_worker(update_q, notification_q, app, datastore):
|
||||
"""Add a new async worker (for dynamic scaling)"""
|
||||
global running_async_tasks
|
||||
|
||||
if not async_loop:
|
||||
logger.error("Async loop not running, cannot add worker")
|
||||
return False
|
||||
|
||||
worker_id = len(running_async_tasks)
|
||||
logger.info(f"Adding async worker {worker_id}")
|
||||
|
||||
task_future = asyncio.run_coroutine_threadsafe(
|
||||
start_single_async_worker(worker_id, update_q, notification_q, app, datastore), async_loop
|
||||
)
|
||||
running_async_tasks.append(task_future)
|
||||
return True
|
||||
|
||||
|
||||
def remove_worker():
|
||||
"""Remove an async worker (for dynamic scaling)"""
|
||||
global running_async_tasks
|
||||
|
||||
if not running_async_tasks:
|
||||
return False
|
||||
|
||||
# Cancel the last worker
|
||||
task_future = running_async_tasks.pop()
|
||||
task_future.cancel()
|
||||
logger.info(f"Removed async worker, {len(running_async_tasks)} workers remaining")
|
||||
return True
|
||||
|
||||
|
||||
def get_worker_count():
|
||||
"""Get current number of async workers"""
|
||||
return len(running_async_tasks)
|
||||
|
||||
|
||||
def get_running_uuids():
|
||||
"""Get list of UUIDs currently being processed by async workers"""
|
||||
return list(currently_processing_uuids)
|
||||
|
||||
|
||||
def set_uuid_processing(uuid, processing=True):
|
||||
"""Mark a UUID as being processed or completed"""
|
||||
global currently_processing_uuids
|
||||
if processing:
|
||||
currently_processing_uuids.add(uuid)
|
||||
logger.debug(f"Started processing UUID: {uuid}")
|
||||
else:
|
||||
currently_processing_uuids.discard(uuid)
|
||||
logger.debug(f"Finished processing UUID: {uuid}")
|
||||
|
||||
|
||||
def is_watch_running(watch_uuid):
|
||||
"""Check if a specific watch is currently being processed"""
|
||||
return watch_uuid in get_running_uuids()
|
||||
|
||||
|
||||
def queue_item_async_safe(update_q, item):
|
||||
"""Queue an item for async queue processing"""
|
||||
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 or closed for queueing item")
|
||||
|
||||
|
||||
def shutdown_workers():
|
||||
"""Shutdown all async workers fast and aggressively"""
|
||||
global async_loop, async_loop_thread, running_async_tasks
|
||||
|
||||
# Check if we're in pytest environment - if so, be more gentle with logging
|
||||
import os
|
||||
in_pytest = "pytest" in os.sys.modules or "PYTEST_CURRENT_TEST" in os.environ
|
||||
|
||||
if not in_pytest:
|
||||
logger.info("Fast shutdown of async workers initiated...")
|
||||
|
||||
# Cancel all async tasks immediately
|
||||
for task_future in running_async_tasks:
|
||||
if not task_future.done():
|
||||
task_future.cancel()
|
||||
|
||||
# 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
|
||||
|
||||
running_async_tasks.clear()
|
||||
async_loop = None
|
||||
|
||||
# Give async thread minimal time to finish, then continue
|
||||
if async_loop_thread and async_loop_thread.is_alive():
|
||||
async_loop_thread.join(timeout=1.0) # Only 1 second timeout
|
||||
if async_loop_thread.is_alive() and not in_pytest:
|
||||
logger.info("Async thread still running after timeout - continuing with shutdown")
|
||||
async_loop_thread = None
|
||||
|
||||
if not in_pytest:
|
||||
logger.info("Async workers fast shutdown complete")
|
||||
|
||||
|
||||
|
||||
|
||||
def adjust_async_worker_count(new_count, update_q=None, notification_q=None, app=None, datastore=None):
|
||||
"""
|
||||
Dynamically adjust the number of async workers.
|
||||
|
||||
Args:
|
||||
new_count: Target number of workers
|
||||
update_q, notification_q, app, datastore: Required for adding new workers
|
||||
|
||||
Returns:
|
||||
dict: Status of the adjustment operation
|
||||
"""
|
||||
global running_async_tasks
|
||||
|
||||
current_count = get_worker_count()
|
||||
|
||||
if new_count == current_count:
|
||||
return {
|
||||
'status': 'no_change',
|
||||
'message': f'Worker count already at {current_count}',
|
||||
'current_count': current_count
|
||||
}
|
||||
|
||||
if new_count > current_count:
|
||||
# Add workers
|
||||
workers_to_add = new_count - current_count
|
||||
logger.info(f"Adding {workers_to_add} async workers (from {current_count} to {new_count})")
|
||||
|
||||
if not all([update_q, notification_q, app, datastore]):
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': 'Missing required parameters to add workers',
|
||||
'current_count': current_count
|
||||
}
|
||||
|
||||
for i in range(workers_to_add):
|
||||
worker_id = len(running_async_tasks)
|
||||
task_future = asyncio.run_coroutine_threadsafe(
|
||||
start_single_async_worker(worker_id, update_q, notification_q, app, datastore),
|
||||
async_loop
|
||||
)
|
||||
running_async_tasks.append(task_future)
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': f'Added {workers_to_add} workers',
|
||||
'previous_count': current_count,
|
||||
'current_count': new_count
|
||||
}
|
||||
|
||||
else:
|
||||
# Remove workers
|
||||
workers_to_remove = current_count - new_count
|
||||
logger.info(f"Removing {workers_to_remove} async workers (from {current_count} to {new_count})")
|
||||
|
||||
removed_count = 0
|
||||
for _ in range(workers_to_remove):
|
||||
if running_async_tasks:
|
||||
task_future = running_async_tasks.pop()
|
||||
task_future.cancel()
|
||||
# Wait for the task to actually stop
|
||||
try:
|
||||
task_future.result(timeout=5) # 5 second timeout
|
||||
except Exception:
|
||||
pass # Task was cancelled, which is expected
|
||||
removed_count += 1
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': f'Removed {removed_count} workers',
|
||||
'previous_count': current_count,
|
||||
'current_count': current_count - removed_count
|
||||
}
|
||||
|
||||
|
||||
def get_worker_status():
|
||||
"""Get status information about async workers"""
|
||||
return {
|
||||
'worker_type': 'async',
|
||||
'worker_count': get_worker_count(),
|
||||
'running_uuids': get_running_uuids(),
|
||||
'async_loop_running': async_loop is not None,
|
||||
}
|
||||
|
||||
|
||||
def check_worker_health(expected_count, update_q=None, notification_q=None, app=None, datastore=None):
|
||||
"""
|
||||
Check if the expected number of async workers are running and restart any missing ones.
|
||||
|
||||
Args:
|
||||
expected_count: Expected number of workers
|
||||
update_q, notification_q, app, datastore: Required for restarting workers
|
||||
|
||||
Returns:
|
||||
dict: Health check results
|
||||
"""
|
||||
global running_async_tasks
|
||||
|
||||
current_count = get_worker_count()
|
||||
|
||||
if current_count == expected_count:
|
||||
return {
|
||||
'status': 'healthy',
|
||||
'expected_count': expected_count,
|
||||
'actual_count': current_count,
|
||||
'message': f'All {expected_count} async workers running'
|
||||
}
|
||||
|
||||
# Check for crashed async workers
|
||||
dead_workers = []
|
||||
alive_count = 0
|
||||
|
||||
for i, task_future in enumerate(running_async_tasks[:]):
|
||||
if task_future.done():
|
||||
try:
|
||||
result = task_future.result()
|
||||
dead_workers.append(i)
|
||||
logger.warning(f"Async worker {i} completed unexpectedly")
|
||||
except Exception as e:
|
||||
dead_workers.append(i)
|
||||
logger.error(f"Async worker {i} crashed: {e}")
|
||||
else:
|
||||
alive_count += 1
|
||||
|
||||
# Remove dead workers from tracking
|
||||
for i in reversed(dead_workers):
|
||||
if i < len(running_async_tasks):
|
||||
running_async_tasks.pop(i)
|
||||
|
||||
missing_workers = expected_count - alive_count
|
||||
restarted_count = 0
|
||||
|
||||
if missing_workers > 0 and all([update_q, notification_q, app, datastore]):
|
||||
logger.info(f"Restarting {missing_workers} crashed async workers")
|
||||
|
||||
for i in range(missing_workers):
|
||||
worker_id = alive_count + i
|
||||
try:
|
||||
task_future = asyncio.run_coroutine_threadsafe(
|
||||
start_single_async_worker(worker_id, update_q, notification_q, app, datastore),
|
||||
async_loop
|
||||
)
|
||||
running_async_tasks.append(task_future)
|
||||
restarted_count += 1
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to restart worker {worker_id}: {e}")
|
||||
|
||||
return {
|
||||
'status': 'repaired' if restarted_count > 0 else 'degraded',
|
||||
'expected_count': expected_count,
|
||||
'actual_count': alive_count,
|
||||
'dead_workers': len(dead_workers),
|
||||
'restarted_workers': restarted_count,
|
||||
'message': f'Found {len(dead_workers)} dead workers, restarted {restarted_count}'
|
||||
}
|
||||
Reference in New Issue
Block a user