From 00804d58b433b65a31755a9d1649a99d1fbb8b91 Mon Sep 17 00:00:00 2001 From: JK Date: Tue, 22 Sep 2026 05:07:49 +0900 Subject: [PATCH 1/5] Fix thread-unsafe dctx sharing in NDArray.get_1d_span_numpy get_1d_span_numpy borrowed the SChunk's shared dctx when non-NULL, but a Blosc2 decompression context is mutable. Concurrent calls from multiple threads (as done by the indexing query planner) can corrupt each other's decoded output. Always create a private per-call context instead, still associated with the SChunk via dparams.schunk so codecs/filters that resolve per-schunk state during decompression keep working. --- src/blosc2/blosc2_ext.pyx | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 4b257c686..d4c3eedb4 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -3874,11 +3874,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 +3901,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 = BLOSC2_DPARAMS_DEFAULTS + 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 +3926,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) From 1efb917c41c6eb1c8624e148621bf125c9973368 Mon Sep 17 00:00:00 2001 From: JK Date: Tue, 22 Sep 2026 05:07:54 +0900 Subject: [PATCH 2/5] Fix segfault in NDArray.get_slice_numpy under concurrent reads b2nd_get_slice_cbuffer has no per-call context parameter, so concurrent reads against the same NDArray corrupt shared native state and segfault. Add a per-instance lock around this call; independent NDArray objects are unaffected and still read in parallel. --- src/blosc2/blosc2_ext.pyx | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index d4c3eedb4..f2fe041f3 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -3779,10 +3779,14 @@ cdef class slice_flatter: cdef class NDArray: cdef b2nd_array_t* array + cdef PyThread_type_lock read_lock def __init__(self, array, base=None): self._dtype = None self.array = PyCapsule_GetPointer(array, "b2nd_array_t*") + self.read_lock = PyThread_allocate_lock() + if self.read_lock == NULL: + raise MemoryError("Could not allocate NDArray read lock") self.base = base # add reference to base if NDArray is a view @property @@ -3851,11 +3855,16 @@ 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_, - view.buf, buffershape_, view.len), - "Error while getting the buffer") + PyThread_acquire_lock(self.read_lock, 1) + try: + rc = b2nd_get_slice_cbuffer(self.array, start_, stop_, + 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 @@ -4548,6 +4557,8 @@ cdef class NDArray: def __dealloc__(self): if self.array != NULL: _check_rc(b2nd_free(self.array), "Error while freeing the array") + if self.read_lock != NULL: + PyThread_free_lock(self.read_lock) cdef b2nd_context_t* create_b2nd_context(shape, chunks, blocks, dtype, kwargs): From 4a88750d3be788a9c4984758e46be5592ddb0fae Mon Sep 17 00:00:00 2001 From: JK Date: Thu, 24 Sep 2026 20:33:44 +0900 Subject: [PATCH 3/5] Fix NDArray read synchronization --- src/blosc2/blosc2_ext.pyx | 17 ++++++++++++----- tests/ctable/test_ctable_indexing.py | 16 ++++++++++++++++ tests/ndarray/test_resize.py | 20 ++++++++++++++++++++ 3 files changed, 48 insertions(+), 5 deletions(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index f2fe041f3..d16719324 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -3780,14 +3780,21 @@ 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 = PyCapsule_GetPointer(array, "b2nd_array_t*") - self.read_lock = PyThread_allocate_lock() - if self.read_lock == NULL: - raise MemoryError("Could not allocate NDArray read lock") 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 = (base).read_lock @property def c_array(self): @@ -3916,7 +3923,7 @@ cdef class NDArray: # 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 = BLOSC2_DPARAMS_DEFAULTS + 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) @@ -4557,7 +4564,7 @@ cdef class NDArray: def __dealloc__(self): if self.array != NULL: _check_rc(b2nd_free(self.array), "Error while freeing the array") - if self.read_lock != NULL: + if self.owns_read_lock and self.read_lock != NULL: PyThread_free_lock(self.read_lock) diff --git a/tests/ctable/test_ctable_indexing.py b/tests/ctable/test_ctable_indexing.py index dd3e00604..dae3b9c77 100644 --- a/tests/ctable/test_ctable_indexing.py +++ b/tests/ctable/test_ctable_indexing.py @@ -1224,6 +1224,22 @@ 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) + + 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 diff --git a/tests/ndarray/test_resize.py b/tests/ndarray/test_resize.py index f2af3d646..ce5bf2ed4 100644 --- a/tests/ndarray/test_resize.py +++ b/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,21 @@ 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_ + + +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))) From 5c2ae8ebcc2c594f557382a8d87200791f20a404 Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Thu, 24 Sep 2026 13:50:11 +0200 Subject: [PATCH 4/5] Skip concurrent read test on WASM --- tests/ndarray/test_resize.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/ndarray/test_resize.py b/tests/ndarray/test_resize.py index ce5bf2ed4..5ca7da7e7 100644 --- a/tests/ndarray/test_resize.py +++ b/tests/ndarray/test_resize.py @@ -99,6 +99,7 @@ def test_expand_dims(shape, axis, chunks, blocks, fill_value): 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) From 0f58225b3aa50c6cc9f55b4ef89ee240f67059df Mon Sep 17 00:00:00 2001 From: JK Date: Thu, 24 Sep 2026 21:28:14 +0900 Subject: [PATCH 5/5] Avoid GIL deadlock during NDArray reads --- src/blosc2/blosc2_ext.pyx | 5 ++++- tests/ctable/test_ctable_indexing.py | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index d16719324..079b37afa 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -3864,7 +3864,10 @@ cdef class NDArray: cdef Py_buffer view cdef int rc PyObject_GetBuffer(arr, &view, PyBUF_SIMPLE) - PyThread_acquire_lock(self.read_lock, 1) + # 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_, view.buf, buffershape_, view.len) diff --git a/tests/ctable/test_ctable_indexing.py b/tests/ctable/test_ctable_indexing.py index dae3b9c77..2342ed273 100644 --- a/tests/ctable/test_ctable_indexing.py +++ b/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 @@ -1240,6 +1241,24 @@ def add_offset(input, output, offset): 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