diff --git a/changedetectionio/tests/util.py b/changedetectionio/tests/util.py index 7b7ca936b..03e6fa213 100644 --- a/changedetectionio/tests/util.py +++ b/changedetectionio/tests/util.py @@ -164,11 +164,12 @@ def wait_for_all_checks(client=None): if q_length == 0 and not any_workers_busy: if empty_since is None: empty_since = time.time() - elif time.time() - empty_since >= 0.15: # Shorter wait + # Longer stabilization period to ensure async workers have finished all updates + elif time.time() - empty_since >= 0.5: # Increased from 0.15 to 0.5 break else: empty_since = None - + attempt += 1 time.sleep(0.3) diff --git a/changedetectionio/worker_handler.py b/changedetectionio/worker_handler.py index 65b5aca65..8a369dbad 100644 --- a/changedetectionio/worker_handler.py +++ b/changedetectionio/worker_handler.py @@ -18,6 +18,9 @@ worker_threads = [] # List of WorkerThread objects # Track currently processing UUIDs for async workers - maps {uuid: worker_id} currently_processing_uuids = {} +# Track worker states - maps {worker_id: state} where state is 'waiting', 'processing', etc. +worker_states = {} + # Configuration - async workers only USE_ASYNC_WORKERS = True @@ -104,10 +107,11 @@ class WorkerThread: def start_async_workers(n_workers, update_q, notification_q, app, datastore): """Start async workers, each with its own thread and event loop for isolation""" - global worker_threads, currently_processing_uuids + global worker_threads, currently_processing_uuids, worker_states - # Clear any stale UUID tracking state + # Clear any stale state currently_processing_uuids.clear() + worker_states.clear() # Start each worker in its own thread with its own event loop logger.info(f"Starting {n_workers} async workers (isolated threads)") @@ -209,6 +213,25 @@ def set_uuid_processing(uuid, worker_id=None, processing=True): logger.debug(f"Worker {worker_id} finished processing UUID: {uuid}") +def set_worker_state(worker_id, state): + """Set the current state of a worker ('waiting', 'processing', etc.)""" + global worker_states + try: + worker_states[worker_id] = state + logger.trace(f"Worker {worker_id} state: {state}") + except Exception as e: + # Fallback in case logger.trace isn't available + logger.debug(f"Worker {worker_id} state: {state}") + + +def are_all_workers_idle(): + """Check if all workers are in 'waiting' state (idle and waiting for queue items)""" + # Must have states for all workers, otherwise they haven't started yet + if len(worker_states) != len(worker_threads): + return False + return all(state == 'waiting' for state in worker_states.values()) + + def is_watch_running(watch_uuid): """Check if a specific watch is currently being processed by any worker""" return watch_uuid in currently_processing_uuids