From 74c1eb7ae5307c7959704935d20262c26222c6f7 Mon Sep 17 00:00:00 2001 From: Nazir Bilal Yavuz Date: Wed, 16 Sep 2026 13:30:15 +0300 Subject: [PATCH v2 6/8] aio: Issue checkpointer fsyncs asynchronously ProcessSyncRequests() previously fsynced pending files one at a time. For checkpoints with many files, this serialized I/O that storage could perform concurrently. Submit the fsyncs through AIO, keeping a bounded set in flight and reaping completions in submission order. Relation requests may use all io_max_concurrency slots. For SLRU requests, apply a transient-descriptor limit to the total in-flight count so their open files do not exhaust the descriptor reserve. Reshape the sync handler API so handlers open the file, assign an AIO target, start the fsync, and record how to close it. sync.c manages the in-flight operations, errors, retries, and pendingOps bookkeeping. Absorbing requests while fsyncs are in flight requires some care: - Recheck cancellation when an operation is reaped because a request can be canceled after its fsync starts. - Defer completion bookkeeping until the pendingOps scan ends because dynahash permits removing only the entry most recently returned. - Keep a new request for a file already in flight until the next checkpoint cycle because the running fsync might not cover its write. Error cleanup drains and closes all outstanding operations while preserving the first durability error that requires PANIC. Otherwise, preserve the original error and leave retries to the next checkpoint. Do not reinterpret an already-handled deletion failure as a failed retry during cleanup. Relation files can be reopened through the smgr target, allowing I/O workers to perform their fsyncs. Open the exact segment path so inactive segments following a truncated partial segment remain syncable. Add a target close callback to release transient worker-side descriptors after execution. SLRU files use the generic sync target and execute synchronously with the worker I/O method, although io_uring can still overlap them. Measure time spent starting each request and waiting for operations that are not already complete. Include only successful attempts in checkpoint longest and aggregate statistics. Add a pgstat helper that accepts an end timestamp. Discussion: https://postgr.es/m/CAN55FZ0vLWJQNB%3DHuHXG2wabFjXJd6OWTa3%3DkRzwObdZD9poHQ%40mail.gmail.com --- src/backend/access/transam/clog.c | 6 +- src/backend/access/transam/commit_ts.c | 6 +- src/backend/access/transam/multixact.c | 12 +- src/backend/access/transam/slru.c | 35 +- src/backend/storage/aio/aio_io.c | 8 + src/backend/storage/aio/aio_target.c | 39 +- src/backend/storage/file/fd.c | 39 +- src/backend/storage/smgr/md.c | 84 +++- src/backend/storage/smgr/smgr.c | 46 +- src/backend/storage/sync/sync.c | 632 ++++++++++++++++++++----- src/backend/utils/activity/pgstat_io.c | 47 +- src/include/access/clog.h | 2 +- src/include/access/commit_ts.h | 2 +- src/include/access/multixact.h | 4 +- src/include/access/slru.h | 2 +- src/include/pgstat.h | 4 + src/include/storage/aio.h | 6 + src/include/storage/aio_internal.h | 1 + src/include/storage/fd.h | 3 +- src/include/storage/md.h | 3 +- src/include/storage/sync.h | 55 +++ src/test/modules/test_slru/test_slru.c | 45 +- src/tools/pgindent/typedefs.list | 3 + 23 files changed, 893 insertions(+), 191 deletions(-) diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c index 6f7f6b86eb6..89fb77ea5da 100644 --- a/src/backend/access/transam/clog.c +++ b/src/backend/access/transam/clog.c @@ -1117,8 +1117,8 @@ clog_redo(XLogReaderState *record) /* * Entrypoint for sync.c to sync clog files. */ -int -clogsyncfiletag(const FileTag *ftag, char *path) +void +clogsyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) { - return SlruSyncFileTag(XactCtl, ftag, path); + SlruSyncFileTag(XactCtl, ioh, entry); } diff --git a/src/backend/access/transam/commit_ts.c b/src/backend/access/transam/commit_ts.c index 9e6fd5d4657..7cbbad383b2 100644 --- a/src/backend/access/transam/commit_ts.c +++ b/src/backend/access/transam/commit_ts.c @@ -1028,8 +1028,8 @@ commit_ts_redo(XLogReaderState *record) /* * Entrypoint for sync.c to sync commit_ts files. */ -int -committssyncfiletag(const FileTag *ftag, char *path) +void +committssyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) { - return SlruSyncFileTag(CommitTsCtl, ftag, path); + SlruSyncFileTag(CommitTsCtl, ioh, entry); } diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c index 70a4ea69486..deb0b110ee8 100644 --- a/src/backend/access/transam/multixact.c +++ b/src/backend/access/transam/multixact.c @@ -3003,17 +3003,17 @@ multixact_redo(XLogReaderState *record) /* * Entrypoint for sync.c to sync offsets files. */ -int -multixactoffsetssyncfiletag(const FileTag *ftag, char *path) +void +multixactoffsetssyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) { - return SlruSyncFileTag(MultiXactOffsetCtl, ftag, path); + SlruSyncFileTag(MultiXactOffsetCtl, ioh, entry); } /* * Entrypoint for sync.c to sync members files. */ -int -multixactmemberssyncfiletag(const FileTag *ftag, char *path) +void +multixactmemberssyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) { - return SlruSyncFileTag(MultiXactMemberCtl, ftag, path); + SlruSyncFileTag(MultiXactMemberCtl, ioh, entry); } diff --git a/src/backend/access/transam/slru.c b/src/backend/access/transam/slru.c index 885fd068535..b1e513ac3b6 100644 --- a/src/backend/access/transam/slru.c +++ b/src/backend/access/transam/slru.c @@ -68,6 +68,7 @@ #include "access/xlogutils.h" #include "miscadmin.h" #include "pgstat.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/shmem.h" #include "storage/shmem_internal.h" @@ -1880,26 +1881,32 @@ SlruScanDirectory(SlruDesc *ctl, SlruScanCallback callback, void *data) * build the path), but they just forward to this common implementation that * performs the fsync. */ -int -SlruSyncFileTag(SlruDesc *ctl, const FileTag *ftag, char *path) +void +SlruSyncFileTag(SlruDesc *ctl, struct PgAioHandle *ioh, InflightSyncEntry *entry) { int fd; - int save_errno; - int result; - SlruFileName(ctl, path, ftag->segno); + SlruFileName(ctl, entry->path, entry->tag.segno); - fd = OpenTransientFile(path, O_RDWR | PG_BINARY); + fd = OpenTransientFile(entry->path, O_RDWR | PG_BINARY); if (fd < 0) - return -1; + { + entry->started = false; + entry->open_errno = errno; + return; + } - pgstat_report_wait_start(WAIT_EVENT_SLRU_FLUSH_SYNC); - result = pg_fsync(fd); - pgstat_report_wait_end(); - save_errno = errno; + /* + * Use the generic sync target. SLRU segments are not smgr relations and + * cannot be reopened from a FileTag in another process, so this fsync + * will run synchronously in worker mode. + */ + pgaio_io_set_target(ioh, PGAIO_TID_SYNC); - CloseTransientFile(fd); + /* Start the asynchronous fsync; the fd is closed once it completes. */ + pgaio_io_start_fsync(ioh, fd, false, WAIT_EVENT_SLRU_FLUSH_SYNC); - errno = save_errno; - return result; + entry->started = true; + entry->close_method = SYNC_CLOSE_TRANSIENT; + entry->close_file = fd; } diff --git a/src/backend/storage/aio/aio_io.c b/src/backend/storage/aio/aio_io.c index 0525643cc2b..a5f5c286fb0 100644 --- a/src/backend/storage/aio/aio_io.c +++ b/src/backend/storage/aio/aio_io.c @@ -173,6 +173,14 @@ pgaio_io_perform_synchronously(PgAioHandle *ioh) Assert(result <= INT_MAX); ioh->result = result < 0 ? -errno : result; + /* + * If we, rather than the process that staged the IO, opened the file, + * close it again. Has to happen after the result has been determined, as + * closing may clobber errno, and before the completion is processed, as + * that can recycle the handle. + */ + pgaio_io_close_reopened(ioh); + pgaio_io_process_completion(ioh, ioh->result); END_CRIT_SECTION(); diff --git a/src/backend/storage/aio/aio_target.c b/src/backend/storage/aio/aio_target.c index 1a0088fa382..534382b423b 100644 --- a/src/backend/storage/aio/aio_target.c +++ b/src/backend/storage/aio/aio_target.c @@ -41,6 +41,13 @@ static const PgAioTargetInfo *pgaio_target_info[] = { [PGAIO_TID_SYNC] = &aio_sync_target_info, }; +/* + * The IO whose descriptor this process reopened and whose target requires the + * descriptor to be released after execution. Only set between + * pgaio_io_reopen() and pgaio_io_close_reopened(). + */ +static PgAioHandle *pgaio_reopened_ioh = NULL; + /* * describe_identity callback for PGAIO_TID_SYNC. As we do not store the path @@ -143,8 +150,38 @@ pgaio_io_reopen(PgAioHandle *ioh) Assert(ioh->target > PGAIO_TID_INVALID && ioh->target < PGAIO_TID_COUNT); Assert(ioh->op > PGAIO_OP_INVALID && ioh->op < PGAIO_OP_COUNT); + Assert(pgaio_reopened_ioh == NULL); result = pgaio_target_info[ioh->target]->reopen(ioh); - return result; + if (result < 0) + return result; + + /* + * Remember that this process, rather than the one that staged the IO, + * owns the file descriptor now, so that pgaio_io_close_reopened() can + * release it once the IO has been executed. + */ + if (pgaio_target_info[ioh->target]->close != NULL) + pgaio_reopened_ioh = ioh; + + return 0; +} + +/* + * Internal: Counterpart to pgaio_io_reopen(), releasing the file descriptor it + * acquired. Does nothing unless this process reopened this very IO and its + * target needs the descriptor to be released. + * + * This has to be called before the IO's completion is processed, as that can + * make the handle be reused for an unrelated IO. + */ +void +pgaio_io_close_reopened(PgAioHandle *ioh) +{ + if (pgaio_reopened_ioh != ioh) + return; + + pgaio_reopened_ioh = NULL; + pgaio_target_info[ioh->target]->close(ioh); } diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c index 245ce6eccdd..27184557b8f 100644 --- a/src/backend/storage/file/fd.c +++ b/src/backend/storage/file/fd.c @@ -2258,6 +2258,29 @@ FileStartReadV(PgAioHandle *ioh, File file, return 0; } +int +FileStartSync(PgAioHandle *ioh, File file, bool datasync, + uint32 wait_event_info) +{ + int returnCode; + Vfd *vfdP; + + Assert(FileIsValid(file)); + + DO_DB(elog(LOG, "FileStartSync: %d (%s)", + file, VfdCache[file].fileName)); + + returnCode = FileAccess(file); + if (returnCode < 0) + return returnCode; + + vfdP = &VfdCache[file]; + + pgaio_io_start_fsync(ioh, vfdP->fd, datasync, wait_event_info); + + return 0; +} + ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, pgoff_t offset, uint32 wait_event_info) @@ -3602,14 +3625,18 @@ do_syncfs(const char *path) * * Callers that fsync files opened with OpenTransientFile() hold one * AllocateDesc for each in-flight IO. At most max_safe_fds / 3 of those can - * be allocated at a time (see reserveAllocatedDesc()), and the callers of - * interest also traverse directories (see walkdir()), which needs - * AllocateDescs of its own. So, hand out at most half of the budget. + * be allocated at a time (see reserveAllocatedDesc()). Keep half of that + * budget available for directory traversal and other AllocateDesc users. + * + * Other callers only need to respect the AIO handle limit. */ int -GetFsyncConcurrencyLimit(void) +GetFsyncConcurrencyLimit(bool uses_transient_fd) { - return Max(1, Min(io_max_concurrency, max_safe_fds / 6)); + if (uses_transient_fd) + return Max(1, Min(io_max_concurrency, max_safe_fds / 6)); + + return io_max_concurrency; } /* @@ -3720,7 +3747,7 @@ SyncDataDirectory(void) begin_startup_progress_phase(); sync_state_data.elevel = LOG; - sync_state_data.max_inflight = GetFsyncConcurrencyLimit(); + sync_state_data.max_inflight = GetFsyncConcurrencyLimit(true); sync_state_data.head = 0; sync_state_data.count = 0; sync_state_data.entries = palloc0(sizeof(DataDirSyncEntry) * diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index 780c88c0630..053f121ef8a 100644 --- a/src/backend/storage/smgr/md.c +++ b/src/backend/storage/smgr/md.c @@ -1505,6 +1505,22 @@ mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) return FileGetRawDesc(v->mdfd_vfd); } +/* + * Open the exact segment containing blocknum for fsync, or return -1 with + * errno set. + */ +int +mdfsyncfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum) +{ + MdPathStr path; + BlockNumber segno; + + segno = blocknum / ((BlockNumber) RELSEG_SIZE); + path = _mdfd_segpath(reln, forknum, segno); + + return OpenTransientFile(path.str, _mdfd_open_flags()); +} + /* * register_dirty_segment() -- Mark a relation segment as needing fsync * @@ -1896,26 +1912,27 @@ _mdnblocks(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg) } /* - * Sync a file to disk, given a file tag. Write the path into an output - * buffer so the caller can use it in error messages. + * Sync a file to disk, given a file tag. * - * Return 0 on success, -1 on failure, with errno set. + * Starts an asynchronous fsync on the given AIO handle and records in "entry" + * the path (for error messages), whether an IO was started, and how the file + * is to be closed once the fsync has completed. */ -int -mdsyncfiletag(const FileTag *ftag, char *path) +void +mdsyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry) { + FileTag *ftag = &entry->tag; SMgrRelation reln = smgropen(ftag->rlocator, INVALID_PROC_NUMBER); + BlockNumber segfirstblock = ftag->segno * ((BlockNumber) RELSEG_SIZE); File file; - instr_time io_start; bool need_to_close; - int result, - save_errno; + instr_time io_start; /* See if we already have the file open, or need to open it. */ if (ftag->segno < reln->md_num_open_segs[ftag->forknum]) { file = reln->md_seg_fds[ftag->forknum][ftag->segno].mdfd_vfd; - strlcpy(path, FilePathName(file), MAXPGPATH); + strlcpy(entry->path, FilePathName(file), MAXPGPATH); need_to_close = false; } else @@ -1923,28 +1940,53 @@ mdsyncfiletag(const FileTag *ftag, char *path) MdPathStr p; p = _mdfd_segpath(reln, ftag->forknum, ftag->segno); - strlcpy(path, p.str, MD_PATH_STR_MAXLEN); + strlcpy(entry->path, p.str, MD_PATH_STR_MAXLEN); - file = PathNameOpenFile(path, _mdfd_open_flags()); + file = PathNameOpenFile(entry->path, _mdfd_open_flags()); if (file < 0) - return -1; + { + entry->started = false; + entry->open_errno = errno; + return; + } need_to_close = true; } + pgaio_io_set_target_smgr(ioh, reln, ftag->forknum, segfirstblock, + 0, false); + + /* + * As with asynchronous reads, measure the time spent starting the IO. + * Synchronous execution includes the fsync itself; otherwise this only + * measures submission. + */ io_start = pgstat_prepare_io_time(track_io_timing); - /* Sync the file. */ - result = FileSync(file, WAIT_EVENT_DATA_FILE_SYNC); - save_errno = errno; + if (FileStartSync(ioh, file, false, WAIT_EVENT_DATA_FILE_SYNC) < 0) + { + entry->started = false; + entry->open_errno = errno; + if (need_to_close) + FileClose(file); + return; + } - if (need_to_close) - FileClose(file); + pgstat_count_io_op_time(IOOBJECT_RELATION, IOCONTEXT_NORMAL, IOOP_FSYNC, + io_start, 1, 0); - pgstat_count_io_op_time(IOOBJECT_RELATION, IOCONTEXT_NORMAL, - IOOP_FSYNC, io_start, 1, 0); + entry->started = true; - errno = save_errno; - return result; + /* + * If we opened the segment ourselves it has to be closed once the fsync + * has completed; segments owned by smgr are left to smgr to manage. + */ + if (need_to_close) + { + entry->close_method = SYNC_CLOSE_VFD; + entry->close_file = (int) file; + } + else + entry->close_method = SYNC_CLOSE_NONE; } /* diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c index 83417e1bcc2..325386ec58e 100644 --- a/src/backend/storage/smgr/smgr.c +++ b/src/backend/storage/smgr/smgr.c @@ -68,6 +68,7 @@ #include "miscadmin.h" #include "storage/aio.h" #include "storage/bufmgr.h" +#include "storage/fd.h" #include "storage/ipc.h" #include "storage/md.h" #include "storage/smgr.h" @@ -123,6 +124,7 @@ typedef struct f_smgr void (*smgr_immedsync) (SMgrRelation reln, ForkNumber forknum); void (*smgr_registersync) (SMgrRelation reln, ForkNumber forknum); int (*smgr_fd) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); + int (*smgr_fsync_fd) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum); } f_smgr; static const f_smgr smgrsw[] = { @@ -148,6 +150,7 @@ static const f_smgr smgrsw[] = { .smgr_immedsync = mdimmedsync, .smgr_registersync = mdregistersync, .smgr_fd = mdfd, + .smgr_fsync_fd = mdfsyncfd, } }; @@ -166,12 +169,14 @@ static void smgrshutdown(int code, Datum arg); static void smgrdestroy(SMgrRelation reln); static int smgr_aio_reopen(PgAioHandle *ioh); +static void smgr_aio_close(PgAioHandle *ioh); static char *smgr_aio_describe_identity(const PgAioTargetData *sd); const PgAioTargetInfo aio_smgr_target_info = { .name = "smgr", .reopen = smgr_aio_reopen, + .close = smgr_aio_close, .describe_identity = smgr_aio_describe_identity, }; @@ -1001,6 +1006,18 @@ smgrfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) return fd; } +/* + * Open the exact segment containing blocknum for fsync, returning a transient + * descriptor, or -1 with errno set if the file cannot be opened. + */ +static int +smgrfsyncfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum) +{ + Assert(!INTERRUPTS_CAN_BE_PROCESSED()); + + return smgrsw[reln->smgr_which].smgr_fsync_fd(reln, forknum, blocknum); +} + /* * AtEOXact_SMgr * @@ -1098,7 +1115,10 @@ smgr_aio_reopen(PgAioHandle *ioh) Assert(off == od->write.offset); return 0; case PGAIO_OP_FSYNC: - fd = smgrfd(reln, sd->smgr.forkNum, sd->smgr.blockNum, &off); + fd = smgrfsyncfd(reln, sd->smgr.forkNum, sd->smgr.blockNum); + /* We need errno for only checkpointer fsyncs for now */ + if (fd < 0) + return -errno; od->fsync.fd = fd; return 0; } @@ -1106,6 +1126,21 @@ smgr_aio_reopen(PgAioHandle *ioh) pg_unreachable(); } +/* + * Release a transient descriptor opened for a worker-executed fsync. + */ +static void +smgr_aio_close(PgAioHandle *ioh) +{ + PgAioOpData *od = pgaio_io_get_op_data(ioh); + + if (pgaio_io_get_op(ioh) == PGAIO_OP_FSYNC) + { + (void) CloseTransientFile(od->fsync.fd); + od->fsync.fd = -1; + } +} + /* * Callback for the smgr AIO target, describing the target of the IO. */ @@ -1121,7 +1156,14 @@ smgr_aio_describe_identity(const PgAioTargetData *sd) sd->smgr.forkNum); if (sd->smgr.nblocks == 0) - desc = psprintf(_("file \"%s\""), path.str); + { + BlockNumber segno = sd->smgr.blockNum / ((BlockNumber) RELSEG_SIZE); + + if (segno > 0) + desc = psprintf(_("file \"%s.%u\""), path.str, segno); + else + desc = psprintf(_("file \"%s\""), path.str); + } else if (sd->smgr.nblocks == 1) desc = psprintf(_("block %u in file \"%s\""), sd->smgr.blockNum, diff --git a/src/backend/storage/sync/sync.c b/src/backend/storage/sync/sync.c index 2c964b6f3d9..2442492d6a9 100644 --- a/src/backend/storage/sync/sync.c +++ b/src/backend/storage/sync/sync.c @@ -26,7 +26,10 @@ #include "pgstat.h" #include "portability/instr_time.h" #include "postmaster/bgwriter.h" +#include "storage/aio.h" +#include "storage/bufmgr.h" #include "storage/fd.h" +#include "storage/ipc.h" #include "storage/latch.h" #include "storage/md.h" #include "utils/hsearch.h" @@ -54,11 +57,22 @@ */ typedef uint16 CycleCtr; /* can be any convenient integer size */ -typedef struct +typedef struct PendingFsyncEntry { FileTag tag; /* identifies handler and file */ CycleCtr cycle_ctr; /* sync_cycle_ctr of oldest request */ bool canceled; /* canceled is true if we canceled "recently" */ + + /* + * Set when a request arrives for a tag that already has an entry, and + * cleared whenever an fsync for it is started. If it is set once that + * fsync completes, the request came in while the fsync was in flight, so + * the fsync cannot be assumed to have covered it. + */ + bool re_requested; + + /* fsync is done, pending hash-table bookkeeping */ + bool sync_completed; } PendingFsyncEntry; typedef struct @@ -68,10 +82,43 @@ typedef struct bool canceled; /* true if request has been canceled */ } PendingUnlinkEntry; +/* + * Transient state used while processing a batch of fsync requests. A single + * SyncState instance lives on the stack of ProcessSyncRequests() so that no + * partial state survives across calls. + */ +typedef struct SyncState +{ + dlist_head inflight; /* InflightSyncEntry being fsync'd right now */ + dlist_head retry; /* InflightSyncEntry to be retried */ + int inflight_count; /* number of entries in "inflight" */ + int absorb_counter; + + /* stats */ + int processed; + instr_time longest; + instr_time total_elapsed; +} SyncState; + static HTAB *pendingOps = NULL; static List *pendingUnlinks = NIL; static MemoryContext pendingOpsCxt; /* context for the above */ +/* + * Context for the InflightSyncEntry structs allocated while a batch of fsync + * requests is being processed. It is kept separate from pendingOpsCxt (which + * must survive for the lifetime of the process, as it holds pendingOps + * itself), so that it can be reset between batches. + */ +static MemoryContext inflightSyncCxt; + +/* + * All InflightSyncEntry structs that have not yet been freed. Unlike the + * lists in SyncState, this survives an error so that handler-owned files can + * be closed before their entries are discarded. + */ +static dlist_head activeSyncEntries = DLIST_STATIC_INIT(activeSyncEntries); + static CycleCtr sync_cycle_ctr = 0; static CycleCtr checkpoint_cycle_ctr = 0; @@ -84,10 +131,11 @@ static CycleCtr checkpoint_cycle_ctr = 0; */ typedef struct SyncOps { - int (*sync_syncfiletag) (const FileTag *ftag, char *path); + void (*sync_syncfiletag) (PgAioHandle *ioh, InflightSyncEntry *entry); int (*sync_unlinkfiletag) (const FileTag *ftag, char *path); bool (*sync_filetagmatches) (const FileTag *ftag, const FileTag *candidate); + bool uses_transient_fd; } SyncOps; /* @@ -98,23 +146,28 @@ static const SyncOps syncsw[] = { [SYNC_HANDLER_MD] = { .sync_syncfiletag = mdsyncfiletag, .sync_unlinkfiletag = mdunlinkfiletag, - .sync_filetagmatches = mdfiletagmatches + .sync_filetagmatches = mdfiletagmatches, + .uses_transient_fd = false }, /* pg_xact */ [SYNC_HANDLER_CLOG] = { - .sync_syncfiletag = clogsyncfiletag + .sync_syncfiletag = clogsyncfiletag, + .uses_transient_fd = true }, /* pg_commit_ts */ [SYNC_HANDLER_COMMIT_TS] = { - .sync_syncfiletag = committssyncfiletag + .sync_syncfiletag = committssyncfiletag, + .uses_transient_fd = true }, /* pg_multixact/offsets */ [SYNC_HANDLER_MULTIXACT_OFFSET] = { - .sync_syncfiletag = multixactoffsetssyncfiletag + .sync_syncfiletag = multixactoffsetssyncfiletag, + .uses_transient_fd = true }, /* pg_multixact/members */ [SYNC_HANDLER_MULTIXACT_MEMBER] = { - .sync_syncfiletag = multixactmemberssyncfiletag + .sync_syncfiletag = multixactmemberssyncfiletag, + .uses_transient_fd = true } }; @@ -155,6 +208,10 @@ InitSync(void) &hash_ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); pendingUnlinks = NIL; + + inflightSyncCxt = AllocSetContextCreate(TopMemoryContext, + "Inflight sync context", + ALLOCSET_DEFAULT_SIZES); } } @@ -281,25 +338,373 @@ SyncPostCheckpoint(void) } /* - * ProcessSyncRequests() -- Process queued fsync requests. + * Close the file that a sync handler opened for an in-flight fsync. */ -void -ProcessSyncRequests(void) +static void +sync_close_file(InflightSyncEntry *entry) { - static bool sync_in_progress = false; + switch (entry->close_method) + { + case SYNC_CLOSE_NONE: + break; + case SYNC_CLOSE_TRANSIENT: + CloseTransientFile(entry->close_file); + break; + case SYNC_CLOSE_VFD: + FileClose((File) entry->close_file); + break; + default: + pg_unreachable(); + } + + entry->close_method = SYNC_CLOSE_NONE; +} + +static void +sync_free_entry(InflightSyncEntry *entry) +{ + dlist_delete_from(&activeSyncEntries, &entry->cleanup_node); + pfree(entry); +} + +/* + * Error cleanup callback for ProcessSyncRequests(). + */ +static void +sync_cleanup_inflight(int code, Datum arg) +{ + int save_errno = 0; + char path[MAXPGPATH]; + + /* Do not let an interrupt abandon the remaining IOs during cleanup. */ + HOLD_INTERRUPTS(); + + while (!dlist_is_empty(&activeSyncEntries)) + { + dlist_node *node = dlist_pop_head_node(&activeSyncEntries); + InflightSyncEntry *entry; + int result; + + entry = dlist_container(InflightSyncEntry, cleanup_node, node); + + if (entry->started) + { + pgaio_wref_wait(&entry->iow); + result = -entry->ioret.result.result; + } + else + result = entry->open_errno; + + /* + * These IOs have no error-reporting completion callback. Even though + * the checkpoint has already failed, we must not discard a durability + * error that requires PANIC, including a recorded failure to open the + * file. As in sync_drain_one(), ignore canceled requests and allow a + * first failure that could be due to deletion. Leave retries to the + * next checkpoint. + * + * Save the first such error without allocating memory or raising + * another ERROR, so that all outstanding IOs are cleaned up. + */ + if (result != 0 && save_errno == 0 && + !entry->hash_entry->canceled && + (!FILE_POSSIBLY_DELETED(result) || entry->retry_count > 0) && + data_sync_elevel(ERROR) == PANIC) + { + save_errno = result; + strlcpy(path, entry->path, sizeof(path)); + } + + sync_close_file(entry); + pfree(entry); + } + + RESUME_INTERRUPTS(); + + if (save_errno != 0) + { + errno = save_errno; + ereport(data_sync_elevel(ERROR), + (errcode_for_file_access(), + errmsg("could not fsync file \"%s\": %m", path))); + } +} + +static void +sync_start_one(SyncState *sync_state, InflightSyncEntry *entry) +{ + struct PgAioHandle *ioh; + instr_time io_start; + + entry->started = false; + entry->open_errno = 0; + entry->close_method = SYNC_CLOSE_NONE; + INSTR_TIME_SET_ZERO(entry->io_time); + pgaio_wref_clear(&entry->iow); + + /* + * Any request that arrives from here on may cover data that the fsync + * started below does not, so start out with a clean slate. This has to + * happen before the IO is submitted; requests absorbed in between are + * covered by the fsync, so treating them as newer is merely conservative. + */ + entry->hash_entry->re_requested = false; + + ioh = pgaio_io_acquire(CurrentResourceOwner, &entry->ioret); + pgaio_io_get_wref(ioh, &entry->iow); + + /* + * The handler opens the file, assigns the target and stages the fsync. + * Hold interrupts so that the referenced descriptor cannot be closed + * during submission. + */ + HOLD_INTERRUPTS(); + INSTR_TIME_SET_CURRENT(io_start); + syncsw[entry->tag.handler].sync_syncfiletag(ioh, entry); + INSTR_TIME_SET_CURRENT(entry->io_time); + RESUME_INTERRUPTS(); + INSTR_TIME_SUBTRACT(entry->io_time, io_start); + + if (!entry->started) + pgaio_io_release(ioh); + + Assert(!entry->started || + (entry->close_method == SYNC_CLOSE_TRANSIENT) == + syncsw[entry->tag.handler].uses_transient_fd); + + dlist_push_tail(&sync_state->inflight, &entry->node); + sync_state->inflight_count++; + + Assert(sync_state->inflight_count <= io_max_concurrency); +} + +static void +sync_drain_one(SyncState *sync_state) +{ + dlist_node *node; + InflightSyncEntry *entry; + int result; + instr_time io_time; + + Assert(sync_state->inflight_count > 0); + + node = dlist_pop_head_node(&sync_state->inflight); + entry = dlist_container(InflightSyncEntry, node, node); + sync_state->inflight_count--; + if (entry->started) + { + if (entry->ioret.result.status == PGAIO_RS_UNKNOWN && + !pgaio_wref_check_done(&entry->iow)) + { + instr_time io_start; + instr_time io_end; + bool track_stat_time; + + track_stat_time = entry->tag.handler == SYNC_HANDLER_MD && + track_io_timing; + INSTR_TIME_SET_CURRENT(io_start); + pgaio_wref_wait(&entry->iow); + INSTR_TIME_SET_CURRENT(io_end); + + /* + * The operation itself was counted when it was started. Add only + * time actually spent waiting for relation IO to complete. + */ + if (track_stat_time) + pgstat_count_io_op_time_end(IOOBJECT_RELATION, + IOCONTEXT_NORMAL, IOOP_FSYNC, + io_start, io_end, 0, 0); + + INSTR_TIME_ACCUM_DIFF(entry->io_time, io_end, io_start); + } + + /* + * We did not register a completion callback, so the distilled status + * is always PGAIO_RS_OK and the raw fsync() return value (0 on + * success, -errno on failure) is available in ->result.result. + */ + result = -entry->ioret.result.result; + } + else + result = entry->open_errno; + + sync_close_file(entry); + io_time = entry->io_time; + + if (!result) + { + if (INSTR_TIME_GT(io_time, sync_state->longest)) + sync_state->longest = io_time; + INSTR_TIME_ADD(sync_state->total_elapsed, io_time); + + sync_state->processed++; + + if (log_checkpoints) + elog(DEBUG1, "checkpoint sync: number=%d file=%s time=%.3f ms", + sync_state->processed, + entry->path, + INSTR_TIME_GET_MILLISEC(io_time)); + + entry->hash_entry->sync_completed = true; + sync_free_entry(entry); + } + else + { + /* + * The request may have been canceled after we started the fsync, e.g. + * because the relation was dropped in the meantime and an intervening + * AbsorbSyncRequests() picked up the cancel message. Since + * mdunlink() queues the "cancel" before actually unlinking, a + * cancellation means the failure is expected and the entry can simply + * be dropped. + * + * The upstream, synchronous code checked this at the top of its retry + * loop; because the fsync is now in flight while requests are being + * absorbed, we have to re-check it here. + */ + if (entry->hash_entry->canceled) + { + entry->hash_entry->sync_completed = true; + sync_free_entry(entry); + return; + } + + /* + * It is possible that the relation has been dropped or truncated + * since the fsync request was entered. Therefore, allow ENOENT, but + * only if we didn't fail already on this file. + */ + errno = result; + if (!FILE_POSSIBLY_DELETED(errno) || entry->retry_count > 0) + ereport(data_sync_elevel(ERROR), + (errcode_for_file_access(), + errmsg("could not fsync file \"%s\": %m", + entry->path))); + else + ereport(DEBUG1, + (errcode_for_file_access(), + errmsg_internal("could not fsync file \"%s\" but retrying: %m", + entry->path))); + + /* Cleanup must not check this already-handled result as a retry. */ + entry->started = false; + entry->open_errno = 0; + entry->retry_count++; + dlist_push_tail(&sync_state->retry, &entry->node); + } +} + +static void +sync_drain_all(SyncState *sync_state) +{ + while (sync_state->inflight_count) + sync_drain_one(sync_state); +} + +/* + * Make room for another fsync using the limit appropriate for its handler. + */ +static void +sync_ensure_room(SyncState *sync_state, const FileTag *tag) +{ + int max_inflight; + + max_inflight = + GetFsyncConcurrencyLimit(syncsw[tag->handler].uses_transient_fd); + while (sync_state->inflight_count >= max_inflight) + sync_drain_one(sync_state); +} + +/* + * Finish requests whose fsyncs have completed. + * + * The main hash scan may only remove the entry it most recently returned, so + * completion processing is deferred until it ends. This second scan can then + * remove each completed entry as the current entry. Recheck the hash entry + * now because requests absorbed since the fsync completed may require it to + * remain for the next checkpoint cycle. + */ +static void +sync_process_completed(void) +{ HASH_SEQ_STATUS hstat; PendingFsyncEntry *entry; - int absorb_counter; - /* Statistics on sync times */ - int processed = 0; - instr_time sync_start, - sync_end, - sync_diff; - uint64 elapsed; - uint64 longest = 0; - uint64 total_elapsed = 0; + hash_seq_init(&hstat, pendingOps); + while ((entry = (PendingFsyncEntry *) hash_seq_search(&hstat)) != NULL) + { + if (!entry->sync_completed) + continue; + + /* + * We are done with this entry, unless a request for it arrived while + * the fsync was in flight. A cancel supersedes any such request, as + * RememberSyncRequest() clears "canceled" when it records a new one. + */ + if (!entry->re_requested || entry->canceled) + { + if (hash_search(pendingOps, &entry->tag, HASH_REMOVE, NULL) == NULL) + elog(ERROR, "pendingOps corrupted"); + } + else + entry->sync_completed = false; + } +} + +/* + * Reissue any fsync requests that previously failed with an ignorable error. + * + * The fsync table could contain requests to fsync segments that have been + * deleted (unlinked) by the time we get to them. Rather than just hoping an + * ENOENT (or EACCES on Windows) error can be ignored, what we do on error is + * absorb pending requests and then retry. Since mdunlink() queues a "cancel" + * message before actually unlinking, the fsync request is guaranteed to be + * marked canceled after the absorb if it really was this case. + */ +static void +sync_process_retries(SyncState *sync_state) +{ + if (dlist_is_empty(&sync_state->retry)) + return; + + AbsorbSyncRequests(); + + while (!dlist_is_empty(&sync_state->retry)) + { + dlist_node *node = dlist_pop_head_node(&sync_state->retry); + InflightSyncEntry *entry = dlist_container(InflightSyncEntry, node, node); + + if (entry->hash_entry->canceled) + { + /* Safe to remove here, the scan has already finished. */ + if (hash_search(pendingOps, &entry->hash_entry->tag, + HASH_REMOVE, NULL) == NULL) + elog(ERROR, "pendingOps corrupted"); + sync_free_entry(entry); + continue; + } + + sync_ensure_room(sync_state, &entry->tag); + sync_start_one(sync_state, entry); + } + + sync_drain_all(sync_state); + sync_process_completed(); +} + +/* + * Process queued fsync requests. The public wrapper ensures that any error + * closes files owned by in-flight entries. + */ +static void +ProcessSyncRequestsInternal(void) +{ + static bool sync_in_progress = false; + + HASH_SEQ_STATUS hstat; + PendingFsyncEntry *entry; + SyncState sync_state; /* * This is only called during checkpoints, and checkpoints should only @@ -350,6 +755,7 @@ ProcessSyncRequests(void) while ((entry = (PendingFsyncEntry *) hash_seq_search(&hstat)) != NULL) { entry->cycle_ctr = sync_cycle_ctr; + entry->sync_completed = false; } } @@ -359,17 +765,27 @@ ProcessSyncRequests(void) /* Set flag to detect failure if we don't reach the end of the loop */ sync_in_progress = true; + /* + * The limit for each request depends on whether its handler holds a + * transient descriptor until completion. + */ + dlist_init(&sync_state.inflight); + dlist_init(&sync_state.retry); + sync_state.inflight_count = 0; + sync_state.processed = 0; + INSTR_TIME_SET_ZERO(sync_state.longest); + INSTR_TIME_SET_ZERO(sync_state.total_elapsed); + + Assert(dlist_is_empty(&activeSyncEntries)); + MemoryContextReset(inflightSyncCxt); + /* Now scan the hashtable for fsync requests to process */ - absorb_counter = FSYNCS_PER_ABSORB; + sync_state.absorb_counter = FSYNCS_PER_ABSORB; hash_seq_init(&hstat, pendingOps); while ((entry = (PendingFsyncEntry *) hash_seq_search(&hstat)) != NULL) { - int failures; - /* - * If the entry is new then don't process it this time; it is new. - * Note "continue" bypasses the hash-remove call at the bottom of the - * loop. + * Leave requests added in this cycle for the next checkpoint. */ if (entry->cycle_ctr == sync_cycle_ctr) continue; @@ -378,103 +794,92 @@ ProcessSyncRequests(void) Assert((CycleCtr) (entry->cycle_ctr + 1) == sync_cycle_ctr); /* - * If fsync is off then we don't have to bother opening the file at - * all. (We delay checking until this point so that changing fsync on - * the fly behaves sensibly.) + * If in checkpointer, we want to absorb pending requests every so + * often to prevent overflow of the fsync request queue. It is + * unspecified whether newly-added entries will be visited by + * hash_seq_search, but we don't care since we don't need to process + * them anyway. */ - if (enableFsync) + if (enableFsync && --sync_state.absorb_counter <= 0) { - /* - * If in checkpointer, we want to absorb pending requests every so - * often to prevent overflow of the fsync request queue. It is - * unspecified whether newly-added entries will be visited by - * hash_seq_search, but we don't care since we don't need to - * process them anyway. - */ - if (--absorb_counter <= 0) - { - AbsorbSyncRequests(); - absorb_counter = FSYNCS_PER_ABSORB; - } + AbsorbSyncRequests(); + sync_state.absorb_counter = FSYNCS_PER_ABSORB; + } + + if (!enableFsync || entry->canceled) + { + /* We are done with this entry, remove it */ + if (hash_search(pendingOps, &entry->tag, HASH_REMOVE, NULL) == NULL) + elog(ERROR, "pendingOps corrupted"); + } + else + { + InflightSyncEntry *inflight_entry; + + sync_ensure_room(&sync_state, &entry->tag); /* - * The fsync table could contain requests to fsync segments that - * have been deleted (unlinked) by the time we get to them. Rather - * than just hoping an ENOENT (or EACCES on Windows) error can be - * ignored, what we do on error is absorb pending requests and - * then retry. Since mdunlink() queues a "cancel" message before - * actually unlinking, the fsync request is guaranteed to be - * marked canceled after the absorb if it really was this case. - * DROP DATABASE likewise has to tell us to forget fsync requests - * before it starts deletions. + * Mark the entry as already dealt with in this cycle. It must + * remain in the hash table until its fsync completes and the scan + * ends. If a new request arrives meanwhile, this cycle counter + * leaves the entry to be processed by the next checkpoint. */ - for (failures = 0; !entry->canceled; failures++) - { - char path[MAXPGPATH]; - - INSTR_TIME_SET_CURRENT(sync_start); - if (syncsw[entry->tag.handler].sync_syncfiletag(&entry->tag, - path) == 0) - { - /* Success; update statistics about sync timing */ - INSTR_TIME_SET_CURRENT(sync_end); - sync_diff = sync_end; - INSTR_TIME_SUBTRACT(sync_diff, sync_start); - elapsed = INSTR_TIME_GET_MICROSEC(sync_diff); - if (elapsed > longest) - longest = elapsed; - total_elapsed += elapsed; - processed++; - - if (log_checkpoints) - elog(DEBUG1, "checkpoint sync: number=%d file=%s time=%.3f ms", - processed, - path, - (double) elapsed / 1000); - - break; /* out of retry loop */ - } - - /* - * It is possible that the relation has been dropped or - * truncated since the fsync request was entered. Therefore, - * allow ENOENT, but only if we didn't fail already on this - * file. - */ - if (!FILE_POSSIBLY_DELETED(errno) || failures > 0) - ereport(data_sync_elevel(ERROR), - (errcode_for_file_access(), - errmsg("could not fsync file \"%s\": %m", - path))); - else - ereport(DEBUG1, - (errcode_for_file_access(), - errmsg_internal("could not fsync file \"%s\" but retrying: %m", - path))); - - /* - * Absorb incoming requests and check to see if a cancel - * arrived for this relation fork. - */ - AbsorbSyncRequests(); - absorb_counter = FSYNCS_PER_ABSORB; /* might as well... */ - } /* end retry loop */ + entry->cycle_ctr = sync_cycle_ctr; + + inflight_entry = MemoryContextAllocZero(inflightSyncCxt, + sizeof(InflightSyncEntry)); + inflight_entry->tag = entry->tag; + inflight_entry->hash_entry = entry; + dlist_push_tail(&activeSyncEntries, + &inflight_entry->cleanup_node); + + sync_start_one(&sync_state, inflight_entry); } + } + + sync_drain_all(&sync_state); + sync_process_completed(); + + /* + * A second failure raises an error, so normally one retry pass is enough. + * Keep an explicit bound in case that changes. + */ + for (int failures = 0; failures < 5; failures++) + { + if (dlist_is_empty(&sync_state.retry)) + break; + + sync_process_retries(&sync_state); + } - /* We are done with this entry, remove it */ - if (hash_search(pendingOps, &entry->tag, HASH_REMOVE, NULL) == NULL) - elog(ERROR, "pendingOps corrupted"); - } /* end loop over hashtable entries */ + if (!dlist_is_empty(&sync_state.inflight) || + !dlist_is_empty(&sync_state.retry)) + elog(PANIC, "in-flight sync requests remain after ProcessSyncRequests"); /* Return sync performance metrics for report at checkpoint end */ - CheckpointStats.ckpt_sync_rels = processed; - CheckpointStats.ckpt_longest_sync = longest; - CheckpointStats.ckpt_agg_sync_time = total_elapsed; + CheckpointStats.ckpt_sync_rels = sync_state.processed; + CheckpointStats.ckpt_longest_sync = INSTR_TIME_GET_MICROSEC(sync_state.longest); + CheckpointStats.ckpt_agg_sync_time = INSTR_TIME_GET_MICROSEC(sync_state.total_elapsed); /* Flag successful completion of ProcessSyncRequests */ sync_in_progress = false; } +/* + * ProcessSyncRequests() -- Process queued fsync requests. + */ +void +ProcessSyncRequests(void) +{ + PG_ENSURE_ERROR_CLEANUP(sync_cleanup_inflight, (Datum) 0); + { + ProcessSyncRequestsInternal(); + } + PG_END_ENSURE_ERROR_CLEANUP(sync_cleanup_inflight, (Datum) 0); + + Assert(dlist_is_empty(&activeSyncEntries)); +} + /* * RememberSyncRequest() -- callback from checkpointer side of sync request * @@ -554,11 +959,20 @@ RememberSyncRequest(const FileTag *ftag, SyncRequestType type) ftag, HASH_ENTER, &found); + + /* + * If an entry already existed, an fsync for it may be in flight right + * now, in which case it cannot be assumed to cover this request; see + * sync_drain_one(). + */ + entry->re_requested = found; + /* if new entry, or was previously canceled, initialize it */ if (!found || entry->canceled) { entry->cycle_ctr = sync_cycle_ctr; entry->canceled = false; + entry->sync_completed = false; } /* diff --git a/src/backend/utils/activity/pgstat_io.c b/src/backend/utils/activity/pgstat_io.c index 8ec1aad5078..761847fac05 100644 --- a/src/backend/utils/activity/pgstat_io.c +++ b/src/backend/utils/activity/pgstat_io.c @@ -99,6 +99,26 @@ pgstat_prepare_io_time(bool track_io_guc) return io_start; } +/* + * Like pgstat_count_io_op_time_end() except IO end time is not supplied, it + * is calculated inside of the function. + */ +void +pgstat_count_io_op_time(IOObject io_object, IOContext io_context, IOOp io_op, + instr_time start_time, uint32 cnt, uint64 bytes) +{ + if (!INSTR_TIME_IS_ZERO(start_time)) + { + instr_time cur_time; + + INSTR_TIME_SET_CURRENT(cur_time); + pgstat_count_io_op_time_end(io_object, io_context, io_op, + start_time, cur_time, cnt, bytes); + } + else + pgstat_count_io_op(io_object, io_context, io_op, cnt, bytes); +} + /* * Like pgstat_count_io_op() except it also accumulates time. * @@ -111,42 +131,41 @@ pgstat_prepare_io_time(bool track_io_guc) * activity of temporary blocks, so these are ignored here. */ void -pgstat_count_io_op_time(IOObject io_object, IOContext io_context, IOOp io_op, - instr_time start_time, uint32 cnt, uint64 bytes) +pgstat_count_io_op_time_end(IOObject io_object, IOContext io_context, IOOp io_op, + instr_time start_time, instr_time end_time, uint32 cnt, + uint64 bytes) { if (!INSTR_TIME_IS_ZERO(start_time)) { - instr_time io_time; - - INSTR_TIME_SET_CURRENT(io_time); - INSTR_TIME_SUBTRACT(io_time, start_time); + Assert(!INSTR_TIME_GT(start_time, end_time)); + INSTR_TIME_SUBTRACT(end_time, start_time); if (io_object != IOOBJECT_WAL) { if (io_op == IOOP_WRITE || io_op == IOOP_EXTEND) { - pgstat_count_buffer_write_time(INSTR_TIME_GET_MICROSEC(io_time)); + pgstat_count_buffer_write_time(INSTR_TIME_GET_MICROSEC(end_time)); if (io_object == IOOBJECT_RELATION) - INSTR_TIME_ADD(pgBufferUsage.shared_blk_write_time, io_time); + INSTR_TIME_ADD(pgBufferUsage.shared_blk_write_time, end_time); else if (io_object == IOOBJECT_TEMP_RELATION) - INSTR_TIME_ADD(pgBufferUsage.local_blk_write_time, io_time); + INSTR_TIME_ADD(pgBufferUsage.local_blk_write_time, end_time); } else if (io_op == IOOP_READ) { - pgstat_count_buffer_read_time(INSTR_TIME_GET_MICROSEC(io_time)); + pgstat_count_buffer_read_time(INSTR_TIME_GET_MICROSEC(end_time)); if (io_object == IOOBJECT_RELATION) - INSTR_TIME_ADD(pgBufferUsage.shared_blk_read_time, io_time); + INSTR_TIME_ADD(pgBufferUsage.shared_blk_read_time, end_time); else if (io_object == IOOBJECT_TEMP_RELATION) - INSTR_TIME_ADD(pgBufferUsage.local_blk_read_time, io_time); + INSTR_TIME_ADD(pgBufferUsage.local_blk_read_time, end_time); } } INSTR_TIME_ADD(PendingIOStats.pending_times[io_object][io_context][io_op], - io_time); + end_time); /* Add the per-backend count */ pgstat_count_backend_io_op_time(io_object, io_context, io_op, - io_time); + end_time); } pgstat_count_io_op(io_object, io_context, io_op, cnt, bytes); diff --git a/src/include/access/clog.h b/src/include/access/clog.h index 7894998c763..e089106f7fe 100644 --- a/src/include/access/clog.h +++ b/src/include/access/clog.h @@ -47,7 +47,7 @@ extern void CheckPointCLOG(void); extern void ExtendCLOG(TransactionId newestXact); extern void TruncateCLOG(TransactionId oldestXact, Oid oldestxid_datoid); -extern int clogsyncfiletag(const FileTag *ftag, char *path); +extern void clogsyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); /* XLOG stuff */ #define CLOG_ZEROPAGE 0x00 diff --git a/src/include/access/commit_ts.h b/src/include/access/commit_ts.h index 825ccda90ed..fa4880e0d03 100644 --- a/src/include/access/commit_ts.h +++ b/src/include/access/commit_ts.h @@ -38,7 +38,7 @@ extern void SetCommitTsLimit(TransactionId oldestXact, TransactionId newestXact); extern void AdvanceOldestCommitTsXid(TransactionId oldestXact); -extern int committssyncfiletag(const FileTag *ftag, char *path); +extern void committssyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); /* XLOG stuff */ #define COMMIT_TS_ZEROPAGE 0x00 diff --git a/src/include/access/multixact.h b/src/include/access/multixact.h index 6be5299ab68..3f980b4120d 100644 --- a/src/include/access/multixact.h +++ b/src/include/access/multixact.h @@ -114,8 +114,8 @@ extern bool MultiXactIdPrecedes(MultiXactId multi1, MultiXactId multi2); extern bool MultiXactIdPrecedesOrEquals(MultiXactId multi1, MultiXactId multi2); -extern int multixactoffsetssyncfiletag(const FileTag *ftag, char *path); -extern int multixactmemberssyncfiletag(const FileTag *ftag, char *path); +extern void multixactoffsetssyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); +extern void multixactmemberssyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); extern void AtEOXact_MultiXact(void); extern void AtPrepare_MultiXact(void); diff --git a/src/include/access/slru.h b/src/include/access/slru.h index b4adb1789c7..0e91df5609c 100644 --- a/src/include/access/slru.h +++ b/src/include/access/slru.h @@ -240,7 +240,7 @@ typedef bool (*SlruScanCallback) (SlruDesc *ctl, char *filename, int64 segpage, extern bool SlruScanDirectory(SlruDesc *ctl, SlruScanCallback callback, void *data); extern void SlruDeleteSegment(SlruDesc *ctl, int64 segno); -extern int SlruSyncFileTag(SlruDesc *ctl, const FileTag *ftag, char *path); +extern void SlruSyncFileTag(SlruDesc *ctl, struct PgAioHandle *ioh, struct InflightSyncEntry *entry); /* SlruScanDirectory public callbacks */ extern bool SlruScanDirCbReportPresence(SlruDesc *ctl, char *filename, diff --git a/src/include/pgstat.h b/src/include/pgstat.h index 187d82c96fe..8c91aa4ef85 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -703,6 +703,10 @@ extern instr_time pgstat_prepare_io_time(bool track_io_guc); extern void pgstat_count_io_op_time(IOObject io_object, IOContext io_context, IOOp io_op, instr_time start_time, uint32 cnt, uint64 bytes); +extern void pgstat_count_io_op_time_end(IOObject io_object, IOContext io_context, + IOOp io_op, instr_time start_time, + instr_time end_time, uint32 cnt, + uint64 bytes); extern PgStat_IO *pgstat_fetch_stat_io(void); extern const char *pgstat_get_io_context_name(IOContext io_context); diff --git a/src/include/storage/aio.h b/src/include/storage/aio.h index 7f056f43e78..6ea230ccad5 100644 --- a/src/include/storage/aio.h +++ b/src/include/storage/aio.h @@ -176,6 +176,12 @@ struct PgAioTargetInfo */ int (*reopen) (PgAioHandle *ioh); + /* + * Optional counterpart to reopen, releasing the file descriptor it + * acquired once the IO has been executed. + */ + void (*close) (PgAioHandle *ioh); + /* describe the target of the IO, used for log messages and views */ char *(*describe_identity) (const PgAioTargetData *sd); diff --git a/src/include/storage/aio_internal.h b/src/include/storage/aio_internal.h index 35d4f3f310d..ccf0fe4c7bf 100644 --- a/src/include/storage/aio_internal.h +++ b/src/include/storage/aio_internal.h @@ -360,6 +360,7 @@ extern int pgaio_io_get_iovec_length(PgAioHandle *ioh, struct iovec **iov); /* aio_target.c */ extern bool pgaio_io_can_reopen(PgAioHandle *ioh); extern int pgaio_io_reopen(PgAioHandle *ioh); +extern void pgaio_io_close_reopened(PgAioHandle *ioh); extern const char *pgaio_io_get_target_name(PgAioHandle *ioh); diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h index c79f3312544..49f93b6a7ad 100644 --- a/src/include/storage/fd.h +++ b/src/include/storage/fd.h @@ -138,6 +138,7 @@ extern int FilePrefetch(File file, pgoff_t offset, pgoff_t amount, uint32 wait_e extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, pgoff_t offset, uint32 wait_event_info); extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, pgoff_t offset, uint32 wait_event_info); extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, pgoff_t offset, uint32 wait_event_info); +extern int FileStartSync(struct PgAioHandle *ioh, File file, bool datasync, uint32 wait_event_info); extern int FileSync(File file, uint32 wait_event_info); extern int FileZero(File file, pgoff_t offset, pgoff_t amount, uint32 wait_event_info); extern int FileFallocate(File file, pgoff_t offset, pgoff_t amount, uint32 wait_event_info); @@ -219,7 +220,7 @@ extern int fsync_fname_ext(const char *fname, bool isdir, bool ignore_perm, int extern int durable_rename(const char *oldfile, const char *newfile, int elevel); extern int durable_unlink(const char *fname, int elevel); extern void SyncDataDirectory(void); -extern int GetFsyncConcurrencyLimit(void); +extern int GetFsyncConcurrencyLimit(bool uses_transient_fd); extern int data_sync_elevel(int elevel); static inline ssize_t diff --git a/src/include/storage/md.h b/src/include/storage/md.h index b8d10329eb8..9f03354c2bc 100644 --- a/src/include/storage/md.h +++ b/src/include/storage/md.h @@ -53,12 +53,13 @@ extern void mdtruncate(SMgrRelation reln, ForkNumber forknum, extern void mdimmedsync(SMgrRelation reln, ForkNumber forknum); extern void mdregistersync(SMgrRelation reln, ForkNumber forknum); extern int mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); +extern int mdfsyncfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum); extern void ForgetDatabaseSyncRequests(Oid dbid); extern void DropRelationFiles(RelFileLocator *delrels, int ndelrels, bool isRedo); /* md sync callbacks */ -extern int mdsyncfiletag(const FileTag *ftag, char *path); +extern void mdsyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); extern int mdunlinkfiletag(const FileTag *ftag, char *path); extern bool mdfiletagmatches(const FileTag *ftag, const FileTag *candidate); diff --git a/src/include/storage/sync.h b/src/include/storage/sync.h index 88290500bc9..8d48151730f 100644 --- a/src/include/storage/sync.h +++ b/src/include/storage/sync.h @@ -13,6 +13,9 @@ #ifndef SYNC_H #define SYNC_H +#include "lib/ilist.h" +#include "portability/instr_time.h" +#include "storage/aio_types.h" #include "storage/relfilelocator.h" /* @@ -55,6 +58,58 @@ typedef struct FileTag uint64 segno; } FileTag; +struct PendingFsyncEntry; +struct PgAioHandle; + +/* + * How the file opened by a sync handler must be closed once its asynchronous + * fsync has completed. + */ +typedef enum SyncFileCloseMethod +{ + SYNC_CLOSE_NONE = 0, /* nothing to close */ + SYNC_CLOSE_TRANSIENT, /* CloseTransientFile(close_file) */ + SYNC_CLOSE_VFD, /* FileClose((File) close_file) */ +} SyncFileCloseMethod; + +/* + * State for a single in-flight asynchronous fsync request. A sync handler + * opens the file to be synced, fills in the fields it is responsible for, and + * starts an asynchronous fsync on the AIO handle it is given. + */ +typedef struct InflightSyncEntry +{ + FileTag tag; /* identifies handler and file */ + + char path[MAXPGPATH]; + + /* + * Set by the handler: whether it started an asynchronous fsync on the + * passed-in AIO handle. If the file could not be opened, the handler + * sets started = false and open_errno to the errno of the failed open. + */ + bool started; + int open_errno; + + /* set by the handler: how to close the opened file after completion */ + SyncFileCloseMethod close_method; + int close_file; /* fd, or File, depending on close_method */ + + struct PendingFsyncEntry *hash_entry; + + int retry_count; + instr_time io_time; /* time spent starting and waiting */ + + PgAioReturn ioret; + PgAioWaitRef iow; + + /* membership in the inflight / retry lists */ + dlist_node node; + + /* membership in the error-cleanup list */ + dlist_node cleanup_node; +} InflightSyncEntry; + extern void InitSync(void); extern void SyncPreCheckpoint(void); extern void SyncPostCheckpoint(void); diff --git a/src/test/modules/test_slru/test_slru.c b/src/test/modules/test_slru/test_slru.c index 40efffdbf62..ccac09a9285 100644 --- a/src/test/modules/test_slru/test_slru.c +++ b/src/test/modules/test_slru/test_slru.c @@ -17,10 +17,13 @@ #include "access/slru.h" #include "access/transam.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/ipc.h" #include "storage/shmem.h" +#include "storage/sync.h" #include "utils/builtins.h" +#include "utils/resowner.h" PG_MODULE_MAGIC; @@ -152,15 +155,47 @@ Datum test_slru_page_sync(PG_FUNCTION_ARGS) { int64 pageno = PG_GETARG_INT64(0); - FileTag ftag; - char path[MAXPGPATH]; + InflightSyncEntry entry = {0}; + PgAioHandle *ioh; + int result; /* note that this flushes the full file a segment is located in */ - ftag.segno = pageno / SLRU_PAGES_PER_SEGMENT; - SlruSyncFileTag(TestSlruCtl, &ftag, path); + entry.tag.segno = pageno / SLRU_PAGES_PER_SEGMENT; + + /* + * SlruSyncFileTag() now performs the fsync asynchronously. Drive it the + * same way sync.c does: acquire an AIO handle, let the handler start the + * fsync, wait for its completion and close the file it opened. + */ + ioh = pgaio_io_acquire(CurrentResourceOwner, &entry.ioret); + pgaio_io_get_wref(ioh, &entry.iow); + + HOLD_INTERRUPTS(); + SlruSyncFileTag(TestSlruCtl, ioh, &entry); + RESUME_INTERRUPTS(); + + if (entry.started) + { + pgaio_wref_wait(&entry.iow); + result = -entry.ioret.result.result; + CloseTransientFile(entry.close_file); + } + else + { + pgaio_io_release(ioh); + result = entry.open_errno; + } + + if (result != 0) + { + errno = result; + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not fsync file \"%s\": %m", entry.path))); + } elog(NOTICE, "Called SlruSyncFileTag() for segment %" PRIu64 " on path %s", - ftag.segno, path); + entry.tag.segno, entry.path); PG_RETURN_VOID(); } diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 21bf56da4fa..c5ed5bc6c0f 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1338,6 +1338,7 @@ IndexVacuumInfo IndxInfo InferClause InferenceElem +InflightSyncEntry InfoItem InhInfo InheritableSocket @@ -3050,12 +3051,14 @@ SupportRequestSimplify SupportRequestSimplifyAggref SupportRequestWFuncMonotonic Syn +SyncFileCloseMethod SyncOps SyncRepConfigData SyncRepStandbyData SyncRequestHandler SyncRequestType SyncStandbySlotsConfigData +SyncState SyncingRelationsState SysCacheIdentifier SysFKRelationship -- 2.47.3