diff --git a/src/python/jw/pkg/lib/ExecContext.py b/src/python/jw/pkg/lib/ExecContext.py index a282dc62..f4986c5e 100644 --- a/src/python/jw/pkg/lib/ExecContext.py +++ b/src/python/jw/pkg/lib/ExecContext.py @@ -17,6 +17,24 @@ from .FileContext import FileContext as Base from .log import DEBUG, ERR, NOTICE, log _US = '\x1f' # unlikely to appear in numeric output + +STAT_GNU_FORMAT = _US.join( + [ + '%f', # st_mode in hex + '%i', # st_ino + '%d', # st_dev + '%h', # st_nlink + '%U', # st_uid + '%G', # st_gid + '%s', # st_size + '%.9X', # st_atime + '%.9Y', # st_mtime + '%.9Z', # st_ctime + '%o', # st_blksize hint + '%b', # st_blocks + '%r', # st_rdev + ] +) _BILLION = Decimal(1_000_000_000) def _looks_like_option_error(stderr: str | None) -> bool: @@ -54,7 +72,10 @@ def _parse_epoch(value: str) -> tuple[int, float, int]: ns = int((dec * _BILLION).to_integral_value(rounding = ROUND_FLOOR)) return sec, float(dec), ns -def _build_stat_result(fields: list[str], mode_base: int) -> StatResult: +def parse_stat_fields(stdout: str, mode_base: int = 16) -> StatResult: # export + return build_stat_result(stdout.split(_US), mode_base = mode_base) + +def build_stat_result(fields: list[str], mode_base: int) -> StatResult: # export if len(fields) != 13: raise ValueError( f'unexpected stat output: expected 13 fields, got {len(fields)}: {fields!r}' @@ -621,27 +642,9 @@ class ExecContext(Base): ) # GNU coreutils stat - gnu_format = _US.join( - [ - '%f', # st_mode in hex - '%i', # st_ino - '%d', # st_dev - '%h', # st_nlink - '%U', # st_uid - '%G', # st_gid - '%s', # st_size - '%.9X', # st_atime - '%.9Y', # st_mtime - '%.9Z', # st_ctime - '%o', # st_blksize hint - '%b', # st_blocks - '%r', # st_rdev - ] - ) - - result = await __stat(['--printf', gnu_format]) + result = await __stat(['--printf', STAT_GNU_FORMAT]) if result.status == 0 and result.stdout is not None: - return _build_stat_result(result.stdout_str.split(_US), mode_base = 16) + return parse_stat_fields(result.stdout_str, mode_base = 16) if not _looks_like_option_error(result.stderr_str_or_none): # log(DEBUG, f'GNU stat attempt failed on "{path}" ({str(e)})') @@ -671,7 +674,7 @@ class ExecContext(Base): if result.status != 0 or stdout is None: _raise_stat_error(path, result) assert stdout is not None # Just there to pacify the linter - return _build_stat_result(stdout.rstrip('\n').split(_US), mode_base = 8) + return parse_stat_fields(stdout.rstrip('\n'), mode_base = 8) @override async def _chown(self, path: str, owner: str | None, group: str | None) -> None: diff --git a/src/python/jw/pkg/lib/FcCache.py b/src/python/jw/pkg/lib/FcCache.py new file mode 100644 index 00000000..8f52e963 --- /dev/null +++ b/src/python/jw/pkg/lib/FcCache.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import errno +import stat + +from typing import TYPE_CHECKING, Iterable, override + +if TYPE_CHECKING: + from .base import Result + from .FileContext import FileContext + from .file_types import FileType + +from .base import StatResult +from .ExecContext import STAT_GNU_FORMAT, ExecContext, parse_stat_fields +from .FcWrapper import FcWrapper + +class FcCache(FcWrapper): # export + """A write-through read cache in front of a file context. + + stat and get results are cached: stat per (path, + follow_symlinks), get per (path, wd, throw). Successful stats + and FileNotFoundErrors are cached; other errors are re-probed, + so a transient transport failure does not poison the cache. + Writes pass through to the wrapped context and drop the cache + entries of the affected paths. + + warm() probes a set of paths without following symlinks and + fills the stat cache with the results: an absent path is + absent, and a non-link is the same node, for both follow + variants, so the result is cached for both; a link is cached + without following, its followed variant probed on first use. + If the wrapped context is an exec context, one host command + stats all of the paths; otherwise, one stat per path runs. + Paths containing tab or newline characters are not supported. + """ + + def __init__(self, wrapped: FileContext) -> None: + super().__init__(wrapped) + self.__stat_cache: dict[tuple[str, bool], StatResult | Exception] = {} + self.__get_cache: dict[tuple[str, str | None, bool], Result] = {} + + @override + async def _stat(self, path: str, follow_symlinks: bool) -> StatResult: + key = (path, follow_symlinks) + hit = self.__stat_cache.get(key) + if hit is not None: + if isinstance(hit, StatResult): + return hit + raise hit + try: + st = await super()._stat(path, follow_symlinks) + except FileNotFoundError as e: + self.__stat_cache[key] = e + raise + self.__stat_cache[key] = st + return st + + @override + async def _get( + self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str + ) -> Result: + key = (path, wd, throw) + hit = self.__get_cache.get(key) + if hit is None: + hit = await super()._get(path, wd, throw, verbose, title) + self.__get_cache[key] = hit + return hit + + @override + async def _file_exists(self, path: str) -> bool: + # -- Reroute the existence check through the cached stat + # instead of the wrapped context's own fast path. A path + # below a file cannot exist: the exec contexts report it + # as absent, so the local one is treated the same + try: + await self._stat(path, False) + except FileNotFoundError: + return False + except NotADirectoryError: + return False + return True + + @override + async def _is_dir(self, path: str, follow_symlinks: bool) -> bool: + try: + return stat.S_ISDIR((await self._stat(path, follow_symlinks)).mode) + except FileNotFoundError: + return False + except NotADirectoryError: + return False + + async def warm(self, paths: Iterable[str]) -> None: + """Probe a set of paths without following symlinks and fill + the stat cache, as the class docstring describes. + + A path whose both follow variants are already in the cache + is skipped, so a warm of a set that is cached already does + nothing. If the wrapped context is an exec context, one + host command stats all of the remaining paths; otherwise, + one stat per path runs. + """ + composed = [] + for p in paths: + if '\t' in p or '\n' in p: + raise ValueError(f'warm: path {p!r} holds a tab or newline') + c = self._chroot(p) + if (c, False) not in self.__stat_cache or \ + (c, True) not in self.__stat_cache: + composed.append(c) + if not composed: + return + ec = self.wrapped + if not isinstance(ec, ExecContext): + for p in composed: + try: + st = await super()._stat(p, False) + except FileNotFoundError as e: + self.__cache_probe(p, e) + except NotADirectoryError: + # -- A path below a file is not statable; it is + # not cached, and file_exists() judges it + # absent + pass + else: + self.__cache_probe(p, st) + return + script = ( + 'while IFS= read -r p; do ' + "printf '%s\\t' \"$p\"; " + f"stat --printf '{STAT_GNU_FORMAT}\\n' \"$p\" 2>/dev/null || " + "printf 'ABSENT\\n'; " + 'done' + ) + ret = await ec.run( + ['sh', '-c', script], + cmd_input = ('\n'.join(composed) + '\n').encode(), + mod_env = {'LC_ALL': 'C'}, + throw = True, + ) + self.__parse_probe(ret.stdout_str, composed) + + def __cache_probe(self, path: str, result: StatResult | FileNotFoundError) -> None: + # -- Cache a probe that did not follow symlinks: an absent + # path is absent, and a non-link is the same node, for + # both follow variants; a link is cached without + # following, its followed variant probed on first use + self.__stat_cache[(path, False)] = result + if isinstance(result, FileNotFoundError) or not stat.S_ISLNK(result.mode): + self.__stat_cache[(path, True)] = result + + def __parse_probe(self, stdout: str, composed: list[str]) -> None: + # -- One line per path: the path, a tab, then the stat + # fields or ABSENT + lines = stdout.splitlines() + if len(lines) != len(composed): + raise RuntimeError( + f'warm: probe reported {len(lines)} lines for {len(composed)} paths' + ) + for p, line in zip(composed, lines): + name, sep, rest = line.partition('\t') + if name != p or not sep: + raise RuntimeError(f'warm: probe line does not name "{p}": {line!r}') + if rest == 'ABSENT': + self.__cache_probe(p, FileNotFoundError(errno.ENOENT, f'stat: {p}')) + else: + self.__cache_probe(p, parse_stat_fields(rest)) + + def __invalidate(self, *paths: str) -> None: + # -- A write changes the stat and get answers of its paths + for p in paths: + for fl in (True, False): + self.__stat_cache.pop((p, fl), None) + self.__get_cache = {k: r for k, r in self.__get_cache.items() if k[0] != p} + + @override + async def _put( + self, + path: str, + content: bytes, + wd: str | None, + throw: bool, + verbose: bool | None, + title: str, + owner: str | None, + group: str | None, + mode: str | None, + atomic: bool, + ftype: FileType, + ) -> Result: + r = await super()._put( + path, + content, + wd, + throw, + verbose, + title, + owner, + group, + mode, + atomic, + ftype, + ) + self.__invalidate(path) + return r + + @override + async def _mkdir(self, path: str, mode: int, ftype: FileType) -> None: + await super()._mkdir(path, mode, ftype) + self.__invalidate(path) + + @override + async def _unlink(self, path: str) -> None: + await super()._unlink(path) + self.__invalidate(path) + + @override + async def _erase(self, path: str) -> None: + await super()._erase(path) + self.__invalidate(path) + + @override + async def _rename(self, src: str, dst: str) -> None: + await super()._rename(src, dst) + self.__invalidate(src, dst) + + @override + async def _chown(self, path: str, owner: str | None, group: str | None) -> None: + await super()._chown(path, owner, group) + self.__invalidate(path) + + @override + async def _chmod(self, path: str, mode: int) -> None: + await super()._chmod(path, mode) + self.__invalidate(path) diff --git a/test/unit/python/jw/pkg/lib/FcCache/Makefile b/test/unit/python/jw/pkg/lib/FcCache/Makefile new file mode 100644 index 00000000..de693650 --- /dev/null +++ b/test/unit/python/jw/pkg/lib/FcCache/Makefile @@ -0,0 +1,7 @@ +TOPDIR = ../../../../../../.. + +include $(TOPDIR)/make/proj.mk +include $(JWBDIR)/make/py-run.mk + +all: +test: run diff --git a/test/unit/python/jw/pkg/lib/FcCache/test.py b/test/unit/python/jw/pkg/lib/FcCache/test.py new file mode 100644 index 00000000..8beee776 --- /dev/null +++ b/test/unit/python/jw/pkg/lib/FcCache/test.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import asyncio +import os +import tempfile + +from typing import TYPE_CHECKING, override + +if TYPE_CHECKING: + from jw.pkg.lib.file_types import FileType + +from jw.pkg.lib.base import StatResult +from jw.pkg.lib.ec.Local import Local +from jw.pkg.lib.ExecContext import ExecContext +from jw.pkg.lib.FcCache import FcCache +from jw.pkg.lib.file_types import FT_DIR, FT_REG +from jw.pkg.lib.FileContext import FileContext +from jw.pkg.lib.Result import Result + +class FakeCtx(FileContext): + """An in-memory FileContext that counts its primitive calls""" + + def __init__(self, uri: str, chroot: bool = False) -> None: + super().__init__(uri, chroot = chroot) + self.files: dict[str, bytes] = {} + self.dirs: set[str] = set() + self.counts: dict[str, int] = {} + + def count(self, op: str) -> None: + self.counts[op] = self.counts.get(op, 0) + 1 + + def _stat_of(self, path: str) -> StatResult: + if path in self.dirs: + return StatResult(0o40755, '0', '0', 40, 0.0, 0.0, 0.0) + if path in self.files: + return StatResult(0o100644, '0', '0', len(self.files[path]), 0.0, 0.0, 0.0) + raise FileNotFoundError(path) + + @override + async def _get( + self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str + ) -> Result: + self.count('get') + if wd is not None: + path = wd + '/' + path + if path not in self.files: + raise FileNotFoundError(path) + return Result(stdout = self.files[path]) + + @override + async def _put( + self, + path: str, + content: bytes, + wd: str | None, + throw: bool, + verbose: bool | None, + title: str, + owner: str | None, + group: str | None, + mode: str | None, + atomic: bool, + ftype: FileType, + ) -> Result: + self.count('put') + if wd is not None: + path = wd + '/' + path + self.files[path] = content + return Result(stdout = b'') + + @override + async def _mkdir(self, path: str, mode: int, ftype: FileType) -> None: + self.count('mkdir') + self.dirs.add(path) + + @override + async def _unlink(self, path: str) -> None: + self.count('unlink') + self.files.pop(path, None) + self.dirs.discard(path) + + @override + async def _stat(self, path: str, follow_symlinks: bool) -> StatResult: + self.count('stat') + return self._stat_of(path) + +class CountingLocal(Local): + """A Local context that counts its executed commands""" + + def __init__(self) -> None: + super().__init__('local') + self.runs = 0 + + @override + async def _run( + self, + cmd: list[str], + wd: str | None, + verbose: bool, + cmd_input: bytes | None, + mod_env: dict[str, str] | None, + interactive: bool, + log_prefix: str, + ) -> Result: + self.runs += 1 + return await super()._run( + cmd, wd, verbose, cmd_input, mod_env, interactive, log_prefix + ) + +async def main() -> None: + fake = FakeCtx('local:///tmp/fc-cache-fake', chroot = True) + fc = FcCache(fake) + await fc.mkdir('d', 0o755, FT_DIR) + await fc.put('d/a', b'x', ftype = FT_REG) + + # -- Successful stats are cached + st = await fc.stat('d/a') + assert st.size == 1 + assert (await fc.stat('d/a')).size == 1 + assert fake.counts['stat'] == 1 + + # -- Absences are cached, too + assert await fc.file_exists('d/missing') is False + assert await fc.file_exists('d/missing') is False + assert fake.counts['stat'] == 2 + + # -- Gets are cached per (path, wd, throw) + ret = await fc.get('d/a') + assert (await fc.get('d/a')).stdout == b'x' + assert fake.counts['get'] == 1 + ret = await fc._get('a', '/tmp/fc-cache-fake/d', True, None, 'get') + assert fake.counts['get'] == 2 + assert ret.stdout == b'x' + + # -- Writes drop the cache entries of their paths + await fc.put('d/a', b'xy', ftype = FT_REG) + assert (await fc.get('d/a')).stdout == b'xy' + assert fake.counts['get'] == 3 + assert (await fc.stat('d/a')).size == 2 + assert fake.counts['stat'] == 3 + + # -- Unlink drops them as well + await fc.unlink('d/a') + assert await fc.file_exists('d/a') is False + assert fake.counts['stat'] == 4 + assert '/tmp/fc-cache-fake/d/a' not in fake.files + + # -- The wrapper itself is not an exec context + assert not isinstance(fc, ExecContext) + + # -- warm() on a non-exec context: one stat per path, then cached + fake2 = FakeCtx('local:///tmp/fc-cache-warm', chroot = True) + fc2 = FcCache(fake2) + await fc2.put('p/q', b'z', ftype = FT_REG) + await fc2.warm(['p', 'p/q', 'p/r']) + assert fake2.counts['stat'] == 3 + await fc2.warm(['p', 'p/q', 'p/r']) + assert fake2.counts['stat'] == 3 + assert await fc2.file_exists('p/r') is False + assert fake2.counts['stat'] == 3 + + # -- warm() on an exec context: one host command for all paths + with tempfile.TemporaryDirectory() as td: + f = os.path.join(td, 'there') + with open(f, 'w') as fh: + fh.write('hello\n') + cl = CountingLocal() + fc3 = FcCache(cl) + missing = os.path.join(td, 'not-there') + await fc3.warm([f, missing]) + assert cl.runs == 1 + + st = await fc3.stat(f) + assert cl.runs == 1 + assert st.size == 6 + assert await fc3.file_exists(missing) is False + assert cl.runs == 1 + try: + await fc3.stat(missing) + assert False, 'stat of a warmed absence must raise' + except FileNotFoundError: + pass + assert cl.runs == 1 + + # -- A put through the cache lands in the wrapped context and + # drops its cache entries, so the next stat is a fresh one. + # Local stats without a command, so file_exists adds no run + g = os.path.join(td, 'new') + await fc3.put(g, b'fresh\n', ftype = FT_REG) + assert cl.runs == 2 + assert (await fc3.get(g)).stdout == b'fresh\n' + assert cl.runs == 3 + assert await fc3.file_exists(g) is True + assert cl.runs == 3 + assert await fc3.file_exists(g) is True + assert cl.runs == 3 + +asyncio.run(main())