持续优化UDP

config copy.jsonbak2

@@ -1,5 +1,16 @@
 {
   "strategy": "top3",
+  "redundancy": 1,
+  "direct_redundancy": 1,
+  "direct_max_redundancy": 2,
+  "direct_redundancy_v4": 2,
+  "direct_redundancy_v6": 2,
+  "direct_ipv6_enabled": true,
+  "direct_open_timeout": 6.0,
+  "relay_open_timeout": 6.0,
+  "tcp_connect_happy_eyeballs_delay": 0.25,
+  "tcp_warmup_bytes": 1024288,
+  "tcp_loser_grace_ms": 1000,
   "udp_redundancy": 2,
   "udp_direct_redundancy": 2,
   "udp_direct_redundancy_v4": 2,

config.json

@@ -1,16 +1,5 @@
 {
   "strategy": "top3",
-  "redundancy": 1,
-  "direct_redundancy": 1,
-  "direct_max_redundancy": 2,
-  "direct_redundancy_v4": 2,
-  "direct_redundancy_v6": 2,
-  "direct_ipv6_enabled": true,
-  "direct_open_timeout": 6.0,
-  "relay_open_timeout": 6.0,
-  "tcp_connect_happy_eyeballs_delay": 0.25,
-  "tcp_warmup_bytes": 1024288,
-  "tcp_loser_grace_ms": 1000,
   "udp_redundancy": 2,
   "udp_direct_redundancy": 2,
   "udp_direct_redundancy_v4": 2,
@@ -18,9 +7,9 @@
   "probe_interval": 3,
   "relay_reconnect_delay": 1,
   "relay_reconnect_max_delay": 10,
-  "udp_always_broadcast": true,
-  "udp_copy_interval_ms": 0,
-  "udp_failover_idle_ms": 600,
+  "udp_always_broadcast": false,
+  "udp_copy_interval_ms": 1,
+  "udp_failover_idle_ms": 450,
   "socks_host": "127.0.0.1",
   "socks_port": 19180,
   "relays": [

scripts/benchmark_local.py

@@ -158,6 +158,13 @@ def summarize_udp_stream(samples: list[float], success_times: list[float], timeo
     )
 
 
+@dataclass
+class UdpRoundResult:
+    round_no: int
+    direct: DnsResult
+    proxy: DnsResult | None
+
+
 def fmt_ms(value: float) -> str:
     return f"{value:.2f}ms"
 
@@ -522,6 +529,7 @@ async def bench_udp_socks(args, proxy: tuple[str, int] | None) -> float | None:
     proxy_results: list[DnsResult] = []
     total = args.count * (2 if proxy else 1)
     started_at = time.perf_counter()
+    round_results: list[UdpRoundResult] = []
 
     print("UDP SOCKS 链路测试开始")
     print(f"  DNS 目标: {args.dns_server_host}:{args.dns_server_port}")
@@ -538,6 +546,7 @@ async def bench_udp_socks(args, proxy: tuple[str, int] | None) -> float | None:
         if isinstance(direct_result, DnsResult):
             direct_results.append(direct_result)
         if proxy is None:
+            round_results.append(UdpRoundResult(index + 1, direct_result if isinstance(direct_result, DnsResult) else DnsResult(0.0, 0.0, 0.0, False, error="执行失败"), None))
             continue
         proxy_result = await run_step(
             f"UDP SOCKS 第{index + 1}步",
@@ -549,6 +558,13 @@ async def bench_udp_socks(args, proxy: tuple[str, int] | None) -> float | None:
         )
         if isinstance(proxy_result, DnsResult):
             proxy_results.append(proxy_result)
+        round_results.append(
+            UdpRoundResult(
+                index + 1,
+                direct_result if isinstance(direct_result, DnsResult) else DnsResult(0.0, 0.0, 0.0, False, error="执行失败"),
+                proxy_result if isinstance(proxy_result, DnsResult) else DnsResult(0.0, 0.0, 0.0, False, error="执行失败"),
+            )
+        )
 
     direct_ok = [item for item in direct_results if item.ok]
     proxy_ok = [item for item in proxy_results if item.ok]
@@ -558,6 +574,14 @@ async def bench_udp_socks(args, proxy: tuple[str, int] | None) -> float | None:
     proxy_stall_count = sum(1 for value in proxy_query_values if value >= args.udp_stall_ms)
 
     print(f"UDP 目标: {args.dns_server_host}:{args.dns_server_port}")
+    print("  逐轮结果:")
+    for item in round_results:
+        direct_part = f"直连 query={fmt_ms(item.direct.query_ms)} total={fmt_ms(item.direct.total_ms)} ok={item.direct.ok}"
+        if item.proxy is None:
+            print(f"    第{item.round_no}轮: {direct_part}")
+        else:
+            proxy_part = f"SOCKS associate={fmt_ms(item.proxy.associate_ms)} query={fmt_ms(item.proxy.query_ms)} total={fmt_ms(item.proxy.total_ms)} ok={item.proxy.ok}"
+            print(f"    第{item.round_no}轮: {direct_part}; {proxy_part}")
     print(
         f"  直连: 成功 {len(direct_ok)}/{len(direct_results)}，查询均值 {fmt_ms(summarize(direct_query_values)['avg'])}，"
         f" p95 {fmt_ms(summarize(direct_query_values)['p95'])}，抖动 {fmt_ms(stdev_or_zero(direct_query_values))}，"
@@ -691,35 +715,22 @@ async def amain(args) -> int:
 
     print("本地基准测试开始")
     print(f"  样本数: {args.count}")
-    print(f"  TCP 目标: {args.http_url}")
     print(f"  UDP 目标: {args.dns_server_host}:{args.dns_server_port}")
     print(f"  SOCKS: {proxy[0]}:{proxy[1]}" if proxy else "  SOCKS: 未配置或未启动")
-    print(f"  Transparent 排除用户: {args.runtime_user}")
-    print("  说明: TCP 比较的是“直连基线 vs transparent_edge 实际链路”，UDP 比较的是“直连 vs socks_edge 实际链路”")
-    if args.http_url.startswith("https://"):
-        print("  提示: 当前 TCP 目标是 HTTPS，更适合做“网页体验参考”；如果要看透明首包纯开销，仍建议使用 http:// 目标")
     print("")
 
-    tcp_pct = None
     udp_pct = None
-    script_path = Path(__file__).resolve()
 
-    if args.mode in ("tcp", "all"):
-        tcp_pct = await bench_tcp_transparent(args, script_path)
-        print("")
-    if args.mode in ("udp", "all"):
-        udp_pct = await bench_udp_socks(args, proxy)
-        if args.udp_stream_count > 0:
-            await bench_udp_stream(args, proxy)
+    udp_pct = await bench_udp_socks(args, proxy)
+    if args.udp_stream_count > 0:
+        await bench_udp_stream(args, proxy)
 
-    changes = [value for value in (tcp_pct, udp_pct) if value is not None]
+    changes = [value for value in (udp_pct,) if value is not None]
     if not changes:
         print("中文总结：无可用结果")
         return 0
 
     parts = []
-    if tcp_pct is not None:
-        parts.append(f"TCP={verdict_from_diff(tcp_pct)}")
     if udp_pct is not None:
         parts.append(f"UDP={verdict_from_diff(udp_pct)}")
     print(f"中文总结：{overall_verdict(changes)}（{', '.join(parts)}）")
@@ -731,16 +742,12 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--config", default="/home/mynetspeeder/config.json", help="配置文件路径")
     parser.add_argument("--proxy-host", default="", help="SOCKS5 地址，默认读取 config.json")
     parser.add_argument("--proxy-port", type=int, default=0, help="SOCKS5 端口，默认读取 config.json")
-    parser.add_argument("--http-url", default="https://spectrum.ieee.org/", help="TCP/透明链路测试 URL，可用 https:// 做网页体验参考")
     parser.add_argument("--dns-server", default="8.8.8.8:53", help="UDP/SOCKS 链路 DNS 目标，格式 host:port")
-    parser.add_argument("--mode", choices=("tcp", "udp", "all"), default="all", help="只测 TCP、只测 UDP 或都测")
-    parser.add_argument("--count", type=int, default=4, help="每类测试样本数，默认 4")
+    parser.add_argument("--count", type=int, default=5, help="UDP 测试轮数，默认 5")
     parser.add_argument("--timeout", type=float, default=3.0, help="单次测试超时秒数")
     parser.add_argument("--udp-stall-ms", type=float, default=120.0, help="UDP 单次查询超过该值记为一次疑似卡顿")
     parser.add_argument("--udp-stream-count", type=int, default=20, help="UDP 连续流稳定性测试样本数，设为 0 则关闭")
     parser.add_argument("--udp-stream-interval-ms", type=float, default=200.0, help="UDP 连续流相邻样本间隔毫秒")
-    parser.add_argument("--runtime-user", default=os.environ.get("MYNETSPEEDER_USER", "mynetspeeder"), help="transparent 排除用户")
-    parser.add_argument("--edge-log", default="/var/log/mynetspeeder-edge.log", help="transparent edge 日志路径")
     parser.add_argument("--child-http", default="", help=argparse.SUPPRESS)
     parser.add_argument("--child-timeout", type=float, default=3.0, help=argparse.SUPPRESS)
     return parser

socks_edge.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import itertools
+from collections import deque
 import socket
 import struct
 from dataclasses import dataclass, field
@@ -13,6 +14,9 @@ from .protocol import AUTH, STATUS_OK, TCP_CLOSE, TCP_DATA, TCP_OPEN, TCP_STATUS
 from .scheduler import Scheduler
 
 SOCKS_VERSION = 5
+UDP_WARMUP_BROADCAST_PACKETS = 6
+UDP_SHADOW_PROBE_INTERVAL_SEC = 0.25
+UDP_FAST_FAILOVER_MISSES = 3
 
 
 async def read_exact(reader: asyncio.StreamReader, size: int) -> bytes:
@@ -102,6 +106,10 @@ class UdpFlowState:
     relay_failures: dict[str, int] = field(default_factory=dict)
     relay_error_seen: set[str] = field(default_factory=set)
     path_last_seen: dict[str, float] = field(default_factory=dict)
+    packet_client_addrs: dict[int, tuple[str, int]] = field(default_factory=dict)
+    direct_pending_clients: dict[str, deque[tuple[int, tuple[str, int]]]] = field(default_factory=dict)
+    last_probe_at: float = 0.0
+    winner_miss_streak: int = 0
 
     def touch(self, now: float) -> None:
         self.last_activity = now
@@ -263,9 +271,11 @@ class UdpAssociateServer(asyncio.DatagramProtocol):
             )
             self.client_flows[flow_key] = flow
         flow.touch(now)
+        flow.client_addr = (addr[0], addr[1])
         flow.packets_sent += 1
         packet_id = next(self.packet_counter)
-        asyncio.create_task(self.edge.forward_udp(flow, payload, packet_id, self))
+        flow.packet_client_addrs[packet_id] = (addr[0], addr[1])
+        asyncio.create_task(self.edge.forward_udp(flow, payload, packet_id, (addr[0], addr[1]), self))
         self._log_udp_summary()
 
     def _reset_client_state(self, addr) -> None:
@@ -286,12 +296,12 @@ class UdpAssociateServer(asyncio.DatagramProtocol):
             return
         await self._deliver_flow_packet(flow, frame.packet_id, frame.payload, link.node.name)
 
-    async def handle_from_direct(self, flow: UdpFlowState, path_name: str, payload: bytes) -> None:
+    async def handle_from_direct(self, flow: UdpFlowState, path_name: str, payload: bytes, packet_id: int = 0, client_addr: tuple[str, int] | None = None) -> None:
         if self.transport is None or self.client_addr is None:
             return
-        await self._deliver_flow_packet(flow, 0, payload, path_name)
+        await self._deliver_flow_packet(flow, packet_id, payload, path_name, client_addr)
 
-    async def _deliver_flow_packet(self, flow: UdpFlowState, packet_id: int, payload: bytes, source_name: str) -> None:
+    async def _deliver_flow_packet(self, flow: UdpFlowState, packet_id: int, payload: bytes, source_name: str, client_addr: tuple[str, int] | None = None) -> None:
         if self.transport is None or self.client_addr is None:
             return
         packet = self._build_socks_udp(flow.target_host, flow.target_port, payload)
@@ -299,8 +309,10 @@ class UdpAssociateServer(asyncio.DatagramProtocol):
         flow.touch(now)
         flow.path_last_seen[source_name] = now
         flow.packets_received += 1
+        target_addr = client_addr or flow.packet_client_addrs.pop(packet_id, None) or flow.client_addr
         if flow.winner_name is None:
             flow.winner_name = source_name
+            flow.winner_miss_streak = 0
             self.win_counts[source_name] = self.win_counts.get(source_name, 0) + 1
             self._log_udp_summary(force=True)
         elif flow.winner_name != source_name:
@@ -308,10 +320,14 @@ class UdpAssociateServer(asyncio.DatagramProtocol):
             winner_last_seen = flow.path_last_seen.get(flow.winner_name, 0.0)
             if winner_last_seen and now - winner_last_seen >= (self.edge.config.udp_failover_idle_ms / 1000):
                 flow.winner_name = source_name
+                flow.winner_miss_streak = 0
                 self.win_counts[source_name] = self.win_counts.get(source_name, 0) + 1
                 self._log_udp_summary(force=True)
+        else:
+            flow.winner_miss_streak = 0
         if flow.winner_name == source_name:
-            self.transport.sendto(packet, self.client_addr)
+            if target_addr is not None:
+                self.transport.sendto(packet, target_addr)
 
     def set_flow_candidates(self, flow: UdpFlowState, candidate_names: tuple[str, ...]) -> None:
         if not flow.candidate_names:
@@ -506,7 +522,12 @@ class SocksEdge:
                 data = await loop.sock_recv(sock, 65535)
                 if not data:
                     break
-                await udp_server.handle_from_direct(flow, path_name, data)
+                pending = flow.direct_pending_clients.get(path_name)
+                packet_id = 0
+                client_addr = flow.client_addr
+                if pending:
+                    packet_id, client_addr = pending.popleft()
+                await udp_server.handle_from_direct(flow, path_name, data, packet_id, client_addr)
         except Exception:
             pass
         finally:
@@ -515,7 +536,7 @@ class SocksEdge:
             with contextlib.suppress(Exception):
                 sock.close()
 
-    async def forward_udp(self, flow: UdpFlowState, payload: bytes, packet_id: int, udp_server: UdpAssociateServer) -> None:
+    async def forward_udp(self, flow: UdpFlowState, payload: bytes, packet_id: int, client_addr: tuple[str, int], udp_server: UdpAssociateServer) -> None:
         await self._ensure_udp_direct_paths(flow, udp_server)
         meta = encode_json({"host": flow.target_host, "port": flow.target_port})
         links = self._selected_udp_links()
@@ -528,12 +549,27 @@ class SocksEdge:
             return
         active_direct_names = list(direct_names)
         active_links = links
-        if not (self.config.udp_always_broadcast or flow.winner_name is None):
+        now = asyncio.get_running_loop().time()
+        warmup_mode = flow.packets_sent <= UDP_WARMUP_BROADCAST_PACKETS
+        shadow_probe = (
+            flow.winner_name is not None
+            and now - flow.last_probe_at >= UDP_SHADOW_PROBE_INTERVAL_SEC
+        )
+        if shadow_probe:
+            flow.last_probe_at = now
+        broadcast_mode = self.config.udp_always_broadcast or flow.winner_name is None or warmup_mode or shadow_probe
+        if not broadcast_mode:
             winner_last_seen = flow.path_last_seen.get(flow.winner_name, 0.0) if flow.winner_name else 0.0
-            if winner_last_seen and asyncio.get_running_loop().time() - winner_last_seen >= (self.config.udp_failover_idle_ms / 1000):
+            winner_stale = bool(winner_last_seen and now - winner_last_seen >= (self.config.udp_failover_idle_ms / 1000))
+            if not winner_stale:
+                flow.winner_miss_streak += 1
+            if winner_stale or flow.winner_miss_streak >= UDP_FAST_FAILOVER_MISSES:
                 flow.winner_name = None
-            active_direct_names = [name for name in active_direct_names if name == flow.winner_name]
-            active_links = [link for link in active_links if link.node.name == flow.winner_name]
+                flow.winner_miss_streak = 0
+                broadcast_mode = True
+            else:
+                active_direct_names = [name for name in active_direct_names if name == flow.winner_name]
+                active_links = [link for link in active_links if link.node.name == flow.winner_name]
         if not active_direct_names and not active_links:
             if direct_names:
                 active_direct_names = [direct_names[0]]
@@ -547,9 +583,14 @@ class SocksEdge:
                 if sock is None:
                     continue
                 try:
+                    flow.direct_pending_clients.setdefault(path_name, deque()).append((packet_id, client_addr))
                     await asyncio.get_running_loop().sock_sendall(sock, payload)
                     sent_any = True
                 except Exception as exc:
+                    pending = flow.direct_pending_clients.get(path_name)
+                    if pending:
+                        with contextlib.suppress(Exception):
+                            pending.pop()
                     flow.direct_failures.add(path_name)
                     flow.direct_sockets.pop(path_name, None)
                     task = flow.direct_tasks.pop(path_name, None)