shadow-gossipsub: select muxer and discovery (static | kad-dht) (#316)

radiken · web-flow · commit 2075b9f7aa72 · 2026-07-17T21:02:52.000+01:00
Expose muxer, discovery and start_sleep on ExpConfig and wire them into shadow.yaml.
diff --git a/src/deployments/experiments/libp2p/multi_shadow_gossipsub.py b/src/deployments/experiments/libp2p/multi_shadow_gossipsub.py
@@ -1,15 +1,29 @@
-# 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 <base.yaml> --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
 from src.deployments.experiments.multi_experiment import Multiple
 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
@@ -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-<run_id>-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
@@ -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
