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 000000000..b769fa28a Binary files /dev/null and b/doc/guides/optim_tips/tip_16_prefetch.png differ diff --git a/doc/guides/optimization_tips.md b/doc/guides/optimization_tips.md index eb3cdc27f..718991592 100644 --- a/doc/guides/optimization_tips.md +++ b/doc/guides/optimization_tips.md @@ -514,6 +514,28 @@ With 8 concurrent readers doing random slice reads over a 269 MB array, mmap was *Benchmark for this tip: [`tip_10_mmap_many_readers.py`](https://github.com/Blosc/python-blosc2/blob/main/bench/optim_tips/tip_10_mmap_many_readers.py)* +## Reading cold data from disk? Turn on read-ahead + +A slice of an on-disk array reads the blocks it needs one after the other, and every read that misses the page cache waits for the storage. On a network file system (NFS, SMB, ...) each of those waits is a round trip to the server. {func}`blosc2.set_prefetch` starts a few background threads that fetch the chunks a slice needs in large, parallel requests, while the slice decompresses the chunks that have already arrived. It works for `a[...]`, {meth}`NDArray.slice `, 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 64affa623..f55ef4723 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -14,8 +14,11 @@ import ast import atexit import pathlib import sys +import functools import time import warnings +import weakref +from threading import RLock import _ctypes @@ -264,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() @@ -365,6 +371,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) @@ -500,6 +508,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, @@ -622,7 +632,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, @@ -872,7 +882,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): @@ -1246,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.") @@ -1707,11 +1730,105 @@ 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 + + +# 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 + 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 = _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) @@ -1881,6 +1998,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 +2032,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 +2076,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 +2092,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 +2120,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 +2169,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 +2203,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 +2215,32 @@ 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 cdef c_bool needs_free @@ -2113,6 +2264,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 +2275,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 +2293,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 +2304,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 +2335,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 +2346,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 +2354,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 +2385,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 +2401,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 +2426,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 +2532,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 +2550,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 +2567,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 +2600,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 +2635,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 +2659,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 +2689,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 +2725,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 +2808,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 +3498,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 +3520,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,15 +3534,18 @@ 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) if rc < 0: raise RuntimeError("Could not delete the vlmeta") + @_handle_locked 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 +3554,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 +3566,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 +3585,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 @@ -3401,10 +3594,12 @@ 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 +@_handle_locked def meta_keys(self): cdef blosc2_schunk *schunk = self.c_schunk keys = [] @@ -3631,12 +3826,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, @@ -3698,11 +3902,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: @@ -3814,24 +4025,40 @@ 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), 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: + 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 = _new_handle_lock() + if schunk is not None: + (schunk)._handle_lock = self._handle_lock @property def c_array(self): @@ -3889,6 +4116,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_ @@ -3899,22 +4127,31 @@ 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): + _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) 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 +4234,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 +4252,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 +4285,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 +4319,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 +4361,7 @@ cdef class NDArray: return ndarray + @_handle_locked def set_slice(self, key, ndarray): ndim = self.ndim start, stop = key @@ -4138,6 +4380,7 @@ cdef class NDArray: return self + @_handle_locked def tobytes(self): buffersize = self.size * self.array.sc.typesize buffer = bytes(buffersize) @@ -4146,6 +4389,7 @@ cdef class NDArray: return buffer + @_handle_locked def to_cframe(self): cdef c_bool needs_free cdef uint8_t *cframe @@ -4160,6 +4404,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 +4425,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 +4433,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 +4603,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 +4786,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 +4807,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 +4830,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 +4859,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 +5108,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 +5125,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 +5155,10 @@ 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 + 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") if copy: @@ -4916,6 +5169,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 +5187,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..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. @@ -1449,6 +1512,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/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/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/ndarray/test_concurrent_access.py b/tests/ndarray/test_concurrent_access.py new file mode 100644 index 000000000..186870e50 --- /dev/null +++ b/tests/ndarray/test_concurrent_access.py @@ -0,0 +1,227 @@ +####################################################################### +# 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 itertools +import os +import signal +import threading +import time + +import numpy as np +import pytest + +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): + """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 + + +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 write_region(array, k): + array[8:24, 0:100] = VALUES[8:24, 0:100] + (k % 2) + + +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}, + ) + + def longest_gap(): + """Longest time a ticker thread did not run during one read, as a fraction of the read.""" + 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] + 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))) == [] + + +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_vlmeta(a, k): + a.schunk.vlmeta[f"k{k % 5}"] = list(range(k % 50)) + + +def swap_dparams(a, k): + a.schunk.dparams = blosc2.DParams(nthreads=1 + k % 2) + + +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(): + 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) + + assert run_for(SECONDS, 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( + 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() 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) diff --git a/tests/test_prefetch.py b/tests/test_prefetch.py new file mode 100644 index 000000000..8ee342f12 --- /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(blosc2.IS_WASM or not hasattr(os, "fork"), reason="needs native 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