Timing tune

changedetectionio/queue_handlers.py

@@ -49,12 +49,9 @@ class RecheckPriorityQueue:
             self._priority_items = []
             self._lock = threading.RLock()
 
-            # Async interface: threading.Event to wake async waiters across multiple event loops
-            # Each worker (in its own thread with own event loop) waits on this shared event
-            # threading.Event works across threads/loops, unlike asyncio.Event (loop-bound)
-            # This allows scaling to 100+ workers without thread pool exhaustion
-            self._async_event = threading.Event()
-            self._async_event_lock = threading.Lock()
+            # No event signaling needed - pure polling approach
+            # Workers check queue every 50ms (latency acceptable: 0-500ms)
+            # Scales to 1000+ workers: each sleeping worker = ~4KB coroutine, not thread
 
             # Signals for UI updates
             self.queue_length_signal = signal('queue_length')
@@ -87,10 +84,6 @@ class RecheckPriorityQueue:
                     heapq.heapify(self._priority_items)
                     raise  # Re-raise to be caught by outer exception handler
 
-            # Signal async waiters (workers) that item is available
-            # This wakes all async workers waiting in async_get() without consuming threads
-            self._signal_async_waiters()
-
             # Signal emission after successful queue - log but don't fail the operation
             # Item is already safely queued, so signal failure shouldn't affect queue state
             try:
@@ -165,20 +158,22 @@ class RecheckPriorityQueue:
             logger.critical(f"CRITICAL: Failed to async put item {self._get_item_uuid(item)}: {str(e)}")
             return False
 
-    async def async_get(self, executor=None, timeout=1.0):
+    async def async_get(self, executor=None, timeout=0.9):
         """
-        Truly async get - NO executor threads consumed while waiting!
+        Pure async polling - NO executor threads consumed!
 
-        SCALABILITY: With 200 workers, this approach uses:
-        - 0 threads while waiting (pure coroutine suspension)
-        - vs run_in_executor which would block 200 threads
+        SCALABILITY: With 2000 workers, this approach uses:
+        - 0 threads while waiting (pure coroutine suspension via asyncio.sleep)
+        - vs run_in_executor which would block 2000 threads = executor exhaustion
 
-        MULTI-LOOP SUPPORT: Uses threading.Event which works across
-        multiple event loops (each worker has its own loop in its own thread).
+        POLLING: Checks queue every 50ms (latency: 0-50ms, acceptable for web monitoring)
+
+        MULTI-LOOP SUPPORT: Works across multiple event loops since each worker
+        has its own loop in its own thread. No event signaling = no loop binding issues.
 
         Args:
-            executor: Ignored (kept for API compatibility)
-            timeout: Maximum time to wait in seconds
+            executor: Ignored (kept for API compatibility with old code)
+            timeout: Maximum time to wait in seconds (default 0.9s)
 
         Returns:
             Item from queue
@@ -189,11 +184,10 @@ class RecheckPriorityQueue:
         logger.trace(f"RecheckQueue.async_get() called, timeout={timeout}")
         import asyncio
 
-        start_time = asyncio.get_event_loop().time()
-        end_time = start_time + timeout
+        end_time = asyncio.get_event_loop().time() + timeout
 
         while True:
-            # Try to get item without blocking
+            # Try to get item (non-blocking, thread-safe)
             with self._lock:
                 if self._priority_items:
                     item = heapq.heappop(self._priority_items)
@@ -213,23 +207,16 @@ class RecheckPriorityQueue:
                     logger.trace(f"RecheckQueue.async_get() successfully retrieved item: {self._get_item_uuid(item)}")
                     return item
 
-            # No item available - check if we should continue waiting
+            # No item - check timeout
             remaining = end_time - asyncio.get_event_loop().time()
             if remaining <= 0:
                 logger.trace(f"RecheckQueue.async_get() timed out - queue is empty")
                 raise queue.Empty()
 
-            # Check if event is signaled (non-blocking, thread-safe)
-            # threading.Event.is_set() works across multiple event loops
-            if self._async_event.is_set():
-                # Event signaled - clear it and loop back to check queue
-                self._async_event.clear()
-                continue
-
-            # No signal yet - sleep briefly and check again
-            # Short sleep (10ms) keeps workers responsive without busy-waiting
-            # 200 workers sleeping = 0 threads blocked (pure coroutine suspension)
-            sleep_time = min(0.01, remaining)  # 10ms or remaining time
+            # Sleep before next check (pure coroutine suspension - no thread blocked!)
+            # 50ms poll interval: workers check queue 20 times/sec
+            # With 2000 workers: 2000 × 4KB = 8MB RAM (vs 2000 threads = 16GB)
+            sleep_time = min(0.05, remaining)  # 50ms or remaining timeout
             await asyncio.sleep(sleep_time)
     
     # UTILITY METHODS
@@ -275,9 +262,6 @@ class RecheckPriorityQueue:
                 if drained > 0:
                     logger.debug(f"Cleared queue: removed {drained} notifications")
 
-            # Clear the async event
-            self._async_event.clear()
-
             return True
         except Exception as e:
             logger.critical(f"CRITICAL: Failed to clear queue: {str(e)}")
@@ -419,16 +403,6 @@ class RecheckPriorityQueue:
             pass
         return 'unknown'
 
-    def _signal_async_waiters(self):
-        """
-        Wake all async workers waiting for items (thread-safe).
-
-        Uses threading.Event which works across multiple event loops.
-        All workers (each in their own thread/loop) check is_set() and wake up.
-        """
-        # Set the threading.Event - thread-safe, works across all event loops
-        self._async_event.set()
-
     def _emit_put_signals(self, item):
         """Emit signals when item is added"""
         try:

changedetectionio/worker.py

@@ -60,7 +60,7 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec
             # as coroutines (not threads) while waiting for items.
             # Short timeout (0.3s) for fast shutdown with zero performance penalty
             # since workers are coroutines, not threads - timeout just reschedules.
-            queued_item_data = await q.async_get(timeout=0.3)
+            queued_item_data = await q.async_get(timeout=0.9)
 
             # CRITICAL: Claim UUID immediately after getting from queue to prevent race condition
             # in wait_for_all_checks() which checks qsize() and running_uuids separately