agora inbox for [email protected]
help / color / mirror / Atom feed[PATCH] cfe-10-hint_over_cfe-09-test squash commit
12+ messages / 2 participants
[nested] [flat]
* [PATCH] cfe-10-hint_over_cfe-09-test squash commit
@ 2021-04-06 18:23 Bruce Momjian <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Bruce Momjian @ 2021-04-06 18:23 UTC (permalink / raw)
---
doc/src/sgml/config.sgml | 7 +-
src/backend/access/rmgrdesc/xlogdesc.c | 6 +-
src/backend/access/transam/xlog.c | 4 +
src/backend/access/transam/xloginsert.c | 24 +++++
src/backend/replication/logical/decode.c | 1 +
src/backend/storage/buffer/bufmgr.c | 128 +++++++++++++++++------
src/include/access/xlog.h | 14 +--
src/include/access/xloginsert.h | 2 +
src/include/catalog/pg_control.h | 1 +
9 files changed, 143 insertions(+), 44 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index d7b6e368d0..6bcd9c64df 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -3093,9 +3093,10 @@ include_dir 'conf.d'
</para>
<para>
- If data checksums are enabled, hint bit updates are always WAL-logged
- and this setting is ignored. You can use this setting to test how much
- extra WAL-logging would occur if your database had data checksums
+ If data checksums or cluster file encryption is enabled,
+ hint bit updates are always WAL-logged and this setting is
+ ignored. You can use this setting to test how much extra
+ WAL-logging would occur if your database had data checksums
enabled.
</para>
diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c
index e6090a9dad..d5474990d6 100644
--- a/src/backend/access/rmgrdesc/xlogdesc.c
+++ b/src/backend/access/rmgrdesc/xlogdesc.c
@@ -80,7 +80,8 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
appendStringInfoString(buf, xlrec->rp_name);
}
- else if (info == XLOG_FPI || info == XLOG_FPI_FOR_HINT)
+ else if (info == XLOG_FPI || info == XLOG_FPI_FOR_HINT ||
+ info == XLOG_ENCRYPTION_LSN)
{
/* no further information to print */
}
@@ -184,6 +185,9 @@ xlog_identify(uint8 info)
case XLOG_FPI_FOR_HINT:
id = "FPI_FOR_HINT";
break;
+ case XLOG_ENCRYPTION_LSN:
+ id = "ENCRYPTION_LSN";
+ break;
}
return id;
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 51165f6593..dba690bc5e 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -10353,6 +10353,10 @@ xlog_redo(XLogReaderState *record)
UnlockReleaseBuffer(buffer);
}
}
+ else if (info == XLOG_ENCRYPTION_LSN)
+ {
+ /* nothing to do here */
+ }
else if (info == XLOG_BACKUP_END)
{
XLogRecPtr startpoint;
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 7052dc245e..c76ed1f881 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -980,6 +980,30 @@ XLogSaveBufferForHint(Buffer buffer, bool buffer_std)
return recptr;
}
+/*
+ * This function returns either a WAL or fake LSN, for use by encryption.
+ */
+XLogRecPtr
+LSNForEncryption(bool use_wal_lsn)
+{
+ if (use_wal_lsn)
+ {
+ int dummy = 0;
+
+ Assert(FileEncryptionEnabled);
+ /*
+ * Records other than SWITCH_WAL must have content. We use an integer 0 to
+ * follow the restriction.
+ */
+ XLogBeginInsert();
+ XLogSetRecordFlags(XLOG_MARK_UNIMPORTANT);
+ XLogRegisterData((char *) &dummy, sizeof(dummy));
+ return XLogInsert(RM_XLOG_ID, XLOG_ENCRYPTION_LSN);
+ }
+ else
+ return GetFakeLSNForUnloggedRel();
+}
+
/*
* Write a WAL record containing a full image of a page. Caller is responsible
* for writing the page to disk after calling this routine.
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 97be4b0f23..48400fde2a 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -224,6 +224,7 @@ DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
case XLOG_FPW_CHANGE:
case XLOG_FPI_FOR_HINT:
case XLOG_FPI:
+ case XLOG_ENCRYPTION_LSN:
break;
default:
elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 852138f9c9..367fed5e85 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -3766,11 +3766,12 @@ IncrBufferRefCount(Buffer buffer)
* This is essentially the same as MarkBufferDirty, except:
*
* 1. The caller does not write WAL; so if checksums are enabled, we may need
- * to write an XLOG_FPI_FOR_HINT WAL record to protect against torn pages.
+ * to write an XLOG_FPI_FOR_HINT record to protect against torn pages, or
+ * XLOG_ENCRYPTION_LSN to generate a new LSN for the page.
* 2. The caller might have only share-lock instead of exclusive-lock on the
- * buffer's content lock.
+ * buffer's content lock.
* 3. This function does not guarantee that the buffer is always marked dirty
- * (due to a race condition), so it cannot be used for important changes.
+ * (due to a race condition), so it cannot be used for important changes.
*/
void
MarkBufferDirtyHint(Buffer buffer, bool buffer_std)
@@ -3816,51 +3817,110 @@ MarkBufferDirtyHint(Buffer buffer, bool buffer_std)
* If we need to protect hint bit updates from torn writes, WAL-log a
* full page image of the page. This full page image is only necessary
* if the hint bit update is the first change to the page since the
- * last checkpoint.
+ * last checkpoint. If cluster file encryption is enabled, we also
+ * need to generate new page LSNs for all other cases of page writes.
*
* We don't check full_page_writes here because that logic is included
* when we call XLogInsert() since the value changes dynamically.
*/
- if (XLogHintBitIsNeeded() &&
- (pg_atomic_read_u32(&bufHdr->state) & BM_PERMANENT))
+ if (XLogHintBitIsNeeded())
{
/*
- * If we must not write WAL, due to a relfilenode-specific
- * condition or being in recovery, don't dirty the page. We can
- * set the hint, just not dirty the page as a result so the hint
- * is lost when we evict the page or shutdown.
+ * If we must not write WAL during recovery so don't dirty the page.
+ * We can set the hint, just not dirty the page as a result so the
+ * hint is lost when we evict the page or shutdown.
*
* See src/backend/storage/page/README for longer discussion.
+ * XXX Can this be improved?
*/
- if (RecoveryInProgress() ||
- RelFileNodeSkippingWAL(bufHdr->tag.rnode))
+ if (RecoveryInProgress() ||
+ (RelFileNodeSkippingWAL(bufHdr->tag.rnode) &&
+ !FileEncryptionEnabled))
return;
/*
- * If the block is already dirty because we either made a change
- * or set a hint already, then we don't need to write a full page
- * image. Note that aggressive cleaning of blocks dirtied by hint
- * bit setting would increase the call rate. Bulk setting of hint
- * bits would reduce the call rate...
- *
- * We must issue the WAL record before we mark the buffer dirty.
- * Otherwise we might write the page before we write the WAL. That
- * causes a race condition, since a checkpoint might occur between
- * writing the WAL record and marking the buffer dirty. We solve
- * that with a kluge, but one that is already in use during
- * transaction commit to prevent race conditions. Basically, we
- * simply prevent the checkpoint WAL record from being written
- * until we have marked the buffer dirty. We don't start the
- * checkpoint flush until we have marked dirty, so our checkpoint
- * must flush the change to disk successfully or the checkpoint
- * never gets written, so crash recovery will fix.
+ * Non-BM_PERMANENT objects don't need full page images because
+ * they are not restored. WAL-skipped relfilenodes should never
+ * have full page images generated.
+ */
+ if (pg_atomic_read_u32(&bufHdr->state) & BM_PERMANENT &&
+ !RelFileNodeSkippingWAL(bufHdr->tag.rnode))
+ {
+ /*
+ * If the block is already dirty because we either made a change
+ * or set a hint already, then we don't need to write a full
+ * page image. Note that aggressive cleaning of blocks dirtied
+ * by hint bit setting would increase the call rate. Bulk
+ * setting of hint bits would reduce the call rate...
+ *
+ * We must issue the WAL record before we mark the buffer
+ * dirty. Otherwise we might write the page before we write
+ * the WAL. That causes a race condition, since a checkpoint
+ * might occur between writing the WAL record and marking the
+ * buffer dirty. We solve that with a kluge, but one that is
+ * already in use during transaction commit to prevent race
+ * conditions. Basically, we simply prevent the checkpoint WAL
+ * record from being written until we have marked the buffer
+ * dirty. We don't start the checkpoint flush until we have
+ * marked dirty, so our checkpoint must flush the change to disk
+ * successfully or the checkpoint never gets written, so crash
+ * recovery will fix.
+ *
+ * It's possible we may enter here without an xid, so it
+ * is essential that CreateCheckpoint waits for virtual
+ * transactions rather than full transactionids.
+ */
+ MyProc->delayChkpt = delayChkpt = true;
+ lsn = XLogSaveBufferForHint(buffer, buffer_std);
+ }
+
+ /*
+ * Above, for hint bit changes, we might have generated a new page
+ * LSN and a full-page WAL record for a page's first-clean-to-dirty
+ * during a checkpoint for permanent, non-WAL-skipped relfilenodes.
+ * If we didn't (the lsn variable is invalid), and we are
+ * doing cluster file encryption, we must generate a new
+ * page LSN here for either non-permanent relations or page
+ * non-first-clean-to-dirty during a checkpoint. (Cluster file
+ * encryption does not support WAL-skip relfilenodes.) We must
+ * update the page LSN even if the page with the hint bit change is
+ * later overwritten in the file system with an earlier version of
+ * the page during crash recovery.
*
- * It's possible we may enter here without an xid, so it is
- * essential that CreateCheckpoint waits for virtual transactions
- * rather than full transactionids.
+ * XXX Can we rely on the full page write above with no lock being
+ * held to avoid torn pages? Above, the LSN and page image are
+ * tied together, but here is just the page LSN update.
*/
- MyProc->delayChkpt = delayChkpt = true;
- lsn = XLogSaveBufferForHint(buffer, buffer_std);
+ if (XLogRecPtrIsInvalid(lsn) && FileEncryptionEnabled)
+ {
+ /*
+ * For cluster file encryption we need a new page LSN because
+ * the LSN is used, with the page number and permanent flag, as
+ * part of the nonce, and the nonce must be unique for every
+ * page write. If we reencrypt a page with hint bit changes
+ * using the same nonce as previous writes, it would expose the
+ * hint bit change locations. To avoid this, we write a simple
+ * WAL record to advance the lsn, which can then be assigned to
+ * the page below.
+ *
+ * Above we are relying on the full page writes to revert
+ * any partial pages writes caused by this LSN change for
+ * permanent, non-WAL-skip relfilenodes. For non-permanent
+ * relations, they crash recover as empty. For WAL-skip
+ * relfilenodes, they recover with their original contents, so
+ * that works too.
+ */
+ /* XXX Do we need the checkpoint delay here? */
+ MyProc->delayChkpt = delayChkpt = true;
+ /*
+ * XXX We probably don't need to replay this WAL on the primary
+ * since the full page image is restored, but do we have
+ * to replay this on the repicas (for relations that are
+ * replicated)?
+ */
+ lsn = LSNForEncryption(
+ pg_atomic_read_u32(&bufHdr->state) & BM_PERMANENT);
+ }
}
buf_state = LockBufHdr(bufHdr);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index ab66a660a5..52f5a42f50 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -203,13 +203,15 @@ extern PGDLLIMPORT int wal_level;
/*
* Is a full-page image needed for hint bit updates?
*
- * Normally, we don't WAL-log hint bit updates, but if checksums are enabled,
- * we have to protect them against torn page writes. When you only set
- * individual bits on a page, it's still consistent no matter what combination
- * of the bits make it to disk, but the checksum wouldn't match. Also WAL-log
- * them if forced by wal_log_hints=on.
+ * Normally, we don't WAL-log hint bit updates, but if checksums or encryption
+ * is enabled, we have to protect them against torn page writes. When you
+ * only set individual bits on a page, it's still consistent no matter what
+ * combination of the bits make it to disk, but the checksum wouldn't match.
+ * Cluster file encryption requires a new LSN for hint bit changes, and can't
+ * tolerate torn pages. Also WAL-log them if forced by wal_log_hints=on.
*/
-#define XLogHintBitIsNeeded() (DataChecksumsEnabled() || wal_log_hints)
+#define XLogHintBitIsNeeded() \
+ (DataChecksumsEnabled() || FileEncryptionEnabled || wal_log_hints)
/* Do we need to WAL-log information required only for Hot Standby and logical replication? */
#define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA)
diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index f1d8c39edf..e7ee674dd4 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -61,6 +61,8 @@ extern void log_newpage_range(Relation rel, ForkNumber forkNum,
BlockNumber startblk, BlockNumber endblk, bool page_std);
extern XLogRecPtr XLogSaveBufferForHint(Buffer buffer, bool buffer_std);
+extern XLogRecPtr LSNForEncryption(bool use_wal_lsn);
+
extern void InitXLogInsert(void);
#endif /* XLOGINSERT_H */
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index 6216d1faf5..e23693da4f 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -76,6 +76,7 @@ typedef struct CheckPoint
#define XLOG_END_OF_RECOVERY 0x90
#define XLOG_FPI_FOR_HINT 0xA0
#define XLOG_FPI 0xB0
+#define XLOG_ENCRYPTION_LSN 0xC0
/*
--
2.20.1
--gatW/ieO32f1wygP
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment; filename="cfe-11-gist_over_cfe-10-hint.diff"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.5 22/30] aio: Implement smgr/md/fd write support
@ 2025-02-26 16:26 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-02-26 16:26 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 2 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 189 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 22 +++
7 files changed, 252 insertions(+)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index f44f3b939d6..e2fd896646e 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -113,6 +113,8 @@ extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t o
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index bf714a8896d..7b28c3d482c 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
struct PgAioHandleCallbacks;
extern const struct PgAioHandleCallbacks aio_md_readv_cb;
+extern const struct PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(struct PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(struct PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 85562fa0cc0..86fa07b110f 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -110,6 +110,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(struct PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index ff7b2f25c99..f0414085eba 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 996458fac71..a9c90bc4e59 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2346,6 +2346,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(struct PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index d01ae5f6a09..4c0e5371875 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const struct PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const struct PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1107,6 +1114,56 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1495,6 +1552,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2016,3 +2107,101 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
);
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ md_writev_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * The smgr API operates in blocks, therefore convert the result from
+ * bytes to blocks.
+ */
+ result.result /= BLCKSZ;
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ md_writev_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != ARS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = ARS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index ea8b42ee4e8..6cbf8939fdc 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -102,6 +102,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (struct PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -129,6 +134,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -691,6 +697,22 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
buffers, nblocks, skipFsync);
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO with the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ */
+void
+smgrstartwritev(struct PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--abvteypvk35ocehs
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.5-0023-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.7 27/35] aio: Implement smgr/md/fd write support
@ 2025-03-09 23:35 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-09 23:35 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 189 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 22 +++
7 files changed, 251 insertions(+)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index f76f74ba166..fb6ac058a09 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index be8c4c2d60d..65abebefbfb 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2346,6 +2346,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 11d4d5a7aea..a61fc14805e 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1107,6 +1114,56 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1495,6 +1552,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2017,3 +2108,101 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * The smgr API operates in blocks, therefore convert the result from
+ * bytes to blocks.
+ */
+ result.result /= BLCKSZ;
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != ARS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = ARS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 607c14ee173..088b189543b 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -102,6 +102,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -129,6 +134,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -691,6 +697,22 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
buffers, nblocks, skipFsync);
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO with the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--vsphh7g5lukufvxs
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.7-0028-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.8 29/38] aio: Implement smgr/md/fd write support
@ 2025-03-09 23:35 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-09 23:35 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 189 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 22 +++
7 files changed, 251 insertions(+)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index f76f74ba166..fb6ac058a09 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 34588ed5167..46cbcb2b688 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2346,6 +2346,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 11d4d5a7aea..a61fc14805e 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1107,6 +1114,56 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1495,6 +1552,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2017,3 +2108,101 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * The smgr API operates in blocks, therefore convert the result from
+ * bytes to blocks.
+ */
+ result.result /= BLCKSZ;
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != ARS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = ARS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 53fd6af4799..cd515954e74 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -112,6 +112,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -139,6 +144,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -769,6 +775,22 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO with the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--ow5flh3n247znjrs
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.8-0030-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.6 26/34] aio: Implement smgr/md/fd write support
@ 2025-03-09 23:35 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-09 23:35 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 189 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 22 +++
7 files changed, 251 insertions(+)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index f76f74ba166..fb6ac058a09 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index be8c4c2d60d..65abebefbfb 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2346,6 +2346,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 11d4d5a7aea..a61fc14805e 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1107,6 +1114,56 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1495,6 +1552,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2017,3 +2108,101 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * The smgr API operates in blocks, therefore convert the result from
+ * bytes to blocks.
+ */
+ result.result /= BLCKSZ;
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != ARS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = ARS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 607c14ee173..088b189543b 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -102,6 +102,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -129,6 +134,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -691,6 +697,22 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
buffers, nblocks, skipFsync);
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO with the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--6g5xner6ro2tsnwz
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.6-0027-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.9 21/30] aio: Implement smgr/md/fd write support
@ 2025-03-15 16:29 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-15 16:29 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 189 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 22 +++
7 files changed, 251 insertions(+)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index 2537b1914d8..d1ea1ee145b 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 34588ed5167..46cbcb2b688 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2346,6 +2346,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 11d4d5a7aea..a61fc14805e 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1107,6 +1114,56 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1495,6 +1552,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2017,3 +2108,101 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * The smgr API operates in blocks, therefore convert the result from
+ * bytes to blocks.
+ */
+ result.result /= BLCKSZ;
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != ARS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = ARS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 06b2acf8fea..86be5cb61bc 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -112,6 +112,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -139,6 +144,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -769,6 +775,22 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO with the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--23jbdfobqrqxnmx5
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.9-0022-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.14 20/29] aio: Implement smgr/md/fd write support
@ 2025-03-18 18:40 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw)
TODO:
- Right now the sync.c integration with smgr.c/md.c isn't properly safe to use
in a critical section
The only reason it doesn't immediately fail is that it's reasonably rare
that RegisterSyncRequest() fails *and* either:
- smgropen()->hash_search(HASH_ENTER) decides to resize the hash table, even
though the lookup is guaranteed to succeed for io_method=worker.
- an io_method=uring completion is run in a different backend and smgropen()
needs to build a new entry and thus needs to allocate memory
For a bit I thought this could be worked around easily enough by not doing
an smgropen() in mdsyncfiletag(), or adding a "fallible" smgropen() and
instead just opening the file directly. That actually does kinda solve the
problem, but only because the memory allocation in PathNameOpenFile()
uses malloc(), not palloc() and thus doesn't trigger
- temp_file_limit implementation
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 201 ++++++++++++++++++++++++-
src/backend/storage/smgr/smgr.c | 29 ++++
7 files changed, 269 insertions(+), 1 deletion(-)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index c0063d6950e..67026f5f6b4 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 0e8299dd556..2138d47dab9 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2348,6 +2348,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_start_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index eedba1e5794..f6bdc7e37bc 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1115,6 +1122,64 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+ int ret;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ ret = FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+ if (ret != 0)
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not start writing blocks %u..%u in file \"%s\": %m",
+ blocknum,
+ blocknum + nblocks_this_segment - 1,
+ FilePathName(v->mdfd_vfd))));
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1503,6 +1568,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -1997,7 +2096,7 @@ md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
* AIO error reporting callback for mdstartreadv().
*
* Errors are encoded as follows:
- * - PgAioResult.error_data != 0 encodes IO that failed with errno != 0
+ * - PgAioResult.error_data != 0 encodes IO that failed with that errno
* - PgAioResult.error_data == 0 encodes IO that didn't read all data
*/
static void
@@ -2037,3 +2136,103 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * As explained above smgrstartwritev(), the smgr API operates on the
+ * level of blocks, rather than bytes. Convert.
+ */
+ result.result /= BLCKSZ;
+
+ Assert(result.result <= td->smgr.nblocks);
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != PGAIO_RS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = PGAIO_RS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index bf8f57b410a..e56d6a2597b 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -115,6 +115,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -142,6 +147,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -787,6 +793,29 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO using the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ *
+ * Completion callbacks above smgr will be passed the result as the number of
+ * successfully written blocks if the write [partially] succeeds. This
+ * maintains the abstraction that smgr operates on the level of blocks, rather
+ * than bytes.
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ HOLD_INTERRUPTS();
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+ RESUME_INTERRUPTS();
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--m2na7lgr3zfazgom
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.14-0021-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.10 18/28] aio: Implement smgr/md/fd write support
@ 2025-03-18 18:40 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 189 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 22 +++
7 files changed, 251 insertions(+)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index 2537b1914d8..d1ea1ee145b 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 34588ed5167..46cbcb2b688 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2346,6 +2346,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 11d4d5a7aea..a61fc14805e 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1107,6 +1114,56 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1495,6 +1552,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2017,3 +2108,101 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * The smgr API operates in blocks, therefore convert the result from
+ * bytes to blocks.
+ */
+ result.result /= BLCKSZ;
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = ARS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != ARS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = ARS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 70bf6fd824a..51eaea69550 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -115,6 +115,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -142,6 +147,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -772,6 +778,22 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO with the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--w6dfit2y42fwvotd
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.10-0019-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.11 19/27] aio: Implement smgr/md/fd write support
@ 2025-03-18 18:40 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 199 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 29 ++++
7 files changed, 268 insertions(+)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index 3cdd2a6b195..03e44b76075 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index f573aa8e7ca..8f143c10d36 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2348,6 +2348,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 2218f17e7a0..9d3dec2710c 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1115,6 +1122,64 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+ int ret;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ ret = FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+ if (ret != 0)
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not start writing blocks %u..%u in file \"%s\": %m",
+ blocknum,
+ blocknum + nblocks_this_segment - 1,
+ FilePathName(v->mdfd_vfd))));
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1503,6 +1568,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2027,3 +2126,103 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * As explained above smgrstartwritev(), the smgr API operates on the
+ * level of blocks, rather than bytes. Convert.
+ */
+ result.result /= BLCKSZ;
+
+ Assert(result.result <= td->smgr.nblocks);
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != PGAIO_RS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = PGAIO_RS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 545888dcdfc..aec2c0c565a 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -115,6 +115,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -142,6 +147,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -787,6 +793,29 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO using the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ *
+ * Completion callbacks above smgr will be passed the result as the number of
+ * successfully written blocks if the write [partially] succeeds. This
+ * maintains the abstraction that smgr operates on the level of blocks, rather
+ * than bytes.
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ HOLD_INTERRUPTS();
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+ RESUME_INTERRUPTS();
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--bjnmbpad43bpmfxt
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.11-0020-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.12 19/28] aio: Implement smgr/md/fd write support
@ 2025-03-18 18:40 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw)
TODO:
- register_dirty_segment_aio() can error out in edge cases
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 201 ++++++++++++++++++++++++-
src/backend/storage/smgr/smgr.c | 29 ++++
7 files changed, 269 insertions(+), 1 deletion(-)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index 566060c12cb..12b0f1ebfa6 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index f573aa8e7ca..8f143c10d36 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2348,6 +2348,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_prep_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 2218f17e7a0..f4a7af201ba 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1115,6 +1122,64 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+ int ret;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ ret = FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+ if (ret != 0)
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not start writing blocks %u..%u in file \"%s\": %m",
+ blocknum,
+ blocknum + nblocks_this_segment - 1,
+ FilePathName(v->mdfd_vfd))));
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1503,6 +1568,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -1987,7 +2086,7 @@ md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
* AIO error reporting callback for mdstartreadv().
*
* Errors are encoded as follows:
- * - PgAioResult.error_data != 0 encodes IO that failed with errno != 0
+ * - PgAioResult.error_data != 0 encodes IO that failed with that errno
* - PgAioResult.error_data == 0 encodes IO that didn't read all data
*/
static void
@@ -2027,3 +2126,103 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * As explained above smgrstartwritev(), the smgr API operates on the
+ * level of blocks, rather than bytes. Convert.
+ */
+ result.result /= BLCKSZ;
+
+ Assert(result.result <= td->smgr.nblocks);
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != PGAIO_RS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = PGAIO_RS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index bf8f57b410a..e56d6a2597b 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -115,6 +115,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -142,6 +147,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -787,6 +793,29 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO using the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ *
+ * Completion callbacks above smgr will be passed the result as the number of
+ * successfully written blocks if the write [partially] succeeds. This
+ * maintains the abstraction that smgr operates on the level of blocks, rather
+ * than bytes.
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ HOLD_INTERRUPTS();
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+ RESUME_INTERRUPTS();
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--5i73spx2p4vwf7fe
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.12-0020-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.13 19/28] aio: Implement smgr/md/fd write support
@ 2025-03-18 18:40 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw)
TODO:
- Right now the sync.c integration with smgr.c/md.c isn't properly safe to use
in a critical section
The only reason it doesn't immediately fail is that it's reasonably rare
that RegisterSyncRequest() fails *and* either:
- smgropen()->hash_search(HASH_ENTER) decides to resize the hash table, even
though the lookup is guaranteed to succeed for io_method=worker.
- an io_method=uring completion is run in a different backend and smgropen()
needs to build a new entry and thus needs to allocate memory
For a bit I thought this could be worked around easily enough by not doing
an smgropen() in mdsyncfiletag(), or adding a "fallible" smgropen() and
instead just opening the file directly. That actually does kinda solve the
problem, but only because the memory allocation in PathNameOpenFile()
uses malloc(), not palloc() and thus doesn't trigger
- temp_file_limit implementation
---
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 201 ++++++++++++++++++++++++-
src/backend/storage/smgr/smgr.c | 29 ++++
7 files changed, 269 insertions(+), 1 deletion(-)
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index c0063d6950e..67026f5f6b4 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 9e1cdf7ddfe..4aecf2cfb97 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2348,6 +2348,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_start_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index eedba1e5794..f6bdc7e37bc 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1115,6 +1122,64 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+ int ret;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ ret = FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+ if (ret != 0)
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not start writing blocks %u..%u in file \"%s\": %m",
+ blocknum,
+ blocknum + nblocks_this_segment - 1,
+ FilePathName(v->mdfd_vfd))));
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1503,6 +1568,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -1997,7 +2096,7 @@ md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
* AIO error reporting callback for mdstartreadv().
*
* Errors are encoded as follows:
- * - PgAioResult.error_data != 0 encodes IO that failed with errno != 0
+ * - PgAioResult.error_data != 0 encodes IO that failed with that errno
* - PgAioResult.error_data == 0 encodes IO that didn't read all data
*/
static void
@@ -2037,3 +2136,103 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * As explained above smgrstartwritev(), the smgr API operates on the
+ * level of blocks, rather than bytes. Convert.
+ */
+ result.result /= BLCKSZ;
+
+ Assert(result.result <= td->smgr.nblocks);
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != PGAIO_RS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = PGAIO_RS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index bf8f57b410a..e56d6a2597b 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -115,6 +115,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -142,6 +147,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -787,6 +793,29 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO using the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ *
+ * Completion callbacks above smgr will be passed the result as the number of
+ * successfully written blocks if the write [partially] succeeds. This
+ * maintains the abstraction that smgr operates on the level of blocks, rather
+ * than bytes.
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ HOLD_INTERRUPTS();
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+ RESUME_INTERRUPTS();
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--pro7bqageygxfsvg
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.13-0020-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH v2.15 09/18] aio: Implement smgr/md/fd write support
@ 2025-03-29 17:28 Andres Freund <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Andres Freund @ 2025-03-29 17:28 UTC (permalink / raw)
TODO:
- Right now the sync.c integration with smgr.c/md.c isn't properly safe to use
in a critical section
The only reason it doesn't immediately fail is that it's reasonably rare
that RegisterSyncRequest() fails *and* either:
- smgropen()->hash_search(HASH_ENTER) decides to resize the hash table, even
though the lookup is guaranteed to succeed for io_method=worker.
- an io_method=uring completion is run in a different backend and smgropen()
needs to build a new entry and thus needs to allocate memory
For a bit I thought this could be worked around easily enough by not doing
an smgropen() in mdsyncfiletag(), or adding a "fallible" smgropen() and
instead just opening the file directly. That actually does kinda solve the
problem, but only because the memory allocation in PathNameOpenFile()
uses malloc(), not palloc() and thus doesn't trigger
- temp_file_limit implementation
---
src/include/storage/aio.h | 1 +
src/include/storage/fd.h | 1 +
src/include/storage/md.h | 5 +
src/include/storage/smgr.h | 5 +
src/backend/storage/aio/aio_callback.c | 1 +
src/backend/storage/file/fd.c | 28 ++++
src/backend/storage/smgr/md.c | 199 +++++++++++++++++++++++++
src/backend/storage/smgr/smgr.c | 29 ++++
8 files changed, 269 insertions(+)
diff --git a/src/include/storage/aio.h b/src/include/storage/aio.h
index 9fe9d9ad9fa..bfe0d93683b 100644
--- a/src/include/storage/aio.h
+++ b/src/include/storage/aio.h
@@ -194,6 +194,7 @@ typedef enum PgAioHandleCallbackID
PGAIO_HCB_INVALID = 0,
PGAIO_HCB_MD_READV,
+ PGAIO_HCB_MD_WRITEV,
PGAIO_HCB_SHARED_BUFFER_READV,
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index b77d8e5e30e..2cc7c5a4761 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -112,6 +112,7 @@ extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event
extern ssize_t FileReadV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileStartReadV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
+extern int FileStartWriteV(struct PgAioHandle *ioh, File file, int iovcnt, off_t offset, uint32 wait_event_info);
extern int FileSync(File file, uint32 wait_event_info);
extern int FileZero(File file, off_t offset, off_t amount, uint32 wait_event_info);
extern int FileFallocate(File file, off_t offset, off_t amount, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 9d7131eff43..47ae6c36c94 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,7 @@
#include "storage/sync.h"
extern const PgAioHandleCallbacks aio_md_readv_cb;
+extern const PgAioHandleCallbacks aio_md_writev_cb;
/* md storage manager functionality */
extern void mdinit(void);
@@ -45,6 +46,10 @@ extern void mdstartreadv(PgAioHandle *ioh,
extern void mdwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks, bool skipFsync);
+extern void mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync);
extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 856ebcda350..f00b3763ac9 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -108,6 +108,11 @@ extern void smgrwritev(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+extern void smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
diff --git a/src/backend/storage/aio/aio_callback.c b/src/backend/storage/aio/aio_callback.c
index bf42778a48c..abe5b191b42 100644
--- a/src/backend/storage/aio/aio_callback.c
+++ b/src/backend/storage/aio/aio_callback.c
@@ -41,6 +41,7 @@ static const PgAioHandleCallbacksEntry aio_handle_cbs[] = {
CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
+ CALLBACK_ENTRY(PGAIO_HCB_MD_WRITEV, aio_md_writev_cb),
CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 0e8299dd556..2138d47dab9 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2348,6 +2348,34 @@ retry:
return returnCode;
}
+int
+FileStartWriteV(PgAioHandle *ioh, File file,
+ int iovcnt, off_t offset,
+ uint32 wait_event_info)
+{
+ int returnCode;
+ Vfd *vfdP;
+
+ Assert(FileIsValid(file));
+
+ DO_DB(elog(LOG, "FileStartWriteV: %d (%s) " INT64_FORMAT " %d",
+ file, VfdCache[file].fileName,
+ (int64) offset,
+ iovcnt));
+
+ returnCode = FileAccess(file);
+ if (returnCode < 0)
+ return returnCode;
+
+ vfdP = &VfdCache[file];
+
+ /* FIXME: think about / reimplement temp_file_limit */
+
+ pgaio_io_start_writev(ioh, vfdP->fd, iovcnt, offset);
+
+ return 0;
+}
+
int
FileSync(File file, uint32 wait_event_info)
{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index bc7d94df870..bacfe5f121b 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -155,12 +155,19 @@ static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum,
static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
static void md_readv_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
+static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data);
+static void md_writev_report(PgAioResult result, const PgAioTargetData *target_data, int elevel);
const PgAioHandleCallbacks aio_md_readv_cb = {
.complete_shared = md_readv_complete,
.report = md_readv_report,
};
+const PgAioHandleCallbacks aio_md_writev_cb = {
+ .complete_shared = md_writev_complete,
+ .report = md_writev_report,
+};
+
static inline int
_mdfd_open_flags(void)
@@ -1143,6 +1150,64 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
}
}
+/*
+ * mdstartwritev() -- Asynchronous version of mdrwritev().
+ */
+void
+mdstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ off_t seekpos;
+ MdfdVec *v;
+ BlockNumber nblocks_this_segment;
+ struct iovec *iov;
+ int iovcnt;
+ int ret;
+
+ v = _mdfd_getseg(reln, forknum, blocknum, false,
+ EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+ seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+ Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+ nblocks_this_segment =
+ Min(nblocks,
+ RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+ if (nblocks_this_segment != nblocks)
+ elog(ERROR, "write crossing segment boundary");
+
+ iovcnt = pgaio_io_get_iovec(ioh, &iov);
+
+ Assert(nblocks <= iovcnt);
+
+ iovcnt = buffers_to_iovec(iov, unconstify(void **, buffers), nblocks_this_segment);
+
+ Assert(iovcnt <= nblocks_this_segment);
+
+ if (!(io_direct_flags & IO_DIRECT_DATA))
+ pgaio_io_set_flag(ioh, PGAIO_HF_BUFFERED);
+
+ pgaio_io_set_target_smgr(ioh,
+ reln,
+ forknum,
+ blocknum,
+ nblocks,
+ skipFsync);
+ pgaio_io_register_callbacks(ioh, PGAIO_HCB_MD_WRITEV, 0);
+
+ ret = FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+ if (ret != 0)
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not start writing blocks %u..%u in file \"%s\": %m",
+ blocknum,
+ blocknum + nblocks_this_segment - 1,
+ FilePathName(v->mdfd_vfd))));
+}
+
/*
* mdwriteback() -- Tell the kernel to write pages back to storage.
@@ -1531,6 +1596,40 @@ register_dirty_segment(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg)
}
}
+/*
+ * Like register_dirty_segment(), except for use by AIO. In the completion
+ * callback we don't have access to the MdfdVec (the completion callback might
+ * be executed in a different backend than the issuing backend), therefore we
+ * have to implement this slightly differently.
+ */
+static void
+register_dirty_segment_aio(RelFileLocator locator, ForkNumber forknum, uint64 segno)
+{
+ FileTag tag;
+
+ INIT_MD_FILETAG(tag, locator, forknum, segno);
+
+ /*
+ * Can't block here waiting for checkpointer to accept our sync request,
+ * as checkpointer might be waiting for this AIO to finish if offloaded to
+ * a worker.
+ */
+ if (!RegisterSyncRequest(&tag, SYNC_REQUEST, false /* retryOnError */ ))
+ {
+ char path[MAXPGPATH];
+
+ ereport(DEBUG1,
+ (errmsg_internal("could not forward fsync request because request queue is full")));
+
+ /* reuse mdsyncfiletag() to avoid duplicating code */
+ if (mdsyncfiletag(&tag, path))
+ ereport(data_sync_elevel(ERROR),
+ (errcode_for_file_access(),
+ errmsg("could not fsync file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* register_unlink_segment() -- Schedule a file to be deleted after next checkpoint
*/
@@ -2065,3 +2164,103 @@ md_readv_report(PgAioResult result, const PgAioTargetData *td, int elevel)
td->smgr.nblocks * (size_t) BLCKSZ));
}
}
+
+/*
+ * AIO completion callback for mdstartwritev().
+ */
+static PgAioResult
+md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result, uint8 cb_data)
+{
+ PgAioTargetData *td = pgaio_io_get_target_data(ioh);
+ PgAioResult result = prior_result;
+
+ if (prior_result.result < 0)
+ {
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ /* For "hard" errors, track the error number in error_data */
+ result.error_data = -prior_result.result;
+ result.result = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ /*
+ * As explained above smgrstartwritev(), the smgr API operates on the
+ * level of blocks, rather than bytes. Convert.
+ */
+ result.result /= BLCKSZ;
+
+ Assert(result.result <= td->smgr.nblocks);
+
+ if (result.result == 0)
+ {
+ /* consider 0 blocks written a failure */
+ result.status = PGAIO_RS_ERROR;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ result.error_data = 0;
+
+ pgaio_result_report(result, td, LOG);
+
+ return result;
+ }
+
+ if (result.status != PGAIO_RS_ERROR &&
+ result.result < td->smgr.nblocks)
+ {
+ /* partial writes should be retried at upper level */
+ result.status = PGAIO_RS_PARTIAL;
+ result.id = PGAIO_HCB_MD_WRITEV;
+ }
+
+ if (!td->smgr.skip_fsync)
+ register_dirty_segment_aio(td->smgr.rlocator, td->smgr.forkNum,
+ td->smgr.blockNum / ((BlockNumber) RELSEG_SIZE));
+
+ return result;
+}
+
+/*
+ * AIO error reporting callback for mdstartwritev().
+ */
+static void
+md_writev_report(PgAioResult result, const PgAioTargetData *td, int elevel)
+{
+ RelPathStr path;
+
+ path = relpathbackend(td->smgr.rlocator,
+ td->smgr.is_temp ? MyProcNumber : INVALID_PROC_NUMBER,
+ td->smgr.forkNum);
+
+ if (result.error_data != 0)
+ {
+ errno = result.error_data; /* for errcode_for_file_access() */
+
+ ereport(elevel,
+ errcode_for_file_access(),
+ errmsg("could not write blocks %u..%u in file \"%s\": %m",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks,
+ path.str)
+ );
+ }
+ else
+ {
+ /*
+ * NB: This will typically only be output in debug messages, while
+ * retrying a partial IO.
+ */
+ ereport(elevel,
+ errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not write blocks %u..%u in file \"%s\": wrote only %zu of %zu bytes",
+ td->smgr.blockNum,
+ td->smgr.blockNum + td->smgr.nblocks - 1,
+ path.str,
+ result.result * (size_t) BLCKSZ,
+ td->smgr.nblocks * (size_t) BLCKSZ
+ )
+ );
+ }
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index d532fea39d9..85daf686e08 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -115,6 +115,11 @@ typedef struct f_smgr
BlockNumber blocknum,
const void **buffers, BlockNumber nblocks,
bool skipFsync);
+ void (*smgr_startwritev) (PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum,
+ BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks,
+ bool skipFsync);
void (*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
BlockNumber blocknum, BlockNumber nblocks);
BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -142,6 +147,7 @@ static const f_smgr smgrsw[] = {
.smgr_readv = mdreadv,
.smgr_startreadv = mdstartreadv,
.smgr_writev = mdwritev,
+ .smgr_startwritev = mdstartwritev,
.smgr_writeback = mdwriteback,
.smgr_nblocks = mdnblocks,
.smgr_truncate = mdtruncate,
@@ -795,6 +801,29 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
RESUME_INTERRUPTS();
}
+/*
+ * smgrstartwritev() -- asynchronous version of smgrwritev()
+ *
+ * This starts an asynchronous writev IO using the IO handle `ioh`. Other than
+ * `ioh` all parameters are the same as smgrwritev().
+ *
+ * Completion callbacks above smgr will be passed the result as the number of
+ * successfully written blocks if the write [partially] succeeds. This
+ * maintains the abstraction that smgr operates on the level of blocks, rather
+ * than bytes.
+ */
+void
+smgrstartwritev(PgAioHandle *ioh,
+ SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+ const void **buffers, BlockNumber nblocks, bool skipFsync)
+{
+ HOLD_INTERRUPTS();
+ smgrsw[reln->smgr_which].smgr_startwritev(ioh,
+ reln, forknum, blocknum, buffers,
+ nblocks, skipFsync);
+ RESUME_INTERRUPTS();
+}
+
/*
* smgrwriteback() -- Trigger kernel writeback for the supplied range of
* blocks.
--
2.48.1.76.g4e746b1a31.dirty
--xevce4sdbnyxplun
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.15-0010-aio-Add-bounce-buffers.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
end of thread, other threads:[~2025-03-29 17:28 UTC | newest]
Thread overview: 12+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-04-06 18:23 [PATCH] cfe-10-hint_over_cfe-09-test squash commit Bruce Momjian <[email protected]>
2025-02-26 16:26 [PATCH v2.5 22/30] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-09 23:35 [PATCH v2.7 27/35] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-09 23:35 [PATCH v2.8 29/38] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-09 23:35 [PATCH v2.6 26/34] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-15 16:29 [PATCH v2.9 21/30] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-18 18:40 [PATCH v2.14 20/29] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-18 18:40 [PATCH v2.10 18/28] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-18 18:40 [PATCH v2.11 19/27] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-18 18:40 [PATCH v2.12 19/28] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-18 18:40 [PATCH v2.13 19/28] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
2025-03-29 17:28 [PATCH v2.15 09/18] aio: Implement smgr/md/fd write support Andres Freund <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox