Hi,
I worked on using the currently proposed streaming read API [1] in ANALYZE.
The patch is attached. 0001 is the not yet merged streaming read API code
changes that can be applied to the master, 0002 is the actual code.
The blocks to analyze are obtained by using the streaming read API now.
- Since streaming read API is already doing prefetch, I removed the #ifdef
USE_PREFETCH code from acquire_sample_rows().
- Changed 'while (BlockSampler_HasMore(&bs))' to 'while (nblocks)' because
the prefetch mechanism in the streaming read API will advance 'bs' before
returning buffers.
- Removed BlockNumber and BufferAccessStrategy from the declaration of
scan_analyze_next_block(), passing pgsr (PgStreamingRead) instead of them.
I counted syscalls of analyzing ~5GB table. It can be seen that the patched
version did ~1300 less read calls.
Patched:
% time seconds usecs/call calls errors syscall
------ ----------- ----------- --------- --------- ----------------
39.67 0.012128 0 29809 pwrite64
36.96 0.011299 0 28594 pread64
23.24 0.007104 0 27611 fadvise64
Master (21a71648d3):
% time seconds usecs/call calls errors syscall
------ ----------- ----------- --------- --------- ----------------
38.94 0.016457 0 29816 pwrite64
36.79 0.015549 0 29850 pread64
23.91 0.010106 0 29848 fadvise64
Any kind of feedback would be appreciated.
[1]:
https://www.postgresql.org/message-id/CA%2BhUKGJkOiOCa%2Bmag4BF%2BzHo7qo%3Do9CFheB8%3Dg6uT5TUm2gkvA%40mail.gmail.com
--
Regards,
Nazir Bilal Yavuz
Microsoft
From 509e55997c084f831fcbcb46cabe79d4f97aa93e Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.mu...@gmail.com>
Date: Sat, 22 Jul 2023 17:31:54 +1200
Subject: [PATCH v1 1/2] Streaming read API changes that are not committed to
master yet
Discussion: https://www.postgresql.org/message-id/CA%2BhUKGJkOiOCa%2Bmag4BF%2BzHo7qo%3Do9CFheB8%3Dg6uT5TUm2gkvA%40mail.gmail.com
---
src/include/storage/bufmgr.h | 22 +
src/include/storage/streaming_read.h | 52 ++
src/backend/storage/Makefile | 2 +-
src/backend/storage/aio/Makefile | 14 +
src/backend/storage/aio/meson.build | 5 +
src/backend/storage/aio/streaming_read.c | 472 ++++++++++++++++++
src/backend/storage/buffer/bufmgr.c | 592 +++++++++++++++--------
src/backend/storage/buffer/localbuf.c | 14 +-
src/backend/storage/meson.build | 1 +
src/tools/pgindent/typedefs.list | 2 +
10 files changed, 953 insertions(+), 223 deletions(-)
create mode 100644 src/include/storage/streaming_read.h
create mode 100644 src/backend/storage/aio/Makefile
create mode 100644 src/backend/storage/aio/meson.build
create mode 100644 src/backend/storage/aio/streaming_read.c
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index d51d46d3353..a38f1acb37a 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -14,6 +14,7 @@
#ifndef BUFMGR_H
#define BUFMGR_H
+#include "port/pg_iovec.h"
#include "storage/block.h"
#include "storage/buf.h"
#include "storage/bufpage.h"
@@ -158,6 +159,11 @@ extern PGDLLIMPORT int32 *LocalRefCount;
#define BUFFER_LOCK_SHARE 1
#define BUFFER_LOCK_EXCLUSIVE 2
+/*
+ * Maximum number of buffers for multi-buffer I/O functions. This is set to
+ * allow 128kB transfers, unless BLCKSZ and IOV_MAX imply a a smaller maximum.
+ */
+#define MAX_BUFFERS_PER_TRANSFER Min(PG_IOV_MAX, (128 * 1024) / BLCKSZ)
/*
* prototypes for functions in bufmgr.c
@@ -177,6 +183,18 @@ extern Buffer ReadBufferWithoutRelcache(RelFileLocator rlocator,
ForkNumber forkNum, BlockNumber blockNum,
ReadBufferMode mode, BufferAccessStrategy strategy,
bool permanent);
+extern Buffer PrepareReadBuffer(BufferManagerRelation bmr,
+ ForkNumber forkNum,
+ BlockNumber blockNum,
+ BufferAccessStrategy strategy,
+ bool *foundPtr);
+extern void CompleteReadBuffers(BufferManagerRelation bmr,
+ Buffer *buffers,
+ ForkNumber forknum,
+ BlockNumber blocknum,
+ int nblocks,
+ bool zero_on_error,
+ BufferAccessStrategy strategy);
extern void ReleaseBuffer(Buffer buffer);
extern void UnlockReleaseBuffer(Buffer buffer);
extern bool BufferIsExclusiveLocked(Buffer buffer);
@@ -247,9 +265,13 @@ extern void LockBufferForCleanup(Buffer buffer);
extern bool ConditionalLockBufferForCleanup(Buffer buffer);
extern bool IsBufferCleanupOK(Buffer buffer);
extern bool HoldingBufferPinThatDelaysRecovery(void);
+extern void ZeroBuffer(Buffer buffer, ReadBufferMode mode);
extern bool BgBufferSync(struct WritebackContext *wb_context);
+extern void LimitAdditionalPins(uint32 *additional_pins);
+extern void LimitAdditionalLocalPins(uint32 *additional_pins);
+
/* in buf_init.c */
extern void InitBufferPool(void);
extern Size BufferShmemSize(void);
diff --git a/src/include/storage/streaming_read.h b/src/include/storage/streaming_read.h
new file mode 100644
index 00000000000..c4d3892bb26
--- /dev/null
+++ b/src/include/storage/streaming_read.h
@@ -0,0 +1,52 @@
+#ifndef STREAMING_READ_H
+#define STREAMING_READ_H
+
+#include "storage/bufmgr.h"
+#include "storage/fd.h"
+#include "storage/smgr.h"
+
+/* Default tuning, reasonable for many users. */
+#define PGSR_FLAG_DEFAULT 0x00
+
+/*
+ * I/O streams that are performing maintenance work on behalf of potentially
+ * many users.
+ */
+#define PGSR_FLAG_MAINTENANCE 0x01
+
+/*
+ * We usually avoid issuing prefetch advice automatically when sequential
+ * access is detected, but this flag explicitly disables it, for cases that
+ * might not be correctly detected. Explicit advice is known to perform worse
+ * than letting the kernel (at least Linux) detect sequential access.
+ */
+#define PGSR_FLAG_SEQUENTIAL 0x02
+
+/*
+ * We usually ramp up from smaller reads to larger ones, to support users who
+ * don't know if it's worth reading lots of buffers yet. This flag disables
+ * that, declaring ahead of time that we'll be reading all available buffers.
+ */
+#define PGSR_FLAG_FULL 0x04
+
+struct PgStreamingRead;
+typedef struct PgStreamingRead PgStreamingRead;
+
+/* Callback that returns the next block number to read. */
+typedef BlockNumber (*PgStreamingReadBufferCB) (PgStreamingRead *pgsr,
+ void *pgsr_private,
+ void *per_buffer_private);
+
+extern PgStreamingRead *pg_streaming_read_buffer_alloc(int flags,
+ void *pgsr_private,
+ size_t per_buffer_private_size,
+ BufferAccessStrategy strategy,
+ BufferManagerRelation bmr,
+ ForkNumber forknum,
+ PgStreamingReadBufferCB next_block_cb);
+
+extern void pg_streaming_read_prefetch(PgStreamingRead *pgsr);
+extern Buffer pg_streaming_read_buffer_get_next(PgStreamingRead *pgsr, void **per_buffer_private);
+extern void pg_streaming_read_free(PgStreamingRead *pgsr);
+
+#endif
diff --git a/src/backend/storage/Makefile b/src/backend/storage/Makefile
index 8376cdfca20..eec03f6f2b4 100644
--- a/src/backend/storage/Makefile
+++ b/src/backend/storage/Makefile
@@ -8,6 +8,6 @@ subdir = src/backend/storage
top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
-SUBDIRS = buffer file freespace ipc large_object lmgr page smgr sync
+SUBDIRS = aio buffer file freespace ipc large_object lmgr page smgr sync
include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/storage/aio/Makefile b/src/backend/storage/aio/Makefile
new file mode 100644
index 00000000000..bcab44c802f
--- /dev/null
+++ b/src/backend/storage/aio/Makefile
@@ -0,0 +1,14 @@
+#
+# Makefile for storage/aio
+#
+# src/backend/storage/aio/Makefile
+#
+
+subdir = src/backend/storage/aio
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+
+OBJS = \
+ streaming_read.o
+
+include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/storage/aio/meson.build b/src/backend/storage/aio/meson.build
new file mode 100644
index 00000000000..39aef2a84a2
--- /dev/null
+++ b/src/backend/storage/aio/meson.build
@@ -0,0 +1,5 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+backend_sources += files(
+ 'streaming_read.c',
+)
diff --git a/src/backend/storage/aio/streaming_read.c b/src/backend/storage/aio/streaming_read.c
new file mode 100644
index 00000000000..bb5b863be76
--- /dev/null
+++ b/src/backend/storage/aio/streaming_read.c
@@ -0,0 +1,472 @@
+#include "postgres.h"
+
+#include "storage/streaming_read.h"
+#include "utils/rel.h"
+
+/*
+ * Element type for PgStreamingRead's circular array of block ranges.
+ *
+ * For hits, need_to_complete is false and there is just one block per
+ * range, already pinned and ready for use.
+ *
+ * For misses, need_to_complete is true and buffers[] holds a range of
+ * blocks that are contiguous in storage (though the buffers may not be
+ * contiguous in memory), so we can complete them with a single call to
+ * CompleteReadBuffers().
+ */
+typedef struct PgStreamingReadRange
+{
+ bool advice_issued;
+ bool need_complete;
+ BlockNumber blocknum;
+ int nblocks;
+ int per_buffer_data_index[MAX_BUFFERS_PER_TRANSFER];
+ Buffer buffers[MAX_BUFFERS_PER_TRANSFER];
+} PgStreamingReadRange;
+
+struct PgStreamingRead
+{
+ int max_ios;
+ int ios_in_progress;
+ int ios_in_progress_trigger;
+ int max_pinned_buffers;
+ int pinned_buffers;
+ int pinned_buffers_trigger;
+ int next_tail_buffer;
+ int ramp_up_pin_limit;
+ int ramp_up_pin_stall;
+ bool finished;
+ void *pgsr_private;
+ PgStreamingReadBufferCB callback;
+ BufferAccessStrategy strategy;
+ BufferManagerRelation bmr;
+ ForkNumber forknum;
+
+ bool advice_enabled;
+
+ /* Next expected block, for detecting sequential access. */
+ BlockNumber seq_blocknum;
+
+ /* Space for optional per-buffer private data. */
+ size_t per_buffer_data_size;
+ void *per_buffer_data;
+ int per_buffer_data_next;
+
+ /* Circular buffer of ranges. */
+ int size;
+ int head;
+ int tail;
+ PgStreamingReadRange ranges[FLEXIBLE_ARRAY_MEMBER];
+};
+
+static PgStreamingRead *
+pg_streaming_read_buffer_alloc_internal(int flags,
+ void *pgsr_private,
+ size_t per_buffer_data_size,
+ BufferAccessStrategy strategy)
+{
+ PgStreamingRead *pgsr;
+ int size;
+ int max_ios;
+ uint32 max_pinned_buffers;
+
+
+ /*
+ * Decide how many assumed I/Os we will allow to run concurrently. That
+ * is, advice to the kernel to tell it that we will soon read. This
+ * number also affects how far we look ahead for opportunities to start
+ * more I/Os.
+ */
+ if (flags & PGSR_FLAG_MAINTENANCE)
+ max_ios = maintenance_io_concurrency;
+ else
+ max_ios = effective_io_concurrency;
+
+ /*
+ * The desired level of I/O concurrency controls how far ahead we are
+ * willing to look ahead. We also clamp it to at least
+ * MAX_BUFFER_PER_TRANFER so that we can have a chance to build up a full
+ * sized read, even when max_ios is zero.
+ */
+ max_pinned_buffers = Max(max_ios * 4, MAX_BUFFERS_PER_TRANSFER);
+
+ /*
+ * The *_io_concurrency GUCs, we might have 0. We want to allow at least
+ * one, to keep our gating logic simple.
+ */
+ max_ios = Max(max_ios, 1);
+
+ /*
+ * Don't allow this backend to pin too many buffers. For now we'll apply
+ * the limit for the shared buffer pool and the local buffer pool, without
+ * worrying which it is.
+ */
+ LimitAdditionalPins(&max_pinned_buffers);
+ LimitAdditionalLocalPins(&max_pinned_buffers);
+ Assert(max_pinned_buffers > 0);
+
+ /*
+ * pgsr->ranges is a circular buffer. When it is empty, head == tail.
+ * When it is full, there is an empty element between head and tail. Head
+ * can also be empty (nblocks == 0), therefore we need two extra elements
+ * for non-occupied ranges, on top of max_pinned_buffers to allow for the
+ * maxmimum possible number of occupied ranges of the smallest possible
+ * size of one.
+ */
+ size = max_pinned_buffers + 2;
+
+ pgsr = (PgStreamingRead *)
+ palloc0(offsetof(PgStreamingRead, ranges) +
+ sizeof(pgsr->ranges[0]) * size);
+
+ pgsr->max_ios = max_ios;
+ pgsr->per_buffer_data_size = per_buffer_data_size;
+ pgsr->max_pinned_buffers = max_pinned_buffers;
+ pgsr->pgsr_private = pgsr_private;
+ pgsr->strategy = strategy;
+ pgsr->size = size;
+
+#ifdef USE_PREFETCH
+
+ /*
+ * This system supports prefetching advice. As long as direct I/O isn't
+ * enabled, and the caller hasn't promised sequential access, we can use
+ * it.
+ */
+ if ((io_direct_flags & IO_DIRECT_DATA) == 0 &&
+ (flags & PGSR_FLAG_SEQUENTIAL) == 0)
+ pgsr->advice_enabled = true;
+#endif
+
+ /*
+ * We start off building small ranges, but double that quickly, for the
+ * benefit of users that don't know how far ahead they'll read. This can
+ * be disabled by users that already know they'll read all the way.
+ */
+ if (flags & PGSR_FLAG_FULL)
+ pgsr->ramp_up_pin_limit = INT_MAX;
+ else
+ pgsr->ramp_up_pin_limit = 1;
+
+ /*
+ * We want to avoid creating ranges that are smaller than they could be
+ * just because we hit max_pinned_buffers. We only look ahead when the
+ * number of pinned buffers falls below this trigger number, or put
+ * another way, we stop looking ahead when we wouldn't be able to build a
+ * "full sized" range.
+ */
+ pgsr->pinned_buffers_trigger =
+ Max(1, (int) max_pinned_buffers - MAX_BUFFERS_PER_TRANSFER);
+
+ /* Space the callback to store extra data along with each block. */
+ if (per_buffer_data_size)
+ pgsr->per_buffer_data = palloc(per_buffer_data_size * max_pinned_buffers);
+
+ return pgsr;
+}
+
+/*
+ * Create a new streaming read object that can be used to perform the
+ * equivalent of a series of ReadBuffer() calls for one fork of one relation.
+ * Internally, it generates larger vectored reads where possible by looking
+ * ahead.
+ */
+PgStreamingRead *
+pg_streaming_read_buffer_alloc(int flags,
+ void *pgsr_private,
+ size_t per_buffer_data_size,
+ BufferAccessStrategy strategy,
+ BufferManagerRelation bmr,
+ ForkNumber forknum,
+ PgStreamingReadBufferCB next_block_cb)
+{
+ PgStreamingRead *result;
+
+ result = pg_streaming_read_buffer_alloc_internal(flags,
+ pgsr_private,
+ per_buffer_data_size,
+ strategy);
+ result->callback = next_block_cb;
+ result->bmr = bmr;
+ result->forknum = forknum;
+
+ return result;
+}
+
+/*
+ * Start building a new range. This is called after the previous one
+ * reached maximum size, or the callback's next block can't be merged with it.
+ *
+ * Since the previous head range has now reached its full potential size, this
+ * is also a good time to issue 'prefetch' advice, because we know that'll
+ * soon be reading. In future, we could start an actual I/O here.
+ */
+static PgStreamingReadRange *
+pg_streaming_read_new_range(PgStreamingRead *pgsr)
+{
+ PgStreamingReadRange *head_range;
+
+ head_range = &pgsr->ranges[pgsr->head];
+ Assert(head_range->nblocks > 0);
+
+ /*
+ * If a call to CompleteReadBuffers() will be needed, and we can issue
+ * advice to the kernel to get the read started. We suppress it if the
+ * access pattern appears to be completely sequential, though, because on
+ * some systems that interfers with the kernel's own sequential read ahead
+ * heurstics and hurts performance.
+ */
+ if (pgsr->advice_enabled)
+ {
+ BlockNumber blocknum = head_range->blocknum;
+ int nblocks = head_range->nblocks;
+
+ if (head_range->need_complete && blocknum != pgsr->seq_blocknum)
+ {
+ SMgrRelation smgr =
+ pgsr->bmr.smgr ? pgsr->bmr.smgr :
+ RelationGetSmgr(pgsr->bmr.rel);
+
+ Assert(!head_range->advice_issued);
+
+ smgrprefetch(smgr, pgsr->forknum, blocknum, nblocks);
+
+ /*
+ * Count this as an I/O that is concurrently in progress, though
+ * we don't really know if the kernel generates a physical I/O.
+ */
+ head_range->advice_issued = true;
+ pgsr->ios_in_progress++;
+ }
+
+ /* Remember the block after this range, for sequence detection. */
+ pgsr->seq_blocknum = blocknum + nblocks;
+ }
+
+ /* Create a new head range. There must be space. */
+ Assert(pgsr->size > pgsr->max_pinned_buffers);
+ Assert((pgsr->head + 1) % pgsr->size != pgsr->tail);
+ if (++pgsr->head == pgsr->size)
+ pgsr->head = 0;
+ head_range = &pgsr->ranges[pgsr->head];
+ head_range->nblocks = 0;
+
+ return head_range;
+}
+
+static void
+pg_streaming_read_look_ahead(PgStreamingRead *pgsr)
+{
+ /*
+ * If we're still ramping up, we may have to stall to wait for buffers to
+ * be consumed first before we do any more prefetching.
+ */
+ if (pgsr->ramp_up_pin_stall > 0)
+ {
+ Assert(pgsr->pinned_buffers > 0);
+ return;
+ }
+
+ /*
+ * If we're finished or can't start more I/O, then don't look ahead.
+ */
+ if (pgsr->finished || pgsr->ios_in_progress == pgsr->max_ios)
+ return;
+
+ /*
+ * We'll also wait until the number of pinned buffers falls below our
+ * trigger level, so that we have the chance to create a full range.
+ */
+ if (pgsr->pinned_buffers >= pgsr->pinned_buffers_trigger)
+ return;
+
+ do
+ {
+ BufferManagerRelation bmr;
+ ForkNumber forknum;
+ BlockNumber blocknum;
+ Buffer buffer;
+ bool found;
+ bool need_complete;
+ PgStreamingReadRange *head_range;
+ void *per_buffer_data;
+
+ /* Do we have a full-sized range? */
+ head_range = &pgsr->ranges[pgsr->head];
+ if (head_range->nblocks == lengthof(head_range->buffers))
+ {
+ Assert(head_range->need_complete);
+ head_range = pg_streaming_read_new_range(pgsr);
+
+ /*
+ * Give up now if I/O is saturated, or we wouldn't be able form
+ * another full range after this due to the pin limit.
+ */
+ if (pgsr->pinned_buffers >= pgsr->pinned_buffers_trigger ||
+ pgsr->ios_in_progress == pgsr->max_ios)
+ break;
+ }
+
+ per_buffer_data = (char *) pgsr->per_buffer_data +
+ pgsr->per_buffer_data_size * pgsr->per_buffer_data_next;
+
+ /* Find out which block the callback wants to read next. */
+ blocknum = pgsr->callback(pgsr, pgsr->pgsr_private, per_buffer_data);
+ if (blocknum == InvalidBlockNumber)
+ {
+ pgsr->finished = true;
+ break;
+ }
+ bmr = pgsr->bmr;
+ forknum = pgsr->forknum;
+
+ Assert(pgsr->pinned_buffers < pgsr->max_pinned_buffers);
+
+ buffer = PrepareReadBuffer(bmr,
+ forknum,
+ blocknum,
+ pgsr->strategy,
+ &found);
+ pgsr->pinned_buffers++;
+
+ need_complete = !found;
+
+ /* Is there a head range that we can't extend? */
+ head_range = &pgsr->ranges[pgsr->head];
+ if (head_range->nblocks > 0 &&
+ (!need_complete ||
+ !head_range->need_complete ||
+ head_range->blocknum + head_range->nblocks != blocknum))
+ {
+ /* Yes, time to start building a new one. */
+ head_range = pg_streaming_read_new_range(pgsr);
+ Assert(head_range->nblocks == 0);
+ }
+
+ if (head_range->nblocks == 0)
+ {
+ /* Initialize a new range beginning at this block. */
+ head_range->blocknum = blocknum;
+ head_range->need_complete = need_complete;
+ head_range->advice_issued = false;
+ }
+ else
+ {
+ /* We can extend an existing range by one block. */
+ Assert(head_range->blocknum + head_range->nblocks == blocknum);
+ Assert(head_range->need_complete);
+ }
+
+ head_range->per_buffer_data_index[head_range->nblocks] = pgsr->per_buffer_data_next++;
+ head_range->buffers[head_range->nblocks] = buffer;
+ head_range->nblocks++;
+
+ if (pgsr->per_buffer_data_next == pgsr->max_pinned_buffers)
+ pgsr->per_buffer_data_next = 0;
+
+ } while (pgsr->pinned_buffers < pgsr->max_pinned_buffers &&
+ pgsr->ios_in_progress < pgsr->max_ios &&
+ pgsr->pinned_buffers < pgsr->ramp_up_pin_limit);
+
+ /* If we've hit the ramp-up limit, insert a stall. */
+ if (pgsr->pinned_buffers >= pgsr->ramp_up_pin_limit)
+ {
+ /* Can't get here if an earlier stall hasn't finished. */
+ Assert(pgsr->ramp_up_pin_stall == 0);
+ /* Don't do any more prefetching until these buffers are consumed. */
+ pgsr->ramp_up_pin_stall = pgsr->ramp_up_pin_limit;
+ /* Double it. It will soon be out of the way. */
+ pgsr->ramp_up_pin_limit *= 2;
+ }
+
+ if (pgsr->ranges[pgsr->head].nblocks > 0)
+ pg_streaming_read_new_range(pgsr);
+}
+
+Buffer
+pg_streaming_read_buffer_get_next(PgStreamingRead *pgsr, void **per_buffer_data)
+{
+ pg_streaming_read_look_ahead(pgsr);
+
+ /* See if we have one buffer to return. */
+ while (pgsr->tail != pgsr->head)
+ {
+ PgStreamingReadRange *tail_range;
+
+ tail_range = &pgsr->ranges[pgsr->tail];
+
+ /*
+ * Do we need to perform an I/O before returning the buffers from this
+ * range?
+ */
+ if (tail_range->need_complete)
+ {
+ CompleteReadBuffers(pgsr->bmr,
+ tail_range->buffers,
+ pgsr->forknum,
+ tail_range->blocknum,
+ tail_range->nblocks,
+ false,
+ pgsr->strategy);
+ tail_range->need_complete = false;
+
+ /*
+ * We don't really know if the kernel generated an physical I/O
+ * when we issued advice, let alone when it finished, but it has
+ * certainly finished after a read call returns.
+ */
+ if (tail_range->advice_issued)
+ pgsr->ios_in_progress--;
+ }
+
+ /* Are there more buffers available in this range? */
+ if (pgsr->next_tail_buffer < tail_range->nblocks)
+ {
+ int buffer_index;
+ Buffer buffer;
+
+ buffer_index = pgsr->next_tail_buffer++;
+ buffer = tail_range->buffers[buffer_index];
+
+ Assert(BufferIsValid(buffer));
+
+ /* We are giving away ownership of this pinned buffer. */
+ Assert(pgsr->pinned_buffers > 0);
+ pgsr->pinned_buffers--;
+
+ if (pgsr->ramp_up_pin_stall > 0)
+ pgsr->ramp_up_pin_stall--;
+
+ if (per_buffer_data)
+ *per_buffer_data = (char *) pgsr->per_buffer_data +
+ tail_range->per_buffer_data_index[buffer_index] *
+ pgsr->per_buffer_data_size;
+
+ return buffer;
+ }
+
+ /* Advance tail to next range, if there is one. */
+ if (++pgsr->tail == pgsr->size)
+ pgsr->tail = 0;
+ pgsr->next_tail_buffer = 0;
+ }
+
+ Assert(pgsr->pinned_buffers == 0);
+
+ return InvalidBuffer;
+}
+
+void
+pg_streaming_read_free(PgStreamingRead *pgsr)
+{
+ Buffer buffer;
+
+ /* Stop looking ahead, and unpin anything that wasn't consumed. */
+ pgsr->finished = true;
+ while ((buffer = pg_streaming_read_buffer_get_next(pgsr, NULL)) != InvalidBuffer)
+ ReleaseBuffer(buffer);
+
+ if (pgsr->per_buffer_data)
+ pfree(pgsr->per_buffer_data);
+ pfree(pgsr);
+}
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 07575ef3129..ef5367c6cc7 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -472,7 +472,7 @@ ForgetPrivateRefCountEntry(PrivateRefCountEntry *ref)
)
-static Buffer ReadBuffer_common(SMgrRelation smgr, char relpersistence,
+static Buffer ReadBuffer_common(BufferManagerRelation bmr,
ForkNumber forkNum, BlockNumber blockNum,
ReadBufferMode mode, BufferAccessStrategy strategy,
bool *hit);
@@ -795,15 +795,9 @@ ReadBufferExtended(Relation reln, ForkNumber forkNum, BlockNumber blockNum,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot access temporary tables of other sessions")));
- /*
- * Read the buffer, and update pgstat counters to reflect a cache hit or
- * miss.
- */
- pgstat_count_buffer_read(reln);
- buf = ReadBuffer_common(RelationGetSmgr(reln), reln->rd_rel->relpersistence,
+ buf = ReadBuffer_common(BMR_REL(reln),
forkNum, blockNum, mode, strategy, &hit);
- if (hit)
- pgstat_count_buffer_hit(reln);
+
return buf;
}
@@ -827,8 +821,9 @@ ReadBufferWithoutRelcache(RelFileLocator rlocator, ForkNumber forkNum,
SMgrRelation smgr = smgropen(rlocator, InvalidBackendId);
- return ReadBuffer_common(smgr, permanent ? RELPERSISTENCE_PERMANENT :
- RELPERSISTENCE_UNLOGGED, forkNum, blockNum,
+ return ReadBuffer_common(BMR_SMGR(smgr, permanent ? RELPERSISTENCE_PERMANENT :
+ RELPERSISTENCE_UNLOGGED),
+ forkNum, blockNum,
mode, strategy, &hit);
}
@@ -998,7 +993,7 @@ ExtendBufferedRelTo(BufferManagerRelation bmr,
bool hit;
Assert(extended_by == 0);
- buffer = ReadBuffer_common(bmr.smgr, bmr.relpersistence,
+ buffer = ReadBuffer_common(bmr,
fork, extend_to - 1, mode, strategy,
&hit);
}
@@ -1012,18 +1007,11 @@ ExtendBufferedRelTo(BufferManagerRelation bmr,
* *hit is set to true if the request was satisfied from shared buffer cache.
*/
static Buffer
-ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
+ReadBuffer_common(BufferManagerRelation bmr, ForkNumber forkNum,
BlockNumber blockNum, ReadBufferMode mode,
BufferAccessStrategy strategy, bool *hit)
{
- BufferDesc *bufHdr;
- Block bufBlock;
- bool found;
- IOContext io_context;
- IOObject io_object;
- bool isLocalBuf = SmgrIsTemp(smgr);
-
- *hit = false;
+ Buffer buffer;
/*
* Backward compatibility path, most code should use ExtendBufferedRel()
@@ -1042,177 +1030,333 @@ ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
if (mode == RBM_ZERO_AND_LOCK || mode == RBM_ZERO_AND_CLEANUP_LOCK)
flags |= EB_LOCK_FIRST;
- return ExtendBufferedRel(BMR_SMGR(smgr, relpersistence),
- forkNum, strategy, flags);
+ *hit = false;
+
+ return ExtendBufferedRel(bmr, forkNum, strategy, flags);
}
- TRACE_POSTGRESQL_BUFFER_READ_START(forkNum, blockNum,
- smgr->smgr_rlocator.locator.spcOid,
- smgr->smgr_rlocator.locator.dbOid,
- smgr->smgr_rlocator.locator.relNumber,
- smgr->smgr_rlocator.backend);
-
- if (isLocalBuf)
- {
- /*
- * We do not use a BufferAccessStrategy for I/O of temporary tables.
- * However, in some cases, the "strategy" may not be NULL, so we can't
- * rely on IOContextForStrategy() to set the right IOContext for us.
- * This may happen in cases like CREATE TEMPORARY TABLE AS...
- */
- io_context = IOCONTEXT_NORMAL;
- io_object = IOOBJECT_TEMP_RELATION;
- bufHdr = LocalBufferAlloc(smgr, forkNum, blockNum, &found);
- if (found)
- pgBufferUsage.local_blks_hit++;
- else if (mode == RBM_NORMAL || mode == RBM_NORMAL_NO_LOG ||
- mode == RBM_ZERO_ON_ERROR)
- pgBufferUsage.local_blks_read++;
- }
- else
- {
- /*
- * lookup the buffer. IO_IN_PROGRESS is set if the requested block is
- * not currently in memory.
- */
- io_context = IOContextForStrategy(strategy);
- io_object = IOOBJECT_RELATION;
- bufHdr = BufferAlloc(smgr, relpersistence, forkNum, blockNum,
- strategy, &found, io_context);
- if (found)
- pgBufferUsage.shared_blks_hit++;
- else if (mode == RBM_NORMAL || mode == RBM_NORMAL_NO_LOG ||
- mode == RBM_ZERO_ON_ERROR)
- pgBufferUsage.shared_blks_read++;
- }
+ buffer = PrepareReadBuffer(bmr,
+ forkNum,
+ blockNum,
+ strategy,
+ hit);
/* At this point we do NOT hold any locks. */
- /* if it was already in the buffer pool, we're done */
- if (found)
+ if (mode == RBM_ZERO_AND_CLEANUP_LOCK || mode == RBM_ZERO_AND_LOCK)
+ {
+ /* if we just want zeroes and a lock, we're done */
+ ZeroBuffer(buffer, mode);
+ }
+ else if (!*hit)
+ {
+ /* we might need to perform I/O */
+ CompleteReadBuffers(bmr,
+ &buffer,
+ forkNum,
+ blockNum,
+ 1,
+ mode == RBM_ZERO_ON_ERROR,
+ strategy);
+ }
+
+ return buffer;
+}
+
+/*
+ * Prepare to read a block. The buffer is pinned. If this is a 'hit', then
+ * the returned buffer can be used immediately. Otherwise, a physical read
+ * should be completed with CompleteReadBuffers(), or the buffer should be
+ * zeroed with ZeroBuffer(). PrepareReadBuffer() followed by
+ * CompleteReadBuffers() or ZeroBuffer() is equivalent to ReadBuffer(), but
+ * the caller has the opportunity to combine reads of multiple neighboring
+ * blocks into one CompleteReadBuffers() call.
+ *
+ * *foundPtr is set to true for a hit, and false for a miss.
+ */
+Buffer
+PrepareReadBuffer(BufferManagerRelation bmr,
+ ForkNumber forkNum,
+ BlockNumber blockNum,
+ BufferAccessStrategy strategy,
+ bool *foundPtr)
+{
+ BufferDesc *bufHdr;
+ bool isLocalBuf;
+ IOContext io_context;
+ IOObject io_object;
+
+ Assert(blockNum != P_NEW);
+
+ if (bmr.rel)
+ {
+ bmr.smgr = RelationGetSmgr(bmr.rel);
+ bmr.relpersistence = bmr.rel->rd_rel->relpersistence;
+ }
+
+ isLocalBuf = SmgrIsTemp(bmr.smgr);
+ if (isLocalBuf)
+ {
+ io_context = IOCONTEXT_NORMAL;
+ io_object = IOOBJECT_TEMP_RELATION;
+ }
+ else
+ {
+ io_context = IOContextForStrategy(strategy);
+ io_object = IOOBJECT_RELATION;
+ }
+
+ TRACE_POSTGRESQL_BUFFER_READ_START(forkNum, blockNum,
+ bmr.smgr->smgr_rlocator.locator.spcOid,
+ bmr.smgr->smgr_rlocator.locator.dbOid,
+ bmr.smgr->smgr_rlocator.locator.relNumber,
+ bmr.smgr->smgr_rlocator.backend);
+
+ ResourceOwnerEnlarge(CurrentResourceOwner);
+ if (isLocalBuf)
+ {
+ bufHdr = LocalBufferAlloc(bmr.smgr, forkNum, blockNum, foundPtr);
+ if (*foundPtr)
+ pgBufferUsage.local_blks_hit++;
+ }
+ else
+ {
+ bufHdr = BufferAlloc(bmr.smgr, bmr.relpersistence, forkNum, blockNum,
+ strategy, foundPtr, io_context);
+ if (*foundPtr)
+ pgBufferUsage.shared_blks_hit++;
+ }
+ if (bmr.rel)
+ {
+ /*
+ * While pgBufferUsage's "read" counter isn't bumped unless we reach
+ * CompleteReadBuffers() (so, not for hits, and not for buffers that
+ * are zeroed instead), the per-relation stats always count them.
+ */
+ pgstat_count_buffer_read(bmr.rel);
+ if (*foundPtr)
+ pgstat_count_buffer_hit(bmr.rel);
+ }
+ if (*foundPtr)
{
- /* Just need to update stats before we exit */
- *hit = true;
VacuumPageHit++;
pgstat_count_io_op(io_object, io_context, IOOP_HIT);
-
if (VacuumCostActive)
VacuumCostBalance += VacuumCostPageHit;
TRACE_POSTGRESQL_BUFFER_READ_DONE(forkNum, blockNum,
- smgr->smgr_rlocator.locator.spcOid,
- smgr->smgr_rlocator.locator.dbOid,
- smgr->smgr_rlocator.locator.relNumber,
- smgr->smgr_rlocator.backend,
- found);
-
- /*
- * In RBM_ZERO_AND_LOCK mode the caller expects the page to be locked
- * on return.
- */
- if (!isLocalBuf)
- {
- if (mode == RBM_ZERO_AND_LOCK)
- LWLockAcquire(BufferDescriptorGetContentLock(bufHdr),
- LW_EXCLUSIVE);
- else if (mode == RBM_ZERO_AND_CLEANUP_LOCK)
- LockBufferForCleanup(BufferDescriptorGetBuffer(bufHdr));
- }
-
- return BufferDescriptorGetBuffer(bufHdr);
+ bmr.smgr->smgr_rlocator.locator.spcOid,
+ bmr.smgr->smgr_rlocator.locator.dbOid,
+ bmr.smgr->smgr_rlocator.locator.relNumber,
+ bmr.smgr->smgr_rlocator.backend,
+ true);
}
- /*
- * if we have gotten to this point, we have allocated a buffer for the
- * page but its contents are not yet valid. IO_IN_PROGRESS is set for it,
- * if it's a shared buffer.
- */
- Assert(!(pg_atomic_read_u32(&bufHdr->state) & BM_VALID)); /* spinlock not needed */
-
- bufBlock = isLocalBuf ? LocalBufHdrGetBlock(bufHdr) : BufHdrGetBlock(bufHdr);
-
- /*
- * Read in the page, unless the caller intends to overwrite it and just
- * wants us to allocate a buffer.
- */
- if (mode == RBM_ZERO_AND_LOCK || mode == RBM_ZERO_AND_CLEANUP_LOCK)
- MemSet((char *) bufBlock, 0, BLCKSZ);
- else
- {
- instr_time io_start = pgstat_prepare_io_time(track_io_timing);
-
- smgrread(smgr, forkNum, blockNum, bufBlock);
-
- pgstat_count_io_op_time(io_object, io_context,
- IOOP_READ, io_start, 1);
-
- /* check for garbage data */
- if (!PageIsVerifiedExtended((Page) bufBlock, blockNum,
- PIV_LOG_WARNING | PIV_REPORT_STAT))
- {
- if (mode == RBM_ZERO_ON_ERROR || zero_damaged_pages)
- {
- ereport(WARNING,
- (errcode(ERRCODE_DATA_CORRUPTED),
- errmsg("invalid page in block %u of relation %s; zeroing out page",
- blockNum,
- relpath(smgr->smgr_rlocator, forkNum))));
- MemSet((char *) bufBlock, 0, BLCKSZ);
- }
- else
- ereport(ERROR,
- (errcode(ERRCODE_DATA_CORRUPTED),
- errmsg("invalid page in block %u of relation %s",
- blockNum,
- relpath(smgr->smgr_rlocator, forkNum))));
- }
- }
-
- /*
- * In RBM_ZERO_AND_LOCK / RBM_ZERO_AND_CLEANUP_LOCK mode, grab the buffer
- * content lock before marking the page as valid, to make sure that no
- * other backend sees the zeroed page before the caller has had a chance
- * to initialize it.
- *
- * Since no-one else can be looking at the page contents yet, there is no
- * difference between an exclusive lock and a cleanup-strength lock. (Note
- * that we cannot use LockBuffer() or LockBufferForCleanup() here, because
- * they assert that the buffer is already valid.)
- */
- if ((mode == RBM_ZERO_AND_LOCK || mode == RBM_ZERO_AND_CLEANUP_LOCK) &&
- !isLocalBuf)
- {
- LWLockAcquire(BufferDescriptorGetContentLock(bufHdr), LW_EXCLUSIVE);
- }
-
- if (isLocalBuf)
- {
- /* Only need to adjust flags */
- uint32 buf_state = pg_atomic_read_u32(&bufHdr->state);
-
- buf_state |= BM_VALID;
- pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
- }
- else
- {
- /* Set BM_VALID, terminate IO, and wake up any waiters */
- TerminateBufferIO(bufHdr, false, BM_VALID, true);
- }
-
- VacuumPageMiss++;
- if (VacuumCostActive)
- VacuumCostBalance += VacuumCostPageMiss;
-
- TRACE_POSTGRESQL_BUFFER_READ_DONE(forkNum, blockNum,
- smgr->smgr_rlocator.locator.spcOid,
- smgr->smgr_rlocator.locator.dbOid,
- smgr->smgr_rlocator.locator.relNumber,
- smgr->smgr_rlocator.backend,
- found);
-
return BufferDescriptorGetBuffer(bufHdr);
}
+static inline bool
+CompleteReadBuffersCanStartIO(Buffer buffer)
+{
+ if (BufferIsLocal(buffer))
+ {
+ BufferDesc *bufHdr = GetLocalBufferDescriptor(-buffer - 1);
+
+ return (pg_atomic_read_u32(&bufHdr->state) & BM_VALID) == 0;
+ }
+ else
+ return StartBufferIO(GetBufferDescriptor(buffer - 1), true);
+}
+
+/*
+ * Complete a set reads prepared with PrepareReadBuffers(). The buffers must
+ * cover a cluster of neighboring block numbers.
+ *
+ * Typically this performs one physical vector read covering the block range,
+ * but if some of the buffers have already been read in the meantime by any
+ * backend, zero or multiple reads may be performed.
+ */
+void
+CompleteReadBuffers(BufferManagerRelation bmr,
+ Buffer *buffers,
+ ForkNumber forknum,
+ BlockNumber blocknum,
+ int nblocks,
+ bool zero_on_error,
+ BufferAccessStrategy strategy)
+{
+ bool isLocalBuf;
+ IOContext io_context;
+ IOObject io_object;
+
+ if (bmr.rel)
+ {
+ bmr.smgr = RelationGetSmgr(bmr.rel);
+ bmr.relpersistence = bmr.rel->rd_rel->relpersistence;
+ }
+
+ isLocalBuf = SmgrIsTemp(bmr.smgr);
+ if (isLocalBuf)
+ {
+ io_context = IOCONTEXT_NORMAL;
+ io_object = IOOBJECT_TEMP_RELATION;
+ }
+ else
+ {
+ io_context = IOContextForStrategy(strategy);
+ io_object = IOOBJECT_RELATION;
+ }
+
+ /*
+ * We count all these blocks as read by this backend. This is traditional
+ * behavior, but might turn out to be not true if we find that someone
+ * else has beaten us and completed the read of some of these blocks. In
+ * that case the system globally double-counts, but we traditionally don't
+ * count this as a "hit", and we don't have a separate counter for "miss,
+ * but another backend completed the read".
+ */
+ if (isLocalBuf)
+ pgBufferUsage.local_blks_read += nblocks;
+ else
+ pgBufferUsage.shared_blks_read += nblocks;
+
+ for (int i = 0; i < nblocks; ++i)
+ {
+ int io_buffers_len;
+ Buffer io_buffers[MAX_BUFFERS_PER_TRANSFER];
+ void *io_pages[MAX_BUFFERS_PER_TRANSFER];
+ instr_time io_start;
+ BlockNumber io_first_block;
+
+#ifdef USE_ASSERT_CHECKING
+
+ /*
+ * We could get all the information from buffer headers, but it can be
+ * expensive to access buffer header cache lines so we make the caller
+ * provide all the information we need, and assert that it is
+ * consistent.
+ */
+ {
+ RelFileLocator xlocator;
+ ForkNumber xforknum;
+ BlockNumber xblocknum;
+
+ BufferGetTag(buffers[i], &xlocator, &xforknum, &xblocknum);
+ Assert(RelFileLocatorEquals(bmr.smgr->smgr_rlocator.locator, xlocator));
+ Assert(xforknum == forknum);
+ Assert(xblocknum == blocknum + i);
+ }
+#endif
+
+ /* Skip this block if someone else has already completed it. */
+ if (!CompleteReadBuffersCanStartIO(buffers[i]))
+ {
+ /*
+ * Report this as a 'hit' for this backend, even though it must
+ * have started out as a miss in PrepareReadBuffer().
+ */
+ TRACE_POSTGRESQL_BUFFER_READ_DONE(forknum, blocknum + i,
+ bmr.smgr->smgr_rlocator.locator.spcOid,
+ bmr.smgr->smgr_rlocator.locator.dbOid,
+ bmr.smgr->smgr_rlocator.locator.relNumber,
+ bmr.smgr->smgr_rlocator.backend,
+ true);
+ continue;
+ }
+
+ /* We found a buffer that we need to read in. */
+ io_buffers[0] = buffers[i];
+ io_pages[0] = BufferGetBlock(buffers[i]);
+ io_first_block = blocknum + i;
+ io_buffers_len = 1;
+
+ /*
+ * How many neighboring-on-disk blocks can we can scatter-read into
+ * other buffers at the same time?
+ */
+ while ((i + 1) < nblocks &&
+ CompleteReadBuffersCanStartIO(buffers[i + 1]))
+ {
+ /* Must be consecutive block numbers. */
+ Assert(BufferGetBlockNumber(buffers[i + 1]) ==
+ BufferGetBlockNumber(buffers[i]) + 1);
+
+ io_buffers[io_buffers_len] = buffers[++i];
+ io_pages[io_buffers_len++] = BufferGetBlock(buffers[i]);
+ }
+
+ io_start = pgstat_prepare_io_time(track_io_timing);
+ smgrreadv(bmr.smgr, forknum, io_first_block, io_pages, io_buffers_len);
+ pgstat_count_io_op_time(io_object, io_context, IOOP_READ, io_start,
+ io_buffers_len);
+
+ /* Verify each block we read, and terminate the I/O. */
+ for (int j = 0; j < io_buffers_len; ++j)
+ {
+ BufferDesc *bufHdr;
+ Block bufBlock;
+
+ if (isLocalBuf)
+ {
+ bufHdr = GetLocalBufferDescriptor(-io_buffers[j] - 1);
+ bufBlock = LocalBufHdrGetBlock(bufHdr);
+ }
+ else
+ {
+ bufHdr = GetBufferDescriptor(io_buffers[j] - 1);
+ bufBlock = BufHdrGetBlock(bufHdr);
+ }
+
+ /* check for garbage data */
+ if (!PageIsVerifiedExtended((Page) bufBlock, io_first_block + j,
+ PIV_LOG_WARNING | PIV_REPORT_STAT))
+ {
+ if (zero_on_error || zero_damaged_pages)
+ {
+ ereport(WARNING,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("invalid page in block %u of relation %s; zeroing out page",
+ io_first_block + j,
+ relpath(bmr.smgr->smgr_rlocator, forknum))));
+ memset(bufBlock, 0, BLCKSZ);
+ }
+ else
+ ereport(ERROR,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("invalid page in block %u of relation %s",
+ io_first_block + j,
+ relpath(bmr.smgr->smgr_rlocator, forknum))));
+ }
+
+ /* Terminate I/O and set BM_VALID. */
+ if (isLocalBuf)
+ {
+ uint32 buf_state = pg_atomic_read_u32(&bufHdr->state);
+
+ buf_state |= BM_VALID;
+ pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
+ }
+ else
+ {
+ /* Set BM_VALID, terminate IO, and wake up any waiters */
+ TerminateBufferIO(bufHdr, false, BM_VALID, true);
+ }
+
+ /* Report I/Os as completing individually. */
+ TRACE_POSTGRESQL_BUFFER_READ_DONE(forknum, io_first_block + j,
+ bmr.smgr->smgr_rlocator.locator.spcOid,
+ bmr.smgr->smgr_rlocator.locator.dbOid,
+ bmr.smgr->smgr_rlocator.locator.relNumber,
+ bmr.smgr->smgr_rlocator.backend,
+ false);
+ }
+
+ VacuumPageMiss += io_buffers_len;
+ if (VacuumCostActive)
+ VacuumCostBalance += VacuumCostPageMiss * io_buffers_len;
+ }
+}
+
/*
* BufferAlloc -- subroutine for ReadBuffer. Handles lookup of a shared
* buffer. If no buffer exists already, selects a replacement
@@ -1224,11 +1368,8 @@ ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
*
* The returned buffer is pinned and is already marked as holding the
* desired page. If it already did have the desired page, *foundPtr is
- * set true. Otherwise, *foundPtr is set false and the buffer is marked
- * as IO_IN_PROGRESS; ReadBuffer will now need to do I/O to fill it.
- *
- * *foundPtr is actually redundant with the buffer's BM_VALID flag, but
- * we keep it for simplicity in ReadBuffer.
+ * set true. Otherwise, *foundPtr is set false. A read should be
+ * performed with CompleteReadBuffers().
*
* io_context is passed as an output parameter to avoid calling
* IOContextForStrategy() when there is a shared buffers hit and no IO
@@ -1287,19 +1428,10 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
{
/*
* We can only get here if (a) someone else is still reading in
- * the page, or (b) a previous read attempt failed. We have to
- * wait for any active read attempt to finish, and then set up our
- * own read attempt if the page is still not BM_VALID.
- * StartBufferIO does it all.
+ * the page, (b) a previous read attempt failed, or (c) someone
+ * called PrepareReadBuffer() but not yet CompleteReadBuffers().
*/
- if (StartBufferIO(buf, true))
- {
- /*
- * If we get here, previous attempts to read the buffer must
- * have failed ... but we shall bravely try again.
- */
- *foundPtr = false;
- }
+ *foundPtr = false;
}
return buf;
@@ -1364,19 +1496,10 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
{
/*
* We can only get here if (a) someone else is still reading in
- * the page, or (b) a previous read attempt failed. We have to
- * wait for any active read attempt to finish, and then set up our
- * own read attempt if the page is still not BM_VALID.
- * StartBufferIO does it all.
+ * the page, (b) a previous read attempt failed, or (c) someone
+ * called PrepareReadBuffer() but not yet CompleteReadBuffers().
*/
- if (StartBufferIO(existing_buf_hdr, true))
- {
- /*
- * If we get here, previous attempts to read the buffer must
- * have failed ... but we shall bravely try again.
- */
- *foundPtr = false;
- }
+ *foundPtr = false;
}
return existing_buf_hdr;
@@ -1408,15 +1531,9 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
LWLockRelease(newPartitionLock);
/*
- * Buffer contents are currently invalid. Try to obtain the right to
- * start I/O. If StartBufferIO returns false, then someone else managed
- * to read it before we did, so there's nothing left for BufferAlloc() to
- * do.
+ * Buffer contents are currently invalid.
*/
- if (StartBufferIO(victim_buf_hdr, true))
- *foundPtr = false;
- else
- *foundPtr = true;
+ *foundPtr = false;
return victim_buf_hdr;
}
@@ -1770,7 +1887,7 @@ again:
* pessimistic, but outside of toy-sized shared_buffers it should allow
* sufficient pins.
*/
-static void
+void
LimitAdditionalPins(uint32 *additional_pins)
{
uint32 max_backends;
@@ -2373,7 +2490,12 @@ PinBuffer(BufferDesc *buf, BufferAccessStrategy strategy)
else
{
/*
- * If we previously pinned the buffer, it must surely be valid.
+ * If we previously pinned the buffer, it is likely to be valid, but
+ * it may not be if PrepareReadBuffer() was called and
+ * CompleteReadBuffers() hasn't been called yet. We'll check by
+ * loading the flags without locking. This is racy, but it's OK to
+ * return false spuriously: when CompleteReadBuffers() calls
+ * StartBufferIO(), it'll see that it's now valid.
*
* Note: We deliberately avoid a Valgrind client request here.
* Individual access methods can optionally superimpose buffer page
@@ -2382,7 +2504,7 @@ PinBuffer(BufferDesc *buf, BufferAccessStrategy strategy)
* that the buffer page is legitimately non-accessible here. We
* cannot meddle with that.
*/
- result = true;
+ result = (pg_atomic_read_u32(&buf->state) & BM_VALID) != 0;
}
ref->refcount++;
@@ -4833,6 +4955,46 @@ ConditionalLockBuffer(Buffer buffer)
LW_EXCLUSIVE);
}
+/*
+ * Zero a buffer, and lock it as RBM_ZERO_AND_LOCK or
+ * RBM_ZERO_AND_CLEANUP_LOCK would. The buffer must be already pinned. It
+ * does not have to be valid, but it is valid and locked on return.
+ */
+void
+ZeroBuffer(Buffer buffer, ReadBufferMode mode)
+{
+ BufferDesc *bufHdr;
+ uint32 buf_state;
+
+ Assert(mode == RBM_ZERO_AND_LOCK || mode == RBM_ZERO_AND_CLEANUP_LOCK);
+
+ if (BufferIsLocal(buffer))
+ bufHdr = GetLocalBufferDescriptor(-buffer - 1);
+ else
+ {
+ bufHdr = GetBufferDescriptor(buffer - 1);
+ if (mode == RBM_ZERO_AND_LOCK)
+ LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+ else
+ LockBufferForCleanup(buffer);
+ }
+
+ memset(BufferGetPage(buffer), 0, BLCKSZ);
+
+ if (BufferIsLocal(buffer))
+ {
+ buf_state = pg_atomic_read_u32(&bufHdr->state);
+ buf_state |= BM_VALID;
+ pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
+ }
+ else
+ {
+ buf_state = LockBufHdr(bufHdr);
+ buf_state |= BM_VALID;
+ UnlockBufHdr(bufHdr, buf_state);
+ }
+}
+
/*
* Verify that this backend is pinning the buffer exactly once.
*
diff --git a/src/backend/storage/buffer/localbuf.c b/src/backend/storage/buffer/localbuf.c
index 1f02fed250e..6956d4e5b49 100644
--- a/src/backend/storage/buffer/localbuf.c
+++ b/src/backend/storage/buffer/localbuf.c
@@ -109,10 +109,9 @@ PrefetchLocalBuffer(SMgrRelation smgr, ForkNumber forkNum,
* LocalBufferAlloc -
* Find or create a local buffer for the given page of the given relation.
*
- * API is similar to bufmgr.c's BufferAlloc, except that we do not need
- * to do any locking since this is all local. Also, IO_IN_PROGRESS
- * does not get set. Lastly, we support only default access strategy
- * (hence, usage_count is always advanced).
+ * API is similar to bufmgr.c's BufferAlloc, except that we do not need to do
+ * any locking since this is all local. We support only default access
+ * strategy (hence, usage_count is always advanced).
*/
BufferDesc *
LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber blockNum,
@@ -288,7 +287,7 @@ GetLocalVictimBuffer(void)
}
/* see LimitAdditionalPins() */
-static void
+void
LimitAdditionalLocalPins(uint32 *additional_pins)
{
uint32 max_pins;
@@ -298,9 +297,10 @@ LimitAdditionalLocalPins(uint32 *additional_pins)
/*
* In contrast to LimitAdditionalPins() other backends don't play a role
- * here. We can allow up to NLocBuffer pins in total.
+ * here. We can allow up to NLocBuffer pins in total, but it might not be
+ * initialized yet so read num_temp_buffers.
*/
- max_pins = (NLocBuffer - NLocalPinnedBuffers);
+ max_pins = (num_temp_buffers - NLocalPinnedBuffers);
if (*additional_pins >= max_pins)
*additional_pins = max_pins;
diff --git a/src/backend/storage/meson.build b/src/backend/storage/meson.build
index 40345bdca27..739d13293fb 100644
--- a/src/backend/storage/meson.build
+++ b/src/backend/storage/meson.build
@@ -1,5 +1,6 @@
# Copyright (c) 2022-2024, PostgreSQL Global Development Group
+subdir('aio')
subdir('buffer')
subdir('file')
subdir('freespace')
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index d808aad8b05..35657008761 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2094,6 +2094,8 @@ PgStat_TableCounts
PgStat_TableStatus
PgStat_TableXactStatus
PgStat_WalStats
+PgStreamingRead
+PgStreamingReadRange
PgXmlErrorContext
PgXmlStrictness
Pg_finfo_record
--
2.43.0
From ead994a29826be6033c3acc8fca90a6f6cc58a30 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavu...@gmail.com>
Date: Mon, 19 Feb 2024 14:30:47 +0300
Subject: [PATCH v1 2/2] Use streaming read API in ANALYZE
ANALYZE command gets random tuples using BlockSampler algorithm. Use
streaming reads to get these tuples by using BlockSampler algorithm in
streaming read API prefetch logic.
---
src/include/access/tableam.h | 16 ++--
src/backend/access/heap/heapam_handler.c | 11 +--
src/backend/commands/analyze.c | 97 ++++++++----------------
3 files changed, 45 insertions(+), 79 deletions(-)
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index 5f8474871d2..7e6e99ba71d 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -21,6 +21,7 @@
#include "access/sdir.h"
#include "access/xact.h"
#include "executor/tuptable.h"
+#include "storage/streaming_read.h"
#include "utils/rel.h"
#include "utils/snapshot.h"
@@ -648,9 +649,9 @@ typedef struct TableAmRoutine
BufferAccessStrategy bstrategy);
/*
- * Prepare to analyze block `blockno` of `scan`. The scan has been started
- * with table_beginscan_analyze(). See also
- * table_scan_analyze_next_block().
+ * Prepare to analyze next block of `scan`. Next block is decided by
+ * callback function of `pgsr`. The scan has been started with
+ * table_beginscan_analyze(). See also table_scan_analyze_next_block().
*
* The callback may acquire resources like locks that are held until
* table_scan_analyze_next_tuple() returns false. It e.g. can make sense
@@ -665,8 +666,7 @@ typedef struct TableAmRoutine
* isn't one yet.
*/
bool (*scan_analyze_next_block) (TableScanDesc scan,
- BlockNumber blockno,
- BufferAccessStrategy bstrategy);
+ PgStreamingRead *pgsr);
/*
* See table_scan_analyze_next_tuple().
@@ -1714,11 +1714,9 @@ table_relation_vacuum(Relation rel, struct VacuumParams *params,
* Returns false if block is unsuitable for sampling, true otherwise.
*/
static inline bool
-table_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
- BufferAccessStrategy bstrategy)
+table_scan_analyze_next_block(TableScanDesc scan, PgStreamingRead *pgsr)
{
- return scan->rs_rd->rd_tableam->scan_analyze_next_block(scan, blockno,
- bstrategy);
+ return scan->rs_rd->rd_tableam->scan_analyze_next_block(scan, pgsr);
}
/*
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 680a50bf8b1..7ffaf8ac402 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -993,10 +993,10 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
}
static bool
-heapam_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
- BufferAccessStrategy bstrategy)
+heapam_scan_analyze_next_block(TableScanDesc scan, PgStreamingRead *pgsr)
{
HeapScanDesc hscan = (HeapScanDesc) scan;
+ BlockNumber *current_block;
/*
* We must maintain a pin on the target page's buffer to ensure that
@@ -1007,10 +1007,11 @@ heapam_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
* doing much work per tuple, the extra lock traffic is probably better
* avoided.
*/
- hscan->rs_cblock = blockno;
+ hscan->rs_cbuf = pg_streaming_read_buffer_get_next(pgsr, (void **) ¤t_block);
+ hscan->rs_cblock = *current_block;
hscan->rs_cindex = FirstOffsetNumber;
- hscan->rs_cbuf = ReadBufferExtended(scan->rs_rd, MAIN_FORKNUM,
- blockno, RBM_NORMAL, bstrategy);
+
+ Assert(BufferIsValid(hscan->rs_cbuf));
LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_SHARE);
/* in heap all blocks can contain tuples, so always return true */
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index a03495d6c95..7576afcf655 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -1112,6 +1112,26 @@ examine_attribute(Relation onerel, int attnum, Node *index_expr)
return stats;
}
+/*
+ * Prefetch callback function to get next block number while using
+ * BlockSampling algorithm
+ */
+static BlockNumber
+pg_block_sampling_streaming_read_next(PgStreamingRead *pgsr,
+ void *pgsr_private,
+ void *per_io_data)
+{
+ BlockSamplerData *bs = pgsr_private;
+ BlockNumber *current_block = per_io_data;
+
+ if (BlockSampler_HasMore(bs))
+ *current_block = BlockSampler_Next(bs);
+ else
+ *current_block = InvalidBlockNumber;
+
+ return *current_block;
+}
+
/*
* acquire_sample_rows -- acquire a random sample of rows from the table
*
@@ -1164,10 +1184,7 @@ acquire_sample_rows(Relation onerel, int elevel,
TableScanDesc scan;
BlockNumber nblocks;
BlockNumber blksdone = 0;
-#ifdef USE_PREFETCH
- int prefetch_maximum = 0; /* blocks to prefetch if enabled */
- BlockSamplerData prefetch_bs;
-#endif
+ PgStreamingRead *pgsr = NULL;
Assert(targrows > 0);
@@ -1180,13 +1197,6 @@ acquire_sample_rows(Relation onerel, int elevel,
randseed = pg_prng_uint32(&pg_global_prng_state);
nblocks = BlockSampler_Init(&bs, totalblocks, targrows, randseed);
-#ifdef USE_PREFETCH
- prefetch_maximum = get_tablespace_maintenance_io_concurrency(onerel->rd_rel->reltablespace);
- /* Create another BlockSampler, using the same seed, for prefetching */
- if (prefetch_maximum)
- (void) BlockSampler_Init(&prefetch_bs, totalblocks, targrows, randseed);
-#endif
-
/* Report sampling block numbers */
pgstat_progress_update_param(PROGRESS_ANALYZE_BLOCKS_TOTAL,
nblocks);
@@ -1197,68 +1207,23 @@ acquire_sample_rows(Relation onerel, int elevel,
scan = table_beginscan_analyze(onerel);
slot = table_slot_create(onerel, NULL);
-#ifdef USE_PREFETCH
-
- /*
- * If we are doing prefetching, then go ahead and tell the kernel about
- * the first set of pages we are going to want. This also moves our
- * iterator out ahead of the main one being used, where we will keep it so
- * that we're always pre-fetching out prefetch_maximum number of blocks
- * ahead.
- */
- if (prefetch_maximum)
- {
- for (int i = 0; i < prefetch_maximum; i++)
- {
- BlockNumber prefetch_block;
-
- if (!BlockSampler_HasMore(&prefetch_bs))
- break;
-
- prefetch_block = BlockSampler_Next(&prefetch_bs);
- PrefetchBuffer(scan->rs_rd, MAIN_FORKNUM, prefetch_block);
- }
- }
-#endif
+ pgsr = pg_streaming_read_buffer_alloc(PGSR_FLAG_DEFAULT,
+ &bs,
+ sizeof(BlockSamplerData),
+ vac_strategy,
+ BMR_REL(scan->rs_rd),
+ MAIN_FORKNUM,
+ pg_block_sampling_streaming_read_next);
/* Outer loop over blocks to sample */
- while (BlockSampler_HasMore(&bs))
+ while (nblocks)
{
bool block_accepted;
- BlockNumber targblock = BlockSampler_Next(&bs);
-#ifdef USE_PREFETCH
- BlockNumber prefetch_targblock = InvalidBlockNumber;
-
- /*
- * Make sure that every time the main BlockSampler is moved forward
- * that our prefetch BlockSampler also gets moved forward, so that we
- * always stay out ahead.
- */
- if (prefetch_maximum && BlockSampler_HasMore(&prefetch_bs))
- prefetch_targblock = BlockSampler_Next(&prefetch_bs);
-#endif
vacuum_delay_point();
- block_accepted = table_scan_analyze_next_block(scan, targblock, vac_strategy);
+ block_accepted = table_scan_analyze_next_block(scan, pgsr);
-#ifdef USE_PREFETCH
-
- /*
- * When pre-fetching, after we get a block, tell the kernel about the
- * next one we will want, if there's any left.
- *
- * We want to do this even if the table_scan_analyze_next_block() call
- * above decides against analyzing the block it picked.
- */
- if (prefetch_maximum && prefetch_targblock != InvalidBlockNumber)
- PrefetchBuffer(scan->rs_rd, MAIN_FORKNUM, prefetch_targblock);
-#endif
-
- /*
- * Don't analyze if table_scan_analyze_next_block() indicated this
- * block is unsuitable for analyzing.
- */
if (!block_accepted)
continue;
@@ -1309,7 +1274,9 @@ acquire_sample_rows(Relation onerel, int elevel,
pgstat_progress_update_param(PROGRESS_ANALYZE_BLOCKS_DONE,
++blksdone);
+ nblocks--;
}
+ pg_streaming_read_free(pgsr);
ExecDropSingleTupleTableSlot(slot);
table_endscan(scan);
--
2.43.0
