[Support] - Implemented zlib::StreamCompression class.

[grimar]

Summary

Patch introduces zlib streaming API. I am using it in LLD.
(https://reviews.llvm.org/D31941).

Patch implements only compression (no decompression),
because that is what I was need and has simple API that allows to feed compressor
object with chunks of data and get compressed result part by part from client.

Docs uded in implementation were:
http://zlib.net/zlib_how.html
http://zlib.net/manual.html

[grimar]

• Last minux fix for llvm_unreachable messages text.

[dblaikie]

Thanks for working on this - it'll be great to have streaming compression APIs for LLVM (& especially in llvm-dwp - it could use streaming compression (& streaming decompression, but having one side's a great start/example!))

include/llvm/Support/Compression.h

60

It'd be tidier if this could be Expected<StreamCompression> instead (avoiding the unique_ptr indirection, etc) & helpful if the StreamCompression class was movable.

67

Should the OutFn be part of the StreamCompression object - rather than passed in for each chunk of compression?

67

Also "IsLastChunk" is probably a somewhat clunky API - might be nicer to have a separate "finalize" function for once all the chunks are done?

72

Would SmallVector be useful here?

unittests/Support/CompressionTest.cpp

82

No need for this variable - you should be able to pass 'In' directly as the ArrayRef parameter and rely on implicit conversion.

105–106

Rather than writing a for loop, can you test that the uncompressed data is equal to the original input data?

[ruiu]

I wonder why you want a streaming compression support to write compressed .debug sections. All you need is to call zlib's compress2 only once for each section, no?

[grimar]

Thanks for review, David ! All comments addressed.

include/llvm/Support/Compression.h

60

Done.

67

Should the OutFn be part of the StreamCompression object - rather than passed in for each chunk of compression?

Yep, that simpler. Changed.

67

Also "IsLastChunk" is probably a somewhat clunky API - might be nicer to have a separate "finalize" function for once all the chunks are done?

Done.

72

I am not sure honestly. I would expect a few kilobytes output buffer size is reasonable, may be even more (I think about a megabyte for LLD use case). May be it is ok to have little buffer for some cases. Done.

unittests/Support/CompressionTest.cpp

82

Done.

105–106

Original input data is N input chunks in a row. So that anyways will be a loop here I think, I need to test N intervals.
I changed to test against input chunk instead of calculating value again.

[grimar]

• Updated stubs implementation for case when zlib was not available(defined) (update of signatures and names, code was not compilable in that case).

[grimar]

Was not used in LLD, abandoning for now.

include/llvm/Support/Compression.h

 //===-- llvm/Support/Compression.h ---Compression----------------*- C++ -*-===//
 //
 //                     The LLVM Compiler Infrastructure
 //
 // This file is distributed under the University of Illinois Open Source
 // License. See LICENSE.TXT for details.
 //
 //===----------------------------------------------------------------------===//
 //
 // This file contains basic functions for compression/uncompression.
 //
 //===----------------------------------------------------------------------===//
 
 #ifndef LLVM_SUPPORT_COMPRESSION_H
 #define LLVM_SUPPORT_COMPRESSION_H
 
+#include "llvm/ADT/ArrayRef.h"
 #include "llvm/Support/DataTypes.h"
 
 namespace llvm {
 template <typename T> class SmallVectorImpl;
+template <typename T> class Expected;
 class Error;
 class StringRef;
 
 namespace zlib {
 
 enum CompressionLevel {
   NoCompression,
   DefaultCompression,
Error uncompress(StringRef InputBuffer, char *UncompressedBuffer,
                  size_t &UncompressedSize);
 
 Error uncompress(StringRef InputBuffer,
                  SmallVectorImpl<char> &UncompressedBuffer,
                  size_t UncompressedSize);
 
 uint32_t crc32(StringRef Buffer);
 
+/// StreamCompression class implements API that allows streaming compression.
+class StreamCompression final {
+public:
+  StreamCompression(StreamCompression&& O);
+  ~StreamCompression();
+
+  /// Creates instance of this class.
+  /// OutBufSize - size of internal buffer for decompression.
+  static Expected<StreamCompression> create(size_t OutBufSize);
+
+  /// Compressed input chunk of data.
+  /// InChuck - chunk to compress.
+  /// IsLastChunk - flag that should be set if input chunk is the last. That
+  ///               allows streaming to finalize compression.
+  /// Out - vector to receive compressed data.
+  Error compressChunk(ArrayRef<uint8_t> InChunk, SmallVectorImpl<uint8_t> &Out);
+
+  /// Should be called after all chunks were proccessed to finalize compression.
+  /// Out - vector to receive compressed data.
+  Error endCompress(SmallVectorImpl<uint8_t> &Out);
+
+private:
+  StreamCompression(size_t ChunkSize);
+
+  Error doDeflate(SmallVectorImpl<uint8_t> &Out, bool Flush);
+
+  SmallVector<uint8_t, 64> OutBuff;
+
+  void *ZStreamImpl = nullptr;
+};
+
 }  // End of namespace zlib
 
 } // End of namespace llvm
 
 #endif
 

lib/Support/Compression.cpp

Error zlib::uncompress(StringRef InputBuffer,
   UncompressedBuffer.resize(UncompressedSize);
   return E;
 }
 
 uint32_t zlib::crc32(StringRef Buffer) {
   return ::crc32(0, (const Bytef *)Buffer.data(), Buffer.size());
 }
 
+zlib::StreamCompression::StreamCompression(size_t OutBufSize)
+    : OutBuff(OutBufSize) {
+  ZStreamImpl = new z_stream();
+}
+
+zlib::StreamCompression::StreamCompression(StreamCompression &&O) {
+  std::swap(ZStreamImpl, O.ZStreamImpl);
+  std::swap(OutBuff, O.OutBuff);
+}
+
+zlib::StreamCompression::~StreamCompression() {
+  delete (z_stream *)ZStreamImpl;
+}
+
+Expected<zlib::StreamCompression>
+zlib::StreamCompression::create(size_t OutBufSize) {
+  zlib::StreamCompression Ret(OutBufSize);
+
+  z_stream *ZStream = (z_stream *)Ret.ZStreamImpl;
+  ZStream->zalloc = Z_NULL;
+  ZStream->zfree = Z_NULL;
+  ZStream->opaque = (voidpf)(0);
+
+  int Res = deflateInit(ZStream, Z_DEFAULT_COMPRESSION);
+  if (Res != Z_OK)
+    return createError(convertZlibCodeToString(Res));
+  return Ret;
+}
+
+Error zlib::StreamCompression::compressChunk(ArrayRef<uint8_t> InChunk,
+                                             SmallVectorImpl<uint8_t> &Out) {
+  z_stream *ZStream = (z_stream *)ZStreamImpl;
+  ZStream->avail_in = InChunk.size();
+  ZStream->next_in = (Bytef *)InChunk.data();
+  return doDeflate(Out, false /*Flush*/);
+}
+
+Error zlib::StreamCompression::endCompress(SmallVectorImpl<uint8_t> &Out) {
+  z_stream *ZStream = (z_stream *)ZStreamImpl;
+  ZStream->avail_in = 0;
+  ZStream->next_in = (Bytef *)0;
+  return doDeflate(Out, true /*Flush*/);
+}
+
+Error zlib::StreamCompression::doDeflate(SmallVectorImpl<uint8_t> &Out,
+                                         bool Flush) {
+  z_stream *ZStream = (z_stream *)ZStreamImpl;
+
+  while (true) {
+    ZStream->avail_out = OutBuff.size();
+    ZStream->next_out = &OutBuff[0];
+    int Res = deflate(ZStream, Flush ? Z_FINISH : Z_NO_FLUSH);
+
+    // Z_BUF_ERROR is a normal possible status for deflate() call.
+    // It just indicates here that deflate() could not consume more input.
+    if (Res != Z_OK && Res != Z_STREAM_END && Res != Z_BUF_ERROR) {
+      deflateEnd(ZStream);
+      return createError(convertZlibCodeToString(Res));
+    }
+
+    ArrayRef<uint8_t> Data(OutBuff.data(), OutBuff.size() - ZStream->avail_out);
+    if (Data.empty())
+      break;
+    Out.insert(Out.end(), Data.begin(), Data.end());
+  }
+
+  if (Flush)
+    deflateEnd(ZStream);
+  return Error::success();
+}
+
 #else
 bool zlib::isAvailable() { return false; }
 Error zlib::compress(StringRef InputBuffer,
                      SmallVectorImpl<char> &CompressedBuffer,
                      CompressionLevel Level) {
   llvm_unreachable("zlib::compress is unavailable");
 }
 Error zlib::uncompress(StringRef InputBuffer, char *UncompressedBuffer,
                        size_t &UncompressedSize) {
   llvm_unreachable("zlib::uncompress is unavailable");
 }
 Error zlib::uncompress(StringRef InputBuffer,
                        SmallVectorImpl<char> &UncompressedBuffer,
                        size_t UncompressedSize) {
   llvm_unreachable("zlib::uncompress is unavailable");
 }
 uint32_t zlib::crc32(StringRef Buffer) {
   llvm_unreachable("zlib::crc32 is unavailable");
 }
+zlib::StreamCompression::StreamCompression(size_t OutBufSize)
+    : OutBuff(OutBufSize) {
+  llvm_unreachable("zlib::compress is unavailable");
+}
+zlib::StreamCompression::StreamCompression(StreamCompression &&O) {
+  llvm_unreachable("zlib::compress is unavailable");
+}
+zlib::StreamCompression::~StreamCompression() {
+  llvm_unreachable("zlib::compress is unavailable");
+}
+Expected<zlib::StreamCompression>
+zlib::StreamCompression::create(size_t OutBufSize) {
+  llvm_unreachable("zlib::compress is unavailable");
+}
+Error zlib::StreamCompression::compressChunk(ArrayRef<uint8_t> InChunk,
+                                             SmallVectorImpl<uint8_t> &Out) {
+  llvm_unreachable("zlib::compress is unavailable");
+}
+Error zlib::StreamCompression::endCompress(SmallVectorImpl<uint8_t> &Out) {
+  llvm_unreachable("zlib::compress is unavailable");
+}
+Error zlib::StreamCompression::doDeflate(SmallVectorImpl<uint8_t> &Out,
+                                         bool Flush) {
+  llvm_unreachable("zlib::compress is unavailable");
+}
 #endif
 

unittests/Support/CompressionTest.cpp

 }
 
 TEST(CompressionTest, ZlibCRC32) {
   EXPECT_EQ(
       0x414FA339U,
       zlib::crc32(StringRef("The quick brown fox jumps over the lazy dog")));
 }
 
