From: Andres Freund Date: Tue, 23 Jul 2024 10:00:06 -0700 Subject: [PATCH v2.0 14/17] aio: Add IO queue helper This is likely never going to anywhere - Thomas Munro is working on something more complete. But I needed a way to exercise aio for checkpointer / bgwriter. --- src/include/storage/io_queue.h | 33 +++++ src/backend/storage/aio/Makefile | 1 + src/backend/storage/aio/io_queue.c | 195 ++++++++++++++++++++++++++++ src/backend/storage/aio/meson.build | 1 + src/tools/pgindent/typedefs.list | 2 + 5 files changed, 232 insertions(+) create mode 100644 src/include/storage/io_queue.h create mode 100644 src/backend/storage/aio/io_queue.c diff --git a/src/include/storage/io_queue.h b/src/include/storage/io_queue.h new file mode 100644 index 00000000000..28077158d6d --- /dev/null +++ b/src/include/storage/io_queue.h @@ -0,0 +1,33 @@ +/*------------------------------------------------------------------------- + * + * io_queue.h + * Mechanism for tracking many IOs + * + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/storage/io_queue.h + * + *------------------------------------------------------------------------- + */ +#ifndef IO_QUEUE_H +#define IO_QUEUE_H + +#include "storage/bufmgr.h" + +struct IOQueue; +typedef struct IOQueue IOQueue; + +struct PgAioHandleRef; + +extern IOQueue *io_queue_create(int depth, int flags); +extern void io_queue_track(IOQueue *ioq, const struct PgAioHandleRef *ior); +extern void io_queue_wait_one(IOQueue *ioq); +extern void io_queue_wait_all(IOQueue *ioq); +extern bool io_queue_is_empty(IOQueue *ioq); +extern void io_queue_reserve(IOQueue *ioq); +extern struct PgAioHandle *io_queue_get_io(IOQueue *ioq); +extern void io_queue_free(IOQueue *ioq); + +#endif /* IO_QUEUE_H */ diff --git a/src/backend/storage/aio/Makefile b/src/backend/storage/aio/Makefile index 2a5e72a8024..3fb527ed0d1 100644 --- a/src/backend/storage/aio/Makefile +++ b/src/backend/storage/aio/Makefile @@ -13,6 +13,7 @@ OBJS = \ aio_io.o \ aio_init.o \ aio_subject.o \ + io_queue.o \ method_worker.o \ method_io_uring.o \ read_stream.o diff --git a/src/backend/storage/aio/io_queue.c b/src/backend/storage/aio/io_queue.c new file mode 100644 index 00000000000..4dda2f4e20e --- /dev/null +++ b/src/backend/storage/aio/io_queue.c @@ -0,0 +1,195 @@ +/*------------------------------------------------------------------------- + * + * io_queue.c + * Mechanism for tracking many IOs + * + * Portions Copyright (c) 2024, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/backend/storage/aio/io_queue.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "storage/io_queue.h" + +#include "storage/aio.h" + + +typedef struct TrackedIO +{ + PgAioHandleRef ior; + dlist_node node; +} TrackedIO; + +struct IOQueue +{ + int depth; + int unsubmitted; + + bool has_reserved; + + dclist_head idle; + dclist_head in_progress; + + TrackedIO tracked_ios[FLEXIBLE_ARRAY_MEMBER]; +}; + + +IOQueue * +io_queue_create(int depth, int flags) +{ + size_t sz; + IOQueue *ioq; + + sz = offsetof(IOQueue, tracked_ios) + + sizeof(TrackedIO) * depth; + + ioq = palloc0(sz); + + ioq->depth = 0; + + for (int i = 0; i < depth; i++) + { + TrackedIO *tio = &ioq->tracked_ios[i]; + + pgaio_io_ref_clear(&tio->ior); + dclist_push_tail(&ioq->idle, &tio->node); + } + + return ioq; +} + +void +io_queue_wait_one(IOQueue *ioq) +{ + while (!dclist_is_empty(&ioq->in_progress)) + { + /* FIXME: Should we really pop here already? */ + dlist_node *node = dclist_pop_head_node(&ioq->in_progress); + TrackedIO *tio = dclist_container(TrackedIO, node, node); + + pgaio_io_ref_wait(&tio->ior); + dclist_push_head(&ioq->idle, &tio->node); + } +} + +void +io_queue_reserve(IOQueue *ioq) +{ + if (ioq->has_reserved) + return; + + if (dclist_is_empty(&ioq->idle)) + io_queue_wait_one(ioq); + + Assert(!dclist_is_empty(&ioq->idle)); + + ioq->has_reserved = true; +} + +PgAioHandle * +io_queue_get_io(IOQueue *ioq) +{ + PgAioHandle *ioh; + + io_queue_reserve(ioq); + + Assert(!dclist_is_empty(&ioq->idle)); + + if (!io_queue_is_empty(ioq)) + { + ioh = pgaio_io_get_nb(CurrentResourceOwner, NULL); + if (ioh == NULL) + { + /* + * Need to wait for all IOs, blocking might not be legal in the + * context. + * + * XXX: This doesn't make a whole lot of sense, we're also + * blocking here. What was I smoking when I wrote the above? + */ + io_queue_wait_all(ioq); + ioh = pgaio_io_get(CurrentResourceOwner, NULL); + } + } + else + { + ioh = pgaio_io_get(CurrentResourceOwner, NULL); + } + + return ioh; +} + +void +io_queue_track(IOQueue *ioq, const struct PgAioHandleRef *ior) +{ + dlist_node *node; + TrackedIO *tio; + + Assert(ioq->has_reserved); + ioq->has_reserved = false; + + Assert(!dclist_is_empty(&ioq->idle)); + + node = dclist_pop_head_node(&ioq->idle); + tio = dclist_container(TrackedIO, node, node); + + tio->ior = *ior; + + dclist_push_tail(&ioq->in_progress, &tio->node); + + ioq->unsubmitted++; + + /* + * XXX: Should have some smarter logic here. We don't want to wait too + * long to submit, that'll mean we're more likely to block. But we also + * don't want to have the overhead of submitting every IO individually. + */ + if (ioq->unsubmitted >= 4) + { + pgaio_submit_staged(); + ioq->unsubmitted = 0; + } +} + +void +io_queue_wait_all(IOQueue *ioq) +{ + while (!dclist_is_empty(&ioq->in_progress)) + { + /* wait for the last IO to minimize unnecessary wakeups */ + dlist_node *node = dclist_tail_node(&ioq->in_progress); + TrackedIO *tio = dclist_container(TrackedIO, node, node); + + if (!pgaio_io_ref_check_done(&tio->ior)) + { + ereport(DEBUG3, + errmsg("io_queue_wait_all for io:%d", + pgaio_io_ref_get_id(&tio->ior)), + errhidestmt(true), + errhidecontext(true)); + + pgaio_io_ref_wait(&tio->ior); + } + + dclist_delete_from(&ioq->in_progress, &tio->node); + dclist_push_head(&ioq->idle, &tio->node); + } +} + +bool +io_queue_is_empty(IOQueue *ioq) +{ + return dclist_is_empty(&ioq->in_progress); +} + +void +io_queue_free(IOQueue *ioq) +{ + io_queue_wait_all(ioq); + + pfree(ioq); +} diff --git a/src/backend/storage/aio/meson.build b/src/backend/storage/aio/meson.build index 8960223194a..6d64c75a49c 100644 --- a/src/backend/storage/aio/meson.build +++ b/src/backend/storage/aio/meson.build @@ -5,6 +5,7 @@ backend_sources += files( 'aio_io.c', 'aio_init.c', 'aio_subject.c', + 'io_queue.c', 'method_io_uring.c', 'method_worker.c', 'read_stream.c', diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index be8be9fbff0..6f39abcdf3c 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1171,6 +1171,7 @@ IOContext IOFuncSelector IOObject IOOp +IOQueue IO_STATUS_BLOCK IPCompareMethod ITEM @@ -2959,6 +2960,7 @@ TocEntry TokenAuxData TokenizedAuthLine TrackItem +TrackedIO TransApplyAction TransInvalidationInfo TransState -- 2.45.2.827.g557ae147e6 --ww2auydviafoh7lh Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.0-0015-bufmgr-use-AIO-in-checkpointer-bgwriter.patch"