perf: persist drive listing worker

Author: NSPC911

src/rovr/app.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
 import asyncio
-import multiprocessing
 import os
 import sys
 import threading
 from contextlib import suppress
 from io import TextIOWrapper
 from os import path
 from subprocess import Popen, TimeoutExpired
+from time import monotonic
 from typing import Callable, Iterable
 
 from rich.console import RenderableType
@@ -63,7 +63,7 @@
     PreviewContainer,
 )
 from rovr.footer import Clipboard, MetadataContainer, ProcessContainer
-from rovr.functions import drive_workers, multiprocessing_utils
+from rovr.functions import drive_workers
 from rovr.functions.cwd import chdir, getcwd
 from rovr.functions.path import (
     dump_exc,
@@ -641,19 +641,19 @@ def cd(
     @work(thread=True)
     def watch_for_changes_and_update(self) -> None:
         cwd = getcwd()
-        file_list: FileList = self.query_one(FileList)
         pins_path = path.join(RovrVars.ROVRCONFIG, "pins.json")
         with suppress(OSError):
             self._pins_mtime = path.getmtime(pins_path)
         state_path = path.join(RovrVars.ROVRSTATE, "state.toml")
-        state_mtime = None
         with suppress(OSError):
             state_mtime = path.getmtime(state_path)
-        drive_update_every = int(config["interface"]["drive_watcher_frequency"])
-        count: int = -2
+        drive_update_every = max(1.0, config["interface"]["drive_watcher_frequency"])
+        next_drive_check = 0.0
+        drive_watcher = drive_workers.DriveWatcher(
+            sys.platform, config["settings"]["drive_exclude"], drive_update_every
+        )
         style_available: bool = self.CUSTOM_STYLE_AVAILABLE
         custom_style_path = path.join(RovrVars.ROVRCONFIG, "style.tcss")
-        new_drives: list[str] | None = None
         cwd_mtime: float | None = None
         pin_sidebar = self.query_one(PinnedSidebar)
 
@@ -663,14 +663,11 @@ def watch_for_changes_and_update(self) -> None:
 
         while True:
             if self._shutdown_event.wait(timeout=1):
-                return
+                break
             if i_should_shut_down():
-                return
+                break
             if (file_list := self.file_list).parent is None or not file_list.is_running:
                 continue
-            count += 1
-            if count >= drive_update_every:
-                count = 0
             try:
                 new_cwd = getcwd()
                 if not self.file_list.file_list_pause_check:
@@ -683,12 +680,10 @@ def watch_for_changes_and_update(self) -> None:
                     else:
                         # only rescan when the directory mtime changed;
                         # renames/creates/deletes always bump it
-                        new_cwd_mtime = None
                         with suppress(OSError):
                             new_cwd_mtime = path.getmtime(cwd)
                         if new_cwd_mtime != cwd_mtime:
                             cwd_mtime = new_cwd_mtime
-                            items = None
                             with suppress(OSError):
                                 items = get_filtered_dir_names(
                                     cwd,
@@ -710,11 +705,9 @@ def watch_for_changes_and_update(self) -> None:
                 )
 
             if i_should_shut_down():
-                return
+                break
 
             # check pins.json
-            new_mtime = None
-            reload_called: bool = False
             with suppress(OSError):
                 new_mtime = path.getmtime(pins_path)
             if new_mtime != self._pins_mtime:
@@ -726,12 +719,10 @@ def watch_for_changes_and_update(self) -> None:
                     # really is no issue here, thanks to any AI
                     # models raising false issues on thread safety
                     pin_sidebar.reload_pins()
-                    reload_called = True
             if i_should_shut_down():
-                return
+                break
 
             # check state.toml
-            new_state_mtime = None
             with suppress(OSError):
                 new_state_mtime = path.getmtime(state_path)
             if new_state_mtime != state_mtime:
@@ -741,53 +732,30 @@ def watch_for_changes_and_update(self) -> None:
                     self.app.call_from_thread(state_manager._load_state)
                     self.app.call_from_thread(state_manager.restore_state)
             if i_should_shut_down():
-                return
+                break
 
             # check drives
-            if count == 0 and not reload_called:
-                try:
-                    if self.MULTIPROCESSING_PROCESS_ALLOWED:
-                        # Run drive check in a separate process using multiprocessing.Process
-                        # Using Queue to get the result back from the process
-                        result_queue: multiprocessing.Queue[list[str]] = (
-                            multiprocessing.Queue()
-                        )
-
-                        process = multiprocessing.Process(
-                            target=drive_workers.get_mounted_drives_worker,
-                            args=(result_queue, sys.platform, config),
-                        )
-                        multiprocessing_utils.start_process(process)
-                        process.join(timeout=2.0)
-
-                        if process.is_alive():
-                            # Timeout - terminate the process
-                            process.terminate()
-                            process.join(timeout=0.5)
-                            if process.is_alive():
-                                process.kill()
-                        elif not result_queue.empty():
-                            # Process completed successfully
-                            new_drives = result_queue.get_nowait()
-                    else:
-                        new_drives = drive_workers.get_mounted_drives(
-                            sys.platform, config
-                        )
-                    if new_drives is not None and new_drives != pin_sidebar.DRIVES:
-                        pin_sidebar.reload_pins()
-                except Exception as exc:
-                    if multiprocessing_process_error_checker(self, exc):
-                        count = -1  # try again immediately on next loop
-                    else:
-                        self.notify(
-                            f"{type(exc).__name__}: {exc}",
-                            title="Drives Watcher",
-                            severity="warning",
-                            markup=False,
-                        )
-                        dump_exc(self, exc)
+            try:
+                new_drives = None
+                if self.MULTIPROCESSING_PROCESS_ALLOWED:
+                    new_drives = drive_watcher.poll()
+                elif monotonic() >= next_drive_check:
+                    drive_watcher.close()
+                    next_drive_check = monotonic() + drive_update_every
+                    new_drives = drive_workers.get_mounted_drives(sys.platform, config)
+                if new_drives is not None and new_drives != pin_sidebar.DRIVES:
+                    pin_sidebar.reload_pins(drives=new_drives)
+            except Exception as exc:
+                if not multiprocessing_process_error_checker(self, exc):
+                    self.notify(
+                        f"{type(exc).__name__}: {exc}",
+                        title="Drives Watcher",
+                        severity="warning",
+                        markup=False,
+                    )
+                    dump_exc(self, exc)
             if i_should_shut_down():
-                return
+                break
 
             # check highlighted file mtime
             if not self.file_list.file_list_pause_check:
@@ -823,7 +791,7 @@ def watch_for_changes_and_update(self) -> None:
                                     dir_entry
                                 )
             if i_should_shut_down():
-                return
+                break
 
             if not self.CUSTOM_STYLE_AVAILABLE:
                 if not style_available and path.exists(custom_style_path):
@@ -835,6 +803,7 @@ def watch_for_changes_and_update(self) -> None:
                     )
                 elif not path.exists(custom_style_path):
                     style_available = False
