diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 9 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/roots.py | 63 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 74 |
3 files changed, 113 insertions, 33 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index fa9ef81..69c7e66 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -1789,6 +1789,15 @@ class NodeDaemon: for indexer in self._indexers: await indexer.stop() + # The thread each group's roots are read from (roots.py `io_executor`). + # Created on the first read, so a group nobody downloaded from never + # started one and this is a no-op for it. + groups = self._webrtc._ctx.get("groups") if self._webrtc else None + for group in (groups or {}).values(): + roots = group.get("roots") + if roots is not None: + roots.close_io() + if self._quic_server: await self._quic_server.stop() 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. diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index d79674d..e748be9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -139,7 +139,8 @@ from meshbay_node.media_probe import ( probe_video as _probe_video, ) from meshbay_node.roots import ( - ROOT_NOT_SERVED, RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, + ROOT_NOT_SERVED, RootSet, entry_abs_path, off_disk, SAFE_UPLOAD_NAME, safe_subdir, + _free_name, ) log = logging.getLogger(__name__) @@ -3877,12 +3878,9 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return # A real index entry, asked for without a lease: browsing, or a client @@ -3913,7 +3911,8 @@ class WebRTCPeerSession: file_id[:12], chunk_index, getattr(self._channel, "bufferedAmount", "?")) file_hash = bytes.fromhex(entry.id) - chunk_data = _read_and_encrypt( + chunk_data = await off_disk( + ctx["roots"], _read_and_encrypt, ctx["gek"], file_path, chunk_index, file_hash, entry.id) # Backpressure. Without it the node hands the whole window to the # channel at once and the reader sees the first chunk, then nothing for @@ -4007,12 +4006,9 @@ class WebRTCPeerSession: if not entry: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return media_cache = self._ctx.get("media_cache") @@ -4074,12 +4070,9 @@ class WebRTCPeerSession: if not entry: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return media_cache = self._ctx.get("media_cache") @@ -5882,7 +5875,7 @@ class WebRTCPeerSession: self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") return - self._exec_file_delete(ctx, file_id, entry) + await self._exec_file_delete(ctx, file_id, entry) async def _admin_exec_invite_create( self, pending: dict, transcript: bytes, sig: bytes, @@ -5918,15 +5911,15 @@ class WebRTCPeerSession: "username": result.get("username", ""), }) - def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: + async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal == ROOT_NOT_SERVED: # Frozen, not gone: removing the entry would lose a file that is # still on a drive the node cannot read right now. self._send({"type": "error", "detail": ROOT_NOT_SERVED}) return - if file_path.exists(): - file_path.unlink() + if file_path is not None: + await off_disk(ctx["roots"], file_path.unlink) log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) @@ -6123,12 +6116,9 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return gek = ctx.get("gek") @@ -6581,6 +6571,24 @@ class WebRTCPeerSession: await self._pc.close() +def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: + """ + Where an entry is, and whether it is readable — or the refusal to send. + + Both halves are syscalls: `resolve()` walks the path and `exists()` stats + it, and a stat is what *wakes* a sleeping disk. Leaving either on the event + loop and offloading only the read would move the stall rather than remove + it, and the read would then find the disk already awake. Blocking; called + through `off_disk`. + """ + path = entry_abs_path(roots, entry) + if path is None: + return None, ROOT_NOT_SERVED + if not path.exists(): + return None, "File not on disk" + return path, None + + def _read_and_encrypt( gek: bytes, file_path: Path, @@ -6588,7 +6596,7 @@ def _read_and_encrypt( file_hash: bytes, file_id: str = "", ) -> dict: - """Read one chunk off disk and encrypt it. Blocking; the caller keeps it short.""" + """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) |