more msd async

kvmd/inotify.py

@@ -34,6 +34,7 @@
 
 from .logging import get_logger
 
+from . import aiotools
 from . import libc
 
 
@@ -189,11 +190,12 @@ def __init__(self) -> None:
 
         self.__events_queue: "asyncio.Queue[InotifyEvent]" = asyncio.Queue()
 
-    def watch(self, path: str, mask: int) -> None:
+    async def watch(self, path: str, mask: int) -> None:
         path = os.path.normpath(path)
         assert path not in self.__wd_by_path, path
         get_logger().info("Watching for %s", path)
-        wd = _inotify_check(libc.inotify_add_watch(self.__fd, _fs_encode(path), mask))
+        # Асинхронно, чтобы не висло на NFS
+        wd = _inotify_check(await aiotools.run_async(libc.inotify_add_watch, self.__fd, _fs_encode(path), mask))
         self.__wd_by_path[path] = wd
         self.__path_by_wd[wd] = path
 

kvmd/plugins/msd/otg/__init__.py

@@ -431,10 +431,10 @@ async def __watch_inotify(self) -> None:
 
                 with Inotify() as inotify:
                     for path in [
-                        *self.__storage.get_watchable_paths(),
+                        *(await self.__storage.get_watchable_paths()),
                         *self.__drive.get_watchable_paths(),
                     ]:
-                        inotify.watch(path, InotifyMask.ALL_MODIFY_EVENTS)
+                        await inotify.watch(path, InotifyMask.ALL_MODIFY_EVENTS)
 
                     # После установки вотчеров еще раз проверяем стейт, чтобы ничего не потерять
                     await self.__reload_state()
@@ -471,7 +471,7 @@ async def __reload_state(self, notify: bool=True) -> None:
                     await self.__storage.remount_rw(False)
                     await self.__setup_initial()
 
-                storage_state = self.__get_storage_state()
+                storage_state = await self.__get_storage_state()
 
             except Exception:
                 logger.exception("Error while reloading MSD state; switching to offline")
@@ -514,8 +514,8 @@ async def __setup_initial(self) -> None:
 
     # =====
 
-    def __get_storage_state(self) -> _StorageState:
-        images = self.__storage.get_images()
+    async def __get_storage_state(self) -> _StorageState:
+        images = await self.__storage.get_images()
         space = self.__storage.get_space(fatal=True)
         assert space
         return _StorageState(

kvmd/plugins/msd/otg/storage.py

@@ -21,6 +21,7 @@
 
 
 import os
+import operator
 import dataclasses
 
 from typing import Generator
@@ -31,6 +32,7 @@
 
 from ....logging import get_logger
 
+from .... import aiotools
 from .... import aiohelpers
 
 from .. import MsdError
@@ -130,26 +132,42 @@ def __init__(self, path: str, remount_cmd: list[str]) -> None:
         self.__path = path
         self.__remount_cmd = remount_cmd
 
-    def get_watchable_paths(self) -> list[str]:
-        paths: list[str] = []
-        for (root_path, dirs, _) in os.walk(self.__path):
-            dirs[:] = list(self.__filtered(dirs))
-            paths.append(root_path)
-        return paths
+    async def get_watchable_paths(self) -> list[str]:
+        return (await aiotools.run_async(self.__get_watchable_paths))
 
-    def get_images(self) -> dict[str, Image]:
+    async def get_images(self) -> dict[str, Image]:
+        return (await aiotools.run_async(self.__get_images))
+
+    def __get_watchable_paths(self) -> list[str]:
+        return list(map(operator.itemgetter(0), self.__walk(with_files=False)))
+
+    def __get_images(self) -> dict[str, Image]:
         images: dict[str, Image] = {}
-        for (root_path, dirs, files) in os.walk(self.__path):
-            dirs[:] = list(self.__filtered(dirs))
-            for file in self.__filtered(files):
-                name = os.path.relpath(os.path.join(root_path, file), self.__path)
+        for (_, files) in self.__walk(with_files=True):
+            for path in files:
+                name = os.path.relpath(path, self.__path)
                 images[name] = self.get_image_by_name(name)
         return images
 
-    def __filtered(self, items: list[str]) -> Generator[str, None, None]:
-        for item in sorted(map(str.strip, items)):
-            if not item.startswith(".") and item != "lost+found":
-                yield item
+    def __walk(self, with_files: bool, root_path: (str | None)=None) -> Generator[tuple[str, list[str]], None, None]:
+        if root_path is None:
+            root_path = self.__path
+        files: list[str] = []
+        with os.scandir(root_path) as dir_iter:
+            for item in sorted(dir_iter, key=operator.attrgetter("name")):
+                if item.name.startswith(".") or item.name == "lost+found":
+                    continue
+                try:
+                    if item.is_dir(follow_symlinks=False):
+                        item.stat()  # Проверяем, не сдохла ли смонтированная NFS
+                        yield from self.__walk(with_files, item.path)
+                    elif with_files and item.is_file(follow_symlinks=False):
+                        files.append(item.path)
+                except Exception:
+                    pass
+        yield (root_path, files)
+
+    # =====
 
     def get_image_by_name(self, name: str) -> Image:
         assert name
@@ -170,6 +188,8 @@ def __get_image(self, name: str, path: str, in_storage: bool) -> Image:
         assert path
         return Image(name, path, (self if in_storage else None))
 
+    # =====
+
     def get_space(self, fatal: bool) -> (StorageSpace | None):
         try:
             st = os.statvfs(self.__path)

kvmd/plugins/ugpio/otgconf.py

@@ -82,8 +82,8 @@ async def run(self) -> None:
                     await asyncio.sleep(5)
 
                 with Inotify() as inotify:
-                    inotify.watch(os.path.dirname(self.__udc_path), InotifyMask.ALL_MODIFY_EVENTS)
-                    inotify.watch(self.__profile_path, InotifyMask.ALL_MODIFY_EVENTS)
+                    await inotify.watch(os.path.dirname(self.__udc_path), InotifyMask.ALL_MODIFY_EVENTS)
+                    await inotify.watch(self.__profile_path, InotifyMask.ALL_MODIFY_EVENTS)
                     self._notifier.notify()
                     while True:
                         need_restart = False