# -*- coding: utf-8 -*- """局域网直连 vs 跨网络中转 —— 可重复的实测对比。 为什么要有这个： 产品一直讲"局域网优先"，但从来没有数字。没有数字就无法判断一次 改动是优化还是自我感觉良好，也回答不了"局域网到底快多少、稳多少"。 做法（全部在本机跑，不依赖外网）： 1. 起一个本地中转（relay/relay.py） 2. 起**真实被控端**（windows/run_agent.py），它同时： * 监听局域网 TCP -> 直连那条路 * 登记到本地中转 -> 跨网络那条路 3. 两条路各打 N 次同一个命令（cmd:ping，服务端只回一个时间戳， 几乎不干活），量握手耗时、中位/P90 往返、成功率 两条路都用**同一套 Python 客户端写法**（换行分隔的 JSON）， 所以量到的差异来自路径本身，而不是客户端实现。 LAN : TCP 直连 -> 被控端 中转 : HTTP 长轮询 -> 中转 -> 被控端 用法: python _tests/bench_lan_vs_relay.py [--rounds 40] 注意（踩过的坑）： * run_agent.py 读的是 data_dir()，优先认 YKXMZ_DATA 环境变量。 配置必须写到同一个目录，否则表现成"怎么都不登记"。 * run_agent.py 有单实例锁（端口 8767）。残留进程会让新实例直接退出， 所以跑之前最好确认没有游离的 agent。 """ from __future__ import annotations import argparse import json import os import shutil import socket import statistics import subprocess import sys import tempfile import time import urllib.error import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "windows")) sys.path.insert(0, str(ROOT / "_tests")) _UA = {"User-Agent": "LanRemote/1.8.2"} def free_port() -> int: with socket.socket() as probe: probe.bind(("127.0.0.1", 0)) return probe.getsockname()[1] def percentile(values: list, q: float) -> int: if not values: return 0 ordered = sorted(values) index = min(int(len(ordered) * q), len(ordered) - 1) return int(ordered[index]) # ---------------------------------------------------------------------- # 局域网：TCP 直连，换行分隔 JSON # ---------------------------------------------------------------------- class LanClient: def __init__(self, host: str, port: int): self.host = host self.port = port self.sock = None self.buf = b"" def connect(self, code: str, timeout: float = 8.0) -> dict: self.sock = socket.create_connection((self.host, self.port), timeout) self.sock.settimeout(timeout) self.buf = b"" return self.call({"cmd": "pair", "deviceId": "bench-lan", "deviceName": "基准测试手表", "code": code}, timeout) def call(self, obj: dict, timeout: float = 30.0) -> dict: self.sock.settimeout(timeout) payload = (json.dumps(obj, ensure_ascii=False) + "\n").encode("utf-8") self.sock.sendall(payload) deadline = time.time() + timeout while True: while b"\n" in self.buf: line, _, self.buf = self.buf.partition(b"\n") if not line.strip(): continue message = json.loads(line.decode("utf-8")) if "event" in message: continue return message remaining = deadline - time.time() if remaining <= 0: raise TimeoutError("直连等回复超时") self.sock.settimeout(remaining) chunk = self.sock.recv(65536) if not chunk: raise ConnectionError("直连被对端关闭") self.buf += chunk def close(self) -> None: if self.sock is not None: try: self.sock.close() except OSError: pass self.sock = None # ---------------------------------------------------------------------- # 跨网络：HTTP 长轮询 # ---------------------------------------------------------------------- class RelayClient: def __init__(self, base: str): self.base = base.rstrip("/") self.session = "" self.buf = b"" def _post(self, path: str, body: bytes = b"", headers=None, timeout: float = 40.0): req = urllib.request.Request(self.base + path, data=body, method="POST", headers=dict(_UA)) for name, value in (headers or {}).items(): req.add_header(name, value) with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, resp.read() def attach(self, code: str) -> dict: body = json.dumps({"code": code, "name": "基准测试手表"}, ensure_ascii=False).encode("utf-8") _status, raw = self._post("/v1/client/attach", body) reply = json.loads(raw.decode("utf-8")) self.session = reply.get("session", "") return reply def call(self, obj: dict, timeout: float = 40.0) -> dict: payload = (json.dumps(obj, ensure_ascii=False) + "\n").encode("utf-8") self._post("/v1/client/send", payload, {"X-Session": self.session}, timeout) deadline = time.time() + timeout while True: while b"\n" in self.buf: line, _, self.buf = self.buf.partition(b"\n") if not line.strip(): continue message = json.loads(line.decode("utf-8")) if "event" in message: continue return message remaining = deadline - time.time() if remaining <= 0: raise TimeoutError("中转等回复超时") _status, chunk = self._post("/v1/client/poll", b"", {"X-Session": self.session}, min(30.0, remaining)) self.buf += chunk def measure(name: str, connect_fn, ping_fn, rounds: int) -> dict: """跑一条路：先握手，再打 rounds 次 ping。""" result = {"name": name, "connected": False, "error": "", "connect_ms": 0, "ok": 0, "total": rounds, "samples": []} client = None try: t0 = time.time() client, reply = connect_fn() result["connect_ms"] = int((time.time() - t0) * 1000) if not reply.get("ok"): result["error"] = str(reply.get("error") or "握手被拒") return result result["connected"] = True # 热身一次，排除首包/建链的影响 try: ping_fn(client) except Exception: # noqa: BLE001 pass for i in range(rounds): try: t1 = time.time() pong = ping_fn(client) cost = (time.time() - t1) * 1000 if pong.get("ok"): result["ok"] += 1 result["samples"].append(cost) elif not result["error"]: result["error"] = "第 %d 次被拒: %s" % (i + 1, pong) except Exception as exc: # noqa: BLE001 if not result["error"]: result["error"] = "第 %d 次失败: %s" % (i + 1, exc) except Exception as exc: # noqa: BLE001 result["error"] = "连接阶段失败: %s" % exc finally: if client is not None: try: client.close() except Exception: # noqa: BLE001 pass return result def describe(result: dict) -> str: samples = result["samples"] median = int(statistics.median(samples)) if samples else 0 rate = (result["ok"] * 100.0 / result["total"]) if result["total"] else 0 return ("握手 %4d ms 成功 %2d/%2d (%5.1f%%) " "中位 %4d P90 %4d 最小 %4d 最大 %4d ms" % (result["connect_ms"], result["ok"], result["total"], rate, median, percentile(samples, 0.9), int(min(samples)) if samples else 0, int(max(samples)) if samples else 0)) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--rounds", type=int, default=40, help="局域网那条路的 ping 次数") parser.add_argument("--relay-rounds", type=int, default=2, help="中转那条路的 ping 次数（默认少，见代码里的说明）") parser.add_argument("--relay-timeout", type=float, default=8.0, help="中转那条路单次等待上限（秒）") args = parser.parse_args() work = Path(tempfile.mkdtemp(prefix="lan_bench_")) relay_proc = None agent_proc = None try: relay_port = free_port() agent_port = free_port() relay_url = "http://127.0.0.1:%d" % relay_port code = "246813" print("=" * 66) print(" 局域网直连 vs 跨网络中转 （每种路径 %d 次 cmd:ping）" % args.rounds) print("=" * 66) # ---------------- 1. 本地中转 ---------------- print("一、起本地中转（端口 %d）" % relay_port) relay_data = work / "relaydata" relay_data.mkdir(parents=True, exist_ok=True) env = dict(os.environ) env.update({ "RELAY_PORT": str(relay_port), "RELAY_BIND": "127.0.0.1", "RELAY_DATA": str(relay_data), "RELAY_POLL": "2", "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1", }) relay_proc = subprocess.Popen( [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env, stdout=open(work / "relay.log", "wb"), stderr=subprocess.STDOUT) up = False for _ in range(60): time.sleep(0.5) try: with urllib.request.urlopen(relay_url + "/v1/health", timeout=2) as resp: if resp.status == 200: up = True break except Exception: # noqa: BLE001 continue if not up: print(" [失败] 中转没起来") print((work / "relay.log").read_text(encoding="utf-8", errors="replace")[-700:]) return 1 print(" 中转就绪") # ---------------- 2/3/4. 每条路各用一台**全新的**被控端 ---------------- from lanremote.config import Store # # 为什么要各起一台： # 一开始两条路共用同一台被控端（先走局域网、再走中转）， # 结果局域网那 30 次跑完、轮到中转时整个卡住不返回。 # 真实被控端同时被"一条 TCP 会话 + 一条中转会话"占用时行为 # 和单独用一条不一样。搞不清楚就别共用了 —— 每条路起一台干净实例， # 既排除了互相干扰，两边条件也完全一致，对比才有意义。 def start_agent(): data = Path(tempfile.mkdtemp(prefix="bench_agent_", dir=str(work))) store = Store(data) store.set_pairing_code(code) store.set("setup_done", True) store.set("allow_new_pairing", True) store.set("close_pair_after_first", False) store.set("server_name", "基准测试电脑") store.set("relay_url", relay_url) store.set("port", agent_port) env2 = dict(os.environ) env2.update({ "YKXMZ_DATA": str(data), # 用户自己的被控端很可能正在跑，占着默认的单实例锁端口 8767。 # 不换端口的话，测试实例一启动就"已有实例在运行，退出"， # 表现成"怎么都不登记到中转"（这个坑真踩了）。 "YKXMZ_LOCK_PORT": str(free_port()), "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1", }) proc = subprocess.Popen( [sys.executable, "-u", str(ROOT / "windows" / "run_agent.py"), "--silent", "--port", str(agent_port)], cwd=str(ROOT / "windows"), env=env2, stdout=open(data / "stdout.log", "wb"), stderr=subprocess.STDOUT) # 等它登记到中转 for _ in range(45): time.sleep(1.0) if proc.poll() is not None: raise RuntimeError("被控端启动即退出（码 %s）：\n%s" % (proc.returncode, (data / "silent.log").read_text( encoding="utf-8", errors="replace")[-400:])) try: with urllib.request.urlopen(relay_url + "/v1/health", timeout=4) as resp: snap = json.loads(resp.read().decode("utf-8")) except Exception: # noqa: BLE001 continue if any(d.get("online") for d in snap.get("devices", [])): return proc, data raise RuntimeError("被控端 45 秒内没登记到中转") def stop_agent(proc): if proc is None: return try: proc.terminate() proc.wait(timeout=6) except Exception: # noqa: BLE001 try: proc.kill() except Exception: # noqa: BLE001 pass agent_proc, _data = start_agent() print(" 被控端就绪（局域网在听，中转已上线）") print() # ---- 局域网那条路 ---- lan = measure( "局域网", lambda: (lambda c: (c, c.connect(code)))(LanClient("127.0.0.1", agent_port)), lambda c: c.call({"cmd": "ping"}, 20.0), args.rounds) print(" 局域网 %s" % describe(lan)) if lan["error"]: print(" %s" % lan["error"]) stop_agent(agent_proc) agent_proc = None time.sleep(2.0) # ---- 跨网络那条路（换一台干净的被控端）---- # # ⚠ 这里的轮数**故意比局域网少**，而且超时给得短。 # 原因：当前版本经中转只稳定支撑前 1~2 次交互，之后被控端 # 就不再回包（详见 _build/diag_relay_leg.py 的时间线： # attach/pair/ping#1 全 0ms 返回，ping#2 开始 relay 一直回 0 字节）。 # 这是中转链路的**产品 bug**，不是本测试的毛病 —— # 仓库现成的 _tests/test_relay_local.py 也是同样形状。 # 所以这里只做**有界**测量：拿得到就打点，拿不到就如实报错， # 绝不让整个基准卡死。 agent_proc, _data = start_agent() # attach 只是"接入中转"，协议层还得 pair 一次才算鉴权完成 —— # 这一步等价于局域网那边的 connect()，两边口径才一致。 def relay_connect(): client = RelayClient(relay_url) attach = client.attach(code) if not attach.get("ok"): return client, {"ok": False, "error": attach.get("error") or "接入中转被拒"} reply = client.call({"cmd": "pair", "deviceId": "bench-relay", "deviceName": "基准测试手表", "code": code}, args.relay_timeout) return client, reply relay = measure( "中转", relay_connect, lambda c: c.call({"cmd": "ping"}, args.relay_timeout), args.relay_rounds) print(" 中转 %s" % describe(relay)) if relay["error"]: print(" %s" % relay["error"]) print() # ---------------- 5. 对比 ---------------- print("=" * 66) print(" 对比") print("=" * 66) both = lan["connected"] and relay["connected"] and lan["samples"] \ and relay["samples"] if both: lm = statistics.median(lan["samples"]) rm = statistics.median(relay["samples"]) print(" 中位延迟 局域网 %6.1f ms 中转 %6.1f ms -> 局域网快 %.1f 倍" % (lm, rm, (rm / lm) if lm > 0 else 0)) print(" P90 延迟 局域网 %6d ms 中转 %6d ms" % (percentile(lan["samples"], 0.9), percentile(relay["samples"], 0.9))) print(" 握手耗时 局域网 %6d ms 中转 %6d ms" % (lan["connect_ms"], relay["connect_ms"])) print(" 成功率 局域网 %5.1f%% 中转 %5.1f%%" % (lan["ok"] * 100.0 / lan["total"], relay["ok"] * 100.0 / relay["total"])) else: print(" 有路径没跑通，无法对比") if not lan["connected"]: print(" 局域网: %s" % lan["error"]) if not relay["connected"]: print(" 中转 : %s" % relay["error"]) print() print("BENCH_RESULT lan_median=%.0f lan_p90=%d lan_ok=%d/%d " "lan_connect=%d relay_median=%.0f relay_p90=%d relay_ok=%d/%d " "relay_connect=%d" % (statistics.median(lan["samples"]) if lan["samples"] else 0, percentile(lan["samples"], 0.9), lan["ok"], lan["total"], lan["connect_ms"], statistics.median(relay["samples"]) if relay["samples"] else 0, percentile(relay["samples"], 0.9), relay["ok"], relay["total"], relay["connect_ms"])) print(" 两条路都跑通了" if both else " **有路径没跑通**") return 0 if both else 1 finally: for proc in (agent_proc, relay_proc): if proc is not None: try: proc.terminate() proc.wait(timeout=6) except Exception: # noqa: BLE001 try: proc.kill() except Exception: # noqa: BLE001 pass time.sleep(0.4) shutil.rmtree(work, ignore_errors=True) if __name__ == "__main__": raise SystemExit(main())