diff options
author | Michael Cahill <michael.cahill@mongodb.com> | 2015-05-12 14:47:28 +1000 |
---|---|---|
committer | Michael Cahill <michael.cahill@mongodb.com> | 2015-05-12 14:47:28 +1000 |
commit | 9880d7563f74ec1e7033bae0e2f5aa312abc7b20 (patch) | |
tree | 9d1893587cc34c117619340b8f2f673666a7e206 /ext | |
parent | 8f6b8fd4647186cc6ce68cb31ecd085c4323157e (diff) | |
parent | df5a23d353d141afcbdb6cdced0538c87e99d389 (diff) | |
download | mongo-9880d7563f74ec1e7033bae0e2f5aa312abc7b20.tar.gz |
Merge branch 'develop' into encryption-api
Conflicts:
ext/compressors/lz4/lz4_compress.c
Diffstat (limited to 'ext')
-rw-r--r-- | ext/compressors/lz4/lz4_compress.c | 283 |
1 files changed, 208 insertions, 75 deletions
diff --git a/ext/compressors/lz4/lz4_compress.c b/ext/compressors/lz4/lz4_compress.c index 6b602b3d45c..0906e1d131d 100644 --- a/ext/compressors/lz4/lz4_compress.c +++ b/ext/compressors/lz4/lz4_compress.c @@ -48,19 +48,41 @@ typedef struct { } LZ4_COMPRESSOR; /* + * LZ4 decompression requires the exact compressed byte count returned by the + * LZ4_compress and LZ4_compress_destSize functions. WiredTiger doesn't track + * that value, store it in the destination buffer. + * + * Additionally, LZ4_compress_destSize may compress into the middle of a record, + * and after decompression we return the length to the last record successfully + * decompressed, not the number of bytes decompressed; store that value in the + * destination buffer as well. + * + * Use fixed-size, 4B values (WiredTiger never writes buffers larger than 4GB). + * + * The unused field is available for a mode flag if one is needed in the future, + * we guarantee it's 0. + */ +typedef struct { + uint32_t compressed_len; /* True compressed length */ + uint32_t uncompressed_len; /* True uncompressed source length */ + uint32_t useful_len; /* Decompression return value */ + uint32_t unused; /* Guaranteed to be 0 */ +} LZ4_PREFIX; + +/* * lz4_error -- * Output an error message, and return a standard error code. */ static int lz4_error( - WT_COMPRESSOR *compressor, WT_SESSION *session, const char *call, int zret) + WT_COMPRESSOR *compressor, WT_SESSION *session, const char *call, int error) { WT_EXTENSION_API *wt_api; wt_api = ((LZ4_COMPRESSOR *)compressor)->wt_api; (void)wt_api->err_printf(wt_api, - session, "lz4 error: %s: %d", call, zret); + session, "lz4 error: %s: %d", call, error); return (WT_ERROR); } @@ -74,39 +96,34 @@ lz4_compress(WT_COMPRESSOR *compressor, WT_SESSION *session, uint8_t *dst, size_t dst_len, size_t *result_lenp, int *compression_failed) { - char *lz4buf; - size_t lz4_len; + LZ4_PREFIX prefix; + int lz4_len; - /* - * The buffer should always be large enough due to the lz4_pre_size - * call, but be paranoid and error if it isn't. - */ - if (dst_len < src_len + sizeof(size_t)) - return (lz4_error(compressor, session, - "LZ4 compress buffer too small", 0)); + (void)compressor; /* Unused parameters */ + (void)session; + (void)dst_len; - /* Store the length of the compressed block in the first 8 bytes. */ - lz4buf = (char *)dst + sizeof(size_t); - lz4_len = (size_t)LZ4_compress((const char *)src, lz4buf, (int)src_len); + /* Compress, starting after the prefix bytes. */ + lz4_len = LZ4_compress( + (const char *)src, (char *)dst + sizeof(LZ4_PREFIX), (int)src_len); /* - * Flag no-compression if the result was larger than the original - * size or compression failed. + * If compression succeeded and the compressed length is smaller than + * the original size, return success. */ - if (lz4_len == 0 || lz4_len + sizeof(size_t) >= src_len) - *compression_failed = 1; - else { - /* - * On decompression, lz4 requires the exact compressed byte - * count (the current value of lz4_len). WiredTiger does not - * preserve that value, so save lz4_len at the beginning of the - * destination buffer. - */ - *(size_t *)dst = lz4_len; - *result_lenp = lz4_len + sizeof(size_t); + if (lz4_len != 0 && (size_t)lz4_len + sizeof(LZ4_PREFIX) < src_len) { + prefix.compressed_len = (uint32_t)lz4_len; + prefix.uncompressed_len = (uint32_t)src_len; + prefix.useful_len = (uint32_t)src_len; + prefix.unused = 0; + memcpy(dst, &prefix, sizeof(LZ4_PREFIX)); + + *result_lenp = (size_t)lz4_len + sizeof(LZ4_PREFIX); *compression_failed = 0; + return (0); } + *compression_failed = 1; return (0); } @@ -121,40 +138,143 @@ lz4_decompress(WT_COMPRESSOR *compressor, WT_SESSION *session, size_t *result_lenp) { WT_EXTENSION_API *wt_api; - char *compressed_data; + LZ4_PREFIX prefix; int decoded; - size_t src_data_len; + uint8_t *dst_tmp; + + (void)src_len; /* Unused parameters */ wt_api = ((LZ4_COMPRESSOR *)compressor)->wt_api; - /* Retrieve compressed length from start of the data buffer. */ - src_data_len = *(size_t *)src; - if (src_data_len + sizeof(size_t) > src_len) { - (void)wt_api->err_printf(wt_api, - session, - "lz4_decompress: stored size exceeds buffer size"); - return (WT_ERROR); + /* + * Retrieve the true length of the compressed block and source and the + * decompressed bytes to return from the start of the source buffer. + */ + memcpy(&prefix, src, sizeof(LZ4_PREFIX)); + + /* + * Decompress, starting after the prefix bytes. Use safe decompression: + * we rely on decompression to detect corruption. + * + * Two code paths, one with and one without a bounce buffer. When doing + * raw compression, we compress to a target size irrespective of row + * boundaries, and return to our caller a "useful" compression length + * based on the last complete row that was compressed. Our caller stores + * that length, not the length of bytes actually compressed by LZ4. In + * other words, our caller doesn't know how many bytes will result from + * decompression, likely hasn't provided us a large enough buffer, and + * we have to allocate a scratch buffer. + */ + if (dst_len < prefix.uncompressed_len) { + if ((dst_tmp = wt_api->scr_alloc( + wt_api, session, (size_t)prefix.uncompressed_len)) == NULL) + return (ENOMEM); + + decoded = LZ4_decompress_safe( + (const char *)src + sizeof(LZ4_PREFIX), (char *)dst_tmp, + (int)prefix.compressed_len, (int)prefix.uncompressed_len); + + if (decoded >= 0) + memcpy(dst, dst_tmp, dst_len); + wt_api->scr_free(wt_api, session, dst_tmp); + } else + decoded = LZ4_decompress_safe( + (const char *)src + sizeof(LZ4_PREFIX), + (char *)dst, (int)prefix.compressed_len, (int)dst_len); + + if (decoded >= 0) { + *result_lenp = prefix.useful_len; + return (0); } - /* Skip over the data size to the start of compressed data. */ - compressed_data = (char *)src + sizeof(size_t); + return ( + lz4_error(compressor, session, "LZ4 decompress error", decoded)); +} + +/* + * lz4_find_slot -- + * Find the slot containing the target offset (binary search). + */ +static inline uint32_t +lz4_find_slot(int target_arg, uint32_t *offsets, uint32_t slots) +{ + uint32_t base, indx, limit, target; + + indx = 1; /* -Wuninitialized */ + + target = (uint32_t)target_arg; /* Type conversion */ + + /* Fast check if we consumed it all, it's a likely result. */ + if (target >= offsets[slots]) + return (slots); /* - * The destination buffer length should always be sufficient because - * wiredtiger keeps track of the byte count before compression. Use - * safe decompression: we may be relying on decompression to detect - * corruption. + * Figure out which slot we got to: binary search. Note the test of + * offset (slot + 1), that's (end-byte + 1) for slot. */ - decoded = LZ4_decompress_safe( - compressed_data, (char *)dst, (int)src_data_len, (int)dst_len); + for (base = 0, limit = slots; limit != 0; limit >>= 1) { + indx = base + (limit >> 1); + if (target > offsets[indx + 1]) { + base = indx + 1; + --limit; + } + } + + return (indx); +} + +/* + * lz4_compress_raw -- + * Pack records into a specified on-disk page size. + */ +static int +lz4_compress_raw(WT_COMPRESSOR *compressor, WT_SESSION *session, + size_t page_max, int split_pct, size_t extra, + uint8_t *src, uint32_t *offsets, uint32_t slots, + uint8_t *dst, size_t dst_len, int final, + size_t *result_lenp, uint32_t *result_slotsp) +{ + LZ4_PREFIX prefix; + int lz4_len; + uint32_t slot; + int sourceSize, targetDestSize; + + (void)compressor; /* Unused parameters */ + (void)session; + (void)split_pct; + (void)final; + + sourceSize = (int)offsets[slots]; /* Type conversion */ + targetDestSize = + (int)((dst_len < page_max ? dst_len : page_max) - extra); - if (decoded < 0) - return (lz4_error(compressor, session, - "LZ4 decompress error", decoded)); + /* Compress, starting after the prefix bytes. */ + lz4_len = LZ4_compress_destSize((const char *)src, + (char *)dst + sizeof(LZ4_PREFIX), &sourceSize, targetDestSize); - /* return the uncompressed data length */ - *result_lenp = dst_len; + /* + * If compression succeeded and the compressed length is smaller than + * the original size, return success. + */ + if (lz4_len != 0) { + /* Find the first slot we didn't compress. */ + slot = lz4_find_slot(sourceSize, offsets, slots); + + if ((size_t)lz4_len + sizeof(LZ4_PREFIX) < offsets[slot]) { + prefix.compressed_len = (uint32_t)lz4_len; + prefix.uncompressed_len = (uint32_t)sourceSize; + prefix.useful_len = offsets[slot]; + prefix.unused = 0; + memcpy(dst, &prefix, sizeof(LZ4_PREFIX)); + + *result_slotsp = slot; + *result_lenp = (size_t)lz4_len + sizeof(LZ4_PREFIX); + return (0); + } + } + *result_slotsp = 0; + *result_lenp = 1; return (0); } @@ -164,18 +284,18 @@ lz4_decompress(WT_COMPRESSOR *compressor, WT_SESSION *session, */ static int lz4_pre_size(WT_COMPRESSOR *compressor, WT_SESSION *session, - uint8_t *src, size_t src_len, - size_t *result_lenp) + uint8_t *src, size_t src_len, size_t *result_lenp) { - (void)compressor; + (void)compressor; /* Unused parameters */ (void)session; (void)src; /* - * LZ4 can use more space than the input data size, use the library - * calculation of that overhead (plus our overhead) to be safe. + * In block mode, LZ4 can use more space than the input data size, use + * the library calculation of that overhead (plus our overhead) to be + * safe. */ - *result_lenp = LZ4_COMPRESSBOUND(src_len) + sizeof(size_t); + *result_lenp = LZ4_COMPRESSBOUND(src_len) + sizeof(LZ4_PREFIX); return (0); } @@ -186,39 +306,30 @@ lz4_pre_size(WT_COMPRESSOR *compressor, WT_SESSION *session, static int lz4_terminate(WT_COMPRESSOR *compressor, WT_SESSION *session) { - (void)session; + (void)session; /* Unused parameters */ - /* Free the allocated memory. */ free(compressor); - return (0); } -int lz4_extension_init(WT_CONNECTION *, WT_CONFIG_ARG *); - /* - * lz4_extension_init -- - * A simple shared library compression example. + * lz4_add_compressor -- + * Add a LZ4 compressor. */ -int -lz4_extension_init(WT_CONNECTION *connection, WT_CONFIG_ARG *config) +static int +lz_add_compressor(WT_CONNECTION *connection, int raw, const char *name) { LZ4_COMPRESSOR *lz4_compressor; - (void)config; /* Unused parameters */ - + /* + * There are two almost identical LZ4 compressors: one using raw + * compression to target a specific block size, and one without. + */ if ((lz4_compressor = calloc(1, sizeof(LZ4_COMPRESSOR))) == NULL) return (errno); - /* - * Allocate a local compressor structure, with a WT_COMPRESSOR structure - * as the first field, allowing us to treat references to either type of - * structure as a reference to the other type. - * - * Heap memory (not static), because it can support multiple databases. - */ lz4_compressor->compressor.compress = lz4_compress; - lz4_compressor->compressor.compress_raw = NULL; + lz4_compressor->compressor.compress_raw = raw ? lz4_compress_raw : NULL; lz4_compressor->compressor.decompress = lz4_decompress; lz4_compressor->compressor.pre_size = lz4_pre_size; lz4_compressor->compressor.terminate = lz4_terminate; @@ -227,7 +338,29 @@ lz4_extension_init(WT_CONNECTION *connection, WT_CONFIG_ARG *config) /* Load the compressor */ return (connection->add_compressor( - connection, "lz4", (WT_COMPRESSOR *)lz4_compressor, NULL)); + connection, name, (WT_COMPRESSOR *)lz4_compressor, NULL)); +} + +int lz4_extension_init(WT_CONNECTION *, WT_CONFIG_ARG *); + +/* + * lz4_extension_init -- + * WiredTiger LZ4 compression extension - called directly when LZ4 support + * is built in, or via wiredtiger_extension_init when LZ4 support is included + * via extension loading. + */ +int +lz4_extension_init(WT_CONNECTION *connection, WT_CONFIG_ARG *config) +{ + int ret; + + (void)config; /* Unused parameters */ + + if ((ret = lz_add_compressor(connection, 1, "lz4")) != 0) + return (ret); + if ((ret = lz_add_compressor(connection, 0, "lz4-noraw")) != 0) + return (ret); + return (0); } /* |