Here is a work-in-progress patch set for converting ready_list to a
priority queue. On my machine, Tom's 100k-table example [0] takes 11.5
minutes without these patches and 1.5 minutes with them.
One item that requires more thought is binaryheap's use of Datum. AFAICT
the Datum definitions live in postgres.h and aren't available to frontend
code. I think we'll either need to move the Datum definitions to c.h or to
adjust binaryheap to use "void *".
[0] https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us
--
Nathan Bossart
Amazon Web Services: https://aws.amazon.com
>From bbf7b32f0abd0e9e2f7a0759cc79fcccbb5f9281 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nat...@postgresql.org>
Date: Wed, 19 Jul 2023 15:54:00 -0700
Subject: [PATCH v2 1/4] misc binaryheap fixes
---
src/backend/postmaster/pgarch.c | 12 ++++++------
src/backend/replication/logical/reorderbuffer.c | 2 +-
2 files changed, 7 insertions(+), 7 deletions(-)
diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c
index 46af349564..c28e6f2070 100644
--- a/src/backend/postmaster/pgarch.c
+++ b/src/backend/postmaster/pgarch.c
@@ -108,7 +108,7 @@ static ArchiveModuleState *archive_module_state;
* archive_status. Minimizing the number of directory scans when there are
* many files to archive can significantly improve archival rate.
*
- * arch_heap is a max-heap that is used during the directory scan to track
+ * arch_heap is a min-heap that is used during the directory scan to track
* the highest-priority files to archive. After the directory scan
* completes, the file names are stored in ascending order of priority in
* arch_files. pgarch_readyXlog() returns files from arch_files until it
@@ -248,7 +248,7 @@ PgArchiverMain(void)
arch_files = palloc(sizeof(struct arch_files_state));
arch_files->arch_files_size = 0;
- /* Initialize our max-heap for prioritizing files to archive. */
+ /* Initialize our min-heap for prioritizing files to archive. */
arch_files->arch_heap = binaryheap_allocate(NUM_FILES_PER_DIRECTORY_SCAN,
ready_file_comparator, NULL);
@@ -624,7 +624,7 @@ pgarch_readyXlog(char *xlog)
basename[basenamelen] = '\0';
/*
- * Store the file in our max-heap if it has a high enough priority.
+ * Store the file in our min-heap if it has a high enough priority.
*/
if (arch_files->arch_heap->bh_size < NUM_FILES_PER_DIRECTORY_SCAN)
{
@@ -644,15 +644,15 @@ pgarch_readyXlog(char *xlog)
* Remove the lowest priority file and add the current one to the
* heap.
*/
- arch_file = DatumGetCString(binaryheap_remove_first(arch_files->arch_heap));
+ arch_file = DatumGetCString(binaryheap_first(arch_files->arch_heap));
strcpy(arch_file, basename);
- binaryheap_add(arch_files->arch_heap, CStringGetDatum(arch_file));
+ binaryheap_replace_first(arch_files->arch_heap, CStringGetDatum(arch_file));
}
}
FreeDir(rldir);
/* If no files were found, simply return. */
- if (arch_files->arch_heap->bh_size == 0)
+ if (binaryheap_empty(arch_files->arch_heap))
return false;
/*
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 26d252bd87..05bbcecd29 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -1381,7 +1381,7 @@ ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
int32 off;
/* nothing there anymore */
- if (state->heap->bh_size == 0)
+ if (binaryheap_empty(state->heap))
return NULL;
off = DatumGetInt32(binaryheap_first(state->heap));
--
2.25.1
>From 8f0339263dd21bb04c1b96a0e850a9b57a2e3f8b Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nat...@postgresql.org>
Date: Thu, 20 Jul 2023 09:45:03 -0700
Subject: [PATCH v2 2/4] make binaryheap available to frontend
---
src/backend/lib/Makefile | 1 -
src/backend/lib/meson.build | 1 -
src/common/Makefile | 1 +
src/{backend/lib => common}/binaryheap.c | 15 ++++++++++++
src/common/meson.build | 1 +
src/include/lib/binaryheap.h | 29 ++++++++++++++++++++++++
6 files changed, 46 insertions(+), 2 deletions(-)
rename src/{backend/lib => common}/binaryheap.c (96%)
diff --git a/src/backend/lib/Makefile b/src/backend/lib/Makefile
index 9dad31398a..b6cefd9cca 100644
--- a/src/backend/lib/Makefile
+++ b/src/backend/lib/Makefile
@@ -13,7 +13,6 @@ top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
OBJS = \
- binaryheap.o \
bipartite_match.o \
bloomfilter.o \
dshash.o \
diff --git a/src/backend/lib/meson.build b/src/backend/lib/meson.build
index 974cab8776..b4e88f54ae 100644
--- a/src/backend/lib/meson.build
+++ b/src/backend/lib/meson.build
@@ -1,7 +1,6 @@
# Copyright (c) 2022-2023, PostgreSQL Global Development Group
backend_sources += files(
- 'binaryheap.c',
'bipartite_match.c',
'bloomfilter.c',
'dshash.c',
diff --git a/src/common/Makefile b/src/common/Makefile
index 113029bf7b..cc5c54dcee 100644
--- a/src/common/Makefile
+++ b/src/common/Makefile
@@ -48,6 +48,7 @@ LIBS += $(PTHREAD_LIBS)
OBJS_COMMON = \
archive.o \
base64.o \
+ binaryheap.o \
checksum_helper.o \
compression.o \
config_info.o \
diff --git a/src/backend/lib/binaryheap.c b/src/common/binaryheap.c
similarity index 96%
rename from src/backend/lib/binaryheap.c
rename to src/common/binaryheap.c
index 1737546757..7a590eec4a 100644
--- a/src/backend/lib/binaryheap.c
+++ b/src/common/binaryheap.c
@@ -11,10 +11,17 @@
*-------------------------------------------------------------------------
*/
+#ifndef FRONTEND
#include "postgres.h"
+#else
+#include "postgres_fe.h"
+#endif
#include <math.h>
+#ifdef FRONTEND
+#include "common/logging.h"
+#endif
#include "lib/binaryheap.h"
static void sift_down(binaryheap *heap, int node_off);
@@ -109,7 +116,11 @@ void
binaryheap_add_unordered(binaryheap *heap, Datum d)
{
if (heap->bh_size >= heap->bh_space)
+#ifdef FRONTEND
+ pg_fatal("out of binary heap slots");
+#else
elog(ERROR, "out of binary heap slots");
+#endif
heap->bh_has_heap_property = false;
heap->bh_nodes[heap->bh_size] = d;
heap->bh_size++;
@@ -141,7 +152,11 @@ void
binaryheap_add(binaryheap *heap, Datum d)
{
if (heap->bh_size >= heap->bh_space)
+#ifdef FRONTEND
+ pg_fatal("out of binary heap slots");
+#else
elog(ERROR, "out of binary heap slots");
+#endif
heap->bh_nodes[heap->bh_size] = d;
heap->bh_size++;
sift_up(heap, heap->bh_size - 1);
diff --git a/src/common/meson.build b/src/common/meson.build
index 53942a9a61..3b97497d1a 100644
--- a/src/common/meson.build
+++ b/src/common/meson.build
@@ -3,6 +3,7 @@
common_sources = files(
'archive.c',
'base64.c',
+ 'binaryheap.c',
'checksum_helper.c',
'compression.c',
'controldata_utils.c',
diff --git a/src/include/lib/binaryheap.h b/src/include/lib/binaryheap.h
index 52f7b06b25..06e3878cf7 100644
--- a/src/include/lib/binaryheap.h
+++ b/src/include/lib/binaryheap.h
@@ -11,6 +11,35 @@
#ifndef BINARYHEAP_H
#define BINARYHEAP_H
+/*
+ * XXX: This is obviously a hack.
+ */
+#ifdef FRONTEND
+
+typedef uintptr_t Datum;
+
+/*
+ * DatumGetPointer
+ * Returns pointer value of a datum.
+ */
+static inline Pointer
+DatumGetPointer(Datum X)
+{
+ return (Pointer) X;
+}
+
+/*
+ * PointerGetDatum
+ * Returns datum representation for a pointer.
+ */
+static inline Datum
+PointerGetDatum(const void *X)
+{
+ return (Datum) X;
+}
+
+#endif
+
/*
* For a max-heap, the comparator must return <0 iff a < b, 0 iff a == b,
* and >0 iff a > b. For a min-heap, the conditions are reversed.
--
2.25.1
>From 533f82b9ee4bfb5dee6cfc8e5c7e8091087e3e0b Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nat...@postgresql.org>
Date: Thu, 20 Jul 2023 09:52:20 -0700
Subject: [PATCH v2 3/4] expand binaryheap api
---
src/common/binaryheap.c | 28 ++++++++++++++++++++++++++++
src/include/lib/binaryheap.h | 3 +++
2 files changed, 31 insertions(+)
diff --git a/src/common/binaryheap.c b/src/common/binaryheap.c
index 7a590eec4a..526e56bbd0 100644
--- a/src/common/binaryheap.c
+++ b/src/common/binaryheap.c
@@ -211,6 +211,34 @@ binaryheap_remove_first(binaryheap *heap)
return result;
}
+/*
+ * binaryheap_remove_node
+ *
+ * Removes the given node from the heap. O(log n) worst case.
+ */
+void
+binaryheap_remove_node(binaryheap *heap, int n)
+{
+ int cmp;
+
+ Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property);
+ Assert(n >= 0 && n < heap->bh_size);
+
+ /* compare last node to the one that is being removed */
+ cmp = heap->bh_compare(heap->bh_nodes[--heap->bh_size],
+ heap->bh_nodes[n],
+ heap->bh_arg);
+
+ /* remove the last node, placing it in the vacated entry */
+ heap->bh_nodes[n] = heap->bh_nodes[heap->bh_size];
+
+ /* sift as needed to preserve the heap property */
+ if (cmp > 0)
+ sift_up(heap, n);
+ else if (cmp < 0)
+ sift_down(heap, n);
+}
+
/*
* binaryheap_replace_first
*
diff --git a/src/include/lib/binaryheap.h b/src/include/lib/binaryheap.h
index 06e3878cf7..9f09aff29f 100644
--- a/src/include/lib/binaryheap.h
+++ b/src/include/lib/binaryheap.h
@@ -76,8 +76,11 @@ extern void binaryheap_build(binaryheap *heap);
extern void binaryheap_add(binaryheap *heap, Datum d);
extern Datum binaryheap_first(binaryheap *heap);
extern Datum binaryheap_remove_first(binaryheap *heap);
+extern void binaryheap_remove_node(binaryheap *heap, int n);
extern void binaryheap_replace_first(binaryheap *heap, Datum d);
#define binaryheap_empty(h) ((h)->bh_size == 0)
+#define binaryheap_size(h) ((h)->bh_size)
+#define binaryheap_get_node(h, n) ((h)->bh_nodes[n])
#endif /* BINARYHEAP_H */
--
2.25.1
>From 997999700689d0a8dc4d01efc59b168974362992 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nat...@postgresql.org>
Date: Thu, 20 Jul 2023 10:19:08 -0700
Subject: [PATCH v2 4/4] use priority queue for pg_restore ready_list
---
src/bin/pg_dump/pg_backup_archiver.c | 155 ++++++---------------------
1 file changed, 35 insertions(+), 120 deletions(-)
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 39ebcfec32..7c0e50353d 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -34,6 +34,7 @@
#include "compress_io.h"
#include "dumputils.h"
#include "fe_utils/string_utils.h"
+#include "lib/binaryheap.h"
#include "lib/stringinfo.h"
#include "libpq/libpq-fs.h"
#include "parallel.h"
@@ -44,24 +45,6 @@
#define TEXT_DUMP_HEADER "--\n-- PostgreSQL database dump\n--\n\n"
#define TEXT_DUMPALL_HEADER "--\n-- PostgreSQL database cluster dump\n--\n\n"
-/*
- * State for tracking TocEntrys that are ready to process during a parallel
- * restore. (This used to be a list, and we still call it that, though now
- * it's really an array so that we can apply qsort to it.)
- *
- * tes[] is sized large enough that we can't overrun it.
- * The valid entries are indexed first_te .. last_te inclusive.
- * We periodically sort the array to bring larger-by-dataLength entries to
- * the front; "sorted" is true if the valid entries are known sorted.
- */
-typedef struct _parallelReadyList
-{
- TocEntry **tes; /* Ready-to-dump TocEntrys */
- int first_te; /* index of first valid entry in tes[] */
- int last_te; /* index of last valid entry in tes[] */
- bool sorted; /* are valid entries currently sorted? */
-} ParallelReadyList;
-
static ArchiveHandle *_allocAH(const char *FileSpec, const ArchiveFormat fmt,
const pg_compress_specification compression_spec,
@@ -110,16 +93,12 @@ static void restore_toc_entries_postfork(ArchiveHandle *AH,
static void pending_list_header_init(TocEntry *l);
static void pending_list_append(TocEntry *l, TocEntry *te);
static void pending_list_remove(TocEntry *te);
-static void ready_list_init(ParallelReadyList *ready_list, int tocCount);
-static void ready_list_free(ParallelReadyList *ready_list);
-static void ready_list_insert(ParallelReadyList *ready_list, TocEntry *te);
-static void ready_list_remove(ParallelReadyList *ready_list, int i);
-static void ready_list_sort(ParallelReadyList *ready_list);
static int TocEntrySizeCompare(const void *p1, const void *p2);
+static int TocEntrySizeCompareBinaryHeap(Datum p1, Datum p2, void *arg);
static void move_to_ready_list(TocEntry *pending_list,
- ParallelReadyList *ready_list,
+ binaryheap *ready_list,
RestorePass pass);
-static TocEntry *pop_next_work_item(ParallelReadyList *ready_list,
+static TocEntry *pop_next_work_item(binaryheap *ready_list,
ParallelState *pstate);
static void mark_dump_job_done(ArchiveHandle *AH,
TocEntry *te,
@@ -134,7 +113,7 @@ static bool has_lock_conflicts(TocEntry *te1, TocEntry *te2);
static void repoint_table_dependencies(ArchiveHandle *AH);
static void identify_locking_dependencies(ArchiveHandle *AH, TocEntry *te);
static void reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
- ParallelReadyList *ready_list);
+ binaryheap *ready_list);
static void mark_create_done(ArchiveHandle *AH, TocEntry *te);
static void inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te);
@@ -4023,13 +4002,15 @@ static void
restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
TocEntry *pending_list)
{
- ParallelReadyList ready_list;
+ binaryheap *ready_list;
TocEntry *next_work_item;
pg_log_debug("entering restore_toc_entries_parallel");
/* Set up ready_list with enough room for all known TocEntrys */
- ready_list_init(&ready_list, AH->tocCount);
+ ready_list = binaryheap_allocate(AH->tocCount,
+ TocEntrySizeCompareBinaryHeap,
+ NULL);
/*
* The pending_list contains all items that we need to restore. Move all
@@ -4040,7 +4021,7 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
* process in the current restore pass.
*/
AH->restorePass = RESTORE_PASS_MAIN;
- move_to_ready_list(pending_list, &ready_list, AH->restorePass);
+ move_to_ready_list(pending_list, ready_list, AH->restorePass);
/*
* main parent loop
@@ -4054,7 +4035,7 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
for (;;)
{
/* Look for an item ready to be dispatched to a worker */
- next_work_item = pop_next_work_item(&ready_list, pstate);
+ next_work_item = pop_next_work_item(ready_list, pstate);
if (next_work_item != NULL)
{
/* If not to be restored, don't waste time launching a worker */
@@ -4064,7 +4045,7 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
next_work_item->dumpId,
next_work_item->desc, next_work_item->tag);
/* Update its dependencies as though we'd completed it */
- reduce_dependencies(AH, next_work_item, &ready_list);
+ reduce_dependencies(AH, next_work_item, ready_list);
/* Loop around to see if anything else can be dispatched */
continue;
}
@@ -4075,7 +4056,7 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
/* Dispatch to some worker */
DispatchJobForTocEntry(AH, pstate, next_work_item, ACT_RESTORE,
- mark_restore_job_done, &ready_list);
+ mark_restore_job_done, ready_list);
}
else if (IsEveryWorkerIdle(pstate))
{
@@ -4089,7 +4070,7 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
/* Advance to next restore pass */
AH->restorePass++;
/* That probably allows some stuff to be made ready */
- move_to_ready_list(pending_list, &ready_list, AH->restorePass);
+ move_to_ready_list(pending_list, ready_list, AH->restorePass);
/* Loop around to see if anything's now ready */
continue;
}
@@ -4119,9 +4100,9 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
}
/* There should now be nothing in ready_list. */
- Assert(ready_list.first_te > ready_list.last_te);
+ Assert(binaryheap_empty(ready_list));
- ready_list_free(&ready_list);
+ binaryheap_free(ready_list);
pg_log_info("finished main parallel loop");
}
@@ -4221,77 +4202,6 @@ pending_list_remove(TocEntry *te)
}
-/*
- * Initialize the ready_list with enough room for up to tocCount entries.
- */
-static void
-ready_list_init(ParallelReadyList *ready_list, int tocCount)
-{
- ready_list->tes = (TocEntry **)
- pg_malloc(tocCount * sizeof(TocEntry *));
- ready_list->first_te = 0;
- ready_list->last_te = -1;
- ready_list->sorted = false;
-}
-
-/*
- * Free storage for a ready_list.
- */
-static void
-ready_list_free(ParallelReadyList *ready_list)
-{
- pg_free(ready_list->tes);
-}
-
-/* Add te to the ready_list */
-static void
-ready_list_insert(ParallelReadyList *ready_list, TocEntry *te)
-{
- ready_list->tes[++ready_list->last_te] = te;
- /* List is (probably) not sorted anymore. */
- ready_list->sorted = false;
-}
-
-/* Remove the i'th entry in the ready_list */
-static void
-ready_list_remove(ParallelReadyList *ready_list, int i)
-{
- int f = ready_list->first_te;
-
- Assert(i >= f && i <= ready_list->last_te);
-
- /*
- * In the typical case where the item to be removed is the first ready
- * entry, we need only increment first_te to remove it. Otherwise, move
- * the entries before it to compact the list. (This preserves sortedness,
- * if any.) We could alternatively move the entries after i, but there
- * are typically many more of those.
- */
- if (i > f)
- {
- TocEntry **first_te_ptr = &ready_list->tes[f];
-
- memmove(first_te_ptr + 1, first_te_ptr, (i - f) * sizeof(TocEntry *));
- }
- ready_list->first_te++;
-}
-
-/* Sort the ready_list into the desired order */
-static void
-ready_list_sort(ParallelReadyList *ready_list)
-{
- if (!ready_list->sorted)
- {
- int n = ready_list->last_te - ready_list->first_te + 1;
-
- if (n > 1)
- qsort(ready_list->tes + ready_list->first_te, n,
- sizeof(TocEntry *),
- TocEntrySizeCompare);
- ready_list->sorted = true;
- }
-}
-
/* qsort comparator for sorting TocEntries by dataLength */
static int
TocEntrySizeCompare(const void *p1, const void *p2)
@@ -4314,6 +4224,15 @@ TocEntrySizeCompare(const void *p1, const void *p2)
return 0;
}
+static int
+TocEntrySizeCompareBinaryHeap(Datum p1, Datum p2, void *arg)
+{
+ const void *te1 = (const void *) DatumGetPointer(p1);
+ const void *te2 = (const void *) DatumGetPointer(p2);
+
+ return TocEntrySizeCompare(te1, te2);
+}
+
/*
* Move all immediately-ready items from pending_list to ready_list.
@@ -4324,7 +4243,7 @@ TocEntrySizeCompare(const void *p1, const void *p2)
*/
static void
move_to_ready_list(TocEntry *pending_list,
- ParallelReadyList *ready_list,
+ binaryheap *ready_list,
RestorePass pass)
{
TocEntry *te;
@@ -4341,7 +4260,7 @@ move_to_ready_list(TocEntry *pending_list,
/* Remove it from pending_list ... */
pending_list_remove(te);
/* ... and add to ready_list */
- ready_list_insert(ready_list, te);
+ binaryheap_add(ready_list, PointerGetDatum(te));
}
}
}
@@ -4358,20 +4277,16 @@ move_to_ready_list(TocEntry *pending_list,
* no remaining dependencies, but we have to check for lock conflicts.
*/
static TocEntry *
-pop_next_work_item(ParallelReadyList *ready_list,
+pop_next_work_item(binaryheap *ready_list,
ParallelState *pstate)
{
- /*
- * Sort the ready_list so that we'll tackle larger jobs first.
- */
- ready_list_sort(ready_list);
-
/*
* Search the ready_list until we find a suitable item.
*/
- for (int i = ready_list->first_te; i <= ready_list->last_te; i++)
+ for (int i = 0; i < binaryheap_size(ready_list); i++)
{
- TocEntry *te = ready_list->tes[i];
+ Datum ted = binaryheap_get_node(ready_list, i);
+ TocEntry *te = (TocEntry *) DatumGetPointer(ted);
bool conflicts = false;
/*
@@ -4397,7 +4312,7 @@ pop_next_work_item(ParallelReadyList *ready_list,
continue;
/* passed all tests, so this item can run */
- ready_list_remove(ready_list, i);
+ binaryheap_remove_node(ready_list, i);
return te;
}
@@ -4443,7 +4358,7 @@ mark_restore_job_done(ArchiveHandle *AH,
int status,
void *callback_data)
{
- ParallelReadyList *ready_list = (ParallelReadyList *) callback_data;
+ binaryheap *ready_list = (binaryheap *) callback_data;
pg_log_info("finished item %d %s %s",
te->dumpId, te->desc, te->tag);
@@ -4708,7 +4623,7 @@ identify_locking_dependencies(ArchiveHandle *AH, TocEntry *te)
*/
static void
reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
- ParallelReadyList *ready_list)
+ binaryheap *ready_list)
{
int i;
@@ -4737,7 +4652,7 @@ reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
/* Remove it from pending list ... */
pending_list_remove(otherte);
/* ... and add to ready_list */
- ready_list_insert(ready_list, otherte);
+ binaryheap_add(ready_list, PointerGetDatum(otherte));
}
}
}
--
2.25.1
