From 09e7a47f1a1f9eff9622390d9d1fbdd8255e2fff Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Wed, 28 Oct 2026 21:32:42 +0100 Subject: [PATCH 1/5] Release the GIL in NDArray slice reads NDArray.get_slice_numpy() decompressed with the GIL held, so threads reading separate arrays, or separate handles of one file, ran one at a time; on NFS this included the wait for the server. Release the GIL around b2nd_get_slice_cbuffer(). The GIL was also what kept two threads from using one super-chunk at the same time. Replace the per-NDArray read_lock with one reentrant lock per super-chunk, shared by the SChunk wrapper, the NDArray, its views and vlmeta, and hold it in every entry point into C. Keep the GIL when a Python postfilter, codec or filter could run during the read, and on wasm32. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SwMZj7Wi7Sc8HtQpELyNXB --- src/blosc2/blosc2_ext.pyx | 173 ++++++++++++++++++++---- src/blosc2/core.py | 4 + tests/ndarray/test_concurrent_access.py | 161 ++++++++++++++++++++++ 3 files changed, 310 insertions(+), 28 deletions(-) create mode 100644 tests/ndarray/test_concurrent_access.py diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 64affa623..47740ecbe 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -14,8 +14,10 @@ import ast import atexit import pathlib import sys +import functools import time import warnings +from threading import RLock import _ctypes @@ -365,6 +367,8 @@ cdef extern from "blosc2.h": blosc2_context* blosc2_create_dctx(blosc2_dparams dparams) nogil + int blosc2_ctx_get_dparams(blosc2_context *ctx, blosc2_dparams *dparams) + void blosc2_free_ctx(blosc2_context * context) nogil int blosc2_set_maskout(blosc2_context *ctx, c_bool *maskout, int nblocks) @@ -622,7 +626,7 @@ cdef extern from "b2nd.h": int b2nd_free(b2nd_array_t *array) int b2nd_get_slice_cbuffer(b2nd_array_t *array, int64_t *start, int64_t *stop, - void *buffer, int64_t *buffershape, int64_t buffersize) + void *buffer, int64_t *buffershape, int64_t buffersize) nogil int b2nd_set_slice_cbuffer(void *buffer, int64_t *buffershape, int64_t buffersize, int64_t *start, int64_t *stop, b2nd_array_t *array) int b2nd_get_slice(b2nd_context_t *ctx, b2nd_array_t **array, b2nd_array_t *src, const int64_t *start, @@ -1707,11 +1711,30 @@ cdef get_chunk_repeatval(blosc2_cparams cparams, const int32_t nbytes, raise RuntimeError("Problems when creating the repeated values chunk") +def _handle_locked(method): + """Run `method` holding the handle lock of its first argument (an SChunk, NDArray or vlmeta). + + There is one such lock per C super-chunk, shared by the SChunk wrapper, the NDArray built on it + and that array's views. NDArray.get_slice_numpy() decompresses without the GIL, so every other + operation that uses the super-chunk's C state (its contexts, frame, metalayers) must hold this + lock instead of relying on the GIL. It is reentrant because Python callbacks (prefilters, + postfilters, UDFs) can call back into the same array. + """ + @functools.wraps(method) + def locked(self, *args, **kwargs): + with self._handle_lock: + return method(self, *args, **kwargs) + return locked + + cdef class SChunk: cdef blosc2_schunk *schunk cdef c_bool _is_view + cdef readonly object _handle_lock def __init__(self, _schunk=None, chunksize=2 ** 24, data=None, **kwargs): + # Replaced by the NDArray's lock when this is an NDArray's SChunk (see NDArray.__init__) + self._handle_lock = RLock() # hold on to a bytestring of urlpath for the lifetime of the instance # because its value is referenced via a C-pointer urlpath = kwargs.get("urlpath", None) @@ -1881,6 +1904,7 @@ cdef class SChunk: def contiguous(self): return self.schunk.storage.contiguous + @_handle_locked def get_cparams(self): if self.schunk.storage.cparams.compcode in blosc2.Codec._value2member_map_: codec = blosc2.Codec(self.schunk.storage.cparams.compcode) @@ -1914,6 +1938,7 @@ cdef class SChunk: return cparams + @_handle_locked def update_cparams(self, new_cparams): cdef blosc2_cparams* cparams = self.schunk.storage.cparams codec = new_cparams.codec @@ -1957,9 +1982,11 @@ cdef class SChunk: self.schunk.filters = self.schunk.storage.cparams.filters self.schunk.filters_meta = self.schunk.storage.cparams.filters_meta + @_handle_locked def get_dparams(self): return blosc2.DParams(nthreads=self.schunk.storage.dparams.nthreads) + @_handle_locked def update_dparams(self, new_dparams): cdef blosc2_dparams* dparams = self.schunk.storage.dparams dparams.nthreads = new_dparams.nthreads @@ -1971,6 +1998,7 @@ cdef class SChunk: if self.schunk.dctx == NULL: raise RuntimeError("Could not create decompression context") + @_handle_locked def append_data(self, data): cdef Py_buffer buf PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE) @@ -1998,6 +2026,7 @@ cdef class SChunk: raise RuntimeError("Could not append the chunk") return rc + @_handle_locked def fill_special(self, nitems, special_value, value): if value is None: return blosc2_schunk_fill_special(self.schunk, nitems, special_value, self.chunksize) @@ -2046,6 +2075,7 @@ cdef class SChunk: return self.nchunks + @_handle_locked def decompress_chunk(self, nchunk, dst=None): cdef uint8_t *chunk cdef c_bool needs_free @@ -2079,6 +2109,7 @@ cdef class SChunk: if size < 0: raise RuntimeError(f"Error while decompressing the specified chunk, error code: {size}") + @_handle_locked def get_chunk(self, nchunk): cdef uint8_t *chunk cdef c_bool needs_free @@ -2090,6 +2121,7 @@ cdef class SChunk: free(chunk) return ret_chunk + @_handle_locked def get_lazychunk(self, nchunk): cdef uint8_t *chunk cdef c_bool needs_free @@ -2113,6 +2145,7 @@ cdef class SChunk: free(chunk) return ret_chunk + @_handle_locked def get_vlblock(self, nchunk, nblock): cdef uint8_t *block cdef int32_t destsize @@ -2123,12 +2156,14 @@ cdef class SChunk: free(block) return ret_block + @_handle_locked def delete_chunk(self, nchunk): rc = blosc2_schunk_delete_chunk(self.schunk, nchunk) if rc < 0: raise RuntimeError("Could not delete the desired chunk") return rc + @_handle_locked def append_chunk(self, chunk): cdef const uint8_t[:] typed_view_chunk mem_view_chunk = memoryview(chunk) @@ -2139,6 +2174,7 @@ cdef class SChunk: raise RuntimeError("Could not append the desired chunk") return rc + @_handle_locked def insert_chunk(self, nchunk, chunk): cdef const uint8_t[:] typed_view_chunk mem_view_chunk = memoryview(chunk) @@ -2149,6 +2185,7 @@ cdef class SChunk: raise RuntimeError("Could not insert the desired chunk") return rc + @_handle_locked def insert_data(self, nchunk, data, copy): cdef blosc2_context *cctx cdef Py_buffer buf @@ -2179,6 +2216,7 @@ cdef class SChunk: raise RuntimeError("Could not insert the desired chunk") return rc + @_handle_locked def update_chunk(self, nchunk, chunk): cdef const uint8_t[:] typed_view_chunk mem_view_chunk = memoryview(chunk) @@ -2189,6 +2227,7 @@ cdef class SChunk: raise RuntimeError("Could not update the desired chunk") return rc + @_handle_locked def reorder_offsets(self, order): cdef np.ndarray[np.int64_t, ndim=1] offsets_order = np.ascontiguousarray(order, dtype=np.int64) rc = blosc2_schunk_reorder_offsets(self.schunk, offsets_order.data) @@ -2196,6 +2235,7 @@ cdef class SChunk: raise RuntimeError("Could not reorder the chunk offsets") return None + @_handle_locked def update_data(self, nchunk, data, copy): cdef Py_buffer buf PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE) @@ -2226,6 +2266,7 @@ cdef class SChunk: return rc # This is used internally for prefiltering + @_handle_locked def _prefilter_data(self, nchunk, data, chunk_data): cdef Py_buffer buf PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE) @@ -2241,6 +2282,7 @@ cdef class SChunk: raise RuntimeError("The result could not fit ") return size + @_handle_locked def get_slice(self, start=0, stop=None, out=None): cdef int64_t nitems = self.schunk.nbytes // self.schunk.typesize start, stop, _ = slice(start, stop, 1).indices(nitems) @@ -2265,6 +2307,7 @@ cdef class SChunk: if rc < 0: raise RuntimeError("Error while getting the slice") + @_handle_locked def set_slice(self, value, start=0, stop=None): cdef int64_t nitems = self.schunk.nbytes // self.schunk.typesize start, stop = self._massage_key(start, stop, nitems) @@ -2370,6 +2413,7 @@ cdef class SChunk: if rc < 0: raise RuntimeError("Error while setting the slice") + @_handle_locked def to_cframe(self): cdef c_bool needs_free cdef uint8_t *cframe @@ -2387,9 +2431,11 @@ cdef class SChunk: return out + @_handle_locked def _avoid_cframe_free(self, avoid_cframe_free): blosc2_schunk_avoid_cframe_free(self.schunk, avoid_cframe_free) + @_handle_locked def lock(self): """Take the exclusive frame lock and hold it across several operations. @@ -2402,12 +2448,14 @@ cdef class SChunk: if rc < 0: raise RuntimeError("Could not lock the super-chunk frame") + @_handle_locked def unlock(self): """Release the frame lock taken with `lock()`.""" cdef int rc = blosc2_schunk_unlock(self.schunk) if rc < 0: raise RuntimeError("Could not unlock the super-chunk frame") + @_handle_locked def refresh(self): cdef int rc = blosc2_schunk_refresh(self.schunk) _check_rc(rc, "Error while refreshing the schunk") @@ -2433,6 +2481,7 @@ cdef class SChunk: return start, stop + @_handle_locked def _set_postfilter(self, func, dtype_input, dtype_output=None): # Get user data func_id = func.__name__ @@ -2467,7 +2516,11 @@ cdef class SChunk: if self.schunk.dctx == NULL: raise RuntimeError("Could not create decompression context") - cpdef remove_postfilter(self, func_name, _new_ctx=True): + @_handle_locked + def remove_postfilter(self, func_name, _new_ctx=True): + self._remove_postfilter(func_name, _new_ctx) + + cdef _remove_postfilter(self, func_name, _new_ctx): if func_name is not None: del blosc2.postfilter_funcs[func_name] @@ -2487,6 +2540,7 @@ cdef class SChunk: # Avoid creating new dctx when calling this from the __dealloc__ self.schunk.dctx = NULL + @_handle_locked def _set_filler(self, func, inputs_id, dtype_output): if self.schunk.storage.cparams.nthreads > 1: raise AttributeError("compress `nthreads` must be 1 when assigning a prefilter") @@ -2516,6 +2570,7 @@ cdef class SChunk: if self.schunk.cctx == NULL: raise RuntimeError("Could not create compression context") + @_handle_locked def _set_prefilter(self, func, dtype_input, dtype_output=None): if self.schunk.storage.cparams.nthreads > 1: raise AttributeError("compress `nthreads` must be 1 when assigning a prefilter") @@ -2551,7 +2606,11 @@ cdef class SChunk: if self.schunk.cctx == NULL: raise RuntimeError("Could not create compression context") - cpdef remove_prefilter(self, func_name, _new_ctx=True): + @_handle_locked + def remove_prefilter(self, func_name, _new_ctx=True): + self._remove_prefilter(func_name, _new_ctx) + + cdef _remove_prefilter(self, func_name, _new_ctx): cdef udf_udata* udf_data cdef user_filters_udata* udata cdef mm_udata* mm_data @@ -2630,9 +2689,9 @@ cdef class SChunk: if self.schunk != NULL and not self._is_view: # Free prefilters and postfilters params if self.schunk.storage.cparams.prefilter != NULL: - self.remove_prefilter(func_name=None, _new_ctx=False) + self._remove_prefilter(None, False) if self.schunk.storage.dparams.postfilter != NULL: - self.remove_postfilter(func_name=None, _new_ctx=False) + self._remove_postfilter(None, False) # Free the C-Blosc2 super-chunk with the GIL held so threadpool # teardown cannot race with active miniexpr workers. @@ -3320,6 +3379,13 @@ cdef class vlmeta: def __init__(self, schunk): self.schunk = schunk + @property + def _handle_lock(self): + # Looked up on every call: an NDArray swaps its own lock into the owning SChunk after + # that SChunk (and this object) were created + return self._owner._handle_lock + + @_handle_locked def set_vlmeta(self, name, content, **cparams): cdef blosc2_cparams ccparams create_cparams_from_kwargs(&ccparams, cparams) @@ -3335,6 +3401,7 @@ cdef class vlmeta: if rc < 0: raise RuntimeError + @_handle_locked def get_vlmeta(self, name): name = name.encode("utf-8") if isinstance(name, str) else name rc = blosc2_vlmeta_exists(self.schunk, name) @@ -3348,6 +3415,7 @@ cdef class vlmeta: raise RuntimeError return content[:content_len] + @_handle_locked def del_vlmeta(self, name): name = name.encode("utf-8") if isinstance(name, str) else name rc = blosc2_vlmeta_delete(self.schunk, name) @@ -3357,6 +3425,7 @@ cdef class vlmeta: def nvlmetalayers(self): return self.schunk.nvlmetalayers + @_handle_locked def get_names(self): cdef char** names = malloc(self.schunk.nvlmetalayers * sizeof (char *)) rc = blosc2_vlmeta_get_names(self.schunk, names) @@ -3365,6 +3434,7 @@ cdef class vlmeta: res = [names[i].decode("utf-8") for i in range(rc)] return res + @_handle_locked def to_dict(self): cdef char** names = malloc(self.schunk.nvlmetalayers * sizeof (char*)) rc = blosc2_vlmeta_get_names(self.schunk, names) @@ -3376,12 +3446,14 @@ cdef class vlmeta: return res +@_handle_locked def meta__contains__(self, name): cdef blosc2_schunk *schunk = self.c_schunk name = name.encode("utf-8") if isinstance(name, str) else name n = blosc2_meta_exists(schunk, name) return False if n < 0 else True +@_handle_locked def meta__getitem__(self, name): cdef blosc2_schunk *schunk = self.c_schunk name = name.encode("utf-8") if isinstance(name, str) else name @@ -3393,6 +3465,7 @@ def meta__getitem__(self, name): return res +@_handle_locked def meta__setitem__(self, name, content): cdef blosc2_schunk *schunk = self.c_schunk name = name.encode("utf-8") if isinstance(name, str) else name @@ -3405,6 +3478,7 @@ def meta__len__(self): cdef blosc2_schunk *schunk = self.c_schunk return schunk.nmetalayers +@_handle_locked def meta_keys(self): cdef blosc2_schunk *schunk = self.c_schunk keys = [] @@ -3814,24 +3888,39 @@ cdef class slice_flatter: self.done = 1 +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).""" + 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: + return False + return dparams.postfilter == NULL + + cdef class NDArray: cdef b2nd_array_t* array - cdef PyThread_type_lock read_lock - cdef c_bool owns_read_lock + cdef readonly object _handle_lock def __init__(self, array, base=None): self._dtype = None self.array = PyCapsule_GetPointer(array, "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 + # One handle lock per C super-chunk (see _handle_locked): expand_dims/squeeze views share + # their base's, and the SChunk wrapper over the same super-chunk (blosc2.NDArray._schunk, + # created before this runs) is switched to it. + schunk = getattr(self, "_schunk", None) + if base is not None: + self._handle_lock = (base)._handle_lock + elif schunk is not None: + self._handle_lock = (schunk)._handle_lock 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 + self._handle_lock = RLock() + if schunk is not None: + (schunk)._handle_lock = self._handle_lock @property def c_array(self): @@ -3899,22 +3988,29 @@ cdef class NDArray: buffershape_[i] = stop_[i] - start_[i] cdef Py_buffer view - cdef int rc + cdef int rc = 0 PyObject_GetBuffer(arr, &view, PyBUF_SIMPLE) - # 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) + # Waiting for the lock releases the GIL (RLock does that), so a holder that needs the + # GIL back, e.g. for a Python postfilter, can finish. + with self._handle_lock: + # Decompress without the GIL, 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. + if _can_release_gil(self.array.sc): + with nogil: + rc = b2nd_get_slice_cbuffer(self.array, start_, stop_, + view.buf, buffershape_, view.len) + else: + rc = b2nd_get_slice_cbuffer(self.array, start_, stop_, + view.buf, buffershape_, view.len) finally: - PyThread_release_lock(self.read_lock) - PyBuffer_Release(&view) + PyBuffer_Release(&view) _check_rc(rc, "Error while getting the buffer") return arr + @_handle_locked def get_1d_span_numpy(self, arr, int64_t nchunk, int32_t start, int32_t nitems): if self.ndim != 1: raise ValueError("get_1d_span_numpy is only supported for 1-D arrays") @@ -3997,6 +4093,7 @@ cdef class NDArray: return arr + @_handle_locked def get_sparse_numpy(self, arr, coords): cdef np.ndarray[np.int64_t, ndim=1, mode="c"] coords_ = np.ascontiguousarray(coords, dtype=np.int64) cdef Py_buffer view @@ -4014,6 +4111,7 @@ cdef class NDArray: _check_rc(rc, "Error while getting the sparse selection") return arr + @_handle_locked def get_oindex_numpy(self, arr, key): """ Orthogonal indexing. Key is a tuple of lists of integer indices. @@ -4046,6 +4144,7 @@ cdef class NDArray: free(key_) return arr + @_handle_locked def set_oindex_numpy(self, key, arr): """ Orthogonal indexing. Set elements of self with arr using key. @@ -4079,6 +4178,7 @@ cdef class NDArray: return arr + @_handle_locked def get_slice(self, key, mask, **kwargs): start, stop = key shape = tuple(sp - st for sp, st in zip(stop, start)) @@ -4120,6 +4220,7 @@ cdef class NDArray: return ndarray + @_handle_locked def set_slice(self, key, ndarray): ndim = self.ndim start, stop = key @@ -4138,6 +4239,7 @@ cdef class NDArray: return self + @_handle_locked def tobytes(self): buffersize = self.size * self.array.sc.typesize buffer = bytes(buffersize) @@ -4146,6 +4248,7 @@ cdef class NDArray: return buffer + @_handle_locked def to_cframe(self): cdef c_bool needs_free cdef uint8_t *cframe @@ -4160,6 +4263,7 @@ cdef class NDArray: return out + @_handle_locked def copy(self, dtype, **kwargs): chunks = kwargs.pop("chunks", self.chunks) blocks = kwargs.pop("blocks", self.blocks) @@ -4180,6 +4284,7 @@ cdef class NDArray: return ndarray + @_handle_locked def resize(self, new_shape): cdef int64_t new_shape_[B2ND_MAX_DIM] for i, s in enumerate(new_shape): @@ -4187,6 +4292,7 @@ cdef class NDArray: _check_rc(b2nd_resize(self.array, new_shape_, NULL), "Error while resizing the array") + @_handle_locked def refresh(self): cdef int rc = b2nd_refresh(self.array) _check_rc(rc, "Error while refreshing the array") @@ -4356,6 +4462,7 @@ cdef class NDArray: return udata + @_handle_locked def _set_pref_expr(self, expression, inputs, fp_accuracy, aux_reduc=None, jit=None, candidate_blocks=None): # Set prefilter for miniexpr @@ -4538,6 +4645,7 @@ cdef class NDArray: "status": _me_compile_status_name(rc), } + @_handle_locked def _set_pref_matmul(self, inputs, fp_accuracy): # Set prefilter for miniexpr cdef blosc2_cparams* cparams = self.array.sc.storage.cparams @@ -4558,6 +4666,7 @@ cdef class NDArray: if self.array.sc.cctx == NULL: raise RuntimeError("Could not create compression context") + @_handle_locked def _set_pref_udf(self, func, inputs_id): if self.array.sc.storage.cparams.nthreads > 1: raise AttributeError("compress `nthreads` must be 1 when assigning a prefilter") @@ -4580,6 +4689,7 @@ cdef class NDArray: if self.array.sc.cctx == NULL: raise RuntimeError("Could not create compression context") + @_handle_locked def _set_postf_udf(self, func, inputs_id): if self.array.sc.storage.dparams.nthreads > 1: raise AttributeError("decompress `nthreads` must be 1 when assigning a postfilter") @@ -4608,8 +4718,6 @@ 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): @@ -4859,6 +4967,7 @@ def ndarray_from_cframe(cframe, copy=False): return ndarray +@_handle_locked def array_get_slice_nchunks(array: NDArray, key): start, stop = key cdef int64_t[B2ND_MAX_DIM] start_, stop_ @@ -4875,6 +4984,7 @@ def array_get_slice_nchunks(array: NDArray, key): return res +@_handle_locked def schunk_get_slice_nchunks(schunk: SChunk, key): start, stop = key nitems = schunk.nbytes // schunk.typesize @@ -4904,8 +5014,13 @@ def concat(arr1: NDArray, arr2: NDArray, axis: int, **kwargs): raise RuntimeError("Error while creating the context for concatenation") cdef b2nd_array_t *array - _check_rc(b2nd_concatenate(ctx, arr1.array, arr2.array, axis, copy, &array), - "Error while concatenating the arrays") + cdef int rc = 0 + # Both handle locks, always in the same order, so that concat(a, b) and concat(b, a) in two + # threads cannot deadlock (they are the same lock for views of one array) + first, second = sorted((arr1._handle_lock, arr2._handle_lock), key=id) + with first, second: + rc = b2nd_concatenate(ctx, arr1.array, arr2.array, axis, copy, &array) + _check_rc(rc, "Error while concatenating the arrays") _check_rc(b2nd_free_ctx(ctx), "Error while freeing the context") if copy: @@ -4916,6 +5031,7 @@ def concat(arr1: NDArray, arr2: NDArray, axis: int, **kwargs): # Return the first array, which now contains the concatenated data return arr1 +@_handle_locked def expand_dims(arr1: NDArray, axis_mask: list[bool], final_dims: int) -> blosc2.NDArray: """ Add new dummy axis to NDArray object at specified dimension. @@ -4933,6 +5049,7 @@ def expand_dims(arr1: NDArray, axis_mask: list[bool], final_dims: int) -> blosc2 return blosc2.NDArray(_schunk=PyCapsule_New(view.sc, "blosc2_schunk*", NULL), _array=PyCapsule_New(view, "b2nd_array_t*", NULL), _base=new_base) +@_handle_locked def squeeze(arr1: NDArray, axis_mask: list[bool]) -> blosc2.NDArray: """ Remove axis from NDArray object at specified dimensions. diff --git a/src/blosc2/core.py b/src/blosc2/core.py index b2ebfb75e..21a4eafbc 100644 --- a/src/blosc2/core.py +++ b/src/blosc2/core.py @@ -1449,6 +1449,10 @@ def set_releasegil(gilstate: bool) -> bool: small performance penalty with releasing the GIL that will more harshly penalize small block sizes. + Reading slices of an :ref:`NDArray` releases the GIL while decompressing + regardless of this setting, unless a Python postfilter, codec or filter + could run during the read. + Examples -------- >>> oldReleaseState = blosc2.set_releasegil(True) diff --git a/tests/ndarray/test_concurrent_access.py b/tests/ndarray/test_concurrent_access.py new file mode 100644 index 000000000..e41418584 --- /dev/null +++ b/tests/ndarray/test_concurrent_access.py @@ -0,0 +1,161 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +# NDArray.get_slice_numpy() decompresses without the GIL. These tests check that other threads +# really run meanwhile, and that every other operation on the same array still waits for it. + +import threading +import time + +import numpy as np +import pytest + +import blosc2 + +SHAPE, CHUNKS, BLOCKS = (64, 256, 256), (8, 256, 256), (2, 64, 256) + + +def run_for(seconds, *targets): + """Run each target in a loop in its own thread for `seconds`; return the errors they raised.""" + stop = time.monotonic() + seconds + errors = [] + + def loop(target): + try: + while time.monotonic() < stop and not errors: + target() + except Exception as e: # report anything a racing thread hits + errors.append(e) + + threads = [threading.Thread(target=loop, args=(t,)) for t in targets] + for t in threads: + t.start() + for t in threads: + t.join(timeout=seconds + 60) + assert not any(t.is_alive() for t in threads), "threads did not finish (deadlock?)" + return errors + + +@pytest.fixture +def array_and_values(tmp_path, request): + values = np.arange(np.prod(SHAPE), dtype=np.float32).reshape(SHAPE) + kwargs = {"urlpath": tmp_path / "a.b2nd", "mode": "w"} if request.param == "disk" else {} + return blosc2.asarray(values, chunks=CHUNKS, blocks=BLOCKS, **kwargs), values + + +def check_region(array, values): + """Read [8:24, 10:200]; a concurrent writer only ever writes values or values + 1 into + [8:24, 0:100], one chunk (8 rows) at a time.""" + got, want = array[8:24, 10:200], values[8:24, 10:200] + for c in (0, 8): + g, w = got[c : c + 8], want[c : c + 8] + np.testing.assert_array_equal(g[:, 90:], w[:, 90:]) + assert np.array_equal(g[:, :90], w[:, :90]) or np.array_equal(g[:, :90], w[:, :90] + 1) + + +def test_handle_lock_is_shared(): + a = blosc2.asarray(np.arange(100).reshape(10, 10), chunks=(5, 5)) + view = blosc2.expand_dims(a, axis=0) + for other in (a.schunk, a.schunk.vlmeta, view, view.schunk): + assert other._handle_lock is a._handle_lock + + +def test_get_slice_releases_the_gil(monkeypatch): + # Python-registered codecs or filters (from other tests in this process) keep the GIL + monkeypatch.setattr(blosc2, "ucodecs_registry", {}) + monkeypatch.setattr(blosc2, "ufilters_registry", {}) + a = blosc2.linspace( + 0, + 1, + 32 * 1024 * 1024, + shape=(4096, 8192), + dtype=np.float32, + cparams={"codec": blosc2.Codec.ZSTD, "clevel": 1}, + dparams={"nthreads": 1}, + ) + 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() + _ = a[:] + t1 = time.perf_counter() + finally: + stop.set() + t.join() + inside = [s for s in stamps if t0 <= s <= t1] + gaps = np.diff([t0, *inside, t1]) + # Holding the GIL while decompressing leaves one gap as long as the whole read + assert gaps.max() < 0.5 * (t1 - t0), f"longest gap {gaps.max():.3f} s of a {t1 - t0:.3f} s read" + + +@pytest.mark.parametrize("array_and_values", ["memory", "disk"], indirect=True) +def test_read_while_writing_same_handle(array_and_values): + a, values = array_and_values + k = [0] + + def write(): + k[0] += 1 + a[8:24, 0:100] = values[8:24, 0:100] + (k[0] % 2) + + assert run_for(1.0, lambda: check_region(a, values), write) == [] + + +@pytest.mark.parametrize("array_and_values", ["memory"], indirect=True) +@pytest.mark.parametrize("other", ["schunk", "vlmeta", "dparams", "resize"]) +def test_read_while_other_access_same_handle(array_and_values, other): + a, values = array_and_values + k = [0] + + def access(): + k[0] += 1 + if other == "schunk": + n = 1 + k[0] % 6 + a.schunk.update_chunk(n, a.schunk.get_chunk(n)) + a.schunk.decompress_chunk(k[0] % 8) + elif other == "vlmeta": + a.schunk.vlmeta[f"k{k[0] % 5}"] = list(range(k[0] % 50)) + elif other == "dparams": + a.schunk.dparams = blosc2.DParams(nthreads=1 + k[0] % 2) + elif other == "resize": + # grow and shrink past the region the reader checks + a.resize((SHAPE[0] + 8 * (k[0] % 2), *SHAPE[1:])) + + assert run_for(1.0, lambda: check_region(a, values), access) == [] + + +def test_concurrent_reads_with_python_postfilter(): + values = np.arange(np.prod(SHAPE), dtype=np.float32).reshape(SHAPE) + a = blosc2.asarray(values, chunks=CHUNKS, blocks=BLOCKS, dparams={"nthreads": 1}) + + @a.schunk.postfilter(np.float32, np.float32) + def add_one(inp, out, offset): + out[:] = inp + 1 + + def read(): + np.testing.assert_array_equal(a[8:24, 10:200], values[8:24, 10:200] + 1) + + assert run_for(1.0, read, read) == [] + + +def test_concat_both_orders_does_not_deadlock(): + a = blosc2.asarray(np.zeros((16, 64), dtype=np.int32), chunks=(8, 64)) + b = blosc2.asarray(np.ones((16, 64), dtype=np.int32), chunks=(8, 64)) + errors = run_for( + 1.0, + lambda: blosc2.concat([a, b], axis=0), + lambda: blosc2.concat([b, a], axis=0), + lambda: a[:], + lambda: b[:], + ) + assert errors == [] From ddd70de032f3095352cce05b51ecedba9fe8b8c6 Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Wed, 28 Oct 2026 22:26:22 +0100 Subject: [PATCH 2/5] Make GIL-free slice reads safe around fork, exit and lazy expressions Hold the operands' handle locks while the miniexpr prefilter of a lazy expression reads their super-chunks from C. Count the reads that run without the GIL, so that exit and the registration of Python codecs and filters wait for them, and free handle locks that other threads held at a fork in the child. Lock the metalayer counters too. Skip the new tests on wasm32, shorten them, retry the GIL check on busy machines, and add tests for lazy expressions and fork. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SwMZj7Wi7Sc8HtQpELyNXB --- src/blosc2/blosc2_ext.pyx | 139 ++++++++++++++--- src/blosc2/lazyexpr.py | 10 +- tests/ndarray/test_concurrent_access.py | 192 ++++++++++++++++-------- 3 files changed, 257 insertions(+), 84 deletions(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 47740ecbe..1ccf65d5b 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -17,6 +17,7 @@ import sys import functools import time import warnings +import weakref from threading import RLock import _ctypes @@ -876,7 +877,13 @@ blosc2_init() @atexit.register def destroy(): - blosc2_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): + blosc2_destroy() def register_wasm_jit_helpers(uintptr_t instantiate_ptr, uintptr_t free_ptr): @@ -1727,6 +1734,81 @@ def _handle_locked(method): return locked +# Every handle lock, so that a forked child can free those that other threads held at the fork +_all_handle_locks = weakref.WeakSet() + + +def _new_handle_lock(): + lock = RLock() + _all_handle_locks.add(lock) + return lock + + +cdef class HandleLocks: + """Reusable context manager that holds the handle locks of `arrays` (objects without one are + skipped), always in the same order, so that threads locking overlapping sets cannot deadlock. + For C code that reads other arrays' super-chunks outside their methods: concat() and the + miniexpr prefilter of lazy expressions. + """ + cdef list locks + + def __init__(self, arrays): + locks = {id(lock): lock for lock in (getattr(a, "_handle_lock", None) for a in arrays) + if lock is not None} + self.locks = [locks[key] for key in sorted(locks)] + + def __enter__(self): + cdef Py_ssize_t n = 0 + try: + for lock in self.locks: + lock.acquire() + n += 1 + except BaseException: + for lock in reversed(self.locks[:n]): + lock.release() + raise + return self + + def __exit__(self, *exc_info): + for lock in reversed(self.locks): + 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 + + +def _wait_for_nogil_reads(timeout=None): + """Wait until no slice read 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. + """ + deadline = None if timeout is None else time.monotonic() + timeout + while _nogil_reads: + if deadline is not None and time.monotonic() > deadline: + return False + time.sleep(0.0005) # releases the GIL, so the readers 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 + for lock in list(_all_handle_locks): + if not lock._is_owned(): + lock._at_fork_reinit() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_reset_after_fork) + + cdef class SChunk: cdef blosc2_schunk *schunk cdef c_bool _is_view @@ -1734,7 +1816,7 @@ cdef class SChunk: def __init__(self, _schunk=None, chunksize=2 ** 24, data=None, **kwargs): # Replaced by the NDArray's lock when this is an NDArray's SChunk (see NDArray.__init__) - self._handle_lock = RLock() + self._handle_lock = _new_handle_lock() # hold on to a bytestring of urlpath for the lifetime of the instance # because its value is referenced via a C-pointer urlpath = kwargs.get("urlpath", None) @@ -3422,6 +3504,7 @@ cdef class vlmeta: if rc < 0: raise RuntimeError("Could not delete the vlmeta") + @_handle_locked def nvlmetalayers(self): return self.schunk.nvlmetalayers @@ -3474,6 +3557,7 @@ def meta__setitem__(self, name, content): raise ValueError("The length of the content in a metalayer cannot change.") blosc2_meta_update(schunk, name, content, len(content)) +@_handle_locked def meta__len__(self): cdef blosc2_schunk *schunk = self.c_schunk return schunk.nmetalayers @@ -3705,12 +3789,21 @@ def register_codec(codec_name, id, encoder=None, decoder=None, version=1): else: codec.decoder = general_decoder - rc = blosc2_register_codec(&codec) - if rc < 0: - raise RuntimeError("Error while registering codec") + # 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 + _nogil_reads_blocked += 1 + try: + _wait_for_nogil_reads() + rc = blosc2_register_codec(&codec) + if rc < 0: + raise RuntimeError("Error while registering codec") - if encoder and decoder: - blosc2.ucodecs_registry[id] = (codec_name, encoder, decoder) + if encoder and decoder: + blosc2.ucodecs_registry[id] = (codec_name, encoder, decoder) + finally: + _nogil_reads_blocked -= 1 cdef int general_forward(const uint8_t* input_buffer, uint8_t* output_buffer, int32_t size, @@ -3772,11 +3865,18 @@ def register_filter(id, forward, backward, filter_name): filter.name = malloc(strlen(filter_name_) + 1) strcpy(filter.name, filter_name_) - rc = blosc2_register_filter(&filter) - if rc < 0: - raise RuntimeError("Error while registering filter") - if forward and backward: - blosc2.ufilters_registry[id] = (forward, backward) + # As in register_codec() + global _nogil_reads_blocked + _nogil_reads_blocked += 1 + try: + _wait_for_nogil_reads() + rc = blosc2_register_filter(&filter) + if rc < 0: + raise RuntimeError("Error while registering filter") + if forward and backward: + blosc2.ufilters_registry[id] = (forward, backward) + finally: + _nogil_reads_blocked -= 1 cdef _check_rc(rc, message): if rc < 0: @@ -3892,8 +3992,9 @@ 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).""" - if 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), 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: return False cdef blosc2_dparams dparams if sc.dctx == NULL or blosc2_ctx_get_dparams(sc.dctx, &dparams) < 0: @@ -3918,7 +4019,7 @@ cdef class NDArray: elif schunk is not None: self._handle_lock = (schunk)._handle_lock else: - self._handle_lock = RLock() + self._handle_lock = _new_handle_lock() if schunk is not None: (schunk)._handle_lock = self._handle_lock @@ -3978,6 +4079,7 @@ cdef class NDArray: return dtype def get_slice_numpy(self, arr, key): + global _nogil_reads start, stop = key cdef int64_t[B2ND_MAX_DIM] start_, stop_ @@ -3998,9 +4100,11 @@ cdef class NDArray: # of the same file) run in parallel; everything else that uses this super-chunk # holds the lock meanwhile. if _can_release_gil(self.array.sc): + _nogil_reads += 1 with nogil: rc = b2nd_get_slice_cbuffer(self.array, start_, stop_, view.buf, buffershape_, view.len) + _nogil_reads -= 1 else: rc = b2nd_get_slice_cbuffer(self.array, start_, stop_, view.buf, buffershape_, view.len) @@ -5015,10 +5119,7 @@ def concat(arr1: NDArray, arr2: NDArray, axis: int, **kwargs): cdef b2nd_array_t *array cdef int rc = 0 - # Both handle locks, always in the same order, so that concat(a, b) and concat(b, a) in two - # threads cannot deadlock (they are the same lock for views of one array) - first, second = sorted((arr1._handle_lock, arr2._handle_lock), key=id) - with first, second: + 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(b2nd_free_ctx(ctx), "Error while freeing the context") diff --git a/src/blosc2/lazyexpr.py b/src/blosc2/lazyexpr.py index 6cf988169..28af9bbb8 100644 --- a/src/blosc2/lazyexpr.py +++ b/src/blosc2/lazyexpr.py @@ -1983,10 +1983,13 @@ def _miniexpr_eligible_operand(op): nct = res_eval.schunk.nchunks if bpc > 0 and _cb_anchor.shape[0] >= nct * bpc: chunk_has_candidate = _cb_anchor[: nct * bpc].reshape(nct, bpc).any(axis=1) + # The prefilter reads the operands' super-chunks from C, outside their methods + operand_locks = blosc2.blosc2_ext.HandleLocks(_pref_ops.values()) for nchunk in range(res_eval.schunk.nchunks): if chunk_has_candidate is not None and not chunk_has_candidate[nchunk]: continue - res_eval.schunk.update_data(nchunk, data, copy=False) + with operand_locks: + res_eval.schunk.update_data(nchunk, data, copy=False) # _cb_anchor (if any) stays referenced until fast_eval returns, which # is after the finally below removes the prefilter — so the buffer # safely outlives every prefilter dereference. @@ -2924,8 +2927,11 @@ def reduce_slices( # noqa: C901 data = np.empty(res_eval.schunk.chunksize, dtype=np.uint8) chunk_data = np.empty(res_eval.schunk.chunksize + blosc2.MAX_OVERHEAD, dtype=np.uint8) # Exercise prefilter for each chunk + # The prefilter reads the operands' super-chunks from C, outside their methods + operand_locks = blosc2.blosc2_ext.HandleLocks(operands.values()) for nchunk in range(res_eval.schunk.nchunks): - res_eval.schunk._prefilter_data(nchunk, data, chunk_data) + with operand_locks: + res_eval.schunk._prefilter_data(nchunk, data, chunk_data) except Exception as e: use_miniexpr = False if callable(expression) and _is_dsl_kernel_expression(expression): diff --git a/tests/ndarray/test_concurrent_access.py b/tests/ndarray/test_concurrent_access.py index e41418584..186870e50 100644 --- a/tests/ndarray/test_concurrent_access.py +++ b/tests/ndarray/test_concurrent_access.py @@ -8,6 +8,9 @@ # NDArray.get_slice_numpy() decompresses without the GIL. These tests check that other threads # really run meanwhile, and that every other operation on the same array still waits for it. +import itertools +import os +import signal import threading import time @@ -16,7 +19,16 @@ import blosc2 +pytestmark = pytest.mark.skipif(blosc2.IS_WASM, reason="Pyodide cannot start threads") + SHAPE, CHUNKS, BLOCKS = (64, 256, 256), (8, 256, 256), (2, 64, 256) +VALUES = np.arange(np.prod(SHAPE), dtype=np.float32).reshape(SHAPE) +# Per stress test; without the locking, the races show up within the first few reads +SECONDS = 0.5 + + +def make_array(**kwargs): + return blosc2.asarray(VALUES, chunks=CHUNKS, blocks=BLOCKS, **kwargs) def run_for(seconds, *targets): @@ -40,21 +52,20 @@ def loop(target): return errors -@pytest.fixture -def array_and_values(tmp_path, request): - values = np.arange(np.prod(SHAPE), dtype=np.float32).reshape(SHAPE) - kwargs = {"urlpath": tmp_path / "a.b2nd", "mode": "w"} if request.param == "disk" else {} - return blosc2.asarray(values, chunks=CHUNKS, blocks=BLOCKS, **kwargs), values +def check_region(array): + """Read [8:24, 10:200]; a concurrent write_region() only ever writes VALUES or VALUES + 1 into + [8:24, 0:100], one chunk (8 rows) at a time.""" + got, want = array[8:24, 10:200], VALUES[8:24, 10:200] + for chunk in (slice(0, 8), slice(8, 16)): + got_rows, want_rows = got[chunk], want[chunk] + np.testing.assert_array_equal(got_rows[:, 90:], want_rows[:, 90:]) + assert np.array_equal(got_rows[:, :90], want_rows[:, :90]) or np.array_equal( + got_rows[:, :90], want_rows[:, :90] + 1 + ) -def check_region(array, values): - """Read [8:24, 10:200]; a concurrent writer only ever writes values or values + 1 into - [8:24, 0:100], one chunk (8 rows) at a time.""" - got, want = array[8:24, 10:200], values[8:24, 10:200] - for c in (0, 8): - g, w = got[c : c + 8], want[c : c + 8] - np.testing.assert_array_equal(g[:, 90:], w[:, 90:]) - assert np.array_equal(g[:, :90], w[:, :90]) or np.array_equal(g[:, :90], w[:, :90] + 1) +def write_region(array, k): + array[8:24, 0:100] = VALUES[8:24, 0:100] + (k % 2) def test_handle_lock_is_shared(): @@ -77,85 +88,140 @@ def test_get_slice_releases_the_gil(monkeypatch): cparams={"codec": blosc2.Codec.ZSTD, "clevel": 1}, dparams={"nthreads": 1}, ) - stamps, stop = [], threading.Event() - def ticker(): - while not stop.is_set(): - stamps.append(time.perf_counter()) - time.sleep(0) + def longest_gap(): + """Longest time a ticker thread did not run during one read, as a fraction of the read.""" + stamps, stop = [], threading.Event() - t = threading.Thread(target=ticker) - t.start() - try: - t0 = time.perf_counter() - _ = a[:] - t1 = time.perf_counter() - finally: - stop.set() - t.join() - inside = [s for s in stamps if t0 <= s <= t1] - gaps = np.diff([t0, *inside, t1]) - # Holding the GIL while decompressing leaves one gap as long as the whole read - assert gaps.max() < 0.5 * (t1 - t0), f"longest gap {gaps.max():.3f} s of a {t1 - t0:.3f} s read" + 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() + _ = a[:] + 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) + + # Holding the GIL while decompressing leaves one gap as long as nearly the whole read, every + # time; a busy machine can delay the ticker now and then, hence a few attempts + gaps = [] + while len(gaps) < 5 and (not gaps or gaps[-1] >= 0.5): + gaps.append(longest_gap()) + assert gaps[-1] < 0.5, f"longest gaps, as fractions of the read: {gaps}" + + +@pytest.mark.parametrize("on_disk", [False, True], ids=["memory", "disk"]) +def test_read_while_writing_same_handle(tmp_path, on_disk): + a = make_array(urlpath=tmp_path / "a.b2nd", mode="w") if on_disk else make_array() + k = itertools.count() + assert run_for(SECONDS, lambda: check_region(a), lambda: write_region(a, next(k))) == [] -@pytest.mark.parametrize("array_and_values", ["memory", "disk"], indirect=True) -def test_read_while_writing_same_handle(array_and_values): - a, values = array_and_values - k = [0] +def update_chunks(a, k): + n = 1 + k % 6 + a.schunk.update_chunk(n, a.schunk.get_chunk(n)) + a.schunk.decompress_chunk(k % 8) - def write(): - k[0] += 1 - a[8:24, 0:100] = values[8:24, 0:100] + (k[0] % 2) - assert run_for(1.0, lambda: check_region(a, values), write) == [] +def write_vlmeta(a, k): + a.schunk.vlmeta[f"k{k % 5}"] = list(range(k % 50)) -@pytest.mark.parametrize("array_and_values", ["memory"], indirect=True) -@pytest.mark.parametrize("other", ["schunk", "vlmeta", "dparams", "resize"]) -def test_read_while_other_access_same_handle(array_and_values, other): - a, values = array_and_values - k = [0] +def swap_dparams(a, k): + a.schunk.dparams = blosc2.DParams(nthreads=1 + k % 2) - def access(): - k[0] += 1 - if other == "schunk": - n = 1 + k[0] % 6 - a.schunk.update_chunk(n, a.schunk.get_chunk(n)) - a.schunk.decompress_chunk(k[0] % 8) - elif other == "vlmeta": - a.schunk.vlmeta[f"k{k[0] % 5}"] = list(range(k[0] % 50)) - elif other == "dparams": - a.schunk.dparams = blosc2.DParams(nthreads=1 + k[0] % 2) - elif other == "resize": - # grow and shrink past the region the reader checks - a.resize((SHAPE[0] + 8 * (k[0] % 2), *SHAPE[1:])) - assert run_for(1.0, lambda: check_region(a, values), access) == [] +def resize(a, k): + # grow and shrink past the region check_region() reads + a.resize((SHAPE[0] + 8 * (k % 2), *SHAPE[1:])) + + +@pytest.mark.parametrize( + "access", [update_chunks, write_vlmeta, swap_dparams, resize], ids=lambda f: f.__name__ +) +def test_read_while_other_access_same_handle(access): + a = make_array() + k = itertools.count() + assert run_for(SECONDS, lambda: check_region(a), lambda: access(a, next(k))) == [] def test_concurrent_reads_with_python_postfilter(): - values = np.arange(np.prod(SHAPE), dtype=np.float32).reshape(SHAPE) - a = blosc2.asarray(values, chunks=CHUNKS, blocks=BLOCKS, dparams={"nthreads": 1}) + a = make_array(dparams={"nthreads": 1}) @a.schunk.postfilter(np.float32, np.float32) def add_one(inp, out, offset): out[:] = inp + 1 def read(): - np.testing.assert_array_equal(a[8:24, 10:200], values[8:24, 10:200] + 1) + np.testing.assert_array_equal(a[8:24, 10:200], VALUES[8:24, 10:200] + 1) + + assert run_for(SECONDS, read, read) == [] + - assert run_for(1.0, read, read) == [] +def test_lazy_expression_while_reading_operands(tmp_path): + # The miniexpr prefilter reads on-disk operands from C, so it must hold their locks too + a = make_array(urlpath=tmp_path / "a.b2nd", mode="w") + b = make_array(urlpath=tmp_path / "b.b2nd", mode="w") + + def compute(): + np.testing.assert_array_equal((a + b).compute()[8:24], 2 * VALUES[8:24]) + + assert run_for(SECONDS, compute, compute, lambda: check_region(a), lambda: check_region(b)) == [] def test_concat_both_orders_does_not_deadlock(): a = blosc2.asarray(np.zeros((16, 64), dtype=np.int32), chunks=(8, 64)) b = blosc2.asarray(np.ones((16, 64), dtype=np.int32), chunks=(8, 64)) errors = run_for( - 1.0, + SECONDS, lambda: blosc2.concat([a, b], axis=0), lambda: blosc2.concat([b, a], axis=0), lambda: a[:], lambda: b[:], ) assert errors == [] + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") +@pytest.mark.filterwarnings("ignore::DeprecationWarning") # fork() with threads running +def test_fork_while_another_thread_holds_the_lock(): + # As when forking while another thread reads without the GIL: the child must not inherit + # a lock that no thread of its own will ever release + a = make_array(dparams={"nthreads": 1}) + held, release = threading.Event(), threading.Event() + + def hold(): + with a._handle_lock: + held.set() + release.wait() + + t = threading.Thread(target=hold) + t.start() + try: + assert held.wait(timeout=10) + pid = os.fork() + if pid == 0: + ok = False + try: + ok = np.array_equal(a[8:24, 10:200], VALUES[8:24, 10:200]) + finally: + os._exit(0 if ok else 1) + deadline = time.monotonic() + 30 + while (status := os.waitpid(pid, os.WNOHANG))[0] == 0: + if time.monotonic() > deadline: + os.kill(pid, signal.SIGKILL) + os.waitpid(pid, 0) + pytest.fail("the child hung on a lock held by a thread of the parent") + time.sleep(0.01) + assert os.waitstatus_to_exitcode(status[1]) == 0 + finally: + release.set() + t.join() From 0a77d14f49829797a1f13eec4996063e913149cb Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Tue, 3 Nov 2026 10:32:24 +0100 Subject: [PATCH 3/5] Add SChunk.get_chunk_range() Offset and size of a chunk in the file of a super-chunk stored as a contiguous frame on disk, from the frame header and the chunk index, without reading the chunk (C-Blosc2's blosc2_schunk_get_chunk_range()). Callers that know which chunks they will need can read them, or ask the operating system to read them ahead, themselves. The size is 0 for chunks stored as special values; a RuntimeError for chunks that do not exist and super-chunks without a frame. Co-Authored-By: Claude Opus 5.5 --- src/blosc2/blosc2_ext.pyx | 27 ++++++++++++++++++++++++ tests/test_chunk_range.py | 44 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+) create mode 100644 tests/test_chunk_range.py diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 1ccf65d5b..7a8548a0f 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -505,6 +505,8 @@ cdef extern from "blosc2.h": int blosc2_schunk_get_chunk(blosc2_schunk *schunk, int64_t nchunk, uint8_t ** chunk, c_bool *needs_free) nogil + int blosc2_schunk_get_chunk_range(blosc2_schunk *schunk, int64_t nchunk, int64_t *offset, + int64_t *nbytes) int blosc2_schunk_get_lazychunk(blosc2_schunk *schunk, int64_t nchunk, uint8_t ** chunk, c_bool *needs_free) nogil int blosc2_schunk_get_vlblock(blosc2_schunk *schunk, int64_t nchunk, int32_t nblock, @@ -2203,6 +2205,31 @@ cdef class SChunk: free(chunk) return ret_chunk + @_handle_locked + def get_chunk_range(self, nchunk): + """Where chunk `nchunk` is in the file of a super-chunk stored as a contiguous frame on disk. + + Returns + ------- + out: tuple of two ints + The position of the chunk's first byte in the file and its size there: the distance to the next + chunk, which is the compressed size for chunks written in order and otherwise at most the largest + possible chunk size. The size is 0 for chunks stored as special values (zeros, NaNs, + uninitialized), which have no bytes in the file. No chunk is read to find this: it comes from + the frame header and the chunk index, which opening the frame has read already. + + Raises + ------ + RuntimeError + If the super-chunk is not a contiguous frame on disk, or `nchunk` is out of range. + """ + cdef int64_t offset + cdef int64_t nbytes + rc = blosc2_schunk_get_chunk_range(self.schunk, nchunk, &offset, &nbytes) + if rc < 0: + raise RuntimeError(f"Could not get the file range of chunk {nchunk}") + return offset, nbytes + @_handle_locked def get_lazychunk(self, nchunk): cdef uint8_t *chunk diff --git a/tests/test_chunk_range.py b/tests/test_chunk_range.py new file mode 100644 index 000000000..5322c50cc --- /dev/null +++ b/tests/test_chunk_range.py @@ -0,0 +1,44 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +import numpy as np +import pytest + +import blosc2 + + +@pytest.mark.parametrize("nthreads", [1, 4]) +def test_chunk_range(tmp_path, nthreads): + path = str(tmp_path / "chunk_range.b2nd") + a = blosc2.zeros( + (40, 50, 61), + dtype=np.int32, + chunks=(10, 20, 16), + blocks=(5, 8, 8), + urlpath=path, + cparams={"nthreads": nthreads}, + ) + # data in part of the array only, so that the other chunks stay special values (zeros) in the index + values = np.arange(40 * 30 * 61, dtype=np.int64).reshape(40, 30, 61) * 2654435761 % 1000003 + a[:, :30] = values.astype(np.int32) + del a + a = blosc2.open(path, mode="r") + with open(path, "rb") as f: + raw = f.read() + nspecial = 0 + for nchunk in range(a.schunk.nchunks): + offset, nbytes = a.schunk.get_chunk_range(nchunk) + if nbytes == 0: + nspecial += 1 + continue + # written in order: exactly the chunk's bytes + assert raw[offset : offset + nbytes] == a.schunk.get_chunk(nchunk) + assert nspecial > 0 + with pytest.raises(RuntimeError): + a.schunk.get_chunk_range(a.schunk.nchunks) + with pytest.raises(RuntimeError): + blosc2.SChunk(chunksize=100).get_chunk_range(0) From 65943a1aeea22111950e7c54b6d98ed92c8b786c Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Tue, 3 Nov 2026 10:33:52 +0100 Subject: [PATCH 4/5] Add blosc2.set_prefetch(): read-ahead for on-disk arrays blosc2.set_prefetch(nthreads, max_ahead=0, piece_size=0) turns on C-Blosc2's read-ahead (blosc2_set_prefetch(); BLOSC_PREFETCH= in the environment does the same at import): slices of arrays and super-chunks on disk (a[...], NDArray.slice(), orthogonal selections, SChunk slices) read the chunks they need, and that are not in the page cache, with a few background threads in large pieces, and decompress each chunk as soon as it is in. Of chunks that a[...] or slice() covers only in part, just the blocks they need are read. Off by default; a no-op on Windows. The docstring says which settings to use where (network file systems, local disks, many reader processes); blosc2.open() points to it for files on network storage, and a new optimization tip (with its benchmark script) shows the effect on cold reads: 1.7x for 40 cold patch reads on a local NVMe disk, 3-5x for 3D patches on NFS. Co-Authored-By: Claude Opus 5.5 --- bench/optim_tips/tip_16_prefetch.py | 90 ++++++++++++ doc/guides/optim_tips/tip_16_prefetch.png | Bin 0 -> 33438 bytes doc/guides/optimization_tips.md | 22 +++ doc/reference/low_level.rst | 1 + src/blosc2/__init__.py | 2 + src/blosc2/blosc2_ext.pyx | 10 ++ src/blosc2/core.py | 63 +++++++++ src/blosc2/schunk.py | 4 + tests/test_prefetch.py | 164 ++++++++++++++++++++++ 9 files changed, 356 insertions(+) create mode 100644 bench/optim_tips/tip_16_prefetch.py create mode 100644 doc/guides/optim_tips/tip_16_prefetch.png create mode 100644 tests/test_prefetch.py diff --git a/bench/optim_tips/tip_16_prefetch.py b/bench/optim_tips/tip_16_prefetch.py new file mode 100644 index 000000000..4ce021fff --- /dev/null +++ b/bench/optim_tips/tip_16_prefetch.py @@ -0,0 +1,90 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +# Tip 16: blosc2.set_prefetch() lets a few background threads read the chunks a +# slice needs, in large parallel requests and ahead of decompression. Without +# it, a slice of an on-disk array reads its blocks one after the other, and +# every read that misses the page cache waits for the storage. +# +# The benefit depends on the storage latency: on a local NVMe disk cold patches +# get ~1.5-3x faster, on NFS ~3-5x. Every read here is cold: the file is +# dropped from the page cache before each one (Linux; elsewhere the reads are +# warm, and both variants measure the same). + +import os +from pathlib import Path + +import numpy as np +from common import fmt_bytes, measure, save_plot + +import blosc2 + +SHAPE, CHUNKS, BLOCKS, PATCH = (4, 256, 256, 256), (1, 64, 128, 128), (1, 16, 32, 128), (128, 128, 128) +URLPATH = str(Path(__file__).parent / "tip_16.b2nd") +N_READS = 40 + + +def _build(): + rng = np.random.default_rng(0) + data = blosc2.zeros(SHAPE, dtype=np.float32, chunks=CHUNKS, blocks=BLOCKS, urlpath=URLPATH, mode="w") + for c in range(SHAPE[0]): + data[c] = rng.normal(size=SHAPE[1:]).astype(np.float32).round(2) + + +_build() + +_starts = np.random.default_rng(1).integers( + 0, [s - p + 1 for s, p in zip(SHAPE[1:], PATCH, strict=True)], size=(N_READS, 3) +) + + +def _drop_from_page_cache(): + if hasattr(os, "posix_fadvise"): + fd = os.open(URLPATH, os.O_RDONLY) + os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED) + os.close(fd) + + +def _cold_patch_reads(): + total = 0.0 + for start in _starts: + _drop_from_page_cache() + arr = blosc2.open(URLPATH, mmap_mode="r") + total += arr[ + (slice(None),) + tuple(slice(s, s + p) for s, p in zip(start, PATCH, strict=True)) + ].sum() + del arr # unmapped, so that the next drop reaches every page + return total + + +def naive(): + return _cold_patch_reads() + + +def tip(): + blosc2.set_prefetch(4) # per process; BLOSC_PREFETCH=4 in the environment does the same + return _cold_patch_reads() + + +if __name__ == "__main__": + naive_t, naive_m = measure(__file__, "naive") + tip_t, tip_m = measure(__file__, "tip") + + print(f"naive without read-ahead : {naive_t:.3f}s peak {fmt_bytes(naive_m)}") + print(f"tip set_prefetch(4) : {tip_t:.3f}s peak {fmt_bytes(tip_m)}") + print(f"speedup: {naive_t / tip_t:.1f}x") + + save_plot( + "tip_16_prefetch.png", + f"set_prefetch(4) vs no read-ahead — {N_READS} cold patch reads", + "no read-ahead", + "set_prefetch(4)", + naive_t, + tip_t, + naive_m, + tip_m, + ) diff --git a/doc/guides/optim_tips/tip_16_prefetch.png b/doc/guides/optim_tips/tip_16_prefetch.png new file mode 100644 index 0000000000000000000000000000000000000000..b769fa28a0643c19b9af14722f9328797760d140 GIT binary patch literal 33438 zcmeFZ2{e}L-!`mix0~mbqFqRasH6;`-9Wd5%$ZdZGP?;G+PhLo>J}nHGP?~S^SDzf ziVQb#8Pwzdz|_|@Ao|KyS`_A>s#;lto6O?wbowQ+^*|7&+9yYzvFit$M3wO zrlPoNIoEOy4vtlnQzz6pI2J$W;8^ha_uuf7tl&qH_($65q^^^O?IkA{QwKASGp0^< zS8ScGSe*aE+04Pw!q!GybpL+Q10sK1c5y|9}Gik}6Zo~XXf47J^>h`34hkc_+(+xhprUi1tjlvTZ z??W6T!dpXRM+ZX0MQ@pISx6p$gG0H3(X`;#w;X{xuP^!atp@M^s_#A*j(z@njnE%| z98D{k&CfiZ$hl&($j6US#~Cq{4NHFejr^uESS-!nW7)E0%YXk}O;=ZUuY%wpGe>p7 zf(4ch$xTv>Oe#xT;>striB=cmz>1RbW-Ic+qes1WD?k4H{QXdC0pq#p@iUZs}!Nn#q`7X)DaD9l?!5 z518GRcItil@uOOnMV-P#i=+SeP^-S`y2h3Z{<3l2-W)|gM=BZ3L)Y9QSxNGCJGpau zlDy_-$$gTv=waN;y?wVY6XEEIrP0=J;^d^q#Jp;Xs?U?KYUu5LV%yNqpa(`oq&|E0 z%+Yi3mT5w6uI|Z`C$qL#3^JzQi&rPa#GLRyGV=CZmr;aXM z!J%x^{XRrUP_X95_fG1!&wrX-yts##_qdLcYi#Cm`UbqM#)?2;3FiUbtSe0!wHdce zd#^9$d0@JQ>%hgg=UAR2?*z=Y+=}oRZ)#r7S?YSRt0FLT|AkwrMdQtxK_9BC8&@4} zc{(=cMD6|fXJyb{jq~T9c}@==UbS6PEiW%GpWUEO9r^ms?e=yKjt8X+IRa!HzTW1u z`0=e{`HB?*1qB5yt*u>OzI^%mw*O7RaIr9j{6)3ic(lJAC6M`}egkUrvyw z_Q&kM@TRv?tVGk(v$$;VkZoIX!BGC`U;~Fc$BrN0v~3&xY_vk(^l(Xzu*+EQS={aN z)L0)aW^TG&VsToA>HznoeSw z?KRBTT(@o=*FKY@vD6*%o(3B6Dq75JYq}FXDv)|VSyg`KMFsQP>|_slpr?QSEwtCA zY}MgC*rIcI9Roo-q@7N!Ir7aLGkS6Q+Qe|Xv8vqU#jNfcDaZTi{X@oyPI8guJnqVd z`A%nbbss;Aj;1oSboxfC<!_%x+_c}cd2@fs+)SF=_(0O9$hj-QXO_n)OZGz**uH?Cj*%z3c6uSRi)mY$woEYtA!>FTBk zuUS`axof&)H1!BUnL&|q~c@rhb=;tem(^ zmCuyuTcha`sm~b8m1W8$rOf`th&t{wHc}~O)%!8ZvM$cgx+?q51_8^mTdM-!;0>;B zb*ei<*7b=KH?Rbw54T-Ce)HzdKpP$Xc3TcL{@*z2pCM{)-Rv9tvV^RbmP2VpX!dWtk*4aZQ$0+;k=6Xlb;VrG#XqG)@gMcE}kAzn&=oV znX?gg@BO$l+Oel~T2;319@feG=eia=BiN=oc~jW!SGFjy@7vSoW+zGlJyoXJEc1(IW`RsmdHMD&`_JDf zvtGDpk)NPO+?LJ0uX9H#gy}z}t0i``COa^kgII@O?3fXY=LqdVH+}KukUVhW-_`b zwsB}KLE6i@)cMAbtb&CL7v3XNpnxd#Fg)D))$7+=5#q&%3P)8s5uh`emxfyM$S1iZ zEHC~XwDl6twr#h$ci-h&70$xB0!v1V@EyqODGKhiW^_g=7(+q*p? zA}?mNBzAV(&eE{J&#+avj6K+a4|$>o>^x2;)Tyub^&TOi11rn+VW0WE ze0hwSpXb;U#Q)IqA0GV#UNTg$A(<-{Q2|uuruD?_V+)Wk&$sP zT*ief^4iFq{dS*^7R;EOxvV?4TQ%a!6=wE_x9{H{%)Ziee&x19?;9pXnL&(%v(esryg6D^d@s@GUIzu4bh6>ha%-m~SDvhqQ>pWlw_JB$@gvU!m&pIwHOR}PF*^e&{VsE!d*9?16_2EMn{|V0-cB4@F=xa{nPyNM6iGJ-~bFJid z_FrBt>x_@|JV55ZaI{LsnOTSlB9C~@!^1C8Bc4m~{`URkr5St8FnSx)BKMwq@Zf<- zj^W8=$@y-P)m`o{c1JcPjW4y#sIKnHP3fPHv08JpGk%Rm?jL{w{(kuQ@n!7KZLG$$ z!@E_&E@OxA^6|wOWJvJyzYco3P15dZ`lSzCyu!kjl1{x3zV}A^H-~ptE1UfL9rWhj zR`OIf#gpAafnr#KqIqA}mz__%`s!`+xlFP`FwNyQlQR?TwzxrOppKDLPi{(uvyc}S zxdRI++M4xEVE1n6mP3F1QTi#Zq#`d`e50bHHQiQHOeZ_)2$%;Ch>KfS(gRAl4%>v`E)|B)H80gXJ4puCoNaCO2w(bM zuQ%x+3dS1HkXEzFuaHHxy0V2FM5i02^Y@B~B)9Z5?psMlNP_;l?c1-!#KmdLL?9~# z>U)WnUo|5ES63vKx1^$&Q%Xuo;o`*@GAO)|lon?^l{f-XKi>k4v&y%=_e}0YNRf0K zyPRho>Ds-Oto#da?xpK{?jJ14AN%4f#)X)Cb@anw>&mL2(&l_B-P!}V=_XZQF*3}x zE%nr&e_pT2Fsoc1AbB6~af{!!!-vQg%_*5lX_X9 z@r28%^o&E+%`v&Tl7L^#Zhq-HB(bipu2>XC!s6nY+MOw#;!80q%CAUo&?=%G^qRd! ziFLHivScLbF_mf`32IdAU9oy~_J~G6+G?%cbJz-&C z;p{k_Xh%pN;1vUL|E5s&5O-B>t^Y~1unx4 z!fA!is={Shv(v+ABhF?$$4AD-lAnhK2kYdG1fDDLELy*Q{nx&}M1sKSwTG-4Q?=aO z@(T+KxAT}I2)cc~zLbyR%PE$s?m?X)U@L)FOR88%J;1|An~G$aG(7wATSpmxU{$IkwI~PpUjb2We{0hq*kbjR;1gu7o%JW zMkcQa%u(6+Xd%@l@r!82xmx=<48t{QRWjKC7j#{jj^a zAt2kL?oEdS%PRbUnbOS-_qW~Jnds!Dsb$CzemDN87>HKeN+!|iug-*oJwB}DcLCyJ zKh!le_$f9vHmhZ7Xk(rJ@+8sJ-(JV-G?>+>Eh8$-T3pp|i2ky*)exz?4s}XeQ>HoH zxWqG?nrU7g#J6hOq2!jY9|<08NYRf+rDP~>UR95`&M0uPC8ZoLE^)VH_p}S@E?P{g zVS#i@>YsoAX35M>L3x*BsanYD?(W|A&Na3CNB4&(Y~W^#!R9P!`!6>f92~AhnzIfG z3JO{_zesGgy23~b_?>B$Z2z{ z9KZS<`YR#<_-v`V4LcDW-an zA{z0c=sLz8jB)9bpChN4Q{t@-m3xImMC$S0lle~veexV!y>ewe(gC|(Lsjs=febzS z`ipay%*-fBKS#L{;B~T-5JkUsbey%B%-h}>={c3igt+3m;Gk&@{JI! zlPm)LK`&&sj078dHjTf?3w@>8(A1W5_x`eXKc~|UY#L-brh80}D=Atk)%xu4OrWwF z`oz2@eX$cXQ>H3+uT+gVlKA=aX9u&9(NXK^)v-=i<49{Ti5Gy zSZbjrzEl;Kc<6#qZv|yX)^CjC!NYI5j1FhD>>XbB=9f*^2#EiiP>SQ&@#@h+>uq<5L|zI}Tdb z@3C&lJv%)ytP`!^chIR<&4)>zk8C;67m0v0h6qv&gqpFw_(;p|UtW_K``2H8C2cxQ zot&Byl9xBeZ8rs_&xRVeR=b?Ebh{TH7uTO&&JzG~M#edLsCe3j)U;sMvJZGnjS0Fi z?`9QM7+pBY0@~%?eJ44R9{R>fv?_i^C*Mf1w?Rl;(Qj!Fm@O)~$a1`t@qBzAO|2&VL$=S5M^$6%_C4>+5s2)-DHSAl6sS{LTDB zC_jZu*3Bu_u`4$>Hv#KBzw^G#7~ZHQR#SgVUZzTzWK^~MTw$}lrVffBeu{qf75iK{ zF|lfA`^$UXQ}LS8&SffKQ(4wLyD10A#vWwqn3|f7jE@I_?^;==cJ11=yz%7nu40dA zq3y9}Ygk^xMH7!6K0G^AEv=|_=7Kb557M6U$?0)}Srj8lEydBJC!0q9jy6k8P0ePG zXs76PJvU;FJbwIG$X_?3I36Wqn!~ZzZbx|JuI0Ww8{IGwOE@#xl19d~fi|^7vAXMr zG?SI`PR=DR)H6-LXQMloy;J4hoCKsXr2k+aA0H5L!eR8vDLKut2(9tFOo#D_ZQj01 zQ4q3UN;mcO)x`$}1?{GgEVpadE=@JHr(i0R^w{Tmwk7sz7dxgqbSYuUr?-q?X=yKA zz5T(HCr<`Br;|_|K8mSbk+*2+>UxF8J(_9MxlJ24YF@SKKoQSRL1{(t6G%+C57LC$ zqJs@AoFlh;cUSo1%oL)jYzP;PY&-nd{_tU^mV?b-Sw%|QA2wVLBluLAI>+1R8G zX8+8{$jr2^6y{puc4;NZpZTriC{?}pCGC!abf^F=i`{ZX_0-+FcQejB z>jyCEiS#m3?`z?7X>*M~WYdy>5*lw|kEo~$MpDn~f8-fb(A(4$B=gfWlIrlfG%ZqZ z_vfMlLzvlxxF9Gjd^+>ehm4|I+iY^9njNKEoSD`H5@0k`WG*X|J;2&D(=F1jBQeec zpQ6^t|5O82RR;pLzpxE^I|<4Az{=xfj_^m73=dXHP7)Sj#oMRR^CA zboBhvhtODL11&3g%{L!DoRM`Of5hj*DMu43dYZMH3u~at?O}mw*>4nez!W;Sxw7&- zLW2GE*FxDJpZ`{(P|o6U>j7zex={C~fOMd2J9I@&U7d!N=5Zsh{Y(~x{F`q{Hxv9Ymz;Vx}8?Q^p(lv-|=ma`o#A`Gp%3oX*dH3<1* zEsoWKz}A2&%>^!LJYGM)FAtDH5pu|wLH>r2vKXD3sc958ZY;nebN9j2R1%p*V`h3* z2J%-`sO%dAnz9>Xy}Gku)7GttDE2`DXd|?oJ$v@y_#iWWn{DCQrE7Q|rKP3uT>JKq zP&rb0lzD`!g1W`a&qD>n4fv3jre+MuipC|NURcJ~@AfB-Em+YYu%{SBvfE@=a9~(i z@|Q2?0h=gb>oh^tW9cW2#k`(!i=>arx`~!IHcc&(XzHobtQ%$nZl9*jj^~v00STNn zGKv7|%(Ck^wu+^ReY!EP*RQNksjKn$(JUvOxNw_ zcp6j)@&MIg;y1B<`#$RC+L`7J&eMJ1-D<4`lpmYqeE8m3?)TuqTBe(^v2j^@d;1pg zOUizoCE&tIRoeorrC?|GyiC%K-ES5X8>^K&=ssMeV>*NKL;=BB*drX1{t1}8)fwPg z0!7_^+qOczl!-L28G5UmmzNj0s&0-=c*=W(-g~D-^I~2~Zv%Cu(*w4S6y*mkYGYo` z>@$72nADZxm)`Hm8Hx=yIzX=FI`VaSPTE^2NS8>p3>H2xNXBg}ft0F%=FzCvng_w{ zT?5W^p6)@Pd(gEOz{_QcOjVLfR&$zPWsr~gz?bQ%}UZ-Bef*kzh zJjS_klTU*a?v&m^t)Nk*5a9hRU?g5j=EAJ0c9b>cy(Ap{Ermku(g&0`Y$N>ebrFKUAb+0>~J%Ss8Kc+0GqvA4z;fFnaglLrRN* zI^w8s$;|i@sAA`+1E0>y$;p9IQ}^Ax7i9e5rU*$*ebnjFfIYngA+yg?wV{(f-~-o? z`o@2fN7n5{Yd~W1=y|CO=RrN$$*%$ZO_}ki9lw70vP)iGp54zlEmra~1Cv^y)Kiso z=I`HD)rLw~9kl;)g6K3T&MZGZy$=316(q@-cF*aQLFce$&43F(2XopRR)`foMPO?F zf;Ad#bnW}e4^IwPcWA8Pe+SZl2)6CB{n)8*iUX zQ1s^^U=-}tC1?p4^qD{Z{1dR~h3D*~xgjJPiKL0Vo`^btR|0fz-n!LX)l}p@L8|og z+jg3D4GpD=75_Mfd5l0j=Ea82%g-0uwd)wy&u#-5Y0waYqu*6c_{zOX=gA7Qft0bKN6 zT#gcmyrw?dQ=Rrrra~E&g6L||TW8D6OBQjx-N&*nrsSFtVpk$oKO?GyJ&Ic0lDID9>|*SOfA9Vs}gn~UCWZUUvN z8APFWnvr#Zb?olsX?jMZ9%mJGuEFI9H8!sNCi_da==sz=aY=;&?4lhXa_9Q>5yoSt z++U`M*Nw4=Z2gpF6@2K*r>AlrQ(+n- zN>h$rjER#!^M=}|?#HLrkL5nEdZ^90UuUC+R$IVSoByR2FLsLbhL@3PmTq$UI;Ra7 z)mN<9OZ)qCQ-&R1F!BoH*^D&{yfcY~U|%23&td{zYH>GvGHp6A4BID;IVrDu-#1<-jZN&h_gtmdkK+(o&c!zTHP zDjtQuuRphezL3Li=V$kg9BRR*Onw<3O6wmlT0uU|!4anb{rmrX_kZT#|L^PIw7SxW zeJtM&7w0YW%Q|uy2Zy9@LFS@;e?NP6%-jb*G;tdX!A8u?1p<+jxji-VU2#~-pe9`V)xSc2C2E+zM^aMv`t|Gm^@#zq-~GKl(OR_v0|R48jz9C{ zkQVAmMy~yZGZ9D5B_<|nrFOyrlo4(VPO$H}GSnRyZLEcZ0|U+U4+Ad|#f7GC-e57a zV)=4)5JY2ddHweY3MT5MZRn|bolup-T+)BySQww+t^*B;_6B?I?~qSLF~giiNp`7` zvh~oF$3%b^nmI5zR;SAPaJc=EzrSjph#QottZS1a(iWT?9FG(u-=)KEqaRt;+S)p_ z@bb$W%ckI6vYta_kxc}1W=XZk+{+bWy+EO<>IDGp45-ut3JT{~Q!L}Tq@D=()QayQ zh=f7>*%nX6?LHH#nfm_5vJEe;G#%*b>9K-mDijxssePyo??N43(in)c>eMQ@)Fdu{ zJgVNe#t-~zHhY7BNW(@8c;a$64Y8MZj;>Y^m zddu{5m{R(?@*AFh;spk1|N$NsYB3&~OemB`Eh@6Mc*6B1IgYDh-q?^ngO!*VI_ zb2XNMp&_364^kZC22WJL8KZP0upYbTtNjAjt!a5byu7Z%{p>rl4{Z(l-dzncK0V6y-0O7{s1SmQKte@bMTaBD!NnyQbOTwutzL68 zv>0f@k5gMSN<8PI%U?@YI6(syl$O>L6%{4VO0YGo1ZQY3tQtl2o7V9v^rD1?o3m`= zx^+7_IXSTx$>i!ZrW$G?;IDM?p?EH4SS%eF6gL`J5NANS7k6O!hPs0X0xlWPY`T1 zUYJVtkB7$ z{PtaCWoAL_^Ia&@`WsT7M0ieFP#hlbIo`VS`b{^O2ZbpTT$j$ldz3P{J*jpGOEMXA z1*3`*q8?1LAvT5I6+qG?Dhy)D>z=a}U$$ZeU462DEP%XSO6;<9ae<9I@}7l;AxWY8 zO#gy1Tj{swv7ARf+%R<1se#P6XHw6FE_fVOwX?I!`m#uEF3Ft-Ybr4&DZCt;<%R1g zDeV%y=4PE)q<|t}S5D3d_K8G?La^0vf8H+s@nuCp-$l8(pN?B^oi*LnR9QP?}rDS@X#u>ngoj z_#0?3uondCH$tof!`@V(sRc?1dFs+Od;UNbwGtRuh&0AXIxRH#H`YbdU@&wuEx23> zu{_uqO(jC{-hK7c;RMfq|JD}b~Zt1MjI|O1Bp@)#0W$cDg zeQLECScXif%rDNrUQUVac@`pWZuzwu;9! zv!lAN&hqwuAnG-PuERod*1mf6YF4%^q!sI9B@it7Y`vzRkc!uJv^$)=pS~%p05{mQ zX;UAn|GEHGxkT&;LWv@R*1WvA!fK3b0Y~X;`W5-I*RM|=ZgmPUy4rCwQ_i6X=z{o6 zA-&U&r021=31%~yh(mAuK{*6<`pdd;50*2`#Z(>1(L)5OBAL1gBxiocm(_Pa>E zFHDoFBrLt{h^00u~TGx+A!t5Y;0Ux9a} z(4dS{5ZHgUCzmB?2ZopPX-yk+gROt9Q31k{# z8h&+q9V4l{B3~8!9X2sP1(>1^IOo}m7caONcR~>`q$eWhek%IQjz1>K>`agRTw+6Z zfPiX43?+a#Hb64%*|X>EP=L0%4!79-T|6_c?=jUEPnJa>6MOjqpU+GXXk&=x z747XM&1wID5F(%4EM}Ap*k%mKyQylc;~}%iV~-y_iUD~IQ;J@Seoh0z|C#kG_C-N- zD0eQj`SkoxLKVaMBw<$ZXnD>gA~WH%{I*M}6Ng=yz@(IVldCdquasthW7FQ8mS+Iv z9%D6%q4HkENQDBnw!@ZcNeDEaf#CL_|eF>KDjEkT;XCZ6$6OkG)JwNH{h3 zqg)=JJq_R+ufPr?H9@#EwH7GNmXb7H_nM=A$Kk_=NlyZ?Hllyvpl#cc0!mE^u}Ts= zOHfPs4ljeQuYsCwiuZs95$?L|HWGH9f5x?#ejopz2<$l3STY+6&tfYKyNuN6U5MCI zC3DmA&`*>Jj-v$xSf$u#-p7oPB8EuCMc7+zm`XN0QVo_f4a+90!gQi`58k?U%M!(a z+fZK57`hBDg9x6QD49!vN2K2JJ+{&p8Ui<^qN*xeKLW3p>~4rCAK!VXHNtuzs`(Sq42xFZ)U&yekAE0*@x2ut8jGpVg6Q;o4qY1(Y>`xpK+f(2=e*$x9Dw3O^kWi9Bx@wptGzp{r z+E`y`@7mGQLg%;K8oZMfR@uKMAUC=J7iVQNZR)T<(@2$p9i*G*9w&R`$k(nevuPI; z3d@%-2Oh9~(hIN=O(Zl(3@(Y#eDB`At>)lB=vNuiDME{R$g9|p+lq{(0tCk;e(4@k z=1vF-z*cXAw)^_~>p+08@>+_bdc_+|OqNrhIsulJXh|Uz7BF3kURK4SXYQrm3$jO$ zit4f&jp!t>LBwqEUa+Wwty3}X6H^PV!`FAP`ytXw1N@&+1fCG&LpWUs>lST=u5LZy zF0eYC>(rl%sff2o5MpWko&wL~3y*1MS@=I_ufCRhQ!vUtSYaDsu!<(h9q0nZL35sC zJdrE`0TXr7jI#B=K?s8leT@b1CKnI|xH>Z^jh`5OLqj$$l|~Q=0PTQ`ES@MT(TL`Q zd9n>X;+oOs+QUssP`J5kAWQH=JUf&0oi#s?a0oSCbr;5i#n!SKu&P-whG|{9RtTFW z0|rP(n?+g|OHWC9ej@)hnHubaBnhWwkkTB~tv?H_dE5u~6A6a+RIHmY{X!|7aZhXx zz}R5bGK4uZGR=dfZVPBaYH=FakjRhbLbez~;JECqH9KrjzJ97GK{a7Mn1=UJp6Ulc zzYS`IBoc4y$jtP_Hn=hDb<3aF7N6^C{5(8t>on9dncTH{yQB$GE>OQ^o57Dqyn*(} zn?dxaa9Oc`VS)_sM<7ocuChPdrqxn7^6}%V4Wpl6>>F%;9x;uiyriHD zo&(}n=WeiXG7H{{ztD191u&UIG$X*wh^zrLO%UOUn0B+T;HE@_0Wna?IM>upph@fx zHJL5fZ|e1QcRM!-Fv6s+M+654lXP+9$dyXeC&Xq)ULIkjQM!r(z#2Ndl1DZZfCt3RK& zf2UCp&b?4AF?`OFHLw=^T=im&=#m>rmwyhnhml|WI4dSCbAu}{VZH7Sud#@-E~G44 zT3VzE-z$dgQSUq7CYO3wkfF$64Boz*|0DJ1(&8Ff*46M(fI3|%RMuf?b5CS`V2?NP z{ud1KzhWBy^Lt>D$#}c}19RGU;+VnAZ;KxkaY4k$-~s?rlsCf_QesETmDz|U(Vl@I?X!!no0z>RtD*UxVXfffFI-IU?f0#&>3i=Z`43 zGQzpJP?i)4vr@jA=LBE zhGQHYc7N_7({6GeL3j!diG$}~{Z5+i;7-aW0xWe$IKboA2t~Mm-~pb` zG8_auCj6)?T+_A0LyMc9`2PJnN&&R^&~QmPK)E-m$?@^=XcR~QX{P{-hLid${1w%Q zaOK|1tkD%liW@R20Iz;P5)Bg*lbc;-FJB(K@@XH5)oe~a+(vf3U)SUqn((IZNm9<_ z+FB868zg*|2@)+Bf;Rr#VbYGoal;Mwqf_>@qs2H z2)j)T8KwJN%6PLasttzr}KSz~RRde)F&IPkjPimmEFklzPoFM;ef%lW? z#SY-3fC(hHkCdqBDzZdn$p!x%WZ{(GR&hy6q5XSa7{#S;5R~d!cc-=JWjflGhdR4;#B^ENMv#h!0N1$(p~WCgESr5*WuxL97`KR9VyLwggiG? zS8{Rm=kZfERoR;vMudc%v)zx8ORgXj#<77k+46Ct0~Mr4SHo~~)@j;C9v-7Kd-9n@ zzdl1aMRZWd!a79YqmYnLMhY|if)Bxolv>n^OhH>LL zy)6ns1rSUu_v7=T8MGJ?-X3f!apV#jgMd)r>u~PfxJWpCVsQi5GB%iyesGW^%Uh4i zoKi>sx7%266j+}=1SO4&7cYAFEX6Ev1sghtc>d^EdIn91#4wA(nve3&KmUY6cZhK7 z7tG5)@D#1Q20!}uUeB^-9-2YVTs^9Yr>I!-3JU0m4rC4foS6v>2}wlZY>5tL6el61 zFebGOX<|jsrmnFmqRV3?xK^_+zFTi*ZY~UOAtasUYu1ozG4nWC1BcnP(tMB%Y)IQE zyHAm~rf_dd6qbJyp1rH$lDWAR7(!C7612r;HL4QNLmLRxso$}dOCJHdg&wl*AJ|70 z?=z$Ix+Swy^fu3FYf>;cxTFbD6p4^X=bl7FO}nVXiTJhJ-LT|I!{Y?~sLRfi>=oaWFg=oa zC$E+)VHo0~P?!V?my<15`npt5HUd~bhBP!NdAP7h=Pl0Jgyl_#&Ns%q|op{VtQ(MsLW43>p{E8?Q1J%V^esxV?= zBr*J7oMF}Pt1$45Ac6QOh&d|vv5iV6@yAnOoB1i|l+T&_ zmL|*LDOE#+e*x+cG+S*`2Odo_P-DF=LOm_!?b}n7TFhWQT}O;U7;}CKn)83%@E~#b zE?K(N>dUJ;#4DXyhFxXH6|ek-2MyqmK4N2HNZm!MWlSVL1w;w$)irDy4YWSi!-kRs zN}ae<(DAC46>oG# z?UAw$tG0ekJm2$tr@m*eeK}F9wAG^K$O4O4^y7;!Ee1J~Lc<2{y7v+64#nh}q;@6Z4heeb7>r+6c+`gf>Sh7Y7?OZYmZoMBjr4*1RvI?8OxKX&ThlyQF&{?k%IkxY6L6HU<`|d z76YD<4u5D@&>Vt?(=ja0Pgdv+1>xjk^uoQrUJwP168f@WU5_Mf{4j9UpS_VFR#Q_$ z3`)>)Gmn#3pYKL7LGOniNL3Xkf|5Dm+3sx>cfuvGZ!A&#gL9DXaf8o1R*}c`1PaV- z{kGWCoTND%K&z+aw=ncwD{t7X{1GhU3t%Ylt8Gm?adpWvMSRam8{wId11GTo%(K>? zMHAH&A}$+~(T^BQ-LezNgSc>=Re8t0f3HEIoQl5b32z4a4sD5_`{ zuQCvy;nl0T4Qa+{ME>0ydB{4HoXt=KI-EqZ3csYpyQnCMs!et@IERSAc)3)u4f(L#61NfuD)zC#oRnTRwRG(V|BGE5esbc`{Jf-fbTWzUBqU% z4y}xSLtTu+GKifIaUec9Sp;V+7-9?|WStPJ9HMz@iXB!UN4!1N8UGl(ZAdkw<6H`d z%96?MZ3N;X>Pi$Z5kbZz?dvnl#zq|!pEa4)a&)NE5E_c!*q0fQ%;eauOjO=bar67g zH^hUDo=bBdgkm+Zeh-4>dVGm%e7;o?0rV6N$93#=b#qgq!LMg@PGmEMNH;{3A`OLL z?7~y-6SpyJ!2RtdB>>u_Hvx#gAAJIWe1uj7YawBI81fcQGf;#0$Vk)>zW9<6dalIP*d7s8tNtCK;D*>9Y-@y&Ib=Xsa+H)EsicBv@JJc zflZUISn=iQVUS5pL8a1Yz9!%T5Vnc`IZ@=Wi7e3e=FHA5C?Fk-jT)-5TAj|ve;ors z?AVB^8N%&2=3@q3^z?4wTTMTSi<(FQ2u`GzVG7K@AO*b|jnJp&4=jO_Ls*#rUgYZp zeF9DsN?}w-W7NInjrsvP>V0*;?Qs3N?)95D&4I3j-i46^y`fqFzYhQm$uTA%Yz$z> z=&83sqaYim0Slx!j6MKD7qNiA15QWd0=&li;-QM*ndSme?p&&lNI`kYOaY>#lPs8G zkVhl$F84DjH=ZuU`8THmE73;PKiL!Mk}C?RgwsY+0-X7=Q&f)Fe_QwhZFt~l>Bvq% zSYpAY8CBsy4<3BJ_YEtoe-oU3Sh;E-?D?9yPdVaW{mAg}y_EM}7_I2?Ur>Axe8JFW z=Et*qc?JJ-XyJP)XqU|re@W1qPTVOISZ843H}}E%HQ|CPg~Lp|4;(l!hVrdB<+Q45 zBjPh31*3OVxQaWoD=d}~30`*m(*;X-QPDKFam$uHlqqKWY#h84iqKJ z$=qx~Yd&Iw5)IuOJtX0BD3cB#VizHbCDI7#3_*%xpqp@bfPA3euRxq@1}TA$g6(FP zM@>&pr^P^!WL!9bUw3zd8D}AgaTr_)b6yd}M>ZR6=0cQ*2l##rU>;85gB;4`g1pa3 z=6&>NBXYXXfdjQRtuTY|QLwOn0VU1<9Fmgkj{aGq%dko14>|rJ5ckTm2V*^hH{lO( zBI%tY?mJG`=p{lvnj&tRZRgLz4@D2EP)uZjQS@@{qEH&1D-c*)1qPEyF`&R>NmI+# zC=BEAvHv^Z+S9K$Eq?6U&l>mD!4rLTr;Cd{r5~=grqHvG&;Ktq?cc*90qBK`5+uD)MD$DoYUqj6ShQpX`b|_?f}*H^cc&%O)H+= zXy)_&AxzSwv5)w75F7Q=tjQ|(!|I?B4u*{bav-HNV#{l(t3N}#2yw|TU$)F%TbE40 z+Kn6eNC-%q$R7ShILSNLTOY7>;ci$J_GC)s;-aI5lE&3$DyL1*HkS!To}usV^v3&c zc$&a|j_3Iwt1kQxXgBC1;A8F<#%v6!EfO}7xc}GRe@j4=!-^^W1%~}J6ipy?fkD2Y zq@t4j1T~O69uv3t3sLrSG3zg1CiyNSR@>4>k9OF_8N0iLTgP2(U z&-m}a9e3b2zM8EhU>rqnDV#~xGpJR)eM9hGi0wmD`XIm@M;G$61t@v~_0ikZF<^%u zhE)jWtR)BJfD9nqQ4)$V-$^JE990kgb2ya-UPney2{)SRc(hm>VD$x>Yj7Bik&Mfb zCjJ-8S1Yvz&?0;-H$F9L-I}jO&O{h&*N9WL*G+{bTd(k{y~j;N#LpYC4NzJ2y%UJk zL<0yxF#t$1B8r3}Fh~Y<&W6|;k!9fICirba~`M)h(qK{QdP1`@CNcLP08mCu1dZF?eS?q za9RvbPLYmw{q@r@MN*FuVG-DqRE|Vp6UV*CSE=huK}QqF)>MGaIvg65N&qV9O&lK= z)v-avu5x>xo#o)T^aZaLEgO6kbbHWZpeIZ7l4rzM*izhrWMA=6Oo^l|A3r~_qh$NE zYhJ|DBl@b%EGCo1rxG&n5f5ad0>uJUw+JDSgPiizFSrX=>iHKJS6p0-T0L}jx_wUC zU=HWD&7TiNrcjXpmbw0}%`YLYouZK=EGC9C(}TcSvltcKD}{khKXwiebkr5Qh?GwU(IP`u67i!E%?`#ZD>UFPU%ArB`8LMK zerBX#B0KVymw2rDqXlu-6?d`0uw}@ZcYbK01oNTy+~hh?Uf3@>I5BvgE@|82>m$;o ze>W~uC4&M4lA`%dRy?*b_#8bF29Roi zC&;llGX?)z$J{6JYD`xD%Itq2vscSaU&+cLMBm>e{J+2Xi|9M?ABaB9v}MV#Iwb0K znOB9KQ+$XI;_LIZv-tVsXiuX#PGlrsxg7OY3#~xWv72xbU8$=rGCrYMop5{`Vq1_t zbZmFh(V%i0tLf=1K26R^``4!jf$T{1L5G`xX$Vf0aJz>MP%32*j-AQDQ6awaI5{DJ zuzV;ePSZ%u32pckunuVHU?bPAO>m)ndiM~~wq6!p>>M1sop2c6>+*6ssvyFcS-Yao zg~*;{@`>Gpw8;o4AS|oN&-XSwRs$IrQ+ora1w7$vYE{DFGim@g05LcbRk` z+2nO?CZ{-H=aW_;g4#&~6iQQP=LFn>nu;}tCwn>-fsS8dI0in*hvY^)ei(&CQ-&Ea zzY*vRu~tR~tHxV;K7N(PU>Q2$WVj}D*R#@CSb!SP@9`eOw#<{_UpAAVHG13Ecf~tl z;PaY!&${?#ZFH_`|SjgfQ_f#d}9^(jIP@6T@MFNuS6o05%y^CS+=yJSI1#-k+Y zqn#Fghz6ta;dU{|s=r2`^K0}mv*5K@uxFh`J;BBVF@vV3r>$CY?MWvqypds~7C$Q|)uJ0-&}&M9As`JLl*VxKF7?M_~9mq?e;bzvPNzKxu?gAUwmjZ{HkP z(b3WKoD=&$s4^HCjSp@*O04P*5n&k_ecIzWLTGIK*IEHljv4=RbV2vf-BuXucjIdvi<>2DX!S3tyZ*$BoRE;!3 zAzgRpM()(A-}kMPo8zc*p?dHTbW}IR?TX^n3FiUzhL}g3MQzVdArv-T9K@wd8gdDO zC@RN-F<7=d=9Ju@2vh zV77<|-(4MPtG^iv2gnpfVwFOY4Ch32@ZMm*0x!>-as-`ugvkZvMw*C8>m3d?nLpDK zXH1$Cg8K8pe+4YKr(jB@@e%nPeyTdqYlJgI*R%zw2$lO}%eelU^-{3AM6H4V0Ux|6 zjSp67WL^sG1^8?8TjZ*mzpnZj*`F43`}XezUBJA~EKG)|!mnRF1#L+C+Hct|n>owO z#{&9%{?)wbq0EOaZBkFpxZ^O2)xr8`0KpV%vp2(R=`H?C{d)-F@iOQrEu%JsD&mln zJBIXa$Mq{Vt;N|ts3)}D3&8p{{7Kl_?ql;}!(!sig!NJS7ce-#-UYwI*JG}EG2y@d zyU<@Mop;{<1mH^+EZF^z`Hs??PerTE);&z8Y9Ic;lUu_g%-pAN`f(JBoVjCe4CZJ%W3YFId`Y`jj3eoUy9@ z)?~#d>RG?impA{JqJJX`_3J5)r`?jl3lglt%sC8xmkj>r?tYRH@U>53 zKJfiN_%7lYkOTSBxG;}r@!kKszPPH>g6}AIeP#H{_Vvfi1_)?u;&eHM5|gU#Wu3(G zSnzbn?9#>F-Wti*uenLDgT{HZ;OH2%>@>R=qvOsDNEEy)@BE_n0H6h~^7WepPJ0SJ z++*6L1@9#WLH zJG1k9`J+&pfyGu{U?GNsNxFvP&k~>j%W*tf~g81D^zF47x$~*ofg-SO{2c zuY6a(kzB<2O1s`*wt01EYLUB(j^5DpaJL2ifV)^hM~Bm6lpU;BVsj>|^Eb6T+meuI zv(uc7=VpApeyFV2e9h3D-58hO#_7Ek2sk(QxRQ#Jlk=V)zK*MfazERdB$$nL4Q$o> zF_R;a)H9Z-ffh3;#SW#FF{`K*bo z5{uC|cE9}RjQo|`P-iQGJr1PNV&Xo8TK%LRmXZnMx+Cwb`Tcvj6NAAkB;q)BDhy`b zXROqcbB>q1;MK%stZH6PO1pi1e*XLu#>w0`>MT1*>=?}ZN~dlImw0(|Ni0`~0@mk| zDXMQ4%F!OF>4wYPkBPOSf8DS~qUr zu4mDZ3kx?26Ll}=sH8qW}M+5LE28P^g79Lu89IG`jAbc;-*%F@{71%vCGY@%v0wW`Bw{5mX zTN5uV{Mu5fH*Q=dhs5KNuo&y>H-gp1DKffpa;t>)iUn4keGo|1i;C(P`TjjUPo4>N57Y`W%;_vF-fZJ_ zsx$jV+JvN~8*Jp!Tp84$Wst1k_t*B~HyML*aeS4G4_Z5SnyxOf0&S8Crp>Un_`*$g zvhHj&Y#Q}FwF{>DD?>H~t@!ASKIG>x!1f9aX1?RMT1(g`^!4>^SDD3-_!{Cr_pO~G^h7kp9*=k&8DAr< zURJBpT3dVV5>>fEQ>?74?DeZuMJPGrAYnGeMu5E}?q*x;YfN`_VZvsmk)x`YY;1p3G$ofPXxq@6?~!YNQ$EgJ>Q=+)9GanobB6HJ1uk- zP$5*Z z#wfddv3-VR)?V8KpQVdvOS5^Iwnb$xcDzh)%r9k4b!B1m7IwRAQhX}{+7wx(Z+f$Z z)iJezl{n3*O0YIwNov@3ul#U3+)^4P@~V9OwEWhzlizS&1-)3B5mV{KlB4M^wlx1UWeUn9Ur;iHb7I#1^PTYln zyFpDNmzve!226mZ|Y-j^X^t3_8=@Q(?5~oigARyq9Axnv(aUj5` z2Itwcs5a<8cpMDNwJ>iXcPMt4rM0Y6Chc*((L^tVq` z@ov27;XX4Wdt{{0l)Hq&r-l-{r*k`o{a0p>V~5CBk(#oB>=c6+!ZLXam0KfY@wm)= zp&RRdBk_&R==NjkKK35rTw8EFgE2r-!vKZW&gl1mHeHN4SJQHTngL5eO7G)w;Z-Wo z?L-#Y)`4lPu!1AS#+i;ihCAAdGF|7lDtNIuUfXGj*D?*VrwX4{JaXg*8qF_UY*Zef z`-n4+0e4j4rhmMu=e44LbnEk_PRVOsKsjk&IG*kto0XMykSvZjmpn#cpB)IHNnSGO zlflX&RU63FlSEF;war&u^KQB@G7^6s9~2lkF((&Eyhv>opiw*r%uU0uxh45dc2GCr zN>NL>Y}U~!9*(>(G6%@@Zh~n5e|1PP$MA7HEs;4Ita1NdgMe7SlCiMX0tOtK=vd;l zp2==qk0Ehi+@|$tLUVijUuMaE7Vc=7D-}~z@u^mIq_nc1P_rG4BMjh0pfR&U>HTyv zcECk*=y8Oh+t=QmxTXqVGC4yHpca|QUKS>;a)d_4MkdbijeR`}ax&3-*iH0YF;aFf z)ct;Y*q|2vn6{68^x;;d8D2()Fzx0qTTjmyhrezyr+4mv)BkSc(;W@J*#{6D4hbJ7 zz5lik;ys6@%4P-n68#&dUR5?atPM*9RtD$7vsF9~DC8|jp7R*J11g2(-`fvgGEjl( z620OwU1#71a-o*n2E4jsUhYvyVKIPI{Xh=1xj zA$i90=dqHZly4sJM)Mf;`JOC0)3BB6++~i{J)_)zPDmhqlpwhSr;l}w<Ua+ax{e zAhlwFdI^emWJR6n=N$M&cu&DK->_BBBu$N)u`6b|UjA!cdAiN2R*eRF&CMExyhE)p zbAYfj5YFnSf0YsblG9Ka-hC8$w%XVf6pghqCLk%nKed4$02?awG@>^Pnu;c*al7SR zZ)eHLYh@@O_7X=7VzjGg%R^a*ks38+MJ>)`yP#0VYf=9cNo7L~z2b~n7;&2II zeYh2VO%hOBAty^LfmnGGV$OU+iifG2F(IX7Wta-}?9=zkZ6=hWqPy~Zdweo+ayK^? z%kJ_W%8gQK*vN8=PyzJWxkkz2lZq`%=<$tPTuXR^4bk5ce4uUIg1=M zr@okHFS9r-^7xlmjbxpO#W&iJ2#$uS&=m6&jA;pqjD!{umm`tFs=KW3-_J64y*5^|KyrjR$zzH`u|}p~wa6a3F)h_RiL*VjYS==!Q%M zcD6Lw>QFvD)>V83AzK4#4mGzcp6%K#csnNOMKmD}?=u*b=Q0NIEAr$P{1jwb}af^)^yi7^}zHx3yVZ|ZR`6SW%AhLLf zpUi!9EDiie=K_NhQ-d4-If2;4Vnlk}n|{n-{oy?Jj@CeGRrbltY@clo2{IcM%pu7< zWp~#?xgYpVYExX^TF4p|tmMVODZ{5+yMu-`*%as5SrE4APJox38yWNH(Eg)RNd2%> zYWolI`ZBr1`x1H)yu8m=G8=H&karbwH`LIl@4p#%y4kF+Z-8{$pyD-W+@H7J`;|4R zxXsvDM*04iZipL6K#d}-^Ni3}`e0p+|Llh_mclVd%3YAo-oaT^4w~{OJ2B$o4yb^l z3HYn?dm|q$hDkK?@rm7zR<+j-S|W>FD!)6i7Y;xpsoIXC=~s&W0A}d%e03+cjcS$2 zjxszp%ewJ{0dlk6Oj5l9kA~L>i~K@Ny`aLw&ZBKQWzyFjb`hgsMGz%m3z1tgev?^z z6T)?^duYWH+Mh*krnS*I0sS5#yWT-wM$ZY&g{pbXc=31ghWCHgo*0V9B@c7@ln&0yWfTCpH$T{sH=lD*SB;qsUPU3H9Ir-W z6Rq_8qN926tUl)YQj$6ng?|W$Tjw*dlJU4ByLaybx8892Tz>e5@uf=oER#gnJU?o21}o5s*E+w~_6wa4 zHCes2@I!$zkoWZI7v01hNx4(M9&?7gA+*ka;?@($R|p;ftcC$d>j@rvp#;wH=+SVUXY;e) z|BlT%-bf0dR&Db<}kJzTctVucUsk3gSkt$hbs&|iT>FUd3nwxOr& z^OYojNQ_Xiel4?BI4H#xNP9#twH0xkGl_-p++AfFAgXcy!*q=_eAd{XoMANS#p|Dv zg-#TM6NdOA2G_N2wt~UmWz%*&k*S?d)Z>W;&7cVK_5uJZt6aX-j;>jh58$%DU*xm3x6dVWnE92tmHn#TnKW8e1@a5Z4qA9{S@Ki#lY)8@wo34927??Bcz9PqC>f z$ZNOld0s5q7~4fg2N&Ac>q8M~_JGiXxB3RE3{na!{>trU-aNl9%f70GC63(2buSjF z4qQ#~slTWxA_nePy}ZOlwI^5ZtD^u>kfBstL$)5!7&vvsq>kZx08#%OoAE-0eS0rA zWq^S+irt>hGq=3gQeXe|3#xb-dkmDuYw(v$^p>0uIH=>=eH#sFD8dCse?s-_5qDiq z`35Z%r?rC;S=nub4J`7a*2>ISIH1G|&zz-WcNXY!FOP)p0nIeiIyTOanzD=%9(r5A zb3&1Dwq)T zOyW}OT4T%IzK3vzD2D&-g!zhs$3uZRa!;LR@!!UCqF+Oq7dY|auwc5CI;yB6{D5*_rKSpK!rTl3wf`+O5kPq(pJMXbuu2HGC3zQ% zh>^)uWdrgAqC^W)@au%Rj{>d1{F~=y1ckO%(O1@~>g_FsA*h>>cmTeWfNY_L^{`;x z^0C-E$d*>gnA7*|nbj;N1=Z0R_s}cYfrsdWdMyk_E)JlTYI2BMl9Z zJ=r~VHeR>2fI1HWy8_D|-9>3>X?gX;@Bt8Di$E|oYlK!WtattO1xNG4nUk%r#M)Yq zQkB^2k6x_p1ei85YH*dZq{iuW4%h6oMd2-ZP&$zH!XDD_mo01VF&E5hsrD%?LkN}P;#m`!yGk^xw27}CpGvnlVV`%9 zfYsQj#Z*BVOlF*Wp2ph*3}x`0&HTcwyqeQnuJ{IIlF@92TkDUp{ivR1=)5|XZ09*W z4av^xYT$^*!Kx|(U(u7`ZUGbHx!IFjr@gms1w#iOld?X3c|xNi)b#q@*t!mwiM=M` zRT1oS8SL+O&UiNq>;8-&MWi$5k*j1tw{aTzjdQ1b7c-Y*lP`wV7jz zj=xD$Q-U}robHpfogBq2FJ`OxA!k!KI5JVyQ6jE^DBo!Hs0Qs_;RY$$6K9Sk+A^% zrJa|Pg(2EMW=MU#RgWh2BdaEBro9%7;NKw#Q{23x0SQej$X*8zJWn3s7^D%GeR zpa`vnUG6_GZ7drgV(ibm@$00Og_2Ut(SW+fxGQkk%ZE#1cWqXljC%T1cDZ-B&{0*R zU(t0tP{}jLhj@j|W1(Sx?NiXA5c@7r(=z68n%lOp`0D?E3zlQ1Qd@{^|+dmHLK5+F)Exl8~}s0 zd!{%ak^Rz^m>96i+Sc}2W6Z!VVWUuh5w$34d|4}feO(I#etl2R=@E^=zVPRMC2)x$ zfCGdkBS3}E@VXeaqAQT*Vmp5lQ+Z@>YBxkgT#_TQKE#epk$|{@=S`wAI5N09F;)>kKmd;d*-Yj3USU)Vj51 zXfGxqVa;8dW=M%@Xe66__-oI|=;m9uQZXogi5O|N>xnwJ)5O@kv?k#zEa~5T%QPqi z<8SjxO0{`+7lEo)R_O-Mh1s))`M{gY#ylb#qe&#fSnOHuWkz*utogW6zw@Vmnai#q z&D_=f%MSHHeB<5JS%ZbO`&^seJEi85;)31!zdd|7!f@m|sRTagSbhEngE#lem?E$z0I+cHLo~RkSum-`iDxQAXwN5lNI9 zDy8{r7nJ|`&3@1^3tTv59GlkApjB534a0D-Ax=q@mdyD8DLLc$Y2)-A6k^=Rz~zf1 zmPqCY7a~pf&+~Zz(U-ZxtOyTV8wzK4pwqTREdBE$WEoGGxLnQq;PfKu= zBlamPnIxi4#lCZ>6pa8nG2T=qpj8M#p}_gYr^FFjM0(`3wOf{y^L;xiy~|;ySB9u2 zd9rKbGMEbO>chfQhY%7D2s^XN`i07!o45X0g)y9#2tCik1AW6!VFzpVCfsFc-9wTZ zWIut{mCSdGpYBYyr^xxmXmY+q|puZ0||)?fLwo90Hz-qJ#r6b%*^h;B`Zos+X51@hR$bXFIGlcxPz5v z-Qpt?S?xalBloHj6gKoc!nob!ekj;ZbULjMA4QR};hMPD?nl%g;IVgGE!FK&b3brG z$;(T$w}E2f&B)?RY9NH~5R{jm>7*aid8Q+ArrGo6&;2K;6G=sP*u5PmG)8aCb)1cp z$1<;H4AkJpL7ZIVGP`55{>Rx1W&$=tQ2S8NW;U={6ouubCC03WAu7-hH4)-;3)=sO z02`nO@<6fU%;eujl>cD<-9b9k>I1;6fLi_gR zdKf;>i^mzf5Gx7<3InueQSAdL&(OYqaRAG}D3=ccGAndU&FzL&nkkpv<7C%|WdjEV zKi2{6%SbD?3R-Kz+l>y`4~N9V%loTAk!_yAJBEga z2)j+Kh%B_!sxvEmp&Ls7by7&*es*kEHdYgBzf}W(bC5hW>L?wbgqpYuO~3Mhc-bh& zR1M4lSmVOHHr>ZF@Df9eKRFdYt?X}s>w_zJmr+yPQ`&|?k% zX<}V4E|EO^Ljt%<754*$qj$6t$~`Dz(6Hqdc;^UH1@fS6B8>j79ecU}!D4;wt|ICc z!Z;`-slg}D^zrw$8kRnd&}m^#H4ebU+13{qc(S0_?pc)k~O8ze#)(m4`=!7#N)} zBE}jlb@f1!50GdIA#;T=(zRk50_aLmObmFNrIVl;A%7^d@X7l6l;O@x`l_URn|#KI zAeSL*=mC-yLREp37P4WI_LNCRbA2fMYCzDdp{SzelMH{(ml9^Vcgg_^$=vtF=g(}` zRuqRmF$cI{pozPy5c;~zIoHbaC6hAa5^Ns--MVTu>hKKG)q!Zq6|GG&du*WDH`+bn~)E?x|H$;xNXeO@n0JE zZlu9|;E^O>3!;Jlc<RH3uHapkRaU;7QsZ}p*%Sva9+OnoEGu!wByPGA1+X3k7ZB)rKn99bU$3VAZzH_u!-p|`Q@#5^XKDKDxv=&jYIloS z4>~+r)r)EQ&h6|Tm8ch1IjsY&m1b?Wr%#)4Qd93||He~f$-;@2-&q!b?VqqTyv_qm zF9BQ$OkLjgM)ZHaPr!*@t);E)PgkIL%xCkPm}=(z0J3|tqyea0ARxOT?cn3|)tQ3+ zKR3HzgliIJ2Y{V1tYYC@s(elU}Fac&cTy zIsp_+LGU!!G{)0iD-E&A-TP={Vp6V+tvCak6>-X({_HNTFUh5>h z|Dd;1>gSW>;X7y6Xg>!#(ag>QHLrmVf9%_qL|5*Oq^X93#oJayhMN}Z31e?rGx_yz z?P_mRboXfJ3de|h_Wia^$E(DCaeD#lpEs!64jn~AqK{Sz-+)z*Ssr`zqSBi7?f>-p z-^Z!beu+4-W~F{dDhsxC{2`D0m$CCd7DavdE%5KD^nbtE4C)fMZTs`x1F#=&JFom3 kl*Md&;P(IbgXMDLvmvW=-nXjoi?`, orthogonal selections and {class}`SChunk ` slices: + +```python +# Avoid (data not in the page cache yet): one read per block, one after the other +arr = blosc2.open(path, mmap_mode="r") +patch = arr[:, 10:138, 20:148, 30:158] + +# Prefer: let a few threads fetch the chunks ahead of decompression (once per process; +# BLOSC_PREFETCH=4 in the environment does the same) +blosc2.set_prefetch(4) +arr = blosc2.open(path, mmap_mode="r") +patch = arr[:, 10:138, 20:148, 30:158] +``` + +![set_prefetch vs no read-ahead](optim_tips/tip_16_prefetch.png) + +On a local NVMe disk, 40 cold patch reads were **~1.7x faster**; on NFS, cold patches of 3D images get 3-5x faster. Chunks already in the page cache are not read again, so warm reads cost about the same as without read-ahead. The threads mostly wait for the storage, but the kernel spends CPU on every byte they read, more so the more threads read one file at once: 4 threads with the default 1 MiB pieces give most of the speed on network and local storage alike; on local disks, `set_prefetch(2, piece_size=8 << 20)` costs less CPU for some of the speed. Of a chunk that `a[...]` or `slice()` covers only in part, only the blocks the slice needs are read. With many reader processes at once, 1-2 threads each are enough; when the CPU is the limit (tens of processes reading cold data on one node), read-ahead can be slower than none. The read-ahead threads are separate from the decompression threads (`nthreads`), so a program with many reader processes can keep `nthreads=1` and still use them. + +*Benchmark for this tip: [`tip_16_prefetch.py`](https://github.com/Blosc/python-blosc2/blob/main/bench/optim_tips/tip_16_prefetch.py)* + ## Use `.b2z` to group related data into one memory-mapped file `.b2z` packs related data — a {class}`~blosc2.CTable`'s columns and indexes, hierarchies of arrays, or both — into a single-file container that blosc2 reads *in place*: opened read-only, nothing is ever unpacked, and with `mmap_mode="r"` every member is memory-mapped at its offset inside the container. Beyond the convenience of one file to copy or archive, this buys you: diff --git a/doc/reference/low_level.rst b/doc/reference/low_level.rst index 154830b8b..1979c08a1 100644 --- a/doc/reference/low_level.rst +++ b/doc/reference/low_level.rst @@ -43,6 +43,7 @@ Set / get compression params register_filter set_blocksize set_nthreads + set_prefetch set_releasegil set_compressor get_compressor diff --git a/src/blosc2/__init__.py b/src/blosc2/__init__.py index 39008fd0b..315895465 100644 --- a/src/blosc2/__init__.py +++ b/src/blosc2/__init__.py @@ -491,6 +491,7 @@ def _raise(exc): set_blocksize, set_compressor, set_nthreads, + set_prefetch, set_releasegil, unpack, unpack_array, @@ -1116,6 +1117,7 @@ def _raise(exc): "set_blocksize", "set_compressor", "set_nthreads", + "set_prefetch", "set_releasegil", "sign", "signbit", diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 7a8548a0f..f55ef4723 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -267,6 +267,9 @@ cdef extern from "blosc2.h": void blosc2_set_threads_callback(blosc2_threads_callback callback, void *callback_data) int16_t blosc2_set_nthreads(int16_t nthreads) + int BLOSC2_PREFETCH_MAX_NTHREADS + int BLOSC2_PREFETCH_MIN_PIECE_SIZE + int blosc2_set_prefetch(int nthreads, int64_t max_ahead, int64_t piece_size) const char* blosc1_get_compressor() @@ -1259,6 +1262,13 @@ def free_resources(): if rc < 0: raise ValueError("Could not free the resources") +def set_prefetch(int nthreads, int64_t max_ahead, int64_t piece_size): + rc = blosc2_set_prefetch(nthreads, max_ahead, piece_size) + if rc < 0: + raise ValueError(f"nthreads must be in [0, {BLOSC2_PREFETCH_MAX_NTHREADS}], max_ahead must not be " + f"negative and piece_size must be 0 or at least {BLOSC2_PREFETCH_MIN_PIECE_SIZE}.") + return rc + def set_nthreads(nthreads): if nthreads > INT_MAX: raise ValueError("nthreads must be less or equal than 2^31 - 1.") diff --git a/src/blosc2/core.py b/src/blosc2/core.py index 21a4eafbc..28dd44c93 100644 --- a/src/blosc2/core.py +++ b/src/blosc2/core.py @@ -1323,6 +1323,69 @@ def set_nthreads(nthreads: int) -> int: return rc +def set_prefetch(nthreads: int, max_ahead: int = 0, piece_size: int = 0) -> int: + """Read ahead when slicing arrays and super-chunks stored as contiguous frames on disk. + + A slice of an on-disk array reads the blocks it needs one after the other, each read at most as large + as the kernel's read-ahead window. On a network file system every one of those reads is a round trip, + so a cold slice can wait for hundreds of them in a row. With read-ahead, the chunks a slice needs (and + that are not in the page cache yet) are read by :paramref:`nthreads` background threads in pieces of + at most :paramref:`piece_size` bytes, in the order the slice decompresses them and at most + :paramref:`max_ahead` bytes ahead of it, and each chunk is decompressed as soon as it is in. Of a + chunk that a slice (``a[...]``, :meth:`NDArray.slice`) covers only in part, only the blocks it needs + are read (after the chunk's header). The data goes to the page cache, so ``mmap_mode="r"`` stays + zero-copy, and cached pages are not read again. + + It applies to :class:`NDArray` slices (``a[...]``, :meth:`NDArray.slice`), orthogonal selections + (``a[[1, 5, 9], ...]``, :meth:`NDArray.get_oselection_numpy`) and :class:`SChunk` slices + (``schunk[a:b]``, :meth:`SChunk.get_slice`). + + Which settings to use: reads that are not cached get faster on any storage, most on high-latency + ones. The threads mostly wait for the storage, but the kernel spends CPU on every byte they read, and + more per byte the more threads read one file at once. + + * 4 threads with the default piece size give most of the speed, on network file systems (NFS, + SMB, ...) and on local disks; more threads help a little more on network storage and cost more + CPU. + * On local disks, fewer threads (1-2) or larger pieces (4-8 MiB) cost less CPU, for some of the + speed (how much depends on the disk). + * With many reader processes at once, 1-2 threads each are enough; when the CPU is the limit, + none may be better. + + Parameters + ---------- + nthreads: int + Number of read-ahead threads (at most 256); 0 turns read-ahead off (the default). + max_ahead: int + Most bytes read ahead of the chunk being decompressed; 0 for the default (64 MiB). + piece_size: int + Largest read handed to one thread, in bytes; 0 for the default (1 MiB), else at least 64 KiB. + Each thread keeps a buffer as large as the largest piece it has read. + + Returns + ------- + out: int + The previous number of read-ahead threads. + + Notes + ----- + The setting and the threads are per process; a process created with ``fork()`` starts its own threads + on first use. The ``BLOSC_PREFETCH`` environment variable, set to a number of threads, turns read-ahead + on when blosc2 is imported. It works with the default and the memory-mapped file I/O on POSIX systems + (Linux, macOS); on Windows it has no effect. Skipping cached pages asks the kernel (``cachestat()`` on + Linux >= 6.5, else ``mincore()``), which on Linux answers only for files the process owns or may write + to (before 6.5: only files it owns); other files are read again even if cached, which is correct but + costs a copy. + + Examples + -------- + >>> old = blosc2.set_prefetch(4) # network storage: 4 threads, 1 MiB pieces, up to 64 MiB ahead + >>> _ = blosc2.set_prefetch(2, piece_size=8 << 20) # less CPU: 2 threads, 8 MiB pieces + >>> _ = blosc2.set_prefetch(old) + """ + return blosc2_ext.set_prefetch(nthreads, max_ahead, piece_size) + + def compressor_list(plugins: bool = False) -> list: """ Returns a list of compressors (codecs) available in the C library. diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index aa058e044..c00db2d3b 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -3109,6 +3109,10 @@ def open( # noqa: C901 * When opening a :ref:`LazyExpr` keep in mind the note above regarding operands. + * Files on network storage (NFS, SMB, ...): turn on read-ahead with + :func:`blosc2.set_prefetch` (or the ``BLOSC_PREFETCH`` environment variable), so that + slices fetch the chunks they need in parallel instead of one block after the other. + Examples -------- >>> import blosc2 diff --git a/tests/test_prefetch.py b/tests/test_prefetch.py new file mode 100644 index 000000000..18af602cd --- /dev/null +++ b/tests/test_prefetch.py @@ -0,0 +1,164 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +import os +import signal +import sys +import time + +import numpy as np +import pytest + +import blosc2 + +# Slices: everything, inside one chunk, across chunks, one item, through the chunks that stay special values +SLICES = [ + np.s_[:], + np.s_[11:19, 21:39, 17:31], + np.s_[3:37, 7:44, 9:58], + np.s_[39, 49, 60], + np.s_[:, 30:, :30], +] + + +@pytest.fixture +def on_disk(tmp_path): + path = str(tmp_path / "prefetch.b2nd") + a = blosc2.zeros((40, 50, 61), dtype=np.int32, chunks=(10, 20, 16), blocks=(5, 8, 8), urlpath=path) + # data in part of the array only, so that the other chunks stay special values (zeros) in the index + values = np.arange(40 * 30 * 61, dtype=np.int64).reshape(40, 30, 61) * 2654435761 % 1000003 + a[:, :30] = values.astype(np.int32) + expected = a[:] + del a + return path, expected + + +def evict(path): + """Drop the file from the page cache, so that read-ahead has something to read (where the OS lets us).""" + if hasattr(os, "posix_fadvise"): + fd = os.open(path, os.O_RDONLY) + try: + os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED) + finally: + os.close(fd) + + +@pytest.fixture +def prefetch(): + yield blosc2.set_prefetch + blosc2.set_prefetch(0) + + +# (nthreads, max_ahead, piece_size): several threads; a budget smaller than any chunk; the smallest pieces; +# pieces larger than the file +SETTINGS = [(4, 0, 0), (1, 1, 0), (3, 0, 64 << 10), (2, 0, 4 << 20)] + + +def read_all_ways(path, mmap_mode): + """Every slice through the paths that read ahead: [], slice(), orthogonal selection; and SChunk slices.""" + a = blosc2.open(path, mode="r", mmap_mode=mmap_mode) + got = [] + for sl in SLICES: + got.append(a[sl]) + if all(isinstance(k, slice) for k in np.index_exp[sl]): + got.append(a.slice(sl)[:]) + key = (np.index_exp[sl] + (slice(None),) * 3)[:3] + got.append( + a.get_oselection_numpy( + [list(range(*k.indices(n))) for k, n in zip(key, a.shape, strict=True)] + ) + ) + nitems = a.schunk.nbytes // a.schunk.typesize + got.extend(a.schunk[start : min(start + 20000, nitems)] for start in (0, 7919, nitems - 5000)) + del a + return got + + +@pytest.mark.skipif(sys.platform == "win32", reason="read-ahead is a no-op on Windows") +@pytest.mark.parametrize("mmap_mode", [None, "r"]) +@pytest.mark.parametrize(("nthreads", "max_ahead", "piece_size"), SETTINGS) +def test_slices(on_disk, prefetch, mmap_mode, nthreads, max_ahead, piece_size): + path, _ = on_disk + prefetch(0) + expected = read_all_ways(path, mmap_mode) + prefetch(nthreads, max_ahead, piece_size) + evict(path) + for _ in range(2): # cold, then from the page cache + got = read_all_ways(path, mmap_mode) + assert len(got) == len(expected) + for g, e in zip(got, expected, strict=True): + if isinstance(e, bytes): + assert g == e + else: + np.testing.assert_array_equal(g, e) + + +@pytest.mark.skipif(sys.platform == "win32", reason="read-ahead is a no-op on Windows") +@pytest.mark.parametrize("mmap_mode", [None, "r"]) +def test_partial_chunks(tmp_path, prefetch, mmap_mode): + # Chunks of 2 MiB with 128 KiB blocks of random data, compressed with several threads (blocks stored out of + # order): of the chunks a slice covers in part, only the blocks it needs are read ahead, in several pieces + path = str(tmp_path / "partial.b2nd") + data = np.random.default_rng(0).random((64, 256, 256), dtype=np.float32) + blosc2.asarray( + data, chunks=(32, 128, 128), blocks=(8, 32, 128), urlpath=path, mode="w", cparams={"nthreads": 4} + ) + prefetch(4) + for sl in [np.s_[5:40, 100:150, 10:200], np.s_[31:33, :, 120:140], np.s_[:, 3:4, :]]: + evict(path) + a = blosc2.open(path, mode="r", mmap_mode=mmap_mode) + np.testing.assert_array_equal(a[sl], data[sl]) + np.testing.assert_array_equal(a.slice(sl)[:], data[sl]) + del a + + +def test_set_prefetch(prefetch): + assert prefetch(2) == 0 + assert prefetch(0, 1 << 20) == 2 + assert prefetch(1, 0, 64 << 10) == 0 + with pytest.raises(ValueError): + prefetch(-1) + with pytest.raises(ValueError): + prefetch(1, -1) + with pytest.raises(ValueError): + prefetch(1, 0, (64 << 10) - 1) + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs fork()") +def test_fork(on_disk, prefetch): + path, expected = on_disk + prefetch(4) + # One decompression thread: C-Blosc2's decompression threads do not survive fork() either, independently + # of read-ahead (a child decompressing with nthreads > 1 after the parent did hangs) + old_nthreads = blosc2.set_nthreads(1) + try: + evict(path) + blosc2.open(path, mode="r")[:] # the read-ahead threads run in this process now + evict(path) + _fork_and_read(path, expected) + finally: + blosc2.set_nthreads(old_nthreads) + + +def _fork_and_read(path, expected): + pid = os.fork() + if pid == 0: # the child has to start its own threads + try: + got = blosc2.open(path, mode="r", mmap_mode="r")[3:37, 7:44, 9:58] + os._exit(0 if np.array_equal(got, expected[3:37, 7:44, 9:58]) else 1) + except BaseException: + os._exit(2) + for _ in range(600): + done, status = os.waitpid(pid, os.WNOHANG) + if done: + break + time.sleep(0.1) + else: + os.kill(pid, signal.SIGKILL) + os.waitpid(pid, 0) + pytest.fail("the forked child hung") + assert os.waitstatus_to_exitcode(status) == 0 From 5def38ad13ad91fcf7d505ff855c55f5da3d3abe Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Wed, 4 Nov 2026 10:30:22 +0100 Subject: [PATCH 5/5] Restrict the prefetch fork test to native platforms Pyodide exposes os.fork as a stub that raises ENOSYS. Keep the read-ahead correctness tests enabled on WASM, but run the fork lifecycle test only where native fork is supported. Co-Authored-By: Codex --- tests/test_prefetch.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_prefetch.py b/tests/test_prefetch.py index 18af602cd..8ee342f12 100644 --- a/tests/test_prefetch.py +++ b/tests/test_prefetch.py @@ -128,7 +128,7 @@ def test_set_prefetch(prefetch): prefetch(1, 0, (64 << 10) - 1) -@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs fork()") +@pytest.mark.skipif(blosc2.IS_WASM or not hasattr(os, "fork"), reason="needs native fork()") def test_fork(on_disk, prefetch): path, expected = on_disk prefetch(4)