From 1845c07e610d9651759f9ce4239c042074916e7c Mon Sep 17 00:00:00 2001 From: Nazir Bilal Yavuz Date: Tue, 25 Aug 2026 11:27:22 +0300 Subject: [PATCH v1 3/4] 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. Each in-flight SLRU fsync holds a transient file descriptor, so limit the depth with GetFsyncConcurrencyLimit() rather than io_max_concurrency to avoid exhausting 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. Relation files can be reopened through the smgr target, allowing I/O workers to perform their fsyncs. SLRU files use the generic sync target and execute synchronously with the worker I/O method, although io_uring can still overlap them. Because handlers now return before an fsync completes, md.c reports submission time to pg_stat_io, as it does for asynchronous reads. Methods that execute the fsync immediately still report its duration, and fsync counts are unchanged. The per-file checkpoint timing statistics now measure submission-to-reap time rather than fsync duration. Submission-order reaping can overstate individual times, and overlapping operations can make the aggregate exceed the wall-clock time of the sync phase. --- 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/file/fd.c | 23 ++ src/backend/storage/smgr/md.c | 68 +++- src/backend/storage/sync/sync.c | 537 ++++++++++++++++++++----- 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/storage/fd.h | 1 + src/include/storage/md.h | 2 +- src/include/storage/sync.h | 56 +++ src/test/modules/test_slru/test_slru.c | 45 ++- src/tools/pgindent/typedefs.list | 3 + 16 files changed, 645 insertions(+), 159 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 d688815083c..deb866cd7f0 100644 --- a/src/backend/access/transam/multixact.c +++ b/src/backend/access/transam/multixact.c @@ -2998,17 +2998,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/file/fd.c b/src/backend/storage/file/fd.c index 0fcae14541d..f91ee488b89 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) diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index 780c88c0630..e1fca54d9d9 100644 --- a/src/backend/storage/smgr/md.c +++ b/src/backend/storage/smgr/md.c @@ -1896,26 +1896,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 +1924,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/sync/sync.c b/src/backend/storage/sync/sync.c index 2c964b6f3d9..4263881ddd8 100644 --- a/src/backend/storage/sync/sync.c +++ b/src/backend/storage/sync/sync.c @@ -26,7 +26,9 @@ #include "pgstat.h" #include "portability/instr_time.h" #include "postmaster/bgwriter.h" +#include "storage/aio.h" #include "storage/fd.h" +#include "storage/ipc.h" #include "storage/latch.h" #include "storage/md.h" #include "utils/hsearch.h" @@ -54,11 +56,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 +81,44 @@ 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 max_inflight; /* max number of concurrent fsyncs */ + 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,7 +131,7 @@ 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); @@ -155,6 +202,10 @@ InitSync(void) &hash_ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); pendingUnlinks = NIL; + + inflightSyncCxt = AllocSetContextCreate(TopMemoryContext, + "Inflight sync context", + ALLOCSET_DEFAULT_SIZES); } } @@ -281,25 +332,295 @@ 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) +{ + while (!dlist_is_empty(&activeSyncEntries)) + { + dlist_node *node = dlist_pop_head_node(&activeSyncEntries); + InflightSyncEntry *entry; + + entry = dlist_container(InflightSyncEntry, cleanup_node, node); + + if (entry->started) + pgaio_wref_wait(&entry->iow); + + sync_close_file(entry); + pfree(entry); + } +} + +static void +sync_start_one(SyncState *sync_state, InflightSyncEntry *entry) +{ + struct PgAioHandle *ioh; + instr_time io_start; + + INSTR_TIME_SET_CURRENT(io_start); + entry->start_time = io_start; + + entry->started = false; + entry->open_errno = 0; + entry->close_method = SYNC_CLOSE_NONE; + 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(); + syncsw[entry->tag.handler].sync_syncfiletag(ioh, entry); + RESUME_INTERRUPTS(); + + if (!entry->started) + pgaio_io_release(ioh); + + dlist_push_tail(&sync_state->inflight, &entry->node); + sync_state->inflight_count++; +} + +static void +sync_drain_one(SyncState *sync_state) +{ + dlist_node *node; + InflightSyncEntry *entry; + int result; + + 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) + { + pgaio_wref_wait(&entry->iow); + + /* + * 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); + + if (!result) + { + instr_time io_time; + + /* + * These values measure submission-to-reap time, not necessarily fsync + * duration. Submission-order reaping can overstate individual + * durations, and the aggregate can exceed wall-clock time because + * fsyncs overlap. + */ + INSTR_TIME_SET_CURRENT(io_time); + INSTR_TIME_SUBTRACT(io_time, entry->start_time); + + 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))); + + 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); +} + +/* + * 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; + } + + Assert(sync_state->inflight_count <= sync_state->max_inflight); + if (sync_state->inflight_count == sync_state->max_inflight) + sync_drain_one(sync_state); + + 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 +671,7 @@ ProcessSyncRequests(void) while ((entry = (PendingFsyncEntry *) hash_seq_search(&hstat)) != NULL) { entry->cycle_ctr = sync_cycle_ctr; + entry->sync_completed = false; } } @@ -359,13 +681,26 @@ ProcessSyncRequests(void) /* Set flag to detect failure if we don't reach the end of the loop */ sync_in_progress = true; + /* + * Bound concurrent fsyncs by both the AIO handle and transient descriptor + * budgets. + */ + dlist_init(&sync_state.inflight); + dlist_init(&sync_state.retry); + sync_state.inflight_count = 0; + sync_state.max_inflight = GetFsyncConcurrencyLimit(); + 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 @@ -378,103 +713,94 @@ 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; + + Assert(sync_state.inflight_count <= sync_state.max_inflight); + if (sync_state.inflight_count == sync_state.max_inflight) + sync_drain_one(&sync_state); /* - * 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); } + } - /* 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 */ + 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); + } + + 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 +880,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/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/storage/fd.h b/src/include/storage/fd.h index c79f3312544..f39469058b7 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); diff --git a/src/include/storage/md.h b/src/include/storage/md.h index b8d10329eb8..53f75802ac0 100644 --- a/src/include/storage/md.h +++ b/src/include/storage/md.h @@ -58,7 +58,7 @@ 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..a72ce4c3fd3 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,59 @@ 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 start_time; + + 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 a95b09859b5..4fd08012e1b 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1361,6 +1361,7 @@ IndexVacuumInfo IndxInfo InferClause InferenceElem +InflightSyncEntry InfoItem InhInfo InheritableSocket @@ -3078,12 +3079,14 @@ SupportRequestSimplify SupportRequestSimplifyAggref SupportRequestWFuncMonotonic Syn +SyncFileCloseMethod SyncOps SyncRepConfigData SyncRepStandbyData SyncRequestHandler SyncRequestType SyncStandbySlotsConfigData +SyncState SyncingRelationsState SysCacheIdentifier SysFKRelationship -- 2.47.3