diff --git a/src/components/unideskcprovider/test_unideskcprovider.py b/src/components/unideskcprovider/test_unideskcprovider.py index 6ee555fd..7d21f222 100644 --- a/src/components/unideskcprovider/test_unideskcprovider.py +++ b/src/components/unideskcprovider/test_unideskcprovider.py @@ -11,6 +11,8 @@ import yaml import unideskcprovider as provider from unideskcprovider import ( + DataChannel, + ProviderManager, ProviderConfig, UiBus, UpdateManager, @@ -135,6 +137,42 @@ class ProtocolHelpersTest(unittest.TestCase): self.assertIn(b'"length":3', header) self.assertEqual(frame[-3:], b"abc") + def test_recycles_only_the_oldest_expired_idle_channel(self): + manager = ProviderManager.__new__(ProviderManager) + manager.state_lock = provider.threading.RLock() + manager.logger = MagicMock() + expired = DataChannel(manager, "expired") + expired.status = "ready" + expired.idle_since = 100.0 + newer = DataChannel(manager, "newer") + newer.status = "ready" + newer.idle_since = 200.0 + claimed = DataChannel(manager, "claimed") + claimed.status = "claimed" + claimed.session_id = "session" + claimed.idle_since = 0.0 + manager.channels = {value.id: value for value in (newer, claimed, expired)} + for channel in manager.channels.values(): + channel.close = MagicMock() + + self.assertTrue(manager.recycle_idle_channel(now=1000.0)) + expired.close.assert_called_once_with("空闲通道主动轮换") + newer.close.assert_not_called() + claimed.close.assert_not_called() + + def test_keeps_recent_idle_channels(self): + manager = ProviderManager.__new__(ProviderManager) + manager.state_lock = provider.threading.RLock() + manager.logger = MagicMock() + channel = DataChannel(manager, "recent") + channel.status = "ready" + channel.idle_since = 500.0 + channel.close = MagicMock() + manager.channels = {channel.id: channel} + + self.assertFalse(manager.recycle_idle_channel(now=1000.0)) + channel.close.assert_not_called() + class UpdateManagerTest(unittest.TestCase): @staticmethod diff --git a/src/components/unideskcprovider/unideskcprovider.py b/src/components/unideskcprovider/unideskcprovider.py index 1cc247a9..5b36614d 100644 --- a/src/components/unideskcprovider/unideskcprovider.py +++ b/src/components/unideskcprovider/unideskcprovider.py @@ -48,7 +48,7 @@ else: APP_NAME = "UniDesk Windows Provider" -APP_VERSION = "0.3.0" +APP_VERSION = "0.3.1" CONFIG_DIR = Path.home() / ".unidesk" CONFIG_PATH = CONFIG_DIR / "provider.yaml" LOG_DIR = CONFIG_DIR / "logs" @@ -61,6 +61,7 @@ POOL_SIZE = 32 MAX_HEADER_BYTES = 64 * 1024 MAX_PAYLOAD_BYTES = 16 * 1024 * 1024 HEARTBEAT_SECONDS = 15 +DATA_CHANNEL_MAX_IDLE_SECONDS = 10 * 60 DEFAULT_UPDATE_METADATA_URL = ( "https://apps.hwpod.com/v1/unideskcprovider/update.json" ) @@ -669,6 +670,7 @@ class DataChannel: self.buffer = bytearray() self.pending: list[tuple[dict, bytes]] = [] self.pending_bytes = 0 + self.idle_since = time.monotonic() self.write_lock = threading.Lock() self.closed = threading.Event() @@ -690,6 +692,7 @@ class DataChannel: } ) self.status = "ready" + self.idle_since = time.monotonic() self.manager.logger.info("TCP 通道已就绪: %s", self.id) self.manager.pool_changed() while not self.closed.is_set(): @@ -879,6 +882,7 @@ class ProviderManager: while not self.stop_event.is_set(): if time.time() - last_heartbeat >= HEARTBEAT_SECONDS: self._send_heartbeat() + self.recycle_idle_channel() last_heartbeat = time.time() try: opcode, payload = control.receive() @@ -1055,8 +1059,26 @@ class ProviderManager: channel.pending_bytes = 0 if not channel.closed.is_set(): channel.status = "ready" + channel.idle_since = time.monotonic() self.pool_changed() + def recycle_idle_channel(self, now: float | None = None) -> bool: + current = time.monotonic() if now is None else now + with self.state_lock: + candidates = [ + channel + for channel in self.channels.values() + if channel.status == "ready" + and channel.session_id is None + and current - channel.idle_since >= DATA_CHANNEL_MAX_IDLE_SECONDS + ] + channel = min(candidates, key=lambda value: value.idle_since) if candidates else None + if channel is None: + return False + self.logger.info("主动轮换空闲 TCP 通道: %s", channel.id) + channel.close("空闲通道主动轮换") + return True + def handle_data_frame(self, channel: DataChannel, header: dict, payload: bytes) -> None: session_id = str(header.get("sessionId") or "") with self.state_lock: @@ -1603,16 +1625,22 @@ class ProviderApp: elif kind == "restart": self.restart_requested = True self.quit() + return self.root.after(120, self._poll_events) def _append_log(self, line: str) -> None: - self.log_text.configure(state="normal") - self.log_text.insert("end", line + "\n") - lines = int(self.log_text.index("end-1c").split(".")[0]) - if lines > 1000: - self.log_text.delete("1.0", f"{lines - 1000}.0") - self.log_text.see("end") - self.log_text.configure(state="disabled") + try: + if not self.log_text.winfo_exists(): + return + self.log_text.configure(state="normal") + self.log_text.insert("end", line + "\n") + lines = int(self.log_text.index("end-1c").split(".")[0]) + if lines > 1000: + self.log_text.delete("1.0", f"{lines - 1000}.0") + self.log_text.see("end") + self.log_text.configure(state="disabled") + except tk.TclError: + return def _set_connection(self, state: str) -> None: self.connection_state = state diff --git a/src/components/unideskcprovider/update.json b/src/components/unideskcprovider/update.json index b467e028..5e237737 100644 --- a/src/components/unideskcprovider/update.json +++ b/src/components/unideskcprovider/update.json @@ -1,9 +1,9 @@ { "schemaVersion": 1, "channel": "stable", - "latestVersion": "0.3.0", + "latestVersion": "0.3.1", "downloadUrl": "unideskcprovider.py", - "bytes": 71119, - "sha256": "98dbdd9e28e2dabe0b89da22924543a36502f5e5bebda9d1e27a4d677ed7f746", - "releaseNotes": "将 Windows Provider TCP warm pool 提升到 32 条，并增加面向人类和 Agent 的 L1 快速部署入口。" + "bytes": 72278, + "sha256": "075a41d81d551b5ef2b75aa7958a7cf38da3c2865bc999e79cc440c1b3b4195d", + "releaseNotes": "主动轮换长期空闲 TCP 通道，修复托盘退出期间的日志控件异常，并保持 32 条 warm pool。" }