Increase max queue size and many CPU performance fixes

changedetectionio/blueprint/watchlist/__init__.py

@@ -84,7 +84,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
             app_rss_token=datastore.data['settings']['application'].get('rss_access_token'),
             datastore=datastore,
             errored_count=errored_count,
-            extra_classes='has-queue' if len(update_q.queue) else '',
+            extra_classes='has-queue' if not update_q.empty() else '',
             form=form,
             generate_tag_colors=processors.generate_processor_badge_colors,
             guid=datastore.data['app_guid'],
@@ -95,8 +95,8 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
             processor_badge_css=processors.get_processor_badge_css(),
             processor_badge_texts=processors.get_processor_badge_texts(),
             processor_descriptions=processors.get_processor_descriptions(),
-            queue_size=len(update_q.queue),
-            queued_uuids=[q_uuid.item['uuid'] for q_uuid in update_q.queue],
+            queue_size=update_q.qsize(),
+            queued_uuids=update_q.get_queued_uuids(),
             search_q=request.args.get('q', '').strip(),
             sort_attribute=request.args.get('sort') if request.args.get('sort') else request.cookies.get('sort'),
             sort_order=request.args.get('order') if request.args.get('order') else request.cookies.get('order'),

changedetectionio/flask_app.py

@@ -55,7 +55,7 @@ extra_stylesheets = []
 # Use bulletproof janus-based queues for sync/async reliability  
 update_q = RecheckPriorityQueue()
 notification_q = NotificationQueue()
-MAX_QUEUE_SIZE = 2000
+MAX_QUEUE_SIZE = 5000
 
 app = Flask(__name__,
             static_url_path="",
@@ -1005,6 +1005,9 @@ def ticker_thread_check_time_launch_checks():
         # Get a list of watches by UUID that are currently fetching data
         running_uuids = worker_handler.get_running_uuids()
 
+        # Build set of queued UUIDs once for O(1) lookup instead of O(n) per watch
+        queued_uuids = {q_item.item['uuid'] for q_item in update_q.queue}
+
         # Re #232 - Deepcopy the data incase it changes while we're iterating through it all
         watch_uuid_list = []
         while True:
@@ -1081,7 +1084,7 @@ def ticker_thread_check_time_launch_checks():
             seconds_since_last_recheck = now - watch['last_checked']
 
             if seconds_since_last_recheck >= (threshold + watch.jitter_seconds) and seconds_since_last_recheck >= recheck_time_minimum_seconds:
-                if not uuid in running_uuids and uuid not in [q_uuid.item['uuid'] for q_uuid in update_q.queue]:
+                if not uuid in running_uuids and uuid not in queued_uuids:
 
                     # Proxies can be set to have a limit on seconds between which they can be called
                     watch_proxy = datastore.get_preferred_proxy_for_watch(uuid=uuid)

changedetectionio/queue_handlers.py

@@ -176,7 +176,16 @@ class RecheckPriorityQueue:
     def empty(self) -> bool:
         """Check if queue is empty"""
         return self.qsize() == 0
-    
+
+    def get_queued_uuids(self) -> list:
+        """Get list of all queued UUIDs efficiently with single lock"""
+        try:
+            with self._lock:
+                return [item.item['uuid'] for item in self._priority_items if hasattr(item, 'item') and 'uuid' in item.item]
+        except Exception as e:
+            logger.critical(f"CRITICAL: Failed to get queued UUIDs: {str(e)}")
+            return []
+
     def close(self):
         """Close the janus queue"""
         try:

changedetectionio/realtime/socket_server.py

@@ -150,11 +150,8 @@ def handle_watch_update(socketio, **kwargs):
         # Get list of watches that are currently running
         running_uuids = worker_handler.get_running_uuids()
 
-        # Get list of watches in the queue
-        queue_list = []
-        for q_item in update_q.queue:
-            if hasattr(q_item, 'item') and 'uuid' in q_item.item:
-                queue_list.append(q_item.item['uuid'])
+        # Get list of watches in the queue (efficient single-lock method)
+        queue_list = update_q.get_queued_uuids()
 
         # Get the error texts from the watch
         error_texts = watch.compile_error_texts()