+TEST(CompressionTest, ZlibStreaming) {
+  // 1. Create input chunk. It will be used 4 times,
+  // total size of uncompressed data will be 4 * sizeof(chunk).
+  const size_t Size = 512;
+  const size_t ChunksCnt = 4;
+  std::vector<uint8_t> In(512);
+  for (size_t I = 0; I < Size; ++I)
+    In[I] = I & 255;
+
+  // 2. Feed streaming compression with chunks created.
+  SmallVector<uint8_t, 1> Compressed;
+  Expected<zlib::StreamCompression> SC = zlib::StreamCompression::create(128);
+  EXPECT_TRUE((bool)SC);
+  for (int I = 0; I < ChunksCnt; ++I) {
+    Error E = SC->compressChunk(In, Compressed);
+    EXPECT_FALSE(E);
+    consumeError(std::move(E));
+  }
+  Error E = SC->endCompress(Compressed);
+  EXPECT_FALSE(E);
+  consumeError(std::move(E));
+
+  // 3. Decompress using regular decompression API.
+  SmallString<32> Uncompressed;
+  E = zlib::uncompress(
+      StringRef((const char *)Compressed.begin(), Compressed.size()),
+      Uncompressed, Size * ChunksCnt);
+  EXPECT_FALSE(E);
+  consumeError(std::move(E));
+
+  // 4. Check that decompressed data equals to compressed.
+  for (size_t I = 0; I < Size * ChunksCnt; ++I)
+    EXPECT_EQ((uint8_t)Uncompressed[I], In[I % Size]);
+}
+
 #endif
 
 }