tweaks

changedetectionio/queue_handlers.py

@@ -158,22 +158,21 @@ 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=0.9):
+    async def async_get(self, executor=None, timeout=1.0):
         """
-        Pure async polling - NO executor threads consumed!
+        Efficient async get using executor for blocking call.
 
-        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
+        HYBRID APPROACH: Best of both worlds
+        - Uses run_in_executor for efficient blocking (no polling overhead)
+        - Single timeout (no double-timeout race condition)
+        - Scales well: executor sized to match worker count
 
-        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.
+        With FETCH_WORKERS=10: 10 threads blocked max (acceptable)
+        With FETCH_WORKERS=200: Need executor with 200+ threads (see worker_pool.py)
 
         Args:
-            executor: Ignored (kept for API compatibility with old code)
-            timeout: Maximum time to wait in seconds (default 0.9s)
+            executor: ThreadPoolExecutor (sized to match worker count)
+            timeout: Maximum time to wait in seconds
 
         Returns:
             Item from queue
@@ -183,41 +182,24 @@ class RecheckPriorityQueue:
         """
         logger.trace(f"RecheckQueue.async_get() called, timeout={timeout}")
         import asyncio
+        try:
+            # Use run_in_executor to call sync get efficiently
+            # No outer asyncio.wait_for wrapper = no double timeout issue!
+            loop = asyncio.get_event_loop()
+            item = await loop.run_in_executor(
+                executor,
+                lambda: self.get(block=True, timeout=timeout)
+            )
 
-        end_time = asyncio.get_event_loop().time() + timeout
+            logger.trace(f"RecheckQueue.async_get() successfully retrieved item: {self._get_item_uuid(item)}")
+            return item
 
-        while True:
-            # Try to get item (non-blocking, thread-safe)
-            with self._lock:
-                if self._priority_items:
-                    item = heapq.heappop(self._priority_items)
-
-                    # Drain sync notification queue to keep in sync
-                    try:
-                        self._notification_queue.get_nowait()
-                    except queue.Empty:
-                        pass
-
-                    # Emit signals
-                    try:
-                        self._emit_get_signals()
-                    except Exception as signal_e:
-                        logger.error(f"Failed to emit get signals but item retrieved successfully: {signal_e}")
-
-                    logger.trace(f"RecheckQueue.async_get() successfully retrieved item: {self._get_item_uuid(item)}")
-                    return item
-
-            # 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()
-
-            # 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)
+        except queue.Empty:
+            logger.trace(f"RecheckQueue.async_get() timed out - queue is empty")
+            raise
+        except Exception as e:
+            logger.critical(f"CRITICAL: Failed to async get item from queue: {type(e).__name__}: {str(e)}")
+            raise
     
     # UTILITY METHODS
     def qsize(self) -> int:

changedetectionio/worker.py

@@ -55,12 +55,11 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec
         watch = None
 
         try:
-            # Pure async queue - no executor threads consumed!
-            # With 100-200 workers, this scales perfectly as workers suspend
-            # 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.9)
+            # Efficient blocking via run_in_executor (no polling overhead!)
+            # Worker blocks in threading.Queue.get() which uses Condition.wait()
+            # Executor must be sized to match worker count (see worker_pool.py: 50 threads default)
+            # Single timeout (no double-timeout wrapper) = no race condition
+            queued_item_data = await q.async_get(executor=executor, timeout=1.0)
 
             # CRITICAL: Claim UUID immediately after getting from queue to prevent race condition
             # in wait_for_all_checks() which checks qsize() and running_uuids separately