diff --git a/changes/274.bugfix b/changes/274.bugfix new file mode 100644 index 00000000..42f84be9 --- /dev/null +++ b/changes/274.bugfix @@ -0,0 +1,24 @@ +Several fixes to compressed block data reading: + +Fixed reading compressed arrays. Every access to a compressed block's data +after the first returned the raw compressed bytes instead of the decompressed +data, so that, for example, reading successive tiles with +``asdf_ndarray_read_tile_2d`` returned garbage and could crash reading past the +end of the compressed data. + +Fixed lazy decompression of LZ4-compressed arrays larger than 4 MB, which +returned zeros for all data past the first 4 MB, and could read past the end of +a buffer for the last, partial chunk. Tile reads now return +``ASDF_NDARRAY_ERR_OUT_OF_BOUNDS`` rather than reading out of bounds when the +array's data is smaller than its shape requires. + +Finally, fixed a race condition if multiple threads are reading the same +decompressed data buffer (e.g. as returned by `asdf_block_data` or +`asdf_ndarray_data`), and each encounter page faults. If multiple threads +cause page faults in the same range handled by a UFFD, the kernel queues these +and handles one at a time, even if the faulting address has had its page filled +by a previous fault handler call for one one of the other threads. In this +case the subsequent page faults, which have already been handled, should just +wake the faulting thread and allow it to resume. Instead it fell through to +attempting to copy the data again, resulting in an error (``EEXIST`` from +``ioctl``) causing the fault handler thread to exit. diff --git a/changes/276.bugfix b/changes/276.bugfix new file mode 100644 index 00000000..75d07816 --- /dev/null +++ b/changes/276.bugfix @@ -0,0 +1,3 @@ +Fixed auto-selection of decompression mode. Lazy decompression was only being +used if explicitly requested and not as the default on supported platforms +(i.e. Linux). diff --git a/src/block.c b/src/block.c index 31c7d2cb..0b2c85e8 100644 --- a/src/block.c +++ b/src/block.c @@ -686,8 +686,23 @@ const void *asdf_block_data_impl(asdf_block_t *block, size_t *size, bool decompr if (!block) return NULL; - /* Cached data from a previous stream open */ + /* Cached data from a previous stream open; block->data is always the raw + * (possibly compressed) data, so return the decompressed data if asked */ if (block->data && block->should_close) { + if (decompress) { + if (!block->comp_state && asdf_block_comp_open(block) != 0) { + ASDF_LOG(block->file, ASDF_LOG_ERROR, "failed to open compressed block data"); + return NULL; + } + + if (block->comp_state) { + if (size) + *size = block->comp_state->dest_size; + + return block->comp_state->dest; + } + } + if (size) *size = block->avail_size; diff --git a/src/compression/compression.c b/src/compression/compression.c index b6ace1c3..31139753 100644 --- a/src/compression/compression.c +++ b/src/compression/compression.c @@ -95,6 +95,28 @@ static int asdf_create_temp_file(size_t data_size, const char *tmp_dir, int *out } +/** + * Decompress a chunk of a block into the state's work buffer + * + * Thin wrapper around the compressor's ``decomp`` callback, used by both + * eager and lazy decompression. In eager mode the work buffer is the whole + * destination mapping, so one call decompresses the entire block in place; in + * lazy mode it is a separate chunk-sized buffer (see handle_pagefault). + * + * ``offset_hint`` is the offset into the decompressed data that the caller + * needs, and on success ``*offset_out`` is set to the offset of the chunk + * actually written to the work buffer, which is not necessarily the chunk + * containing ``offset_hint``. zlib and bzip2 can only decompress + * sequentially, so they always produce the next chunk after the previous call + * and the caller must keep calling until it gets the chunk it needs. Other + * schemes such as ASDF's LZ4 format allow random access (up to the LZ4 block + * size), though that is not implemented yet (see #90). + * + * Currently, if ``offset_hint`` is before data the compressor has already + * produced, it returns 0 without writing anything or setting ``*offset_out``. + * + * Returns 0 on success or a compressor-specific nonzero error code. + */ static int asdf_block_decomp_offset( asdf_block_comp_state_t *state, size_t *offset_out, size_t offset_hint) { assert(state); @@ -252,6 +274,29 @@ static inline asdf_block_comp_uffd_evt_t wait_for_event( } +/** + * Resolve a userfaultfd page fault by decompressing the faulting page + * + * Called from asdf_block_comp_userfaultfd_handler with a fault on an address + * in the destination mapping ``state->dest``. This decompresses chunks of + * ``state->work_buf_size`` bytes (a whole number of pages) via + * asdf_block_decomp_offset and copies each one into the mapping at the offset + * the compressor reports, using UFFDIO_COPY, until the chunk containing the + * faulting page has been copied. Chunks copied before then use + * UFFDIO_COPY_MODE_DONTWAKE; only the copy that fills the faulting page wakes + * the faulting thread. A partial last chunk is rounded up to a whole page, + * padded with zeros from the work buffer. + * + * The faulting page may already have been filled by the time its fault is + * handled: when several threads fault at once, their faults are queued, and + * handling one may fill pages that other threads are waiting on. Pages + * filled with DONTWAKE do not wake those threads (nor discard their queued + * faults), so if the compressor produces nothing for the faulting page this + * just wakes the faulting thread with UFFDIO_WAKE. + * + * Returns 0 on success, the decompressor's nonzero error code, or -1 from a + * failed ioctl with errno set. + */ static int handle_pagefault( asdf_block_comp_userfaultfd_t *uffd, struct uffd_msg *msg, size_t page_size) { asdf_block_comp_state_t *state = uffd->comp_state; @@ -262,8 +307,22 @@ static int handle_pagefault( size_t offset = (fault_addr - (uintptr_t)state->dest) & ~page_mask; int ret = 0; + ASDF_LOG( + state->file, + ASDF_LOG_TRACE, + "lazy decompression %p: page fault at offset %zu", + (void *)state->dest, + offset); + while (!atomic_load(&uffd->stop)) { - size_t got_offset = 0; + // Sentinel for the compressor producing nothing, because the data at + // the requested offset was already produced. This used to default to + // got_offset = 0 but then there was no way to distinguish from + // asdf_block_decomp_offset doing nothing and returning without setting + // got_offset, versus returning that the chunk at offset 0 was + // decompressed (return value is 0 in either case; may be better to + // have a separate return value indicating 'already done'...?) + size_t got_offset = SIZE_MAX; memset(state->work_buf, 0, chunk_size); ret = asdf_block_decomp_offset(state, &got_offset, offset); @@ -271,13 +330,34 @@ static int handle_pagefault( if (ret != 0) return ret; + if (got_offset == SIZE_MAX) { + // The faulting page is already filled; just wake the faulting + // thread + ASDF_LOG( + state->file, + ASDF_LOG_TRACE, + "lazy decompression %p: page at offset %zu already filled; waking faulting " + "thread", + (void *)state->dest, + offset); + struct uffdio_range wake_range = {.start = fault_addr, .len = page_size}; + return ioctl(uffd->uffd, UFFDIO_WAKE, &wake_range); + } + size_t dst = (uintptr_t)state->dest + got_offset; size_t src = (size_t)uffd->work_buf; - size_t len = - ((got_offset + chunk_size) > state->dest_size ? (state->dest_size - got_offset) - : chunk_size); + size_t dest_size = state->dest_size; + size_t next_offset = got_offset + chunk_size; + /* Here len is the size of the chunk read from the decompression buffer: + * usually just chunk_size, or otherwise the trailing bytes at the last + * page */ + size_t len = next_offset > dest_size ? (dest_size - got_offset) : chunk_size; + /* align to page size */ len = (len + page_size - 1) & ~page_mask; - uint64_t mode = (got_offset == offset) ? 0 : UFFDIO_COPY_MODE_DONTWAKE; + /* The decompressed chunk contains the faulting address; if not we must + * continue decompression until it does */ + bool has_fault = offset >= got_offset && offset - got_offset < len; + uint64_t mode = has_fault ? 0 : UFFDIO_COPY_MODE_DONTWAKE; struct uffdio_copy uffd_copy = { // Copy back to the (page-aligned) destination in the user's mmap @@ -295,7 +375,16 @@ static int handle_pagefault( if (ret != 0) return ret; - if (got_offset == offset) + ASDF_LOG( + state->file, + ASDF_LOG_TRACE, + "lazy decompression %p: filled [%zu, %zu)%s", + (void *)state->dest, + got_offset, + got_offset + len, + has_fault ? " and woke faulting thread" : ""); + + if (has_fault) break; } @@ -303,6 +392,25 @@ static int handle_pagefault( } +/** Main userfaultfd fault handler thread + * + * The way UFFD works in UFFDIO_REGISTER_MODE_MISSING (as configured via ioctl + * in asdf_block_decomp_lazy), is if a page fault occurs the faulting thread + * (typically the main thread) is blocked until another ioctl is received + * *on* the same UFFD file descriptor with UFFDIO_COPY or UFFDIO_ZEROPAGE, + * signaling that the fault has been handled in some way (either copying of + * data into the faulted page address which we do here, or zeroing the page). + * + * The UFFD mechanism itself does not know anything about this thread--it just + * runs in the background and blocks (via poll()) until the UFFD is ready, or + * a separate 'event' file descriptor that we manage which signals the thread + * to stop. The ``wait_for_events()`` function implements the polling, and + * when the event is a UFFD page fault, we pass to ``handle_pagefault()`` also + * defined above. It fills the faulting page with data from the decompressor + * and sets the UFFDIO_COPY ioctl, unblocking the main thread and allowing it + * to continue from before the page fault--it will retry reading the now filled + * memory page and continue. + */ static void *asdf_block_comp_userfaultfd_handler(void *arg) { asdf_block_comp_userfaultfd_t *uffd = arg; asdf_block_comp_state_t *state = uffd->comp_state; @@ -317,16 +425,30 @@ static void *asdf_block_comp_userfaultfd_handler(void *arg) { fds[1].fd = uffd->evtfd; fds[1].events = POLLIN; + // Reason the handler stopped, for logging + const char *reason = "shutdown requested"; + + ASDF_LOG( + state->file, + ASDF_LOG_TRACE, + "lazy decompression %p: handler started (%zu bytes, %zu-byte chunks)", + (void *)state->dest, + state->dest_size, + state->work_buf_size); + // Wait for an event, exiting if we get a stop signal or a ready event (or error) while (!atomic_load(&uffd->stop)) { info = state->compressor->info(state->userdata); - if (!info || info->status == ASDF_COMPRESSOR_DONE) + if (!info || info->status == ASDF_COMPRESSOR_DONE) { + reason = "decompression complete"; break; + } evt = wait_for_event(fds, &msg); switch (evt) { case ASDF_BLOCK_COMP_UFFD_EVT_ERROR: + reason = "error waiting for events"; ASDF_ERROR_SYSTEM(state->file, errno); break; case ASDF_BLOCK_COMP_UFFD_EVT_NOT_READY: @@ -335,6 +457,7 @@ static void *asdf_block_comp_userfaultfd_handler(void *arg) { break; case ASDF_BLOCK_COMP_UFFD_EVT_PAGEFAULT: if (handle_pagefault(uffd, &msg, page_size)) { + reason = "error handling page fault"; ASDF_ERROR_SYSTEM(state->file, errno); break; } @@ -347,6 +470,13 @@ static void *asdf_block_comp_userfaultfd_handler(void *arg) { break; } + ASDF_LOG( + state->file, + ASDF_LOG_TRACE, + "lazy decompression %p: handler exiting: %s", + (void *)state->dest, + reason); + // If the thread exits for any reason make sure to deregister UFFD for // for the range handled; of course if the thread crashes we're in deep // water (main thread may hang) @@ -358,7 +488,7 @@ static void *asdf_block_comp_userfaultfd_handler(void *arg) { } -#define FEATURE_IS_SET(bits, bit) (((bits) & (bit)) == (bit)) +#define FEATURE_IS_SET(bits, bit) (((bits) & (1ULL << (bit))) != 0) /** * Probe the running kernel for basic userfaultfd support. @@ -755,16 +885,14 @@ static asdf_block_comp_state_t *asdf_block_comp_state_create( "using eager decompression mode, since lazy mode is not possible when " "decompressing to a temp file on disk"); use_lazy_mode = false; - mode = ASDF_BLOCK_DECOMP_MODE_EAGER; } else { // If the user explicitly requested lazy mode, disable file backing instead ASDF_LOG( block->file, ASDF_LOG_WARN, - "using eager decompression mode, since lazy mode is not possible when " - "decompressing to a temp file on disk"); + "lazy decompression mode requested, which is not possible when " + "decompressing to a temp file on disk; decompressing in memory instead"); use_file_backing = false; - mode = ASDF_BLOCK_DECOMP_MODE_LAZY; } } else if (!use_lazy_mode && mode == ASDF_BLOCK_DECOMP_MODE_LAZY) { ASDF_LOG( @@ -772,10 +900,13 @@ static asdf_block_comp_state_t *asdf_block_comp_state_create( ASDF_LOG_WARN, "lazy decompression mode requested, but the runtime check for kernel " "support failed"); - mode = ASDF_BLOCK_DECOMP_MODE_EAGER; } } + // Resolve AUTO (and LAZY when not possible) to the effective mode; only + // the effective mode is stored on the state + mode = use_lazy_mode ? ASDF_BLOCK_DECOMP_MODE_LAZY : ASDF_BLOCK_DECOMP_MODE_EAGER; + if (use_file_backing && !use_lazy_mode) { if (asdf_create_temp_file(dest_size, config->decomp.tmp_dir, &state->fd) != 0) { free(state); diff --git a/src/compression/lz4.c b/src/compression/lz4.c index 43f60e62..cfc4dd50 100644 --- a/src/compression/lz4.c +++ b/src/compression/lz4.c @@ -51,6 +51,8 @@ typedef struct { asdf_file_t *file; uint8_t *data; size_t data_size; + /** Total size of the decompressed data */ + size_t dest_size; size_t pos; size_t progress; @@ -65,8 +67,9 @@ typedef struct { /** * Intermediate buffer for decompressed blocks * - * These are always the same size so no need for a separate capacity (except for the last - * block which may be smaller) + * These are normally all the same size (except for the last block which + * may be smaller), but the capacity is tracked separately in case a + * malformed file declares a later block larger than the first. * * Later will maybe have a collection of these since for some cases, like tile reading, * it will be useful to have multiple fully decompressed but partially read blocks in memory @@ -74,6 +77,7 @@ typedef struct { */ struct { uint8_t *buf; + size_t cap; size_t size; size_t pos; } block; @@ -114,7 +118,7 @@ static int asdf_compressor_lz4_read_header(asdf_compressor_lz4_userdata_t *lz4) static asdf_compressor_userdata_t *asdf_compressor_lz4_init( - const asdf_block_t *block, UNUSED(const void *dest), UNUSED(size_t dest_size)) { + const asdf_block_t *block, UNUSED(const void *dest), size_t dest_size) { asdf_compressor_lz4_userdata_t *userdata = NULL; userdata = calloc(1, sizeof(asdf_compressor_lz4_userdata_t)); @@ -128,6 +132,7 @@ static asdf_compressor_userdata_t *asdf_compressor_lz4_init( userdata->file = block->file; userdata->data = block->data; userdata->data_size = block->avail_size; + userdata->dest_size = dest_size; userdata->pos = 0; userdata->progress = 0; @@ -172,8 +177,13 @@ static const asdf_compressor_info_t *asdf_compressor_lz4_info( static int asdf_compresser_lz4_read_block(asdf_compressor_lz4_userdata_t *lz4) { - if (asdf_compressor_lz4_read_header(lz4) != 0) - return 0; + if (asdf_compressor_lz4_read_header(lz4) != 0) { + ASDF_LOG( + lz4->file, + ASDF_LOG_ERROR, + "LZ4 compressed data ended before the full decompressed size was reached"); + return -1; + } if (lz4->header.block_size <= 0) { // Zero-width compressed block encountered--I guess done? @@ -192,17 +202,30 @@ static int asdf_compresser_lz4_read_block(asdf_compressor_lz4_userdata_t *lz4) { return -1; } + if ((size_t)lz4->header.block_size > lz4->data_size - lz4->pos) { + ASDF_LOG( + lz4->file, + ASDF_LOG_ERROR, + "LZ4 block size (%d) exceeds the remaining compressed data (%zu); aborting " + "decompression", + lz4->header.block_size, + lz4->data_size - lz4->pos); + return -1; + } + // Allocate buffer for decompressed block - // This is always the same size through the entire compressed block - // except possibly for the last LZ4 block which may be smaller but - // never larger - if (!lz4->block.buf) { - lz4->block.buf = malloc(lz4->header.decomp_block_size); + // This is normally the same size through the entire compressed block + // except possibly for the last LZ4 block which may be smaller + if ((size_t)lz4->header.decomp_block_size > lz4->block.cap) { + uint8_t *buf = realloc(lz4->block.buf, lz4->header.decomp_block_size); - if (!lz4->block.buf) { + if (!buf) { ASDF_ERROR_OOM(lz4->file); return -1; } + + lz4->block.buf = buf; + lz4->block.cap = lz4->header.decomp_block_size; } lz4->block.size = lz4->header.decomp_block_size; @@ -219,6 +242,16 @@ static int asdf_compresser_lz4_read_block(asdf_compressor_lz4_userdata_t *lz4) { return ret; } + if ((size_t)ret != lz4->block.size) { + ASDF_LOG( + lz4->file, + ASDF_LOG_ERROR, + "LZ4 block decompressed to %d bytes, but its header declares %zu bytes", + ret, + lz4->block.size); + return -1; + } + lz4->pos += lz4->header.block_size; lz4->header.pos = 0; return 0; @@ -326,7 +359,9 @@ static int asdf_compressor_lz4_decomp( if (offset_hint < lz4->progress) return 0; - size_t need = buf_size; + // The last chunk may be partial; the remainder of buf is left untouched + size_t remaining = lz4->dest_size - lz4->progress; + size_t need = buf_size < remaining ? buf_size : remaining; size_t pos = 0; while (need > 0) { @@ -339,6 +374,14 @@ static int asdf_compressor_lz4_decomp( return ret; avail = lz4->block.size; + + if (avail == 0) { + ASDF_LOG( + lz4->file, + ASDF_LOG_ERROR, + "empty LZ4 block encountered; aborting decompression"); + return -1; + } } size_t take = avail < need ? avail : need; @@ -353,7 +396,7 @@ static int asdf_compressor_lz4_decomp( lz4->progress += buf_size; - if (lz4->progress >= (uint32_t)lz4->header.decomp_block_size) + if (lz4->progress >= lz4->dest_size) lz4->info.status = ASDF_COMPRESSOR_DONE; return 0; diff --git a/src/core/ndarray.c b/src/core/ndarray.c index 5ea44fd7..918aa7ca 100644 --- a/src/core/ndarray.c +++ b/src/core/ndarray.c @@ -1897,7 +1897,9 @@ asdf_ndarray_err_t asdf_ndarray_read_tile_ndim( size_t data_size = 0; const void *data = asdf_ndarray_data(ndarray, &data_size); - if (data_size < src_tile_size) + // The tile is already known to lie within the array's shape, so it suffices to check that + // the data actually covers the full array (not just a tile's worth of bytes anywhere in it) + if (data_size < src_tile_size || data_size < asdf_ndarray_nbytes(ndarray)) return ASDF_NDARRAY_ERR_OUT_OF_BOUNDS; // Determine the copy strategy to use; right now this just handles whether-or-not byteswap diff --git a/src/core/ndarray_convert.c b/src/core/ndarray_convert.c index 7648f581..e28a7914 100644 --- a/src/core/ndarray_convert.c +++ b/src/core/ndarray_convert.c @@ -97,7 +97,7 @@ static atomic_bool conversion_table_initialized = false; const char *_src = (const char *)src; \ for (size_t idx = 0; idx < count; idx++) { \ src_t val; \ - LOAD_UNALIGNED(val, _src + idx * sizeof(src_t)); \ + LOAD_UNALIGNED(val, _src + (idx * sizeof(src_t))); \ _DO_BSWAP_##bswap(src_t, val); \ _dst[idx] = (dst_t)val; \ } \ @@ -119,7 +119,7 @@ static atomic_bool conversion_table_initialized = false; int overflow = 0; \ for (size_t idx = 0; idx < count; idx++) { \ src_t val; \ - LOAD_UNALIGNED(val, _src + idx * sizeof(src_t)); \ + LOAD_UNALIGNED(val, _src + (idx * sizeof(src_t))); \ _DO_BSWAP_##bswap(src_t, val); \ if (val < (src_t)(minval)) { \ _dst[idx] = minval; \ @@ -151,7 +151,7 @@ static atomic_bool conversion_table_initialized = false; int overflow = 0; \ for (size_t idx = 0; idx < count; idx++) { \ src_t val; \ - LOAD_UNALIGNED(val, _src + idx * sizeof(src_t)); \ + LOAD_UNALIGNED(val, _src + (idx * sizeof(src_t))); \ _DO_BSWAP_##bswap(src_t, val); \ if (isinf(val)) { \ _dst[idx] = (dst_t)val; \ @@ -182,7 +182,7 @@ static atomic_bool conversion_table_initialized = false; int overflow = 0; \ for (size_t idx = 0; idx < count; idx++) { \ src_t val; \ - LOAD_UNALIGNED(val, _src + idx * sizeof(src_t)); \ + LOAD_UNALIGNED(val, _src + (idx * sizeof(src_t))); \ _DO_BSWAP_##bswap(src_t, val); \ if (val < (src_t)(minval)) { \ _dst[idx] = (half)(-INFINITY); \ @@ -213,7 +213,7 @@ static atomic_bool conversion_table_initialized = false; int overflow = 0; \ for (size_t idx = 0; idx < count; idx++) { \ src_t val; \ - LOAD_UNALIGNED(val, _src + idx * sizeof(src_t)); \ + LOAD_UNALIGNED(val, _src + (idx * sizeof(src_t))); \ _DO_BSWAP_##bswap(src_t, val); \ if (val > (src_t)(maxval)) { \ _dst[idx] = (half)(INFINITY); \ @@ -250,7 +250,7 @@ static atomic_bool conversion_table_initialized = false; int overflow = 0; \ for (size_t idx = 0; idx < count; idx++) { \ half hval; \ - LOAD_UNALIGNED(hval, _src + idx * sizeof(half)); \ + LOAD_UNALIGNED(hval, _src + (idx * sizeof(half))); \ _DO_BSWAP_##bswap(half, hval); \ float val = (float)hval; \ if (isnan(val)) { \ @@ -279,7 +279,7 @@ static atomic_bool conversion_table_initialized = false; int overflow = 0; \ for (size_t idx = 0; idx < count; idx++) { \ src_t val; \ - LOAD_UNALIGNED(val, _src + idx * sizeof(src_t)); \ + LOAD_UNALIGNED(val, _src + (idx * sizeof(src_t))); \ _DO_BSWAP_##bswap(src_t, val); \ if (val > (src_t)(maxval)) { \ _dst[idx] = maxval; \ @@ -303,7 +303,7 @@ static atomic_bool conversion_table_initialized = false; int overflow = 0; \ for (size_t idx = 0; idx < count; idx++) { \ src_t val; \ - LOAD_UNALIGNED(val, csrc + idx * sizeof(src_t)); \ + LOAD_UNALIGNED(val, csrc + (idx * sizeof(src_t))); \ _DO_BSWAP_##bswap(src_t, val); \ if (val < (src_t)(0)) { \ val = (src_t)(0); \ @@ -343,7 +343,7 @@ static atomic_bool conversion_table_initialized = false; #define DEFINE_IDENTITY_CONVERSION(src_name, src_t) \ static int convert_##src_name##_to_##src_name( \ void *dst, const void *src, size_t nelem, size_t elsize) { \ - memcpy(dst, src, nelem *elsize); \ + memcpy(dst, src, (nelem * elsize)); \ return 0; \ } \ _DEFINE_GENERIC_CONV_FN(src_t, src_t, src_name##_to_##src_name##_bswap, 1) diff --git a/tests/test-compression.c b/tests/test-compression.c index 07bc144f..070c5458 100644 --- a/tests/test-compression.c +++ b/tests/test-compression.c @@ -2,6 +2,7 @@ #define _GNU_SOURCE /* for memmem */ #endif #include +#include #include #include #include @@ -18,6 +19,19 @@ #include "config.h" #include "file.h" +#if defined(HAVE_USERFAULTFD) && defined(HAVE_DECL_SYS_USERFAULTFD) +#include +#include +#include +#include +#include + +#ifndef UFFD_USER_MODE_ONLY +/* Missing from older Linux headers; see the same fallback in compression.c */ +#define UFFD_USER_MODE_ONLY 1 +#endif +#endif + #include "munit.h" #include "util.h" @@ -38,9 +52,9 @@ */ static char *comp_params[] = {"zlib", "bzp2", "lz4", NULL}; #ifdef ASDF_BLOCK_DECOMP_LAZY_AVAILABLE -static char *mode_params[] = {"eager", "lazy", NULL}; +static char *mode_params[] = {"auto", "eager", "lazy", NULL}; #else -static char *mode_params[] = {"eager", NULL}; +static char *mode_params[] = {"auto", "eager", NULL}; #endif static MunitParameterEnum comp_mode_test_params[] = { {"comp", comp_params}, @@ -55,7 +69,31 @@ static MunitParameterEnum comp_test_params[] = { }; +/** + * Lazy decompression chunk sizes, in pages, or "default" to leave + * decomp.chunk_size unset + */ +static char *chunk_params[] = {"1", "4", "default", NULL}; +static MunitParameterEnum comp_mode_chunk_test_params[] = { + {"comp", comp_params}, + {"mode", mode_params}, + {"chunk", chunk_params}, + {NULL, NULL} +}; + + +static size_t chunk_size_from_param(const char *chunk) { + if (strcmp(chunk, "default") == 0) + return 0; + + return strtoul(chunk, NULL, 10) * (size_t)sysconf(_SC_PAGESIZE); +} + + static asdf_block_decomp_mode_t decomp_mode_from_param(const char *mode) { + if (strcmp(mode, "auto") == 0) + return ASDF_BLOCK_DECOMP_MODE_AUTO; + if (strcmp(mode, "eager") == 0) return ASDF_BLOCK_DECOMP_MODE_EAGER; @@ -169,6 +207,20 @@ static int test_compressed_file( fisher_yates_shuffle(pages, num_pages); + // Always read page 1 first: with lazy decompression in chunks of + // more than one page, the first page fault is then guaranteed not to + // be on the first page of a chunk, ensuring better determinism in + // failure modes. The bug in #274 is triggered particularly when the + // decompression chunk_size is > page_size, and the first fault falls + // on a page other than the first (0-th) page within that chunk. + for (int idx = 0; idx < num_pages; idx++) { + if (pages[idx] == 1) { + pages[idx] = pages[0]; + pages[0] = 1; + break; + } + } + for (int idx = 0; idx < num_pages; idx++) { size_t page_idx = pages[idx]; //munit_logf(MUNIT_LOG_DEBUG, "checking page %zu\n", page_idx); @@ -179,6 +231,13 @@ static int test_compressed_file( free(pages); } + // In lazy mode errors in the decompression handler only occur once the + // data is actually read, so check again + error = asdf_error(file); + if (error) + munit_logf(MUNIT_LOG_ERROR, "error after reading the ndarray data: %s", error); + assert_null(error); + const asdf_block_t *block = asdf_ndarray_block(ndarray); assert_not_null(block); assert_not_null(block->comp_state); @@ -277,6 +336,211 @@ MU_TEST(read_compressed_block_to_file_on_threshold) { } +/* + * Probes for whether lazy decompression is possible on this system + * + * These mirror the library's internal (and not exported) runtime checks, so + * that the tests can determine which decompression mode the library should + * select. + */ +#if defined(HAVE_USERFAULTFD) && defined(HAVE_DECL_SYS_USERFAULTFD) +/** + * Open a userfaultfd and perform the API handshake; returns -1 on failure + * + * If ``ioctls`` is non-``NULL`` it receives the bitmask of ioctls supported + * by the kernel, as returned by the handshake. + */ +static int open_uffd(uint64_t *ioctls) { + struct uffdio_api uffd_api = {.api = UFFD_API}; + long uffd = syscall(SYS_userfaultfd, O_CLOEXEC | O_NONBLOCK | UFFD_USER_MODE_ONLY); + + // Pre-5.11 kernels reject the unknown flag; retry without it + if (uffd < 0 && errno == EINVAL) + uffd = syscall(SYS_userfaultfd, O_CLOEXEC | O_NONBLOCK); + + if (uffd < 0) + return -1; + + if (ioctl((int)uffd, UFFDIO_API, &uffd_api) == -1) { + close((int)uffd); + return -1; + } + + if (ioctls) + *ioctls = uffd_api.ioctls; + + return (int)uffd; +} + + +/** + * Test whether the kernel supports lazy decompression with userfaultfd + */ +static bool lazy_available(void) { + uint64_t ioctls = 0; + int uffd = open_uffd(&ioctls); + + if (uffd < 0) + return false; + + close(uffd); + // The ioctls bitmask is indexed by the _UFFDIO_* bit numbers + return (ioctls & (1ULL << _UFFDIO_REGISTER)) != 0; +} + + +/** + * Test whether userfaultfd can handle a mapping of a temp file in the default + * temp dir + * + * This generally only succeeds if the temp dir is on a memory-backed + * filesystem such as tmpfs. The result determines whether an explicit + * request for lazy decompression takes precedence over decompressing to a + * temp file. + */ +static bool lazy_file_backing_available(void) { + size_t page_size = sysconf(_SC_PAGESIZE); + const char *tmp_dir = getenv("ASDF_TMPDIR"); + char path[4096]; + bool available = false; + int uffd = -1; + int fd = -1; + void *map = MAP_FAILED; + struct uffdio_register uffd_reg = {0}; + + if (!tmp_dir || !tmp_dir[0]) + tmp_dir = getenv("TMPDIR"); + + if (!tmp_dir || !tmp_dir[0]) + tmp_dir = "/tmp"; + + uffd = open_uffd(NULL); + + if (uffd < 0) + goto finish; + + snprintf(path, sizeof(path), "%s/libasdf-test-XXXXXX", tmp_dir); + fd = mkstemp(path); + + if (fd < 0) + goto finish; + + unlink(path); + + if (ftruncate(fd, (off_t)page_size) != 0) + goto finish; + + map = mmap(NULL, page_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); + + if (map == MAP_FAILED) + goto finish; + + uffd_reg.range.start = (uintptr_t)map; + uffd_reg.range.len = page_size; + uffd_reg.mode = UFFDIO_REGISTER_MODE_MISSING; + available = ioctl(uffd, UFFDIO_REGISTER, &uffd_reg) == 0; + +finish: + if (map != MAP_FAILED) + munmap(map, page_size); + + if (fd >= 0) + close(fd); + + if (uffd >= 0) + close(uffd); + + return available; +} +#else +static bool lazy_available(void) { + return false; +} + + +static bool lazy_file_backing_available(void) { + return false; +} +#endif + + +/** + * Test that the effective decompression mode is selected correctly + * + * Eager mode is always eager; auto and lazy modes are lazy if and only if + * lazy decompression is available on this system, and otherwise fall back to + * eager. + * + * Regression test for #276. + */ +MU_TEST(decomp_mode_selection) { + const char *comp = munit_parameters_get(params, "comp"); + const char *filename = get_fixture_file_path("compressed.asdf"); + asdf_block_decomp_mode_t mode = decomp_mode_from_param(munit_parameters_get(params, "mode")); + asdf_config_t config = {.decomp = {.mode = mode}}; + asdf_block_decomp_mode_t expected = ASDF_BLOCK_DECOMP_MODE_EAGER; + + if (mode != ASDF_BLOCK_DECOMP_MODE_EAGER && lazy_available()) + expected = ASDF_BLOCK_DECOMP_MODE_LAZY; + + asdf_file_t *file = asdf_open_file_ex(filename, "r", &config); + assert_not_null(file); + asdf_ndarray_t *ndarray = NULL; + assert_int(asdf_get_ndarray(file, comp, &ndarray), ==, ASDF_VALUE_OK); + assert_not_null(ndarray); + assert_not_null(asdf_ndarray_data(ndarray, NULL)); + const asdf_block_t *block = asdf_ndarray_block(ndarray); + assert_not_null(block); + assert_not_null(block->comp_state); + assert_int(block->comp_state->mode, ==, expected); + asdf_ndarray_destroy(ndarray); + asdf_close(file); + return MUNIT_OK; +} + + +/** + * Test the effective decompression mode when decompressing to a temp file + * + * Lazy decompression is incompatible with decompressing to a temp file, so + * in auto mode decompression to a temp file takes precedence and is eager, + * whereas in lazy mode lazy decompression (in memory) takes precedence if it + * is available, including the check that userfaultfd works with the temp + * dir (see lazy_file_backing_available). + * + * Regression test for #276. + */ +MU_TEST(decomp_mode_selection_to_file) { + const char *comp = munit_parameters_get(params, "comp"); + const char *filename = get_fixture_file_path("compressed.asdf"); + asdf_block_decomp_mode_t mode = decomp_mode_from_param(munit_parameters_get(params, "mode")); + asdf_config_t config = {.decomp = {.mode = mode, .max_memory_bytes = 1}}; + asdf_block_decomp_mode_t expected = ASDF_BLOCK_DECOMP_MODE_EAGER; + bool expect_own_fd = true; + + if (mode == ASDF_BLOCK_DECOMP_MODE_LAZY && lazy_available() && + lazy_file_backing_available()) { + expected = ASDF_BLOCK_DECOMP_MODE_LAZY; + expect_own_fd = false; + } + + asdf_file_t *file = asdf_open_file_ex(filename, "r", &config); + assert_not_null(file); + asdf_ndarray_t *ndarray = NULL; + assert_int(asdf_get_ndarray(file, comp, &ndarray), ==, ASDF_VALUE_OK); + assert_not_null(ndarray); + assert_not_null(asdf_ndarray_data(ndarray, NULL)); + const asdf_block_t *block = asdf_ndarray_block(ndarray); + assert_not_null(block); + assert_not_null(block->comp_state); + assert_int(block->comp_state->mode, ==, expected); + assert_int(block->comp_state->own_fd, ==, expect_own_fd); + asdf_ndarray_destroy(ndarray); + asdf_close(file); + return MUNIT_OK; +} + + /** * Test opening a compressed block in lazy read mode, but without reading it * @@ -388,6 +652,12 @@ MU_TEST(compressed_block_no_hang_on_segfault) { /** * Test decompressed block lazy random access + * + * The pages of the array are read in random order, so with chunks larger + * than one page the first fault in a chunk is usually not on its first page. + * With the default chunk size the whole array fits in one chunk. + * + * Regression test for #274. */ MU_TEST(read_compressed_block_lazy_random_access) { const char *comp = munit_parameters_get(params, "comp"); @@ -397,7 +667,7 @@ MU_TEST(read_compressed_block_lazy_random_access) { .decomp = { // Run the test in eager mode too as a control .mode = decomp_mode_from_param(munit_parameters_get(params, "mode")), - .chunk_size = 4096 + .chunk_size = chunk_size_from_param(munit_parameters_get(params, "chunk")) } }; @@ -408,6 +678,148 @@ MU_TEST(read_compressed_block_lazy_random_access) { } +/* + * The concurrent fault test needs lazy decompression with userfaultfd (so it + * is Linux-only anyway) and pthread barriers, which are optional in POSIX and + * notably missing on macOS; where either is unavailable the test is skipped. + */ +#if defined(HAVE_USERFAULTFD) && defined(HAVE_DECL_SYS_USERFAULTFD) && \ + defined(_POSIX_BARRIERS) && _POSIX_BARRIERS > 0 +/** Argument for concurrent_fault_thread */ +typedef struct { + pthread_barrier_t *barrier; + const volatile uint8_t *addr; + uint8_t value; +} concurrent_fault_arg_t; + + +static void *concurrent_fault_thread(void *arg) { + concurrent_fault_arg_t *fault_arg = arg; + pthread_barrier_wait(fault_arg->barrier); + fault_arg->value = *fault_arg->addr; + return NULL; +} + + +#define CONCURRENT_FAULT_THREADS 8 + + +/** + * Test several threads faulting on lazily-decompressed pages at once + * + * The data pointer is shared between threads (not the asdf_ndarray_t itself, + * which is not currently thread-safe) which each read a different + * not-yet-decompressed page at the same time. Each of their faults is + * reported to the decompression handler separately, and whichever it handles + * first decompresses all pages up to that one. So faults on earlier pages are + * generally still queued (in the waiting threads) for pages it has already + * filled (without waking their threads), and the handler has to cope with + * them. + * + * This is the regression tested here: the UFFDIO_COPY ioctl called from the + * subsequently awakened threads would get an EEXIST error which was not handled + * properly. + * + * One-page chunks are used so that this is independent of the chunk size + * handling tested in read_compressed_block_lazy_random_access. + */ +MU_TEST(read_compressed_block_lazy_concurrent_fault) { + const char *comp = munit_parameters_get(params, "comp"); + const char *filename = get_fixture_file_path("compressed.asdf"); + + if (!lazy_available()) { + munit_log(MUNIT_LOG_INFO, "lazy decompression not available; skipping test..."); + return MUNIT_SKIP; + } + + asdf_config_t config = { + .decomp = { + .mode = ASDF_BLOCK_DECOMP_MODE_LAZY, + .chunk_size = (size_t)sysconf(_SC_PAGESIZE) + } + }; + asdf_file_t *file = asdf_open_ex(filename, "r", &config); + assert_not_null(file); + asdf_ndarray_t *ndarray = NULL; + assert_int(asdf_get_ndarray(file, comp, &ndarray), ==, ASDF_VALUE_OK); + assert_not_null(ndarray); + size_t size = 0; + const uint8_t *data = asdf_ndarray_data(ndarray, &size); + assert_not_null(data); + + // Same data layout as in test_compressed_file: 100 4 KiB "pages" of the + // repeating pattern 0 to 255, except the first byte of each is its index + const size_t page_size = 4096; + const size_t num_pages = 100; + assert_size(size, ==, page_size * num_pages); + uint8_t *expected = malloc(size); + + if (!expected) + return MUNIT_ERROR; + + for (size_t idx = 0; idx < size; idx++) + expected[idx] = (idx % page_size == 0) ? (idx / page_size) % 256 : idx % 256; + + // Spread the faults over the first part of the array, so that there is + // still data after them to be decompressed by later faults + pthread_barrier_t barrier; + pthread_t threads[CONCURRENT_FAULT_THREADS]; + concurrent_fault_arg_t args[CONCURRENT_FAULT_THREADS]; + assert_int(pthread_barrier_init(&barrier, NULL, CONCURRENT_FAULT_THREADS), ==, 0); + + for (int idx = 0; idx < CONCURRENT_FAULT_THREADS; idx++) { + size_t fault_offset = (size_t)(idx + 1) * 5 * page_size; + args[idx] = (concurrent_fault_arg_t){.barrier = &barrier, .addr = data + fault_offset}; + // Each thread is started and blocks at the barrier--once all + // CONCURRENT_FAULT_THREADS(=8) reach the pthread_barrier_wait call + // the barrier is unblocked and the first thread to run causes the + // first page fault and becomes blocked. Subsequent fault_threads can + // also become blocked at a page fault, with each unhandled fault + // queued by the kernel as a pending UFFD message sent over the fd. + // The single fault handler thread reads the queued faults one at a + // time, sequentially (though in arbitrary order). Handling a fault + // on a later page (in terms of offset) will fill prior pages as well, + // at least in the current sequential decompressors. But + // UFFDIO_COPY_MODE_DONTWAKE, which neither wakes the thread waiting + // on the page fault, nor removes queued faults on other threads. + // So those faults are still handled afters, for pages that have now + // already been filled, and the handler must simply wake their threads + // rather than attempt to fill their faulted pages again (this resulted + // in an EEXIST error from the UFFDIO_COPY ioctl, causing the fault + // handler thread to exit, but this is symptom not the cause of the + // problem). + assert_int(pthread_create(&threads[idx], NULL, concurrent_fault_thread, &args[idx]), ==, 0); + } + + for (int idx = 0; idx < CONCURRENT_FAULT_THREADS; idx++) { + pthread_join(threads[idx], NULL); + assert_uint8(args[idx].value, ==, *(expected + (args[idx].addr - data))); + } + + pthread_barrier_destroy(&barrier); + + const char *error = asdf_error(file); + if (error) + munit_logf(MUNIT_LOG_ERROR, "error after concurrent faults: %s", error); + assert_null(error); + + // The rest of the data is still decompressed correctly + assert_memory_equal(size, data, expected); + free(expected); + asdf_ndarray_destroy(ndarray); + asdf_close(file); + return MUNIT_OK; +} +#else +MU_TEST(read_compressed_block_lazy_concurrent_fault) { + munit_log( + MUNIT_LOG_INFO, + "lazy decompression or pthread barriers not available; skipping test..."); + return MUNIT_SKIP; +} +#endif + + /** * Write a small, highly-compressible ndarray with a given compressor and * verify the round-trip. @@ -561,6 +973,135 @@ static const char *write_compressed_ndarray_to_file( } +/** + * Test repeated access to the data of a compressed ndarray + * + * Regression test for #274. + * + * Every access after the first must still return the decompressed data, not + * the raw compressed bytes of the block. Each tile read accesses the + * ndarray's data anew, so reading tiles one after another is the most common + * way to hit this. + */ +MU_TEST(read_compressed_block_repeated_access) { + const char *comp = munit_parameters_get(params, "comp"); + const char *filename = get_fixture_file_path("compressed.asdf"); + asdf_config_t config = { + .decomp = { + .mode = decomp_mode_from_param(munit_parameters_get(params, "mode")) + } + }; + asdf_file_t *file = asdf_open_file_ex(filename, "r", &config); + assert_not_null(file); + asdf_ndarray_t *ndarray = NULL; + asdf_value_err_t err = asdf_get_ndarray(file, comp, &ndarray); + assert_int(err, ==, ASDF_VALUE_OK); + assert_not_null(ndarray); + + // Same data layout as in test_compressed_file: 100 pages of the repeating + // pattern 0 to 255, except the first byte of each page is the page index + const size_t page_size = 4096; + const size_t num_pages = 100; + const uint64_t tile_shape[] = {page_size}; + uint64_t origin[] = {0}; + uint8_t expected[4096] = {0}; + uint8_t *tile = NULL; + + for (size_t idx = 0; idx < page_size; idx++) + expected[idx] = idx % 256; + + // First access, reading the first page + expected[0] = 0; + assert_int( + asdf_ndarray_read_tile_ndim(ndarray, origin, tile_shape, ASDF_DATATYPE_SOURCE, (void **)&tile), + ==, + ASDF_NDARRAY_OK); + assert_memory_equal(page_size, tile, expected); + free(tile); + tile = NULL; + + // Subsequent accesses all return the same decompressed data + size_t size = 0; + const void *data = asdf_ndarray_data(ndarray, &size); + assert_not_null(data); + assert_size(size, ==, page_size * num_pages); + size_t size2 = 0; + assert_ptr_equal(asdf_ndarray_data(ndarray, &size2), data); + assert_size(size2, ==, size); + + // Read the last page + origin[0] = page_size * (num_pages - 1); + expected[0] = num_pages - 1; + assert_int( + asdf_ndarray_read_tile_ndim(ndarray, origin, tile_shape, ASDF_DATATYPE_SOURCE, (void **)&tile), + ==, + ASDF_NDARRAY_OK); + assert_memory_equal(page_size, tile, expected); + free(tile); + + asdf_ndarray_destroy(ndarray); + asdf_close(file); + return MUNIT_OK; +} + + +/** + * Test reading a compressed array large enough to span multiple chunks + * + * Regression test for #274. + * + * The array in ``compressed.asdf`` decompresses in a single chunk (and is a + * single LZ4 block) so it does not exercise decompression of multiple chunks + * in lazy mode. This writes an array larger than the 4 MB LZ4 block size + * (and the default lazy decompression chunk size), and which is not a + * multiple of either the block size or the page size, so that the last chunk + * and the last page are both partial. + */ +MU_TEST(read_compressed_block_multi_chunk) { + const char *comp = munit_parameters_get(params, "comp"); + /* Large enough to span two LZ4 blocks plus a partial one, and not a + * multiple of page size */ + const size_t n = 9000000; + uint8_t *ref = malloc(n); + + if (!ref) + return MUNIT_ERROR; + + // Use a prime period so the pattern does not line up with page or chunk + // boundaries + for (size_t idx = 0; idx < n; idx++) + ref[idx] = idx % 251; + + const char *path = write_compressed_ndarray_to_file(comp, fixture->tempfile_prefix, ref, n); + + if (!path) { + free(ref); + return MUNIT_ERROR; + } + + asdf_config_t config = { + .decomp = { + .mode = decomp_mode_from_param(munit_parameters_get(params, "mode")) + } + }; + asdf_file_t *file = asdf_open_file_ex(path, "r", &config); + assert_not_null(file); + asdf_ndarray_t *ndarray = NULL; + assert_int(asdf_get_ndarray(file, "data", &ndarray), ==, ASDF_VALUE_OK); + assert_not_null(ndarray); + size_t size = 0; + const uint8_t *data = asdf_ndarray_data(ndarray, &size); + assert_not_null(data); + assert_size(size, ==, n); + assert_memory_equal(n, data, ref); + asdf_ndarray_destroy(ndarray); + asdf_close(file); + free(ref); + free((void *)path); + return MUNIT_OK; +} + + /** * Re-emit an already-compressed block verbatim (no decompression/recompression) * @@ -1079,8 +1620,13 @@ MU_TEST_SUITE( MU_RUN_TEST(read_compressed_block_to_file, comp_test_params), MU_RUN_TEST(read_compressed_block_to_file_on_threshold, comp_test_params), MU_RUN_TEST(open_close_compressed_block, comp_mode_test_params), - MU_RUN_TEST(read_compressed_block_lazy_random_access, comp_mode_test_params), + MU_RUN_TEST(read_compressed_block_lazy_random_access, comp_mode_chunk_test_params), + MU_RUN_TEST(read_compressed_block_lazy_concurrent_fault, comp_test_params), MU_RUN_TEST(compressed_block_no_hang_on_segfault, comp_mode_test_params), + MU_RUN_TEST(decomp_mode_selection, comp_mode_test_params), + MU_RUN_TEST(decomp_mode_selection_to_file, comp_mode_test_params), + MU_RUN_TEST(read_compressed_block_repeated_access, comp_mode_test_params), + MU_RUN_TEST(read_compressed_block_multi_chunk, comp_mode_test_params), MU_RUN_TEST(reemit_compressed_verbatim, comp_test_params), MU_RUN_TEST(recompress_block), MU_RUN_TEST(copy_compressed_block, comp_test_params),