Release v1.0.5 adaptive fast service

.gitignore

@@ -13,6 +13,7 @@ venv/
 build/
 dist/
 release/
+*.spec
 
 *.log
 *.tmp

AGENTS.md

@@ -1,5 +1,9 @@
 # PGU-NetFileSyncWin 项目维护说明
 
+当前应用版本：`v1.0.5`（2026-09-27）
+
+本版本变更：极速服务 V3、CPU 自适应并发、HTTP 长连接、服务版本显示与旧版升级、连接断开后的 user linger、下载重试稳定性和项目文档整理。
+
 ## 项目定位
 
 PGU-NetFileSyncWin 是一个面向 Windows 的 PyQt5 远程文件管理与高速下载工具。它用于连接 PGU 设备、浏览远程文件，并将文件或文件夹下载到 Windows 本机。
@@ -122,21 +126,24 @@ remote_path=/home/clear/Data
 
 ```python
 FAST_HTTP_PORT = 5924
-FAST_HTTP_WORKERS = 12
-FAST_HTTP_BIG_FILE_WORKERS = 4
+FAST_HTTP_WORKERS = 8
+FAST_HTTP_BIG_FILE_WORKERS = 3
+FAST_HTTP_TIMEOUT = 45
 FAST_SPLIT_THRESHOLD = 64 * 1024 * 1024
 FAST_HTTP_TOKEN = "PGU4_SYNC_FAST_V2"
-FAST_SERVICE_VERSION = "2"
+FAST_SERVICE_VERSION = "3"
 ```
 
 策略：
 
-- 小文件按文件并发，最多 12 路。
+- 连接时同时检测 Windows 本机与远端设备在线 CPU 线程数，总并发自动取两者较小值且不保存。
+- 小文件按文件并发，最多 8 路，并复用每个下载线程的 HTTP 连接，优先保证吞吐稳定。
 - 大文件超过 64MB 后使用 Range 分片。
-- 单个大文件最多 4 路分片。
-- 分片失败自动重试。
+- 单个大文件最多 3 路分片，避免远端磁盘和HTTP线程争用。
+- 分片失败在块内断点续传并自动重试，避免从块头重复下载。
+- 极速HTTP使用45秒读取超时，避免短暂磁盘/网络抖动被过早判定失败。
 - 失败汇总只在任务结束后提示。
-- V2 使用 UTF-8 URL-safe Base64 传输远程路径，支持中文目录、中文文件名和空格。
+- V3 按远端逻辑 CPU 数量建立并发槽位，上限为 `min(16, CPU线程数)` 且至少 2 路；同时扩大连接等待队列并启用 HTTP/1.1 长连接。路径仍使用 UTF-8 URL-safe Base64，支持中文目录、中文文件名和空格。
 
 ## 远端极速服务
 
