agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Andres Freund <[email protected]>
Subject: [PATCH v2.1 17/20] aio: Add IO queue helper
Date: Wed, 4 Sep 2024 16:15:42 -0400
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 3bcb8a0b2ed..f3a7f9e63d6 100644
--- a/src/backend/storage/aio/Makefile
+++ b/src/backend/storage/aio/Makefile
@@ -13,6 +13,7 @@ OBJS = \
aio_init.o \
aio_io.o \
aio_subject.o \
+ io_queue.o \
method_io_uring.o \
method_sync.o \
method_worker.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 537f23d446d..e8a88e615c0 100644
--- a/src/backend/storage/aio/meson.build
+++ b/src/backend/storage/aio/meson.build
@@ -5,6 +5,7 @@ backend_sources += files(
'aio_init.c',
'aio_io.c',
'aio_subject.c',
+ 'io_queue.c',
'method_io_uring.c',
'method_sync.c',
'method_worker.c',
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 46d31cf2b9f..a38141b4e50 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1172,6 +1172,7 @@ IOContext
IOFuncSelector
IOObject
IOOp
+IOQueue
IO_STATUS_BLOCK
IPCompareMethod
ITEM
@@ -2960,6 +2961,7 @@ TocEntry
TokenAuxData
TokenizedAuthLine
TrackItem
+TrackedIO
TransApplyAction
TransInvalidationInfo
TransState
--
2.45.2.827.g557ae147e6
--ggp33qz5xusicvk6
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v2.1-0018-bufmgr-use-AIO-in-checkpointer-bgwriter.patch"
view thread (17+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH v2.1 17/20] aio: Add IO queue helper
In-Reply-To: <no-message-id-51749@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox