remove more references to the global concurrency limit

Author: d-v-b

src/zarr/core/common.py

@@ -1,15 +1,11 @@
 from __future__ import annotations
 
-import asyncio
 import functools
 import math
 import operator
-import threading
 import warnings
-import weakref
 from collections.abc import Iterable, Mapping, Sequence
 from enum import Enum
-from itertools import starmap
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -29,7 +25,7 @@
 from zarr.errors import ZarrRuntimeWarning
 
 if TYPE_CHECKING:
-    from collections.abc import Awaitable, Callable, Iterator
+    from collections.abc import Iterator
 
 
 ZARR_JSON = "zarr.json"
@@ -96,139 +92,6 @@ def ceildiv(a: float, b: float) -> int:
     return math.ceil(a / b)
 
 
-T = TypeVar("T", bound=tuple[Any, ...])
-V = TypeVar("V")
-
-
-# Global semaphore management for per-process concurrency limiting
-# Use WeakKeyDictionary to automatically clean up semaphores when event loops are garbage collected
-_global_semaphores: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Semaphore] = (
-    weakref.WeakKeyDictionary()
-)
-# Use threading.Lock instead of asyncio.Lock to coordinate across event loops
-_global_semaphore_lock = threading.Lock()
-
-
-def get_global_semaphore() -> asyncio.Semaphore:
-    """
-    Get the global semaphore for the current event loop.
-
-    This ensures that all concurrent operations across the process share the same
-    concurrency limit, preventing excessive concurrent task creation when multiple
-    arrays or operations are running simultaneously.
-
-    The semaphore is lazily created per event loop and uses the configured
-    `async.concurrency` value from zarr config. The semaphore is cached per event
-    loop, so subsequent calls return the same semaphore instance.
-
-    Note: Config changes after the first call will not affect the semaphore limit.
-    To apply new config values, use :func:`reset_global_semaphores` to clear the cache.
-
-    Returns
-    -------
-    asyncio.Semaphore
-        The global semaphore for this event loop.
-
-    Raises
-    ------
-    RuntimeError
-        If called outside of an async context (no running event loop).
-
-    See Also
-    --------
-    reset_global_semaphores : Clear the global semaphore cache
-    """
-    loop = asyncio.get_running_loop()
-
-    # Acquire lock FIRST to prevent TOCTOU race condition
-    with _global_semaphore_lock:
-        if loop not in _global_semaphores:
-            limit = zarr_config.get("async.concurrency")
-            _global_semaphores[loop] = asyncio.Semaphore(limit)
-        return _global_semaphores[loop]
-
-
-def reset_global_semaphores() -> None:
-    """
-    Clear all cached global semaphores.
-
-    This is useful when you want config changes to take effect, or for testing.
-    The next call to :func:`get_global_semaphore` will create a new semaphore
-    using the current configuration.
-
-    Warning: This should only be called when no async operations are in progress,
-    as it will invalidate all existing semaphore references.
-
-    Examples
-    --------
-    >>> import zarr
-    >>> zarr.config.set({"async.concurrency": 50})
-    >>> reset_global_semaphores()  # Apply new config
-    """
-    with _global_semaphore_lock:
-        _global_semaphores.clear()
-
-
-async def concurrent_map(
-    items: Iterable[T],
-    func: Callable[..., Awaitable[V]],
-    limit: int | None = None,
-    *,
-    use_global_semaphore: bool = True,
-) -> list[V]:
-    """
-    Execute an async function concurrently over multiple items with concurrency limiting.
-
-    Parameters
-    ----------
-    items : Iterable[T]
-        Items to process, where each item is a tuple of arguments to pass to func.
-    func : Callable[..., Awaitable[V]]
-        Async function to execute for each item.
-    limit : int | None, optional
-        If provided and use_global_semaphore is False, creates a local semaphore
-        with this limit. If None, no concurrency limiting is applied.
-    use_global_semaphore : bool, default True
-        If True, uses the global per-process semaphore for concurrency limiting,
-        ensuring all concurrent operations share the same limit. If False, uses
-        the `limit` parameter for local limiting (legacy behavior).
-
-    Returns
-    -------
-    list[V]
-        Results from executing func on all items.
-    """
-    if use_global_semaphore:
-        if limit is not None:
-            raise ValueError(
-                "Cannot specify both use_global_semaphore=True and a limit value. "
-                "Either use the global semaphore (use_global_semaphore=True, limit=None) "
-                "or specify a local limit (use_global_semaphore=False, limit=<int>)."
-            )
-        # Use the global semaphore for process-wide concurrency limiting
-        sem = get_global_semaphore()
-
-        async def run(item: tuple[Any]) -> V:
-            async with sem:
-                return await func(*item)
-
-        return await asyncio.gather(*[asyncio.ensure_future(run(item)) for item in items])
-
-    elif limit is None:
-        # No concurrency limiting
-        return await asyncio.gather(*list(starmap(func, items)))
-
-    else:
-        # Legacy mode: create local semaphore with specified limit
-        sem = asyncio.Semaphore(limit)
-
-        async def run(item: tuple[Any]) -> V:
-            async with sem:
-                return await func(*item)
-
-        return await asyncio.gather(*[asyncio.ensure_future(run(item)) for item in items])
-
-
 E = TypeVar("E", bound=Enum)
 
 

