diff --git a/CHANGELOG.md b/CHANGELOG.md index 8840bb7..1696982 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## 0.7.1 (2026-10-23) + +Faster opens over http. No placement change, no unit table change, no API change. + +- The documents a call needs from one directory are read at once instead of in turn. An + http open of a zarr v2 level made its eight requests one after another (five probe keys, + the level's `.zattrs`, the parent's `zarr.json` and `.zattrs`); now they go in three + rounds. At 10 ms an answer, `wa.open` took 98 ms and takes 52 (cellmap-flow's + measurement). A server that never answers still costs one key's attempts, not five. + ## 0.7.0 (2026-10-23) Opens time out. No placement change and no unit table change. diff --git a/docs/spec.md b/docs/spec.md index c87ece8..5354659 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -490,7 +490,9 @@ this document and that one differ, fix one of them. 5. **Stores are read through tensorstore's kvstore only**, never zarr-python: `zarr.json`, `.zarray`, `.zattrs`, `.zgroup`, `attributes.json`, `info`. Format detection probes keys - in this order: `zarr.json`, `.zarray`, `.zgroup`, `info`, `attributes.json`; when none is + in this order: `zarr.json`, `.zarray`, `.zgroup`, `info`, `attributes.json`, the reads + begun all at once and the order deciding which document wins (0.7.1: one round trip + where an http open made five in a row, cellmap-flow's measurement); when none is found, the parent is probed for `info` and the last path segment is taken as a precomputed scale key (mesh-n-bone's case), then `s0` is probed so a group with no document of its own (an N5 directory holding `s0, s1, ...`) is still a group. The diff --git a/pyproject.toml b/pyproject.toml index a374b66..515a422 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "worldarray" -version = "0.7.0" +version = "0.7.1" description = "Read zarr, N5 and Neuroglancer precomputed volumes in world coordinates through tensorstore: correct voxel placement, pyramid levels, padded region reads, resampling. numpy + tensorstore only." readme = "README.md" license = { file = "LICENSE" } diff --git a/src/worldarray/__init__.py b/src/worldarray/__init__.py index 253dcf5..4e443aa 100644 --- a/src/worldarray/__init__.py +++ b/src/worldarray/__init__.py @@ -28,7 +28,7 @@ parse_chunk_key, ) -__version__ = "0.7.0" +__version__ = "0.7.1" __all__ = [ "ArrayMeta", diff --git a/src/worldarray/store.py b/src/worldarray/store.py index 60acce4..d0bde2f 100644 --- a/src/worldarray/store.py +++ b/src/worldarray/store.py @@ -254,18 +254,48 @@ def _metadata_kvstore(spec_json: str, gcs_url: str | None) -> ts.KvStore: return _wait(lambda: ts.KvStore.open(metadata_spec(spec)), spec, _where(spec), "open") -def _read_key(spec: dict, key: str) -> bytes | None: - kv = _metadata_kvstore( +def _metadata_kv(spec: dict) -> ts.KvStore: + return _metadata_kvstore( json.dumps(spec, sort_keys=True), os.environ.get("TENSORSTORE_GCS_HTTP_URL") ) - result = _wait(lambda: kv.read(key), spec, _where(spec, key), "read") - return None if result.state == "missing" else bytes(result.value) + + +def _read_keys(spec: dict, keys: Sequence[str]) -> dict[str, bytes | None | BaseException]: + """``keys`` under one way in, their reads begun at once: one round trip, not one per key + (an http open probes five). Each value is the document, None when it is missing, or the + error its read raised; after an error other than a refusal the rest are cancelled and + left out, so a server that never answers costs one key's attempts, not five.""" + kv = _metadata_kv(spec) + started = {key: kv.read(key) for key in keys} + out: dict[str, bytes | None | BaseException] = {} + for key in keys: + first = [started[key]] + + def start(key=key, first=first): + return first.pop() if first else kv.read(key) + + try: + result = _wait(start, spec, _where(spec, key), "read") + except Exception as e: # noqa: BLE001 the caller decides what a failed key means + out[key] = e + if not is_refused(e): + for future in started.values(): + future.cancel() + return out + continue + out[key] = None if result.state == "missing" else bytes(result.value) + return out + + +def _read_key(spec: dict, key: str) -> bytes | None: + value = _read_keys(spec, [key])[key] + if isinstance(value, BaseException): + raise value + return value def _write_key(spec: dict, key: str, data: bytes) -> None: - kv = _metadata_kvstore( - json.dumps(spec, sort_keys=True), os.environ.get("TENSORSTORE_GCS_HTTP_URL") - ) + kv = _metadata_kv(spec) _wait(lambda: kv.write(key, data), spec, _where(spec, key), "write") @@ -292,20 +322,21 @@ def write_json(self, key: str, doc: dict) -> None: self.write(key, (json.dumps(doc, indent=2) + "\n").encode()) def read_many(self, keys: Sequence[str]) -> dict[str, bytes | None]: - """All of ``keys`` under one way in. A bucket with listing off answers a refusal for a - missing key, so a refusal on one key counts as missing when another key was read - under the same way; only when every key is refused is the next way tried.""" + """All of ``keys`` under one way in, read at once (``_read_keys``). A bucket with + listing off answers a refusal for a missing key, so a refusal on one key counts as + missing when another key was read under the same way; only when every key is refused + is the next way tried.""" def attempt(spec: dict) -> dict[str, bytes | None]: found: dict[str, bytes | None] = {} refusals: dict[str, BaseException] = {} - for key in keys: - try: - found[key] = _read_key(spec, key) - except Exception as e: - if not is_refused(e): - raise - refusals[key] = e + for key, value in _read_keys(spec, keys).items(): + if isinstance(value, BaseException): + if not is_refused(value): + raise value + refusals[key] = value + else: + found[key] = value if refusals and len(refusals) == len(keys): raise Refused(next(iter(refusals.values())).args[0] if refusals else "refused") for key in refusals: diff --git a/tests/test_access.py b/tests/test_access.py index 9d2a8ff..615fd19 100644 --- a/tests/test_access.py +++ b/tests/test_access.py @@ -2,7 +2,9 @@ from __future__ import annotations +import functools import gc +import http.server import logging import os import socket @@ -56,6 +58,21 @@ def attempt(spec): store.with_access(specs, attempt) +def _per_key(read_key): + """A ``store._read_keys`` stand-in built from a one-key fake.""" + + def read_keys(spec, keys): + out = {} + for key in keys: + try: + out[key] = read_key(spec, key) + except Exception as e: # noqa: BLE001 + out[key] = e + return out + + return read_keys + + def test_a_refused_probe_key_counts_as_missing_when_another_key_was_read(monkeypatch): present = {"zarr.json": b'{"node_type": "array"}'} @@ -66,7 +83,7 @@ def fake_read_key(spec, key): return present[key] raise RuntimeError("PERMISSION_DENIED: listing is off, so a missing key is a 403") - monkeypatch.setattr(store, "_read_key", fake_read_key) + monkeypatch.setattr(store, "_read_keys", _per_key(fake_read_key)) loc = store.Location("s3://bucket/vol.zarr/s0") found = loc.read_many(store.PROBE_KEYS) assert found["zarr.json"] == present["zarr.json"] @@ -79,7 +96,7 @@ def fake_read_key(spec, key): raise RuntimeError("PERMISSION_DENIED") return b"{}" if key == ".zarray" else None - monkeypatch.setattr(store, "_read_key", fake_read_key) + monkeypatch.setattr(store, "_read_keys", _per_key(fake_read_key)) found = store.Location("s3://bucket/vol.zarr/s0").read_many(store.PROBE_KEYS) assert found[".zarray"] == b"{}" and found["zarr.json"] is None @@ -358,3 +375,27 @@ def test_a_server_that_never_answers_times_out_the_open(silent_server): with pytest.raises(TimeoutError, match="127.0.0.1.*2 attempts"): open_it(f"{silent_server}/vol.zarr/s0") assert time.monotonic() - start < 10 + + +def test_an_http_open_probes_its_keys_at_once(tmp_path): + """Five probe keys, one round trip: each answer here takes 0.3 s.""" + write_zarr2(tmp_path / "vol.zarr" / "s0", pattern((4, 4, 4))) + + class Slow(http.server.SimpleHTTPRequestHandler): + def do_GET(self): + time.sleep(0.3) + super().do_GET() + + def log_message(self, *args): + pass + + handler = functools.partial(Slow, directory=str(tmp_path)) + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + start = time.monotonic() + node = wa.detect(f"http://127.0.0.1:{server.server_address[1]}/vol.zarr/s0") + assert node.format == "zarr2" and node.kind == "array" + assert time.monotonic() - start < 1.0 # in turn: 1.5 s + finally: + server.shutdown() diff --git a/uv.lock b/uv.lock index 48f0ffe..9b5dd72 100644 --- a/uv.lock +++ b/uv.lock @@ -1247,7 +1247,7 @@ wheels = [ [[package]] name = "worldarray" -version = "0.7.0" +version = "0.7.1" source = { editable = "." } dependencies = [ { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },