diff --git a/iaa.cpp b/iaa.cpp index 3acb2c4..50c10ab 100644 --- a/iaa.cpp +++ b/iaa.cpp @@ -197,7 +197,8 @@ int CompressIAA(uint8_t* input, uint32_t* input_length, uint8_t* output, int UncompressIAA(uint8_t* input, uint32_t* input_length, uint8_t* output, uint32_t* output_length, qpl_path_t execution_path, - int window_bits, bool* end_of_stream, bool detect_gzip_ext) { + int window_bits, bool* end_of_stream, bool detect_gzip_ext, + bool* window_too_large) { Log(LogLevel::LOG_INFO, "UncompressIAA() Line ", __LINE__, " input_length ", *input_length, "\n"); @@ -235,6 +236,14 @@ int UncompressIAA(uint8_t* input, uint32_t* input_length, uint8_t* output, qpl_status status = qpl_execute_job(job); if (status != QPL_STS_OK && status != QPL_STS_MORE_OUTPUT_NEEDED) { + // QPL_STS_BAD_DIST_ERR means the stream referenced a match further back + // than IAA's 4 kB history buffer. Unlike the other failures this one is not + // about this call: it says the producer used a larger window, and every + // remaining block of the same stream will be rejected for the same reason. + // Report it separately so the caller can stop submitting. + if (status == QPL_STS_BAD_DIST_ERR && window_too_large != nullptr) { + *window_too_large = true; + } Log(LogLevel::LOG_ERROR, "UncompressIAA() Line ", __LINE__, " qpl_execute_job status ", status, "\n"); return 1; diff --git a/iaa.h b/iaa.h index 94e1b5b..77d4e73 100644 --- a/iaa.h +++ b/iaa.h @@ -55,10 +55,18 @@ int CompressIAA(uint8_t* input, uint32_t* input_length, uint8_t* output, int window_bits, uint32_t max_compressed_size = 0, bool gzip_ext = false); -int UncompressIAA(uint8_t* input, uint32_t* input_length, uint8_t* output, - uint32_t* output_length, qpl_path_t execution_path, - int window_bits, bool* end_of_stream, - bool detect_gzip_ext = false); +// window_too_large, when non-null, is set to true if the job was rejected +// because the stream references match distances beyond IAA's fixed 4 kB history +// buffer (QPL_STS_BAD_DIST_ERR). That is a property of whichever compressor +// produced the stream, not of the individual block, so a caller that sees it +// can stop offering the rest of that stream to IAA. It is never set to false; +// the caller owns initialisation. +VISIBLE_FOR_TESTING int UncompressIAA(uint8_t* input, uint32_t* input_length, + uint8_t* output, uint32_t* output_length, + qpl_path_t execution_path, + int window_bits, bool* end_of_stream, + bool detect_gzip_ext = false, + bool* window_too_large = nullptr); VISIBLE_FOR_TESTING bool SupportedOptionsIAA(int window_bits, uint32_t input_length, diff --git a/tests/zlib_accel_test.cpp b/tests/zlib_accel_test.cpp index c6a1dcb..f1e4c62 100644 --- a/tests/zlib_accel_test.cpp +++ b/tests/zlib_accel_test.cpp @@ -6773,6 +6773,287 @@ TEST_F(ConfigLoaderTest, MapShardsInvalidNonPowerOfTwo) { SetConfig(MAP_SHARDS, saved_shards); } +#ifdef USE_IAA +// IAA's decompressor has a fixed 4 kB history buffer, so it cannot decode a +// stream whose producer used a larger window -- zlib's default is 32 kB. QPL +// reports that as QPL_STS_BAD_DIST_ERR, and it is the one decompress failure +// that predicts the next call: the window belongs to the compressor, not to the +// block. IsIAADecompressible() cannot see it for raw deflate or gzip, where +// there is no header to read the window out of, so the only way to know is to +// be told once and remember. +class IAAWindowRejectionTest : public ::testing::Test {}; + +// Run one whole stream through strm and check the bytes. Returns the last +// inflate() code so the caller can assert on it, or Z_DATA_ERROR if the output +// came back wrong. +int InflateWholeStream(z_streamp strm, const std::string& compressed, + const char* expected, size_t expected_length) { + std::vector output(expected_length + 1024); + strm->next_in = + reinterpret_cast(const_cast(compressed.data())); + strm->avail_in = static_cast(compressed.size()); + strm->next_out = output.data(); + strm->avail_out = static_cast(output.size()); + int ret = Z_OK; + for (int guard = 0; guard < 128; guard++) { + ret = inflate(strm, Z_NO_FLUSH); + if (ret != Z_OK && ret != Z_BUF_ERROR) { + break; + } + } + if (ret != Z_STREAM_END) { + return ret; + } + if (strm->total_out != expected_length || + memcmp(output.data(), expected, expected_length) != 0) { + return Z_DATA_ERROR; + } + return Z_STREAM_END; +} + +// Every case below except the first needs QPL to get far enough into a job to +// report the oversized window. Without a usable device it fails at job +// initialization instead, which leaves the flag correctly clear -- so the test +// would be asserting the opposite of what it means to. Probe with a stream IAA +// can definitely decode: a short one, whose matches cannot reach back 4 kB +// because the whole payload is smaller than that. +bool IAAHardwareDecompressWorks() { + const size_t input_length = 2048; + char* input = GenerateSeededCompressibleBlock(input_length, /*seed=*/0x4144); + if (input == nullptr) { + return false; + } + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + std::string compressed; + size_t output_upper_bound = 0; + ExecutionPath compress_path = UNDEFINED; + int ret = ZlibCompress(input, input_length, &compressed, -15, Z_FINISH, + &output_upper_bound, &compress_path); + DestroyBlock(input); + if (ret != Z_STREAM_END) { + return false; + } + + std::vector output(input_length + 1024); + uint32_t input_len = static_cast(compressed.size()); + uint32_t output_len = static_cast(output.size()); + bool end_of_stream = false; + ret = UncompressIAA(reinterpret_cast(&compressed[0]), &input_len, + output.data(), &output_len, qpl_path_hardware, + /*window_bits=*/-15, &end_of_stream); + return ret == 0 && end_of_stream && output_len == input_length; +} + +// The contract UncompressIAA() now offers its callers, checked on QPL's +// software path so that it holds on a host with no device: the software path +// rejects an oversized window for the same reason and with the same status. +TEST_F(IAAWindowRejectionTest, UncompressIAAReportsOversizedWindow) { + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + + const size_t input_length = 64 * 1024; + char* input = GenerateSeededCompressibleBlock(input_length, /*seed=*/0x7e11); + ASSERT_NE(input, nullptr); + + // Two encodings of the same bytes. GenerateSeededCompressibleBlock() repeats + // a string every 8192 bytes, so with zlib's full window the first is + // guaranteed to contain a match distance IAA cannot reach; restricted to 4 + // kB, the second cannot contain one. + std::string wide; + std::string narrow; + size_t output_upper_bound = 0; + ExecutionPath compress_path = UNDEFINED; + ASSERT_EQ(ZlibCompress(input, input_length, &wide, -15, Z_FINISH, + &output_upper_bound, &compress_path), + Z_STREAM_END); + ASSERT_EQ(ZlibCompress(input, input_length, &narrow, -12, Z_FINISH, + &output_upper_bound, &compress_path), + Z_STREAM_END); + + std::vector output(input_length + 1024); + uint32_t input_len = static_cast(wide.size()); + uint32_t output_len = static_cast(output.size()); + bool end_of_stream = false; + bool window_too_large = false; + EXPECT_NE(UncompressIAA(reinterpret_cast(&wide[0]), &input_len, + output.data(), &output_len, qpl_path_software, + /*window_bits=*/-15, &end_of_stream, + /*detect_gzip_ext=*/false, &window_too_large), + 0); + EXPECT_TRUE(window_too_large); + + // A stream IAA can follow decodes, and leaves the flag alone. Never setting + // it to false is what lets a caller pass one bool through a whole stream. + input_len = static_cast(narrow.size()); + output_len = static_cast(output.size()); + end_of_stream = false; + bool narrow_window_too_large = false; + EXPECT_EQ(UncompressIAA(reinterpret_cast(&narrow[0]), &input_len, + output.data(), &output_len, qpl_path_software, + /*window_bits=*/-12, &end_of_stream, + /*detect_gzip_ext=*/false, &narrow_window_too_large), + 0); + EXPECT_FALSE(narrow_window_too_large); + EXPECT_TRUE(end_of_stream); + EXPECT_EQ(output_len, input_length); + EXPECT_EQ(memcmp(output.data(), input, input_length), 0); + + // Omitting the out-parameter has to stay legal: most callers do not care + // which failure they got. + input_len = static_cast(wide.size()); + output_len = static_cast(output.size()); + end_of_stream = false; + EXPECT_NE(UncompressIAA(reinterpret_cast(&wide[0]), &input_len, + output.data(), &output_len, qpl_path_software, + /*window_bits=*/-15, &end_of_stream), + 0); + + DestroyBlock(input); +} + +// The point of the whole change. A rejection has to outlive inflateReset(), +// because a reset is exactly what the callers that matter do between documents: +// Lucene resets its Inflater once per stored field. Clearing the flag on reset +// would forget the lesson before it was ever acted on. +TEST_F(IAAWindowRejectionTest, StreamFlagSurvivesInflateReset) { + if (!IAAHardwareDecompressWorks()) { + GTEST_SKIP() << "no usable IAA device: QPL cannot reach the point where it " + "reports an oversized history window"; + } + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + SetUncompressPath(IAA, /*zlib_fallback=*/true, false); + + const size_t input_length = 64 * 1024; + char* input = GenerateSeededCompressibleBlock(input_length, /*seed=*/0x7e12); + ASSERT_NE(input, nullptr); + + std::string compressed; + size_t output_upper_bound = 0; + ExecutionPath compress_path = UNDEFINED; + ASSERT_EQ(ZlibCompress(input, input_length, &compressed, -15, Z_FINISH, + &output_upper_bound, &compress_path), + Z_STREAM_END); + + z_stream stream; + memset(&stream, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&stream, -15), Z_OK); + EXPECT_FALSE(InflateIAAWindowRejected(&stream)); + + // The rejection costs one submission and then falls through to zlib, so the + // bytes are still right. + EXPECT_EQ(InflateWholeStream(&stream, compressed, input, input_length), + Z_STREAM_END); + EXPECT_TRUE(InflateIAAWindowRejected(&stream)); + + ASSERT_EQ(inflateReset(&stream), Z_OK); + EXPECT_TRUE(InflateIAAWindowRejected(&stream)); + // inflateReset() clears the path, so a stream that had forgotten the + // rejection would be dispatched to IAA again here. + EXPECT_EQ(GetInflateExecutionPath(&stream), UNDEFINED); + EXPECT_EQ(InflateWholeStream(&stream, compressed, input, input_length), + Z_STREAM_END); + EXPECT_NE(GetInflateExecutionPath(&stream), IAA); + EXPECT_TRUE(InflateIAAWindowRejected(&stream)); + + // inflateReset2() restarts the stream in a new format, and is the other way + // back to an undefined path. + ASSERT_EQ(inflateReset2(&stream, -15), Z_OK); + EXPECT_TRUE(InflateIAAWindowRejected(&stream)); + + ASSERT_EQ(inflateEnd(&stream), Z_OK); + DestroyBlock(input); +} + +// A copy decodes the rest of the same stream, so it inherits the verdict. The +// settings are rebuilt member by member in SetFromCopy(), not assigned, so this +// is the kind of field that gets silently dropped. +TEST_F(IAAWindowRejectionTest, StreamFlagPropagatesThroughInflateCopy) { + if (!IAAHardwareDecompressWorks()) { + GTEST_SKIP() << "no usable IAA device: QPL cannot reach the point where it " + "reports an oversized history window"; + } + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + SetUncompressPath(IAA, /*zlib_fallback=*/true, false); + + const size_t input_length = 64 * 1024; + char* input = GenerateSeededCompressibleBlock(input_length, /*seed=*/0x7e13); + ASSERT_NE(input, nullptr); + + std::string compressed; + size_t output_upper_bound = 0; + ExecutionPath compress_path = UNDEFINED; + ASSERT_EQ(ZlibCompress(input, input_length, &compressed, -15, Z_FINISH, + &output_upper_bound, &compress_path), + Z_STREAM_END); + + z_stream source; + memset(&source, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&source, -15), Z_OK); + EXPECT_EQ(InflateWholeStream(&source, compressed, input, input_length), + Z_STREAM_END); + ASSERT_TRUE(InflateIAAWindowRejected(&source)); + + z_stream dest; + memset(&dest, 0, sizeof(z_stream)); + ASSERT_EQ(inflateCopy(&dest, &source), Z_OK); + EXPECT_TRUE(InflateIAAWindowRejected(&dest)); + // And the copy keeps it across its own reset, like the original. + ASSERT_EQ(inflateReset(&dest), Z_OK); + EXPECT_TRUE(InflateIAAWindowRejected(&dest)); + EXPECT_EQ(InflateWholeStream(&dest, compressed, input, input_length), + Z_STREAM_END); + EXPECT_NE(GetInflateExecutionPath(&dest), IAA); + + ASSERT_EQ(inflateEnd(&dest), Z_OK); + ASSERT_EQ(inflateEnd(&source), Z_OK); + DestroyBlock(input); +} + +// A stream IAA can serve must not be tarred by another stream's rejection: the +// flag is per stream, and there is no process-wide counter behind it. +TEST_F(IAAWindowRejectionTest, RejectionDoesNotAffectOtherStreams) { + if (!IAAHardwareDecompressWorks()) { + GTEST_SKIP() << "no usable IAA device: QPL cannot reach the point where it " + "reports an oversized history window"; + } + SetCompressPath(ZLIB, /*zlib_fallback=*/true, false, false); + SetUncompressPath(IAA, /*zlib_fallback=*/true, false); + + const size_t input_length = 64 * 1024; + char* input = GenerateSeededCompressibleBlock(input_length, /*seed=*/0x7e14); + ASSERT_NE(input, nullptr); + + std::string wide; + std::string narrow; + size_t output_upper_bound = 0; + ExecutionPath compress_path = UNDEFINED; + ASSERT_EQ(ZlibCompress(input, input_length, &wide, -15, Z_FINISH, + &output_upper_bound, &compress_path), + Z_STREAM_END); + ASSERT_EQ(ZlibCompress(input, input_length, &narrow, -12, Z_FINISH, + &output_upper_bound, &compress_path), + Z_STREAM_END); + + z_stream rejected; + memset(&rejected, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&rejected, -15), Z_OK); + EXPECT_EQ(InflateWholeStream(&rejected, wide, input, input_length), + Z_STREAM_END); + ASSERT_TRUE(InflateIAAWindowRejected(&rejected)); + + z_stream served; + memset(&served, 0, sizeof(z_stream)); + ASSERT_EQ(inflateInit2(&served, -15), Z_OK); + EXPECT_EQ(InflateWholeStream(&served, narrow, input, input_length), + Z_STREAM_END); + EXPECT_FALSE(InflateIAAWindowRejected(&served)); + EXPECT_EQ(GetInflateExecutionPath(&served), IAA); + + ASSERT_EQ(inflateEnd(&served), Z_OK); + ASSERT_EQ(inflateEnd(&rejected), Z_OK); + DestroyBlock(input); +} +#endif // USE_IAA + // The shim keeps per-stream state in maps keyed by z_streamp, and every entry // point that consumes that state has to cope with the entry being absent: a // stream that was never initialized at all, one whose *Init failed, or a diff --git a/zlib_accel.cpp b/zlib_accel.cpp index e4aea5e..016829b 100644 --- a/zlib_accel.cpp +++ b/zlib_accel.cpp @@ -325,6 +325,14 @@ struct InflateSettings { struct inflate_state* isal_strm = nullptr; // See DeflateSettings::stream_end_reached. bool stream_end_reached = false; + // Set once IAA has rejected a block of this stream for referencing a match + // beyond its 4 kB history buffer. Deliberately NOT cleared by inflateReset: + // the window is a property of the compressor that produced the bytes, and a + // reset starts a new stream from the same producer in every caller that + // matters here (Lucene resets its Inflater once per stored-field document). + // Clearing it would make the flag useless, since almost every rejection + // arrives on a stream that is about to be reset. + bool iaa_window_too_large = false; }; // isal_strm is a raw pointer, so destroying a settings object does not free the @@ -439,6 +447,12 @@ class InflateStreamSettings { settings->path = source.path; settings->isal_strm = isal_clone; settings->stream_end_reached = source.stream_end_reached; + // A copy decodes the rest of the same stream, so it inherits what IAA + // already said about that stream's history window. This has to be copied + // out by hand like every other field: the settings are rebuilt member by + // member here, not assigned, so a new member is silently dropped + // otherwise. + settings->iaa_window_too_large = source.iaa_window_too_large; map.Set(dest, std::move(settings)); } catch (...) { Log(LogLevel::LOG_ERROR, @@ -515,6 +529,14 @@ static void ResetDeflateStreamState( // Same for the inflate side. A reset stream is ready to decode again; leaving // the terminal state set would wedge every later inflate() at Z_STREAM_END. +// +// iaa_window_too_large deliberately does NOT belong here. It records what IAA +// said about the compressor that produced these bytes, and a reset stream is +// almost always the same caller decoding more output from the same producer -- +// Lucene resets its Inflater once per stored-field document. Clearing it here +// would make the flag useless: it would be forgotten before it was ever +// consulted, and the shim would go back to submitting jobs it knows will be +// rejected. static void ResetInflateStreamState( const std::shared_ptr& settings) { if (settings == nullptr) { @@ -1278,7 +1300,12 @@ int ZEXPORT inflate(z_streamp strm, int flush) { uint32_t output_len = strm->avail_out; #ifdef USE_IAA + // IsIAADecompressible cannot see match distances, so for raw deflate and + // gzip it has no header to read and is guessing. iaa_window_too_large is + // what a wrong guess, once made, costs being remembered: IAA has already + // told us this stream's producer used a window it cannot follow. iaa_available = configs[USE_IAA_UNCOMPRESS] && + !inflate_settings->iaa_window_too_large && SupportedOptionsIAA(inflate_settings->window_bits, input_len, output_len) && IsIAADecompressible(strm->next_in, input_len, @@ -1316,9 +1343,10 @@ int ZEXPORT inflate(z_streamp strm, int flush) { if (path_selected == IAA) { #ifdef USE_IAA in_call = true; - ret = UncompressIAA(strm->next_in, &input_len, strm->next_out, - &output_len, qpl_path_hardware, - inflate_settings->window_bits, &end_of_stream); + ret = UncompressIAA( + strm->next_in, &input_len, strm->next_out, &output_len, + qpl_path_hardware, inflate_settings->window_bits, &end_of_stream, + /*detect_gzip_ext=*/false, &inflate_settings->iaa_window_too_large); SetInflatePath(inflate_settings, IAA); // IAA inflate is stateless in this wrapper. If stream end was not // reached, use zlib for stateful continuation. @@ -1938,6 +1966,11 @@ bool InflateOwnsIgzipState(z_streamp strm) { return inflate_settings != nullptr && inflate_settings->isal_strm != nullptr; } +bool InflateIAAWindowRejected(z_streamp strm) { + auto inflate_settings = inflate_stream_settings.Get(strm); + return inflate_settings != nullptr && inflate_settings->iaa_window_too_large; +} + enum class FileMode { NONE, READ, WRITE, APPEND }; // What a gzopen/gzdopen mode string asks for beyond the open(2) flags. zlib @@ -2337,6 +2370,11 @@ static int GzreadAcceleratorUncompress(GzipFile* gz, uint8_t* input, bool igzip_available = false; #ifdef USE_IAA + // No remembered window rejection here, unlike inflate(): gzread already pins + // a file to zlib for good on its first accelerator failure of any kind (see + // use_zlib_for_decompression at the call site), and nothing clears that for + // the life of the GzipFile. One wasted submission per file is all this path + // can spend, and there is no second one to suppress. iaa_available = configs[USE_IAA_UNCOMPRESS] && SupportedOptionsIAA(kWindowBitsGzip, *input_length, *output_length) && diff --git a/zlib_accel.h b/zlib_accel.h index ed21080..e219779 100644 --- a/zlib_accel.h +++ b/zlib_accel.h @@ -23,4 +23,10 @@ ExecutionPath GetGzipFileExecutionPath(gzFile file); bool DeflateOwnsIgzipState(z_streamp strm); bool InflateOwnsIgzipState(z_streamp strm); +// True once IAA has rejected a block of this stream for referencing a match +// beyond its 4 kB history buffer. Tests need it because the record deliberately +// survives inflateReset(), and nothing else about the stream reveals that it is +// being kept. Always false in a build without IAA support. +bool InflateIAAWindowRejected(z_streamp strm); + #pragma GCC visibility pop