From 2f1019a9685951d5d2bb905c120fa4f4be8b0eb6 Mon Sep 17 00:00:00 2001 From: RickZ <121943721+zhy1658858023@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:49:52 +0800 Subject: [PATCH 1/3] feat(proxy): track instance inflight load --- proxy/README.md | 34 ++++++++++--- proxy/proxy.py | 53 +++++++++++++++---- proxy/resource/instance_pool.py | 84 +++++++++++++++++++++++++++++-- proxy/resource/p_control_plane.py | 16 ++++++ proxy/strategy/factory.py | 27 +++++----- proxy/strategy/least_load.py | 57 +++++++++++++++++++++ 6 files changed, 237 insertions(+), 34 deletions(-) create mode 100644 proxy/strategy/least_load.py diff --git a/proxy/README.md b/proxy/README.md index 179440d..b7dec20 100644 --- a/proxy/README.md +++ b/proxy/README.md @@ -40,6 +40,7 @@ proxy/ ├── strategy/ │ ├── base.py # Base Instance selection strategy │ ├── round_robin.py # Round-robin Instance selection +│ ├── least_load.py # Experimental least-load Instance selection │ ├── least_inflight.py # Reserved for future strategy extension │ └── factory.py # Strategy builder ├── queue/ @@ -83,7 +84,7 @@ Common options: |---|---| | `--host` | Proxy service-plane bind host. The demo also uses it as the advertised host. | | `--port` | Proxy service-plane bind port. The demo also uses it as the advertised port. | -| `--strategy` | Local Instance selection strategy. Currently `round_robin` is the active demo path. | +| `--strategy` | Local Instance selection strategy. Defaults to `round_robin`; use `least_load` for the experimental load-aware selector. | | `--injection-strategy` | Knowledge injection strategy. Use `default` or `iws`. | | `--ready-release-policy` | Ready queue release policy: `ordered` or `text_bypass`. | | `--kdn-links-json` | Optional static KDN topology metadata. | @@ -151,7 +152,7 @@ POST /v1/instance/unregister GET /v1/instance/list?include_dead=true ``` -Instances register static information such as `instance_id`, `host`, `port`, `endpoints`, `tags`, `weight`, and `meta`. Heartbeats refresh `last_seen_at` and can optionally report lightweight load fields such as `inflight`, `qps_1m`, and `gpu_util`. +Instances register static information such as `instance_id`, `host`, `port`, `endpoints`, `tags`, `weight`, and `meta`. Heartbeats refresh `last_seen_at` and can optionally report lightweight non-lifecycle load fields such as `qps_1m` and `gpu_util`; Proxy request lifecycle hooks maintain `inflight`. ### Instance resource snapshots @@ -160,6 +161,7 @@ After PR #87, demo Instances can report host resource snapshots to the Proxy con ```text POST /v1/instance/resource_snapshot GET /debug/instance_resources +GET /debug/instance_loads ``` The reporting path is: @@ -240,11 +242,31 @@ For streaming chat completion, the Proxy forwards the downstream SSE stream and ## Instance selection -The active demo strategy is: +### Proxy Instance selection strategies -| Strategy | Description | -|---|---| -| `round_robin` | Selects alive Instances in round-robin order. | +| Strategy | Aliases | Status | Description | +|---|---|---|---| +| `round_robin` | `round_robin`, `round-robin`, `rr` | Default | Selects alive Instances in round-robin order. | +| `least_load` | `least_load`, `least-load`, `ll` | Experimental | Selects the lowest known Instance load by Proxy-maintained `load.inflight`, then uses `qps_1m` as a secondary signal. Missing metrics remain unknown rather than zero; if all candidates lack usable load metrics, selection falls back to round-robin. | +| `kv_aware` | Planned | Planned | Future strategy intended to consider KVCache locality/inventory together with runtime load. | + +Select `least_load` from the demo CLI: + +```bash +python3 test/demo_proxy.py --strategy least_load +``` + +Or select it through the environment: + +```bash +PROXY_INSTANCE_STRATEGY=least_load python3 test/demo_proxy.py +``` + +The Proxy increments the selected Instance's local inflight counter before enqueueing a request and decrements it when the response path finishes, fails, or is cancelled. Inspect these counters and queue-depth hints with: + +```bash +curl -sS http://127.0.0.1:8002/debug/instance_loads | python3 -m json.tool +``` If no alive Instance is available, the Proxy returns: diff --git a/proxy/proxy.py b/proxy/proxy.py index f19e369..02451c1 100644 --- a/proxy/proxy.py +++ b/proxy/proxy.py @@ -393,7 +393,12 @@ def _sse_meta_event(task: ProxyTask) -> bytes: ).encode("utf-8") -async def _wrap_chat_stream_with_meta(task: ProxyTask, queue_mgr: QueueManager) -> AsyncGenerator[bytes, None]: +async def _wrap_chat_stream_with_meta( + task: ProxyTask, + queue_mgr: QueueManager, + instance_pool: InstancePool, + instance_id: str, +) -> AsyncGenerator[bytes, None]: """ Forward downstream chat SSE, but delay [DONE] and insert one cacheroute_meta event first. """ @@ -423,11 +428,14 @@ async def _wrap_chat_stream_with_meta(task: ProxyTask, queue_mgr: QueueManager) task.trace["stream_exception_ms"] = int(time.time() * 1000) logger.exception("[Proxy] stream wrapper failed rid=%s", task.request_id) finally: - if pending: - yield pending - pending = b"" - yield _sse_meta_event(task) - yield b"data: [DONE]\n\n" + try: + if pending: + yield pending + pending = b"" + yield _sse_meta_event(task) + yield b"data: [DONE]\n\n" + finally: + instance_pool.end_request(instance_id) def select_instance(app: FastAPI, req_obj: SchedulerRequest): @@ -593,8 +601,15 @@ async def proxy_chat_completions(request: FastAPIRequest): # ==================================== # Send into the queue: enqueue -> manager -> forward # ==================================== + instance_pool: InstancePool = proxy.state.instance_pool # type: ignore + request_counted = instance_pool.begin_request(chosen.instance_id) + if not request_counted: + return JSONResponse( + status_code=503, + content={"error": "no_instance", "detail": "selected instance is no longer registered"}, + ) try: - # 1) Wrap the task (note: chosen comes from RR and has instance_id/host/port fields) + # 1) Wrap the task (note: chosen comes from strategy and has instance_id/host/port fields) task = ProxyTask( request_id=getattr(req_obj, "Request_ID", None), req_obj=req_obj, @@ -611,10 +626,12 @@ async def proxy_chat_completions(request: FastAPIRequest): await queue_mgr.enqueue_prepare(task) - stream_gen = _wrap_chat_stream_with_meta(task, queue_mgr) + stream_gen = _wrap_chat_stream_with_meta(task, queue_mgr, instance_pool, chosen.instance_id) return StreamingResponse(stream_gen, media_type="text/event-stream") except Exception as e: + if request_counted: + instance_pool.end_request(chosen.instance_id) logger.exception("[Proxy] failed to call Worker(chat)") return JSONResponse( status_code=502, @@ -762,6 +779,13 @@ async def proxy_completions(request: FastAPIRequest): # ================================== # Send into the queue: enqueue -> drain -> forward # ================================== + instance_pool: InstancePool = proxy.state.instance_pool # type: ignore + request_counted = instance_pool.begin_request(chosen.instance_id) + if not request_counted: + return JSONResponse( + status_code=503, + content={"error": "no_instance", "detail": "selected instance is no longer registered"}, + ) try: task = ProxyTask( request_id=getattr(req_obj, "Request_ID", None), @@ -780,9 +804,14 @@ async def proxy_completions(request: FastAPIRequest): await queue_mgr.enqueue_prepare(task) content_bytes = b"" - async for chunk in queue_mgr.iter_response(task): - if chunk: - content_bytes += chunk + try: + async for chunk in queue_mgr.iter_response(task): + if chunk: + content_bytes += chunk + finally: + if request_counted: + instance_pool.end_request(chosen.instance_id) + request_counted = False # completions is non-streaming: the worker should return a one-shot JSON response if not content_bytes: @@ -806,6 +835,8 @@ async def proxy_completions(request: FastAPIRequest): ) except Exception as e: + if request_counted: + instance_pool.end_request(chosen.instance_id) logger.exception("[Proxy] failed to call Worker(completions)") return JSONResponse( status_code=502, diff --git a/proxy/resource/instance_pool.py b/proxy/resource/instance_pool.py index 25b8921..61bf0d3 100644 --- a/proxy/resource/instance_pool.py +++ b/proxy/resource/instance_pool.py @@ -57,7 +57,7 @@ class InstancePool: """ In-memory instance pool (TTL based). - upsert: register/update instance static fields - - heartbeat: refresh last_seen and optionally update load fields + - heartbeat: refresh last_seen and optionally update non-lifecycle load fields - list(include_dead=False): returns alive instances by default """ def __init__(self, ttl_s: int = 30): @@ -120,9 +120,8 @@ def heartbeat( if not it: return False it.last_seen_at = now - # Only update when provided to avoid overwriting with 0/None - if inflight is not None: - it.load.inflight = int(inflight) + # Inflight is Proxy-maintained via begin_request/end_request. + # Keep heartbeat load updates for non-lifecycle signals only. if qps_1m is not None: it.load.qps_1m = float(qps_1m) if gpu_util is not None: @@ -145,6 +144,81 @@ def report_resource_snapshot( return True + def begin_request(self, instance_id: str) -> bool: + """Increment the Proxy-maintained inflight counter for an Instance.""" + with self._lock: + it = self._items.get(instance_id) + if not it: + return False + current = int(it.load.inflight or 0) + it.load.inflight = current + 1 + return True + + def end_request(self, instance_id: str) -> bool: + """Decrement the Proxy-maintained inflight counter without going below zero.""" + with self._lock: + it = self._items.get(instance_id) + if not it: + return False + current = int(it.load.inflight or 0) + it.load.inflight = max(0, current - 1) + return True + + def snapshot_instance_loads( + self, + queue_depths: Optional[Dict[str, Any]] = None, + include_dead: bool = True, + ) -> Dict[str, Any]: + """Return per-Instance load inputs used by Proxy-local strategies and debug views.""" + now = int(time.time()) + per_instance_queue = {} + if queue_depths: + raw = queue_depths.get("per_instance", {}) + if isinstance(raw, dict): + per_instance_queue = raw + + with self._lock: + items = list(self._items.values()) + + instances: List[Dict[str, Any]] = [] + for it in items: + is_alive = (now - int(it.last_seen_at)) <= self._ttl_s + if not include_dead and not is_alive: + continue + queue_item = per_instance_queue.get(it.instance_id, {}) + inflight = it.load.inflight + qps_1m = it.load.qps_1m + score = { + "primary": "inflight" if inflight is not None else None, + "primary_value": inflight, + "primary_source": "proxy_lifecycle_counter" if inflight is not None else "unavailable", + "secondary": "qps_1m" if qps_1m is not None else None, + "secondary_value": qps_1m, + "secondary_source": "instance_heartbeat" if qps_1m is not None else "unavailable", + } + instances.append({ + "instance_id": it.instance_id, + "is_alive": is_alive, + "inflight": inflight, + "qps_1m": qps_1m, + "prepare_queue_depth": queue_item.get("prepare_queue_depth"), + "ready_queue_depth": queue_item.get("ready_queue_depth"), + "active_prepare": queue_item.get("active_prepare"), + "active_ready": queue_item.get("active_ready"), + "least_load_score": score, + }) + + return { + "ttl_s": self._ttl_s, + "generated_at": time.time(), + "metric_source": { + "inflight": "proxy_lifecycle_counter", + "qps_1m": "instance_heartbeat", + "queue_depth": "proxy_queue_manager" if queue_depths is not None else "unavailable", + }, + "instances": instances, + } + def build_pool_resource_snapshot( self, proxy_id: str, @@ -216,7 +290,7 @@ def build_pool_resource_snapshot( metric_source = { "instances": "proxy_instance_pool_ttl", "resource": "instance_resource_snapshot" if reporting else "unavailable", - "inflight_total": "instance_heartbeat" if inflight_values else "unavailable", + "inflight_total": "proxy_lifecycle_counter" if inflight_values else "unavailable", "qps_1m_total": "instance_heartbeat" if qps_values else "unavailable", "load_ratio": "derived_from_inflight_capacity" if load_ratio is not None else "unavailable", "capacity": "proxy_config", diff --git a/proxy/resource/p_control_plane.py b/proxy/resource/p_control_plane.py index 2610914..d2c99ba 100644 --- a/proxy/resource/p_control_plane.py +++ b/proxy/resource/p_control_plane.py @@ -339,6 +339,22 @@ async def debug_instance_resources(include_dead: bool = True) -> Dict[str, Any]: }) return {"instances": resources} + +@_control_plane.get("/debug/instance_loads") +async def debug_instance_loads(include_dead: bool = True) -> Dict[str, Any]: + pool = get_pool() + queue_depths = None + if _queue_snapshot_provider is not None: + try: + queue_depths = _queue_snapshot_provider() or {} + except Exception: + logger.warning( + "[ProxyCP] queue snapshot provider failed; instance load queue metrics unavailable", + exc_info=True, + ) + snapshot = pool.snapshot_instance_loads(queue_depths=queue_depths, include_dead=include_dead) + return {"ok": True, **snapshot} + @_control_plane.post("/v1/topology/report") async def report_topology(req: TopologyReportReq) -> Dict[str, Any]: merged = 0 diff --git a/proxy/strategy/factory.py b/proxy/strategy/factory.py index 35b1ad9..0f37a8b 100644 --- a/proxy/strategy/factory.py +++ b/proxy/strategy/factory.py @@ -1,12 +1,15 @@ -# proxy/strategy/factory.py -from __future__ import annotations - -from .base import BaseInstanceStrategy -from .round_robin import RoundRobinStrategy - - -def build_instance_strategy(name: str) -> BaseInstanceStrategy: - n = (name or "").strip().lower() - if n in ("rr", "round_robin", "round-robin"): - return RoundRobinStrategy() - raise ValueError(f"unknown instance strategy: {name}") +# proxy/strategy/factory.py +from __future__ import annotations + +from .base import BaseInstanceStrategy +from .least_load import LeastLoadStrategy +from .round_robin import RoundRobinStrategy + + +def build_instance_strategy(name: str) -> BaseInstanceStrategy: + n = (name or "").strip().lower() + if n in ("rr", "round_robin", "round-robin"): + return RoundRobinStrategy() + if n in ("ll", "least_load", "least-load"): + return LeastLoadStrategy() + raise ValueError(f"unknown instance strategy: {name}") diff --git a/proxy/strategy/least_load.py b/proxy/strategy/least_load.py new file mode 100644 index 0000000..fa2faf6 --- /dev/null +++ b/proxy/strategy/least_load.py @@ -0,0 +1,57 @@ +# proxy/strategy/least_load.py +"""Implements least-load instance selection for the proxy.""" +from __future__ import annotations + +from typing import Any, List, Optional + +from .base import BaseInstanceStrategy, InstanceLike +from .round_robin import RoundRobinStrategy + + +class LeastLoadStrategy(BaseInstanceStrategy): + """ + Select the Instance with the lowest known load. + + The strategy prefers known ``load.inflight`` values maintained by the Proxy + request lifecycle. It uses known ``load.qps_1m`` as a secondary signal, + without treating missing metrics as zero. When no useful load metrics are + known, it falls back to round-robin. + """ + + name = "least_load" + + def __init__(self) -> None: + self._rr = RoundRobinStrategy() + + def select(self, instances: List[InstanceLike], hint: Optional[Any] = None) -> InstanceLike: + if not instances: + raise RuntimeError("no instances") + + known_inflight = [it for it in instances if self._inflight(it) is not None] + if known_inflight: + min_inflight = min(self._inflight(it) for it in known_inflight) + inflight_ties = [it for it in known_inflight if self._inflight(it) == min_inflight] + return self._select_by_qps_or_rr(inflight_ties) + + return self._select_by_qps_or_rr(instances) + + def _select_by_qps_or_rr(self, instances: List[InstanceLike]) -> InstanceLike: + known_qps = [it for it in instances if self._qps_1m(it) is not None] + if not known_qps: + return self._rr.select(instances) + + min_qps = min(self._qps_1m(it) for it in known_qps) + qps_ties = [it for it in known_qps if self._qps_1m(it) == min_qps] + return self._rr.select(qps_ties) + + @staticmethod + def _inflight(instance: InstanceLike) -> Optional[int]: + load = getattr(instance, "load", None) + value = getattr(load, "inflight", None) + return int(value) if value is not None else None + + @staticmethod + def _qps_1m(instance: InstanceLike) -> Optional[float]: + load = getattr(instance, "load", None) + value = getattr(load, "qps_1m", None) + return float(value) if value is not None else None From 55b32e921b671a7779428703e6379eb9a0b43892 Mon Sep 17 00:00:00 2001 From: RickZ <121943721+zhy1658858023@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:08:54 +0800 Subject: [PATCH 2/3] feat(proxy): maintain local instance inflight counters --- proxy/README.md | 10 +++------- proxy/resource/instance_pool.py | 22 ++++++++++++---------- proxy/strategy/least_load.py | 7 +++---- 3 files changed, 18 insertions(+), 21 deletions(-) diff --git a/proxy/README.md b/proxy/README.md index b7dec20..85de188 100644 --- a/proxy/README.md +++ b/proxy/README.md @@ -164,6 +164,8 @@ GET /debug/instance_resources GET /debug/instance_loads ``` +Use `/debug/instance_loads` to inspect Proxy-maintained per-Instance `inflight` counters, `qps_1m` when present, and queue-depth hints when the queue snapshot provider is available. + The reporting path is: ```text @@ -247,7 +249,7 @@ For streaming chat completion, the Proxy forwards the downstream SSE stream and | Strategy | Aliases | Status | Description | |---|---|---|---| | `round_robin` | `round_robin`, `round-robin`, `rr` | Default | Selects alive Instances in round-robin order. | -| `least_load` | `least_load`, `least-load`, `ll` | Experimental | Selects the lowest known Instance load by Proxy-maintained `load.inflight`, then uses `qps_1m` as a secondary signal. Missing metrics remain unknown rather than zero; if all candidates lack usable load metrics, selection falls back to round-robin. | +| `least_load` | `least_load`, `least-load`, `ll` | Experimental | Selects the lowest known Instance load by `load.inflight`, then uses `qps_1m` as a secondary signal. Missing metrics remain unknown rather than zero; if all candidates lack usable load metrics, selection falls back to round-robin. | | `kv_aware` | Planned | Planned | Future strategy intended to consider KVCache locality/inventory together with runtime load. | Select `least_load` from the demo CLI: @@ -262,12 +264,6 @@ Or select it through the environment: PROXY_INSTANCE_STRATEGY=least_load python3 test/demo_proxy.py ``` -The Proxy increments the selected Instance's local inflight counter before enqueueing a request and decrements it when the response path finishes, fails, or is cancelled. Inspect these counters and queue-depth hints with: - -```bash -curl -sS http://127.0.0.1:8002/debug/instance_loads | python3 -m json.tool -``` - If no alive Instance is available, the Proxy returns: ```text diff --git a/proxy/resource/instance_pool.py b/proxy/resource/instance_pool.py index 61bf0d3..40970f7 100644 --- a/proxy/resource/instance_pool.py +++ b/proxy/resource/instance_pool.py @@ -57,7 +57,7 @@ class InstancePool: """ In-memory instance pool (TTL based). - upsert: register/update instance static fields - - heartbeat: refresh last_seen and optionally update non-lifecycle load fields + - heartbeat: refresh last_seen and optionally update load fields - list(include_dead=False): returns alive instances by default """ def __init__(self, ttl_s: int = 30): @@ -90,6 +90,8 @@ def upsert( it.weight = float(weight) if meta: it.meta.update(meta) + if it.load.inflight is None: + it.load.inflight = 0 it.last_seen_at = now return it @@ -101,6 +103,7 @@ def upsert( tags=tags or [], weight=float(weight), meta=meta or {}, + load=InstanceLoad(inflight=0), registered_at=now, last_seen_at=now, ) @@ -188,14 +191,6 @@ def snapshot_instance_loads( queue_item = per_instance_queue.get(it.instance_id, {}) inflight = it.load.inflight qps_1m = it.load.qps_1m - score = { - "primary": "inflight" if inflight is not None else None, - "primary_value": inflight, - "primary_source": "proxy_lifecycle_counter" if inflight is not None else "unavailable", - "secondary": "qps_1m" if qps_1m is not None else None, - "secondary_value": qps_1m, - "secondary_source": "instance_heartbeat" if qps_1m is not None else "unavailable", - } instances.append({ "instance_id": it.instance_id, "is_alive": is_alive, @@ -205,7 +200,14 @@ 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"), - "least_load_score": score, + "least_load_score": { + "primary": "inflight" if inflight is not None else None, + "primary_value": inflight, + "primary_source": "proxy_lifecycle_counter" if inflight is not None else "unavailable", + "secondary": "qps_1m" if qps_1m is not None else None, + "secondary_value": qps_1m, + "secondary_source": "instance_heartbeat" if qps_1m is not None else "unavailable", + }, }) return { diff --git a/proxy/strategy/least_load.py b/proxy/strategy/least_load.py index fa2faf6..500697f 100644 --- a/proxy/strategy/least_load.py +++ b/proxy/strategy/least_load.py @@ -12,10 +12,9 @@ class LeastLoadStrategy(BaseInstanceStrategy): """ Select the Instance with the lowest known load. - The strategy prefers known ``load.inflight`` values maintained by the Proxy - request lifecycle. It uses known ``load.qps_1m`` as a secondary signal, - without treating missing metrics as zero. When no useful load metrics are - known, it falls back to round-robin. + The strategy prefers known ``load.inflight`` values. It uses known + ``load.qps_1m`` as a secondary signal, without treating missing metrics as + zero. When no useful load metrics are known, it falls back to round-robin. """ name = "least_load" From 9371b799797786116d6475cb3a4ca698ce314d6e Mon Sep 17 00:00:00 2001 From: RickZ <121943721+zhy1658858023@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:44:58 +0800 Subject: [PATCH 3/3] feat(proxy): add instance inflight counters --- proxy/README.md | 25 +++------------ proxy/resource/instance_pool.py | 2 +- proxy/strategy/factory.py | 27 +++++++--------- proxy/strategy/least_load.py | 56 --------------------------------- 4 files changed, 18 insertions(+), 92 deletions(-) delete mode 100644 proxy/strategy/least_load.py diff --git a/proxy/README.md b/proxy/README.md index 85de188..447e5b6 100644 --- a/proxy/README.md +++ b/proxy/README.md @@ -40,7 +40,6 @@ proxy/ ├── strategy/ │ ├── base.py # Base Instance selection strategy │ ├── round_robin.py # Round-robin Instance selection -│ ├── least_load.py # Experimental least-load Instance selection │ ├── least_inflight.py # Reserved for future strategy extension │ └── factory.py # Strategy builder ├── queue/ @@ -84,7 +83,7 @@ Common options: |---|---| | `--host` | Proxy service-plane bind host. The demo also uses it as the advertised host. | | `--port` | Proxy service-plane bind port. The demo also uses it as the advertised port. | -| `--strategy` | Local Instance selection strategy. Defaults to `round_robin`; use `least_load` for the experimental load-aware selector. | +| `--strategy` | Local Instance selection strategy. Currently `round_robin` is the active demo path. | | `--injection-strategy` | Knowledge injection strategy. Use `default` or `iws`. | | `--ready-release-policy` | Ready queue release policy: `ordered` or `text_bypass`. | | `--kdn-links-json` | Optional static KDN topology metadata. | @@ -244,25 +243,11 @@ For streaming chat completion, the Proxy forwards the downstream SSE stream and ## Instance selection -### Proxy Instance selection strategies +The active demo strategy is: -| Strategy | Aliases | Status | Description | -|---|---|---|---| -| `round_robin` | `round_robin`, `round-robin`, `rr` | Default | Selects alive Instances in round-robin order. | -| `least_load` | `least_load`, `least-load`, `ll` | Experimental | Selects the lowest known Instance load by `load.inflight`, then uses `qps_1m` as a secondary signal. Missing metrics remain unknown rather than zero; if all candidates lack usable load metrics, selection falls back to round-robin. | -| `kv_aware` | Planned | Planned | Future strategy intended to consider KVCache locality/inventory together with runtime load. | - -Select `least_load` from the demo CLI: - -```bash -python3 test/demo_proxy.py --strategy least_load -``` - -Or select it through the environment: - -```bash -PROXY_INSTANCE_STRATEGY=least_load python3 test/demo_proxy.py -``` +| Strategy | Description | +|---|---| +| `round_robin` | Selects alive Instances in round-robin order. | If no alive Instance is available, the Proxy returns: diff --git a/proxy/resource/instance_pool.py b/proxy/resource/instance_pool.py index 40970f7..38fea8b 100644 --- a/proxy/resource/instance_pool.py +++ b/proxy/resource/instance_pool.py @@ -172,7 +172,7 @@ def snapshot_instance_loads( queue_depths: Optional[Dict[str, Any]] = None, include_dead: bool = True, ) -> Dict[str, Any]: - """Return per-Instance load inputs used by Proxy-local strategies and debug views.""" + """Return per-Instance local load counters and optional queue-depth hints.""" now = int(time.time()) per_instance_queue = {} if queue_depths: diff --git a/proxy/strategy/factory.py b/proxy/strategy/factory.py index 0f37a8b..35b1ad9 100644 --- a/proxy/strategy/factory.py +++ b/proxy/strategy/factory.py @@ -1,15 +1,12 @@ -# proxy/strategy/factory.py -from __future__ import annotations - -from .base import BaseInstanceStrategy -from .least_load import LeastLoadStrategy -from .round_robin import RoundRobinStrategy - - -def build_instance_strategy(name: str) -> BaseInstanceStrategy: - n = (name or "").strip().lower() - if n in ("rr", "round_robin", "round-robin"): - return RoundRobinStrategy() - if n in ("ll", "least_load", "least-load"): - return LeastLoadStrategy() - raise ValueError(f"unknown instance strategy: {name}") +# proxy/strategy/factory.py +from __future__ import annotations + +from .base import BaseInstanceStrategy +from .round_robin import RoundRobinStrategy + + +def build_instance_strategy(name: str) -> BaseInstanceStrategy: + n = (name or "").strip().lower() + if n in ("rr", "round_robin", "round-robin"): + return RoundRobinStrategy() + raise ValueError(f"unknown instance strategy: {name}") diff --git a/proxy/strategy/least_load.py b/proxy/strategy/least_load.py deleted file mode 100644 index 500697f..0000000 --- a/proxy/strategy/least_load.py +++ /dev/null @@ -1,56 +0,0 @@ -# proxy/strategy/least_load.py -"""Implements least-load instance selection for the proxy.""" -from __future__ import annotations - -from typing import Any, List, Optional - -from .base import BaseInstanceStrategy, InstanceLike -from .round_robin import RoundRobinStrategy - - -class LeastLoadStrategy(BaseInstanceStrategy): - """ - Select the Instance with the lowest known load. - - The strategy prefers known ``load.inflight`` values. It uses known - ``load.qps_1m`` as a secondary signal, without treating missing metrics as - zero. When no useful load metrics are known, it falls back to round-robin. - """ - - name = "least_load" - - def __init__(self) -> None: - self._rr = RoundRobinStrategy() - - def select(self, instances: List[InstanceLike], hint: Optional[Any] = None) -> InstanceLike: - if not instances: - raise RuntimeError("no instances") - - known_inflight = [it for it in instances if self._inflight(it) is not None] - if known_inflight: - min_inflight = min(self._inflight(it) for it in known_inflight) - inflight_ties = [it for it in known_inflight if self._inflight(it) == min_inflight] - return self._select_by_qps_or_rr(inflight_ties) - - return self._select_by_qps_or_rr(instances) - - def _select_by_qps_or_rr(self, instances: List[InstanceLike]) -> InstanceLike: - known_qps = [it for it in instances if self._qps_1m(it) is not None] - if not known_qps: - return self._rr.select(instances) - - min_qps = min(self._qps_1m(it) for it in known_qps) - qps_ties = [it for it in known_qps if self._qps_1m(it) == min_qps] - return self._rr.select(qps_ties) - - @staticmethod - def _inflight(instance: InstanceLike) -> Optional[int]: - load = getattr(instance, "load", None) - value = getattr(load, "inflight", None) - return int(value) if value is not None else None - - @staticmethod - def _qps_1m(instance: InstanceLike) -> Optional[float]: - load = getattr(instance, "load", None) - value = getattr(load, "qps_1m", None) - return float(value) if value is not None else None