journal: add LZ4 as optional compressor

Add liblz4 as an optional dependency when requested with --enable-lz4,
and use it in preference to liblzma for journal blob and coredump
compression. To retain backwards compatibility, XZ is used to
decompress old blobs.

Things will function correctly only with lz4-119.

Based on the benchmarks found on the web, lz4 seems to be the best
choice for "quick" compressors atm.

For pkg-config status, see http://code.google.com/p/lz4/issues/detail?id=135.

1 files changed, 385 insertions, 38 deletions

diff --git a/src/journal/compress.c b/src/journal/compress.c
index 37c55a8728..49d694ac28 100644
--- a/src/journal/compress.c
+++ b/src/journal/compress.c
@@ -23,13 +23,32 @@
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
-#include <lzma.h>
+
+#ifdef HAVE_XZ
+#  include <lzma.h>
+#endif
+
+#ifdef HAVE_LZ4
+#  include <lz4.h>
+#endif
 
 #include "compress.h"
 #include "macro.h"
 #include "util.h"
+#include "sparse-endian.h"
+#include "journal-def.h"
+
+#define ALIGN_8(l) ALIGN_TO(l, sizeof(size_t))
+
+static const char* const object_compressed_table[_OBJECT_COMPRESSED_MAX] = {
+        [OBJECT_COMPRESSED_XZ] = "XZ",
+        [OBJECT_COMPRESSED_LZ4] = "LZ4",
+};
 
