diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index de90e254c..86431f114 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -910,12 +910,12 @@ blosc2_init() @atexit.register def destroy(): - # Daemon threads can still be inside a slice read without the GIL (see - # NDArray.get_slice_numpy), and blosc2_destroy() frees thread pools and plugins such a - # read may be using. If they do not finish soon, leave the cleanup to the OS. - global _nogil_reads_blocked - _nogil_reads_blocked += 1 - if _wait_for_nogil_reads(timeout=1.0): + # Daemon threads can still be inside a C call without the GIL (see _nogil_enter()), and + # blosc2_destroy() frees the global context and the thread pools such a call may be using. + # If they do not finish soon, leave the cleanup to the OS. + global _nogil_calls_blocked + _nogil_calls_blocked += 1 + if _wait_for_nogil_calls(timeout=1.0): blosc2_destroy() @@ -2037,6 +2037,15 @@ cdef class NativeGraphHandle: return result +cdef inline int _graph_execute_sources(const void *handle, const b2_array_source *sources, int n, + const b2_array_view *destination, void *output, size_t capacity, + unsigned raise_mask, b2_graph_report *report, + b2_graph_error *error) noexcept nogil: + if b2_graph_output_available(): + return b2_graph_execute_into(handle, NULL, sources, n, destination, raise_mask, report, error) + return b2_graph_execute_sources(handle, sources, n, output, capacity, raise_mask, report, error) + + cdef class NativeGraphSchedule: cdef void *_handle @@ -2235,12 +2244,17 @@ cdef class NativeGraphSchedule: normalized += bound.nbytes owned.append(bound) sources[i].view.name = name - with nogil: - if b2_graph_output_available(): - rc = b2_graph_execute_into(self._handle,NULL,sources,n,&destination,raise_mask,&report,&error) - else: - rc = b2_graph_execute_sources(self._handle, sources, n, np.PyArray_DATA(output), capacity, - raise_mask, &report, &error) + cdef void *output_data = np.PyArray_DATA(output) + # Native span readers decompress with C-Blosc2 from the graph's callbacks, which must hold the GIL + # for chunks of a codec or filter registered from Python (any chunk could use one) + if _nogil_enter(not (native_spans and _python_ids_registered)): + with nogil: + rc = _graph_execute_sources(self._handle, sources, n, &destination, output_data, capacity, + raise_mask, &report, &error) + _nogil_exit() + else: + rc = _graph_execute_sources(self._handle, sources, n, &destination, output_data, capacity, + raise_mask, &report, &error) for state in states: if state.error is not None: raise state.error @@ -2990,15 +3004,56 @@ cdef class MenudetResidentPrefilter: free(self.state.workers) free(self.placeholder) + cdef int _schedule(self, int32_t *encoded, size_t *chunks, size_t *discarded) noexcept nogil: + """evaluate()'s chunk loop: adds to *chunks and *discarded, leaves the last result in *encoded and + returns the first error code, or 0. The disposable prefilter ignores its scheduling source and + writes directly to the real sink. Borrow that same sink's valid address range as the driver: no + dummy input allocation or numerical copy.""" + cdef int64_t origin = 0, live + cdef int32_t nbytes + cdef int tid, rc = 0 + cdef blosc2_prefilter_params tail_params + while origin < self.count: + live = min(self.chunk_items, self.count - origin) + nbytes = (live * self.itemsize) + self.state.resident_origin = origin + self.state.resident_chunk_items = live + if live <= self.state.block_items: + # A single-block chunk needs no thread pool and must not + # let C-Blosc2's tuner shrink context->blocksize. + tail_params = self.parameters + tail_params.tid = 0 + tail_params.nchunk = chunks[0] + tail_params.nblock = 0 + tail_params.output = self.placeholder + tail_params.output_size = 0 + menudet_menudet_prefilter(&tail_params) + encoded[0] = BLOSC2_MAX_OVERHEAD + else: + encoded[0] = blosc2_compress_ctx(self.context, + self.state.numpy_output + origin * self.itemsize, nbytes, + self.placeholder, self.placeholder_capacity) + chunks[0] += 1 + if encoded[0] <= 0: + rc = encoded[0] + break + discarded[0] += encoded[0] + for tid in range(self.nworkers): + if self.state.workers[tid].status: + rc = self.state.workers[tid].status + break + if rc: + break + origin += live + return rc + def evaluate(self, inputs, out=None): cdef np.ndarray array, output cdef list owned = [] cdef int i, tid, rc = 0, active_workers = 0 - cdef int32_t encoded = 0, nbytes - cdef int64_t origin = 0, live + cdef int32_t encoded = 0 cdef size_t discarded = 0, chunks = 0 cdef menudet_prefilter_worker *worker - cdef blosc2_prefilter_params tail_params if self.context == NULL: raise RuntimeError("Resident prefilter context is not initialized") if self.running: @@ -3038,42 +3093,12 @@ cdef class MenudetResidentPrefilter: worker.evaluated_blocks = 0 worker.output_zero_fill_bytes = 0 memset(&worker.error, 0, sizeof(worker.error)) - # The disposable prefilter ignores its scheduling source and writes - # directly to the real sink. Borrow that same sink's valid address - # range as the driver: no dummy input allocation or numerical copy. - with nogil: - while origin < self.count: - live = min(self.chunk_items, self.count - origin) - nbytes = (live * self.itemsize) - self.state.resident_origin = origin - self.state.resident_chunk_items = live - if live <= self.state.block_items: - # A single-block chunk needs no thread pool and must not - # let C-Blosc2's tuner shrink context->blocksize. - tail_params = self.parameters - tail_params.tid = 0 - tail_params.nchunk = chunks - tail_params.nblock = 0 - tail_params.output = self.placeholder - tail_params.output_size = 0 - menudet_menudet_prefilter(&tail_params) - encoded = BLOSC2_MAX_OVERHEAD - else: - encoded = blosc2_compress_ctx(self.context, - self.state.numpy_output + origin * self.itemsize, nbytes, - self.placeholder, self.placeholder_capacity) - chunks += 1 - if encoded <= 0: - rc = encoded - break - discarded += encoded - for tid in range(self.nworkers): - if self.state.workers[tid].status: - rc = self.state.workers[tid].status - break - if rc: - break - origin += live + if _nogil_enter(True): + with nogil: + rc = self._schedule(&encoded, &chunks, &discarded) + _nogil_exit() + else: + rc = self._schedule(&encoded, &chunks, &discarded) menudet_prefilter_merge(&self.state) if self.state.status: try: @@ -3351,12 +3376,13 @@ cpdef compress(src, int32_t typesize=8, int clevel=9, filter=blosc2.Filter.SHUFF cdef int32_t len_dest = len(dest) cdef int size cdef int filter_ = filter.value if isinstance(filter, Enum) else 0 - if RELEASEGIL: - _dest = dest + _dest = dest + if _nogil_enter(RELEASEGIL): with nogil: size = blosc2_compress(clevel, filter_, typesize, buf.buf, buf.len, _dest, len_dest) + _nogil_exit() else: - size = blosc2_compress(clevel, filter_, typesize, buf.buf, buf.len, dest, len_dest) + size = blosc2_compress(clevel, filter_, typesize, buf.buf, buf.len, _dest, len_dest) PyBuffer_Release(&buf) if size > 0: return dest[:size] @@ -3489,10 +3515,7 @@ cdef inline c_bool _uses_python_filter(const blosc2_cparams *cparams) noexcept: 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.""" + callback can run, i.e. its prefilter, codec and filters are not Python ones.""" cdef blosc2_cparams cparams if not RELEASEGIL or cctx == NULL or blosc2_ctx_get_cparams(cctx, &cparams) < 0: return False @@ -3503,9 +3526,10 @@ cdef int _compress_ctx(blosc2_context *cctx, const void *src, int32_t srcsize, v 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): + if _nogil_enter(_can_compress_without_gil(cctx)): with nogil: rc = blosc2_compress_ctx(cctx, src, srcsize, dest, destsize) + _nogil_exit() else: rc = blosc2_compress_ctx(cctx, src, srcsize, dest, destsize) return rc @@ -3515,9 +3539,10 @@ cdef int _decompress_ctx(blosc2_context *dctx, const void *src, int32_t srcsize, """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: + if _nogil_enter(RELEASEGIL and not _python_ids_registered): with nogil: rc = blosc2_decompress_ctx(dctx, src, srcsize, dest, destsize) + _nogil_exit() else: rc = blosc2_decompress_ctx(dctx, src, srcsize, dest, destsize) return rc @@ -3553,9 +3578,10 @@ cdef int64_t _store_buffer(blosc2_schunk *schunk, _write_op op, int64_t nchunk, cdef int64_t rc if nbytes > BLOSC2_MAX_BUFFERSIZE: raise ValueError(f"Cannot compress more than {BLOSC2_MAX_BUFFERSIZE} bytes into one chunk") - if _can_compress_without_gil(schunk.cctx): + if _nogil_enter(_can_compress_without_gil(schunk.cctx)): with nogil: rc = _store_buffer_nogil(schunk, op, nchunk, src, nbytes) + _nogil_exit() else: rc = _store_buffer_nogil(schunk, op, nchunk, src, nbytes) return rc @@ -3799,9 +3825,10 @@ def vlcompress(srcs, **kwargs): cctx = blosc2_create_cctx(cparams) if cctx == NULL: raise RuntimeError("Could not create the compression context") - if _can_compress_without_gil(cctx): + if _nogil_enter(_can_compress_without_gil(cctx)): with nogil: size = blosc2_vlcompress_ctx(cctx, src_ptrs, srcsizes, nblocks, _dest, len_dest) + _nogil_exit() else: size = blosc2_vlcompress_ctx(cctx, src_ptrs, srcsizes, nblocks, _dest, len_dest) finally: @@ -4042,32 +4069,54 @@ cdef class HandleLocks: lock.release() -# Slice reads running without the GIL (see NDArray.get_slice_numpy), and whether new ones must -# keep it (> 0). Both only change with the GIL held. -cdef int _nogil_reads = 0 -cdef int _nogil_reads_blocked = 0 +# C calls running without the GIL (see _nogil_enter()), and whether new ones must keep it (> 0). +# Both only change with the GIL held. +cdef int _nogil_calls = 0 +cdef int _nogil_calls_blocked = 0 + + +cdef inline c_bool _nogil_enter(c_bool release) noexcept: + """Whether a C call may run without the GIL: if `release` (the caller's own conditions) and no + _wait_for_nogil_calls() caller needs such calls to stop. If so, the call counts as running until + the caller, holding the GIL again, calls _nogil_exit(). Every C-Blosc2 call in this module that + releases the GIL goes through here, except opening a frame and lock(): nothing needs to wait for + them, as they run no codec or filter from Python and use neither the global context nor (unless + BLOSC_NTHREADS asks for it) a thread pool, and most of them can wait for another thread's frame + lock. miniexpr's native graph and artifact calls reach C-Blosc2 only through native span + readers, so of those only NativeGraphSchedule._execute_sources_bound() counts.""" + global _nogil_calls + if release and not _nogil_calls_blocked: + _nogil_calls += 1 + return True + return False + + +cdef inline void _nogil_exit() noexcept: + """End a call that _nogil_enter() let run without the GIL; the caller holds the GIL again.""" + global _nogil_calls + _nogil_calls -= 1 -def _wait_for_nogil_reads(timeout=None): - """Wait until no slice read runs without the GIL; return False if `timeout` (s) ran out. +def _wait_for_nogil_calls(timeout=None): + """Wait until no C call runs without the GIL; return False if `timeout` (s) ran out. - For changes to C state that such reads use (registering a Python codec or filter, exit). - The caller increments `_nogil_reads_blocked` first, so that no new ones start meanwhile. + For changes to C state that such calls use (registering a Python codec or filter, exit). + The caller increments `_nogil_calls_blocked` first, so that no new ones start meanwhile. """ deadline = None if timeout is None else time.monotonic() + timeout - while _nogil_reads: + while _nogil_calls: if deadline is not None and time.monotonic() > deadline: return False - time.sleep(0.0005) # releases the GIL, so the readers can finish + time.sleep(0.0005) # releases the GIL, so the calls can finish return True def _reset_after_fork(): - # The threads that held handle locks or read without the GIL at the fork do not exist in - # the child, so nothing would ever release those locks there - global _nogil_reads, _nogil_reads_blocked - _nogil_reads = 0 - _nogil_reads_blocked = 0 + # The threads that held handle locks or ran C calls without the GIL at the fork do not exist + # in the child, so nothing would ever release those locks there + global _nogil_calls, _nogil_calls_blocked + _nogil_calls = 0 + _nogil_calls_blocked = 0 for lock in list(_all_handle_locks): if not lock._is_owned(): lock._at_fork_reinit() @@ -4143,7 +4192,8 @@ cdef class SChunk: create_storage(&storage, kwargs) if self.mode == "r": - with nogil: # A lock holder in another thread must be able to resume. + # A lock holder in another thread must be able to resume; not counted (see _nogil_enter()) + with nogil: if storage.io != NULL: # mmap or locking: open through the user-defined io self.schunk = blosc2_schunk_open_offset_udio(storage.urlpath, 0, storage.io) @@ -4587,11 +4637,8 @@ cdef class SChunk: self.schunk.current_nchunk = nchunk # prefilter needs this value to be set cdef int size try: - if _can_compress_without_gil(self.schunk.cctx): - with nogil: - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk_buf.buf, chunk_buf.len) - else: - size = blosc2_compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk_buf.buf, chunk_buf.len) + size = _compress_ctx(self.schunk.cctx, buf.buf, buf.len, chunk_buf.buf, + chunk_buf.len) finally: PyBuffer_Release(&buf) PyBuffer_Release(&chunk_buf) @@ -4718,7 +4765,7 @@ cdef class SChunk: `holding_lock()` context manager instead of calling this directly. """ cdef int rc - with nogil: # may block waiting for other processes + with nogil: # may block waiting for other processes; not counted (see _nogil_enter()) rc = blosc2_schunk_lock(self.schunk) if rc < 0: raise RuntimeError("Could not lock the super-chunk frame") @@ -5843,7 +5890,8 @@ def open(urlpath, mode, offset, **kwargs): if mmap_mode is None: io = &_locking_io if locking else NULL - with nogil: # Opening can wait for another Python thread's frame lock. + # Opening can wait for another Python thread's frame lock; not counted (see _nogil_enter()) + with nogil: if io != NULL: schunk = blosc2_schunk_open_offset_udio(path, frame_offset, io) else: @@ -5861,7 +5909,7 @@ def open(urlpath, mode, offset, **kwargs): io = malloc(sizeof(blosc2_io)) io.id = BLOSC2_IO_FILESYSTEM_MMAP io.params = mmap_file - with nogil: + with nogil: # not counted (see _nogil_enter()) schunk = blosc2_schunk_open_offset_udio(path, frame_offset, io) if schunk == NULL: @@ -5997,9 +6045,10 @@ cdef schunk_to_cframe(blosc2_schunk* schunk, c_bool release_gil): dest.out = NULL dest.alloc_failed = False cdef int64_t cframe_len - if release_gil: + if _nogil_enter(release_gil): with nogil: cframe_len = blosc2_schunk_to_buffer_alloc(schunk, cframe_alloc, &dest) + _nogil_exit() else: cframe_len = blosc2_schunk_to_buffer_alloc(schunk, cframe_alloc, &dest) out = dest.out if dest.out != NULL else None @@ -6092,13 +6141,13 @@ def register_codec(codec_name, id, encoder=None, decoder=None, version=1): else: codec.decoder = general_decoder - # 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, _python_ids_registered - _nogil_reads_blocked += 1 + # A C call that found no Python codec could otherwise meet `decoder` without the GIL: keep new + # calls from releasing it and wait for those that did. Nothing below releases the GIL before + # the registry is updated. + global _nogil_calls_blocked, _python_ids_registered + _nogil_calls_blocked += 1 try: - _wait_for_nogil_reads() + _wait_for_nogil_calls() rc = blosc2_register_codec(&codec) if rc < 0: raise RuntimeError("Error while registering codec") @@ -6107,7 +6156,7 @@ def register_codec(codec_name, id, encoder=None, decoder=None, version=1): blosc2.ucodecs_registry[id] = (codec_name, encoder, decoder) _python_codec_ids[id] = _python_ids_registered = True finally: - _nogil_reads_blocked -= 1 + _nogil_calls_blocked -= 1 cdef int general_forward(const uint8_t* input_buffer, uint8_t* output_buffer, int32_t size, @@ -6170,10 +6219,10 @@ def register_filter(id, forward, backward, filter_name): strcpy(filter.name, filter_name_) # As in register_codec() - global _nogil_reads_blocked, _python_ids_registered - _nogil_reads_blocked += 1 + global _nogil_calls_blocked, _python_ids_registered + _nogil_calls_blocked += 1 try: - _wait_for_nogil_reads() + _wait_for_nogil_calls() rc = blosc2_register_filter(&filter) if rc < 0: raise RuntimeError("Error while registering filter") @@ -6181,7 +6230,7 @@ def register_filter(id, forward, backward, filter_name): blosc2.ufilters_registry[id] = (forward, backward) _python_filter_ids[id] = _python_ids_registered = True finally: - _nogil_reads_blocked -= 1 + _nogil_calls_blocked -= 1 cdef _check_rc(rc, message): # Before anything else: a Python callback may have failed in the C call that gave rc @@ -6299,9 +6348,8 @@ cdef c_bool _can_release_gil(blosc2_schunk *sc) except? False: """Whether decompressing from `sc` may run without the GIL, i.e. no Python callback can run on the way: no postfilter on its decompression context and no codec or filter registered from Python (any chunk could use one, whatever the super-chunk's own cparams say). Never on wasm32, - where releasing the GIL is known to give wrong results (see blosc2.set_releasegil), nor while - _wait_for_nogil_reads() callers need the reads to stop.""" - if _nogil_reads_blocked or blosc2.IS_WASM or blosc2.ucodecs_registry or blosc2.ufilters_registry: + where releasing the GIL is known to give wrong results (see blosc2.set_releasegil).""" + if blosc2.IS_WASM or blosc2.ucodecs_registry or blosc2.ufilters_registry: return False cdef blosc2_dparams dparams if sc.dctx == NULL or blosc2_ctx_get_dparams(sc.dctx, &dparams) < 0: @@ -6693,16 +6741,14 @@ cdef class NDArray: callback can run on the way, so that threads reading other arrays (or other handles of the same file) run in parallel; everything else that uses this super-chunk holds the lock meanwhile.""" - global _nogil_reads cdef int rc - if _can_release_gil(self.array.sc): - _nogil_reads += 1 + if _nogil_enter(_can_release_gil(self.array.sc)): with nogil: if into: 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) - _nogil_reads -= 1 + _nogil_exit() elif into: rc = b2nd_get_slice_cbuffer_into(self.array, start_, stop_, buf, buffershape_, buffersize) else: @@ -6807,6 +6853,9 @@ cdef class NDArray: # teams. Zero preserves the existing caller/storage decompression policy. if nthreads: dparams.nthreads = nthreads + # A codec or filter registered from Python (any chunk could use one) needs the GIL, on this thread + if _python_ids_registered: + dparams.nthreads = 1 dparams.schunk = self.array.sc dparams.typesize = self.array.sc.typesize dctx = blosc2_create_dctx(dparams) @@ -6824,8 +6873,15 @@ cdef class NDArray: want_nbytes = nitems * self.array.sc.typesize # This invocation owns its context, chunk reference and destination; # release the GIL so independent bounded graph readers can decompress - # concurrently instead of serializing all native source callbacks. - with nogil: + # concurrently instead of serializing all native source callbacks + # (unless a codec or filter is registered from Python, see above). + if _nogil_enter(not _python_ids_registered): + with nogil: + rc = blosc2_getitem_bytes_ctx(dctx, chunk, lazychunk_cbytes, + start * self.array.sc.typesize, want_nbytes, + view.buf, view.len) + _nogil_exit() + else: rc = blosc2_getitem_bytes_ctx(dctx, chunk, lazychunk_cbytes, start * self.array.sc.typesize, want_nbytes, view.buf, view.len) diff --git a/tests/test_nogil_calls.py b/tests/test_nogil_calls.py new file mode 100644 index 000000000..dc5ed1134 --- /dev/null +++ b/tests/test_nogil_calls.py @@ -0,0 +1,159 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +# Every C call that runs without the GIL is counted, so that the exit (blosc2_destroy() frees the global +# context and the thread pools) and register_codec()/register_filter() wait for it. Both tests run in a +# new interpreter: no codec or filter registered from Python by other tests may keep the GIL. + +import json +import subprocess +import sys + +import pytest + +import blosc2 + +pytestmark = pytest.mark.skipif(blosc2.IS_WASM, reason="Pyodide cannot start threads or processes") + +COUNTED_SCRIPT = """ +import json +import threading +import time + +import numpy as np +import blosc2 +from blosc2 import blosc2_ext + +blosc2.set_releasegil(True) +data = (np.arange(2**20) % 1000).astype(np.int32) # 4 MiB +cparams = {"typesize": 4, "nthreads": 1, "codec": blosc2.Codec.ZSTD, "clevel": 5} +chunk = blosc2.compress2(data, **cparams) +schunk = blosc2.SChunk(chunksize=data.nbytes, cparams=cparams) +sparse = blosc2.SChunk(chunksize=data.nbytes // 8, data=data, cparams=cparams, contiguous=False) +array = blosc2.asarray(data.reshape(1024, 1024), chunks=(256, 1024), cparams=cparams) +flat = blosc2.asarray(data, chunks=(2**18,), cparams=cparams) +span = np.empty(2**18, dtype=np.int32) + + +def append(): + schunk.append_data(data) + schunk.delete_chunk(0) + + +calls = { + "compress": lambda: blosc2.compress(data, typesize=4, clevel=5, codec=blosc2.Codec.ZSTD), + "compress2": lambda: blosc2.compress2(data, **cparams), + "decompress2": lambda: blosc2.decompress2(chunk), + "vlcompress": lambda: blosc2_ext.vlcompress([data[:1000], data[1000:]], **cparams), + "append_data": append, + "to_cframe": sparse.to_cframe, + "NDArray slice": lambda: array[:], + "get_1d_span_numpy": lambda: flat.get_1d_span_numpy(span, 0, 0, span.size), +} + +# Native graph and Menudet routes, where this build and NumPy support them +values = np.arange(2**20, dtype=np.float64) +try: + from blosc2.graph_plan import NativeGraph + from blosc2.storage_readers import ReaderState, StorageReader + + if not blosc2_ext.native_graph_sources_available(): + raise NotImplementedError + compressed = blosc2.asarray(values, chunks=(2**18,), blocks=(2**14,)) + schedule = NativeGraph.from_expression("x + 1", {"x": "float64"}).specialize( + {"x": (values.dtype, values.shape)}, tile_items=2**14 + ) + + def graph_with_native_spans(): + reader = StorageReader(compressed, values.shape) + reader.start = 0 + schedule.execute_sources({"x": compressed}, {"x": ReaderState(reader)}, native_spans=True) + + calls["graph with native spans"] = graph_with_native_spans +except Exception: + pass +try: + from blosc2.native_graph import prepare_resident + + if not blosc2_ext.menudet_resident_prefilter_available(): + raise NotImplementedError + with blosc2.jit_options(jit=False, jit_backend="cc", math_backend="system"): + resident = prepare_resident("x*x - 1", {"x": values}, jit=False, optimization="aggressive")[2] + executor = resident.kernel._handle.resident_prefilter( + values.shape, workers=2, block_items=2**14, chunk_items=2**18, graph=resident.plan._handle + ) + calls["resident prefilter"] = lambda: executor.evaluate({"x": values}) +except Exception: + pass +problems = [] +for name, call in calls.items(): + stop = threading.Event() + + def loop(): + while not stop.is_set(): + call() + + thread = threading.Thread(target=loop) + thread.start() + deadline = time.monotonic() + 10 + seen = False + try: + while not seen and time.monotonic() < deadline: + time.sleep(0.001) + # With the GIL held here, the other thread is either inside the call without it, or not in it + seen = not blosc2_ext._wait_for_nogil_calls(timeout=0) + finally: + stop.set() + thread.join() + if not seen: + problems.append(f"{name}: not counted") + if not blosc2_ext._wait_for_nogil_calls(timeout=0): + problems.append(f"{name}: still counted after it returned") +print(json.dumps(problems)) +""" + +# Two daemon threads compress() (the global context) while the interpreter exits +EXIT_SCRIPT = """ +import threading +import time + +import numpy as np +import blosc2 + +blosc2.set_releasegil(True) +# Blocks this large make the global context's temporaries big enough to be mmap()ed: freeing them under +# the compressing thread crashes it instead of going unnoticed +blosc2.set_blocksize(4 * 2**20) +data = (np.arange(4 * 2**20) % 1000).astype(np.int32) # 16 MiB + + +def compress(): + while True: + blosc2.compress(data, typesize=4, codec=blosc2.Codec.ZSTD, clevel=3) + + +for _ in range(2): + threading.Thread(target=compress, daemon=True).start() +time.sleep(0.2) +""" + + +def run_python(code): + return subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, timeout=120) + + +def test_calls_without_the_gil_are_counted(): + result = run_python(COUNTED_SCRIPT) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout) == [] + + +def test_exit_waits_for_daemon_threads_in_calls_without_the_gil(): + # Without the waiting, 4 out of 5 runs crashed with SIGSEGV on Linux + for _ in range(3): + result = run_python(EXIT_SCRIPT) + assert result.returncode == 0, result.stderr diff --git a/tests/test_span_reads_python_callbacks.py b/tests/test_span_reads_python_callbacks.py new file mode 100644 index 000000000..69173c218 --- /dev/null +++ b/tests/test_span_reads_python_callbacks.py @@ -0,0 +1,94 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +# Span reads (NDArray.get_1d_span_numpy() and the native span readers of graph executions) decompress +# without the GIL. Chunks of a codec or filter registered from Python need it: their callbacks run Python. +# Each case runs in a child interpreter, because the bug crashes it. + +import subprocess +import sys + +import pytest + +import blosc2 + +pytestmark = pytest.mark.skipif(blosc2.IS_WASM, reason="Pyodide cannot start processes") + +SETUP = """ +import zlib + +import numpy as np +import blosc2 + + +def encoder(src, dst, meta, schunk): + out = zlib.compress(src.tobytes(), 1) + dst[: len(out)] = np.frombuffer(out, np.uint8) + return len(out) + + +def decoder(src, dst, meta, schunk): + out = zlib.decompress(src.tobytes()) + dst[: len(out)] = np.frombuffer(out, np.uint8) + return len(out) + + +def reverse(src, dst, meta, schunk): + dst[:] = src[::-1] + + +if KIND == "codec": + blosc2.register_codec("span_zlib", 180, encoder, decoder) + cparams = {"codec": 180, "nthreads": 1, "filters": [], "splitmode": blosc2.SplitMode.NEVER_SPLIT} +else: + blosc2.register_filter(181, reverse, reverse, "span_reverse") + cparams = {"filters": [181], "filters_meta": [0], "nthreads": 1} +""" + +SPAN_SCRIPT = """ +a = blosc2.asarray(np.arange(2**16, dtype=np.int32), chunks=(2**14,), blocks=(2**12,), cparams=cparams, + dparams={"nthreads": 1}) +out = np.empty(2**13, dtype=np.int32) +for _ in range(20): + a.get_1d_span_numpy(out, 1, 0, out.size, NTHREADS) +assert (out == np.arange(2**14, 2**14 + out.size)).all() +""" + +GRAPH_SCRIPT = """ +from blosc2.graph_plan import NativeGraph +from blosc2.storage_readers import ReaderState, StorageReader + +values = np.arange(2**16, dtype=np.float64) +x = blosc2.asarray(values, chunks=(2**14,), blocks=(2**12,), cparams=cparams, dparams={"nthreads": 1}) +reader = StorageReader(x, values.shape) +reader.start = 0 +schedule = NativeGraph.from_expression("x + 1", {"x": "float64"}).specialize( + {"x": (values.dtype, values.shape)}, tile_items=2**12 +) +for _ in range(5): + result = schedule.execute_sources({"x": x}, {"x": ReaderState(reader)}, native_spans=True)[0] +assert (result == values + 1).all() +""" + + +def run_child(kind, script, nthreads=0): + code = f"KIND = {kind!r}\nNTHREADS = {nthreads}\n" + SETUP + script + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, timeout=120) + assert result.returncode == 0, result.stderr + + +@pytest.mark.parametrize("kind", ["codec", "filter"]) +@pytest.mark.parametrize("nthreads", [0, 4]) +def test_get_1d_span_numpy_python_callbacks(kind, nthreads): + run_child(kind, SPAN_SCRIPT, nthreads) + + +@pytest.mark.parametrize("kind", ["codec", "filter"]) +def test_graph_native_spans_python_callbacks(kind, native_graph_runtime): + if not blosc2.blosc2_ext.native_graph_sources_available(): + pytest.skip("Installed extension does not implement graph storage readers") + run_child(kind, GRAPH_SCRIPT)