diff --git a/src/deployments/experiments/libp2p/multi_shadow_gossipsub.py b/src/deployments/experiments/libp2p/multi_shadow_gossipsub.py index 0ac8f738..ae3e2e3b 100644 --- a/src/deployments/experiments/libp2p/multi_shadow_gossipsub.py +++ b/src/deployments/experiments/libp2p/multi_shadow_gossipsub.py @@ -1,5 +1,10 @@ -# Drives shadow-gossipsub at multiple scales via the Multiple meta-experiment. -# Run with: uv run python deployment.py multi-shadow-gossipsub --skip-check +# Drives shadow-gossipsub across scales and/or muxers via the Multiple meta-experiment. +# Defaults give the scale sweep; env vars turn it into the muxer regression matrix, e.g. +# SHADOW_MUXERS=mplex,yamux,quic SHADOW_SIZES=1000 SHADOW_DISCOVERY=kad-dht \ +# uv run python deployment.py multi-shadow-gossipsub --values --skip-check +# The base (message count, timing, image, resources) comes from --values; the runner +# only overlays the swept dimensions on top. +import os from typing import Any, Iterable, List from src.deployments.experiments.libp2p.shadow_gossipsub import ShadowGossipsubExperiment @@ -7,9 +12,18 @@ from src.deployments.registry import experiment +def _env_list(name: str, default: str) -> list: + return [x.strip() for x in os.environ.get(name, default).split(",") if x.strip()] + + +SIZES = [int(x) for x in _env_list("SHADOW_SIZES", "10,30,100")] +MUXERS = _env_list("SHADOW_MUXERS", "yamux") +DISCOVERY = os.environ.get("SHADOW_DISCOVERY", "static") + + @experiment(name="multi-shadow-gossipsub") class MultiShadowGossipsub(Multiple): - """Run shadow-gossipsub multiple times at different scales.""" + """Run shadow-gossipsub across scales and/or muxers (see env vars above).""" def model_post_init(self, __context: Any) -> None: self.config.name = ShadowGossipsubExperiment.name @@ -20,8 +34,10 @@ def get_params_list(self) -> List[dict]: return list(self.exp_params()) def exp_params(self) -> Iterable[dict]: - for num_nodes in (10, 30, 100): - yield {"num_nodes": num_nodes} + for size in SIZES: + for muxer in MUXERS: + yield {"num_nodes": size, "muxer": muxer, "discovery": DISCOVERY} def get_name_from_params(self, params: dict) -> str: - return f"num_nodes_{params['num_nodes']}" + disc = "kad" if params["discovery"] == "kad-dht" else params["discovery"] + return f"n{params['num_nodes']}-{params['muxer']}-{disc}" diff --git a/src/deployments/experiments/libp2p/shadow_gossipsub.py b/src/deployments/experiments/libp2p/shadow_gossipsub.py index 9c424ed8..f7942c1a 100644 --- a/src/deployments/experiments/libp2p/shadow_gossipsub.py +++ b/src/deployments/experiments/libp2p/shadow_gossipsub.py @@ -1,7 +1,7 @@ # Shadow GossipSub experiment: N nim libp2p peers + 1 publisher inside Shadow on a # single k8s pod. See the "Using Shadow at DST" runbook in Notion. import logging -from typing import Optional +from typing import Literal, Optional from pydantic import BaseModel, ConfigDict, NonNegativeFloat, NonNegativeInt @@ -31,11 +31,24 @@ class ExpConfig(BaseModel): message_size_bytes: NonNegativeInt = 1000 delay_seconds: NonNegativeFloat = 2.0 connect_to: NonNegativeInt = 2 + muxer: Literal["yamux", "mplex", "quic"] = "yamux" + discovery: Literal["static", "kad-dht"] = "static" # CONNECTTO dial vs kad bootstrap + start_sleep: NonNegativeInt = 60 # node STARTSLEEP before mesh formation # Timing (simulated seconds). Publisher starts after the mesh forms (~60s). publisher_start_s: NonNegativeInt = 90 sim_stop_time_s: NonNegativeInt = 180 # storeMetrics scrape cadence (s); short so the last scrape is post-traffic. metrics_interval_s: NonNegativeInt = 15 + # Determinism + diagnostics. seed is rendered into shadow.yaml (Shadow default 1); + # strace is global and heavy — small-N diagnosis only. + seed: NonNegativeInt = 1 + model_unblocked_syscall_latency: bool = False + strace_logging_mode: str = "off" # off | standard | deterministic + # Floor (µs) for lsquic engine tick re-arms; needs the tick-floor node image. + # 0 = stock lsquic behavior (which livelocks quic under Shadow). + lsquic_tick_floor_us: NonNegativeInt = 0 + # Per-pod process start stagger (pod-i starts at 5000 + i*jitter ms); 0 = lockstep. + start_jitter_ms: NonNegativeInt = 0 # Job-pod resources, sized for ~10 peers; bump for bigger sims. cpu_request: str = "2" cpu_limit: str = "4" @@ -59,8 +72,10 @@ async def _run(self): self.log_event("run_start") cfg = self.config namespace = self.namespace - # unique per run (output folder name carries a random suffix) - run_id = self.output_folder.name.lower().replace("_", "-")[:50].strip("-") + # unique per run (output folder name carries a random suffix). Cap the length + # so the derived `shadow--reader` log-reader pod stays within the k8s + # 63-char name limit. + run_id = self.output_folder.name.lower().replace("_", "-")[:45].strip("-") cm_name = f"shadow-{run_id}" job_name = f"shadow-{run_id}" pvc_name = f"shadow-{run_id}-data" @@ -70,7 +85,15 @@ async def _run(self): sim_stop_time_s=cfg.sim_stop_time_s, publisher_start_s=cfg.publisher_start_s, connect_to=cfg.connect_to, + muxer=cfg.muxer, + discovery=cfg.discovery, + start_sleep=cfg.start_sleep, metrics_interval_s=cfg.metrics_interval_s, + seed=cfg.seed, + model_unblocked_syscall_latency=cfg.model_unblocked_syscall_latency, + strace_logging_mode=cfg.strace_logging_mode, + lsquic_tick_floor_us=cfg.lsquic_tick_floor_us, + start_jitter_ms=cfg.start_jitter_ms, ) publisher_config = render_publisher_config( num_nodes=cfg.num_nodes, diff --git a/src/deployments/shadow/builders.py b/src/deployments/shadow/builders.py index a26e3b23..ccf06a42 100644 --- a/src/deployments/shadow/builders.py +++ b/src/deployments/shadow/builders.py @@ -1,6 +1,6 @@ # Builders for Shadow simulator runs: config values -> kubernetes-client objects # and yaml dicts. Pure data, no I/O. See the "Using Shadow at DST" runbook. -from typing import Optional +from typing import Literal, Optional from kubernetes.client import ( V1Capabilities, @@ -42,45 +42,95 @@ ) -def render_shadow_yaml( +def _peer_env( *, num_nodes: int, - sim_stop_time_s: int, - publisher_start_s: int, - connect_to: int = 2, - muxer: str = "yamux", - metrics_interval_s: int = 15, - requester_app_path: str = _REQUESTER_APP_PATH, + connect_to: int, + muxer: str, + discovery: str, + start_sleep: int, + metrics_interval_s: int, + lsquic_tick_floor_us: int, ) -> dict: - """Build the shadow.yaml dict: N peer hosts running `./main` + a publisher host - running the pod-api-requester in batch mode against the peers' `/publish` - endpoints. The traffic shape (message count, size, pacing) lives in the - requester's own config (see `render_publisher_config`), mounted at - `{_CONFIG_MOUNT}/{_PUBLISHER_CONFIG}`.""" - if connect_to >= num_nodes: - raise ValueError(f"connect_to ({connect_to}) must be smaller than num_nodes ({num_nodes}).") - - peer_env = { + """Environment for a relay (pod-*) node.""" + env = { "PEERS": str(num_nodes), "CONNECTTO": str(connect_to), "SHADOWENV": "true", # env.nim requires the literal string "true" "MUXER": muxer, + "DISCOVERY": discovery, + "STARTSLEEP": str(start_sleep), "METRICS_INTERVAL_S": str(metrics_interval_s), } - peer_process = { - "path": "./main", - "start_time": "5s", - "expected_final_state": "running", # daemon: don't error when alive at stop_time - "environment": peer_env, - } - hosts = { + if lsquic_tick_floor_us > 0: + # Needs the tick-floor test-node image (stock images ignore the env var). + env["LSQUIC_TICK_FLOOR_US"] = str(lsquic_tick_floor_us) + if discovery == "kad-dht": + env["NODE_ROLE"] = "RoleNormal" + env["SERVICE"] = "bootstrap-0" + return env + + +def _peer_hosts(num_nodes: int, peer_env: dict, start_jitter_ms: int) -> dict: + """The N relay hosts (pod-0..pod-(N-1)). start_jitter_ms staggers per-pod process + start so peers don't wake and dial at one simulated instant (lockstep wakes force + simultaneous-dial collisions that never occur on real hosts).""" + return { f"pod-{i}": { "network_node_id": 0, - "processes": [peer_process], + "processes": [ + { + "path": "./main", + "start_time": f"{5000 + i * start_jitter_ms}ms", + # daemon: don't error when alive at stop_time + "expected_final_state": "running", + "environment": peer_env, + } + ], } for i in range(num_nodes) } - hosts["publisher"] = { + + +def _bootstrap_host( + *, + num_nodes: int, + muxer: str, + start_sleep: int, + metrics_interval_s: int, + lsquic_tick_floor_us: int, +) -> dict: + """The kad-dht anchor (bootstrap-0); peers discover through it by hostname.""" + env = { + "PEERS": str(num_nodes), + "SHADOWENV": "true", + "MUXER": muxer, + "DISCOVERY": "kad-dht", + "NODE_ROLE": "RoleBootstrap", + # the single anchor must accept every node's bootstrap dial, so lift its cap + # above the network size (default is 250). + "MAXCONNECTIONS": str(num_nodes + 100), + "STARTSLEEP": str(start_sleep), + "METRICS_INTERVAL_S": str(metrics_interval_s), + } + if lsquic_tick_floor_us > 0: + env["LSQUIC_TICK_FLOOR_US"] = str(lsquic_tick_floor_us) + return { + "network_node_id": 0, + "processes": [ + { + "path": "./main", + "start_time": "5s", + "expected_final_state": "running", + "environment": env, + } + ], + } + + +def _publisher_host(publisher_start_s: int, requester_app_path: str) -> dict: + """The publisher host: runs the pod-api-requester in batch mode against the peers.""" + return { "network_node_id": 0, "processes": [ { @@ -94,16 +144,75 @@ def render_shadow_yaml( } ], } - return { + + +def render_shadow_yaml( + *, + num_nodes: int, + sim_stop_time_s: int, + publisher_start_s: int, + connect_to: int = 2, + muxer: Literal["yamux", "mplex", "quic"] = "yamux", + discovery: Literal["static", "kad-dht"] = "static", + start_sleep: int = 60, + metrics_interval_s: int = 15, + seed: int = 1, + model_unblocked_syscall_latency: bool = False, + strace_logging_mode: str = "off", + lsquic_tick_floor_us: int = 0, + start_jitter_ms: int = 0, + requester_app_path: str = _REQUESTER_APP_PATH, +) -> dict: + """Build the shadow.yaml dict: N peer hosts running `./main` + a publisher host + running the pod-api-requester in batch mode against the peers' `/publish` + endpoints. The traffic shape (message count, size, pacing) lives in the + requester's own config (see `render_publisher_config`), mounted at + `{_CONFIG_MOUNT}/{_PUBLISHER_CONFIG}`. + + discovery selects mesh formation: "static" dials CONNECTTO peers by pod-N + hostname; "kad-dht" adds a `bootstrap-0` anchor host that peers discover + through (Shadow resolves it by hostname, so no k8s Service is needed).""" + if connect_to >= num_nodes: + raise ValueError(f"connect_to ({connect_to}) must be smaller than num_nodes ({num_nodes}).") + + peer_env = _peer_env( + num_nodes=num_nodes, + connect_to=connect_to, + muxer=muxer, + discovery=discovery, + start_sleep=start_sleep, + metrics_interval_s=metrics_interval_s, + lsquic_tick_floor_us=lsquic_tick_floor_us, + ) + hosts = _peer_hosts(num_nodes, peer_env, start_jitter_ms) + if discovery == "kad-dht": + hosts["bootstrap-0"] = _bootstrap_host( + num_nodes=num_nodes, + muxer=muxer, + start_sleep=start_sleep, + metrics_interval_s=metrics_interval_s, + lsquic_tick_floor_us=lsquic_tick_floor_us, + ) + hosts["publisher"] = _publisher_host(publisher_start_s, requester_app_path) + # Always render the seed so the run's shadow.yaml records it (Shadow defaults to 1). + config = { "general": { "stop_time": f"{sim_stop_time_s}s", "progress": True, + "seed": seed, }, "network": { "graph": {"type": "1_gbit_switch"}, }, "hosts": hosts, } + if model_unblocked_syscall_latency: + config["general"]["model_unblocked_syscall_latency"] = True + if strace_logging_mode != "off": + # Global (all hosts) and voluminous — a straced host writes ~100s of MB per + # simulated minute of activity. Diagnostics at small N only. + config["experimental"] = {"strace_logging_mode": strace_logging_mode} + return config def render_publisher_config( diff --git a/src/deployments/shadow/tests/test_builders.py b/src/deployments/shadow/tests/test_builders.py index ff217f35..543bef07 100644 --- a/src/deployments/shadow/tests/test_builders.py +++ b/src/deployments/shadow/tests/test_builders.py @@ -4,6 +4,10 @@ from ruamel.yaml import YAML from src.deployments.shadow.builders import ( + _bootstrap_host, + _peer_env, + _peer_hosts, + _publisher_host, build_configmap, build_pvc, build_shadow_job, @@ -69,6 +73,77 @@ def test_request_and_action_cross_references_resolve(self): assert cfg["actions"][0]["targets"] == [cfg["targets"][0]["name"]] +# --------------------------------------------------------------------------- # +# host builders (the pieces render_shadow_yaml assembles) +# --------------------------------------------------------------------------- # +class TestHostBuilders: + def _env(self, **over): + base = dict( + num_nodes=5, + connect_to=2, + muxer="yamux", + discovery="static", + start_sleep=60, + metrics_interval_s=15, + lsquic_tick_floor_us=0, + ) + base.update(over) + return _peer_env(**base) + + def test_peer_env_static_omits_kad_and_tickfloor(self): + env = self._env() + assert env["MUXER"] == "yamux" and env["DISCOVERY"] == "static" + assert "NODE_ROLE" not in env and "SERVICE" not in env + assert "LSQUIC_TICK_FLOOR_US" not in env + + def test_peer_env_kad_adds_role_and_service(self): + env = self._env(discovery="kad-dht") + assert env["NODE_ROLE"] == "RoleNormal" + assert env["SERVICE"] == "bootstrap-0" + + def test_peer_env_tickfloor_only_when_set(self): + assert ( + _peer_env( + num_nodes=5, + connect_to=2, + muxer="quic", + discovery="static", + start_sleep=60, + metrics_interval_s=15, + lsquic_tick_floor_us=10000, + )["LSQUIC_TICK_FLOOR_US"] + == "10000" + ) + + def test_peer_hosts_stagger_start_time_by_jitter(self): + hosts = _peer_hosts(3, {"MUXER": "yamux"}, start_jitter_ms=50) + assert [hosts[f"pod-{i}"]["processes"][0]["start_time"] for i in range(3)] == [ + "5000ms", + "5050ms", + "5100ms", + ] + assert all(hosts[h]["processes"][0]["expected_final_state"] == "running" for h in hosts) + + def test_bootstrap_host_lifts_connection_cap(self): + h = _bootstrap_host( + num_nodes=1000, + muxer="yamux", + start_sleep=60, + metrics_interval_s=15, + lsquic_tick_floor_us=0, + ) + env = h["processes"][0]["environment"] + assert env["NODE_ROLE"] == "RoleBootstrap" + assert env["MAXCONNECTIONS"] == "1100" # num_nodes + 100 + + def test_publisher_host_runs_requester_batch(self): + h = _publisher_host(90, "/app/api_requester.py") + proc = h["processes"][0] + assert proc["path"] == "/usr/bin/python3" + assert "--mode batch" in proc["args"] + assert proc["start_time"] == "90s" + + # --------------------------------------------------------------------------- # # render_shadow_yaml (peer hosts + publisher host) # --------------------------------------------------------------------------- # @@ -108,7 +183,7 @@ def test_peer_process_is_daemon_with_env(self): ) peer = sy["hosts"]["pod-0"]["processes"][0] assert peer["path"] == "./main" - assert peer["start_time"] == "5s" + assert peer["start_time"] == "5000ms" assert peer["expected_final_state"] == "running" env = peer["environment"] assert env["PEERS"] == "3"