src/zarr/core/group.py

@@ -44,7 +44,6 @@
     NodeType,
     ShapeLike,
     ZarrFormat,
-    get_global_semaphore,
     parse_shapelike,
 )
 from zarr.core.config import config
@@ -1441,13 +1440,10 @@ async def _members(
             )
 
             raise ValueError(msg)
-        # Use global semaphore for process-wide concurrency limiting
-        semaphore = get_global_semaphore()
         async for member in _iter_members_deep(
             self,
             max_depth=max_depth,
             skip_keys=skip_keys,
-            semaphore=semaphore,
             use_consolidated_for_children=use_consolidated_for_children,
         ):
             yield member
@@ -3324,13 +3320,11 @@ async def create_nodes(
         The created nodes in the order they are created.
     """
 
-    # Use global semaphore for process-wide concurrency limiting
-    semaphore = get_global_semaphore()
     create_tasks: list[Coroutine[None, None, str]] = []
 
     for key, value in nodes.items():
         # make the key absolute
-        create_tasks.extend(_persist_metadata(store, key, value, semaphore=semaphore))
+        create_tasks.extend(_persist_metadata(store, key, value))
 
     created_object_keys = []
 
@@ -3476,28 +3470,16 @@ def _ensure_consistent_zarr_format(
     )
 
 
-async def _getitem_semaphore(
-    node: AsyncGroup, key: str, semaphore: asyncio.Semaphore | None
-) -> AnyAsyncArray | AsyncGroup:
+async def _getitem(node: AsyncGroup, key: str) -> AnyAsyncArray | AsyncGroup:
     """
-    Wrap Group.getitem with an optional semaphore.
-
-    If the semaphore parameter is an
-    asyncio.Semaphore instance, then the getitem operation is performed inside an async context
-    manager provided by that semaphore. If the semaphore parameter is None, then getitem is invoked
-    without a context manager.
+    Fetch a child node from a group by key.
     """
-    if semaphore is not None:
-        async with semaphore:
-            return await node.getitem(key)
-    else:
-        return await node.getitem(key)
+    return await node.getitem(key)
 
 
 async def _iter_members(
     node: AsyncGroup,
     skip_keys: tuple[str, ...],
-    semaphore: asyncio.Semaphore | None,
 ) -> AsyncGenerator[tuple[str, AnyAsyncArray | AsyncGroup], None]:
     """
     Iterate over the arrays and groups contained in a group.
@@ -3508,8 +3490,6 @@ async def _iter_members(
         The group to traverse.
     skip_keys : tuple[str, ...]
         A tuple of keys to skip when iterating over the possible members of the group.
-    semaphore : asyncio.Semaphore | None
-        An optional semaphore to use for concurrency control.
 
     Yields
     ------
@@ -3520,10 +3500,7 @@ async def _iter_members(
     keys = [key async for key in node.store.list_dir(node.path)]
     keys_filtered = tuple(filter(lambda v: v not in skip_keys, keys))
 
-    node_tasks = tuple(
-        asyncio.create_task(_getitem_semaphore(node, key, semaphore), name=key)
-        for key in keys_filtered
-    )
+    node_tasks = tuple(asyncio.create_task(_getitem(node, key), name=key) for key in keys_filtered)
 
     for fetched_node_coro in asyncio.as_completed(node_tasks):
         try:
@@ -3550,7 +3527,6 @@ async def _iter_members_deep(
     *,
     max_depth: int | None,
     skip_keys: tuple[str, ...],
-    semaphore: asyncio.Semaphore | None = None,
     use_consolidated_for_children: bool = True,
 ) -> AsyncGenerator[tuple[str, AnyAsyncArray | AsyncGroup], None]:
     """
@@ -3565,8 +3541,6 @@ async def _iter_members_deep(
         The maximum depth of recursion.
     skip_keys : tuple[str, ...]
         A tuple of keys to skip when iterating over the possible members of the group.
-    semaphore : asyncio.Semaphore | None
-        An optional semaphore to use for concurrency control.
     use_consolidated_for_children : bool, default True
         Whether to use the consolidated metadata of child groups loaded
         from the store. Note that this only affects groups loaded from the
@@ -3585,7 +3559,7 @@ async def _iter_members_deep(
         new_depth = None
     else:
         new_depth = max_depth - 1
-    async for name, node in _iter_members(group, skip_keys=skip_keys, semaphore=semaphore):
+    async for name, node in _iter_members(group, skip_keys=skip_keys):
         is_group = isinstance(node, AsyncGroup)
         if (
             is_group
@@ -3599,9 +3573,7 @@ async def _iter_members_deep(
         yield name, node
         if is_group and do_recursion:
             node = cast("AsyncGroup", node)
-            to_recurse[name] = _iter_members_deep(
-                node, max_depth=new_depth, skip_keys=skip_keys, semaphore=semaphore
-            )
+            to_recurse[name] = _iter_members_deep(node, max_depth=new_depth, skip_keys=skip_keys)
 
     for prefix, subgroup_iter in to_recurse.items():
         async for name, node in subgroup_iter:
@@ -3811,9 +3783,7 @@ async def get_node(store: Store, path: str, zarr_format: ZarrFormat) -> AnyAsync
             raise ValueError(f"Unexpected zarr format: {zarr_format}")  # pragma: no cover
 
 
-async def _set_return_key(
-    *, store: Store, key: str, value: Buffer, semaphore: asyncio.Semaphore | None = None
-) -> str:
+async def _set_return_key(*, store: Store, key: str, value: Buffer) -> str:
     """
     Write a value to storage at the given key. The key is returned.
     Useful when saving values via routines that return results in execution order,
@@ -3828,31 +3798,23 @@ async def _set_return_key(
         The key to save the value to.
     value : Buffer
         The value to save.
-    semaphore : asyncio.Semaphore | None
-        An optional semaphore to use to limit the number of concurrent writes.
     """
-
-    if semaphore is not None:
-        async with semaphore:
-            await store.set(key, value)
-    else:
-        await store.set(key, value)
+    await store.set(key, value)
     return key
 
 
 def _persist_metadata(
     store: Store,
     path: str,
     metadata: ArrayV2Metadata | ArrayV3Metadata | GroupMetadata,
-    semaphore: asyncio.Semaphore | None = None,
 ) -> tuple[Coroutine[None, None, str], ...]:
     """
     Prepare to save a metadata document to storage, returning a tuple of coroutines that must be awaited.
     """
 
     to_save = metadata.to_buffer_dict(default_buffer_prototype())
     return tuple(
-        _set_return_key(store=store, key=_join_paths([path, key]), value=value, semaphore=semaphore)
+        _set_return_key(store=store, key=_join_paths([path, key]), value=value)
         for key, value in to_save.items()
     )