-bool compress_blob(const void *src, uint64_t src_size, void *dst, uint64_t *dst_size) {
+DEFINE_STRING_TABLE_LOOKUP(object_compressed, int);
+
+int compress_blob_xz(const void *src, uint64_t src_size, void *dst, uint64_t *dst_size) {
+#ifdef HAVE_XZ
         lzma_ret ret;
         size_t out_pos = 0;
 
@@ -38,25 +57,54 @@ bool compress_blob(const void *src, uint64_t src_size, void *dst, uint64_t *dst_
         assert(dst);
         assert(dst_size);
 
-        /* Returns false if we couldn't compress the data or the
+        /* Returns < 0 if we couldn't compress the data or the
          * compressed result is longer than the original */
 
         ret = lzma_easy_buffer_encode(LZMA_PRESET_DEFAULT, LZMA_CHECK_NONE, NULL,
-                                      src, src_size, dst, &out_pos, src_size);
+                                      src, src_size, dst, &out_pos, src_size - 1);
         if (ret != LZMA_OK)
-                return false;
-
-        /* Is it actually shorter? */
-        if (out_pos == src_size)
-                return false;
+                return -ENOBUFS;
 
         *dst_size = out_pos;
-        return true;
+        return 0;
+#else
+        return -EPROTONOSUPPORT;
+#endif
 }
 
-bool uncompress_blob(const void *src, uint64_t src_size,
-                     void **dst, uint64_t *dst_alloc_size, uint64_t* dst_size, uint64_t dst_max) {
+int compress_blob_lz4(const void *src, uint64_t src_size, void *dst, uint64_t *dst_size) {
+#ifdef HAVE_LZ4
+        int r;
+
+        assert(src);
+        assert(src_size > 0);
+        assert(dst);
+        assert(dst_size);
+
+        /* Returns < 0 if we couldn't compress the data or the
+         * compressed result is longer than the original */
+
+        if (src_size < 9)
+                return -ENOBUFS;
 
+        r = LZ4_compress_limitedOutput(src, dst + 8, src_size, src_size - 8 - 1);
+        if (r <= 0)
+                return -ENOBUFS;
+
+        *(le64_t*) dst = htole64(src_size);
+        *dst_size = r + 8;
+
+        return 0;
+#else
+        return -EPROTONOSUPPORT;
+#endif
+}
+
+
+int decompress_blob_xz(const void *src, uint64_t src_size,
+                       void **dst, uint64_t *dst_alloc_size, uint64_t* dst_size, uint64_t dst_max) {
+
+#ifdef HAVE_XZ
         _cleanup_(lzma_end) lzma_stream s = LZMA_STREAM_INIT;
         lzma_ret ret;
         uint64_t space;
@@ -70,7 +118,7 @@ bool uncompress_blob(const void *src, uint64_t src_size,
 
         ret = lzma_stream_decoder(&s, UINT64_MAX, 0);
         if (ret != LZMA_OK)
-                return false;
+                return -ENOMEM;
 
         space = MIN(src_size * 2, dst_max ?: (uint64_t) -1);
         if (!greedy_realloc(dst, dst_alloc_size, space, 1))
@@ -89,15 +137,13 @@ bool uncompress_blob(const void *src, uint64_t src_size,
 
                 if (ret == LZMA_STREAM_END)
                         break;
-
-                if (ret != LZMA_OK)
-                        return false;
+                else if (ret != LZMA_OK)
+                        return -ENOMEM;
 
                 if (dst_max > 0 && (space - s.avail_out) >= dst_max)
                         break;
-
-                if (dst_max > 0 && space == dst_max)
-                        return false;
+                else if (dst_max > 0 && space == dst_max)
+                        return -ENOBUFS;
 
                 used = space - s.avail_out;
                 space = MIN(2 * space, dst_max ?: (uint64_t) -1);
@@ -109,18 +155,75 @@ bool uncompress_blob(const void *src, uint64_t src_size,
         }
 
         *dst_size = space - s.avail_out;
-        return true;
+        return 0;
+#else
+        return -EPROTONOSUPPORT;
+#endif
 }
 
-bool uncompress_startswith(const void *src, uint64_t src_size,
-                           void **buffer, uint64_t *buffer_size,
-                           const void *prefix, uint64_t prefix_len,
-                           uint8_t extra) {
+int decompress_blob_lz4(const void *src, uint64_t src_size,
+                        void **dst, uint64_t *dst_alloc_size, uint64_t* dst_size, uint64_t dst_max) {
+
+#ifdef HAVE_LZ4
+        char* out;
+        uint64_t size;
+        int r;
 
+        assert(src);
+        assert(src_size > 0);
+        assert(dst);
+        assert(dst_alloc_size);
+        assert(dst_size);
+        assert(*dst_alloc_size == 0 || *dst);
+
+        if (src_size <= 8)
+                return -EBADMSG;
+
+        size = le64toh( *(le64_t*)src );
+        if (size > *dst_alloc_size) {
+                out = realloc(*dst, size);
+                if (!out)
+                        return -ENOMEM;
+                *dst = out;
+                *dst_alloc_size = size;
+        } else
+                out = *dst;
+
+        r = LZ4_decompress_safe(src + 8, out, src_size - 8, size);
+        if (r < 0 || (uint64_t) r != size)
+                return -EBADMSG;
+
+        *dst_size = size;
+        return 0;
+#else
+        return -EPROTONOSUPPORT;
+#endif
+}
+
+int decompress_blob(int compression,
+                    const void *src, uint64_t src_size,
+                    void **dst, uint64_t *dst_alloc_size, uint64_t* dst_size, uint64_t dst_max) {
+        if (compression == OBJECT_COMPRESSED_XZ)
+                return decompress_blob_xz(src, src_size,
+                                          dst, dst_alloc_size, dst_size, dst_max);
+        else if (compression == OBJECT_COMPRESSED_LZ4)
+                return decompress_blob_lz4(src, src_size,
+                                           dst, dst_alloc_size, dst_size, dst_max);
+        else
+                return -EBADMSG;
+}
+
+
+int decompress_startswith_xz(const void *src, uint64_t src_size,
+                             void **buffer, uint64_t *buffer_size,
+                             const void *prefix, uint64_t prefix_len,
+                             uint8_t extra) {
+
+#ifdef HAVE_XZ
         _cleanup_(lzma_end) lzma_stream s = LZMA_STREAM_INIT;
         lzma_ret ret;
 
-        /* Checks whether the uncompressed blob starts with the
+        /* Checks whether the decompressed blob starts with the
          * mentioned prefix. The byte extra needs to follow the
          * prefix */
 
@@ -133,10 +236,10 @@ bool uncompress_startswith(const void *src, uint64_t src_size,
 
         ret = lzma_stream_decoder(&s, UINT64_MAX, 0);
         if (ret != LZMA_OK)
-                return false;
+                return -EBADMSG;
 
-        if (!(greedy_realloc(buffer, buffer_size, prefix_len + 1, 1)))
-                return false;
+        if (!(greedy_realloc(buffer, buffer_size, ALIGN_8(prefix_len + 1), 1)))
+                return -ENOMEM;
 
         s.next_in = src;
         s.avail_in = src_size;
@@ -148,25 +251,88 @@ bool uncompress_startswith(const void *src, uint64_t src_size,
                 ret = lzma_code(&s, LZMA_FINISH);
 
                 if (ret != LZMA_STREAM_END && ret != LZMA_OK)
-                        return false;
+                        return -EBADMSG;
 
                 if (*buffer_size - s.avail_out >= prefix_len + 1)
                         return memcmp(*buffer, prefix, prefix_len) == 0 &&
                                 ((const uint8_t*) *buffer)[prefix_len] == extra;
 
                 if (ret == LZMA_STREAM_END)
-                        return false;
+                        return 0;
 
                 s.avail_out += *buffer_size;
 
                 if (!(greedy_realloc(buffer, buffer_size, *buffer_size * 2, 1)))
-                        return false;
+                        return -ENOMEM;
 
                 s.next_out = *buffer + *buffer_size - s.avail_out;
         }
+
+#else
+        return -EPROTONOSUPPORT;
+#endif
+}
+
+int decompress_startswith_lz4(const void *src, uint64_t src_size,
+                              void **buffer, uint64_t *buffer_size,
+                              const void *prefix, uint64_t prefix_len,
+                              uint8_t extra) {
+#ifdef HAVE_LZ4
+        /* Checks whether the decompressed blob starts with the
+         * mentioned prefix. The byte extra needs to follow the
+         * prefix */
+
+        int r;
+
+        assert(src);
+        assert(src_size > 0);
+        assert(buffer);
+        assert(buffer_size);
+        assert(prefix);
+        assert(*buffer_size == 0 || *buffer);
+
+        if (src_size <= 8)
+                return -EBADMSG;
+
+        if (!(greedy_realloc(buffer, buffer_size, ALIGN_8(prefix_len + 1), 1)))
+                return -ENOMEM;
+
+        r = LZ4_decompress_safe_partial(src + 8, *buffer, src_size - 8,
+                                        prefix_len + 1, *buffer_size);
+
+        if (r < 0)
+                return -EBADMSG;
+        if ((unsigned) r >= prefix_len + 1)
+                return memcmp(*buffer, prefix, prefix_len) == 0 &&
+                        ((const uint8_t*) *buffer)[prefix_len] == extra;
+        else
+                return 0;
+
+#else
+        return -EPROTONOSUPPORT;
+#endif
 }
 
-int compress_stream(int fdf, int fdt, uint32_t preset, off_t max_bytes) {
+int decompress_startswith(int compression,
+                          const void *src, uint64_t src_size,
+                          void **buffer, uint64_t *buffer_size,
+                          const void *prefix, uint64_t prefix_len,
+                          uint8_t extra) {
+        if (compression == OBJECT_COMPRESSED_XZ)
+                return decompress_startswith_xz(src, src_size,
+                                                buffer, buffer_size,
+                                                prefix, prefix_len,
+                                                extra);
+        else if (compression == OBJECT_COMPRESSED_LZ4)
+                return decompress_startswith_lz4(src, src_size,
+                                                 buffer, buffer_size,
+                                                 prefix, prefix_len,
+                                                 extra);
+        else
+                return -EBADMSG;
+}
+
+int compress_stream_xz(int fdf, int fdt, off_t max_bytes) {
         _cleanup_(lzma_end) lzma_stream s = LZMA_STREAM_INIT;
         lzma_ret ret;
 
@@ -176,7 +342,7 @@ int compress_stream(int fdf, int fdt, uint32_t preset, off_t max_bytes) {
         assert(fdf >= 0);
         assert(fdt >= 0);
 
-        ret = lzma_easy_encoder(&s, preset, LZMA_CHECK_CRC64);
+        ret = lzma_easy_encoder(&s, LZMA_PRESET_DEFAULT, LZMA_CHECK_CRC64);
         if (ret != LZMA_OK) {
                 log_error("Failed to initialize XZ encoder: code %d", ret);
                 return -EINVAL;
@@ -230,7 +396,7 @@ int compress_stream(int fdf, int fdt, uint32_t preset, off_t max_bytes) {
                                 return errno ? -errno : -EIO;
 
                         if (ret == LZMA_STREAM_END) {
-                                log_debug("Compression finished (%zu -> %zu bytes, %.1f%%)",
+                                log_debug("XZ compression finished (%zu -> %zu bytes, %.1f%%)",
                                           s.total_in, s.total_out,
                                           (double) s.total_out / s.total_in * 100);
 
@@ -240,7 +406,91 @@ int compress_stream(int fdf, int fdt, uint32_t preset, off_t max_bytes) {
         }
 }
 
-int decompress_stream(int fdf, int fdt, off_t max_bytes) {
+#define LZ4_BUFSIZE (512*1024)
+
+int compress_stream_lz4(int fdf, int fdt, off_t max_bytes) {
+
+#ifdef HAVE_LZ4
+
+        _cleanup_free_ char *buf1 = NULL, *buf2 = NULL, *out = NULL;
+        char *buf;
+        LZ4_stream_t lz4_data = {};
+        le32_t header;
+        size_t total_in = 0, total_out = sizeof(header);
+        ssize_t n;
+
+        assert(fdf >= 0);
+        assert(fdt >= 0);
+
+        buf1 = malloc(LZ4_BUFSIZE);
+        buf2 = malloc(LZ4_BUFSIZE);
+        out = malloc(LZ4_COMPRESSBOUND(LZ4_BUFSIZE));
+        if (!buf1 || !buf2 || !out)
+                return log_oom();
+
+        buf = buf1;
+        for (;;) {
+                size_t m;
+                int r;
+
+                m = LZ4_BUFSIZE;
+                if (max_bytes != -1 && m > (size_t) max_bytes - total_in)
+                        m = max_bytes - total_in;
+
+                n = read(fdf, buf, m);
+                if (n < 0)
+                        return -errno;
+                if (n == 0)
+                        break;
+
+                total_in += n;
+
+                r = LZ4_compress_limitedOutput_continue(&lz4_data, buf, out, n, n);
+                if (r == 0) {
+                        log_debug("Compressed size exceeds original, aborting compression.");
+                        return -ENOBUFS;
+                }
+
+                header = htole32(r);
+                errno = 0;
+
+                n = write(fdt, &header, sizeof(header));
+                if (n < 0)
+                        return -errno;
+                if (n != sizeof(header))
+                        return errno ? -errno : -EIO;
+
+                n = loop_write(fdt, out, r, false);
+                if (n < 0)
+                        return n;
+                if (n != r)
+                        return errno ? -errno : -EIO;
+
+                total_out += sizeof(header) + r;
+
+                buf = buf == buf1 ? buf2 : buf1;
+        }
+
+        header = htole32(0);
+        n = write(fdt, &header, sizeof(header));
+        if (n < 0)
+                return -errno;
+        if (n != sizeof(header))
+                return errno ? -errno : -EIO;
+
+        log_debug("LZ4 compression finished (%zu -> %zu bytes, %.1f%%)",
+                  total_in, total_out,
+                  (double) total_out / total_in * 100);
+
+        return 0;
+#else
+        return -EPROTONOSUPPORT;
+#endif
+}
+
+int decompress_stream_xz(int fdf, int fdt, off_t max_bytes) {
+
+#ifdef HAVE_XZ
         _cleanup_(lzma_end) lzma_stream s = LZMA_STREAM_INIT;
         lzma_ret ret;
 
@@ -253,7 +503,7 @@ int decompress_stream(int fdf, int fdt, off_t max_bytes) {
         ret = lzma_stream_decoder(&s, UINT64_MAX, 0);
         if (ret != LZMA_OK) {
                 log_error("Failed to initialize XZ decoder: code %d", ret);
-                return -EINVAL;
+                return -ENOMEM;
         }
 
         for (;;) {
@@ -289,7 +539,7 @@ int decompress_stream(int fdf, int fdt, off_t max_bytes) {
 
                         if (max_bytes != -1) {
                                 if (max_bytes < n)
-                                        return -E2BIG;
+                                        return -EFBIG;
 
                                 max_bytes -= n;
                         }
@@ -302,7 +552,7 @@ int decompress_stream(int fdf, int fdt, off_t max_bytes) {
                                 return errno ? -errno : -EIO;
 
                         if (ret == LZMA_STREAM_END) {
-                                log_debug("Decompression finished (%zu -> %zu bytes, %.1f%%)",
+                                log_debug("XZ decompression finished (%zu -> %zu bytes, %.1f%%)",
                                           s.total_in, s.total_out,
                                           (double) s.total_out / s.total_in * 100);
 
@@ -310,4 +560,101 @@ int decompress_stream(int fdf, int fdt, off_t max_bytes) {
                         }
                 }
         }
+#else
+        log_error("Cannot decompress file. Compiled without XZ support.");
+        return -EPROTONOSUPPORT;
+#endif
+}
+
+int decompress_stream_lz4(int fdf, int fdt, off_t max_bytes) {
+
+#ifdef HAVE_LZ4
+        _cleanup_free_ char *buf = NULL, *out = NULL;
+        size_t buf_size = 0;
+        LZ4_streamDecode_t lz4_data = {};
+        le32_t header;
+        size_t total_in = sizeof(header), total_out = 0;
+
+        assert(fdf >= 0);
+        assert(fdt >= 0);
+
+        out = malloc(4*LZ4_BUFSIZE);
+        if (!out)
+                return log_oom();
+
+        for (;;) {
+                ssize_t n, m;
+                int r;
+
+                n = read(fdf, &header, sizeof(header));
+                if (n < 0)
+                        return -errno;
+                if (n != sizeof(header))
+                        return errno ? -errno : -EIO;
+
+                m = le32toh(header);
+                if (m == 0)
+                        break;
+
+                /* We refuse to use a bigger decompression buffer than
+                 * the one used for compression by 4 times. This means
+                 * that compression buffer size can be enlarged 4
+                 * times. This can be changed, but old binaries might
+                 * not accept buffers compressed by newer binaries then.
+                 */
+                if (m > LZ4_COMPRESSBOUND(LZ4_BUFSIZE * 4)) {
+                        log_error("Compressed stream block too big: %zd bytes", m);
+                        return -EBADMSG;
+                }
+
+                total_in += sizeof(header) + m;
+
+                if (!GREEDY_REALLOC(buf, buf_size, m))
+                        return log_oom();
+
+                errno = 0;
+                n = loop_read(fdf, buf, m, false);
+                if (n < 0)
+                        return n;
+                if (n != m)
+                        return errno ? -errno : -EIO;
+
+                r = LZ4_decompress_safe_continue(&lz4_data, buf, out, m, 4*LZ4_BUFSIZE);
+                if (r <= 0)
+                        log_error("LZ4 decompression failed.");
+
+                total_out += r;
+
+                if (max_bytes != -1 && total_out > (size_t) max_bytes) {
+                        log_debug("Decompressed stream longer than %zd bytes", max_bytes);
+                        return -EFBIG;
+                }
+
+                errno = 0;
+                n = loop_write(fdt, out, r, false);
+                if (n < 0)
+                        return n;
+                if (n != r)
+                        return errno ? -errno : -EIO;
+        }
+
+        log_debug("LZ4 decompression finished (%zu -> %zu bytes, %.1f%%)",
+                  total_in, total_out,
+                  (double) total_out / total_in * 100);
+
+        return 0;
+#else
+        log_error("Cannot decompress file. Compiled without LZ4 support.");
+        return -EPROTONOSUPPORT;
+#endif
+}
+
+int decompress_stream(const char *filename, int fdf, int fdt, off_t max_bytes) {
+
+        if (endswith(filename, ".lz4"))
+                return decompress_stream_lz4(fdf, fdt, max_bytes);
+        else if (endswith(filename, ".xz"))
+                return decompress_stream_xz(fdf, fdt, max_bytes);
+        else
+                return -EPROTONOSUPPORT;
 }