diff --git a/lib/adler32.c b/lib/adler32.c index d5f39d8f..8785500b 100644 --- a/lib/adler32.c +++ b/lib/adler32.c @@ -135,7 +135,14 @@ typedef u32 (*adler32_func_t)(u32 adler, const u8 *p, size_t len); #ifdef arch_select_adler32_func static u32 dispatch_adler32(u32 adler, const u8 *p, size_t len); -static volatile adler32_func_t adler32_impl = dispatch_adler32; +/* + * Resolved to the best implementation on the first call. Accessed with relaxed atomics: the + * first-call resolution is a benign race (every thread computes the same pointer, a pure function + * of the CPU), but a plain load racing with the store is undefined behavior and is flagged by + * ThreadSanitizer. Relaxed ordering suffices because no other memory is published through it. + */ +static adler32_func_t adler32_impl = dispatch_adler32; +#define adler32_impl_load() __atomic_load_n(&adler32_impl, __ATOMIC_RELAXED) /* Choose the best implementation at runtime. */ static u32 dispatch_adler32(u32 adler, const u8 *p, size_t len) @@ -145,12 +152,12 @@ static u32 dispatch_adler32(u32 adler, const u8 *p, size_t len) if (f == NULL) f = DEFAULT_IMPL; - adler32_impl = f; + __atomic_store_n(&adler32_impl, f, __ATOMIC_RELAXED); return f(adler, p, len); } #else /* The best implementation is statically known, so call it directly. */ -#define adler32_impl DEFAULT_IMPL +#define adler32_impl_load() (DEFAULT_IMPL) #endif LIBDEFLATEAPI u32 @@ -158,5 +165,5 @@ libdeflate_adler32(u32 adler, const void *buffer, size_t len) { if (buffer == NULL) /* Return initial value. */ return 1; - return adler32_impl(adler, buffer, len); + return adler32_impl_load()(adler, buffer, len); } diff --git a/lib/arm/cpu_features.c b/lib/arm/cpu_features.c index 42f81f5e..18b8018d 100644 --- a/lib/arm/cpu_features.c +++ b/lib/arm/cpu_features.c @@ -202,7 +202,7 @@ static const struct cpu_feature arm_cpu_feature_table[] = { {ARM_CPU_FEATURE_DOTPROD, "dotprod"}, }; -volatile u32 libdeflate_arm_cpu_features = 0; +u32 libdeflate_arm_cpu_features = 0; void libdeflate_init_arm_cpu_features(void) { @@ -224,7 +224,8 @@ void libdeflate_init_arm_cpu_features(void) disable_cpu_features_for_testing(&features, arm_cpu_feature_table, ARRAY_LEN(arm_cpu_feature_table)); - libdeflate_arm_cpu_features = features | ARM_CPU_FEATURES_KNOWN; + __atomic_store_n(&libdeflate_arm_cpu_features, + features | ARM_CPU_FEATURES_KNOWN, __ATOMIC_RELAXED); } #endif /* ARM_CPU_FEATURES_KNOWN */ diff --git a/lib/arm/cpu_features.h b/lib/arm/cpu_features.h index dc9ab8ad..2fd291e1 100644 --- a/lib/arm/cpu_features.h +++ b/lib/arm/cpu_features.h @@ -51,15 +51,25 @@ (defined(_WIN32) && defined(ARCH_ARM64))) /* Runtime ARM CPU feature detection is supported. */ # define ARM_CPU_FEATURES_KNOWN (1U << 31) -extern volatile u32 libdeflate_arm_cpu_features; +extern u32 libdeflate_arm_cpu_features; void libdeflate_init_arm_cpu_features(void); +/* + * Resolved on the first call. Accessed with relaxed atomics: the first-call initialization is a + * benign race (every thread computes the same features bitmask, a pure function of the CPU), but + * a plain load racing with the store in libdeflate_init_arm_cpu_features() is undefined behavior + * and is flagged by ThreadSanitizer. Relaxed ordering suffices because no other memory is + * published through it. + */ static inline u32 get_arm_cpu_features(void) { - if (libdeflate_arm_cpu_features == 0) + u32 features = __atomic_load_n(&libdeflate_arm_cpu_features, __ATOMIC_RELAXED); + if (features == 0) { libdeflate_init_arm_cpu_features(); - return libdeflate_arm_cpu_features; + features = __atomic_load_n(&libdeflate_arm_cpu_features, __ATOMIC_RELAXED); + } + return features; } #else static inline u32 get_arm_cpu_features(void) { return 0; } diff --git a/lib/crc32.c b/lib/crc32.c index a0ec0223..4af101b0 100644 --- a/lib/crc32.c +++ b/lib/crc32.c @@ -235,7 +235,14 @@ typedef u32 (*crc32_func_t)(u32 crc, const u8 *p, size_t len); #ifdef arch_select_crc32_func static u32 dispatch_crc32(u32 crc, const u8 *p, size_t len); -static volatile crc32_func_t crc32_impl = dispatch_crc32; +/* + * Resolved to the best implementation on the first call. Accessed with relaxed atomics: the + * first-call resolution is a benign race (every thread computes the same pointer, a pure function + * of the CPU), but a plain load racing with the store is undefined behavior and is flagged by + * ThreadSanitizer. Relaxed ordering suffices because no other memory is published through it. + */ +static crc32_func_t crc32_impl = dispatch_crc32; +#define crc32_impl_load() __atomic_load_n(&crc32_impl, __ATOMIC_RELAXED) /* Choose the best implementation at runtime. */ static u32 dispatch_crc32(u32 crc, const u8 *p, size_t len) @@ -245,12 +252,12 @@ static u32 dispatch_crc32(u32 crc, const u8 *p, size_t len) if (f == NULL) f = DEFAULT_IMPL; - crc32_impl = f; + __atomic_store_n(&crc32_impl, f, __ATOMIC_RELAXED); return f(crc, p, len); } #else /* The best implementation is statically known, so call it directly. */ -#define crc32_impl DEFAULT_IMPL +#define crc32_impl_load() (DEFAULT_IMPL) #endif LIBDEFLATEAPI u32 @@ -258,5 +265,5 @@ libdeflate_crc32(u32 crc, const void *p, size_t len) { if (p == NULL) /* Return initial value. */ return 0; - return ~crc32_impl(~crc, p, len); + return ~crc32_impl_load()(~crc, p, len); } diff --git a/lib/decompress_template.h b/lib/decompress_template.h index 8c874c36..71698db4 100644 --- a/lib/decompress_template.h +++ b/lib/decompress_template.h @@ -62,16 +62,67 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, u32 bitsleft = 0; size_t overread_count = 0; - bool is_final_block; + /* Initialized to keep the streaming suspension paths (which store it into the + * decompressor unconditionally, guarded by 'in_block' on resume) away from an + * indeterminate read when suspending before the first block header is decoded. */ + bool is_final_block = false; unsigned block_type; unsigned num_litlen_syms; unsigned num_offset_syms; bitbuf_t litlen_tablemask; u32 entry; +#ifdef DEFLATE_STREAMING + /* + * Streaming state. 'window_nbytes' bytes of previously produced output + * precede 'out' (for resolving back-references). The checkpoint (cp_*) + * holds the sub-byte-aligned resume state at the last resumable + * position: a block boundary, or a symbol boundary inside a Huffman + * block ('cp_in_block' distinguishes the two). On suspension we roll + * back to it. Checkpointing at symbol granularity (not just block + * boundaries) is what keeps streaming decompression linear-time even + * when a single DEFLATE block spans the whole stream, as produced by + * e.g. zlib-ng at compression level 1. + */ + const size_t window_nbytes = d->window_nbytes; + const u8 *cp_in_next = in_next; + u8 *cp_out_next = out_next; + bitbuf_t cp_bitbuf = d->saved_bitbuf; + u32 cp_bitsleft = d->saved_bitsleft; + bool cp_in_block = false; + + bitbuf = d->saved_bitbuf; + bitsleft = d->saved_bitsleft; + + if (d->in_block) { + /* + * Resuming at a symbol boundary inside a Huffman block: the + * litlen/offset decode tables in *d still describe the block's + * codes (nothing rebuilds them between the suspension and now), + * so skip the block header and go straight back to decoding + * symbols. + */ + is_final_block = d->block_is_final; + cp_in_block = true; + goto have_decode_tables; + } +#endif + next_block: /* Starting to read the next block */ ; +#ifdef DEFLATE_STREAMING + /* Checkpoint the byte-aligned resume state at this block boundary. */ + { + u32 cp_bl = (u8)bitsleft; + SAFETY_CHECK(overread_count <= (cp_bl >> 3)); + cp_in_next = in_next - ((cp_bl >> 3) - overread_count); + cp_out_next = out_next; + cp_bitsleft = cp_bl & 7; + cp_bitbuf = bitbuf & (((bitbuf_t)1 << cp_bitsleft) - 1); + cp_in_block = false; + } +#endif STATIC_ASSERT(CAN_CONSUME(1 + 2 + 5 + 5 + 4 + 3)); REFILL_BITS(); @@ -268,15 +319,35 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, bitbuf = 0; bitsleft = 0; +#ifdef DEFLATE_STREAMING + if (in_end - in_next < 4) { + if (!d->end_of_input) + goto need_more_input; + SAFETY_CHECK(0); + } +#else SAFETY_CHECK(in_end - in_next >= 4); +#endif len = get_unaligned_le16(in_next); nlen = get_unaligned_le16(in_next + 2); in_next += 4; SAFETY_CHECK(len == (u16)~nlen); if (unlikely(len > out_end - out_next)) +#ifdef DEFLATE_STREAMING + goto need_more_output; +#else return LIBDEFLATE_INSUFFICIENT_SPACE; +#endif +#ifdef DEFLATE_STREAMING + if ((size_t)(in_end - in_next) < len) { + if (!d->end_of_input) + goto need_more_input; + SAFETY_CHECK(0); + } +#else SAFETY_CHECK(len <= in_end - in_next); +#endif memcpy(out_next, in_next, len); in_next += len; @@ -547,7 +618,11 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, offset += EXTRACT_VARBITS8(saved_bitbuf, entry) >> (u8)(entry >> 8); /* Validate the match offset; needed even in the fastloop. */ +#ifdef DEFLATE_STREAMING + SAFETY_CHECK((size_t)offset <= (size_t)(out_next - (u8 *)out) + window_nbytes); +#else SAFETY_CHECK(offset <= out_next - (const u8 *)out); +#endif src = out_next - offset; dst = out_next; out_next += length; @@ -683,6 +758,36 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, const u8 *src; u8 *dst; +#ifdef DEFLATE_STREAMING + /* + * Checkpoint this symbol boundary, so that a suspension inside + * the block (input exhausted, or output full) resumes here + * instead of rolling back to the block start. Without this, a + * single DEFLATE block spanning the whole stream would be + * re-decoded from its start on every refill, making streaming + * decompression quadratic in the block's compressed size. + * Suspensions can only trigger from this loop (the fastloop's + * entry conditions leave it enough input and output slack), so + * checkpointing here is sufficient and keeps the fastloop free + * of extra work. + * + * Implicit appended zero bytes can never have been consumed at + * a symbol boundary of a valid stream (they can only appear + * past the final block, or when truncated data was declared + * complete via 'end_of_input'), so reaching this point with + * more overread bytes than unconsumed whole bytes in the + * bitbuffer means the data is bad. + */ + { + u32 cp_bl = (u8)bitsleft; + SAFETY_CHECK(overread_count <= (cp_bl >> 3)); + cp_in_next = in_next - ((cp_bl >> 3) - overread_count); + cp_out_next = out_next; + cp_bitsleft = cp_bl & 7; + cp_bitbuf = bitbuf & (((bitbuf_t)1 << cp_bitsleft) - 1); + cp_in_block = true; + } +#endif REFILL_BITS(); entry = d->u.litlen_decode_table[bitbuf & litlen_tablemask]; saved_bitbuf = bitbuf; @@ -698,7 +803,11 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, length = entry >> 16; if (entry & HUFFDEC_LITERAL) { if (unlikely(out_next == out_end)) +#ifdef DEFLATE_STREAMING + goto need_more_output; +#else return LIBDEFLATE_INSUFFICIENT_SPACE; +#endif *out_next++ = length; continue; } @@ -706,7 +815,11 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, goto block_done; length += EXTRACT_VARBITS8(saved_bitbuf, entry) >> (u8)(entry >> 8); if (unlikely(length > out_end - out_next)) +#ifdef DEFLATE_STREAMING + goto need_more_output; +#else return LIBDEFLATE_INSUFFICIENT_SPACE; +#endif if (!CAN_CONSUME(LENGTH_MAXBITS + OFFSET_MAXBITS)) REFILL_BITS(); @@ -724,7 +837,11 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, bitbuf >>= (u8)entry; bitsleft -= entry; +#ifdef DEFLATE_STREAMING + SAFETY_CHECK((size_t)offset <= (size_t)(out_next - (u8 *)out) + window_nbytes); +#else SAFETY_CHECK(offset <= out_next - (const u8 *)out); +#endif src = out_next - offset; dst = out_next; out_next += length; @@ -769,6 +886,35 @@ FUNCNAME(struct libdeflate_decompressor * restrict d, return LIBDEFLATE_SHORT_OUTPUT; } return LIBDEFLATE_SUCCESS; + +#ifdef DEFLATE_STREAMING + /* + * Suspension points: roll back to the last checkpoint (a block + * boundary, or a symbol boundary inside the current Huffman block) and + * report how much input/output was fully consumed/produced up to it. + * 'is_final_block' describes the current block, which is also the + * checkpoint's block whenever 'cp_in_block' is set: the checkpoint is + * re-taken at every block boundary, so it can never lag behind in a + * previous block. + */ +need_more_input: + d->saved_bitbuf = cp_bitbuf; + d->saved_bitsleft = cp_bitsleft; + d->in_block = cp_in_block; + d->block_is_final = is_final_block; + *actual_in_nbytes_ret = cp_in_next - (const u8 *)in; + *actual_out_nbytes_ret = cp_out_next - (u8 *)out; + return LIBDEFLATE_STREAM_NEED_INPUT; + +need_more_output: + d->saved_bitbuf = cp_bitbuf; + d->saved_bitsleft = cp_bitsleft; + d->in_block = cp_in_block; + d->block_is_final = is_final_block; + *actual_in_nbytes_ret = cp_in_next - (const u8 *)in; + *actual_out_nbytes_ret = cp_out_next - (u8 *)out; + return LIBDEFLATE_STREAM_NEED_OUTPUT; +#endif } #undef FUNCNAME diff --git a/lib/deflate_compress.c b/lib/deflate_compress.c index 4a1f3276..775bc057 100644 --- a/lib/deflate_compress.c +++ b/lib/deflate_compress.c @@ -463,6 +463,13 @@ struct libdeflate_compressor { /* The compression level with which this compressor was created */ unsigned compression_level; + /* + * If true, deflate_flush_block() forces every block to be non-final, so + * that the output of one compression call can be concatenated with more + * DEFLATE data. Set transiently by libdeflate_deflate_compress_stream_chunk(). + */ + bool stream_chunk; + /* Anything of this size or less we won't bother trying to compress. */ size_t max_passthrough_size; @@ -1710,6 +1717,14 @@ deflate_flush_block(struct libdeflate_compressor *c, const struct deflate_sequence *sequences, bool is_final_block) { + /* + * In stream-chunk mode, never terminate the DEFLATE stream: keep every + * block non-final so the caller can append further blocks. Only the + * block-header BFINAL bit value changes, not its cost, so the output + * size accounting below is unaffected. + */ + if (c->stream_chunk) + is_final_block = false; /* * It is hard to get compilers to understand that writes to 'os->next' * don't alias 'os'. That hurts performance significantly, as @@ -3911,6 +3926,8 @@ libdeflate_alloc_compressor_ex(int compression_level, c->compression_level = compression_level; + c->stream_chunk = false; + /* * The higher the compression level, the more we should bother trying to * compress very small inputs. @@ -4064,6 +4081,118 @@ libdeflate_deflate_compress(struct libdeflate_compressor *c, return os.next - (u8 *)out; } +/* + * Emit a "sync flush": a non-final empty stored block that also flushes any + * pending bits and aligns the output to a byte boundary. This is the same + * construct as zlib's Z_SYNC_FLUSH and lets us concatenate independently + * compressed DEFLATE segments into a single stream. + */ +static bool +deflate_emit_sync_flush(struct deflate_output_bitstream *os) +{ + bitbuf_t bitbuf = os->bitbuf; + unsigned bitcount = os->bitcount; + u8 *out_next = os->next; + + /* BFINAL=0, BTYPE=00 (3 bits, all zero). */ + bitcount += 3; + while (bitcount >= 8) { + if (out_next >= os->end) { os->overflow = true; return false; } + *out_next++ = (u8)bitbuf; + bitbuf >>= 8; + bitcount -= 8; + } + /* Align to a byte boundary (the remaining header bits are zero). */ + if (bitcount > 0) { + if (out_next >= os->end) { os->overflow = true; return false; } + *out_next++ = (u8)bitbuf; + bitcount = 0; + } + /* LEN = 0, NLEN = 0xFFFF. */ + if (os->end - out_next < 4) { os->overflow = true; return false; } + *out_next++ = 0x00; + *out_next++ = 0x00; + *out_next++ = 0xFF; + *out_next++ = 0xFF; + + os->bitbuf = 0; + os->bitcount = 0; + os->next = out_next; + return true; +} + +/* + * Emit the input as one or more non-final stored (uncompressed) blocks. Used for + * stream chunks too small to be worth compressing. Assumes byte-aligned output. + */ +static bool +deflate_emit_stored_nonfinal(struct deflate_output_bitstream *os, + const u8 *in, size_t in_nbytes) +{ + u8 *out_next = os->next; + size_t offset = 0; + + do { + size_t len = in_nbytes - offset; + if (len > UINT16_MAX) + len = UINT16_MAX; + if ((size_t)(os->end - out_next) < 5 + len) { + os->overflow = true; + return false; + } + *out_next++ = 0x00; /* BFINAL=0, BTYPE=00, + alignment padding */ + put_unaligned_le16((u16)len, out_next); + out_next += 2; + put_unaligned_le16((u16)~len, out_next); + out_next += 2; + if (len) { + memcpy(out_next, in + offset, len); + out_next += len; + } + offset += len; + } while (offset < in_nbytes); + + os->next = out_next; + return true; +} + +/* + * Compress one chunk of a DEFLATE stream. Unlike libdeflate_deflate_compress(), + * the output is NOT terminated: all blocks are non-final and the output ends on + * a byte boundary, so the result can be concatenated with more chunks and then a + * final block. Returns the number of bytes written, or 0 if the output buffer is + * too small. + */ +LIBDEFLATEAPI size_t +libdeflate_deflate_compress_stream_chunk(struct libdeflate_compressor *c, + const void *in, size_t in_nbytes, + void *out, size_t out_nbytes_avail) +{ + struct deflate_output_bitstream os; + + os.bitbuf = 0; + os.bitcount = 0; + os.next = out; + os.end = os.next + out_nbytes_avail; + os.overflow = false; + + if (in_nbytes > c->max_passthrough_size && c->impl != NULL) { + c->stream_chunk = true; + (*c->impl)(c, in, in_nbytes, &os); + c->stream_chunk = false; + if (os.overflow) + return 0; + if (!deflate_emit_sync_flush(&os)) + return 0; + } else { + /* Small input (or level 0): store it, already byte-aligned. */ + if (!deflate_emit_stored_nonfinal(&os, in, in_nbytes)) + return 0; + } + + return os.next - (u8 *)out; +} + LIBDEFLATEAPI void libdeflate_free_compressor(struct libdeflate_compressor *c) { diff --git a/lib/deflate_decompress.c b/lib/deflate_decompress.c index 63726c7a..0e65b926 100644 --- a/lib/deflate_decompress.c +++ b/lib/deflate_decompress.c @@ -244,6 +244,7 @@ do { \ bitbuf |= (bitbuf_t)*in_next++ << \ (u8)bitsleft; \ } else { \ + OVERREAD_HANDLER(); \ overread_count++; \ SAFETY_CHECK(overread_count <= \ sizeof(bitbuf_t)); \ @@ -253,6 +254,15 @@ do { \ } \ } while (0) +/* + * Hook invoked by REFILL_BITS() when it would overread the input. It is a no-op + * for the normal (one-shot) decompressor, and is redefined to suspend at a block + * boundary for the streaming instantiation of the template. + */ +#ifndef OVERREAD_HANDLER +# define OVERREAD_HANDLER() /* nothing */ +#endif + /* * REFILL_BITS_IN_FASTLOOP() is like REFILL_BITS(), but it doesn't check for the * end of the input. It can only be used in the fastloop. @@ -672,6 +682,24 @@ struct libdeflate_decompressor { bool static_codes_loaded; unsigned litlen_tablebits; + /* + * Streaming decompression state (libdeflate_deflate_decompress_stream()). + * Used only by the DEFLATE_STREAMING instantiation of the decode template; + * the normal one-shot path never touches these. + * + * 'saved_bitbuf'/'saved_bitsleft' hold the sub-byte bit position of the + * resume point. When 'in_block' is set, the resume point is a symbol + * boundary inside a Huffman block whose BFINAL flag is 'block_is_final' + * and whose litlen/offset decode tables are the ones stored above; + * otherwise it is a block boundary. + */ + bool end_of_input; + bool in_block; + bool block_is_final; + size_t window_nbytes; + bitbuf_t saved_bitbuf; + u32 saved_bitsleft; + /* The free() function for this struct, chosen at allocation time */ free_func_t free_func; }; @@ -1099,7 +1127,14 @@ dispatch_decomp(struct libdeflate_decompressor *d, void *out, size_t out_nbytes_avail, size_t *actual_in_nbytes_ret, size_t *actual_out_nbytes_ret); -static volatile decompress_func_t decompress_impl = dispatch_decomp; +/* + * Resolved to the best implementation on the first call. Accessed with relaxed atomics: the + * first-call resolution is a benign race (every thread computes the same pointer, a pure function + * of the CPU), but a plain load racing with the store is undefined behavior and is flagged by + * ThreadSanitizer. Relaxed ordering suffices because no other memory is published through it. + */ +static decompress_func_t decompress_impl = dispatch_decomp; +#define decompress_impl_load() __atomic_load_n(&decompress_impl, __ATOMIC_RELAXED) /* Choose the best implementation at runtime. */ static enum libdeflate_result @@ -1113,15 +1148,77 @@ dispatch_decomp(struct libdeflate_decompressor *d, if (f == NULL) f = DEFAULT_IMPL; - decompress_impl = f; + __atomic_store_n(&decompress_impl, f, __ATOMIC_RELAXED); return f(d, in, in_nbytes, out, out_nbytes_avail, actual_in_nbytes_ret, actual_out_nbytes_ret); } #else /* The best implementation is statically known, so call it directly. */ -# define decompress_impl DEFAULT_IMPL +# define decompress_impl_load() (DEFAULT_IMPL) +#endif + +/* + * Streaming decompressor instantiations (libdeflate_deflate_decompress_stream()). + * Same decoder, but DEFLATE_STREAMING makes it suspend at block boundaries and + * OVERREAD_HANDLER() turns an input overread into a clean suspension. We mirror + * the one-shot arch dispatch above so the streaming path also gets the x86 BMI2 + * implementation at runtime; otherwise streaming would be stuck on the generic + * code on BMI2-capable CPUs. + */ +#define DEFLATE_STREAMING 1 +#undef OVERREAD_HANDLER +#define OVERREAD_HANDLER() do { if (!d->end_of_input) goto need_more_input; } while (0) + +#define FUNCNAME deflate_decompress_stream_default +#undef ATTRIBUTES +#undef EXTRACT_VARBITS +#undef EXTRACT_VARBITS8 +#include "decompress_template.h" + +#undef DEFAULT_STREAM_IMPL +#undef arch_select_stream_decompress_func +#if defined(ARCH_X86_32) || defined(ARCH_X86_64) +# include "x86/decompress_stream_impl.h" +#endif + +#ifndef DEFAULT_STREAM_IMPL +# define DEFAULT_STREAM_IMPL deflate_decompress_stream_default #endif +#ifdef arch_select_stream_decompress_func +static enum libdeflate_result +dispatch_stream_decomp(struct libdeflate_decompressor *d, + const void *in, size_t in_nbytes, + void *out, size_t out_nbytes_avail, + size_t *actual_in_nbytes_ret, size_t *actual_out_nbytes_ret); + +/* See decompress_impl above: relaxed-atomic access to the runtime-resolved implementation pointer. */ +static decompress_func_t stream_decompress_impl = dispatch_stream_decomp; +#define stream_decompress_impl_load() __atomic_load_n(&stream_decompress_impl, __ATOMIC_RELAXED) + +static enum libdeflate_result +dispatch_stream_decomp(struct libdeflate_decompressor *d, + const void *in, size_t in_nbytes, + void *out, size_t out_nbytes_avail, + size_t *actual_in_nbytes_ret, size_t *actual_out_nbytes_ret) +{ + decompress_func_t f = arch_select_stream_decompress_func(); + + if (f == NULL) + f = DEFAULT_STREAM_IMPL; + + __atomic_store_n(&stream_decompress_impl, f, __ATOMIC_RELAXED); + return f(d, in, in_nbytes, out, out_nbytes_avail, + actual_in_nbytes_ret, actual_out_nbytes_ret); +} +#else +# define stream_decompress_impl_load() (DEFAULT_STREAM_IMPL) +#endif + +#undef DEFLATE_STREAMING +#undef OVERREAD_HANDLER +#define OVERREAD_HANDLER() /* nothing */ + /* * This is the main DEFLATE decompression routine. See libdeflate.h for the * documentation. @@ -1137,8 +1234,8 @@ libdeflate_deflate_decompress_ex(struct libdeflate_decompressor *d, size_t *actual_in_nbytes_ret, size_t *actual_out_nbytes_ret) { - return decompress_impl(d, in, in_nbytes, out, out_nbytes_avail, - actual_in_nbytes_ret, actual_out_nbytes_ret); + return decompress_impl_load()(d, in, in_nbytes, out, out_nbytes_avail, + actual_in_nbytes_ret, actual_out_nbytes_ret); } LIBDEFLATEAPI enum libdeflate_result @@ -1152,6 +1249,35 @@ libdeflate_deflate_decompress(struct libdeflate_decompressor *d, NULL, actual_out_nbytes_ret); } +LIBDEFLATEAPI void +libdeflate_deflate_decompress_stream_reset(struct libdeflate_decompressor *d) +{ + d->end_of_input = false; + d->in_block = false; + d->block_is_final = false; + d->window_nbytes = 0; + d->saved_bitbuf = 0; + d->saved_bitsleft = 0; + d->static_codes_loaded = false; +} + +LIBDEFLATEAPI enum libdeflate_result +libdeflate_deflate_decompress_stream(struct libdeflate_decompressor *d, + int end_of_input, + const void *in, size_t in_nbytes, + void *out, size_t out_nbytes_avail, + size_t window_nbytes, + size_t *actual_in_nbytes_ret, + size_t *actual_out_nbytes_ret) +{ + d->end_of_input = (end_of_input != 0); + d->window_nbytes = window_nbytes; + return stream_decompress_impl_load()(d, in, in_nbytes, + out, out_nbytes_avail, + actual_in_nbytes_ret, + actual_out_nbytes_ret); +} + LIBDEFLATEAPI struct libdeflate_decompressor * libdeflate_alloc_decompressor_ex(const struct libdeflate_options *options) { diff --git a/lib/x86/cpu_features.c b/lib/x86/cpu_features.c index a4e09ada..31e7f805 100644 --- a/lib/x86/cpu_features.c +++ b/lib/x86/cpu_features.c @@ -86,7 +86,7 @@ static const struct cpu_feature x86_cpu_feature_table[] = { {X86_CPU_FEATURE_AVXVNNI, "avx_vnni"}, }; -volatile u32 libdeflate_x86_cpu_features = 0; +u32 libdeflate_x86_cpu_features = 0; static inline bool os_supports_avx512(u64 xcr0) @@ -207,7 +207,8 @@ void libdeflate_init_x86_cpu_features(void) disable_cpu_features_for_testing(&features, x86_cpu_feature_table, ARRAY_LEN(x86_cpu_feature_table)); - libdeflate_x86_cpu_features = features | X86_CPU_FEATURES_KNOWN; + __atomic_store_n(&libdeflate_x86_cpu_features, + features | X86_CPU_FEATURES_KNOWN, __ATOMIC_RELAXED); } #endif /* X86_CPU_FEATURES_KNOWN */ diff --git a/lib/x86/cpu_features.h b/lib/x86/cpu_features.h index cb225b98..99b80180 100644 --- a/lib/x86/cpu_features.h +++ b/lib/x86/cpu_features.h @@ -53,15 +53,25 @@ #if defined(__GNUC__) || defined(__clang__) || defined(_MSC_VER) /* Runtime x86 CPU feature detection is supported. */ # define X86_CPU_FEATURES_KNOWN (1U << 31) -extern volatile u32 libdeflate_x86_cpu_features; +extern u32 libdeflate_x86_cpu_features; void libdeflate_init_x86_cpu_features(void); +/* + * Resolved on the first call. Accessed with relaxed atomics: the first-call initialization is a + * benign race (every thread computes the same features bitmask, a pure function of the CPU), but + * a plain load racing with the store in libdeflate_init_x86_cpu_features() is undefined behavior + * and is flagged by ThreadSanitizer. Relaxed ordering suffices because no other memory is + * published through it. + */ static inline u32 get_x86_cpu_features(void) { - if (libdeflate_x86_cpu_features == 0) + u32 features = __atomic_load_n(&libdeflate_x86_cpu_features, __ATOMIC_RELAXED); + if (features == 0) { libdeflate_init_x86_cpu_features(); - return libdeflate_x86_cpu_features; + features = __atomic_load_n(&libdeflate_x86_cpu_features, __ATOMIC_RELAXED); + } + return features; } /* * x86 intrinsics are also supported. Include the headers needed to use them. diff --git a/lib/x86/decompress_stream_impl.h b/lib/x86/decompress_stream_impl.h new file mode 100644 index 00000000..1ca374d5 --- /dev/null +++ b/lib/x86/decompress_stream_impl.h @@ -0,0 +1,44 @@ +#ifndef LIB_X86_DECOMPRESS_STREAM_IMPL_H +#define LIB_X86_DECOMPRESS_STREAM_IMPL_H + +#include "cpu_features.h" + +/* + * BMI2-optimized streaming decompression function. This is the streaming + * counterpart of x86/decompress_impl.h: the includer has already defined + * DEFLATE_STREAMING and OVERREAD_HANDLER, so instantiating the template here + * with __attribute__((target("bmi2"))) gives the streaming decoder the same + * BMI2 path the one-shot decoder gets, selected at runtime via CPUID. + */ +#if defined(__GNUC__) || defined(__clang__) || MSVC_PREREQ(1930) +# define deflate_decompress_stream_bmi2 deflate_decompress_stream_bmi2 +# define FUNCNAME deflate_decompress_stream_bmi2 +# define ATTRIBUTES _target_attribute("bmi2") +# ifndef __clang__ +# ifdef ARCH_X86_64 +# define EXTRACT_VARBITS(word, count) _bzhi_u64((word), (count)) +# define EXTRACT_VARBITS8(word, count) _bzhi_u64((word), (count)) +# else +# define EXTRACT_VARBITS(word, count) _bzhi_u32((word), (count)) +# define EXTRACT_VARBITS8(word, count) _bzhi_u32((word), (count)) +# endif +# endif +# include "../decompress_template.h" +#endif + +#if defined(deflate_decompress_stream_bmi2) && HAVE_BMI2_NATIVE +#define DEFAULT_STREAM_IMPL deflate_decompress_stream_bmi2 +#else +static inline decompress_func_t +arch_select_stream_decompress_func(void) +{ +#ifdef deflate_decompress_stream_bmi2 + if (HAVE_BMI2(get_x86_cpu_features())) + return deflate_decompress_stream_bmi2; +#endif + return NULL; +} +#define arch_select_stream_decompress_func arch_select_stream_decompress_func +#endif + +#endif /* LIB_X86_DECOMPRESS_STREAM_IMPL_H */ diff --git a/libdeflate.h b/libdeflate.h index d6c885f2..75ade2db 100644 --- a/libdeflate.h +++ b/libdeflate.h @@ -86,6 +86,23 @@ libdeflate_deflate_compress(struct libdeflate_compressor *compressor, const void *in, size_t in_nbytes, void *out, size_t out_nbytes_avail); +/* + * libdeflate_deflate_compress_stream_chunk() compresses one chunk of a DEFLATE + * stream WITHOUT terminating it. All emitted blocks are non-final and the output + * ends on a byte boundary (via an empty stored "sync flush" block), so the + * outputs of consecutive calls can be concatenated. After the last chunk, append + * a final block (e.g. the two bytes 0x03 0x00, an empty final block) to terminate + * the stream. This is a ClickHouse addition that lets libdeflate back a streaming + * compressor while still producing a single valid DEFLATE/gzip/zlib member. + * + * Returns the number of bytes written, or 0 if 'out_nbytes_avail' was too small; + * size 'libdeflate_deflate_compress_bound(in_nbytes) + 8' to be safe. + */ +LIBDEFLATEAPI size_t +libdeflate_deflate_compress_stream_chunk(struct libdeflate_compressor *compressor, + const void *in, size_t in_nbytes, + void *out, size_t out_nbytes_avail); + /* * libdeflate_deflate_compress_bound() returns a worst-case upper bound on the * number of bytes of compressed data that may be produced by compressing any @@ -205,6 +222,16 @@ enum libdeflate_result { /* The data would have decompressed to more than 'out_nbytes_avail' * bytes. */ LIBDEFLATE_INSUFFICIENT_SPACE = 3, + + /* Streaming only (libdeflate_deflate_decompress_stream()): the input ran + * out and 'end_of_input' was false. All input up to the last resumable + * position (a symbol boundary within the current block, or a block + * boundary) was consumed. Provide more input and call again. */ + LIBDEFLATE_STREAM_NEED_INPUT = 4, + + /* Streaming only: the output buffer filled up. Drain it (carrying the + * last 32 KiB as the window) and call again. */ + LIBDEFLATE_STREAM_NEED_OUTPUT = 5, }; /* @@ -257,6 +284,53 @@ libdeflate_deflate_decompress_ex(struct libdeflate_decompressor *decompressor, size_t *actual_in_nbytes_ret, size_t *actual_out_nbytes_ret); +/* + * Streaming raw-DEFLATE decompression (a ClickHouse addition). + * + * Decompresses as much of the (possibly partial) input as it can, suspending at + * symbol boundaries so that arbitrarily large streams - including streams whose + * single DEFLATE block spans the whole input, as produced by e.g. zlib-ng at + * compression level 1 - can be decompressed in linear time with bounded memory. + * Reuses libdeflate's fast table-driven decoder. + * + * Window handling: back-references reach up to 32 KiB. The caller must place the + * last 'window_nbytes' (<= 32768) bytes of previously produced output immediately + * BEFORE 'out' (contiguously), and pass that count as 'window_nbytes'. For the + * first call of a stream, 'window_nbytes' is 0. + * + * 'end_of_input' must be nonzero only when 'in'..'in'+'in_nbytes' contains the + * end of the DEFLATE stream. + * + * Returns: + * LIBDEFLATE_SUCCESS - reached the final block (stream complete). + * LIBDEFLATE_STREAM_NEED_INPUT - consumed input up to the last resumable + * position (a symbol boundary within the current + * block, or a block boundary); supply more input + * (from *actual_in_nbytes_ret) and call again. + * LIBDEFLATE_STREAM_NEED_OUTPUT - output buffer full; drain it, slide the + * window, and call again. More output always + * follows, since the suspended item itself did + * not fit. + * LIBDEFLATE_BAD_DATA - the input was invalid. + * + * *actual_in_nbytes_ret / *actual_out_nbytes_ret receive the bytes consumed / + * produced (both required, must be non-NULL). Call + * libdeflate_deflate_decompress_stream_reset() before starting a new stream on a + * reused decompressor. + */ +LIBDEFLATEAPI enum libdeflate_result +libdeflate_deflate_decompress_stream(struct libdeflate_decompressor *decompressor, + int end_of_input, + const void *in, size_t in_nbytes, + void *out, size_t out_nbytes_avail, + size_t window_nbytes, + size_t *actual_in_nbytes_ret, + size_t *actual_out_nbytes_ret); + +/* Reset streaming state so the decompressor can start a new stream. */ +LIBDEFLATEAPI void +libdeflate_deflate_decompress_stream_reset(struct libdeflate_decompressor *decompressor); + /* * Like libdeflate_deflate_decompress(), but assumes the zlib wrapper format * instead of raw DEFLATE.