feat: expose blosc.use_threads via zarr.config

changes/4531.feature.md

@@ -0,0 +1 @@
+Expose `blosc.use_threads` via `zarr.config`, allowing users to enable Blosc's internal threading for faster decoding of large single chunks.

src/zarr/core/config.py

@@ -28,6 +28,7 @@
 """
 
 from __future__ import annotations
+import importlib
 
 from typing import TYPE_CHECKING, Any, Literal, cast
 
@@ -57,6 +58,19 @@ class Config(DConfig):  # type: ignore[misc]
 
     """
 
+
+    def set(self, arg=None, **kwargs):
+        result = super().set(arg, **kwargs)
+        numcodecs = importlib.import_module("numcodecs")
+        numcodecs.blosc.use_threads = self.get("blosc.use_threads", False)
+        return result
+
+    def refresh(self, *args, **kwargs):
+        result = super().refresh(*args, **kwargs)
+        numcodecs = importlib.import_module("numcodecs")
+        numcodecs.blosc.use_threads = self.get("blosc.use_threads", False)
+        return result
+
     def reset(self) -> None:
         self.clear()
         self.refresh()
@@ -107,6 +121,9 @@ def enable_gpu(self) -> ConfigSet:
             "async": {"concurrency": 10, "timeout": None},
             "threading": {"max_workers": None},
             "json_indent": 2,
+            "blosc": {
+                "use_threads": False,
+            },
             "codec_pipeline": {
                 # FusedCodecPipeline is the faster synchronous pipeline, but it stays
                 # opt-in for now so behavior is unchanged for existing users. Early

tests/test_config.py

@@ -483,3 +483,11 @@ def test_deprecated_config(key: str) -> None:
     with pytest.raises(ValueError):
         with zarr.config.set({key: "foo"}):
             pass
+
+
+def test_blosc_use_threads_config() -> None:
+    import numcodecs
+
+    assert numcodecs.blosc.use_threads is False
+    with config.set({"blosc.use_threads": True}):
+        assert numcodecs.blosc.use_threads is True