diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/roots.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/roots.py | 63 |
1 files changed, 63 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index 7854e28..67778ca 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -28,8 +28,10 @@ are one directory. from __future__ import annotations +import asyncio import logging import re +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path @@ -171,6 +173,49 @@ class RootSet: # like a deletion all over again on the pass after it. auto_ejected: list[str] = field(default_factory=list) + # The thread every blocking filesystem call on these roots is made from. + # + # **Why it exists at all.** A root that has spun down, or that lives on a + # network mount, answers its first syscall in seconds rather than + # microseconds. Made from the event loop that is the whole node: no other + # group is served, no stream is fed, no chat message is delivered and the + # hub socket is not read, for as long as the platter takes to come back. + # A client's connection attempt times out and has to be made again, which + # is what this was found by. It is `AV9`'s lesson with the disk in the + # place of the mail server. + # + # **Why one worker and not a pool.** The same reason the indexer's executor + # has one: two interleaved reads on a spinning drive seek-thrash against + # each other rather than go faster, measured there on a USB disk. One + # worker also keeps every read of these roots in the order it was asked + # for, which costs nothing — the protocol addresses a chunk by index, so + # no caller depends on that order — and leaves no way for two threads to + # be inside the same file at once. + # + # **Why per root set and not one for the node.** A node serves several + # groups, and their roots are not all on the same volume. A single worker + # would put the sleeping USB drive of one group in front of the SSD of + # another, which is the symptom this removes, one level down. + # + # Created on first use, so a RootSet that never reads anything — most of + # them, in tests — never starts a thread. Not compared and not printed: it + # is machinery, not part of what a root set *is*, and `daemon.py` compares + # root sets to decide whether a reload changed anything. + _io: ThreadPoolExecutor | None = field( + default=None, init=False, repr=False, compare=False) + + @property + def io_executor(self) -> ThreadPoolExecutor: + if self._io is None: + self._io = ThreadPoolExecutor(max_workers=1, thread_name_prefix="rootio") + return self._io + + def close_io(self) -> None: + """Stop the disk thread. Safe to call twice, and on a set that never read.""" + if self._io is not None: + self._io.shutdown(wait=False) + self._io = None + # ── Construction ───────────────────────────────────────────────────────── @classmethod @@ -401,6 +446,24 @@ def entry_abs_path(roots: RootSet, entry) -> Path | None: return (parent / entry.name) if parent else None +async def off_disk(roots: RootSet, fn, *args): + """ + Run one blocking filesystem call on the thread that serves `roots`. + + Every syscall against a group's content goes through here, `resolve()` and + `exists()` included: a `stat` is what *wakes* a sleeping disk, so a check + left on the event loop pays the spin-up in full and the read that follows + it finds the disk already awake. Offloading only the read would move the + stall, not remove it. + + `fn` must not touch anything the loop also touches — it runs on another + thread. Reading and encrypting a chunk qualifies; updating a session's + state does not. + """ + loop = asyncio.get_running_loop() + return await loop.run_in_executor(roots.io_executor, fn, *args) + + def _refuse_nesting(new: Root, existing: list[Root]) -> None: """ No root may contain another, compared case-insensitively. |