diff --git a/changedetectionio/blueprint/watchlist/__init__.py b/changedetectionio/blueprint/watchlist/__init__.py index e78bc9146..8999b6dd4 100644 --- a/changedetectionio/blueprint/watchlist/__init__.py +++ b/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'), diff --git a/changedetectionio/flask_app.py b/changedetectionio/flask_app.py index 1d64747a5..e3e7417d4 100644 --- a/changedetectionio/flask_app.py +++ b/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) diff --git a/changedetectionio/queue_handlers.py b/changedetectionio/queue_handlers.py index 8ddaee159..519843d57 100644 --- a/changedetectionio/queue_handlers.py +++ b/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: diff --git a/changedetectionio/realtime/socket_server.py b/changedetectionio/realtime/socket_server.py index 4bd180c4d..e4acd0d29 100644 --- a/changedetectionio/realtime/socket_server.py +++ b/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()