diff --git a/src/arcadedb_embedded/aot.py b/src/arcadedb_embedded/aot.py new file mode 100644 index 00000000..b4c685b0 --- /dev/null +++ b/src/arcadedb_embedded/aot.py @@ -0,0 +1,479 @@ +""" +ArcadeDB Python Bindings - opt-in JDK AOT cache (Project Leyden, JEP 483/514/515) + +``ARCADEDB_AOT=1`` (or ``start_jvm(aot=True)``) lets the JVM keep a per-user +ahead-of-time cache of the classes it loads and links at start-up, which makes +the next start faster (cold start). It is off by default. + +A cache records the absolute jar paths and each jar's mtime and size, so it +cannot ship inside the wheel: the first process on a machine *trains* it +(``-XX:AOTCacheOutput=``, the JVM writes the file when it exits normally), and +later processes *use* it (``-XX:AOTCache=... -XX:AOTMode=auto``, where the JVM +ignores a cache it cannot use instead of failing). + +Everything here degrades to a normal start. No function raises to the caller of +``start_jvm``: any failure (unwritable directory, full disk, unsupported JDK, +unusable cache) leaves the JVM arguments untouched and logs one debug line. + +Concurrency: one process at a time trains, chosen by an exclusive lock file +that holds the trainer's pid. The others start normally. The trained file is +written under a unique temporary name; the *next* start promotes it with an +atomic rename after the trainer is gone and after the bundled JVM has accepted +it. A cache that is missing, truncated or altered is rebuilt, never trusted. +""" + +import hashlib +import json +import os +import platform +import re +import secrets +import subprocess # nosec B404 - only runs the bundled JRE's java to validate a cache +import sys +import tempfile +import time +from pathlib import Path +from typing import Iterable, Optional + +from ._logging import get_logger + +_LOGGER = get_logger(__name__) + +ENV_ENABLE = "ARCADEDB_AOT" +ENV_CACHE = "ARCADEDB_AOT_CACHE" + +# One-step cache creation (-XX:AOTCacheOutput) arrived with JDK 25 (JEP 514). +MIN_JAVA_MAJOR = 25 +# A lock older than this is stale even if its pid looks alive (pid reuse). +LOCK_STALE_SECONDS = 30 * 60 +# Upper bound for the one-off check that the JVM accepts a freshly trained cache. +VALIDATE_TIMEOUT_SECONDS = 60 + +# Flags of the user that already decide how the JVM treats an AOT/CDS archive; +# with any of them present the bindings add nothing. +_USER_FLAG_PREFIXES = ( + "-XX:AOTCache", + "-XX:AOTMode", + "-XX:AOTConfiguration", + "-XX:SharedArchiveFile", + "-XX:ArchiveClassesAtExit", + "-Xshare:", + "-XX:-UseSharedSpaces", +) + +_TRUE = {"1", "true", "yes", "on"} + + +def enabled(aot: Optional[bool]) -> bool: + """The keyword wins; without it ``ARCADEDB_AOT`` decides. Default off.""" + if aot is not None: + return bool(aot) + return os.environ.get(ENV_ENABLE, "").strip().lower() in _TRUE + + +def cache_dir() -> Path: + """Per-user cache directory by platform (stdlib only; never raises). + + Linux ``$XDG_CACHE_HOME/arcadedb/aot`` or ``~/.cache/arcadedb/aot``, macOS + ``~/Library/Caches/arcadedb/aot``, Windows ``%LOCALAPPDATA%\\arcadedb\\Cache\\aot``; + the system temp directory when no home directory can be found. + """ + if sys.platform == "win32": + local = os.environ.get("LOCALAPPDATA") + if local: + return Path(local) / "arcadedb" / "Cache" / "aot" + elif sys.platform != "darwin": + xdg = os.environ.get("XDG_CACHE_HOME") + # The XDG spec ignores a relative value. + if xdg and os.path.isabs(xdg): + return Path(xdg) / "arcadedb" / "aot" + try: + home = Path.home() + except (RuntimeError, KeyError, OSError): + return Path(tempfile.gettempdir()) / "arcadedb" / "aot" + if sys.platform == "win32": + return home / "AppData" / "Local" / "arcadedb" / "Cache" / "aot" + if sys.platform == "darwin": + return home / "Library" / "Caches" / "arcadedb" / "aot" + return home / ".cache" / "arcadedb" / "aot" + + +def jpype_support_jar() -> Optional[Path]: + """JPype's ``org.jpype.jar``, which JPype appends to the JVM class path. + + The cache records the whole class path, so this jar (an absolute path inside + the Python environment) is part of what the cache is valid for. None when it + cannot be found, then no cache is used: it could not be checked. + """ + try: + import jpype + + jar = Path(jpype.__file__).resolve().parent.parent / "org.jpype.jar" + return jar if jar.is_file() else None + except Exception: # noqa: BLE001 + return None + + +def jre_identity(jre_dir: Path) -> Optional[tuple]: + """``(major, token)`` of the bundled JRE, or None when it cannot be read. + + ``token`` stands for ``java.vm.version``: the ``JAVA_VERSION`` of the JRE's + ``release`` file plus the size of the JVM library (a rebuilt JRE with the + same version string changes the size). Read from files, so no JVM is needed. + """ + try: + text = (Path(jre_dir) / "release").read_text("utf-8", errors="replace") + match = re.search(r'^JAVA_VERSION="([^"]+)"', text, re.MULTILINE) + if not match: + return None + version = match.group(1) + major = int(re.match(r"(\d+)", version).group(1)) + size = 0 + for rel in ( + ("lib", "server", "libjvm.so"), + ("lib", "server", "libjvm.dylib"), + ("bin", "server", "jvm.dll"), + ): + lib = Path(jre_dir).joinpath(*rel) + if lib.exists(): + size = lib.stat().st_size + break + return major, f"{version}+{size}" + except (OSError, ValueError, AttributeError): + return None + + +def cache_key( + package_version: str, + jre_token: str, + jar_sha256: str, + classpath: str, + jar_stats: str = "", +) -> str: + """Hash of everything a cache is only valid for (32 hex characters). + + Package version, bundled JRE version, the sha256 over the sorted (name, + content) of every jar (``jar_fingerprint``), the exact class path string (the + cache stores the class path in order), the size and mtime of each jar (the JVM + refuses a cache when either changed, even if the bytes did not) and the platform. + """ + h = hashlib.sha256() + for part in ( + package_version, + jre_token, + jar_sha256, + classpath, + jar_stats, + sys.platform, + platform.machine(), + ): + h.update(part.encode("utf-8", errors="surrogateescape")) + h.update(b"\0") + return h.hexdigest()[:32] + + +def _jar_stats(classpath: str) -> str: + parts = [] + for jar in classpath.split(os.pathsep): + st = os.stat(jar) + parts.append(f"{os.path.basename(jar)}:{st.st_size}:{st.st_mtime_ns}") + return ";".join(sorted(parts)) + + +def _sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + h.update(chunk) + return h.hexdigest() + + +def _pid_alive(pid: int) -> bool: + """True if the process exists (conservative: True when unsure).""" + if pid <= 0: + return False + if sys.platform == "win32": + try: + import ctypes + + handle = ctypes.windll.kernel32.OpenProcess(0x1000, False, pid) + if not handle: + return ctypes.windll.kernel32.GetLastError() != 87 # not invalid pid + code = ctypes.c_ulong() + ctypes.windll.kernel32.GetExitCodeProcess(handle, ctypes.byref(code)) + ctypes.windll.kernel32.CloseHandle(handle) + return code.value == 259 # STILL_ACTIVE + except Exception: # noqa: BLE001 - unsure means alive + return True + try: + os.kill(pid, 0) # signal 0 only checks existence + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return True + return True + + +def _write_json_atomic(path: Path, data: dict) -> None: + tmp = path.with_name(f"{path.name}.{os.getpid()}.{secrets.token_hex(4)}.part") + try: + tmp.write_text(json.dumps(data), encoding="utf-8") + os.replace(tmp, path) + finally: + try: + tmp.unlink() + except OSError: + pass + + +def _read_json(path: Path) -> Optional[dict]: + try: + data = json.loads(path.read_text("utf-8")) + return data if isinstance(data, dict) else None + except (OSError, ValueError): + return None + + +def _unlink(*paths) -> None: + for p in paths: + try: + Path(p).unlink() + except OSError: + pass + + +def _remove_training_files(tmp: Path) -> None: + """Remove a training output and the side files the JVM derives from its name.""" + try: + for p in tmp.parent.glob(tmp.name + "*"): + _unlink(p) + except OSError: + pass + + +def _jvm_accepts( + jre_dir: Path, cache: Path, classpath: str, jvm_flags: Iterable[str] +) -> bool: + """Ask the bundled JVM to start with ``AOTMode=on`` (refuses an unusable cache).""" + java = Path(jre_dir) / "bin" / ("java.exe" if sys.platform == "win32" else "java") + if not java.exists(): + # No launcher to ask: fall back on the file being non-empty. A bad file + # still cannot hurt, AOTMode=auto ignores a cache it cannot use. + return cache.stat().st_size > 0 + cmd = [ + str(java), + f"-XX:AOTCache={cache}", + "-XX:AOTMode=on", + *jvm_flags, + "-cp", + classpath, + "-version", + ] + result = subprocess.run( # nosec B603 - argv list, no shell: the bundled java, fixed flags, our own cache path + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, # the JVM reports start-up errors on stdout + timeout=VALIDATE_TIMEOUT_SECONDS, + check=False, + ) + if result.returncode != 0: + _LOGGER.debug( + "JVM refused the AOT cache (exit %s): %s", + result.returncode, + result.stdout.decode("utf-8", "replace").strip()[-400:], + ) + return result.returncode == 0 + + +class _Paths: + def __init__(self, final: Path, key: str): + self.final = final + self.key = key + self.meta = final.with_name(final.name + ".json") + self.lock = final.with_name(final.name + ".lock") + + +def _usable_final(p: _Paths) -> bool: + """Final file present, recorded for this key, same size and same sha256.""" + meta = _read_json(p.meta) + if not meta or meta.get("key") != p.key: + return False + try: + if p.final.stat().st_size != meta.get("size"): + return False + return _sha256_file(p.final) == meta.get("sha256") + except OSError: + return False + + +def _recover(p: _Paths, jre_dir: Path, classpath: str, jvm_flags: list) -> None: + """Finish a finished or dead training run: promote its file or clean it up. + + Does nothing while the trainer lives (and its lock is fresh). Of several + processes arriving together, one wins the right to recover by renaming the + lock file (atomic); the others carry on. + """ + info = _read_json(p.lock) + if info is None and not p.lock.exists(): + _remove_orphans(p) + return + if info is not None: + age = time.time() - float(info.get("started", 0)) + if _pid_alive(int(info.get("pid", 0))) and age < LOCK_STALE_SECONDS: + return + else: + # Unreadable: a trainer between creating the lock and writing it, or junk. + try: + if time.time() - p.lock.stat().st_mtime < 60: + return + except OSError: + return + claim = p.lock.with_name(f"{p.lock.name}.{os.getpid()}.claim") + try: + os.rename(p.lock, claim) + except OSError: + return # somebody else recovers + try: + tmp = Path(info["tmp"]) if info and info.get("tmp") else None + promoted = False + if tmp is not None and tmp.exists() and tmp.stat().st_size > 0: + if info.get("key") == p.key and _jvm_accepts( + jre_dir, tmp, classpath, jvm_flags + ): + size = tmp.stat().st_size + digest = _sha256_file(tmp) + os.replace(tmp, p.final) + _write_json_atomic( + p.meta, {"key": p.key, "size": size, "sha256": digest} + ) + promoted = True + _LOGGER.debug("AOT cache %s promoted (%d bytes)", p.final, size) + if not promoted and tmp is not None: + _LOGGER.debug("AOT training output %s unusable, removed", tmp) + _remove_training_files(tmp) + if tmp is not None: + # The JVM may leave a side file next to its output (name + suffix). + for extra in tmp.parent.glob(tmp.name + ".*"): + _unlink(extra) + finally: + _unlink(claim) + + +def _remove_orphans(p: _Paths) -> None: + """Remove training outputs no lock refers to and that are long finished.""" + try: + for f in p.final.parent.glob(p.final.name + ".*.tmp*"): + if time.time() - f.stat().st_mtime > LOCK_STALE_SECONDS: + _unlink(f) + except OSError: + pass + + +def _try_lock(p: _Paths, tmp: Path) -> bool: + """Become the one trainer: create the lock exclusively. False if taken.""" + try: + fd = os.open(p.lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + return False + with os.fdopen(fd, "w", encoding="utf-8") as fh: + json.dump( + { + "pid": os.getpid(), + "started": time.time(), + "tmp": str(tmp), + "key": p.key, + }, + fh, + ) + return True + + +def _user_flags_present(args: Iterable[str]) -> bool: + return any(str(a).startswith(_USER_FLAG_PREFIXES) for a in args) + + +def jvm_flags_for( + jvm_args: list, + jre_dir: Path, + classpath: str, + package_version: str, + jar_sha256: str, +) -> list: + """The AOT flags to add to ``jvm_args`` (an empty list means a normal start). + + Never raises. Assumes the caller checked ``enabled()``. + """ + try: + return _plan(jvm_args, Path(jre_dir), classpath, package_version, jar_sha256) + except Exception as exc: # noqa: BLE001 - AOT is an optimisation only + _LOGGER.debug("AOT cache skipped: %s: %s", type(exc).__name__, exc) + return [] + + +def _plan( + jvm_args: list, + jre_dir: Path, + classpath: str, + package_version: str, + jar_sha256: str, +) -> list: + if _user_flags_present(jvm_args): + _LOGGER.debug("AOT cache skipped: AOT/CDS flags given in the JVM arguments") + return [] + ident = jre_identity(jre_dir) + if ident is None or ident[0] < MIN_JAVA_MAJOR: + _LOGGER.debug( + "AOT cache skipped: bundled JRE %s lacks one-step AOT cache (needs JDK %d)", + ident, + MIN_JAVA_MAJOR, + ) + return [] + support = jpype_support_jar() + if support is None: + _LOGGER.debug("AOT cache skipped: org.jpype.jar not found") + return [] + # What the JVM will really see; the cache stores it and must match it. + classpath = classpath + os.pathsep + str(support) + key = cache_key( + package_version, ident[1], jar_sha256, classpath, _jar_stats(classpath) + ) + + override = os.environ.get(ENV_CACHE) + if override: + final = Path(override).expanduser() + if not final.name or str(override).endswith(("/", os.sep)): + raise ValueError(f"{ENV_CACHE} must name a file, got {override!r}") + else: + final = cache_dir() / f"{key}.aot" + final.parent.mkdir(parents=True, exist_ok=True) + p = _Paths(final, key) + + # The JVM accepts a cache only with the settings it was trained with (compact + # object headers, modules, class path), so the validation run gets all of them. + _recover(p, jre_dir, classpath, list(jvm_args)) + + if p.final.exists(): + if _usable_final(p): + _LOGGER.debug("AOT cache in use: %s", p.final) + return [f"-XX:AOTCache={p.final}", "-XX:AOTMode=auto"] + _LOGGER.debug("AOT cache %s is stale or damaged, retraining", p.final) + _unlink(p.final, p.meta) + + tmp = p.final.with_name(f"{p.final.name}.{os.getpid()}.{secrets.token_hex(4)}.tmp") + if not _try_lock(p, tmp): + _LOGGER.debug("AOT cache: another process is training, starting normally") + return [] + _LOGGER.debug("AOT cache training into %s", tmp) + return [f"-XX:AOTCacheOutput={tmp}"] + + +def release_training(flags: Iterable[str]) -> None: + """Give up a training run whose JVM did not start (frees the lock and file).""" + for flag in flags: + if flag.startswith("-XX:AOTCacheOutput="): + tmp = Path(flag.split("=", 1)[1]) + _remove_training_files(tmp) + final_name = tmp.name.split(".aot.")[0] + ".aot" + _unlink(tmp.with_name(final_name + ".lock")) diff --git a/src/arcadedb_embedded/jvm.py b/src/arcadedb_embedded/jvm.py index 3961e1fb..edea1489 100644 --- a/src/arcadedb_embedded/jvm.py +++ b/src/arcadedb_embedded/jvm.py @@ -22,6 +22,7 @@ import jpype import jpype.imports +from . import aot as _aot from ._logging import get_logger from .exceptions import ArcadeDBError @@ -309,6 +310,7 @@ def start_jvm( jvm_args: Optional[Union[Iterable[str], str]] = None, common_pool_parallelism: Optional[int] = None, interrupt: Optional[bool] = None, + aot: Optional[bool] = None, ): """ Start the JVM with ArcadeDB JARs if not already started. @@ -409,6 +411,21 @@ def start_jvm( # Explicit thread cap for reproducible benchmarks start_jvm(heap_size="8g", common_pool_parallelism=8) + aot (optional) + Opt in to the JDK AOT cache (Project Leyden): a per-user cache of loaded + and linked classes that makes later starts faster. ``None`` (the default) + follows ``ARCADEDB_AOT`` (``1`` = on), which is off when unset; ``True`` + and ``False`` override it. The first process trains the cache and the JVM + writes it when that process exits normally. Any problem falls back to a + normal start. Ignored when the JVM is already started or when ``jvm_args`` + carry their own ``-XX:AOT*`` or ``-Xshare`` flags. + + ARCADEDB_AOT (optional) + ``1`` turns the AOT cache on (see ``aot``). ``ARCADEDB_AOT_CACHE=`` + names the cache file; default: a file keyed by package version, JRE and + jars in the per-user cache directory (Linux ``$XDG_CACHE_HOME/arcadedb/aot`` + or ``~/.cache/arcadedb/aot``). + ARCADEDB_LOG_DIR (optional) Directory for the engine log (arcadedb.log.*), the JVM crash log and the generated logging.properties. Default: a per-user state directory, @@ -488,6 +505,7 @@ def start_jvm( log_dir=log_dir, ) _check_initial_heap(jvm_args) + aot_flags = _aot_flags(aot, jvm_args, classpath) try: # Always use bundled JRE @@ -495,9 +513,23 @@ def start_jvm( # handles SIGINT and ends the process with status 130, bypassing Python's # signal handler, KeyboardInterrupt, `finally` and `atexit` (#118). _JVM_INTERRUPT = bool(interrupt) - jpype.startJVM( - jvm_path, *jvm_args, classpath=classpath, interrupt=_JVM_INTERRUPT - ) + try: + jpype.startJVM( + jvm_path, + *jvm_args, + *aot_flags, + classpath=classpath, + interrupt=_JVM_INTERRUPT, + ) + except Exception: + # A JVM cannot be created twice in one process (JPype cannot retry + # after a failed start), so there is no fallback here: free the + # training lock and report as usual. The flags are only added for a + # JRE that supports them, so this is not expected. + _aot.release_training(aot_flags) + raise + # AOT flags stay out of the recorded config: they differ per process + # (training file name) and must not make a later start_jvm() look changed. _JVM_CONFIG = tuple(jvm_args) _LOG_DIR_USED = log_dir except Exception as e: @@ -527,6 +559,25 @@ def start_jvm( faulthandler.disable() +def _aot_flags(aot: Optional[bool], jvm_args: list, classpath: str) -> list: + """AOT cache flags for this start; empty when off, unsupported or on any error.""" + if not _aot.enabled(aot): + return [] + try: + from . import __version__ + + return _aot.jvm_flags_for( + jvm_args, + _extract_runtime_resource("jre"), + classpath, + __version__, + jar_fingerprint()["sha256"], + ) + except Exception as exc: # noqa: BLE001 - an optimisation must never fail a start + _LOGGER.debug("AOT cache skipped: %s: %s", type(exc).__name__, exc) + return [] + + def _restore_python_signal_handlers() -> None: """Give SIGINT and SIGTERM back to Python after JPype took them (``interrupt=False``). diff --git a/tests/test_aot.py b/tests/test_aot.py new file mode 100644 index 00000000..3c4bb39a --- /dev/null +++ b/tests/test_aot.py @@ -0,0 +1,287 @@ +"""Opt-in JDK AOT cache (arcadedb_embedded.aot): key, paths, fallbacks. No JVM is started.""" + +import hashlib +import json +import os +import time +from pathlib import Path + +import pytest + +from arcadedb_embedded import aot, jvm + + +def _fake_jre(tmp_path: Path, version="25.0.4.1", libsize=1234) -> Path: + jre = tmp_path / "jre" + (jre / "lib" / "server").mkdir(parents=True) + (jre / "release").write_text(f'JAVA_VERSION="{version}"\nMODULES="java.base"\n') + (jre / "lib" / "server" / "libjvm.so").write_bytes(b"x" * libsize) + return jre + + +@pytest.fixture +def env(tmp_path, monkeypatch): + """A fake JRE, one jar and org.jpype.jar, and a cache directory under tmp_path.""" + jre = _fake_jre(tmp_path) + jar = tmp_path / "a.jar" + jar.write_bytes(b"jar") + support = tmp_path / "org.jpype.jar" + support.write_bytes(b"jpype") + monkeypatch.setattr(aot, "jpype_support_jar", lambda: support) + monkeypatch.delenv(aot.ENV_CACHE, raising=False) + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache")) + monkeypatch.setattr(aot.sys, "platform", "linux") + return dict(jre=jre, cp=str(jar), dir=tmp_path / "cache" / "arcadedb" / "aot") + + +def _plan(env, args=(), version="1.0", sha="s"): + return aot.jvm_flags_for(list(args), env["jre"], env["cp"], version, sha) + + +def test_enabled_keyword_beats_environment(monkeypatch): + monkeypatch.delenv(aot.ENV_ENABLE, raising=False) + assert aot.enabled(None) is False + monkeypatch.setenv(aot.ENV_ENABLE, "1") + assert aot.enabled(None) is True + assert aot.enabled(False) is False + monkeypatch.setenv(aot.ENV_ENABLE, "0") + assert aot.enabled(None) is False + assert aot.enabled(True) is True + + +def test_cache_dir_by_platform(monkeypatch, tmp_path): + monkeypatch.setattr(aot.sys, "platform", "linux") + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + assert aot.cache_dir() == tmp_path / "arcadedb" / "aot" + monkeypatch.setenv("XDG_CACHE_HOME", "relative/ignored") + assert aot.cache_dir() == Path.home() / ".cache" / "arcadedb" / "aot" + monkeypatch.setattr(aot.sys, "platform", "darwin") + assert aot.cache_dir() == Path.home() / "Library" / "Caches" / "arcadedb" / "aot" + monkeypatch.setattr(aot.sys, "platform", "win32") + monkeypatch.setenv("LOCALAPPDATA", str(tmp_path)) + assert aot.cache_dir() == tmp_path / "arcadedb" / "Cache" / "aot" + + +def test_jre_identity(tmp_path): + assert aot.jre_identity(_fake_jre(tmp_path, "25.0.4.1", 10)) == (25, "25.0.4.1+10") + assert aot.jre_identity(tmp_path / "nowhere") is None + bad = tmp_path / "bad" + bad.mkdir() + (bad / "release").write_text("no version here") + assert aot.jre_identity(bad) is None + + +def test_key_depends_on_every_input_and_is_stable(): + base = ("1.0", "25.0.4.1+9", "sha", "cp", "stats") + key = aot.cache_key(*base) + assert key == aot.cache_key(*base) + assert len(key) == 32 + for i in range(len(base)): + changed = list(base) + changed[i] += "x" + assert aot.cache_key(*changed) != key, i + + +def test_first_start_trains_with_a_unique_tmp_and_a_lock(env): + flags = _plan(env) + assert len(flags) == 1 and flags[0].startswith("-XX:AOTCacheOutput=") + tmp = Path(flags[0].split("=", 1)[1]) + assert tmp.parent == env["dir"] and tmp.name.endswith(".tmp") + assert str(os.getpid()) in tmp.name + locks = list(env["dir"].glob("*.lock")) + assert len(locks) == 1 + assert json.loads(locks[0].read_text())["pid"] == os.getpid() + + +def test_second_live_process_starts_normally(env): + assert _plan(env)[0].startswith("-XX:AOTCacheOutput=") + assert _plan(env) == [] # the first trainer is this (alive) process + + +def test_dead_trainer_output_is_promoted_then_used(env, monkeypatch): + tmp = Path(_plan(env)[0].split("=", 1)[1]) + tmp.write_bytes(b"cache-bytes") + lock = next(env["dir"].glob("*.lock")) + info = json.loads(lock.read_text()) + info["pid"] = 2**22 + 12345 # not a live pid + lock.write_text(json.dumps(info)) + monkeypatch.setattr(aot, "_pid_alive", lambda pid: False) + monkeypatch.setattr(aot, "_jvm_accepts", lambda *a, **k: True) + flags = _plan(env) + assert flags[0].startswith("-XX:AOTCache=") and flags[1] == "-XX:AOTMode=auto" + final = Path(flags[0].split("=", 1)[1]) + assert final.read_bytes() == b"cache-bytes" and not tmp.exists() + assert not list(env["dir"].glob("*.lock")) and not list(env["dir"].glob("*.tmp")) + # and a third start just uses it + assert _plan(env) == flags + + +def test_output_the_jvm_refuses_is_removed_and_retrained(env, monkeypatch): + tmp = Path(_plan(env)[0].split("=", 1)[1]) + tmp.write_bytes(b"partial") + monkeypatch.setattr(aot, "_pid_alive", lambda pid: False) + monkeypatch.setattr(aot, "_jvm_accepts", lambda *a, **k: False) + flags = _plan(env) + assert flags[0].startswith("-XX:AOTCacheOutput=") + assert not tmp.exists() and not list(env["dir"].glob("*.aot")) + + +def test_missing_output_of_a_dead_trainer_clears_the_lock(env, monkeypatch): + _plan(env) # lock held, no output ever written (os._exit, crash) + monkeypatch.setattr(aot, "_pid_alive", lambda pid: False) + assert _plan(env)[0].startswith("-XX:AOTCacheOutput=") + + +def test_old_lock_of_a_live_pid_is_stale(env, monkeypatch): + _plan(env) + lock = next(env["dir"].glob("*.lock")) + info = json.loads(lock.read_text()) + info["started"] = time.time() - aot.LOCK_STALE_SECONDS - 5 + lock.write_text(json.dumps(info)) + assert _plan(env)[0].startswith("-XX:AOTCacheOutput=") # took over + + +def _good_cache(env, monkeypatch, payload=b"cache-bytes"): + tmp = Path(_plan(env)[0].split("=", 1)[1]) + tmp.write_bytes(payload) + monkeypatch.setattr(aot, "_pid_alive", lambda pid: False) + monkeypatch.setattr(aot, "_jvm_accepts", lambda *a, **k: True) + flags = _plan(env) + return Path(flags[0].split("=", 1)[1]), flags + + +@pytest.mark.parametrize( + "damage", ["truncate", "flip", "empty", "meta-gone", "meta-junk"] +) +def test_damaged_cache_is_never_used_and_is_rebuilt(env, monkeypatch, damage): + final, _ = _good_cache(env, monkeypatch) + meta = final.with_name(final.name + ".json") + if damage == "truncate": + final.write_bytes(b"cache") + elif damage == "flip": + final.write_bytes(b"cache-byteX") + elif damage == "empty": + final.write_bytes(b"") + elif damage == "meta-gone": + meta.unlink() + else: + meta.write_text("{nope") + monkeypatch.undo() # real pid check again; the earlier trainer is gone + # undo() also restored the real sys.platform, so put back the fixture's "linux": on macOS and Windows + # cache_dir() would otherwise leave XDG_CACHE_HOME for ~/Library/Caches or %LOCALAPPDATA% + monkeypatch.setattr(aot.sys, "platform", "linux") + monkeypatch.setenv("XDG_CACHE_HOME", str(env["dir"].parents[1])) + monkeypatch.setattr( + aot, "jpype_support_jar", lambda: Path(env["cp"]).parent / "org.jpype.jar" + ) + flags = _plan(env) + assert flags[0].startswith("-XX:AOTCacheOutput=") + assert not final.exists() + + +@pytest.mark.parametrize( + "change", + [ + dict(version="2.0"), + dict(sha="other-jars"), + ], +) +def test_other_version_or_jars_get_another_cache_file(env, monkeypatch, change): + final, _ = _good_cache(env, monkeypatch) + flags = _plan(env, **change) + assert flags[0].startswith("-XX:AOTCacheOutput=") + assert final.exists() # the old one is left alone + + +def test_touched_jar_changes_the_key(env, monkeypatch): + final, _ = _good_cache(env, monkeypatch) + jar = Path(env["cp"]) + st = jar.stat() + os.utime(jar, ns=(st.st_atime_ns, st.st_mtime_ns + 5_000_000_000)) + assert _plan(env)[0].startswith("-XX:AOTCacheOutput=") + + +def test_user_flags_win(env): + for flag in ( + "-XX:AOTCache=/x.aot", + "-XX:AOTMode=off", + "-Xshare:off", + "-XX:SharedArchiveFile=/x", + ): + assert _plan(env, args=[flag]) == [] + assert not env["dir"].exists() + + +def test_unsupported_or_unreadable_jre_falls_back(tmp_path, env): + old = _fake_jre(tmp_path / "old", "21.0.5") + assert aot.jvm_flags_for([], old, env["cp"], "1", "s") == [] + assert aot.jvm_flags_for([], tmp_path / "nowhere", env["cp"], "1", "s") == [] + assert not env["dir"].exists() + + +def test_missing_jpype_jar_falls_back(env, monkeypatch): + monkeypatch.setattr(aot, "jpype_support_jar", lambda: None) + assert _plan(env) == [] + + +def test_unwritable_location_never_raises(env, tmp_path, monkeypatch): + blocker = tmp_path / "cache" + blocker.write_text("a file where the directory should be") + assert _plan(env) == [] + blocker.unlink() + monkeypatch.setattr( + aot.Path, "mkdir", lambda *a, **k: (_ for _ in ()).throw(PermissionError("ro")) + ) + assert _plan(env) == [] + + +def test_lock_failure_is_a_normal_start(env, monkeypatch): + def boom(*a, **k): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(aot.os, "open", boom) + assert _plan(env) == [] + + +def test_override_path_is_used_and_keyed(env, tmp_path, monkeypatch): + target = tmp_path / "mine" / "c.aot" + monkeypatch.setenv(aot.ENV_CACHE, str(target)) + flags = _plan(env) + assert flags[0].startswith("-XX:AOTCacheOutput=") and str(target.parent) in flags[0] + assert (target.parent / "c.aot.lock").exists() + # a cache recorded for another key is not trusted + target.write_bytes(b"x") + (target.parent / "c.aot.json").write_text( + json.dumps( + {"key": "other", "size": 1, "sha256": hashlib.sha256(b"x").hexdigest()} + ) + ) + assert not aot._usable_final(aot._Paths(target, "this-key")) + assert aot._usable_final(aot._Paths(target, "other")) + + +def test_release_training_removes_lock_and_output(env): + flag = _plan(env)[0] + tmp = Path(flag.split("=", 1)[1]) + tmp.write_bytes(b"x") + aot.release_training([flag]) + assert not tmp.exists() and not list(env["dir"].glob("*.lock")) + + +def test_pid_alive(): + assert aot._pid_alive(os.getpid()) + assert not aot._pid_alive(0) + + +def test_jvm_aot_flags_off_by_default_and_never_raises(monkeypatch): + monkeypatch.delenv(aot.ENV_ENABLE, raising=False) + monkeypatch.setattr( + aot, "jvm_flags_for", lambda *a, **k: pytest.fail("called while off") + ) + assert jvm._aot_flags(None, [], "cp") == [] + monkeypatch.setattr( + aot, "jvm_flags_for", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("x")) + ) + monkeypatch.setattr(jvm, "_extract_runtime_resource", lambda n: Path(".")) + monkeypatch.setattr(jvm, "jar_fingerprint", lambda **k: {"sha256": "s"}) + assert jvm._aot_flags(True, [], "cp") == []