Support benchmark smoke runs on small CI hosts

Author: zzylol

.github/workflows/asap-sketch-benchmark.yml

@@ -33,7 +33,7 @@ jobs:
       - name: Test streaming window correctness
         run: cargo test --manifest-path asap-precompute-rs/Cargo.toml --release --features stream-benchmark --bin asap-stream-bench
       - name: Run persistent OTLP network benchmark
-        run: ./benchmarks/run-nightly.sh
+        run: ./benchmarks/run-nightly.sh --allow-shared-infrastructure
       - name: Verify streaming measurements
         run: |
           python3 - <<'PYCODE'
@@ -62,8 +62,10 @@ jobs:
           for path in root.glob("traffic-*-sps/window-*/*mbit/*/*/config.json"):
               config = json.loads(path.read_text())
               assigned = list(config["component_cores"].values())
-              assert len(assigned) == 5 and len(set(assigned)) == 5
-              assert set(config["generator_cores"]).isdisjoint(assigned)
+              assert len(assigned) == 4 and len(set(assigned)) == 4
+              assert config["placement"] in {"exclusive-capacity", "shared-infrastructure-smoke"}
+              if config["placement"] == "exclusive-capacity":
+                  assert set(config["generator_cores"]).isdisjoint(assigned)
           assert (root / "throughput.svg").is_file()
           assert (root / "resources.svg").is_file()
           PYCODE

benchmarks/STREAMING.md

@@ -82,6 +82,12 @@ No component has a Docker memory limit. The exact mapping is recorded as
 `--traffic-rates 10000 25000 50000 100000 200000 300000 400000` controls offered traffic per source and
 `--window-points 16384 65536 262144` controls aggregation volume per source.
 Together they form the default two-dimensional ingestion-rate/window-size sweep.
+
+`--allow-shared-infrastructure` exists only for correctness smoke tests on CI
+runners with fewer than nine CPUs. It keeps the four evaluated processors on
+distinct CPUs but allows generator and backend infrastructure to overlap them.
+The resulting `placement` is recorded as `shared-infrastructure-smoke`; those
+runs must not be used for throughput capacity or resource-efficiency claims.
 `--batch-size` independently controls the OTLP request size.
 Bandwidth shaping uses Linux `tc tbf` on each branch's **data** interface only
 (`NET_ADMIN` only on those containers), with a 32 KiB burst and 100 ms queue.

benchmarks/run-streaming.py

@@ -98,16 +98,30 @@ def setup(self):
         # least one CPU available for it. Memory is intentionally uncapped for
         # every role; the harness reports actual RSS instead.
         required = self.generator_core_count + len(PROCESSOR_ROLES) + 1
-        if len(cores) < required:
+        shared_infrastructure = len(cores) < required and self.args.allow_shared_infrastructure
+        if len(cores) < required and not shared_infrastructure:
             raise RuntimeError(
                 f"dedicated placement needs {required} CPUs: {self.generator_core_count} "
                 f"traffic-generator CPUs plus one CPU for each of {len(PROCESSOR_ROLES)} processors "
                 "and at least one backend CPU; "
                 f"only {len(cores)} are available"
             )
-        generators = cores[:self.generator_core_count]
-        component_cores = dict(zip(PROCESSOR_ROLES, cores[self.generator_core_count:self.generator_core_count + len(PROCESSOR_ROLES)]))
-        backend_cores = cores[self.generator_core_count + len(PROCESSOR_ROLES):]
+        if shared_infrastructure:
+            if len(cores) < len(PROCESSOR_ROLES):
+                raise RuntimeError(
+                    f"shared-infrastructure smoke placement still needs {len(PROCESSOR_ROLES)} CPUs "
+                    f"for distinct processor assignments; only {len(cores)} are available"
+                )
+            # Correctness smoke mode for small CI runners. The four evaluated
+            # processors remain distinct, while generator/backend infrastructure
+            # may overlap them. Results from this placement are not capacity data.
+            generators = cores[:min(self.generator_core_count, len(cores))]
+            component_cores = dict(zip(PROCESSOR_ROLES, cores[:len(PROCESSOR_ROLES)]))
+            backend_cores = cores
+        else:
+            generators = cores[:self.generator_core_count]
+            component_cores = dict(zip(PROCESSOR_ROLES, cores[self.generator_core_count:self.generator_core_count + len(PROCESSOR_ROLES)]))
+            backend_cores = cores[self.generator_core_count + len(PROCESSOR_ROLES):]
         for suffix in ("data", "control"):
             network = self.prefix + "-" + suffix
             self.command("network", "create", "--internal", network)
@@ -191,6 +205,7 @@ def generator_ready():
                   "warmup_seconds": self.args.warmup, "observation_seconds": self.observation_seconds,
                   "generator_cores": generators, "component_cores": component_cores,
                   "backend_cores": backend_cores,
+                  "placement": "shared-infrastructure-smoke" if shared_infrastructure else "exclusive-capacity",
                   "measurement_clock": "CLOCK_MONOTONIC, shared host kernel", "max_source_skew_windows": 1024, "pdata_channel_capacity": 8, "exporter_max_in_flight": 1,
                   "transport": "standard OTLP/HTTP protobuf, uncompressed, persistent connections",
                   "data_interfaces": self.data_ifaces,
@@ -461,6 +476,8 @@ def main():
                         help="target signals/s per source to sweep using OTAP's traffic_generator receiver")
     parser.add_argument("--generator-cores", type=int, default=4,
                         help="OTAP traffic_generator pipeline workers per source; each measured processor gets one dedicated CPU and the backend gets all remaining CPUs")
+    parser.add_argument("--allow-shared-infrastructure", action="store_true",
+                        help="CI correctness smoke mode: keep processors on distinct CPUs but allow generator/backend overlap; results are not capacity measurements")
     parser.add_argument("--profile-cpu", action="store_true", help="exclusive synchronous processor thread CPU scopes")
     parser.add_argument("--account-cpu", action="store_true", help="match CPU/input boundaries with warmup and final drains; implied by --profile-cpu")
     parser.add_argument("--perf-command", default="", help="optional host sampler, e.g. 'sudo -n perf'; records four worker processes")