journal: support zstd compression for large objects in journal files

This commit is contained in:
Lennart Poettering
2020-06-25 15:02:18 +02:00
parent 0dbe57ee86
commit 8653185a9e
10 changed files with 260 additions and 46 deletions
+174 -20
View File
@@ -59,6 +59,7 @@ static int zstd_ret_to_errno(size_t ret) {
static const char* const object_compressed_table[_OBJECT_COMPRESSED_MAX] = {
[OBJECT_COMPRESSED_XZ] = "XZ",
[OBJECT_COMPRESSED_LZ4] = "LZ4",
[OBJECT_COMPRESSED_ZSTD] = "ZSTD",
/* If we add too many more entries here, it's going to grow quite large (and be mostly sparse), since
* the array key is actually a bitmask, not a plain enum */
};
@@ -133,6 +134,29 @@ int compress_blob_lz4(const void *src, uint64_t src_size,
#endif
}
int compress_blob_zstd(
const void *src, uint64_t src_size,
void *dst, size_t dst_alloc_size, size_t *dst_size) {
#if HAVE_ZSTD
size_t k;
assert(src);
assert(src_size > 0);
assert(dst);
assert(dst_alloc_size > 0);
assert(dst_size);
k = ZSTD_compress(dst, dst_alloc_size, src, src_size, 0);
if (ZSTD_isError(k))
return zstd_ret_to_errno(k);
*dst_size = k;
return 0;
#else
return -EPROTONOSUPPORT;
#endif
}
int decompress_blob_xz(const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max) {
@@ -233,15 +257,74 @@ int decompress_blob_lz4(const void *src, uint64_t src_size,
#endif
}
int decompress_blob(int compression,
const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max) {
int decompress_blob_zstd(
const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max) {
#if HAVE_ZSTD
size_t space;
assert(src);
assert(src_size > 0);
assert(dst);
assert(dst_alloc_size);
assert(dst_size);
assert(*dst_alloc_size == 0 || *dst);
if (src_size > SIZE_MAX/2) /* Overflow? */
return -ENOBUFS;
space = src_size * 2;
if (dst_max > 0 && space > dst_max)
space = dst_max;
if (!greedy_realloc(dst, dst_alloc_size, space, 1))
return -ENOMEM;
for (;;) {
size_t k;
k = ZSTD_decompress(*dst, *dst_alloc_size, src, src_size);
if (!ZSTD_isError(k)) {
*dst_size = k;
return 0;
}
if (ZSTD_getErrorCode(k) != ZSTD_error_dstSize_tooSmall)
return zstd_ret_to_errno(k);
if (dst_max > 0 && space >= dst_max) /* Already at max? */
return -ENOBUFS;
if (space > SIZE_MAX / 2) /* Overflow? */
return -ENOBUFS;
space *= 2;
if (dst_max > 0 && space > dst_max)
space = dst_max;
if (!greedy_realloc(dst, dst_alloc_size, space, 1))
return -ENOMEM;
}
#else
return -EPROTONOSUPPORT;
#endif
}
int decompress_blob(
int compression,
const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max) {
if (compression == OBJECT_COMPRESSED_XZ)
return decompress_blob_xz(src, src_size,
dst, dst_alloc_size, dst_size, dst_max);
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);
return decompress_blob_lz4(
src, src_size,
dst, dst_alloc_size, dst_size, dst_max);
else if (compression == OBJECT_COMPRESSED_ZSTD)
return decompress_blob_zstd(
src, src_size,
dst, dst_alloc_size, dst_size, dst_max);
else
return -EBADMSG;
}
@@ -367,21 +450,92 @@ int decompress_startswith_lz4(const void *src, uint64_t src_size,
#endif
}
int decompress_startswith(int compression,
const void *src, uint64_t src_size,
void **buffer, size_t *buffer_size,
const void *prefix, size_t prefix_len,
uint8_t extra) {
int decompress_startswith_zstd(
const void *src, uint64_t src_size,
void **buffer, size_t *buffer_size,
const void *prefix, size_t prefix_len,
uint8_t extra) {
#if HAVE_ZSTD
_cleanup_(ZSTD_freeDCtxp) ZSTD_DCtx *dctx = NULL;
size_t k;
assert(src);
assert(src_size > 0);
assert(buffer);
assert(buffer_size);
assert(prefix);
assert(*buffer_size == 0 || *buffer);
dctx = ZSTD_createDCtx();
if (!dctx)
return -ENOMEM;
if (!(greedy_realloc(buffer, buffer_size, MAX(ZSTD_DStreamOutSize(), prefix_len + 1), 1)))
return -ENOMEM;
ZSTD_inBuffer input = {
.src = src,
.size = src_size,
};
ZSTD_outBuffer output = {
.dst = *buffer,
.size = *buffer_size,
};
for (;;) {
k = ZSTD_decompressStream(dctx, &output, &input);
if (ZSTD_isError(k)) {
log_debug("ZSTD decoder failed: %s", ZSTD_getErrorName(k));
return zstd_ret_to_errno(k);
}
if (output.pos >= prefix_len + 1)
return memcmp(*buffer, prefix, prefix_len) == 0 &&
((const uint8_t*) *buffer)[prefix_len] == extra;
if (input.pos >= input.size)
return 0;
if (*buffer_size > SIZE_MAX/2)
return -ENOBUFS;
if (!(greedy_realloc(buffer, buffer_size, *buffer_size * 2, 1)))
return -ENOMEM;
output.dst = *buffer;
output.size = *buffer_size;
}
#else
return -EPROTONOSUPPORT;
#endif
}
int decompress_startswith(
int compression,
const void *src, uint64_t src_size,
void **buffer, size_t *buffer_size,
const void *prefix, size_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);
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);
return decompress_startswith_lz4(
src, src_size,
buffer, buffer_size,
prefix, prefix_len,
extra);
else if (compression == OBJECT_COMPRESSED_ZSTD)
return decompress_startswith_zstd(
src, src_size,
buffer, buffer_size,
prefix, prefix_len,
extra);
else
return -EBADMSG;
}
+13 -1
View File
@@ -12,11 +12,17 @@ int compress_blob_xz(const void *src, uint64_t src_size,
void *dst, size_t dst_alloc_size, size_t *dst_size);
int compress_blob_lz4(const void *src, uint64_t src_size,
void *dst, size_t dst_alloc_size, size_t *dst_size);
int compress_blob_zstd(const void *src, uint64_t src_size,
void *dst, size_t dst_alloc_size, size_t *dst_size);
static inline int compress_blob(const void *src, uint64_t src_size,
void *dst, size_t dst_alloc_size, size_t *dst_size) {
int r;
#if HAVE_LZ4
#if HAVE_ZSTD
r = compress_blob_zstd(src, src_size, dst, dst_alloc_size, dst_size);
if (r == 0)
return OBJECT_COMPRESSED_ZSTD;
#elif HAVE_LZ4
r = compress_blob_lz4(src, src_size, dst, dst_alloc_size, dst_size);
if (r == 0)
return OBJECT_COMPRESSED_LZ4;
@@ -32,6 +38,8 @@ int decompress_blob_xz(const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max);
int decompress_blob_lz4(const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max);
int decompress_blob_zstd(const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max);
int decompress_blob(int compression,
const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max);
@@ -44,6 +52,10 @@ int decompress_startswith_lz4(const void *src, uint64_t src_size,
void **buffer, size_t *buffer_size,
const void *prefix, size_t prefix_len,
uint8_t extra);
int decompress_startswith_zstd(const void *src, uint64_t src_size,
void **buffer, size_t *buffer_size,
const void *prefix, size_t prefix_len,
uint8_t extra);
int decompress_startswith(int compression,
const void *src, uint64_t src_size,
void **buffer, size_t *buffer_size,
+17 -7
View File
@@ -46,11 +46,11 @@ typedef enum ObjectType {
enum {
OBJECT_COMPRESSED_XZ = 1 << 0,
OBJECT_COMPRESSED_LZ4 = 1 << 1,
OBJECT_COMPRESSION_MASK = (OBJECT_COMPRESSED_XZ | OBJECT_COMPRESSED_LZ4),
OBJECT_COMPRESSED_ZSTD = 1 << 2,
OBJECT_COMPRESSION_MASK = (OBJECT_COMPRESSED_XZ | OBJECT_COMPRESSED_LZ4 | OBJECT_COMPRESSED_ZSTD),
_OBJECT_COMPRESSED_MAX = OBJECT_COMPRESSION_MASK,
};
struct ObjectHeader {
uint8_t type;
uint8_t flags;
@@ -152,19 +152,29 @@ enum {
HEADER_INCOMPATIBLE_COMPRESSED_XZ = 1 << 0,
HEADER_INCOMPATIBLE_COMPRESSED_LZ4 = 1 << 1,
HEADER_INCOMPATIBLE_KEYED_HASH = 1 << 2,
HEADER_INCOMPATIBLE_COMPRESSED_ZSTD = 1 << 3,
};
#define HEADER_INCOMPATIBLE_ANY \
(HEADER_INCOMPATIBLE_COMPRESSED_XZ| \
HEADER_INCOMPATIBLE_COMPRESSED_LZ4| \
HEADER_INCOMPATIBLE_KEYED_HASH)
#define HEADER_INCOMPATIBLE_ANY \
(HEADER_INCOMPATIBLE_COMPRESSED_XZ | \
HEADER_INCOMPATIBLE_COMPRESSED_LZ4 | \
HEADER_INCOMPATIBLE_KEYED_HASH | \
HEADER_INCOMPATIBLE_COMPRESSED_ZSTD)
#if HAVE_XZ && HAVE_LZ4
#if HAVE_XZ && HAVE_LZ4 && HAVE_ZSTD
# define HEADER_INCOMPATIBLE_SUPPORTED HEADER_INCOMPATIBLE_ANY
#elif HAVE_XZ && HAVE_LZ4
# define HEADER_INCOMPATIBLE_SUPPORTED (HEADER_INCOMPATIBLE_COMPRESSED_XZ|HEADER_INCOMPATIBLE_COMPRESSED_LZ4|HEADER_INCOMPATIBLE_KEYED_HASH)
#elif HAVE_XZ && HAVE_ZSTD
# define HEADER_INCOMPATIBLE_SUPPORTED (HEADER_INCOMPATIBLE_COMPRESSED_XZ|HEADER_INCOMPATIBLE_COMPRESSED_ZSTD|HEADER_INCOMPATIBLE_KEYED_HASH)
#elif HAVE_LZ4 && HAVE_ZSTD
# define HEADER_INCOMPATIBLE_SUPPORTED (HEADER_INCOMPATIBLE_COMPRESSED_LZ4|HEADER_INCOMPATIBLE_COMPRESSED_ZSTD|HEADER_INCOMPATIBLE_KEYED_HASH)
#elif HAVE_XZ
# define HEADER_INCOMPATIBLE_SUPPORTED (HEADER_INCOMPATIBLE_COMPRESSED_XZ|HEADER_INCOMPATIBLE_KEYED_HASH)
#elif HAVE_LZ4
# define HEADER_INCOMPATIBLE_SUPPORTED (HEADER_INCOMPATIBLE_COMPRESSED_LZ4|HEADER_INCOMPATIBLE_KEYED_HASH)
#elif HAVE_ZSTD
# define HEADER_INCOMPATIBLE_SUPPORTED (HEADER_INCOMPATIBLE_COMPRESSED_ZSTD|HEADER_INCOMPATIBLE_KEYED_HASH)
#else
# define HEADER_INCOMPATIBLE_SUPPORTED HEADER_INCOMPATIBLE_KEYED_HASH
#endif
+14 -7
View File
@@ -392,7 +392,7 @@ JournalFile* journal_file_close(JournalFile *f) {
ordered_hashmap_free_free(f->chain_cache);
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
free(f->compress_buffer);
#endif
@@ -424,6 +424,7 @@ static int journal_file_init_header(JournalFile *f, JournalFile *template) {
h.incompatible_flags |= htole32(
f->compress_xz * HEADER_INCOMPATIBLE_COMPRESSED_XZ |
f->compress_lz4 * HEADER_INCOMPATIBLE_COMPRESSED_LZ4 |
f->compress_zstd * HEADER_INCOMPATIBLE_COMPRESSED_ZSTD |
f->keyed_hash * HEADER_INCOMPATIBLE_KEYED_HASH);
h.compatible_flags = htole32(
@@ -491,7 +492,7 @@ static bool warn_wrong_flags(const JournalFile *f, bool compatible) {
f->path, type, flags & ~any);
flags = (flags & any) & ~supported;
if (flags) {
const char* strv[4];
const char* strv[5];
unsigned n = 0;
_cleanup_free_ char *t = NULL;
@@ -503,6 +504,8 @@ static bool warn_wrong_flags(const JournalFile *f, bool compatible) {
strv[n++] = "xz-compressed";
if (flags & HEADER_INCOMPATIBLE_COMPRESSED_LZ4)
strv[n++] = "lz4-compressed";
if (flags & HEADER_INCOMPATIBLE_COMPRESSED_ZSTD)
strv[n++] = "zstd-compressed";
if (flags & HEADER_INCOMPATIBLE_KEYED_HASH)
strv[n++] = "keyed-hash";
}
@@ -602,6 +605,7 @@ static int journal_file_verify_header(JournalFile *f) {
f->compress_xz = JOURNAL_HEADER_COMPRESSED_XZ(f->header);
f->compress_lz4 = JOURNAL_HEADER_COMPRESSED_LZ4(f->header);
f->compress_zstd = JOURNAL_HEADER_COMPRESSED_ZSTD(f->header);
f->seal = JOURNAL_HEADER_SEALED(f->header);
@@ -1451,7 +1455,7 @@ int journal_file_find_data_object_with_hash(
goto next;
if (o->object.flags & OBJECT_COMPRESSION_MASK) {
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
uint64_t l;
size_t rsize = 0;
@@ -1620,7 +1624,7 @@ static int journal_file_append_data(
o->data.hash = htole64(hash);
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
if (JOURNAL_FILE_COMPRESS(f) && size >= f->compress_threshold_bytes) {
size_t rsize = 0;
@@ -3224,7 +3228,7 @@ void journal_file_print_header(JournalFile *f) {
"Sequential number ID: %s\n"
"State: %s\n"
"Compatible flags:%s%s\n"
"Incompatible flags:%s%s%s%s\n"
"Incompatible flags:%s%s%s%s%s\n"
"Header size: %"PRIu64"\n"
"Arena size: %"PRIu64"\n"
"Data hash table size: %"PRIu64"\n"
@@ -3249,6 +3253,7 @@ void journal_file_print_header(JournalFile *f) {
(le32toh(f->header->compatible_flags) & ~HEADER_COMPATIBLE_ANY) ? " ???" : "",
JOURNAL_HEADER_COMPRESSED_XZ(f->header) ? " COMPRESSED-XZ" : "",
JOURNAL_HEADER_COMPRESSED_LZ4(f->header) ? " COMPRESSED-LZ4" : "",
JOURNAL_HEADER_COMPRESSED_ZSTD(f->header) ? " COMPRESSED-ZSTD" : "",
JOURNAL_HEADER_KEYED_HASH(f->header) ? " KEYED-HASH" : "",
(le32toh(f->header->incompatible_flags) & ~HEADER_INCOMPATIBLE_ANY) ? " ???" : "",
le64toh(f->header->header_size),
@@ -3370,7 +3375,9 @@ int journal_file_open(
.prot = prot_from_flags(flags),
.writable = (flags & O_ACCMODE) != O_RDONLY,
#if HAVE_LZ4
#if HAVE_ZSTD
.compress_zstd = compress,
#elif HAVE_LZ4
.compress_lz4 = compress,
#elif HAVE_XZ
.compress_xz = compress,
@@ -3845,7 +3852,7 @@ int journal_file_copy_entry(JournalFile *from, JournalFile *to, Object *o, uint6
return -E2BIG;
if (o->object.flags & OBJECT_COMPRESSION_MASK) {
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
size_t rsize = 0;
r = decompress_blob(o->object.flags & OBJECT_COMPRESSION_MASK,
+6 -2
View File
@@ -67,6 +67,7 @@ typedef struct JournalFile {
bool writable:1;
bool compress_xz:1;
bool compress_lz4:1;
bool compress_zstd:1;
bool seal:1;
bool defrag_on_close:1;
bool close_fd:1;
@@ -106,7 +107,7 @@ typedef struct JournalFile {
unsigned last_seen_generation;
uint64_t compress_threshold_bytes;
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
void *compress_buffer;
size_t compress_buffer_size;
#endif
@@ -196,6 +197,9 @@ static inline bool VALID_EPOCH(uint64_t u) {
#define JOURNAL_HEADER_COMPRESSED_LZ4(h) \
FLAGS_SET(le32toh((h)->incompatible_flags), HEADER_INCOMPATIBLE_COMPRESSED_LZ4)
#define JOURNAL_HEADER_COMPRESSED_ZSTD(h) \
FLAGS_SET(le32toh((h)->incompatible_flags), HEADER_INCOMPATIBLE_COMPRESSED_ZSTD)
#define JOURNAL_HEADER_KEYED_HASH(h) \
FLAGS_SET(le32toh((h)->incompatible_flags), HEADER_INCOMPATIBLE_KEYED_HASH)
@@ -264,7 +268,7 @@ int journal_file_map_field_hash_table(JournalFile *f);
static inline bool JOURNAL_FILE_COMPRESS(JournalFile *f) {
assert(f);
return f->compress_xz || f->compress_lz4;
return f->compress_xz || f->compress_lz4 || f->compress_zstd;
}
uint64_t journal_file_hash_data(JournalFile *f, const void *data, size_t sz);
+10 -3
View File
@@ -925,9 +925,10 @@ int journal_file_verify(
goto fail;
}
if ((o->object.flags & OBJECT_COMPRESSED_XZ) &&
(o->object.flags & OBJECT_COMPRESSED_LZ4)) {
error(p, "Objected with double compression");
if (!!(o->object.flags & OBJECT_COMPRESSED_XZ) +
!!(o->object.flags & OBJECT_COMPRESSED_LZ4) +
!!(o->object.flags & OBJECT_COMPRESSED_ZSTD) > 1) {
error(p, "Object has multiple compression flags set");
r = -EINVAL;
goto fail;
}
@@ -944,6 +945,12 @@ int journal_file_verify(
goto fail;
}
if ((o->object.flags & OBJECT_COMPRESSED_ZSTD) && !JOURNAL_HEADER_COMPRESSED_ZSTD(f->header)) {
error(p, "ZSTD compressed object in file without ZSTD compression");
r = -EBADMSG;
goto fail;
}
switch (o->object.type) {
case OBJECT_DATA:
+2 -2
View File
@@ -2327,7 +2327,7 @@ _public_ int sd_journal_get_data(sd_journal *j, const char *field, const void **
compression = o->object.flags & OBJECT_COMPRESSION_MASK;
if (compression) {
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
r = decompress_startswith(compression,
o->data.payload, l,
&f->compress_buffer, &f->compress_buffer_size,
@@ -2394,7 +2394,7 @@ static int return_data(sd_journal *j, JournalFile *f, Object *o, const void **da
compression = o->object.flags & OBJECT_COMPRESSION_MASK;
if (compression) {
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
size_t rsize;
int r;
+5 -2
View File
@@ -17,7 +17,7 @@ typedef int (compress_t)(const void *src, uint64_t src_size, void *dst,
typedef int (decompress_t)(const void *src, uint64_t src_size,
void **dst, size_t *dst_alloc_size, size_t* dst_size, size_t dst_max);
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
static usec_t arg_duration;
static size_t arg_start;
@@ -143,7 +143,7 @@ static void test_compress_decompress(const char* label, const char* type,
#endif
int main(int argc, char *argv[]) {
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
test_setup_logging(LOG_INFO);
if (argc >= 2) {
@@ -167,6 +167,9 @@ int main(int argc, char *argv[]) {
#endif
#if HAVE_LZ4
test_compress_decompress("LZ4", i, compress_blob_lz4, decompress_blob_lz4);
#endif
#if HAVE_ZSTD
test_compress_decompress("ZSTD", i, compress_blob_zstd, decompress_blob_zstd);
#endif
}
return 0;
+17
View File
@@ -339,8 +339,25 @@ int main(int argc, char *argv[]) {
#endif
#if HAVE_ZSTD
test_compress_decompress("ZSTD", compress_blob_zstd, decompress_blob_zstd,
text, sizeof(text), false);
test_compress_decompress("ZSTD", compress_blob_zstd, decompress_blob_zstd,
data, sizeof(data), true);
test_decompress_startswith("ZSTD",
compress_blob_zstd, decompress_startswith_zstd,
text, sizeof(text), false);
test_decompress_startswith("ZSTD",
compress_blob_zstd, decompress_startswith_zstd,
data, sizeof(data), true);
test_decompress_startswith("ZSTD",
compress_blob_zstd, decompress_startswith_zstd,
huge, HUGE_SIZE, true);
test_compress_stream("ZSTD", "zstdcat",
compress_stream_zstd, decompress_stream_zstd, srcfile);
test_decompress_startswith_short("ZSTD", compress_blob_zstd, decompress_startswith_zstd);
#else
log_info("/* ZSTD test skipped */");
#endif
+2 -2
View File
@@ -157,7 +157,7 @@ static void test_empty(void) {
(void) journal_file_close(f4);
}
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
static bool check_compressed(uint64_t compress_threshold, uint64_t data_size) {
dual_timestamp ts;
JournalFile *f;
@@ -251,7 +251,7 @@ int main(int argc, char *argv[]) {
test_non_empty();
test_empty();
#if HAVE_XZ || HAVE_LZ4
#if HAVE_XZ || HAVE_LZ4 || HAVE_ZSTD
test_min_compress_size();
#endif