From 203cf306618d1440e74e7b192edc3b13b3c679a7 Mon Sep 17 00:00:00 2001 From: Nazir Bilal Yavuz Date: Tue, 25 Aug 2026 15:00:40 +0300 Subject: [PATCH v1 4/4] Allow IO workers to execute SLRU fsyncs The SLRU fsyncs introduced by the preceding commit use the generic PGAIO_TID_SYNC target. It does not store paths in shared memory, so I/O workers cannot reopen the files and the checkpointer must perform the operations itself. Add PGAIO_TID_SYNC_FILETAG, which identifies a file by the FileTag registered with sync.c. A FileTag contains enough information to reopen the file in another process. Dispatch reopening through an optional SyncOps callback, implemented for SLRUs using the new SlruOpenFileTag(). SLRUs that are not registered with sync.c continue to use PGAIO_TID_SYNC. Define the shared FileTag representation in storage/aio_types.h so PgAioTargetData can store it directly without duplicating its layout. Describe the target and worker-side open errors using the SLRU name and segment number. Unlike the smgr target, the FileTag target opens a new descriptor for each operation instead of using a cache. Add an optional PgAioTargetInfo close callback so I/O workers release these descriptors after executing the operation. SLRU_FLUSH_SYNC is now reported by the I/O worker rather than the checkpointer, as for relation fsyncs. SyncDataDirectory() continues to use PGAIO_TID_SYNC because its files have no FileTag. --- src/backend/access/transam/clog.c | 8 +- src/backend/access/transam/commit_ts.c | 8 +- src/backend/access/transam/multixact.c | 16 +++- src/backend/access/transam/slru.c | 32 ++++++- src/backend/storage/aio/aio_io.c | 8 ++ src/backend/storage/aio/aio_target.c | 36 +++++++ src/backend/storage/sync/sync.c | 125 ++++++++++++++++++++++++- src/include/access/clog.h | 1 + src/include/access/commit_ts.h | 1 + src/include/access/multixact.h | 2 + src/include/access/slru.h | 1 + src/include/storage/aio.h | 13 ++- src/include/storage/aio_internal.h | 1 + src/include/storage/aio_types.h | 14 +++ src/include/storage/sync.h | 19 ++-- src/tools/pgindent/typedefs.list | 1 + 16 files changed, 263 insertions(+), 23 deletions(-) diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c index 89fb77ea5da..0fe0491d877 100644 --- a/src/backend/access/transam/clog.c +++ b/src/backend/access/transam/clog.c @@ -1115,10 +1115,16 @@ clog_redo(XLogReaderState *record) } /* - * Entrypoint for sync.c to sync clog files. + * Entrypoints for sync.c to sync and reopen clog files. */ void clogsyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) { SlruSyncFileTag(XactCtl, ioh, entry); } + +int +clogopenfiletag(const FileTag *ftag) +{ + return SlruOpenFileTag(XactCtl, ftag); +} diff --git a/src/backend/access/transam/commit_ts.c b/src/backend/access/transam/commit_ts.c index 7cbbad383b2..4bb95e415ac 100644 --- a/src/backend/access/transam/commit_ts.c +++ b/src/backend/access/transam/commit_ts.c @@ -1026,10 +1026,16 @@ commit_ts_redo(XLogReaderState *record) } /* - * Entrypoint for sync.c to sync commit_ts files. + * Entrypoints for sync.c to sync and reopen commit_ts files. */ void committssyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) { SlruSyncFileTag(CommitTsCtl, ioh, entry); } + +int +committsopenfiletag(const FileTag *ftag) +{ + return SlruOpenFileTag(CommitTsCtl, ftag); +} diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c index deb866cd7f0..11f5f266075 100644 --- a/src/backend/access/transam/multixact.c +++ b/src/backend/access/transam/multixact.c @@ -2996,7 +2996,7 @@ multixact_redo(XLogReaderState *record) } /* - * Entrypoint for sync.c to sync offsets files. + * Entrypoints for sync.c to sync and reopen offsets files. */ void multixactoffsetssyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) @@ -3004,11 +3004,23 @@ multixactoffsetssyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) SlruSyncFileTag(MultiXactOffsetCtl, ioh, entry); } +int +multixactoffsetsopenfiletag(const FileTag *ftag) +{ + return SlruOpenFileTag(MultiXactOffsetCtl, ftag); +} + /* - * Entrypoint for sync.c to sync members files. + * Entrypoints for sync.c to sync and reopen members files. */ void multixactmemberssyncfiletag(struct PgAioHandle *ioh, InflightSyncEntry *entry) { SlruSyncFileTag(MultiXactMemberCtl, ioh, entry); } + +int +multixactmembersopenfiletag(const FileTag *ftag) +{ + return SlruOpenFileTag(MultiXactMemberCtl, ftag); +} diff --git a/src/backend/access/transam/slru.c b/src/backend/access/transam/slru.c index b1e513ac3b6..8e0d9af16df 100644 --- a/src/backend/access/transam/slru.c +++ b/src/backend/access/transam/slru.c @@ -1897,11 +1897,16 @@ SlruSyncFileTag(SlruDesc *ctl, struct PgAioHandle *ioh, InflightSyncEntry *entry } /* - * 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. + * If this SLRU is registered with sync.c, identify the file by its + * FileTag, so that the fsync can be executed by an IO worker, which will + * reopen the file with SlruOpenFileTag(). Otherwise there is no handler + * to reopen the file through, so use the generic sync target, whose IOs + * cannot be handed off to a worker. */ - pgaio_io_set_target(ioh, PGAIO_TID_SYNC); + if (ctl->options.sync_handler != SYNC_HANDLER_NONE) + pgaio_io_set_target_sync_filetag(ioh, &entry->tag); + else + pgaio_io_set_target(ioh, PGAIO_TID_SYNC); /* Start the asynchronous fsync; the fd is closed once it completes. */ pgaio_io_start_fsync(ioh, fd, false, WAIT_EVENT_SLRU_FLUSH_SYNC); @@ -1910,3 +1915,22 @@ SlruSyncFileTag(SlruDesc *ctl, struct PgAioHandle *ioh, InflightSyncEntry *entry entry->close_method = SYNC_CLOSE_TRANSIENT; entry->close_file = fd; } + +/* + * Counterpart to SlruSyncFileTag(), opening the segment identified by ftag in + * a process that did not stage the IO. As with SlruSyncFileTag(), individual + * SLRUs have to provide the handler function, so that the correct "SlruCtl" + * is used. + * + * Returns a file descriptor opened with OpenTransientFile(), or -1 with errno + * set. + */ +int +SlruOpenFileTag(SlruDesc *ctl, const FileTag *ftag) +{ + char path[MAXPGPATH]; + + SlruFileName(ctl, path, ftag->segno); + + return OpenTransientFile(path, O_RDWR | PG_BINARY); +} diff --git a/src/backend/storage/aio/aio_io.c b/src/backend/storage/aio/aio_io.c index 324fb6911e2..070a5ff70b3 100644 --- a/src/backend/storage/aio/aio_io.c +++ b/src/backend/storage/aio/aio_io.c @@ -169,6 +169,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 82b24b0026d..795e8ac7046 100644 --- a/src/backend/storage/aio/aio_target.c +++ b/src/backend/storage/aio/aio_target.c @@ -17,6 +17,7 @@ #include "storage/aio.h" #include "storage/aio_internal.h" #include "storage/smgr.h" +#include "storage/sync.h" static char *pgaio_sync_describe_identity(const PgAioTargetData *sd); @@ -39,8 +40,16 @@ static const PgAioTargetInfo *pgaio_target_info[] = { }, [PGAIO_TID_SMGR] = &aio_smgr_target_info, [PGAIO_TID_SYNC] = &aio_sync_target_info, + [PGAIO_TID_SYNC_FILETAG] = &aio_sync_filetag_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 @@ -141,6 +150,33 @@ 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); pgaio_target_info[ioh->target]->reopen(ioh); + + /* + * 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; +} + +/* + * 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/sync/sync.c b/src/backend/storage/sync/sync.c index 4263881ddd8..35f87793171 100644 --- a/src/backend/storage/sync/sync.c +++ b/src/backend/storage/sync/sync.c @@ -132,9 +132,19 @@ static CycleCtr checkpoint_cycle_ctr = 0; typedef struct SyncOps { void (*sync_syncfiletag) (PgAioHandle *ioh, InflightSyncEntry *entry); + + /* + * Optional. Reopen the file identified by ftag, so that an fsync started + * by sync_syncfiletag() can be executed in a different process, e.g. an + * IO worker. Returns a file descriptor opened with OpenTransientFile(), + * or -1 with errno set. Handlers that provide this must use the + * PGAIO_TID_SYNC_FILETAG target (see pgaio_io_set_target_sync_filetag()). + */ + int (*sync_openfiletag) (const FileTag *ftag); int (*sync_unlinkfiletag) (const FileTag *ftag, char *path); bool (*sync_filetagmatches) (const FileTag *ftag, const FileTag *candidate); + const char *sync_target_name; } SyncOps; /* @@ -149,22 +159,129 @@ static const SyncOps syncsw[] = { }, /* pg_xact */ [SYNC_HANDLER_CLOG] = { - .sync_syncfiletag = clogsyncfiletag + .sync_syncfiletag = clogsyncfiletag, + .sync_openfiletag = clogopenfiletag, + .sync_target_name = "pg_xact" }, /* pg_commit_ts */ [SYNC_HANDLER_COMMIT_TS] = { - .sync_syncfiletag = committssyncfiletag + .sync_syncfiletag = committssyncfiletag, + .sync_openfiletag = committsopenfiletag, + .sync_target_name = "pg_commit_ts" }, /* pg_multixact/offsets */ [SYNC_HANDLER_MULTIXACT_OFFSET] = { - .sync_syncfiletag = multixactoffsetssyncfiletag + .sync_syncfiletag = multixactoffsetssyncfiletag, + .sync_openfiletag = multixactoffsetsopenfiletag, + .sync_target_name = "pg_multixact/offsets" }, /* pg_multixact/members */ [SYNC_HANDLER_MULTIXACT_MEMBER] = { - .sync_syncfiletag = multixactmemberssyncfiletag + .sync_syncfiletag = multixactmemberssyncfiletag, + .sync_openfiletag = multixactmembersopenfiletag, + .sync_target_name = "pg_multixact/members" } }; +static void sync_aio_reopen(PgAioHandle *ioh); +static void sync_aio_close(PgAioHandle *ioh); +static char *sync_aio_describe_identity(const PgAioTargetData *sd); + +/* + * Target info for files identified by a FileTag (see PGAIO_TID_SYNC_FILETAG). + * Unlike PGAIO_TID_SYNC, a FileTag contains everything needed to find the + * file again in another process, so such IOs can be executed by IO workers. + */ +const PgAioTargetInfo aio_sync_filetag_target_info = { + .name = "sync_filetag", + .reopen = sync_aio_reopen, + .close = sync_aio_close, + .describe_identity = sync_aio_describe_identity, +}; + +/* + * Set up ioh to operate on the file identified by ftag. + */ +void +pgaio_io_set_target_sync_filetag(PgAioHandle *ioh, const FileTag *ftag) +{ + PgAioTargetData *sd = pgaio_io_get_target_data(ioh); + + Assert(syncsw[ftag->handler].sync_openfiletag != NULL); + Assert(syncsw[ftag->handler].sync_target_name != NULL); + + pgaio_io_set_target(ioh, PGAIO_TID_SYNC_FILETAG); + + sd->sync_filetag = *ftag; +} + +static FileTag +sync_aio_filetag(const PgAioTargetData *sd) +{ + return sd->sync_filetag; +} + +/* + * reopen callback for PGAIO_TID_SYNC_FILETAG, to open the file in the process + * executing the IO. + */ +static void +sync_aio_reopen(PgAioHandle *ioh) +{ + PgAioTargetData *sd = pgaio_io_get_target_data(ioh); + PgAioOpData *od = pgaio_io_get_op_data(ioh); + FileTag ftag = sync_aio_filetag(sd); + int fd; + + /* + * The caller needs to prevent interrupts from being processed, otherwise + * the FD could be closed again before we get to executing the IO. + */ + Assert(!INTERRUPTS_CAN_BE_PROCESSED()); + + Assert(pgaio_io_get_op(ioh) == PGAIO_OP_FSYNC); + + fd = syncsw[ftag.handler].sync_openfiletag(&ftag); + if (fd < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not open segment " UINT64_FORMAT " of SLRU \"%s\": %m", + ftag.segno, + syncsw[ftag.handler].sync_target_name))); + + od->fsync.fd = fd; +} + +/* + * close callback for PGAIO_TID_SYNC_FILETAG, releasing the descriptor + * acquired by sync_aio_reopen(). + * + * Called in a critical section, so a failure to close cannot be reported. + * That is not a meaningful loss: the data has already been flushed by the + * fsync, and the descriptor is not written to. + */ +static void +sync_aio_close(PgAioHandle *ioh) +{ + PgAioOpData *od = pgaio_io_get_op_data(ioh); + + (void) CloseTransientFile(od->fsync.fd); + od->fsync.fd = -1; +} + +/* + * describe_identity callback for PGAIO_TID_SYNC_FILETAG. + */ +static char * +sync_aio_describe_identity(const PgAioTargetData *sd) +{ + FileTag ftag = sync_aio_filetag(sd); + + return psprintf(_("segment " UINT64_FORMAT " of SLRU \"%s\""), + ftag.segno, + syncsw[ftag.handler].sync_target_name); +} + /* * Initialize data structures for the file sync tracking. */ diff --git a/src/include/access/clog.h b/src/include/access/clog.h index e089106f7fe..fc06794d186 100644 --- a/src/include/access/clog.h +++ b/src/include/access/clog.h @@ -48,6 +48,7 @@ extern void ExtendCLOG(TransactionId newestXact); extern void TruncateCLOG(TransactionId oldestXact, Oid oldestxid_datoid); extern void clogsyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); +extern int clogopenfiletag(const FileTag *ftag); /* XLOG stuff */ #define CLOG_ZEROPAGE 0x00 diff --git a/src/include/access/commit_ts.h b/src/include/access/commit_ts.h index fa4880e0d03..ebb87d9534e 100644 --- a/src/include/access/commit_ts.h +++ b/src/include/access/commit_ts.h @@ -39,6 +39,7 @@ extern void SetCommitTsLimit(TransactionId oldestXact, extern void AdvanceOldestCommitTsXid(TransactionId oldestXact); extern void committssyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); +extern int committsopenfiletag(const FileTag *ftag); /* XLOG stuff */ #define COMMIT_TS_ZEROPAGE 0x00 diff --git a/src/include/access/multixact.h b/src/include/access/multixact.h index 3f980b4120d..3f5233a3bcd 100644 --- a/src/include/access/multixact.h +++ b/src/include/access/multixact.h @@ -115,7 +115,9 @@ extern bool MultiXactIdPrecedesOrEquals(MultiXactId multi1, MultiXactId multi2); extern void multixactoffsetssyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); +extern int multixactoffsetsopenfiletag(const FileTag *ftag); extern void multixactmemberssyncfiletag(PgAioHandle *ioh, InflightSyncEntry *entry); +extern int multixactmembersopenfiletag(const FileTag *ftag); 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 0e91df5609c..43429392c7d 100644 --- a/src/include/access/slru.h +++ b/src/include/access/slru.h @@ -241,6 +241,7 @@ extern bool SlruScanDirectory(SlruDesc *ctl, SlruScanCallback callback, void *da extern void SlruDeleteSegment(SlruDesc *ctl, int64 segno); extern void SlruSyncFileTag(SlruDesc *ctl, struct PgAioHandle *ioh, struct InflightSyncEntry *entry); +extern int SlruOpenFileTag(SlruDesc *ctl, const FileTag *ftag); /* SlruScanDirectory public callbacks */ extern bool SlruScanDirCbReportPresence(SlruDesc *ctl, char *filename, diff --git a/src/include/storage/aio.h b/src/include/storage/aio.h index 428504c0472..904ab185458 100644 --- a/src/include/storage/aio.h +++ b/src/include/storage/aio.h @@ -119,9 +119,10 @@ typedef enum PgAioTargetID PGAIO_TID_INVALID = 0, PGAIO_TID_SMGR, PGAIO_TID_SYNC, + PGAIO_TID_SYNC_FILETAG, } PgAioTargetID; -#define PGAIO_TID_COUNT (PGAIO_TID_SYNC + 1) +#define PGAIO_TID_COUNT (PGAIO_TID_SYNC_FILETAG + 1) /* @@ -171,6 +172,16 @@ struct PgAioTargetInfo */ void (*reopen) (PgAioHandle *ioh); + /* + * Optional counterpart to reopen, releasing the file descriptor it + * acquired. Called in the process that reopened the IO, after the IO has + * been executed. Targets whose reopen callback hands out a descriptor + * that is cached and reused, like smgr's, do not need this. + * + * This is called in a critical section, so it must not raise errors. + */ + 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 9ca4087aa7f..9d2e92c6538 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 void 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/aio_types.h b/src/include/storage/aio_types.h index 17b59aeed7c..90bc4c73989 100644 --- a/src/include/storage/aio_types.h +++ b/src/include/storage/aio_types.h @@ -23,6 +23,18 @@ typedef struct PgAioHandle PgAioHandle; typedef struct PgAioHandleCallbacks PgAioHandleCallbacks; typedef struct PgAioTargetInfo PgAioTargetInfo; +/* + * A tag identifying a file handled by sync.c. This is defined here so that + * PgAioTargetData can store it without duplicating its representation. + */ +typedef struct PgAioSyncFileTag +{ + int16 handler; /* SyncRequestHandler value */ + int16 forknum; /* ForkNumber */ + RelFileLocator rlocator; /* physical relation identifier */ + uint64 segno; +} PgAioSyncFileTag; + /* * A reference to an IO that can be used to wait for the IO (using * pgaio_wref_wait()) to complete. @@ -69,6 +81,8 @@ typedef union PgAioTargetData bool is_temp:1; /* proc can be inferred by owning AIO */ bool skip_fsync:1; } smgr; + + PgAioSyncFileTag sync_filetag; } PgAioTargetData; diff --git a/src/include/storage/sync.h b/src/include/storage/sync.h index a72ce4c3fd3..4a2cca50703 100644 --- a/src/include/storage/sync.h +++ b/src/include/storage/sync.h @@ -46,17 +46,11 @@ typedef enum SyncRequestHandler } SyncRequestHandler; /* - * A tag identifying a file. Currently it has the members required for md.c's - * usage, but sync.c has no knowledge of the internal structure, and it is - * liable to change as required by future handlers. + * A tag identifying a file. Its representation is shared with AIO target + * data, so changes are automatically visible to processes that reopen files + * on behalf of sync.c. */ -typedef struct FileTag -{ - int16 handler; /* SyncRequestHandler value, saving space */ - int16 forknum; /* ForkNumber, saving space */ - RelFileLocator rlocator; - uint64 segno; -} FileTag; +typedef PgAioSyncFileTag FileTag; struct PendingFsyncEntry; struct PgAioHandle; @@ -119,4 +113,9 @@ extern void RememberSyncRequest(const FileTag *ftag, SyncRequestType type); extern bool RegisterSyncRequest(const FileTag *ftag, SyncRequestType type, bool retryOnError); +/* AIO support */ +extern PGDLLIMPORT const PgAioTargetInfo aio_sync_filetag_target_info; +extern void pgaio_io_set_target_sync_filetag(PgAioHandle *ioh, + const FileTag *ftag); + #endif /* SYNC_H */ diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 4fd08012e1b..0812aae6793 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2285,6 +2285,7 @@ PgAioOpData PgAioResult PgAioResultStatus PgAioReturn +PgAioSyncFileTag PgAioTargetData PgAioTargetID PgAioTargetInfo -- 2.47.3