From a5d75428cdfa4b14a6526c1678dab8a7fe10a68d Mon Sep 17 00:00:00 2001 From: Jakub Wartak Date: Wed, 5 Aug 2026 08:53:31 +0200 Subject: [PATCH v06082026 09/10] pg_basebackup: add support for Direct I/O and Async I/O using liburing Profiling pg_basebackup when doing high-data transfer reveals that it is highly CPU-bound on kernel side inside write() due to kernel's side page-cache memory allocations/cgroups memory accounting and copying the data from userspace back to kernel side, just to be copied again to the device. To avoid that we can do Direct Memory Transfers from our userspace buffers directly to the storage device, however we cannot do use synchronous API for that due to high individual device latency. On Linux however, we are already using liburing for the backend code, so this commit teaches pg_basebackup/astreamers (for writing plain and tar formats) to write using AIO+DIO with deep I/O queues. It realistic benchmarks and with adequate sequential storage bandwidth and no SSL encryption this makes possible for pg_basebackup -Fp to go from ~1610MB/s to ~2975MB/s as long as the data is hot in page-cache of the source server and the network and single stream TCP socket performance is able to keep up. This commits also exposes undocumented/debugging PG_BASEBACKUP_NODIO env variable, that when set allows disabling this optimization. Author: Jakub Wartak --- src/bin/pg_basebackup/Makefile | 2 +- src/bin/pg_basebackup/meson.build | 2 +- src/bin/pg_basebackup/pg_basebackup.c | 9 +- src/fe_utils/astreamer_file.c | 465 +++++++++++++++++++++++++- src/fe_utils/meson.build | 2 +- src/include/fe_utils/astreamer.h | 5 +- 6 files changed, 463 insertions(+), 22 deletions(-) diff --git a/src/bin/pg_basebackup/Makefile b/src/bin/pg_basebackup/Makefile index df94fc27d02..448abb71e99 100644 --- a/src/bin/pg_basebackup/Makefile +++ b/src/bin/pg_basebackup/Makefile @@ -42,7 +42,7 @@ BBOBJS = \ all: pg_basebackup pg_createsubscriber pg_receivewal pg_recvlogical pg_basebackup: $(BBOBJS) $(OBJS) | submake-libpq submake-libpgport submake-libpgfeutils - $(CC) $(CFLAGS) $(BBOBJS) $(OBJS) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X) + $(CC) $(CFLAGS) $(BBOBJS) $(OBJS) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) $(LIBURING_LIBS) -o $@$(X) pg_createsubscriber: pg_createsubscriber.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils $(CC) $(CFLAGS) $^ $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X) diff --git a/src/bin/pg_basebackup/meson.build b/src/bin/pg_basebackup/meson.build index d70ce5786a2..e4cb673e933 100644 --- a/src/bin/pg_basebackup/meson.build +++ b/src/bin/pg_basebackup/meson.build @@ -7,7 +7,7 @@ common_sources = files( 'walmethods.c', ) -pg_basebackup_deps = [frontend_code, libpq, lz4, zlib, zstd] +pg_basebackup_deps = [frontend_code, libpq, lz4, zlib, zstd, liburing] pg_basebackup_common = static_library('libpg_basebackup_common', common_sources, dependencies: pg_basebackup_deps, diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index 708b0727941..490395510e9 100644 --- a/src/bin/pg_basebackup/pg_basebackup.c +++ b/src/bin/pg_basebackup/pg_basebackup.c @@ -1158,7 +1158,8 @@ CreateBackupStreamer(char *archive_name, char *spclocation, streamer = astreamer_extractor_new(directory, get_tablespace_mapping, progress_update_filename, - backup_target_clientblackhole); + backup_target_clientblackhole, + verbose); } else { @@ -1196,7 +1197,7 @@ CreateBackupStreamer(char *archive_name, char *spclocation, */ if (compress->algorithm == PG_COMPRESSION_NONE) streamer = astreamer_plain_writer_new(archive_filename, - archive_file); + archive_file, verbose); else if (compress->algorithm == PG_COMPRESSION_GZIP) { if (!backup_target_clientblackhole) @@ -1209,7 +1210,7 @@ CreateBackupStreamer(char *archive_name, char *spclocation, if (!backup_target_clientblackhole) strlcat(archive_filename, ".lz4", sizeof(archive_filename)); streamer = astreamer_plain_writer_new(archive_filename, - archive_file); + archive_file, verbose); streamer = astreamer_lz4_compressor_new(streamer, compress); } else if (compress->algorithm == PG_COMPRESSION_ZSTD) @@ -1217,7 +1218,7 @@ CreateBackupStreamer(char *archive_name, char *spclocation, if (!backup_target_clientblackhole) strlcat(archive_filename, ".zst", sizeof(archive_filename)); streamer = astreamer_plain_writer_new(archive_filename, - archive_file); + archive_file, verbose); streamer = astreamer_zstd_compressor_new(streamer, compress); } else diff --git a/src/fe_utils/astreamer_file.c b/src/fe_utils/astreamer_file.c index 32df6b09fb9..d4d34e5d412 100644 --- a/src/fe_utils/astreamer_file.c +++ b/src/fe_utils/astreamer_file.c @@ -17,17 +17,61 @@ #include #include +#ifdef USE_LIBURING +#include +#endif #include "common/file_perm.h" #include "common/logging.h" #include "fe_utils/astreamer.h" +#ifdef USE_LIBURING +/* + * Parameters for the io_uring + O_DIRECT write path. Aim is to keep deep + * queue of the I/O devices populated by throwing large number of independent + * synchronous/DIRECT_IO requests, but in asynchronous way. We avoid latency + * of single synchronous write, and utilize the device up to the max what it + * allows. + */ +#define DIO_ALIGN 4096 /* O_DIRECT aligment */ +#define DIO_BUFSZ (1024 * 1024) /* size of each direct I/O write */ +#define DIO_NBUF 32 /* queue depth, XXX:expose it via getopt? */ +#define DIO_MIN_SIZE DIO_BUFSZ /* fsize threshold for activating O_DIRECT writes */ +#define DIO_SUBMIT_BATCH 8 /* how many SQEs to batch */ + +/* + * State for the dio/io_uring writer. Used by plain(file) and tar extractors. + * We buffer data as we recieve it, into the "pool" of buffers and submit them + * using DIO_BUFSZ sizes. + */ +typedef struct dio_writer +{ + struct io_uring ring; /* see io_uring(7) */ + bool ring_ready; /* was the ring and buffer initialized */ + char *pool; /* buffer memory, max size: DIO_NBUF * DIO_BUFSZ */ + bool busy[DIO_NBUF]; /* is buffer in flight? */ + size_t buflen[DIO_NBUF]; /* bytes submitted for in-flight buffer */ + int fd; /* DIO fd: if fd > 0 then it's active, -1 otherwise */ + const char *filename; /* for error handling */ + pgoff_t offset; /* offset of the next write */ + pgoff_t written; /* bytes written */ + int curidx; /* buffer being filled, or -1 */ + int curlen; /* bytes filled in current buffer */ + int inflight; /* prepared writes not yet reaped */ + int unsubmitted; /* SQEs prepared but not yet submitted */ + bool notified; /* already logged a fall-back-to-buffered? */ +} dio_writer; +#endif + typedef struct astreamer_plain_writer { astreamer base; char *pathname; - FILE *file; + FILE *file; /* if NULL, then dio.fd is used */ bool should_close_file; +#ifdef USE_LIBURING + dio_writer dio; +#endif } astreamer_plain_writer; typedef struct astreamer_extractor @@ -37,8 +81,12 @@ typedef struct astreamer_extractor const char *(*link_map) (const char *); void (*report_output_file) (const char *); char filename[MAXPGPATH]; - int fd; + int fd; /* if -1, then dio.fd is used */ bool discard_backup; + bool verbose; +#ifdef USE_LIBURING + dio_writer dio; +#endif } astreamer_extractor; static void astreamer_plain_writer_content(astreamer *streamer, @@ -66,6 +114,13 @@ static int create_file_for_extract(const char *filename, mode_t mode, bool discard_backup, pgoff_t size); static void write_file_range(int fd, const char *filename, const char *data, int len); +#ifdef USE_LIBURING +static bool dio_writer_start(dio_writer *dw, const char *filename, + pgoff_t prealloc_size, bool verbose); +static void dio_writer_write(dio_writer *dw, const char *data, int len); +static void dio_writer_finish(dio_writer *dw); +static void dio_writer_destroy(dio_writer *dw); +#endif static const astreamer_ops astreamer_extractor_ops = { .content = astreamer_extractor_content, @@ -83,7 +138,7 @@ static const astreamer_ops astreamer_extractor_ops = { * there. */ astreamer * -astreamer_plain_writer_new(char *pathname, FILE *file) +astreamer_plain_writer_new(char *pathname, FILE *file, bool verbose) { astreamer_plain_writer *streamer; @@ -93,13 +148,33 @@ astreamer_plain_writer_new(char *pathname, FILE *file) streamer->pathname = pstrdup(pathname); streamer->file = file; +#ifdef USE_LIBURING + streamer->dio.fd = -1; +#endif if (file == NULL) { - streamer->file = fopen(pathname, "wb"); - if (streamer->file == NULL) - pg_fatal("could not create file \"%s\": %m", pathname); - streamer->should_close_file = true; +#ifdef USE_LIBURING + + /* + * Fallback to classic buffered writes in case of: + * - env variable PG_BASEBACKUP_NODIO is set (debugging) + * - pipe/stdout is used + * - /dev/null is being used (-t client-blackhole) + * - filesystems without O_DIRECT support + */ + if (getenv("PG_BASEBACKUP_NODIO") == NULL && + dio_writer_start(&streamer->dio, streamer->pathname, 0, verbose) == true) { + /* do not use buffered writes, as the dio.fd is going to be used */ + streamer->file = NULL; + } else +#endif + { + streamer->file = fopen(pathname, "wb"); + if (streamer->file == NULL) + pg_fatal("could not create file \"%s\": %m", pathname); + streamer->should_close_file = true; + } } return &streamer->base; @@ -120,6 +195,13 @@ astreamer_plain_writer_content(astreamer *streamer, if (len == 0) return; +#ifdef USE_LIBURING + if (mystreamer->dio.fd != -1) + { + dio_writer_write(&mystreamer->dio, data, len); + return; + } +#endif write_file_range(fileno(mystreamer->file), mystreamer->pathname, data, len); } @@ -134,6 +216,13 @@ astreamer_plain_writer_finalize(astreamer *streamer) mystreamer = (astreamer_plain_writer *) streamer; +#ifdef USE_LIBURING + if (mystreamer->dio.fd != -1) + { + dio_writer_finish(&mystreamer->dio); + return; + } +#endif if (mystreamer->should_close_file && fclose(mystreamer->file) != 0) pg_fatal("could not close file \"%s\": %m", mystreamer->pathname); @@ -156,6 +245,9 @@ astreamer_plain_writer_free(astreamer *streamer) Assert(mystreamer->base.bbs_next == NULL); pfree(mystreamer->pathname); +#ifdef USE_LIBURING + dio_writer_destroy(&mystreamer->dio); +#endif pfree(mystreamer); } @@ -186,7 +278,7 @@ astreamer * astreamer_extractor_new(const char *basepath, const char *(*link_map) (const char *), void (*report_output_file) (const char *), - bool discard_backup) + bool discard_backup, bool verbose) { astreamer_extractor *streamer; @@ -197,7 +289,11 @@ astreamer_extractor_new(const char *basepath, streamer->link_map = link_map; streamer->report_output_file = report_output_file; streamer->discard_backup = discard_backup; + streamer->verbose = verbose; streamer->fd = -1; +#ifdef USE_LIBURING + streamer->dio.fd = -1; +#endif return &streamer->base; } @@ -220,6 +316,9 @@ astreamer_extractor_content(astreamer *streamer, astreamer_member *member, { case ASTREAMER_MEMBER_HEADER: Assert(mystreamer->fd == -1); +#ifdef USE_LIBURING + Assert(mystreamer->dio.fd == -1); +#endif if (!path_is_safe_for_extraction(member->pathname)) pg_fatal("tar member has unsafe path name: \"%s\"", @@ -238,11 +337,31 @@ astreamer_extractor_content(astreamer *streamer, astreamer_member *member, * Dispatch based on file type. */ if (member->is_regular) - mystreamer->fd = - create_file_for_extract(mystreamer->filename, - member->mode, - mystreamer->discard_backup, - member->size); + { +#ifdef USE_LIBURING + /* + * Fallback to classic buffered writes in case of: + * - small files + * - env variable PG_BASEBACKUP_NODIO is set (debugging) + * - pipe/stdout is used + * - /dev/null is being used (-t client-blackhole) + * - filesystems without O_DIRECT support + */ + if (member->size >= DIO_MIN_SIZE && + !mystreamer->discard_backup && + getenv("PG_BASEBACKUP_NODIO") == NULL && + dio_writer_start(&mystreamer->dio, mystreamer->filename, + member->size, mystreamer->verbose)) { + /* do not use buffered writes, as the dio.fd is going to be used */ + mystreamer->fd = -1; + } else +#endif + mystreamer->fd = + create_file_for_extract(mystreamer->filename, + member->mode, + mystreamer->discard_backup, + member->size); + } else if (member->is_directory) { if (!mystreamer->discard_backup) @@ -272,6 +391,14 @@ astreamer_extractor_content(astreamer *streamer, astreamer_member *member, break; case ASTREAMER_MEMBER_CONTENTS: +#ifdef USE_LIBURING + if (mystreamer->dio.fd != -1) + { + if (len > 0) + dio_writer_write(&mystreamer->dio, data, len); + break; + } +#endif if (mystreamer->fd == -1) break; @@ -281,6 +408,13 @@ astreamer_extractor_content(astreamer *streamer, astreamer_member *member, break; case ASTREAMER_MEMBER_TRAILER: +#ifdef USE_LIBURING + if (mystreamer->dio.fd != -1) + { + dio_writer_finish(&mystreamer->dio); + break; + } +#endif if (mystreamer->fd == -1) break; if (close(mystreamer->fd) != 0) @@ -463,6 +597,9 @@ astreamer_extractor_finalize(astreamer *streamer) = (astreamer_extractor *) streamer; Assert(mystreamer->fd == -1); +#ifdef USE_LIBURING + Assert(mystreamer->dio.fd == -1); +#endif } /* @@ -474,5 +611,307 @@ astreamer_extractor_free(astreamer *streamer) astreamer_extractor *mystreamer = (astreamer_extractor *) streamer; pfree(mystreamer->basepath); +#ifdef USE_LIBURING + dio_writer_destroy(&mystreamer->dio); +#endif pfree(mystreamer); } + +#ifdef USE_LIBURING +/* + * IO_uring/liburing uses concept of two rings (submissions and completion). + * See io_uring_queue_init(3) and io_uring(7) for more information and + * especially https://github.com/axboe/liburing/blob/master/examples/io_uring-cp.c + * for nice example on how to use it. + */ + +/* Take (consume) one completion event from the ring */ +static void +dio_wait_one(dio_writer *dw) +{ + int idx; + int ret; + struct io_uring_cqe *cqe; + + ret = io_uring_wait_cqe(&dw->ring, &cqe); + if (ret < 0) + { + errno = -ret; + pg_fatal("could not wait for io_uring_wait_cqe(): %m"); + } + + idx = (int) io_uring_cqe_get_data64(cqe); + if (cqe->res < 0) + { + errno = -cqe->res; + pg_fatal("could not write to file \"%s\": %m", dw->filename); + } + + if ((size_t) cqe->res != dw->buflen[idx]) + { + /* short write? */ + errno = ENOSPC; + pg_fatal("could not write to file \"%s\": %m", dw->filename); + } + + io_uring_cqe_seen(&dw->ring, cqe); + dw->busy[idx] = false; + dw->inflight--; +} + +/* + * Common routine for flushing (submitting) earlier prepared, but not yet + * submitted SQEs. + */ +static void +dio_flush_sq(dio_writer *dw) +{ + int ret; + + if (dw->unsubmitted == 0) + return; + + ret = io_uring_submit(&dw->ring); + if (ret < 0) + { + errno = -ret; + pg_fatal("could not submit io_uring (io_uring_enter(2) failure?): %m"); + } + + /* Everything was submitted */ + dw->unsubmitted = 0; +} + +/* Get free buffer index. If none are available, wait for some to finish */ +static int +dio_get_free_buf(dio_writer *dw) +{ + for (;;) + { + for (int i = 0; i < DIO_NBUF; i++) + { + if (dw->busy[i] == false) + { + /* + * Mark it busy till we finish the write (technically we need CQE for + * this buffer to arrive, and then can mark it as non-busy again). + * */ + dw->busy[i] = true; + return i; + } + } + /* All buffers are busy, so we wait for writes to finish */ + dio_flush_sq(dw); + dio_wait_one(dw); + } +} + +/* + * Prepare the SQE from the buffer with full O_DIRECT write. We really + * submit the SQEs to the kernel only (flush them) only once every + * couple of times to avoid constant syscall tax (AKA io_uring batching). + * + * buffer's idx needs to have it's lenth aligned to DIO_ALIGN (O_DIRECT + * requirement). + */ +static void +dio_submit(dio_writer *dw, int idx, size_t len) +{ + struct io_uring_sqe *sqe; + + sqe = io_uring_get_sqe(&dw->ring); + /* this should never happen? liburing/examples uses abort/asserts for this */ + if (sqe == NULL) + pg_fatal("io_uring submission queue full: %m"); + + io_uring_prep_write(sqe, dw->fd, + dw->pool + (size_t) idx * DIO_BUFSZ, + len, dw->offset); + io_uring_sqe_set_data64(sqe, idx); + + dw->buflen[idx] = len; + dw->offset += len; + dw->inflight++; + dw->unsubmitted++; + + if (dw->unsubmitted >= DIO_SUBMIT_BATCH) + dio_flush_sq(dw); +} + +/* + * Start writing to a file through the DIO+AIO path. Returns true if that is + * supported (dw->fd is set). + * + * If prealloc_size is known it can be used to preallocate file which helps + * greatly to avoid slow writes with O_DIRECT on some filesystems. + */ +static bool +dio_writer_start(dio_writer *dw, const char *filename, pgoff_t prealloc_size, bool verbose) +{ + int fd; + int ret; + struct stat st; + + fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC | O_DIRECT | PG_BINARY, + pg_file_create_mode); + if (fd < 0) + { + if (verbose && !dw->notified) + { + pg_log_info("could not open \"%s\" with O_DIRECT, using buffered I/O instead: %m", + filename); + dw->notified = true; + } + return false; + } + + /* + * Only regular files can be written with O_DIRECT. However, the /dev/null + * device (used with -t client-blackhole mode) and other special files can + * be still opened with O_DIRECT, but must go through the buffered path(?). + * Ensure we opened regular file before setting up the io_uring, so we + * don't allocate the buffer pool just to fall back. + */ + if (fstat(fd, &st) != 0 || !S_ISREG(st.st_mode)) + { + if (verbose && !dw->notified) + { + pg_log_info("\"%s\" is not a regular file, using buffered I/O instead", + filename); + dw->notified = true; + } + close(fd); + return false; + } + + /* Lazy setup of the io_uring */ + if (!dw->ring_ready) + { + ret = io_uring_queue_init(DIO_NBUF, &dw->ring, 0); + if (ret != 0) + { + errno = -ret; + /* XXX: shouldn't we warn about it non-verbose mode? */ + if (verbose && !dw->notified) + { + pg_log_info("could not set up io_uring, using buffered I/O instead: %m"); + dw->notified = true; + } + close(fd); + return false; + } + + /* O_DIRECT writes require memory aligment */ + if (posix_memalign((void **) &dw->pool, DIO_ALIGN, + (size_t) DIO_NBUF * DIO_BUFSZ) != 0) + pg_fatal("unable to allocate aligned memory for direct I/O writes"); + + dw->ring_ready = true; + if (verbose) + pg_log_info("using O_DIRECT with io_uring for large file writes"); + } + + /* If size is known (in plain mode), preallocate the space */ +#ifdef HAVE_POSIX_FALLOCATE + if (prealloc_size > 0) + { + int rc = posix_fallocate(fd, 0, prealloc_size); + + if (rc == ENOSPC) + { + errno = rc; + pg_fatal("could not preallocate file \"%s\": %m", filename); + } + } +#endif + + dw->fd = fd; + dw->filename = filename; + dw->offset = 0; + dw->written = 0; + dw->curidx = -1; + dw->curlen = 0; + return true; +} + +/* + * Handle buffering on DIO side (in dio->pool buffers) before submiting + * full (big) writes as required by O_DIRECT. + */ +static void +dio_writer_write(dio_writer *dw, const char *data, int len) +{ + dw->written += len; + + while (len > 0) + { + char *buf; + int space; + int n; + + if (dw->curidx < 0) + { + dw->curidx = dio_get_free_buf(dw); + dw->curlen = 0; + } + + buf = dw->pool + (size_t) dw->curidx * DIO_BUFSZ; + space = DIO_BUFSZ - dw->curlen; + n = Min(space, len); + /* append to the pool buffer */ + memcpy(buf + dw->curlen, data, n); + dw->curlen += n; + data += n; + len -= n; + + if (dw->curlen == DIO_BUFSZ) + { + dio_submit(dw, dw->curidx, DIO_BUFSZ); + dw->curidx = -1; + } + } +} + +/* Send/wait for remaining in-flight writes, truncate and close the DIO */ +static void +dio_writer_finish(dio_writer *dw) +{ + /* Flush partial buffer */ + if (dw->curidx >= 0 && dw->curlen > 0) + { + char *buf = dw->pool + (size_t) dw->curidx * DIO_BUFSZ; + /* We need to pad stuff */ + size_t padded = TYPEALIGN(DIO_ALIGN, dw->curlen); + + memset(buf + dw->curlen, 0, padded - dw->curlen); + dio_submit(dw, dw->curidx, padded); + dw->curidx = -1; + } + + /* Submit any batched writes and wait for them all to finish */ + dio_flush_sq(dw); + while (dw->inflight > 0) + dio_wait_one(dw); + + /* Truncate the file to the right size (we could pre-allocate too much) */ + if (ftruncate(dw->fd, dw->written) != 0) + pg_fatal("could not truncate file \"%s\": %m", dw->filename); + + if (close(dw->fd) != 0) + pg_fatal("could not close file \"%s\": %m", dw->filename); + dw->fd = -1; +} + +/* Free the buffers and the io_uring */ +static void +dio_writer_destroy(dio_writer *dw) +{ + if (dw->ring_ready) + { + io_uring_queue_exit(&dw->ring); + free(dw->pool); + dw->ring_ready = false; + dw->fd = -1; + } +} +#endif /* USE_LIBURING */ diff --git a/src/fe_utils/meson.build b/src/fe_utils/meson.build index 86befca192e..bf508da183c 100644 --- a/src/fe_utils/meson.build +++ b/src/fe_utils/meson.build @@ -34,7 +34,7 @@ fe_utils = static_library('libpgfeutils', c_pch: pch_postgres_fe_h, include_directories: [postgres_inc, libpq_inc], c_args: host_system == 'windows' ? ['-DFD_SETSIZE=1024'] : [], - dependencies: frontend_common_code, + dependencies: [frontend_common_code, liburing], kwargs: default_lib_args + { 'install': install_internal_static_lib, }, diff --git a/src/include/fe_utils/astreamer.h b/src/include/fe_utils/astreamer.h index 7206dc0c48b..6eae8ebbb6d 100644 --- a/src/include/fe_utils/astreamer.h +++ b/src/include/fe_utils/astreamer.h @@ -210,13 +210,14 @@ astreamer_buffer_until(astreamer *streamer, const char **data, int *len, * Functions for creating astreamer objects of various types. See the header * comments for each of these functions for details. */ -extern astreamer *astreamer_plain_writer_new(char *pathname, FILE *file); +extern astreamer *astreamer_plain_writer_new(char *pathname, FILE *file, + bool verbose); extern astreamer *astreamer_gzip_writer_new(char *pathname, FILE *file, pg_compress_specification *compress); extern astreamer *astreamer_extractor_new(const char *basepath, const char *(*link_map) (const char *), void (*report_output_file) (const char *), - bool discard_backup); + bool discard_backup, bool verbose); extern astreamer *astreamer_gzip_decompressor_new(astreamer *next); extern astreamer *astreamer_lz4_compressor_new(astreamer *next, -- 2.43.0