From cfa1764ccc16552988c0deb7190c0b8ffb5b2215 Mon Sep 17 00:00:00 2001 From: Jelte Fennema-Nio Date: Sat, 4 Jul 2026 11:11:09 +0200 Subject: [PATCH v3] pg_dump: Remove TerminateThread() call When pg_dump or pg_restore --jobs N is interrupted with ^C on Windows, we cancel all queries, but we don't want the cancellations to be reported as errors to the user in the short time before the whole process exits. That was previously achieved by calling TerminateThread() on each worker thread before sending the cancel message, but that doesn't appear to be 100% safe: the implementations of write() and the socket calls inside PQcancel() might acquire user space locks that were held by the terminated threads. (write() certainly does that.) Instead of silencing the threads in such a sketchy way this now sets a volatile flag (before sending any cancel requests) that tells the threads to not log errors anymore. Instead of a volatile, it would be better to use an atomic operation here, that has to wait until we add support for atomics on the frontend though. Note that this also stops using pg_fatal and exit to exit from workers on failure and instead use pg_log_error combined with exit_nicely. If a query fails in a worker we want it to kill the worker not the whole process. On Unix that's currently the same thing, but on Windows workers are threads. --- src/bin/pg_dump/parallel.c | 50 ++++++++++---------- src/bin/pg_dump/pg_backup_archiver.c | 69 +++++++++++++++------------- src/bin/pg_dump/pg_backup_db.c | 30 ++++++++---- src/bin/pg_dump/pg_backup_utils.c | 36 +++++++++++++++ src/bin/pg_dump/pg_backup_utils.h | 21 +++++++++ 5 files changed, 139 insertions(+), 67 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index b77d2650df0..1bbcf91cd1f 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -531,11 +531,12 @@ WaitForTerminatingWorkers(ParallelState *pstate) * might be that only the leader gets signaled. * * On Windows, the cancel handler runs in a separate thread, because that's - * how SetConsoleCtrlHandler works. We make it stop worker threads, send - * cancels on all active connections, and then return FALSE, which will allow - * the process to die. For safety's sake, we use a critical section to - * protect the PGcancel structures against being changed while the signal - * thread runs. + * how SetConsoleCtrlHandler works. Because the workers are threads in this + * same process, we set a flag (is_cancel_in_progress()) so they stay quiet about + * the query cancellations instead of cluttering the screen, then send cancels + * on all active connections and return FALSE, which will allow the process to + * die. For safety's sake, we use a critical section to protect the PGcancel + * structures against being changed while the signal thread runs. */ #ifndef WIN32 @@ -641,34 +642,30 @@ consoleHandler(DWORD dwCtrlType) if (dwCtrlType == CTRL_C_EVENT || dwCtrlType == CTRL_BREAK_EVENT) { + /* + * Tell worker threads to stay quiet about the query cancellations + * we're about to send them; otherwise they'd report them as errors + * and clutter the user's screen. This must be set before we send any + * cancel, so that a worker is guaranteed to see it by the time its + * query fails as a result. + */ + set_cancel_in_progress(); + /* Critical section prevents changing data we look at here */ EnterCriticalSection(&signal_info_lock); /* - * If in parallel mode, stop worker threads and send QueryCancel to - * their connected backends. The main point of stopping the worker - * threads is to keep them from reporting the query cancels as errors, - * which would clutter the user's screen. We needn't stop the leader - * thread since it won't be doing much anyway. Do this before - * canceling the main transaction, else we might get invalid-snapshot - * errors reported before we can stop the workers. Ignore errors, - * there's not much we can do about them anyway. + * If in parallel mode, send QueryCancel to each worker's connected + * backend. Do this before canceling the main transaction, else we + * might get invalid-snapshot errors reported before we can stop the + * workers. Ignore errors, there's not much we can do about them + * anyway. */ if (signal_info.pstate != NULL) { for (i = 0; i < signal_info.pstate->numWorkers; i++) { - ParallelSlot *slot = &(signal_info.pstate->parallelSlot[i]); - ArchiveHandle *AH = slot->AH; - HANDLE hThread = (HANDLE) slot->hThread; - - /* - * Using TerminateThread here may leave some resources leaked, - * but it doesn't matter since we're about to end the whole - * process. - */ - if (hThread != INVALID_HANDLE_VALUE) - TerminateThread(hThread, 0); + ArchiveHandle *AH = signal_info.pstate->parallelSlot[i].AH; if (AH != NULL && AH->connCancel != NULL) (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); @@ -687,9 +684,8 @@ consoleHandler(DWORD dwCtrlType) /* * Report we're quitting, using nothing more complicated than - * write(2). (We might be able to get away with using pg_log_*() - * here, but since we terminated other threads uncleanly above, it - * seems better to assume as little as possible.) + * write(2). We should be able to use pg_log_*() here, but for now we + * stay aligned with the sigTermHandler behavior. */ if (progname) { diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 0557eb6d6ed..77cc50e0607 100644 --- a/src/bin/pg_dump/pg_backup_archiver.c +++ b/src/bin/pg_dump/pg_backup_archiver.c @@ -1896,47 +1896,52 @@ ahwrite(const void *ptr, size_t size, size_t nmemb, ArchiveHandle *AH) void warn_or_exit_horribly(ArchiveHandle *AH, const char *fmt, ...) { - va_list ap; - - switch (AH->stage) + /* Stay quiet if this is a result of our own cancellation. */ + if (!is_cancel_in_progress()) { + va_list ap; - case STAGE_NONE: - /* Do nothing special */ - break; + switch (AH->stage) + { - case STAGE_INITIALIZING: - if (AH->stage != AH->lastErrorStage) - pg_log_info("while INITIALIZING:"); - break; + case STAGE_NONE: + /* Do nothing special */ + break; - case STAGE_PROCESSING: - if (AH->stage != AH->lastErrorStage) - pg_log_info("while PROCESSING TOC:"); - break; + case STAGE_INITIALIZING: + if (AH->stage != AH->lastErrorStage) + pg_log_info("while INITIALIZING:"); + break; - case STAGE_FINALIZING: - if (AH->stage != AH->lastErrorStage) - pg_log_info("while FINALIZING:"); - break; - } - if (AH->currentTE != NULL && AH->currentTE != AH->lastErrorTE) - { - pg_log_info("from TOC entry %d; %u %u %s %s %s", - AH->currentTE->dumpId, - AH->currentTE->catalogId.tableoid, - AH->currentTE->catalogId.oid, - AH->currentTE->desc ? AH->currentTE->desc : "(no desc)", - AH->currentTE->tag ? AH->currentTE->tag : "(no tag)", - AH->currentTE->owner ? AH->currentTE->owner : "(no owner)"); + case STAGE_PROCESSING: + if (AH->stage != AH->lastErrorStage) + pg_log_info("while PROCESSING TOC:"); + break; + + case STAGE_FINALIZING: + if (AH->stage != AH->lastErrorStage) + pg_log_info("while FINALIZING:"); + break; + } + if (AH->currentTE != NULL && AH->currentTE != AH->lastErrorTE) + { + pg_log_info("from TOC entry %d; %u %u %s %s %s", + AH->currentTE->dumpId, + AH->currentTE->catalogId.tableoid, + AH->currentTE->catalogId.oid, + AH->currentTE->desc ? AH->currentTE->desc : "(no desc)", + AH->currentTE->tag ? AH->currentTE->tag : "(no tag)", + AH->currentTE->owner ? AH->currentTE->owner : "(no owner)"); + } + + va_start(ap, fmt); + pg_log_generic_v(PG_LOG_ERROR, PG_LOG_PRIMARY, fmt, ap); + va_end(ap); } + AH->lastErrorStage = AH->stage; AH->lastErrorTE = AH->currentTE; - va_start(ap, fmt); - pg_log_generic_v(PG_LOG_ERROR, PG_LOG_PRIMARY, fmt, ap); - va_end(ap); - if (AH->public.exit_on_error) exit_nicely(1); else diff --git a/src/bin/pg_dump/pg_backup_db.c b/src/bin/pg_dump/pg_backup_db.c index ec0ddf1d718..17c0b7cbdf2 100644 --- a/src/bin/pg_dump/pg_backup_db.c +++ b/src/bin/pg_dump/pg_backup_db.c @@ -207,10 +207,14 @@ notice_processor(void *arg, const char *message) static void die_on_query_failure(ArchiveHandle *AH, const char *query) { - pg_log_error("query failed: %s", - PQerrorMessage(AH->connection)); - pg_log_error_detail("Query was: %s", query); - exit(1); + if (!is_cancel_in_progress()) + { + pg_log_error("query failed: %s", + PQerrorMessage(AH->connection)); + pg_log_error_detail("Query was: %s", query); + } + + exit_nicely(1); } void @@ -396,8 +400,13 @@ ExecuteSqlCommandBuf(Archive *AHX, const char *buf, size_t bufLen) */ if (AH->pgCopyIn && PQputCopyData(AH->connection, buf, bufLen) <= 0) - pg_fatal("error returned by PQputCopyData: %s", - PQerrorMessage(AH->connection)); + { + /* Stay quiet if this is a result of our own cancellation. */ + if (!is_cancel_in_progress()) + pg_log_error("error returned by PQputCopyData: %s", + PQerrorMessage(AH->connection)); + exit_nicely(1); + } } else if (AH->outputKind == OUTPUT_OTHERDATA) { @@ -445,8 +454,13 @@ EndDBCopyMode(Archive *AHX, const char *tocEntryTag) PGresult *res; if (PQputCopyEnd(AH->connection, NULL) <= 0) - pg_fatal("error returned by PQputCopyEnd: %s", - PQerrorMessage(AH->connection)); + { + /* Stay quiet if this is a result of our own cancellation. */ + if (!is_cancel_in_progress()) + pg_log_error("error returned by PQputCopyEnd: %s", + PQerrorMessage(AH->connection)); + exit_nicely(1); + } /* Check command status and return to normal libpq state */ res = PQgetResult(AH->connection); diff --git a/src/bin/pg_dump/pg_backup_utils.c b/src/bin/pg_dump/pg_backup_utils.c index 0368f7623a7..a291b9953e6 100644 --- a/src/bin/pg_dump/pg_backup_utils.c +++ b/src/bin/pg_dump/pg_backup_utils.c @@ -21,6 +21,42 @@ /* Globals exported by this file */ const char *progname = NULL; +#ifdef WIN32 + +/* + * Flag telling worker threads to stay quiet about query failures because we're + * cancelling their queries as part of tearing down the process. See the + * comment in pg_backup_utils.h. + * + * The cancel thread writes it while worker threads read it, so it's marked + * volatile to keep the compiler from caching the value. A plain volatile bool + * isn't a real memory barrier, but it's good enough here: the flag is only ever + * flipped one way (false to true) and a worker briefly observing the stale + * false just means it prints one error before the process dies. The only goal + * of this flag is to make sure workers don't log "query cancelled" errors + * during the shutdown process. When such an error is received that acts as a + * full barrier so for the messages that we care about a stale value won't be + * observed. + * + * XXX: This should be swapped out for a proper atomic when we have those in + * the frontend code. + */ +static volatile bool cancelInProgress = false; + +void +set_cancel_in_progress(void) +{ + cancelInProgress = true; +} + +bool +is_cancel_in_progress(void) +{ + return cancelInProgress; +} + +#endif /* WIN32 */ + #define MAX_ON_EXIT_NICELY 20 static struct diff --git a/src/bin/pg_dump/pg_backup_utils.h b/src/bin/pg_dump/pg_backup_utils.h index 9e98ed11619..31e66b61198 100644 --- a/src/bin/pg_dump/pg_backup_utils.h +++ b/src/bin/pg_dump/pg_backup_utils.h @@ -31,6 +31,27 @@ extern void set_dump_section(const char *arg, int *dumpSections); extern void on_exit_nicely(on_exit_nicely_callback function, void *arg); pg_noreturn extern void exit_nicely(int code); +/* + * On Windows the parallel workers are threads inside the leader process. When + * a cancel is processed there, the leader sends cancels to the workers' + * in-flight queries; without this flag each worker would then report the + * resulting "canceling statement due to user request" error and clutter the + * screen in the brief window before the whole process exits. The cancel + * thread sets this flag before sending any cancel, and worker threads check it + * before reporting a query failure. It's written and read from different + * threads, so it's marked volatile; see the comment in pg_backup_utils.c. + * + * On other platforms the workers are separate processes that just _exit() when + * cancelled, so they never reach the error-reporting code; there the check is + * compiled out to a constant false and the flag doesn't exist. + */ +#ifdef WIN32 +extern void set_cancel_in_progress(void); +extern bool is_cancel_in_progress(void); +#else +#define is_cancel_in_progress() false +#endif + /* In pg_dump, we modify pg_fatal to call exit_nicely instead of exit */ #undef pg_fatal #define pg_fatal(...) do { \ base-commit: 07211f64ace0150c92a00769a1cfe8b9305b9e78 -- 2.54.0