diff options
| -rw-r--r-- | libbutl/lz4.cxx | 468 | ||||
| -rw-r--r-- | libbutl/lz4.hxx | 138 |
2 files changed, 409 insertions, 197 deletions
diff --git a/libbutl/lz4.cxx b/libbutl/lz4.cxx index 6a91a12..a28ed78 100644 --- a/libbutl/lz4.cxx +++ b/libbutl/lz4.cxx @@ -34,44 +34,6 @@ namespace butl { namespace lz4 { - struct cctx - { - LZ4F_cctx* ctx; - - operator LZ4F_cctx* () const {return ctx;}; - - cctx () - { - if (LZ4F_isError (LZ4F_createCompressionContext (&ctx, LZ4F_VERSION))) - throw bad_alloc (); - } - - ~cctx () - { - LZ4F_errorCode_t e (LZ4F_freeCompressionContext (ctx)); - assert (!LZ4F_isError (e)); - } - }; - - struct dctx - { - LZ4F_dctx* ctx; - - operator LZ4F_dctx* () const {return ctx;}; - - dctx () - { - if (LZ4F_isError (LZ4F_createDecompressionContext (&ctx, LZ4F_VERSION))) - throw bad_alloc (); - } - - ~dctx () - { - LZ4F_errorCode_t e (LZ4F_freeDecompressionContext (ctx)); - assert (!LZ4F_isError (e)); - } - }; - static inline size_t block_size (LZ4F_blockSizeID_t id) { @@ -132,161 +94,316 @@ namespace butl throw_exception (LZ4F_getErrorCode (r)); } - // Return the compressed size. + // compression // - uint64_t - compress (ofdstream& os, ifdstream& is, - int level, - int block_id, - optional<uint64_t> content_size) + + compressor:: + ~compressor () + { + if (LZ4F_cctx* ctx = static_cast<LZ4F_cctx*> (ctx_)) + { + LZ4F_errorCode_t e (LZ4F_freeCompressionContext (ctx)); + assert (!LZ4F_isError (e)); + } + } + + inline void compressor:: + init_preferences (void* vp) const + { + LZ4F_preferences_t* p (static_cast<LZ4F_preferences_t*> (vp)); + + p->autoFlush = 1; + p->favorDecSpeed = 0; + p->compressionLevel = level_; + p->frameInfo.blockMode = LZ4F_blockLinked; + p->frameInfo.blockSizeID = static_cast<LZ4F_blockSizeID_t> (block_id_); + p->frameInfo.blockChecksumFlag = LZ4F_noBlockChecksum; + p->frameInfo.contentChecksumFlag = LZ4F_contentChecksumEnabled; + p->frameInfo.contentSize = content_size_ + ? static_cast<unsigned long long> (*content_size_) + : 0; + } + + void compressor:: + begin (int level, + int block_id, + optional<uint64_t> content_size) { assert (block_id >= 4 && block_id <= 7); + level_ = level; + block_id_ = block_id; + content_size_ = content_size; + LZ4F_preferences_t prefs = LZ4F_INIT_PREFERENCES; - prefs.autoFlush = 1; - prefs.favorDecSpeed = 0; - prefs.compressionLevel = level; - prefs.frameInfo.blockMode = LZ4F_blockLinked; - prefs.frameInfo.blockSizeID = static_cast<LZ4F_blockSizeID_t> (block_id); - prefs.frameInfo.blockChecksumFlag = LZ4F_noBlockChecksum; - prefs.frameInfo.contentChecksumFlag = LZ4F_contentChecksumEnabled; - prefs.frameInfo.contentSize = - content_size ? static_cast<unsigned long long> (*content_size) : 0; + init_preferences (&prefs); // Input/output buffer capacities. // - size_t ic (block_size (prefs.frameInfo.blockSizeID)); - size_t oc; - - // Input/output buffers. + // To be binary compatible with the lz4 utility we have to compress + // files that fit into the block with a single *_compressFrame() call + // instead of *_compressBegin()/*_compressUpdate(). And to determine the + // output buffer capacity we must use *_compressFrameBound() instead of + // *_compressBound(). The problem is, at this stage (before filling the + // input buffer), we don't know which case it will be. // - unique_ptr<char[]> ibg (new char[ic]); char* ib (ibg.get ()); - unique_ptr<char[]> obg; char* ob; - - // Read into the input buffer returning the number of bytes read and - // updating the total read and the eof flag. + // However, in our case (autoFlush=1), *Bound() < *FrameBound() and so + // we can always use the latter at the cost of slight overhead. Also, + // using *FrameBound() allows us to call *Begin() and *Update() without + // flushing the buffer in between (this insight is based on studying the + // implementation of the *Bound() functions). // - // Note that we could try to do direct fd read/write but that would - // complicate things quite a bit (error handling, stream state, etc). + // Actually, we can use content_size (we can get away with much smaller + // buffers for small inputs). We just need to verify the caller is not + // lying to us (failed that, we may end up with strange error like + // insufficient output buffer space). // - uint64_t it (0); - bool eof (false); - auto read = [&is, ib, ic, &it, &eof] () -> size_t - { - eof = butl::eof (is.read (ib, ic)); - size_t n (static_cast<size_t> (is.gcount ())); - it += n; - return n; - }; + ic = block_size (prefs.frameInfo.blockSizeID); - // Write the specified number of bytes from the output buffer updating - // the total written. - // - uint64_t ot (0); - auto write = [&os, &ob, &ot] (size_t n) + if (content_size_ && *content_size_ < ic) { - os.write (ob, static_cast<streamsize> (n)); - ot += n; - }; + // This is nuanced: we need to add an extra byte in order to detect + // EOF. + // + ic = static_cast<size_t> (*content_size_) + 1; + } + + oc = LZ4F_compressFrameBound (ic, &prefs); + + begin_ = true; + } + + void compressor:: + next (bool end) + { + LZ4F_cctx* ctx; // Unlike the decompression case below, compression cannot fail due to // invalid content. So any LZ4F_*() function failure is either due to a // programming bug or argument inconsistencies (e.g., content size does // not match actual). - // To be binary compatible with the lz4 utility we have to compress - // files that fit into the block with a single LZ4F_compressFrame() - // call. - // - size_t in (read ()); - size_t on; - - if (eof && in < ic) // Should be really <= but that's not lz4-compatible. + if (begin_) { - oc = LZ4F_compressFrameBound (in, &prefs); - obg.reset ((ob = new char[oc])); - - on = LZ4F_compressFrame (ob, oc, ib, in, &prefs); - if (LZ4F_isError (on)) - throw_exception (on); + begin_ = false; - write (on); + LZ4F_preferences_t prefs = LZ4F_INIT_PREFERENCES; + init_preferences (&prefs); - // Verify specified and actual content sizes match. + // If we've allocated smaller buffers based on content_size_, then + // verify the input size matches what's promised. // - // LZ4F_compressFrame() does not fail if it doesn't match instead - // replacing it with the actual value. + // Note also that LZ4F_compressFrame() does not fail if it doesn't + // match instead replacing it with the actual value. // - if (content_size && *content_size != it) - throw_exception (LZ4F_ERROR_frameSize_wrong); - } - else - { - cctx ctx; - - oc = LZ4F_compressBound (ic, &prefs); - obg.reset ((ob = new char[oc])); - - // Write the header. - // - on = LZ4F_compressBegin (ctx, ob, oc, &prefs); - if (LZ4F_isError (on)) - throw_exception (on); - - write (on); + size_t bs (block_size (prefs.frameInfo.blockSizeID)); + if (content_size_ && *content_size_ < bs) + { + if (!end || in != *content_size_) + throw_exception (LZ4F_ERROR_frameSize_wrong); + } - // Keep compressing, writing, and reading chunks of content. + // Must be < for lz4 compatibility (see EOF nuance above for the + // likely reason). // - for (;;) + if (end && in < bs) { - on = LZ4F_compressUpdate (ctx, ob, oc, ib, in, nullptr); + on = LZ4F_compressFrame (ob, oc, ib, in, &prefs); if (LZ4F_isError (on)) throw_exception (on); - if (on != 0) // LZ4F_compressUpdate() may just buffer the data. - write (on); + return; + } + else + { + if (LZ4F_isError ( + LZ4F_createCompressionContext (&ctx, LZ4F_VERSION))) + throw bad_alloc (); + + ctx_ = ctx; - if (eof) - break; + // Write the header. + // + on = LZ4F_compressBegin (ctx, ob, oc, &prefs); + if (LZ4F_isError (on)) + throw_exception (on); - in = read (); + // Fall through. } + } + else + { + ctx = static_cast<LZ4F_cctx*> (ctx_); + on = 0; + } - // Write the end marker. - // + size_t n; + + n = LZ4F_compressUpdate (ctx, ob + on, oc - on, ib, in, nullptr); + if (LZ4F_isError (n)) + throw_exception (n); + + on += n; + + // Write the end marker. + // + if (end) + { // Note that this call also verifies specified and actual content // sizes match. // - on = LZ4F_compressEnd (ctx, ob, oc, nullptr); - if (LZ4F_isError (on)) - throw_exception (on); + n = LZ4F_compressEnd (ctx, ob + on, oc - on, nullptr); + if (LZ4F_isError (n)) + throw_exception (n); + + on += n; + } + } + + uint64_t + compress (ofdstream& os, ifdstream& is, + int level, + int block_id, + optional<uint64_t> content_size) + { + compressor c; + + // Input/output buffer guards. + // + unique_ptr<char[]> ibg; + unique_ptr<char[]> obg; + + // First determine required buffer capacities. + // + c.begin (level, block_id, content_size); + + ibg.reset ((c.ib = new char[c.ic])); + obg.reset ((c.ob = new char[c.oc])); + + // Read into the input buffer updating the eof flag. + // + // Note that we could try to do direct fd read/write but that would + // complicate things quite a bit (error handling, stream state, etc). + // + bool eof (false); + auto read = [&is, &c, &eof] () + { + eof = butl::eof (is.read (c.ib, c.ic)); + c.in = static_cast<size_t> (is.gcount ()); + }; + + // Write from the output buffer updating the total written. + // + uint64_t ot (0); + auto write = [&os, &c, &ot] () + { + os.write (c.ob, static_cast<streamsize> (c.on)); + ot += c.on; + }; + + // Keep reading, compressing, and writing chunks of content. + // + while (!eof) + { + read (); - write (on); + c.next (eof); + + if (c.on != 0) // next() may just buffer the data. + write (); } return ot; } - uint64_t - decompress (ofdstream& os, ifdstream& is) + // decompression + // + + static_assert (sizeof (decompressor::hb) == LZ4F_HEADER_SIZE_MAX, + "LZ4 header size mismatch"); + + decompressor:: + decompressor () + : hn (0), in (0), on (0) + { + LZ4F_dctx* ctx; + + if (LZ4F_isError (LZ4F_createDecompressionContext (&ctx, LZ4F_VERSION))) + throw bad_alloc (); + + ctx_ = ctx; + } + + decompressor:: + ~decompressor () + { + LZ4F_errorCode_t e ( + LZ4F_freeDecompressionContext (static_cast<LZ4F_dctx*> (ctx_))); + assert (!LZ4F_isError (e)); + } + + size_t decompressor:: + begin () + { + LZ4F_dctx* ctx (static_cast<LZ4F_dctx*> (ctx_)); + + LZ4F_frameInfo_t info = LZ4F_INIT_FRAMEINFO; + + // Input hint and end as signalled by the LZ4F_*() functions. + // + size_t h, e; + + h = LZ4F_getFrameInfo (ctx, &info, hb, &(e = hn)); + if (LZ4F_isError (h)) + throw_exception (h); + + // Use the block size for the output buffer capacity and compressed + // bound plus the header size for the input. The expectation is that + // LZ4F_decompress() should never hint for more than that. + // + oc = block_size (info.blockSizeID); + ic = LZ4F_compressBound (oc, nullptr) + LZ4F_BLOCK_HEADER_SIZE; + + assert (h <= ic); + + // Move over whatever is left in the header buffer to be beginning. + // + hn -= e; + memmove (hb, hb + e, hn); + + return h; + } + + size_t decompressor:: + next () { - // The LZ4F_*() decompression functions return a hint of how much data - // they want on the next call. So the plan is to allocate the input - // buffer large enough to hold anything that can be asked for and then - // fill it in in the asked chunks. This way we avoid having to shift the - // unread data, etc. + LZ4F_dctx* ctx (static_cast<LZ4F_dctx*> (ctx_)); + + size_t h, e; + + // Note that LZ4F_decompress() verifies specified and actual content + // sizes match (similar to compression). // - dctx ctx; + h = LZ4F_decompress (ctx, ob, &(on = oc), ib, &(e = in), nullptr); + if (LZ4F_isError (h)) + throw_exception (h); - // Input/output buffer capacities and sizes. + // We expect LZ4F_decompress() to consume what it asked for. // - size_t ic, oc; - size_t in, on; + assert (e == in && h <= ic); + + return h; + } - // Input/output buffers. + uint64_t + decompress (ofdstream& os, ifdstream& is) + { + decompressor d; + + // Input/output buffer guards. // - unique_ptr<char[]> ibg; char* ib; - unique_ptr<char[]> obg; char* ob; + unique_ptr<char[]> ibg; + unique_ptr<char[]> obg; // Read into the specified buffer returning the number of bytes read and // updating the eof flag. @@ -309,73 +426,46 @@ namespace butl // the total written. // uint64_t ot (0); - auto write = [&os, &ob, &ot] (size_t n) + auto write = [&os, &ot] (char* b, size_t n) { - os.write (ob, static_cast<streamsize> (n)); + os.write (b, static_cast<streamsize> (n)); ot += n; }; - // Input hint and end as signalled by the LZ4F_*() functions. - // - size_t ih, ie; + size_t h; // Input hint. - // Read the header. + // First read in the header and allocate the buffers. // - LZ4F_frameInfo_t info = LZ4F_INIT_FRAMEINFO; - { - char hb[LZ4F_HEADER_SIZE_MAX]; - in = read (hb, sizeof (hb)); + d.hn = read (d.hb, sizeof (d.hb)); + h = d.begin (); - ih = LZ4F_getFrameInfo (ctx, &info, hb, &(ie = in)); - if (LZ4F_isError (ih)) - throw_exception (ih); + ibg.reset ((d.ib = new char[d.ic])); + obg.reset ((d.ob = new char[d.oc])); - // Use the block size for the output buffer capacity and compressed - // bound plus the header size for the input. The expectation is that - // LZ4F_decompress() should never hint for more than that. - // - oc = block_size (info.blockSizeID); - ic = LZ4F_compressBound (oc, nullptr) + LZ4F_BLOCK_HEADER_SIZE; - - assert (ih <= ic); - - ibg.reset ((ib = new char[ic])); - obg.reset ((ob = new char[oc])); + // Copy over whatever is left in the header buffer and read up to + // the hinted size. + // + memcpy (d.ib, d.hb, (d.in = d.hn)); - // Copy over whatever is left in the header buffer and read up to - // the hinted size. - // - in -= ie; - memcpy (ib, hb + ie, in); - in += read (ib + in, ih - in); - } + if (h > d.in) + d.in += read (d.ib + d.in, h - d.in); // Keep decompressing, writing, and reading chunks of compressed // content. // - // Note that LZ4F_decompress() verifies specified and actual content - // sizes match (similar to compression). - // for (;;) { - ih = LZ4F_decompress (ctx, ob, &(on = oc), ib, &(ie = in), nullptr); - if (LZ4F_isError (ih)) - throw_exception (ih); - - // We expect LZ4F_decompress() to consume what it asked for. - // - assert (ie == in); + h = d.next (); - write (on); + write (d.ob, d.on); - if (ih == 0) + if (h == 0) break; if (eof) throw invalid_argument ("incomplete compressed content"); - assert (ih <= ic); - in = read (ib, ih); + d.in = read (d.ib, h); } return ot; diff --git a/libbutl/lz4.hxx b/libbutl/lz4.hxx index bf679c5..98175c1 100644 --- a/libbutl/lz4.hxx +++ b/libbutl/lz4.hxx @@ -4,6 +4,7 @@ #pragma once #include <cstdint> +#include <cstddef> #include <libbutl/optional.mxx> #include <libbutl/fdstream.mxx> @@ -14,16 +15,14 @@ namespace butl { namespace lz4 { - //@@ TODO: allow (re-)using external buffers, contexts? - // Read the content from the input stream, compress it using the specified // compression level and block size, and write the compressed content to // the output stream. If content size is specified, then include it into // the compressed content header. Return the compressed content size. // // This function may throw std::bad_alloc as well as exceptions thrown by - // fdstream read/write functions. It may also throw invalid_argument in - // case of argument inconsistencies (e.g., content size does not match + // fdstream read/write functions. It may also throw std::invalid_argument + // in case of argument inconsistencies (e.g., content size does not match // actual) with what() returning the error description. The input stream // is expected to throw on badbit (but not failbit). The output stream is // expected to throw on badbit or failbit. @@ -52,14 +51,75 @@ namespace butl int block_size_id, optional<std::uint64_t> content_size); + // Low-level iterative compression API. + // + // This API may throw std::bad_alloc in case of memory allocation errors + // and std::invalid_argument in case of argument inconsistencies (e.g., + // content size does not match actual) with what() returning the error + // description. + // + // See the implementation of the compress() function above for usage + // example. + // + // @@ TODO: reset support. + // + struct LIBBUTL_SYMEXPORT compressor + { + // Buffer, current size (part filled with data), and capacity. + // + char* ib; std::size_t in, ic; // Input. + char* ob; std::size_t on, oc; // Output. + + // As a first step call begin(). This function sets the required input + // and output buffer capacities (ic, oc). + // + // The caller normally allocates the input and output buffers and fills + // the input buffer. + // + void + begin (int compression_level, + int block_size_id, + optional<std::uint64_t> content_size); + + // Then call next() to compress the next chunk of input passing true on + // reaching EOF. Note that the input buffer should be filled to capacity + // unless end is true and the output buffer must be flushed before each + // subsequent call to next(). + // + void + next (bool end); + + // Not copyable or movable. + // + compressor (const compressor&) = delete; + compressor (compressor&&) = delete; + compressor& operator= (const compressor&) = delete; + compressor& operator= (compressor&&) = delete; + + // Implementation details. + // + compressor (): ctx_ (nullptr) {} + ~compressor (); + + public: + void + init_preferences (void*) const; + + void* ctx_; + int level_; + int block_id_; + optional<std::uint64_t> content_size_; + bool begin_; + }; + // Read the compressed content from the input stream, decompress it, and // write the decompressed content to the output stream. Return the // decompressed content size. // // This function may throw std::bad_alloc as well as exceptions thrown by - // fdstream read/write functions. It may also throw invalid_argument if - // the compressed content is invalid with what() returning the error + // fdstream read/write functions. It may also throw std::invalid_argument + // if the compressed content is invalid with what() returning the error // description. The input stream is expected to throw on badbit (but not // failbit). The output stream is expected to throw on badbit or failbit. // @@ -67,10 +127,72 @@ namespace butl // mode. // // Note that this function does not require the input stream to reach EOF - // at the end of compressed content. So if you have this a requirement, - // you will need to enforce it yourself. + // at the end of compressed content. So if you have this requirement, you + // will need to enforce it yourself. // LIBBUTL_SYMEXPORT std::uint64_t decompress (ofdstream&, ifdstream&); + + // Low-level iterative decompression API. + // + // This API may throw std::bad_alloc in case of memory allocation errors + // and std::invalid_argument if the compressed content is invalid with + // what() returning the error description. + // + // See the implementation of the decompress() function above for usage + // example. + // + // The LZ4F_*() decompression functions return a hint of how much data + // they want on the next call. So the plan is to allocate the input + // buffer large enough to hold anything that can be asked for and then + // fill it in in the asked chunks. This way we avoid having to shift the + // unread data around. + // + // @@ TODO: reset support. + // + struct LIBBUTL_SYMEXPORT decompressor + { + // Buffer, current size (part filled with data), and capacity. + // + char hb[19]; std::size_t hn ; // Header. + char* ib; std::size_t in, ic; // Input. + char* ob; std::size_t on, oc; // Output. + + // As a first step, fill in the header buffer and call begin(). This + // function sets the required input and output buffer capacities (ic, + // oc) and the number of bytes left in the header buffer (hn) and + // returns the number of bytes expected by the following call to next(). + // + // The caller normally allocates the input and output buffers, copies + // remaining header buffer data over to the input buffer, and then fills + // in the remainder of the input buffer up to what's expected by the + // call to next(). + // + std::size_t + begin (); + + // Then call next() to decompress the next chunk of input. This function + // returns the number of bytes expected by the following call to next() + // or 0 if no further input is expected. Note that the output buffer + // must be flushed before each subsequent call to next(). + // + std::size_t + next (); + + // Not copyable or movable. + // + decompressor (const decompressor&) = delete; + decompressor (decompressor&&) = delete; + decompressor& operator= (const decompressor&) = delete; + decompressor& operator= (decompressor&&) = delete; + + // Implementation details. + // + decompressor (); + ~decompressor (); + + public: + void* ctx_; + }; } } |