@@ -159,10 +166,11 @@ FAST_SERVICE_VERSION = "2"
 crontab @reboot
 ```
 
-极速服务脚本带持久版本标记。连接设备时：
+极速服务脚本带持久版本标记，并在权限允许时启用 systemd user linger，避免 SSH 会话结束后服务退出。连接设备时：
 
-- 当前 V2 服务正常：直接启用极速通道并刷新自启配置。
-- 当前 V2 已安装但未运行：自动清理残留进程并重启，不显示升级。
+- 当前 V3 服务正常：直接启用极速通道并刷新自启配置，GUI 显示 `极速 v3`。
+- 检测到旧版本时显示实际版本，例如 `极速 v2 · 可升级`，按钮提供 `升级极速`。
+- 当前 V3 已安装但未运行：自动清理残留进程并重启，不显示升级。
 - 自动恢复失败：显示 `修复极速`。
 - 仅脚本确实为旧版本时显示 `升级极速`。
 - crontab 清理必须匹配 `pgu4_fast_http.py.*--port 5924`，避免带引号的旧命令累积。

README.md

@@ -23,6 +23,16 @@ PGU-NetFileSyncWin 是一个 Windows 桌面文件同步工具，用于连接 PGU
 - 使用速度条显示实时速度，下载完成后显示用时和平均速度。
 - 使用同目录 `PGU-NetFileSyncWin.ini` 保存连接配置，便于和 exe 一起复制使用。
 
+## 当前版本
+
+当前应用版本：`v1.0.5`
+
+本版本重点修复极速下载稳定性：客户端会同时考虑 Windows 本机和 PGU 设备的 CPU 线程数；极速服务按远端 CPU 建立并发槽位，支持 HTTP 长连接、连接队列扩容、服务版本显示和旧版一键升级。
+
+- 高速通道区域显示已安装的极速服务版本，例如 `极速 v3`。
+- 检测到旧版时显示实际版本，并提供 `升级极速`。
+- 大量小文件下载复用连接，减少速度波动和重复重试。
+
 ## 发布版
 
 推荐使用发布页中的文件：

USER.md

@@ -152,7 +152,9 @@ IP、端口、用户名和远程路径可以直接用记事本修改。密码请
 
 左侧高速通道区域会显示当前极速服务状态：
 
+- `极速 v3`：当前已安装并运行 V3 极速服务。
 - `安装极速`：PGU 设备上还没有极速服务，点击后安装。
+- `极速 v2 · 可升级`：检测到旧版本，点击 `升级极速` 后更新。
 - `升级极速`：检测到旧服务，点击后升级。
 - `修复极速`：当前版本已安装但自动启动失败，点击后修复。
 - `停止极速`：极速服务正在运行，点击后停止。
@@ -161,7 +163,7 @@ IP、端口、用户名和远程路径可以直接用记事本修改。密码请
 
 安装极速服务后，PGU 设备重启时会通过 systemd 和单条 crontab 备用项自动启动。软件会清理重复旧自启命令；当前版本没有运行时会先自动恢复，不会误报为旧版本。
 
-极速 V2 支持中文远程目录、中文文件名、空格以及中文 Windows 下载目录。
+极速 V3 支持中文远程目录、中文文件名、空格以及中文 Windows 下载目录；同时按远端 CPU 数量限制服务并发，使用 HTTP 长连接减少大量小文件下载时的波动。
 
 ## 10. 下载过程
 

pgu4_sync.py

@@ -83,7 +83,7 @@ from PyQt5.QtWidgets import (
 
 APP_ORG = "ClearCore"
 APP_NAME = "PGU4-Sync"
-APP_VERSION = "1.0.4"
+APP_VERSION = "1.0.5"
 CONFIG_FILE_NAME = "PGU-NetFileSyncWin.ini"
 PASSWORD_PROTECTED_KEY = "password_protected"
 PASSWORD_LEGACY_KEY = "password"
@@ -100,13 +100,14 @@ CHUNK_SIZE = 16 * 1024 * 1024
 SPLIT_THRESHOLD = 32 * 1024 * 1024
 PROGRESS_EMIT_STEP = 512 * 1024
 DOWNLOAD_RETRIES = 3
-DOWNLOAD_RETRY_BACKOFF = 0.6
+DOWNLOAD_RETRY_BACKOFF = 0.8
 FAST_HTTP_PORT = 5924
 FAST_HTTP_CHUNK = 4 * 1024 * 1024
-FAST_HTTP_WORKERS = 12
-FAST_HTTP_BIG_FILE_WORKERS = 4
+FAST_HTTP_WORKERS = 8
+FAST_HTTP_BIG_FILE_WORKERS = 3
+FAST_HTTP_TIMEOUT = 45
 FAST_HTTP_TOKEN = "PGU4_SYNC_FAST_V2"
-FAST_SERVICE_VERSION = "2"
+FAST_SERVICE_VERSION = "3"
 FAST_SPLIT_THRESHOLD = 64 * 1024 * 1024
 FAST_REMOTE_APP_DIR = ".local/share/pgu4_sync"
 FAST_REMOTE_STATE_DIR = ".local/state/pgu4_sync"
@@ -122,13 +123,31 @@ FAST_HTTP_SERVER_SCRIPT = r'''#!/usr/bin/env python3
 import argparse
 import base64
 import os
+import threading
 import urllib.parse
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
-PGU_FAST_SERVICE_VERSION = "2"
+PGU_FAST_SERVICE_VERSION = "3"
+
+FAST_SERVER_CPU_COUNT = max(1, os.cpu_count() or 1)
+FAST_SERVER_WORKER_LIMIT = max(2, min(16, FAST_SERVER_CPU_COUNT))
+
+class FastThreadingHTTPServer(ThreadingHTTPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+    request_queue_size = max(32, FAST_SERVER_WORKER_LIMIT * 8)
+
+    def __init__(self, server_address, handler_class):
+        super().__init__(server_address, handler_class)
+        self.request_slots = threading.BoundedSemaphore(FAST_SERVER_WORKER_LIMIT)
+
+    def process_request_thread(self, request, client_address):
+        with self.request_slots:
+            super().process_request_thread(request, client_address)
 
 class Handler(BaseHTTPRequestHandler):
     token = ""
+    protocol_version = "HTTP/1.1"
 
     def log_message(self, fmt, *args):
         return
@@ -143,9 +162,13 @@ class Handler(BaseHTTPRequestHandler):
             self.send_error(403)
             return
         if parsed.path == "/health":
+            body = b"ok-v3"
             self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.send_header("X-PGU-CPU-Count", str(FAST_SERVER_CPU_COUNT))
+            self.send_header("X-PGU-Worker-Limit", str(FAST_SERVER_WORKER_LIMIT))
             self.end_headers()
-            self.wfile.write(b"ok-v2")
+            self.wfile.write(body)
             return
         if parsed.path != "/file":
             self.send_error(404)
@@ -186,7 +209,10 @@ class Handler(BaseHTTPRequestHandler):
                 data = f.read(min(1024 * 1024, remaining))
                 if not data:
                     break
-                self.wfile.write(data)
+                try:
+                    self.wfile.write(data)
+                except (BrokenPipeError, ConnectionResetError):
+                    return
                 remaining -= len(data)
 
 def main():
@@ -196,7 +222,7 @@ def main():
     parser.add_argument("--token", required=True)
     args = parser.parse_args()
     Handler.token = args.token
-    server = ThreadingHTTPServer((args.host, args.port), Handler)
+    server = FastThreadingHTTPServer((args.host, args.port), Handler)
     server.serve_forever()
 
 if __name__ == "__main__":
@@ -237,6 +263,7 @@ class FastHttpProfile:
     host: str
     port: int
     token: str
+    version: str = ""
 
 
 @dataclass(frozen=True)
@@ -379,7 +406,7 @@ class RemoteTableModel(QAbstractTableModel):
 
 
 class ConnectWorker(QObject):
-    connected = pyqtSignal()
+    connected = pyqtSignal(int)
     failed = pyqtSignal(str)
 
     def __init__(self, factory: SSHClientFactory, profile: ConnectionProfile) -> None:
@@ -388,13 +415,26 @@ class ConnectWorker(QObject):
         self.profile = profile
 
     def run(self) -> None:
+        client = None
         try:
             self.factory.set_profile(self.profile)
             client = self.factory.open_client()
-            client.close()
-            self.connected.emit()
+            _stdin, stdout, _stderr = client.exec_command(
+                "getconf _NPROCESSORS_ONLN 2>/dev/null || "
+                "nproc 2>/dev/null || grep -c '^processor' /proc/cpuinfo",
+                timeout=5,
+            )
+            output = stdout.read().decode(errors="ignore").strip().splitlines()
+            try:
+                remote_workers = max(1, min(256, int(output[0].strip()))) if output else 0
+            except (TypeError, ValueError):
+                remote_workers = 0
+            self.connected.emit(remote_workers)
         except Exception as exc:
             self.failed.emit(str(exc))
+        finally:
+            if client is not None:
+                client.close()
 
 
 class ListWorker(QObject):
@@ -699,7 +739,7 @@ class UpdateDownloadWorker(QObject):
 
 
 class FastServiceWorker(QObject):
-    ready = pyqtSignal(str, int, str)
+    ready = pyqtSignal(str, int, str, str)
     failed = pyqtSignal(str)
 
     def __init__(self, factory: SSHClientFactory, host: str, port: int, token: str, replace_existing: bool = False) -> None:
@@ -715,7 +755,7 @@ class FastServiceWorker(QObject):
             if self._has_existing_fast_service() and not self.replace_existing:
                 if self._healthcheck():
                     self._ensure_autostart_only()
-                    self.ready.emit(self.host, self.port, self.token)
+                    self.ready.emit(self.host, self.port, self.token, FAST_SERVICE_VERSION)
                     return
                 else:
                     raise RuntimeError(
@@ -751,6 +791,8 @@ class FastServiceWorker(QObject):
                 f"done; "
                 f"pkill -f '[p]gu4_fast_http.py --port {self.port}' >/dev/null 2>&1 || true; "
                 f"chmod +x {self._shq(script_path)}; "
+                "user_name=$(id -un); loginctl enable-linger \"$user_name\" >/dev/null 2>&1 "
+                "|| sudo -n loginctl enable-linger \"$user_name\" >/dev/null 2>&1 || true; "
                 f"(crontab -l 2>/dev/null | grep -v 'pgu4_fast_http.py.*--port {self.port}' ; printf '%s\\n' {self._shq(cron_line)}) | crontab -; "
                 f"systemctl --user daemon-reload >/dev/null 2>&1 || true; "
                 f"systemctl --user enable {FAST_REMOTE_SERVICE_NAME} >/dev/null 2>&1 || true; "
@@ -773,7 +815,7 @@ class FastServiceWorker(QObject):
             deadline = time.time() + 8
             while time.time() < deadline:
                 if self._healthcheck():
-                    self.ready.emit(self.host, self.port, self.token)
+                    self.ready.emit(self.host, self.port, self.token, FAST_SERVICE_VERSION)
                     return
                 time.sleep(0.2)
             client = self.factory.open_client()
@@ -814,6 +856,8 @@ class FastServiceWorker(QObject):
             )
             cmd = (
                 f"chmod +x {self._shq(script_path)}; "
+                "user_name=$(id -un); loginctl enable-linger \"$user_name\" >/dev/null 2>&1 "
+                "|| sudo -n loginctl enable-linger \"$user_name\" >/dev/null 2>&1 || true; "
                 f"(crontab -l 2>/dev/null | grep -v 'pgu4_fast_http.py.*--port {self.port}' ; printf '%s\\n' {self._shq(cron_line)}) | crontab -; "
                 f"systemctl --user daemon-reload >/dev/null 2>&1 || true; "
                 f"systemctl --user enable {FAST_REMOTE_SERVICE_NAME} >/dev/null 2>&1 || true"
@@ -854,7 +898,7 @@ class FastServiceWorker(QObject):
             path = "/health?token={}".format(urllib.parse.quote(self.token))
             conn.request("GET", path)
             resp = conn.getresponse()
-            ok = resp.status == 200 and resp.read().startswith(b"ok-v2")
+            ok = resp.status == 200 and resp.read().startswith(b"ok-v3")
             conn.close()
             return ok
         except Exception:
@@ -897,7 +941,7 @@ class StopFastServiceWorker(QObject):
 
 
 class CheckFastServiceWorker(QObject):
-    available = pyqtSignal(str, int, str)
+    available = pyqtSignal(str, int, str, str)
     legacy = pyqtSignal(str)
     repair = pyqtSignal(str)
     unavailable = pyqtSignal()
@@ -910,28 +954,41 @@ class CheckFastServiceWorker(QObject):
         self.token = token
 
     def run(self) -> None:
-        if self._healthcheck():
+        healthy, version = self._healthcheck_details()
+        if healthy and version == FAST_SERVICE_VERSION:
             try:
                 self._ensure_autostart()
             except Exception:
                 pass
-            self.available.emit(self.host, self.port, self.token)
+            self.available.emit(self.host, self.port, self.token, version or FAST_SERVICE_VERSION)
+            return
+
+        if healthy and version:
+            self.legacy.emit(f"检测到 5924 上已有旧极速服务，当前版本 v{version}")
             return
 
         installed_state = self._installed_service_state()
         if installed_state == "current":
             if self._restart_current_service():
-                self.available.emit(self.host, self.port, self.token)
+                self.available.emit(self.host, self.port, self.token, FAST_SERVICE_VERSION)
             else:
-                self.repair.emit("当前版本极速服务已安装，但自动启动失败")
+                version = self._installed_service_version()
+                version_text = f"v{version}" if version else "当前版本"
+                self.repair.emit(f"极速服务 {version_text} 已安装，但自动启动失败")
             return
 
         if installed_state == "legacy" or self._has_legacy_service():
-            self.legacy.emit(f"检测到 5924 上已有旧极速服务或旧 token 服务")
+            version = self._installed_service_version()
+            version_text = f"v{version}" if version else "未知版本"
+            self.legacy.emit(f"检测到 5924 上已有旧极速服务，当前版本 {version_text}")
         else:
             self.unavailable.emit()
 
     def _healthcheck(self) -> bool:
+        healthy, _version = self._healthcheck_details()
+        return healthy
+
+    def _healthcheck_details(self) -> tuple[bool, str]:
         try:
             conn = http.client.HTTPConnection(self.host, self.port, timeout=1.5)
             path = "/health?token={}".format(
@@ -939,11 +996,14 @@ class CheckFastServiceWorker(QObject):
             )
             conn.request("GET", path)
             resp = conn.getresponse()
-            ok = resp.status == 200 and resp.read().startswith(b"ok-v2")
+            body = resp.read()
+            match = re.match(rb"ok-v(\d+)", body)
+            version = match.group(1).decode() if match else ""
+            ok = resp.status == 200 and bool(match)
             conn.close()
-            return ok
+            return ok, version
         except Exception:
-            return False
+            return False, ""
 
     def _installed_service_state(self) -> str:
         try:
@@ -956,7 +1016,7 @@ class CheckFastServiceWorker(QObject):
                 f"script=\"$HOME/{script_rel}\"; "
                 f"if [ ! -f \"$script\" ]; then echo missing; "
                 f"elif grep -q 'PGU_FAST_SERVICE_VERSION = \"{FAST_SERVICE_VERSION}\"' \"$script\" "
-                f"|| grep -q 'ok-v2' \"$script\"; then echo current; "
+                f"|| grep -q 'ok-v3' \"$script\"; then echo current; "
                 f"else echo legacy; fi"
             )
             _stdin, stdout, _stderr = client.exec_command(cmd, timeout=4)
@@ -966,6 +1026,22 @@ class CheckFastServiceWorker(QObject):
         except Exception:
             return "missing"
 
+    def _installed_service_version(self) -> str:
+        try:
+            client = self.factory.open_client()
+            script_rel = posixpath.join(FAST_REMOTE_APP_DIR, FAST_REMOTE_SCRIPT_NAME)
+            cmd = (
+                f"script=\"$HOME/{script_rel}\"; "
+                "grep -o 'PGU_FAST_SERVICE_VERSION = \"[0-9][0-9]*\"' \"$script\" "
+                "| grep -o '[0-9][0-9]*' | head -1"
+            )
+            _stdin, stdout, _stderr = client.exec_command(cmd, timeout=4)
+            version = stdout.read().decode(errors="ignore").strip()
+            client.close()
+            return version
+        except Exception:
+            return ""
+
     def _restart_current_service(self) -> bool:
         try:
             self._ensure_autostart()
@@ -978,6 +1054,8 @@ class CheckFastServiceWorker(QObject):
                 f"tr '\\0' ' ' < /proc/$pid/cmdline | grep -q 'pgu4_fast_http.py' "
                 f"&& kill $pid >/dev/null 2>&1 || true; done; "
                 f"pkill -f '[p]gu4_fast_http.py --port {self.port}' >/dev/null 2>&1 || true; "
+                "user_name=$(id -un); loginctl enable-linger \"$user_name\" >/dev/null 2>&1 "
+                "|| sudo -n loginctl enable-linger \"$user_name\" >/dev/null 2>&1 || true; "
                 f"systemctl --user daemon-reload >/dev/null 2>&1 || true; "
                 f"systemctl --user enable {FAST_REMOTE_SERVICE_NAME} >/dev/null 2>&1 || true; "
                 f"if ! systemctl --user restart {FAST_REMOTE_SERVICE_NAME} >/dev/null 2>&1; then "
@@ -1052,6 +1130,8 @@ class CheckFastServiceWorker(QObject):
             )
             cmd = (
                 f"chmod +x {shq(script_path)}; "
+                "user_name=$(id -un); loginctl enable-linger \"$user_name\" >/dev/null 2>&1 "
+                "|| sudo -n loginctl enable-linger \"$user_name\" >/dev/null 2>&1 || true; "
                 f"(crontab -l 2>/dev/null | grep -v 'pgu4_fast_http.py.*--port {self.port}' ; printf '%s\\n' {shq(cron_line)}) | crontab -; "
                 f"systemctl --user daemon-reload >/dev/null 2>&1 || true; "
                 f"systemctl --user enable {FAST_REMOTE_SERVICE_NAME} >/dev/null 2>&1 || true"
@@ -1484,6 +1564,8 @@ class DownloadJob(QRunnable):
         self._thread_local = threading.local()
         self._clients: list[paramiko.SSHClient] = []
         self._clients_lock = threading.Lock()
+        self._http_connections: list[http.client.HTTPConnection] = []
+        self._http_connections_lock = threading.Lock()
         self._file_locks: dict[Path, threading.Lock] = {}
         self._progress_lock = threading.Lock()
         self._progress_by_path: dict[str, int] = {}
@@ -1522,6 +1604,7 @@ class DownloadJob(QRunnable):
             for worker in workers:
                 worker.join(timeout=2)
             self._close_thread_clients()
+            self._close_http_connections()
             self.signals.all_done.emit(cancelled or self.controller.cancel_event.is_set())
 
     def _transfer_worker_count(self) -> int:
@@ -1709,9 +1792,12 @@ class DownloadJob(QRunnable):
             self.signals.item_done.emit(file.remote_path)
             return
         last_error = None
-        for attempt in range(1, DOWNLOAD_RETRIES + 1):
+        outer_attempts = 1 if self.fast_http is not None else DOWNLOAD_RETRIES
+        for attempt in range(1, outer_attempts + 1):
             try:
                 if self.fast_http is not None:
+                    # The HTTP path owns retries so a broken connection can
+                    # resume the unfinished range instead of restarting it.
                     self._download_chunk_http(file, offset, length)
                 else:
                     self._download_chunk_sftp_once(file, offset, length)
@@ -1721,13 +1807,21 @@ class DownloadJob(QRunnable):
                 self._reset_thread_client()
                 if self.controller.cancel_event.is_set():
                     return
-                if attempt < DOWNLOAD_RETRIES:
+                if attempt < DOWNLOAD_RETRIES and self.fast_http is None:
                     self.signals.status.emit(
                         f"重试 {attempt}/{DOWNLOAD_RETRIES - 1}: {Path(file.remote_path).name}"
                     )
-                    time.sleep(DOWNLOAD_RETRY_BACKOFF * attempt)
+                    self._retry_wait(attempt)
         self.signals.failed.emit(file.remote_path, str(last_error) if last_error else "下载失败")
 
+    def _retry_wait(self, attempt: int) -> None:
+        deadline = time.monotonic() + DOWNLOAD_RETRY_BACKOFF * (2 ** max(0, attempt - 1))
+        while time.monotonic() < deadline:
+            if self.controller.cancel_event.is_set():
+                return
+            self.controller.wait_if_paused()
+            time.sleep(min(0.1, max(0.0, deadline - time.monotonic())))
+
     def _download_chunk_sftp_once(self, file: FileTask, offset: int, length: int) -> None:
         if self.fast_http is not None:
             return
@@ -1782,41 +1876,89 @@ class DownloadJob(QRunnable):
     def _download_chunk_http_locked(self, file: FileTask, offset: int, length: int) -> None:
         written = 0
         emit_acc = 0
-        conn = None
         path = "/file?token={}&path_b64={}".format(
             urllib.parse.quote(self.fast_http.token, safe=""),
             urllib.parse.quote(encode_fast_remote_path(file.remote_path), safe=""),
         )
-        try:
-            conn = http.client.HTTPConnection(self.fast_http.host, self.fast_http.port, timeout=15)
-            headers = {"Range": f"bytes={offset}-{offset + length - 1}"}
-            conn.request("GET", path, headers=headers)
-            resp = conn.getresponse()
-            if resp.status not in (200, 206):
-                raise RuntimeError(f"HTTP {resp.status} {resp.reason}")
-            with file.local_path.open("r+b") as local:
-                while written < length:
-                    self.controller.wait_if_paused()
-                    if self.controller.cancel_event.is_set():
-                        return
-                    data = resp.read(min(FAST_HTTP_CHUNK, length - written))
-                    if not data:
-                        break
-                    with self._file_locks[file.local_path]:
-                        local.seek(offset + written)
-                        local.write(data)
-                    written += len(data)
-                    emit_acc += len(data)
-                    if emit_acc >= PROGRESS_EMIT_STEP:
+        last_error = None
+        for attempt in range(1, DOWNLOAD_RETRIES + 1):
+            try:
+                self.controller.wait_if_paused()
+                if self.controller.cancel_event.is_set():
+                    return
+                conn = self._get_http_connection()
+                range_start = offset + written
+                headers = {
+                    "Range": f"bytes={range_start}-{offset + length - 1}",
+                    "Connection": "keep-alive",
+                }
+                conn.request("GET", path, headers=headers)
+                resp = conn.getresponse()
+                expected_status = 206
+                if resp.status != expected_status:
+                    raise RuntimeError(f"HTTP {resp.status} {resp.reason}")
+                with file.local_path.open("r+b") as local:
+                    while written < length:
+                        self.controller.wait_if_paused()
+                        if self.controller.cancel_event.is_set():
+                            self._reset_thread_http_connection()
+                            return
+                        data = resp.read(min(FAST_HTTP_CHUNK, length - written))
+                        if not data:
+                            break
+                        with self._file_locks[file.local_path]:
+                            local.seek(offset + written)
+                            local.write(data)
+                        written += len(data)
+                        emit_acc += len(data)
+                        if emit_acc >= PROGRESS_EMIT_STEP:
+                            self._emit_chunk_progress(file, offset, written)
+                            emit_acc = 0
+                    if emit_acc:
                         self._emit_chunk_progress(file, offset, written)
-                        emit_acc = 0
-                if emit_acc:
-                    self._emit_chunk_progress(file, offset, written)
-            if written < length and not self.controller.cancel_event.is_set():
+                if written >= length:
+                    return
                 raise IOError(f"连接提前结束: {written}/{length}")
-        finally:
-            if conn is not None:
-                conn.close()
+            except Exception as exc:
+                last_error = exc
+                self._reset_thread_http_connection()
+                if self.controller.cancel_event.is_set():
+                    return
+                if attempt < DOWNLOAD_RETRIES:
+                    self.signals.status.emit(
+                        f"极速通道重试 {attempt}/{DOWNLOAD_RETRIES - 1}: {Path(file.remote_path).name}"
+                    )
+                    self._retry_wait(attempt)
+        raise last_error or IOError("极速通道下载失败")
+
+    def _get_http_connection(self) -> http.client.HTTPConnection:
+        conn = getattr(self._thread_local, "http_conn", None)
+        if conn is None:
+            conn = http.client.HTTPConnection(
+                self.fast_http.host,
+                self.fast_http.port,
+                timeout=FAST_HTTP_TIMEOUT,
+            )
+            self._thread_local.http_conn = conn
+            with self._http_connections_lock:
+                self._http_connections.append(conn)
+        return conn
+
+    def _reset_thread_http_connection(self) -> None:
+        conn = getattr(self._thread_local, "http_conn", None)
+        if conn is None:
+            return
+        try:
+            conn.close()
+        except Exception:
+            pass
+        with self._http_connections_lock:
+            if conn in self._http_connections:
+                self._http_connections.remove(conn)
+        try:
+            delattr(self._thread_local, "http_conn")
+        except Exception:
+            pass
 
     def _emit_chunk_progress(self, file: FileTask, offset: int, chunk_written: int) -> None:
         if chunk_written <= 0:
@@ -1871,6 +2013,16 @@ class DownloadJob(QRunnable):
             except Exception:
                 pass
 
+    def _close_http_connections(self) -> None:
+        with self._http_connections_lock:
+            connections = list(self._http_connections)
+            self._http_connections.clear()
+        for conn in connections:
+            try:
+                conn.close()
+            except Exception:
+                pass
+
 
 class RemoteTableView(QTableView):
     external_drag_requested = pyqtSignal()
@@ -2165,7 +2317,9 @@ class MainWindow(QMainWindow):
         self.download_signals = DownloadSignals()
         self.controller: TransferController | None = None
         self.fast_http: FastHttpProfile | None = None
-        self.max_workers = get_logical_thread_count()
+        self.local_workers = get_logical_thread_count()
+        self.remote_workers = 0
+        self.max_workers = self.local_workers
         self._threads: list[QThread] = []
         self._active_download = False
         self._active_delete = False
@@ -2176,6 +2330,7 @@ class MainWindow(QMainWindow):
         self._download_start_time = 0.0
         self._current_speed = 0.0
         self._channel_label = "普通通道"
+        self._fast_service_version = ""
         self._download_errors: list[tuple[str, str]] = []
         self._fast_service_legacy = False
         self._fast_service_repair = False
@@ -3707,7 +3862,9 @@ class MainWindow(QMainWindow):
         else:
             self.start_fast_service()
 
-    def set_fast_button_state(self, state: str) -> None:
+    def set_fast_button_state(self, state: str, version: str | None = None) -> None:
+        if version is not None:
+            self._fast_service_version = version.strip().lstrip("vV")
         self.fast_btn.setProperty("fastState", state)
         self.fast_btn.style().unpolish(self.fast_btn)
         self.fast_btn.style().polish(self.fast_btn)
@@ -3720,7 +3877,14 @@ class MainWindow(QMainWindow):
                 "stop": "极速已启用",
                 "busy": "极速检测中",
             }
-            self.fast_status_badge.setText(labels.get(state, "极速未启用"))
+            label = labels.get(state, "极速未启用")
+            if self._fast_service_version and state in {"stop", "upgrade", "repair"}:
+                label = f"极速 v{self._fast_service_version}"
+                if state == "upgrade":
+                    label += " · 可升级"
+                elif state == "repair":
+                    label += " · 需修复"
+            self.fast_status_badge.setText(label)
             self.fast_status_badge.setProperty("fastState", state)
             self.fast_status_badge.style().unpolish(self.fast_status_badge)
             self.fast_status_badge.style().polish(self.fast_status_badge)
@@ -3768,7 +3932,7 @@ class MainWindow(QMainWindow):
         thread = QThread(self)
         worker.moveToThread(thread)
         thread.worker = worker
-        worker.available.connect(lambda h, p, t: self.on_fast_service_detected(h, p, t, thread))
+        worker.available.connect(lambda h, p, t, v: self.on_fast_service_detected(h, p, t, v, thread))
         worker.legacy.connect(lambda msg: self.on_fast_service_legacy(msg, thread))
         worker.repair.connect(lambda msg: self.on_fast_service_repair(msg, thread))
         worker.unavailable.connect(lambda: self.on_fast_service_unavailable(thread))
@@ -3777,17 +3941,18 @@ class MainWindow(QMainWindow):
         self._threads.append(thread)
         thread.start()
 
-    def on_fast_service_detected(self, host: str, port: int, token: str, thread: QThread) -> None:
+    def on_fast_service_detected(self, host: str, port: int, token: str, version: str, thread: QThread) -> None:
         if self._closing:
             self._finish_thread(thread)
             return
-        self.fast_http = FastHttpProfile(host, port, token)
+        self.fast_http = FastHttpProfile(host, port, token, version)
+        self._fast_service_version = version
         self._fast_service_legacy = False
         self._fast_service_repair = False
         self.fast_btn.setEnabled(True)
         self.fast_btn.setText("停止极速")
-        self.set_fast_button_state("stop")
-        self.statusBar().showMessage("检测到远端极速服务，已启用极速通道", 5000)
+        self.set_fast_button_state("stop", version)
+        self.statusBar().showMessage(f"检测到远端极速服务 v{version}，已启用极速通道", 5000)
         self._finish_thread(thread)
 
     def on_fast_service_unavailable(self, thread: QThread) -> None:
@@ -3795,6 +3960,7 @@ class MainWindow(QMainWindow):
             self._finish_thread(thread)
             return
         self.fast_http = None
+        self._fast_service_version = ""
         self._fast_service_legacy = False
         self._fast_service_repair = False
         self.fast_btn.setEnabled(True)
@@ -3810,9 +3976,11 @@ class MainWindow(QMainWindow):
         self.fast_http = None
         self._fast_service_legacy = True
         self._fast_service_repair = False
+        match = re.search(r"当前版本\s+v?([0-9]+)", message)
+        self._fast_service_version = match.group(1) if match else ""
         self.fast_btn.setEnabled(True)
         self.fast_btn.setText("升级极速")
-        self.set_fast_button_state("upgrade")
+        self.set_fast_button_state("upgrade", self._fast_service_version)
         self.statusBar().showMessage(message + "，可点击升级极速", 7000)
         self._finish_thread(thread)
 
@@ -3823,9 +3991,11 @@ class MainWindow(QMainWindow):
         self.fast_http = None
         self._fast_service_legacy = False
         self._fast_service_repair = True
+        match = re.search(r"极速服务\s+v?([0-9]+)", message)
+        self._fast_service_version = match.group(1) if match else self._fast_service_version
         self.fast_btn.setEnabled(True)
         self.fast_btn.setText("修复极速")
-        self.set_fast_button_state("repair")
+        self.set_fast_button_state("repair", self._fast_service_version)
         self.statusBar().showMessage(message + "，可点击修复极速", 7000)
         self._finish_thread(thread)
 
@@ -3854,7 +4024,7 @@ class MainWindow(QMainWindow):
         thread = QThread(self)
         worker.moveToThread(thread)
         thread.worker = worker
-        worker.ready.connect(lambda host, port, t: self.on_fast_service_ready(host, port, t, thread))
+        worker.ready.connect(lambda host, port, t, v: self.on_fast_service_ready(host, port, t, v, thread))
         worker.failed.connect(lambda msg: self.on_fast_service_failed(msg, thread))
         thread.started.connect(worker.run)
         thread.finished.connect(lambda: self._threads.remove(thread) if thread in self._threads else None)
@@ -3862,17 +4032,18 @@ class MainWindow(QMainWindow):
         thread.start()
         QTimer.singleShot(20000, lambda: self.on_fast_service_timeout(thread))
 
-    def on_fast_service_ready(self, host: str, port: int, token: str, thread: QThread) -> None:
+    def on_fast_service_ready(self, host: str, port: int, token: str, version: str, thread: QThread) -> None:
         if self._closing:
             self._finish_thread(thread)
             return
-        self.fast_http = FastHttpProfile(host, port, token)
+        self.fast_http = FastHttpProfile(host, port, token, version)
+        self._fast_service_version = version
         self._fast_service_legacy = False
         self._fast_service_repair = False
         self.fast_btn.setEnabled(True)
         self.fast_btn.setText("停止极速")
-        self.set_fast_button_state("stop")
-        self.statusBar().showMessage(f"远端极速服务已运行，已启用极速通道: http://{host}:{port}", 5000)
+        self.set_fast_button_state("stop", version)
+        self.statusBar().showMessage(f"远端极速服务 v{version} 已运行，已启用极速通道: http://{host}:{port}", 5000)
         self._finish_thread(thread)
 
     def on_fast_service_failed(self, message: str, thread: QThread) -> None:
@@ -3880,6 +4051,8 @@ class MainWindow(QMainWindow):
             self._finish_thread(thread)
             return
         self.fast_http = None
+        if not self._fast_service_legacy and not self._fast_service_repair:
+            self._fast_service_version = ""
         self.fast_btn.setEnabled(True)
         if self._fast_service_repair:
             self.fast_btn.setText("修复极速")
@@ -3975,10 +4148,10 @@ class MainWindow(QMainWindow):
         self.statusBar().showMessage("正在连接设备...")
         worker = ConnectWorker(self.factory, profile)
         thread = self._run_worker(worker, worker.run)
-        worker.connected.connect(lambda: self.on_connected(profile, thread))
+        worker.connected.connect(lambda remote_workers: self.on_connected(profile, remote_workers, thread))
         worker.failed.connect(lambda msg: self.on_connect_failed(msg, thread))
 
-    def on_connected(self, profile: ConnectionProfile, thread: QThread) -> None:
+    def on_connected(self, profile: ConnectionProfile, remote_workers: int, thread: QThread) -> None:
         if self._closing:
             self._finish_thread(thread)
             return
@@ -3986,6 +4159,18 @@ class MainWindow(QMainWindow):
         self.settings.setValue("port", profile.port)
         self.settings.setValue("username", profile.username)
         save_password_setting(self.settings, profile.password)
+        self.remote_workers = remote_workers
+        self.max_workers = adaptive_worker_count(self.local_workers, remote_workers)
+        if hasattr(self, "worker_label"):
+            self.worker_label.setText(
+                f"{self.max_workers} 线程 · {min(FAST_HTTP_WORKERS, self.max_workers)} 路并发"
+            )
+            if remote_workers > 0:
+                self.worker_label.setToolTip(
+                    f"Windows 本机 {self.local_workers} 线程，远端设备 {remote_workers} 线程，自动取较小值"
+                )
+            else:
+                self.worker_label.setToolTip(f"Windows 本机 {self.local_workers} 线程")
         self._connected = True
         self._path_history = []
         self._history_index = -1
@@ -5058,6 +5243,12 @@ def get_logical_thread_count() -> int:
     return max(1, os.cpu_count() or 1)
 
 
+def adaptive_worker_count(local_workers: int, remote_workers: int) -> int:
+    local = max(1, int(local_workers or 1))
+    remote = max(0, int(remote_workers or 0))
+    return min(local, remote) if remote else local
+
+
 class POINT(ctypes.Structure):
     _fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)]
 

tests/test_commercial_ui.py

@@ -7,7 +7,7 @@ os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
 from PyQt5.QtCore import QItemSelectionModel
 from PyQt5.QtWidgets import QApplication, QStatusBar
 
-from pgu4_sync import ConnectionDialog, MainWindow, RemoteEntry
+from pgu4_sync import ConnectionDialog, MainWindow, RemoteEntry, adaptive_worker_count
 
 
 class CommercialUiTests(unittest.TestCase):
@@ -156,3 +156,18 @@ class CommercialUiTests(unittest.TestCase):
         self.assertEqual("root", dialog.user_edit.text())
         self.assertEqual("Clear8492..", dialog.pass_edit.text())
         self.assertEqual("/nvme-data/pgu_data", dialog.selected_remote_path)
+
+    def test_worker_count_uses_smaller_local_or_remote_cpu_count(self):
+        self.assertEqual(8, adaptive_worker_count(32, 8))
+        self.assertEqual(6, adaptive_worker_count(6, 8))
+        self.assertEqual(32, adaptive_worker_count(32, 0))
+
+    def test_fast_service_version_is_visible_and_upgradeable(self):
+        window = MainWindow()
+        self.addCleanup(window.close)
+
+        window.set_fast_button_state("stop", "3")
+        self.assertEqual("极速 v3", window.fast_status_badge.text())
+
+        window.set_fast_button_state("upgrade", "2")
+        self.assertEqual("极速 v2 · 可升级", window.fast_status_badge.text())

tests/test_fast_service_recovery.py

@@ -60,7 +60,12 @@ class FastServiceRecoveryTests(unittest.TestCase):
         self.assertEqual([], repair)
 
     def test_service_script_has_persistent_version_marker(self):
-        self.assertIn('PGU_FAST_SERVICE_VERSION = "2"', FAST_HTTP_SERVER_SCRIPT)
+        self.assertIn('PGU_FAST_SERVICE_VERSION = "3"', FAST_HTTP_SERVER_SCRIPT)
+        self.assertIn('protocol_version = "HTTP/1.1"', FAST_HTTP_SERVER_SCRIPT)
+        self.assertIn("request_queue_size", FAST_HTTP_SERVER_SCRIPT)
+        self.assertIn("BoundedSemaphore", FAST_HTTP_SERVER_SCRIPT)
+        self.assertIn("X-PGU-Worker-Limit", FAST_HTTP_SERVER_SCRIPT)
         source = inspect.getsource(FastServiceWorker)
         self.assertIn("recv_exit_status", source)
         self.assertIn("pgu4_fast_http.py.*--port", source)
+        self.assertIn("enable-linger", source)

tests/test_unicode_paths.py

@@ -162,7 +162,7 @@ class UnicodePathTests(unittest.TestCase):
                         "/health?token=" + urllib.parse.quote(token, safe=""),
                     )
                     response = conn.getresponse()
-                    if response.status == 200 and response.read() == b"ok-v2":
+                    if response.status == 200 and response.read() == b"ok-v3":
                         conn.close()
                         break
                     conn.close()

tests/test_update.py

@@ -35,7 +35,7 @@ class UpdateTests(unittest.TestCase):
         self.assertEqual([True], finished)
 
     def test_version_comparison(self):
-        self.assertTrue(is_newer_version("1.0.5", APP_VERSION))
+        self.assertTrue(is_newer_version("1.0.6", APP_VERSION))
         self.assertTrue(is_newer_version("v1.1.0", "1.0.9"))
         self.assertTrue(is_newer_version("1.0.10", "1.0.2"))
         self.assertFalse(is_newer_version("v1.0.4", APP_VERSION))