jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Add forced TCP bench mode and verify observed source timestamp wraps

master
evgeny 22 hours ago
parent
57bf4e490d
commit
46ac9a03b9
2 changed files with 33 additions and 6 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 11
      tests/stream_netns/run.py
  2. 28
      tests/stream_netns/scenario.py

11
tests/stream_netns/run.py
Unescape View File

@ -91,7 +91,9 @@ def main():
parser.add_argument("--duration", type=int, default=120, help="Baseline duration in real seconds")
parser.add_argument("--faults", action="store_true", help="Degrade WAN, recover and check unsubscribe")
parser.add_argument("--fault-duration", type=int, default=30)
parser.add_argument("--tcp", action="store_true", help="Also expose TCP sockets, permitting ETCP multilink")
transport = parser.add_mutually_exclusive_group()
transport.add_argument("--tcp", action="store_true", help="Also expose TCP sockets, permitting ETCP multilink")
transport.add_argument("--tcp-only", action="store_true", help="Expose only TCP, requiring the real STCP transport")
parser.add_argument("--inside", action="store_true", help=argparse.SUPPRESS)
parser.add_argument("--host-netns", help=argparse.SUPPRESS)
args = parser.parse_args()
@ -120,6 +122,8 @@ def main():
command.append("--faults")
if args.tcp:
command.append("--tcp")
if args.tcp_only:
command.append("--tcp-only")
print(f"Artifacts: {directory}", flush=True)
child = subprocess.Popen(command)
try:
@ -135,6 +139,11 @@ def main():
(directory / "host-after.json").write_text(json.dumps(after, indent=2) + "\n")
unchanged = stable_snapshot(before) == stable_snapshot(after) and os.readlink("/proc/self/ns/net") == original_namespace
(directory / "host-check.json").write_text(json.dumps({"unchanged": unchanged}, indent=2) + "\n")
result_path = directory / "result.json"
result = json.loads(result_path.read_text()) if result_path.exists() else {"pass": False, "error": "Isolated runner did not complete"}
result.update(host_unchanged=unchanged, launcher_exit=status)
result["pass"] = result["pass"] and unchanged and status == 0
result_path.write_text(json.dumps(result, indent=2) + "\n")
if not unchanged:
print("FAIL: main network namespace snapshot changed during the run", file=sys.stderr)
return 1

28
tests/stream_netns/scenario.py
Unescape View File

@ -11,7 +11,7 @@ from topology import NODES, WANS
class Agent:
def __init__(self, network, node, binary, seed, tcp):
def __init__(self, network, node, binary, seed, tcp, tcp_only):
self.node = node
directory = network.directory / node.name
directory.mkdir()
@ -20,7 +20,7 @@ class Agent:
self.identity = identity["node"]
kind = "nat" if node.nat else "public"
sockets = "".join(f"[server:{transport}]\naddr={node.ip}:32000\ntype={kind}\ntransport={transport}\n"
for transport in (["udp", "tcp"] if tcp else ["udp"]))
for transport in (["tcp"] if tcp_only else ["udp", "tcp"] if tcp else ["udp"]))
config = (f"[global]\nmy_node_name={node.name}\nmy_private_key={identity['private']}\n"
f"my_public_key={identity['public']}\nclient_type={node.client_type}\nauto_sockets=no\n"
f"db_path={directory / 'db'}\ndebug_level=warn\n"
@ -81,6 +81,9 @@ class Scenario:
self.started = time.monotonic()
self.phases = []
self.latest = {}
self.last_metadata = {}
self.clock_previous = {}
self.clock_wraps = 0
def sample(self, phase):
snapshot = {name: agent.rpc("stats") for name, agent in self.agents.items()}
@ -91,6 +94,16 @@ class Scenario:
raise RuntimeError(f"{name}: payload/order validation failed; see samples.jsonl and utun.log")
if any(s.get("stored_bytes", 0) > 32 * 1024 * 1024 for s in statistics["streams"]):
raise RuntimeError(f"{name}: stream data buffer exceeded 32 MiB")
for stream in statistics["streams"]:
self.last_metadata.setdefault((name, stream["source"]), {}).update(stream)
path = next((p for p in stream.get("paths", []) if p["sub"] == 0), {})
if stream["publisher"] and "source_timestamp_ms" in path:
key = (name, stream["source"])
current = (path["seq"], path["source_timestamp_ms"])
previous = self.clock_previous.get(key)
if previous and current[0] > previous[0] and current[1] < previous[1]:
self.clock_wraps += 1
self.clock_previous[key] = current
return snapshot
def wait(self, phase, predicate, timeout):
@ -121,7 +134,7 @@ class Scenario:
def run(self):
binary = Path(self.args.binary).resolve()
for node in NODES:
self.agents[node.name] = Agent(self.network, node, binary, self.args.seed, self.args.tcp)
self.agents[node.name] = Agent(self.network, node, binary, self.args.seed, self.args.tcp, self.args.tcp_only)
owner = self.agents["n01"]
group = owner.rpc("create")["group"]
print(f"Joining real CHAT group {group}", flush=True)
@ -181,6 +194,8 @@ class Scenario:
generated = self.stream(final[self.args.publisher], source_id)["generated"]
if generated < 10:
raise RuntimeError("Publisher generated insufficient data")
if self.args.duration >= 66 and not self.clock_wraps:
raise RuntimeError("Full-duration run did not observe a source timestamp wrap")
for name in names:
stream = self.stream(final[name], source_id)
if stream["next_seq"] != generated:
@ -190,17 +205,20 @@ class Scenario:
f"start={stream['start_seq']}, generated={generated}")
self.report(final, source_id)
return {"pass": True, "group": group, "publisher": self.args.publisher, "generated": generated,
"phases": self.phases, "subscribers_checked": len(names)}
"phases": self.phases, "subscribers_checked": len(names), "source_timestamp_wraps": self.clock_wraps,
"transport": "tcp" if self.args.tcp_only else "udp+tcp" if self.args.tcp else "udp",
"rate_bps": self.args.rate}
def report(self, final, source):
rows = []
for node in NODES:
stream = self.stream(final[node.name], source)
metadata = self.last_metadata[(node.name, source)]
rows.append(dict(node=node.name, country=node.country, nat=node.nat or "public",
received=stream["received"], gap_batches=stream["gap_batches"], errors=stream["errors"],
p50_ms=stream["latency_p50_ms"], p95_ms=stream["latency_p95_ms"],
p99_ms=stream["latency_p99_ms"], maximum_ms=stream["latency_max_ms"],
migrations=stream.get("migrations", 0), probes=stream.get("probes", 0)))
migrations=metadata.get("migrations", 0), probes=metadata.get("probes", 0)))
with (self.network.directory / "summary.csv").open("w", newline="") as output:
writer = csv.DictWriter(output, fieldnames=list(rows[0]))
writer.writeheader(); writer.writerows(rows)

Write Preview
Loading…
Cancel
Save