xmz7891/ykxmz
1
0
Fork 0
mirror of https://gitee.com/xmz7891/ykxmz.git synced 2026-10-18 14:50:45 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
aac43e559dc33501d670a5826d69c0e4d8cfc010
ykxmz/_tests/bench_lan_vs_relay.py
T
xmz7891 8880e7033d 修「经中转几次后永久哑掉」——「按一下没反应」的真因 
== 现象 ==

用户报：
  * 局域网能搜到电脑，IP/端口/配对码都对，连上后**没有任何反应**
  * 走中转「已配对」，但点按钮没反应

局域网那一半已经修过（preferLan）。这次抓到的是**经中转**那一半，
而且比预想的严重得多。

== 证据链（插桩复现出来的）==

被控端日志里直接抛了未捕获异常，线程当场死：

    Exception in thread client-relay:
      server.py:407  session.send(reply)
      relay.py:166   self._link.send_to_client(data)
      relay.py:357   self._post("/v1/agent/send", ...)
      relay.py:766   raise RelayError(f"请求失败: {last_error}")
    lanremote.relay.RelayError: 请求失败: Idle

之后的状态：
  * client-relay 线程没了 —— 电脑再也不往手表回任何东西
  * relay-link 线程还活着、还在长轮询 —— 中转那边看着「设备在线」
  * **手表看到的就是「连上了，但按一下毫无反应」**

写入仓库的测试 _tests/test_relay_local.py 一直是这个形状：
前面几项 PASS、跑到经中转发命令就卡死。之前没人深挖。

== 根因 1：一条 HTTP/1.1 连接被两个线程同时用 ==

_post() 里锁只包住「取连接」那一段：

    with self._http_lock:
        connection = self._http      # 锁到这里就放了
    try:
        connection.request("POST", ...)   # 锁外发请求

于是两个线程会同时用同一条 connection：
  * relay-link 线程跑 /v1/agent/poll —— 长轮询，一条请求挂 180 秒
  * client-relay 线程同时 POST /v1/agent/send

HTTP/1.1 一条连接同一时刻只能有一个在途请求，第二个线程一进来
就撞上 http.client 状态机，抛 CannotSendRequest("Idle")。

_post 的「重试一次」也救不了：它 _drop_http() 之后，另一个线程
手里还攥着那条旧连接对象在发，照样坏。

修法：
  * **发命令用独立连接**（_post 加 dedicated 参数）。
    长轮询那条常年被占着，发命令不该去挤它，各走各的。
  * 两条连接各自一把锁**锁住整个「发请求 + 读响应」周期**，
    不再只锁取连接那一下。
  * 普通请求若发现缓存连接正忙（长轮询在途），临时另开一条用完即关。
  * _drop_http(dedicated=) 区分清哪条；finally 里一定归还占用
    （否则那条连接会被永久标记为忙）。

== 根因 2：回包失败会打死整个会话线程 ==

server.py 的 _client_loop 里：

    if reply is not None:
        session.send(reply)      # <- 不在任何 try 里
    ...
    except OSError:              # <- 外层只捕获 OSError
        pass

而中转链路上 send 抛的是 RelayError，直接冒出去把线程打死。
**死一次就永久哑**，代价太大。

修法：session.send() 单独兜住所有异常，失败就断开这个会话
（让手表重连），绝不让线程消失。外层也补一个兜底 except。

== 修复效果 ==

    修复前：pair OK -> ping OK -> ping 卡死 -> 线程崩溃，永久哑
    修复后：pair 21ms -> ping 82ms -> ping 4ms -> ping 6ms，零异常

_tests/test_relay_local.py：**之前完全卡死 -> 现在 21/21 全过**

这个测试之前一直是红的，说明「35 个阶段的回归全绿」这件事
本身就不成立过 —— 某个阶段卡住时看起来像环境问题而不是缺陷。

== 同批修掉验证阶段找到的其它问题 ==

1. **TTL 清理 + 关掉接入开关 = 设备永久锁死**（严重）
   register() 用 `device_id in self.devices` 判断老设备，但超时清理
   把设备从 devices 和 codes 都删了。电脑关机过夜（必然 > 6 小时）
   第二天开机就会被自己的开关挡住，而且**重启也救不回来**
   （清 codes 时没落盘，重启后又当孤儿清一遍，自我维系）。
   管理员在页面上还看不到这台设备，不知道该放开谁。
   修法：状态文件里单独存 knownDevices 设备历史，超时清理绝不动它；
   register 改查这份历史；清理后补 _save()；delete_device 才真正忘记。

