diff --git a/README.md b/README.md index 139cbaa..16a9549 100644 --- a/README.md +++ b/README.md @@ -281,16 +281,19 @@ unset LD_PRELOAD deflate/inflate and related functions - deflateInit, deflateInit2, deflateSetDictionary, deflateParams, deflateCopy, deflate, deflateEnd, deflateReset, deflateResetKeep -- inflateInit, inflateInit2, inflateSetDictionary, inflateCopy, inflate, inflateEnd, inflateReset, inflateReset2, inflateResetKeep +- inflateInit, inflateInit2, inflateSetDictionary, inflateCopy, inflate, inflateEnd, inflateReset, inflateReset2, inflateResetKeep, inflateSync For deflate, offload is supported for Z_FINISH flush option. Support for additional options will be added in later releases. For deflateSetDictionary/inflateSetDictionary, zlib-accel simply sets the execution path to zlib, as dictionary compression is currently not supported for accelerators. +A dictionary a decompressor was never told about is detected from the FDICT bit in the zlib header, which `inflate` reads before choosing an engine. The bit is in the second header byte, so a first call carrying only one byte of a zlib-format stream is also pinned to zlib: an accelerator handed that byte consumes it into its own header buffer, and by the time it reports the dictionary the bytes zlib would need to parse the header itself are no longer in the caller's buffer. The cost is that such a stream stays on zlib even when it turns out to carry no dictionary. deflateParams is intercepted only to keep the recorded compression level current, so that a level set after initialization is still seen by path selection, and to give up an IGZIP stream that was built for a level the call supersedes; the call itself is always forwarded to zlib. inflateReset2 is intercepted because it is the only zlib entry point that changes `windowBits` on a live stream: the recorded window size and format have to be refreshed, or path selection keeps deciding on the format the stream was initialized with, and an IGZIP decompression state built for the old format is discarded rather than reset (ISA-L's reset deliberately preserves its window and format settings). It also clears the terminal state described below, since a stream it restarts is ready to decode again. As with `deflateParams`, the call is forwarded to zlib first and the recorded state is only updated when zlib accepts the new `windowBits`. Once a stream has returned `Z_STREAM_END`, `deflate`/`inflate` answer every later call from that terminal state instead of dispatching it to an engine. What matches zlib is the return code, the `msg` string, and the fact that no input is consumed and no output is written — for `deflate`, `Z_STREAM_END` for `Z_FINISH` with no remaining input, `Z_BUF_ERROR` when input remains or there is no output room, and `Z_STREAM_ERROR` for any other flush or for a null `next_out`/`next_in`; for `inflate`, `Z_STREAM_END` for every flush, and `Z_STREAM_ERROR` for a null `next_out` or for a null `next_in` with input pending. `msg` follows zlib's own rule of writing it only where zlib rejects through its internal `ERR_RETURN` macro, so an out-of-range flush on `deflate` — and every `inflate` rejection, which zlib returns without setting `msg` — leaves the field as the caller left it. `data_type` is *not* written, for the reason given above: no offloaded call can compute it, and this gate is no better placed to guess. These calls are counted separately as `deflate_stream_end_count`/`inflate_stream_end_count` when built with `ENABLE_STATISTICS`, so the per-engine counters still add up to `deflate_count`/`inflate_count`. This matters because an offloaded stream never feeds zlib's own deflate/inflate state, so without it a call after `Z_STREAM_END` would be dispatched from scratch and could append a second stream to a finished one. `deflateReset`, `inflateReset`, `inflateReset2`, `deflateResetKeep`, and `inflateResetKeep` clear the terminal state, and `deflateCopy`/`inflateCopy` carry it to the copy, so a copy of a finished stream refuses input exactly as its source does. The gate covers `deflate`/`inflate` only. Two other entry points called on a finished stream still report a different return code than zlib would, in either direction, without moving any bytes and without changing how the stream answers a later `Z_FINISH`. `deflateParams` returns `Z_OK` where zlib returns `Z_STREAM_ERROR`: zlib reaches that error through an internal `deflate(strm, Z_BLOCK)` that it only performs once its own encoder has flushed something, and an offloaded stream never advanced that state. `deflateSetDictionary` returns `Z_STREAM_ERROR` where zlib returns `Z_OK`: zlib accepts a dictionary on a finished stream, having cleared the wrapper flag that its "before compression begins" check tests when it wrote the trailer, whereas the mid-stream rejection above also catches a stream that is merely finished. Applications that inspect these return codes on a finished stream should disable offload for those streams. `deflateResetKeep`/`inflateResetKeep` (declared in zlib.h among the functions zlib does not document) are intercepted because they restart a stream without going through `deflateReset`/`inflateReset`, so a terminal state they left set would wedge the stream at `Z_STREAM_END`. `inflateResetKeep` additionally pins the stream to zlib: keeping the window is the only reason to call it rather than `inflateReset`, so the next stream may reference the previous stream's bytes, which no accelerator can see — the same treatment `inflateSetDictionary` gets. A later `inflateReset` lifts the pin, being the reset that discards the history. `deflateResetKeep` needs no pin: what it keeps only affects how zlib would encode the next stream, and an offloaded stream emits a self-contained one instead. Both are forwarded to zlib first, and the recorded state is only updated when zlib reports success. The pin decides which engine decodes the next stream; it cannot supply the history. If the previous stream was offloaded, zlib's window never received it, so a next stream that does reference those bytes fails with `Z_DATA_ERROR` where unaccelerated zlib would have decoded it. This is a limitation of offloading rather than of the pin: the bytes exist only in the output the accelerator already returned to the caller, and whether a later stream will reference them is unknowable while the previous one is still being decoded. The pin is what makes that case fail as a zlib data error on the stream that needs the history, instead of an accelerator decoding the lookback against an unrelated window and reporting success. Applications that carry decode history across streams this way should disable offload for those streams. Used as a plain restart — the history-independent case — `inflateResetKeep` decodes correctly on every path, but the pin still applies: the stream stays on zlib until an `inflateReset` lifts it, so an application that restarts per message with `inflateResetKeep` gets no acceleration for the life of that `z_stream`. Any IGZIP decompression state the stream held is released at the pin rather than kept for a path that can no longer be selected, and a later `inflateReset` builds a new one. +When a decode fails part-way through a stream, `inflate` reports `Z_DATA_ERROR` after handing back the bytes the engine had already decoded in that call, and every later call on the stream reports `Z_DATA_ERROR` too — the same way zlib latches a data error. The failed stream is *not* handed to zlib to finish: zlib's own inflate state never saw the earlier chunks, so the input left in `next_in` begins inside a deflate block it would read as the start of a new stream, which on a raw deflate stream means junk counted as output and, at some input alignments, a `Z_STREAM_END` on a short decode. A failure on the *first* call of a stream is different and still goes to zlib, which can take the stream from its start. Only a stateful engine can fail mid-stream, so this applies to IGZIP, including where QAT or IAA reached it through `igzip_fallback`. The latched calls are counted as `inflate_failed_stream_count` when built with `ENABLE_STATISTICS`, for the same reason `inflate_stream_end_count` exists: no engine executes them, and the per-engine counters still have to add up to `inflate_count`. +`inflateSync` is intercepted because it is the one call that legitimately resumes a stream somewhere other than where the shim left it: it searches the input for a full-flush point and discards the decode state, so decoding continues there with no history. A stream that has already had input consumed is pinned to zlib at that point, and any latched data error is cleared. zlib performed the search, so zlib's state is the one left at the flush point; an accelerator still holds the bits it read ahead of the failure, and whether it resumes correctly would depend on how far it happened to have read. As with the `inflateResetKeep` pin, the stream stays on zlib for the rest of its life, and any IGZIP decompression state it held is released at the next `inflate`. A stream that has consumed nothing keeps its engine — it is still at its own start. deflateCopy/inflateCopy are intercepted so that the copy gets its own per-stream state: zlib duplicates the stream it owns, but zlib-accel keys its own state on the `z_stream` pointer, so without this the copy would be unknown to the shim and silently run on zlib. The copy inherits the settings and execution path of the source, and for inflate it also gets an independent copy of the IGZIP decompression state, so either stream can be used, reset, or ended without affecting the other. `inflateCopy` is supported on every path; `deflateCopy` has one restriction, described under IGZIP above. utility functions diff --git a/statistics.cpp b/statistics.cpp index 8ade3ee..57bd584 100644 --- a/statistics.cpp +++ b/statistics.cpp @@ -16,16 +16,27 @@ using namespace config; const std::array stat_names{ - {"deflate_count", "deflate_error_count", - "deflate_qat_count", "deflate_qat_error_count", - "deflate_iaa_count", "deflate_iaa_error_count", - "deflate_igzip_count", "deflate_igzip_error_count", - "deflate_zlib_count", "deflate_stream_end_count", - "inflate_count", "inflate_error_count", - "inflate_qat_count", "inflate_qat_error_count", - "inflate_iaa_count", "inflate_iaa_error_count", - "inflate_igzip_count", "inflate_igzip_error_count", - "inflate_zlib_count", "inflate_stream_end_count"}}; + {"deflate_count", + "deflate_error_count", + "deflate_qat_count", + "deflate_qat_error_count", + "deflate_iaa_count", + "deflate_iaa_error_count", + "deflate_igzip_count", + "deflate_igzip_error_count", + "deflate_zlib_count", + "deflate_stream_end_count", + "inflate_count", + "inflate_error_count", + "inflate_qat_count", + "inflate_qat_error_count", + "inflate_iaa_count", + "inflate_iaa_error_count", + "inflate_igzip_count", + "inflate_igzip_error_count", + "inflate_zlib_count", + "inflate_stream_end_count", + "inflate_failed_stream_count"}}; thread_local std::array stats{}; diff --git a/statistics.h b/statistics.h index 3588368..4971280 100644 --- a/statistics.h +++ b/statistics.h @@ -11,9 +11,10 @@ #define VISIBLE_FOR_TESTING __attribute__((visibility("default"))) // *_STREAM_END_COUNT counts the calls answered from the terminal state a stream -// reached earlier, which no engine executes. Without it DEFLATE_COUNT and -// INFLATE_COUNT would no longer be the sum of their per-engine counters. Keep -// this enum and stat_names in statistics.cpp index-parallel. +// reached earlier, and INFLATE_FAILED_STREAM_COUNT the calls answered from a +// latched decode failure; no engine executes either. Without them DEFLATE_COUNT +// and INFLATE_COUNT would no longer be the sum of their per-engine counters. +// Keep this enum and stat_names in statistics.cpp index-parallel. enum class Statistic : size_t { DEFLATE_COUNT = 0, DEFLATE_ERROR_COUNT, @@ -35,6 +36,7 @@ enum class Statistic : size_t { INFLATE_IGZIP_ERROR_COUNT, INFLATE_ZLIB_COUNT, INFLATE_STREAM_END_COUNT, + INFLATE_FAILED_STREAM_COUNT, STATS_COUNT }; diff --git a/tests/zlib_accel_test.cpp b/tests/zlib_accel_test.cpp index c6a1dcb..e276343 100644 --- a/tests/zlib_accel_test.cpp +++ b/tests/zlib_accel_test.cpp @@ -8047,6 +8047,454 @@ TEST_F(GzipFileTest, GzwriteAndGzreadRejectALengthThatDoesNotFitInInt) { DestroyBlock(input); } +#ifdef USE_IGZIP +class InflateMidstreamErrorRegressionTest : public ::testing::Test { + protected: + void SetUp() override { + saved_use_zlib_uncompress_ = GetConfig(USE_ZLIB_UNCOMPRESS); + saved_use_iaa_uncompress_ = GetConfig(USE_IAA_UNCOMPRESS); + saved_use_qat_uncompress_ = GetConfig(USE_QAT_UNCOMPRESS); + saved_use_igzip_uncompress_ = GetConfig(USE_IGZIP_UNCOMPRESS); + saved_use_zlib_compress_ = GetConfig(USE_ZLIB_COMPRESS); + saved_use_iaa_compress_ = GetConfig(USE_IAA_COMPRESS); + saved_use_qat_compress_ = GetConfig(USE_QAT_COMPRESS); + saved_use_igzip_compress_ = GetConfig(USE_IGZIP_COMPRESS); + // Written unconditionally by SetCompressPath/SetUncompressPath as well. + saved_iaa_prepend_empty_block_ = GetConfig(IAA_PREPEND_EMPTY_BLOCK); + saved_qat_allow_chunking_ = GetConfig(QAT_COMPRESSION_ALLOW_CHUNKING); + saved_igzip_fallback_ = GetConfig(IGZIP_FALLBACK); + saved_ignore_dictionary_ = GetConfig(IGNORE_ZLIB_DICTIONARY); + } + + // Restored here rather than at the end of each helper: an ASSERT_* failure + // returns from the helper, which would skip an inline restore and leave the + // rest of the suite running on this test's configuration. + void TearDown() override { + SetConfig(USE_ZLIB_UNCOMPRESS, saved_use_zlib_uncompress_); + SetConfig(USE_IAA_UNCOMPRESS, saved_use_iaa_uncompress_); + SetConfig(USE_QAT_UNCOMPRESS, saved_use_qat_uncompress_); + SetConfig(USE_IGZIP_UNCOMPRESS, saved_use_igzip_uncompress_); + SetConfig(USE_ZLIB_COMPRESS, saved_use_zlib_compress_); + SetConfig(USE_IAA_COMPRESS, saved_use_iaa_compress_); + SetConfig(USE_QAT_COMPRESS, saved_use_qat_compress_); + SetConfig(USE_IGZIP_COMPRESS, saved_use_igzip_compress_); + SetConfig(IAA_PREPEND_EMPTY_BLOCK, saved_iaa_prepend_empty_block_); + SetConfig(QAT_COMPRESSION_ALLOW_CHUNKING, saved_qat_allow_chunking_); + SetConfig(IGZIP_FALLBACK, saved_igzip_fallback_); + SetConfig(IGNORE_ZLIB_DICTIONARY, saved_ignore_dictionary_); + } + + uint32_t saved_use_zlib_uncompress_ = 0; + uint32_t saved_use_iaa_uncompress_ = 0; + uint32_t saved_use_qat_uncompress_ = 0; + uint32_t saved_use_igzip_uncompress_ = 0; + uint32_t saved_use_zlib_compress_ = 0; + uint32_t saved_use_iaa_compress_ = 0; + uint32_t saved_use_qat_compress_ = 0; + uint32_t saved_use_igzip_compress_ = 0; + uint32_t saved_iaa_prepend_empty_block_ = 0; + uint32_t saved_qat_allow_chunking_ = 0; + uint32_t saved_igzip_fallback_ = 0; + uint32_t saved_ignore_dictionary_ = 0; +}; + +// Build a stream that decodes cleanly up to a point and is invalid after it. +// Z_SYNC_FLUSH ends the good part at a byte boundary with the stream still +// open, and the byte appended there opens a block whose type is the reserved +// 11b, which every conforming decoder must reject. The whole payload is +// decodable from the prefix, which is what makes the delivered byte count +// assertable. +static void BuildStreamWithInvalidTail(const char* input, size_t input_length, + int window_bits, + std::vector* stream_out) { + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + + z_stream stream; + memset(&stream, 0, sizeof(z_stream)); + ASSERT_EQ(deflateInit2(&stream, Z_DEFAULT_COMPRESSION, Z_DEFLATED, + window_bits, 8, Z_DEFAULT_STRATEGY), + Z_OK); + + std::vector buffer(deflateBound(&stream, input_length) + 4096); + stream.next_in = reinterpret_cast(const_cast(input)); + stream.avail_in = static_cast(input_length); + stream.next_out = buffer.data(); + stream.avail_out = static_cast(buffer.size()); + + ASSERT_EQ(deflate(&stream, Z_SYNC_FLUSH), Z_OK); + ASSERT_EQ(stream.avail_in, 0u); + const size_t prefix_length = buffer.size() - stream.avail_out; + // Z_DATA_ERROR, not Z_OK: the stream is deliberately left unfinished, which + // is what deflateEnd() reports when it is ended before Z_FINISH. + ASSERT_EQ(deflateEnd(&stream), Z_DATA_ERROR); + + buffer.resize(prefix_length + 64); + buffer[prefix_length] = 0x07; + for (size_t i = prefix_length + 1; i < buffer.size(); i++) { + buffer[i] = 0x5a; + } + *stream_out = buffer; +} + +// Feed a stream to inflate() in fixed-size chunks, stopping at the first error +// or at the end of the stream. Reports what the caller needs to judge the +// failure: whether a completion was ever claimed, and what the last call said. +// Resumable: the cursor comes from total_in, so a caller that resynchronized +// the stream in between continues from where the resync left it. +static void InflateInChunks(z_streamp stream, const std::vector& input, + size_t chunk_length, int* last_ret, + bool* saw_stream_end) { + *last_ret = Z_OK; + *saw_stream_end = false; + size_t fed = stream->total_in; + stream->avail_in = 0; + for (int guard = 0; guard < 1024; guard++) { + if (stream->avail_in == 0 && fed < input.size()) { + const size_t take = std::min(chunk_length, input.size() - fed); + stream->next_in = const_cast(input.data()) + fed; + stream->avail_in = static_cast(take); + fed += take; + } + *last_ret = inflate(stream, Z_NO_FLUSH); + if (*last_ret == Z_STREAM_END) { + *saw_stream_end = true; + return; + } + if (*last_ret < 0) { + return; + } + if (stream->avail_in == 0 && fed >= input.size() && + *last_ret == Z_BUF_ERROR) { + return; + } + } + FAIL() << "inflate() made no progress in 1024 calls"; +} + +// The finding: a decode failure part-way through a stream used to pin the +// stream to zlib and hand it the rest, but zlib's inflate state never saw the +// earlier chunks, so next_in pointed into the middle of a deflate block that +// zlib read as the start of a fresh stream. On a raw stream there is no header +// to reject that, so zlib emitted junk it counted as output and could report +// Z_STREAM_END -- a short decode reported as success -- and the output the +// failing call had already decoded was dropped on the way out. +static void RunMidstreamInflateErrorRegression(ExecutionPath accel_path, + int window_bits, uint32_t seed) { + const size_t input_length = 96 * 1024; + char* input = GenerateSeededCompressibleBlock(input_length, seed); + ASSERT_NE(input, nullptr); + + std::vector compressed; + BuildStreamWithInvalidTail(input, input_length, window_bits, &compressed); + ASSERT_FALSE(compressed.empty()); + + SetUncompressPath(accel_path, /*zlib_fallback=*/true, false); + if (accel_path != IGZIP) { + SetConfig(USE_IGZIP_UNCOMPRESS, 1); + SetConfig(IGZIP_FALLBACK, 1); + } + + z_stream stream; + memset(&stream, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&stream, window_bits), Z_OK); + + std::vector output(input_length + 4096, 0); + stream.next_out = reinterpret_cast(output.data()); + stream.avail_out = static_cast(output.size()); + + int last_ret = Z_OK; + bool saw_stream_end = false; + // Chunked deliberately: one call per stream is the case the fall-through was + // written for, and it stays supported -- see the first-call test below. + InflateInChunks(&stream, compressed, 4096, &last_ret, &saw_stream_end); + + // The stream is broken, so the one answer that must never appear is success. + EXPECT_FALSE(saw_stream_end); + EXPECT_EQ(last_ret, Z_DATA_ERROR); + + // Everything before the invalid block is decodable, and an engine that + // decoded it has to hand it over -- zlib returns the decodable prefix and + // then the error too. + const size_t produced = output.size() - stream.avail_out; + EXPECT_EQ(produced, input_length); + EXPECT_EQ(memcmp(output.data(), input, std::min(produced, input_length)), 0); + + // Answered in place rather than delegated: a stream this far in cannot be + // handed to zlib at all, so the path must not have moved to ZLIB. + EXPECT_NE(GetInflateExecutionPath(&stream), ZLIB); + + // A failed stream stays failed. zlib holds its own state at BAD and keeps + // answering Z_DATA_ERROR; the engine here has no such state of its own. + stream.next_in = compressed.data(); + stream.avail_in = static_cast(compressed.size()); + EXPECT_EQ(inflate(&stream, Z_NO_FLUSH), Z_DATA_ERROR); + + EXPECT_EQ(inflateEnd(&stream), Z_OK); + DestroyBlock(input); +} + +// The other side of the same gate. When the very first call fails, next_in +// still addresses the first byte of the stream, so zlib can take it from the +// start -- which is what the fall-through was for, and it has to keep working. +static void RunFirstCallInflateErrorRegression(ExecutionPath accel_path, + int window_bits, uint32_t seed) { + const size_t input_length = 32 * 1024; + char* input = GenerateSeededCompressibleBlock(input_length, seed); + ASSERT_NE(input, nullptr); + + std::vector compressed; + BuildStreamWithInvalidTail(input, input_length, window_bits, &compressed); + ASSERT_FALSE(compressed.empty()); + // Smash the first block instead of the appended one, leaving the zlib header + // valid so the stream is refused for its content rather than its format. + const size_t header_length = (window_bits >= 8) ? 2 : 0; + ASSERT_GT(compressed.size(), header_length); + compressed[header_length] = 0x07; + + SetUncompressPath(accel_path, /*zlib_fallback=*/true, false); + if (accel_path != IGZIP) { + SetConfig(USE_IGZIP_UNCOMPRESS, 1); + SetConfig(IGZIP_FALLBACK, 1); + } + + z_stream stream; + memset(&stream, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&stream, window_bits), Z_OK); + + std::vector output(input_length + 4096, 0); + stream.next_out = reinterpret_cast(output.data()); + stream.avail_out = static_cast(output.size()); + + int last_ret = Z_OK; + bool saw_stream_end = false; + InflateInChunks(&stream, compressed, 4096, &last_ret, &saw_stream_end); + + EXPECT_FALSE(saw_stream_end); + EXPECT_EQ(last_ret, Z_DATA_ERROR); + EXPECT_EQ(output.size() - stream.avail_out, 0u); + // Nothing was consumed, so this is the one case zlib can still be given. + EXPECT_EQ(GetInflateExecutionPath(&stream), ZLIB); + + EXPECT_EQ(inflateEnd(&stream), Z_OK); + DestroyBlock(input); +} + +// inflateSync() is the way back from a failed stream: it skips to the next +// full-flush point and decoding resumes there. Recovery has to survive the +// mid-stream gate above -- the resync clears the latched error, or a stream +// that resynchronized successfully would answer Z_DATA_ERROR forever. +static void RunInflateSyncAfterMidstreamErrorRegression( + ExecutionPath accel_path, int window_bits, uint32_t seed) { + const size_t segment_length = 32 * 1024; + const size_t input_length = 3 * segment_length; + char* input = GenerateSeededCompressibleBlock(input_length, seed); + ASSERT_NE(input, nullptr); + + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + + z_stream cstream; + memset(&cstream, 0, sizeof(z_stream)); + ASSERT_EQ(deflateInit2(&cstream, Z_DEFAULT_COMPRESSION, Z_DEFLATED, + window_bits, 8, Z_DEFAULT_STRATEGY), + Z_OK); + + std::vector compressed(deflateBound(&cstream, input_length) + 4096); + cstream.next_out = compressed.data(); + cstream.avail_out = static_cast(compressed.size()); + + // Z_FULL_FLUSH discards the window, so each segment decodes from its own + // start. That is what a resync can recover to. + size_t mark[3] = {0, 0, 0}; + for (int segment = 0; segment < 3; segment++) { + cstream.next_in = + reinterpret_cast(input) + segment * segment_length; + cstream.avail_in = static_cast(segment_length); + const int ret = deflate(&cstream, segment == 2 ? Z_FINISH : Z_FULL_FLUSH); + ASSERT_EQ(ret, segment == 2 ? Z_STREAM_END : Z_OK); + mark[segment] = compressed.size() - cstream.avail_out; + } + const size_t compressed_length = compressed.size() - cstream.avail_out; + ASSERT_EQ(deflateEnd(&cstream), Z_OK); + compressed.resize(compressed_length); + + // Smash the middle segment from its byte-aligned start, keeping its trailing + // flush marker so the resync still has a point to find. + compressed[mark[0]] = 0x07; + for (size_t i = mark[0] + 1; i + 8 < mark[1]; i++) { + compressed[i] = 0x5a; + } + + SetUncompressPath(accel_path, /*zlib_fallback=*/true, false); + if (accel_path != IGZIP) { + SetConfig(USE_IGZIP_UNCOMPRESS, 1); + SetConfig(IGZIP_FALLBACK, 1); + } + + z_stream stream; + memset(&stream, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&stream, window_bits), Z_OK); + + std::vector output(input_length + 4096, 0); + stream.next_out = reinterpret_cast(output.data()); + stream.avail_out = static_cast(output.size()); + + int last_ret = Z_OK; + bool saw_stream_end = false; + InflateInChunks(&stream, compressed, 4096, &last_ret, &saw_stream_end); + ASSERT_EQ(last_ret, Z_DATA_ERROR); + const size_t produced_before_sync = output.size() - stream.avail_out; + ASSERT_GE(produced_before_sync, segment_length); + EXPECT_EQ(memcmp(output.data(), input, segment_length), 0); + + // inflateSync() searches the input it is given, and the flush point it needs + // is past what the failing call had been fed. + stream.next_in = compressed.data() + stream.total_in; + stream.avail_in = static_cast(compressed.size() - stream.total_in); + ASSERT_EQ(inflateSync(&stream), Z_OK); + + // Segment 3 comes back, which it cannot if the resync left the error latched. + InflateInChunks(&stream, compressed, 4096, &last_ret, &saw_stream_end); + EXPECT_NE(last_ret, Z_DATA_ERROR); + const size_t produced = output.size() - stream.avail_out; + ASSERT_GE(produced, segment_length); + EXPECT_EQ(memcmp(output.data() + produced - segment_length, + input + 2 * segment_length, segment_length), + 0); + // The resync is the one mid-stream handoff to zlib, and the only engine that + // can honor it: zlib performed the search, so its state is the one left at + // the flush point. A backend still holds the bits it read ahead of the + // failure. + EXPECT_EQ(GetInflateExecutionPath(&stream), ZLIB); + + EXPECT_EQ(inflateEnd(&stream), Z_OK); + DestroyBlock(input); +} + +// The other way a stream reaches the gate with input already consumed: a zlib +// header delivered one byte per call. FDICT lives in the second byte, so the +// first call is the shim's only chance to keep the stream off a backend, and it +// cannot yet see the bit. A backend given that byte swallows it into its own +// header buffer and reports the dictionary on the next call, by which time the +// bytes zlib would need to parse the header itself are gone from the caller's +// buffer -- so the request has to be answered the way zlib answers it, which +// means not offloading the byte at all. +static void RunFragmentedZlibHeaderDictionaryRegression( + ExecutionPath accel_path, uint32_t seed) { + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + SetUncompressPath(accel_path, /*zlib_fallback=*/true, false); + SetConfig(IGNORE_ZLIB_DICTIONARY, 0); + + const size_t input_length = 32 * 1024; + char* input = GenerateSeededCompressibleBlock(input_length, seed); + ASSERT_NE(input, nullptr); + + const unsigned char dict[] = "fragmented-header-preset-dictionary"; + const uInt dict_length = static_cast(sizeof(dict) - 1); + + z_stream cstream; + memset(&cstream, 0, sizeof(z_stream)); + ASSERT_EQ(deflateInit2(&cstream, Z_DEFAULT_COMPRESSION, Z_DEFLATED, 15, 8, + Z_DEFAULT_STRATEGY), + Z_OK); + ASSERT_EQ(deflateSetDictionary(&cstream, dict, dict_length), Z_OK); + + std::vector compressed(deflateBound(&cstream, input_length) + 4096); + cstream.next_in = reinterpret_cast(input); + cstream.avail_in = static_cast(input_length); + cstream.next_out = compressed.data(); + cstream.avail_out = static_cast(compressed.size()); + ASSERT_EQ(deflate(&cstream, Z_FINISH), Z_STREAM_END); + compressed.resize(compressed.size() - cstream.avail_out); + ASSERT_EQ(deflateEnd(&cstream), Z_OK); + ASSERT_GT(compressed.size(), 2u); + // FDICT, second header byte -- the bit the first call cannot see. + ASSERT_NE(compressed[1] & 0x20, 0); + + z_stream stream; + memset(&stream, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&stream, 15), Z_OK); + + std::vector output(input_length + 4096, 0); + stream.next_out = reinterpret_cast(output.data()); + stream.avail_out = static_cast(output.size()); + + size_t fed = 0; + int ret = Z_OK; + bool asked_for_dictionary = false; + for (size_t guard = 0; guard < 4 * compressed.size() + 64; guard++) { + if (stream.avail_in == 0 && fed < compressed.size()) { + stream.next_in = compressed.data() + fed; + stream.avail_in = 1; + fed++; + } + ret = inflate(&stream, Z_NO_FLUSH); + if (ret == Z_NEED_DICT) { + asked_for_dictionary = true; + ASSERT_EQ(inflateSetDictionary(&stream, dict, dict_length), Z_OK); + continue; + } + ASSERT_NE(ret, Z_DATA_ERROR); + if (ret == Z_STREAM_END || ret == Z_BUF_ERROR) { + break; + } + } + + // zlib asks for the dictionary and then decodes the stream; the pin is what + // lets the shim do the same. + EXPECT_TRUE(asked_for_dictionary); + EXPECT_EQ(ret, Z_STREAM_END); + EXPECT_EQ(stream.total_out, input_length); + EXPECT_EQ(memcmp(output.data(), input, input_length), 0); + EXPECT_EQ(GetInflateExecutionPath(&stream), ZLIB); + + EXPECT_EQ(inflateEnd(&stream), Z_OK); + DestroyBlock(input); +} + +TEST_F(InflateMidstreamErrorRegressionTest, IGZIPRawErrorIsNotReportedAsEnd) { + RunMidstreamInflateErrorRegression(IGZIP, -15, /*seed=*/0x11f4); +} + +TEST_F(InflateMidstreamErrorRegressionTest, + IGZIPZlibErrorDeliversDecodedBytes) { + RunMidstreamInflateErrorRegression(IGZIP, 15, /*seed=*/0x11f5); +} + +TEST_F(InflateMidstreamErrorRegressionTest, + IGZIPGzipErrorDeliversDecodedBytes) { + RunMidstreamInflateErrorRegression(IGZIP, 31, /*seed=*/0x11f6); +} + +TEST_F(InflateMidstreamErrorRegressionTest, + IGZIPFirstCallErrorStillReachesZlib) { + RunFirstCallInflateErrorRegression(IGZIP, 15, /*seed=*/0x11f7); +} + +TEST_F(InflateMidstreamErrorRegressionTest, + IGZIPInflateSyncRecoversAfterError) { + RunInflateSyncAfterMidstreamErrorRegression(IGZIP, -15, /*seed=*/0x11f8); +} + +TEST_F(InflateMidstreamErrorRegressionTest, + IGZIPFragmentedZlibHeaderStillAsksForDictionary) { + RunFragmentedZlibHeaderDictionaryRegression(IGZIP, /*seed=*/0x11fb); +} + +#ifdef USE_QAT +TEST_F(InflateMidstreamErrorRegressionTest, + QATFallbackRawErrorIsNotReportedAsEnd) { + RunMidstreamInflateErrorRegression(QAT, -15, /*seed=*/0x11f9); +} +#endif + +#ifdef USE_IAA +TEST_F(InflateMidstreamErrorRegressionTest, + IAAFallbackRawErrorIsNotReportedAsEnd) { + RunMidstreamInflateErrorRegression(IAA, -15, /*seed=*/0x11fa); +} +#endif +#endif // USE_IGZIP + class ShardedMapTest : public ::testing::Test {}; TEST_F(ShardedMapTest, BasicSetAndGet) { diff --git a/zlib_accel.cpp b/zlib_accel.cpp index e4aea5e..6ea4c77 100644 --- a/zlib_accel.cpp +++ b/zlib_accel.cpp @@ -71,6 +71,7 @@ static int (*orig_inflateReset)(z_streamp strm); static int (*orig_inflateResetKeep)(z_streamp strm); static int (*orig_inflateReset2)(z_streamp strm, int windowBits); static int (*orig_inflateCopy)(z_streamp dest, z_streamp source); +static int (*orig_inflateSync)(z_streamp strm); static int (*orig_compress)(Bytef* dest, uLongf* destLen, const Bytef* source, uLong sourceLen); static int (*orig_compress2)(Bytef* dest, uLongf* destLen, const Bytef* source, @@ -181,6 +182,8 @@ static int init_zlib_accel(void) { LOAD_SYMBOL(orig_inflateCopy, int (*)(z_streamp, z_streamp), "inflateCopy"); + LOAD_SYMBOL(orig_inflateSync, int (*)(z_streamp), "inflateSync"); + // Load compress/uncompress functions LOAD_SYMBOL(orig_compress, int (*)(Bytef*, uLongf*, const Bytef*, uLong), "compress"); @@ -325,6 +328,16 @@ struct InflateSettings { struct inflate_state* isal_strm = nullptr; // See DeflateSettings::stream_end_reached. bool stream_end_reached = false; + // Set once an offloaded call has consumed input from this stream. From that + // point on next_in addresses the middle of a stream zlib's own inflate state + // has never seen, so the remainder cannot be handed to orig_inflate: it would + // be parsed as the start of a new stream, with no history behind it. + bool bytes_consumed = false; + // Set once a mid-stream decode failure has been reported. zlib latches a data + // error and answers every later call with it until inflateSync() or a reset + // clears the state; ISA-L does not, and keeps parsing whatever follows the + // rejected bytes as a new block header, so the latch has to live here. + bool data_error = false; }; // isal_strm is a raw pointer, so destroying a settings object does not free the @@ -439,6 +452,8 @@ class InflateStreamSettings { settings->path = source.path; settings->isal_strm = isal_clone; settings->stream_end_reached = source.stream_end_reached; + settings->bytes_consumed = source.bytes_consumed; + settings->data_error = source.data_error; map.Set(dest, std::move(settings)); } catch (...) { Log(LogLevel::LOG_ERROR, @@ -522,6 +537,8 @@ static void ResetInflateStreamState( } SetInflatePath(settings, UNDEFINED); settings->stream_end_reached = false; + settings->bytes_consumed = false; + settings->data_error = false; if (settings->isal_strm != nullptr) { #ifdef USE_IGZIP ResetUncompressIGZIP(settings->isal_strm); @@ -529,6 +546,49 @@ static void ResetInflateStreamState( } } +// Hand the caller what an offloaded inflate call consumed and produced. Every +// such call goes through here, so this is also where a stream is recorded as +// consumed from -- zlib's inflate state saw none of these bytes. +static void AdvanceInflateStream( + z_streamp strm, const std::shared_ptr& settings, + uint32_t input_len, uint32_t output_len) { + strm->next_in += input_len; + strm->avail_in -= input_len; + strm->total_in += input_len; + strm->next_out += output_len; + strm->avail_out -= output_len; + strm->total_out += output_len; + if (input_len > 0 && settings != nullptr) { + settings->bytes_consumed = true; + } +} + +#ifdef USE_IGZIP +// An IGZIP decode failure part-way through a stream cannot be handed to zlib. +// zlib's inflate state saw none of the earlier chunks, so the bytes left in +// next_in begin inside a deflate block with no history behind them: zlib parses +// them as a fresh stream, which a zlib or gzip header check rejects but raw +// deflate cannot, and whatever ISA-L had already decoded in the failing call is +// dropped on the way. Deliver that output -- zlib hands back the decodable +// prefix before reporting the error too -- and report the error here. +// +// A stream that has consumed nothing yet is the case the fall-through was +// written for: next_in still addresses the first byte, so zlib can take the +// whole stream from the start, and such a call is left to do exactly that. +// Returns true when the error was answered here. +static bool HandleMidStreamIGZIPInflateError( + z_streamp strm, const std::shared_ptr& settings, + uint32_t input_len, uint32_t output_len, int* ret) { + if (settings == nullptr || !settings->bytes_consumed) { + return false; + } + AdvanceInflateStream(strm, settings, input_len, output_len); + settings->data_error = true; + *ret = Z_DATA_ERROR; + return true; +} +#endif // USE_IGZIP + // zlib's Z_NO_COMPRESSION (0) asks for stored, uncompressed deflate blocks. No // backend can produce those: ISA-L's level 0 is still LZ77+Huffman ("fastest"), // and QAT and IAA take no level argument at all -- all three would silently @@ -947,6 +1007,26 @@ int ZEXPORT deflate(z_streamp strm, int flush) { } } +#ifdef USE_IGZIP + // The compress-side counterpart of inflate()'s mid-stream gate. A failed + // IGZIP call on a stream that has already emitted bytes must not reach + // orig_deflate: zlib's deflate state holds none of that stream, so it would + // start a second one -- a fresh header -- in the middle of the output and + // report Z_OK, the worst shape available. Nothing reaches this today, and the + // reason is an ISA-L internal rather than a contract: isal_deflate refuses + // only flush values CompressIGZIP screens beforehand and a level/level_buf + // mismatch fixed at InitCompressIGZIP, which is never rebuilt while ISA-L + // owns the stream. Refuse rather than rest on that. + if (!in_call && ret != 0 && strm->total_out > 0 && + IgzipOwnsDeflateStream(deflate_settings)) { + Log(LogLevel::LOG_ERROR, "deflate Line ", __LINE__, ", strm ", + static_cast(strm), + ", igzip mid-stream error, refusing to hand the stream to zlib\n"); + INCREMENT_STAT(DEFLATE_ERROR_COUNT); + return Z_STREAM_ERROR; + } +#endif + if (in_call || configs[USE_ZLIB_COMPRESS] || deflate_settings->path == ZLIB) { // Distinguish "no zlib to delegate to" from "zlib rejected the data": the // former is an unusable library, not a data problem. @@ -1211,6 +1291,21 @@ int ZEXPORT inflate(z_streamp strm, int flush) { return ret; } + // A stream that has already failed mid-stream stays failed. zlib holds its + // own state at BAD and answers Z_DATA_ERROR until inflateSync() or a reset + // clears it; the backend that failed here has no such state, so the flag is + // what makes a repeated call answer the same way instead of resuming on the + // bytes that follow the ones it rejected. Placed below the parameter checks + // above, which zlib also answers ahead of its own error state. + if (inflate_settings->data_error) { + Log(LogLevel::LOG_INFO, "inflate Line ", __LINE__, ", strm ", + static_cast(strm), ", stream already failed, return code ", + Z_DATA_ERROR, "\n"); + INCREMENT_STAT(INFLATE_FAILED_STREAM_COUNT); + INCREMENT_STAT(INFLATE_ERROR_COUNT); + return Z_DATA_ERROR; + } + int ret = 1; bool end_of_stream = true; bool iaa_available = false; @@ -1248,10 +1343,23 @@ int ZEXPORT inflate(z_streamp strm, int flush) { // Early detection: if this is a zlib-format stream with the FDICT bit set // in the header, pin to ZLIB immediately so dictionary streams never reach // any accelerator (QAT/IAA/IGZIP don't support preset dictionaries). + // + // A single byte of input is pinned too, because the bit lives in the second + // one and this is the only chance to see it: a backend consumes that byte + // into its own header buffer, so by the time it reports the dictionary the + // bytes zlib would need to parse the header itself are gone from the caller's + // buffer. zlib carries a partial header across calls in its bit buffer and + // asks for the dictionary once it holds all of it, which is why handing it + // the stream here is what makes a header split across calls behave. The cost + // is a stream whose first call happens to carry one byte staying on zlib even + // when it turns out to have no dictionary. An empty call is not pinned: it + // consumes nothing, so the next one still gets to look. if (!in_call && inflate_settings->path == UNDEFINED && inflate_settings->window_bits >= 8 && - inflate_settings->window_bits <= kWindowBitsZlib && strm->avail_in >= 2 && - (strm->next_in[1] & ZLIB_FDICT_MASK)) { + inflate_settings->window_bits <= kWindowBitsZlib && + strm->next_in != nullptr && + (strm->avail_in == 1 || + (strm->avail_in >= 2 && (strm->next_in[1] & ZLIB_FDICT_MASK)))) { SetInflatePath(inflate_settings, ZLIB); } @@ -1276,6 +1384,11 @@ int ZEXPORT inflate(z_streamp strm, int flush) { if (!in_call && strm->avail_in > 0 && inflate_settings->path != ZLIB) { uint32_t input_len = strm->avail_in; uint32_t output_len = strm->avail_out; +#ifdef USE_IGZIP + // Set when an IGZIP failure was answered in place rather than delegated, + // which is the one accelerator error that must not reach zlib below. + bool igzip_midstream_error = false; +#endif #ifdef USE_IAA iaa_available = configs[USE_IAA_UNCOMPRESS] && @@ -1362,9 +1475,17 @@ int ZEXPORT inflate(z_streamp strm, int flush) { Log(LogLevel::LOG_ERROR, " strm=", static_cast(strm), " source=igzip", " total_in=", strm->total_in, " total_out=", strm->total_out, " adler=", strm->adler, "\n"); - SetInflatePath(inflate_settings, ZLIB); + igzip_midstream_error = HandleMidStreamIGZIPInflateError( + strm, inflate_settings, input_len, output_len, &ret); + if (!igzip_midstream_error) { + SetInflatePath(inflate_settings, ZLIB); + } } else if (path_action == IGZIP_INFLATE_PATH_FALLBACK_DATA_ERROR) { - SetInflatePath(inflate_settings, ZLIB); + igzip_midstream_error = HandleMidStreamIGZIPInflateError( + strm, inflate_settings, input_len, output_len, &ret); + if (!igzip_midstream_error) { + SetInflatePath(inflate_settings, ZLIB); + } } else if (path_action == IGZIP_INFLATE_PATH_SET_IGZIP && inflate_settings->path != ZLIB) { SetInflatePath(inflate_settings, IGZIP); @@ -1400,9 +1521,17 @@ int ZEXPORT inflate(z_streamp strm, int flush) { " source=igzip (", (path_selected == QAT) ? "QAT" : "IAA", " fallback)", " total_in=", strm->total_in, " total_out=", strm->total_out, " adler=", strm->adler, "\n"); - SetInflatePath(inflate_settings, ZLIB); + igzip_midstream_error = HandleMidStreamIGZIPInflateError( + strm, inflate_settings, input_len, output_len, &ret); + if (!igzip_midstream_error) { + SetInflatePath(inflate_settings, ZLIB); + } } else if (path_action == IGZIP_INFLATE_PATH_FALLBACK_DATA_ERROR) { - SetInflatePath(inflate_settings, ZLIB); + igzip_midstream_error = HandleMidStreamIGZIPInflateError( + strm, inflate_settings, input_len, output_len, &ret); + if (!igzip_midstream_error) { + SetInflatePath(inflate_settings, ZLIB); + } } else if (path_action == IGZIP_INFLATE_PATH_SET_IGZIP && inflate_settings->path != ZLIB) { SetInflatePath(inflate_settings, IGZIP); @@ -1412,13 +1541,22 @@ int ZEXPORT inflate(z_streamp strm, int flush) { } #endif // USE_IGZIP accelerator fallback +#ifdef USE_IGZIP + if (igzip_midstream_error) { + Log(LogLevel::LOG_INFO, "inflate Line ", __LINE__, ", strm ", + static_cast(strm), ", igzip mid-stream error, return code ", + ret, ", bytes_in ", input_len, ", bytes_out ", output_len, + ", avail_in ", strm->avail_in, ", avail_out ", strm->avail_out, + ", path ", static_cast(inflate_settings->path), ", path_name ", + ExecutionPathName(inflate_settings->path), ", window_bits ", + inflate_settings->window_bits, "\n"); + INCREMENT_STAT(INFLATE_ERROR_COUNT); + return ret; + } +#endif + if (ret == 0) { - strm->next_in += input_len; - strm->avail_in -= input_len; - strm->total_in += input_len; - strm->next_out += output_len; - strm->avail_out -= output_len; - strm->total_out += output_len; + AdvanceInflateStream(strm, inflate_settings, input_len, output_len); if (end_of_stream) { ret = Z_STREAM_END; } else if (input_len > 0 || output_len > 0) { @@ -1448,8 +1586,10 @@ int ZEXPORT inflate(z_streamp strm, int flush) { // Z_DATA_ERROR reaches zlib: those path actions set the path to ZLIB and // leave ret non-zero, so the update block above is skipped and control // arrives here with strm->next_in never advanced. zlib therefore re-reads - // the original, untouched input. The fall-through is deliberate, not - // accidental. + // the original, untouched input. That holds only while the stream has + // consumed nothing, which is why the pin is now conditional: once an earlier + // call has advanced next_in, the input zlib would re-read starts inside a + // deflate block, and the failure is answered above instead of delegated. if (in_call || configs[USE_ZLIB_UNCOMPRESS] || inflate_settings->path == ZLIB) { // refer to comment in deflate @@ -1569,6 +1709,60 @@ int ZEXPORT inflateResetKeep(z_streamp strm) { return ret; } +// inflateSync() is the one legitimate mid-stream handoff to zlib, and the way +// back from a failed stream: it searches the input for a full-flush point, +// advances next_in to it and discards the decode state, so decoding resumes +// there with no history. That is a request zlib has just satisfied on its own +// state -- the search is what leaves it at a block boundary in TYPE mode -- so +// the stream is pinned to zlib for the rest of its life and the latched data +// error is cleared with it. +// +// The pin is what makes the resume work at all. A backend cannot honor it: the +// bits it buffered ahead of the failure are still in its own state, and next_in +// moving underneath it changes nothing, so whether it resumes correctly depends +// on how far it had read -- ISA-L picks up the search only when the failure +// happened to leave it byte-aligned at the flush point. zlib, which found the +// point, always can, and everything after a full flush is self-contained by +// construction. A stream that has consumed nothing keeps its path: it is at its +// own start, where every engine can still take it. +int ZEXPORT inflateSync(z_streamp strm) { + Log(LogLevel::LOG_INFO, "inflateSync Line ", __LINE__, ", strm ", + static_cast(strm), ", avail_in ", + strm != nullptr ? strm->avail_in : 0, "\n"); + + if (orig_inflateSync == nullptr) { + return Z_VERSION_ERROR; + } + + // Decide before calling zlib, not after. zlib builds inflateSync() on top of + // inflateReset(), so on a libz whose internal call is interposable that call + // lands in this file's inflateReset() and clears the very fields the decision + // reads -- and this is the one entry point whose whole purpose is to pin. + auto inflate_settings = inflate_stream_settings.Get(strm); + const bool pin_to_zlib = inflate_settings != nullptr && + inflate_settings->bytes_consumed && + !inflate_settings->stream_end_reached; + + const int ret = orig_inflateSync(strm); + + // Z_OK is a completed search and Z_DATA_ERROR one that ran out of input + // looking; both leave zlib's state holding the outcome, and zlib answers a + // failed search with Z_STREAM_ERROR on every later inflate() call, which is + // its own state to report. Z_BUF_ERROR and Z_STREAM_ERROR are refusals that + // change nothing, so a stream that only got those keeps its engine. + if (ret != Z_OK && ret != Z_DATA_ERROR) { + return ret; + } + + if (pin_to_zlib) { + inflate_settings->data_error = false; + // Any ISA-L stream this puts out of reach is handed back by the next + // inflate(), the same way inflateResetKeep()'s pin releases it. + SetInflatePath(inflate_settings, ZLIB); + } + return ret; +} + // inflateReset2() is the only zlib entry point that changes windowBits on a // live stream, so it is the only one that can restart a finished stream without // going through inflateReset(). It has to be intercepted for two reasons: the