fix: use threads scheduler in da.store (#2183)

Author: ilan-gold

docs/release-notes/2183.fix.md

@@ -0,0 +1 @@
+Unblock version restriction on `dask` distributed writing by using threading scheduler always (see {pr}`2172`) {user}`ilan-gold`

src/anndata/_io/specs/methods.py

@@ -506,26 +506,16 @@ def write_basic_dask_dask_dense(
     dataset_kwargs: Mapping[str, Any] = MappingProxyType({}),
 ):
     import dask.array as da
-    import dask.config as dc
-
-    is_distributed = dc.get("scheduler", None) == "dask.distributed"
-    is_h5 = isinstance(f, H5Group)
-    if is_distributed and is_h5:
-        msg = "Cannot write dask arrays to hdf5 when using distributed scheduler"
-        raise ValueError(msg)
 
     dataset_kwargs = dataset_kwargs.copy()
+    is_h5 = isinstance(f, H5Group)
     if not is_h5:
         dataset_kwargs = zarr_v3_compressor_compat(dataset_kwargs)
-        # See https://github.com/dask/dask/issues/12109
-        if Version(version("dask")) < Version("2025.4.0") and is_distributed:
-            msg = "Writing dense data with a distributed scheduler to zarr could produce corrupted data with a Lock and will error without one when dask is older than 2025.4.0: https://github.com/dask/dask/issues/12109"
-            raise RuntimeError(msg)
     if is_zarr_v2() or is_h5:
         g = f.require_dataset(k, shape=elem.shape, dtype=elem.dtype, **dataset_kwargs)
     else:
         g = f.require_array(k, shape=elem.shape, dtype=elem.dtype, **dataset_kwargs)
-    da.store(elem, g)
+    da.store(elem, g, scheduler="threads")
 
 
 @_REGISTRY.register_read(H5Array, IOSpec("array", "0.2.0"))

tests/test_dask.py

@@ -4,14 +4,12 @@
 
 from __future__ import annotations
 
-from importlib.metadata import version
 from pathlib import Path
 from typing import TYPE_CHECKING
 
 import numpy as np
 import pandas as pd
 import pytest
-from packaging.version import Version
 
 import anndata as ad
 from anndata._core.anndata import AnnData
@@ -135,17 +133,6 @@ def test_dask_distributed_write(
         adata.obsm["b"] = da.random.random((M, 10))
         adata.varm["a"] = da.random.random((N, 10))
         orig = adata
-        is_h5 = diskfmt == "h5ad"
-        is_corrupted_dask = Version(version("dask")) < Version("2025.4.0")
-        if is_corrupted_dask or is_h5:
-            with pytest.raises(
-                ValueError if is_h5 else RuntimeError,
-                match=r"Cannot write dask arrays to hdf5"
-                if is_h5
-                else r"Writing dense data with a distributed scheduler to zarr",
-            ):
-                ad.io.write_elem(g, "", orig)
-            return
         ad.io.write_elem(g, "", orig)
         # TODO: See https://github.com/zarr-developers/zarr-python/issues/2716
         g = as_group(pth, mode="r")