refactor: just remove the concurrency limits entirely

Author: d-v-b

docs/user-guide/performance.md

@@ -190,29 +190,7 @@ scenarios.
 
 ### Concurrent I/O operations
 
-For latency-sensitve storage backends like HTTP and cloud object storage, Zarr uses asynchronous I/O internally to enable concurrent reads and writes across multiple chunks.
-Concurrency is controlled at the **store level**. Many of the stores defined in `zarr-python` accept a concurrency limit on construction via the `concurrency_limit` parameter.
-
-```python
-import zarr
-
-# Local filesystem store with custom concurrency limit
-store = zarr.storage.LocalStore("data/my_array.zarr", concurrency_limit=64)
-
-# Remote store with higher concurrency for network I/O
-from obstore.store import S3Store
-store = zarr.storage.ObjectStore(S3Store.from_url("s3://bucket/path"), concurrency_limit=128)
-```
-
-Higher concurrency values can improve throughput when:
-- Working with remote storage (e.g., S3, GCS) where network latency is high
-- Reading/writing many small chunks in parallel
-- The storage backend can handle many concurrent requests
-
-Lower concurrency values may be beneficial when:
-- Working with local storage with limited I/O bandwidth
-- Memory is constrained (each concurrent operation requires buffer space)
-- Using Zarr within a parallel computing framework (see below)
+For latency-sensitive storage backends like HTTP and cloud object storage, Zarr uses asynchronous I/O internally to enable concurrent reads and writes across multiple chunks. Zarr does not impose its own concurrency limits — storage backends are expected to manage their own concurrency constraints (e.g., connection pool sizes, rate limits). If you need to limit concurrency for a particular backend, configure it at the storage layer (e.g., via fsspec or obstore options).
 
 ### Thread pool size (`threading.max_workers`)
 
@@ -238,32 +216,21 @@ concurrently.
 
 ### Using Zarr with Dask
 
-[Dask](https://www.dask.org/) is a popular parallel computing library that works well with Zarr for processing large arrays. When using Zarr with Dask, it's important to consider the interaction between Dask's thread pool and the store's concurrency limit.
+[Dask](https://www.dask.org/) is a popular parallel computing library that works well with Zarr for processing large arrays. When using Zarr with Dask, it's important to consider the interaction between Dask's thread pool and Zarr's internal thread pool.
 
-**Important**: When using many Dask threads, you may need to reduce the store's `concurrency_limit` and Zarr's `threading.max_workers` setting to avoid creating too many concurrent operations. The total number of concurrent I/O operations can be roughly estimated as:
-
-```
-total_concurrency ≈ dask_threads × store_concurrency_limit
-```
-
-For example, if you're running Dask with 10 threads and a store concurrency limit of 64, you could potentially have up to 640 concurrent operations, which may overwhelm your storage system or cause memory issues.
-
-**Recommendation**: When using Dask with many threads, configure concurrency settings:
+**Recommendation**: When using Dask with many threads, reduce Zarr's internal thread pool to avoid thread contention:
 
 ```python
 import zarr
 import dask.array as da
 
-# Create store with reduced concurrency limit for Dask workloads
-store = zarr.storage.LocalStore("data/large_array.zarr", concurrency_limit=4)
-
-# Also limit Zarr's internal thread pool
+# Limit Zarr's internal thread pool
 zarr.config.set({
     'threading.max_workers': 4,  # Limit Zarr's internal thread pool
 })
 
 # Open Zarr array
-z = zarr.open_array(store=store, mode='r')
+z = zarr.open_array("data/large_array.zarr", mode='r')
 
 # Create Dask array from Zarr array
 arr = da.from_array(z, chunks=z.chunks)
@@ -272,13 +239,6 @@ arr = da.from_array(z, chunks=z.chunks)
 result = arr.mean(axis=0).compute()
 ```
 
-**Configuration guidelines for Dask workloads**:
-
-- `concurrency_limit` (per-store): Controls the maximum number of concurrent async I/O operations for a given store. Start with a lower value (e.g., 4-8) when using many Dask threads.
-- `threading.max_workers` (global config): Controls Zarr's internal thread pool size for blocking operations (defaults to CPU count). Reduce this to avoid thread contention with Dask's scheduler.
-
-You may need to experiment with different values to find the optimal balance for your workload. Monitor your system's resource usage and adjust these settings based on whether your storage system or CPU is the bottleneck.
-
 ### Thread safety and process safety
 
 Zarr arrays are designed to be thread-safe for concurrent reads and writes from multiple threads within the same process. However, proper synchronization is required when writing to overlapping regions from multiple threads.

src/zarr/abc/codec.py

@@ -246,7 +246,6 @@ async def decode_partial(
         -------
         Iterable[NDBuffer | None]
         """
-        # Store handles concurrency limiting internally
         return await asyncio.gather(*[self._decode_partial_single(*info) for info in batch_info])
 
 
@@ -280,7 +279,6 @@ async def encode_partial(
             The ByteSetter is used to write the necessary bytes and fetch bytes for existing chunk data.
             The chunk spec contains information about the chunk.
         """
-        # Store handles concurrency limiting internally
         await asyncio.gather(*[self._encode_partial_single(*info) for info in batch_info])
 
 

src/zarr/core/config.py

@@ -29,32 +29,13 @@
 
 from __future__ import annotations
 
-import os
-import warnings
 from typing import TYPE_CHECKING, Any, Literal, cast
 
 from donfig import Config as DConfig
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping
-
     from donfig.config_obj import ConfigSet
 
-# Config keys that have been moved from global config to per-store parameters.
-# Maps old config key to a warning message.
-_warn_on_set: dict[str, str] = {
-    "async.concurrency": (
-        "The 'async.concurrency' configuration key has no effect. "
-        "Concurrency limits are now set per-store via the 'concurrency_limit' "
-        "parameter. For example: zarr.storage.LocalStore(..., concurrency_limit=10)."
-    ),
-}
-
-# Environment variable forms of the keys above (ZARR_ASYNC__CONCURRENCY -> async.concurrency)
-_warn_on_set_env: dict[str, str] = {
-    "ZARR_ASYNC__CONCURRENCY": _warn_on_set["async.concurrency"],
-}
-
 
 class BadConfigError(ValueError):
     _msg = "bad Config: %r"
@@ -74,25 +55,6 @@ class Config(DConfig):  # type: ignore[misc]
 
     """
 
-    def set(self, arg: Mapping[str, Any] | None = None, **kwargs: Any) -> ConfigSet:
-        # Check for keys that now belong to per-store config
-        if arg is not None:
-            for key in arg:
-                if key in _warn_on_set:
-                    warnings.warn(_warn_on_set[key], UserWarning, stacklevel=2)
-        for key in kwargs:
-            normalized = key.replace("__", ".")
-            if normalized in _warn_on_set:
-                warnings.warn(_warn_on_set[normalized], UserWarning, stacklevel=2)
-        return super().set(arg, **kwargs)
-
-    def refresh(self, **kwargs: Any) -> None:
-        # Warn if env vars are being used for removed config keys
-        for env_key, message in _warn_on_set_env.items():
-            if env_key in os.environ:
-                warnings.warn(message, UserWarning, stacklevel=2)
-        super().refresh(**kwargs)
-
     def reset(self) -> None:
         self.clear()
         self.refresh()

src/zarr/storage/__init__.py

@@ -10,12 +10,10 @@
 from zarr.storage._logging import LoggingStore
 from zarr.storage._memory import GpuMemoryStore, MemoryStore
 from zarr.storage._obstore import ObjectStore
-from zarr.storage._utils import ConcurrencyLimiter
 from zarr.storage._wrapper import WrapperStore
 from zarr.storage._zip import ZipStore
 
 __all__ = [
-    "ConcurrencyLimiter",
     "FsspecStore",
     "GpuMemoryStore",
     "LocalStore",

src/zarr/storage/_fsspec.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import json
 import warnings
 from contextlib import suppress
@@ -18,7 +17,6 @@
 from zarr.core.buffer import Buffer
 from zarr.errors import ZarrUserWarning
 from zarr.storage._common import _dereference_path
-from zarr.storage._utils import ConcurrencyLimiter, with_concurrency_limit
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Iterable
@@ -69,7 +67,7 @@ def _make_async(fs: AbstractFileSystem) -> AsyncFileSystem:
     return AsyncFileSystemWrapper(fs, asynchronous=True)
 
 
-class FsspecStore(Store, ConcurrencyLimiter):
+class FsspecStore(Store):
     """
     Store for remote data based on FSSpec.
 
@@ -84,9 +82,6 @@ class FsspecStore(Store, ConcurrencyLimiter):
         filesystem scheme.
     allowed_exceptions : tuple[type[Exception], ...]
         When fetching data, these cases will be deemed to correspond to missing keys.
-    concurrency_limit : int, optional
-        Maximum number of concurrent I/O operations. Default is 50.
-        Set to None for unlimited concurrency.
 
     Attributes
     ----------
@@ -130,10 +125,8 @@ def __init__(
         read_only: bool = False,
         path: str = "/",
         allowed_exceptions: tuple[type[Exception], ...] = ALLOWED_EXCEPTIONS,
-        concurrency_limit: int | None = 50,
     ) -> None:
-        Store.__init__(self, read_only=read_only)
-        ConcurrencyLimiter.__init__(self, concurrency_limit)
+        super().__init__(read_only=read_only)
         self.fs = fs
         self.path = path
         self.allowed_exceptions = allowed_exceptions
@@ -259,7 +252,6 @@ def with_read_only(self, read_only: bool = False) -> FsspecStore:
             path=self.path,
             allowed_exceptions=self.allowed_exceptions,
             read_only=read_only,
-            concurrency_limit=self.concurrency_limit,
         )
 
     async def clear(self) -> None:
@@ -282,7 +274,6 @@ def __eq__(self, other: object) -> bool:
             and self.fs == other.fs
         )
 
-    @with_concurrency_limit
     async def get(
         self,
         key: str,
@@ -325,7 +316,6 @@ async def get(
         else:
             return value
 
-    @with_concurrency_limit
     async def set(
         self,
         key: str,
@@ -346,24 +336,6 @@ async def set(
             raise NotImplementedError
         await self.fs._pipe_file(path, value.to_bytes())
 
-    async def _set_many(self, values: Iterable[tuple[str, Buffer]]) -> None:
-        # Override to avoid deadlock from calling decorated set() method
-        if not self._is_open:
-            await self._open()
-        self._check_writable()
-
-        async def _set_with_limit(key: str, value: Buffer) -> None:
-            if not isinstance(value, Buffer):
-                raise TypeError(
-                    f"FsspecStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead."
-                )
-            path = _dereference_path(self.path, key)
-            async with self._limit():
-                await self.fs._pipe_file(path, value.to_bytes())
-
-        await asyncio.gather(*[_set_with_limit(key, value) for key, value in values])
-
-    @with_concurrency_limit
     async def delete(self, key: str) -> None:
         # docstring inherited
         self._check_writable()