remove unused or unneeded routines

Author: d-v-b

src/zarr/core/sync.py

@@ -13,7 +13,7 @@
 from zarr.core.config import config
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine
+    from collections.abc import AsyncIterator, Coroutine
     from typing import Any
 
 logger = logging.getLogger(__name__)
@@ -211,17 +211,3 @@ async def iter_to_list() -> list[T]:
             return [item async for item in async_iterator]
 
         return self._sync(iter_to_list())
-
-
-async def _with_semaphore(
-    func: Callable[[], Awaitable[T]], semaphore: asyncio.Semaphore | None = None
-) -> T:
-    """
-    Await the result of invoking the no-argument-callable ``func`` within the context manager
-    provided by a Semaphore, if one is provided. Otherwise, just await the result of invoking
-    ``func``.
-    """
-    if semaphore is None:
-        return await func()
-    async with semaphore:
-        return await func()

src/zarr/storage/_local.py

@@ -19,7 +19,6 @@
 )
 from zarr.core.buffer import Buffer
 from zarr.core.buffer.core import default_buffer_prototype
-from zarr.storage._utils import ConcurrencyLimiter, with_concurrency_limit
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Iterable, Iterator
@@ -86,7 +85,7 @@ def _put(path: Path, value: Buffer, exclusive: bool = False) -> int:
         return f.write(view)
 
 
-class LocalStore(Store, ConcurrencyLimiter):
+class LocalStore(Store):
     """
     Store for the local file system.
 
@@ -96,9 +95,6 @@ class LocalStore(Store, ConcurrencyLimiter):
         Directory to use as root of store.
     read_only : bool
         Whether the store is read-only
-    concurrency_limit : int, optional
-        Maximum number of concurrent I/O operations. Default is 100.
-        Set to None for unlimited concurrency.
 
     Attributes
     ----------
@@ -119,24 +115,21 @@ def __init__(
         root: Path | str,
         *,
         read_only: bool = False,
-        concurrency_limit: int | None = 100,
     ) -> None:
         if isinstance(root, str):
             root = Path(root)
         if not isinstance(root, Path):
             raise TypeError(
                 f"'root' must be a string or Path instance. Got an instance of {type(root)} instead."
             )
-        Store.__init__(self, read_only=read_only)
-        ConcurrencyLimiter.__init__(self, concurrency_limit)
+        super().__init__(read_only=read_only)
         self.root = root
 
     def with_read_only(self, read_only: bool = False) -> Self:
         # docstring inherited
         return type(self)(
             root=self.root,
             read_only=read_only,
-            concurrency_limit=self.concurrency_limit,
         )
 
     @classmethod
@@ -199,7 +192,6 @@ def __repr__(self) -> str:
     def __eq__(self, other: object) -> bool:
         return isinstance(other, type(self)) and self.root == other.root
 
-    @with_concurrency_limit
     async def get(
         self,
         key: str,
@@ -225,19 +217,8 @@ async def get_partial_values(
         key_ranges: Iterable[tuple[str, ByteRequest | None]],
     ) -> list[Buffer | None]:
         # docstring inherited
-        # We directly call the I/O functions here, wrapped with the semaphore,
-        # to avoid deadlock from calling the decorated get() method.
-
-        async def _get_with_limit(key: str, byte_range: ByteRequest | None) -> Buffer | None:
-            path = self.root / key
-            try:
-                async with self._limit():
-                    return await asyncio.to_thread(_get, path, prototype, byte_range)
-            except (FileNotFoundError, IsADirectoryError, NotADirectoryError):
-                return None
-
         return await asyncio.gather(
-            *[_get_with_limit(key, byte_range) for key, byte_range in key_ranges]
+            *[self.get(key, prototype, byte_range) for key, byte_range in key_ranges]
         )
 
     async def set(self, key: str, value: Buffer) -> None:
@@ -251,7 +232,6 @@ async def set_if_not_exists(self, key: str, value: Buffer) -> None:
         except FileExistsError:
             pass
 
-    @with_concurrency_limit
     async def _set(self, key: str, value: Buffer, exclusive: bool = False) -> None:
         if not self._is_open:
             await self._open()
@@ -264,7 +244,6 @@ async def _set(self, key: str, value: Buffer, exclusive: bool = False) -> None:
         path = self.root / key
         await asyncio.to_thread(_put, path, value, exclusive=exclusive)
 
-    @with_concurrency_limit
     async def delete(self, key: str) -> None:
         """
         Remove a key from the store.

src/zarr/testing/store_concurrency.py

@@ -75,56 +75,36 @@ def test_concurrency_limit_custom(self, store_kwargs: dict[str, Any]) -> None:
         assert store.concurrency_limit is None
         assert store._semaphore is None
 
-    async def test_concurrency_limit_enforced(self, store: S) -> None:
-        """Test that the concurrency limit is actually enforced during execution.
+    async def test_concurrency_limit_enforced(self, store_kwargs: dict[str, Any]) -> None:
+        """Test that store operations acquire the semaphore.
 
-        This test verifies that when many operations are submitted concurrently,
-        only up to the concurrency limit are actually executing at once.
+        Exhausts all semaphore slots externally, then verifies that a store
+        operation blocks until a slot is released.
         """
-        semaphore = self._get_semaphore(store)
-        if semaphore is None:
-            pytest.skip("Store has no concurrency limit")
-
-        assert isinstance(store, ConcurrencyLimiter)
-        assert store.concurrency_limit is not None  # type: ignore[unreachable]
-        limit = store.concurrency_limit
-
-        # We'll monitor the semaphore's available count
-        # When it reaches 0, that means `limit` operations are running
-        min_available = limit
-
-        async def monitored_operation(key: str, value: B) -> None:
-            nonlocal min_available
-            # Check semaphore state right after we're scheduled
-            await asyncio.sleep(0)  # Yield to ensure we're in the queue
-            available = semaphore._value
-            min_available = min(min_available, available)
+        if not issubclass(self.store_cls, ConcurrencyLimiter):
+            pytest.skip("Store does not support concurrency limits")
 
-            # Now do the actual operation (which will acquire the semaphore)
-            await store.set(key, value)
+        limit = 3
+        store = self.store_cls(**{**store_kwargs, "concurrency_limit": limit})  # type: ignore[unreachable]
+        await store._ensure_open()
 
-        # Launch more operations than the limit to ensure contention
-        num_ops = limit * 2
-        items = [
-            (f"limit_test_key_{i}", self.buffer_cls.from_bytes(f"value_{i}".encode()))
-            for i in range(num_ops)
-        ]
+        sem = store._semaphore
+        assert sem is not None
 
-        await asyncio.gather(*[monitored_operation(k, v) for k, v in items])
+        # Exhaust all slots
+        for _ in range(limit):
+            await sem.acquire()
+        assert sem._value == 0  # type: ignore[attr-defined]
 
-        # The semaphore should have been fully utilized (reached 0 or close to it)
-        # This indicates that `limit` operations were running concurrently
-        assert min_available < limit, (
-            f"Semaphore was never fully utilized. "
-            f"Min available: {min_available}, Limit: {limit}. "
-            f"This suggests operations aren't running concurrently."
-        )
+        # A store operation should block because no slots are available
+        buf = self.buffer_cls.from_bytes(b"x")
+        task = asyncio.create_task(store.set("key", buf))
+        await asyncio.sleep(0)  # let the task attempt to acquire
+        assert not task.done()
 
-        # Ideally it should reach 0, but allow some slack for timing
-        assert min_available <= 5, (
-            f"Semaphore only reached {min_available} available slots. "
-            f"Expected close to 0 with limit {limit}."
-        )
+        # Release one slot — the task should now complete
+        sem.release()
+        await asyncio.wait_for(task, timeout=5.0)
 
     async def test_batch_write_no_deadlock(self, store: S) -> None:
         """Test that batch writes don't deadlock when exceeding concurrency limit."""

tests/test_global_concurrency.py

@@ -15,7 +15,6 @@
 from zarr.storage import LocalStore
 from zarr.storage._local import _atomic_write
 from zarr.testing.store import StoreTests
-from zarr.testing.store_concurrency import StoreConcurrencyTests
 from zarr.testing.utils import assert_bytes_equal
 
 if TYPE_CHECKING:
@@ -206,14 +205,3 @@ def test_atomic_write_exclusive_preexisting(tmp_path: pathlib.Path) -> None:
     assert path.read_bytes() == b"xyz"
     assert list(path.parent.iterdir()) == [path]  # no temp files
 
-
-class TestLocalStoreConcurrency(StoreConcurrencyTests[LocalStore, cpu.Buffer]):
-    """Test LocalStore concurrency limiting behavior."""
-
-    store_cls = LocalStore
-    buffer_cls = cpu.Buffer
-    expected_concurrency_limit = 100  # LocalStore default
-
-    @pytest.fixture
-    def store_kwargs(self, tmpdir: str) -> dict[str, str]:
-        return {"root": str(tmpdir)}

tests/test_store/test_local.py

@@ -14,7 +14,6 @@
 from zarr.errors import ZarrUserWarning
 from zarr.storage import GpuMemoryStore, MemoryStore
 from zarr.testing.store import StoreTests
-from zarr.testing.store_concurrency import StoreConcurrencyTests
 from zarr.testing.utils import gpu_test
 
 if TYPE_CHECKING:
@@ -182,15 +181,3 @@ def test_from_dict(self) -> None:
             result = GpuMemoryStore.from_dict(d)
         for v in result._store_dict.values():
             assert type(v) is gpu.Buffer
-
-
-class TestMemoryStoreConcurrency(StoreConcurrencyTests[MemoryStore, cpu.Buffer]):
-    """Test MemoryStore concurrency limiting behavior."""
-
-    store_cls = MemoryStore
-    buffer_cls = cpu.Buffer
-    expected_concurrency_limit = None  # MemoryStore has no limit (fast in-memory ops)
-
-    @pytest.fixture
-    def store_kwargs(self) -> dict[str, Any]:
-        return {"store_dict": None}