A bunch of frontend tools, including psql, still used PQcancel to send cancel requests to the server. That function is insecure, because it does not use encryption to send the cancel request. This starts using the new cancellation APIs (introduced in 61461a300) for all these frontend tools. These APIs use the same encryption settings as the connection that's being cancelled. Since these APIs are not signal-safe this required a refactor to not send the cancel request in a signal handler anymore, but instead using a dedicated thread. Similar logic was already used for Windows anyway, so this also has the benefit that it makes the cancel logic more uniform across our supported platforms.
Because this is fixing a security issue, it would be nice if we could backport it. I'm not sure that's realistic though, given the size/complexity of the change. I'm curious what others think about that. To be clear, I'm only really talking about backporting to PG17 and PG18 because those contain the new cancellation APIs in libpq. Backporting to even older versions would also require backporting the new cancellation APIs in libpq, which seems like a no-go. A possible follow up improvement to pg_dump would be to use threads for its parallel workers on UNIX as well. Then the Windows and Unix implementations would get even more aligned. I started looking into that, but that quickly became quite a big refactor, touching a lot of code unrelated to cancellation. So it seems better to do that in a separate follow-on patch if people are interested.
From 0a1d3b5894502a25a2c5ed5a283251e3296ae413 Mon Sep 17 00:00:00 2001 From: Jelte Fennema-Nio <[email protected]> Date: Sat, 13 Dec 2025 18:18:13 +0100 Subject: [PATCH v1 1/2] Move Windows pthread compatibility functions to src/port This is in preparation of a follow-up commit which will start to use these functions in more places than just libpq. --- configure.ac | 1 + src/interfaces/libpq/Makefile | 1 - src/interfaces/libpq/meson.build | 2 +- src/port/meson.build | 1 + src/{interfaces/libpq => port}/pthread-win32.c | 5 +++-- 5 files changed, 6 insertions(+), 4 deletions(-) rename src/{interfaces/libpq => port}/pthread-win32.c (94%) diff --git a/configure.ac b/configure.ac index 4ba3967db0f..e73e1287bf4 100644 --- a/configure.ac +++ b/configure.ac @@ -1887,6 +1887,7 @@ if test "$PORTNAME" = "win32"; then AC_LIBOBJ(dirmod) AC_LIBOBJ(kill) AC_LIBOBJ(open) + AC_LIBOBJ(pthread-win32) AC_LIBOBJ(system) AC_LIBOBJ(win32common) AC_LIBOBJ(win32dlopen) diff --git a/src/interfaces/libpq/Makefile b/src/interfaces/libpq/Makefile index 9fe321147fc..98cfe42db34 100644 --- a/src/interfaces/libpq/Makefile +++ b/src/interfaces/libpq/Makefile @@ -75,7 +75,6 @@ endif ifeq ($(PORTNAME), win32) OBJS += \ - pthread-win32.o \ win32.o endif diff --git a/src/interfaces/libpq/meson.build b/src/interfaces/libpq/meson.build index b259c998fa2..eb8bf7b45bf 100644 --- a/src/interfaces/libpq/meson.build +++ b/src/interfaces/libpq/meson.build @@ -20,7 +20,7 @@ libpq_sources = files( libpq_so_sources = [] # for shared lib, in addition to the above if host_system == 'windows' - libpq_sources += files('pthread-win32.c', 'win32.c') + libpq_sources += files('win32.c') libpq_so_sources += rc_lib_gen.process(win32ver_rc, extra_args: [ '--NAME', 'libpq', '--FILEDESC', 'PostgreSQL Access Library',]) diff --git a/src/port/meson.build b/src/port/meson.build index fc7b059fee5..c1e53a9a341 100644 --- a/src/port/meson.build +++ b/src/port/meson.build @@ -31,6 +31,7 @@ if host_system == 'windows' 'dirmod.c', 'kill.c', 'open.c', + 'pthread-win32.c', 'system.c', 'win32common.c', 'win32dlopen.c', diff --git a/src/interfaces/libpq/pthread-win32.c b/src/port/pthread-win32.c similarity index 94% rename from src/interfaces/libpq/pthread-win32.c rename to src/port/pthread-win32.c index db75d491b90..f3e5e08cb09 100644 --- a/src/interfaces/libpq/pthread-win32.c +++ b/src/port/pthread-win32.c @@ -4,13 +4,14 @@ * partial pthread implementation for win32 * * Copyright (c) 2004-2025, PostgreSQL Global Development Group +* * IDENTIFICATION -* src/interfaces/libpq/pthread-win32.c +* src/port/pthread-win32.c * *------------------------------------------------------------------------- */ -#include "postgres_fe.h" +#include "c.h" #include "pthread-win32.h" base-commit: c5ae07a90a0f3594e5053a26f3c99b041df427d3 -- 2.52.0
From d96d56209f4f9f0021cb12606b381b909018a2b9 Mon Sep 17 00:00:00 2001 From: Jelte Fennema-Nio <[email protected]> Date: Sat, 13 Dec 2025 13:05:50 +0100 Subject: [PATCH v1 2/2] Don't use the deprecated and insecure PQcancel in our frontend tools anymore A bunch of frontend tools, including psql, still used PQcancel to send cancel requests to the server. That function is insecure, because it does not use encryption to send the cancel request. This starts using the new cancellation APIs (introduced in 61461a300) for all these frontend tools. These APIs use the same encryption settings as the connection that's being cancelled. Since these APIs are not signal-safe this required a refactor to not send the cancel request in a signal handler anymore, but instead using a dedicated thread. Similar logic was already used for Windows anyway, so this also has the benefit that it makes the cancel logic more uniform across our supported platforms. --- meson.build | 2 +- src/bin/pg_dump/Makefile | 2 +- src/bin/pg_dump/meson.build | 2 + src/bin/pg_dump/parallel.c | 401 ++++++++++++++------------- src/bin/pg_dump/pg_backup_archiver.c | 2 +- src/bin/pg_dump/pg_backup_archiver.h | 8 +- src/bin/pg_dump/pg_backup_db.c | 7 +- src/fe_utils/Makefile | 2 +- src/fe_utils/cancel.c | 314 +++++++++++++-------- 9 files changed, 415 insertions(+), 325 deletions(-) diff --git a/meson.build b/meson.build index 1256094fa57..202c7c5dda4 100644 --- a/meson.build +++ b/meson.build @@ -3337,7 +3337,7 @@ frontend_code = declare_dependency( include_directories: [postgres_inc], link_with: [fe_utils, common_static, pgport_static], sources: generated_headers_stamp, - dependencies: [os_deps, libintl], + dependencies: [os_deps, libintl, thread_dep], ) backend_both_deps += [ diff --git a/src/bin/pg_dump/Makefile b/src/bin/pg_dump/Makefile index fa795883e9f..0500eb21b0d 100644 --- a/src/bin/pg_dump/Makefile +++ b/src/bin/pg_dump/Makefile @@ -21,7 +21,7 @@ export LZ4 export ZSTD export with_icu -override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS) +override CPPFLAGS := -I$(libpq_srcdir) -I$(top_srcdir)/src/port $(CPPFLAGS) LDFLAGS_INTERNAL += -L$(top_builddir)/src/fe_utils -lpgfeutils $(libpq_pgport) OBJS = \ diff --git a/src/bin/pg_dump/meson.build b/src/bin/pg_dump/meson.build index f3c669f484e..ac2130afdfe 100644 --- a/src/bin/pg_dump/meson.build +++ b/src/bin/pg_dump/meson.build @@ -22,6 +22,8 @@ pg_dump_common_sources = files( pg_dump_common = static_library('libpgdump_common', pg_dump_common_sources, c_pch: pch_postgres_fe_h, + # port needs to be in include path due to pthread-win32.h + include_directories: ['../../port'], dependencies: [frontend_code, libpq, lz4, zlib, zstd], kwargs: internal_lib_args, ) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 086adcdc502..f34e9651099 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -58,8 +58,12 @@ #include <signal.h> #include <unistd.h> #include <fcntl.h> +#include <pthread.h> +#else +#include "pthread-win32.h" #endif +#include "common/logging.h" #include "fe_utils/string_utils.h" #include "parallel.h" #include "pg_backup_utils.h" @@ -167,6 +171,7 @@ typedef struct DumpSignalInformation ArchiveHandle *myAH; /* database connection to issue cancel for */ ParallelState *pstate; /* parallel state, if any */ bool handler_set; /* signal handler set up in this process? */ + bool cancel_requested; /* cancel requested via signal? */ #ifndef WIN32 bool am_worker; /* am I a worker process? */ #endif @@ -174,8 +179,18 @@ typedef struct DumpSignalInformation static volatile DumpSignalInformation signal_info; -#ifdef WIN32 -static CRITICAL_SECTION signal_info_lock; +/* + * Mutex protecting signal_info during cancel operations. + */ +static pthread_mutex_t signal_info_lock; + +#ifndef WIN32 +/* + * On Unix, we use a self-pipe to wake up a cancel thread from the signal + * handler, since it's not safe to call PQcancelBlocking from a signal handler. + */ +static int cancel_pipe[2] = {-1, -1}; +static pthread_t cancel_thread; #endif /* @@ -209,6 +224,7 @@ static void WaitForTerminatingWorkers(ParallelState *pstate); static void set_cancel_handler(void); static void set_cancel_pstate(ParallelState *pstate); static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH); +static void StopWorkers(void); static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot); static int GetIdleWorker(ParallelState *pstate); static bool HasEveryWorkerTerminated(ParallelState *pstate); @@ -410,32 +426,9 @@ ShutdownWorkersHard(ParallelState *pstate) /* * Force early termination of any commands currently in progress. */ -#ifndef WIN32 - /* On non-Windows, send SIGTERM to each worker process. */ - for (i = 0; i < pstate->numWorkers; i++) - { - pid_t pid = pstate->parallelSlot[i].pid; - - if (pid != 0) - kill(pid, SIGTERM); - } -#else - - /* - * On Windows, send query cancels directly to the workers' backends. Use - * a critical section to ensure worker threads don't change state. - */ - EnterCriticalSection(&signal_info_lock); - for (i = 0; i < pstate->numWorkers; i++) - { - ArchiveHandle *AH = pstate->parallelSlot[i].AH; - char errbuf[1]; - - if (AH != NULL && AH->connCancel != NULL) - (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); - } - LeaveCriticalSection(&signal_info_lock); -#endif + pthread_mutex_lock(&signal_info_lock); + StopWorkers(); + pthread_mutex_unlock(&signal_info_lock); /* Now wait for them to terminate. */ WaitForTerminatingWorkers(pstate); @@ -519,74 +512,54 @@ WaitForTerminatingWorkers(ParallelState *pstate) * could leave a SQL command (e.g., CREATE INDEX on a large table) running * for a long time. Instead, we try to send a cancel request and then die. * pg_dump probably doesn't really need this, but we might as well use it - * there too. Note that sending the cancel directly from the signal handler - * is safe because PQcancel() is written to make it so. + * there too. * - * In parallel operation on Unix, each process is responsible for canceling - * its own connection (this must be so because nobody else has access to it). - * Furthermore, the leader process should attempt to forward its signal to - * each child. In simple manual use of pg_dump/pg_restore, forwarding isn't - * needed because typing control-C at the console would deliver SIGINT to - * every member of the terminal process group --- but in other scenarios it - * might be that only the leader gets signaled. + * On Unix, the signal handler wakes up a dedicated cancel thread via a + * self-pipe, which then sends the cancel and calls _exit(). This thread also + * forwards the signal to each child so they can also cancel their queries. In + * simple manual use of pg_dump/pg_restore, forwarding isn't needed because + * typing control-C at the console would deliver SIGINT to every member of the + * terminal process group --- but in other scenarios it 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 + * protect the PGcancelConn structures against being changed while the signal * thread runs. */ -#ifndef WIN32 - /* - * Signal handler (Unix only) + * Cancel all active queries and print termination message. */ static void -sigTermHandler(SIGNAL_ARGS) +CancelBackends(void) { - int i; - char errbuf[1]; + pthread_mutex_lock(&signal_info_lock); - /* - * Some platforms allow delivery of new signals to interrupt an active - * signal handler. That could muck up our attempt to send PQcancel, so - * disable the signals that set_cancel_handler enabled. - */ - pqsignal(SIGINT, SIG_IGN); - pqsignal(SIGTERM, SIG_IGN); - pqsignal(SIGQUIT, SIG_IGN); + signal_info.cancel_requested = true; /* - * If we're in the leader, forward signal to all workers. (It seems best - * to do this before PQcancel; killing the leader transaction will result - * in invalid-snapshot errors from active workers, which maybe we can - * quiet by killing workers first.) Ignore any errors. + * Stop workers first to avoid invalid-snapshot errors if the leader + * cancels before workers. */ - if (signal_info.pstate != NULL) - { - for (i = 0; i < signal_info.pstate->numWorkers; i++) - { - pid_t pid = signal_info.pstate->parallelSlot[i].pid; + StopWorkers(); - if (pid != 0) - kill(pid, SIGTERM); - } - } + if (signal_info.myAH != NULL && signal_info.myAH->cancelConn != NULL) + (void) PQcancelBlocking(signal_info.myAH->cancelConn); - /* - * Send QueryCancel if we have a connection to send to. Ignore errors, - * there's not much we can do about them anyway. - */ - if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL) - (void) PQcancel(signal_info.myAH->connCancel, errbuf, sizeof(errbuf)); + pthread_mutex_unlock(&signal_info_lock); /* - * Report we're quitting, using nothing more complicated than write(2). - * When in parallel operation, only the leader process should do this. + * Print termination message. In parallel operation, only the leader + * should print this. On Windows, workers are threads in the same process + * and the console handler only runs in the leader context, so we can + * always print it. */ +#ifndef WIN32 if (!signal_info.am_worker) +#endif { if (progname) { @@ -595,172 +568,203 @@ sigTermHandler(SIGNAL_ARGS) } write_stderr("terminated by user\n"); } - - /* - * And die, using _exit() not exit() because the latter will invoke atexit - * handlers that can fail if we interrupted related code. - */ - _exit(1); } /* - * Enable cancel interrupt handler, if not already done. + * Stop all worker processes/threads. + * + * On Unix, send SIGTERM to each worker process; their signal handlers will + * send cancel requests to their backends. + * + * On Windows, workers are threads in the same process, so we send cancel + * requests directly to their backends. + * + * Caller must hold signal_info_lock. */ static void -set_cancel_handler(void) +StopWorkers(void) { - /* - * When forking, signal_info.handler_set will propagate into the new - * process, but that's fine because the signal handler state does too. - */ - if (!signal_info.handler_set) + int i; + + if (signal_info.pstate == NULL) + return; + + for (i = 0; i < signal_info.pstate->numWorkers; i++) { - signal_info.handler_set = true; +#ifndef WIN32 + pid_t pid = signal_info.pstate->parallelSlot[i].pid; + + if (pid != 0) + kill(pid, SIGTERM); +#else + ArchiveHandle *AH = signal_info.pstate->parallelSlot[i].AH; - pqsignal(SIGINT, sigTermHandler); - pqsignal(SIGTERM, sigTermHandler); - pqsignal(SIGQUIT, sigTermHandler); + if (AH != NULL && AH->cancelConn != NULL) + (void) PQcancelBlocking(AH->cancelConn); +#endif } } -#else /* WIN32 */ +#ifdef WIN32 /* * Console interrupt handler --- runs in a newly-started thread. * - * After stopping other threads and sending cancel requests on all open - * connections, we return FALSE which will allow the default ExitProcess() - * action to be taken. + * Send cancel requests on all open connections and return FALSE to allow + * the default ExitProcess() action to terminate the process. */ static BOOL WINAPI consoleHandler(DWORD dwCtrlType) { - int i; - char errbuf[1]; - if (dwCtrlType == CTRL_C_EVENT || dwCtrlType == CTRL_BREAK_EVENT) { - /* Critical section prevents changing data we look at here */ - EnterCriticalSection(&signal_info_lock); + CancelBackends(); + } - /* - * 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 (signal_info.pstate != NULL) + /* Always return FALSE to allow signal handling to continue */ + return FALSE; +} + +#else /* !WIN32 */ + +/* + * Cancel thread main function. Waits for the signal handler to write to the + * pipe, then cancels backends and calls _exit(). + */ +static void * +cancel_thread_main(void *arg) +{ + for (;;) + { + char buf[16]; + ssize_t rc; + + /* Wait for signal handler to wake us up */ + rc = read(cancel_pipe[0], buf, sizeof(buf)); + if (rc <= 0) { - 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); - - if (AH != NULL && AH->connCancel != NULL) - (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); - } + if (errno == EINTR) + continue; + /* Pipe closed or error - exit thread */ + break; } - /* - * Send QueryCancel to leader connection, if enabled. Ignore errors, - * there's not much we can do about them anyway. - */ - if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL) - (void) PQcancel(signal_info.myAH->connCancel, - errbuf, sizeof(errbuf)); - - LeaveCriticalSection(&signal_info_lock); + CancelBackends(); /* - * 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.) + * And die, using _exit() not exit() because the latter will invoke + * atexit handlers that can fail if we interrupted related code. */ - if (progname) - { - write_stderr(progname); - write_stderr(": "); - } - write_stderr("terminated by user\n"); + _exit(1); } - /* Always return FALSE to allow signal handling to continue */ - return FALSE; + return NULL; } +/* + * Signal handler (Unix only). Wakes up the cancel thread by writing to the + * pipe. + */ +static void +sigTermHandler(SIGNAL_ARGS) +{ + int save_errno = errno; + char c = 1; + + /* Wake up the cancel thread - write() is async-signal-safe */ + if (cancel_pipe[1] >= 0) + (void) write(cancel_pipe[1], &c, 1); + + errno = save_errno; +} + +#endif /* WIN32 */ + /* * Enable cancel interrupt handler, if not already done. */ static void set_cancel_handler(void) { - if (!signal_info.handler_set) + if (signal_info.handler_set) + return; + + signal_info.handler_set = true; + + pthread_mutex_init(&signal_info_lock, NULL); + +#ifdef WIN32 + SetConsoleCtrlHandler(consoleHandler, TRUE); +#else + + /* + * Set up the self-pipe for communication between signal handler and + * cancel thread. We use a pipe because write() is async-signal-safe. + */ + if (pipe(cancel_pipe) < 0) { - signal_info.handler_set = true; + pg_log_error("could not create pipe for cancel: %m"); + exit(1); + } - InitializeCriticalSection(&signal_info_lock); + /* + * Create a thread to handle cancel requests. The signal handler will + * write to the pipe to wake up this thread. + */ + { + int rc; - SetConsoleCtrlHandler(consoleHandler, TRUE); + rc = pthread_create(&cancel_thread, NULL, cancel_thread_main, NULL); + if (rc != 0) + { + pg_log_error("could not create cancel thread: %s", strerror(rc)); + exit(1); + } } -} -#endif /* WIN32 */ + pqsignal(SIGINT, sigTermHandler); + pqsignal(SIGTERM, sigTermHandler); + pqsignal(SIGQUIT, sigTermHandler); +#endif +} /* * set_archive_cancel_info * - * Fill AH->connCancel with cancellation info for the specified database + * Fill AH->cancelConn with cancellation info for the specified database * connection; or clear it if conn is NULL. */ void set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) { - PGcancel *oldConnCancel; + PGcancelConn *oldCancelConn; /* - * Activate the interrupt handler if we didn't yet in this process. On - * Windows, this also initializes signal_info_lock; therefore it's - * important that this happen at least once before we fork off any - * threads. + * Activate the interrupt handler if we didn't yet in this process. This + * also initializes signal_info_lock; therefore it's important that this + * happen at least once before we fork off any threads. */ set_cancel_handler(); /* - * On Unix, we assume that storing a pointer value is atomic with respect - * to any possible signal interrupt. On Windows, use a critical section. + * Use mutex to prevent the cancel handler from using the pointer while + * we're changing it. */ - -#ifdef WIN32 - EnterCriticalSection(&signal_info_lock); -#endif + pthread_mutex_lock(&signal_info_lock); /* Free the old one if we have one */ - oldConnCancel = AH->connCancel; + oldCancelConn = AH->cancelConn; /* be sure interrupt handler doesn't use pointer while freeing */ - AH->connCancel = NULL; + AH->cancelConn = NULL; - if (oldConnCancel != NULL) - PQfreeCancel(oldConnCancel); + if (oldCancelConn != NULL) + PQcancelFinish(oldCancelConn); /* Set the new one if specified */ if (conn) - AH->connCancel = PQgetCancel(conn); + AH->cancelConn = PQcancelCreate(conn); /* * On Unix, there's only ever one active ArchiveHandle per process, so we @@ -776,49 +780,35 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) signal_info.myAH = AH; #endif -#ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); -#endif + pthread_mutex_unlock(&signal_info_lock); } /* * set_cancel_pstate * * Set signal_info.pstate to point to the specified ParallelState, if any. - * We need this mainly to have an interlock against Windows signal thread. + * We need this mainly to have an interlock against the cancel handler thread. */ static void set_cancel_pstate(ParallelState *pstate) { -#ifdef WIN32 - EnterCriticalSection(&signal_info_lock); -#endif - + pthread_mutex_lock(&signal_info_lock); signal_info.pstate = pstate; - -#ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); -#endif + pthread_mutex_unlock(&signal_info_lock); } /* * set_cancel_slot_archive * * Set ParallelSlot's AH field to point to the specified archive, if any. - * We need this mainly to have an interlock against Windows signal thread. + * We need this mainly to have an interlock against the cancel handler thread. */ static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) { -#ifdef WIN32 - EnterCriticalSection(&signal_info_lock); -#endif - + pthread_mutex_lock(&signal_info_lock); slot->AH = AH; - -#ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); -#endif + pthread_mutex_unlock(&signal_info_lock); } @@ -933,7 +923,7 @@ ParallelBackupStart(ArchiveHandle *AH) /* * Temporarily disable query cancellation on the leader connection. This - * ensures that child processes won't inherit valid AH->connCancel + * ensures that child processes won't inherit valid AH->cancelConn * settings and thus won't try to issue cancels against the leader's * connection. No harm is done if we fail while it's disabled, because * the leader connection is idle at this point anyway. @@ -991,6 +981,17 @@ ParallelBackupStart(ArchiveHandle *AH) /* instruct signal handler that we're in a worker now */ signal_info.am_worker = true; + /* + * Reset cancel handler state so that the worker will set up its + * own cancel thread when it calls set_archive_cancel_info(). + * Threads don't survive fork(), so we can't use the leader's. + * Also close the inherited pipe fds. + */ + signal_info.handler_set = false; + close(cancel_pipe[0]); + close(cancel_pipe[1]); + cancel_pipe[0] = cancel_pipe[1] = -1; + /* close read end of Worker -> Leader */ closesocket(pipeWM[PIPE_READ]); /* close write end of Leader -> Worker */ @@ -1407,8 +1408,18 @@ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait) if (!msg) { - /* If do_wait is true, we must have detected EOF on some socket */ - if (do_wait) + /* + * If do_wait is true, we must have detected EOF on some socket. If + * it's due to a cancel request, that's expected, otherwise it's a + * problem. + */ + bool cancel_requested; + + pthread_mutex_lock(&signal_info_lock); + cancel_requested = signal_info.cancel_requested; + pthread_mutex_unlock(&signal_info_lock); + + if (do_wait && !cancel_requested) pg_fatal("a worker process died unexpectedly"); return false; } diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 4a63f7392ae..4f00c3f020b 100644 --- a/src/bin/pg_dump/pg_backup_archiver.c +++ b/src/bin/pg_dump/pg_backup_archiver.c @@ -5161,7 +5161,7 @@ CloneArchive(ArchiveHandle *AH) /* The clone will have its own connection, so disregard connection state */ clone->connection = NULL; - clone->connCancel = NULL; + clone->cancelConn = NULL; clone->currUser = NULL; clone->currSchema = NULL; clone->currTableAm = NULL; diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h index 325b53fc9bd..f7a260683db 100644 --- a/src/bin/pg_dump/pg_backup_archiver.h +++ b/src/bin/pg_dump/pg_backup_archiver.h @@ -288,8 +288,12 @@ struct _archiveHandle char *savedPassword; /* password for ropt->username, if known */ char *use_role; PGconn *connection; - /* If connCancel isn't NULL, SIGINT handler will send a cancel */ - PGcancel *volatile connCancel; + + /* + * If connCancel isn't NULL, SIGINT handler will trigger the cancel thread + * send a cancel. + */ + PGcancelConn *cancelConn; int connectToDB; /* Flag to indicate if direct DB connection is * required */ diff --git a/src/bin/pg_dump/pg_backup_db.c b/src/bin/pg_dump/pg_backup_db.c index 5c349279beb..0cc29a8aa70 100644 --- a/src/bin/pg_dump/pg_backup_db.c +++ b/src/bin/pg_dump/pg_backup_db.c @@ -84,7 +84,7 @@ ReconnectToServer(ArchiveHandle *AH, const char *dbname) /* * Note: we want to establish the new connection, and in particular update - * ArchiveHandle's connCancel, before closing old connection. Otherwise + * ArchiveHandle's cancelConn, before closing old connection. Otherwise * an ill-timed SIGINT could try to access a dead connection. */ AH->connection = NULL; /* dodge error check in ConnectDatabaseAhx */ @@ -164,12 +164,11 @@ void DisconnectDatabase(Archive *AHX) { ArchiveHandle *AH = (ArchiveHandle *) AHX; - char errbuf[1]; if (!AH->connection) return; - if (AH->connCancel) + if (AH->cancelConn) { /* * If we have an active query, send a cancel before closing, ignoring @@ -177,7 +176,7 @@ DisconnectDatabase(Archive *AHX) * helpful during pg_fatal(). */ if (PQtransactionStatus(AH->connection) == PQTRANS_ACTIVE) - (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); + (void) PQcancelBlocking(AH->cancelConn); /* * Prevent signal handler from sending a cancel after this. diff --git a/src/fe_utils/Makefile b/src/fe_utils/Makefile index b9fd566ff87..eee084e8ed0 100644 --- a/src/fe_utils/Makefile +++ b/src/fe_utils/Makefile @@ -17,7 +17,7 @@ subdir = src/fe_utils top_builddir = ../.. include $(top_builddir)/src/Makefile.global -override CPPFLAGS := -DFRONTEND -I$(libpq_srcdir) $(CPPFLAGS) +override CPPFLAGS := -DFRONTEND -I$(libpq_srcdir) -I$(top_srcdir)/src/port $(CPPFLAGS) OBJS = \ archive.o \ diff --git a/src/fe_utils/cancel.c b/src/fe_utils/cancel.c index f434718eac8..c1c5b7055a8 100644 --- a/src/fe_utils/cancel.c +++ b/src/fe_utils/cancel.c @@ -16,9 +16,17 @@ #include "postgres_fe.h" +#include <signal.h> #include <unistd.h> +#ifdef WIN32 +#include "pthread-win32.h" +#else +#include <pthread.h> +#endif + #include "common/connect.h" +#include "common/logging.h" #include "fe_utils/cancel.h" #include "fe_utils/string_utils.h" @@ -36,11 +44,22 @@ (void) rc_; \ } while (0) + +/* + * Cancel connection that should be used to send cancel requests. + */ +static PGcancelConn *cancelConn = NULL; + +/* + * Generation counter for cancelConn. Incremented each time cancelConn is + * changed. Used to detect if cancelConn was replaced while we were using it. + */ +static uint64 cancelConnGeneration = 0; + /* - * Contains all the information needed to cancel a query issued from - * a database connection to the backend. + * Mutex protecting cancelConn and cancelConnGeneration. */ -static PGcancel *volatile cancelConn = NULL; +static pthread_mutex_t cancelConnLock = PTHREAD_MUTEX_INITIALIZER; /* * Predetermined localized error strings --- needed to avoid trying @@ -58,186 +77,241 @@ static const char *cancel_not_sent_msg = NULL; */ volatile sig_atomic_t CancelRequested = false; -#ifdef WIN32 -static CRITICAL_SECTION cancelConnLock; -#endif - /* * Additional callback for cancellations. */ static void (*cancel_callback) (void) = NULL; +#ifndef WIN32 +/* + * On Unix, we use a self-pipe to wake up the cancel thread from the signal + * handler, since pthread_cond_signal() is not async-signal-safe. + */ +static int cancel_pipe[2] = {-1, -1}; +static pthread_t cancel_thread; +#endif + /* - * SetCancelConn - * - * Set cancelConn to point to the current database connection. + * Send a cancel request to the connection, if one is set. */ -void -SetCancelConn(PGconn *conn) +static void +SendCancelRequest(void) { - PGcancel *oldCancelConn; + PGcancelConn *cc; + uint64 generation; + bool putConnectionBack = false; + + /* + * We take the cancel connection out of the global. This ensures that + * ResetCancelConn or SetCancelConn won't free it while we're using it. + */ + pthread_mutex_lock(&cancelConnLock); + cc = cancelConn; + generation = cancelConnGeneration; + cancelConn = NULL; + pthread_mutex_unlock(&cancelConnLock); -#ifdef WIN32 - EnterCriticalSection(&cancelConnLock); -#endif + if (cc == NULL) + return; - /* Free the old one if we have one */ - oldCancelConn = cancelConn; + write_stderr(cancel_sent_msg); - /* be sure handle_sigint doesn't use pointer while freeing */ - cancelConn = NULL; + if (!PQcancelBlocking(cc)) + { + char *errmsg = PQcancelErrorMessage(cc); - if (oldCancelConn != NULL) - PQfreeCancel(oldCancelConn); + write_stderr(cancel_not_sent_msg); + if (errmsg) + write_stderr(errmsg); + } + /* Reset for possible reuse */ + PQcancelReset(cc); + + /* + * Put the cancel connection back if it wasn't replaced while we were + * using it. + */ + pthread_mutex_lock(&cancelConnLock); + if (cancelConnGeneration == generation) + { + /* Generation unchanged, put it back for reuse */ + cancelConn = cc; + putConnectionBack = true; + } + pthread_mutex_unlock(&cancelConnLock); - cancelConn = PQgetCancel(conn); + /* If it was replaced, we free it, because we were the last user */ + if (!putConnectionBack) + PQcancelFinish(cc); +} -#ifdef WIN32 - LeaveCriticalSection(&cancelConnLock); -#endif + +/* + * Helper to replace cancelConn with a new value. + */ +static void +SetCancelConnInternal(PGcancelConn *newCancelConn) +{ + PGcancelConn *oldCancelConn; + + pthread_mutex_lock(&cancelConnLock); + oldCancelConn = cancelConn; + cancelConn = newCancelConn; + cancelConnGeneration++; + pthread_mutex_unlock(&cancelConnLock); + + if (oldCancelConn != NULL) + PQcancelFinish(oldCancelConn); +} + +/* + * SetCancelConn + * + * Set cancelConn to point to a cancel connection for the given database + * connection. This creates a new PGcancelConn that can be used to send + * cancel requests. + */ +void +SetCancelConn(PGconn *conn) +{ + SetCancelConnInternal(PQcancelCreate(conn)); } /* * ResetCancelConn * - * Free the current cancel connection, if any, and set to NULL. + * Clear cancelConn, preventing any pending cancel from being sent. */ void ResetCancelConn(void) { - PGcancel *oldCancelConn; + SetCancelConnInternal(NULL); +} -#ifdef WIN32 - EnterCriticalSection(&cancelConnLock); -#endif - oldCancelConn = cancelConn; +#ifdef WIN32 +/* + * Console control handler for Windows. + * + * This runs in a separate thread created by the OS, so we can safely call + * the blocking cancel API directly. + */ +static BOOL WINAPI +consoleHandler(DWORD dwCtrlType) +{ + if (dwCtrlType == CTRL_C_EVENT || + dwCtrlType == CTRL_BREAK_EVENT) + { + CancelRequested = true; - /* be sure handle_sigint doesn't use pointer while freeing */ - cancelConn = NULL; + if (cancel_callback != NULL) + cancel_callback(); - if (oldCancelConn != NULL) - PQfreeCancel(oldCancelConn); + SendCancelRequest(); -#ifdef WIN32 - LeaveCriticalSection(&cancelConnLock); -#endif + return TRUE; + } + else + /* Return FALSE for any signals not being handled */ + return FALSE; } +#else /* !WIN32 */ /* - * Code to support query cancellation - * - * Note that sending the cancel directly from the signal handler is safe - * because PQcancel() is written to make it so. We use write() to report - * to stderr because it's better to use simple facilities in a signal - * handler. - * - * On Windows, the signal canceling happens on a separate thread, because - * that's how SetConsoleCtrlHandler works. The PQcancel function is safe - * for this (unlike PQrequestCancel). However, a CRITICAL_SECTION is required - * to protect the PGcancel structure against being changed while the signal - * thread is using it. + * Cancel thread main function. Waits for the signal handler to write to the + * pipe, then sends a cancel request. */ +static void * +cancel_thread_main(void *arg) +{ + for (;;) + { + char buf[16]; + ssize_t rc; -#ifndef WIN32 + /* Wait for signal handler to wake us up */ + rc = read(cancel_pipe[0], buf, sizeof(buf)); + if (rc <= 0) + { + if (errno == EINTR) + continue; + /* Pipe closed or error - exit thread */ + break; + } + + SendCancelRequest(); + } + + return NULL; +} /* - * handle_sigint - * - * Handle interrupt signals by canceling the current command, if cancelConn - * is set. + * Signal handler for SIGINT. Sets CancelRequested and wakes up the cancel + * thread by writing to the pipe. */ static void handle_sigint(SIGNAL_ARGS) { - char errbuf[256]; + int save_errno = errno; + char c = 1; CancelRequested = true; if (cancel_callback != NULL) cancel_callback(); - /* Send QueryCancel if we are processing a database query */ - if (cancelConn != NULL) - { - if (PQcancel(cancelConn, errbuf, sizeof(errbuf))) - { - write_stderr(cancel_sent_msg); - } - else - { - write_stderr(cancel_not_sent_msg); - write_stderr(errbuf); - } - } + /* Wake up the cancel thread - write() is async-signal-safe */ + if (cancel_pipe[1] >= 0) + (void) write(cancel_pipe[1], &c, 1); + + errno = save_errno; } +#endif /* WIN32 */ + + /* * setup_cancel_handler * - * Register query cancellation callback for SIGINT. + * Set up handler for SIGINT (Unix) or console events (Windows) to send a + * cancel request to the server. Also sets up the infrastructure to send + * cancel requests asynchronously. */ void setup_cancel_handler(void (*query_cancel_callback) (void)) { +#ifndef WIN32 + int rc; +#endif cancel_callback = query_cancel_callback; - cancel_sent_msg = _("Cancel request sent\n"); + cancel_sent_msg = _("Sending cancel request\n"); cancel_not_sent_msg = _("Could not send cancel request: "); - pqsignal(SIGINT, handle_sigint); -} - -#else /* WIN32 */ - -static BOOL WINAPI -consoleHandler(DWORD dwCtrlType) -{ - char errbuf[256]; +#ifdef WIN32 + SetConsoleCtrlHandler(consoleHandler, TRUE); +#else - if (dwCtrlType == CTRL_C_EVENT || - dwCtrlType == CTRL_BREAK_EVENT) + /* + * Set up the self-pipe for communication between signal handler and + * cancel thread. We use a pipe because write() is async-signal-safe. + */ + if (pipe(cancel_pipe) < 0) { - CancelRequested = true; - - if (cancel_callback != NULL) - cancel_callback(); - - /* Send QueryCancel if we are processing a database query */ - EnterCriticalSection(&cancelConnLock); - if (cancelConn != NULL) - { - if (PQcancel(cancelConn, errbuf, sizeof(errbuf))) - { - write_stderr(cancel_sent_msg); - } - else - { - write_stderr(cancel_not_sent_msg); - write_stderr(errbuf); - } - } - - LeaveCriticalSection(&cancelConnLock); - - return TRUE; + pg_log_error("could not create pipe for cancel: %m"); + exit(1); } - else - /* Return FALSE for any signals not being handled */ - return FALSE; -} -void -setup_cancel_handler(void (*callback) (void)) -{ - cancel_callback = callback; - cancel_sent_msg = _("Cancel request sent\n"); - cancel_not_sent_msg = _("Could not send cancel request: "); + /* Start the cancel thread if not already running */ - InitializeCriticalSection(&cancelConnLock); + rc = pthread_create(&cancel_thread, NULL, cancel_thread_main, NULL); + if (rc != 0) + { + pg_log_error("could not create cancel thread: %s", strerror(rc)); + exit(1); + } - SetConsoleCtrlHandler(consoleHandler, TRUE); + pqsignal(SIGINT, handle_sigint); +#endif } - -#endif /* WIN32 */ -- 2.52.0
