Run cache, write path: with CYMSIM_CACHE_KEY a computed run is uploaded unless the entry holds at least as many steps; per-segment device, jax, commit and host in the entry metadata; failed uploads are logged, not raised
3 changed files+242−17
README.mdmodified+1−1View file
@@ -26,7 +26,7 @@ cymsim cache-key my_run.yaml --show # the run's cache key (inde
2626 cymsim run my_run.yaml --cache https://artifact-cache.tempory.net/api/apps/cymsim # take the run from the run cache, or continue a shorter cached run
2727 ```
2828
29-With `--cache URL` (or `CYMSIM_CACHE_URL` set), a run is looked up in a run cache under its key (`cymsim cache-key`), which leaves out the duration and the output naming. The cache holds the longest run of each key so far as a `.cymsim` file without the acceleration record: a run it covers is cut from it without building the model, and a longer run continues from it (`--resume`), identical to an uninterrupted run on the same device. Downloads are kept in `~/.cache/cymsim` (`CYMSIM_CACHE_DIR`) and revalidated by ETag. See [cache.py](src/cymsim/cache.py).
29+With `--cache URL` (or `CYMSIM_CACHE_URL` set), a run is looked up in a run cache under its key (`cymsim cache-key`), which leaves out the duration and the output naming. The cache holds the longest run of each key so far as a `.cymsim` file without the acceleration record: a run it covers is cut from it without building the model, and a longer run continues from it (`--resume`), identical to an uninterrupted run on the same device. Downloads are kept in `~/.cache/cymsim` (`CYMSIM_CACHE_DIR`) and revalidated by ETag. With the cache app's passcode in `CYMSIM_CACHE_KEY`, a run that was computed (from rest or continued) is uploaded, unless the entry meanwhile holds at least as much; without it the cache is read-only. The entry's metadata records the device, jax version, commit and host of each segment of integration. See [cache.py](src/cymsim/cache.py).
3030
3131 A `.cymsim` file is an HDF5 file holding the configuration, the mesh, the listener signals, displacement frames at a low rate, optionally the normal acceleration of every node at the audio rate, and the integrator state; the layout is specified in [docs/format.md](docs/format.md). It is meant for taking the shell's motion into other programs (other listeners, rooms, acoustic solvers) and is the file the web companion cymsimweb saves and loads. A partial file written during a run is a consistent prefix of the complete one.
3232
src/cymsim/cache.pymodified+137−7View file
@@ -14,15 +14,26 @@ An entry is the .cymsim file of the longest run of its key so far, without the a
1414 record. ``cached_run`` reads it at ``<url>/entries/<key>`` (public): a run it covers is cut from
1515 it, a run it does not cover continues from it, and without an entry the run starts from rest.
1616 Downloads are kept in CYMSIM_CACHE_DIR (default ~/.cache/cymsim) and revalidated by ETag.
17+
18+With the app's passcode (CYMSIM_CACHE_KEY), a run that was computed rather than cut is uploaded,
19+unless the entry meanwhile holds at least as many steps (checked just before the upload; the
20+worker has no conditional write). The entry's metadata records, per segment of integration, the
21+device, jax version, cymsim commit and host; continuing on a different device than the last
22+segment is allowed but logged, since float32 runs differ bitwise between devices. A failed upload
23+is logged and does not fail the run.
1724 """
1825 from __future__ import annotations
1926
2027 import dataclasses
28+import datetime as _dt
2129 import hashlib
2230 import json
2331 import logging
2432 import os
33+import socket
34+import subprocess
2535 import urllib.error
36+import urllib.parse
2637 import urllib.request
2738 from pathlib import Path
2839
@@ -31,6 +42,9 @@ from .config import CONTINUATION_FIELDS, SimulationConfig
3142 log = logging.getLogger("cymsim")
3243
3344 MODEL_VERSION = 1
45+MAX_ENTRY_BYTES = 100 * 1024 * 1024 # the artifact-cache limit
46+MAX_META_BYTES = 4096
47+MAX_SEGMENTS = 20 # segments of history kept in an entry's metadata
3448
3549
3650 def _non_default(obj):
@@ -70,14 +84,118 @@ def cache_dir() -> Path:
7084 return Path(os.environ.get("CYMSIM_CACHE_DIR") or Path.home() / ".cache" / "cymsim")
7185
7286
87+def _request(url: str, method: str = "GET", data: bytes | None = None, headers: dict | None = None):
88+ from . import __version__
89+ # (Cloudflare refuses urllib's default User-Agent)
90+ return urllib.request.Request(url, data=data, method=method,
91+ headers={"User-Agent": f"cymsim/{__version__}", **(headers or {})})
92+
93+
94+def entry_meta(url: str, key: str) -> dict | None:
95+ """The metadata cymsim stored with the entry ``key`` ({} if it stored none), or None if there
96+ is no entry."""
97+ try:
98+ with urllib.request.urlopen(_request(f"{url.rstrip('/')}/meta/{key}"), timeout=60) as resp:
99+ return json.loads(resp.read()).get("meta") or {}
100+ except urllib.error.HTTPError as e:
101+ if e.code == 404:
102+ return None
103+ raise
104+
105+
106+def device() -> str:
107+ import jax
108+ d = jax.devices()[0]
109+ return f"{d.platform}:{getattr(d, 'device_kind', '?')}"
110+
111+
112+def code_version() -> dict:
113+ """The git commit of the cymsim checkout this package runs from, if it is one."""
114+ root = Path(__file__).resolve().parents[2]
115+ try:
116+ git = lambda *a: subprocess.run(["git", "-C", str(root), *a], capture_output=True, text=True,
117+ timeout=10, check=True).stdout.strip()
118+ return {"commit": git("rev-parse", "HEAD"), "dirty": bool(git("status", "--porcelain", "--untracked-files=no"))}
119+ except (OSError, subprocess.SubprocessError):
120+ return {"commit": None, "dirty": None}
121+
122+
123+def _segment(n_from: int, n_to: int) -> dict:
124+ import jax
125+
126+ from . import __version__
127+ return {"from": n_from, "to": n_to, "device": device(), "jax": jax.__version__, "cymsim": __version__,
128+ **code_version(), "host": socket.gethostname(),
129+ "date": _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")}
130+
131+
132+def _meta_header(result, segments: list[dict]) -> str:
133+ meta = {"model_version": MODEL_VERSION, "n_steps_done": int(result.n_steps_done),
134+ "seconds": len(result.pressure_rayleigh) / result.fs, "complete": bool(result.complete),
135+ "fs": int(result.fs), "substeps": int(result.substeps), "n_nodes": int(result.n_nodes),
136+ "segments": segments[-MAX_SEGMENTS:], "config": key_fields(result.config)}
137+ text = json.dumps(meta, separators=(",", ":"))
138+ while len(text.encode()) > MAX_META_BYTES and (meta.get("config") is not None or len(meta["segments"]) > 1):
139+ if meta.get("config") is not None:
140+ meta["config"] = None # the file holds the configuration anyway
141+ else:
142+ meta["segments"] = meta["segments"][1:]
143+ text = json.dumps(meta, separators=(",", ":"))
144+ return urllib.parse.quote(text)
145+
146+
147+def publish(result, url: str, key: str, passcode: str, segments: list[dict], file: Path | None = None,
148+ directory: Path | None = None) -> bool:
149+ """Upload ``result`` as the entry ``key`` unless the entry holds at least as many steps.
150+ ``file`` is the result's .cymsim file if already written. The upload becomes the local copy.
151+ Returns whether it was uploaded; failures are logged, not raised."""
152+ from .cymsim_file import save_cymsim
153+
154+ try:
155+ current = entry_meta(url, key)
156+ if current is not None and int(current.get("n_steps_done", 0)) >= result.n_steps_done:
157+ log.info("cache: the entry %s already holds %d steps (this run %d); not replacing it",
158+ key, int(current.get("n_steps_done", 0)), result.n_steps_done)
159+ return False
160+ path = (directory or cache_dir()) / f"{key}.cymsim"
161+ path.parent.mkdir(parents=True, exist_ok=True)
162+ tmp = path.with_name(path.name + ".part")
163+ if file is None:
164+ save_cymsim(result, tmp)
165+ data = tmp.read_bytes()
166+ else:
167+ data = Path(file).read_bytes()
168+ tmp.write_bytes(data)
169+ if len(data) > MAX_ENTRY_BYTES:
170+ log.warning("cache: %s is %.0f MB, above the %d MB limit; not uploaded", key, len(data) / 1e6,
171+ MAX_ENTRY_BYTES >> 20)
172+ tmp.unlink()
173+ return False
174+ req = _request(f"{url.rstrip('/')}/entries/{key}", "PUT", data,
175+ {"X-Passcode": passcode, "Content-Type": "application/x-hdf5",
176+ "X-Artifact-Meta": _meta_header(result, segments)})
177+ with urllib.request.urlopen(req, timeout=600) as resp:
178+ etag = json.loads(resp.read()).get("etag")
179+ tmp.replace(path)
180+ etag_path = path.with_name(path.name + ".etag")
181+ if etag:
182+ etag_path.write_text(f'"{etag}"')
183+ else:
184+ etag_path.unlink(missing_ok=True)
185+ log.info("cache: uploaded %s (%.3f s, %.1f MB)", key, len(result.pressure_rayleigh) / result.fs, len(data) / 1e6)
186+ return True
187+ except (urllib.error.URLError, OSError) as e:
188+ detail = e.read().decode(errors="replace") if isinstance(e, urllib.error.HTTPError) else ""
189+ log.error("cache: upload of %s failed: %s %s", key, e, detail)
190+ return False
191+
192+
73193 def fetch(url: str, key: str, directory: Path | None = None) -> Path | None:
74194 """The local copy of the entry ``key`` at the cache ``url``, downloaded or revalidated, or None
75195 if there is no such entry. If the cache cannot be reached, an earlier local copy is used."""
76196 path = (directory or cache_dir()) / f"{key}.cymsim"
77197 etag_path = path.with_name(path.name + ".etag")
78- from . import __version__
79- # (Cloudflare refuses urllib's default User-Agent)
80- req = urllib.request.Request(f"{url.rstrip('/')}/entries/{key}", headers={"User-Agent": f"cymsim/{__version__}"})
198+ req = _request(f"{url.rstrip('/')}/entries/{key}")
81199 if path.exists() and etag_path.exists():
82200 req.add_header("If-None-Match", etag_path.read_text())
83201 try:
@@ -112,9 +230,11 @@ def fetch(url: str, key: str, directory: Path | None = None) -> Path | None:
112230
113231
114232 def cached_run(config: SimulationConfig, url: str, save_path: str | None = None,
115- save_every: float | None = None, directory: Path | None = None):
116- """The run of ``config``, taken from or continued from the cache at ``url`` where possible
117- (see the module docstring). Nothing is uploaded."""
233+ save_every: float | None = None, directory: Path | None = None, passcode: str | None = None):
234+ """The run of ``config``, taken from or continued from the cache at ``url`` where possible,
235+ and uploaded if it was computed and a passcode is given (``passcode`` or CYMSIM_CACHE_KEY);
236+ see the module docstring."""
237+ passcode = passcode or os.environ.get("CYMSIM_CACHE_KEY") or None
118238 from .cymsim_file import load_cymsim, save_cymsim
119239 from .simulate import Simulator, cut_run
120240
@@ -139,4 +259,14 @@ def cached_run(config: SimulationConfig, url: str, save_path: str | None = None,
139259 save_cymsim(cut, save_path)
140260 return cut
141261 log.info("cache: %s holds %.3f s; continuing from there", key, len(prev.pressure_rayleigh) / prev.fs)
142- return Simulator(config).run(save_path=save_path, save_every=save_every, resume=prev)
262+ segments = []
263+ if prev is not None:
264+ segments = list((entry_meta(url, key) or {}).get("segments") or [])
265+ if segments and segments[-1].get("to") == prev.n_steps_done and segments[-1].get("device") != device():
266+ log.warning("cache: the cached run was integrated on %s and continues on %s; float32 runs "
267+ "differ bitwise between devices", segments[-1].get("device"), device())
268+ result = Simulator(config).run(save_path=save_path, save_every=save_every, resume=prev)
269+ if passcode:
270+ segments.append(_segment(prev.n_steps_done if prev is not None else 0, result.n_steps_done))
271+ publish(result, url, key, passcode, segments, file=save_path, directory=directory)
272+ return result
tests/test_cache_run.pymodified+104−9View file
@@ -1,15 +1,17 @@
11 """Reading runs from a cache: a local HTTP server stands in for the artifact-cache app."""
22 import copy
33 import hashlib
4+import json
45 import logging
56 import threading
67 import urllib.error
8+import urllib.parse
79 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
810
911 import pytest
1012 from test_resume import _assert_same_files
1113
12-from cymsim.cache import cache_key, cached_run, fetch
14+from cymsim.cache import cache_key, cached_run, entry_meta, fetch, key_fields, publish
1315 from cymsim.cli import main
1416 from cymsim.cymsim_file import load_cymsim
1517 from cymsim.simulate import Simulator
@@ -17,15 +19,42 @@ from cymsim.simulate import Simulator
1719 PREFIX = "/api/apps/cymsim/entries/"
1820
1921
22+PASSCODE = "secret"
23+
24+
2025 class FakeCache:
21- """GET <prefix><key> with ETag revalidation, as the artifact-cache worker serves it."""
26+ """The artifact-cache worker's GET (with ETag revalidation), GET meta and PUT (with passcode)."""
2227
2328 def __init__(self):
24- self.entries, self.requests = {}, []
29+ self.entries, self.metas, self.requests = {}, {}, []
2530 cache = self
2631
2732 class Handler(BaseHTTPRequestHandler):
33+ def do_PUT(self):
34+ key = self.path[len(PREFIX):]
35+ data = self.rfile.read(int(self.headers["Content-Length"]))
36+ if self.headers.get("X-Passcode") != PASSCODE:
37+ status, body = 401, b'{"error":"Invalid passcode"}'
38+ else:
39+ cache.entries[key] = data
40+ cache.metas[key] = json.loads(urllib.parse.unquote(self.headers.get("X-Artifact-Meta") or "null"))
41+ status, body = 201, json.dumps({"etag": hashlib.md5(data).hexdigest()}).encode()
42+ self.send_response(status)
43+ self.send_header("Content-Length", str(len(body)))
44+ self.end_headers()
45+ self.wfile.write(body)
46+ cache.requests.append(("PUT", key, status))
47+
2848 def do_GET(self):
49+ meta_prefix = PREFIX.replace("/entries/", "/meta/")
50+ if self.path.startswith(meta_prefix):
51+ key = self.path[len(meta_prefix):]
52+ body = json.dumps({"key": key, "meta": cache.metas.get(key)}).encode()
53+ self.send_response(200 if key in cache.entries else 404)
54+ self.send_header("Content-Length", str(len(body)))
55+ self.end_headers()
56+ self.wfile.write(body)
57+ return
2958 key = self.path[len(PREFIX):] if self.path.startswith(PREFIX) else None
3059 data = cache.entries.get(key)
3160 if data is None:
@@ -42,7 +71,7 @@ class FakeCache:
4271 self.end_headers()
4372 if status == 200:
4473 self.wfile.write(data)
45- cache.requests.append((key, status))
74+ cache.requests.append(("GET", key, status))
4675
4776 def log_message(self, *args):
4877 pass
@@ -51,12 +80,14 @@ class FakeCache:
5180 self.url = f"http://127.0.0.1:{self.server.server_address[1]}{PREFIX[:-len('entries/')]}"
5281 threading.Thread(target=self.server.serve_forever, daemon=True).start()
5382
54- def put(self, cfg, path):
83+ def put(self, cfg, path, meta=None):
5584 self.entries[cache_key(cfg)] = path.read_bytes()
85+ self.metas[cache_key(cfg)] = meta
5686
5787
5888 @pytest.fixture
59-def cache():
89+def cache(monkeypatch):
90+ monkeypatch.delenv("CYMSIM_CACHE_KEY", raising=False)
6091 c = FakeCache()
6192 yield c
6293 c.server.shutdown()
@@ -85,7 +116,7 @@ def test_without_an_entry_the_run_starts_from_rest(coarse_config, runs, cache, t
85116 out = tmp_path / "out.cymsim"
86117 cached_run(_config(coarse_config, 0.05), cache.url, save_path=str(out), directory=tmp_path / "c")
87118 _assert_same_files(out, runs[0.05])
88- assert cache.requests == [(cache_key(_config(coarse_config, 0.05)), 404)]
119+ assert cache.requests == [("GET", cache_key(_config(coarse_config, 0.05)), 404)]
89120
90121
91122 def test_a_covering_entry_is_cut_without_building_the_model(coarse_config, runs, cache, tmp_path, monkeypatch):
@@ -137,7 +168,7 @@ def test_local_copies(coarse_config, runs, cache, tmp_path):
137168 p = fetch(cache.url, key, d)
138169 assert p == d / f"{key}.cymsim" and p.read_bytes() == runs[0.02].read_bytes()
139170 assert fetch(cache.url, key, d) == p
140- assert [s for _, s in cache.requests] == [200, 304]
171+ assert [s for _, _, s in cache.requests] == [200, 304]
141172 cache.entries[key] = runs[0.05].read_bytes() # replaced remotely: downloaded again
142173 assert fetch(cache.url, key, d).read_bytes() == runs[0.05].read_bytes()
143174 dead = "http://127.0.0.1:9/api/apps/cymsim" # unreachable: the local copy serves
@@ -169,5 +200,69 @@ def test_cli_cache(coarse_config, runs, cache, tmp_path, monkeypatch):
169200 monkeypatch.setenv("CYMSIM_CACHE_URL", cache.url)
170201 assert main(["run", str(ypath), "-o", str(out), "--quiet"]) == 0
171202 assert main(["run", str(ypath), "--no-cache", "-o", str(out), "--quiet"]) == 0
172- assert [s for _, s in cache.requests] == [200, 304]
203+ assert [s for _, _, s in cache.requests] == [200, 304]
173204 assert main(["run", str(ypath), "--cache", cache.url, "--resume", str(out), "--quiet"]) == 2
205+
206+
207+def _load_entry(cache, key, path):
208+ path.write_bytes(cache.entries[key])
209+ return path
210+
211+
212+def test_a_computed_run_is_uploaded(coarse_config, runs, cache, tmp_path):
213+ cfg, d = _config(coarse_config, 0.02), tmp_path / "c"
214+ key = cache_key(cfg)
215+ cached_run(cfg, cache.url, directory=d, passcode=PASSCODE)
216+ _assert_same_files(_load_entry(cache, key, tmp_path / "e.cymsim"), runs[0.02])
217+ meta = entry_meta(cache.url, key)
218+ assert meta["n_steps_done"] == load_cymsim(runs[0.02]).n_steps_done and meta["complete"]
219+ (seg,) = meta["segments"]
220+ assert seg["from"] == 0 and seg["to"] == meta["n_steps_done"] and seg["device"].startswith("cpu")
221+ assert meta["config"] == json.loads(json.dumps(key_fields(cfg)))
222+ # the upload is the local copy: the next lookup is a 304 and uploads nothing
223+ res = cached_run(cfg, cache.url, directory=d, passcode=PASSCODE)
224+ assert res.complete and cache.requests[-1] == ("GET", key, 304)
225+ assert [r for r in cache.requests if r[0] == "PUT"] == [("PUT", key, 201)]
226+
227+
228+def test_a_continued_run_replaces_the_shorter_entry(coarse_config, runs, cache, tmp_path, caplog):
229+ cfg = _config(coarse_config, 0.05)
230+ key = cache_key(cfg)
231+ n = load_cymsim(runs[0.02]).n_steps_done
232+ cache.put(cfg, runs[0.02], {"n_steps_done": n, "segments": [{"from": 0, "to": n, "device": "gpu:Fake"}]})
233+ with caplog.at_level(logging.WARNING, logger="cymsim"):
234+ cached_run(cfg, cache.url, directory=tmp_path / "c", passcode=PASSCODE)
235+ assert "integrated on gpu:Fake" in caplog.text
236+ _assert_same_files(_load_entry(cache, key, tmp_path / "e.cymsim"), runs[0.05])
237+ segs = cache.metas[key]["segments"]
238+ assert [(g["from"], g["to"]) for g in segs] == [(0, n), (n, load_cymsim(runs[0.05]).n_steps_done)]
239+
240+
241+def test_reading_without_a_passcode_uploads_nothing(coarse_config, cache, tmp_path):
242+ cached_run(_config(coarse_config, 0.02), cache.url, directory=tmp_path / "c")
243+ assert not cache.entries and all(r[0] == "GET" for r in cache.requests)
244+
245+
246+def test_a_cut_run_is_not_uploaded(coarse_config, runs, cache, tmp_path):
247+ cfg = _config(coarse_config, 0.02)
248+ cache.put(cfg, runs[0.05], {"n_steps_done": load_cymsim(runs[0.05]).n_steps_done})
249+ cached_run(cfg, cache.url, directory=tmp_path / "c", passcode=PASSCODE)
250+ assert all(r[0] == "GET" for r in cache.requests)
251+
252+
253+def test_a_longer_entry_is_never_replaced(coarse_config, runs, cache, tmp_path):
254+ cfg = _config(coarse_config, 0.02)
255+ long_steps = load_cymsim(runs[0.05]).n_steps_done
256+ cache.put(cfg, runs[0.05], {"n_steps_done": long_steps})
257+ assert not publish(load_cymsim(runs[0.02]), cache.url, cache_key(cfg), PASSCODE, [], directory=tmp_path / "c")
258+ assert all(r[0] == "GET" for r in cache.requests)
259+ assert cache.entries[cache_key(cfg)] == runs[0.05].read_bytes()
260+
261+
262+def test_a_failed_upload_keeps_the_run(coarse_config, runs, cache, tmp_path, caplog):
263+ cfg, out = _config(coarse_config, 0.02), tmp_path / "out.cymsim"
264+ with caplog.at_level(logging.ERROR, logger="cymsim"):
265+ res = cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c", passcode="wrong")
266+ assert res.complete and "upload of" in caplog.text and "Invalid passcode" in caplog.text
267+ assert ("PUT", cache_key(cfg), 401) in cache.requests and not cache.entries
268+ _assert_same_files(out, runs[0.02])