From 8bfe6a4a4924fd7fce75c437d195df147c520d26 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Fri, 11 Jul 2025 23:48:31 +0100 Subject: [PATCH] feat: add update_backpressure_value method to BackpressureHandler class Co-authored-by: aider (openrouter/openai/o3-mini-high) --- amqp/adapter/backpressure_handler.py | 34 ++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/amqp/adapter/backpressure_handler.py b/amqp/adapter/backpressure_handler.py index 854c2af..53c4eec 100644 --- a/amqp/adapter/backpressure_handler.py +++ b/amqp/adapter/backpressure_handler.py @@ -161,6 +161,40 @@ class BackpressureHandler: """Update the last data message time""" self.last_data_message_time = time.time() + async def update_backpressure_value(self, current: int, maximum: int): + """ + Update the current backpressure value and check for overload conditions. + This method is called by the AMQService.backpressure() method to update + the current parallel executions count and check for overload conditions. + + Args: + current: Current value of the backpressure metric + maximum: Maximum value of the backpressure metric + """ + logging_info( + "Backpressure: Updating backpressure value, current=%s, maximum=%s", + current, + maximum + ) + + # Update the current parallel executions count + self.update_parallel_executions(current) + + # Update the last data message time to indicate activity + self.update_last_data_message_time() + + # Check for overload conditions + await self.check_overload_condition() + + # If maximum has changed, update the parallel workers threshold + if maximum != self.parallel_workers and maximum > 0: + logging_info( + "Backpressure: Updating maximum parallel workers from %s to %s", + self.parallel_workers, + maximum + ) + self.parallel_workers = maximum + def start_backpressure_monitor(self) -> Thread: # Check if CPU monitoring is enabled in configuration cpu_monitoring_enabled = self.config.backpressure.cpu_monitoring_enabled