Add is_concurrency_safe function

- Introduced is_concurrency_safe function to determine if the file system
supports concurrent access.
- Modified get_array_metadata to read metadata files sequentially for
non-concurrency-safe file systems.
- Enhanced compatibility with synchronous file systems by avoiding deadlocks
when accessing metadata concurrently.
- Updated logic to conditionally use asyncio.gather based on the concurrency
safety of the underlying file system.

Author: dgegen

src/zarr/core/array.py

@@ -196,26 +196,42 @@ def create_codec_pipeline(metadata: ArrayMetadata) -> CodecPipeline:
     raise TypeError  # pragma: no cover
 
 
+def is_concurrency_safe(store: Store) -> bool:
+    fs = getattr(store, "fs", None)
+    return getattr(fs, "asynchronous", True)
+
+
 async def get_array_metadata(
     store_path: StorePath, zarr_format: ZarrFormat | None = 3
 ) -> dict[str, JSON]:
+    concurrency_safe = is_concurrency_safe(store_path.store)
+
     if zarr_format == 2:
-        zarray_bytes, zattrs_bytes = await gather(
-            (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype),
-            (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype),
-        )
+        if concurrency_safe:
+            zarray_bytes, zattrs_bytes = await gather(
+                (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype),
+                (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype),
+            )
+        else:
+            zarray_bytes = await (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype)
+            zattrs_bytes = await (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype)
         if zarray_bytes is None:
             raise FileNotFoundError(store_path)
     elif zarr_format == 3:
         zarr_json_bytes = await (store_path / ZARR_JSON).get(prototype=cpu_buffer_prototype)
         if zarr_json_bytes is None:
             raise FileNotFoundError(store_path)
     elif zarr_format is None:
-        zarr_json_bytes, zarray_bytes, zattrs_bytes = await gather(
-            (store_path / ZARR_JSON).get(prototype=cpu_buffer_prototype),
-            (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype),
-            (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype),
-        )
+        if concurrency_safe:
+            zarr_json_bytes, zarray_bytes, zattrs_bytes = await gather(
+                (store_path / ZARR_JSON).get(prototype=cpu_buffer_prototype),
+                (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype),
+                (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype),
+            )
+        else:
+            zarr_json_bytes = await (store_path / ZARR_JSON).get(prototype=cpu_buffer_prototype)
+            zarray_bytes = await (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype)
+            zattrs_bytes = await (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype)
         if zarr_json_bytes is not None and zarray_bytes is not None:
             # warn and favor v3
             msg = f"Both zarr.json (Zarr format 3) and .zarray (Zarr format 2) metadata objects exist at {store_path}. Zarr v3 will be used."
@@ -1426,7 +1442,11 @@ async def _save_metadata(self, metadata: ArrayMetadata, ensure_parents: bool = F
                     ]
                 )
 
-        await gather(*awaitables)
+        if is_concurrency_safe(self.store_path.store):
+            await gather(*awaitables)
+        else:
+            for awaitable in awaitables:
+                await awaitable
 
     async def _set_selection(
         self,

src/zarr/core/group.py

@@ -30,6 +30,7 @@
     _build_parents,
     _parse_deprecated_compressor,
     create_array,
+    is_concurrency_safe,
 )
 from zarr.core.attributes import Attributes
 from zarr.core.buffer import default_buffer_prototype
@@ -533,17 +534,23 @@ async def open(
             if zarr_json_bytes is None:
                 raise FileNotFoundError(store_path)
         elif zarr_format is None:
+            paths = [
+                (store_path / ZARR_JSON).get(),
+                (store_path / ZGROUP_JSON).get(),
+                (store_path / ZATTRS_JSON).get(),
+                (store_path / consolidated_key).get(),
+            ]
             (
                 zarr_json_bytes,
                 zgroup_bytes,
                 zattrs_bytes,
                 maybe_consolidated_metadata_bytes,
-            ) = await asyncio.gather(
-                (store_path / ZARR_JSON).get(),
-                (store_path / ZGROUP_JSON).get(),
-                (store_path / ZATTRS_JSON).get(),
-                (store_path / str(consolidated_key)).get(),
+            ) = (
+                await asyncio.gather(*paths)
+                if is_concurrency_safe(store_path.store)
+                else [await path for path in paths]
             )
+
             if zarr_json_bytes is not None and zgroup_bytes is not None:
                 # warn and favor v3
                 msg = f"Both zarr.json (Zarr format 3) and .zgroup (Zarr format 2) metadata objects exist at {store_path}. Zarr format 3 will be used."
@@ -3453,11 +3460,23 @@ async def _read_metadata_v2(store: Store, path: str) -> ArrayV2Metadata | GroupM
     """
     # TODO: consider first fetching array metadata, and only fetching group metadata when we don't
     # find an array
-    zarray_bytes, zgroup_bytes, zattrs_bytes = await asyncio.gather(
-        store.get(_join_paths([path, ZARRAY_JSON]), prototype=default_buffer_prototype()),
-        store.get(_join_paths([path, ZGROUP_JSON]), prototype=default_buffer_prototype()),
-        store.get(_join_paths([path, ZATTRS_JSON]), prototype=default_buffer_prototype()),
-    )
+    print(f"Reading metadata from {path} in store {store}", file=sys.stderr)
+    if is_concurrency_safe(store):
+        zarray_bytes, zgroup_bytes, zattrs_bytes = await asyncio.gather(
+            store.get(_join_paths([path, ZARRAY_JSON]), prototype=default_buffer_prototype()),
+            store.get(_join_paths([path, ZGROUP_JSON]), prototype=default_buffer_prototype()),
+            store.get(_join_paths([path, ZATTRS_JSON]), prototype=default_buffer_prototype()),
+        )
+    else:
+        zarray_bytes = await store.get(
+            _join_paths([path, ZARRAY_JSON]), prototype=default_buffer_prototype()
+        )
+        zgroup_bytes = await store.get(
+            _join_paths([path, ZGROUP_JSON]), prototype=default_buffer_prototype()
+        )
+        zattrs_bytes = await store.get(
+            _join_paths([path, ZATTRS_JSON]), prototype=default_buffer_prototype()
+        )
 
     if zattrs_bytes is None:
         zattrs = {}