public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 06/17] Allow user to use password instead of encryption key. 20+ messages / 5 participants [nested] [flat]
* [PATCH 06/17] Allow user to use password instead of encryption key. @ 2019-07-05 14:24 Antonin Houska <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Antonin Houska @ 2019-07-05 14:24 UTC (permalink / raw) pg_keytool reads the password from stdin, uses "key derivation function (KDF) parameters" stored in the data directory to derive the key and writes the key to standard output. The cluster can be created this way initdb -D data -K "echo securepwd | pg_keytool -D data -w" and started either this way pg_ctl -D data -K "echo securepwd | pg_keytool -D data -w" start or this way postgres -D data and (in another session) echo securepwd | pg_keytool -D data -ws Both initdb and pg_ctl can substitute the data directory for %D pattern in the key encryption command. Thus user uses pg_keytool to derive the password, he can in fact say initdb -D data -K "echo securepwd | pg_keytool -D %D -w" and pg_ctl -D data -K "echo securepwd | pg_keytool -D %D -w" start respectively. --- src/bin/initdb/initdb.c | 26 +++- src/bin/pg_ctl/pg_ctl.c | 2 +- src/bin/pg_keytool/pg_keytool.c | 87 +++++++++--- src/fe_utils/encryption.c | 273 +++++++++++++++++++++++++++++++++++++- src/include/fe_utils/encryption.h | 8 +- src/include/storage/encryption.h | 8 ++ 6 files changed, 379 insertions(+), 25 deletions(-) diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c index f2a582e975..bfa8f5fad4 100644 --- a/src/bin/initdb/initdb.c +++ b/src/bin/initdb/initdb.c @@ -3004,11 +3004,31 @@ initialize_data_directory(void) write_version_file(NULL); /* - * If the cluster will be encrypted, run the command to generate the - * encryption key. + * If the cluster will be encrypted, write the KDF file so that encryption + * key can be derived from password. */ if (encryption_key_command) - run_encryption_key_command(encryption_key); + { + /* + * XXX Since execution of encryption_key_command produce the key (as + * opposed to password), we don't know if the command received the key + * itself or a password. If DBA provided initdb with a key, he will + * never use password in the future (there was no KDF so far so the + * key could not be derived from password, and the password can hardly + * be derived from the key), so the KDF file may be useless. We don't + * have enough information to recognize this special case, so just + * initialize and write the KDF unconditionally. + */ + init_kdf(); + write_kdf_file(pg_data); + + /* + * The key command is allowed to use pg_keytool, which in turn needs + * the KDF parameters. The KDF parameters are now available so we can + * run the command. + */ + run_encryption_key_command(encryption_key, pg_data); + } /* Select suitable configuration settings */ set_null_conf(); diff --git a/src/bin/pg_ctl/pg_ctl.c b/src/bin/pg_ctl/pg_ctl.c index 7e414ac048..e8e4eee162 100644 --- a/src/bin/pg_ctl/pg_ctl.c +++ b/src/bin/pg_ctl/pg_ctl.c @@ -881,7 +881,7 @@ do_start(void) * If encryption key is needed, retrieve it before trying to start * postmaster. */ - run_encryption_key_command(encryption_key); + run_encryption_key_command(encryption_key, pg_data); /* * Where should the key be sent? diff --git a/src/bin/pg_keytool/pg_keytool.c b/src/bin/pg_keytool/pg_keytool.c index 322625da41..c5151ad57c 100644 --- a/src/bin/pg_keytool/pg_keytool.c +++ b/src/bin/pg_keytool/pg_keytool.c @@ -43,6 +43,7 @@ usage(const char *progname) printf(_("Usage:\n")); printf(_(" %s [OPTION]...\n"), progname); printf(_("\nOptions:\n")); + printf(_(" -D, --pgdata=DATADIR data directory\n")); /* Display default host */ env = getenv("PGHOST"); printf(_(" -h, --host=HOSTNAME database server host or socket directory (default: \"%s\")\n"), @@ -52,8 +53,9 @@ usage(const char *progname) printf(_(" -p, --port=PORT database server port (default: \"%s\")\n"), env ? env : DEF_PGPORT_STR); printf(_(" -s, send output to database server\n")); + printf(_(" -w expect password on input, not a key\n")); printf(_(" -?, --help show this help, then exit\n\n")); - printf(_("Key is read from stdin and sent either to stdout or to PostgreSQL server being started\n")); + printf(_("Password or key is read from stdin. Key is sent to PostgreSQL server being started\n")); } #endif /* USE_ENCRYPTION */ @@ -69,9 +71,12 @@ main(int argc, char **argv) int c; char *host = NULL; char *port_str = NULL; + char *DataDir = NULL; bool to_server = false; + bool expect_password = false; int i, n; int optindex; + char password[ENCRYPTION_PWD_MAX_LENGTH]; char key_chars[ENCRYPTION_KEY_CHARS]; static struct option long_options[] = @@ -99,11 +104,15 @@ main(int argc, char **argv) } } - while ((c = getopt_long(argc, argv, "h:p:s", + while ((c = getopt_long(argc, argv, "h:D:p:sw", long_options, &optindex)) != -1) { switch (c) { + case 'D': + DataDir = optarg; + break; + case 'h': host = pg_strdup(optarg); break; @@ -116,6 +125,10 @@ main(int argc, char **argv) to_server = true; break; + case 'w': + expect_password = true; + break; + case '?': /* Actual help option given */ if (strcmp(argv[optind - 1], "-?") == 0) @@ -146,34 +159,78 @@ main(int argc, char **argv) exit(1); } - /* Read the key. */ + /* + * The KDF file is needed to derive the key from password, and this file + * is located in the data directory. + */ + if (expect_password && DataDir == NULL) + { + pg_log_error("%s: no data directory specified", progname); + pg_log_error("Try \"%s --help\" for more information.", progname); + exit(EXIT_FAILURE); + } + + /* + * Read the credentials (key or password). + */ n = 0; /* Key length in characters (two characters per hexadecimal digit) */ while ((c = getchar()) != EOF && c != '\n') { - if (n >= ENCRYPTION_KEY_CHARS) + if (!expect_password) { - pg_log_error("The key is too long"); - exit(EXIT_FAILURE); + if (n >= ENCRYPTION_KEY_CHARS) + { + pg_log_error("The key is too long"); + exit(EXIT_FAILURE); + } + + key_chars[n++] = c; } + else + { + if (n >= ENCRYPTION_PWD_MAX_LENGTH) + { + pg_log_error("The password is too long"); + exit(EXIT_FAILURE); + } - key_chars[n++] = c; + password[n++] = c; + } } - if (n < ENCRYPTION_KEY_CHARS) + /* If password was received, turn it into encryption key. */ + if (!expect_password) { - pg_log_error("The key is too short"); - exit(EXIT_FAILURE); - } + if (n < ENCRYPTION_KEY_CHARS) + { + pg_log_error("The key is too short"); + exit(EXIT_FAILURE); + } - for (i = 0; i < ENCRYPTION_KEY_LENGTH; i++) + for (i = 0; i < ENCRYPTION_KEY_LENGTH; i++) + { + if (sscanf(key_chars + 2 * i, "%2hhx", encryption_key + i) == 0) + { + pg_log_error("Invalid character in encryption key at position %d", + 2 * i); + exit(EXIT_FAILURE); + } + } + } + else { - if (sscanf(key_chars + 2 * i, "%2hhx", encryption_key + i) == 0) + if (n < ENCRYPTION_PWD_MIN_LENGTH) { - pg_log_error("Invalid character in encryption key at position %d", - 2 * i); + pg_log_error("The password is too short"); exit(EXIT_FAILURE); } + + /* Read the KDF parameters. */ + read_kdf_file(DataDir); + + /* Run the KDF. */ + derive_key_from_password(encryption_key, password, n); } /* diff --git a/src/fe_utils/encryption.c b/src/fe_utils/encryption.c index ca37c9f373..134b3bde9b 100644 --- a/src/fe_utils/encryption.c +++ b/src/fe_utils/encryption.c @@ -21,31 +21,294 @@ #include "common/file_perm.h" #include "common/logging.h" #include "fe_utils/encryption.h" -#include "storage/encryption.h" #include "libpq-fe.h" #include "libpq-int.h" #include "libpq/pqcomm.h" char *encryption_key_command = NULL; +#define KDF_PARAMS_FILE "global/kdf_params" +#define KDF_PARAMS_FILE_SIZE 512 + +/* + * Key derivation function. + */ +typedef enum KDFKind +{ + KDF_OPENSSL_PKCS5_PBKDF2_HMAC_SHA = 0 +} KFDKind; + +typedef struct KDFParamsPBKDF2 +{ + unsigned long int niter; + unsigned char salt[ENCRYPTION_KDF_SALT_LEN]; +} KDFParamsPBKDF2; + +/* + * Parameters of the key derivation function. + * + * The parameters are generated by initdb and stored into a file, which is + * then read during PG startup. This is similar to storing various settings in + * pg_control. However an existing KDF file is read only, so it does not have + * to be stored in shared memory. + */ +typedef struct KDFParamsData +{ + KFDKind function; + + /* + * Function-specific parameters. + */ + union + { + KDFParamsPBKDF2 pbkdf2; + } data; + + /* CRC of all above ... MUST BE LAST! */ + pg_crc32c crc; +} KDFParamsData; + +extern KDFParamsData *KDFParams; + +/* + * Pointer to the KDF parameters. + */ +KDFParamsData *KDFParams = NULL; + +/* Initialize KDF file. */ +void +init_kdf(void) +{ + KDFParamsPBKDF2 *params; + struct timeval tv; + uint64 salt; + + /* + * The initialization should not be repeated. + */ + Assert(KDFParams == NULL); + + KDFParams = palloc0(KDF_PARAMS_FILE_SIZE); + KDFParams->function = KDF_OPENSSL_PKCS5_PBKDF2_HMAC_SHA; + params = &KDFParams->data.pbkdf2; + + /* + * Currently we derive the salt in the same way as system identifier, + * however these two values are not supposed to match. XXX Is it worth the + * effort if initdb derives the system identifier, passes it to this + * function and also sends it to the bootstrap process? Not sure. + */ + gettimeofday(&tv, NULL); + salt = ((uint64) tv.tv_sec) << 32; + salt |= ((uint64) tv.tv_usec) << 12; + salt |= getpid() & 0xFFF; + + memcpy(params->salt, &salt, sizeof(uint64)); + params->niter = ENCRYPTION_KDF_NITER; +} + +/* + * Write KDFParamsData to file. + */ +void +write_kdf_file(char *dir) +{ + char path[MAXPGPATH]; + int fd; + + Assert(KDFParams != NULL); + + /* Account for both file separator and terminating NULL character. */ + if ((strlen(dir) + 1 + strlen(KDF_PARAMS_FILE) + 1) > MAXPGPATH) + { + pg_log_fatal("KDF directory is too long"); + exit(EXIT_FAILURE); + } + + snprintf(path, MAXPGPATH, "%s/%s", dir, KDF_PARAMS_FILE); + + /* Contents are protected with a CRC */ + INIT_CRC32C(KDFParams->crc); + COMP_CRC32C(KDFParams->crc, + (char *) KDFParams, + offsetof(KDFParamsData, crc)); + FIN_CRC32C(KDFParams->crc); + + fd = open(path, O_WRONLY | O_CREAT | PG_BINARY, + pg_file_create_mode); + if (fd < 0) + { + pg_log_fatal("could not create key derivation file \"%s\": %m", path); + exit(EXIT_FAILURE); + } + + if (write(fd, KDFParams, KDF_PARAMS_FILE_SIZE) != KDF_PARAMS_FILE_SIZE) + { + /* if write didn't set errno, assume problem is no disk space */ + if (errno == 0) + errno = ENOSPC; + pg_log_fatal("could not write to key derivation file \"%s\": %m", + path); + exit(EXIT_FAILURE); + } + + if (close(fd)) + { + pg_log_fatal("could not close key setup file: %m"); + exit(EXIT_FAILURE); + } +} + +/* + * Read KDFParamsData from file and store it in local memory. + * + * If dir is NULL, assume we're in the data directory. + * + * postmaster should call the function early enough for any other process to + * inherit valid pointer to the data. + */ +void +read_kdf_file(char *dir) +{ + pg_crc32c crc; + char path[MAXPGPATH]; + int fd; + + /* Account for both file separator and terminating NULL character. */ + if ((strlen(dir) + 1 + strlen(KDF_PARAMS_FILE) + 1) > MAXPGPATH) + { + pg_log_fatal("KDF directory is too long"); + exit(EXIT_FAILURE); + } + + snprintf(path, MAXPGPATH, "%s/%s", dir, KDF_PARAMS_FILE); + + KDFParams = palloc0(KDF_PARAMS_FILE_SIZE); + fd = open(path, O_RDONLY | PG_BINARY, S_IRUSR); + + if (fd < 0) + { + pg_log_fatal("could not open key setup file \"%s\": %m", path); + exit(EXIT_FAILURE); + } + + if (read(fd, KDFParams, sizeof(KDFParamsData)) != sizeof(KDFParamsData)) + { + pg_log_fatal("could not read from key setup file \"%s\": %m", path); + exit(EXIT_FAILURE); + } + + close(fd); + + /* Now check the CRC. */ + INIT_CRC32C(crc); + COMP_CRC32C(crc, + (char *) KDFParams, + offsetof(KDFParamsData, crc)); + FIN_CRC32C(crc); + + if (!EQ_CRC32C(crc, KDFParams->crc)) + { + pg_log_fatal("incorrect checksum in key setup file \"%s\"", path); + exit(EXIT_FAILURE); + } + + + if (KDFParams->function != KDF_OPENSSL_PKCS5_PBKDF2_HMAC_SHA) + { + pg_log_fatal("unsupported KDF function"); + exit(EXIT_FAILURE); + } +} + +/* + * Run the key derivation function and initialize encryption_key variable. + */ +void +derive_key_from_password(unsigned char *encryption_key, const char *password, + int len) +{ + KDFParamsPBKDF2 *params; + int rc; + + params = &KDFParams->data.pbkdf2; + rc = PKCS5_PBKDF2_HMAC(password, + len, + params->salt, + ENCRYPTION_KDF_SALT_LEN, + params->niter, + EVP_sha1(), + ENCRYPTION_KEY_LENGTH, + encryption_key); + + if (rc != 1) + { + pg_log_fatal("failed to derive key from password"); + exit(EXIT_FAILURE); + } +} + /* * Run the command that is supposed to generate encryption key and store it - * where encryption_key points to. + * where encryption_key points to. If valid string is passed for data_dir, + * it's used to replace '%D' pattern in the command. */ void -run_encryption_key_command(unsigned char *encryption_key) +run_encryption_key_command(unsigned char *encryption_key, char *data_dir) { FILE *fp; + char cmd[MAXPGPATH]; + char *sp, *dp, *endp; char *buf; int read_len, i, c; Assert(encryption_key_command != NULL && strlen(encryption_key_command) > 0); - fp = popen(encryption_key_command, "r"); + /* + * Replace %D pattern in the command with the actual data directory path. + */ + dp = cmd; + endp = cmd + MAXPGPATH - 1; + *endp = '\0'; + for (sp = encryption_key_command; *sp; sp++) + { + if (*sp == '%') + { + if (sp[1] == 'D') + { + if (data_dir == NULL) + { + pg_log_fatal("data directory is not known, %%D pattern cannot be replaced"); + exit(EXIT_FAILURE); + } + + sp++; + strlcpy(dp, data_dir, endp - dp); + make_native_path(dp); + dp += strlen(dp); + } + else if (dp < endp) + *dp++ = *sp; + else + break; + } + else + { + if (dp < endp) + *dp++ = *sp; + else + break; + } + } + *dp = '\0'; + + pg_log_debug("executing encryption key command \"%s\"", cmd); + + fp = popen(cmd, "r"); if (fp == NULL) { - pg_log_fatal("Failed to execute \"%s\"", encryption_key_command); + pg_log_fatal("Failed to execute \"%s\"", cmd); exit(EXIT_FAILURE); } diff --git a/src/include/fe_utils/encryption.h b/src/include/fe_utils/encryption.h index 7307557ed6..da302fe494 100644 --- a/src/include/fe_utils/encryption.h +++ b/src/include/fe_utils/encryption.h @@ -15,6 +15,12 @@ /* Executable to retrieve the encryption key. */ extern char *encryption_key_command; -extern void run_encryption_key_command(unsigned char *encryption_key); +extern void init_kdf(void); +extern void write_kdf_file(char *dir); +extern void read_kdf_file(char *dir); +extern void derive_key_from_password(unsigned char *encryption_key, + const char *password, int len); +extern void run_encryption_key_command(unsigned char *encryption_key, + char *data_dir); extern bool send_key_to_postmaster(const char *host, const char *port, const unsigned char *encryption_Key); diff --git a/src/include/storage/encryption.h b/src/include/storage/encryption.h index 4f7b96d4f3..72bcae0972 100644 --- a/src/include/storage/encryption.h +++ b/src/include/storage/encryption.h @@ -57,6 +57,14 @@ typedef enum CipherKind PG_CIPHER_AES_BLOCK_CBC_256_STREAM_CTR_256 } CipherKind; +/* + * TODO Tune these values. + */ +#define ENCRYPTION_PWD_MIN_LENGTH 8 +#define ENCRYPTION_PWD_MAX_LENGTH 16 +#define ENCRYPTION_KDF_NITER 1048576 +#define ENCRYPTION_KDF_SALT_LEN sizeof(uint64) + /* Key to encrypt / decrypt data. */ extern unsigned char encryption_key[]; -- 2.13.7 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v04-0007-Refactoring-patch.-This-only-moves-some-code-to-XLog.patch ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v9 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..ea17ca4559 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,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; @@ -4344,38 +4261,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4318,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; } @@ -4447,7 +4364,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); @@ -4465,7 +4382,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); } @@ -4708,11 +4625,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; @@ -4730,18 +4647,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 --CE+1k2dSO48ffgeK Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v9-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v6 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 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..4c86f7b448 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,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will usually be able to pick one of the first few items, which will + * generally be relatively high priority. */ - 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 +4314,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 +4360,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 +4378,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 +4621,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 +4643,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 --YZ5djTAD1cGYuMQK Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v6-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v10 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 198 ++++++++------------------- 1 file changed, 58 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..b30b49702d 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,25 @@ 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 opposite of qsort comparator for max-heap */ + 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; @@ -4344,38 +4262,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4319,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; } @@ -4447,7 +4365,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); @@ -4465,7 +4383,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); } @@ -4708,11 +4626,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; @@ -4730,18 +4648,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 --y0ulUmNC+osPPQO6 Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v10-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v9 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..ea17ca4559 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,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; @@ -4344,38 +4261,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4318,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; } @@ -4447,7 +4364,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); @@ -4465,7 +4382,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); } @@ -4708,11 +4625,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; @@ -4730,18 +4647,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 --CE+1k2dSO48ffgeK Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v9-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..95b8f69ecf 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,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; @@ -4344,38 +4261,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4318,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; } @@ -4447,7 +4364,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); @@ -4465,7 +4382,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); } @@ -4708,11 +4625,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; @@ -4730,18 +4647,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 --gBBFr7Ir9EOA20Yy Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v8-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..95b8f69ecf 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,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; @@ -4344,38 +4261,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4318,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; } @@ -4447,7 +4364,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); @@ -4465,7 +4382,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); } @@ -4708,11 +4625,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; @@ -4730,18 +4647,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 --gBBFr7Ir9EOA20Yy Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v8-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v9 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..ea17ca4559 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,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; @@ -4344,38 +4261,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4318,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; } @@ -4447,7 +4364,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); @@ -4465,7 +4382,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); } @@ -4708,11 +4625,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; @@ -4730,18 +4647,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 --CE+1k2dSO48ffgeK Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v9-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v10 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 198 ++++++++------------------- 1 file changed, 58 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..b30b49702d 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,25 @@ 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 opposite of qsort comparator for max-heap */ + 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; @@ -4344,38 +4262,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4319,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; } @@ -4447,7 +4365,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); @@ -4465,7 +4383,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); } @@ -4708,11 +4626,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; @@ -4730,18 +4648,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 --y0ulUmNC+osPPQO6 Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v10-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..95b8f69ecf 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,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; @@ -4344,38 +4261,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4318,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; } @@ -4447,7 +4364,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); @@ -4465,7 +4382,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); } @@ -4708,11 +4625,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; @@ -4730,18 +4647,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 --gBBFr7Ir9EOA20Yy Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v8-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 197 ++++++++------------------- 1 file changed, 57 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..95b8f69ecf 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,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; @@ -4344,38 +4261,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4318,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; } @@ -4447,7 +4364,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); @@ -4465,7 +4382,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); } @@ -4708,11 +4625,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; @@ -4730,18 +4647,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 --gBBFr7Ir9EOA20Yy Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v8-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v10 3/4] Convert pg_restore's ready_list to a priority queue. @ 2023-07-20 17:19 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-07-20 17:19 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. Suggested-by: Tom Lane Tested-by: Pierre Ducroquet Reviewed-by: Tom Lane Discussion: https://postgr.es/m/3612876.1689443232%40sss.pgh.pa.us --- src/bin/pg_dump/pg_backup_archiver.c | 198 ++++++++------------------- 1 file changed, 58 insertions(+), 140 deletions(-) diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4d83381d84..b30b49702d 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, @@ -111,16 +94,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, @@ -135,7 +114,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); @@ -2384,7 +2363,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, @@ -3984,7 +3963,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 @@ -4027,24 +4006,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 @@ -4058,7 +4039,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 */ @@ -4068,7 +4049,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; } @@ -4079,7 +4060,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)) { @@ -4093,7 +4074,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; } @@ -4122,10 +4103,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"); } @@ -4225,80 +4206,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; @@ -4318,17 +4228,25 @@ 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 opposite of qsort comparator for max-heap */ + 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; @@ -4344,38 +4262,38 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; /* @@ -4401,7 +4319,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; } @@ -4447,7 +4365,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); @@ -4465,7 +4383,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); } @@ -4708,11 +4626,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; @@ -4730,18 +4648,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 --y0ulUmNC+osPPQO6 Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v10-0004-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v7 4/5] Convert pg_restore's ready_list to a priority queue. @ 2023-09-04 22:35 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-09-04 22:35 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. --- src/bin/pg_dump/pg_backup_archiver.c | 201 ++++++++------------------- 1 file changed, 61 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..ff7349537e 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,13 @@ 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(const void *p1, const 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 +114,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 +2360,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 +3960,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 +4003,27 @@ 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, + sizeof(TocEntry *), + 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 +4037,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 +4047,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 +4058,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 +4072,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 +4101,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 +4204,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 +4226,24 @@ TocEntrySizeCompare(const void *p1, const void *p2) return 0; } +/* binaryheap comparator for sorting TocEntries by dataLength */ +static int +TocEntrySizeCompareBinaryheap(const void *p1, const 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,40 +4259,42 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; bool conflicts = false; + binaryheap_nth(ready_heap, i, &te); + /* * 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 @@ -4397,7 +4318,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_nth(ready_heap, i); return te; } @@ -4443,7 +4364,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 +4382,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 +4625,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 +4647,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 --DocE+STaALJfprDB Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v7-0005-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v7 4/5] Convert pg_restore's ready_list to a priority queue. @ 2023-09-04 22:35 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-09-04 22:35 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. --- src/bin/pg_dump/pg_backup_archiver.c | 201 ++++++++------------------- 1 file changed, 61 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..ff7349537e 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,13 @@ 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(const void *p1, const 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 +114,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 +2360,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 +3960,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 +4003,27 @@ 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, + sizeof(TocEntry *), + 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 +4037,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 +4047,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 +4058,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 +4072,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 +4101,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 +4204,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 +4226,24 @@ TocEntrySizeCompare(const void *p1, const void *p2) return 0; } +/* binaryheap comparator for sorting TocEntries by dataLength */ +static int +TocEntrySizeCompareBinaryheap(const void *p1, const 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,40 +4259,42 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; bool conflicts = false; + binaryheap_nth(ready_heap, i, &te); + /* * 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 @@ -4397,7 +4318,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_nth(ready_heap, i); return te; } @@ -4443,7 +4364,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 +4382,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 +4625,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 +4647,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 --DocE+STaALJfprDB Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v7-0005-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* [PATCH v7 4/5] Convert pg_restore's ready_list to a priority queue. @ 2023-09-04 22:35 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Nathan Bossart @ 2023-09-04 22:35 UTC (permalink / raw) Presently, we spend a lot of time sorting this list so that we pick the largest items first. With many tables, this sorting can become a significant bottleneck. There are a couple of reports from the field about this, and it is easily reproducible, so this is not a hypothetical issue. This commit improves the performance of pg_restore with many tables by converting its ready_list to a priority queue, i.e., a binary heap. We will first try to run the highest priority item, but if it cannot be chosen due to the lock heuristic, we'll do a sequential scan through the heap nodes until we find one that is runnable. This means that we might end up picking an item with much lower priority, but since we expect that we'll typically be able to choose one of the first few nodes, we should usually pick an item with a relatively high priority. On my machine, a basic test with 100,000 tables takes 11.5 minutes without this patch and 1.5 minutes with it. Pierre Ducroquet claims to see a speedup from 30 minutes to 23 minutes for a real-world dump of over 50,000 tables. --- src/bin/pg_dump/pg_backup_archiver.c | 201 ++++++++------------------- 1 file changed, 61 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..ff7349537e 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,13 @@ 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(const void *p1, const 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 +114,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 +2360,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 +3960,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 +4003,27 @@ 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, + sizeof(TocEntry *), + 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 +4037,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 +4047,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 +4058,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 +4072,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 +4101,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 +4204,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 +4226,24 @@ TocEntrySizeCompare(const void *p1, const void *p2) return 0; } +/* binaryheap comparator for sorting TocEntries by dataLength */ +static int +TocEntrySizeCompareBinaryheap(const void *p1, const 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,40 +4259,42 @@ 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. Note that we do a + * sequential scan through the heap nodes, so even though we will first + * try to choose the highest-priority item, we might end up picking + * something with a much lower priority. However, it is expected that we + * will typically be able to pick one of the first few items, which should + * usually have a relatively high priority. */ - 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; bool conflicts = false; + binaryheap_nth(ready_heap, i, &te); + /* * 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 @@ -4397,7 +4318,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_nth(ready_heap, i); return te; } @@ -4443,7 +4364,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 +4382,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 +4625,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 +4647,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 --DocE+STaALJfprDB Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v7-0005-Remove-open-coded-binary-heap-in-pg_dump_sort.c.patch" ^ permalink raw reply [nested|flat] 20+ messages in thread
* Re: Don't keep closed WAL segment in page cache after replay @ 2026-02-18 09:51 Anthonin Bonnefoy <[email protected]> 0 siblings, 1 reply; 20+ messages in thread From: Anthonin Bonnefoy @ 2026-02-18 09:51 UTC (permalink / raw) To: Kirill Reshke <[email protected]>; +Cc: Fujii Masao <[email protected]>; pgsql-hackers Thanks for the review! On Tue, Feb 17, 2026 at 9:38 AM Kirill Reshke <[email protected]> wrote: > This does not immediately strike me as good reasoning. We have, for > example, pg_stat_get_wal_senders (or WalSndInitStopping from > walsenders.h) function which accesses exactly WalSndCtl->walsnds. > Why don't we simply have another utility function that will return the > number of active walsenders? That's true it is an option. I've switched to this approach and created a WalSndRunning. We only need to know if there's at least one wal sender running, no need to have a precise number. I've also added StandbyMode as a condition to restrict this to replicas. XLogPageRead may be used by the primary when starting up, and will likely re-read the WAL so releasing cached pages should be avoided on the primary. Attachments: [application/octet-stream] v3-0001-Don-t-keep-closed-WAL-segments-in-page-cache-afte.patch (3.6K, ../../CAO6_Xqric0=7TVC-dLySU6_mtG3ie8kJY=87EjDeYjqGc3uNNA@mail.gmail.com/2-v3-0001-Don-t-keep-closed-WAL-segments-in-page-cache-afte.patch) download | inline diff: From 781306ce190509749d4b45b76894ae1790b7c892 Mon Sep 17 00:00:00 2001 From: Anthonin Bonnefoy <[email protected]> Date: Wed, 2 Jul 2025 09:58:52 +0200 Subject: Don't keep closed WAL segments in page cache after replay On a standby, the recovery process reads the WAL segments, applies changes and closes the segment. When closed, the segments will still be in page cache memory until they are evicted due to inactivity. The segments may be re-read if archive_mode is set to always, wal_summarizer is enabled or if the standby is used for replication and has an active walsender. Outside of those circumstances, the WAL segments won't be re-read and keeping them in the page cache generates unnecessary memory pressure. If the standby doesn't archive wal, doesn't have wal_summarize and doesn't have an active walsender, a POSIX_FADV_DONTNEED is sent before closing a replayed WAL segment to immediately free any cached pages. --- src/backend/access/transam/xlogrecovery.c | 16 ++++++++++++++++ src/backend/replication/walsender.c | 23 +++++++++++++++++++++++ src/include/replication/walsender.h | 1 + 3 files changed, 40 insertions(+) diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index c0c2744d45b..762b517e8ef 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -49,6 +49,7 @@ #include "pgstat.h" #include "postmaster/bgwriter.h" #include "postmaster/startup.h" +#include "postmaster/walsummarizer.h" #include "replication/slot.h" #include "replication/slotsync.h" #include "replication/walreceiver.h" @@ -3369,6 +3370,21 @@ XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, } } + /* + * Once replayed, WAL segment files may be re-read in several cases: + * archive_mode is set to always, summarize_wal is enabled or the + * standby acts as a walsender for either logical or physical + * replication. Outside of those conditions, the WAL segment files + * shouldn't be re-read and we can signal the kernel to release any + * cached pages. + */ +#if defined(USE_POSIX_FADVISE) && defined(POSIX_FADV_DONTNEED) + { + if (StandbyMode && XLogArchiveMode != ARCHIVE_MODE_ALWAYS && + !summarize_wal && !WalSndRunning()) + (void) posix_fadvise(readFile, 0, 0, POSIX_FADV_DONTNEED); + } +#endif close(readFile); readFile = -1; readSource = XLOG_FROM_ANY; diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2cde8ebc729..8ae3ef155d9 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -3697,6 +3697,29 @@ WalSndRqstFileReload(void) } } +/* + * Return true if there's at least one active walsender process + */ +bool +WalSndRunning(void) +{ + int i; + + for (i = 0; i < max_wal_senders; i++) + { + WalSnd *walsnd = &WalSndCtl->walsnds[i]; + + SpinLockAcquire(&walsnd->mutex); + if (walsnd->pid > 0) + { + SpinLockRelease(&walsnd->mutex); + return true; + } + SpinLockRelease(&walsnd->mutex); + } + return false; +} + /* * Handle PROCSIG_WALSND_INIT_STOPPING signal. */ diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h index a4df3b8e0ae..7c197e71392 100644 --- a/src/include/replication/walsender.h +++ b/src/include/replication/walsender.h @@ -48,6 +48,7 @@ extern void WalSndInitStopping(void); extern void WalSndWaitStopping(void); extern void HandleWalSndInitStopping(void); extern void WalSndRqstFileReload(void); +extern bool WalSndRunning(void); /* * Remember that we want to wakeup walsenders later -- 2.52.0 ^ permalink raw reply [nested|flat] 20+ messages in thread
* Re: Don't keep closed WAL segment in page cache after replay @ 2026-03-03 14:13 Anthonin Bonnefoy <[email protected]> parent: Anthonin Bonnefoy <[email protected]> 0 siblings, 1 reply; 20+ messages in thread From: Anthonin Bonnefoy @ 2026-03-03 14:13 UTC (permalink / raw) To: Hüseyin Demir <[email protected]>; +Cc: [email protected] On Mon, Mar 2, 2026 at 9:15 AM Hüseyin Demir <[email protected]> wrote: > The idea looks good and efficient albeit I have some feedback. The first one is about logical replication slot. > > Inside the patch, it checks if there is an active walsender process. Is it possible to create a replication slot and wait until a subscriber will connect it. During this time due to patch PostgreSQL will close the WAL segments on the memory and once the subscriber connects it has to read the WAL files from disk. But it's a trade-off and can be decided by others too. Good point. This is also the case for physical replication slots, if there's at least one used replication slot, then it's very likely a walsender will start at some point and would need to read the WAL. And having to read a large amount of WAL files from disk is likely not desirable, so it's probably better to add this as a condition. > Secondly, when it comes to using spinlock to check the running walsender processes it can lead inefficient recovery process. Because assuming that a database with max_wal_sender set to 20+ and producing more than 4-5TB WAL in a day it can cause additional +100-200 spinlocks each second on walreceiver. Put simply, WalSndRunning() scans every walsender slot with spinlocks on every segment switch, contending with all active walsenders updating their own slots. On high-throughput standbys this creates unnecessary cross-process spinlock contention in the recovery hot path — the > exact path that should be as lean as possible for fast replay. Maybe you can implement a single pg_atomic_uint32 counter in WalSndCtlData and achieve the same result with zero contention. True. I think I've assumed that a segment closing is rare enough that it would be fine to go through all walsenders, but there are certainly clusters that can generate a large amount of segments. I've updated the patch with the suggested approach: - a new atomic counter tracking the number of active walsenders - a similar atomic counter for the number of used replication slots. Otherwise, we would have a similar issue of going through all max_replication_slots to check if one is used. - Both are now used as condition to send POSIX_FADV_DONTNEED or not Regards, Anthonin Bonnefoy Attachments: [application/octet-stream] v4-0001-Don-t-keep-closed-WAL-segments-in-page-cache-afte.patch (8.2K, ../../CAO6_Xqp0s3aVyAhx3K_dsdPBqe4_OoD3Wcd__-CgEcvN+HqPxQ@mail.gmail.com/2-v4-0001-Don-t-keep-closed-WAL-segments-in-page-cache-afte.patch) download | inline diff: From 8a26458d2542234f4a0ff294ab9c34c252479f32 Mon Sep 17 00:00:00 2001 From: Anthonin Bonnefoy <[email protected]> Date: Wed, 2 Jul 2025 09:58:52 +0200 Subject: Don't keep closed WAL segments in page cache after replay On a standby, the recovery process reads the WAL segments, applies changes and closes the segment. When closed, the segments will still be in page cache memory until they are evicted due to inactivity. The segments may be re-read if archive_mode is set to always, wal_summarizer is enabled or if the standby is used for replication and has an active walsender. The presence of a replication slots is also a likely indicator that a walsender will be started, and need to read the WAL segments. Outside of those circumstances, the WAL segments won't be re-read and keeping them in the page cache generates unnecessary memory pressure. A POSIX_FADV_DONTNEED is sent before closing a replayed WAL segment to immediately free any cached pages. --- src/backend/access/transam/xlogrecovery.c | 21 +++++++++++++++++++++ src/backend/replication/slot.c | 16 ++++++++++++++++ src/backend/replication/walsender.c | 14 +++++++++++++- src/include/replication/slot.h | 10 +++++----- src/include/replication/walsender.h | 1 + src/include/replication/walsender_private.h | 4 ++++ 6 files changed, 60 insertions(+), 6 deletions(-) diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index ecd66fd86a4..b5ffef515c7 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -49,6 +49,7 @@ #include "pgstat.h" #include "postmaster/bgwriter.h" #include "postmaster/startup.h" +#include "postmaster/walsummarizer.h" #include "replication/slot.h" #include "replication/slotsync.h" #include "replication/walreceiver.h" @@ -3370,6 +3371,26 @@ XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, } } + /* + * Once replayed, WAL segment files may be re-read in several cases: + * archive_mode is set to always, summarize_wal is enabled or the + * standby acts as a walsender for either logical or physical + * replication. Outside of those conditions, the WAL segment files + * shouldn't be re-read and we can signal the kernel to release any + * cached pages. We also check if any replication slot is used: If + * present, it's likely a walsender will start and will need to read + * the WAL segment files, so we may as well keep the pages in cache. + */ +#if defined(USE_POSIX_FADVISE) && defined(POSIX_FADV_DONTNEED) + { + if (StandbyMode && + XLogArchiveMode != ARCHIVE_MODE_ALWAYS && + !summarize_wal && + WalSndNumActive() == 0 && + ReplicationSlotNumUsed() == 0) + (void) posix_fadvise(readFile, 0, 0, POSIX_FADV_DONTNEED); + } +#endif close(readFile); readFile = -1; readSource = XLOG_FROM_ANY; diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 28c7019402b..9ea7272df4a 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -220,6 +220,7 @@ ReplicationSlotsShmemInit(void) /* First time through, so initialize */ MemSet(ReplicationSlotCtl, 0, ReplicationSlotsShmemSize()); + pg_atomic_init_u32(&ReplicationSlotCtl->used_replication_slots, 0); for (i = 0; i < max_replication_slots; i++) { @@ -522,6 +523,9 @@ ReplicationSlotCreate(const char *name, bool db_specific, if (SlotIsLogical(slot)) pgstat_create_replslot(slot); + /* Update the used_replication_slots counter with the created slot */ + pg_atomic_fetch_add_u32(&ReplicationSlotCtl->used_replication_slots, 1); + /* * Now that the slot has been marked as in_use and active, it's safe to * let somebody else try to allocate a slot. @@ -579,6 +583,15 @@ ReplicationSlotIndex(ReplicationSlot *slot) return slot - ReplicationSlotCtl->replication_slots; } +/* + * Return the number of used replication slots + */ +int +ReplicationSlotNumUsed(void) +{ + return pg_atomic_read_u32(&ReplicationSlotCtl->used_replication_slots); +} + /* * If the slot at 'index' is unused, return false. Otherwise 'name' is set to * the slot's name and true is returned. @@ -1143,6 +1156,9 @@ ReplicationSlotDropPtr(ReplicationSlot *slot) if (SlotIsLogical(slot)) pgstat_drop_replslot(slot); + /* Decrement the used_replication_slots counter */ + pg_atomic_sub_fetch_u32(&ReplicationSlotCtl->used_replication_slots, 1); + /* * We release this at the very end, so that nobody starts trying to create * a slot while we're still cleaning up the detritus of the old one. diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2cde8ebc729..0cbcd136fd5 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -3092,7 +3092,7 @@ InitWalSenderSlot(void) SpinLockRelease(&walsnd->mutex); /* don't need the lock anymore */ MyWalSnd = walsnd; - + pg_atomic_fetch_add_u32(&WalSndCtl->active_wal_senders, 1); break; } } @@ -3117,6 +3117,7 @@ WalSndKill(int code, Datum arg) /* Mark WalSnd struct as no longer being in use. */ walsnd->pid = 0; SpinLockRelease(&walsnd->mutex); + pg_atomic_fetch_sub_u32(&WalSndCtl->active_wal_senders, 1); } /* XLogReaderRoutine->segment_open callback */ @@ -3697,6 +3698,15 @@ WalSndRqstFileReload(void) } } +/* + * Returns the number of active walsender processes + */ +int +WalSndNumActive(void) +{ + return pg_atomic_read_u32(&WalSndCtl->active_wal_senders); +} + /* * Handle PROCSIG_WALSND_INIT_STOPPING signal. */ @@ -3785,6 +3795,8 @@ WalSndShmemInit(void) SpinLockInit(&walsnd->mutex); } + pg_atomic_init_u32(&WalSndCtl->active_wal_senders, 0); + ConditionVariableInit(&WalSndCtl->wal_flush_cv); ConditionVariableInit(&WalSndCtl->wal_replay_cv); ConditionVariableInit(&WalSndCtl->wal_confirm_rcv_cv); diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 4b4709f6e2c..491b8e53db5 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -292,11 +292,10 @@ typedef struct ReplicationSlot */ typedef struct ReplicationSlotCtlData { - /* - * This array should be declared [FLEXIBLE_ARRAY_MEMBER], but for some - * reason you can't do that in an otherwise-empty struct. - */ - ReplicationSlot replication_slots[1]; + /* Number of used replication slots */ + pg_atomic_uint32 used_replication_slots; + + ReplicationSlot replication_slots[FLEXIBLE_ARRAY_MEMBER]; } ReplicationSlotCtlData; /* @@ -370,6 +369,7 @@ extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, TransactionId snapshotConflictHorizon); extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock); extern int ReplicationSlotIndex(ReplicationSlot *slot); +extern int ReplicationSlotNumUsed(void); extern bool ReplicationSlotName(int index, Name name); extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot); extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok); diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h index a4df3b8e0ae..b0a3d9a72ba 100644 --- a/src/include/replication/walsender.h +++ b/src/include/replication/walsender.h @@ -48,6 +48,7 @@ extern void WalSndInitStopping(void); extern void WalSndWaitStopping(void); extern void HandleWalSndInitStopping(void); extern void WalSndRqstFileReload(void); +extern int WalSndNumActive(void); /* * Remember that we want to wakeup walsenders later diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h index 5de674d5410..98618a14e34 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -16,6 +16,7 @@ #include "lib/ilist.h" #include "nodes/nodes.h" #include "nodes/replnodes.h" +#include "port/atomics.h" #include "replication/syncrep.h" #include "storage/condition_variable.h" #include "storage/shmem.h" @@ -113,6 +114,9 @@ typedef struct */ ConditionVariable wal_confirm_rcv_cv; + /* Number of active walsenders. */ + pg_atomic_uint32 active_wal_senders; + WalSnd walsnds[FLEXIBLE_ARRAY_MEMBER]; } WalSndCtlData; -- 2.52.0 ^ permalink raw reply [nested|flat] 20+ messages in thread
* Re: Don't keep closed WAL segment in page cache after replay @ 2026-03-04 01:40 Japin Li <[email protected]> parent: Anthonin Bonnefoy <[email protected]> 0 siblings, 1 reply; 20+ messages in thread From: Japin Li @ 2026-03-04 01:40 UTC (permalink / raw) To: Anthonin Bonnefoy <[email protected]>; +Cc: Hüseyin Demir <[email protected]>; [email protected] Hi, Anthonin Date: Wed, 04 Mar 2026 09:37:27 +0800 On Tue, 03 Mar 2026 at 15:13, Anthonin Bonnefoy <[email protected]> wrote: > On Mon, Mar 2, 2026 at 9:15 AM Hüseyin Demir <[email protected]> wrote: >> The idea looks good and efficient albeit I have some feedback. The first one is about logical replication slot. >> >> Inside the patch, it checks if there is an active walsender >> process. Is it possible to create a replication slot and wait until >> a subscriber will connect it. During this time due to patch >> PostgreSQL will close the WAL segments on the memory and once the >> subscriber connects it has to read the WAL files from disk. But it's >> a trade-off and can be decided by others too. > > Good point. This is also the case for physical replication slots, if > there's at least one used replication slot, then it's very likely a > walsender will start at some point and would need to read the WAL. And > having to read a large amount of WAL files from disk is likely not > desirable, so it's probably better to add this as a condition. > >> Secondly, when it comes to using spinlock to check the running >> walsender processes it can lead inefficient recovery >> process. Because assuming that a database with max_wal_sender set to >> 20+ and producing more than 4-5TB WAL in a day it can cause >> additional +100-200 spinlocks each second on walreceiver. Put >> simply, WalSndRunning() scans every walsender slot with spinlocks on >> every segment switch, contending with all active walsenders updating >> their own slots. On high-throughput standbys this creates >> unnecessary cross-process spinlock contention in the recovery hot >> path — the >> exact path that should be as lean as possible for fast replay. Maybe you can implement a single pg_atomic_uint32 counter in WalSndCtlData and achieve the same result with zero contention. > > True. I think I've assumed that a segment closing is rare enough that > it would be fine to go through all walsenders, but there are certainly > clusters that can generate a large amount of segments. > > I've updated the patch with the suggested approach: > - a new atomic counter tracking the number of active walsenders > - a similar atomic counter for the number of used replication slots. > Otherwise, we would have a similar issue of going through all > max_replication_slots to check if one is used. > - Both are now used as condition to send POSIX_FADV_DONTNEED or not > > Regards, > Anthonin Bonnefoy I noticed two different comment styles in the v4 patch: 1. +/* + * Return the number of used replication slots + */ +int +ReplicationSlotNumUsed(void) 2. +/* + * Returns the number of active walsender processes + */ +int +WalSndNumActive(void) Both "Return ..." and "Returns ..." styles exist in the PostgreSQL codebase, but within the same patch, would it be better to use a consistent style? I'd like to use the imperative/singular form. What do you think? > > [2. text/x-diff; v4-0001-Don-t-keep-closed-WAL-segments-in-page-cache-afte.patch]... -- Regards, Japin Li ChengDu WenWu Information Technology Co., Ltd. ^ permalink raw reply [nested|flat] 20+ messages in thread
* Re: Don't keep closed WAL segment in page cache after replay @ 2026-03-04 07:38 Anthonin Bonnefoy <[email protected]> parent: Japin Li <[email protected]> 0 siblings, 1 reply; 20+ messages in thread From: Anthonin Bonnefoy @ 2026-03-04 07:38 UTC (permalink / raw) To: Japin Li <[email protected]>; +Cc: Hüseyin Demir <[email protected]>; [email protected] On Wed, Mar 4, 2026 at 2:40 AM Japin Li <[email protected]> wrote: > Both "Return ..." and "Returns ..." styles exist in the PostgreSQL codebase, > but within the same patch, would it be better to use a consistent style? That makes sense, I've updated the comment. I think I usually look around the code to copy the comments' style, but as you said, both are present in PostgreSQL codebase. Regards, Anthonin Bonnefoy Attachments: [application/octet-stream] v5-0001-Don-t-keep-closed-WAL-segments-in-page-cache-afte.patch (8.2K, ../../CAO6_XqoDPt9+VOGkVHx5QsaBZjL7OQVhnkwb+rmtRq3pdm-ojw@mail.gmail.com/2-v5-0001-Don-t-keep-closed-WAL-segments-in-page-cache-afte.patch) download | inline diff: From ad0a3cfe10bdd2cccc4274849c4a77898b06e13c Mon Sep 17 00:00:00 2001 From: Anthonin Bonnefoy <[email protected]> Date: Wed, 2 Jul 2025 09:58:52 +0200 Subject: Don't keep closed WAL segments in page cache after replay On a standby, the recovery process reads the WAL segments, applies changes and closes the segment. When closed, the segments will still be in page cache memory until they are evicted due to inactivity. The segments may be re-read if archive_mode is set to always, wal_summarizer is enabled or if the standby is used for replication and has an active walsender. The presence of a replication slots is also a likely indicator that a walsender will be started, and need to read the WAL segments. Outside of those circumstances, the WAL segments won't be re-read and keeping them in the page cache generates unnecessary memory pressure. A POSIX_FADV_DONTNEED is sent before closing a replayed WAL segment to immediately free any cached pages. --- src/backend/access/transam/xlogrecovery.c | 21 +++++++++++++++++++++ src/backend/replication/slot.c | 16 ++++++++++++++++ src/backend/replication/walsender.c | 14 +++++++++++++- src/include/replication/slot.h | 10 +++++----- src/include/replication/walsender.h | 1 + src/include/replication/walsender_private.h | 4 ++++ 6 files changed, 60 insertions(+), 6 deletions(-) diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index ecd66fd86a4..b5ffef515c7 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -49,6 +49,7 @@ #include "pgstat.h" #include "postmaster/bgwriter.h" #include "postmaster/startup.h" +#include "postmaster/walsummarizer.h" #include "replication/slot.h" #include "replication/slotsync.h" #include "replication/walreceiver.h" @@ -3370,6 +3371,26 @@ XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, } } + /* + * Once replayed, WAL segment files may be re-read in several cases: + * archive_mode is set to always, summarize_wal is enabled or the + * standby acts as a walsender for either logical or physical + * replication. Outside of those conditions, the WAL segment files + * shouldn't be re-read and we can signal the kernel to release any + * cached pages. We also check if any replication slot is used: If + * present, it's likely a walsender will start and will need to read + * the WAL segment files, so we may as well keep the pages in cache. + */ +#if defined(USE_POSIX_FADVISE) && defined(POSIX_FADV_DONTNEED) + { + if (StandbyMode && + XLogArchiveMode != ARCHIVE_MODE_ALWAYS && + !summarize_wal && + WalSndNumActive() == 0 && + ReplicationSlotNumUsed() == 0) + (void) posix_fadvise(readFile, 0, 0, POSIX_FADV_DONTNEED); + } +#endif close(readFile); readFile = -1; readSource = XLOG_FROM_ANY; diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 28c7019402b..9ea7272df4a 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -220,6 +220,7 @@ ReplicationSlotsShmemInit(void) /* First time through, so initialize */ MemSet(ReplicationSlotCtl, 0, ReplicationSlotsShmemSize()); + pg_atomic_init_u32(&ReplicationSlotCtl->used_replication_slots, 0); for (i = 0; i < max_replication_slots; i++) { @@ -522,6 +523,9 @@ ReplicationSlotCreate(const char *name, bool db_specific, if (SlotIsLogical(slot)) pgstat_create_replslot(slot); + /* Update the used_replication_slots counter with the created slot */ + pg_atomic_fetch_add_u32(&ReplicationSlotCtl->used_replication_slots, 1); + /* * Now that the slot has been marked as in_use and active, it's safe to * let somebody else try to allocate a slot. @@ -579,6 +583,15 @@ ReplicationSlotIndex(ReplicationSlot *slot) return slot - ReplicationSlotCtl->replication_slots; } +/* + * Return the number of used replication slots + */ +int +ReplicationSlotNumUsed(void) +{ + return pg_atomic_read_u32(&ReplicationSlotCtl->used_replication_slots); +} + /* * If the slot at 'index' is unused, return false. Otherwise 'name' is set to * the slot's name and true is returned. @@ -1143,6 +1156,9 @@ ReplicationSlotDropPtr(ReplicationSlot *slot) if (SlotIsLogical(slot)) pgstat_drop_replslot(slot); + /* Decrement the used_replication_slots counter */ + pg_atomic_sub_fetch_u32(&ReplicationSlotCtl->used_replication_slots, 1); + /* * We release this at the very end, so that nobody starts trying to create * a slot while we're still cleaning up the detritus of the old one. diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2cde8ebc729..13fa279658f 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -3092,7 +3092,7 @@ InitWalSenderSlot(void) SpinLockRelease(&walsnd->mutex); /* don't need the lock anymore */ MyWalSnd = walsnd; - + pg_atomic_fetch_add_u32(&WalSndCtl->active_wal_senders, 1); break; } } @@ -3117,6 +3117,7 @@ WalSndKill(int code, Datum arg) /* Mark WalSnd struct as no longer being in use. */ walsnd->pid = 0; SpinLockRelease(&walsnd->mutex); + pg_atomic_fetch_sub_u32(&WalSndCtl->active_wal_senders, 1); } /* XLogReaderRoutine->segment_open callback */ @@ -3697,6 +3698,15 @@ WalSndRqstFileReload(void) } } +/* + * Return the number of active walsender processes + */ +int +WalSndNumActive(void) +{ + return pg_atomic_read_u32(&WalSndCtl->active_wal_senders); +} + /* * Handle PROCSIG_WALSND_INIT_STOPPING signal. */ @@ -3785,6 +3795,8 @@ WalSndShmemInit(void) SpinLockInit(&walsnd->mutex); } + pg_atomic_init_u32(&WalSndCtl->active_wal_senders, 0); + ConditionVariableInit(&WalSndCtl->wal_flush_cv); ConditionVariableInit(&WalSndCtl->wal_replay_cv); ConditionVariableInit(&WalSndCtl->wal_confirm_rcv_cv); diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 4b4709f6e2c..491b8e53db5 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -292,11 +292,10 @@ typedef struct ReplicationSlot */ typedef struct ReplicationSlotCtlData { - /* - * This array should be declared [FLEXIBLE_ARRAY_MEMBER], but for some - * reason you can't do that in an otherwise-empty struct. - */ - ReplicationSlot replication_slots[1]; + /* Number of used replication slots */ + pg_atomic_uint32 used_replication_slots; + + ReplicationSlot replication_slots[FLEXIBLE_ARRAY_MEMBER]; } ReplicationSlotCtlData; /* @@ -370,6 +369,7 @@ extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, TransactionId snapshotConflictHorizon); extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock); extern int ReplicationSlotIndex(ReplicationSlot *slot); +extern int ReplicationSlotNumUsed(void); extern bool ReplicationSlotName(int index, Name name); extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot); extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok); diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h index a4df3b8e0ae..b0a3d9a72ba 100644 --- a/src/include/replication/walsender.h +++ b/src/include/replication/walsender.h @@ -48,6 +48,7 @@ extern void WalSndInitStopping(void); extern void WalSndWaitStopping(void); extern void HandleWalSndInitStopping(void); extern void WalSndRqstFileReload(void); +extern int WalSndNumActive(void); /* * Remember that we want to wakeup walsenders later diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h index 5de674d5410..98618a14e34 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -16,6 +16,7 @@ #include "lib/ilist.h" #include "nodes/nodes.h" #include "nodes/replnodes.h" +#include "port/atomics.h" #include "replication/syncrep.h" #include "storage/condition_variable.h" #include "storage/shmem.h" @@ -113,6 +114,9 @@ typedef struct */ ConditionVariable wal_confirm_rcv_cv; + /* Number of active walsenders. */ + pg_atomic_uint32 active_wal_senders; + WalSnd walsnds[FLEXIBLE_ARRAY_MEMBER]; } WalSndCtlData; -- 2.52.0 ^ permalink raw reply [nested|flat] 20+ messages in thread
* Re: Don't keep closed WAL segment in page cache after replay @ 2026-03-04 15:55 Andres Freund <[email protected]> parent: Anthonin Bonnefoy <[email protected]> 0 siblings, 0 replies; 20+ messages in thread From: Andres Freund @ 2026-03-04 15:55 UTC (permalink / raw) To: Anthonin Bonnefoy <[email protected]>; +Cc: Japin Li <[email protected]>; Hüseyin Demir <[email protected]>; [email protected] Hi, On 2026-03-04 08:38:24 +0100, Anthonin Bonnefoy wrote: > From ad0a3cfe10bdd2cccc4274849c4a77898b06e13c Mon Sep 17 00:00:00 2001 > From: Anthonin Bonnefoy <[email protected]> > Date: Wed, 2 Jul 2025 09:58:52 +0200 > Subject: Don't keep closed WAL segments in page cache after replay > > On a standby, the recovery process reads the WAL segments, applies > changes and closes the segment. When closed, the segments will still be > in page cache memory until they are evicted due to inactivity. The > segments may be re-read if archive_mode is set to always, wal_summarizer > is enabled or if the standby is used for replication and has an active > walsender. > > The presence of a replication slots is also a likely indicator that > a walsender will be started, and need to read the WAL segments. > > Outside of those circumstances, the WAL segments won't be re-read and > keeping them in the page cache generates unnecessary memory pressure. > A POSIX_FADV_DONTNEED is sent before closing a replayed WAL segment to > immediately free any cached pages. I am quite sceptical that this is a good idea. Have you actually measured benefits? I skimmed the thread and didn't see anything. It's pretty cheap for the kernel to replace a clean page from the page cache with different content. If you [crash-]restart the replica this will make it way more expensive. If you have twophase commits where we need to read 2PC details from the WAL, this will make it more expensive. If somebody takes a base backup, this ... I think you'd have to have pretty convincing benchmarks showing that this is a good idea before we should even remotely consider applying this. Greetings, Andres Freund ^ permalink raw reply [nested|flat] 20+ messages in thread
end of thread, other threads:[~2026-03-04 15:55 UTC | newest] Thread overview: 20+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2019-07-05 14:24 [PATCH 06/17] Allow user to use password instead of encryption key. Antonin Houska <[email protected]> 2023-07-20 17:19 [PATCH v6 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v10 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v9 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v10 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v10 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v9 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v9 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-07-20 17:19 [PATCH v8 3/4] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-09-04 22:35 [PATCH v7 4/5] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-09-04 22:35 [PATCH v7 4/5] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2023-09-04 22:35 [PATCH v7 4/5] Convert pg_restore's ready_list to a priority queue. Nathan Bossart <[email protected]> 2026-02-18 09:51 Re: Don't keep closed WAL segment in page cache after replay Anthonin Bonnefoy <[email protected]> 2026-03-03 14:13 ` Re: Don't keep closed WAL segment in page cache after replay Anthonin Bonnefoy <[email protected]> 2026-03-04 01:40 ` Re: Don't keep closed WAL segment in page cache after replay Japin Li <[email protected]> 2026-03-04 07:38 ` Re: Don't keep closed WAL segment in page cache after replay Anthonin Bonnefoy <[email protected]> 2026-03-04 15:55 ` Re: Don't keep closed WAL segment in page cache after replay Andres Freund <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox