继续优化

README.md

@@ -34,7 +34,7 @@
 {
   "strategy": "top2",
   "redundancy": 2,
-  "tcp_warmup_bytes": 65536,
+  "tcp_warmup_bytes": 262144,
   "probe_interval": 15,
   "relays": [
     {"name": "hk1", "host": "1.2.3.4", "port": 9009, "token": "demo", "weight": 100},
@@ -118,7 +118,7 @@ python3 -m mynetspeeder probe --config /home/mynetspeeder/config.json --once
 - `edge` 会同时尝试：
   - 当前主 VPS 直接连目标
   - 若干在线子节点 relay 代为连接目标
-- 前 `tcp_warmup_bytes` 的上行数据会并发发给所有候选路径
+- 前 `tcp_warmup_bytes` 的上行数据会更积极地并发发给所有候选路径
 - 谁先回下行数据，谁成为胜出路径
 - 其它路径会关闭
 

config.json

@@ -1,7 +1,7 @@
 {
   "strategy": "top2",
   "redundancy": 2,
-  "tcp_warmup_bytes": 65536,
+  "tcp_warmup_bytes": 262144,
   "probe_interval": 3,
   "relays": [{"name": "hk1", "host": "141.140.15.30", "port": 9009, "token": "130", "weight": 100}]
 }

config.py

@@ -22,7 +22,7 @@ class Config:
     relays: list[RelayNode]
     strategy: Strategy = "top2"
     redundancy: int = 2
-    tcp_warmup_bytes: int = 65536
+    tcp_warmup_bytes: int = 262144
     probe_interval: float = 15.0
 
     @classmethod
@@ -33,6 +33,6 @@ class Config:
             relays=relays,
             strategy=raw.get("strategy", "top2"),
             redundancy=raw.get("redundancy", 2),
-            tcp_warmup_bytes=raw.get("tcp_warmup_bytes", 65536),
+            tcp_warmup_bytes=raw.get("tcp_warmup_bytes", 262144),
             probe_interval=raw.get("probe_interval", 15.0),
         )

demo-config copy.json

@@ -1,7 +1,7 @@
 {
   "strategy": "top2",
   "redundancy": 2,
-  "tcp_warmup_bytes": 65536,
+  "tcp_warmup_bytes": 262144,
   "probe_interval": 3,
   "relays": []
 }

socks_edge.py

@@ -77,6 +77,7 @@ class TcpRaceSession:
     links: list[RelayLink]
     warmup_bytes: int
     winning_link: RelayLink | None = None
+    winner_name: str | None = None
     opened: int = 0
     open_errors: list[str] = field(default_factory=list)
     uplink_bytes: int = 0
@@ -84,6 +85,7 @@ class TcpRaceSession:
     open_event: asyncio.Event = field(default_factory=asyncio.Event)
     winner_event: asyncio.Event = field(default_factory=asyncio.Event)
     pump_task: asyncio.Task | None = None
+    win_counts: Dict[str, int] = field(default_factory=dict)
 
     async def start(self) -> None:
         meta = encode_json({"host": self.target_host, "port": self.target_port})
@@ -128,6 +130,11 @@ class TcpRaceSession:
         if frame.kind == TCP_DATA:
             if self.winning_link is None:
                 self.winning_link = link
+                self.winner_name = link.node.name
+                self.win_counts[link.node.name] = self.win_counts.get(link.node.name, 0) + 1
+                node_total = self.win_counts[link.node.name]
+                relay_detail = ", ".join(f"{name}={count}" for name, count in sorted(self.win_counts.items())) or "none"
+                print(f"[edge] tcp win session={self.session_id} target={self.target_host}:{self.target_port} winner={link.node.name} node_total={node_total} win_breakdown={relay_detail}")
                 self.winner_event.set()
                 await self._close_losers(except_link=link)
             if link is self.winning_link:

transparent_edge.py

@@ -169,6 +169,8 @@ class TransparentSession:
     writer: asyncio.StreamWriter
     paths: list[BasePath]
     warmup_bytes: int
+    stats: dict[str, int]
+    target_stats: dict[tuple[str, int], dict[str, int]]
     opened_count: int = 0
     status_count: int = 0
     errors: list[str] = field(default_factory=list)
@@ -179,6 +181,20 @@ class TransparentSession:
     closed: bool = False
     pump_task: asyncio.Task | None = None
 
+    def _record_win(self, winner: BasePath) -> None:
+        self.stats[winner.name] = self.stats.get(winner.name, 0) + 1
+        key = (self.target.host, self.target.port)
+        target_stats = self.target_stats.setdefault(key, {})
+        target_stats[winner.name] = target_stats.get(winner.name, 0) + 1
+        direct_wins = self.stats.get("direct", 0)
+        relay_wins = sum(count for name, count in self.stats.items() if name != "direct")
+        target_direct = target_stats.get("direct", 0)
+        target_relay = sum(count for name, count in target_stats.items() if name != "direct")
+        relay_detail = ", ".join(f"{name}={count}" for name, count in sorted(self.stats.items()) if name != "direct") or "none"
+        target_detail = ", ".join(f"{name}={count}" for name, count in sorted(target_stats.items()) if name != "direct") or "none"
+        target_pref = "relay" if target_relay > target_direct else "direct"
+        print(f"[edge] tcp win session={self.session_id} target={self.target.host}:{self.target.port} winner={winner.name} direct={direct_wins} relay={relay_wins} relay_breakdown={relay_detail} target_pref={target_pref} target_direct={target_direct} target_relay={target_relay} target_breakdown={target_detail}")
+
     async def start(self) -> None:
         await asyncio.gather(*(path.open(self.target) for path in self.paths), return_exceptions=True)
         await asyncio.wait_for(self.open_event.wait(), timeout=10)
@@ -223,7 +239,7 @@ class TransparentSession:
         if event == "data":
             if self.winner is None:
                 self.winner = path
-                print(f"[edge] session={self.session_id} winner={path.name} target={self.target.host}:{self.target.port}")
+                self._record_win(path)
                 self.winner_event.set()
                 await self._close_losers(path)
             if path is self.winner and payload is not None:
@@ -456,6 +472,8 @@ class TransparentEdge:
         self.udp_flows: dict[tuple[PeerAddress, TargetAddress], UdpFlow] = {}
         self.udp_flow_ids = itertools.count(1)
         self.udp_gc_task: asyncio.Task | None = None
+        self.tcp_win_counts: dict[str, int] = {}
+        self.tcp_target_wins: dict[tuple[str, int], dict[str, int]] = {}
 
     async def start(self) -> None:
         await self.manager.start()
@@ -502,7 +520,7 @@ class TransparentEdge:
         try:
             target = self._get_original_dst(writer)
             session_id = next(self.session_ids)
-            session = TransparentSession(session_id=session_id, target=target, reader=reader, writer=writer, paths=[], warmup_bytes=self.config.tcp_warmup_bytes)
+            session = TransparentSession(session_id=session_id, target=target, reader=reader, writer=writer, paths=[], warmup_bytes=self.config.tcp_warmup_bytes, stats=self.tcp_win_counts, target_stats=self.tcp_target_wins)
             paths: list[BasePath] = [DirectTcpPath(name="direct", on_frame=lambda path, event, payload, s=session: self._handle_tcp_session(s, path, event, payload))]
             for connection in self.manager.available():
                 stream_id = next(self.stream_ids)