diff --git a/changes/4205.bugfix.md b/changes/4205.bugfix.md new file mode 100644 index 0000000000..0492febb7d --- /dev/null +++ b/changes/4205.bugfix.md @@ -0,0 +1,9 @@ +Fixed several small correctness issues from the codec-pipeline performance work: construction-time +codec warnings (e.g. sharding's "disables partial reads" warning) no longer fire twice per array +open — including for `FusedCodecPipeline`, which previously re-warned via its own codec-chain +reconstruction and, on the async fallback path, on every decode/encode call; `concurrent_iter` now +schedules its tasks eagerly, matching its documented contract; an invalid +`codec_pipeline.max_workers` config/environment value now warns and falls back to the default +instead of raising mid-read; and `FusedCodecPipeline`'s async fallback helpers now cancel +already-spawned fetch/decode/write tasks instead of abandoning them in the background when one +fails. diff --git a/src/zarr/core/array.py b/src/zarr/core/array.py index f75ef72415..cd51dad50c 100644 --- a/src/zarr/core/array.py +++ b/src/zarr/core/array.py @@ -228,6 +228,12 @@ def create_codec_pipeline(metadata: ArrayMetadata, *, store: Store | None = None pass if isinstance(metadata, ArrayV3Metadata): + # The pipeline built here is a throwaway: `evolve_from_array_spec` below + # reconstructs codecs against the evolved spec. `from_codecs` is the + # chain's first construction, so its advisory warnings (e.g. sharding's + # "disables partial reads" warning) fire here; `evolve_from_array_spec` + # re-splits the same already-warned-about chain via + # `codecs_from_list_unchecked`, so it does not re-emit them. pipeline = get_pipeline_class().from_codecs(metadata.codecs) from zarr.core.metadata.v3 import RegularChunkGridMetadata diff --git a/src/zarr/core/chunk_utils.py b/src/zarr/core/chunk_utils.py index d93793f853..b26d5478b2 100644 --- a/src/zarr/core/chunk_utils.py +++ b/src/zarr/core/chunk_utils.py @@ -238,7 +238,7 @@ class ChunkTransform: ) def __post_init__(self) -> None: - from zarr.core.codec_pipeline import codecs_from_list + from zarr.core.codec_pipeline import codecs_from_list_unchecked # _codec_supports_sync, not a bare isinstance check: a codec can satisfy # the SupportsSyncCodec protocol structurally yet be unable to run @@ -253,7 +253,11 @@ def __post_init__(self) -> None: f"All codecs must implement SupportsSyncCodec. The following do not: {names}" ) - aa, ab, bb = codecs_from_list(list(self.codecs)) + # `ChunkTransform` is built from a codec chain that already went + # through `codecs_from_list` when the owning pipeline was constructed + # (see `FusedCodecPipeline.evolve_from_array_spec`), so re-splitting it + # here must not re-emit that chain's advisory warnings. + aa, ab, bb = codecs_from_list_unchecked(list(self.codecs)) # SupportsSyncCodec was verified above; the cast is purely for mypy. self._aa_codecs = cast("tuple[SupportsSyncCodec[NDBuffer, NDBuffer], ...]", tuple(aa)) self._ab_codec = cast("SupportsSyncCodec[NDBuffer, Buffer]", ab) diff --git a/src/zarr/core/codec_pipeline.py b/src/zarr/core/codec_pipeline.py index ca760ece59..92fd0970fe 100644 --- a/src/zarr/core/codec_pipeline.py +++ b/src/zarr/core/codec_pipeline.py @@ -4,7 +4,7 @@ import threading from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field -from itertools import batched, pairwise +from itertools import batched, chain, pairwise from typing import TYPE_CHECKING, Any, cast from warnings import warn @@ -54,10 +54,23 @@ def _resolve_max_workers() -> int: """Helper for getting the maximum number of workers available to the `FusedCodecPipeline`""" import os as _os + default = _os.cpu_count() or 1 cfg = config.get("codec_pipeline.max_workers", default=None) if cfg is None: - return _os.cpu_count() or 1 - return max(1, int(cfg)) + return default + try: + return max(1, int(cfg)) + except (TypeError, ValueError): + # This value arrives via the config/env layer (e.g. + # `ZARR_CODEC_PIPELINE__MAX_WORKERS`), so tolerate bad input here + # instead of raising mid-read. + warn( + f"Ignoring invalid `codec_pipeline.max_workers` config value {cfg!r}; " + f"falling back to {default}.", + category=ZarrUserWarning, + stacklevel=2, + ) + return default def _get_pool(max_workers: int) -> ThreadPoolExecutor: @@ -169,6 +182,23 @@ def pipeline_supports_partial_encode( return isinstance(array_bytes_codec, ArrayBytesCodecPartialEncodeMixin) +async def _cancel_and_drain(futures: Iterable[asyncio.Future[Any]]) -> None: + """Cancel every not-yet-done future/task and await its outcome. + + Used to clean up work spawned by a drain loop (`asyncio.as_completed` + + `await`) when the loop exits early via exception. Without this, tasks + already spawned keep running unattended after the caller has moved on, + and an eventual failure surfaces as an unraisable "exception was never + retrieved" warning instead of being observed here. + """ + pending = [f for f in futures if not f.done()] + if len(pending) == 0: + return + for f in pending: + f.cancel() + await asyncio.gather(*pending, return_exceptions=True) + + async def _fetch_and_decode_as_completed( batch: Sequence[tuple[ByteGetter | None, ArraySpec]], transform: ChunkTransform, @@ -201,20 +231,29 @@ def _decode(buffer: Buffer | None, chunk_spec: ArraySpec) -> NDBuffer | None: _fetch, config.get("async.concurrency"), ) - for fetch_coro in asyncio.as_completed(fetch_tasks): - idx, buffer = await fetch_coro - chunk_spec = batch[idx][1] - # Bridge both paths to asyncio.Future so the final collection loop - # can `await` uniformly without blocking the event loop. For the - # pool path that means `wrap_future` (not `pool.submit(...).result()`, - # which would block the loop thread for the duration of every decode - # — freezing any unrelated coroutines sharing this loop). - if pool is None: - decode_futures[idx].set_result(_decode(buffer, chunk_spec)) - else: - decode_futures[idx] = asyncio.wrap_future(pool.submit(_decode, buffer, chunk_spec)) + try: + for fetch_coro in asyncio.as_completed(fetch_tasks): + idx, buffer = await fetch_coro + chunk_spec = batch[idx][1] + # Bridge both paths to asyncio.Future so the final collection loop + # can `await` uniformly without blocking the event loop. For the + # pool path that means `wrap_future` (not `pool.submit(...).result()`, + # which would block the loop thread for the duration of every decode + # — freezing any unrelated coroutines sharing this loop). + if pool is None: + decode_futures[idx].set_result(_decode(buffer, chunk_spec)) + else: + decode_futures[idx] = asyncio.wrap_future(pool.submit(_decode, buffer, chunk_spec)) - return await asyncio.gather(*decode_futures) + return await asyncio.gather(*decode_futures) + finally: + # On the happy path every future here is already done, so this is a + # no-op; on failure it stops abandoned fetches/decodes from + # continuing to run unattended after this function has raised. A + # single call over both iterables (not two sequential calls) so that + # outer-task cancellation during the first drain can't skip the + # second, leaving its futures/tasks unobserved. + await _cancel_and_drain(chain(fetch_tasks, decode_futures)) async def _encode_and_write_as_completed( @@ -263,10 +302,20 @@ async def _write(idx: int, chunk_bytes: Buffer | None) -> None: # Kick off each chunk's write the instant its encode lands, so writes of # already-compressed chunks proceed while the rest are still encoding. write_tasks: list[asyncio.Task[None]] = [] - for encode_coro in asyncio.as_completed(encode_futures): - idx, chunk_bytes = await encode_coro - write_tasks.append(asyncio.ensure_future(_write(idx, chunk_bytes))) - await asyncio.gather(*write_tasks) + try: + for encode_coro in asyncio.as_completed(encode_futures): + idx, chunk_bytes = await encode_coro + write_tasks.append(asyncio.ensure_future(_write(idx, chunk_bytes))) + await asyncio.gather(*write_tasks) + finally: + # On the happy path every future here is already done, so this is a + # no-op; on failure (an encode or a write raising) it stops + # already-spawned writes from continuing in the background after + # this function has raised. A single call over both iterables (not + # two sequential calls) so that outer-task cancellation during the + # first drain can't skip the second, leaving its futures/tasks + # unobserved. + await _cancel_and_drain(chain(write_tasks, encode_futures)) async def _async_read_fallback( @@ -468,7 +517,11 @@ class AsyncChunkTransform: _bb_codecs: tuple[BytesBytesCodec, ...] = field(init=False, repr=False, compare=False) def __post_init__(self) -> None: - aa, ab, bb = codecs_from_list(list(self.codecs)) + # `AsyncChunkTransform` is (re)constructed per decode/encode call from a + # codec chain that already went through `codecs_from_list` when the + # pipeline itself was built, so re-splitting it here must not re-emit + # that chain's advisory warnings on every call. + aa, ab, bb = codecs_from_list_unchecked(list(self.codecs)) self._aa_codecs = aa self._ab_codec = ab self._bb_codecs = bb @@ -532,7 +585,19 @@ class BatchedCodecPipeline(CodecPipeline): batch_size: int def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self: - return type(self).from_codecs(evolve_codecs(self, array_spec)) + # Re-splits an already-`codecs_from_list`-validated (and warned-about) + # chain against the evolved spec, so this uses the quiet variant rather + # than routing through `from_codecs` (which would re-warn). + evolved_codecs = evolve_codecs(self, array_spec) + array_array_codecs, array_bytes_codec, bytes_bytes_codecs = codecs_from_list_unchecked( + evolved_codecs + ) + return type(self)( + array_array_codecs=array_array_codecs, + array_bytes_codec=array_bytes_codec, + bytes_bytes_codecs=bytes_bytes_codecs, + batch_size=self.batch_size, + ) @classmethod def from_codecs(cls, codecs: Iterable[Codec], *, batch_size: int | None = None) -> Self: @@ -794,14 +859,20 @@ async def write( def codecs_from_list( codecs: Iterable[Codec], ) -> tuple[tuple[ArrayArrayCodec, ...], ArrayBytesCodec, tuple[BytesBytesCodec, ...]]: + """Split `codecs` into `(array_array, array_bytes, bytes_bytes)`, validating order. + + Emits user-facing advisory warnings about the codec chain (e.g. sharding's + "disables partial reads" warning). Use this for the FIRST construction of a + codec chain from user-supplied codecs. Use `codecs_from_list_unchecked` when + re-splitting a chain that was already validated and warned about by a prior + `codecs_from_list` call (e.g. `evolve_from_array_spec` re-splitting the same + codecs against an evolved spec) — re-warning there would fire the same + advisory once per reconstruction instead of once per user-facing chain. + """ from zarr.codecs.sharding import ShardingCodec codecs = tuple(codecs) # materialize to avoid generator consumption issues - array_array: tuple[ArrayArrayCodec, ...] = () - array_bytes_maybe: ArrayBytesCodec | None = None - bytes_bytes: tuple[BytesBytesCodec, ...] = () - if any(isinstance(codec, ShardingCodec) for codec in codecs) and len(codecs) > 1: warn( "Combining a `sharding_indexed` codec disables partial reads and " @@ -809,6 +880,23 @@ def codecs_from_list( category=ZarrUserWarning, stacklevel=3, ) + return codecs_from_list_unchecked(codecs) + + +def codecs_from_list_unchecked( + codecs: Iterable[Codec], +) -> tuple[tuple[ArrayArrayCodec, ...], ArrayBytesCodec, tuple[BytesBytesCodec, ...]]: + """Split `codecs` into `(array_array, array_bytes, bytes_bytes)`, validating order. + + Same structural validation as `codecs_from_list` (raises on bad codec + ordering or a missing/duplicate array-bytes codec) but does NOT emit + user-facing advisory warnings. See `codecs_from_list` for when to use each. + """ + codecs = tuple(codecs) # materialize to avoid generator consumption issues + + array_array: tuple[ArrayArrayCodec, ...] = () + array_bytes_maybe: ArrayBytesCodec | None = None + bytes_bytes: tuple[BytesBytesCodec, ...] = () for prev_codec, cur_codec in pairwise((None, *codecs)): if isinstance(cur_codec, ArrayArrayCodec): @@ -911,8 +999,11 @@ def from_codecs(cls, codecs: Iterable[Codec], *, batch_size: int | None = None) ) def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self: + # Re-splits an already-`codecs_from_list`-validated (and warned-about) + # chain against the evolved spec, so this uses the quiet variant to + # avoid re-emitting the same advisory warning on every array open. evolved_codecs = evolve_codecs(self.codecs, array_spec) - aa, ab, bb = codecs_from_list(evolved_codecs) + aa, ab, bb = codecs_from_list_unchecked(evolved_codecs) try: sync_transform: ChunkTransform | None = ChunkTransform(codecs=evolved_codecs) diff --git a/src/zarr/core/common.py b/src/zarr/core/common.py index 4114cb7645..1541683b09 100644 --- a/src/zarr/core/common.py +++ b/src/zarr/core/common.py @@ -93,26 +93,26 @@ def concurrent_iter[T: tuple[Any, ...], V]( items: Iterable[T], func: Callable[..., Awaitable[V]], limit: int | None = None, -) -> Iterator[asyncio.Task[V]]: +) -> list[asyncio.Task[V]]: """Launch `func(*item)` for each item concurrently, returning the tasks. When `limit` is set, no more than `limit` calls are in flight at once. Tasks are returned in input order; callers that want completion order should wrap the result in `asyncio.as_completed`. - Note on `ensure_future`: when the result is passed to `asyncio.gather` or - `asyncio.as_completed`, those already wrap awaitables into tasks, so the - `ensure_future` here is redundant. It matters for callers that iterate and - await tasks one at a time — without eager scheduling, each coroutine would - only start when individually awaited, serializing the work and defeating - the semaphore. It also makes the return type honest (real `Task`s support - `.cancel()`, `.done()`, callbacks) rather than bare coroutines. + Every task is scheduled (via `ensure_future`) before this function + returns, not on first iteration of the result. That matters for callers + that await the returned tasks one at a time — without eager scheduling, + each coroutine would only start when individually awaited, serializing + the work and defeating the semaphore. It also makes the return type + honest (real `Task`s support `.cancel()`, `.done()`, callbacks) rather + than bare coroutines. See https://docs.python.org/3/library/asyncio-task.html#coroutines: "Note that simply calling a coroutine will not schedule it to be executed:" """ if limit is None: - return (asyncio.ensure_future(func(*item)) for item in items) + return [asyncio.ensure_future(func(*item)) for item in items] sem = asyncio.Semaphore(limit) @@ -120,7 +120,7 @@ async def run(item: T) -> V: async with sem: return await func(*item) - return (asyncio.ensure_future(run(item)) for item in items) + return [asyncio.ensure_future(run(item)) for item in items] async def concurrent_map[T: tuple[Any, ...], V]( diff --git a/tests/benchmarks/test_e2e.py b/tests/benchmarks/test_e2e.py index de69fca59b..9720778d8f 100644 --- a/tests/benchmarks/test_e2e.py +++ b/tests/benchmarks/test_e2e.py @@ -63,7 +63,8 @@ def _data(shape: tuple[int]) -> np.ndarray: noise_level = 1 pattern = (np.sin(np.linspace(0, 2 * np.pi, period)) * 50 + 128).round().astype(np.uint8) data = np.tile(pattern, int(np.ceil(n / period)))[:n].astype(np.int16) - data += np.random.randint(-noise_level, noise_level + 1, size=n, dtype=np.int16) + rng = np.random.default_rng(0) + data += rng.integers(-noise_level, noise_level + 1, size=n, dtype=np.int16) return np.clip(data, 0, 255).astype(np.uint8) @@ -189,7 +190,7 @@ def test_read_array( get_data: Callable[[tuple[int]], np.ndarray | int], ) -> None: """ - Test the time required to fill an array with a single value + Test the time required to read the entirety of an array """ arr = create_array( bench_store, diff --git a/tests/test_codecs/test_codecs.py b/tests/test_codecs/test_codecs.py index 01ac02920f..8b4585503c 100644 --- a/tests/test_codecs/test_codecs.py +++ b/tests/test_codecs/test_codecs.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import warnings from dataclasses import dataclass from typing import TYPE_CHECKING, Any @@ -22,7 +23,7 @@ from zarr.core.metadata.v3 import ArrayV3Metadata from zarr.dtype import UInt8 from zarr.errors import ZarrUserWarning -from zarr.storage import StorePath +from zarr.storage import MemoryStore, StorePath if TYPE_CHECKING: from zarr.abc.codec import Codec @@ -375,6 +376,49 @@ def test_invalid_metadata_create_array() -> None: ) +@pytest.mark.parametrize( + "pipeline_path", + [ + "zarr.core.codec_pipeline.BatchedCodecPipeline", + "zarr.core.codec_pipeline.FusedCodecPipeline", + ], +) +def test_sharding_warning_fires_once_per_open(pipeline_path: str) -> None: + """Construction-time codec warnings (e.g. sharding's partial-reads warning) + must fire exactly once per array open, not once per internal codec-chain + reconstruction. + + `create_codec_pipeline` builds a throwaway pipeline via `from_codecs` (which + warns) and then calls `evolve_from_array_spec` on it, which re-splits the + (already-warned-about) codec chain against the evolved spec. That re-split + goes through `codecs_from_list_unchecked` rather than `codecs_from_list`, so + it does not re-emit the warning. `FusedCodecPipeline` additionally builds a + `ChunkTransform` (and, on the async fallback path, an `AsyncChunkTransform` + per call) from the same evolved codec chain, which must use the same quiet + variant. + """ + with config.set({"codec_pipeline.path": pipeline_path}): + store = MemoryStore() + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + zarr.create_array( + store, + shape=(16, 16), + chunks=(16, 16), + dtype=np.dtype("uint8"), + fill_value=0, + serializer=ShardingCodec(chunk_shape=(8, 8)), + compressors=[GzipCodec()], + ) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + zarr.open_array(store, mode="r") + + matches = [w for w in caught if "disables partial reads" in str(w.message)] + assert len(matches) == 1 + + @pytest.mark.parametrize("store", ["local", "memory"], indirect=["store"]) async def test_resize(store: Store) -> None: data = np.zeros((16, 18), dtype="uint16") diff --git a/tests/test_common.py b/tests/test_common.py index 2fe0743e14..5d8df326da 100644 --- a/tests/test_common.py +++ b/tests/test_common.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from collections.abc import Iterable from typing import TYPE_CHECKING, get_args @@ -9,6 +10,7 @@ from zarr.core.common import ( ANY_ACCESS_MODE, AccessModeLiteral, + concurrent_iter, parse_int, parse_name, parse_shapelike, @@ -32,6 +34,32 @@ def test_access_modes() -> None: assert set(ANY_ACCESS_MODE) == set(get_args(AccessModeLiteral)) +async def test_concurrent_iter_schedules_eagerly() -> None: + """`concurrent_iter` must return already-scheduled tasks, not a lazy generator. + + Its docstring promises `func(*item)` is launched concurrently for every + item up front; a caller that awaits the returned tasks one at a time + (rather than via `gather`/`as_completed`, which force iteration) relies + on that eager scheduling to get any overlap at all. + """ + started = [False, False, False] + + async def mark(i: int) -> int: + started[i] = True + return i + + tasks = concurrent_iter([(0,), (1,), (2,)], mark) + + # Give the event loop one chance to run before awaiting anything + # individually. If `concurrent_iter` were lazy, nothing would have been + # scheduled yet and `started` would still be all-False here. + await asyncio.sleep(0) + assert started == [True, True, True] + + results = [await t for t in tasks] + assert results == [0, 1, 2] + + # todo: test def test_concurrent_map() -> None: ... diff --git a/tests/test_fused_pipeline.py b/tests/test_fused_pipeline.py index fd86936853..7fa3ef2277 100644 --- a/tests/test_fused_pipeline.py +++ b/tests/test_fused_pipeline.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio from dataclasses import dataclass, field, replace from typing import TYPE_CHECKING, Any @@ -25,8 +26,9 @@ from zarr.storage import MemoryStore, StorePath if TYPE_CHECKING: + from zarr.abc.store import ByteRequest from zarr.core.array_spec import ArraySpec - from zarr.core.buffer import Buffer, NDBuffer + from zarr.core.buffer import Buffer, BufferPrototype, NDBuffer @pytest.mark.parametrize( @@ -439,6 +441,103 @@ def test_thread_pool_read_worker_exception_propagates() -> None: arr[:] +def test_resolve_max_workers_warns_and_falls_back_on_invalid_config() -> None: + """`codec_pipeline.max_workers` arrives via the config/env layer (e.g. + `ZARR_CODEC_PIPELINE__MAX_WORKERS`), so garbage input should warn and fall + back to the default rather than raising mid-read. + """ + import os + + import zarr.core.codec_pipeline as cp_mod + from zarr.errors import ZarrUserWarning + + default = os.cpu_count() or 1 + with zarr_config.set({"codec_pipeline.max_workers": "fast"}): + with pytest.warns(ZarrUserWarning, match="max_workers"): + result = cp_mod._resolve_max_workers() + assert result == default + + +async def test_encode_and_write_as_completed_cancels_stray_writes_on_failure() -> None: + """A failing write must not leave sibling writes running in the background. + + `_encode_and_write_as_completed` fires one write task per chunk as soon as + its encode completes, then `gather`s them. Plain `gather` (without + `return_exceptions=True`) re-raises the first exception without cancelling + the other in-flight tasks, so a still-running write would keep going after + the caller has already seen the exception -- and its eventual outcome is + never retrieved (an unraisable "Task exception was never retrieved" + warning if it later fails). + """ + from zarr.core.array_spec import ArrayConfig, ArraySpec + from zarr.core.buffer import default_buffer_prototype + from zarr.core.buffer.cpu import NDBuffer as CPUNDBuffer + from zarr.core.chunk_utils import ChunkTransform + from zarr.core.codec_pipeline import _encode_and_write_as_completed + from zarr.core.dtype import get_data_type_from_native_dtype + + write_started = asyncio.Event() + write_finished = False + + class _SlowByteSetter: + async def get( + self, prototype: BufferPrototype, byte_range: ByteRequest | None = None + ) -> Buffer | None: + return None + + async def set(self, value: Buffer) -> None: + nonlocal write_finished + write_started.set() + await asyncio.sleep(0.2) + write_finished = True + + async def delete(self) -> None: + pass + + async def set_if_not_exists(self, default: Buffer) -> None: + pass + + class _FailingByteSetter: + async def get( + self, prototype: BufferPrototype, byte_range: ByteRequest | None = None + ) -> Buffer | None: + return None + + async def set(self, value: Buffer) -> None: + raise RuntimeError("simulated write failure") + + async def delete(self) -> None: + pass + + async def set_if_not_exists(self, default: Buffer) -> None: + pass + + zdtype = get_data_type_from_native_dtype(np.dtype("uint8")) + chunk_spec = ArraySpec( + shape=(1,), + dtype=zdtype, + fill_value=zdtype.cast_scalar(0), + config=ArrayConfig(order="C", write_empty_chunks=True), + prototype=default_buffer_prototype(), + ) + chunk_array = CPUNDBuffer.from_numpy_array(np.zeros(1, dtype="uint8")) + transform = ChunkTransform(codecs=(BytesCodec(),)) + + batch = [ + (_SlowByteSetter(), chunk_array, chunk_spec), + (_FailingByteSetter(), chunk_array, chunk_spec), + ] + + with pytest.raises(RuntimeError, match="simulated write failure"): + await _encode_and_write_as_completed(batch, transform) # type: ignore[arg-type] + + assert write_started.is_set() + # Give the slow write's sleep long enough to finish if it were left + # running unattended in the background instead of being cancelled. + await asyncio.sleep(0.3) + assert not write_finished, "the slow write should have been cancelled, not left running" + + def test_concurrent_reads_shared_transform_with_pool() -> None: """Concurrent decode through the shared ChunkTransform produces correct data. diff --git a/tests/test_store/test_local.py b/tests/test_store/test_local.py index 61e48a269f..90d214ee2c 100644 --- a/tests/test_store/test_local.py +++ b/tests/test_store/test_local.py @@ -122,58 +122,6 @@ async def test_move( ): await store2.move(destination) - # --- byte-range-write tests: disabled --- - # Byte-range-write support (set_range / set_range_sync / SupportsSetRange) - # was removed from this PR pending a decision on the store interface. These - # tests are known-good and kept commented out to restore once that lands. - # def test_supports_set_range(self, store: LocalStore) -> None: - # """LocalStore should implement SupportsSetRange.""" - # assert isinstance(store, SupportsSetRange) - # - # @pytest.mark.parametrize( - # ("start", "patch", "expected"), - # [ - # (0, b"XX", b"XXAAAAAAAA"), - # (3, b"XX", b"AAAXXAAAAA"), - # (8, b"XX", b"AAAAAAAAXX"), - # (0, b"ZZZZZZZZZZ", b"ZZZZZZZZZZ"), - # (5, b"B", b"AAAAABAAAA"), - # (0, b"BCDE", b"BCDEAAAAAA"), - # ], - # ids=["start", "middle", "end", "full-overwrite", "single-byte", "multi-byte-start"], - # ) - # async def test_set_range( - # self, store: LocalStore, start: int, patch: bytes, expected: bytes - # ) -> None: - # """set_range should overwrite bytes at the given offset.""" - # await store.set("test/key", cpu.Buffer.from_bytes(b"AAAAAAAAAA")) - # await store.set_range("test/key", cpu.Buffer.from_bytes(patch), start=start) - # result = await store.get("test/key", prototype=cpu.buffer_prototype) - # assert result is not None - # assert result.to_bytes() == expected - # - # @pytest.mark.parametrize( - # ("start", "patch", "expected"), - # [ - # (0, b"XX", b"XXAAAAAAAA"), - # (3, b"XX", b"AAAXXAAAAA"), - # (8, b"XX", b"AAAAAAAAXX"), - # (0, b"ZZZZZZZZZZ", b"ZZZZZZZZZZ"), - # (5, b"B", b"AAAAABAAAA"), - # (0, b"BCDE", b"BCDEAAAAAA"), - # ], - # ids=["start", "middle", "end", "full-overwrite", "single-byte", "multi-byte-start"], - # ) - # def test_set_range_sync( - # self, store: LocalStore, start: int, patch: bytes, expected: bytes - # ) -> None: - # """set_range_sync should overwrite bytes at the given offset.""" - # sync(store.set("test/key", cpu.Buffer.from_bytes(b"AAAAAAAAAA"))) - # store.set_range_sync("test/key", cpu.Buffer.from_bytes(patch), start=start) - # result = store.get_sync(key="test/key", prototype=cpu.buffer_prototype) - # assert result is not None - # assert result.to_bytes() == expected - @pytest.mark.parametrize("exclusive", [True, False]) def test_atomic_write_successful(tmp_path: pathlib.Path, exclusive: bool) -> None: diff --git a/tests/test_store/test_memory.py b/tests/test_store/test_memory.py index a976f3738e..013dae7044 100644 --- a/tests/test_store/test_memory.py +++ b/tests/test_store/test_memory.py @@ -126,59 +126,6 @@ def test_write_does_not_alias_source_array( np.testing.assert_array_equal(array[:], expected) - # --- byte-range-write tests: disabled --- - # Byte-range-write support (set_range / set_range_sync / SupportsSetRange) - # was removed from this PR pending a decision on the store interface. These - # tests are known-good and kept commented out to restore once that lands. - # def test_supports_set_range(self, store: MemoryStore) -> None: - # """MemoryStore should implement SupportsSetRange.""" - # assert isinstance(store, SupportsSetRange) - # - # @pytest.mark.parametrize( - # ("start", "patch", "expected"), - # [ - # (0, b"XX", b"XXAAAAAAAA"), - # (3, b"XX", b"AAAXXAAAAA"), - # (8, b"XX", b"AAAAAAAAXX"), - # (0, b"ZZZZZZZZZZ", b"ZZZZZZZZZZ"), - # (5, b"B", b"AAAAABAAAA"), - # (0, b"BCDE", b"BCDEAAAAAA"), - # ], - # ids=["start", "middle", "end", "full-overwrite", "single-byte", "multi-byte-start"], - # ) - # async def test_set_range( - # self, store: MemoryStore, start: int, patch: bytes, expected: bytes - # ) -> None: - # """set_range should overwrite bytes at the given offset.""" - # await store.set("test/key", cpu.Buffer.from_bytes(b"AAAAAAAAAA")) - # await store.set_range("test/key", cpu.Buffer.from_bytes(patch), start=start) - # result = await store.get("test/key", prototype=cpu.buffer_prototype) - # assert result is not None - # assert result.to_bytes() == expected - # - # @pytest.mark.parametrize( - # ("start", "patch", "expected"), - # [ - # (0, b"XX", b"XXAAAAAAAA"), - # (3, b"XX", b"AAAXXAAAAA"), - # (8, b"XX", b"AAAAAAAAXX"), - # (0, b"ZZZZZZZZZZ", b"ZZZZZZZZZZ"), - # (5, b"B", b"AAAAABAAAA"), - # (0, b"BCDE", b"BCDEAAAAAA"), - # ], - # ids=["start", "middle", "end", "full-overwrite", "single-byte", "multi-byte-start"], - # ) - # def test_set_range_sync( - # self, store: MemoryStore, start: int, patch: bytes, expected: bytes - # ) -> None: - # """set_range_sync should overwrite bytes at the given offset.""" - # store._is_open = True - # store._store_dict["test/key"] = cpu.Buffer.from_bytes(b"AAAAAAAAAA") - # store.set_range_sync("test/key", cpu.Buffer.from_bytes(patch), start=start) - # result = store.get_sync(key="test/key", prototype=cpu.buffer_prototype) - # assert result is not None - # assert result.to_bytes() == expected - # TODO: fix this warning @pytest.mark.filterwarnings("ignore:Unclosed client session:ResourceWarning")