2. **管理页 JS 语法错误 -> 所有按钮失效**（严重）
   admin.py 里 ADMIN_JS 是普通三引号字符串，我在 Python 源码里写的
   `\n` 被变成真实换行，拼进 JS 就是字符串里的裸换行 ->
   整段 <script> 不执行 -> 改码/删除/批量删除/全选全变幽灵按键。
   元凶是我加改码功能那次提交。改成 `\\n`。

3. **--relay / --relay-status 崩溃**（严重）
   RelayLink.start(url) 的 url 是必填参数，我把默认地址改空白时
   run_agent.py 两处调用没跟着改。一处被 try 吞成「测试失败：missing
   1 required positional argument」（用户以为地址填错了），
   一处直接报栈。两处都补上 url。

4. 非法 deviceId（控制字符 / 路径分隔符）现在被拒（白名单：
   字母数字 + - _ . :）

5. enable_relay() 第二次调用不再谎报成功（加 RelayLink.running 属性，
   返回值反映真实运行状态；不再用「对象建过没有」判断）

6. snapshot() 清孤儿码后补 _save()，磁盘和内存不再脱节

== 局域网全面优化（task-2，另一个智能体完成）==

4 个真问题：
  * **网段判断写死 /24** —— 在 /16 网络上 10.8.1.5 和 10.8.2.5
    被判成不同网络，明明能直连却去走中转还费流量。
    改成按网卡真实掩码比较。
  * **广播只发一个包** —— 丢一个就整轮搜不到。改成 5 个时间点重发。
  * **只探一个网段** —— 多网卡手表搜不到另一网段的电脑。改成全探。
  * **走中转之后再也回不到局域网** —— 出门再回家永远挂中转。
    新增定期回切（先便宜地 TCP ping，通了才动真格，
    不破坏正在工作的中转连接）。

实测：局域网握手 2ms、50/50 = 100%、中位 0ms。
（跨网络那半之前拿不到数字，就是被上面那个线程崩溃挡住的 ——
  现已修复。）

顺带：instance.py 单实例锁端口加 YKXMZ_LOCK_PORT 覆盖（默认不变）。
用户自己那台被控端正跑在这台机器上占着 8767/TCP + 8766/UDP，
是这次排查里好几个「测试怎么都不通」的根因。

== 服务端一键部署脚本（task-1，另一个智能体完成）==

relay/deploy.sh + relay/deploy-tune.sh，随服务端 zip 发布：
  * 自动检查并安装 cloudflared
  * 部署程序、装 4 个 systemd 服务、网络调优、配隧道、自检打公网地址
  * **不传 --hostname 且没有 TUNNEL_TOKEN 时明确提示要填**，
    不会静默用一个写死的域名（用户明确说过这个中转站是他自己用的，
    不对外开放，绝不能把私人域名烧成默认值）
  * 重跑 = 升级，已有数据/配对码/管理密码/凭据都不动

顺带修了发布包一直缺 site_shots/ 的老 bug
（package_remote.py 用 iterdir()+is_file() 收文件，目录进不去，
新机器装完官网截图全是 404）。

== 默认中转地址改成空白 ==

用户要求：软件默认不带任何中转地址，由用户自己填。
四处硬编码全部清空（server.py / MainActivity.java / run_agent.py /
relay.py 的注释和文档示例），全仓已无非注释的硬编码地址。

== 文档同步（task-5，另一个智能体完成）==

README / BEFORE-INSTALL / AFTER-INSTALL / 打包模板：
  * 新增「软件里不带任何中转地址」告示，讲清跨网络要自己部署
  * 修正 Android 版本（写的是 9.0/API28，实际 API 24 / Android 7.0）
  * 删掉「地址已经内置好了」（现在是错的）
  * AGENT_README 原来把作者自己的中转地址当成「内置地址」发给用户，已重写
  * RELAY_README 原来写「两条隧道同时跑会互相争抢」，
    和真机/CHANGELOG 矛盾；按实测改成「两种协议同时在跑（故意的）」
2026-10-07 17:59:39 +08:00

452 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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())
Reference in New Issue View Git Blame Copy Permalink