update docs, warn when config is used to set concurrency limits

Author: d-v-b

changes/3547.misc.md

@@ -1 +1,2 @@
-Moved concurrency limits to a global per-event loop setting instead of per-array call.
+Moved concurrency-limiting functionality to store classes. The global configuration object no longer
+controls concurrency limits. Concurrency limits, if applicable, must now be specified when constructing a store.

docs/user-guide/config.md

@@ -30,7 +30,7 @@ Configuration options include the following:
 - Default Zarr format `default_zarr_version`
 - Default array order in memory `array.order`
 - Whether empty chunks are written to storage `array.write_empty_chunks`
-- Async and threading options, e.g. `async.concurrency` and `threading.max_workers`
+- Threading options, e.g. `threading.max_workers`
 - Selections of implementations of codecs, codec pipelines and buffers
 - Enabling GPU support with `zarr.config.enable_gpu()`. See GPU support for more.
 

docs/user-guide/performance.md

@@ -191,20 +191,18 @@ scenarios.
 ### Concurrent I/O operations
 
 Zarr uses asynchronous I/O internally to enable concurrent reads and writes across multiple chunks.
-The level of concurrency is controlled by the `async.concurrency` configuration setting, which
-determines the maximum number of concurrent I/O operations.
-
-The default value is 10, which is a conservative value. You may get improved performance by tuning
-the concurrency limit. You can adjust this value based on your specific needs:
+Concurrency is controlled at the **store level** — each store instance can have its own concurrency
+limit, set via the `concurrency_limit` parameter when creating the store.
 
 ```python
 import zarr
 
-# Set concurrency for the current session
-zarr.config.set({'async.concurrency': 128})
+# Local filesystem store with custom concurrency limit
+store = zarr.storage.LocalStore("data/my_array.zarr", concurrency_limit=64)
 
-# Or use environment variable
-# export ZARR_ASYNC_CONCURRENCY=128
+# 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:
@@ -217,32 +215,36 @@ Lower concurrency values may be beneficial when:
 - Memory is constrained (each concurrent operation requires buffer space)
 - Using Zarr within a parallel computing framework (see below)
 
+Set `concurrency_limit=None` to disable the concurrency limit entirely.
+
 ### 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 Zarr's concurrency settings.
+[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.
 
-**Important**: When using many Dask threads, you may need to reduce both Zarr's `async.concurrency` and `threading.max_workers` settings to avoid creating too many concurrent operations. The total number of concurrent I/O operations can be roughly estimated as:
+**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 × zarr_async_concurrency
+total_concurrency ≈ dask_threads × store_concurrency_limit
 ```
 
-For example, if you're running Dask with 10 threads and Zarr's default concurrency of 64, you could potentially have up to 640 concurrent operations, which may overwhelm your storage system or cause memory issues.
+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 Zarr's concurrency settings:
+**Recommendation**: When using Dask with many threads, configure concurrency settings:
 
 ```python
 import zarr
 import dask.array as da
 
-# If using Dask with many threads (e.g., 8-16), reduce Zarr's concurrency settings
+# 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
 zarr.config.set({
-    'async.concurrency': 4,      # Limit concurrent async operations
     'threading.max_workers': 4,  # Limit Zarr's internal thread pool
 })
 
 # Open Zarr array
-z = zarr.open_array('data/large_array.zarr', mode='r')
+z = zarr.open_array(store=store, mode='r')
 
 # Create Dask array from Zarr array
 arr = da.from_array(z, chunks=z.chunks)
@@ -253,8 +255,8 @@ result = arr.mean(axis=0).compute()
 
 **Configuration guidelines for Dask workloads**:
 
-- `async.concurrency`: Controls the maximum number of concurrent async I/O operations. Start with a lower value (e.g., 4-8) when using many Dask threads.
-- `threading.max_workers`: Controls Zarr's internal thread pool size for blocking operations (defaults to CPU count). Reduce this to avoid thread contention with Dask's scheduler.
+- `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.
 

src/zarr/core/config.py

@@ -29,13 +29,32 @@
 
 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"
@@ -55,6 +74,25 @@ 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()
@@ -98,7 +136,7 @@ def enable_gpu(self) -> ConfigSet:
                 "write_empty_chunks": False,
                 "target_shard_size_bytes": None,
             },
-            "async": {"concurrency": 10, "timeout": None},
+            "async": {"timeout": None},
             "threading": {"max_workers": None},
             "json_indent": 2,
             "codec_pipeline": {

tests/test_config.py

@@ -55,7 +55,7 @@ def test_config_defaults_set() -> None:
                     "write_empty_chunks": False,
                     "target_shard_size_bytes": None,
                 },
-                "async": {"concurrency": 10, "timeout": None},
+                "async": {"timeout": None},
                 "threading": {"max_workers": None},
                 "json_indent": 2,
                 "codec_pipeline": {
@@ -101,15 +101,14 @@ def test_config_defaults_set() -> None:
         ]
     )
     assert config.get("array.order") == "C"
-    assert config.get("async.concurrency") == 10
     assert config.get("async.timeout") is None
     assert config.get("codec_pipeline.batch_size") == 1
     assert config.get("json_indent") == 2
 
 
 @pytest.mark.parametrize(
     ("key", "old_val", "new_val"),
-    [("array.order", "C", "F"), ("async.concurrency", 10, 128), ("json_indent", 2, 0)],
+    [("array.order", "C", "F"), ("json_indent", 2, 0)],
 )
 def test_config_defaults_can_be_overridden(key: str, old_val: Any, new_val: Any) -> None:
     assert config.get(key) == old_val
@@ -347,3 +346,15 @@ def test_deprecated_config(key: str) -> None:
     with pytest.raises(ValueError):
         with zarr.config.set({key: "foo"}):
             pass
+
+
+def test_async_concurrency_config_warns() -> None:
+    """Test that setting async.concurrency emits a warning directing users to per-store config."""
+    with pytest.warns(UserWarning, match="async.concurrency.*no effect"):
+        with zarr.config.set({"async.concurrency": 20}):
+            pass
+
+    # Also test the kwarg form
+    with pytest.warns(UserWarning, match="async.concurrency.*no effect"):
+        with zarr.config.set(async__concurrency=20):
+            pass