diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer/indexer.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 122 |
1 files changed, 97 insertions, 25 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index db1b11c..c1b151a 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -26,7 +26,7 @@ import asyncio import logging import time from concurrent.futures import ThreadPoolExecutor -from dataclasses import dataclass +from dataclasses import dataclass, field as dataclass_field from pathlib import Path from typing import Callable, Awaitable @@ -141,6 +141,20 @@ class IndexProgress: # Basename only, deliberately not the full path — enough to show progress # without broadcasting the operator's directory structure. current_dir: str = "" + # What the operator's progress band names. `root` and `queued` are root + # names, so they stay on the loopback API like `current_dir`; members get + # `root_pos`, a position in the roots table they already opened from the + # sealed index, and the length of the queue. + root: str = "" + root_pos: int = -1 + # "scan" (a root walked for the first time), "rescan" (one that came back), + # "reconcile" (files the watcher missed), "watch" (a burst of events), or + # "" when idle. + kind: str = "" + files_done: int = 0 + files_total: int = 0 + # Roots waiting for the scan lock, in the order they will be walked. + queued: list[str] = dataclass_field(default_factory=list) def _virtual_dir(root: Root, file_path: Path) -> str: @@ -323,6 +337,8 @@ class DirectoryIndexer: self._job_running = False self._burst_scanned = 0 self._burst_total = 0 + self._burst_files_done = 0 + self._burst_files_total = 0 self._burst_dir = "" # Ids whose entry this indexer threw away and rebuilt from disk, since # the last time a consumer drained this. A rebuilt entry carries only @@ -373,20 +389,27 @@ class DirectoryIndexer: async def _initial_scan(self) -> None: self.roots.refresh_availability() total = 0 - for root in self.roots: - if not root.available: - log.warning("Root %r is not readable at startup (%s) — its files " - "are not indexed yet and will appear when it returns", - root.name, root.path) - continue - total += await self._scan_root(root) + waiting = [r.name for r in self.roots if r.available] + self._queue(waiting) + try: + for root in self.roots: + if not root.available: + log.warning("Root %r is not readable at startup (%s) — its files " + "are not indexed yet and will appear when it returns", + root.name, root.path) + continue + waiting.remove(root.name) + self._unqueue([root.name]) + total += await self._scan_root(root) + finally: + self._unqueue(waiting) self._index.version = int(time.time()) self._index.roots = self.roots.describe() self._report_collisions() log.info("Initial scan complete: %d files across %d root(s)", total, len(self.roots)) - async def _scan_root(self, root: Root) -> int: + async def _scan_root(self, root: Root, *, kind: str = "scan") -> int: log.info("Scanning %s (root %r) ...", root.path, root.name) count = 0 loop = asyncio.get_event_loop() @@ -402,7 +425,7 @@ class DirectoryIndexer: # had not yet turned on, even though the node was already several # seconds into walking and sizing a large root — total_bytes is # unknown at this point, so it starts at 0 and is corrected below. - self._begin_job(root.name) + self._begin_job(root.name, root=root, kind=kind) try: try: files = await loop.run_in_executor(self._executor, _walk_root, root) @@ -418,11 +441,13 @@ class DirectoryIndexer: sized = await loop.run_in_executor(self._executor, _size_files, files) self.progress.total_bytes = sum(size for _, size in sized) + self.progress.files_total = len(sized) self.progress.current_dir = "" for file_path, size in sized: self.progress.current_dir = file_path.parent.name entry = await self._hash_or_cached(root, file_path) self.progress.scanned_bytes += size + self.progress.files_done += 1 if entry: self._index.add_entry(entry) count += 1 @@ -433,18 +458,30 @@ class DirectoryIndexer: self._end_job() return count - def _begin_job(self, current_dir: str, total_bytes: int = 0) -> None: + def _begin_job(self, current_dir: str, total_bytes: int = 0, *, + root: Root, kind: str, files_total: int = 0) -> None: """`progress` now describes a whole-root walk, whatever a burst is doing.""" self._job_running = True - self.progress.scanning = True - self.progress.scanned_bytes = 0 - self.progress.total_bytes = total_bytes - self.progress.current_dir = current_dir + p = self.progress + p.scanning = True + p.scanned_bytes = 0 + p.total_bytes = total_bytes + p.files_done = 0 + p.files_total = files_total + p.current_dir = current_dir + p.kind = kind + p.root = root.name + p.root_pos = next((i for i, r in enumerate(self.roots) + if r.folded == root.folded), -1) def _end_job(self) -> None: self._job_running = False - self.progress.scanning = False - self.progress.current_dir = "" + p = self.progress + p.scanning = False + p.current_dir = "" + p.kind = "" + p.root = "" + p.root_pos = -1 if self._burst_inflight > 0: # A burst that started during the walk is still hashing. self._show_burst() @@ -456,11 +493,25 @@ class DirectoryIndexer: if self._burst_inflight > 0: p.scanning = True p.current_dir = self._burst_dir + p.kind = "watch" + p.root = "" + p.root_pos = -1 elif not self._pending_timers: p.scanning = False p.current_dir = "" + p.kind = "" p.scanned_bytes = self._burst_scanned p.total_bytes = self._burst_total + p.files_done = self._burst_files_done + p.files_total = self._burst_files_total + + def _queue(self, names: list[str]) -> None: + self.progress.queued.extend(names) + + def _unqueue(self, names: list[str]) -> None: + for name in names: + if name in self.progress.queued: + self.progress.queued.remove(name) async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: """ @@ -633,6 +684,7 @@ class DirectoryIndexer: """ old_names = {r.folded for r in self.roots} new_names = {r.folded for r in roots} + self.progress.queued[:] = [n for n in self.progress.queued if fold(n) in new_names] for root in self.roots: if root.folded not in new_names: @@ -655,6 +707,7 @@ class DirectoryIndexer: await self.on_change(self) return + self._queue([r.name for r in added]) task = asyncio.create_task(self._scan_added_roots(added)) self._scan_tasks.add(task) task.add_done_callback(self._scan_tasks.discard) @@ -668,9 +721,12 @@ class DirectoryIndexer: return any(r.folded == root.folded and r.path == root.path for r in self.roots) async def _scan_added_roots(self, added: list[Root]) -> None: + waiting = [r.name for r in added] try: async with self._scan_lock: for root in added: + waiting.remove(root.name) + self._unqueue([root.name]) # A later retarget may have removed it while this waited. if not self._holds(root): continue @@ -689,6 +745,8 @@ class DirectoryIndexer: raise except Exception: log.exception("Scanning the added root(s) failed") + finally: + self._unqueue(waiting) # ── Reconciliation ──────────────────────────────────────────────────────── @@ -818,12 +876,14 @@ class DirectoryIndexer: except OSError: added_sized.append((p, 0)) - self._begin_job("", sum(size for _, size in added_sized)) + self._begin_job("", sum(size for _, size in added_sized), root=root, + kind="reconcile", files_total=len(added_sized)) try: for added, size in added_sized: self.progress.current_dir = added.parent.name entry = await self._hash_or_cached(root, added) self.progress.scanned_bytes += size + self.progress.files_done += 1 if not entry: continue # The index is keyed by **content hash**, so two identical @@ -883,7 +943,7 @@ class DirectoryIndexer: """ carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)} self._drop_root_entries(root) - count = await self._scan_root(root) + count = await self._scan_root(root, kind="rescan") for entry in self._entries_under(root): old = carried.get((entry.id, entry.name, entry.path)) if old is None: @@ -1002,12 +1062,20 @@ class DirectoryIndexer: # A whole-root walk like any other, so it waits its turn: run beside # an added root's scan, the two reset each other's progress and read # the same drive in alternation. - async with self._scan_lock: - # Ejected or removed again while it waited. `_rescan_root` drops - # the entries before it walks, so going ahead would empty a root - # that is not there to be read. - if self._holds(root) and not root.ejected and root.is_live(): - await self._rescan_root(root) + self._queue([root.name]) + waiting = True + try: + async with self._scan_lock: + self._unqueue([root.name]) + waiting = False + # Ejected or removed again while it waited. `_rescan_root` + # drops the entries before it walks, so going ahead would + # empty a root that is not there to be read. + if self._holds(root) and not root.ejected and root.is_live(): + await self._rescan_root(root) + finally: + if waiting: + self._unqueue([root.name]) self._restart_observer() self._index.roots = self.roots.describe() self._index.version = int(time.time()) @@ -1044,7 +1112,10 @@ class DirectoryIndexer: if self._burst_inflight <= 0: self._burst_scanned = 0 self._burst_total = 0 + self._burst_files_done = 0 + self._burst_files_total = 0 self._burst_total += size + self._burst_files_total += 1 self._burst_dir = file_path.parent.name self._burst_sizes[key] = size self._burst_inflight += 1 @@ -1118,6 +1189,7 @@ class DirectoryIndexer: size = self._burst_sizes.pop(str(file_path), None) if size is not None: self._burst_scanned += size + self._burst_files_done += 1 self._burst_inflight -= 1 self._show_burst() |