+        drive_watcher.close()
 
     @work(exclusive=True)
     async def on_resize(self, event: events.Resize) -> None:

src/rovr/core/pinned_sidebar.py

@@ -1,5 +1,3 @@
-import multiprocessing
-import sys
 from os import R_OK, access, path
 from threading import Lock
 from typing import ClassVar, cast
@@ -13,12 +11,9 @@
 
 from rovr.classes.mixins import Action, Actionable, CursorNavigationMixin
 from rovr.classes.textual_options import PinnedSidebarOption
-from rovr.functions import drive_workers as drive_utils
 from rovr.functions import icons as icon_utils
-from rovr.functions import multiprocessing_utils
 from rovr.functions import path as path_utils
 from rovr.functions import pins as pin_utils
-from rovr.functions.utils import multiprocessing_process_error_checker
 from rovr.variables.constants import bindings, config
 
 
@@ -36,11 +31,11 @@ class PinnedSidebar(
     ACTIONS: list[Action] = [Action("focus_search", config["keybinds"]["focus_search"])]
 
     @work(exclusive=True, thread=True)
-    def reload_pins(self) -> None:
+    def reload_pins(self, drives: list[str] | None = None) -> None:
         with self.tlock:
-            self._reload_pins()
+            self._reload_pins(drives)
 
-    def _reload_pins(self) -> None:
+    def _reload_pins(self, drives: list[str] | None = None) -> None:
         """Reload pins shown
 
         Raises:
@@ -161,44 +156,16 @@ def _reload_pins(self) -> None:
         )
         if self.app.return_code is not None:
             return
-        self.app.call_from_thread(self.set_options, self.list_of_options)
-        if prev_highlighted < len(self.list_of_options):
-            self.app.call_from_thread(setattr, self, "highlighted", prev_highlighted)
-            self.refresh_drives(id_list, None)
-        else:
-            self.refresh_drives(id_list, prev_highlighted)
+        self.refresh_drives(id_list, prev_highlighted, drives)
 
     def refresh_drives(
-        self, id_list: list[str], prev_highlighted: int | None = None
+        self,
+        id_list: list[str],
+        prev_highlighted: int | None = None,
+        drives: list[str] | None = None,
     ) -> None:
-        # force refresh
-        try:
-            if self.app.MULTIPROCESSING_PROCESS_ALLOWED:
-                result_queue: multiprocessing.Queue[list[str]] = multiprocessing.Queue()
-                process = multiprocessing.Process(
-                    target=drive_utils.get_mounted_drives_worker,
-                    args=(result_queue, sys.platform, config),
-                )
-                multiprocessing_utils.start_process(process)
-                process.join(timeout=2.0)
-
-                if process.is_alive():
-                    process.terminate()
-                    process.join(timeout=0.5)
-                    if process.is_alive():
-                        process.kill()
-                    return
-
-                if result_queue.empty():
-                    return
-
-                drives = result_queue.get_nowait()
-            else:
-                drives = drive_utils.get_mounted_drives(sys.platform, config)
-        except Exception as exc:
-            if not multiprocessing_process_error_checker(self.app, exc):
-                return
-            drives = drive_utils.get_mounted_drives(sys.platform, config)
+        if drives is None:
+            drives = self.DRIVES
         self.DRIVES = drives
         new_options: list[PinnedSidebarOption] = []
         for drive in drives: