| 84 | } |
| 85 | |
| 86 | Status GzipDecompressor::ProcessBlockStreaming(int64_t input_length, const uint8_t* input, |
| 87 | int64_t* input_bytes_read, int64_t* output_length, uint8_t** output, |
| 88 | bool* stream_end) { |
| 89 | if (!reuse_buffer_ || out_buffer_ == nullptr) { |
| 90 | buffer_length_ = STREAM_OUT_BUF_SIZE; |
| 91 | out_buffer_ = memory_pool_->TryAllocate(buffer_length_); |
| 92 | if (UNLIKELY(out_buffer_ == nullptr)) { |
| 93 | string details = Substitute(DECOMPRESSOR_MEM_LIMIT_EXCEEDED, "Gzip", |
| 94 | buffer_length_); |
| 95 | return memory_pool_->mem_tracker()->MemLimitExceeded( |
| 96 | nullptr, details, buffer_length_); |
| 97 | } |
| 98 | } |
| 99 | *output = out_buffer_; |
| 100 | |
| 101 | stream_.next_in = const_cast<Bytef*>(reinterpret_cast<const Bytef*>(input)); |
| 102 | stream_.avail_in = input_length; |
| 103 | stream_.next_out = reinterpret_cast<Bytef*>(*output); |
| 104 | stream_.avail_out = buffer_length_; |
| 105 | |
| 106 | *stream_end = false; |
| 107 | *input_bytes_read = 0; |
| 108 | *output_length = 0; |
| 109 | while (stream_.avail_out > 0 && stream_.avail_in > 0) { |
| 110 | *stream_end = false; |
| 111 | // inflate() performs one or both of the following actions: |
| 112 | // Decompress more input starting at next_in and update next_in and avail_in |
| 113 | // accordingly. |
| 114 | // Provide more output starting at next_out and update next_out and avail_out |
| 115 | // accordingly. |
| 116 | // inflate() returns Z_OK if some progress has been made (more input processed |
| 117 | // or more output produced) |
| 118 | int ret = inflate(&stream_, Z_SYNC_FLUSH); |
| 119 | *input_bytes_read = input_length - stream_.avail_in; |
| 120 | *output_length = buffer_length_ - stream_.avail_out; |
| 121 | VLOG_ROW << "inflate() ret=" << ret << " consumed=" << *input_bytes_read |
| 122 | << " produced=" << *output_length << " stream: " << DebugStreamState(); |
| 123 | |
| 124 | if (ret == Z_DATA_ERROR) { |
| 125 | return Status(TErrorCode::COMPRESSED_FILE_BLOCK_CORRUPTED, "Gzip"); |
| 126 | } else if (ret == Z_BUF_ERROR) { |
| 127 | // Z_BUF_ERROR indicates that inflate() could not consume more input or |
| 128 | // produce more output. inflate() can be called again with more output space |
| 129 | // or more available input. |
| 130 | VLOG_ROW << "inflate() ret=" << ret << ", cannot make progress, need more input"; |
| 131 | return Status::OK(); |
| 132 | } else if (ret == Z_STREAM_END) { |
| 133 | *stream_end = true; |
| 134 | ret = inflateReset(&stream_); |
| 135 | if (ret != Z_OK) { |
| 136 | return Status(TErrorCode::COMPRESSED_FILE_DECOMPRESSOR_ERROR, "Gzip", |
| 137 | "inflateReset()", ret); |
| 138 | } |
| 139 | } else if (ret != Z_OK) { |
| 140 | return Status(TErrorCode::COMPRESSED_FILE_DECOMPRESSOR_ERROR, "Gzip", |
| 141 | "inflate()", ret); |
| 142 | } |
| 143 | DCHECK_EQ(ret, Z_OK); |