d8208685a2
Reactor/loop-pool engine with TCP/µTP/MSE transports, per-connection pipelining, priority-driven piece selection with endgame, and the Python FFI test harness. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
206 lines
7.4 KiB
Python
206 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import secrets
|
|
import sys
|
|
import time
|
|
import traceback
|
|
|
|
ROOT = "/work"
|
|
sys.path.insert(0, os.path.join(ROOT, "harness"))
|
|
|
|
from common import load_metadata, resolve_ipv4, write_manifest # noqa: E402
|
|
from engine_ffi import Engine, EngineConfig, ERROR_NAMES, STATE_ERROR, STATE_NAMES # noqa: E402
|
|
|
|
|
|
LIB = "/work/build/libtorrentpeer.so"
|
|
|
|
|
|
def make_peer_id() -> bytes:
|
|
return b"-PC0001-" + secrets.token_bytes(12)
|
|
|
|
|
|
def load_clients(path: str, only: set[str] | None) -> list[dict]:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
clients = json.load(f)
|
|
if only:
|
|
clients = [c for c in clients if c["name"] in only]
|
|
if not clients:
|
|
raise ValueError("no clients selected")
|
|
return clients
|
|
|
|
|
|
def download_from_client(meta, client: dict, timeout: float, lib_path: str) -> dict:
|
|
ip = resolve_ipv4(client["host"])
|
|
port = int(client["port"])
|
|
cfg = EngineConfig(
|
|
loop_count=1,
|
|
slots_per_loop=int(client.get("slots_per_loop", 1024)),
|
|
max_pipeline=int(client.get("max_pipeline", 256)),
|
|
request_timeout_ms=int(client.get("request_timeout_ms", 10000)),
|
|
encryption=int(client.get("encryption", 0)),
|
|
utp=int(client.get("utp", 0)),
|
|
connect_timeout_ms=int(client.get("connect_timeout_ms", 5000)),
|
|
fallback=int(client.get("fallback", 0)),
|
|
)
|
|
|
|
started = time.time()
|
|
buffers = [bytearray(meta.piece_len(i)) for i in range(meta.num_pieces)]
|
|
received = [0] * meta.num_pieces
|
|
done = bytearray(meta.num_pieces)
|
|
done_count = 0
|
|
last_progress = started
|
|
last_bytes = 0
|
|
status_snap = None
|
|
|
|
with Engine(cfg, lib_path=lib_path, poll_batch=2048) as eng:
|
|
tid = eng.add_torrent(
|
|
meta.info_hash,
|
|
make_peer_id(),
|
|
meta.piece_length,
|
|
meta.total_size,
|
|
meta.num_pieces,
|
|
)
|
|
eng.set_priorities(tid, [1] * meta.num_pieces)
|
|
eng.add_peer(tid, ip, port)
|
|
|
|
while done_count < meta.num_pieces:
|
|
status_snap = eng.status(tid)
|
|
if status_snap.state == STATE_ERROR:
|
|
raise RuntimeError(f"engine error: {ERROR_NAMES[status_snap.error]}")
|
|
|
|
descs = eng.poll_ready()
|
|
if not descs:
|
|
eng.wait(200)
|
|
now = time.time()
|
|
if status_snap.bytes_received != last_bytes:
|
|
last_bytes = status_snap.bytes_received
|
|
last_progress = now
|
|
if now - started > timeout or now - last_progress > timeout:
|
|
raise TimeoutError(
|
|
f"stalled after {now - started:.1f}s: "
|
|
f"{done_count}/{meta.num_pieces} pieces, "
|
|
f"state={STATE_NAMES[status_snap.state]}, "
|
|
f"connected={status_snap.peers_connected}, "
|
|
f"failed={status_snap.peers_failed}, "
|
|
f"outstanding={status_snap.outstanding}"
|
|
)
|
|
continue
|
|
|
|
for block in descs:
|
|
buf = buffers[block.piece]
|
|
buf[block.begin:block.begin + block.len] = eng.block_data(
|
|
block.loop, block.slot, block.len
|
|
)
|
|
eng.release(block.loop, block.slot)
|
|
received[block.piece] += block.len
|
|
if not done[block.piece] and received[block.piece] >= meta.piece_len(block.piece):
|
|
digest = hashlib.sha1(bytes(buf)).digest()
|
|
if digest != meta.piece_hashes[block.piece]:
|
|
raise ValueError(f"piece {block.piece} hash mismatch")
|
|
done[block.piece] = 1
|
|
done_count += 1
|
|
eng.set_priority(tid, block.piece, 0)
|
|
|
|
status_snap = eng.status(tid)
|
|
|
|
full = hashlib.sha1()
|
|
for buf in buffers:
|
|
full.update(buf)
|
|
elapsed = time.time() - started
|
|
return {
|
|
"name": client["name"],
|
|
"ok": True,
|
|
"host": client["host"],
|
|
"ip": ip,
|
|
"port": port,
|
|
"utp": cfg.utp,
|
|
"encryption": cfg.encryption,
|
|
"fallback": cfg.fallback,
|
|
"pieces": done_count,
|
|
"bytes": meta.total_size,
|
|
"sha1": full.hexdigest(),
|
|
"elapsed_s": elapsed,
|
|
"mbps": (meta.total_size / 1e6 / elapsed) if elapsed > 0 else 0.0,
|
|
"peers_connected": int(status_snap.peers_connected if status_snap else 0),
|
|
"peers_failed": int(status_snap.peers_failed if status_snap else 0),
|
|
"state": STATE_NAMES[status_snap.state] if status_snap else "UNKNOWN",
|
|
"error": ERROR_NAMES[status_snap.error] if status_snap else "OK",
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="Run offline client interop matrix.")
|
|
ap.add_argument("--fixture", required=True)
|
|
ap.add_argument("--clients", required=True)
|
|
ap.add_argument("--results", required=True)
|
|
ap.add_argument("--timeout", type=float, default=90.0)
|
|
ap.add_argument("--lib", default=LIB)
|
|
ap.add_argument("--only", action="append",
|
|
help="client name to run; may be repeated")
|
|
args = ap.parse_args()
|
|
|
|
torrent_path = os.path.join(args.fixture, "test.torrent")
|
|
manifest_path = os.path.join(args.fixture, "manifest.json")
|
|
meta = load_metadata(torrent_path)
|
|
with open(manifest_path, "r", encoding="utf-8") as f:
|
|
manifest = json.load(f)
|
|
|
|
clients = load_clients(args.clients, set(args.only or []) or None)
|
|
print(f"fixture: {meta.name}, {meta.total_size} bytes, "
|
|
f"{meta.num_pieces} pieces", flush=True)
|
|
print(f"clients: {', '.join(c['name'] for c in clients)}", flush=True)
|
|
|
|
results = []
|
|
for client in clients:
|
|
print(f"==> {client['name']} ({client['host']}:{client['port']})", flush=True)
|
|
try:
|
|
result = download_from_client(meta, client, args.timeout, args.lib)
|
|
if result["sha1"] != manifest["sha1"]:
|
|
raise ValueError(
|
|
f"full-file sha1 mismatch: got {result['sha1']} expected {manifest['sha1']}"
|
|
)
|
|
print(f" ok: {result['bytes']/1e6:.1f} MB in "
|
|
f"{result['elapsed_s']:.2f}s ({result['mbps']:.1f} MB/s)",
|
|
flush=True)
|
|
except Exception as exc:
|
|
result = {
|
|
"name": client["name"],
|
|
"ok": False,
|
|
"host": client.get("host"),
|
|
"port": client.get("port"),
|
|
"utp": client.get("utp", 0),
|
|
"encryption": client.get("encryption", 0),
|
|
"fallback": client.get("fallback", 0),
|
|
"error": str(exc),
|
|
"traceback": traceback.format_exc(),
|
|
}
|
|
print(f" FAIL: {exc}", flush=True)
|
|
results.append(result)
|
|
|
|
os.makedirs(os.path.dirname(args.results), exist_ok=True)
|
|
write_manifest(
|
|
args.results,
|
|
fixture=manifest,
|
|
timeout_s=args.timeout,
|
|
results=results,
|
|
passed=sum(1 for r in results if r["ok"]),
|
|
failed=sum(1 for r in results if not r["ok"]),
|
|
)
|
|
|
|
print("-" * 72)
|
|
for r in results:
|
|
if r["ok"]:
|
|
print(f"PASS {r['name']:<24} {r['mbps']:8.1f} MB/s "
|
|
f"{r['state']} enc={r['encryption']} utp={r['utp']}")
|
|
else:
|
|
print(f"FAIL {r['name']:<24} {r['error']}")
|
|
return 0 if all(r["ok"] for r in results) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
|