agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH v1] Add pg_get_query_def() to deparse and print a rewritten SQL statement. 4+ messages / 2 participants [nested] [flat]
* [PATCH v1] Add pg_get_query_def() to deparse and print a rewritten SQL statement. @ 2021-06-27 03:39 Julien Rouhaud <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Julien Rouhaud @ 2021-06-27 03:39 UTC (permalink / raw) --- src/backend/rewrite/rewriteHandler.c | 2 + src/backend/utils/adt/ruleutils.c | 70 ++++++++++++++++++++++++++++ src/include/catalog/pg_proc.dat | 3 ++ src/test/regress/expected/rules.out | 26 +++++++++++ src/test/regress/sql/rules.sql | 3 ++ 5 files changed, 104 insertions(+) diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c index 88a9e95e33..f01b4531bf 100644 --- a/src/backend/rewrite/rewriteHandler.c +++ b/src/backend/rewrite/rewriteHandler.c @@ -1815,6 +1815,8 @@ ApplyRetrieveRule(Query *parsetree, rte->rtekind = RTE_SUBQUERY; rte->subquery = rule_action; rte->security_barrier = RelationIsSecurityView(relation); + if (!rte->alias) + rte->alias = makeAlias(get_rel_name(rte->relid), NULL); /* Clear fields that should not be set in a subquery RTE */ rte->relid = InvalidOid; rte->relkind = 0; diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 3719755a0d..cd39f4ce75 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -49,6 +49,8 @@ #include "nodes/nodeFuncs.h" #include "nodes/pathnodes.h" #include "optimizer/optimizer.h" +#include "parser/analyze.h" +#include "parser/parse_node.h" #include "parser/parse_agg.h" #include "parser/parse_func.h" #include "parser/parse_node.h" @@ -58,6 +60,7 @@ #include "rewrite/rewriteHandler.h" #include "rewrite/rewriteManip.h" #include "rewrite/rewriteSupport.h" +#include "tcop/tcopprot.h" #include "utils/array.h" #include "utils/builtins.h" #include "utils/fmgroids.h" @@ -489,6 +492,73 @@ static void get_reloptions(StringInfo buf, Datum reloptions); #define only_marker(rte) ((rte)->inh ? "" : "ONLY ") +/* return the query as postgres will rewrite */ +Datum +pg_get_query_def(PG_FUNCTION_ARGS) +{ + char *sql = TextDatumGetCString(PG_GETARG_TEXT_PP(0)); + List *parsetree_list; + List *querytree_list; + RawStmt *parsetree; + Query *query; + bool snapshot_set = false; + StringInfoData buf; + StringInfoData res; + ListCell *lc; + + if (strcmp(sql, "") == 0) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("empty statement now allowed"))); + + parsetree_list = pg_parse_query(sql); + + /* only support one statement at a time */ + if (list_length(parsetree_list) != 1) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("a single statement should be provided"))); + + initStringInfo(&res); + + parsetree = linitial_node(RawStmt, parsetree_list); + + /* + * Set up a snapshot if parse analysis/planning will need one. + */ + if (analyze_requires_snapshot(parsetree)) + { + PushActiveSnapshot(GetTransactionSnapshot()); + snapshot_set = true; + } + + querytree_list = pg_analyze_and_rewrite(parsetree, sql, + NULL, 0, NULL); + + /* Done with the snapshot used for parsing/planning */ + if (snapshot_set) + PopActiveSnapshot(); + + foreach(lc, querytree_list) + { + query = (Query *) lfirst(lc); + initStringInfo(&buf); + + if (query->utilityStmt) + appendStringInfo(&res, "%s;\n", sql); + else + { + get_query_def(query, &buf, NIL, NULL, + PRETTYFLAG_INDENT, + WRAP_COLUMN_DEFAULT, 0); + + appendStringInfo(&res, "%s;\n", buf.data); + } + } + pfree(buf.data); + + PG_RETURN_TEXT_P(string_to_text(res.data)); +} /* ---------- * pg_get_ruledef - Do it all and return a text diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index fde251fa4f..b15b40aa7a 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -3650,6 +3650,9 @@ proargtypes => 'oid oid', prosrc => 'oidge' }, # System-view support functions +{ oid => '9246', descr => 'show a query as rewritten', + proname => 'pg_get_query_def', provolatile => 'v', prorettype => 'text', + proargtypes => 'text', prosrc => 'pg_get_query_def' }, { oid => '1573', descr => 'source text of a rule', proname => 'pg_get_ruledef', provolatile => 's', prorettype => 'text', proargtypes => 'oid', prosrc => 'pg_get_ruledef' }, diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index e5ab11275d..dd88b890f4 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -3075,6 +3075,32 @@ select pg_get_viewdef('shoe'::regclass,0) as prettier; WHERE sh.slunit = un.un_name; (1 row) +-- test pg_get_query_def() +SELECT pg_get_query_def('SELECT * FROM shoe') as def; + def +-------------------------------------------------------- + SELECT shoename, + + sh_avail, + + slcolor, + + slminlen, + + slminlen_cm, + + slmaxlen, + + slmaxlen_cm, + + slunit + + FROM ( SELECT sh.shoename, + + sh.sh_avail, + + sh.slcolor, + + sh.slminlen, + + (sh.slminlen * un.un_fact) AS slminlen_cm,+ + sh.slmaxlen, + + (sh.slmaxlen * un.un_fact) AS slmaxlen_cm,+ + sh.slunit + + FROM shoe_data sh, + + unit un + + WHERE (sh.slunit = un.un_name)) shoe; + + +(1 row) + -- -- check multi-row VALUES in rules -- diff --git a/src/test/regress/sql/rules.sql b/src/test/regress/sql/rules.sql index 6ec37c4381..aeac713680 100644 --- a/src/test/regress/sql/rules.sql +++ b/src/test/regress/sql/rules.sql @@ -998,6 +998,9 @@ select pg_get_viewdef('shoe'::regclass) as unpretty; select pg_get_viewdef('shoe'::regclass,true) as pretty; select pg_get_viewdef('shoe'::regclass,0) as prettier; +-- test pg_get_query_def() +SELECT pg_get_query_def('SELECT * FROM shoe') as def; + -- -- check multi-row VALUES in rules -- -- 2.31.1 --2e5tvq4ltenpvm6w-- ^ permalink raw reply [nested|flat] 4+ messages in thread
* [PATCH v2.0 10/17] aio: Implement smgr/md.c aio methods @ 2024-09-01 02:33 Andres Freund <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Andres Freund @ 2024-09-01 02:33 UTC (permalink / raw) --- src/include/storage/aio.h | 17 +- src/include/storage/fd.h | 6 + src/include/storage/md.h | 12 ++ src/include/storage/smgr.h | 21 +++ src/backend/storage/aio/aio_subject.c | 4 + src/backend/storage/file/fd.c | 68 ++++++++ src/backend/storage/smgr/md.c | 217 ++++++++++++++++++++++++++ src/backend/storage/smgr/smgr.c | 91 +++++++++++ 8 files changed, 434 insertions(+), 2 deletions(-) diff --git a/src/include/storage/aio.h b/src/include/storage/aio.h index 65052462b02..acfd50c587c 100644 --- a/src/include/storage/aio.h +++ b/src/include/storage/aio.h @@ -57,9 +57,10 @@ typedef enum PgAioSubjectID { /* intentionally the zero value, to help catch zeroed memory etc */ ASI_INVALID = 0, + ASI_SMGR, } PgAioSubjectID; -#define ASI_COUNT (ASI_INVALID + 1) +#define ASI_COUNT (ASI_SMGR + 1) /* * Flags for an IO that can be set with pgaio_io_set_flag(). @@ -90,7 +91,8 @@ typedef enum PgAioHandleFlags */ typedef enum PgAioHandleSharedCallbackID { - ASC_PLACEHOLDER /* empty enums are invalid */ , + ASC_MD_READV, + ASC_MD_WRITEV, } PgAioHandleSharedCallbackID; @@ -139,6 +141,17 @@ typedef union typedef union PgAioSubjectData { + struct + { + RelFileLocator rlocator; /* physical relation identifier */ + BlockNumber blockNum; /* blknum relative to begin of reln */ + int nblocks; + ForkNumber forkNum:8; /* don't waste 4 byte for four values */ + bool is_temp; /* proc can be inferred by owning AIO */ + bool release_lock; + int8 mode; + } smgr; + /* just as an example placeholder for later */ struct { diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h index 1456ab383a4..e993e1b671f 100644 --- a/src/include/storage/fd.h +++ b/src/include/storage/fd.h @@ -101,6 +101,8 @@ extern PGDLLIMPORT int max_safe_fds; * prototypes for functions in fd.c */ +struct PgAioHandle; + /* Operations on virtual Files --- equivalent to Unix kernel file ops */ extern File PathNameOpenFile(const char *fileName, int fileFlags); extern File PathNameOpenFilePerm(const char *fileName, int fileFlags, mode_t fileMode); @@ -109,6 +111,10 @@ extern void FileClose(File file); extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event_info); 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 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 b72293c79a5..ede77695853 100644 --- a/src/include/storage/md.h +++ b/src/include/storage/md.h @@ -19,6 +19,10 @@ #include "storage/smgr.h" #include "storage/sync.h" +struct PgAioHandleSharedCallbacks; +extern const struct PgAioHandleSharedCallbacks aio_md_readv_cb; +extern const struct PgAioHandleSharedCallbacks aio_md_writev_cb; + /* md storage manager functionality */ extern void mdinit(void); extern void mdopen(SMgrRelation reln); @@ -36,9 +40,16 @@ extern uint32 mdmaxcombine(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum); extern void mdreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); +extern void mdstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks); 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); @@ -46,6 +57,7 @@ extern void mdtruncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks); extern void mdimmedsync(SMgrRelation reln, ForkNumber forknum); extern void mdregistersync(SMgrRelation reln, ForkNumber forknum); +extern int mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); extern void ForgetDatabaseSyncRequests(Oid dbid); extern void DropRelationFiles(RelFileLocator *delrels, int ndelrels, bool isRedo); diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h index 899d0d681c5..66730bc24fa 100644 --- a/src/include/storage/smgr.h +++ b/src/include/storage/smgr.h @@ -73,6 +73,11 @@ typedef SMgrRelationData *SMgrRelation; #define SmgrIsTemp(smgr) \ RelFileLocatorBackendIsTemp((smgr)->smgr_rlocator) +struct PgAioHandle; +struct PgAioSubjectInfo; + +extern const struct PgAioSubjectInfo aio_smgr_subject_info; + extern void smgrinit(void); extern SMgrRelation smgropen(RelFileLocator rlocator, ProcNumber backend); extern bool smgrexists(SMgrRelation reln, ForkNumber forknum); @@ -97,10 +102,19 @@ extern uint32 smgrmaxcombine(SMgrRelation reln, ForkNumber forknum, extern void smgrreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); +extern void smgrstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, + BlockNumber blocknum, + void **buffers, BlockNumber nblocks); 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); @@ -109,6 +123,7 @@ extern void smgrtruncate(SMgrRelation reln, ForkNumber *forknum, int nforks, BlockNumber *nblocks); extern void smgrimmedsync(SMgrRelation reln, ForkNumber forknum); extern void smgrregistersync(SMgrRelation reln, ForkNumber forknum); +extern int smgrfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); extern void AtEOXact_SMgr(void); extern bool ProcessBarrierSmgrRelease(void); @@ -126,4 +141,10 @@ smgrwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, smgrwritev(reln, forknum, blocknum, &buffer, 1, skipFsync); } +extern void pgaio_io_set_subject_smgr(struct PgAioHandle *ioh, + SMgrRelationData *smgr, + ForkNumber forknum, + BlockNumber blocknum, + int nblocks); + #endif /* SMGR_H */ diff --git a/src/backend/storage/aio/aio_subject.c b/src/backend/storage/aio/aio_subject.c index 68e9e80074c..12ab1730f49 100644 --- a/src/backend/storage/aio/aio_subject.c +++ b/src/backend/storage/aio/aio_subject.c @@ -20,6 +20,7 @@ #include "storage/aio_internal.h" #include "storage/buf_internals.h" #include "storage/bufmgr.h" +#include "storage/md.h" #include "storage/smgr.h" #include "utils/memutils.h" @@ -28,9 +29,12 @@ static const PgAioSubjectInfo *aio_subject_info[] = { [ASI_INVALID] = &(PgAioSubjectInfo) { .name = "invalid", }, + [ASI_SMGR] = &aio_smgr_subject_info, }; static const PgAioHandleSharedCallbacks *aio_shared_cbs[] = { + [ASC_MD_READV] = &aio_md_readv_cb, + [ASC_MD_WRITEV] = &aio_md_writev_cb, }; diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c index 368cc9455cf..35bf3c1e7bd 100644 --- a/src/backend/storage/file/fd.c +++ b/src/backend/storage/file/fd.c @@ -95,6 +95,7 @@ #include "pgstat.h" #include "portability/mem.h" #include "postmaster/startup.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/ipc.h" #include "utils/guc.h" @@ -1295,6 +1296,8 @@ LruDelete(File file) vfdP = &VfdCache[file]; + pgaio_closing_fd(vfdP->fd); + /* * Close the file. We aren't expecting this to fail; if it does, better * to leak the FD than to mess up our internal state. @@ -1988,6 +1991,8 @@ FileClose(File file) if (!FileIsNotOpen(file)) { + pgaio_closing_fd(vfdP->fd); + /* close the file */ if (close(vfdP->fd) != 0) { @@ -2211,6 +2216,32 @@ retry: return returnCode; } +int +FileStartReadV(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, "FileStartReadV: %d (%s) " INT64_FORMAT " %d", + file, VfdCache[file].fileName, + (int64) offset, + iovcnt)); + + returnCode = FileAccess(file); + if (returnCode < 0) + return returnCode; + + vfdP = &VfdCache[file]; + + pgaio_io_prep_readv(ioh, vfdP->fd, iovcnt, offset); + + return 0; +} + ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info) @@ -2316,6 +2347,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) { @@ -2499,6 +2558,12 @@ FilePathName(File file) int FileGetRawDesc(File file) { + int returnCode; + + returnCode = FileAccess(file); + if (returnCode < 0) + return returnCode; + Assert(FileIsValid(file)); return VfdCache[file].fd; } @@ -2779,6 +2844,7 @@ FreeDesc(AllocateDesc *desc) result = closedir(desc->desc.dir); break; case AllocateDescRawFD: + pgaio_closing_fd(desc->desc.fd); result = close(desc->desc.fd); break; default: @@ -2847,6 +2913,8 @@ CloseTransientFile(int fd) /* Only get here if someone passes us a file not in allocatedDescs */ elog(WARNING, "fd passed to CloseTransientFile was not obtained from OpenTransientFile"); + pgaio_closing_fd(fd); + return close(fd); } diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index 6cd81a61faa..f96308490d9 100644 --- a/src/backend/storage/smgr/md.c +++ b/src/backend/storage/smgr/md.c @@ -31,6 +31,7 @@ #include "miscadmin.h" #include "pg_trace.h" #include "pgstat.h" +#include "storage/aio.h" #include "storage/bufmgr.h" #include "storage/fd.h" #include "storage/md.h" @@ -931,6 +932,49 @@ mdreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, } } +void +mdstartreadv(PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks) +{ + 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, "read crossing segment boundary"); + + iovcnt = pgaio_io_get_iovec(ioh, &iov); + + Assert(nblocks <= iovcnt); + + iovcnt = buffers_to_iovec(iov, buffers, nblocks_this_segment); + + Assert(iovcnt <= nblocks_this_segment); + + pgaio_io_set_subject_smgr(ioh, + reln, + forknum, + blocknum, + nblocks); + pgaio_io_add_shared_cb(ioh, ASC_MD_READV); + + FileStartReadV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_READ); +} + /* * mdwritev() -- Write the supplied blocks at the appropriate location. * @@ -1036,6 +1080,49 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, } } +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); + + pgaio_io_set_subject_smgr(ioh, + reln, + forknum, + blocknum, + nblocks); + pgaio_io_add_shared_cb(ioh, ASC_MD_WRITEV); + + FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE); +} + /* * mdwriteback() -- Tell the kernel to write pages back to storage. @@ -1357,6 +1444,21 @@ mdimmedsync(SMgrRelation reln, ForkNumber forknum) } } +int +mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) +{ + MdfdVec *v = mdopenfork(reln, forknum, EXTENSION_FAIL); + + v = _mdfd_getseg(reln, forknum, blocknum, false, + EXTENSION_FAIL); + + *off = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE)); + + Assert(*off < (off_t) BLCKSZ * RELSEG_SIZE); + + return FileGetRawDesc(v->mdfd_vfd); +} + /* * register_dirty_segment() -- Mark a relation segment as needing fsync * @@ -1832,3 +1934,118 @@ mdfiletagmatches(const FileTag *ftag, const FileTag *candidate) */ return ftag->rlocator.dbOid == candidate->rlocator.dbOid; } + + + +static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result); +static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result); +static void md_readv_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel); + +const struct PgAioHandleSharedCallbacks aio_md_readv_cb = { + .complete = md_readv_complete, + .error = md_readv_error, +}; + +const struct PgAioHandleSharedCallbacks aio_md_writev_cb = { + .complete = md_writev_complete, +}; + +static PgAioResult +md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + PgAioResult result = prior_result; + + elog(DEBUG3, "%s: %d %d", __func__, prior_result.status, prior_result.result); + + if (prior_result.result < 0) + { + result.status = ARS_ERROR; + result.id = ASC_MD_READV; + result.error_data = -prior_result.result; + result.result = 0; + + md_readv_error(result, sd, LOG); + + return result; + } + + result.result /= BLCKSZ; + + if (result.result == 0) + { + /* consider 0 blocks read a failure */ + result.status = ARS_ERROR; + result.id = ASC_MD_READV; + result.error_data = 0; + + md_readv_error(result, sd, LOG); + } + + if (result.status != ARS_ERROR && + result.result < sd->smgr.nblocks) + { + /* partial reads should be retried at upper level */ + result.id = ASC_MD_READV; + result.status = ARS_PARTIAL; + } + + /* AFIXME: post-read portion of mdreadv() */ + + return result; +} + +static void +md_readv_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel) +{ + MemoryContext oldContext = CurrentMemoryContext; + + /* AFIXME: */ + oldContext = MemoryContextSwitchTo(ErrorContext); + + if (result.error_data != 0) + { + errno = result.error_data; /* for errcode_for_file_access() */ + + ereport(elevel, + errcode_for_file_access(), + errmsg("could not read blocks %u..%u in file \"%s\": %m", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum) + ) + ); + } + else + { + ereport(elevel, + errcode(ERRCODE_DATA_CORRUPTED), + errmsg("could not read blocks %u..%u in file \"%s\": read only %zu of %zu bytes", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks - 1, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum), + result.result * (size_t) BLCKSZ, + subject_data->smgr.nblocks * (size_t) BLCKSZ + ) + ); + } + + MemoryContextSwitchTo(oldContext); +} + + +static PgAioResult +md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result) +{ + elog(DEBUG3, "%s: %d %d", __func__, prior_result.status, prior_result.result); + + if (prior_result.status == ARS_ERROR) + { + /* AFIXME: complain */ + return prior_result; + } + + prior_result.result /= BLCKSZ; + + return prior_result; +} diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c index ee31db85eec..2dacb361a4f 100644 --- a/src/backend/storage/smgr/smgr.c +++ b/src/backend/storage/smgr/smgr.c @@ -53,6 +53,7 @@ #include "access/xlogutils.h" #include "lib/ilist.h" +#include "storage/aio.h" #include "storage/bufmgr.h" #include "storage/ipc.h" #include "storage/md.h" @@ -93,10 +94,19 @@ typedef struct f_smgr void (*smgr_readv) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); + void (*smgr_startreadv) (struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, + BlockNumber blocknum, + void **buffers, BlockNumber nblocks); void (*smgr_writev) (SMgrRelation reln, ForkNumber forknum, 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); @@ -104,6 +114,7 @@ typedef struct f_smgr BlockNumber nblocks); void (*smgr_immedsync) (SMgrRelation reln, ForkNumber forknum); void (*smgr_registersync) (SMgrRelation reln, ForkNumber forknum); + int (*smgr_fd) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); } f_smgr; static const f_smgr smgrsw[] = { @@ -121,12 +132,15 @@ static const f_smgr smgrsw[] = { .smgr_prefetch = mdprefetch, .smgr_maxcombine = mdmaxcombine, .smgr_readv = mdreadv, + .smgr_startreadv = mdstartreadv, .smgr_writev = mdwritev, + .smgr_startwritev = mdstartwritev, .smgr_writeback = mdwriteback, .smgr_nblocks = mdnblocks, .smgr_truncate = mdtruncate, .smgr_immedsync = mdimmedsync, .smgr_registersync = mdregistersync, + .smgr_fd = mdfd, } }; @@ -145,6 +159,14 @@ static void smgrshutdown(int code, Datum arg); static void smgrdestroy(SMgrRelation reln); +static void smgr_aio_reopen(PgAioHandle *ioh); + +const struct PgAioSubjectInfo aio_smgr_subject_info = { + .name = "smgr", + .reopen = smgr_aio_reopen, +}; + + /* * smgrinit(), smgrshutdown() -- Initialize or shut down storage * managers. @@ -620,6 +642,19 @@ smgrreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, nblocks); } +/* + * FILL ME IN + */ +void +smgrstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks) +{ + smgrsw[reln->smgr_which].smgr_startreadv(ioh, + reln, forknum, blocknum, buffers, + nblocks); +} + /* * smgrwritev() -- Write the supplied buffers out. * @@ -651,6 +686,16 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, buffers, nblocks, skipFsync); } +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. @@ -807,6 +852,12 @@ smgrimmedsync(SMgrRelation reln, ForkNumber forknum) smgrsw[reln->smgr_which].smgr_immedsync(reln, forknum); } +int +smgrfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) +{ + return smgrsw[reln->smgr_which].smgr_fd(reln, forknum, blocknum, off); +} + /* * AtEOXact_SMgr * @@ -835,3 +886,43 @@ ProcessBarrierSmgrRelease(void) smgrreleaseall(); return true; } + +void +pgaio_io_set_subject_smgr(PgAioHandle *ioh, + struct SMgrRelationData *smgr, + ForkNumber forknum, + BlockNumber blocknum, + int nblocks) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + + pgaio_io_set_subject(ioh, ASI_SMGR); + + /* backend is implied via IO owner */ + sd->smgr.rlocator = smgr->smgr_rlocator.locator; + sd->smgr.forkNum = forknum; + sd->smgr.blockNum = blocknum; + sd->smgr.nblocks = nblocks; + sd->smgr.is_temp = SmgrIsTemp(smgr); + sd->smgr.release_lock = false; + sd->smgr.mode = RBM_NORMAL; +} + +static void +smgr_aio_reopen(PgAioHandle *ioh) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + PgAioOpData *od = pgaio_io_get_op_data(ioh); + SMgrRelation reln; + ProcNumber procno; + uint32 off; + + if (sd->smgr.is_temp) + procno = pgaio_io_get_owner(ioh); + else + procno = INVALID_PROC_NUMBER; + + reln = smgropen(sd->smgr.rlocator, procno); + od->read.fd = smgrfd(reln, sd->smgr.forkNum, sd->smgr.blockNum, &off); + Assert(off == od->read.offset); +} -- 2.45.2.827.g557ae147e6 --ww2auydviafoh7lh Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.0-0011-bufmgr-Implement-AIO-support.patch" ^ permalink raw reply [nested|flat] 4+ messages in thread
* [PATCH v2.1 13/20] aio: Implement smgr/md.c aio methods @ 2024-09-01 02:33 Andres Freund <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Andres Freund @ 2024-09-01 02:33 UTC (permalink / raw) --- src/include/storage/aio.h | 17 +- src/include/storage/fd.h | 6 + src/include/storage/md.h | 12 ++ src/include/storage/smgr.h | 21 +++ src/backend/storage/aio/aio_subject.c | 4 + src/backend/storage/file/fd.c | 68 ++++++++ src/backend/storage/smgr/md.c | 217 ++++++++++++++++++++++++++ src/backend/storage/smgr/smgr.c | 91 +++++++++++ 8 files changed, 434 insertions(+), 2 deletions(-) diff --git a/src/include/storage/aio.h b/src/include/storage/aio.h index b8c743548c9..07bf92a6b7a 100644 --- a/src/include/storage/aio.h +++ b/src/include/storage/aio.h @@ -57,9 +57,10 @@ typedef enum PgAioSubjectID { /* intentionally the zero value, to help catch zeroed memory etc */ ASI_INVALID = 0, + ASI_SMGR, } PgAioSubjectID; -#define ASI_COUNT (ASI_INVALID + 1) +#define ASI_COUNT (ASI_SMGR + 1) /* * Flags for an IO that can be set with pgaio_io_set_flag(). @@ -90,7 +91,8 @@ typedef enum PgAioHandleFlags */ typedef enum PgAioHandleSharedCallbackID { - ASC_PLACEHOLDER /* empty enums are invalid */ , + ASC_MD_READV, + ASC_MD_WRITEV, } PgAioHandleSharedCallbackID; @@ -139,6 +141,17 @@ typedef union typedef union PgAioSubjectData { + struct + { + RelFileLocator rlocator; /* physical relation identifier */ + BlockNumber blockNum; /* blknum relative to begin of reln */ + int nblocks; + ForkNumber forkNum:8; /* don't waste 4 byte for four values */ + bool is_temp; /* proc can be inferred by owning AIO */ + bool release_lock; + int8 mode; + } smgr; + /* just as an example placeholder for later */ struct { diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h index 1456ab383a4..e993e1b671f 100644 --- a/src/include/storage/fd.h +++ b/src/include/storage/fd.h @@ -101,6 +101,8 @@ extern PGDLLIMPORT int max_safe_fds; * prototypes for functions in fd.c */ +struct PgAioHandle; + /* Operations on virtual Files --- equivalent to Unix kernel file ops */ extern File PathNameOpenFile(const char *fileName, int fileFlags); extern File PathNameOpenFilePerm(const char *fileName, int fileFlags, mode_t fileMode); @@ -109,6 +111,10 @@ extern void FileClose(File file); extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event_info); 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 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 b72293c79a5..ede77695853 100644 --- a/src/include/storage/md.h +++ b/src/include/storage/md.h @@ -19,6 +19,10 @@ #include "storage/smgr.h" #include "storage/sync.h" +struct PgAioHandleSharedCallbacks; +extern const struct PgAioHandleSharedCallbacks aio_md_readv_cb; +extern const struct PgAioHandleSharedCallbacks aio_md_writev_cb; + /* md storage manager functionality */ extern void mdinit(void); extern void mdopen(SMgrRelation reln); @@ -36,9 +40,16 @@ extern uint32 mdmaxcombine(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum); extern void mdreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); +extern void mdstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks); 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); @@ -46,6 +57,7 @@ extern void mdtruncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks); extern void mdimmedsync(SMgrRelation reln, ForkNumber forknum); extern void mdregistersync(SMgrRelation reln, ForkNumber forknum); +extern int mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); extern void ForgetDatabaseSyncRequests(Oid dbid); extern void DropRelationFiles(RelFileLocator *delrels, int ndelrels, bool isRedo); diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h index 899d0d681c5..66730bc24fa 100644 --- a/src/include/storage/smgr.h +++ b/src/include/storage/smgr.h @@ -73,6 +73,11 @@ typedef SMgrRelationData *SMgrRelation; #define SmgrIsTemp(smgr) \ RelFileLocatorBackendIsTemp((smgr)->smgr_rlocator) +struct PgAioHandle; +struct PgAioSubjectInfo; + +extern const struct PgAioSubjectInfo aio_smgr_subject_info; + extern void smgrinit(void); extern SMgrRelation smgropen(RelFileLocator rlocator, ProcNumber backend); extern bool smgrexists(SMgrRelation reln, ForkNumber forknum); @@ -97,10 +102,19 @@ extern uint32 smgrmaxcombine(SMgrRelation reln, ForkNumber forknum, extern void smgrreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); +extern void smgrstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, + BlockNumber blocknum, + void **buffers, BlockNumber nblocks); 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); @@ -109,6 +123,7 @@ extern void smgrtruncate(SMgrRelation reln, ForkNumber *forknum, int nforks, BlockNumber *nblocks); extern void smgrimmedsync(SMgrRelation reln, ForkNumber forknum); extern void smgrregistersync(SMgrRelation reln, ForkNumber forknum); +extern int smgrfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); extern void AtEOXact_SMgr(void); extern bool ProcessBarrierSmgrRelease(void); @@ -126,4 +141,10 @@ smgrwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, smgrwritev(reln, forknum, blocknum, &buffer, 1, skipFsync); } +extern void pgaio_io_set_subject_smgr(struct PgAioHandle *ioh, + SMgrRelationData *smgr, + ForkNumber forknum, + BlockNumber blocknum, + int nblocks); + #endif /* SMGR_H */ diff --git a/src/backend/storage/aio/aio_subject.c b/src/backend/storage/aio/aio_subject.c index 51ee3b3969d..14be8432f5a 100644 --- a/src/backend/storage/aio/aio_subject.c +++ b/src/backend/storage/aio/aio_subject.c @@ -20,6 +20,7 @@ #include "storage/aio_internal.h" #include "storage/buf_internals.h" #include "storage/bufmgr.h" +#include "storage/md.h" #include "storage/smgr.h" #include "utils/memutils.h" @@ -28,9 +29,12 @@ static const PgAioSubjectInfo *aio_subject_info[] = { [ASI_INVALID] = &(PgAioSubjectInfo) { .name = "invalid", }, + [ASI_SMGR] = &aio_smgr_subject_info, }; static const PgAioHandleSharedCallbacks *aio_shared_cbs[] = { + [ASC_MD_READV] = &aio_md_readv_cb, + [ASC_MD_WRITEV] = &aio_md_writev_cb, }; diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c index ec1505802b9..f5ff554f946 100644 --- a/src/backend/storage/file/fd.c +++ b/src/backend/storage/file/fd.c @@ -95,6 +95,7 @@ #include "pgstat.h" #include "portability/mem.h" #include "postmaster/startup.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/ipc.h" #include "utils/guc.h" @@ -1295,6 +1296,8 @@ LruDelete(File file) vfdP = &VfdCache[file]; + pgaio_closing_fd(vfdP->fd); + /* * Close the file. We aren't expecting this to fail; if it does, better * to leak the FD than to mess up our internal state. @@ -1988,6 +1991,8 @@ FileClose(File file) if (!FileIsNotOpen(file)) { + pgaio_closing_fd(vfdP->fd); + /* close the file */ if (close(vfdP->fd) != 0) { @@ -2211,6 +2216,32 @@ retry: return returnCode; } +int +FileStartReadV(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, "FileStartReadV: %d (%s) " INT64_FORMAT " %d", + file, VfdCache[file].fileName, + (int64) offset, + iovcnt)); + + returnCode = FileAccess(file); + if (returnCode < 0) + return returnCode; + + vfdP = &VfdCache[file]; + + pgaio_io_prep_readv(ioh, vfdP->fd, iovcnt, offset); + + return 0; +} + ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info) @@ -2316,6 +2347,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) { @@ -2499,6 +2558,12 @@ FilePathName(File file) int FileGetRawDesc(File file) { + int returnCode; + + returnCode = FileAccess(file); + if (returnCode < 0) + return returnCode; + Assert(FileIsValid(file)); return VfdCache[file].fd; } @@ -2779,6 +2844,7 @@ FreeDesc(AllocateDesc *desc) result = closedir(desc->desc.dir); break; case AllocateDescRawFD: + pgaio_closing_fd(desc->desc.fd); result = close(desc->desc.fd); break; default: @@ -2847,6 +2913,8 @@ CloseTransientFile(int fd) /* Only get here if someone passes us a file not in allocatedDescs */ elog(WARNING, "fd passed to CloseTransientFile was not obtained from OpenTransientFile"); + pgaio_closing_fd(fd); + return close(fd); } diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index 6cd81a61faa..f96308490d9 100644 --- a/src/backend/storage/smgr/md.c +++ b/src/backend/storage/smgr/md.c @@ -31,6 +31,7 @@ #include "miscadmin.h" #include "pg_trace.h" #include "pgstat.h" +#include "storage/aio.h" #include "storage/bufmgr.h" #include "storage/fd.h" #include "storage/md.h" @@ -931,6 +932,49 @@ mdreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, } } +void +mdstartreadv(PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks) +{ + 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, "read crossing segment boundary"); + + iovcnt = pgaio_io_get_iovec(ioh, &iov); + + Assert(nblocks <= iovcnt); + + iovcnt = buffers_to_iovec(iov, buffers, nblocks_this_segment); + + Assert(iovcnt <= nblocks_this_segment); + + pgaio_io_set_subject_smgr(ioh, + reln, + forknum, + blocknum, + nblocks); + pgaio_io_add_shared_cb(ioh, ASC_MD_READV); + + FileStartReadV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_READ); +} + /* * mdwritev() -- Write the supplied blocks at the appropriate location. * @@ -1036,6 +1080,49 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, } } +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); + + pgaio_io_set_subject_smgr(ioh, + reln, + forknum, + blocknum, + nblocks); + pgaio_io_add_shared_cb(ioh, ASC_MD_WRITEV); + + FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE); +} + /* * mdwriteback() -- Tell the kernel to write pages back to storage. @@ -1357,6 +1444,21 @@ mdimmedsync(SMgrRelation reln, ForkNumber forknum) } } +int +mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) +{ + MdfdVec *v = mdopenfork(reln, forknum, EXTENSION_FAIL); + + v = _mdfd_getseg(reln, forknum, blocknum, false, + EXTENSION_FAIL); + + *off = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE)); + + Assert(*off < (off_t) BLCKSZ * RELSEG_SIZE); + + return FileGetRawDesc(v->mdfd_vfd); +} + /* * register_dirty_segment() -- Mark a relation segment as needing fsync * @@ -1832,3 +1934,118 @@ mdfiletagmatches(const FileTag *ftag, const FileTag *candidate) */ return ftag->rlocator.dbOid == candidate->rlocator.dbOid; } + + + +static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result); +static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result); +static void md_readv_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel); + +const struct PgAioHandleSharedCallbacks aio_md_readv_cb = { + .complete = md_readv_complete, + .error = md_readv_error, +}; + +const struct PgAioHandleSharedCallbacks aio_md_writev_cb = { + .complete = md_writev_complete, +}; + +static PgAioResult +md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + PgAioResult result = prior_result; + + elog(DEBUG3, "%s: %d %d", __func__, prior_result.status, prior_result.result); + + if (prior_result.result < 0) + { + result.status = ARS_ERROR; + result.id = ASC_MD_READV; + result.error_data = -prior_result.result; + result.result = 0; + + md_readv_error(result, sd, LOG); + + return result; + } + + result.result /= BLCKSZ; + + if (result.result == 0) + { + /* consider 0 blocks read a failure */ + result.status = ARS_ERROR; + result.id = ASC_MD_READV; + result.error_data = 0; + + md_readv_error(result, sd, LOG); + } + + if (result.status != ARS_ERROR && + result.result < sd->smgr.nblocks) + { + /* partial reads should be retried at upper level */ + result.id = ASC_MD_READV; + result.status = ARS_PARTIAL; + } + + /* AFIXME: post-read portion of mdreadv() */ + + return result; +} + +static void +md_readv_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel) +{ + MemoryContext oldContext = CurrentMemoryContext; + + /* AFIXME: */ + oldContext = MemoryContextSwitchTo(ErrorContext); + + if (result.error_data != 0) + { + errno = result.error_data; /* for errcode_for_file_access() */ + + ereport(elevel, + errcode_for_file_access(), + errmsg("could not read blocks %u..%u in file \"%s\": %m", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum) + ) + ); + } + else + { + ereport(elevel, + errcode(ERRCODE_DATA_CORRUPTED), + errmsg("could not read blocks %u..%u in file \"%s\": read only %zu of %zu bytes", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks - 1, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum), + result.result * (size_t) BLCKSZ, + subject_data->smgr.nblocks * (size_t) BLCKSZ + ) + ); + } + + MemoryContextSwitchTo(oldContext); +} + + +static PgAioResult +md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result) +{ + elog(DEBUG3, "%s: %d %d", __func__, prior_result.status, prior_result.result); + + if (prior_result.status == ARS_ERROR) + { + /* AFIXME: complain */ + return prior_result; + } + + prior_result.result /= BLCKSZ; + + return prior_result; +} diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c index ee31db85eec..2dacb361a4f 100644 --- a/src/backend/storage/smgr/smgr.c +++ b/src/backend/storage/smgr/smgr.c @@ -53,6 +53,7 @@ #include "access/xlogutils.h" #include "lib/ilist.h" +#include "storage/aio.h" #include "storage/bufmgr.h" #include "storage/ipc.h" #include "storage/md.h" @@ -93,10 +94,19 @@ typedef struct f_smgr void (*smgr_readv) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); + void (*smgr_startreadv) (struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, + BlockNumber blocknum, + void **buffers, BlockNumber nblocks); void (*smgr_writev) (SMgrRelation reln, ForkNumber forknum, 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); @@ -104,6 +114,7 @@ typedef struct f_smgr BlockNumber nblocks); void (*smgr_immedsync) (SMgrRelation reln, ForkNumber forknum); void (*smgr_registersync) (SMgrRelation reln, ForkNumber forknum); + int (*smgr_fd) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); } f_smgr; static const f_smgr smgrsw[] = { @@ -121,12 +132,15 @@ static const f_smgr smgrsw[] = { .smgr_prefetch = mdprefetch, .smgr_maxcombine = mdmaxcombine, .smgr_readv = mdreadv, + .smgr_startreadv = mdstartreadv, .smgr_writev = mdwritev, + .smgr_startwritev = mdstartwritev, .smgr_writeback = mdwriteback, .smgr_nblocks = mdnblocks, .smgr_truncate = mdtruncate, .smgr_immedsync = mdimmedsync, .smgr_registersync = mdregistersync, + .smgr_fd = mdfd, } }; @@ -145,6 +159,14 @@ static void smgrshutdown(int code, Datum arg); static void smgrdestroy(SMgrRelation reln); +static void smgr_aio_reopen(PgAioHandle *ioh); + +const struct PgAioSubjectInfo aio_smgr_subject_info = { + .name = "smgr", + .reopen = smgr_aio_reopen, +}; + + /* * smgrinit(), smgrshutdown() -- Initialize or shut down storage * managers. @@ -620,6 +642,19 @@ smgrreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, nblocks); } +/* + * FILL ME IN + */ +void +smgrstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks) +{ + smgrsw[reln->smgr_which].smgr_startreadv(ioh, + reln, forknum, blocknum, buffers, + nblocks); +} + /* * smgrwritev() -- Write the supplied buffers out. * @@ -651,6 +686,16 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, buffers, nblocks, skipFsync); } +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. @@ -807,6 +852,12 @@ smgrimmedsync(SMgrRelation reln, ForkNumber forknum) smgrsw[reln->smgr_which].smgr_immedsync(reln, forknum); } +int +smgrfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) +{ + return smgrsw[reln->smgr_which].smgr_fd(reln, forknum, blocknum, off); +} + /* * AtEOXact_SMgr * @@ -835,3 +886,43 @@ ProcessBarrierSmgrRelease(void) smgrreleaseall(); return true; } + +void +pgaio_io_set_subject_smgr(PgAioHandle *ioh, + struct SMgrRelationData *smgr, + ForkNumber forknum, + BlockNumber blocknum, + int nblocks) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + + pgaio_io_set_subject(ioh, ASI_SMGR); + + /* backend is implied via IO owner */ + sd->smgr.rlocator = smgr->smgr_rlocator.locator; + sd->smgr.forkNum = forknum; + sd->smgr.blockNum = blocknum; + sd->smgr.nblocks = nblocks; + sd->smgr.is_temp = SmgrIsTemp(smgr); + sd->smgr.release_lock = false; + sd->smgr.mode = RBM_NORMAL; +} + +static void +smgr_aio_reopen(PgAioHandle *ioh) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + PgAioOpData *od = pgaio_io_get_op_data(ioh); + SMgrRelation reln; + ProcNumber procno; + uint32 off; + + if (sd->smgr.is_temp) + procno = pgaio_io_get_owner(ioh); + else + procno = INVALID_PROC_NUMBER; + + reln = smgropen(sd->smgr.rlocator, procno); + od->read.fd = smgrfd(reln, sd->smgr.forkNum, sd->smgr.blockNum, &off); + Assert(off == od->read.offset); +} -- 2.45.2.827.g557ae147e6 --ggp33qz5xusicvk6 Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.1-0014-bufmgr-Implement-AIO-support.patch" ^ permalink raw reply [nested|flat] 4+ messages in thread
* [PATCH v2 10/20] aio: Implement smgr/md.c aio methods @ 2024-12-15 17:36 Andres Freund <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Andres Freund @ 2024-12-15 17:36 UTC (permalink / raw) --- src/include/storage/aio.h | 17 +- src/include/storage/fd.h | 6 + src/include/storage/md.h | 12 + src/include/storage/smgr.h | 21 ++ src/backend/storage/aio/aio_subject.c | 4 + src/backend/storage/file/fd.c | 68 ++++++ src/backend/storage/smgr/md.c | 314 ++++++++++++++++++++++++++ src/backend/storage/smgr/smgr.c | 91 ++++++++ 8 files changed, 532 insertions(+), 1 deletion(-) diff --git a/src/include/storage/aio.h b/src/include/storage/aio.h index a1633a0ed3d..d693b0b0bd8 100644 --- a/src/include/storage/aio.h +++ b/src/include/storage/aio.h @@ -55,9 +55,10 @@ typedef enum PgAioSubjectID { /* intentionally the zero value, to help catch zeroed memory etc */ ASI_INVALID = 0, + ASI_SMGR, } PgAioSubjectID; -#define ASI_COUNT (ASI_INVALID + 1) +#define ASI_COUNT (ASI_SMGR + 1) /* * Flags for an IO that can be set with pgaio_io_set_flag(). @@ -100,6 +101,9 @@ typedef enum PgAioHandleFlags typedef enum PgAioHandleSharedCallbackID { ASC_INVALID, + + ASC_MD_READV, + ASC_MD_WRITEV, } PgAioHandleSharedCallbackID; @@ -135,6 +139,17 @@ typedef union typedef union PgAioSubjectData { + struct + { + RelFileLocator rlocator; /* physical relation identifier */ + BlockNumber blockNum; /* blknum relative to begin of reln */ + int nblocks; + ForkNumber forkNum:8; /* don't waste 4 byte for four values */ + bool is_temp; /* proc can be inferred by owning AIO */ + bool release_lock; + int8 mode; + } smgr; + /* just as an example placeholder for later */ struct { diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h index 1456ab383a4..e993e1b671f 100644 --- a/src/include/storage/fd.h +++ b/src/include/storage/fd.h @@ -101,6 +101,8 @@ extern PGDLLIMPORT int max_safe_fds; * prototypes for functions in fd.c */ +struct PgAioHandle; + /* Operations on virtual Files --- equivalent to Unix kernel file ops */ extern File PathNameOpenFile(const char *fileName, int fileFlags); extern File PathNameOpenFilePerm(const char *fileName, int fileFlags, mode_t fileMode); @@ -109,6 +111,10 @@ extern void FileClose(File file); extern int FilePrefetch(File file, off_t offset, off_t amount, uint32 wait_event_info); 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 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 e7671dd6c18..c3a18465c6b 100644 --- a/src/include/storage/md.h +++ b/src/include/storage/md.h @@ -19,6 +19,10 @@ #include "storage/smgr.h" #include "storage/sync.h" +struct PgAioHandleSharedCallbacks; +extern const struct PgAioHandleSharedCallbacks aio_md_readv_cb; +extern const struct PgAioHandleSharedCallbacks aio_md_writev_cb; + /* md storage manager functionality */ extern void mdinit(void); extern void mdopen(SMgrRelation reln); @@ -36,9 +40,16 @@ extern uint32 mdmaxcombine(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum); extern void mdreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); +extern void mdstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks); 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); @@ -46,6 +57,7 @@ extern void mdtruncate(SMgrRelation reln, ForkNumber forknum, BlockNumber old_blocks, BlockNumber nblocks); extern void mdimmedsync(SMgrRelation reln, ForkNumber forknum); extern void mdregistersync(SMgrRelation reln, ForkNumber forknum); +extern int mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); extern void ForgetDatabaseSyncRequests(Oid dbid); extern void DropRelationFiles(RelFileLocator *delrels, int ndelrels, bool isRedo); diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h index 63a186bd346..fe23a7f744f 100644 --- a/src/include/storage/smgr.h +++ b/src/include/storage/smgr.h @@ -73,6 +73,11 @@ typedef SMgrRelationData *SMgrRelation; #define SmgrIsTemp(smgr) \ RelFileLocatorBackendIsTemp((smgr)->smgr_rlocator) +struct PgAioHandle; +struct PgAioSubjectInfo; + +extern const struct PgAioSubjectInfo aio_smgr_subject_info; + extern void smgrinit(void); extern SMgrRelation smgropen(RelFileLocator rlocator, ProcNumber backend); extern bool smgrexists(SMgrRelation reln, ForkNumber forknum); @@ -97,10 +102,19 @@ extern uint32 smgrmaxcombine(SMgrRelation reln, ForkNumber forknum, extern void smgrreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); +extern void smgrstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, + BlockNumber blocknum, + void **buffers, BlockNumber nblocks); 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); @@ -110,6 +124,7 @@ extern void smgrtruncate(SMgrRelation reln, ForkNumber *forknum, int nforks, BlockNumber *nblocks); extern void smgrimmedsync(SMgrRelation reln, ForkNumber forknum); extern void smgrregistersync(SMgrRelation reln, ForkNumber forknum); +extern int smgrfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); extern void AtEOXact_SMgr(void); extern bool ProcessBarrierSmgrRelease(void); @@ -127,4 +142,10 @@ smgrwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, smgrwritev(reln, forknum, blocknum, &buffer, 1, skipFsync); } +extern void pgaio_io_set_subject_smgr(struct PgAioHandle *ioh, + SMgrRelationData *smgr, + ForkNumber forknum, + BlockNumber blocknum, + int nblocks); + #endif /* SMGR_H */ diff --git a/src/backend/storage/aio/aio_subject.c b/src/backend/storage/aio/aio_subject.c index 8694cfafcd1..effb09c11c7 100644 --- a/src/backend/storage/aio/aio_subject.c +++ b/src/backend/storage/aio/aio_subject.c @@ -20,6 +20,7 @@ #include "storage/aio_internal.h" #include "storage/buf_internals.h" #include "storage/bufmgr.h" +#include "storage/md.h" #include "storage/smgr.h" #include "utils/memutils.h" @@ -35,6 +36,7 @@ static const PgAioSubjectInfo *aio_subject_info[] = { [ASI_INVALID] = &(PgAioSubjectInfo) { .name = "invalid", }, + [ASI_SMGR] = &aio_smgr_subject_info, }; @@ -46,6 +48,8 @@ typedef struct PgAioHandleSharedCallbacksEntry static const PgAioHandleSharedCallbacksEntry aio_shared_cbs[] = { #define CALLBACK_ENTRY(id, callback) [id] = {.cb = &callback, .name = #callback} + CALLBACK_ENTRY(ASC_MD_READV, aio_md_readv_cb), + CALLBACK_ENTRY(ASC_MD_WRITEV, aio_md_writev_cb), #undef CALLBACK_ENTRY }; diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c index 7c403fb360e..eeb6288a9b5 100644 --- a/src/backend/storage/file/fd.c +++ b/src/backend/storage/file/fd.c @@ -94,6 +94,7 @@ #include "miscadmin.h" #include "pgstat.h" #include "postmaster/startup.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/ipc.h" #include "utils/guc.h" @@ -1294,6 +1295,8 @@ LruDelete(File file) vfdP = &VfdCache[file]; + pgaio_closing_fd(vfdP->fd); + /* * Close the file. We aren't expecting this to fail; if it does, better * to leak the FD than to mess up our internal state. @@ -1987,6 +1990,8 @@ FileClose(File file) if (!FileIsNotOpen(file)) { + pgaio_closing_fd(vfdP->fd); + /* close the file */ if (close(vfdP->fd) != 0) { @@ -2210,6 +2215,32 @@ retry: return returnCode; } +int +FileStartReadV(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, "FileStartReadV: %d (%s) " INT64_FORMAT " %d", + file, VfdCache[file].fileName, + (int64) offset, + iovcnt)); + + returnCode = FileAccess(file); + if (returnCode < 0) + return returnCode; + + vfdP = &VfdCache[file]; + + pgaio_io_prep_readv(ioh, vfdP->fd, iovcnt, offset); + + return 0; +} + ssize_t FileWriteV(File file, const struct iovec *iov, int iovcnt, off_t offset, uint32 wait_event_info) @@ -2315,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) { @@ -2498,6 +2557,12 @@ FilePathName(File file) int FileGetRawDesc(File file) { + int returnCode; + + returnCode = FileAccess(file); + if (returnCode < 0) + return returnCode; + Assert(FileIsValid(file)); return VfdCache[file].fd; } @@ -2778,6 +2843,7 @@ FreeDesc(AllocateDesc *desc) result = closedir(desc->desc.dir); break; case AllocateDescRawFD: + pgaio_closing_fd(desc->desc.fd); result = close(desc->desc.fd); break; default: @@ -2846,6 +2912,8 @@ CloseTransientFile(int fd) /* Only get here if someone passes us a file not in allocatedDescs */ elog(WARNING, "fd passed to CloseTransientFile was not obtained from OpenTransientFile"); + pgaio_closing_fd(fd); + return close(fd); } diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index 11fccda475f..b1277ed97ae 100644 --- a/src/backend/storage/smgr/md.c +++ b/src/backend/storage/smgr/md.c @@ -31,6 +31,7 @@ #include "miscadmin.h" #include "pg_trace.h" #include "pgstat.h" +#include "storage/aio.h" #include "storage/bufmgr.h" #include "storage/fd.h" #include "storage/md.h" @@ -132,6 +133,22 @@ static MdfdVec *_mdfd_getseg(SMgrRelation reln, ForkNumber forknum, static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg); +static PgAioResult md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result); +static void md_readv_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel); +static PgAioResult md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result); +static void md_writev_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel); + +const struct PgAioHandleSharedCallbacks aio_md_readv_cb = { + .complete = md_readv_complete, + .error = md_readv_error, +}; + +const struct PgAioHandleSharedCallbacks aio_md_writev_cb = { + .complete = md_writev_complete, + .error = md_writev_error, +}; + + static inline int _mdfd_open_flags(void) { @@ -927,6 +944,52 @@ mdreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, } } +void +mdstartreadv(PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks) +{ + 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, "read crossing segment boundary"); + + iovcnt = pgaio_io_get_iovec(ioh, &iov); + + Assert(nblocks <= iovcnt); + + iovcnt = buffers_to_iovec(iov, buffers, nblocks_this_segment); + + Assert(iovcnt <= nblocks_this_segment); + + if (!(io_direct_flags & IO_DIRECT_DATA)) + pgaio_io_set_flag(ioh, AHF_BUFFERED); + + pgaio_io_set_subject_smgr(ioh, + reln, + forknum, + blocknum, + nblocks); + pgaio_io_add_shared_cb(ioh, ASC_MD_READV); + + FileStartReadV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_READ); +} + /* * mdwritev() -- Write the supplied blocks at the appropriate location. * @@ -1032,6 +1095,52 @@ mdwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, } } +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, AHF_BUFFERED); + + pgaio_io_set_subject_smgr(ioh, + reln, + forknum, + blocknum, + nblocks); + pgaio_io_add_shared_cb(ioh, ASC_MD_WRITEV); + + FileStartWriteV(ioh, v->mdfd_vfd, iovcnt, seekpos, WAIT_EVENT_DATA_FILE_WRITE); +} + /* * mdwriteback() -- Tell the kernel to write pages back to storage. @@ -1355,6 +1464,21 @@ mdimmedsync(SMgrRelation reln, ForkNumber forknum) } } +int +mdfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) +{ + MdfdVec *v = mdopenfork(reln, forknum, EXTENSION_FAIL); + + v = _mdfd_getseg(reln, forknum, blocknum, false, + EXTENSION_FAIL); + + *off = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE)); + + Assert(*off < (off_t) BLCKSZ * RELSEG_SIZE); + + return FileGetRawDesc(v->mdfd_vfd); +} + /* * register_dirty_segment() -- Mark a relation segment as needing fsync * @@ -1838,3 +1962,193 @@ mdfiletagmatches(const FileTag *ftag, const FileTag *candidate) */ return ftag->rlocator.dbOid == candidate->rlocator.dbOid; } + +/* + * AIO completion callback for mdstartreadv(). + */ +static PgAioResult +md_readv_complete(PgAioHandle *ioh, PgAioResult prior_result) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + PgAioResult result = prior_result; + + if (prior_result.result < 0) + { + result.status = ARS_ERROR; + result.id = ASC_MD_READV; + /* For "hard" errors, track the error number in error_data */ + result.error_data = -prior_result.result; + result.result = 0; + + md_readv_error(result, sd, LOG); + + return result; + } + + result.result /= BLCKSZ; + + if (result.result == 0) + { + /* consider 0 blocks read a failure */ + result.status = ARS_ERROR; + result.id = ASC_MD_READV; + result.error_data = 0; + + md_readv_error(result, sd, LOG); + } + + if (result.status != ARS_ERROR && + result.result < sd->smgr.nblocks) + { + /* partial reads should be retried at upper level */ + result.status = ARS_PARTIAL; + result.id = ASC_MD_READV; + } + + /* AFIXME: post-read portion of mdreadv() */ + + return result; +} + +/* + * AIO error reporting callback for mdstartreadv(). + */ +static void +md_readv_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel) +{ + MemoryContext oldContext = CurrentMemoryContext; + + /* AFIXME: */ + oldContext = MemoryContextSwitchTo(ErrorContext); + + if (result.error_data != 0) + { + errno = result.error_data; /* for errcode_for_file_access() */ + + ereport(elevel, + errcode_for_file_access(), + errmsg("could not read blocks %u..%u in file \"%s\": %m", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum) + ) + ); + } + 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 read blocks %u..%u in file \"%s\": read only %zu of %zu bytes", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks - 1, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum), + result.result * (size_t) BLCKSZ, + subject_data->smgr.nblocks * (size_t) BLCKSZ + ) + ); + } + + MemoryContextSwitchTo(oldContext); +} + +/* + * AIO completion callback for mdstartwritev(). + */ +static PgAioResult +md_writev_complete(PgAioHandle *ioh, PgAioResult prior_result) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + PgAioResult result = prior_result; + + if (prior_result.result < 0) + { + result.status = ARS_ERROR; + result.id = ASC_MD_WRITEV; + /* For "hard" errors, track the error number in error_data */ + result.error_data = -prior_result.result; + result.result = 0; + + md_writev_error(result, sd, LOG); + + return result; + } + + result.result /= BLCKSZ; + + if (result.result == 0) + { + /* consider 0 blocks written a failure */ + result.status = ARS_ERROR; + result.id = ASC_MD_WRITEV; + result.error_data = 0; + + md_writev_error(result, sd, LOG); + } + + if (result.status != ARS_ERROR && + result.result < sd->smgr.nblocks) + { + /* partial writes should be retried at upper level */ + result.status = ARS_PARTIAL; + result.id = ASC_MD_WRITEV; + } + + if (prior_result.status == ARS_ERROR) + { + /* AFIXME: complain */ + return prior_result; + } + + prior_result.result /= BLCKSZ; + + return prior_result; +} + +/* + * AIO error reporting callback for mdstartwritev(). + */ +static void +md_writev_error(PgAioResult result, const PgAioSubjectData *subject_data, int elevel) +{ + MemoryContext oldContext = CurrentMemoryContext; + + /* AFIXME: */ + oldContext = MemoryContextSwitchTo(ErrorContext); + + 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", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum) + ) + ); + } + 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", + subject_data->smgr.blockNum, + subject_data->smgr.blockNum + subject_data->smgr.nblocks - 1, + relpathperm(subject_data->smgr.rlocator, subject_data->smgr.forkNum), + result.result * (size_t) BLCKSZ, + subject_data->smgr.nblocks * (size_t) BLCKSZ + ) + ); + } + + MemoryContextSwitchTo(oldContext); +} diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c index 36ad34aa6ac..454ebe9c243 100644 --- a/src/backend/storage/smgr/smgr.c +++ b/src/backend/storage/smgr/smgr.c @@ -53,6 +53,7 @@ #include "access/xlogutils.h" #include "lib/ilist.h" +#include "storage/aio.h" #include "storage/bufmgr.h" #include "storage/ipc.h" #include "storage/md.h" @@ -93,10 +94,19 @@ typedef struct f_smgr void (*smgr_readv) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, void **buffers, BlockNumber nblocks); + void (*smgr_startreadv) (struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, + BlockNumber blocknum, + void **buffers, BlockNumber nblocks); void (*smgr_writev) (SMgrRelation reln, ForkNumber forknum, 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); @@ -104,6 +114,7 @@ typedef struct f_smgr BlockNumber old_blocks, BlockNumber nblocks); void (*smgr_immedsync) (SMgrRelation reln, ForkNumber forknum); void (*smgr_registersync) (SMgrRelation reln, ForkNumber forknum); + int (*smgr_fd) (SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off); } f_smgr; static const f_smgr smgrsw[] = { @@ -121,12 +132,15 @@ static const f_smgr smgrsw[] = { .smgr_prefetch = mdprefetch, .smgr_maxcombine = mdmaxcombine, .smgr_readv = mdreadv, + .smgr_startreadv = mdstartreadv, .smgr_writev = mdwritev, + .smgr_startwritev = mdstartwritev, .smgr_writeback = mdwriteback, .smgr_nblocks = mdnblocks, .smgr_truncate = mdtruncate, .smgr_immedsync = mdimmedsync, .smgr_registersync = mdregistersync, + .smgr_fd = mdfd, } }; @@ -145,6 +159,14 @@ static void smgrshutdown(int code, Datum arg); static void smgrdestroy(SMgrRelation reln); +static void smgr_aio_reopen(PgAioHandle *ioh); + +const struct PgAioSubjectInfo aio_smgr_subject_info = { + .name = "smgr", + .reopen = smgr_aio_reopen, +}; + + /* * smgrinit(), smgrshutdown() -- Initialize or shut down storage * managers. @@ -623,6 +645,19 @@ smgrreadv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, nblocks); } +/* + * FILL ME IN + */ +void +smgrstartreadv(struct PgAioHandle *ioh, + SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, + void **buffers, BlockNumber nblocks) +{ + smgrsw[reln->smgr_which].smgr_startreadv(ioh, + reln, forknum, blocknum, buffers, + nblocks); +} + /* * smgrwritev() -- Write the supplied buffers out. * @@ -657,6 +692,16 @@ smgrwritev(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, buffers, nblocks, skipFsync); } +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. @@ -819,6 +864,12 @@ smgrimmedsync(SMgrRelation reln, ForkNumber forknum) smgrsw[reln->smgr_which].smgr_immedsync(reln, forknum); } +int +smgrfd(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, uint32 *off) +{ + return smgrsw[reln->smgr_which].smgr_fd(reln, forknum, blocknum, off); +} + /* * AtEOXact_SMgr * @@ -847,3 +898,43 @@ ProcessBarrierSmgrRelease(void) smgrreleaseall(); return true; } + +void +pgaio_io_set_subject_smgr(PgAioHandle *ioh, + struct SMgrRelationData *smgr, + ForkNumber forknum, + BlockNumber blocknum, + int nblocks) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + + pgaio_io_set_subject(ioh, ASI_SMGR); + + /* backend is implied via IO owner */ + sd->smgr.rlocator = smgr->smgr_rlocator.locator; + sd->smgr.forkNum = forknum; + sd->smgr.blockNum = blocknum; + sd->smgr.nblocks = nblocks; + sd->smgr.is_temp = SmgrIsTemp(smgr); + sd->smgr.release_lock = false; + sd->smgr.mode = RBM_NORMAL; +} + +static void +smgr_aio_reopen(PgAioHandle *ioh) +{ + PgAioSubjectData *sd = pgaio_io_get_subject_data(ioh); + PgAioOpData *od = pgaio_io_get_op_data(ioh); + SMgrRelation reln; + ProcNumber procno; + uint32 off; + + if (sd->smgr.is_temp) + procno = pgaio_io_get_owner(ioh); + else + procno = INVALID_PROC_NUMBER; + + reln = smgropen(sd->smgr.rlocator, procno); + od->read.fd = smgrfd(reln, sd->smgr.forkNum, sd->smgr.blockNum, &off); + Assert(off == od->read.offset); +} -- 2.45.2.746.g06e570c0df.dirty --oy2jwuii6tssbict Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2-0011-bufmgr-Implement-AIO-read-support.patch" ^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2024-12-15 17:36 UTC | newest] Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-06-27 03:39 [PATCH v1] Add pg_get_query_def() to deparse and print a rewritten SQL statement. Julien Rouhaud <[email protected]> 2024-09-01 02:33 [PATCH v2.0 10/17] aio: Implement smgr/md.c aio methods Andres Freund <[email protected]> 2024-09-01 02:33 [PATCH v2.1 13/20] aio: Implement smgr/md.c aio methods Andres Freund <[email protected]> 2024-12-15 17:36 [PATCH v2 10/20] aio: Implement smgr/md.c aio methods 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