clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

refactor: update backpressure logic to handle availability correctly

Browse Source 
Co-authored-by: aider (openrouter/openai/o3-mini-high) <aider@aider.chat>
This commit is contained in:
Stanislav Hejny
2025-07-16 18:04:09 +01:00
parent f8fd9f24f8
commit 9c0cb3a334
1 changed files with 28 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amqp/adapter/backpressure_handler.py
+23 -12
View File
@@ -192,20 +192,32 @@ class BackpressureHandler:
async def check_overload_condition(self):
"""
Check if the current parallel executions exceed the limit (OVERLOAD)
or if the service has been idle for too long (IDLE).
Check if the current availability is too low (OVERLOAD)
or if the service has been idle for too long with high availability (IDLE).
Note: current_availability represents available capacity, not used capacity.
- Low availability (close to 0) means OVERLOAD
- High availability (close to maximum) with no activity means IDLE
"""
current_time = time.time()
last_event_delta = current_time - self.last_backpressure_event_time
# If maximum is not set or invalid, use the highest seen value
if self.maximum_load <= 0 and self.current_load > 0:
self.maximum_load = self.current_load
elif self.current_load > self.maximum_load:
# Update maximum if we see a higher value
self.maximum_load = self.current_load
logging_info(
"Backpressure: Check conditions, current=%s, max=%s, last_activity=%s",
"Backpressure: Check conditions, current_availability=%s, max=%s, last_activity=%s",
self.current_load,
self.maximum_load,
last_event_delta,
)
# Check for OVERLOAD condition
if self.current_load >= round(0.8 * self.maximum_load):
# Check for OVERLOAD condition - low availability (less than 20% of maximum)
if self.current_load <= round(0.2 * self.maximum_load):
# Check if the last backpressure event was not an overload
if self.last_backpressure_event != ScalingRequestAlert.OVERLOAD:
# Trigger the overload event
@@ -214,15 +226,14 @@ class BackpressureHandler:
self.last_backpressure_event_time = current_time
self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
# Check for IDLE condition
elif self.current_load == 0:
# Check for IDLE condition - high availability (more than 80% of maximum) with no activity
elif self.current_load >= round(0.8 * self.maximum_load):
idle_duration = self.config.backpressure.idle_duration
# Check if service has been idle for longer than the configured duration
if (
last_event_delta > idle_duration
and self.last_backpressure_event == ScalingRequestAlert.IDLE
):
logging_info(
"Backpressure: Service has been idle for %s seconds (threshold: %s seconds)",
last_event_delta,
@@ -234,7 +245,7 @@ class BackpressureHandler:
# update / reset time-window so that the IDLE is not sent too often
self.last_backpressure_event_time = current_time
else:
# Don't send IDLE right away when the usage drops to zero / below threshold, but wait for a while
# Don't send IDLE right away when the availability increases, but wait for a while
self.last_backpressure_event = ScalingRequestAlert.IDLE
# If neither OVERLOAD nor IDLE, and it's time for an update, send UPDATE event
@@ -243,7 +254,7 @@ class BackpressureHandler:
self.swarm_service_id,
self.swarm_task_id,
self.maximum_load,
self.maximum_load - self.current_load,
self.current_load, # Current availability is passed directly
ScalingRequestAlert.UPDATE,
)
# Address the message to any adapter capable of supporting BACKPRESSURE request
@@ -353,7 +364,7 @@ class BackpressureHandler:
self.swarm_service_id,
self.swarm_task_id,
self.maximum_load,
self.maximum_load - self.current_load,
self.current_load, # Current availability is passed directly
ScalingRequestAlert.OVERLOAD,
)
# Address the message to any management service instance capable of supporting BACKPRESSURE request
@@ -366,7 +377,7 @@ class BackpressureHandler:
self.swarm_service_id,
self.swarm_task_id,
self.maximum_load,
self.maximum_load - self.current_load,
self.current_load, # Current availability is passed directly
ScalingRequestAlert.IDLE,
)
# Address the message to any adapter capable of supporting BACKPRESSURE request