On Sat, Jul 22, 2023 at 10:57:03PM -0700, Nathan Bossart wrote:
> On Sat, Jul 22, 2023 at 07:47:50PM -0400, Tom Lane wrote:
>> I wonder whether we can't provide some alternate definition or "skin"
>> for binaryheap that preserves the Datum API for backend code that wants
>> that, while providing a void *-based API for frontend code to use.
>
> I can give this a try next, but it might be rather #ifdef-heavy.
Here is a sketch of this approach. It required fewer #ifdefs than I was
expecting. At the moment, this one seems like the winner to me.
--
Nathan Bossart
Amazon Web Services: https://aws.amazon.com
>From 13017359cc6dd35f3e465280373b4b82e87b7c5b Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nat...@postgresql.org>
Date: Sat, 22 Jul 2023 15:04:45 -0700
Subject: [PATCH v5 1/3] 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 | 37 +++++++++++++++++-------
src/common/meson.build | 1 +
src/include/lib/binaryheap.h | 23 ++++++++++-----
6 files changed, 44 insertions(+), 20 deletions(-)
rename src/{backend/lib => common}/binaryheap.c (91%)
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 91%
rename from src/backend/lib/binaryheap.c
rename to src/common/binaryheap.c
index 1737546757..a6be4b42ae 100644
--- a/src/backend/lib/binaryheap.c
+++ b/src/common/binaryheap.c
@@ -6,15 +6,22 @@
* Portions Copyright (c) 2012-2023, PostgreSQL Global Development Group
*
* IDENTIFICATION
- * src/backend/lib/binaryheap.c
+ * src/common/binaryheap.c
*
*-------------------------------------------------------------------------
*/
+#ifdef FRONTEND
+#include "postgres_fe.h"
+#else
#include "postgres.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);
@@ -34,7 +41,7 @@ binaryheap_allocate(int capacity, binaryheap_comparator compare, void *arg)
int sz;
binaryheap *heap;
- sz = offsetof(binaryheap, bh_nodes) + sizeof(Datum) * capacity;
+ sz = offsetof(binaryheap, bh_nodes) + sizeof(bh_elem_type) * capacity;
heap = (binaryheap *) palloc(sz);
heap->bh_space = capacity;
heap->bh_compare = compare;
@@ -106,10 +113,14 @@ parent_offset(int i)
* afterwards.
*/
void
-binaryheap_add_unordered(binaryheap *heap, Datum d)
+binaryheap_add_unordered(binaryheap *heap, bh_elem_type 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++;
@@ -138,10 +149,14 @@ binaryheap_build(binaryheap *heap)
* the heap property.
*/
void
-binaryheap_add(binaryheap *heap, Datum d)
+binaryheap_add(binaryheap *heap, bh_elem_type 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);
@@ -154,7 +169,7 @@ binaryheap_add(binaryheap *heap, Datum d)
* without modifying the heap. The caller must ensure that this
* routine is not used on an empty heap. Always O(1).
*/
-Datum
+bh_elem_type
binaryheap_first(binaryheap *heap)
{
Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property);
@@ -169,10 +184,10 @@ binaryheap_first(binaryheap *heap)
* that this routine is not used on an empty heap. O(log n) worst
* case.
*/
-Datum
+bh_elem_type
binaryheap_remove_first(binaryheap *heap)
{
- Datum result;
+ bh_elem_type result;
Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property);
@@ -204,7 +219,7 @@ binaryheap_remove_first(binaryheap *heap)
* sifting the new node down.
*/
void
-binaryheap_replace_first(binaryheap *heap, Datum d)
+binaryheap_replace_first(binaryheap *heap, bh_elem_type d)
{
Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property);
@@ -221,7 +236,7 @@ binaryheap_replace_first(binaryheap *heap, Datum d)
static void
sift_up(binaryheap *heap, int node_off)
{
- Datum node_val = heap->bh_nodes[node_off];
+ bh_elem_type node_val = heap->bh_nodes[node_off];
/*
* Within the loop, the node_off'th array entry is a "hole" that
@@ -232,7 +247,7 @@ sift_up(binaryheap *heap, int node_off)
{
int cmp;
int parent_off;
- Datum parent_val;
+ bh_elem_type parent_val;
/*
* If this node is smaller than its parent, the heap condition is
@@ -264,7 +279,7 @@ sift_up(binaryheap *heap, int node_off)
static void
sift_down(binaryheap *heap, int node_off)
{
- Datum node_val = heap->bh_nodes[node_off];
+ bh_elem_type node_val = heap->bh_nodes[node_off];
/*
* Within the loop, the node_off'th array entry is a "hole" that
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..ae0af68f0d 100644
--- a/src/include/lib/binaryheap.h
+++ b/src/include/lib/binaryheap.h
@@ -11,11 +11,20 @@
#ifndef BINARYHEAP_H
#define BINARYHEAP_H
+/*
+ * XXX: This needs a big comment.
+ */
+#ifdef FRONTEND
+#define bh_elem_type void *
+#else
+#define bh_elem_type Datum
+#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.
*/
-typedef int (*binaryheap_comparator) (Datum a, Datum b, void *arg);
+typedef int (*binaryheap_comparator) (bh_elem_type a, bh_elem_type b, void *arg);
/*
* binaryheap
@@ -34,7 +43,7 @@ typedef struct binaryheap
bool bh_has_heap_property; /* debugging cross-check */
binaryheap_comparator bh_compare;
void *bh_arg;
- Datum bh_nodes[FLEXIBLE_ARRAY_MEMBER];
+ bh_elem_type bh_nodes[FLEXIBLE_ARRAY_MEMBER];
} binaryheap;
extern binaryheap *binaryheap_allocate(int capacity,
@@ -42,12 +51,12 @@ extern binaryheap *binaryheap_allocate(int capacity,
void *arg);
extern void binaryheap_reset(binaryheap *heap);
extern void binaryheap_free(binaryheap *heap);
-extern void binaryheap_add_unordered(binaryheap *heap, Datum d);
+extern void binaryheap_add_unordered(binaryheap *heap, bh_elem_type d);
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_replace_first(binaryheap *heap, Datum d);
+extern void binaryheap_add(binaryheap *heap, bh_elem_type d);
+extern bh_elem_type binaryheap_first(binaryheap *heap);
+extern bh_elem_type binaryheap_remove_first(binaryheap *heap);
+extern void binaryheap_replace_first(binaryheap *heap, bh_elem_type d);
#define binaryheap_empty(h) ((h)->bh_size == 0)
--
2.25.1
>From 36eefefc9a9597fe9671aa084903a699d16a1144 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nat...@postgresql.org>
Date: Thu, 20 Jul 2023 09:52:20 -0700
Subject: [PATCH v5 2/3] expand binaryheap api
---
src/common/binaryheap.c | 29 +++++++++++++++++++++++++++++
src/include/lib/binaryheap.h | 3 +++
2 files changed, 32 insertions(+)
diff --git a/src/common/binaryheap.c b/src/common/binaryheap.c
index a6be4b42ae..d24a61fd2b 100644
--- a/src/common/binaryheap.c
+++ b/src/common/binaryheap.c
@@ -211,6 +211,35 @@ binaryheap_remove_first(binaryheap *heap)
return result;
}
+/*
+ * binaryheap_remove_node
+ *
+ * Removes the nth node from the heap. The caller must ensure that there are
+ * at least (n - 1) nodes in 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 ae0af68f0d..84c08d7ebb 100644
--- a/src/include/lib/binaryheap.h
+++ b/src/include/lib/binaryheap.h
@@ -56,8 +56,11 @@ extern void binaryheap_build(binaryheap *heap);
extern void binaryheap_add(binaryheap *heap, bh_elem_type d);
extern bh_elem_type binaryheap_first(binaryheap *heap);
extern bh_elem_type binaryheap_remove_first(binaryheap *heap);
+extern void binaryheap_remove_node(binaryheap *heap, int n);
extern void binaryheap_replace_first(binaryheap *heap, bh_elem_type 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 71d8c2f4fb3945d90d1e4af174ff8b40175449bf Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nat...@postgresql.org>
Date: Thu, 20 Jul 2023 10:19:08 -0700
Subject: [PATCH v5 3/3] use priority queue for pg_restore ready_list
---
src/bin/pg_dump/pg_backup_archiver.c | 192 ++++++++-------------------
1 file changed, 52 insertions(+), 140 deletions(-)
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 39ebcfec32..5cfddf3916 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 void move_to_ready_list(TocEntry *pending_list,
- ParallelReadyList *ready_list,
+static int TocEntrySizeCompareQsort(const void *p1, const void *p2);
+static int TocEntrySizeCompareBinaryheap(void *p1, void *p2, void *arg);
+static void move_to_ready_heap(TocEntry *pending_list,
+ binaryheap *ready_heap,
RestorePass pass);
-static TocEntry *pop_next_work_item(ParallelReadyList *ready_list,
+static TocEntry *pop_next_work_item(binaryheap *ready_heap,
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_heap);
static void mark_create_done(ArchiveHandle *AH, TocEntry *te);
static void inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te);
@@ -2380,7 +2359,7 @@ WriteDataChunks(ArchiveHandle *AH, ParallelState *pstate)
}
if (ntes > 1)
- qsort(tes, ntes, sizeof(TocEntry *), TocEntrySizeCompare);
+ qsort(tes, ntes, sizeof(TocEntry *), TocEntrySizeCompareQsort);
for (int i = 0; i < ntes; i++)
DispatchJobForTocEntry(AH, pstate, tes[i], ACT_DUMP,
@@ -3980,7 +3959,7 @@ restore_toc_entries_prefork(ArchiveHandle *AH, TocEntry *pending_list)
(void) restore_toc_entry(AH, next_work_item, false);
- /* Reduce dependencies, but don't move anything to ready_list */
+ /* Reduce dependencies, but don't move anything to ready_heap */
reduce_dependencies(AH, next_work_item, NULL);
}
else
@@ -4023,24 +4002,26 @@ static void
restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
TocEntry *pending_list)
{
- ParallelReadyList ready_list;
+ binaryheap *ready_heap;
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);
+ /* Set up ready_heap with enough room for all known TocEntrys */
+ ready_heap = binaryheap_allocate(AH->tocCount,
+ TocEntrySizeCompareBinaryheap,
+ NULL);
/*
* The pending_list contains all items that we need to restore. Move all
- * items that are available to process immediately into the ready_list.
+ * items that are available to process immediately into the ready_heap.
* After this setup, the pending list is everything that needs to be done
- * but is blocked by one or more dependencies, while the ready list
+ * but is blocked by one or more dependencies, while the ready heap
* contains items that have no remaining dependencies and are OK to
* process in the current restore pass.
*/
AH->restorePass = RESTORE_PASS_MAIN;
- move_to_ready_list(pending_list, &ready_list, AH->restorePass);
+ move_to_ready_heap(pending_list, ready_heap, 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_heap, 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_heap);
/* 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_heap);
}
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_heap(pending_list, ready_heap, AH->restorePass);
/* Loop around to see if anything's now ready */
continue;
}
@@ -4118,10 +4099,10 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
next_work_item ? WFW_ONE_IDLE : WFW_GOT_STATUS);
}
- /* There should now be nothing in ready_list. */
- Assert(ready_list.first_te > ready_list.last_te);
+ /* There should now be nothing in ready_heap. */
+ Assert(binaryheap_empty(ready_heap));
- ready_list_free(&ready_list);
+ binaryheap_free(ready_heap);
pg_log_info("finished main parallel loop");
}
@@ -4221,80 +4202,9 @@ 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)
+TocEntrySizeCompareQsort(const void *p1, const void *p2)
{
const TocEntry *te1 = *(const TocEntry *const *) p1;
const TocEntry *te2 = *(const TocEntry *const *) p2;
@@ -4314,17 +4224,24 @@ TocEntrySizeCompare(const void *p1, const void *p2)
return 0;
}
+/* binaryheap comparator for sorting TocEntries by dataLength */
+static int
+TocEntrySizeCompareBinaryheap(void *p1, void *p2, void *arg)
+{
+ return TocEntrySizeCompareQsort(p1, p2);
+}
+
/*
- * Move all immediately-ready items from pending_list to ready_list.
+ * Move all immediately-ready items from pending_list to ready_heap.
*
* Items are considered ready if they have no remaining dependencies and
* they belong in the current restore pass. (See also reduce_dependencies,
* which applies the same logic one-at-a-time.)
*/
static void
-move_to_ready_list(TocEntry *pending_list,
- ParallelReadyList *ready_list,
+move_to_ready_heap(TocEntry *pending_list,
+ binaryheap *ready_heap,
RestorePass pass)
{
TocEntry *te;
@@ -4340,38 +4257,33 @@ 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);
+ /* ... and add to ready_heap */
+ binaryheap_add(ready_heap, te);
}
}
}
/*
* Find the next work item (if any) that is capable of being run now,
- * and remove it from the ready_list.
+ * and remove it from the ready_heap.
*
* Returns the item, or NULL if nothing is runnable.
*
* To qualify, the item must have no remaining dependencies
* and no requirements for locks that are incompatible with
- * items currently running. Items in the ready_list are known to have
+ * items currently running. Items in the ready_heap are known to have
* 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_heap,
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.
+ * Search the ready_heap 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_heap); i++)
{
- TocEntry *te = ready_list->tes[i];
+ TocEntry *te = (TocEntry *) binaryheap_get_node(ready_heap, i);
bool conflicts = false;
/*
@@ -4397,7 +4309,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_heap, i);
return te;
}
@@ -4443,7 +4355,7 @@ mark_restore_job_done(ArchiveHandle *AH,
int status,
void *callback_data)
{
- ParallelReadyList *ready_list = (ParallelReadyList *) callback_data;
+ binaryheap *ready_heap = (binaryheap *) callback_data;
pg_log_info("finished item %d %s %s",
te->dumpId, te->desc, te->tag);
@@ -4461,7 +4373,7 @@ mark_restore_job_done(ArchiveHandle *AH,
pg_fatal("worker process failed: exit code %d",
status);
- reduce_dependencies(AH, te, ready_list);
+ reduce_dependencies(AH, te, ready_heap);
}
@@ -4704,11 +4616,11 @@ identify_locking_dependencies(ArchiveHandle *AH, TocEntry *te)
/*
* Remove the specified TOC entry from the depCounts of items that depend on
* it, thereby possibly making them ready-to-run. Any pending item that
- * becomes ready should be moved to the ready_list, if that's provided.
+ * becomes ready should be moved to the ready_heap, if that's provided.
*/
static void
reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
- ParallelReadyList *ready_list)
+ binaryheap *ready_heap)
{
int i;
@@ -4726,18 +4638,18 @@ reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
* the current restore pass, and it is currently a member of the
* pending list (that check is needed to prevent double restore in
* some cases where a list-file forces out-of-order restoring).
- * However, if ready_list == NULL then caller doesn't want any list
+ * However, if ready_heap == NULL then caller doesn't want any list
* memberships changed.
*/
if (otherte->depCount == 0 &&
_tocEntryRestorePass(otherte) == AH->restorePass &&
otherte->pending_prev != NULL &&
- ready_list != NULL)
+ ready_heap != NULL)
{
/* Remove it from pending list ... */
pending_list_remove(otherte);
- /* ... and add to ready_list */
- ready_list_insert(ready_list, otherte);
+ /* ... and add to ready_heap */
+ binaryheap_add(ready_heap, otherte);
}
}
}
--
2.25.1
