wehub-resource-sync
693 lines
24 KiB
Python
693 lines
24 KiB
Python
"""Router benchmark: CapacityQueueRouter vs PowerOfTwoChoices.
|
|
|
|
Compares routers across small (8), medium (32), large (128),
|
|
and xlarge (512) replica scales.
|
|
|
|
Measures per configuration:
|
|
- p50 throughput (req/s)
|
|
- p50 client end-to-end latency (ms)
|
|
- p50 app latency (ms)
|
|
- p50 actual child processing latency (ms)
|
|
- Per-replica utilization (p25, p50, p75)
|
|
|
|
Methodology:
|
|
- Parent->Child deployment chain where Child simulates work via asyncio.sleep
|
|
with an exponential distribution (mean/cap configurable via CLI).
|
|
- Closed-loop load generation: N concurrent users each making sequential
|
|
requests through DeploymentHandle, distributed across remote actors.
|
|
- Load level: 100% of theoretical max throughput, which equals to
|
|
num_replicas * max_ongoing_requests.
|
|
- Serve access logs are disabled so logging throughput does not dominate
|
|
low-latency microbenchmark results.
|
|
|
|
Usage (CI):
|
|
python workloads/router_microbenchmark.py
|
|
|
|
Usage (manual):
|
|
python workloads/router_microbenchmark.py -o /tmp/results.json
|
|
|
|
Plot results (offline, after downloading CI output):
|
|
python workloads/plot_router_benchmark.py results.json -o /tmp/plots
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import math
|
|
import random
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Dict, List, Optional
|
|
|
|
import click
|
|
import numpy as np
|
|
import ray
|
|
from ray import serve
|
|
from ray.serve.config import DeploymentActorConfig, RequestRouterConfig
|
|
from ray.serve.experimental.capacity_queue import CapacityQueue
|
|
from ray.serve.handle import DeploymentHandle
|
|
from serve_test_utils import save_test_results
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Load-test configuration
|
|
# ---------------------------------------------------------------------------
|
|
LOAD_LEVEL = 1.0 # fraction of theoretical max
|
|
WARMUP_S = 10.0
|
|
DURATION_S = 60.0
|
|
THROUGHPUT_WINDOW_S = 5.0 # window size for per-window throughput
|
|
MAX_USERS_PER_TASK = 48 # max concurrent users per load-gen task
|
|
LOAD_GEN_START_DELAY_S = 5.0
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scales and router types
|
|
# ---------------------------------------------------------------------------
|
|
NUM_REPLICAS = [512, 128, 32, 8]
|
|
|
|
ROUTER_TYPES = ["pow2", "capacity_queue"]
|
|
|
|
APP_NAME = "router-benchmark"
|
|
LOGGING_CONFIG = {"enable_access_log": False}
|
|
|
|
|
|
# ===================================================================
|
|
# Deployments
|
|
# ===================================================================
|
|
|
|
|
|
@serve.deployment(
|
|
max_queued_requests=-1,
|
|
graceful_shutdown_timeout_s=0.1,
|
|
graceful_shutdown_wait_loop_s=0.1,
|
|
ray_actor_options={"num_cpus": 1},
|
|
)
|
|
class BenchmarkChild:
|
|
"""Simulates work with variable latency drawn from an exponential distribution."""
|
|
|
|
def __init__(
|
|
self,
|
|
simulated_latency_mean_s: float,
|
|
simulated_latency_cap_s: float,
|
|
):
|
|
self._replica_id = serve.get_replica_context().replica_id.unique_id
|
|
self._mean_s = simulated_latency_mean_s
|
|
self._cap_s = simulated_latency_cap_s
|
|
|
|
async def __call__(self) -> dict:
|
|
simulated_latency_s = min(
|
|
random.expovariate(1 / self._mean_s),
|
|
self._cap_s,
|
|
)
|
|
processing_start = time.perf_counter()
|
|
await asyncio.sleep(simulated_latency_s)
|
|
processing_s = time.perf_counter() - processing_start
|
|
return {
|
|
"replica_id": self._replica_id,
|
|
"processing_s": processing_s,
|
|
"simulated_processing_s": simulated_latency_s,
|
|
}
|
|
|
|
|
|
@serve.deployment(
|
|
max_queued_requests=-1,
|
|
graceful_shutdown_timeout_s=0.1,
|
|
graceful_shutdown_wait_loop_s=0.1,
|
|
ray_actor_options={"num_cpus": 1},
|
|
)
|
|
class BenchmarkParent:
|
|
"""Routes requests to the child deployment and returns child replica id."""
|
|
|
|
def __init__(self, child: DeploymentHandle):
|
|
self._child = child
|
|
|
|
async def __call__(self) -> dict:
|
|
app_start = time.perf_counter()
|
|
resp = await self._child.remote()
|
|
resp["app_latency_s"] = time.perf_counter() - app_start
|
|
return resp
|
|
|
|
|
|
# ===================================================================
|
|
# App builders
|
|
# ===================================================================
|
|
|
|
|
|
@dataclass
|
|
class WorkloadConfig:
|
|
simulated_latency_mean_s: float
|
|
simulated_latency_cap_s: float
|
|
max_ongoing_requests_child: int
|
|
max_ongoing_requests_parent: int
|
|
|
|
|
|
def _build_pow2_app(num_replicas: int, wl: WorkloadConfig):
|
|
"""Build app with default Power-of-Two-Choices router."""
|
|
child = BenchmarkChild.options(
|
|
num_replicas=num_replicas,
|
|
max_ongoing_requests=wl.max_ongoing_requests_child,
|
|
).bind(wl.simulated_latency_mean_s, wl.simulated_latency_cap_s)
|
|
return BenchmarkParent.options(
|
|
num_replicas=num_replicas,
|
|
max_ongoing_requests=wl.max_ongoing_requests_parent,
|
|
).bind(child)
|
|
|
|
|
|
def _build_capacity_queue_app(num_replicas: int, wl: WorkloadConfig):
|
|
"""Build app with CapacityQueueRouter."""
|
|
router_config = RequestRouterConfig(
|
|
request_router_class=(
|
|
"ray.serve.experimental.capacity_queue_router:CapacityQueueRouter"
|
|
),
|
|
request_router_kwargs={
|
|
"capacity_queue_actor_name": "capacity_queue",
|
|
"max_fault_retries": 3,
|
|
},
|
|
initial_backoff_s=0.05,
|
|
backoff_multiplier=2.0,
|
|
max_backoff_s=1.0,
|
|
)
|
|
|
|
def _capacity_queue_actors():
|
|
return [
|
|
DeploymentActorConfig(
|
|
name="capacity_queue",
|
|
actor_class=CapacityQueue,
|
|
init_kwargs={
|
|
"acquire_timeout_s": 0.5,
|
|
"token_ttl_s": 5,
|
|
},
|
|
actor_options={"num_cpus": 0},
|
|
),
|
|
]
|
|
|
|
child = BenchmarkChild.options(
|
|
num_replicas=num_replicas,
|
|
max_ongoing_requests=wl.max_ongoing_requests_child,
|
|
request_router_config=router_config,
|
|
deployment_actors=_capacity_queue_actors(),
|
|
).bind(wl.simulated_latency_mean_s, wl.simulated_latency_cap_s)
|
|
return BenchmarkParent.options(
|
|
num_replicas=num_replicas,
|
|
max_ongoing_requests=wl.max_ongoing_requests_parent,
|
|
request_router_config=router_config,
|
|
deployment_actors=_capacity_queue_actors(),
|
|
).bind(child)
|
|
|
|
|
|
def _build_app(router_type: str, num_replicas: int, wl: WorkloadConfig):
|
|
"""Build a Parent->Child app with the given router type and scale."""
|
|
if router_type == "pow2":
|
|
return _build_pow2_app(num_replicas, wl)
|
|
elif router_type == "capacity_queue":
|
|
return _build_capacity_queue_app(num_replicas, wl)
|
|
raise ValueError(f"Unknown router type: {router_type}")
|
|
|
|
|
|
# ===================================================================
|
|
# Load generation
|
|
# ===================================================================
|
|
|
|
|
|
@dataclass
|
|
class RequestResult:
|
|
start_time: float
|
|
end_time: float
|
|
latency_ms: float
|
|
app_latency_ms: float # parent replica -> child replica -> parent replica
|
|
child_replica_id: str
|
|
processing_s: float # actual child wall-clock time spent in simulated work
|
|
simulated_processing_s: float # sampled sleep duration requested by the child
|
|
success: bool
|
|
|
|
|
|
@ray.remote(num_cpus=1)
|
|
class LoadGenTask:
|
|
"""Remote actor that runs a batch of closed-loop users."""
|
|
|
|
def __init__(self, app_name: str):
|
|
self._handle = serve.get_deployment_handle("BenchmarkParent", app_name=app_name)
|
|
|
|
def ready(self) -> bool:
|
|
return True
|
|
|
|
async def run(
|
|
self, num_users: int, warmup_s: float, duration_s: float, start_at: float
|
|
) -> List[Dict]:
|
|
sleep_s = start_at - time.time()
|
|
if sleep_s > 0:
|
|
await asyncio.sleep(sleep_s)
|
|
|
|
start = start_at
|
|
warmup_end = start + warmup_s
|
|
test_end = start + warmup_s + duration_s
|
|
|
|
async def user_loop() -> List[Dict]:
|
|
results = []
|
|
while time.time() < test_end:
|
|
req_start = time.time()
|
|
try:
|
|
resp = await self._handle.remote()
|
|
req_end = time.time()
|
|
if req_start >= warmup_end:
|
|
results.append(
|
|
{
|
|
"start_time": req_start,
|
|
"end_time": req_end,
|
|
"latency_ms": (req_end - req_start) * 1000,
|
|
"app_latency_ms": resp["app_latency_s"] * 1000,
|
|
"child_replica_id": resp["replica_id"],
|
|
"processing_s": resp["processing_s"],
|
|
"simulated_processing_s": resp[
|
|
"simulated_processing_s"
|
|
],
|
|
"success": True,
|
|
}
|
|
)
|
|
except Exception:
|
|
req_end = time.time()
|
|
if req_start >= warmup_end:
|
|
results.append(
|
|
{
|
|
"start_time": req_start,
|
|
"end_time": req_end,
|
|
"latency_ms": (req_end - req_start) * 1000,
|
|
"app_latency_ms": 0.0,
|
|
"child_replica_id": "error",
|
|
"processing_s": 0.0,
|
|
"simulated_processing_s": 0.0,
|
|
"success": False,
|
|
}
|
|
)
|
|
return results
|
|
|
|
user_results = await asyncio.gather(*[user_loop() for _ in range(num_users)])
|
|
return [r for batch in user_results for r in batch]
|
|
|
|
|
|
async def _run_load_test(
|
|
num_concurrent: int,
|
|
warmup_s: float,
|
|
duration_s: float,
|
|
max_users_per_task: int = MAX_USERS_PER_TASK,
|
|
) -> List[RequestResult]:
|
|
"""Run a closed-loop load test and return per-request results.
|
|
|
|
Distributes *num_concurrent* users across multiple remote
|
|
LoadGenTask actors (up to max_users_per_task users each) to
|
|
avoid bottlenecking the driver's event loop at large scales.
|
|
"""
|
|
if max_users_per_task <= 0:
|
|
raise ValueError("max_users_per_task must be positive.")
|
|
|
|
num_tasks = max(1, math.ceil(num_concurrent / max_users_per_task))
|
|
base_users = num_concurrent // num_tasks
|
|
remainder = num_concurrent % num_tasks
|
|
users_per_task = [
|
|
base_users + (1 if i < remainder else 0) for i in range(num_tasks)
|
|
]
|
|
|
|
logger.info(
|
|
f"Load test: {num_concurrent} users across {num_tasks} tasks, "
|
|
f"{warmup_s}s warmup + {duration_s}s measurement, "
|
|
f"max_users_per_task={max_users_per_task}"
|
|
)
|
|
|
|
tasks = [LoadGenTask.remote(APP_NAME) for _ in range(num_tasks)]
|
|
await asyncio.gather(
|
|
*[asyncio.wrap_future(t.ready.remote().future()) for t in tasks]
|
|
)
|
|
# Use a shared clock edge so autoscaling or actor placement delays do not
|
|
# give each load-gen actor a different measurement window.
|
|
start_at = time.time() + LOAD_GEN_START_DELAY_S
|
|
futures = [
|
|
t.run.remote(n, warmup_s, duration_s, start_at)
|
|
for t, n in zip(tasks, users_per_task)
|
|
]
|
|
all_dicts = await asyncio.gather(
|
|
*[asyncio.wrap_future(f.future()) for f in futures]
|
|
)
|
|
|
|
results = []
|
|
for batch in all_dicts:
|
|
for d in batch:
|
|
results.append(
|
|
RequestResult(
|
|
start_time=d["start_time"],
|
|
end_time=d["end_time"],
|
|
latency_ms=d["latency_ms"],
|
|
app_latency_ms=d["app_latency_ms"],
|
|
child_replica_id=d["child_replica_id"],
|
|
processing_s=d["processing_s"],
|
|
simulated_processing_s=d["simulated_processing_s"],
|
|
success=d["success"],
|
|
)
|
|
)
|
|
logger.info(f"Collected {len(results)} results from {num_tasks} tasks")
|
|
|
|
# Clean up load-gen actors
|
|
for t in tasks:
|
|
ray.kill(t)
|
|
|
|
return results
|
|
|
|
|
|
# ===================================================================
|
|
# Metric computation
|
|
# ===================================================================
|
|
|
|
|
|
def _compute_throughput_p50(
|
|
results: List[RequestResult],
|
|
window_s: float = THROUGHPUT_WINDOW_S,
|
|
) -> float:
|
|
"""p50 of per-window throughput (RPS)."""
|
|
successful = [r for r in results if r.success]
|
|
if not successful:
|
|
return 0.0
|
|
|
|
min_t = min(r.start_time for r in successful)
|
|
max_t = max(r.end_time for r in successful)
|
|
duration = max_t - min_t
|
|
if duration <= 0:
|
|
return 0.0
|
|
if duration <= window_s:
|
|
return len(successful) / duration
|
|
|
|
num_windows = int(duration / window_s)
|
|
window_counts = [0] * num_windows
|
|
for r in successful:
|
|
idx = min(int((r.start_time - min_t) / window_s), num_windows - 1)
|
|
window_counts[idx] += 1
|
|
|
|
window_rps = [c / window_s for c in window_counts]
|
|
return float(np.median(window_rps))
|
|
|
|
|
|
def _compute_utilization(
|
|
results: List[RequestResult],
|
|
num_replicas: int,
|
|
duration_s: float,
|
|
max_ongoing_requests_child: int,
|
|
) -> List[float]:
|
|
"""Per-replica utilization as a list (one value per replica).
|
|
|
|
Utilization = fraction of replica slot-time spent processing.
|
|
1.0 means the router has zero overhead (slots always busy).
|
|
Requests that extend past the measurement window are clamped
|
|
so utilization never exceeds 1.0.
|
|
"""
|
|
successful = [r for r in results if r.success]
|
|
if not successful:
|
|
return [0.0] * num_replicas
|
|
|
|
available_s = duration_s * max_ongoing_requests_child
|
|
|
|
# Measurement window: starts at earliest request, spans duration_s
|
|
window_start = min(r.start_time for r in successful)
|
|
window_end = window_start + duration_s
|
|
|
|
busy: Dict[str, float] = {}
|
|
for r in successful:
|
|
# Clamp contribution to time remaining in the measurement window
|
|
contribution = min(r.processing_s, max(0.0, window_end - r.start_time))
|
|
busy[r.child_replica_id] = busy.get(r.child_replica_id, 0.0) + contribution
|
|
|
|
utilizations = [t / available_s for t in busy.values()]
|
|
utilizations.extend([0.0] * max(0, num_replicas - len(busy)))
|
|
return utilizations
|
|
|
|
|
|
# ===================================================================
|
|
# Readiness helpers
|
|
# ===================================================================
|
|
|
|
|
|
async def _wait_for_ready(
|
|
handle: DeploymentHandle,
|
|
timeout_s: float = 300.0,
|
|
num_probes: int = 5,
|
|
):
|
|
"""Block until multiple consecutive probe requests succeed.
|
|
|
|
Sends *num_probes* sequential requests to ensure routing tables are
|
|
populated and the deployment is fully warmed up.
|
|
"""
|
|
start = time.time()
|
|
while time.time() - start < timeout_s:
|
|
try:
|
|
await handle.remote()
|
|
break
|
|
except Exception:
|
|
await asyncio.sleep(2.0)
|
|
else:
|
|
raise TimeoutError(f"App not ready after {timeout_s}s")
|
|
|
|
# Send additional probes to warm routing tables
|
|
for _ in range(num_probes - 1):
|
|
await handle.remote()
|
|
logger.info(f"App ready ({time.time() - start:.1f}s, {num_probes} probes)")
|
|
|
|
|
|
# ===================================================================
|
|
# Benchmark runner
|
|
# ===================================================================
|
|
|
|
|
|
async def run_router_benchmark(
|
|
workload: WorkloadConfig,
|
|
warmup_s: float = WARMUP_S,
|
|
duration_s: float = DURATION_S,
|
|
num_replicas_list: Optional[List[int]] = None,
|
|
router_types: Optional[List[str]] = None,
|
|
max_users_per_task: int = MAX_USERS_PER_TASK,
|
|
) -> Dict:
|
|
"""Run the router benchmark and return results dict.
|
|
|
|
Returns {"perf_metrics": [...], "utilization_raw": {...}} where
|
|
utilization_raw maps "router_type_replicas" to per-replica values.
|
|
"""
|
|
if num_replicas_list is None:
|
|
num_replicas_list = NUM_REPLICAS
|
|
if router_types is None:
|
|
router_types = ROUTER_TYPES
|
|
|
|
perf_metrics: List[Dict] = []
|
|
utilization_raw: Dict[str, List[float]] = {}
|
|
|
|
for router_type in router_types:
|
|
for num_replicas in num_replicas_list:
|
|
num_concurrent = max(
|
|
1,
|
|
int(num_replicas * workload.max_ongoing_requests_child * LOAD_LEVEL),
|
|
)
|
|
prefix = f"router_{router_type}_{num_replicas}"
|
|
|
|
logger.info(
|
|
f"=== {router_type} @ {num_replicas} replicas "
|
|
f"({num_concurrent} users) ==="
|
|
)
|
|
|
|
handle = serve.run(
|
|
_build_app(router_type, num_replicas, workload),
|
|
name=APP_NAME,
|
|
logging_config=LOGGING_CONFIG,
|
|
)
|
|
|
|
try:
|
|
# Scale readiness probes with replica count so routing
|
|
# tables are fully populated before the load test starts.
|
|
num_probes = max(5, num_replicas // 32)
|
|
await _wait_for_ready(handle, num_probes=num_probes)
|
|
|
|
# Scale warmup with replica count
|
|
scaled_warmup = max(warmup_s, warmup_s * (num_replicas / 32))
|
|
|
|
results = await _run_load_test(
|
|
num_concurrent=num_concurrent,
|
|
warmup_s=scaled_warmup,
|
|
duration_s=duration_s,
|
|
max_users_per_task=max_users_per_task,
|
|
)
|
|
|
|
successful = [r for r in results if r.success]
|
|
total = len(results)
|
|
failed = total - len(successful)
|
|
if total:
|
|
logger.info(
|
|
f" {prefix}: {total} total, "
|
|
f"{failed} failed ({failed / total * 100:.1f}%)"
|
|
)
|
|
|
|
# -- throughput --
|
|
tp50 = _compute_throughput_p50(results)
|
|
perf_metrics.append(
|
|
{
|
|
"perf_metric_name": f"{prefix}_p50_throughput_rps",
|
|
"perf_metric_value": round(tp50, 2),
|
|
"perf_metric_type": "THROUGHPUT",
|
|
}
|
|
)
|
|
|
|
# -- latency --
|
|
if successful:
|
|
app_latencies = [r.app_latency_ms for r in successful]
|
|
e2e_latencies = [r.latency_ms for r in successful]
|
|
processing_latencies = [r.processing_s * 1000 for r in successful]
|
|
perf_metrics.append(
|
|
{
|
|
"perf_metric_name": f"{prefix}_p50_latency_ms",
|
|
"perf_metric_value": round(
|
|
float(np.median(e2e_latencies)), 2
|
|
),
|
|
"perf_metric_type": "LATENCY",
|
|
}
|
|
)
|
|
perf_metrics.append(
|
|
{
|
|
"perf_metric_name": f"{prefix}_p50_app_latency_ms",
|
|
"perf_metric_value": round(
|
|
float(np.median(app_latencies)), 2
|
|
),
|
|
"perf_metric_type": "LATENCY",
|
|
}
|
|
)
|
|
perf_metrics.append(
|
|
{
|
|
"perf_metric_name": (f"{prefix}_p50_processing_latency_ms"),
|
|
"perf_metric_value": round(
|
|
float(np.median(processing_latencies)), 2
|
|
),
|
|
"perf_metric_type": "LATENCY",
|
|
}
|
|
)
|
|
|
|
# -- utilization --
|
|
utils = _compute_utilization(
|
|
results,
|
|
num_replicas,
|
|
duration_s,
|
|
workload.max_ongoing_requests_child,
|
|
)
|
|
utilization_raw[prefix] = [round(u, 4) for u in utils]
|
|
for pct, label in [(25, "p25"), (50, "p50"), (75, "p75")]:
|
|
perf_metrics.append(
|
|
{
|
|
"perf_metric_name": f"{prefix}_{label}_utilization",
|
|
"perf_metric_value": round(
|
|
float(np.percentile(utils, pct)), 4
|
|
),
|
|
"perf_metric_type": "THROUGHPUT",
|
|
}
|
|
)
|
|
|
|
finally:
|
|
await serve.shutdown_async()
|
|
# Let the cluster stabilize before the next deploy,
|
|
# especially important at large scales where actor
|
|
# teardown/creation causes resource churn.
|
|
settle_s = max(5, num_replicas // 32)
|
|
logger.info(f"Settling for {settle_s}s before next config...")
|
|
await asyncio.sleep(settle_s)
|
|
|
|
return {"perf_metrics": perf_metrics, "utilization_raw": utilization_raw}
|
|
|
|
|
|
# ===================================================================
|
|
# CLI entry point
|
|
# ===================================================================
|
|
|
|
|
|
@click.command()
|
|
@click.option("--output-path", "-o", type=str, default=None)
|
|
@click.option(
|
|
"--num-replicas",
|
|
"-n",
|
|
multiple=True,
|
|
type=int,
|
|
default=NUM_REPLICAS,
|
|
help="Replica counts to benchmark. Default: 512, 128, 32, 8.",
|
|
)
|
|
@click.option(
|
|
"--router-type",
|
|
"-r",
|
|
multiple=True,
|
|
type=click.Choice(["pow2", "capacity_queue"]),
|
|
required=True,
|
|
help="Routers to benchmark. Repeat flag for multiple.",
|
|
)
|
|
@click.option(
|
|
"--simulated-latency-mean-s",
|
|
type=float,
|
|
required=True,
|
|
help="Mean of the exponential distribution for simulated child work.",
|
|
)
|
|
@click.option(
|
|
"--simulated-latency-cap-s",
|
|
type=float,
|
|
required=True,
|
|
help="Cap on simulated child work latency.",
|
|
)
|
|
@click.option(
|
|
"--max-ongoing-requests-child",
|
|
type=int,
|
|
required=True,
|
|
help="max_ongoing_requests for the child deployment.",
|
|
)
|
|
@click.option(
|
|
"--max-ongoing-requests-parent",
|
|
type=int,
|
|
required=True,
|
|
help="max_ongoing_requests for the parent deployment.",
|
|
)
|
|
@click.option(
|
|
"--max-users-per-task",
|
|
type=int,
|
|
default=MAX_USERS_PER_TASK,
|
|
show_default=True,
|
|
help="Max closed-loop users assigned to each load-generator actor.",
|
|
)
|
|
def main(
|
|
output_path: Optional[str],
|
|
num_replicas: List[int],
|
|
router_type: List[str],
|
|
simulated_latency_mean_s: float,
|
|
simulated_latency_cap_s: float,
|
|
max_ongoing_requests_child: int,
|
|
max_ongoing_requests_parent: int,
|
|
max_users_per_task: int,
|
|
):
|
|
workload = WorkloadConfig(
|
|
simulated_latency_mean_s=simulated_latency_mean_s,
|
|
simulated_latency_cap_s=simulated_latency_cap_s,
|
|
max_ongoing_requests_child=max_ongoing_requests_child,
|
|
max_ongoing_requests_parent=max_ongoing_requests_parent,
|
|
)
|
|
logger.info(
|
|
f"Running router benchmark: replicas={list(num_replicas)} "
|
|
f"routers={list(router_type)} workload={workload} "
|
|
f"max_users_per_task={max_users_per_task}"
|
|
)
|
|
|
|
results = asyncio.run(
|
|
run_router_benchmark(
|
|
workload=workload,
|
|
warmup_s=WARMUP_S,
|
|
duration_s=DURATION_S,
|
|
num_replicas_list=list(num_replicas),
|
|
router_types=list(router_type),
|
|
max_users_per_task=max_users_per_task,
|
|
)
|
|
)
|
|
|
|
logger.info(f"Perf metrics:\n{json.dumps(results['perf_metrics'], indent=4)}")
|
|
save_test_results(results, output_path=output_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|