From 032ad13b97869778b082869b28314bd41a255eea Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sun, 1 Feb 2026 02:44:52 +0000 Subject: [PATCH] Unified watcher that *may* receive events from inotify and other sources. Added change notify messages from control and upload WebSockets. Cleanup debug printouts. --- cista/__main__.py | 2 +- cista/api.py | 6 + cista/protocol.py | 28 +++++ cista/watching.py | 287 ++++++++++++++++++++++------------------------ 4 files changed, 170 insertions(+), 153 deletions(-) diff --git a/cista/__main__.py b/cista/__main__.py index 4f43f07..c196cc1 100644 --- a/cista/__main__.py +++ b/cista/__main__.py @@ -141,7 +141,7 @@ def _main(): settings["listen"] = listen elif not exists: settings["listen"] = ":8000" - operation = config.update_config(settings) + config.update_config(settings) # Prepare to serve url, opts = serve.parse_listen(config.config.listen) if not config.config.path.is_dir(): diff --git a/cista/api.py b/cista/api.py index 3f7fce0..105db8f 100644 --- a/cista/api.py +++ b/cista/api.py @@ -1,5 +1,6 @@ import asyncio import typing +from pathlib import PurePosixPath from secrets import token_bytes import msgspec @@ -53,6 +54,9 @@ async def upload(req, ws): if pos != req.end: d = f"{len(data)} bytes" if isinstance(data, bytes) else data raise ValueError(f"Expected {req.end - pos} more bytes, got {d}") + # Signal the watcher about the uploaded file and its parent directories + path = PurePosixPath(req.name) + watching.notify_change(path, *path.parents) # Report success res = StatusMsg(status="ack", req=req) await asend(ws, res) @@ -87,6 +91,8 @@ async def control(req, ws): while True: cmd = msgspec.json.decode(await ws.recv(), type=ControlTypes) await asyncio.to_thread(cmd) + # Signal the watcher about affected paths + watching.notify_change(*cmd.affected_paths()) await asend(ws, StatusMsg(status="ack", req=cmd)) diff --git a/cista/protocol.py b/cista/protocol.py index c47e729..7ebd221 100644 --- a/cista/protocol.py +++ b/cista/protocol.py @@ -1,6 +1,7 @@ from __future__ import annotations import shutil +from pathlib import PurePosixPath from typing import Any import msgspec @@ -16,6 +17,10 @@ class ControlBase(msgspec.Struct, tag_field="op", tag=str.lower): def __call__(self): raise NotImplementedError + def affected_paths(self) -> list[str]: + """Return list of paths affected by this operation for change notification.""" + return [] + class MkDir(ControlBase): path: str @@ -24,6 +29,9 @@ class MkDir(ControlBase): path = config.config.path / filename.sanitize(self.path) path.mkdir(parents=True, exist_ok=False) + def affected_paths(self) -> list[str]: + return [filename.sanitize(self.path)] + class Rename(ControlBase): path: str @@ -36,6 +44,11 @@ class Rename(ControlBase): path = config.config.path / filename.sanitize(self.path) path.rename(path.with_name(to)) + def affected_paths(self) -> list[str]: + sanitized = filename.sanitize(self.path) + new_path = str(PurePosixPath(sanitized).with_name(filename.sanitize(self.to))) + return [sanitized, new_path] + class Rm(ControlBase): sel: list[str] @@ -49,6 +62,9 @@ class Rm(ControlBase): else: p.unlink() + def affected_paths(self) -> list[str]: + return [filename.sanitize(p) for p in self.sel] + class Mv(ControlBase): sel: list[str] @@ -63,6 +79,13 @@ class Mv(ControlBase): for p in sel: shutil.move(p, dst) + def affected_paths(self) -> list[str]: + dst = filename.sanitize(self.dst) + paths = [filename.sanitize(p) for p in self.sel] + # Include new locations in dst + paths.extend(f"{dst}/{PurePosixPath(p).name}" for p in self.sel) + return paths + class Cp(ControlBase): sel: list[str] @@ -86,6 +109,11 @@ class Cp(ControlBase): else: shutil.copy2(p, dst) + def affected_paths(self) -> list[str]: + dst = filename.sanitize(self.dst) + # Only destinations are new (sources unchanged) + return [f"{dst}/{PurePosixPath(filename.sanitize(p)).name}" for p in self.sel] + ControlTypes = MkDir | Rename | Rm | Mv | Cp diff --git a/cista/watching.py b/cista/watching.py index 7327f92..9bdef50 100644 --- a/cista/watching.py +++ b/cista/watching.py @@ -1,4 +1,5 @@ import asyncio +import queue import shutil import sys import threading @@ -9,13 +10,6 @@ from pathlib import Path, PurePosixPath from stat import S_ISDIR, S_ISREG import msgspec - - -# Debug instrumentation -def _dbg(msg): - print(f"[watch] {time.perf_counter():.3f} {msg}", file=sys.stderr, flush=True) - - from natsort import humansorted, natsort_keygen, ns from sanic.log import logger @@ -53,7 +47,6 @@ def treeiter(rootmod): def treeget(rootmod: list[FileEntry], path: PurePosixPath): - t0 = time.perf_counter() begin = None ret = [] @@ -67,24 +60,16 @@ def treeget(rootmod: list[FileEntry], path: PurePosixPath): break ret.append(entry) - dur = time.perf_counter() - t0 - if dur > 0.01: - _dbg( - f"treeget({path}) scanned {len(rootmod)} entries in {dur * 1000:.1f}ms, found {len(ret)} items" - ) return begin, ret def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int): # Find the first entry greater than the new one # precondition: the new entry doesn't exist - t0 = time.perf_counter() isfile = 0 level = 0 i = 0 - iter_count = 0 for i, rel, entry in treeiter(rootmod): - iter_count += 1 if entry.level > level: # We haven't found item at level, skip subdirectories continue @@ -125,11 +110,6 @@ def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int): else: i += 1 - dur = time.perf_counter() - t0 - if dur > 0.01: - _dbg( - f"treeinspos({relpath}) iterated {iter_count}/{len(rootmod)} entries in {dur * 1000:.1f}ms -> pos {i}" - ) return i @@ -137,11 +117,32 @@ state = State() rootpath: Path = None # type: ignore quit = threading.Event() +# Thread-safe queue for signaling path updates from websockets +_update_queue: queue.Queue[PurePosixPath] = queue.Queue() + + +def notify_change(*paths: PurePosixPath | str): + """Signal that paths have changed. Called from control/upload websockets.""" + for path in paths: + if isinstance(path, str): + path = PurePosixPath(path) + # Convert absolute paths to relative (strip leading /) + if path.is_absolute(): + path = ( + PurePosixPath(*path.parts[1:]) + if len(path.parts) > 1 + else PurePosixPath() + ) + # Skip root paths (empty, '.') to avoid full tree walks + if not path.parts or path.parts == (".",): + continue + _update_queue.put(path) + + ## Filesystem scanning def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]: - t0 = time.perf_counter() path = rootpath / rel ret = [] try: @@ -193,62 +194,34 @@ def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry] logger.error(f"Watching {path=}: {e!r}") if ret: ret[0] = entry - dur = time.perf_counter() - t0 - if dur > 0.05: - _dbg(f"walk({rel}) returned {len(ret)} entries in {dur * 1000:.1f}ms") return ret def update_root(loop): """Full filesystem scan""" - t0 = time.perf_counter() old = state.root new = walk(PurePosixPath()) - t_walk = time.perf_counter() if old != new: update = format_update(old, new) - t_format = time.perf_counter() with state.lock: broadcast(update, loop) state.root = new - t_done = time.perf_counter() - _dbg( - f"update_root: walk={t_walk - t0:.3f}s format={t_format - t_walk:.3f}s broadcast={t_done - t_format:.3f}s total={t_done - t0:.3f}s ({len(new)} entries)" - ) def update_path(rootmod: list[FileEntry], relpath: PurePosixPath, loop): """Called on FS updates, check the filesystem and broadcast any changes.""" - t0 = time.perf_counter() new = walk(relpath) - t_walk = time.perf_counter() obegin, old = treeget(rootmod, relpath) - t_get = time.perf_counter() if old == new: - dur = time.perf_counter() - t0 - if dur > 0.01: - _dbg( - f"update_path({relpath}) no change, walk={t_walk - t0:.3f}s get={t_get - t_walk:.3f}s" - ) return if obegin is not None: del rootmod[obegin : obegin + len(old)] - t_del = time.perf_counter() if new: i = treeinspos(rootmod, relpath, new[0].isfile) - t_inspos = time.perf_counter() rootmod[i:i] = new - t_ins = time.perf_counter() - _dbg( - f"update_path({relpath}) walk={t_walk - t0:.3f}s get={t_get - t_walk:.3f}s del={t_del - t_get:.3f}s inspos={t_inspos - t_del:.3f}s ins={t_ins - t_inspos:.3f}s total={t_ins - t0:.3f}s (old={len(old)} new={len(new)} tree={len(rootmod)})" - ) - else: - _dbg( - f"update_path({relpath}) DELETED walk={t_walk - t0:.3f}s get={t_get - t_walk:.3f}s del={t_del - t_get:.3f}s (old={len(old)})" - ) def update_space(loop): @@ -268,7 +241,6 @@ def update_space(loop): def format_update(old, new): - t0 = time.perf_counter() # Make keep/del/insert diff until one of the lists ends oidx, nidx = 0, 0 oremain, nremain = set(old), set(new) @@ -279,7 +251,6 @@ def format_update(old, new): # candidates exist in both sequences but are not equal (rename/move cases) old_pos = {e: i for i, e in enumerate(old)} new_pos = {e: i for i, e in enumerate(new)} - t_setup = time.perf_counter() while oidx < len(old) and nidx < len(new): iteration_count += 1 @@ -366,13 +337,7 @@ def format_update(old, new): elif nremain: update.append(UpdIns(new[nidx:])) - t_diff = time.perf_counter() - result = msgspec.json.encode({"update": update}).decode() - t_encode = time.perf_counter() - _dbg( - f"format_update: setup={t_setup - t0:.3f}s diff={t_diff - t_setup:.3f}s ({iteration_count} iters) encode={t_encode - t_diff:.3f}s total={t_encode - t0:.3f}s (old={len(old)} new={len(new)} ops={len(update)} msg={len(result)}B)" - ) - return result + return msgspec.json.encode({"update": update}).decode() def format_space(usage): @@ -384,20 +349,11 @@ def format_root(root): def broadcast(msg, loop): - t0 = time.perf_counter() fut = asyncio.run_coroutine_threadsafe(abroadcast(msg), loop) - t_scheduled = time.perf_counter() - result = fut.result() - t_done = time.perf_counter() - if t_done - t0 > 0.01: - _dbg( - f"broadcast: schedule={t_scheduled - t0:.3f}s wait={t_done - t_scheduled:.3f}s total={t_done - t0:.3f}s ({len(msg)}B to {result} clients)" - ) - return result + return fut.result() async def abroadcast(msg): - t0 = time.perf_counter() client_count = 0 try: for queue in pubsub.values(): @@ -406,9 +362,6 @@ async def abroadcast(msg): except Exception: # Log because asyncio would silently eat the error logger.exception("Broadcast error") - dur = time.perf_counter() - t0 - if dur > 0.001: - _dbg(f"abroadcast: {client_count} clients in {dur * 1000:.2f}ms") return client_count @@ -558,6 +511,10 @@ def collapse_paths(paths: set[PurePosixPath]) -> set[PurePosixPath]: """Remove child paths if parent is in set.""" if not paths: return paths + # Filter out root paths (empty or '.') which would cause full tree walks + paths = {p for p in paths if p.parts and p.parts != (".",)} + if not paths: + return set() # Sort by depth (fewest parts first) sorted_paths = sorted(paths, key=lambda p: len(p.parts)) result = set() @@ -574,28 +531,37 @@ def collapse_paths(paths: set[PurePosixPath]) -> set[PurePosixPath]: return result -def watcher_inotify(loop): - """Inotify watcher thread (Linux only)""" - import inotify.adapters +# Debounce settings +DEBOUNCE_DELAY = 0.01 # Wait 10ms after last event +DEBOUNCE_MAX = 0.1 # But no more than 100ms total - modified_flags = frozenset( - ( - "IN_CREATE", - "IN_DELETE", - "IN_DELETE_SELF", - "IN_MODIFY", - "IN_MOVE_SELF", - "IN_MOVED_FROM", - "IN_MOVED_TO", + +def watcher(loop): + """Unified watcher thread handling inotify, websocket signals, and periodic scans.""" + use_inotify = sys.platform == "linux" + inotify_tree = None + modified_flags = frozenset() + + if use_inotify: + import inotify.adapters + + modified_flags = frozenset( + ( + "IN_CREATE", + "IN_DELETE", + "IN_DELETE_SELF", + "IN_MODIFY", + "IN_MOVE_SELF", + "IN_MOVED_FROM", + "IN_MOVED_TO", + ) ) - ) - - # Debounce settings - DEBOUNCE_DELAY = 0.1 # Wait 100ms after last event - DEBOUNCE_MAX = 0.5 # But no more than 500ms total while not quit.is_set(): - inotify_tree = inotify.adapters.InotifyTree(rootpath.as_posix()) + if use_inotify: + import inotify.adapters + + inotify_tree = inotify.adapters.InotifyTree(rootpath.as_posix()) # Initialize the tree from filesystem update_root(loop) @@ -604,11 +570,44 @@ def watcher_inotify(loop): trefresh = time.monotonic() + 300.0 tspace = time.monotonic() + 5.0 - # Pending changes - dirty_paths: set[PurePosixPath] = set() + # Pending changes: path -> {"ws": count, "inotify": count} + dirty_paths: dict[PurePosixPath, dict[str, int]] = {} first_event_time: float | None = None last_event_time: float | None = None - event_count = 0 + + def add_dirty(path: PurePosixPath, source: str) -> bool: + """Add path to dirty set. Returns True if added, False if redundant.""" + nonlocal first_event_time, last_event_time + # Check if already covered by an existing dirty path + for existing in dirty_paths: + if path == existing or ( + len(path.parts) > len(existing.parts) + and path.parts[: len(existing.parts)] == existing.parts + ): + # Count the event even if skipped + dirty_paths[existing][source] = ( + dirty_paths[existing].get(source, 0) + 1 + ) + return False + # Remove any paths that would be covered by this new one + covered = { + p + for p in dirty_paths + if len(p.parts) > len(path.parts) + and p.parts[: len(path.parts)] == path.parts + } + # Aggregate counts from covered paths + counts: dict[str, int] = {source: 1} + for p in covered: + for s, c in dirty_paths[p].items(): + counts[s] = counts.get(s, 0) + c + del dirty_paths[p] + dirty_paths[path] = counts + now = time.monotonic() + if first_event_time is None: + first_event_time = now + last_event_time = now + return True while not quit.is_set(): now = time.monotonic() @@ -627,19 +626,20 @@ def watcher_inotify(loop): if dirty_paths: time_since_last = now - last_event_time if last_event_time else 0 time_since_first = now - first_event_time if first_event_time else 0 - if time_since_last >= DEBOUNCE_DELAY or time_since_first >= DEBOUNCE_MAX: + if ( + time_since_last >= DEBOUNCE_DELAY + or time_since_first >= DEBOUNCE_MAX + ): should_flush = True if should_flush: - t_start = time.perf_counter() + paths_to_process = dirty_paths.copy() + dirty_paths.clear() + first_event_time = None + last_event_time = None # Collapse paths (remove children if parent present) - collapsed = collapse_paths(dirty_paths) - t_collapse = time.perf_counter() - - _dbg( - f"flush: {event_count} events -> {len(dirty_paths)} paths -> {len(collapsed)} collapsed" - ) + collapsed = collapse_paths(set(paths_to_process.keys())) # Process each collapsed path new_root = path_index.root @@ -647,20 +647,13 @@ def watcher_inotify(loop): new_entries = walk(path) new_root = path_index.apply_update(path, new_entries) - t_update = time.perf_counter() - # Broadcast if changed if new_root != state.root: try: update_msg = format_update(state.root, new_root) - t_format = time.perf_counter() with state.lock: broadcast(update_msg, loop) state.root = new_root - t_broadcast = time.perf_counter() - _dbg( - f"inotify->broadcast: collapse={t_collapse - t_start:.3f}s update={t_update - t_collapse:.3f}s format={t_format - t_update:.3f}s broadcast={t_broadcast - t_format:.3f}s TOTAL={t_broadcast - t_start:.3f}s" - ) except Exception: logger.exception("format_update failed; full rescan") try: @@ -676,63 +669,53 @@ def watcher_inotify(loop): broadcast(format_root(fresh), loop) state.root = fresh - # Reset pending state - dirty_paths.clear() - first_event_time = None - last_event_time = None - event_count = 0 + # Collect events from websocket signals (non-blocking) + try: + while True: + path = _update_queue.get_nowait() + add_dirty(path, "ws") + except queue.Empty: + pass - # Collect inotify events (short timeout for responsiveness) - for event in inotify_tree.event_gen(yield_nones=False, timeout_s=0.05): - if quit.is_set(): - return - if not (modified_flags & set(event[1])): - continue + # Collect inotify events if available (short timeout for responsiveness) + if inotify_tree: + for event in inotify_tree.event_gen(yield_nones=False, timeout_s=0.05): + if quit.is_set(): + return + if not (modified_flags & set(event[1])): + continue - # Extract relative path - path = PurePosixPath(event[2]) / event[3] - try: - rel_path = path.relative_to(rootpath) - except ValueError: - continue + # Extract relative path + path = PurePosixPath(event[2]) / event[3] + try: + rel_path = path.relative_to(rootpath) + except ValueError: + continue - # Skip dotfiles - if any(part.startswith(".") for part in rel_path.parts): - continue + # Skip dotfiles + if any(part.startswith(".") for part in rel_path.parts): + continue - dirty_paths.add(rel_path) - event_count += 1 - now = time.monotonic() - if first_event_time is None: - first_event_time = now - last_event_time = now + add_dirty(rel_path, "inotify") - # Don't block too long collecting events - if now - first_event_time >= DEBOUNCE_MAX: - break + # Don't block too long collecting events + now = time.monotonic() + if first_event_time and now - first_event_time >= DEBOUNCE_MAX: + break + else: + # No inotify, just sleep briefly for responsiveness + time.sleep(0.05) - del inotify_tree - - -def watcher_poll(loop): - """Polling version of the watcher thread.""" - while not quit.is_set(): - t0 = time.perf_counter() - update_root(loop) - update_space(loop) - dur = time.perf_counter() - t0 - if dur > 1.0: - logger.debug(f"Reading the full file list took {dur:.1f}s") - quit.wait(0.1 + 8 * dur) + if inotify_tree: + del inotify_tree def start(app): global rootpath config.load_config() rootpath = config.config.path - use_inotify = sys.platform == "linux" app.ctx.watcher = threading.Thread( - target=watcher_inotify if use_inotify else watcher_poll, + target=watcher, args=[app.loop], # Descriptive name for system monitoring name=f"cista-watcher {rootpath}",