Rebased to correct for pg_indent changes.
Applies cleanly.
Compiles cleanly.
Passes regression tests.
Comments and format look good.
No documentation changes needed.
No regression test changes needed.
Performance tests to follow in a day or two.
-Kevin
Index: src/bin/pg_dump/pg_backup_archiver.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/bin/pg_dump/pg_backup_archiver.c,v
retrieving revision 1.172
diff -c -d -r1.172 pg_backup_archiver.c
*** src/bin/pg_dump/pg_backup_archiver.c 11 Jun 2009 14:49:07 -0000
1.172
--- src/bin/pg_dump/pg_backup_archiver.c 17 Jul 2009 02:20:28 -0000
***************
*** 59,70 ****
--- 59,80 ----
#define thandle HANDLE
#endif
+ /* List header for pending-activity lists */
+ typedef struct
+ {
+ TocEntry *head;
+ TocEntry *tail;
+ /* The list link fields in each TocEntry are par_prev and par_next */
+ } TocEntryList;
+
+ /* Arguments needed for a worker child */
typedef struct _restore_args
{
ArchiveHandle *AH;
TocEntry *te;
} RestoreArgs;
+ /* State for each parallel activity slot */
typedef struct _parallel_slot
{
thandle child_id;
***************
*** 117,124 ****
static thandle reap_child(ParallelSlot *slots, int n_slots, int *work_status);
static bool work_in_progress(ParallelSlot *slots, int n_slots);
static int get_next_slot(ParallelSlot *slots, int n_slots);
static TocEntry *get_next_work_item(ArchiveHandle *AH,
! TocEntry **first_unprocessed,
ParallelSlot *slots, int n_slots);
static parallel_restore_result parallel_restore(RestoreArgs *args);
static void mark_work_done(ArchiveHandle *AH, thandle worker, int status,
--- 127,138 ----
static thandle reap_child(ParallelSlot *slots, int n_slots, int *work_status);
static bool work_in_progress(ParallelSlot *slots, int n_slots);
static int get_next_slot(ParallelSlot *slots, int n_slots);
+ static void toc_list_append(TocEntryList *l, TocEntry *te);
+ static void toc_list_remove(TocEntryList *l, TocEntry *te);
+ static void find_ready_items(TocEntryList *pending_list,
+ TocEntryList
*ready_list);
static TocEntry *get_next_work_item(ArchiveHandle *AH,
! TocEntryList *ready_list,
ParallelSlot *slots, int n_slots);
static parallel_restore_result parallel_restore(RestoreArgs *args);
static void mark_work_done(ArchiveHandle *AH, thandle worker, int status,
***************
*** 3065,3071 ****
ParallelSlot *slots;
int work_status;
int next_slot;
! TocEntry *first_unprocessed = AH->toc->next;
TocEntry *next_work_item;
thandle ret_child;
TocEntry *te;
--- 3079,3086 ----
ParallelSlot *slots;
int work_status;
int next_slot;
! TocEntryList pending_list;
! TocEntryList ready_list;
TocEntry *next_work_item;
thandle ret_child;
TocEntry *te;
***************
*** 3087,3094 ****
* faster in a single connection because we avoid all the connection and
* setup overhead.
*/
! while ((next_work_item = get_next_work_item(AH, &first_unprocessed,
!
NULL, 0)) != NULL)
{
if (next_work_item->section == SECTION_DATA ||
next_work_item->section == SECTION_POST_DATA)
--- 3102,3108 ----
* faster in a single connection because we avoid all the connection and
* setup overhead.
*/
! for (next_work_item = AH->toc->next; next_work_item != AH->toc;
next_work_item = next_work_item->next)
{
if (next_work_item->section == SECTION_DATA ||
next_work_item->section == SECTION_POST_DATA)
***************
*** 3100,3106 ****
(void) restore_toc_entry(AH, next_work_item, ropt, false);
- next_work_item->restored = true;
reduce_dependencies(AH, next_work_item);
}
--- 3114,3119 ----
***************
*** 3125,3130 ****
--- 3138,3162 ----
AH->currWithOids = -1;
/*
+ * Initialize the lists of pending and ready items. 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 contains items that
+ * have no remaining dependencies. Note: we don't yet filter out
entries
+ * that aren't going to be restored. They might participate in
+ * dependency chains connecting entries that should be restored, so we
+ * treat them as live until we actually process them.
+ */
+ pending_list.head = pending_list.tail = NULL;
+ ready_list.head = ready_list.tail = NULL;
+ for (; next_work_item != AH->toc; next_work_item = next_work_item->next)
+ {
+ if (next_work_item->depCount > 0)
+ toc_list_append(&pending_list, next_work_item);
+ else
+ toc_list_append(&ready_list, next_work_item);
+ }
+
+ /*
* main parent loop
*
* Keep going until there is no worker still running AND there is no
work
***************
*** 3133,3139 ****
ahlog(AH, 1, "entering main parallel loop\n");
! while ((next_work_item = get_next_work_item(AH, &first_unprocessed,
slots, n_slots)) != NULL ||
work_in_progress(slots, n_slots))
{
--- 3165,3171 ----
ahlog(AH, 1, "entering main parallel loop\n");
! while ((next_work_item = get_next_work_item(AH, &ready_list,
slots, n_slots)) != NULL ||
work_in_progress(slots, n_slots))
{
***************
*** 3149,3156 ****
next_work_item->dumpId,
next_work_item->desc,
next_work_item->tag);
! next_work_item->restored = true;
reduce_dependencies(AH, next_work_item);
continue;
}
--- 3181,3189 ----
next_work_item->dumpId,
next_work_item->desc,
next_work_item->tag);
! toc_list_remove(&ready_list, next_work_item);
reduce_dependencies(AH, next_work_item);
+ find_ready_items(&pending_list, &ready_list);
continue;
}
***************
*** 3165,3171 ****
next_work_item->dumpId,
next_work_item->desc,
next_work_item->tag);
! next_work_item->restored = true;
/* this memory is dealloced in mark_work_done()
*/
args = malloc(sizeof(RestoreArgs));
--- 3198,3204 ----
next_work_item->dumpId,
next_work_item->desc,
next_work_item->tag);
! toc_list_remove(&ready_list, next_work_item);
/* this memory is dealloced in mark_work_done()
*/
args = malloc(sizeof(RestoreArgs));
***************
*** 3194,3199 ****
--- 3227,3233 ----
{
mark_work_done(AH, ret_child, WEXITSTATUS(work_status),
slots, n_slots);
+ find_ready_items(&pending_list, &ready_list);
}
else
{
***************
*** 3218,3231 ****
* dependencies, or some other pathological condition. If so, do it in
the
* single parent connection.
*/
! for (te = AH->toc->next; te != AH->toc; te = te->next)
{
! if (!te->restored)
! {
! ahlog(AH, 1, "processing missed item %d %s %s\n",
! te->dumpId, te->desc, te->tag);
! (void) restore_toc_entry(AH, te, ropt, false);
! }
}
/* The ACLs will be handled back in RestoreArchive. */
--- 3252,3262 ----
* dependencies, or some other pathological condition. If so, do it in
the
* single parent connection.
*/
! for (te = pending_list.head; te; te = te->par_next)
{
! ahlog(AH, 1, "processing missed item %d %s %s\n",
! te->dumpId, te->desc, te->tag);
! (void) restore_toc_entry(AH, te, ropt, false);
}
/* The ACLs will be handled back in RestoreArchive. */
***************
*** 3366,3392 ****
}
return false;
}
/*
* Find the next work item (if any) that is capable of being run now.
*
* To qualify, the item must have no remaining dependencies
! * and no requirement for locks that is incompatible with
! * items currently running.
*
! * first_unprocessed is state data that tracks the location of the first
! * TocEntry that's not marked 'restored'. This avoids O(N^2) search time
! * with long TOC lists. (Even though the constant is pretty small, it'd
! * get us eventually.)
*
* pref_non_data is for an alternative selection algorithm that gives
* preference to non-data items if there is already a data load running.
* It is currently disabled.
*/
static TocEntry *
! get_next_work_item(ArchiveHandle *AH, TocEntry **first_unprocessed,
ParallelSlot *slots, int n_slots)
{
bool pref_non_data = false; /* or get from AH->ropt */
--- 3397,3474 ----
}
return false;
}
+
+
+ /* Append te to the end of the TocEntryList */
+ static void
+ toc_list_append(TocEntryList *l, TocEntry *te)
+ {
+ if (l->tail)
+ l->tail->par_next = te;
+ te->par_prev = l->tail;
+ te->par_next = NULL;
+ l->tail = te;
+ if (l->head == NULL)
+ l->head = te;
+ }
+
+ /* Remove te from the TocEntryList */
+ static void
+ toc_list_remove(TocEntryList *l, TocEntry *te)
+ {
+ if (te->par_prev)
+ te->par_prev->par_next = te->par_next;
+ else
+ l->head = te->par_next;
+ if (te->par_next)
+ te->par_next->par_prev = te->par_prev;
+ else
+ l->tail = te->par_prev;
+ te->par_next = NULL;
+ te->par_prev = NULL;
+ }
+
+ /*
+ * Find pending items that have no remaining dependencies, and move them
+ * to the ready list.
+ */
+ static void
+ find_ready_items(TocEntryList *pending_list, TocEntryList *ready_list)
+ {
+ TocEntry *te;
+ TocEntry *next;
+ for (te = pending_list->head; te; te = next)
+ {
+ /* Save list link in case we move this item to the other list */
+ next = te->par_next;
+ if (te->depCount > 0)
+ continue; /* not ready yet */
+ /* OK, move it */
+ toc_list_remove(pending_list, te);
+ toc_list_append(ready_list, te);
+ }
+ }
+
/*
* Find the next work item (if any) that is capable of being run now.
*
* 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
! * no remaining dependencies, but we have to check for lock conflicts.
*
! * Note that the returned item has *not* been removed from ready_list.
! * The caller must do that after successfully dispatching the item.
*
* pref_non_data is for an alternative selection algorithm that gives
* preference to non-data items if there is already a data load running.
* It is currently disabled.
*/
static TocEntry *
! get_next_work_item(ArchiveHandle *AH, TocEntryList *ready_list,
ParallelSlot *slots, int n_slots)
{
bool pref_non_data = false; /* or get from AH->ropt */
***************
*** 3411,3436 ****
}
/*
! * Advance first_unprocessed if possible.
! */
! for (te = *first_unprocessed; te != AH->toc; te = te->next)
! {
! if (!te->restored)
! break;
! }
! *first_unprocessed = te;
!
! /*
! * Search from first_unprocessed until we find an available item.
*/
! for (; te != AH->toc; te = te->next)
{
bool conflicts = false;
- /* Ignore if already done or still waiting on dependencies */
- if (te->restored || te->depCount > 0)
- continue;
-
/*
* Check to see if the item would need exclusive lock on
something
* that a currently running item also needs lock on, or vice
versa. If
--- 3493,3504 ----
}
/*
! * Search the ready_list until we find a suitable item.
*/
! for (te = ready_list->head; te; te = te->par_next)
{
bool conflicts = false;
/*
* Check to see if the item would need exclusive lock on
something
* that a currently running item also needs lock on, or vice
versa. If
Index: src/bin/pg_dump/pg_backup_archiver.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/bin/pg_dump/pg_backup_archiver.h,v
retrieving revision 1.79
diff -c -d -r1.79 pg_backup_archiver.h
*** src/bin/pg_dump/pg_backup_archiver.h 11 Jun 2009 14:49:07 -0000
1.79
--- src/bin/pg_dump/pg_backup_archiver.h 17 Jul 2009 02:20:28 -0000
***************
*** 314,320 ****
void *formatData; /* TOC Entry data specific to file
format */
/* working state (needed only for parallel restore) */
! bool restored; /* item is in progress or done
*/
bool created; /* set for DATA member if TABLE
was created */
int depCount; /* number of
dependencies not yet restored */
DumpId *lockDeps; /* dumpIds of objects this one needs
lock on */
--- 314,321 ----
void *formatData; /* TOC Entry data specific to file
format */
/* working state (needed only for parallel restore) */
! struct _tocEntry *par_prev; /* list links for pending items */
! struct _tocEntry *par_next;
bool created; /* set for DATA member if TABLE
was created */
int depCount; /* number of
dependencies not yet restored */
DumpId *lockDeps; /* dumpIds of objects this one needs
lock on */
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
