From a1a432f4271aecf0de04e4fe9270b1570f3990c8 Mon Sep 17 00:00:00 2001 From: David Ackerman Date: Fri, 23 Oct 2026 17:10:36 -0400 Subject: [PATCH 1/2] Write without fsync or OS locks, size file I/O apart, time out the create (0.8.0) Measured 2026-10-23 against zarr-python on real CellMap data: tensorstore's fsync cost 6.5 ms a chunk on local disk and nothing on NFS; its "os" file locking about 10 ms a chunk on NFS; and one file I/O thread made a one-thread worker write a block's chunk files one NFS round trip after another. Contexts the package makes now set file_io_sync false and file_io_locking "none" (fsync=True, locking="os" restore them; whole-unit writes stay atomic), and the default context's file I/O pool is at least 8 while decoding stays at thread_budget(). capped_context gains file_io=. From one-thread workers: an NFS label block 159 to 66.5 ms (zarr-python 2: 102), a local one 51.5 to 4.1 ms. The create open is timed out under retries too (cellmap-analyze's question); a retry that finds the array its own timed-out attempt made adopts it after the usual comparison, and that open keeps OS locking so create-if-absent stays atomic. tensorstore>=0.1.68, the first release with file_io_locking. Claude-Session: https://claude.ai/code/session_0141wgF8mkGbjGfBFKfKhp64 --- CHANGELOG.md | 26 +++++++++ README.md | 8 ++- docs/spec.md | 65 ++++++++++++++++----- pyproject.toml | 4 +- src/worldarray/__init__.py | 2 +- src/worldarray/context.py | 116 +++++++++++++++++++++++++++---------- src/worldarray/write.py | 20 ++++++- tests/test_access.py | 33 +++++++++-- tests/test_write.py | 30 ++++++++++ uv.lock | 4 +- 10 files changed, 251 insertions(+), 57 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1696982..691cb35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,31 @@ # Changelog +## 0.8.0 (unreleased) + +Faster writes from the contexts the package makes, and the create open timed out. No +placement change and no unit table change. **Requires tensorstore 0.1.68 or later** (was +0.1.60), the first with the `file_io_locking` setting. + +- `shared_context` and `capped_context` write without fsync (`file_io_sync: false`) and + without OS file locks (`file_io_locking` mode `"none"`); `fsync=True` and `locking="os"` + restore tensorstore's defaults. A whole-unit write stays atomic, a temporary file renamed + into place; only concurrent read-modify-writes of one unit (`partial=True`) lose their + guarantee, which was never given (Behaviour 13). fsync cost 6.5 ms a chunk on a local + disk and nothing on NFS; the locks cost about 10 ms a chunk on NFS. +- The default context's file I/O pool is `thread_budget()` or 8, whichever is larger; + decoding stays at `thread_budget()`. File I/O threads wait on storage rather than + compute, and one of them wrote a block's chunk files one NFS round trip after another. + `capped_context(file_io=)` sizes it apart from `concurrency` (default unchanged). +- From one-thread workers through the default context, against 0.7: a 64^3 label chunk to + NFS 23 to 12.2 ms (zarr-python 2: 13.1), an 8-chunk label block 159 to 66.5 ms + (zarr-python 2: 102), an 8-chunk EM block 330 to 202 ms; to a local disk a label block + 51.5 to 4.1 ms. Through `capped_context(1, network=False)` the NFS label block went 166 + to 91 ms; `file_io=8` gives it the rest. +- The open that creates an array is timed out and retried under `retries` (Behaviour 6). A + retry that finds the array made by an attempt of its own that timed out adopts it after + the usual comparison, even under `exists="error"`. That open keeps OS locking, so + create-if-absent stays atomic. + ## 0.7.1 (2026-10-23) Faster opens over http. No placement change, no unit table change, no API change. diff --git a/README.md b/README.md index cd8d218..78e6203 100644 --- a/README.md +++ b/README.md @@ -150,9 +150,11 @@ wa.metadata(path, level=None, units=None, open=True, retries=None) -> ArrayMeta wa.detect(path, missing_ok=False, retries=None) -> Node # format, array or group, raw attrs wa.open_tensorstore(path, level=None, context=None, retries=None) # retries= times out and retries document reads and the open, as it does data reads -wa.shared_context(cache_bytes=0, threads=None) ; wa.capped_context(cache_bytes=0, concurrency=1, network=True) ; wa.Retry(...) -wa.thread_budget() # the shared context's default thread limit: OMP_NUM_THREADS, the job's slots, or CPUs; - # threads=N limits it to N, threads=0 leaves tensorstore's own pool sizes +wa.shared_context(cache_bytes=0, threads=None, fsync=False, locking="none") ; wa.Retry(...) +wa.capped_context(cache_bytes=0, concurrency=1, network=True, file_io=None, fsync=False, locking="none") +wa.thread_budget() # the shared context's decoding threads: OMP_NUM_THREADS, the job's slots, or CPUs; + # file I/O gets at least 8; threads=N limits both to N, threads=0 leaves tensorstore's sizes + # contexts made here write without fsync or OS file locks; fsync=True, locking="os" restore them wa.list_chunk_keys(path, separator=".") # chunk indices from keys alone, no metadata needed wa.Box ; wa.units.TABLE diff --git a/docs/spec.md b/docs/spec.md index 5354659..b248eaa 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -85,8 +85,8 @@ is asked to convert. the OME document builder and the level rule, chunk keys, `create`, opening for write, aligned writes, `add_level`. 0.3 adds grids (Behaviours 17 to 19): a store-less `Grid`, the block iterator every blockwise repo writes for itself, `write_chunk`, the factor -schedule and the public path helpers. Dependencies at every version: `tensorstore>=0.1.60`, -`numpy`. **No `zarr`, no `dask`, no `pydantic`, no `funlib`.** The package is float-only; a +schedule and the public path helpers. Dependencies at every version: `tensorstore` +(`>=0.1.60`; `>=0.1.68` from 0.8, the first with `file_io_locking`), `numpy`. **No `zarr`, no `dask`, no `pydantic`, no `funlib`.** The package is float-only; a consumer that works in integer Rois builds them in its own subclass (section 6). Python >= 3.10 with no 3.11-only syntax, no upper bounds on numpy or tensorstore. @@ -185,8 +185,11 @@ wa.open_tensorstore(path, level=None, context=None, retries=None) -> tensorstore wa.corner_to_centre(corner, voxel_size, axes=None) ; wa.default_axes(ndim) ; wa.index_range(begin, end, voxel_size, translation) wa.units.TABLE # the unit tables, public: {"length": {...metres per unit}, "time": {...seconds per unit}} wa.select_level(levels, voxel_size, mode) ; wa.Box ; wa.Retry(max_attempts=10, timeout=None, base_delay=1.0) -wa.shared_context(cache_bytes=0, threads=None) ; wa.capped_context(cache_bytes=0, concurrency=1, network=True) ; wa.capped_context_spec(...) - # threads: None = thread_budget(), N = that many, 0 = tensorstore's own pool sizes (0.6) +wa.shared_context(cache_bytes=0, threads=None, fsync=False, locking="none") +wa.capped_context(cache_bytes=0, concurrency=1, network=True, file_io=None, fsync=False, locking="none") ; wa.capped_context_spec(...) + # threads: None = thread_budget() for decoding and at least 8 for file I/O (0.8), N = that many, + # 0 = tensorstore's own pool sizes (0.6); fsync and locking: tensorstore's file_io_sync and + # file_io_locking mode, off and "none" in every context the package makes (0.8, Behaviour 7) # network=False caps decoding and local file I/O only, leaving tensorstore's request pools alone wa.list_chunk_keys(path, separator=".", ndim=None) # chunk indices present under a path, from its keys alone: no # metadata document needed, so a directory of chunk files with no .zarray lists too @@ -232,9 +235,9 @@ where it also chooses the integer scale factor that makes the corner whole. `shared_context(cache_bytes, threads=None)` is one `tensorstore.Context` per budget and thread limit for the process, so the levels of a pyramid and repeat opens share one cache -pool; the default budget is 0. Its decoding and file I/O pools are sized by -`thread_budget()` (0.5), or by `threads` when given, `0` leaving tensorstore's own sizes -(0.6, Behaviour 7). +pool; the default budget is 0. Its decoding pool is sized by `thread_budget()` (0.5) and +its file I/O pool by that or 8, whichever is larger (0.8), or both by `threads` when given, +`0` leaving tensorstore's own sizes (0.6, Behaviour 7). `capped_context` is mesh-n-bone's worker-process context (every tensorstore concurrency resource limited, cache off); it returns a **new** context per call, as those workers inline the spec into every open today, and `capped_context_spec` exports the dict for @@ -665,8 +668,13 @@ this document and that one differ, fix one of them. of the public call doing the reading: `open`, `open_pyramid`, `create`, `metadata`, `detect` and `open_tensorstore`, which all take it; a document read later outside such a call (a zarr v2 `.zattrs` read on demand, a pyramid placing its levels) waits under - `Retry()`. Listings are not timed out, their size being unknown, and neither is the - open that creates an array, which a second attempt would find already made. + `Retry()`. Listings are not timed out, their size being unknown. The open that creates + an array is (0.8, cellmap-analyze's question): its second attempt may find the array its + first made before timing out, so after a timed-out attempt `ALREADY_EXISTS` is not an + error even under `exists="error"`, and `create` adopts the array only if it matches what + was asked for (the comparison an adopt makes), since an identical array that appeared + after our own attempt is ours. That open keeps tensorstore's `"os"` file locking whatever + the caller's context says, as create-if-absent is a conditional write. The watchdog (0.5): tensorstore up to at least 0.1.85 can strand a queued task. A pool worker idle for 20 ms exits, and a task queued as it exits gets no thread while the pool @@ -687,8 +695,7 @@ this document and that one differ, fix one of them. 7. **Contexts.** Without `context`, every open in the process shares `shared_context(0)`: tensorstore's cache is off and the levels of one pyramid share one context. Its - `data_copy_concurrency` and `file_io_concurrency` limits are `thread_budget()`: the - first of `OMP_NUM_THREADS` (dask sets 1 in every worker it spawns; a cluster job script + `data_copy_concurrency` limit is `thread_budget()`: the first of `OMP_NUM_THREADS` (dask sets 1 in every worker it spawns; a cluster job script sets its slots) and `LSB_DJOB_NUMPROC` (an LSF job's slots) that holds a positive integer, else the CPUs `sched_getaffinity` allows: the cap every other threaded library in the job already takes, as Janelia's cluster guidance asks. tensorstore alone sizes @@ -696,7 +703,13 @@ this document and that one differ, fix one of them. workers in a twelve-slot job on a 96-core node could start 96 threads (0.5). A dask worker therefore runs tensorstore on one thread of each pool, which Behaviour 6's watchdog makes safe. SLURM's `SLURM_CPUS_PER_TASK` and SGE's `NSLOTS` stand in for - `LSB_DJOB_NUMPROC` elsewhere. The cap is the default because the process that passes + `LSB_DJOB_NUMPROC` elsewhere. Its `file_io_concurrency` limit is that or + `context.FILE_IO_FLOOR` (8), whichever is larger (0.8): file I/O threads wait on storage + rather than compute, so they take no core from the job, and one of them serialises every + chunk file of a block on NFS's round trips. From one decoding thread, eight 64^3 label + chunks written to NFS took 166 to 176 ms with one file I/O thread and 83 to 89 ms with + four to sixteen (zarr-python 2 took 102); an 8-chunk EM read from /nrs was the same + either way, decoding bound. The cap is the default because the process that passes no context is the one nobody tuned: a quick dask script over `wa.open(path)` is how a job came to start a thousand threads on twelve slots. One thread per pool slows one task (a 256^3 block write took 2.1 s against 1.3 s uncapped) but not a job of many @@ -711,9 +724,30 @@ this document and that one differ, fix one of them. process, made on first use and watched; `threads=None` and the number `thread_budget()` returns are two contexts, as a pickle carries `None` for the other process to size by its own job and a number as that number. + + Writes (0.8). Every context the package makes (`shared_context`, `capped_context`) + sets tensorstore's `file_io_sync` to false and its `file_io_locking` mode to `"none"`, + the arguments `fsync=True` and `locking="os"` restoring tensorstore's defaults. fsync + cost 6.5 ms a chunk on a local NVMe disk (8.2 against 1.5 ms for a 64^3 chunk) and + nothing on NFS, whose client commits a file when it closes it; zarr-python and funlib, + every writer the repos used before, never synced, so a machine crash on a local disk + can lose the last seconds of chunks as it always could, and a killed process loses + nothing. `"os"` locking takes an advisory lock on a `.__lock` file for every write, + about 10 ms of NFS round trips a chunk (21.7 against 12.0 ms for a 64^3 label chunk on + /groups; zarr-python 2 took 12.7). With `"none"` a whole-unit write is still atomic, a + temporary file renamed into place, and only concurrent read-modify-writes of one unit + lose their guarantee, which Behaviour 13 never gave (`partial=True` is unsafe while + anything else writes the unit); a write interrupted mid-way may leave a temporary + `.__lock` file that costs space only. tensorstore's `"lockfile"` mode is not the + default because such a file left by a killed worker blocks every later write of its + unit. Together, from one-thread workers (`OMP_NUM_THREADS=1`) through the default + context: a 64^3 label chunk to NFS took 23 ms in 0.7 and 12.2 in 0.8, an 8-chunk label + block 159 and 66.5, an 8-chunk EM block 330 and 202; to a local disk the label block + took 51.5 and 4.1 ms. `capped_context` sets `data_copy_concurrency`, `file_io_concurrency`, `gcs_request_concurrency`, `s3_request_concurrency` and `http_request_concurrency` to - `concurrency` and the cache to `cache_bytes`, and returns a new context each call, so two + `concurrency` (`file_io_concurrency` to `file_io` when given, 0.8) and the cache to + `cache_bytes`, and returns a new context each call, so two handles never share one capped pool unless the caller passes the same object twice; a `concurrency` of 1 is safe under Behaviour 6's watchdog, but a read or write spanning several chunks then handles them one at a time. On /nrs, from one process alternating @@ -1035,7 +1069,10 @@ context's own pools; `thread_budget` from each variable, a bad value skipped, an shared context's limits. 0.6: `shared_context(threads=)` at `None`, `N` and `0`, one context per pair, a negative count refused, and a pickled handle coming back on the same shared context. 0.7: `metadata`, `open` and `detect` against a server -that accepts and never answers raising `TimeoutError` after the attempts. Opt-in (`WORLDARRAY_STRESS=1`, about a minute, +that accepts and never answers raising `TimeoutError` after the attempts. 0.8: the default +and capped contexts' file I/O limit, sync and locking, `fsync` and `locking` carried +through a pickle, and a create whose first attempt lands after its timeout adopting its own +array under `exists="error"`. Opt-in (`WORLDARRAY_STRESS=1`, about a minute, `tests/test_lost_wakeup.py`): worker processes aim tasks at the moment a one-thread pool's worker exits idle, so a plain context strands some for good (13 of about 7,000 on tensorstore 0.1.78; 40 to 44 of 12,000 to 15,000 with 16 workers on 0.1.75 and 0.1.85) diff --git a/pyproject.toml b/pyproject.toml index 515a422..6c7dc1f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "worldarray" -version = "0.7.1" +version = "0.8.0" 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" } @@ -31,7 +31,7 @@ classifiers = [ # no dask, no funlib (docs/spec.md, section 1). dependencies = [ "numpy>=1.26", - "tensorstore>=0.1.60", + "tensorstore>=0.1.68", ] [project.urls] diff --git a/src/worldarray/__init__.py b/src/worldarray/__init__.py index 4e443aa..02dcb20 100644 --- a/src/worldarray/__init__.py +++ b/src/worldarray/__init__.py @@ -28,7 +28,7 @@ parse_chunk_key, ) -__version__ = "0.7.1" +__version__ = "0.8.0" __all__ = [ "ArrayMeta", diff --git a/src/worldarray/context.py b/src/worldarray/context.py index 1226375..1706f14 100644 --- a/src/worldarray/context.py +++ b/src/worldarray/context.py @@ -4,11 +4,13 @@ thread limit for the whole process, so the levels of a pyramid and repeat opens of an array draw on one cache pool. The default budget is 0: tensorstore's decoded-chunk cache is off, because every consumer either reads each region once or caches decoded chunks -itself. By default its decoding and file I/O pools are sized by ``thread_budget()``, from -the variables a cluster job sets, where tensorstore alone would size them to every CPU on -the node; ``threads=0`` leaves them at tensorstore's sizes (chunkmirage's server). Every -context here, and every one a store is opened on, is watched for tensorstore's lost wakeup -(``watch``). +itself. By default its decoding pool is sized by ``thread_budget()``, from the variables a +cluster job sets, where tensorstore alone would size it to every CPU on the node, and its +file I/O pool to at least ``FILE_IO_FLOOR`` threads, which wait on storage rather than +compute; ``threads=0`` leaves both at tensorstore's sizes (chunkmirage's server). Every +context made here writes without fsync and without OS file locks (``write_settings``). +Every context here, and every one a store is opened on, is watched for tensorstore's lost +wakeup (``watch``). ``capped_context`` is mesh-n-bone's worker-process context: every tensorstore concurrency resource limited, so dozens of worker processes on one host do not multiply into thousands @@ -28,7 +30,7 @@ import tensorstore as ts -_shared: dict[tuple[int, int | None], ts.Context] = {} +_shared: dict[tuple[int, int | None, bool, str], ts.Context] = {} _rebuilt: dict[str, ts.Context] = {} _lock = threading.Lock() @@ -138,38 +140,79 @@ def thread_budget() -> int: return os.cpu_count() or 1 -def shared_context(cache_bytes: int = 0, *, threads: int | None = None) -> ts.Context: - """The process's one context for ``cache_bytes`` and ``threads``, made on first use and - watched. ``threads`` limits its decoding and file I/O pools: ``None`` to - ``thread_budget()``, a positive number to that many, ``0`` not at all (tensorstore's own - sizes, every CPU on the node).""" - key = (int(cache_bytes), None if threads is None else int(threads)) +# File I/O threads mostly wait on storage, so they are not capped like decoding: on NFS, +# eight 64^3 label chunks written from one decoding thread took 166-176 ms with one file I/O +# thread and 83-89 ms with four to sixteen (2026-10-23; zarr-python 2 took 102) +FILE_IO_FLOOR = 8 + + +def write_settings(fsync: bool = False, locking: str = "none") -> dict: + """tensorstore's ``file_io_sync`` and ``file_io_locking`` for a context made here. fsync + is off by default: zarr-python never synced, NFS commits a file on close anyway, and on + a local disk it cost 6 ms a chunk. Locking is ``"none"`` by default: a whole-unit write + stays atomic (a temporary file renamed into place), only concurrent read-modify-writes of + one unit lose the guarantee, which Behaviour 13 never gave, and tensorstore's ``"os"`` + locks cost about 10 ms a chunk on NFS (spec, Behaviour 7).""" + return {"file_io_sync": bool(fsync), "file_io_locking": {"mode": str(locking)}} + + +def shared_context( + cache_bytes: int = 0, + *, + threads: int | None = None, + fsync: bool = False, + locking: str = "none", +) -> ts.Context: + """The process's one context for these arguments, made on first use and watched. + ``threads`` limits its decoding and file I/O pools: ``None`` decoding to + ``thread_budget()`` and file I/O to that or ``FILE_IO_FLOOR``, whichever is larger, a + positive number both to that many, ``0`` neither (tensorstore's own sizes, every CPU on + the node). ``fsync`` and ``locking`` go to ``write_settings``.""" + key = (int(cache_bytes), None if threads is None else int(threads), bool(fsync), str(locking)) if key[1] is not None and key[1] < 0: raise ValueError(f"threads must be None, 0 or positive, not {threads!r}") with _lock: ctx = _shared.get(key) if ctx is None: - spec: dict = {"cache_pool": {"total_bytes_limit": key[0]}} - limit = thread_budget() if threads is None else key[1] - if limit: - spec["data_copy_concurrency"] = spec["file_io_concurrency"] = {"limit": limit} + spec: dict = { + "cache_pool": {"total_bytes_limit": key[0]}, + **write_settings(fsync, locking), + } + if threads is None: + budget = thread_budget() + spec["data_copy_concurrency"] = {"limit": budget} + spec["file_io_concurrency"] = {"limit": max(budget, FILE_IO_FLOOR)} + elif key[1]: + spec["data_copy_concurrency"] = spec["file_io_concurrency"] = {"limit": key[1]} ctx = watch(ts.Context(spec)) _shared[key] = ctx return ctx -def capped_context_spec(cache_bytes: int = 0, concurrency: int = 1, network: bool = True) -> dict: +def capped_context_spec( + cache_bytes: int = 0, + concurrency: int = 1, + network: bool = True, + *, + file_io: int | None = None, + fsync: bool = False, + locking: str = "none", +) -> dict: """The spec dict behind ``capped_context``, for callers building their own opens. ``network=False`` caps only decoding and local file I/O, leaving tensorstore's request pools alone (cellmap-analyze's choice for remote reads). A ``concurrency`` of 1 is safe: the context is watched (see ``KICK_EVERY``). It is not free: a read or write spanning several chunks then decodes and stores them one at a time, half the speed of 4 on - 8-chunk blocks in a two-slot job, 7 to 19% slower in a one-slot one (spec, Behaviour 7).""" + 8-chunk blocks in a two-slot job, 7 to 19% slower in a one-slot one (spec, Behaviour 7). + ``file_io`` sizes the file I/O pool apart from decoding (default ``concurrency``; on NFS + a few file I/O threads halve the time of a multi-chunk write, ``FILE_IO_FLOOR``). + ``fsync`` and ``locking`` go to ``write_settings``.""" limit = {"limit": int(concurrency)} spec = { "data_copy_concurrency": limit, - "file_io_concurrency": limit, + "file_io_concurrency": {"limit": int(file_io if file_io is not None else concurrency)}, "cache_pool": {"total_bytes_limit": int(cache_bytes)}, + **write_settings(fsync, locking), } if network: spec.update( @@ -180,20 +223,31 @@ def capped_context_spec(cache_bytes: int = 0, concurrency: int = 1, network: boo return spec -def capped_context(cache_bytes: int = 0, concurrency: int = 1, network: bool = True) -> ts.Context: - return watch(ts.Context(capped_context_spec(cache_bytes, concurrency, network))) +def capped_context( + cache_bytes: int = 0, + concurrency: int = 1, + network: bool = True, + *, + file_io: int | None = None, + fsync: bool = False, + locking: str = "none", +) -> ts.Context: + spec = capped_context_spec( + cache_bytes, concurrency, network, file_io=file_io, fsync=fsync, locking=locking + ) + return watch(ts.Context(spec)) def recipe_of(context: ts.Context) -> tuple[str, Any]: """How another process rebuilds ``context``: ``("shared", cache_bytes)`` for a shared - context at the default limit (the other process takes its own ``thread_budget()``), - ``("shared", (cache_bytes, threads))`` for one with ``threads`` given, else - ``("spec", json)`` from the context's own spec, so a capped context stays capped across - a pickle (Behaviour 11, mesh-n-bone's workers).""" + context with every other argument at its default (the other process takes its own + ``thread_budget()``), ``("shared", (cache_bytes, threads, fsync, locking))`` otherwise, + else ``("spec", json)`` from the context's own spec, so a capped context stays capped + across a pickle (Behaviour 11, mesh-n-bone's workers).""" with _lock: - for (cache_bytes, threads), ctx in _shared.items(): + for key, ctx in _shared.items(): if ctx is context: - return ("shared", cache_bytes if threads is None else (cache_bytes, threads)) + return ("shared", key[0] if key[1:] == (None, False, "none") else key) spec = context.spec return ("spec", json.dumps(spec.to_json() if spec is not None else {}, sort_keys=True)) @@ -205,8 +259,12 @@ def from_recipe(recipe: tuple[str, Any] | None) -> ts.Context | None: return None kind, value = recipe if kind == "shared": - cache_bytes, threads = value if isinstance(value, (tuple, list)) else (value, None) - return shared_context(int(cache_bytes), threads=threads) + cache_bytes, threads, fsync, locking = ( + (*value, False, "none")[:4] + if isinstance(value, (tuple, list)) + else (value, None, False, "none") + ) + return shared_context(int(cache_bytes), threads=threads, fsync=fsync, locking=locking) with _lock: ctx = _rebuilt.get(value) if ctx is None: diff --git a/src/worldarray/write.py b/src/worldarray/write.py index 01cc960..ba9763d 100644 --- a/src/worldarray/write.py +++ b/src/worldarray/write.py @@ -34,6 +34,7 @@ _DRIVER, Location, Node, + _wait, basename, detect, is_refused, @@ -753,6 +754,11 @@ def create_array( # the group documents first (Behaviour 11), then the array ensure_group_entry(group_path, name, vs, tr, axes, units, axis_types, group_attrs, format=fmt) + # create-if-absent is a conditional write, atomic only under tensorstore's OS locks, + # whatever locking the caller's context chose for its data (write_settings) + creating = watch(ts.Context(ts.Context.Spec({"file_io_locking": {"mode": "os"}}), context)) + starts = 0 + def attempt(kvstore: dict) -> ts.TensorStore: spec = array_spec( kvstore, @@ -768,16 +774,26 @@ def attempt(kvstore: dict) -> ts.TensorStore: separator, store_fill, ) - return ts.open(spec, create=True, open=False, context=watch(context)).result() + + def start(): + nonlocal starts + starts += 1 + return ts.open(spec, create=True, open=False, context=creating) + + return _wait(start, kvstore, path, "create") try: with_access(kvstore_specs(path), attempt) except Exception as e: # noqa: BLE001 if "ALREADY_EXISTS" not in str(e) or is_refused(e): raise - if exists == "error": + if exists == "error" and starts == 1: raise FileExistsError(f"{path} was created by another process meanwhile") from e + # another process made it, or an attempt of ours that timed out did: the caller + # adopts it only if it matches what was asked for (Behaviour 6) _wait_for(path) + if starts > 1 and fmt == "zarr2" and attrs and Location(path).read(".zattrs") is None: + Location(path).write_json(".zattrs", dict(attrs)) return path, fmt, True if fmt == "zarr2" and attrs: Location(path).write_json(".zattrs", dict(attrs)) diff --git a/tests/test_access.py b/tests/test_access.py index 615fd19..f85c217 100644 --- a/tests/test_access.py +++ b/tests/test_access.py @@ -154,7 +154,10 @@ def test_the_thread_budget_follows_the_job(monkeypatch): assert wa.thread_budget() == 6 monkeypatch.setattr(wa.context, "_shared", {}) spec = wa.shared_context(5).spec.to_json() - assert spec["data_copy_concurrency"] == spec["file_io_concurrency"] == {"limit": 6} + assert spec["data_copy_concurrency"] == {"limit": 6} + assert spec["file_io_concurrency"] == { + "limit": wa.context.FILE_IO_FLOOR + } # it waits, not computes assert spec["cache_pool"] == {"total_bytes_limit": 5} @@ -326,7 +329,11 @@ def test_a_shared_context_takes_the_callers_thread_count(monkeypatch): monkeypatch.setenv("OMP_NUM_THREADS", "1") monkeypatch.setattr(wa.context, "_shared", {}) default = wa.shared_context(7) - assert default.spec.to_json()["data_copy_concurrency"] == {"limit": 1} + spec = default.spec.to_json() + assert spec["data_copy_concurrency"] == {"limit": 1} and spec["file_io_concurrency"] == { + "limit": 8 + } + assert spec["file_io_sync"] is False and spec["file_io_locking"] == {"mode": "none"} four = wa.shared_context(7, threads=4).spec.to_json() assert four["data_copy_concurrency"] == four["file_io_concurrency"] == {"limit": 4} machine = wa.shared_context(7, threads=0) @@ -335,8 +342,13 @@ def test_a_shared_context_takes_the_callers_thread_count(monkeypatch): assert "data_copy_concurrency" not in spec and "file_io_concurrency" not in spec assert spec["cache_pool"] == {"total_bytes_limit": 7} assert machine in wa.context._watched - assert wa.context.recipe_of(machine) == ("shared", (7, 0)) - assert wa.context.from_recipe(("shared", (7, 0))) is machine + assert wa.context.recipe_of(machine) == ("shared", (7, 0, False, "none")) + assert wa.context.from_recipe(("shared", (7, 0))) is machine # a 0.6 or 0.7 pickle + durable = wa.shared_context(7, fsync=True, locking="os") + spec = durable.spec.to_json() + assert durable is not default and spec["file_io_sync"] is True + assert spec["file_io_locking"] == {} # "os", tensorstore's default, which its spec leaves out + assert wa.context.from_recipe(wa.context.recipe_of(durable)) is durable assert wa.context.from_recipe(("shared", 7)) is default with pytest.raises(ValueError): wa.shared_context(threads=-1) @@ -399,3 +411,16 @@ def log_message(self, *args): assert time.monotonic() - start < 1.0 # in turn: 1.5 s finally: server.shutdown() + + +def test_capped_contexts_size_file_io_apart_and_write_without_sync_or_locks(): + spec = wa.capped_context_spec(concurrency=1, network=False) + assert spec["data_copy_concurrency"] == spec["file_io_concurrency"] == {"limit": 1} + assert spec["file_io_sync"] is False and spec["file_io_locking"] == {"mode": "none"} + spec = wa.capped_context_spec(concurrency=1, file_io=8, fsync=True, locking="os") + assert spec["file_io_concurrency"] == {"limit": 8} and spec["data_copy_concurrency"] == { + "limit": 1 + } + assert spec["file_io_sync"] is True and spec["file_io_locking"] == {"mode": "os"} # as written + capped = wa.capped_context(concurrency=1, file_io=8) + assert capped.spec.to_json()["file_io_concurrency"] == {"limit": 8} diff --git a/tests/test_write.py b/tests/test_write.py index 339ced6..d0ea922 100644 --- a/tests/test_write.py +++ b/tests/test_write.py @@ -849,3 +849,33 @@ def test_codecs_of_a_path_and_the_fill_write_that_removes_a_key(tmp_path): vol.write_index((0, 0, 0), (4, 4, 4), np.zeros((4, 4, 4), np.uint8)) assert vol.list_chunks() == [] # the one key a write may remove: its own, now all fill assert int(vol.read_index((0, 0, 0), (4, 4, 4)).sum()) == 0 + + +class _NeverAnswers: + def result(self, timeout=None): + raise TimeoutError() + + def cancel(self): + pass + + +def test_a_create_that_lands_after_its_timeout_adopts_its_own_array(tmp_path, monkeypatch): + real_open = ts.open + creates = [] + + def open_landing_late(spec, *args, **kwargs): + if kwargs.get("create"): + creates.append(1) + if len(creates) == 1: # the array is made, but the answer never comes + real_open(spec, *args, **kwargs).result() + return _NeverAnswers() + return real_open(spec, *args, **kwargs) + + monkeypatch.setattr(ts, "open", open_landing_late) + path = str(tmp_path / "vol.zarr/s0") + quick = wa.Retry(timeout=0.2, max_attempts=3, base_delay=0.0) + vol = wa.create(path, (8, 8, 8), "uint8", (4, 4, 4), (2, 2, 2), chunks=(4, 4, 4), retries=quick) + assert len(creates) == 2 and vol.shape == (8, 8, 8) + monkeypatch.setattr(ts, "open", real_open) + with pytest.raises(FileExistsError): + wa.create(path, (8, 8, 8), "uint8", (4, 4, 4), (2, 2, 2), chunks=(4, 4, 4)) diff --git a/uv.lock b/uv.lock index 9b5dd72..599bc6c 100644 --- a/uv.lock +++ b/uv.lock @@ -1247,7 +1247,7 @@ wheels = [ [[package]] name = "worldarray" -version = "0.7.1" +version = "0.8.0" source = { editable = "." } dependencies = [ { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, @@ -1270,7 +1270,7 @@ docs = [ [package.metadata] requires-dist = [ { name = "numpy", specifier = ">=1.26" }, - { name = "tensorstore", specifier = ">=0.1.60" }, + { name = "tensorstore", specifier = ">=0.1.68" }, ] [package.metadata.requires-dev] From 9065322d7008fa1c3c0a63d9c0e196d24f7df4c2 Mon Sep 17 00:00:00 2001 From: David Ackerman Date: Fri, 23 Oct 2026 17:12:13 -0400 Subject: [PATCH 2/2] Date 0.8.0 Claude-Session: https://claude.ai/code/session_0141wgF8mkGbjGfBFKfKhp64 --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 691cb35..8bdee55 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## 0.8.0 (unreleased) +## 0.8.0 (2026-10-23) Faster writes from the contexts the package makes, and the create open timed out. No placement change and no unit table change. **Requires tensorstore 0.1.68 or later** (was