Fix two NDArray thread-safety bugs blocking free-threaded support (#555)

src/blosc2/blosc2_ext.pyx

@@ -3779,11 +3779,22 @@ cdef class slice_flatter:
 
 cdef class NDArray:
     cdef b2nd_array_t* array
+    cdef PyThread_type_lock read_lock
+    cdef c_bool owns_read_lock
 
     def __init__(self, array, base=None):
         self._dtype = None
         self.array = <b2nd_array_t *> PyCapsule_GetPointer(array, <char *> "b2nd_array_t*")
         self.base = base # add reference to base if NDArray is a view
+        if base is None:
+            self.read_lock = PyThread_allocate_lock()
+            if self.read_lock == NULL:
+                raise MemoryError("Could not allocate NDArray read lock")
+            self.owns_read_lock = True
+        else:
+            # expand_dims/squeeze views share the base SChunk, so reads through
+            # all aliases must be protected by the same lock.
+            self.read_lock = (<NDArray>base).read_lock
 
     @property
     def c_array(self):
@@ -3851,11 +3862,19 @@ cdef class NDArray:
             buffershape_[i] = stop_[i] - start_[i]
 
         cdef Py_buffer view
+        cdef int rc
         PyObject_GetBuffer(arr, &view, PyBUF_SIMPLE)
-        _check_rc(b2nd_get_slice_cbuffer(self.array, start_, stop_,
-                                         <void *> view.buf, buffershape_, view.len),
-                  "Error while getting the buffer")
+        # Waiting for a reader already using this SChunk must not retain the
+        # GIL: that reader can need the GIL again from a Python postfilter.
+        with nogil:
+            PyThread_acquire_lock(self.read_lock, 1)
+        try:
+            rc = b2nd_get_slice_cbuffer(self.array, start_, stop_,
+                                        <void *> view.buf, buffershape_, view.len)
+        finally:
+            PyThread_release_lock(self.read_lock)
         PyBuffer_Release(&view)
+        _check_rc(rc, "Error while getting the buffer")
 
         return arr
 
@@ -3874,11 +3893,10 @@ cdef class NDArray:
         cdef int32_t chunk_nbytes
         cdef int32_t chunk_cbytes
         cdef int32_t block_nbytes
-        cdef blosc2_context *dctx = self.array.sc.dctx
+        cdef blosc2_context *dctx
         cdef Py_buffer view
         cdef int rc
         cdef int32_t lazychunk_cbytes
-        cdef c_bool owns_dctx = False
         cdef int32_t want_nbytes
 
         lazychunk_cbytes = blosc2_schunk_get_lazychunk(self.array.sc, nchunk, &chunk, &needs_free)
@@ -3902,9 +3920,16 @@ cdef class NDArray:
                 free(chunk)
             raise ValueError("destination buffer is smaller than the requested decoded span")
 
-        if dctx == NULL:
-            dctx = blosc2_create_dctx(BLOSC2_DPARAMS_DEFAULTS)
-            owns_dctx = True
+        # A Blosc2 decompression context is mutable.  This method is used by
+        # the indexing planner from several Python workers, so it must not
+        # borrow the SChunk's shared context.  It still needs to be
+        # associated with the SChunk (not just BLOSC2_DPARAMS_DEFAULTS),
+        # since some codecs/filters resolve per-schunk state (e.g.
+        # dictionaries) through dparams.schunk during decompression.
+        cdef blosc2_dparams dparams = dereference(self.array.sc.storage.dparams)
+        dparams.schunk = self.array.sc
+        dparams.typesize = self.array.sc.typesize
+        dctx = blosc2_create_dctx(dparams)
         if dctx == NULL:
             PyBuffer_Release(&view)
             if needs_free:
@@ -3920,8 +3945,7 @@ cdef class NDArray:
         rc = blosc2_getitem_bytes_ctx(dctx, chunk, lazychunk_cbytes,
                                       start * self.array.sc.typesize, want_nbytes,
                                       view.buf, view.len)
-        if owns_dctx:
-            blosc2_free_ctx(dctx)
+        blosc2_free_ctx(dctx)
         PyBuffer_Release(&view)
         if needs_free:
             free(chunk)
@@ -4543,6 +4567,8 @@ cdef class NDArray:
     def __dealloc__(self):
         if self.array != NULL:
             _check_rc(b2nd_free(self.array), "Error while freeing the array")
+        if self.owns_read_lock and self.read_lock != NULL:
+            PyThread_free_lock(self.read_lock)
 
 
 cdef b2nd_context_t* create_b2nd_context(shape, chunks, blocks, dtype, kwargs):

tests/ctable/test_ctable_indexing.py

@@ -11,6 +11,7 @@
 import shutil
 import tempfile
 import weakref
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 import numpy as np
@@ -1224,6 +1225,40 @@ def test_wide_sidecar_span_read_is_not_short():
     assert out.tolist() == values[128 + 5 : 128 + 105].tolist()
 
 
+def test_span_read_preserves_schunk_postfilter():
+    """Private span contexts must retain the SChunk's dparams."""
+    values = np.arange(16, dtype=np.int32)
+    arr = blosc2.asarray(values, chunks=(16,))
+    arr.schunk.dparams = blosc2.DParams(nthreads=1)
+
+    @arr.schunk.postfilter(np.int32, np.int32)
+    def add_offset(input, output, offset):
+        output[:] = input + 100
+
+    out = np.empty(6, dtype=np.int32)
+    arr.get_1d_span_numpy(out, 0, 4, len(out))
+
+    np.testing.assert_array_equal(out, values[4:10] + 100)
+
+
+@pytest.mark.skipif(blosc2.IS_WASM, reason="Pyodide cannot start threads")
+def test_concurrent_span_reads_have_independent_contexts():
+    """Concurrent span reads must not share a mutable decompression context."""
+    values = np.arange(1024, dtype=np.int64)
+    arr = blosc2.asarray(values, chunks=(128,))
+    spans = [(chunk, start, 31) for chunk in range(8) for start in (0, 17, 64)]
+
+    def read_span(span):
+        chunk, start, nitems = span
+        out = np.empty(nitems, dtype=values.dtype)
+        arr.get_1d_span_numpy(out, chunk, start, nitems)
+        offset = chunk * arr.chunks[0] + start
+        np.testing.assert_array_equal(out, values[offset : offset + nitems])
+
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        list(pool.map(read_span, spans))
+
+
 def test_coalesce_spans_merges_within_a_block():
     """Spans closer than one block must merge: reading them apart re-reads the block."""
     coalesce = blosc2.indexing._coalesce_spans

tests/ndarray/test_resize.py

@@ -5,6 +5,8 @@
 # SPDX-License-Identifier: BSD-3-Clause
 #######################################################################
 
+from concurrent.futures import ThreadPoolExecutor
+
 import numpy as np
 import pytest
 
@@ -95,3 +97,22 @@ def test_expand_dims(shape, axis, chunks, blocks, fill_value):
     del bloscview
     del bloscarr_
     assert bloscview2[()].shape == bloscview2.shape  # shouldn't fail because still have access to bloscarr_
+
+
+@pytest.mark.skipif(blosc2.IS_WASM, reason="Pyodide cannot start threads")
+def test_expand_dims_aliases_support_concurrent_reads():
+    """Views and their base must synchronize reads through their shared SChunk."""
+    expected = np.arange(64 * 32, dtype=np.int64).reshape(64, 32)
+    base = blosc2.asarray(expected, chunks=(16, 16))
+    view = blosc2.expand_dims(base, axis=0)
+
+    def read_base():
+        for _ in range(20):
+            np.testing.assert_array_equal(base[:], expected)
+
+    def read_view():
+        for _ in range(20):
+            np.testing.assert_array_equal(view[:], expected[np.newaxis, :])
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        list(pool.map(lambda read: read(), (read_base, read_view)))