feat: optimize async io performance and benchmark coverage (#5737)

* docs: align deployment sections across multilingual readmes

* docs: normalize deployment punctuation and AUR guidance

* docs: fix french and russian deployment wording

* perf: optimize async io hot paths and extend benchmarks

* fix: address async io review feedback

* fix: address follow-up async io review comments

* fix: align base64 io error handling in message components

* fix: harden attachment export ids and tune io chunking

* fix: preserve best-effort attachment export and batch writes

* test: expand path conversion and helper coverage

Author: zouyonghe

astrbot/core/backup/exporter.py

@@ -161,10 +161,10 @@ async def export_all(
                 # 3. 导出配置文件
                 if progress_callback:
                     await progress_callback("config", 0, 100, "正在导出配置文件...")
-                if await asyncio.to_thread(os.path.exists, self.config_path):
-                    config_content = await asyncio.to_thread(
-                        Path(self.config_path).read_text, encoding="utf-8"
-                    )
+                config_content = await asyncio.to_thread(
+                    self._read_text_if_exists, self.config_path
+                )
+                if config_content is not None:
                     zf.writestr("config/cmd_config.json", config_content)
                     self._add_checksum("config/cmd_config.json", config_content)
                 if progress_callback:
@@ -361,17 +361,44 @@ async def _export_attachments(
         self, zf: zipfile.ZipFile, attachments: list[dict]
     ) -> None:
         """导出附件文件"""
+        await asyncio.to_thread(self._export_attachments_sync, zf, attachments)
+
+    def _export_attachments_sync(
+        self, zf: zipfile.ZipFile, attachments: list[dict]
+    ) -> None:
+        """在单个线程中批量导出附件，减少高频线程切换。"""
         for attachment in attachments:
+            file_path = attachment.get("path", "")
+            attachment_id = attachment.get("attachment_id")
             try:
-                file_path = attachment.get("path", "")
-                if file_path and await asyncio.to_thread(os.path.exists, file_path):
-                    # 使用 attachment_id 作为文件名
-                    attachment_id = attachment.get("attachment_id", "")
-                    ext = os.path.splitext(file_path)[1]
-                    archive_path = f"files/attachments/{attachment_id}{ext}"
-                    zf.write(file_path, archive_path)
+                if not file_path:
+                    continue
+                if not attachment_id:
+                    logger.warning(
+                        f"跳过附件导出：attachment_id 为空 (path={file_path})"
+                    )
+                    continue
+                # 使用 attachment_id 作为文件名
+                ext = os.path.splitext(file_path)[1]
+                archive_path = f"files/attachments/{attachment_id}{ext}"
+                zf.write(file_path, archive_path)
+            except FileNotFoundError:
+                # 和旧逻辑保持一致：缺失文件直接跳过。
+                continue
+            except OSError as e:
+                logger.warning(
+                    f"导出附件失败 (path={file_path}, attachment_id={attachment_id or 'unknown'}): {e}"
+                )
             except Exception as e:
-                logger.warning(f"导出附件失败: {e}")
+                logger.warning(
+                    f"导出附件时发生非预期错误，已跳过 (path={file_path}, attachment_id={attachment_id or 'unknown'}): {e}"
+                )
+
+    def _read_text_if_exists(self, file_path: str) -> str | None:
+        """Read text file when it exists in a single synchronous call."""
+        if not os.path.exists(file_path):
+            return None
+        return Path(file_path).read_text(encoding="utf-8")
 
     def _model_to_dict(self, record: Any) -> dict:
         """将 SQLModel 实例转换为字典

astrbot/core/message/components.py

@@ -40,6 +40,16 @@
 from astrbot.core.utils.io import download_file, download_image_by_url, file_to_base64
 
 
+def _absolute_path(path: str) -> str:
+    return os.path.abspath(path)
+
+
+def _absolute_path_if_exists(path: str | None) -> str | None:
+    if not path or not os.path.exists(path):
+        return None
+    return os.path.abspath(path)
+
+
 class ComponentType(StrEnum):
     # Basic Segment Types
     Plain = "Plain"  # plain text message
@@ -159,17 +169,18 @@ async def convert_to_file_path(self) -> str:
             return self.file[8:]
         if self.file.startswith("http"):
             file_path = await download_image_by_url(self.file)
-            return await asyncio.to_thread(os.path.abspath, file_path)
+            return await asyncio.to_thread(_absolute_path, file_path)
         if self.file.startswith("base64://"):
             bs64_data = self.file.removeprefix("base64://")
             image_bytes = base64.b64decode(bs64_data)
             file_path = os.path.join(
                 get_astrbot_temp_path(), f"recordseg_{uuid.uuid4()}.jpg"
             )
             await asyncio.to_thread(Path(file_path).write_bytes, image_bytes)
-            return await asyncio.to_thread(os.path.abspath, file_path)
-        if await asyncio.to_thread(os.path.exists, self.file):
-            return await asyncio.to_thread(os.path.abspath, self.file)
+            return await asyncio.to_thread(_absolute_path, file_path)
+        local_path = await asyncio.to_thread(_absolute_path_if_exists, self.file)
+        if local_path:
+            return local_path
         raise Exception(f"not a valid file: {self.file}")
 
     async def convert_to_base64(self) -> str:
@@ -189,10 +200,11 @@ async def convert_to_base64(self) -> str:
             bs64_data = await file_to_base64(file_path)
         elif self.file.startswith("base64://"):
             bs64_data = self.file
-        elif await asyncio.to_thread(os.path.exists, self.file):
-            bs64_data = await file_to_base64(self.file)
         else:
-            raise Exception(f"not a valid file: {self.file}")
+            try:
+                bs64_data = await file_to_base64(self.file)
+            except OSError as exc:
+                raise Exception(f"not a valid file: {self.file}") from exc
         bs64_data = bs64_data.removeprefix("base64://")
         return bs64_data
 
@@ -256,11 +268,15 @@ async def convert_to_file_path(self) -> str:
                 get_astrbot_temp_path(), f"videoseg_{uuid.uuid4().hex}"
             )
             await download_file(url, video_file_path)
-            if await asyncio.to_thread(os.path.exists, video_file_path):
-                return await asyncio.to_thread(os.path.abspath, video_file_path)
+            local_path = await asyncio.to_thread(
+                _absolute_path_if_exists, video_file_path
+            )
+            if local_path:
+                return local_path
             raise Exception(f"download failed: {url}")
-        if await asyncio.to_thread(os.path.exists, url):
-            return await asyncio.to_thread(os.path.abspath, url)
+        local_path = await asyncio.to_thread(_absolute_path_if_exists, url)
+        if local_path:
+            return local_path
         raise Exception(f"not a valid file: {url}")
 
     async def register_to_file_service(self) -> str:
@@ -449,17 +465,18 @@ async def convert_to_file_path(self) -> str:
             return url[8:]
         if url.startswith("http"):
             image_file_path = await download_image_by_url(url)
-            return await asyncio.to_thread(os.path.abspath, image_file_path)
+            return await asyncio.to_thread(_absolute_path, image_file_path)
         if url.startswith("base64://"):
             bs64_data = url.removeprefix("base64://")
             image_bytes = base64.b64decode(bs64_data)
             image_file_path = os.path.join(
                 get_astrbot_temp_path(), f"imgseg_{uuid.uuid4()}.jpg"
             )
             await asyncio.to_thread(Path(image_file_path).write_bytes, image_bytes)
-            return await asyncio.to_thread(os.path.abspath, image_file_path)
-        if await asyncio.to_thread(os.path.exists, url):
-            return await asyncio.to_thread(os.path.abspath, url)
+            return await asyncio.to_thread(_absolute_path, image_file_path)
+        local_path = await asyncio.to_thread(_absolute_path_if_exists, url)
+        if local_path:
+            return local_path
         raise Exception(f"not a valid file: {url}")
 
     async def convert_to_base64(self) -> str:
@@ -480,10 +497,11 @@ async def convert_to_base64(self) -> str:
             bs64_data = await file_to_base64(image_file_path)
         elif url.startswith("base64://"):
             bs64_data = url
-        elif await asyncio.to_thread(os.path.exists, url):
-            bs64_data = await file_to_base64(url)
         else:
-            raise Exception(f"not a valid file: {url}")
+            try:
+                bs64_data = await file_to_base64(url)
+            except OSError as exc:
+                raise Exception(f"not a valid file: {url}") from exc
         bs64_data = bs64_data.removeprefix("base64://")
         return bs64_data
 
@@ -734,8 +752,9 @@ async def get_file(self, allow_return_url: bool = False) -> str:
                 ):
                     path = path[1:]
 
-            if await asyncio.to_thread(os.path.exists, path):
-                return await asyncio.to_thread(os.path.abspath, path)
+            local_path = await asyncio.to_thread(_absolute_path_if_exists, path)
+            if local_path:
+                return local_path
 
         if self.url:
             await self._download_file()
@@ -750,7 +769,7 @@ async def get_file(self, allow_return_url: bool = False) -> str:
                         and path[2] == ":"
                     ):
                         path = path[1:]
-                return await asyncio.to_thread(os.path.abspath, path)
+                return await asyncio.to_thread(_absolute_path, path)
 
         return ""
 
@@ -766,7 +785,7 @@ async def _download_file(self) -> None:
             filename = f"fileseg_{uuid.uuid4().hex}"
         file_path = os.path.join(download_dir, filename)
         await download_file(self.url, file_path)
-        self.file_ = await asyncio.to_thread(os.path.abspath, file_path)
+        self.file_ = await asyncio.to_thread(_absolute_path, file_path)
 
     async def register_to_file_service(self) -> str:
         """将文件注册到文件服务。

astrbot/core/utils/io.py

@@ -9,6 +9,7 @@
 import uuid
 import zipfile
 from pathlib import Path
+from typing import BinaryIO
 
 import aiohttp
 import certifi
@@ -18,6 +19,8 @@
 from .astrbot_path import get_astrbot_data_path, get_astrbot_path, get_astrbot_temp_path
 
 logger = logging.getLogger("astrbot")
+_DOWNLOAD_READ_CHUNK_SIZE = 64 * 1024
+_DOWNLOAD_FLUSH_THRESHOLD = 256 * 1024
 
 
 def on_error(func, path, exc_info) -> None:
@@ -134,29 +137,18 @@ async def download_file(url: str, path: str, show_progress: bool = False) -> Non
                 if resp.status != 200:
                     raise Exception(f"下载文件失败: {resp.status}")
                 total_size = int(resp.headers.get("content-length", 0))
-                downloaded_size = 0
                 start_time = time.time()
                 if show_progress:
                     print(f"文件大小: {total_size / 1024:.2f} KB | 文件地址: {url}")
                 file_obj = await asyncio.to_thread(Path(path).open, "wb")
                 try:
-                    while True:
-                        chunk = await resp.content.read(8192)
-                        if not chunk:
-                            break
-                        await asyncio.to_thread(file_obj.write, chunk)
-                        downloaded_size += len(chunk)
-                        if show_progress:
-                            elapsed_time = (
-                                time.time() - start_time
-                                if time.time() - start_time > 0
-                                else 1
-                            )
-                            speed = downloaded_size / 1024 / elapsed_time  # KB/s
-                            print(
-                                f"\r下载进度: {downloaded_size / total_size:.2%} 速度: {speed:.2f} KB/s",
-                                end="",
-                            )
+                    await _stream_to_file(
+                        resp.content,
+                        file_obj,
+                        total_size=total_size,
+                        start_time=start_time,
+                        show_progress=show_progress,
+                    )
                 finally:
                     await asyncio.to_thread(file_obj.close)
     except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError):
@@ -176,31 +168,73 @@ async def download_file(url: str, path: str, show_progress: bool = False) -> Non
         async with aiohttp.ClientSession() as session:
             async with session.get(url, ssl=ssl_context, timeout=120) as resp:
                 total_size = int(resp.headers.get("content-length", 0))
-                downloaded_size = 0
                 start_time = time.time()
                 if show_progress:
                     print(f"文件大小: {total_size / 1024:.2f} KB | 文件地址: {url}")
                 file_obj = await asyncio.to_thread(Path(path).open, "wb")
                 try:
-                    while True:
-                        chunk = await resp.content.read(8192)
-                        if not chunk:
-                            break
-                        await asyncio.to_thread(file_obj.write, chunk)
-                        downloaded_size += len(chunk)
-                        if show_progress:
-                            elapsed_time = time.time() - start_time
-                            speed = downloaded_size / 1024 / elapsed_time  # KB/s
-                            print(
-                                f"\r下载进度: {downloaded_size / total_size:.2%} 速度: {speed:.2f} KB/s",
-                                end="",
-                            )
+                    await _stream_to_file(
+                        resp.content,
+                        file_obj,
+                        total_size=total_size,
+                        start_time=start_time,
+                        show_progress=show_progress,
+                    )
                 finally:
                     await asyncio.to_thread(file_obj.close)
     if show_progress:
         print()
 
 
+async def _stream_to_file(
+    stream: aiohttp.StreamReader,
+    file_obj: BinaryIO,
+    *,
+    total_size: int,
+    start_time: float,
+    show_progress: bool,
+) -> None:
+    """Stream HTTP response into file with buffered thread-offloaded writes."""
+    downloaded_size = 0
+    known_total = total_size if total_size > 0 else None
+    buffered = bytearray()
+
+    try:
+        while True:
+            chunk = await stream.read(_DOWNLOAD_READ_CHUNK_SIZE)
+            if not chunk:
+                break
+
+            buffered.extend(chunk)
+            downloaded_size += len(chunk)
+
+            if len(buffered) >= _DOWNLOAD_FLUSH_THRESHOLD:
+                await asyncio.to_thread(file_obj.write, bytes(buffered))
+                buffered.clear()
+
+            if show_progress:
+                _print_download_progress(downloaded_size, known_total, start_time)
+    finally:
+        if buffered:
+            # Ensure buffered data is flushed even on cancellation.
+            await asyncio.shield(asyncio.to_thread(file_obj.write, bytes(buffered)))
+
+
+def _print_download_progress(
+    downloaded_size: int, total_size: int | None, start_time: float
+) -> None:
+    elapsed_time = max(time.time() - start_time, 1e-6)
+    speed = downloaded_size / 1024 / elapsed_time  # KB/s
+
+    if total_size:
+        percent = downloaded_size / total_size
+        msg = f"\r下载进度: {percent:.2%} 速度: {speed:.2f} KB/s"
+    else:
+        msg = f"\r已下载: {downloaded_size} 字节 速度: {speed:.2f} KB/s"
+
+    print(msg, end="")
+
+
 async def file_to_base64(file_path: str) -> str:
     data_bytes = await asyncio.to_thread(Path(file_path).read_bytes)
     base64_str = base64.b64encode(data_bytes).decode()

tests/fixtures/helpers.py

@@ -7,6 +7,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Callable
+from urllib.parse import urlparse
 from unittest.mock import AsyncMock, MagicMock
 
 from astrbot.core.message.components import BaseMessageComponent
@@ -24,6 +25,25 @@ def __await__(self):
         return None
 
 
+def get_bound_tcp_port(site: Any) -> int:
+    """Resolve the bound aiohttp TCP site port for tests.
+
+    We prefer the public ``site.name`` first. Some aiohttp test setups with
+    ephemeral ports may not expose a usable port there, so we fall back to
+    ``site._server.sockets`` as a test-only compatibility path.
+    """
+    parsed = urlparse(getattr(site, "name", ""))
+    if parsed.port is not None and parsed.port > 0:
+        return parsed.port
+
+    server = getattr(site, "_server", None)
+    sockets = getattr(server, "sockets", None) if server else None
+    if sockets:
+        return sockets[0].getsockname()[1]
+
+    raise RuntimeError("Unable to resolve bound TCP port from aiohttp site")
+
+
 # ============================================================
 # 平台配置工厂
 # ============================================================