From 7be3c48c50aa3edb40073b04d805de3181ab6685 Mon Sep 17 00:00:00 2001 From: Thomas Munro Date: Wed, 1 Jul 2026 23:47:34 +1200 Subject: [PATCH v1 11/14] pg_dump: Use pg_threads.h and remove fork() path. Previously, POSIX systems forked worker processes and Windows systems used threads to implement parallel dump/restore. Now all systems use threads. On Windows, the ^C handler needs interlocking to make sure that objects it accesses aren't freed concurrently, while on Unix it runs in a signal handler which doesn't allow that possibility. To keep the code uniform, the mutex is acquired unconditionally in the main code path. A follow-up patch will refactor and simplify the SIGINT handling. Discussion: Reviewed-by: --- src/bin/pg_dump/parallel.c | 562 +++++++-------------------- src/bin/pg_dump/parallel.h | 7 +- src/bin/pg_dump/pg_backup_archiver.c | 31 +- src/bin/pg_dump/pg_backup_utils.c | 26 +- 4 files changed, 167 insertions(+), 459 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 81d0865284d..4a095dd08e6 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -16,11 +16,11 @@ /* * Parallel operation works like this: * - * The original, leader process calls ParallelBackupStart(), which forks off - * the desired number of worker processes, which each enter WaitForCommands(). + * The original, leader thread calls ParallelBackupStart(), which starts + * the desired number of worker threads, which each enter WaitForCommands(). * - * The leader process dispatches an individual work item to one of the worker - * processes in DispatchJobForTocEntry(). We send a command string such as + * The leader thread dispatches an individual work item to one of the worker + * threads in DispatchJobForTocEntry(). We send a command string such as * "DUMP 1234" or "RESTORE 1234", where 1234 is the TocEntry ID. * The worker process receives and decodes the command and passes it to the * routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr, @@ -34,15 +34,13 @@ * In principle additional archive-format-specific information might be needed * in commands or worker status responses, but so far that hasn't proved * necessary, since workers have full copies of the ArchiveHandle/TocEntry - * data structures. Remember that we have forked off the workers only after - * we have read in the catalog. That's why our worker processes can also - * access the catalog information. (In the Windows case, the workers are - * threads in the same process. To avoid problems, they work with cloned - * copies of the Archive data structure; see RunWorker().) + * data structures. Remember that we have started the workers only after we + * have read in the catalog. That's why our worker threads can also + * access the catalog information. * - * In the leader process, the workerStatus field for each worker has one of - * the following values: - * WRKR_NOT_STARTED: we've not yet forked this worker + * The workerStatus field for each worker is only accessed by the leader, and + * has one of the following values: + * WRKR_NOT_STARTED: we've not yet started this worker * WRKR_IDLE: it's waiting for a command * WRKR_WORKING: it's working on a command * WRKR_TERMINATED: process ended @@ -52,13 +50,8 @@ #include "postgres_fe.h" -#ifndef WIN32 -#include -#include -#include -#endif - #include +#include #include #include "fe_utils/string_utils.h" @@ -67,6 +60,7 @@ #ifdef WIN32 #include "port/pg_bswap.h" #endif +#include "port/pg_threads.h" /* Mnemonic macros for indexing the fd array returned by pipe(2) */ #define PIPE_READ 0 @@ -74,7 +68,7 @@ #define NO_SLOT (-1) /* Failure result for GetIdleWorker() */ -/* Worker process statuses */ +/* Worker thread statuses */ typedef enum { WRKR_NOT_STARTED = 0, @@ -89,9 +83,9 @@ typedef enum /* * Private per-parallel-worker state (typedef for this is in parallel.h). * - * Much of this is valid only in the leader process (or, on Windows, should - * be touched only by the leader thread). But the AH field should be touched - * only by workers. The pipe descriptors are valid everywhere. + * Much of this is valid only in the leader thread. But the AH field should + * be touched only by worker threads. The pipe descriptors are valid + * everywhere. */ struct ParallelSlot { @@ -108,27 +102,21 @@ struct ParallelSlot int pipeRevRead; /* child's end of the pipes */ int pipeRevWrite; - /* Child process/thread identity info: */ -#ifdef WIN32 - uintptr_t hThread; - unsigned int threadId; -#else - pid_t pid; -#endif + PQExpBuffer buffer; + pg_thrd_t thread; + int result; }; -#ifdef WIN32 -/* - * Structure to hold info passed by _beginthreadex() to the function it calls - * via its single allowed argument. - */ +/* Structure to hold info passed to new threads. */ typedef struct { ArchiveHandle *AH; /* leader database connection */ ParallelSlot *slot; /* this worker's parallel slot */ } WorkerInfo; +#ifdef WIN32 + /* Windows implementation of pipe access */ static int pgpipe(int handles[2]); #define piperead(a,b,c) recv(a,b,c,0) @@ -168,16 +156,11 @@ 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? */ -#ifndef WIN32 - bool am_worker; /* am I a worker process? */ -#endif } DumpSignalInformation; static volatile DumpSignalInformation signal_info; -#ifdef WIN32 -static CRITICAL_SECTION signal_info_lock; -#endif +static pg_mtx_t signal_info_lock = PG_MTX_INIT; /* * Write a simple string to stderr --- must be safe in a signal handler. @@ -192,18 +175,13 @@ static CRITICAL_SECTION signal_info_lock; (void) rc_; \ } while (0) +/* Pointer to each worker thread's ParallelSlot. NULL in main thread. */ +static thread_local ParallelSlot *parallel_slot_thread_local; -#ifdef WIN32 -/* file-scope variables */ -static DWORD tls_index; +static bool parallel_init_done = false; -/* globally visible variables (needed by exit_nicely) */ -bool parallel_init_done = false; -DWORD mainThreadId; -#endif /* WIN32 */ /* Local function prototypes */ -static ParallelSlot *GetMyPSlot(ParallelState *pstate); static void archive_close_connection(int code, void *arg); static void ShutdownWorkersHard(ParallelState *pstate); static void WaitForTerminatingWorkers(ParallelState *pstate); @@ -212,7 +190,6 @@ static void set_cancel_pstate(ParallelState *pstate); static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH); static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot); static int GetIdleWorker(ParallelState *pstate); -static bool HasEveryWorkerTerminated(ParallelState *pstate); static void lockTableForWorker(ArchiveHandle *AH, TocEntry *te); static void WaitForCommands(ArchiveHandle *AH, int pipefd[2]); static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, @@ -238,90 +215,46 @@ static char *readMessageFromPipe(int fd); void init_parallel_dump_utils(void) { -#ifdef WIN32 if (!parallel_init_done) { +#ifdef WIN32 WSADATA wsaData; int err; - /* Prepare for threaded operation */ - tls_index = TlsAlloc(); - mainThreadId = GetCurrentThreadId(); - /* Initialize socket access */ err = WSAStartup(MAKEWORD(2, 2), &wsaData); if (err != 0) pg_fatal("%s() failed: error code %d", "WSAStartup", err); +#endif parallel_init_done = true; } -#endif } /* - * Find the ParallelSlot for the current worker process or thread. - * - * Returns NULL if no matching slot is found (this implies we're the leader). + * Exported for use by exit_nicely(). */ -static ParallelSlot * -GetMyPSlot(ParallelState *pstate) +bool +am_parallel_worker_thread(void) { - int i; - - for (i = 0; i < pstate->numWorkers; i++) - { -#ifdef WIN32 - if (pstate->parallelSlot[i].threadId == GetCurrentThreadId()) -#else - if (pstate->parallelSlot[i].pid == getpid()) -#endif - return &(pstate->parallelSlot[i]); - } - - return NULL; + return parallel_init_done && parallel_slot_thread_local != NULL; } /* * A thread-local version of getLocalPQExpBuffer(). - * - * Non-reentrant but reduces memory leakage: we'll consume one buffer per - * thread, which is much better than one per fmtId/fmtQualifiedId call. */ -#ifdef WIN32 static PQExpBuffer getThreadLocalPQExpBuffer(void) { - /* - * The Tls code goes awry if we use a static var, so we provide for both - * static and auto, and omit any use of the static var when using Tls. We - * rely on TlsGetValue() to return 0 if the value is not yet set. - */ - static PQExpBuffer s_id_return = NULL; PQExpBuffer id_return; - if (parallel_init_done) - id_return = (PQExpBuffer) TlsGetValue(tls_index); - else - id_return = s_id_return; + /* Each worker has its own pre-allocated buffer to reuse repeatedly. */ + id_return = parallel_slot_thread_local->buffer; - if (id_return) /* first time through? */ - { - /* same buffer, just wipe contents */ - resetPQExpBuffer(id_return); - } - else - { - /* new buffer */ - id_return = createPQExpBuffer(); - if (parallel_init_done) - TlsSetValue(tls_index, id_return); - else - s_id_return = id_return; - } + resetPQExpBuffer(id_return); return id_return; } -#endif /* WIN32 */ /* * pg_dump and pg_restore call this to register the cleanup handler @@ -346,7 +279,7 @@ archive_close_connection(int code, void *arg) if (si->pstate) { /* In parallel mode, must figure out who we are */ - ParallelSlot *slot = GetMyPSlot(si->pstate); + ParallelSlot *slot = parallel_slot_thread_local; if (!slot) { @@ -362,9 +295,9 @@ archive_close_connection(int code, void *arg) else { /* - * We're a worker. Shut down our own DB connection if any. On - * Windows, we also have to close our communication sockets, to - * emulate what will happen on Unix when the worker process exits. + * We're a worker. Shut down our own DB connection if any. Also + * close our communication sockets. + * * (Without this, if this is a premature exit, the leader would * fail to detect it because there would be no EOF condition on * the other end of the pipe.) @@ -372,10 +305,8 @@ archive_close_connection(int code, void *arg) if (slot->AH) DisconnectDatabase(&(slot->AH->public)); -#ifdef WIN32 closesocket(slot->pipeRevRead); closesocket(slot->pipeRevWrite); -#endif } } else @@ -409,24 +340,11 @@ ShutdownWorkersHard(ParallelState *pstate) closesocket(pstate->parallelSlot[i].pipeWrite); /* - * Force early termination of any commands currently in progress. + * Force early termination of any commands currently in progress. Send + * query cancels directly to the workers' backends. Use a critical + * section to ensure worker threads don't change state. */ -#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); + pg_mtx_lock(&signal_info_lock); for (i = 0; i < pstate->numWorkers; i++) { ArchiveHandle *AH = pstate->parallelSlot[i].AH; @@ -435,8 +353,7 @@ ShutdownWorkersHard(ParallelState *pstate) if (AH != NULL && AH->connCancel != NULL) (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } - LeaveCriticalSection(&signal_info_lock); -#endif + pg_mtx_unlock(&signal_info_lock); /* Now wait for them to terminate. */ WaitForTerminatingWorkers(pstate); @@ -448,64 +365,17 @@ ShutdownWorkersHard(ParallelState *pstate) static void WaitForTerminatingWorkers(ParallelState *pstate) { - while (!HasEveryWorkerTerminated(pstate)) + for (int i = 0; i < pstate->numWorkers; ++i) { - ParallelSlot *slot = NULL; - int j; - -#ifndef WIN32 - /* On non-Windows, use wait() to wait for next worker to end */ - int status; - pid_t pid = wait(&status); + ParallelSlot *slot = &pstate->parallelSlot[i]; - /* Find dead worker's slot, and clear the PID field */ - for (j = 0; j < pstate->numWorkers; j++) + if (slot->workerStatus != WRKR_TERMINATED) { - slot = &(pstate->parallelSlot[j]); - if (slot->pid == pid) - { - slot->pid = 0; - break; - } - } -#else /* WIN32 */ - /* On Windows, we must use WaitForMultipleObjects() */ - HANDLE *lpHandles = pg_malloc_array(HANDLE, pstate->numWorkers); - int nrun = 0; - DWORD ret; - uintptr_t hThread; - - for (j = 0; j < pstate->numWorkers; j++) - { - if (WORKER_IS_RUNNING(pstate->parallelSlot[j].workerStatus)) - { - lpHandles[nrun] = (HANDLE) pstate->parallelSlot[j].hThread; - nrun++; - } - } - ret = WaitForMultipleObjects(nrun, lpHandles, false, INFINITE); - Assert(ret != WAIT_FAILED); - hThread = (uintptr_t) lpHandles[ret - WAIT_OBJECT_0]; - pg_free(lpHandles); - - /* Find dead worker's slot, and clear the hThread field */ - for (j = 0; j < pstate->numWorkers; j++) - { - slot = &(pstate->parallelSlot[j]); - if (slot->hThread == hThread) - { - /* For cleanliness, close handles for dead threads */ - CloseHandle((HANDLE) slot->hThread); - slot->hThread = (uintptr_t) INVALID_HANDLE_VALUE; - break; - } + if (pg_thrd_join(slot->thread, &slot->result) != pg_thrd_success) + pg_fatal("could not join thread %d", i); + slot->workerStatus = WRKR_TERMINATED; + pstate->te[i] = NULL; } -#endif /* WIN32 */ - - /* On all platforms, update workerStatus and te[] as well */ - Assert(j < pstate->numWorkers); - slot->workerStatus = WRKR_TERMINATED; - pstate->te[j] = NULL; } } @@ -522,72 +392,78 @@ WaitForTerminatingWorkers(ParallelState *pstate) * 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. - * - * 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 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. */ -#ifndef WIN32 - /* - * Signal handler (Unix only) + * Unix: runs in a signal handler (async-signal-safe operations only). + * + * Windows: runs in a system-provided thread and signal_info_lock has been + * acquired by the caller. + * + * The caller will exit after this function returns. */ static void -sigTermHandler(int signo) +handle_async_cancellation(void) { - int i; char errbuf[1]; + int null_device; /* - * 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. + * Report we're quitting, using nothing more complicated than write(2). */ - if (signal_info.pstate != NULL) + if (progname) { - for (i = 0; i < signal_info.pstate->numWorkers; i++) - { - pid_t pid = signal_info.pstate->parallelSlot[i].pid; - - if (pid != 0) - kill(pid, SIGTERM); - } + write_stderr(progname); + write_stderr(": "); } + write_stderr("terminated by user\n"); /* - * Send QueryCancel if we have a connection to send to. Ignore errors, - * there's not much we can do about them anyway. + * Atomically replace STDERR_FILENO with the null device, to swallow + * future write() and fwrite() output to that descriptor. This prevents + * worker threads from reporting cancellation-related errors, which might + * clutter the user's screen if they manage arrive before the whole + * process exits and terminates them. Ignore errors. */ - if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL) - (void) PQcancel(signal_info.myAH->connCancel, errbuf, sizeof(errbuf)); + if ((null_device = open(DEVNULL, O_WRONLY, 0)) >= 0) + dup2(null_device, STDERR_FILENO); /* - * Report we're quitting, using nothing more complicated than write(2). - * When in parallel operation, only the leader process should do this. + * If in parallel mode, send QueryCancel to worker threads' connected + * backends. 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.am_worker) + if (signal_info.pstate != NULL) { - if (progname) + for (int i = 0; i < signal_info.pstate->numWorkers; i++) { - write_stderr(progname); - write_stderr(": "); + ParallelSlot *slot = &(signal_info.pstate->parallelSlot[i]); + ArchiveHandle *AH = slot->AH; + + if (AH != NULL && AH->connCancel != NULL) + PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } - write_stderr("terminated by user\n"); } + /* + * 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) + PQcancel(signal_info.myAH->connCancel, errbuf, sizeof(errbuf)); +} + +#ifndef WIN32 + +/* + * Signal handler (Unix only) + */ +static void +sigTermHandler(int signo) +{ + handle_async_cancellation(); + /* * And die, using _exit() not exit() because the latter will invoke atexit * handlers that can fail if we interrupted related code. @@ -640,66 +516,12 @@ set_cancel_handler(void) static BOOL WINAPI consoleHandler(DWORD dwCtrlType) { - int i; - char errbuf[1]; - if (dwCtrlType == CTRL_C_EVENT || dwCtrlType == CTRL_BREAK_EVENT) { - int null_device; - - /* - * Report we're quitting, using nothing more complicated than - * write(2). - */ - if (progname) - { - write_stderr(progname); - write_stderr(": "); - } - write_stderr("terminated by user\n"); - - /* - * Atomically replace STDERR_FILENO with the null device, to swallow - * future write() and fwrite() output to that descriptor. This - * prevents worker threads from reporting cancellation-related errors, - * which might clutter the user's screen if they manage arrive before - * the whole process exits and terminates them. Ignore errors. - */ - if ((null_device = open(DEVNULL, O_WRONLY | O_BINARY, 0)) >= 0) - dup2(null_device, STDERR_FILENO); - - /* 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. 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; - - if (AH != NULL && AH->connCancel != NULL) - (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); - } - } - - /* - * 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); + pg_mtx_lock(&signal_info_lock); + handle_async_cancellation(); + pg_mtx_unlock(&signal_info_lock); } /* Always return FALSE to allow signal handling to continue */ @@ -716,8 +538,6 @@ set_cancel_handler(void) { signal_info.handler_set = true; - InitializeCriticalSection(&signal_info_lock); - SetConsoleCtrlHandler(consoleHandler, TRUE); } } @@ -746,12 +566,10 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) /* * On Unix, we assume that storing a pointer value is atomic with respect - * to any possible signal interrupt. On Windows, use a critical section. + * to any possible signal interrupt. On Windows, the signal acquires this + * lock. */ - -#ifdef WIN32 - EnterCriticalSection(&signal_info_lock); -#endif + pg_mtx_lock(&signal_info_lock); /* Free the old one if we have one */ oldConnCancel = AH->connCancel; @@ -766,22 +584,13 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) AH->connCancel = PQgetCancel(conn); /* - * On Unix, there's only ever one active ArchiveHandle per process, so we - * can just set signal_info.myAH unconditionally. On Windows, do that - * only in the main thread; worker threads have to make sure their - * ArchiveHandle appears in the pstate data, which is dealt with in - * RunWorker(). + * Set the main thread's myAH, unless we're in a worker. Workers set + * their per-slot AH in RunWorker(). */ -#ifndef WIN32 - signal_info.myAH = AH; -#else - if (mainThreadId == GetCurrentThreadId()) + if (parallel_slot_thread_local == NULL) signal_info.myAH = AH; -#endif -#ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); -#endif + pg_mtx_unlock(&signal_info_lock); } /* @@ -793,15 +602,9 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) static void set_cancel_pstate(ParallelState *pstate) { -#ifdef WIN32 - EnterCriticalSection(&signal_info_lock); -#endif - + pg_mtx_lock(&signal_info_lock); signal_info.pstate = pstate; - -#ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); -#endif + pg_mtx_unlock(&signal_info_lock); } /* @@ -813,22 +616,14 @@ set_cancel_pstate(ParallelState *pstate) static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) { -#ifdef WIN32 - EnterCriticalSection(&signal_info_lock); -#endif - + pg_mtx_lock(&signal_info_lock); slot->AH = AH; - -#ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); -#endif + pg_mtx_unlock(&signal_info_lock); } - /* - * This function is called by both Unix and Windows variants to set up - * and run a worker process. Caller should exit the process (or thread) - * upon return. + * This function is called to set up and run a worker process. Caller should + * exit the thread upon return. */ static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot) @@ -842,11 +637,6 @@ RunWorker(ArchiveHandle *AH, ParallelSlot *slot) /* * Clone the archive so that we have our own state to work with, and in * particular our own database connection. - * - * We clone on Unix as well as Windows, even though technically we don't - * need to because fork() gives us a copy in our own address space - * already. But CloneArchive resets the state information and also clones - * the database connection which both seem kinda helpful. */ AH = CloneArchive(AH); @@ -871,38 +661,34 @@ RunWorker(ArchiveHandle *AH, ParallelSlot *slot) DeCloneArchive(AH); } -/* - * Thread base function for Windows - */ -#ifdef WIN32 -static unsigned __stdcall -init_spawned_worker_win32(WorkerInfo *wi) +static int +worker_thread_main(void *argument) { + WorkerInfo *wi = argument; ArchiveHandle *AH = wi->AH; ParallelSlot *slot = wi->slot; /* Don't need WorkerInfo anymore */ free(wi); + /* For getThreadLocalPQExpBuffer() and am_parallel_worker_thread(). */ + parallel_slot_thread_local = slot; + /* Run the worker ... */ RunWorker(AH, slot); /* Exit the thread */ - _endthreadex(0); return 0; } -#endif /* WIN32 */ /* * This function starts a parallel dump or restore by spawning off the worker - * processes. For Windows, it creates a number of threads; on Unix the - * workers are created with fork(). + * threads. */ ParallelState * ParallelBackupStart(ArchiveHandle *AH) { ParallelState *pstate; - int i; Assert(AH->public.numWorkers > 0); @@ -921,10 +707,13 @@ ParallelBackupStart(ArchiveHandle *AH) pstate->parallelSlot = pg_malloc0_array(ParallelSlot, pstate->numWorkers); -#ifdef WIN32 - /* Make fmtId() and fmtQualifiedId() use thread-local storage */ + /* Make per-thread PQExpBuffer object for getThreadLocalPQExpBuffer(). */ + for (int i = 0; i < pstate->numWorkers; ++i) + if (!(pstate->parallelSlot[i].buffer = createPQExpBuffer())) + pg_fatal("could not allocate PQExpBuffer"); + + /* Make fmtId() and fmtQualifiedId() use per-thread buffer. */ getLocalPQExpBuffer = getThreadLocalPQExpBuffer; -#endif /* * Set the pstate in shutdown_info, to tell the exit handler that it must @@ -947,15 +736,10 @@ ParallelBackupStart(ArchiveHandle *AH) fflush(NULL); /* Create desired number of workers */ - for (i = 0; i < pstate->numWorkers; i++) + for (int i = 0; i < pstate->numWorkers; i++) { -#ifdef WIN32 WorkerInfo *wi; - uintptr_t handle; -#else - pid_t pid; -#endif - ParallelSlot *slot = &(pstate->parallelSlot[i]); + ParallelSlot *slot = &pstate->parallelSlot[i]; int pipeMW[2], pipeWM[2]; @@ -970,72 +754,23 @@ ParallelBackupStart(ArchiveHandle *AH) slot->pipeRevRead = pipeMW[PIPE_READ]; slot->pipeRevWrite = pipeWM[PIPE_WRITE]; -#ifdef WIN32 /* Create transient structure to pass args to worker function */ wi = pg_malloc_object(WorkerInfo); wi->AH = AH; wi->slot = slot; - handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32, - wi, 0, &(slot->threadId)); - slot->hThread = handle; - slot->workerStatus = WRKR_IDLE; -#else /* !WIN32 */ - pid = fork(); - if (pid == 0) - { - /* we are the worker */ - int j; - - /* this is needed for GetMyPSlot() */ - slot->pid = getpid(); - - /* instruct signal handler that we're in a worker now */ - signal_info.am_worker = true; + if (pg_thrd_create(&slot->thread, + worker_thread_main, + wi) != pg_thrd_success) + pg_fatal("could not start worker thread"); - /* close read end of Worker -> Leader */ - closesocket(pipeWM[PIPE_READ]); - /* close write end of Leader -> Worker */ - closesocket(pipeMW[PIPE_WRITE]); - - /* - * Close all inherited fds for communication of the leader with - * previously-forked workers. - */ - for (j = 0; j < i; j++) - { - closesocket(pstate->parallelSlot[j].pipeRead); - closesocket(pstate->parallelSlot[j].pipeWrite); - } - - /* Run the worker ... */ - RunWorker(AH, slot); - - /* We can just exit(0) when done */ - exit(0); - } - else if (pid < 0) - { - /* fork failed */ - pg_fatal("could not create worker process: %m"); - } - - /* In Leader after successful fork */ - slot->pid = pid; slot->workerStatus = WRKR_IDLE; - - /* close read end of Leader -> Worker */ - closesocket(pipeMW[PIPE_READ]); - /* close write end of Worker -> Leader */ - closesocket(pipeWM[PIPE_WRITE]); -#endif /* WIN32 */ } /* - * Having forked off the workers, disable SIGPIPE so that leader isn't - * killed if it tries to send a command to a dead worker. We don't want - * the workers to inherit this setting, though. + * Having started the workers, disable SIGPIPE so that the process isn't + * kill if the leader tries to send a command to a dead worker. */ #ifndef WIN32 pqsignal(SIGPIPE, PG_SIG_IGN); @@ -1063,8 +798,6 @@ ParallelBackupStart(ArchiveHandle *AH) void ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) { - int i; - /* No work if non-parallel */ if (pstate->numWorkers == 1) return; @@ -1073,7 +806,7 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) Assert(IsEveryWorkerIdle(pstate)); /* Close the sockets so that the workers know they can exit */ - for (i = 0; i < pstate->numWorkers; i++) + for (int i = 0; i < pstate->numWorkers; i++) { closesocket(pstate->parallelSlot[i].pipeRead); closesocket(pstate->parallelSlot[i].pipeWrite); @@ -1090,6 +823,9 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) set_cancel_pstate(NULL); /* Release state (mere neatnik-ism, since we're about to terminate) */ + for (int i = 0; i < pstate->numWorkers; ++i) + if (pstate->parallelSlot[i].buffer) + destroyPQExpBuffer(pstate->parallelSlot[i].buffer); free(pstate->te); free(pstate->parallelSlot); free(pstate); @@ -1250,22 +986,6 @@ GetIdleWorker(ParallelState *pstate) return NO_SLOT; } -/* - * Return true iff no worker is running. - */ -static bool -HasEveryWorkerTerminated(ParallelState *pstate) -{ - int i; - - for (i = 0; i < pstate->numWorkers; i++) - { - if (WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) - return false; - } - return true; -} - /* * Return true iff every worker is in the WRKR_IDLE state. */ diff --git a/src/bin/pg_dump/parallel.h b/src/bin/pg_dump/parallel.h index f7557cd089c..06764eefcac 100644 --- a/src/bin/pg_dump/parallel.h +++ b/src/bin/pg_dump/parallel.h @@ -19,6 +19,7 @@ #include #include "pg_backup_archiver.h" +#include "port/pg_threads.h" /* Function to call in leader process on completion of a worker task */ typedef void (*ParallelCompletionPtr) (ArchiveHandle *AH, @@ -60,12 +61,8 @@ typedef struct ParallelState ParallelSlot *parallelSlot; /* private info about each worker */ } ParallelState; -#ifdef WIN32 -extern bool parallel_init_done; -extern DWORD mainThreadId; -#endif - extern void init_parallel_dump_utils(void); +extern bool am_parallel_worker_thread(void); extern bool IsEveryWorkerIdle(ParallelState *pstate); extern void WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 0557eb6d6ed..a0e2b6fe916 100644 --- a/src/bin/pg_dump/pg_backup_archiver.c +++ b/src/bin/pg_dump/pg_backup_archiver.c @@ -92,12 +92,12 @@ static CompressFileHandle *SaveOutput(ArchiveHandle *AH); static void RestoreOutput(ArchiveHandle *AH, CompressFileHandle *savedOutput); static int restore_toc_entry(ArchiveHandle *AH, TocEntry *te, bool is_parallel); -static void restore_toc_entries_prefork(ArchiveHandle *AH, - TocEntry *pending_list); +static void restore_toc_entries_prologue(ArchiveHandle *AH, + TocEntry *pending_list); static void restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate, TocEntry *pending_list); -static void restore_toc_entries_postfork(ArchiveHandle *AH, +static void restore_toc_entries_epilogue(ArchiveHandle *AH, TocEntry *pending_list); static void pending_list_header_init(TocEntry *l); static void pending_list_append(TocEntry *l, TocEntry *te); @@ -735,16 +735,16 @@ RestoreArchive(Archive *AHX) pending_list_header_init(&pending_list); /* This runs PRE_DATA items and then disconnects from the database */ - restore_toc_entries_prefork(AH, &pending_list); + restore_toc_entries_prologue(AH, &pending_list); Assert(AH->connection == NULL); - /* ParallelBackupStart() will actually fork the processes */ + /* ParallelBackupStart() will actually run the threads */ pstate = ParallelBackupStart(AH); restore_toc_entries_parallel(AH, pstate, &pending_list); ParallelBackupEnd(AH, pstate); /* reconnect the leader and see if we missed something */ - restore_toc_entries_postfork(AH, &pending_list); + restore_toc_entries_epilogue(AH, &pending_list); Assert(AH->connection != NULL); } else @@ -4355,12 +4355,12 @@ dumpTimestamp(ArchiveHandle *AH, const char *msg, time_t tim) * added to the pending_list for later phases to deal with. */ static void -restore_toc_entries_prefork(ArchiveHandle *AH, TocEntry *pending_list) +restore_toc_entries_prologue(ArchiveHandle *AH, TocEntry *pending_list) { bool skipped_some; TocEntry *next_work_item; - pg_log_debug("entering restore_toc_entries_prefork"); + pg_log_debug("entering restore_toc_entries_prologue"); /* Adjust dependency information */ fix_dependencies(AH); @@ -4469,12 +4469,11 @@ restore_toc_entries_prefork(ArchiveHandle *AH, TocEntry *pending_list) /* * Main engine for parallel restore. * - * Parallel restore is done in three phases. In this second phase, - * we process entries by dispatching them to parallel worker children - * (processes on Unix, threads on Windows), each of which connects - * separately to the database. Inter-entry dependencies are respected, - * and so is the RestorePass multi-pass structure. When we can no longer - * make any entries ready to process, we exit. Normally, there will be + * Parallel restore is done in three phases. In this second phase, we process + * entries by dispatching them to parallel worker threads, each of which + * connects separately to the database. Inter-entry dependencies are + * respected, and so is the RestorePass multi-pass structure. When we can no + * longer make any entries ready to process, we exit. Normally, there will be * nothing left to do; but if there is, the third phase will mop up. */ static void @@ -4596,12 +4595,12 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate, * at least some chance of completing the restore successfully. */ static void -restore_toc_entries_postfork(ArchiveHandle *AH, TocEntry *pending_list) +restore_toc_entries_epilogue(ArchiveHandle *AH, TocEntry *pending_list) { RestoreOptions *ropt = AH->public.ropt; TocEntry *te; - pg_log_debug("entering restore_toc_entries_postfork"); + pg_log_debug("entering restore_toc_entries_epilogue"); /* * Now reconnect the single parent connection. diff --git a/src/bin/pg_dump/pg_backup_utils.c b/src/bin/pg_dump/pg_backup_utils.c index 0368f7623a7..b545a4a5439 100644 --- a/src/bin/pg_dump/pg_backup_utils.c +++ b/src/bin/pg_dump/pg_backup_utils.c @@ -13,10 +13,9 @@ */ #include "postgres_fe.h" -#ifdef WIN32 #include "parallel.h" -#endif #include "pg_backup_utils.h" +#include "port/pg_threads.h" /* Globals exported by this file */ const char *progname = NULL; @@ -75,18 +74,13 @@ on_exit_nicely(on_exit_nicely_callback function, void *arg) * Run accumulated on_exit_nicely callbacks in reverse order and then exit * without printing any message. * - * If running in a parallel worker thread on Windows, we only exit the thread, - * not the whole process. + * If running in a parallel worker thread, we only exit the thread, not the + * whole process. * - * Note that in parallel operation on Windows, the callback(s) will be run - * by each thread since the list state is necessarily shared by all threads; - * each callback must contain logic to ensure it does only what's appropriate - * for its thread. On Unix, callbacks are also run by each process, but only - * for callbacks established before we fork off the child processes. (It'd - * be cleaner to reset the list after fork(), and let each child establish - * its own callbacks; but then the behavior would be completely inconsistent - * between Windows and Unix. For now, just be sure to establish callbacks - * before forking to avoid inconsistency.) + * Note that in parallel operation, the callback(s) will be run by each thread + * since the list state is necessarily shared by all threads; each callback + * must contain logic to ensure it does only what's appropriate for its + * thread. */ void exit_nicely(int code) @@ -97,10 +91,8 @@ exit_nicely(int code) on_exit_nicely_list[i].function(code, on_exit_nicely_list[i].arg); -#ifdef WIN32 - if (parallel_init_done && GetCurrentThreadId() != mainThreadId) - _endthreadex(code); -#endif + if (am_parallel_worker_thread()) + pg_thrd_exit(code); exit(code); } -- 2.47.3