From c9c481e27ac8bb878acd8f719a0731a4063dbf69 Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Thu, 5 Nov 2026 16:47:18 +0100 Subject: [PATCH 1/2] Keep the GIL for compressions that can run Python callbacks With blosc2.set_releasegil(True), every write path compressed without the GIL, also when a prefilter, a filler, or a codec or filter registered from Python runs during the compression. The callback then calls Python without holding the GIL and the interpreter crashes. This affects SChunk(data=), append_data(), insert_data(), update_data(), growing set_slice(), compress2(), decompress2() and vlcompress(), and also to_cframe(), which can decompress and recompress the chunks with the super-chunk's filters. Compressions now release the GIL only if none of the prefilter, codec and filters of their context is a Python one (_can_compress_without_gil()). The ids registered from Python are kept in C arrays for that, so the check needs no Python lookups, and writes next to an unrelated Python codec still release the GIL. decompress2() cannot know the codec of a chunk beforehand and keeps the GIL while any codec or filter is registered from Python, as slice reads already do. _compress_ctx() and _decompress_ctx() make the decision for all call sites; _check_cparams() and _check_dparams() use the same predicates. Co-Authored-By: Claude Opus 5.5 --- src/blosc2/blosc2_ext.pyx | 152 ++++++++++---------- src/blosc2/core.py | 5 + tests/test_releasegil_callbacks.py | 217 +++++++++++++++++++++++++++++ 3 files changed, 303 insertions(+), 71 deletions(-) create mode 100644 tests/test_releasegil_callbacks.py diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 2910c3455..88bd46922 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -1330,7 +1330,7 @@ def get_compressor(): return blosc1_get_compressor() -cdef c_bool RELEASEGIL = False +cdef c_bool RELEASEGIL = False # never set on wasm32 (see blosc2.set_releasegil) def set_releasegil(c_bool gilstate): global RELEASEGIL @@ -1338,39 +1338,81 @@ def set_releasegil(c_bool gilstate): RELEASEGIL = gilstate return oldstate +# Ids of the codecs and filters registered from Python (register_codec(), register_filter()), whose +# callbacks need the GIL. C-Blosc2 cannot unregister them, so these are never cleared. They let the GIL +# checks of compressions run without Python lookups; slice reads (_can_release_gil()) still look at +# blosc2.ucodecs_registry and blosc2.ufilters_registry, which tests replace to force reads without the GIL. +cdef c_bool _python_codec_ids[256] +cdef c_bool _python_filter_ids[256] +cdef c_bool _python_ids_registered = False + +cdef inline c_bool _prefilter_runs_python(blosc2_prefilter_fn prefilter) noexcept: + """Whether `prefilter` is set and calls Python, i.e. is not the C-only miniexpr or matmul one.""" + return (prefilter != NULL and prefilter != miniexpr_prefilter + and prefilter != matmul_prefilter) + +cdef inline c_bool _uses_python_filter(const blosc2_cparams *cparams) noexcept: + cdef int i + for i in range(BLOSC2_MAX_FILTERS): + if _python_filter_ids[cparams.filters[i]]: + return True + return False + +cdef c_bool _can_compress_without_gil(blosc2_context *cctx) noexcept: + """Whether compressing with `cctx` may release the GIL as set_releasegil() asks: only if no Python + callback can run, i.e. its prefilter, codec and filters are not Python ones. Unlike slice reads, calls + without the GIL here and in _decompress_ctx() are not waited for in register_codec() and + register_filter(): what they register can only meet a call that started with an id nothing was + registered under, which is wrong anyway.""" + cdef blosc2_cparams cparams + if not RELEASEGIL or cctx == NULL or blosc2_ctx_get_cparams(cctx, &cparams) < 0: + return False + return not (_prefilter_runs_python(cparams.prefilter) or _python_codec_ids[cparams.compcode] + or _uses_python_filter(&cparams)) + +cdef int _compress_ctx(blosc2_context *cctx, const void *src, int32_t srcsize, void *dest, + int32_t destsize) noexcept: + """blosc2_compress_ctx(), without the GIL if _can_compress_without_gil(cctx).""" + cdef int rc + if _can_compress_without_gil(cctx): + with nogil: + rc = blosc2_compress_ctx(cctx, src, srcsize, dest, destsize) + else: + rc = blosc2_compress_ctx(cctx, src, srcsize, dest, destsize) + return rc + +cdef int _decompress_ctx(blosc2_context *dctx, const void *src, int32_t srcsize, void *dest, + int32_t destsize) noexcept: + """blosc2_decompress_ctx() with a context without postfilter, without the GIL if set_releasegil() asks + and no codec or filter is registered from Python (the chunk tells which ones it uses).""" + cdef int rc + if RELEASEGIL and not _python_ids_registered: + with nogil: + rc = blosc2_decompress_ctx(dctx, src, srcsize, dest, destsize) + else: + rc = blosc2_decompress_ctx(dctx, src, srcsize, dest, destsize) + return rc + def get_blocksize(): return blosc1_get_blocksize() cdef _check_cparams(blosc2_cparams *cparams): if cparams.nthreads > 1: - if BLOSC2_USER_REGISTERED_CODECS_START <= cparams.compcode <= BLOSC2_USER_REGISTERED_CODECS_STOP\ - and cparams.compcode in blosc2.ucodecs_registry.keys(): + if _python_codec_ids[cparams.compcode]: raise ValueError("Cannot use multi-threading with user defined Python codecs") - - ufilters = [BLOSC2_USER_REGISTERED_FILTERS_START <= filter <= BLOSC2_USER_REGISTERED_FILTERS_STOP - for filter in cparams.filters] - for i in range(len(ufilters)): - if ufilters[i] and cparams.filters[i] in blosc2.ufilters_registry.keys(): - raise ValueError("Cannot use multi-threading with user defined Python filters") - - if cparams.prefilter != NULL and (cparams.prefilter != miniexpr_prefilter and cparams.prefilter != matmul_prefilter): - # Note: miniexpr_prefilter uses miniexpr C API which is thread-friendly, + if _uses_python_filter(cparams): + raise ValueError("Cannot use multi-threading with user defined Python filters") + if _prefilter_runs_python(cparams.prefilter): raise ValueError("`nthreads` must be 1 when a prefilter is set") cdef _check_dparams(blosc2_dparams* dparams, blosc2_cparams* cparams=NULL): if cparams == NULL: return if dparams.nthreads > 1: - if BLOSC2_USER_REGISTERED_CODECS_START <= cparams.compcode <= BLOSC2_USER_REGISTERED_CODECS_STOP\ - and cparams.compcode in blosc2.ucodecs_registry.keys(): + if _python_codec_ids[cparams.compcode]: raise ValueError("Cannot use multi-threading with user defined Python codecs") - - ufilters = [BLOSC2_USER_REGISTERED_FILTERS_START <= filter <= BLOSC2_USER_REGISTERED_FILTERS_STOP - for filter in cparams.filters] - for i in range(len(ufilters)): - if ufilters[i] and cparams.filters[i] in blosc2.ufilters_registry.keys(): - raise ValueError("Cannot use multi-threading with user defined Python filters") - + if _uses_python_filter(cparams): + raise ValueError("Cannot use multi-threading with user defined Python filters") if dparams.postfilter != NULL: raise ValueError("`nthreads` must be 1 when a postfilter is set") @@ -1455,11 +1497,7 @@ def compress2(src, **kwargs): cctx = blosc2_create_cctx(cparams) if cctx == NULL: raise RuntimeError("Could not create the compression context") - if RELEASEGIL: - with nogil: - size = blosc2_compress_ctx(cctx, buf.buf, buf.len, _dest, len_dest) - else: - size = blosc2_compress_ctx(cctx, buf.buf, buf.len, _dest, len_dest) + size = _compress_ctx(cctx, buf.buf, buf.len, _dest, len_dest) blosc2_free_ctx(cctx) PyBuffer_Release(&buf) if size < 0: @@ -1509,11 +1547,7 @@ def decompress2(src, dst=None, **kwargs): view = &typed_view_src[0] # For lazy chunks, blosc2_cbuffer_sizes() only reports the header cbytes. # The decode context needs the full source buffer length. - if RELEASEGIL: - with nogil: - size = blosc2_decompress_ctx(dctx, view, srcsize, buf.buf, nbytes) - else: - size = blosc2_decompress_ctx(dctx, view, srcsize, buf.buf, nbytes) + size = _decompress_ctx(dctx, view, srcsize, buf.buf, nbytes) blosc2_free_ctx(dctx) PyBuffer_Release(&buf) else: @@ -1525,11 +1559,7 @@ def decompress2(src, dst=None, **kwargs): view = &typed_view_src[0] # For lazy chunks, blosc2_cbuffer_sizes() only reports the header cbytes. # The decode context needs the full source buffer length. - if RELEASEGIL: - with nogil: - size = blosc2_decompress_ctx(dctx, view, srcsize, dst_buf, nbytes) - else: - size = blosc2_decompress_ctx(dctx, view, srcsize, dst_buf, nbytes) + size = _decompress_ctx(dctx, view, srcsize, dst_buf, nbytes) blosc2_free_ctx(dctx) if size >= 0: return dst @@ -1584,7 +1614,7 @@ def vlcompress(srcs, **kwargs): cctx = blosc2_create_cctx(cparams) if cctx == NULL: raise RuntimeError("Could not create the compression context") - if RELEASEGIL: + if _can_compress_without_gil(cctx): with nogil: size = blosc2_vlcompress_ctx(cctx, src_ptrs, srcsizes, nblocks, _dest, len_dest) else: @@ -1983,11 +2013,7 @@ cdef class SChunk: csize = (len_chunk + BLOSC2_MAX_OVERHEAD) chunk = malloc(csize) self.schunk.current_nchunk = i - if RELEASEGIL: - with nogil: - comp_size = blosc2_compress_ctx(self.schunk.cctx, buf_ptr + index, len_chunk, chunk, csize) - else: - comp_size = blosc2_compress_ctx(self.schunk.cctx, buf_ptr + index, len_chunk, chunk, csize) + comp_size = _compress_ctx(self.schunk.cctx, buf_ptr + index, len_chunk, chunk, csize) if comp_size < 0: free(chunk) PyBuffer_Release(&buf) @@ -2156,11 +2182,7 @@ cdef class SChunk: cdef int32_t len_chunk = (buf.len + BLOSC2_MAX_OVERHEAD) cdef uint8_t* chunk = malloc(len_chunk) self.schunk.current_nchunk = self.schunk.nchunks - if RELEASEGIL: - with nogil: - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) - else: - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) + size = _compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) PyBuffer_Release(&buf) if size < 0: free(chunk) @@ -2390,12 +2412,7 @@ cdef class SChunk: cdef int32_t len_chunk = (buf.len + BLOSC2_MAX_OVERHEAD) cdef uint8_t* chunk = malloc(len_chunk) self.schunk.current_nchunk = nchunk # prefilter needs this value to be set - if RELEASEGIL: - with nogil: - # No need to create another cctx - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) - else: - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) + size = _compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) PyBuffer_Release(&buf) if size <= 0: free(chunk) @@ -2440,11 +2457,7 @@ cdef class SChunk: cdef int32_t len_chunk = (buf.len + BLOSC2_MAX_OVERHEAD) cdef uint8_t* chunk = malloc(len_chunk) self.schunk.current_nchunk = nchunk # prefilter needs this value to be set - if RELEASEGIL: - with nogil: - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) - else: - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) + size = _compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk, len_chunk) PyBuffer_Release(&buf) if size <= 0: @@ -2551,11 +2564,8 @@ cdef class SChunk: memcpy(data + data_start, buf_ptr + buf_pos, nbytes_copy) chunk = malloc(chunk_nbytes + BLOSC2_MAX_OVERHEAD) self.schunk.current_nchunk = self.schunk.nchunks - 1 - if RELEASEGIL: - with nogil: - comp_rc = blosc2_compress_ctx(self.schunk.cctx, data, chunk_nbytes, chunk, chunk_nbytes + BLOSC2_MAX_OVERHEAD) - else: - comp_rc = blosc2_compress_ctx(self.schunk.cctx, data, chunk_nbytes, chunk, chunk_nbytes + BLOSC2_MAX_OVERHEAD) + comp_rc = _compress_ctx(self.schunk.cctx, data, chunk_nbytes, chunk, + chunk_nbytes + BLOSC2_MAX_OVERHEAD) free(data) if comp_rc < 0: free(chunk) @@ -2584,11 +2594,7 @@ cdef class SChunk: alloc_len = (chunksize + BLOSC2_MAX_OVERHEAD) chunk = malloc(alloc_len) self.schunk.current_nchunk = self.schunk.nchunks - if RELEASEGIL: - with nogil: - comp_rc = blosc2_compress_ctx(self.schunk.cctx, buf_ptr + buf_pos, chunksize, chunk, alloc_len) - else: - comp_rc = blosc2_compress_ctx(self.schunk.cctx, buf_ptr + buf_pos, chunksize, chunk, alloc_len) + comp_rc = _compress_ctx(self.schunk.cctx, buf_ptr + buf_pos, chunksize, chunk, alloc_len) if comp_rc < 0: free(chunk) PyBuffer_Release(&buf) @@ -2613,7 +2619,9 @@ cdef class SChunk: @_handle_locked def to_cframe(self): - return schunk_to_cframe(self.schunk, RELEASEGIL) + # Serializing can decompress (with the postfilter) and recompress chunks and vlmetalayers + return schunk_to_cframe(self.schunk, _can_compress_without_gil(self.schunk.cctx) + and _can_release_gil(self.schunk)) @_handle_locked def _avoid_cframe_free(self, avoid_cframe_free): @@ -3946,7 +3954,7 @@ def register_codec(codec_name, id, encoder=None, decoder=None, version=1): # A slice read that found no Python codec could otherwise meet `decoder` without the GIL: keep # new reads from releasing it and wait for those that did. Nothing below releases the GIL # before the registry is updated. - global _nogil_reads_blocked + global _nogil_reads_blocked, _python_ids_registered _nogil_reads_blocked += 1 try: _wait_for_nogil_reads() @@ -3956,6 +3964,7 @@ def register_codec(codec_name, id, encoder=None, decoder=None, version=1): if encoder and decoder: blosc2.ucodecs_registry[id] = (codec_name, encoder, decoder) + _python_codec_ids[id] = _python_ids_registered = True finally: _nogil_reads_blocked -= 1 @@ -4020,7 +4029,7 @@ def register_filter(id, forward, backward, filter_name): strcpy(filter.name, filter_name_) # As in register_codec() - global _nogil_reads_blocked + global _nogil_reads_blocked, _python_ids_registered _nogil_reads_blocked += 1 try: _wait_for_nogil_reads() @@ -4029,6 +4038,7 @@ def register_filter(id, forward, backward, filter_name): raise RuntimeError("Error while registering filter") if forward and backward: blosc2.ufilters_registry[id] = (forward, backward) + _python_filter_ids[id] = _python_ids_registered = True finally: _nogil_reads_blocked -= 1 diff --git a/src/blosc2/core.py b/src/blosc2/core.py index 04181833c..8209498ba 100644 --- a/src/blosc2/core.py +++ b/src/blosc2/core.py @@ -1519,6 +1519,11 @@ def set_releasegil(gilstate: bool) -> bool: regardless of this setting, unless a Python postfilter, codec or filter could run during the read. + Compressing keeps the GIL regardless of this setting when a Python callback + (a prefilter, a filler, or a codec or filter registered from Python) could + run. Decompressing a chunk keeps it while any codec or filter is registered + from Python, as the chunk may use one. + Examples -------- >>> oldReleaseState = blosc2.set_releasegil(True) diff --git a/tests/test_releasegil_callbacks.py b/tests/test_releasegil_callbacks.py new file mode 100644 index 000000000..7d920f0cc --- /dev/null +++ b/tests/test_releasegil_callbacks.py @@ -0,0 +1,217 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +# Python callbacks (prefilters, fillers, postfilters, codecs and filters registered from Python) need the +# GIL, so the calls that run them must keep it even when blosc2.set_releasegil(True) asks to release it. + +import collections +import threading +import time +import zlib + +import numpy as np +import pytest + +import blosc2 + +# Ids that no other test registers: registering is global and lasts for the whole process +CODEC_ID = 170 +FILTER_ID = 171 + +N = 1000 # items per chunk +DTYPE = np.dtype(np.int32) +CHUNKSIZE = N * DTYPE.itemsize +# Several blocks per chunk, so that every callback runs more than once per chunk +CPARAMS = {"typesize": DTYPE.itemsize, "nthreads": 1, "blocksize": CHUNKSIZE // 4} +DPARAMS = {"nthreads": 1} + +calls = collections.Counter() + + +@pytest.fixture(params=[False, True], ids=["gil-held", "gil-released"]) +def releasegil(request): + old = blosc2.set_releasegil(request.param) + yield + blosc2.set_releasegil(old) + + +def called(name): + calls[name] += 1 + + +def encoder(input, output, meta, schunk): + called("encoder") + out = zlib.compress(input.tobytes()) + if len(out) >= output.size: + return 0 # incompressible: C-Blosc2 stores the block as is + output[: len(out)] = np.frombuffer(out, dtype=np.uint8) + return len(out) + + +def decoder(input, output, meta, schunk): + called("decoder") + out = zlib.decompress(input.tobytes()) + output[: len(out)] = np.frombuffer(out, dtype=np.uint8) + return len(out) + + +def forward(input, output, meta, schunk): + called("forward") + np.bitwise_xor(input, 0x5A, out=output) + + +def backward(input, output, meta, schunk): + called("backward") + np.bitwise_xor(input, 0x5A, out=output) + + +@pytest.fixture(scope="module") +def registered(): + if CODEC_ID not in blosc2.ucodecs_registry: + blosc2.register_codec("test_releasegil", CODEC_ID, encoder, decoder) + if FILTER_ID not in blosc2.ufilters_registry: + blosc2.register_filter(FILTER_ID, forward, backward, "test_releasegil") + + +@pytest.fixture(params=["codec", "filter"]) +def python_cparams(request, registered): + """cparams with the Python codec or filter, and the names of its compressing and decompressing + callbacks.""" + if request.param == "codec": + cparams = {"codec": CODEC_ID, "filters": [blosc2.Filter.NOFILTER], "filters_meta": [0]} + return {**CPARAMS, **cparams}, "encoder", "decoder" + return {**CPARAMS, "filters": [FILTER_ID], "filters_meta": [0]}, "forward", "backward" + + +def data(nitems, start=0): + return (np.arange(start, start + nitems) % 200).astype(DTYPE) + + +def new_schunk(cparams=CPARAMS, **kwargs): + return blosc2.SChunk(chunksize=CHUNKSIZE, cparams=cparams, dparams=DPARAMS, **kwargs) + + +def read(schunk): + out = np.empty(schunk.nbytes // DTYPE.itemsize, dtype=DTYPE) + schunk.get_slice(out=out) + return out + + +WRITE_PATHS = ["append", "insert", "update", "setitem"] + + +def write(schunk, path): + """Write 2 * N items to the empty `schunk` through `path`, return them.""" + values = data(2 * N) + if path == "append": + schunk.append_data(values[:N]) + schunk.append_data(values[N:]) + elif path == "insert": + schunk.append_data(values[N:]) + schunk.insert_data(0, values[:N], copy=False) + elif path == "update": + schunk.append_data(np.zeros(N, dtype=DTYPE)) + schunk.append_data(np.zeros(N, dtype=DTYPE)) + schunk.update_data(0, values[:N], copy=True) + schunk.update_data(1, values[N:], copy=False) + else: + # Growing: appends a partial chunk, then recompresses it in full and appends one more + schunk[: N // 2] = values[: N // 2] + schunk[N // 2 : 2 * N] = values[N // 2 :] + return values + + +@pytest.mark.parametrize("path", WRITE_PATHS) +def test_prefilter_writes(releasegil, path): + schunk = new_schunk() + + @schunk.prefilter(DTYPE, DTYPE) + def pf_test_releasegil(input, output, offset): + output[:] = input | 1 # idempotent: the growing setitem prefilters some items twice + + values = write(schunk, path) + np.testing.assert_array_equal(read(schunk), values | 1) + + +def test_filler(releasegil): + schunk = new_schunk() + values = data(2 * N + N // 2) + + @schunk.filler(((values, DTYPE),), DTYPE) + def fill_test_releasegil(inputs, output, offset): + output[:] = inputs[0] + 1 + + np.testing.assert_array_equal(read(schunk), values + 1) + + +@pytest.mark.parametrize("path", ["init", *WRITE_PATHS]) +def test_python_codec_filter_writes(releasegil, python_cparams, path): + cparams, encode, decode = python_cparams + before = calls.copy() + if path == "init": + values = data(2 * N + N // 2) + schunk = new_schunk(cparams, data=values) + else: + schunk = new_schunk(cparams) + values = write(schunk, path) + np.testing.assert_array_equal(read(schunk), values) + assert calls[encode] > before[encode] + assert calls[decode] > before[decode] + + +def test_to_cframe_postfilter(releasegil): + # Serializing a sparse super-chunk decompresses its chunks with its dparams, postfilter included + values = data(2 * N) + schunk = new_schunk(data=values, contiguous=False) + + @schunk.postfilter(DTYPE, DTYPE) + def post_test_releasegil(input, output, offset): + output[:] = input + + copy = blosc2.schunk_from_cframe(schunk.to_cframe(), copy=True) + np.testing.assert_array_equal(read(copy), values) + + +def longest_gap(func): + """Longest time a ticker thread did not run during func(), as a fraction of that call.""" + stamps, stop = [], threading.Event() + + def ticker(): + while not stop.is_set(): + stamps.append(time.perf_counter()) + time.sleep(0) + + t = threading.Thread(target=ticker) + t.start() + try: + t0 = time.perf_counter() + func() + t1 = time.perf_counter() + finally: + stop.set() + t.join() + inside = [s for s in stamps if t0 <= s <= t1] + return np.diff([t0, *inside, t1]).max() / (t1 - t0) + + +@pytest.mark.skipif(blosc2.IS_WASM, reason="set_releasegil() does nothing on wasm32") +def test_write_releases_the_gil_next_to_python_codecs(registered): + # A codec or filter registered from Python only keeps the GIL for the compressions that use it + values = np.linspace(0, 1, 8 * 1024 * 1024, dtype=np.float32) + schunk = blosc2.SChunk( + chunksize=values.nbytes, cparams={"codec": blosc2.Codec.ZSTD, "clevel": 1, "nthreads": 1} + ) + old = blosc2.set_releasegil(True) + try: + # Holding the GIL leaves one gap as long as nearly the whole write, every time; a busy machine can + # delay the ticker now and then, hence a few attempts (as in test_get_slice_releases_the_gil) + gaps = [] + while len(gaps) < 5 and (not gaps or gaps[-1] >= 0.5): + gaps.append(longest_gap(lambda: schunk.append_data(values))) + finally: + blosc2.set_releasegil(old) + assert gaps[-1] < 0.5, f"longest gaps, as fractions of the write: {gaps}" From ef4446971874310168e0cf65b12ba6a682bbd6f3 Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Thu, 5 Nov 2026 16:48:07 +0100 Subject: [PATCH 2/2] Raise the exceptions of Python callbacks as they are An exception raised in a Python prefilter, filler, postfilter, codec or filter stays set while C-Blosc2 returns an error code. The wrappers then raised their own RuntimeError or ValueError on top of it, which Python turns into "SystemError: returned a result with an exception set". Where the wrappers went on instead (the NDArray slice reads retry failed reads, a growing SChunk.set_slice() did not check its first C call, concat() released its locks first), Python code ran with the exception set. Raise the pending exception right after each such C call fails, so that e.g. a prefilter's KeyError reaches the caller as a KeyError: in _check_rc(), which most NDArray methods use and which now also checks the plain error paths of compress2(), vlcompress(), vldecompress(), vldecompress_block(), get_slice(), set_slice(), get_vlblock(), _prefilter_data() and to_cframe(), and at the error paths with cleanup of the other SChunk data and vlmeta methods, decompress(), decompress2() and the NDArray slice reads. C-Blosc2 stops at the first failing callback, which the tests check. Co-Authored-By: Claude Opus 5.5 --- src/blosc2/blosc2_ext.pyx | 70 +++++++----- tests/test_releasegil_callbacks.py | 165 +++++++++++++++++++++++++++++ 2 files changed, 208 insertions(+), 27 deletions(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 88bd46922..1ba2db629 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -32,6 +32,7 @@ from cpython cimport ( PyObject_GetBuffer, ) from cpython.bool cimport PyBool_Check +from cpython.exc cimport PyErr_Occurred from cpython.long cimport PyLong_Check from cpython.number cimport PyIndex_Check from cpython.pyport cimport PY_SSIZE_T_MAX, PY_SSIZE_T_MIN @@ -1257,11 +1258,10 @@ def decompress(src, dst=None, as_bytearray=False): if dst is None: raise RuntimeError("Could not get a bytes object") size = blosc1_decompress(&typed_view_src[0], dst, len(dst)) - if as_bytearray: - dst = bytearray(dst) if size >= 0: - return dst + return bytearray(dst) if as_bytearray else dst if size < 0: + _raise_callback_error() raise RuntimeError("Cannot decompress") @@ -1393,6 +1393,12 @@ cdef int _decompress_ctx(blosc2_context *dctx, const void *src, int32_t srcsize, rc = blosc2_decompress_ctx(dctx, src, srcsize, dest, destsize) return rc +cdef int _raise_callback_error() except -1: + """Raise the pending exception of a Python callback that failed in the C call that just returned, if + any. Call it right after that call: running Python code with the exception pending ends in a + SystemError.""" + return -1 if PyErr_Occurred() != NULL else 0 + def get_blocksize(): return blosc1_get_blocksize() @@ -1500,9 +1506,8 @@ def compress2(src, **kwargs): size = _compress_ctx(cctx, buf.buf, buf.len, _dest, len_dest) blosc2_free_ctx(cctx) PyBuffer_Release(&buf) - if size < 0: - raise RuntimeError("Could not compress the data") - elif size == 0: + _check_rc(size, "Could not compress the data") + if size == 0: del dest raise RuntimeError("The result could not fit ") return dest[:size] @@ -1564,6 +1569,7 @@ def decompress2(src, dst=None, **kwargs): if size >= 0: return dst if size < 0: + _raise_callback_error() raise ValueError("Error while decompressing, check the src data and/or the dparams") @@ -1628,9 +1634,8 @@ def vlcompress(srcs, **kwargs): free(src_ptrs) free(srcsizes) - if size < 0: - raise RuntimeError("Could not compress the data") - elif size == 0: + _check_rc(size, "Could not compress the data") + if size == 0: del dest raise RuntimeError("The result could not fit ") return dest[:size] @@ -1672,8 +1677,7 @@ def vldecompress(src, **kwargs): # For lazy chunks, blosc2_cbuffer_sizes() only reports the header cbytes. # The decode context needs the full source buffer length. rc = blosc2_vldecompress_ctx(dctx, &typed_view_src[0], srcsize, dests, destsizes, nblocks) - if rc < 0: - raise RuntimeError("Could not decompress the data") + _check_rc(rc, "Could not decompress the data") for i in range(rc): out.append(PyBytes_FromStringAndSize(dests[i], destsizes[i])) free(dests[i]) @@ -1713,8 +1717,7 @@ def vldecompress_block(src, int32_t nblock, **kwargs): &dest, &destsize, ) - if rc < 0: - raise RuntimeError("Could not decompress the block") + _check_rc(rc, "Could not decompress the block") return PyBytes_FromStringAndSize(dest, destsize) finally: if dest != NULL: @@ -2017,6 +2020,7 @@ cdef class SChunk: if comp_size < 0: free(chunk) PyBuffer_Release(&buf) + _raise_callback_error() raise RuntimeError("Could not compress the data") elif comp_size == 0: free(chunk) @@ -2186,6 +2190,7 @@ cdef class SChunk: PyBuffer_Release(&buf) if size < 0: free(chunk) + _raise_callback_error() raise RuntimeError("Could not compress the data") elif size == 0: free(chunk) @@ -2300,6 +2305,7 @@ cdef class SChunk: return dst if size < 0: + _raise_callback_error() raise RuntimeError(f"Error while decompressing the specified chunk, error code: {size}") @_handle_locked @@ -2368,8 +2374,7 @@ cdef class SChunk: cdef uint8_t *block cdef int32_t destsize cbytes = blosc2_schunk_get_vlblock(self.schunk, nchunk, nblock, &block, &destsize) - if cbytes < 0: - raise RuntimeError("Error while getting the vlblock") + _check_rc(cbytes, "Error while getting the vlblock") ret_block = PyBytes_FromStringAndSize(block, destsize) free(block) return ret_block @@ -2417,6 +2422,7 @@ cdef class SChunk: if size <= 0: free(chunk) if size < 0: + _raise_callback_error() raise RuntimeError("Could not compress the data") raise RuntimeError("The result could not fit ") @@ -2463,6 +2469,7 @@ cdef class SChunk: if size <= 0: free(chunk) if size < 0: + _raise_callback_error() raise RuntimeError("Could not compress the data") raise RuntimeError("The result could not fit ") @@ -2487,9 +2494,8 @@ cdef class SChunk: cdef int size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk_buf.buf, chunk_buf.len) PyBuffer_Release(&buf) PyBuffer_Release(&chunk_buf) - if size < 0: - raise RuntimeError("Could not compress the data") - elif size == 0: + _check_rc(size, "Could not compress the data") + if size == 0: raise RuntimeError("The result could not fit ") return size @@ -2515,8 +2521,7 @@ cdef class SChunk: rc = blosc2_schunk_get_slice_buffer(self.schunk, start, stop, out) if rc >= 0: return out - if rc < 0: - raise RuntimeError("Error while getting the slice") + _check_rc(rc, "Error while getting the slice") @_handle_locked def set_slice(self, value, start=0, stop=None): @@ -2546,6 +2551,10 @@ cdef class SChunk: # Increase SChunk's size if start < nitems: rc = blosc2_schunk_set_slice_buffer(self.schunk, start, nitems, buf.buf) + if rc < 0: + PyBuffer_Release(&buf) + _raise_callback_error() + raise RuntimeError("Error while setting the slice") buf_pos = (nitems - start) * self.schunk.typesize if self.schunk.nbytes % self.schunk.chunksize != 0: # Update last chunk before appending any other @@ -2559,6 +2568,7 @@ cdef class SChunk: if rc < 0: free(data) PyBuffer_Release(&buf) + _raise_callback_error() raise RuntimeError("Error while decompressing the chunk") data_start = self.schunk.nbytes - (self.schunk.nchunks - 1) * self.schunk.chunksize memcpy(data + data_start, buf_ptr + buf_pos, nbytes_copy) @@ -2570,6 +2580,7 @@ cdef class SChunk: if comp_rc < 0: free(chunk) PyBuffer_Release(&buf) + _raise_callback_error() raise RuntimeError("Error while compressing the data") elif comp_rc == 0: free(chunk) @@ -2598,6 +2609,7 @@ cdef class SChunk: if comp_rc < 0: free(chunk) PyBuffer_Release(&buf) + _raise_callback_error() raise RuntimeError("Error while compressing the chunk") elif comp_rc == 0: free(chunk) @@ -2614,8 +2626,7 @@ cdef class SChunk: else: rc = blosc2_schunk_set_slice_buffer(self.schunk, start, stop, buf.buf) PyBuffer_Release(&buf) - if rc < 0: - raise RuntimeError("Error while setting the slice") + _check_rc(rc, "Error while setting the slice") @_handle_locked def to_cframe(self): @@ -3605,6 +3616,7 @@ cdef class vlmeta: rc = blosc2_vlmeta_add(self.schunk, name, content, len_content, &ccparams) if rc < 0: + _raise_callback_error() raise RuntimeError @_handle_locked @@ -3618,6 +3630,7 @@ cdef class vlmeta: if rc >= 0: rc = blosc2_vlmeta_get(self.schunk, name, &content, &content_len) if rc < 0: + _raise_callback_error() raise RuntimeError return content[:content_len] @@ -3864,8 +3877,7 @@ cdef schunk_to_cframe(blosc2_schunk* schunk, c_bool release_gil): Py_XDECREF(dest.out) if dest.alloc_failed: raise MemoryError() - if cframe_len < 0: - raise RuntimeError("Error while getting the cframe") + _check_rc(cframe_len, "Error while getting the cframe") return out @@ -4043,6 +4055,8 @@ def register_filter(id, forward, backward, filter_name): _nogil_reads_blocked -= 1 cdef _check_rc(rc, message): + # Before anything else: a Python callback may have failed in the C call that gave rc + _raise_callback_error() if rc < 0: raise RuntimeError(message) @@ -4564,6 +4578,9 @@ cdef class NDArray: rc = b2nd_get_slice_cbuffer_into(self.array, start_, stop_, buf, buffershape_, buffersize) else: rc = b2nd_get_slice_cbuffer(self.array, start_, stop_, buf, buffershape_, buffersize) + if rc < 0: + # The callers retry failed reads, which must not run with a callback's exception set + _raise_callback_error() return rc cdef c_bool _box_layout(self, np.ndarray out, const c_bool *int_dims, int64_t *buffershape_, @@ -4677,8 +4694,7 @@ cdef class NDArray: PyBuffer_Release(&view) if needs_free: free(chunk) - if rc < 0: - raise RuntimeError("Error while decoding the requested span") + _check_rc(rc, "Error while decoding the requested span") if rc != want_nbytes: # A short read used to pass silently and leave the tail of the # destination uninitialised. @@ -5608,7 +5624,7 @@ def concat(arr1: NDArray, arr2: NDArray, axis: int, **kwargs): cdef int rc = 0 with HandleLocks((arr1, arr2)): rc = b2nd_concatenate(ctx, arr1.array, arr2.array, axis, copy, &array) - _check_rc(rc, "Error while concatenating the arrays") + _check_rc(rc, "Error while concatenating the arrays") # before the locks' release() runs _check_rc(b2nd_free_ctx(ctx), "Error while freeing the context") if copy: diff --git a/tests/test_releasegil_callbacks.py b/tests/test_releasegil_callbacks.py index 7d920f0cc..8998da858 100644 --- a/tests/test_releasegil_callbacks.py +++ b/tests/test_releasegil_callbacks.py @@ -7,8 +7,10 @@ # Python callbacks (prefilters, fillers, postfilters, codecs and filters registered from Python) need the # GIL, so the calls that run them must keep it even when blosc2.set_releasegil(True) asks to release it. +# An exception raised in a callback must reach the caller as it is. import collections +import contextlib import threading import time import zlib @@ -30,6 +32,7 @@ DPARAMS = {"nthreads": 1} calls = collections.Counter() +failing = set() # the callbacks that raise KeyError instead (see fail()) @pytest.fixture(params=[False, True], ids=["gil-held", "gil-released"]) @@ -41,6 +44,23 @@ def releasegil(request): def called(name): calls[name] += 1 + if name in failing: + raise KeyError(name) + + +@contextlib.contextmanager +def fail(name): + """Make the callback `name` raise KeyError(name) meanwhile. The KeyError must reach the caller, and + the callback must run only once: C-Blosc2 stops at the first failure, so no Python code runs with the + exception set.""" + ncalls = calls[name] + failing.add(name) + try: + with pytest.raises(KeyError, match=name): + yield + finally: + failing.discard(name) + assert calls[name] == ncalls + 1 def encoder(input, output, meta, schunk): @@ -215,3 +235,148 @@ def test_write_releases_the_gil_next_to_python_codecs(registered): finally: blosc2.set_releasegil(old) assert gaps[-1] < 0.5, f"longest gaps, as fractions of the write: {gaps}" + + +# The ways of writing one chunk to a super-chunk holding a chunk (or half of one, for setitem-append) +FAILING_WRITE_PATHS = ["append", "insert", "update", "setitem", "setitem-grow", "setitem-append"] + + +def schunk_for(path, cparams=CPARAMS): + return new_schunk(cparams, data=data(N // 2 if path == "setitem-append" else N)) + + +def write_chunk(schunk, path): + values = data(N, start=1) + if path == "append": + schunk.append_data(values) + elif path == "insert": + schunk.insert_data(0, values, copy=False) + elif path == "update": + schunk.update_data(0, values, copy=True) + elif path == "setitem": + schunk[: N // 2] = values[: N // 2] + else: + # setitem-grow updates the second half of the full chunk, setitem-append recompresses the half + # chunk in full; both append a chunk then + schunk[N // 2 : 2 * N] = data(3 * N // 2, start=1) + + +@pytest.mark.parametrize("path", FAILING_WRITE_PATHS) +def test_prefilter_raises(releasegil, path): + schunk = schunk_for(path) + before = read(schunk) + + @schunk.prefilter(DTYPE, DTYPE) + def pf_raises_test_releasegil(input, output, offset): + called("prefilter") + output[:] = input + + with fail("prefilter"): + write_chunk(schunk, path) + np.testing.assert_array_equal(read(schunk), before) + + +def test_filler_raises(releasegil): + schunk = new_schunk() + with fail("filler"): + + @schunk.filler(((data(2 * N), DTYPE),), DTYPE) + def fill_raises_test_releasegil(inputs, output, offset): + called("filler") + + assert schunk.nchunks == 0 + + +@pytest.mark.parametrize("path", ["init", *FAILING_WRITE_PATHS]) +def test_python_codec_filter_raises_on_write(releasegil, python_cparams, path): + cparams, encode, _ = python_cparams + if path == "init": + with fail(encode): + new_schunk(cparams, data=data(2 * N)) + return + schunk = schunk_for(path, cparams) + before = read(schunk) + with fail(encode): + write_chunk(schunk, path) + np.testing.assert_array_equal(read(schunk), before) + + +READ_PATHS = ["decompress_chunk", "decompress_chunk-dst", "get_slice", "get_slice-out", "getitem"] + + +def read_chunk(schunk, path): + if path == "decompress_chunk": + schunk.decompress_chunk(0) + elif path == "decompress_chunk-dst": + schunk.decompress_chunk(0, np.empty(N, dtype=DTYPE)) + elif path == "get_slice": + schunk.get_slice(N // 4, N) + elif path == "get_slice-out": + schunk.get_slice(N // 4, N, out=np.empty(3 * N // 4, dtype=DTYPE)) + else: + schunk[N // 4 : N] + + +@pytest.mark.parametrize("path", READ_PATHS) +def test_postfilter_raises(releasegil, path): + schunk = new_schunk(data=data(2 * N)) + + @schunk.postfilter(DTYPE, DTYPE) + def post_raises_test_releasegil(input, output, offset): + called("postfilter") + output[:] = input + + with fail("postfilter"): + read_chunk(schunk, path) + + +@pytest.mark.parametrize("path", READ_PATHS) +def test_python_codec_filter_raises_on_read(releasegil, python_cparams, path): + cparams, _, decode = python_cparams + schunk = new_schunk(cparams, data=data(2 * N)) + with fail(decode): + read_chunk(schunk, path) + + +def test_python_codec_filter_raises_ndarray(releasegil, python_cparams): + cparams, encode, decode = python_cparams + kwargs = { + "cparams": blosc2.CParams(**cparams), + "dparams": blosc2.DParams(**DPARAMS), + "chunks": (20, 50), + "blocks": (10, 50), + } + values = data(40 * 50).reshape(40, 50) + with fail(encode): + blosc2.asarray(values, **kwargs) + array = blosc2.asarray(values, **kwargs) + with fail(encode): + array[5:25] = 0 + with fail(decode): + array[5:25] + with fail(decode): + array.read_into(np.empty((20, 50), dtype=DTYPE), slice(5, 25)) + with fail(decode): + array[[1, 3]] + with fail(decode): + blosc2.concat([array, blosc2.asarray(values, **kwargs)], axis=1) + np.testing.assert_array_equal(array[:], values) + + +def test_python_codec_filter_without_schunk(releasegil, python_cparams): + cparams = python_cparams[0] + # The callbacks raise this themselves: Python codecs and filters only work in a super-chunk + with pytest.raises(RuntimeError, match="without an SChunk"): + blosc2.compress2(data(N), **cparams) + with pytest.raises(RuntimeError, match="without an SChunk"): + blosc2.blosc2_ext.vlcompress([data(N).tobytes()], **cparams) + chunk = new_schunk(cparams, data=data(N)).get_chunk(0) + with pytest.raises(RuntimeError, match="without an SChunk"): + blosc2.decompress2(chunk, **DPARAMS) + old_nthreads = blosc2.set_nthreads(1) # decompress() uses the global number of threads + try: + for as_bytearray in (False, True): + with pytest.raises(RuntimeError, match="without an SChunk"): + blosc2.decompress(chunk, as_bytearray=as_bytearray) + finally: + blosc2.set_nthreads(old_nthreads)