From d4bd47fce04a0ea38b9aa0ac6ea772febd425adf Mon Sep 17 00:00:00 2001 From: RickZ <121943721+zhy1658858023@users.noreply.github.com> Date: Mon, 20 Jul 2026 10:18:58 +0800 Subject: [PATCH] feat: add predicted instance pressure to least_load --- proxy/README.md | 15 ++++++++-- proxy/proxy.py | 6 ++-- proxy/queue/manager.py | 52 +++++++++++++++++++++++++++++++-- proxy/resource/instance_pool.py | 5 ++++ proxy/strategy/least_load.py | 23 ++++++++++++--- 5 files changed, 88 insertions(+), 13 deletions(-) diff --git a/proxy/README.md b/proxy/README.md index 90f13b7..da59849 100644 --- a/proxy/README.md +++ b/proxy/README.md @@ -164,13 +164,19 @@ GET /debug/instance_resources GET /debug/instance_loads ``` -Use `/debug/instance_loads` to inspect Proxy-maintained per-Instance `inflight` counters, `qps_1m` when present, queue-depth hints when the queue snapshot provider is available, and the queue-aware `least_load` score used for Instance selection. +Use `/debug/instance_loads` to inspect Proxy-maintained per-Instance `inflight` counters, `qps_1m` when present, queue-pressure hints when the queue snapshot provider is available, and the queue-aware `least_load` score used for Instance selection. ```bash curl -sS http://127.0.0.1:8002/debug/instance_loads | python3 -m json.tool ``` -`least_load` uses a conservative deterministic score. `load.inflight` remains the primary signal with implicit weight `1.0`. When queue hints are available, the strategy also adds per-Instance active prepare work, active ready work, prepare queue depth, and ready queue depth. `qps_1m` from Instance heartbeats is supported but disabled by default with weight `0.0`. Missing metrics remain unavailable; they are not converted to measured zeroes, and the strategy falls back to round-robin if no load signal is known. Ties at the minimum score are also broken by round-robin. +`least_load` uses a conservative deterministic score with three load-signal layers: + +1. Proxy lifecycle pressure: `inflight`. +2. Instantaneous queue pressure: `active_prepare`, `active_ready`, `prepare_queue_depth`, and `ready_queue_depth`. +3. Predicted reservation pressure from QueueManager timelines: `pending_prefill_count`, `active_decode_count`, and `predicted_total_backlog_ms`. + +`load.inflight` remains the primary signal with implicit weight `1.0`. Queue hints add per-Instance active prepare work, active ready work, prepare queue depth, ready queue depth, pending prefill reservations, active decode reservations, and predicted backlog time. `qps_1m` from Instance heartbeats is supported but disabled by default with weight `0.0`. Missing predicted fields remain unavailable and fall back to the instantaneous queue-aware behavior; missing queue hints fall back to inflight-only scoring. If no load signal is known for any candidate, selection falls back to round-robin. Ties at the minimum score are also broken by round-robin. Default `least_load` weights are: @@ -181,9 +187,12 @@ Default `least_load` weights are: | `active_ready` | `1.0` | Proxy queue manager per-Instance snapshot | | `prepare_queue_depth` | `0.25` | Proxy queue manager per-Instance snapshot | | `ready_queue_depth` | `0.25` | Proxy queue manager per-Instance snapshot | +| `pending_prefill_count` | `1.0` | Proxy QueueManager reservation snapshot | +| `active_decode_count` | `0.5` | Proxy QueueManager reservation snapshot | +| `predicted_total_backlog_ms` | `0.001` | Proxy QueueManager reservation snapshot | | `qps_1m` | `0.0` | Instance heartbeat | -The debug response includes `least_load_score.total`, raw `least_load_score.components`, weighted components, metric sources, queue source availability, and the weights used to compute the score. +The debug response includes the raw pressure fields (`pending_prefill_count`, `active_decode_count`, `next_slot_ready_in_ms`, `prefill_free_in_ms`, and `predicted_total_backlog_ms`) plus `least_load_score.total`, raw `least_load_score.components`, weighted components, metric sources, queue source availability, and the weights used to compute the score. The reporting path is: diff --git a/proxy/proxy.py b/proxy/proxy.py index 7899376..82bc87d 100644 --- a/proxy/proxy.py +++ b/proxy/proxy.py @@ -96,7 +96,7 @@ async def _build_proxy_topology_meta() -> Dict[str, Any]: def _build_pool_resource_snapshot(app: FastAPI) -> Dict[str, Any]: pool: InstancePool = app.state.instance_pool # type: ignore - queue_depths = queue_mgr.queue_depth_snapshot() + queue_depths = queue_mgr.queue_pressure_snapshot() return pool.build_pool_resource_snapshot( proxy_id=PROXY_ID, capacity=PROXY_MAX_CAPACITY, @@ -134,7 +134,7 @@ async def lifespan(app: FastAPI): app.state.instance_pool = InstancePool(ttl_s=ttl_s) # type: ignore p_control_plane.set_pool(app.state.instance_pool) # type: ignore p_control_plane.set_pool_resource_context(PROXY_ID, PROXY_MAX_CAPACITY) - p_control_plane.set_queue_snapshot_provider(lambda: queue_mgr.queue_depth_snapshot()) + p_control_plane.set_queue_snapshot_provider(lambda: queue_mgr.queue_pressure_snapshot()) # --- Load the proxy scheduling strategy for the data plane --- strategy_name = os.environ.get("PROXY_INSTANCE_STRATEGY", "round_robin") @@ -453,7 +453,7 @@ def select_instance(app: FastAPI, req_obj: SchedulerRequest): hint = {"request": req_obj} try: - hint["queue_depths"] = queue_mgr.queue_depth_snapshot() + hint["queue_depths"] = queue_mgr.queue_pressure_snapshot() except Exception: logger.warning( "[Proxy] queue snapshot failed during instance select; least_load will use inflight-only metrics", diff --git a/proxy/queue/manager.py b/proxy/queue/manager.py index 1fb7a6b..2e3e2a8 100644 --- a/proxy/queue/manager.py +++ b/proxy/queue/manager.py @@ -99,12 +99,58 @@ def __init__(self) -> None: def queue_depth_snapshot(self) -> Dict[str, Any]: - """Return coarse queue depths without exposing task payloads. + """Return queue pressure without exposing task payloads. - Scheduler-facing pool_resource uses only the global totals. The per-instance - section stays Proxy-local for debugging. + Scheduler-facing pool_resource uses only the global instantaneous totals. + The per-instance section stays Proxy-local for debugging and Instance + selection. It combines instantaneous queue depths with predicted pressure + derived from reservation timelines. + """ + return self.queue_pressure_snapshot() + + def queue_pressure_snapshot(self) -> Dict[str, Any]: + """Return per-instance instantaneous and predicted queue pressure. + + The snapshot intentionally exposes only aggregate numeric fields. It never + includes task payloads, request bodies, prompts, or knowledge content. """ per_instance = self._qmap.snapshot_depths() + now_s = time.time() + + instance_ids = set(per_instance.keys()) + instance_ids.update(self._instance_pending_tasks.keys()) + instance_ids.update(self._instance_decode_tasks.keys()) + instance_ids.update(self._instance_slot_free_ts_s.keys()) + instance_ids.update(self._instance_prefill_free_ts_s.keys()) + + for instance_id in instance_ids: + item = per_instance.setdefault( + instance_id, + { + "prepare_queue_depth": 0, + "ready_queue_depth": 0, + "active_prepare": 0, + "active_ready": 0, + }, + ) + slot_free_ts = self._instance_slot_free_ts_s.get(instance_id) or [] + next_slot_ready_in_ms = 0.0 + if slot_free_ts: + next_slot_ready_in_ms = max(0.0, min(slot_free_ts) - now_s) * 1000.0 + prefill_free_in_ms = max( + 0.0, + float(self._instance_prefill_free_ts_s.get(instance_id, now_s)) - now_s, + ) * 1000.0 + item.update( + { + "pending_prefill_count": int(len(self._instance_pending_tasks.get(instance_id, []))), + "active_decode_count": int(len(self._instance_decode_tasks.get(instance_id, []))), + "next_slot_ready_in_ms": float(next_slot_ready_in_ms), + "prefill_free_in_ms": float(prefill_free_in_ms), + "predicted_total_backlog_ms": float(max(next_slot_ready_in_ms, prefill_free_in_ms)), + } + ) + prepare_total = sum(item["prepare_queue_depth"] for item in per_instance.values()) ready_total = sum(item["ready_queue_depth"] for item in per_instance.values()) return { diff --git a/proxy/resource/instance_pool.py b/proxy/resource/instance_pool.py index d9e695c..dac634a 100644 --- a/proxy/resource/instance_pool.py +++ b/proxy/resource/instance_pool.py @@ -206,6 +206,11 @@ def snapshot_instance_loads( "ready_queue_depth": queue_item.get("ready_queue_depth"), "active_prepare": queue_item.get("active_prepare"), "active_ready": queue_item.get("active_ready"), + "pending_prefill_count": queue_item.get("pending_prefill_count"), + "active_decode_count": queue_item.get("active_decode_count"), + "next_slot_ready_in_ms": queue_item.get("next_slot_ready_in_ms"), + "prefill_free_in_ms": queue_item.get("prefill_free_in_ms"), + "predicted_total_backlog_ms": queue_item.get("predicted_total_backlog_ms"), "least_load_score": least_load_score, }) diff --git a/proxy/strategy/least_load.py b/proxy/strategy/least_load.py index 2df8c96..0cca486 100644 --- a/proxy/strategy/least_load.py +++ b/proxy/strategy/least_load.py @@ -17,6 +17,9 @@ class LeastLoadWeights: active_ready: float = 1.0 prepare_queue_depth: float = 0.25 ready_queue_depth: float = 0.25 + pending_prefill_count: float = 1.0 + active_decode_count: float = 0.5 + predicted_total_backlog_ms: float = 0.001 qps_1m: float = 0.0 @@ -26,9 +29,10 @@ class LeastLoadStrategy(BaseInstanceStrategy): ``load.inflight`` remains the primary signal with implicit weight 1.0. Queue hints are added when the Proxy queue manager provides per-Instance - measurements. Missing metrics stay unknown rather than becoming measured - zeros. If no candidate has any measurable score component, selection falls - back to round-robin. + measurements, including instantaneous queue depth and predicted reservation + pressure. Missing metrics stay unknown rather than becoming measured zeros. + If no candidate has any measurable score component, selection falls back to + round-robin. """ name = "least_load" @@ -63,7 +67,15 @@ def compute_score(self, instance: InstanceLike, hint: Optional[Any] = None) -> D queue_item = self._queue_info(instance, hint) queue_source = "proxy_queue_manager" if queue_item is not None else "unavailable" if queue_item is not None: - for key in ("active_prepare", "active_ready", "prepare_queue_depth", "ready_queue_depth"): + for key in ( + "active_prepare", + "active_ready", + "prepare_queue_depth", + "ready_queue_depth", + "pending_prefill_count", + "active_decode_count", + "predicted_total_backlog_ms", + ): value = self._number_from_mapping(queue_item, key) if value is not None: components[key] = value @@ -92,6 +104,9 @@ def weights_dict(self) -> Dict[str, float]: "active_ready": self.weights.active_ready, "prepare_queue_depth": self.weights.prepare_queue_depth, "ready_queue_depth": self.weights.ready_queue_depth, + "pending_prefill_count": self.weights.pending_prefill_count, + "active_decode_count": self.weights.active_decode_count, + "predicted_total_backlog_ms": self.weights.predicted_total_backlog_ms, "qps_1m": self.weights.qps_1m, }