From d936a8621339a86e9151bd5b942e25afd1dabc64 Mon Sep 17 00:00:00 2001 From: Georgios Kokolatos Date: Tue, 2 Nov 2021 11:26:51 +0000 Subject: [PATCH v8 2/2] Teach pg_receivewal to use LZ4 compression The program pg_receivewal can use gzip compression to store the received WAL. This commit teaches it to also be able to use LZ4 compression. It is required that the binary is build using the -llz4 flag. It is enabled via the --with-lz4 flag on configuration time. The option `--compression-method` has been expanded to inlude the value [LZ4]. The option `--compress` can not be used with LZ4 compression. Under the hood there is nothing exceptional to be noted. Tar based archives have not yet been taught to use LZ4 compression. If that is felt useful, then it is easy to be added in the future. Tests have been added to verify the creation and correctness of the generated LZ4 files. The later is achieved by the use of LZ4 program, if present in the installation. --- doc/src/sgml/ref/pg_receivewal.sgml | 5 +- src/Makefile.global.in | 1 + src/bin/pg_basebackup/Makefile | 1 + src/bin/pg_basebackup/pg_receivewal.c | 129 +++++++++++++++ src/bin/pg_basebackup/t/020_pg_receivewal.pl | 72 ++++++++- src/bin/pg_basebackup/walmethods.c | 159 ++++++++++++++++++- src/bin/pg_basebackup/walmethods.h | 1 + 7 files changed, 358 insertions(+), 10 deletions(-) diff --git a/doc/src/sgml/ref/pg_receivewal.sgml b/doc/src/sgml/ref/pg_receivewal.sgml index f95cffcd5e..2fb2431b3d 100644 --- a/doc/src/sgml/ref/pg_receivewal.sgml +++ b/doc/src/sgml/ref/pg_receivewal.sgml @@ -268,8 +268,11 @@ PostgreSQL documentation Enables compression of write-ahead logs using the specified method. - Supported values zlib, + Supported values are lz4, zlib, and none. + For the LZ4 method to be available, + PostgreSQL must have been have been compiled + with . diff --git a/src/Makefile.global.in b/src/Makefile.global.in index 533c12fef9..05c54b27de 100644 --- a/src/Makefile.global.in +++ b/src/Makefile.global.in @@ -350,6 +350,7 @@ XGETTEXT = @XGETTEXT@ GZIP = gzip BZIP2 = bzip2 +LZ4 = lz4 DOWNLOAD = wget -O $@ --no-use-server-timestamps #DOWNLOAD = curl -o $@ diff --git a/src/bin/pg_basebackup/Makefile b/src/bin/pg_basebackup/Makefile index 459d514183..387d728345 100644 --- a/src/bin/pg_basebackup/Makefile +++ b/src/bin/pg_basebackup/Makefile @@ -24,6 +24,7 @@ export TAR # used by the command "gzip" to pass down options, so stick with a different # name. export GZIP_PROGRAM=$(GZIP) +export LZ4 override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS) LDFLAGS_INTERNAL += -L$(top_builddir)/src/fe_utils -lpgfeutils $(libpq_pgport) diff --git a/src/bin/pg_basebackup/pg_receivewal.c b/src/bin/pg_basebackup/pg_receivewal.c index 9641f4a2f4..87f8589a71 100644 --- a/src/bin/pg_basebackup/pg_receivewal.c +++ b/src/bin/pg_basebackup/pg_receivewal.c @@ -29,6 +29,10 @@ #include "receivelog.h" #include "streamutil.h" +#ifdef HAVE_LIBLZ4 +#include "lz4frame.h" +#endif + /* Time to sleep between reconnection attempts */ #define RECONNECT_SLEEP_TIME 5 @@ -137,6 +141,15 @@ is_xlogfilename(const char *filename, bool *ispartial, return true; } + /* File looks like a complete LZ4 compressed XLOG file */ + if (fname_len == XLOG_FNAME_LEN + strlen(".lz4") && + strcmp(filename + XLOG_FNAME_LEN, ".lz4") == 0) + { + *ispartial = false; + *wal_compression_method = COMPRESSION_LZ4; + return true; + } + /* File looks like a partial uncompressed XLOG file */ if (fname_len == XLOG_FNAME_LEN + strlen(".partial") && strcmp(filename + XLOG_FNAME_LEN, ".partial") == 0) @@ -155,6 +168,15 @@ is_xlogfilename(const char *filename, bool *ispartial, return true; } + /* File looks like a partial LZ4 compressed XLOG file */ + if (fname_len == XLOG_FNAME_LEN + strlen(".lz4.partial") && + strcmp(filename + XLOG_FNAME_LEN, ".lz4.partial") == 0) + { + *ispartial = true; + *wal_compression_method = COMPRESSION_LZ4; + return true; + } + /* File does not look like something we recognise */ return false; } @@ -285,6 +307,10 @@ FindStreamingStart(uint32 *tli) * lower than 4GB, and then compare it to the size of a completed * segment. The 4 last bytes correspond to the ISIZE member according * to http://www.zlib.org/rfc-gzip.html. + * + * For LZ4 compressed segments, uncompress the file in a throw away + * buffer keeping track of the uncompressed size. Then compare it to + * the size of a completed segment. */ if (!ispartial && wal_compression_method == COMPRESSION_NONE) { @@ -351,6 +377,99 @@ FindStreamingStart(uint32 *tli) continue; } } + else if (!ispartial && compression_method == COMPRESSION_LZ4) + { +#ifdef HAVE_LIBLZ4 +#define LZ4_CHUNK_SZ 4096 + int fd; + int r; + size_t uncompressed_size = 0; + char fullpath[MAXPGPATH * 2]; + char readbuf[LZ4_CHUNK_SZ]; + char outbuf[LZ4_CHUNK_SZ]; + LZ4F_decompressionContext_t ctx = NULL; + LZ4F_errorCode_t status; + + snprintf(fullpath, sizeof(fullpath), "%s/%s", basedir, dirent->d_name); + + fd = open(fullpath, O_RDONLY | PG_BINARY, 0); + if (fd < 0) + { + pg_log_error("could not open file \"%s\": %m", fullpath); + exit(1); + } + + status = LZ4F_createDecompressionContext(&ctx, LZ4F_VERSION); + if (LZ4F_isError(status)) + { + pg_log_error("could not create LZ4 decompression context: %s", + LZ4F_getErrorName(status)); + exit(1); + } + + while (1) + { + char *readp; + char *readend; + + r = read(fd, readbuf, sizeof(readbuf)); + if (r < 0) + { + pg_log_error("could not read file \"%s\": %m", fullpath); + exit(1); + } + + /* Done reading */ + if (r == 0) + break; + + readp = readbuf; + readend = readbuf + r; + while (readp < readend) + { + size_t read_size = 1; + size_t out_size = 1; + + status = LZ4F_decompress(ctx, outbuf, &out_size, + readbuf, &read_size, NULL); + if (LZ4F_isError(status)) + { + pg_log_error("could not decompress file \"%s\": %s", + fullpath, + LZ4F_getErrorName(status)); + exit(1); + } + + readp += read_size; + uncompressed_size += out_size; + } + } + + close(fd); + + status = LZ4F_freeDecompressionContext(ctx); + if (LZ4F_isError(status)) + { + pg_log_error("could not free LZ4 decompression context: %s", + LZ4F_getErrorName(status)); + exit(1); + } + + if (uncompressed_size != WalSegSz) + { + pg_log_warning("compressed segment file \"%s\" has incorrect uncompressed size %ld, skipping", + dirent->d_name, uncompressed_size); + (void) LZ4F_freeDecompressionContext(ctx); + continue; + } +#else + pg_log_error("could not check segment file \"%s\" compressed with LZ4", + dirent->d_name); + pg_log_error("this build does not support compression with %s", + "LZ4"); + exit(1); +#endif + } /* Looks like a valid segment. Remember that we saw it. */ if ((segno > high_segno) || @@ -630,6 +749,8 @@ main(int argc, char **argv) case 'I': if (pg_strcasecmp(optarg, "zlib") == 0) compression_method = COMPRESSION_ZLIB; + else if (pg_strcasecmp(optarg, "lz4") == 0) + compression_method = COMPRESSION_LZ4; else if (pg_strcasecmp(optarg, "none") == 0) compression_method = COMPRESSION_NONE; else @@ -733,6 +854,14 @@ main(int argc, char **argv) "gzip"); exit(1); } +#endif +#ifndef HAVE_LIBLZ4 + if (compression_method == COMPRESSION_LZ4) + { + pg_log_error("this build does not support compression with %s", + "LZ4"); + exit(1); + } #endif } diff --git a/src/bin/pg_basebackup/t/020_pg_receivewal.pl b/src/bin/pg_basebackup/t/020_pg_receivewal.pl index 56c4a0d2af..ef4464556a 100644 --- a/src/bin/pg_basebackup/t/020_pg_receivewal.pl +++ b/src/bin/pg_basebackup/t/020_pg_receivewal.pl @@ -5,7 +5,7 @@ use strict; use warnings; use PostgreSQL::Test::Utils; use PostgreSQL::Test::Cluster; -use Test::More tests => 37; +use Test::More tests => 42; program_help_ok('pg_receivewal'); program_version_ok('pg_receivewal'); @@ -138,13 +138,69 @@ SKIP: "gzip verified the integrity of compressed WAL segments"); } +# Check LZ4 compression if available +SKIP: +{ + skip "postgres was not built with LZ4 support", 5 + if (!check_pg_config("#define HAVE_LIBLZ4 1")); + + # Generate more WAL including one completed, compressed segment. + $primary->psql('postgres', 'SELECT pg_switch_wal();'); + $nextlsn = + $primary->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn();'); + chomp($nextlsn); + $primary->psql('postgres', 'INSERT INTO test_table VALUES (3);'); + + # Stream up to the given position + $primary->command_ok( + [ + 'pg_receivewal', '-D', + $stream_dir, '--verbose', + '--endpos', $nextlsn, + '--no-loop', '--compression-method', + 'lz4' + ], + 'streaming some WAL using --compression-method=lz4'); + + # Verify that the stored files are generated with their expected + # names. + my @lz4_wals = glob "$stream_dir/*.lz4"; + is(scalar(@lz4_wals), 1, + "one WAL segment compressed with LZ4 was created"); + my @lz4_partial_wals = glob "$stream_dir/*.lz4.partial"; + is(scalar(@lz4_partial_wals), + 1, "one partial WAL segment compressed with LZ4 was created"); + + # Verify that the start streaming position is computed correctly by + # comparing it with the partial file generated previously. The name + # of the previous partial, now-completed WAL segment is updated, keeping + # its base number. + $partial_wals[0] =~ s/(\.gz)?\.partial$/.lz4/; + is($lz4_wals[0] eq $partial_wals[0], + 1, "one partial WAL segment is now completed"); + # Update the list of partial wals with the current one. + @partial_wals = @lz4_partial_wals; + + # Check the integrity of the completed segment, if LZ4 is an available + # command. + my $lz4 = $ENV{LZ4}; + skip "program lz4 is not found in your system", 1 + if ( !defined $lz4 + || $lz4 eq '' + || system_log($lz4, '--version') != 0); + + my $lz4_is_valid = system_log($lz4, '-t', @lz4_wals); + is($lz4_is_valid, 0, + "lz4 verified the integrity of compressed WAL segments"); +} + # Verify that the start streaming position is computed and that the value is -# correct regardless of whether ZLIB is available. +# correct regardless of whether any compression is available. $primary->psql('postgres', 'SELECT pg_switch_wal();'); $nextlsn = $primary->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn();'); chomp($nextlsn); -$primary->psql('postgres', 'INSERT INTO test_table VALUES (3);'); +$primary->psql('postgres', 'INSERT INTO test_table VALUES (4);'); $primary->command_ok( [ 'pg_receivewal', '-D', $stream_dir, '--verbose', @@ -152,7 +208,7 @@ $primary->command_ok( ], "streaming some WAL"); -$partial_wals[0] =~ s/(\.gz)?.partial//; +$partial_wals[0] =~ s/(\.gz|\.lz4)?.partial//; ok(-e $partial_wals[0], "check that previously partial WAL is now complete"); # Permissions on WAL files should be default @@ -190,7 +246,7 @@ my $walfile_streamed = $primary->safe_psql( # Switch to a new segment, to make sure that the segment retained by the # slot is still streamed. This may not be necessary, but play it safe. -$primary->psql('postgres', 'INSERT INTO test_table VALUES (4);'); +$primary->psql('postgres', 'INSERT INTO test_table VALUES (5);'); $primary->psql('postgres', 'SELECT pg_switch_wal();'); $nextlsn = $primary->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn();'); @@ -198,7 +254,7 @@ chomp($nextlsn); # Add a bit more data to accelerate the end of the next pg_receivewal # commands. -$primary->psql('postgres', 'INSERT INTO test_table VALUES (5);'); +$primary->psql('postgres', 'INSERT INTO test_table VALUES (6);'); # Check case where the slot does not exist. $primary->command_fails_like( @@ -253,13 +309,13 @@ $standby->promote; # on the new timeline. my $walfile_after_promotion = $standby->safe_psql('postgres', "SELECT pg_walfile_name(pg_current_wal_insert_lsn());"); -$standby->psql('postgres', 'INSERT INTO test_table VALUES (6);'); +$standby->psql('postgres', 'INSERT INTO test_table VALUES (7);'); $standby->psql('postgres', 'SELECT pg_switch_wal();'); $nextlsn = $standby->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn();'); chomp($nextlsn); # This speeds up the operation. -$standby->psql('postgres', 'INSERT INTO test_table VALUES (7);'); +$standby->psql('postgres', 'INSERT INTO test_table VALUES (8);'); # Now try to resume from the slot after the promotion. my $timeline_dir = $primary->basedir . '/timeline_wal'; diff --git a/src/bin/pg_basebackup/walmethods.c b/src/bin/pg_basebackup/walmethods.c index 068b276251..1ec89f1d2b 100644 --- a/src/bin/pg_basebackup/walmethods.c +++ b/src/bin/pg_basebackup/walmethods.c @@ -17,6 +17,10 @@ #include #include #include + +#ifdef HAVE_LIBLZ4 +#include +#endif #ifdef HAVE_LIBZ #include #endif @@ -30,6 +34,9 @@ /* Size of zlib buffer for .tar.gz */ #define ZLIB_OUT_SIZE 4096 +/* Size of LZ4 input chunk for .lz4 */ +#define LZ4_IN_SIZE 4096 + /*------------------------------------------------------------------------- * WalDirectoryMethod - write wal to a directory looking like pg_wal *------------------------------------------------------------------------- @@ -60,6 +67,11 @@ typedef struct DirectoryMethodFile #ifdef HAVE_LIBZ gzFile gzfp; #endif +#ifdef HAVE_LIBLZ4 + LZ4F_compressionContext_t ctx; + size_t lz4bufsize; + void *lz4buf; +#endif } DirectoryMethodFile; static const char * @@ -76,7 +88,8 @@ dir_get_file_name(const char *pathname, const char *temp_suffix) snprintf(filename, MAXPGPATH, "%s%s%s", pathname, - dir_data->compression_method == COMPRESSION_ZLIB ? ".gz" : "", + dir_data->compression_method == COMPRESSION_ZLIB ? ".gz" : + dir_data->compression_method == COMPRESSION_LZ4 ? ".lz4" : "", temp_suffix ? temp_suffix : ""); return filename; @@ -92,6 +105,11 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_ #ifdef HAVE_LIBZ gzFile gzfp = NULL; #endif +#ifdef HAVE_LIBLZ4 + LZ4F_compressionContext_t ctx = NULL; + size_t lz4bufsize = 0; + void *lz4buf = NULL; +#endif filename = dir_get_file_name(pathname, temp_suffix); snprintf(tmppath, sizeof(tmppath), "%s/%s", @@ -126,6 +144,49 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_ } } #endif +#ifdef HAVE_LIBLZ4 + if (dir_data->compression_method == COMPRESSION_LZ4) + { + size_t ctx_out; + size_t header_size; + + ctx_out = LZ4F_createCompressionContext(&ctx, LZ4F_VERSION); + lz4bufsize = LZ4F_compressBound(LZ4_IN_SIZE, NULL); + if (LZ4F_isError(ctx_out)) + { + close(fd); + return NULL; + } + + lz4buf = pg_malloc0(lz4bufsize); + + /* add the header */ + header_size = LZ4F_compressBegin(ctx, lz4buf, lz4bufsize, NULL); + if (LZ4F_isError(header_size)) + { + pg_free(lz4buf); + close(fd); + return NULL; + } + + errno = 0; + if (write(fd, lz4buf, header_size) != header_size) + { + int save_errno = errno; + + (void) LZ4F_compressEnd(ctx, lz4buf, lz4bufsize, NULL); + (void) LZ4F_freeCompressionContext(ctx); + pg_free(lz4buf); + close(fd); + + /* + * If write didn't set errno, assume problem is no disk space. + */ + errno = save_errno ? save_errno : ENOSPC; + return NULL; + } + } +#endif /* Do pre-padding on non-compressed files */ if (pad_to_size && dir_data->compression_method == COMPRESSION_NONE) @@ -176,6 +237,16 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_ if (dir_data->compression_method == COMPRESSION_ZLIB) gzclose(gzfp); else +#endif +#ifdef HAVE_LIBLZ4 + if (dir_data->compression_method == COMPRESSION_LZ4) + { + (void) LZ4F_compressEnd(ctx, lz4buf, lz4bufsize, NULL); + (void) LZ4F_freeCompressionContext(ctx); + pg_free(lz4buf); + close(fd); + } + else #endif close(fd); return NULL; @@ -187,6 +258,15 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_ if (dir_data->compression_method == COMPRESSION_ZLIB) f->gzfp = gzfp; #endif +#ifdef HAVE_LIBLZ4 + if (dir_data->compression_method == COMPRESSION_LZ4) + { + f->ctx = ctx; + f->lz4buf = lz4buf; + f->lz4bufsize = lz4bufsize; + } +#endif + f->fd = fd; f->currpos = 0; f->pathname = pg_strdup(pathname); @@ -209,6 +289,43 @@ dir_write(Walfile f, const void *buf, size_t count) if (dir_data->compression_method == COMPRESSION_ZLIB) r = (ssize_t) gzwrite(df->gzfp, buf, count); else +#endif +#ifdef HAVE_LIBLZ4 + if (dir_data->compression_method == COMPRESSION_LZ4) + { + size_t chunk; + size_t remaining; + const void *inbuf = buf; + + remaining = count; + while (remaining > 0) + { + size_t compressed; + + if (remaining > LZ4_IN_SIZE) + chunk = LZ4_IN_SIZE; + else + chunk = remaining; + + remaining -= chunk; + compressed = LZ4F_compressUpdate(df->ctx, + df->lz4buf, df->lz4bufsize, + inbuf, chunk, + NULL); + + if (LZ4F_isError(compressed)) + return -1; + + if (write(df->fd, df->lz4buf, compressed) != compressed) + return -1; + + inbuf = ((char *) inbuf) + chunk; + } + + /* Our caller keeps track of the uncompressed size. */ + r = (ssize_t) count; + } + else #endif r = write(df->fd, buf, count); if (r > 0) @@ -239,6 +356,25 @@ dir_close(Walfile f, WalCloseMethod method) if (dir_data->compression_method == COMPRESSION_ZLIB) r = gzclose(df->gzfp); else +#endif +#ifdef HAVE_LIBLZ4 + if (dir_data->compression_method == COMPRESSION_LZ4) + { + size_t compressed; + + compressed = LZ4F_compressEnd(df->ctx, + df->lz4buf, df->lz4bufsize, + NULL); + + if (LZ4F_isError(compressed)) + return -1; + + if (write(df->fd, df->lz4buf, compressed) != compressed) + return -1; + + r = close(df->fd); + } + else #endif r = close(df->fd); @@ -293,6 +429,12 @@ dir_close(Walfile f, WalCloseMethod method) } } +#ifdef HAVE_LIBLZ4 + pg_free(df->lz4buf); + /* supports free on NULL */ + LZ4F_freeCompressionContext(df->ctx); +#endif + pg_free(df->pathname); pg_free(df->fullpath); if (df->temp_suffix) @@ -317,6 +459,21 @@ dir_sync(Walfile f) return -1; } #endif +#ifdef HAVE_LIBLZ4 + if (dir_data->compression_method == COMPRESSION_LZ4) + { + DirectoryMethodFile *df = (DirectoryMethodFile *) f; + size_t compressed; + + /* Flush any internal buffers */ + compressed = LZ4F_flush(df->ctx, df->lz4buf, df->lz4bufsize, NULL); + if (LZ4F_isError(compressed)) + return -1; + + if (write(df->fd, df->lz4buf, compressed) != compressed) + return -1; + } +#endif return fsync(((DirectoryMethodFile *) f)->fd); } diff --git a/src/bin/pg_basebackup/walmethods.h b/src/bin/pg_basebackup/walmethods.h index 4fc7b3d2a3..3e378c87b6 100644 --- a/src/bin/pg_basebackup/walmethods.h +++ b/src/bin/pg_basebackup/walmethods.h @@ -21,6 +21,7 @@ typedef enum typedef enum { + COMPRESSION_LZ4, COMPRESSION_ZLIB, COMPRESSION_NONE } WalCompressionMethod; -- 2.25.1