diff --git a/frontends/hub_p2p.py b/frontends/hub_p2p.py index 7417de34..275b7059 100644 --- a/frontends/hub_p2p.py +++ b/frontends/hub_p2p.py @@ -2,6 +2,7 @@ import asyncio import os import socket +import time from pathlib import Path from fastapi import Response @@ -16,11 +17,10 @@ def install(app, *, web_port, token, here): signal = os.environ.get("GA_P2P_SIGNAL", "ws://47.101.182.29:49157/ws") name = os.environ.get("GA_P2P_NAME", "ga-hub-phone") rooms = Path(os.environ.get("GA_P2P_ROOMS", "~/.p2p_ws/rooms.json")).expanduser() - # 多人组成员名带角色前缀:hub: 导出本机 API,cust: 代理对方 API。一条 P2PSocket - # 只能有一个 HTTP 消费者,故同一进程以两个成员身份入组,每对连接角色唯一。 + # 组内每个 hub 一个成员名;连接按需建立:主叫代理对方 API,被叫导出本机 API。 member = os.environ.get("GA_P2P_MEMBER", f"{socket.gethostname()}-{web_port}") state = {"task": None, "invite": None, "status": "idle", "error": None, - "started": False, "group": None, "hubs": {}} + "started": False, "group": None, "proxies": {}} app.state.p2p_pair_open = True async def export(ws): @@ -60,24 +60,29 @@ def install(app, *, web_port, token, here): async def use_hub(ws, peer): proxy = await HTTPLocalProxy(ws).start() - state["hubs"][peer[4:]] = proxy.url + state["proxies"][peer] = proxy try: await proxy.wait_closed() finally: - if state["hubs"].get(peer[4:]) == proxy.url: - del state["hubs"][peer[4:]] + if state["proxies"].get(peer) is proxy: + del state["proxies"][peer] await proxy.close() + async def reap_idle(group): + while True: # 10 分钟没请求的代理断开,下次 /p2p/open 再按需重连 + await asyncio.sleep(60) + for peer, proxy in list(state["proxies"].items()): + if time.monotonic() - proxy.last_used > 600: + group.close(peer) + def start_group(): if state["group"]: return room = load_room(name, path=rooms) - hub, cust = f"hub:{member}", f"cust:{member}" - roles = ((hub, serve_customer, "cust:"), (cust, use_hub, "hub:")) - state["group"] = [asyncio.create_task(Group( - room["signal_url"], room["room"], me, serve, - want=lambda p, other=other: p.startswith(other) and p[len(other):] != member, - ).run()) for me, serve, other in roles] + group = state["group"] = Group( + room["signal_url"], room["room"], member, serve_customer, use_hub, + ) + state["group_tasks"] = [asyncio.create_task(group.run()), asyncio.create_task(reap_idle(group))] async def new_pair(): try: @@ -119,8 +124,24 @@ poll();''', media_type="text/html") @app.get("/p2p/hubs") async def p2p_hubs(): - """同组其他 hub 的本地代理地址 {成员名: http://127.0.0.1:端口},按原 /api/ 访问。""" - return JSONResponse(state["hubs"]) + """同组在线 hub {成员名: 本地代理地址,未连接为 null}。""" + group, proxies = state["group"], state["proxies"] + members = sorted(group.members) if group else [] + return JSONResponse({m: getattr(proxies.get(m), "url", None) for m in members}) + + @app.get("/p2p/open") + async def p2p_open(hub: str): + """按需连接 hub,返回本地代理地址,之后按原 /api/ 访问。""" + group = state["group"] + if not group or hub not in group.members: + return JSONResponse({"error": "hub offline"}, status_code=404) + group.open(hub) + for _ in range(200): + if proxy := state["proxies"].get(hub): + proxy.last_used = time.monotonic() + return JSONResponse({"url": proxy.url}) + await asyncio.sleep(0.1) + return JSONResponse({"error": "connect timeout"}, status_code=504) @app.get("/pair/status") async def pair_status(): diff --git a/frontends/p2p_ws_client.py b/frontends/p2p_ws_client.py index 9f1b8f64..498a272e 100644 --- a/frontends/p2p_ws_client.py +++ b/frontends/p2p_ws_client.py @@ -826,6 +826,7 @@ class HTTPLocalProxy: self._app = self._runner = self._site = self._receiver = None self._pending, self._closed = {}, asyncio.Event() self._counter = 0 + self.last_used = time.monotonic() # 供上层回收空闲代理 async def start(self): if self._runner: @@ -868,6 +869,7 @@ class HTTPLocalProxy: return {"Access-Control-Allow-Origin": origin, "Vary": "Origin"} if allowed else {} async def _handle(self, request): + self.last_used = time.monotonic() _, web, _ = _http_modules() cors_headers = self._cors_headers(request) if request.method == "OPTIONS": @@ -1086,7 +1088,7 @@ async def connect_code( async def retry_forever(attempt, on_error=None, *, stable=30, window=60, budget=8): """永久重试 attempt()。节奏三层: - 1) 本轮活过 stable 秒(正常会话被掉线/被顶号) -> 0.2s 秒级重连; + 1) 本轮活过 stable 秒(正常会话被掉线/被顶号) -> 0.2~1s 秒级重连; 2) 秒断/连不上 -> 抖动指数退避 1,2,4..60s, 防"配对即断"空转; 3) 兜底: 任意 window 秒内最多 budget 次, 远低于信令服务器每 IP 限速。 抖动也避免服务端重启时大量客户端同一秒齐涌。""" @@ -1108,38 +1110,49 @@ async def retry_forever(attempt, on_error=None, *, stable=30, window=60, budget= if on_error: on_error(exc) delay = 0 if loop.time() - now >= stable else min(max(delay * 2, 1), 60) - await asyncio.sleep(0.2 if delay == 0 else delay * random.uniform(0.5, 1.5)) + await asyncio.sleep(random.uniform(0.2, 1) if delay == 0 else delay * random.uniform(0.5, 1.5)) -def pair_room(room: str, a: str, b: str) -> str: - """两成员的专属房间:双方独立算出同一值,且不同于原 1v1 房间。""" - x, y = sorted((a, b)) - return hashlib.sha256(f"{room}\n{x}\n{y}".encode()).hexdigest() +def pair_room(room: str, caller: str, callee: str) -> str: + """主叫到被叫的专属房间:双方独立算出同一值;有方向,A 呼 B 与 B 呼 A 互不干扰。""" + return hashlib.sha256(f"{room}\n{caller}\n{callee}".encode()).hexdigest() class Group: """以成员名 me 加入 UUID room 的多人组。 - 大厅连接只收在线名单(成员名列表,变化时推送,无轮询);对每个在线且 want(peer) - 的成员各维护一条普通 P2PSocket,并交给 serve(socket, peer) 处理直到断开。 - 成员离开才取消其连接;大厅自身断线不影响已建立的两两连接。 + 大厅连接收在线名单(变化时推送,无轮询)并转发呼叫。两两连接按需建立: + open(peer) 经大厅呼叫对方,双方进入 pair_room(主叫, 被叫) 的普通 P2PSocket, + 主叫交给 use(socket, peer),被叫交给 serve(socket, peer)。主叫断线自动重呼; + close(peer) 停止,对方离组时暂停、回来后自动恢复。不 open 就不占连接。 同名重入会顶掉旧连接,因此每个进程须用唯一成员名。 """ - def __init__(self, signal_url, room, me, serve, *, want=lambda peer: True, - access_key=DEFAULT_ACCESS_KEY): + def __init__(self, signal_url, room, me, serve, use, *, access_key=DEFAULT_ACCESS_KEY): self.signal_url, self.room, self.me = signal_url, room, me - self.serve, self.want, self.access_key = serve, want, access_key + self.serve, self.use, self.access_key = serve, use, access_key self.members: set[str] = set() - self._pairs: dict[str, asyncio.Task] = {} + self._ws = None + self._wanted: set[str] = set() + self._out: dict[str, asyncio.Task] = {} # 本端主叫 + self._in: dict[str, asyncio.Task] = {} # 对方呼入 async def run(self): try: await retry_forever(self._lobby) finally: - for task in self._pairs.values(): + for task in [*self._out.values(), *self._in.values()]: task.cancel() + def open(self, peer): + self._wanted.add(peer) + self._sync(self.members) + + def close(self, peer): + self._wanted.discard(peer) + if task := self._out.pop(peer, None): + task.cancel() + async def _lobby(self): room = hashlib.sha256(f"{self.room}:lobby".encode()).hexdigest() sep = "&" if "?" in self.signal_url else "?" @@ -1148,28 +1161,51 @@ class Group: async with websockets.connect( signed_url(url, "ws", room, self.access_key), ping_interval=15, ping_timeout=10, ) as ws: - async for raw in ws: - msg = json.loads(raw) - if msg.get("type") == "members" and msg["seq"] > seq: - seq = msg["seq"] - self._sync(set(msg["members"]) - {self.me}) + self._ws = ws + try: + async for raw in ws: + msg = json.loads(raw) + if msg.get("type") == "members" and msg["seq"] > seq: + seq = msg["seq"] + self._sync(set(msg["members"]) - {self.me}) + elif msg.get("type") == "call": + self._answer(msg["from"]) + finally: + self._ws = None def _sync(self, members): self.members = members - for peer in set(self._pairs) - members: - self._pairs.pop(peer).cancel() - for peer in members - set(self._pairs): - if self.want(peer): - self._pairs[peer] = asyncio.create_task( - retry_forever(lambda peer=peer: self._pair(peer))) + for tasks in (self._out, self._in): + for peer in set(tasks) - members: + tasks.pop(peer).cancel() + for peer in (self._wanted & members) - set(self._out): + self._out[peer] = asyncio.create_task(retry_forever(lambda peer=peer: self._call(peer))) - async def _pair(self, peer): + async def _call(self, peer): + if self._ws is None: + raise ConnectionError("lobby offline") + await self._ws.send(json.dumps({"type": "call", "to": peer})) + await self._link(pair_room(self.room, self.me, peer), self.use, peer) + + def _answer(self, peer): + """每次呼叫只接一次;对方重呼时旧连接必已失效,直接替换。""" + if old := self._in.pop(peer, None): + old.cancel() + + async def once(): + with suppress(Exception): + await self._link(pair_room(self.room, peer, self.me), self.serve, peer) + if self._in.get(peer) is task: + del self._in[peer] + task = self._in[peer] = asyncio.create_task(once()) + + async def _link(self, room, handler, peer): socket = await P2PSocket.connect( - self.signal_url, pair_room(self.room, self.me, peer), - access_key=self.access_key, direct_timeout=15, peer_timeout=None, + self.signal_url, room, access_key=self.access_key, + direct_timeout=15, peer_timeout=15, ) try: - await self.serve(socket, peer) + await handler(socket, peer) finally: await socket.close()