public inbox for [email protected]help / color / mirror / Atom feed
Re: WIP: libpq: add a possibility to not send D(escribe) when executing a prepared statement 2+ messages / 2 participants [nested] [flat]
* Re: WIP: libpq: add a possibility to not send D(escribe) when executing a prepared statement @ 2023-11-28 10:18 Ivan Trofimov <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Ivan Trofimov @ 2023-11-28 10:18 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: Andrey M. Borodin <[email protected]>; [email protected] Hi Tom! Thank you for considering this. > It adds a whole new set of programmer-error possibilities, and I doubt > it saves enough in typical cases to justify creating such a foot-gun. Although I agree it adds a considerable amount of complexity, I'd argue it doesn't bring the complexity to a new level, since matching queries against responses is a concept users of asynchronous processing are already familiar with, especially so when pipelining is in play. In case of a single-row select this can easily save as much as a half of the network traffic, which is likely to be encrypted/decrypted through multiple hops (a connection-pooler, for example), and has to be serialized/parsed on a server, a client, a pooler etc. For example, i have a service which bombards its Postgres database with ~20kRPS of "SELECT * FROM users WHERE id=$1", with "users" being a table of just a bunch of textual ids, a couple of timestamps and some enums in it, and for that service alone this change would save ~10Megabytes of server-originated traffic per second, and i have hundreds of such services at my workplace. I can provide more elaborate network/CPU measurements of different workloads if needed. > Instead, I'm tempted to suggest having PQprepare/PQexecPrepared > maintain a cache that maps statement name to result tupdesc, so that > this is all handled internally to libpq From a perspective of someone who maintains a library built on top of libpq and is familiar with other such libraries, I think this is much easier done on the level above libpq, simply because there is more control of when and how invalidation/eviction is done, and the level above also has a more straightforward way to access the cache across different asynchronous processing points. > I just think that successful use of that option requires a client- > side coding structure that allows tying a previously-obtained > tuple descriptor to the current query with confidence. The proposed > API fails badly at that, or at least leaves it up to the end-user > programmer while providing no tools to help her get it right I understand your concerns of usability/safety of what I propose, and I think I have an idea of how to make this much less of a foot-gun: what if we add a new function PGresult * PQexecPreparedPredescribed(PGconn *conn, const char *stmtName, PGresult* description, ...); which requires both a prepared statement and its tuple descriptor (or these two could even be tied together by a struct), and exposes its implementation (basically what I've prototyped in the patch) in the libpq-int.h? This way users of synchronous API get a nice thing too, which is arguably pretty hard to misuse: if the description isn't available upfront then there's no point to reach for the function added since PQexecPrepared is strictly better performance/usability-wise, and if the description is available it's most likely cached alongside the statement. If a user still manages to provide an erroneous description, well, they either get a parsing error or the erroneous description back, I don't see how libpq could misbehave badly here. Exposure of the implementation in the internal includes gives a possibility for users to juggle the actual foot-gun, but implies they know very well what they are doing, and are ready to be on their own. What do you think of such approach? ^ permalink raw reply [nested|flat] 2+ messages in thread
* Re: pg_dump: use threads for parallel workers on all platforms @ 2026-07-09 22:07 Bryan Green <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Bryan Green @ 2026-07-09 22:07 UTC (permalink / raw) To: Heikki Linnakangas <[email protected]>; PostgreSQL Hackers <[email protected]>; +Cc: Thomas Munro <[email protected]>; Andrew Dunstan <[email protected]> On 7/5/2026 6:21 PM, Heikki Linnakangas wrote: > On 02/07/2026 19:30, Bryan Green wrote: >> None of this is broken; it works. It's threads pretending to be >> processes because the code was written for processes, and the port kept >> the protocol rather than rethinking it. I'd like to stop. >> >> One model everywhere should be threads on all platforms, coordinated by >> an in-process work queue-- a mutex and a couple of condition variables-- >> instead of two worker models bridged by an inter-process protocol. > > +1 > >> To be clear, the unification is on the queue, not on what Windows does >> today. Teaching the non-Windows side to talk to its own threads over a >> socket, a byte at a time, would just be the same trick on more >> platforms-- that's the part worth deleting, not copying. >> >> I've done the Windows half, both to prove it out and because it's the >> coordination layer the non-Windows side would adopt. ... > > I see why you developed this that way, and it makes a lot of sense. > However, it has one downside: the Windows-only code cannot be tested > without Windows. At quick glance, it looks reasonable, but I'm a little > nervous committing more Windows-only code without being able to easily > play with it myself. > > Once you have the final patch ready to switch non-Windows systems to > threaded model too, that gets easier. > I have reworked the patches: 0001-0002 are standalone for Windows. 0003-0005 are dependent on Thomas Munro's pthread.h. >> With the thread rework, fmtId's static return value, is now >> _Thread_local. > +1. This is the first _Thread_local in our codebase. It's in C11, so it > should just work, but we'll see if the buildfarm shows any surprises... > > With this, the getLocalPQExpBuffer hook is never set. I think we can > just remove it, and rename defaultGetLocalPQExpBuffer() to > getLocalPQExpBuffer() directly. > Agreed. > I noticed that we currently call setFmtEncoding() in multiple places in > src/bin/pg_dump. I haven't looked at them closely, but I wonder if > there's some kind of thread-safety hazards there even without these > patches. > There is fmtIdEncoding which is a plain static, and on Windows each dump worker already calls setup_connection() -> setFmtEncoding() on its own thread. Every writer stores the same value (the encoding is fixed for the whole archive), so it's a same-value race. On the restore side the leader sets it once (processEncodingEntry) and the workers only read it. > And what about the 'quote_all_identifiers' global variable? Is that set > correctly in both models? I guess it's inherited through fork(). > getopt_long writes it during option parsing (--quote-all-identifiers points straight at the global), before any worker or connection exists, and nothing touches it again. > - Heikki > I expect changes to be needed. Thomas, I hope this is more of a help than not. 0001-0002 stand alone and touch only Windows: fmtId()'s scratch buffer becomes _Thread_local, and the Windows workers exchange work over an in-process channel rather than a loopback socket driven by select(). They're useful on their own and don't depend on the rest. 0003-0005 do the unification on top of Thomas Munro's pg_threads.h, so you will need his patch [v1-0002-port-Provide-minimal-pg_threads.h-API.patch] to install pg_threads API: 0003 ports the channel's locks to that API (mechanical, no behavior change); 0004 switches all platforms to threads and deletes the fork(), pgpipe() and select() code; 0005 drops the string command protocol for plain structs, now that the leader and workers share an address space. Changes since v1: - v1's _beginthreadex-failure check was committed separately as 75e201bf95, so it's dropped here. - Dropped the standalone 64-worker-cap patch; 0004 removes that limit for free by joining workers individually. - Extended the series past Windows: 0003-0005 unify all platforms on threads atop pg_threads.h. - Per review, fmtId()'s getLocalPQExpBuffer indirection is removed in 0001 now that nothing overrides it. -- Bryan Green EDB: https://www.enterprisedb.com From 653b37f2eb3adc429c8365c922a76d3979bde10f Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Sun, 28 Jun 2026 11:40:58 -0500 Subject: [PATCH v2 1/5] Give fmtId()'s temporary buffer thread-local storage getLocalPQExpBuffer() returns a function-scope static buffer, unsafe when fmtId() and fmtQualifiedId() run in multiple threads. On Windows, where pg_dump's parallel workers are threads, this was patched over at runtime by swapping in a TLS-based buffer from ParallelBackupStart(). Mark the default buffer C11 _Thread_local instead. Each thread gets its own, so the Windows TlsAlloc/TlsGetValue workaround goes. With no caller left to override it the getLocalPQExpBuffer indirection is pointless too: drop the function pointer, fold the default into a plain static getLocalPQExpBuffer(), and remove its now-obsolete extern. --- src/bin/pg_dump/parallel.c | 52 ----------------------------- src/fe_utils/string_utils.c | 13 +++----- src/include/fe_utils/string_utils.h | 1 - 3 files changed, 5 insertions(+), 61 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 4a0d04b646..b30e7db5a6 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -193,9 +193,6 @@ static CRITICAL_SECTION signal_info_lock; #ifdef WIN32 -/* file-scope variables */ -static DWORD tls_index; - /* globally visible variables (needed by exit_nicely) */ bool parallel_init_done = false; DWORD mainThreadId; @@ -243,8 +240,6 @@ init_parallel_dump_utils(void) WSADATA wsaData; int err; - /* Prepare for threaded operation */ - tls_index = TlsAlloc(); mainThreadId = GetCurrentThreadId(); /* Initialize socket access */ @@ -280,48 +275,6 @@ GetMyPSlot(ParallelState *pstate) return 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; - - 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; - } - - return id_return; -} -#endif /* WIN32 */ - /* * pg_dump and pg_restore call this to register the cleanup handler * as soon as they've created the ArchiveHandle. @@ -914,11 +867,6 @@ ParallelBackupStart(ArchiveHandle *AH) pstate->parallelSlot = pg_malloc0_array(ParallelSlot, pstate->numWorkers); -#ifdef WIN32 - /* Make fmtId() and fmtQualifiedId() use thread-local storage */ - getLocalPQExpBuffer = getThreadLocalPQExpBuffer; -#endif - /* * Set the pstate in shutdown_info, to tell the exit handler that it must * clean up workers as well as the main database connection. But we don't diff --git a/src/fe_utils/string_utils.c b/src/fe_utils/string_utils.c index 7a762251f3..f05bbf89bd 100644 --- a/src/fe_utils/string_utils.c +++ b/src/fe_utils/string_utils.c @@ -21,11 +21,8 @@ #include "fe_utils/string_utils.h" #include "mb/pg_wchar.h" -static PQExpBuffer defaultGetLocalPQExpBuffer(void); - /* Globals exported by this file */ int quote_all_identifiers = 0; -PQExpBuffer (*getLocalPQExpBuffer) (void) = defaultGetLocalPQExpBuffer; static int fmtIdEncoding = -1; @@ -34,14 +31,14 @@ static int fmtIdEncoding = -1; * Returns a temporary PQExpBuffer, valid until the next call to the function. * This is used by fmtId and fmtQualifiedId. * - * Non-reentrant and non-thread-safe but reduces memory leakage. You can - * replace this with a custom version by setting the getLocalPQExpBuffer - * function pointer. + * The buffer is thread-local, so this is safe to call from multiple threads + * at once; each thread gets its own. It is not reentrant within a thread, + * but it reduces memory leakage. */ static PQExpBuffer -defaultGetLocalPQExpBuffer(void) +getLocalPQExpBuffer(void) { - static PQExpBuffer id_return = NULL; + static _Thread_local PQExpBuffer id_return = NULL; if (id_return) /* first time through? */ { diff --git a/src/include/fe_utils/string_utils.h b/src/include/fe_utils/string_utils.h index 0680bb3c19..48e3b69afc 100644 --- a/src/include/fe_utils/string_utils.h +++ b/src/include/fe_utils/string_utils.h @@ -21,7 +21,6 @@ /* Global variables controlling behavior of fmtId() and fmtQualifiedId() */ extern PGDLLIMPORT int quote_all_identifiers; -extern PQExpBuffer (*getLocalPQExpBuffer) (void); /* Functions */ extern const char *fmtId(const char *rawid); -- 2.54.0.windows.1 From a405f85281261a273da5258b1397dd6b62cbc53b Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Sun, 28 Jun 2026 12:05:10 -0500 Subject: [PATCH v2 2/5] pg_dump: dispatch parallel workers in-process on Windows On Windows the parallel workers are threads in the leader's process, but they reached the leader through the same transport as Unix worker processes: pgpipe() (a loopback TCP socket pair), select(), and a byte-at-a-time string protocol. That is pointless when the workers share the leader's address space. Give each ParallelSlot an in-process channel instead -- a command slot and a response slot under a critical section, with one condition variable to wake the worker and one to wake the leader. Messages pass as malloc'd strings; dispatch writes to memory rather than a socket. The pipe also signalled worker death through EOF. The channel has none, and exit_nicely() ends only the worker thread, so a failed worker -- for instance a connection that fails when -j exceeds max_connections -- would hang the leader forever. A dying worker now sets slot->workerDied and wakes the leader, and getMessageFromWorker() returns NULL just as the Unix path does on EOF. For the leader to name the dead worker, each worker records its own thread id on entry rather than trusting _beginthreadex()'s output argument, which can lag the thread's start. The Unix path is untouched; the now-unused pgpipe(), select_loop(), and readMessageFromPipe() stay until the worker model is unified on threads. --- src/bin/pg_dump/parallel.c | 195 +++++++++++++++++++++++++++++++------ 1 file changed, 167 insertions(+), 28 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index b30e7db5a6..b89d394910 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -107,6 +107,19 @@ struct ParallelSlot int pipeRevRead; /* child's end of the pipes */ int pipeRevWrite; +#ifdef WIN32 + + /* + * In-process channel used instead of the pipes when workers are threads. + * A message is a malloc'd string; ownership passes to the receiver. + * Protected by msg_lock. + */ + char *cmdMsg; /* command pending for the worker, or NULL */ + char *respMsg; /* response pending for the leader, or NULL */ + bool chanClosed; /* leader closed the command channel (EOF) */ + bool workerDied; /* worker exited without sending a response */ +#endif + /* Child process/thread identity info: */ #ifdef WIN32 uintptr_t hThread; @@ -176,6 +189,15 @@ static volatile DumpSignalInformation signal_info; #ifdef WIN32 static CRITICAL_SECTION signal_info_lock; + +/* + * Synchronization for the in-process channels (see struct ParallelSlot). + * msg_lock protects the per-slot cmdMsg/respMsg/chanClosed/workerDied fields; + * worker_cv wakes a worker, leader_cv wakes the leader. + */ +static CRITICAL_SECTION msg_lock; +static CONDITION_VARIABLE worker_cv; +static CONDITION_VARIABLE leader_cv; #endif /* @@ -210,11 +232,11 @@ 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 void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot); static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait); -static char *getMessageFromLeader(int pipefd[2]); -static void sendMessageToLeader(int pipefd[2], const char *str); +static char *getMessageFromLeader(ParallelSlot *slot); +static void sendMessageToLeader(ParallelSlot *slot, const char *str); static int select_loop(int maxFd, fd_set *workerset); static char *getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker); @@ -242,6 +264,11 @@ init_parallel_dump_utils(void) mainThreadId = GetCurrentThreadId(); + /* Initialize the in-process message-channel synchronization */ + InitializeCriticalSection(&msg_lock); + InitializeConditionVariable(&worker_cv); + InitializeConditionVariable(&leader_cv); + /* Initialize socket access */ err = WSAStartup(MAKEWORD(2, 2), &wsaData); if (err != 0) @@ -314,19 +341,22 @@ 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. - * (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.) + * We're a worker. Shut down our own DB connection if any. */ if (slot->AH) DisconnectDatabase(&(slot->AH->public)); #ifdef WIN32 - closesocket(slot->pipeRevRead); - closesocket(slot->pipeRevWrite); + /* + * Tell the leader we're gone so it stops waiting for our reply. + * On Unix the worker is a process whose exit closes the pipe and + * the leader sees EOF; the in-process channel has no EOF, and + * exit_nicely() ends only this thread, so signal it explicitly. + */ + EnterCriticalSection(&msg_lock); + slot->workerDied = true; + WakeAllConditionVariable(&leader_cv); + LeaveCriticalSection(&msg_lock); #endif } } @@ -351,6 +381,17 @@ ShutdownWorkersHard(ParallelState *pstate) { int i; + /* + * Tell any workers that are waiting for commands that they can exit. + */ +#ifdef WIN32 + EnterCriticalSection(&msg_lock); + for (i = 0; i < pstate->numWorkers; i++) + pstate->parallelSlot[i].chanClosed = true; + WakeAllConditionVariable(&worker_cv); + LeaveCriticalSection(&msg_lock); +#else + /* * Close our write end of the sockets so that any workers waiting for * commands know they can exit. (Note: some of the pipeWrite fields might @@ -359,6 +400,7 @@ ShutdownWorkersHard(ParallelState *pstate) */ for (i = 0; i < pstate->numWorkers; i++) closesocket(pstate->parallelSlot[i].pipeWrite); +#endif /* * Force early termination of any commands currently in progress. @@ -779,12 +821,6 @@ set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot) { - int pipefd[2]; - - /* fetch child ends of pipes */ - pipefd[PIPE_READ] = slot->pipeRevRead; - pipefd[PIPE_WRITE] = slot->pipeRevWrite; - /* * Clone the archive so that we have our own state to work with, and in * particular our own database connection. @@ -807,7 +843,7 @@ RunWorker(ArchiveHandle *AH, ParallelSlot *slot) /* * Execute commands until done. */ - WaitForCommands(AH, pipefd); + WaitForCommands(AH, slot); /* * Disconnect from database and clean up. @@ -830,6 +866,15 @@ init_spawned_worker_win32(WorkerInfo *wi) /* Don't need WorkerInfo anymore */ free(wi); + /* + * Record our thread id so GetMyPSlot() can tell us from the leader. Do + * this before anything that might call exit_nicely(): the cleanup handler + * uses GetMyPSlot(), and mistaking a failing worker for the leader + * deadlocks shutdown. We can't trust the leader to have stored the id + * from _beginthreadex() yet, since this thread may run before that returns. + */ + slot->threadId = GetCurrentThreadId(); + /* Run the worker ... */ RunWorker(AH, slot); @@ -895,11 +940,12 @@ ParallelBackupStart(ArchiveHandle *AH) uintptr_t handle; #else pid_t pid; -#endif - ParallelSlot *slot = &(pstate->parallelSlot[i]); int pipeMW[2], pipeWM[2]; +#endif + ParallelSlot *slot = &(pstate->parallelSlot[i]); +#ifndef WIN32 /* Create communication pipes for this worker */ if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0) pg_fatal("could not create communication channels: %m"); @@ -910,6 +956,7 @@ ParallelBackupStart(ArchiveHandle *AH) /* child's ends of the pipes */ slot->pipeRevRead = pipeMW[PIPE_READ]; slot->pipeRevWrite = pipeWM[PIPE_WRITE]; +#endif #ifdef WIN32 /* Create transient structure to pass args to worker function */ @@ -918,8 +965,13 @@ ParallelBackupStart(ArchiveHandle *AH) wi->AH = AH; wi->slot = slot; + /* + * The worker stores its own thread id (see init_spawned_worker_win32), + * so don't ask _beginthreadex() to report it -- that would race its + * store. + */ handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32, - wi, 0, &(slot->threadId)); + wi, 0, NULL); if (handle == 0) pg_fatal("could not create worker thread: %m"); slot->hThread = handle; @@ -1015,12 +1067,21 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) /* There should not be any unfinished jobs */ Assert(IsEveryWorkerIdle(pstate)); + /* Tell the workers they can exit */ +#ifdef WIN32 + EnterCriticalSection(&msg_lock); + for (i = 0; i < pstate->numWorkers; i++) + pstate->parallelSlot[i].chanClosed = true; + WakeAllConditionVariable(&worker_cv); + LeaveCriticalSection(&msg_lock); +#else /* Close the sockets so that the workers know they can exit */ for (i = 0; i < pstate->numWorkers; i++) { closesocket(pstate->parallelSlot[i].pipeRead); closesocket(pstate->parallelSlot[i].pipeWrite); } +#endif /* Wait for them to exit */ WaitForTerminatingWorkers(pstate); @@ -1281,7 +1342,7 @@ lockTableForWorker(ArchiveHandle *AH, TocEntry *te) * Read and execute commands from the leader until we see EOF on the pipe. */ static void -WaitForCommands(ArchiveHandle *AH, int pipefd[2]) +WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) { char *command; TocEntry *te; @@ -1291,7 +1352,7 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2]) for (;;) { - if (!(command = getMessageFromLeader(pipefd))) + if (!(command = getMessageFromLeader(slot))) { /* EOF, so done */ return; @@ -1319,7 +1380,7 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2]) /* Return status to leader */ buildWorkerResponse(AH, te, act, status, buf, sizeof(buf)); - sendMessageToLeader(pipefd, buf); + sendMessageToLeader(slot, buf); /* command was pg_malloc'd and we are responsible for free()ing it. */ free(command); @@ -1461,9 +1522,21 @@ WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode) * This function is executed in worker processes. */ static char * -getMessageFromLeader(int pipefd[2]) +getMessageFromLeader(ParallelSlot *slot) { - return readMessageFromPipe(pipefd[PIPE_READ]); +#ifdef WIN32 + char *msg; + + EnterCriticalSection(&msg_lock); + while (slot->cmdMsg == NULL && !slot->chanClosed) + SleepConditionVariableCS(&worker_cv, &msg_lock, INFINITE); + msg = slot->cmdMsg; /* NULL here means the channel was closed */ + slot->cmdMsg = NULL; + LeaveCriticalSection(&msg_lock); + return msg; +#else + return readMessageFromPipe(slot->pipeRevRead); +#endif } /* @@ -1472,12 +1545,20 @@ getMessageFromLeader(int pipefd[2]) * This function is executed in worker processes. */ static void -sendMessageToLeader(int pipefd[2], const char *str) +sendMessageToLeader(ParallelSlot *slot, const char *str) { +#ifdef WIN32 + EnterCriticalSection(&msg_lock); + Assert(slot->respMsg == NULL); + slot->respMsg = pg_strdup(str); + WakeAllConditionVariable(&leader_cv); + LeaveCriticalSection(&msg_lock); +#else int len = strlen(str) + 1; - if (pipewrite(pipefd[PIPE_WRITE], str, len) != len) + if (pipewrite(slot->pipeRevWrite, str, len) != len) pg_fatal("could not write to the communication channel: %m"); +#endif } /* @@ -1526,6 +1607,53 @@ select_loop(int maxFd, fd_set *workerset) static char * getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) { +#ifdef WIN32 + int i; + + /* + * Return the first pending response; if none and do_wait, sleep on + * leader_cv until a worker posts one. + */ + EnterCriticalSection(&msg_lock); + for (;;) + { + bool anyDied = false; + + for (i = 0; i < pstate->numWorkers; i++) + { + char *msg; + + if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) + continue; + msg = pstate->parallelSlot[i].respMsg; + if (msg != NULL) + { + pstate->parallelSlot[i].respMsg = NULL; + LeaveCriticalSection(&msg_lock); + *worker = i; + return msg; + } + if (pstate->parallelSlot[i].workerDied) + anyDied = true; + } + + /* + * A worker died without responding: return NULL as the Unix path does + * on EOF, so the caller reports the failure instead of hanging. + */ + if (anyDied) + { + LeaveCriticalSection(&msg_lock); + return NULL; + } + if (!do_wait) + { + LeaveCriticalSection(&msg_lock); + return NULL; + } + SleepConditionVariableCS(&leader_cv, &msg_lock, INFINITE); + } +#else int i; fd_set workerset; int maxFd = -1; @@ -1581,6 +1709,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) } Assert(false); return NULL; +#endif } /* @@ -1591,12 +1720,22 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str) { +#ifdef WIN32 + ParallelSlot *slot = &pstate->parallelSlot[worker]; + + EnterCriticalSection(&msg_lock); + Assert(slot->cmdMsg == NULL); + slot->cmdMsg = pg_strdup(str); + WakeAllConditionVariable(&worker_cv); + LeaveCriticalSection(&msg_lock); +#else int len = strlen(str) + 1; if (pipewrite(pstate->parallelSlot[worker].pipeWrite, str, len) != len) { pg_fatal("could not write to the communication channel: %m"); } +#endif } /* -- 2.54.0.windows.1 From b7317eb533d51b7b3fc56c63de1899e298ed0a99 Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Thu, 9 Jul 2026 12:09:10 -0500 Subject: [PATCH v2 3/5] pg_dump: port the parallel channel to pg_threads.h primitives The in-process command channel used Windows CRITICAL_SECTION and CONDITION_VARIABLE directly. Swap them for the portable pg_threads.h equivalents (pg_mtx_t/pg_cnd_t) so the same channel can serve non-Windows workers once they become threads. Still built only under WIN32; a follow-up removes that restriction along with the fork()/pipe worker path. --- src/bin/pg_dump/parallel.c | 84 +++++++++++++++++++------------------- 1 file changed, 41 insertions(+), 43 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index b89d394910..4959f10d81 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -63,6 +63,7 @@ #include "fe_utils/string_utils.h" #include "parallel.h" #include "pg_backup_utils.h" +#include "port/pg_threads.h" #ifdef WIN32 #include "port/pg_bswap.h" #endif @@ -188,16 +189,16 @@ typedef struct DumpSignalInformation static volatile DumpSignalInformation signal_info; #ifdef WIN32 -static CRITICAL_SECTION signal_info_lock; +static pg_mtx_t signal_info_lock = PG_MTX_INIT; /* * Synchronization for the in-process channels (see struct ParallelSlot). * msg_lock protects the per-slot cmdMsg/respMsg/chanClosed/workerDied fields; * worker_cv wakes a worker, leader_cv wakes the leader. */ -static CRITICAL_SECTION msg_lock; -static CONDITION_VARIABLE worker_cv; -static CONDITION_VARIABLE leader_cv; +static pg_mtx_t msg_lock = PG_MTX_INIT; +static pg_cnd_t worker_cv; +static pg_cnd_t leader_cv; #endif /* @@ -264,10 +265,9 @@ init_parallel_dump_utils(void) mainThreadId = GetCurrentThreadId(); - /* Initialize the in-process message-channel synchronization */ - InitializeCriticalSection(&msg_lock); - InitializeConditionVariable(&worker_cv); - InitializeConditionVariable(&leader_cv); + /* Initialize the in-process message-channel condition variables */ + pg_cnd_init(&worker_cv); + pg_cnd_init(&leader_cv); /* Initialize socket access */ err = WSAStartup(MAKEWORD(2, 2), &wsaData); @@ -353,10 +353,10 @@ archive_close_connection(int code, void *arg) * the leader sees EOF; the in-process channel has no EOF, and * exit_nicely() ends only this thread, so signal it explicitly. */ - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); slot->workerDied = true; - WakeAllConditionVariable(&leader_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&leader_cv); + pg_mtx_unlock(&msg_lock); #endif } } @@ -385,11 +385,11 @@ ShutdownWorkersHard(ParallelState *pstate) * Tell any workers that are waiting for commands that they can exit. */ #ifdef WIN32 - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; - WakeAllConditionVariable(&worker_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&worker_cv); + pg_mtx_unlock(&msg_lock); #else /* @@ -420,7 +420,7 @@ ShutdownWorkersHard(ParallelState *pstate) * 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; @@ -429,7 +429,7 @@ ShutdownWorkersHard(ParallelState *pstate) if (AH != NULL && AH->connCancel != NULL) (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif /* Now wait for them to terminate. */ @@ -647,7 +647,7 @@ consoleHandler(DWORD dwCtrlType) set_cancel_in_progress(); /* Critical section prevents changing data we look at here */ - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); /* * If in parallel mode, send QueryCancel to each worker's connected @@ -675,7 +675,7 @@ consoleHandler(DWORD dwCtrlType) (void) PQcancel(signal_info.myAH->connCancel, errbuf, sizeof(errbuf)); - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); /* * Report we're quitting, using nothing more complicated than @@ -704,8 +704,6 @@ set_cancel_handler(void) { signal_info.handler_set = true; - InitializeCriticalSection(&signal_info_lock); - SetConsoleCtrlHandler(consoleHandler, TRUE); } } @@ -738,7 +736,7 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) */ #ifdef WIN32 - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); #endif /* Free the old one if we have one */ @@ -768,7 +766,7 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) #endif #ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif } @@ -782,13 +780,13 @@ static void set_cancel_pstate(ParallelState *pstate) { #ifdef WIN32 - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); #endif signal_info.pstate = pstate; #ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif } @@ -802,13 +800,13 @@ static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) { #ifdef WIN32 - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); #endif slot->AH = AH; #ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif } @@ -1069,11 +1067,11 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) /* Tell the workers they can exit */ #ifdef WIN32 - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; - WakeAllConditionVariable(&worker_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&worker_cv); + pg_mtx_unlock(&msg_lock); #else /* Close the sockets so that the workers know they can exit */ for (i = 0; i < pstate->numWorkers; i++) @@ -1527,12 +1525,12 @@ getMessageFromLeader(ParallelSlot *slot) #ifdef WIN32 char *msg; - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); while (slot->cmdMsg == NULL && !slot->chanClosed) - SleepConditionVariableCS(&worker_cv, &msg_lock, INFINITE); + pg_cnd_wait(&worker_cv, &msg_lock); msg = slot->cmdMsg; /* NULL here means the channel was closed */ slot->cmdMsg = NULL; - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); return msg; #else return readMessageFromPipe(slot->pipeRevRead); @@ -1548,11 +1546,11 @@ static void sendMessageToLeader(ParallelSlot *slot, const char *str) { #ifdef WIN32 - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); Assert(slot->respMsg == NULL); slot->respMsg = pg_strdup(str); - WakeAllConditionVariable(&leader_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&leader_cv); + pg_mtx_unlock(&msg_lock); #else int len = strlen(str) + 1; @@ -1614,7 +1612,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) * Return the first pending response; if none and do_wait, sleep on * leader_cv until a worker posts one. */ - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); for (;;) { bool anyDied = false; @@ -1629,7 +1627,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) if (msg != NULL) { pstate->parallelSlot[i].respMsg = NULL; - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); *worker = i; return msg; } @@ -1643,15 +1641,15 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) */ if (anyDied) { - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); return NULL; } if (!do_wait) { - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); return NULL; } - SleepConditionVariableCS(&leader_cv, &msg_lock, INFINITE); + pg_cnd_wait(&leader_cv, &msg_lock); } #else int i; @@ -1723,11 +1721,11 @@ sendMessageToWorker(ParallelState *pstate, int worker, const char *str) #ifdef WIN32 ParallelSlot *slot = &pstate->parallelSlot[worker]; - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); Assert(slot->cmdMsg == NULL); slot->cmdMsg = pg_strdup(str); - WakeAllConditionVariable(&worker_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&worker_cv); + pg_mtx_unlock(&msg_lock); #else int len = strlen(str) + 1; -- 2.54.0.windows.1 From 5bf707bcb9e3c1a0ff5d96cc26b886854d1e9005 Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Thu, 9 Jul 2026 13:18:24 -0500 Subject: [PATCH v2 4/5] pg_dump: use threads on all platforms and remove the fork/pipe path Parallel dump and restore used worker processes on non-Windows (fork() plus socketpair pipes and select()) and threads on Windows. Now every platform uses threads, coordinated by the in-process channel that was introduced for Windows. Create workers with pg_thrd_create() and reap them with pg_thrd_join(), which also retires the MAXIMUM_WAIT_OBJECTS worker cap on Windows. A worker finds its own slot through a thread_local pointer rather than matching pid/threadId, and exit_nicely() ends only the calling thread via pg_thrd_exit(). The command channel (cmdMsg/respMsg under msg_lock, worker_cv/leader_cv) becomes the sole transport, so pgpipe(), select(), readMessageFromPipe() and the per-slot pipe descriptors all go. Cancellation is unified: with no worker processes to signal, the Unix handler cancels each worker's backend with PQcancel() just as the Windows console handler does, and the is_cancel_in_progress() flag that silences the resulting worker errors is no longer Windows-only. Requires the pg_threads.h portable threading API. --- src/bin/pg_dump/meson.build | 8 +- src/bin/pg_dump/parallel.c | 923 ++++++------------------------ src/bin/pg_dump/parallel.h | 7 +- src/bin/pg_dump/pg_backup_utils.c | 36 +- src/bin/pg_dump/pg_backup_utils.h | 15 +- 5 files changed, 198 insertions(+), 791 deletions(-) diff --git a/src/bin/pg_dump/meson.build b/src/bin/pg_dump/meson.build index 79bd503684..5fdf18128b 100644 --- a/src/bin/pg_dump/meson.build +++ b/src/bin/pg_dump/meson.build @@ -22,7 +22,7 @@ pg_dump_common_sources = files( pg_dump_common = static_library('libpgdump_common', pg_dump_common_sources, c_pch: pch_postgres_fe_h, - dependencies: [frontend_code, libpq, lz4, zlib, zstd], + dependencies: [frontend_code, libpq, lz4, zlib, zstd, thread_dep], kwargs: internal_lib_args, ) @@ -42,7 +42,7 @@ endif pg_dump = executable('pg_dump', pg_dump_sources, link_with: [pg_dump_common], - dependencies: [frontend_code, libpq, zlib], + dependencies: [frontend_code, libpq, zlib, thread_dep], kwargs: default_bin_args, ) bin_targets += pg_dump @@ -61,7 +61,7 @@ endif pg_dumpall = executable('pg_dumpall', pg_dumpall_sources, link_with: [pg_dump_common], - dependencies: [frontend_code, libpq, zlib], + dependencies: [frontend_code, libpq, zlib, thread_dep], kwargs: default_bin_args, ) bin_targets += pg_dumpall @@ -80,7 +80,7 @@ endif pg_restore = executable('pg_restore', pg_restore_sources, link_with: [pg_dump_common], - dependencies: [frontend_code, libpq, zlib], + dependencies: [frontend_code, libpq, zlib, thread_dep], kwargs: default_bin_args, ) bin_targets += pg_restore diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 4959f10d81..0c98be6f4a 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -16,13 +16,13 @@ /* * 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 + * The worker receives and decodes the command and passes it to the * routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr, * which are routines of the current archive format. That routine performs * the required action (dump or restore) and returns an integer status code. @@ -31,50 +31,42 @@ * DispatchJobForTocEntry(). The callback function does state updating * for the leader control logic in pg_backup_archiver.c. * + * Commands and responses are exchanged through a small in-process channel in + * each worker's ParallelSlot (see below), protected by msg_lock. The leader + * and workers are all threads in the same process. + * * 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. To avoid problems, workers operate on cloned + * copies of the Archive data structure; see RunWorker(). * - * 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 + * WRKR_TERMINATED: worker thread ended * The pstate->te[] entry for each worker is valid when it's in WRKR_WORKING * state, and must be NULL in other states. */ #include "postgres_fe.h" -#ifndef WIN32 -#include <sys/select.h> -#include <sys/wait.h> +#include <fcntl.h> #include <signal.h> #include <unistd.h> -#include <fcntl.h> -#endif #include "fe_utils/string_utils.h" #include "parallel.h" #include "pg_backup_utils.h" #include "port/pg_threads.h" -#ifdef WIN32 -#include "port/pg_bswap.h" -#endif - -/* Mnemonic macros for indexing the fd array returned by pipe(2) */ -#define PIPE_READ 0 -#define PIPE_WRITE 1 #define NO_SLOT (-1) /* Failure result for GetIdleWorker() */ -/* Worker process statuses */ +/* Worker thread statuses */ typedef enum { WRKR_NOT_STARTED = 0, @@ -89,9 +81,8 @@ 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 the owning worker thread. */ struct ParallelSlot { @@ -103,38 +94,22 @@ struct ParallelSlot ArchiveHandle *AH; /* Archive data worker is using */ - int pipeRead; /* leader's end of the pipes */ - int pipeWrite; - int pipeRevRead; /* child's end of the pipes */ - int pipeRevWrite; - -#ifdef WIN32 - /* - * In-process channel used instead of the pipes when workers are threads. - * A message is a malloc'd string; ownership passes to the receiver. - * Protected by msg_lock. + * In-process channel used to exchange messages between the leader and + * this worker. A message is a malloc'd string; ownership passes to the + * receiver. All fields are protected by msg_lock. */ char *cmdMsg; /* command pending for the worker, or NULL */ char *respMsg; /* response pending for the leader, or NULL */ bool chanClosed; /* leader closed the command channel (EOF) */ bool workerDied; /* worker exited without sending a response */ -#endif - /* Child process/thread identity info: */ -#ifdef WIN32 - uintptr_t hThread; - unsigned int threadId; -#else - pid_t pid; -#endif + pg_thrd_t thread; /* worker thread identity */ }; -#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 a newly-started worker thread via its + * single allowed argument. */ typedef struct { @@ -142,20 +117,6 @@ typedef struct ParallelSlot *slot; /* this worker's parallel slot */ } WorkerInfo; -/* Windows implementation of pipe access */ -static int pgpipe(int handles[2]); -#define piperead(a,b,c) recv(a,b,c,0) -#define pipewrite(a,b,c) send(a,b,c,0) - -#else /* !WIN32 */ - -/* Non-Windows implementation of pipe access */ -#define pgpipe(a) pipe(a) -#define piperead(a,b,c) read(a,b,c) -#define pipewrite(a,b,c) write(a,b,c) - -#endif /* WIN32 */ - /* * State info for archive_close_connection() shutdown callback. */ @@ -171,9 +132,8 @@ static ShutdownInformation shutdown_info; * State info for signal handling. * We assume signal_info initializes to zeroes. * - * On Unix, myAH is the leader DB connection in the leader process, and the - * worker's own connection in worker processes. On Windows, we have only one - * instance of signal_info, so myAH is the leader connection and the worker + * Since the workers are threads in the leader process, there's only one + * instance of signal_info: myAH is the leader connection, and the worker * connections must be dug out of pstate->parallelSlot[]. */ typedef struct DumpSignalInformation @@ -181,14 +141,10 @@ 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 pg_mtx_t signal_info_lock = PG_MTX_INIT; /* @@ -199,7 +155,6 @@ static pg_mtx_t signal_info_lock = PG_MTX_INIT; static pg_mtx_t msg_lock = PG_MTX_INIT; static pg_cnd_t worker_cv; static pg_cnd_t leader_cv; -#endif /* * Write a simple string to stderr --- must be safe in a signal handler. @@ -215,35 +170,32 @@ static pg_cnd_t leader_cv; } while (0) -#ifdef WIN32 -/* globally visible variables (needed by exit_nicely) */ -bool parallel_init_done = false; -DWORD mainThreadId; -#endif /* WIN32 */ +/* Pointer to each worker thread's ParallelSlot. NULL in the leader thread. */ +static thread_local ParallelSlot *parallel_slot_thread_local; + +/* Set once init_parallel_dump_utils() has run (needed by exit_nicely). */ +static bool parallel_init_done = false; /* 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); +static void handle_async_cancellation(void); 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 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, ParallelSlot *slot); static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait); static char *getMessageFromLeader(ParallelSlot *slot); static void sendMessageToLeader(ParallelSlot *slot, const char *str); -static int select_loop(int maxFd, fd_set *workerset); static char *getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker); static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str); -static char *readMessageFromPipe(int fd); #define messageStartsWith(msg, prefix) \ (strncmp(msg, prefix, strlen(prefix)) == 0) @@ -257,49 +209,35 @@ static char *readMessageFromPipe(int fd); void init_parallel_dump_utils(void) { -#ifdef WIN32 if (!parallel_init_done) { +#ifdef WIN32 WSADATA wsaData; int err; - mainThreadId = GetCurrentThreadId(); - - /* Initialize the in-process message-channel condition variables */ - pg_cnd_init(&worker_cv); - pg_cnd_init(&leader_cv); - /* Initialize socket access */ err = WSAStartup(MAKEWORD(2, 2), &wsaData); if (err != 0) pg_fatal("%s() failed: error code %d", "WSAStartup", err); +#endif + + /* Initialize the in-process message-channel condition variables */ + pg_cnd_init(&worker_cv); + pg_cnd_init(&leader_cv); parallel_init_done = true; } -#endif } /* - * Find the ParallelSlot for the current worker process or thread. + * Returns true in a parallel worker thread, false in the leader 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; } /* @@ -325,7 +263,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) { @@ -346,18 +284,15 @@ archive_close_connection(int code, void *arg) if (slot->AH) DisconnectDatabase(&(slot->AH->public)); -#ifdef WIN32 /* * Tell the leader we're gone so it stops waiting for our reply. - * On Unix the worker is a process whose exit closes the pipe and - * the leader sees EOF; the in-process channel has no EOF, and - * exit_nicely() ends only this thread, so signal it explicitly. + * The in-process channel has no EOF condition, and exit_nicely() + * ends only this thread, so we must signal the leader explicitly. */ pg_mtx_lock(&msg_lock); slot->workerDied = true; pg_cnd_broadcast(&leader_cv); pg_mtx_unlock(&msg_lock); -#endif } } else @@ -382,43 +317,20 @@ ShutdownWorkersHard(ParallelState *pstate) int i; /* - * Tell any workers that are waiting for commands that they can exit. + * Tell any workers that are waiting for commands that they can exit by + * closing their command channels. */ -#ifdef WIN32 pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); -#else - - /* - * Close our write end of the sockets so that any workers waiting for - * commands know they can exit. (Note: some of the pipeWrite fields might - * still be zero, if we failed to initialize all the workers. Hence, just - * ignore errors here.) - */ - for (i = 0; i < pstate->numWorkers; i++) - closesocket(pstate->parallelSlot[i].pipeWrite); -#endif - - /* - * 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. + * Force early termination of any commands currently in progress by + * sending query cancels directly to the workers' backends. Use + * signal_info_lock to ensure worker threads don't change the AH pointers + * concurrently. */ pg_mtx_lock(&signal_info_lock); for (i = 0; i < pstate->numWorkers; i++) @@ -430,7 +342,6 @@ ShutdownWorkersHard(ParallelState *pstate) (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } pg_mtx_unlock(&signal_info_lock); -#endif /* Now wait for them to terminate. */ WaitForTerminatingWorkers(pstate); @@ -442,64 +353,23 @@ 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 */ + ParallelSlot *slot = &pstate->parallelSlot[i]; int status; - pid_t pid = wait(&status); - - /* Find dead worker's slot, and clear the PID field */ - for (j = 0; j < pstate->numWorkers; j++) - { - 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++) + /* + * Join every worker that was actually started (i.e. is IDLE or + * WORKING). Skip slots that never started or were already reaped; + * their thread handle is not valid to join. + */ + if (WORKER_IS_RUNNING(slot->workerStatus)) { - 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, &status) != pg_thrd_success) + pg_fatal("could not join worker 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; } } @@ -517,62 +387,52 @@ WaitForTerminatingWorkers(ParallelState *pstate) * 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. 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. + * The workers are threads in the leader process on all platforms, so the + * cancel is handled the same way everywhere, in handle_async_cancellation(). */ -#ifndef WIN32 - /* - * Signal handler (Unix only) + * Common cancellation logic for the Unix signal handler and the Windows + * console handler. + * + * Unix: runs in a signal handler (async-signal-safe operations only). + * + * Windows: runs in a system-provided thread with signal_info_lock held by + * the caller. */ static void -sigTermHandler(SIGNAL_ARGS) +handle_async_cancellation(void) { - int i; char errbuf[1]; /* - * 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. + * 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. */ - pqsignal(SIGINT, PG_SIG_IGN); - pqsignal(SIGTERM, PG_SIG_IGN); - pqsignal(SIGQUIT, PG_SIG_IGN); + set_cancel_in_progress(); /* - * 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. + * 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++) + for (int i = 0; i < signal_info.pstate->numWorkers; i++) { - pid_t pid = signal_info.pstate->parallelSlot[i].pid; + ArchiveHandle *AH = signal_info.pstate->parallelSlot[i].AH; - if (pid != 0) - kill(pid, SIGTERM); + if (AH != NULL && AH->connCancel != NULL) + (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } } /* - * Send QueryCancel if we have a connection to send to. Ignore errors, + * 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) @@ -580,17 +440,33 @@ sigTermHandler(SIGNAL_ARGS) /* * Report we're quitting, using nothing more complicated than write(2). - * When in parallel operation, only the leader process should do this. */ - if (!signal_info.am_worker) + if (progname) { - if (progname) - { - write_stderr(progname); - write_stderr(": "); - } - write_stderr("terminated by user\n"); + write_stderr(progname); + write_stderr(": "); } + write_stderr("terminated by user\n"); +} + +#ifndef WIN32 + +/* + * Signal handler (Unix only) + */ +static void +sigTermHandler(SIGNAL_ARGS) +{ + /* + * 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, PG_SIG_IGN); + pqsignal(SIGTERM, PG_SIG_IGN); + pqsignal(SIGQUIT, PG_SIG_IGN); + + handle_async_cancellation(); /* * And die, using _exit() not exit() because the latter will invoke atexit @@ -605,10 +481,6 @@ sigTermHandler(SIGNAL_ARGS) static void set_cancel_handler(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) { signal_info.handler_set = true; @@ -624,70 +496,19 @@ set_cancel_handler(void) /* * 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. + * After sending cancel requests on all open connections, we return FALSE + * which will allow the default ExitProcess() action to be taken. */ static BOOL WINAPI consoleHandler(DWORD dwCtrlType) { - int i; - char errbuf[1]; - 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 */ pg_mtx_lock(&signal_info_lock); - - /* - * 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++) - { - ArchiveHandle *AH = signal_info.pstate->parallelSlot[i].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)); - + handle_async_cancellation(); pg_mtx_unlock(&signal_info_lock); - - /* - * Report we're quitting, using nothing more complicated than - * write(2). We should be able to use pg_log_*() here, but for now we - * stay aligned with the sigTermHandler behavior. - */ - if (progname) - { - write_stderr(progname); - write_stderr(": "); - } - write_stderr("terminated by user\n"); } /* Always return FALSE to allow signal handling to continue */ @@ -723,21 +544,15 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) PGcancel *oldConnCancel; /* - * 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. */ 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. + * Serialize updates to the cancel pointers under signal_info_lock; on + * Windows this also interlocks against the console-handler thread. */ - -#ifdef WIN32 pg_mtx_lock(&signal_info_lock); -#endif /* Free the old one if we have one */ oldConnCancel = AH->connCancel; @@ -752,22 +567,14 @@ 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 leader's myAH, unless we're in a worker thread. Workers make + * sure their ArchiveHandle appears in the pstate data, which is dealt + * with 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 pg_mtx_unlock(&signal_info_lock); -#endif } /* @@ -779,15 +586,9 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) static void set_cancel_pstate(ParallelState *pstate) { -#ifdef WIN32 pg_mtx_lock(&signal_info_lock); -#endif - signal_info.pstate = pstate; - -#ifdef WIN32 pg_mtx_unlock(&signal_info_lock); -#endif } /* @@ -799,34 +600,24 @@ set_cancel_pstate(ParallelState *pstate) static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) { -#ifdef WIN32 pg_mtx_lock(&signal_info_lock); -#endif - slot->AH = AH; - -#ifdef WIN32 pg_mtx_unlock(&signal_info_lock); -#endif } /* - * 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 thread. Caller should + * exit the thread upon return. */ static void 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. + * particular our own database connection. CloneArchive resets the state + * information and also clones the database connection, both of which are + * essential since worker threads share the leader's address space. */ AH = CloneArchive(AH); @@ -852,12 +643,12 @@ RunWorker(ArchiveHandle *AH, ParallelSlot *slot) } /* - * Thread base function for Windows + * Thread start function for all platforms. Matches pg_thrd_start_t. */ -#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; @@ -865,27 +656,24 @@ init_spawned_worker_win32(WorkerInfo *wi) free(wi); /* - * Record our thread id so GetMyPSlot() can tell us from the leader. Do - * this before anything that might call exit_nicely(): the cleanup handler - * uses GetMyPSlot(), and mistaking a failing worker for the leader - * deadlocks shutdown. We can't trust the leader to have stored the id - * from _beginthreadex() yet, since this thread may run before that returns. + * Record our slot so that archive_close_connection() and + * am_parallel_worker_thread() can tell us from the leader. Do this + * before anything that might call exit_nicely(): the cleanup handler uses + * this pointer, and mistaking a failing worker for the leader deadlocks + * shutdown. */ - slot->threadId = GetCurrentThreadId(); + 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) @@ -913,122 +701,44 @@ ParallelBackupStart(ArchiveHandle *AH) /* * Set the pstate in shutdown_info, to tell the exit handler that it must * clean up workers as well as the main database connection. But we don't - * set this in signal_info yet, because we don't want child processes to - * inherit non-NULL signal_info.pstate. + * set this in signal_info yet, because until the workers have something + * to do we want a cancel to just kill the leader connection. */ shutdown_info.pstate = pstate; /* - * Temporarily disable query cancellation on the leader connection. This - * ensures that child processes won't inherit valid AH->connCancel - * 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. + * Temporarily disable query cancellation on the leader connection. No + * harm is done if we fail while it's disabled, because the leader + * connection is idle at this point anyway. */ set_archive_cancel_info(AH, NULL); - /* Ensure stdio state is quiesced before forking */ + /* Ensure stdio state is quiesced before starting workers */ fflush(NULL); /* Create desired number of workers */ for (i = 0; i < pstate->numWorkers; i++) { -#ifdef WIN32 WorkerInfo *wi; - uintptr_t handle; -#else - pid_t pid; - int pipeMW[2], - pipeWM[2]; -#endif ParallelSlot *slot = &(pstate->parallelSlot[i]); -#ifndef WIN32 - /* Create communication pipes for this worker */ - if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0) - pg_fatal("could not create communication channels: %m"); - - /* leader's ends of the pipes */ - slot->pipeRead = pipeWM[PIPE_READ]; - slot->pipeWrite = pipeMW[PIPE_WRITE]; - /* child's ends of the pipes */ - slot->pipeRevRead = pipeMW[PIPE_READ]; - slot->pipeRevWrite = pipeWM[PIPE_WRITE]; -#endif - -#ifdef WIN32 /* Create transient structure to pass args to worker function */ wi = pg_malloc_object(WorkerInfo); - wi->AH = AH; wi->slot = slot; - /* - * The worker stores its own thread id (see init_spawned_worker_win32), - * so don't ask _beginthreadex() to report it -- that would race its - * store. - */ - handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32, - wi, 0, NULL); - if (handle == 0) - pg_fatal("could not create worker thread: %m"); - 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 create 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 + * killed if the leader tries to write to a backend over a broken + * connection. (libpq still uses sockets even though our own worker + * communication no longer does.) */ #ifndef WIN32 pqsignal(SIGPIPE, PG_SIG_IGN); @@ -1040,10 +750,11 @@ ParallelBackupStart(ArchiveHandle *AH) set_archive_cancel_info(AH, AH->connection); /* - * Tell the cancel signal handler to forward signals to worker processes, - * too. (As with query cancel, we did not need this earlier because the - * workers have not yet been given anything to do; if we die before this - * point, any already-started workers will see EOF and quit promptly.) + * Tell the cancel signal handler about the workers so it can cancel their + * backends too. (As with query cancel, we did not need this earlier + * because the workers have not yet been given anything to do; if we die + * before this point, any already-started workers will see their command + * channels close and quit promptly.) */ set_cancel_pstate(pstate); @@ -1065,21 +776,12 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) /* There should not be any unfinished jobs */ Assert(IsEveryWorkerIdle(pstate)); - /* Tell the workers they can exit */ -#ifdef WIN32 + /* Tell the workers they can exit by closing their command channels */ pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); -#else - /* Close the sockets so that the workers know they can exit */ - for (i = 0; i < pstate->numWorkers; i++) - { - closesocket(pstate->parallelSlot[i].pipeRead); - closesocket(pstate->parallelSlot[i].pipeWrite); - } -#endif /* Wait for them to exit */ WaitForTerminatingWorkers(pstate); @@ -1252,22 +954,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. */ @@ -1335,9 +1021,10 @@ lockTableForWorker(ArchiveHandle *AH, TocEntry *te) } /* - * WaitForCommands: main routine for a worker process. + * WaitForCommands: main routine for a worker thread. * - * Read and execute commands from the leader until we see EOF on the pipe. + * Read and execute commands from the leader until the command channel is + * closed. */ static void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) @@ -1412,9 +1099,9 @@ 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 is true, a worker must have died without responding */ if (do_wait) - pg_fatal("a worker process died unexpectedly"); + pg_fatal("a worker thread died unexpectedly"); return false; } @@ -1515,14 +1202,13 @@ WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode) /* * Read one command message from the leader, blocking if necessary * until one is available, and return it as a malloc'd string. - * On EOF, return NULL. + * On channel close, return NULL. * - * This function is executed in worker processes. + * This function is executed in worker threads. */ static char * getMessageFromLeader(ParallelSlot *slot) { -#ifdef WIN32 char *msg; pg_mtx_lock(&msg_lock); @@ -1532,80 +1218,40 @@ getMessageFromLeader(ParallelSlot *slot) slot->cmdMsg = NULL; pg_mtx_unlock(&msg_lock); return msg; -#else - return readMessageFromPipe(slot->pipeRevRead); -#endif } /* * Send a status message to the leader. * - * This function is executed in worker processes. + * This function is executed in worker threads. */ static void sendMessageToLeader(ParallelSlot *slot, const char *str) { -#ifdef WIN32 pg_mtx_lock(&msg_lock); Assert(slot->respMsg == NULL); slot->respMsg = pg_strdup(str); pg_cnd_broadcast(&leader_cv); pg_mtx_unlock(&msg_lock); -#else - int len = strlen(str) + 1; - - if (pipewrite(slot->pipeRevWrite, str, len) != len) - pg_fatal("could not write to the communication channel: %m"); -#endif -} - -/* - * Wait until some descriptor in "workerset" becomes readable. - * Returns -1 on error, else the number of readable descriptors. - */ -static int -select_loop(int maxFd, fd_set *workerset) -{ - int i; - fd_set saveSet = *workerset; - - for (;;) - { - *workerset = saveSet; - i = select(maxFd + 1, workerset, NULL, NULL, NULL); - -#ifndef WIN32 - if (i < 0 && errno == EINTR) - continue; -#else - if (i == SOCKET_ERROR && WSAGetLastError() == WSAEINTR) - continue; -#endif - break; - } - - return i; } - /* - * Check for messages from worker processes. + * Check for messages from worker threads. * * If a message is available, return it as a malloc'd string, and put the * index of the sending worker in *worker. * * If nothing is available, wait if "do_wait" is true, else return NULL. * - * If we detect EOF on any socket, we'll return NULL. It's not great that - * that's hard to distinguish from the no-data-available case, but for now + * If a worker has died without responding, we'll return NULL. It's not great + * that that's hard to distinguish from the no-data-available case, but for now * our one caller is okay with that. * - * This function is executed in the leader process. + * This function is executed in the leader thread. */ static char * getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) { -#ifdef WIN32 int i; /* @@ -1636,8 +1282,8 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) } /* - * A worker died without responding: return NULL as the Unix path does - * on EOF, so the caller reports the failure instead of hanging. + * A worker died without responding: return NULL so the caller reports + * the failure instead of hanging. */ if (anyDied) { @@ -1651,74 +1297,16 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) } pg_cnd_wait(&leader_cv, &msg_lock); } -#else - int i; - fd_set workerset; - int maxFd = -1; - struct timeval nowait = {0, 0}; - - /* construct bitmap of socket descriptors for select() */ - FD_ZERO(&workerset); - for (i = 0; i < pstate->numWorkers; i++) - { - if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) - continue; - FD_SET(pstate->parallelSlot[i].pipeRead, &workerset); - if (pstate->parallelSlot[i].pipeRead > maxFd) - maxFd = pstate->parallelSlot[i].pipeRead; - } - - if (do_wait) - { - i = select_loop(maxFd, &workerset); - Assert(i != 0); - } - else - { - if ((i = select(maxFd + 1, &workerset, NULL, NULL, &nowait)) == 0) - return NULL; - } - - if (i < 0) - pg_fatal("%s() failed: %m", "select"); - - for (i = 0; i < pstate->numWorkers; i++) - { - char *msg; - - if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) - continue; - if (!FD_ISSET(pstate->parallelSlot[i].pipeRead, &workerset)) - continue; - - /* - * Read the message if any. If the socket is ready because of EOF, - * we'll return NULL instead (and the socket will stay ready, so the - * condition will persist). - * - * Note: because this is a blocking read, we'll wait if only part of - * the message is available. Waiting a long time would be bad, but - * since worker status messages are short and are always sent in one - * operation, it shouldn't be a problem in practice. - */ - msg = readMessageFromPipe(pstate->parallelSlot[i].pipeRead); - *worker = i; - return msg; - } - Assert(false); - return NULL; -#endif } /* - * Send a command message to the specified worker process. + * Send a command message to the specified worker thread. * - * This function is executed in the leader process. + * This function is executed in the leader thread. */ static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str) { -#ifdef WIN32 ParallelSlot *slot = &pstate->parallelSlot[worker]; pg_mtx_lock(&msg_lock); @@ -1726,161 +1314,4 @@ sendMessageToWorker(ParallelState *pstate, int worker, const char *str) slot->cmdMsg = pg_strdup(str); pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); -#else - int len = strlen(str) + 1; - - if (pipewrite(pstate->parallelSlot[worker].pipeWrite, str, len) != len) - { - pg_fatal("could not write to the communication channel: %m"); - } -#endif -} - -/* - * Read one message from the specified pipe (fd), blocking if necessary - * until one is available, and return it as a malloc'd string. - * On EOF, return NULL. - * - * A "message" on the channel is just a null-terminated string. - */ -static char * -readMessageFromPipe(int fd) -{ - char *msg; - int msgsize, - bufsize; - int ret; - - /* - * In theory, if we let piperead() read multiple bytes, it might give us - * back fragments of multiple messages. (That can't actually occur, since - * neither leader nor workers send more than one message without waiting - * for a reply, but we don't wish to assume that here.) For simplicity, - * read a byte at a time until we get the terminating '\0'. This method - * is a bit inefficient, but since this is only used for relatively short - * command and status strings, it shouldn't matter. - */ - bufsize = 64; /* could be any number */ - msg = (char *) pg_malloc(bufsize); - msgsize = 0; - for (;;) - { - Assert(msgsize < bufsize); - ret = piperead(fd, msg + msgsize, 1); - if (ret <= 0) - break; /* error or connection closure */ - - Assert(ret == 1); - - if (msg[msgsize] == '\0') - return msg; /* collected whole message */ - - msgsize++; - if (msgsize == bufsize) /* enlarge buffer if needed */ - { - bufsize += 16; /* could be any number */ - msg = (char *) pg_realloc(msg, bufsize); - } - } - - /* Other end has closed the connection */ - pg_free(msg); - return NULL; -} - -#ifdef WIN32 - -/* - * This is a replacement version of pipe(2) for Windows which allows the pipe - * handles to be used in select(). - * - * Reads and writes on the pipe must go through piperead()/pipewrite(). - * - * For consistency with Unix we declare the returned handles as "int". - * This is okay even on WIN64 because system handles are not more than - * 32 bits wide, but we do have to do some casting. - */ -static int -pgpipe(int handles[2]) -{ - pgsocket s, - tmp_sock; - struct sockaddr_in serv_addr; - int len = sizeof(serv_addr); - - /* We have to use the Unix socket invalid file descriptor value here. */ - handles[0] = handles[1] = -1; - - /* - * setup listen socket - */ - if ((s = socket(AF_INET, SOCK_STREAM, 0)) == PGINVALID_SOCKET) - { - pg_log_error("pgpipe: could not create socket: error code %d", - WSAGetLastError()); - return -1; - } - - memset(&serv_addr, 0, sizeof(serv_addr)); - serv_addr.sin_family = AF_INET; - serv_addr.sin_port = pg_hton16(0); - serv_addr.sin_addr.s_addr = pg_hton32(INADDR_LOOPBACK); - if (bind(s, (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR) - { - pg_log_error("pgpipe: could not bind: error code %d", - WSAGetLastError()); - closesocket(s); - return -1; - } - if (listen(s, 1) == SOCKET_ERROR) - { - pg_log_error("pgpipe: could not listen: error code %d", - WSAGetLastError()); - closesocket(s); - return -1; - } - if (getsockname(s, (SOCKADDR *) &serv_addr, &len) == SOCKET_ERROR) - { - pg_log_error("pgpipe: %s() failed: error code %d", "getsockname", - WSAGetLastError()); - closesocket(s); - return -1; - } - - /* - * setup pipe handles - */ - if ((tmp_sock = socket(AF_INET, SOCK_STREAM, 0)) == PGINVALID_SOCKET) - { - pg_log_error("pgpipe: could not create second socket: error code %d", - WSAGetLastError()); - closesocket(s); - return -1; - } - handles[1] = (int) tmp_sock; - - if (connect(handles[1], (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR) - { - pg_log_error("pgpipe: could not connect socket: error code %d", - WSAGetLastError()); - closesocket(handles[1]); - handles[1] = -1; - closesocket(s); - return -1; - } - if ((tmp_sock = accept(s, (SOCKADDR *) &serv_addr, &len)) == PGINVALID_SOCKET) - { - pg_log_error("pgpipe: could not accept connection: error code %d", - WSAGetLastError()); - closesocket(handles[1]); - handles[1] = -1; - closesocket(s); - return -1; - } - handles[0] = (int) tmp_sock; - - closesocket(s); - return 0; } - -#endif /* WIN32 */ diff --git a/src/bin/pg_dump/parallel.h b/src/bin/pg_dump/parallel.h index f7557cd089..06764eefca 100644 --- a/src/bin/pg_dump/parallel.h +++ b/src/bin/pg_dump/parallel.h @@ -19,6 +19,7 @@ #include <limits.h> #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_utils.c b/src/bin/pg_dump/pg_backup_utils.c index 6d7ae9afc5..ce0e059554 100644 --- a/src/bin/pg_dump/pg_backup_utils.c +++ b/src/bin/pg_dump/pg_backup_utils.c @@ -13,25 +13,22 @@ */ #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; -#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 + * The cancel handler 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 memory barrier, but it's good enough here. A lot of things - * happen between set_cancel_in_progress() in the cancel thread and the other + * happen between set_cancel_in_progress() in the cancel handler and the other * threads calling is_cancel_in_progress(), including network operations, * which implicitly act as memory barriers. Furthermore, the flag is only * ever flipped one way (false to true) and a worker briefly observing the @@ -40,7 +37,7 @@ const char *progname = NULL; * cancelled" errors during the shutdown process. * * XXX: This should be swapped out for a proper atomic when we have those in - * the frontend code, so that we wouldn't need to rationalizee all of the + * the frontend code, so that we wouldn't need to rationalize all of the * above. */ static volatile bool cancelInProgress = false; @@ -57,8 +54,6 @@ is_cancel_in_progress(void) return cancelInProgress; } -#endif /* WIN32 */ - #define MAX_ON_EXIT_NICELY 20 static struct @@ -113,18 +108,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) @@ -135,10 +125,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); } diff --git a/src/bin/pg_dump/pg_backup_utils.h b/src/bin/pg_dump/pg_backup_utils.h index 8e0dd478cb..82e8c66048 100644 --- a/src/bin/pg_dump/pg_backup_utils.h +++ b/src/bin/pg_dump/pg_backup_utils.h @@ -32,25 +32,16 @@ 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' + * The parallel workers are threads inside the leader process on all platforms. + * When a cancel is processed, 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 + * handler sets this flag before sending any cancel, and worker threads check * it before reporting a query failure. - * - * 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 underlying 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 -- 2.54.0.windows.1 From fc5463c0a1e0aa3c5b00b1e4a54d9b4e39f7efd2 Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Thu, 9 Jul 2026 15:26:47 -0500 Subject: [PATCH v2 5/5] pg_dump: pass worker commands as structs instead of strings Now that the leader and workers share an address space, there's no reason to serialize each command and response to a string and parse it back. Send a WorkerCommand and WorkerResponse by value through the channel instead, removing buildWorkerCommand()/parseWorkerCommand() and their response counterparts along with the "DUMP %d" / "OK %d %d %d" formatting. This drops the never-used provision for format modules to add format-specific data to a command or response, but a struct can gain a field just as easily should that ever be wanted. --- src/bin/pg_dump/parallel.c | 310 ++++++++++--------------------- src/tools/pgindent/typedefs.list | 2 + 2 files changed, 105 insertions(+), 207 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 0c98be6f4a..29c5a372c4 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -20,12 +20,11 @@ * the desired number of worker threads, which each enter WaitForCommands(). * * 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 receives and decodes the command and passes it to the - * routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr, - * which are routines of the current archive format. That routine performs - * the required action (dump or restore) and returns an integer status code. + * threads in DispatchJobForTocEntry(). We hand over a WorkerCommand naming + * the action to take and the TocEntry to act on. The worker runs the routine + * pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr, which are + * routines of the current archive format. That routine performs the required + * action (dump or restore) and returns an integer status code. * This is passed back to the leader where we pass it to the * ParallelCompletionPtr callback function that was passed to * DispatchJobForTocEntry(). The callback function does state updating @@ -78,6 +77,25 @@ typedef enum #define WORKER_IS_RUNNING(workerStatus) \ ((workerStatus) == WRKR_IDLE || (workerStatus) == WRKR_WORKING) +/* + * A command handed from the leader to a worker: the action to take and the + * TocEntry to take it on. + */ +typedef struct +{ + T_Action act; /* ACT_DUMP or ACT_RESTORE */ + DumpId dumpId; /* TocEntry to act on */ +} WorkerCommand; + +/* + * A worker's response back to the leader. + */ +typedef struct +{ + int status; /* status code from the worker's job */ + int n_errors; /* errors to fold into the leader's count */ +} WorkerResponse; + /* * Private per-parallel-worker state (typedef for this is in parallel.h). * @@ -96,11 +114,14 @@ struct ParallelSlot /* * In-process channel used to exchange messages between the leader and - * this worker. A message is a malloc'd string; ownership passes to the - * receiver. All fields are protected by msg_lock. + * this worker. Since the two share an address space the messages are + * passed by value rather than serialized. All fields are protected by + * msg_lock. */ - char *cmdMsg; /* command pending for the worker, or NULL */ - char *respMsg; /* response pending for the leader, or NULL */ + bool cmdPending; /* a command is waiting for the worker */ + WorkerCommand cmd; /* the pending command */ + bool respPending; /* a response is waiting for the leader */ + WorkerResponse resp; /* the pending response */ bool chanClosed; /* leader closed the command channel (EOF) */ bool workerDied; /* worker exited without sending a response */ @@ -149,7 +170,7 @@ static pg_mtx_t signal_info_lock = PG_MTX_INIT; /* * Synchronization for the in-process channels (see struct ParallelSlot). - * msg_lock protects the per-slot cmdMsg/respMsg/chanClosed/workerDied fields; + * msg_lock protects the per-slot cmd/resp and their pending/closed/died flags; * worker_cv wakes a worker, leader_cv wakes the leader. */ static pg_mtx_t msg_lock = PG_MTX_INIT; @@ -190,15 +211,12 @@ static void lockTableForWorker(ArchiveHandle *AH, TocEntry *te); static void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot); static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait); -static char *getMessageFromLeader(ParallelSlot *slot); -static void sendMessageToLeader(ParallelSlot *slot, const char *str); -static char *getMessageFromWorker(ParallelState *pstate, - bool do_wait, int *worker); -static void sendMessageToWorker(ParallelState *pstate, - int worker, const char *str); - -#define messageStartsWith(msg, prefix) \ - (strncmp(msg, prefix, strlen(prefix)) == 0) +static bool getMessageFromLeader(ParallelSlot *slot, WorkerCommand *cmd); +static void sendMessageToLeader(ParallelSlot *slot, WorkerResponse resp); +static bool getMessageFromWorker(ParallelState *pstate, bool do_wait, + int *worker, WorkerResponse *resp); +static void sendMessageToWorker(ParallelState *pstate, int worker, + WorkerCommand cmd); /* @@ -799,108 +817,6 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) free(pstate); } -/* - * These next four functions handle construction and parsing of the command - * strings and response strings for parallel workers. - * - * Currently, these can be the same regardless of which archive format we are - * processing. In future, we might want to let format modules override these - * functions to add format-specific data to a command or response. - */ - -/* - * buildWorkerCommand: format a command string to send to a worker. - * - * The string is built in the caller-supplied buffer of size buflen. - */ -static void -buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act, - char *buf, int buflen) -{ - if (act == ACT_DUMP) - snprintf(buf, buflen, "DUMP %d", te->dumpId); - else if (act == ACT_RESTORE) - snprintf(buf, buflen, "RESTORE %d", te->dumpId); - else - Assert(false); -} - -/* - * parseWorkerCommand: interpret a command string in a worker. - */ -static void -parseWorkerCommand(ArchiveHandle *AH, TocEntry **te, T_Action *act, - const char *msg) -{ - DumpId dumpId; - int nBytes; - - if (messageStartsWith(msg, "DUMP ")) - { - *act = ACT_DUMP; - sscanf(msg, "DUMP %d%n", &dumpId, &nBytes); - Assert(nBytes == strlen(msg)); - *te = getTocEntryByDumpId(AH, dumpId); - Assert(*te != NULL); - } - else if (messageStartsWith(msg, "RESTORE ")) - { - *act = ACT_RESTORE; - sscanf(msg, "RESTORE %d%n", &dumpId, &nBytes); - Assert(nBytes == strlen(msg)); - *te = getTocEntryByDumpId(AH, dumpId); - Assert(*te != NULL); - } - else - pg_fatal("unrecognized command received from leader: \"%s\"", - msg); -} - -/* - * buildWorkerResponse: format a response string to send to the leader. - * - * The string is built in the caller-supplied buffer of size buflen. - */ -static void -buildWorkerResponse(ArchiveHandle *AH, TocEntry *te, T_Action act, int status, - char *buf, int buflen) -{ - snprintf(buf, buflen, "OK %d %d %d", - te->dumpId, - status, - status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0); -} - -/* - * parseWorkerResponse: parse the status message returned by a worker. - * - * Returns the integer status code, and may update fields of AH and/or te. - */ -static int -parseWorkerResponse(ArchiveHandle *AH, TocEntry *te, - const char *msg) -{ - DumpId dumpId; - int nBytes, - n_errors; - int status = 0; - - if (messageStartsWith(msg, "OK ")) - { - sscanf(msg, "OK %d %d %d%n", &dumpId, &status, &n_errors, &nBytes); - - Assert(dumpId == te->dumpId); - Assert(nBytes == strlen(msg)); - - AH->public.n_errors += n_errors; - } - else - pg_fatal("invalid message received from worker: \"%s\"", - msg); - - return status; -} - /* * Dispatch a job to some free worker. * @@ -919,16 +835,16 @@ DispatchJobForTocEntry(ArchiveHandle *AH, void *callback_data) { int worker; - char buf[256]; + WorkerCommand cmd; /* Get a worker, waiting if none are idle */ while ((worker = GetIdleWorker(pstate)) == NO_SLOT) WaitForWorkers(AH, pstate, WFW_ONE_IDLE); - /* Construct and send command string */ - buildWorkerCommand(AH, te, act, buf, sizeof(buf)); - - sendMessageToWorker(pstate, worker, buf); + /* Hand the worker its command */ + cmd.act = act; + cmd.dumpId = te->dumpId; + sendMessageToWorker(pstate, worker, cmd); /* Remember worker is busy, and which TocEntry it's working on */ pstate->parallelSlot[worker].workerStatus = WRKR_WORKING; @@ -1029,24 +945,17 @@ lockTableForWorker(ArchiveHandle *AH, TocEntry *te) static void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) { - char *command; + WorkerCommand cmd; TocEntry *te; - T_Action act; int status = 0; - char buf[256]; + WorkerResponse resp; - for (;;) + while (getMessageFromLeader(slot, &cmd)) { - if (!(command = getMessageFromLeader(slot))) - { - /* EOF, so done */ - return; - } + te = getTocEntryByDumpId(AH, cmd.dumpId); + Assert(te != NULL); - /* Decode the command */ - parseWorkerCommand(AH, &te, &act, command); - - if (act == ACT_DUMP) + if (cmd.act == ACT_DUMP) { /* Acquire lock on this table within the worker's session */ lockTableForWorker(AH, te); @@ -1054,7 +963,7 @@ WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) /* Perform the dump command */ status = (AH->WorkerJobDumpPtr) (AH, te); } - else if (act == ACT_RESTORE) + else if (cmd.act == ACT_RESTORE) { /* Perform the restore command */ status = (AH->WorkerJobRestorePtr) (AH, te); @@ -1063,12 +972,9 @@ WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) Assert(false); /* Return status to leader */ - buildWorkerResponse(AH, te, act, status, buf, sizeof(buf)); - - sendMessageToLeader(slot, buf); - - /* command was pg_malloc'd and we are responsible for free()ing it. */ - free(command); + resp.status = status; + resp.n_errors = (status == WORKER_IGNORED_ERRORS) ? AH->public.n_errors : 0; + sendMessageToLeader(slot, resp); } } @@ -1092,12 +998,12 @@ static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait) { int worker; - char *msg; + WorkerResponse resp; + ParallelSlot *slot; + TocEntry *te; /* Try to collect a status message */ - msg = getMessageFromWorker(pstate, do_wait, &worker); - - if (!msg) + if (!getMessageFromWorker(pstate, do_wait, &worker, &resp)) { /* If do_wait is true, a worker must have died without responding */ if (do_wait) @@ -1106,23 +1012,13 @@ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait) } /* Process it and update our idea of the worker's status */ - if (messageStartsWith(msg, "OK ")) - { - ParallelSlot *slot = &pstate->parallelSlot[worker]; - TocEntry *te = pstate->te[worker]; - int status; - - status = parseWorkerResponse(AH, te, msg); - slot->callback(AH, te, status, slot->callback_data); - slot->workerStatus = WRKR_IDLE; - pstate->te[worker] = NULL; - } - else - pg_fatal("invalid message received from worker: \"%s\"", - msg); + slot = &pstate->parallelSlot[worker]; + te = pstate->te[worker]; - /* Free the string returned from getMessageFromWorker */ - free(msg); + AH->public.n_errors += resp.n_errors; + slot->callback(AH, te, resp.status, slot->callback_data); + slot->workerStatus = WRKR_IDLE; + pstate->te[worker] = NULL; return true; } @@ -1200,57 +1096,63 @@ WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode) } /* - * Read one command message from the leader, blocking if necessary - * until one is available, and return it as a malloc'd string. - * On channel close, return NULL. + * Wait for the next command from the leader. Returns true and fills *cmd when + * one arrives, or false if the leader closed the channel, meaning the worker + * should exit. * * This function is executed in worker threads. */ -static char * -getMessageFromLeader(ParallelSlot *slot) +static bool +getMessageFromLeader(ParallelSlot *slot, WorkerCommand *cmd) { - char *msg; + bool gotCmd; pg_mtx_lock(&msg_lock); - while (slot->cmdMsg == NULL && !slot->chanClosed) + while (!slot->cmdPending && !slot->chanClosed) pg_cnd_wait(&worker_cv, &msg_lock); - msg = slot->cmdMsg; /* NULL here means the channel was closed */ - slot->cmdMsg = NULL; + gotCmd = slot->cmdPending; + if (gotCmd) + { + *cmd = slot->cmd; + slot->cmdPending = false; + } pg_mtx_unlock(&msg_lock); - return msg; + return gotCmd; } /* - * Send a status message to the leader. + * Send a status response to the leader. * * This function is executed in worker threads. */ static void -sendMessageToLeader(ParallelSlot *slot, const char *str) +sendMessageToLeader(ParallelSlot *slot, WorkerResponse resp) { pg_mtx_lock(&msg_lock); - Assert(slot->respMsg == NULL); - slot->respMsg = pg_strdup(str); + Assert(!slot->respPending); + slot->resp = resp; + slot->respPending = true; pg_cnd_broadcast(&leader_cv); pg_mtx_unlock(&msg_lock); } /* - * Check for messages from worker threads. + * Check for a response from a worker thread. * - * If a message is available, return it as a malloc'd string, and put the - * index of the sending worker in *worker. + * If one is available, fill *resp, put the index of the sending worker in + * *worker, and return true. * - * If nothing is available, wait if "do_wait" is true, else return NULL. + * If nothing is available, wait if "do_wait" is true, else return false. * - * If a worker has died without responding, we'll return NULL. It's not great + * If a worker has died without responding, we'll return false. It's not great * that that's hard to distinguish from the no-data-available case, but for now * our one caller is okay with that. * * This function is executed in the leader thread. */ -static char * -getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) +static bool +getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker, + WorkerResponse *resp) { int i; @@ -1265,53 +1167,47 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) for (i = 0; i < pstate->numWorkers; i++) { - char *msg; - if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) continue; - msg = pstate->parallelSlot[i].respMsg; - if (msg != NULL) + if (pstate->parallelSlot[i].respPending) { - pstate->parallelSlot[i].respMsg = NULL; + *resp = pstate->parallelSlot[i].resp; + pstate->parallelSlot[i].respPending = false; pg_mtx_unlock(&msg_lock); *worker = i; - return msg; + return true; } if (pstate->parallelSlot[i].workerDied) anyDied = true; } /* - * A worker died without responding: return NULL so the caller reports - * the failure instead of hanging. + * A worker died without responding, or nothing is pending and the + * caller doesn't want to wait: report that there's no message. */ - if (anyDied) + if (anyDied || !do_wait) { pg_mtx_unlock(&msg_lock); - return NULL; - } - if (!do_wait) - { - pg_mtx_unlock(&msg_lock); - return NULL; + return false; } pg_cnd_wait(&leader_cv, &msg_lock); } } /* - * Send a command message to the specified worker thread. + * Send a command to the specified worker thread. * * This function is executed in the leader thread. */ static void -sendMessageToWorker(ParallelState *pstate, int worker, const char *str) +sendMessageToWorker(ParallelState *pstate, int worker, WorkerCommand cmd) { ParallelSlot *slot = &pstate->parallelSlot[worker]; pg_mtx_lock(&msg_lock); - Assert(slot->cmdMsg == NULL); - slot->cmdMsg = pg_strdup(str); + Assert(!slot->cmdPending); + slot->cmd = cmd; + slot->cmdPending = true; pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); } diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 59a0f0a0fa..ae5f288444 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3485,11 +3485,13 @@ WordEntryPosVector WordEntryPosVector1 WorkTableScan WorkTableScanState +WorkerCommand WorkerInfo WorkerInfoData WorkerJobDumpPtrType WorkerJobRestorePtrType WorkerNodeInstrumentation +WorkerResponse Working_State WriteBufPtrType WriteBytePtrType -- 2.54.0.windows.1 Attachments: [text/plain] v2-0001-Give-fmtId-s-temporary-buffer-thread-local-storag.patch (5.1K, ../../[email protected]/2-v2-0001-Give-fmtId-s-temporary-buffer-thread-local-storag.patch) download | inline diff: From 653b37f2eb3adc429c8365c922a76d3979bde10f Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Sun, 28 Jun 2026 11:40:58 -0500 Subject: [PATCH v2 1/5] Give fmtId()'s temporary buffer thread-local storage getLocalPQExpBuffer() returns a function-scope static buffer, unsafe when fmtId() and fmtQualifiedId() run in multiple threads. On Windows, where pg_dump's parallel workers are threads, this was patched over at runtime by swapping in a TLS-based buffer from ParallelBackupStart(). Mark the default buffer C11 _Thread_local instead. Each thread gets its own, so the Windows TlsAlloc/TlsGetValue workaround goes. With no caller left to override it the getLocalPQExpBuffer indirection is pointless too: drop the function pointer, fold the default into a plain static getLocalPQExpBuffer(), and remove its now-obsolete extern. --- src/bin/pg_dump/parallel.c | 52 ----------------------------- src/fe_utils/string_utils.c | 13 +++----- src/include/fe_utils/string_utils.h | 1 - 3 files changed, 5 insertions(+), 61 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 4a0d04b646..b30e7db5a6 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -193,9 +193,6 @@ static CRITICAL_SECTION signal_info_lock; #ifdef WIN32 -/* file-scope variables */ -static DWORD tls_index; - /* globally visible variables (needed by exit_nicely) */ bool parallel_init_done = false; DWORD mainThreadId; @@ -243,8 +240,6 @@ init_parallel_dump_utils(void) WSADATA wsaData; int err; - /* Prepare for threaded operation */ - tls_index = TlsAlloc(); mainThreadId = GetCurrentThreadId(); /* Initialize socket access */ @@ -280,48 +275,6 @@ GetMyPSlot(ParallelState *pstate) return 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; - - 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; - } - - return id_return; -} -#endif /* WIN32 */ - /* * pg_dump and pg_restore call this to register the cleanup handler * as soon as they've created the ArchiveHandle. @@ -914,11 +867,6 @@ ParallelBackupStart(ArchiveHandle *AH) pstate->parallelSlot = pg_malloc0_array(ParallelSlot, pstate->numWorkers); -#ifdef WIN32 - /* Make fmtId() and fmtQualifiedId() use thread-local storage */ - getLocalPQExpBuffer = getThreadLocalPQExpBuffer; -#endif - /* * Set the pstate in shutdown_info, to tell the exit handler that it must * clean up workers as well as the main database connection. But we don't diff --git a/src/fe_utils/string_utils.c b/src/fe_utils/string_utils.c index 7a762251f3..f05bbf89bd 100644 --- a/src/fe_utils/string_utils.c +++ b/src/fe_utils/string_utils.c @@ -21,11 +21,8 @@ #include "fe_utils/string_utils.h" #include "mb/pg_wchar.h" -static PQExpBuffer defaultGetLocalPQExpBuffer(void); - /* Globals exported by this file */ int quote_all_identifiers = 0; -PQExpBuffer (*getLocalPQExpBuffer) (void) = defaultGetLocalPQExpBuffer; static int fmtIdEncoding = -1; @@ -34,14 +31,14 @@ static int fmtIdEncoding = -1; * Returns a temporary PQExpBuffer, valid until the next call to the function. * This is used by fmtId and fmtQualifiedId. * - * Non-reentrant and non-thread-safe but reduces memory leakage. You can - * replace this with a custom version by setting the getLocalPQExpBuffer - * function pointer. + * The buffer is thread-local, so this is safe to call from multiple threads + * at once; each thread gets its own. It is not reentrant within a thread, + * but it reduces memory leakage. */ static PQExpBuffer -defaultGetLocalPQExpBuffer(void) +getLocalPQExpBuffer(void) { - static PQExpBuffer id_return = NULL; + static _Thread_local PQExpBuffer id_return = NULL; if (id_return) /* first time through? */ { diff --git a/src/include/fe_utils/string_utils.h b/src/include/fe_utils/string_utils.h index 0680bb3c19..48e3b69afc 100644 --- a/src/include/fe_utils/string_utils.h +++ b/src/include/fe_utils/string_utils.h @@ -21,7 +21,6 @@ /* Global variables controlling behavior of fmtId() and fmtQualifiedId() */ extern PGDLLIMPORT int quote_all_identifiers; -extern PQExpBuffer (*getLocalPQExpBuffer) (void); /* Functions */ extern const char *fmtId(const char *rawid); -- 2.54.0.windows.1 [text/plain] v2-0002-pg_dump-dispatch-parallel-workers-in-process-on-W.patch (13.3K, ../../[email protected]/3-v2-0002-pg_dump-dispatch-parallel-workers-in-process-on-W.patch) download | inline diff: From a405f85281261a273da5258b1397dd6b62cbc53b Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Sun, 28 Jun 2026 12:05:10 -0500 Subject: [PATCH v2 2/5] pg_dump: dispatch parallel workers in-process on Windows On Windows the parallel workers are threads in the leader's process, but they reached the leader through the same transport as Unix worker processes: pgpipe() (a loopback TCP socket pair), select(), and a byte-at-a-time string protocol. That is pointless when the workers share the leader's address space. Give each ParallelSlot an in-process channel instead -- a command slot and a response slot under a critical section, with one condition variable to wake the worker and one to wake the leader. Messages pass as malloc'd strings; dispatch writes to memory rather than a socket. The pipe also signalled worker death through EOF. The channel has none, and exit_nicely() ends only the worker thread, so a failed worker -- for instance a connection that fails when -j exceeds max_connections -- would hang the leader forever. A dying worker now sets slot->workerDied and wakes the leader, and getMessageFromWorker() returns NULL just as the Unix path does on EOF. For the leader to name the dead worker, each worker records its own thread id on entry rather than trusting _beginthreadex()'s output argument, which can lag the thread's start. The Unix path is untouched; the now-unused pgpipe(), select_loop(), and readMessageFromPipe() stay until the worker model is unified on threads. --- src/bin/pg_dump/parallel.c | 195 +++++++++++++++++++++++++++++++------ 1 file changed, 167 insertions(+), 28 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index b30e7db5a6..b89d394910 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -107,6 +107,19 @@ struct ParallelSlot int pipeRevRead; /* child's end of the pipes */ int pipeRevWrite; +#ifdef WIN32 + + /* + * In-process channel used instead of the pipes when workers are threads. + * A message is a malloc'd string; ownership passes to the receiver. + * Protected by msg_lock. + */ + char *cmdMsg; /* command pending for the worker, or NULL */ + char *respMsg; /* response pending for the leader, or NULL */ + bool chanClosed; /* leader closed the command channel (EOF) */ + bool workerDied; /* worker exited without sending a response */ +#endif + /* Child process/thread identity info: */ #ifdef WIN32 uintptr_t hThread; @@ -176,6 +189,15 @@ static volatile DumpSignalInformation signal_info; #ifdef WIN32 static CRITICAL_SECTION signal_info_lock; + +/* + * Synchronization for the in-process channels (see struct ParallelSlot). + * msg_lock protects the per-slot cmdMsg/respMsg/chanClosed/workerDied fields; + * worker_cv wakes a worker, leader_cv wakes the leader. + */ +static CRITICAL_SECTION msg_lock; +static CONDITION_VARIABLE worker_cv; +static CONDITION_VARIABLE leader_cv; #endif /* @@ -210,11 +232,11 @@ 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 void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot); static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait); -static char *getMessageFromLeader(int pipefd[2]); -static void sendMessageToLeader(int pipefd[2], const char *str); +static char *getMessageFromLeader(ParallelSlot *slot); +static void sendMessageToLeader(ParallelSlot *slot, const char *str); static int select_loop(int maxFd, fd_set *workerset); static char *getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker); @@ -242,6 +264,11 @@ init_parallel_dump_utils(void) mainThreadId = GetCurrentThreadId(); + /* Initialize the in-process message-channel synchronization */ + InitializeCriticalSection(&msg_lock); + InitializeConditionVariable(&worker_cv); + InitializeConditionVariable(&leader_cv); + /* Initialize socket access */ err = WSAStartup(MAKEWORD(2, 2), &wsaData); if (err != 0) @@ -314,19 +341,22 @@ 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. - * (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.) + * We're a worker. Shut down our own DB connection if any. */ if (slot->AH) DisconnectDatabase(&(slot->AH->public)); #ifdef WIN32 - closesocket(slot->pipeRevRead); - closesocket(slot->pipeRevWrite); + /* + * Tell the leader we're gone so it stops waiting for our reply. + * On Unix the worker is a process whose exit closes the pipe and + * the leader sees EOF; the in-process channel has no EOF, and + * exit_nicely() ends only this thread, so signal it explicitly. + */ + EnterCriticalSection(&msg_lock); + slot->workerDied = true; + WakeAllConditionVariable(&leader_cv); + LeaveCriticalSection(&msg_lock); #endif } } @@ -351,6 +381,17 @@ ShutdownWorkersHard(ParallelState *pstate) { int i; + /* + * Tell any workers that are waiting for commands that they can exit. + */ +#ifdef WIN32 + EnterCriticalSection(&msg_lock); + for (i = 0; i < pstate->numWorkers; i++) + pstate->parallelSlot[i].chanClosed = true; + WakeAllConditionVariable(&worker_cv); + LeaveCriticalSection(&msg_lock); +#else + /* * Close our write end of the sockets so that any workers waiting for * commands know they can exit. (Note: some of the pipeWrite fields might @@ -359,6 +400,7 @@ ShutdownWorkersHard(ParallelState *pstate) */ for (i = 0; i < pstate->numWorkers; i++) closesocket(pstate->parallelSlot[i].pipeWrite); +#endif /* * Force early termination of any commands currently in progress. @@ -779,12 +821,6 @@ set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot) { - int pipefd[2]; - - /* fetch child ends of pipes */ - pipefd[PIPE_READ] = slot->pipeRevRead; - pipefd[PIPE_WRITE] = slot->pipeRevWrite; - /* * Clone the archive so that we have our own state to work with, and in * particular our own database connection. @@ -807,7 +843,7 @@ RunWorker(ArchiveHandle *AH, ParallelSlot *slot) /* * Execute commands until done. */ - WaitForCommands(AH, pipefd); + WaitForCommands(AH, slot); /* * Disconnect from database and clean up. @@ -830,6 +866,15 @@ init_spawned_worker_win32(WorkerInfo *wi) /* Don't need WorkerInfo anymore */ free(wi); + /* + * Record our thread id so GetMyPSlot() can tell us from the leader. Do + * this before anything that might call exit_nicely(): the cleanup handler + * uses GetMyPSlot(), and mistaking a failing worker for the leader + * deadlocks shutdown. We can't trust the leader to have stored the id + * from _beginthreadex() yet, since this thread may run before that returns. + */ + slot->threadId = GetCurrentThreadId(); + /* Run the worker ... */ RunWorker(AH, slot); @@ -895,11 +940,12 @@ ParallelBackupStart(ArchiveHandle *AH) uintptr_t handle; #else pid_t pid; -#endif - ParallelSlot *slot = &(pstate->parallelSlot[i]); int pipeMW[2], pipeWM[2]; +#endif + ParallelSlot *slot = &(pstate->parallelSlot[i]); +#ifndef WIN32 /* Create communication pipes for this worker */ if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0) pg_fatal("could not create communication channels: %m"); @@ -910,6 +956,7 @@ ParallelBackupStart(ArchiveHandle *AH) /* child's ends of the pipes */ slot->pipeRevRead = pipeMW[PIPE_READ]; slot->pipeRevWrite = pipeWM[PIPE_WRITE]; +#endif #ifdef WIN32 /* Create transient structure to pass args to worker function */ @@ -918,8 +965,13 @@ ParallelBackupStart(ArchiveHandle *AH) wi->AH = AH; wi->slot = slot; + /* + * The worker stores its own thread id (see init_spawned_worker_win32), + * so don't ask _beginthreadex() to report it -- that would race its + * store. + */ handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32, - wi, 0, &(slot->threadId)); + wi, 0, NULL); if (handle == 0) pg_fatal("could not create worker thread: %m"); slot->hThread = handle; @@ -1015,12 +1067,21 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) /* There should not be any unfinished jobs */ Assert(IsEveryWorkerIdle(pstate)); + /* Tell the workers they can exit */ +#ifdef WIN32 + EnterCriticalSection(&msg_lock); + for (i = 0; i < pstate->numWorkers; i++) + pstate->parallelSlot[i].chanClosed = true; + WakeAllConditionVariable(&worker_cv); + LeaveCriticalSection(&msg_lock); +#else /* Close the sockets so that the workers know they can exit */ for (i = 0; i < pstate->numWorkers; i++) { closesocket(pstate->parallelSlot[i].pipeRead); closesocket(pstate->parallelSlot[i].pipeWrite); } +#endif /* Wait for them to exit */ WaitForTerminatingWorkers(pstate); @@ -1281,7 +1342,7 @@ lockTableForWorker(ArchiveHandle *AH, TocEntry *te) * Read and execute commands from the leader until we see EOF on the pipe. */ static void -WaitForCommands(ArchiveHandle *AH, int pipefd[2]) +WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) { char *command; TocEntry *te; @@ -1291,7 +1352,7 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2]) for (;;) { - if (!(command = getMessageFromLeader(pipefd))) + if (!(command = getMessageFromLeader(slot))) { /* EOF, so done */ return; @@ -1319,7 +1380,7 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2]) /* Return status to leader */ buildWorkerResponse(AH, te, act, status, buf, sizeof(buf)); - sendMessageToLeader(pipefd, buf); + sendMessageToLeader(slot, buf); /* command was pg_malloc'd and we are responsible for free()ing it. */ free(command); @@ -1461,9 +1522,21 @@ WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode) * This function is executed in worker processes. */ static char * -getMessageFromLeader(int pipefd[2]) +getMessageFromLeader(ParallelSlot *slot) { - return readMessageFromPipe(pipefd[PIPE_READ]); +#ifdef WIN32 + char *msg; + + EnterCriticalSection(&msg_lock); + while (slot->cmdMsg == NULL && !slot->chanClosed) + SleepConditionVariableCS(&worker_cv, &msg_lock, INFINITE); + msg = slot->cmdMsg; /* NULL here means the channel was closed */ + slot->cmdMsg = NULL; + LeaveCriticalSection(&msg_lock); + return msg; +#else + return readMessageFromPipe(slot->pipeRevRead); +#endif } /* @@ -1472,12 +1545,20 @@ getMessageFromLeader(int pipefd[2]) * This function is executed in worker processes. */ static void -sendMessageToLeader(int pipefd[2], const char *str) +sendMessageToLeader(ParallelSlot *slot, const char *str) { +#ifdef WIN32 + EnterCriticalSection(&msg_lock); + Assert(slot->respMsg == NULL); + slot->respMsg = pg_strdup(str); + WakeAllConditionVariable(&leader_cv); + LeaveCriticalSection(&msg_lock); +#else int len = strlen(str) + 1; - if (pipewrite(pipefd[PIPE_WRITE], str, len) != len) + if (pipewrite(slot->pipeRevWrite, str, len) != len) pg_fatal("could not write to the communication channel: %m"); +#endif } /* @@ -1526,6 +1607,53 @@ select_loop(int maxFd, fd_set *workerset) static char * getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) { +#ifdef WIN32 + int i; + + /* + * Return the first pending response; if none and do_wait, sleep on + * leader_cv until a worker posts one. + */ + EnterCriticalSection(&msg_lock); + for (;;) + { + bool anyDied = false; + + for (i = 0; i < pstate->numWorkers; i++) + { + char *msg; + + if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) + continue; + msg = pstate->parallelSlot[i].respMsg; + if (msg != NULL) + { + pstate->parallelSlot[i].respMsg = NULL; + LeaveCriticalSection(&msg_lock); + *worker = i; + return msg; + } + if (pstate->parallelSlot[i].workerDied) + anyDied = true; + } + + /* + * A worker died without responding: return NULL as the Unix path does + * on EOF, so the caller reports the failure instead of hanging. + */ + if (anyDied) + { + LeaveCriticalSection(&msg_lock); + return NULL; + } + if (!do_wait) + { + LeaveCriticalSection(&msg_lock); + return NULL; + } + SleepConditionVariableCS(&leader_cv, &msg_lock, INFINITE); + } +#else int i; fd_set workerset; int maxFd = -1; @@ -1581,6 +1709,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) } Assert(false); return NULL; +#endif } /* @@ -1591,12 +1720,22 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str) { +#ifdef WIN32 + ParallelSlot *slot = &pstate->parallelSlot[worker]; + + EnterCriticalSection(&msg_lock); + Assert(slot->cmdMsg == NULL); + slot->cmdMsg = pg_strdup(str); + WakeAllConditionVariable(&worker_cv); + LeaveCriticalSection(&msg_lock); +#else int len = strlen(str) + 1; if (pipewrite(pstate->parallelSlot[worker].pipeWrite, str, len) != len) { pg_fatal("could not write to the communication channel: %m"); } +#endif } /* -- 2.54.0.windows.1 [text/plain] v2-0003-pg_dump-port-the-parallel-channel-to-pg_threads.h.patch (8.5K, ../../[email protected]/4-v2-0003-pg_dump-port-the-parallel-channel-to-pg_threads.h.patch) download | inline diff: From b7317eb533d51b7b3fc56c63de1899e298ed0a99 Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Thu, 9 Jul 2026 12:09:10 -0500 Subject: [PATCH v2 3/5] pg_dump: port the parallel channel to pg_threads.h primitives The in-process command channel used Windows CRITICAL_SECTION and CONDITION_VARIABLE directly. Swap them for the portable pg_threads.h equivalents (pg_mtx_t/pg_cnd_t) so the same channel can serve non-Windows workers once they become threads. Still built only under WIN32; a follow-up removes that restriction along with the fork()/pipe worker path. --- src/bin/pg_dump/parallel.c | 84 +++++++++++++++++++------------------- 1 file changed, 41 insertions(+), 43 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index b89d394910..4959f10d81 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -63,6 +63,7 @@ #include "fe_utils/string_utils.h" #include "parallel.h" #include "pg_backup_utils.h" +#include "port/pg_threads.h" #ifdef WIN32 #include "port/pg_bswap.h" #endif @@ -188,16 +189,16 @@ typedef struct DumpSignalInformation static volatile DumpSignalInformation signal_info; #ifdef WIN32 -static CRITICAL_SECTION signal_info_lock; +static pg_mtx_t signal_info_lock = PG_MTX_INIT; /* * Synchronization for the in-process channels (see struct ParallelSlot). * msg_lock protects the per-slot cmdMsg/respMsg/chanClosed/workerDied fields; * worker_cv wakes a worker, leader_cv wakes the leader. */ -static CRITICAL_SECTION msg_lock; -static CONDITION_VARIABLE worker_cv; -static CONDITION_VARIABLE leader_cv; +static pg_mtx_t msg_lock = PG_MTX_INIT; +static pg_cnd_t worker_cv; +static pg_cnd_t leader_cv; #endif /* @@ -264,10 +265,9 @@ init_parallel_dump_utils(void) mainThreadId = GetCurrentThreadId(); - /* Initialize the in-process message-channel synchronization */ - InitializeCriticalSection(&msg_lock); - InitializeConditionVariable(&worker_cv); - InitializeConditionVariable(&leader_cv); + /* Initialize the in-process message-channel condition variables */ + pg_cnd_init(&worker_cv); + pg_cnd_init(&leader_cv); /* Initialize socket access */ err = WSAStartup(MAKEWORD(2, 2), &wsaData); @@ -353,10 +353,10 @@ archive_close_connection(int code, void *arg) * the leader sees EOF; the in-process channel has no EOF, and * exit_nicely() ends only this thread, so signal it explicitly. */ - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); slot->workerDied = true; - WakeAllConditionVariable(&leader_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&leader_cv); + pg_mtx_unlock(&msg_lock); #endif } } @@ -385,11 +385,11 @@ ShutdownWorkersHard(ParallelState *pstate) * Tell any workers that are waiting for commands that they can exit. */ #ifdef WIN32 - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; - WakeAllConditionVariable(&worker_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&worker_cv); + pg_mtx_unlock(&msg_lock); #else /* @@ -420,7 +420,7 @@ ShutdownWorkersHard(ParallelState *pstate) * 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; @@ -429,7 +429,7 @@ ShutdownWorkersHard(ParallelState *pstate) if (AH != NULL && AH->connCancel != NULL) (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif /* Now wait for them to terminate. */ @@ -647,7 +647,7 @@ consoleHandler(DWORD dwCtrlType) set_cancel_in_progress(); /* Critical section prevents changing data we look at here */ - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); /* * If in parallel mode, send QueryCancel to each worker's connected @@ -675,7 +675,7 @@ consoleHandler(DWORD dwCtrlType) (void) PQcancel(signal_info.myAH->connCancel, errbuf, sizeof(errbuf)); - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); /* * Report we're quitting, using nothing more complicated than @@ -704,8 +704,6 @@ set_cancel_handler(void) { signal_info.handler_set = true; - InitializeCriticalSection(&signal_info_lock); - SetConsoleCtrlHandler(consoleHandler, TRUE); } } @@ -738,7 +736,7 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) */ #ifdef WIN32 - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); #endif /* Free the old one if we have one */ @@ -768,7 +766,7 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) #endif #ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif } @@ -782,13 +780,13 @@ static void set_cancel_pstate(ParallelState *pstate) { #ifdef WIN32 - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); #endif signal_info.pstate = pstate; #ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif } @@ -802,13 +800,13 @@ static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) { #ifdef WIN32 - EnterCriticalSection(&signal_info_lock); + pg_mtx_lock(&signal_info_lock); #endif slot->AH = AH; #ifdef WIN32 - LeaveCriticalSection(&signal_info_lock); + pg_mtx_unlock(&signal_info_lock); #endif } @@ -1069,11 +1067,11 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) /* Tell the workers they can exit */ #ifdef WIN32 - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; - WakeAllConditionVariable(&worker_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&worker_cv); + pg_mtx_unlock(&msg_lock); #else /* Close the sockets so that the workers know they can exit */ for (i = 0; i < pstate->numWorkers; i++) @@ -1527,12 +1525,12 @@ getMessageFromLeader(ParallelSlot *slot) #ifdef WIN32 char *msg; - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); while (slot->cmdMsg == NULL && !slot->chanClosed) - SleepConditionVariableCS(&worker_cv, &msg_lock, INFINITE); + pg_cnd_wait(&worker_cv, &msg_lock); msg = slot->cmdMsg; /* NULL here means the channel was closed */ slot->cmdMsg = NULL; - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); return msg; #else return readMessageFromPipe(slot->pipeRevRead); @@ -1548,11 +1546,11 @@ static void sendMessageToLeader(ParallelSlot *slot, const char *str) { #ifdef WIN32 - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); Assert(slot->respMsg == NULL); slot->respMsg = pg_strdup(str); - WakeAllConditionVariable(&leader_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&leader_cv); + pg_mtx_unlock(&msg_lock); #else int len = strlen(str) + 1; @@ -1614,7 +1612,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) * Return the first pending response; if none and do_wait, sleep on * leader_cv until a worker posts one. */ - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); for (;;) { bool anyDied = false; @@ -1629,7 +1627,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) if (msg != NULL) { pstate->parallelSlot[i].respMsg = NULL; - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); *worker = i; return msg; } @@ -1643,15 +1641,15 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) */ if (anyDied) { - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); return NULL; } if (!do_wait) { - LeaveCriticalSection(&msg_lock); + pg_mtx_unlock(&msg_lock); return NULL; } - SleepConditionVariableCS(&leader_cv, &msg_lock, INFINITE); + pg_cnd_wait(&leader_cv, &msg_lock); } #else int i; @@ -1723,11 +1721,11 @@ sendMessageToWorker(ParallelState *pstate, int worker, const char *str) #ifdef WIN32 ParallelSlot *slot = &pstate->parallelSlot[worker]; - EnterCriticalSection(&msg_lock); + pg_mtx_lock(&msg_lock); Assert(slot->cmdMsg == NULL); slot->cmdMsg = pg_strdup(str); - WakeAllConditionVariable(&worker_cv); - LeaveCriticalSection(&msg_lock); + pg_cnd_broadcast(&worker_cv); + pg_mtx_unlock(&msg_lock); #else int len = strlen(str) + 1; -- 2.54.0.windows.1 [text/plain] v2-0004-pg_dump-use-threads-on-all-platforms-and-remove-t.patch (53.0K, ../../[email protected]/5-v2-0004-pg_dump-use-threads-on-all-platforms-and-remove-t.patch) download | inline diff: From 5bf707bcb9e3c1a0ff5d96cc26b886854d1e9005 Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Thu, 9 Jul 2026 13:18:24 -0500 Subject: [PATCH v2 4/5] pg_dump: use threads on all platforms and remove the fork/pipe path Parallel dump and restore used worker processes on non-Windows (fork() plus socketpair pipes and select()) and threads on Windows. Now every platform uses threads, coordinated by the in-process channel that was introduced for Windows. Create workers with pg_thrd_create() and reap them with pg_thrd_join(), which also retires the MAXIMUM_WAIT_OBJECTS worker cap on Windows. A worker finds its own slot through a thread_local pointer rather than matching pid/threadId, and exit_nicely() ends only the calling thread via pg_thrd_exit(). The command channel (cmdMsg/respMsg under msg_lock, worker_cv/leader_cv) becomes the sole transport, so pgpipe(), select(), readMessageFromPipe() and the per-slot pipe descriptors all go. Cancellation is unified: with no worker processes to signal, the Unix handler cancels each worker's backend with PQcancel() just as the Windows console handler does, and the is_cancel_in_progress() flag that silences the resulting worker errors is no longer Windows-only. Requires the pg_threads.h portable threading API. --- src/bin/pg_dump/meson.build | 8 +- src/bin/pg_dump/parallel.c | 923 ++++++------------------------ src/bin/pg_dump/parallel.h | 7 +- src/bin/pg_dump/pg_backup_utils.c | 36 +- src/bin/pg_dump/pg_backup_utils.h | 15 +- 5 files changed, 198 insertions(+), 791 deletions(-) diff --git a/src/bin/pg_dump/meson.build b/src/bin/pg_dump/meson.build index 79bd503684..5fdf18128b 100644 --- a/src/bin/pg_dump/meson.build +++ b/src/bin/pg_dump/meson.build @@ -22,7 +22,7 @@ pg_dump_common_sources = files( pg_dump_common = static_library('libpgdump_common', pg_dump_common_sources, c_pch: pch_postgres_fe_h, - dependencies: [frontend_code, libpq, lz4, zlib, zstd], + dependencies: [frontend_code, libpq, lz4, zlib, zstd, thread_dep], kwargs: internal_lib_args, ) @@ -42,7 +42,7 @@ endif pg_dump = executable('pg_dump', pg_dump_sources, link_with: [pg_dump_common], - dependencies: [frontend_code, libpq, zlib], + dependencies: [frontend_code, libpq, zlib, thread_dep], kwargs: default_bin_args, ) bin_targets += pg_dump @@ -61,7 +61,7 @@ endif pg_dumpall = executable('pg_dumpall', pg_dumpall_sources, link_with: [pg_dump_common], - dependencies: [frontend_code, libpq, zlib], + dependencies: [frontend_code, libpq, zlib, thread_dep], kwargs: default_bin_args, ) bin_targets += pg_dumpall @@ -80,7 +80,7 @@ endif pg_restore = executable('pg_restore', pg_restore_sources, link_with: [pg_dump_common], - dependencies: [frontend_code, libpq, zlib], + dependencies: [frontend_code, libpq, zlib, thread_dep], kwargs: default_bin_args, ) bin_targets += pg_restore diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 4959f10d81..0c98be6f4a 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -16,13 +16,13 @@ /* * 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 + * The worker receives and decodes the command and passes it to the * routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr, * which are routines of the current archive format. That routine performs * the required action (dump or restore) and returns an integer status code. @@ -31,50 +31,42 @@ * DispatchJobForTocEntry(). The callback function does state updating * for the leader control logic in pg_backup_archiver.c. * + * Commands and responses are exchanged through a small in-process channel in + * each worker's ParallelSlot (see below), protected by msg_lock. The leader + * and workers are all threads in the same process. + * * 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. To avoid problems, workers operate on cloned + * copies of the Archive data structure; see RunWorker(). * - * 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 + * WRKR_TERMINATED: worker thread ended * The pstate->te[] entry for each worker is valid when it's in WRKR_WORKING * state, and must be NULL in other states. */ #include "postgres_fe.h" -#ifndef WIN32 -#include <sys/select.h> -#include <sys/wait.h> +#include <fcntl.h> #include <signal.h> #include <unistd.h> -#include <fcntl.h> -#endif #include "fe_utils/string_utils.h" #include "parallel.h" #include "pg_backup_utils.h" #include "port/pg_threads.h" -#ifdef WIN32 -#include "port/pg_bswap.h" -#endif - -/* Mnemonic macros for indexing the fd array returned by pipe(2) */ -#define PIPE_READ 0 -#define PIPE_WRITE 1 #define NO_SLOT (-1) /* Failure result for GetIdleWorker() */ -/* Worker process statuses */ +/* Worker thread statuses */ typedef enum { WRKR_NOT_STARTED = 0, @@ -89,9 +81,8 @@ 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 the owning worker thread. */ struct ParallelSlot { @@ -103,38 +94,22 @@ struct ParallelSlot ArchiveHandle *AH; /* Archive data worker is using */ - int pipeRead; /* leader's end of the pipes */ - int pipeWrite; - int pipeRevRead; /* child's end of the pipes */ - int pipeRevWrite; - -#ifdef WIN32 - /* - * In-process channel used instead of the pipes when workers are threads. - * A message is a malloc'd string; ownership passes to the receiver. - * Protected by msg_lock. + * In-process channel used to exchange messages between the leader and + * this worker. A message is a malloc'd string; ownership passes to the + * receiver. All fields are protected by msg_lock. */ char *cmdMsg; /* command pending for the worker, or NULL */ char *respMsg; /* response pending for the leader, or NULL */ bool chanClosed; /* leader closed the command channel (EOF) */ bool workerDied; /* worker exited without sending a response */ -#endif - /* Child process/thread identity info: */ -#ifdef WIN32 - uintptr_t hThread; - unsigned int threadId; -#else - pid_t pid; -#endif + pg_thrd_t thread; /* worker thread identity */ }; -#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 a newly-started worker thread via its + * single allowed argument. */ typedef struct { @@ -142,20 +117,6 @@ typedef struct ParallelSlot *slot; /* this worker's parallel slot */ } WorkerInfo; -/* Windows implementation of pipe access */ -static int pgpipe(int handles[2]); -#define piperead(a,b,c) recv(a,b,c,0) -#define pipewrite(a,b,c) send(a,b,c,0) - -#else /* !WIN32 */ - -/* Non-Windows implementation of pipe access */ -#define pgpipe(a) pipe(a) -#define piperead(a,b,c) read(a,b,c) -#define pipewrite(a,b,c) write(a,b,c) - -#endif /* WIN32 */ - /* * State info for archive_close_connection() shutdown callback. */ @@ -171,9 +132,8 @@ static ShutdownInformation shutdown_info; * State info for signal handling. * We assume signal_info initializes to zeroes. * - * On Unix, myAH is the leader DB connection in the leader process, and the - * worker's own connection in worker processes. On Windows, we have only one - * instance of signal_info, so myAH is the leader connection and the worker + * Since the workers are threads in the leader process, there's only one + * instance of signal_info: myAH is the leader connection, and the worker * connections must be dug out of pstate->parallelSlot[]. */ typedef struct DumpSignalInformation @@ -181,14 +141,10 @@ 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 pg_mtx_t signal_info_lock = PG_MTX_INIT; /* @@ -199,7 +155,6 @@ static pg_mtx_t signal_info_lock = PG_MTX_INIT; static pg_mtx_t msg_lock = PG_MTX_INIT; static pg_cnd_t worker_cv; static pg_cnd_t leader_cv; -#endif /* * Write a simple string to stderr --- must be safe in a signal handler. @@ -215,35 +170,32 @@ static pg_cnd_t leader_cv; } while (0) -#ifdef WIN32 -/* globally visible variables (needed by exit_nicely) */ -bool parallel_init_done = false; -DWORD mainThreadId; -#endif /* WIN32 */ +/* Pointer to each worker thread's ParallelSlot. NULL in the leader thread. */ +static thread_local ParallelSlot *parallel_slot_thread_local; + +/* Set once init_parallel_dump_utils() has run (needed by exit_nicely). */ +static bool parallel_init_done = false; /* 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); +static void handle_async_cancellation(void); 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 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, ParallelSlot *slot); static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait); static char *getMessageFromLeader(ParallelSlot *slot); static void sendMessageToLeader(ParallelSlot *slot, const char *str); -static int select_loop(int maxFd, fd_set *workerset); static char *getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker); static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str); -static char *readMessageFromPipe(int fd); #define messageStartsWith(msg, prefix) \ (strncmp(msg, prefix, strlen(prefix)) == 0) @@ -257,49 +209,35 @@ static char *readMessageFromPipe(int fd); void init_parallel_dump_utils(void) { -#ifdef WIN32 if (!parallel_init_done) { +#ifdef WIN32 WSADATA wsaData; int err; - mainThreadId = GetCurrentThreadId(); - - /* Initialize the in-process message-channel condition variables */ - pg_cnd_init(&worker_cv); - pg_cnd_init(&leader_cv); - /* Initialize socket access */ err = WSAStartup(MAKEWORD(2, 2), &wsaData); if (err != 0) pg_fatal("%s() failed: error code %d", "WSAStartup", err); +#endif + + /* Initialize the in-process message-channel condition variables */ + pg_cnd_init(&worker_cv); + pg_cnd_init(&leader_cv); parallel_init_done = true; } -#endif } /* - * Find the ParallelSlot for the current worker process or thread. + * Returns true in a parallel worker thread, false in the leader 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; } /* @@ -325,7 +263,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) { @@ -346,18 +284,15 @@ archive_close_connection(int code, void *arg) if (slot->AH) DisconnectDatabase(&(slot->AH->public)); -#ifdef WIN32 /* * Tell the leader we're gone so it stops waiting for our reply. - * On Unix the worker is a process whose exit closes the pipe and - * the leader sees EOF; the in-process channel has no EOF, and - * exit_nicely() ends only this thread, so signal it explicitly. + * The in-process channel has no EOF condition, and exit_nicely() + * ends only this thread, so we must signal the leader explicitly. */ pg_mtx_lock(&msg_lock); slot->workerDied = true; pg_cnd_broadcast(&leader_cv); pg_mtx_unlock(&msg_lock); -#endif } } else @@ -382,43 +317,20 @@ ShutdownWorkersHard(ParallelState *pstate) int i; /* - * Tell any workers that are waiting for commands that they can exit. + * Tell any workers that are waiting for commands that they can exit by + * closing their command channels. */ -#ifdef WIN32 pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); -#else - - /* - * Close our write end of the sockets so that any workers waiting for - * commands know they can exit. (Note: some of the pipeWrite fields might - * still be zero, if we failed to initialize all the workers. Hence, just - * ignore errors here.) - */ - for (i = 0; i < pstate->numWorkers; i++) - closesocket(pstate->parallelSlot[i].pipeWrite); -#endif - - /* - * 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. + * Force early termination of any commands currently in progress by + * sending query cancels directly to the workers' backends. Use + * signal_info_lock to ensure worker threads don't change the AH pointers + * concurrently. */ pg_mtx_lock(&signal_info_lock); for (i = 0; i < pstate->numWorkers; i++) @@ -430,7 +342,6 @@ ShutdownWorkersHard(ParallelState *pstate) (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } pg_mtx_unlock(&signal_info_lock); -#endif /* Now wait for them to terminate. */ WaitForTerminatingWorkers(pstate); @@ -442,64 +353,23 @@ 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 */ + ParallelSlot *slot = &pstate->parallelSlot[i]; int status; - pid_t pid = wait(&status); - - /* Find dead worker's slot, and clear the PID field */ - for (j = 0; j < pstate->numWorkers; j++) - { - 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++) + /* + * Join every worker that was actually started (i.e. is IDLE or + * WORKING). Skip slots that never started or were already reaped; + * their thread handle is not valid to join. + */ + if (WORKER_IS_RUNNING(slot->workerStatus)) { - 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, &status) != pg_thrd_success) + pg_fatal("could not join worker 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; } } @@ -517,62 +387,52 @@ WaitForTerminatingWorkers(ParallelState *pstate) * 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. 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. + * The workers are threads in the leader process on all platforms, so the + * cancel is handled the same way everywhere, in handle_async_cancellation(). */ -#ifndef WIN32 - /* - * Signal handler (Unix only) + * Common cancellation logic for the Unix signal handler and the Windows + * console handler. + * + * Unix: runs in a signal handler (async-signal-safe operations only). + * + * Windows: runs in a system-provided thread with signal_info_lock held by + * the caller. */ static void -sigTermHandler(SIGNAL_ARGS) +handle_async_cancellation(void) { - int i; char errbuf[1]; /* - * 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. + * 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. */ - pqsignal(SIGINT, PG_SIG_IGN); - pqsignal(SIGTERM, PG_SIG_IGN); - pqsignal(SIGQUIT, PG_SIG_IGN); + set_cancel_in_progress(); /* - * 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. + * 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++) + for (int i = 0; i < signal_info.pstate->numWorkers; i++) { - pid_t pid = signal_info.pstate->parallelSlot[i].pid; + ArchiveHandle *AH = signal_info.pstate->parallelSlot[i].AH; - if (pid != 0) - kill(pid, SIGTERM); + if (AH != NULL && AH->connCancel != NULL) + (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf)); } } /* - * Send QueryCancel if we have a connection to send to. Ignore errors, + * 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) @@ -580,17 +440,33 @@ sigTermHandler(SIGNAL_ARGS) /* * Report we're quitting, using nothing more complicated than write(2). - * When in parallel operation, only the leader process should do this. */ - if (!signal_info.am_worker) + if (progname) { - if (progname) - { - write_stderr(progname); - write_stderr(": "); - } - write_stderr("terminated by user\n"); + write_stderr(progname); + write_stderr(": "); } + write_stderr("terminated by user\n"); +} + +#ifndef WIN32 + +/* + * Signal handler (Unix only) + */ +static void +sigTermHandler(SIGNAL_ARGS) +{ + /* + * 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, PG_SIG_IGN); + pqsignal(SIGTERM, PG_SIG_IGN); + pqsignal(SIGQUIT, PG_SIG_IGN); + + handle_async_cancellation(); /* * And die, using _exit() not exit() because the latter will invoke atexit @@ -605,10 +481,6 @@ sigTermHandler(SIGNAL_ARGS) static void set_cancel_handler(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) { signal_info.handler_set = true; @@ -624,70 +496,19 @@ set_cancel_handler(void) /* * 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. + * After sending cancel requests on all open connections, we return FALSE + * which will allow the default ExitProcess() action to be taken. */ static BOOL WINAPI consoleHandler(DWORD dwCtrlType) { - int i; - char errbuf[1]; - 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 */ pg_mtx_lock(&signal_info_lock); - - /* - * 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++) - { - ArchiveHandle *AH = signal_info.pstate->parallelSlot[i].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)); - + handle_async_cancellation(); pg_mtx_unlock(&signal_info_lock); - - /* - * Report we're quitting, using nothing more complicated than - * write(2). We should be able to use pg_log_*() here, but for now we - * stay aligned with the sigTermHandler behavior. - */ - if (progname) - { - write_stderr(progname); - write_stderr(": "); - } - write_stderr("terminated by user\n"); } /* Always return FALSE to allow signal handling to continue */ @@ -723,21 +544,15 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) PGcancel *oldConnCancel; /* - * 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. */ 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. + * Serialize updates to the cancel pointers under signal_info_lock; on + * Windows this also interlocks against the console-handler thread. */ - -#ifdef WIN32 pg_mtx_lock(&signal_info_lock); -#endif /* Free the old one if we have one */ oldConnCancel = AH->connCancel; @@ -752,22 +567,14 @@ 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 leader's myAH, unless we're in a worker thread. Workers make + * sure their ArchiveHandle appears in the pstate data, which is dealt + * with 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 pg_mtx_unlock(&signal_info_lock); -#endif } /* @@ -779,15 +586,9 @@ set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn) static void set_cancel_pstate(ParallelState *pstate) { -#ifdef WIN32 pg_mtx_lock(&signal_info_lock); -#endif - signal_info.pstate = pstate; - -#ifdef WIN32 pg_mtx_unlock(&signal_info_lock); -#endif } /* @@ -799,34 +600,24 @@ set_cancel_pstate(ParallelState *pstate) static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH) { -#ifdef WIN32 pg_mtx_lock(&signal_info_lock); -#endif - slot->AH = AH; - -#ifdef WIN32 pg_mtx_unlock(&signal_info_lock); -#endif } /* - * 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 thread. Caller should + * exit the thread upon return. */ static void 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. + * particular our own database connection. CloneArchive resets the state + * information and also clones the database connection, both of which are + * essential since worker threads share the leader's address space. */ AH = CloneArchive(AH); @@ -852,12 +643,12 @@ RunWorker(ArchiveHandle *AH, ParallelSlot *slot) } /* - * Thread base function for Windows + * Thread start function for all platforms. Matches pg_thrd_start_t. */ -#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; @@ -865,27 +656,24 @@ init_spawned_worker_win32(WorkerInfo *wi) free(wi); /* - * Record our thread id so GetMyPSlot() can tell us from the leader. Do - * this before anything that might call exit_nicely(): the cleanup handler - * uses GetMyPSlot(), and mistaking a failing worker for the leader - * deadlocks shutdown. We can't trust the leader to have stored the id - * from _beginthreadex() yet, since this thread may run before that returns. + * Record our slot so that archive_close_connection() and + * am_parallel_worker_thread() can tell us from the leader. Do this + * before anything that might call exit_nicely(): the cleanup handler uses + * this pointer, and mistaking a failing worker for the leader deadlocks + * shutdown. */ - slot->threadId = GetCurrentThreadId(); + 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) @@ -913,122 +701,44 @@ ParallelBackupStart(ArchiveHandle *AH) /* * Set the pstate in shutdown_info, to tell the exit handler that it must * clean up workers as well as the main database connection. But we don't - * set this in signal_info yet, because we don't want child processes to - * inherit non-NULL signal_info.pstate. + * set this in signal_info yet, because until the workers have something + * to do we want a cancel to just kill the leader connection. */ shutdown_info.pstate = pstate; /* - * Temporarily disable query cancellation on the leader connection. This - * ensures that child processes won't inherit valid AH->connCancel - * 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. + * Temporarily disable query cancellation on the leader connection. No + * harm is done if we fail while it's disabled, because the leader + * connection is idle at this point anyway. */ set_archive_cancel_info(AH, NULL); - /* Ensure stdio state is quiesced before forking */ + /* Ensure stdio state is quiesced before starting workers */ fflush(NULL); /* Create desired number of workers */ for (i = 0; i < pstate->numWorkers; i++) { -#ifdef WIN32 WorkerInfo *wi; - uintptr_t handle; -#else - pid_t pid; - int pipeMW[2], - pipeWM[2]; -#endif ParallelSlot *slot = &(pstate->parallelSlot[i]); -#ifndef WIN32 - /* Create communication pipes for this worker */ - if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0) - pg_fatal("could not create communication channels: %m"); - - /* leader's ends of the pipes */ - slot->pipeRead = pipeWM[PIPE_READ]; - slot->pipeWrite = pipeMW[PIPE_WRITE]; - /* child's ends of the pipes */ - slot->pipeRevRead = pipeMW[PIPE_READ]; - slot->pipeRevWrite = pipeWM[PIPE_WRITE]; -#endif - -#ifdef WIN32 /* Create transient structure to pass args to worker function */ wi = pg_malloc_object(WorkerInfo); - wi->AH = AH; wi->slot = slot; - /* - * The worker stores its own thread id (see init_spawned_worker_win32), - * so don't ask _beginthreadex() to report it -- that would race its - * store. - */ - handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32, - wi, 0, NULL); - if (handle == 0) - pg_fatal("could not create worker thread: %m"); - 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 create 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 + * killed if the leader tries to write to a backend over a broken + * connection. (libpq still uses sockets even though our own worker + * communication no longer does.) */ #ifndef WIN32 pqsignal(SIGPIPE, PG_SIG_IGN); @@ -1040,10 +750,11 @@ ParallelBackupStart(ArchiveHandle *AH) set_archive_cancel_info(AH, AH->connection); /* - * Tell the cancel signal handler to forward signals to worker processes, - * too. (As with query cancel, we did not need this earlier because the - * workers have not yet been given anything to do; if we die before this - * point, any already-started workers will see EOF and quit promptly.) + * Tell the cancel signal handler about the workers so it can cancel their + * backends too. (As with query cancel, we did not need this earlier + * because the workers have not yet been given anything to do; if we die + * before this point, any already-started workers will see their command + * channels close and quit promptly.) */ set_cancel_pstate(pstate); @@ -1065,21 +776,12 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) /* There should not be any unfinished jobs */ Assert(IsEveryWorkerIdle(pstate)); - /* Tell the workers they can exit */ -#ifdef WIN32 + /* Tell the workers they can exit by closing their command channels */ pg_mtx_lock(&msg_lock); for (i = 0; i < pstate->numWorkers; i++) pstate->parallelSlot[i].chanClosed = true; pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); -#else - /* Close the sockets so that the workers know they can exit */ - for (i = 0; i < pstate->numWorkers; i++) - { - closesocket(pstate->parallelSlot[i].pipeRead); - closesocket(pstate->parallelSlot[i].pipeWrite); - } -#endif /* Wait for them to exit */ WaitForTerminatingWorkers(pstate); @@ -1252,22 +954,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. */ @@ -1335,9 +1021,10 @@ lockTableForWorker(ArchiveHandle *AH, TocEntry *te) } /* - * WaitForCommands: main routine for a worker process. + * WaitForCommands: main routine for a worker thread. * - * Read and execute commands from the leader until we see EOF on the pipe. + * Read and execute commands from the leader until the command channel is + * closed. */ static void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) @@ -1412,9 +1099,9 @@ 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 is true, a worker must have died without responding */ if (do_wait) - pg_fatal("a worker process died unexpectedly"); + pg_fatal("a worker thread died unexpectedly"); return false; } @@ -1515,14 +1202,13 @@ WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode) /* * Read one command message from the leader, blocking if necessary * until one is available, and return it as a malloc'd string. - * On EOF, return NULL. + * On channel close, return NULL. * - * This function is executed in worker processes. + * This function is executed in worker threads. */ static char * getMessageFromLeader(ParallelSlot *slot) { -#ifdef WIN32 char *msg; pg_mtx_lock(&msg_lock); @@ -1532,80 +1218,40 @@ getMessageFromLeader(ParallelSlot *slot) slot->cmdMsg = NULL; pg_mtx_unlock(&msg_lock); return msg; -#else - return readMessageFromPipe(slot->pipeRevRead); -#endif } /* * Send a status message to the leader. * - * This function is executed in worker processes. + * This function is executed in worker threads. */ static void sendMessageToLeader(ParallelSlot *slot, const char *str) { -#ifdef WIN32 pg_mtx_lock(&msg_lock); Assert(slot->respMsg == NULL); slot->respMsg = pg_strdup(str); pg_cnd_broadcast(&leader_cv); pg_mtx_unlock(&msg_lock); -#else - int len = strlen(str) + 1; - - if (pipewrite(slot->pipeRevWrite, str, len) != len) - pg_fatal("could not write to the communication channel: %m"); -#endif -} - -/* - * Wait until some descriptor in "workerset" becomes readable. - * Returns -1 on error, else the number of readable descriptors. - */ -static int -select_loop(int maxFd, fd_set *workerset) -{ - int i; - fd_set saveSet = *workerset; - - for (;;) - { - *workerset = saveSet; - i = select(maxFd + 1, workerset, NULL, NULL, NULL); - -#ifndef WIN32 - if (i < 0 && errno == EINTR) - continue; -#else - if (i == SOCKET_ERROR && WSAGetLastError() == WSAEINTR) - continue; -#endif - break; - } - - return i; } - /* - * Check for messages from worker processes. + * Check for messages from worker threads. * * If a message is available, return it as a malloc'd string, and put the * index of the sending worker in *worker. * * If nothing is available, wait if "do_wait" is true, else return NULL. * - * If we detect EOF on any socket, we'll return NULL. It's not great that - * that's hard to distinguish from the no-data-available case, but for now + * If a worker has died without responding, we'll return NULL. It's not great + * that that's hard to distinguish from the no-data-available case, but for now * our one caller is okay with that. * - * This function is executed in the leader process. + * This function is executed in the leader thread. */ static char * getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) { -#ifdef WIN32 int i; /* @@ -1636,8 +1282,8 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) } /* - * A worker died without responding: return NULL as the Unix path does - * on EOF, so the caller reports the failure instead of hanging. + * A worker died without responding: return NULL so the caller reports + * the failure instead of hanging. */ if (anyDied) { @@ -1651,74 +1297,16 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) } pg_cnd_wait(&leader_cv, &msg_lock); } -#else - int i; - fd_set workerset; - int maxFd = -1; - struct timeval nowait = {0, 0}; - - /* construct bitmap of socket descriptors for select() */ - FD_ZERO(&workerset); - for (i = 0; i < pstate->numWorkers; i++) - { - if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) - continue; - FD_SET(pstate->parallelSlot[i].pipeRead, &workerset); - if (pstate->parallelSlot[i].pipeRead > maxFd) - maxFd = pstate->parallelSlot[i].pipeRead; - } - - if (do_wait) - { - i = select_loop(maxFd, &workerset); - Assert(i != 0); - } - else - { - if ((i = select(maxFd + 1, &workerset, NULL, NULL, &nowait)) == 0) - return NULL; - } - - if (i < 0) - pg_fatal("%s() failed: %m", "select"); - - for (i = 0; i < pstate->numWorkers; i++) - { - char *msg; - - if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) - continue; - if (!FD_ISSET(pstate->parallelSlot[i].pipeRead, &workerset)) - continue; - - /* - * Read the message if any. If the socket is ready because of EOF, - * we'll return NULL instead (and the socket will stay ready, so the - * condition will persist). - * - * Note: because this is a blocking read, we'll wait if only part of - * the message is available. Waiting a long time would be bad, but - * since worker status messages are short and are always sent in one - * operation, it shouldn't be a problem in practice. - */ - msg = readMessageFromPipe(pstate->parallelSlot[i].pipeRead); - *worker = i; - return msg; - } - Assert(false); - return NULL; -#endif } /* - * Send a command message to the specified worker process. + * Send a command message to the specified worker thread. * - * This function is executed in the leader process. + * This function is executed in the leader thread. */ static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str) { -#ifdef WIN32 ParallelSlot *slot = &pstate->parallelSlot[worker]; pg_mtx_lock(&msg_lock); @@ -1726,161 +1314,4 @@ sendMessageToWorker(ParallelState *pstate, int worker, const char *str) slot->cmdMsg = pg_strdup(str); pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); -#else - int len = strlen(str) + 1; - - if (pipewrite(pstate->parallelSlot[worker].pipeWrite, str, len) != len) - { - pg_fatal("could not write to the communication channel: %m"); - } -#endif -} - -/* - * Read one message from the specified pipe (fd), blocking if necessary - * until one is available, and return it as a malloc'd string. - * On EOF, return NULL. - * - * A "message" on the channel is just a null-terminated string. - */ -static char * -readMessageFromPipe(int fd) -{ - char *msg; - int msgsize, - bufsize; - int ret; - - /* - * In theory, if we let piperead() read multiple bytes, it might give us - * back fragments of multiple messages. (That can't actually occur, since - * neither leader nor workers send more than one message without waiting - * for a reply, but we don't wish to assume that here.) For simplicity, - * read a byte at a time until we get the terminating '\0'. This method - * is a bit inefficient, but since this is only used for relatively short - * command and status strings, it shouldn't matter. - */ - bufsize = 64; /* could be any number */ - msg = (char *) pg_malloc(bufsize); - msgsize = 0; - for (;;) - { - Assert(msgsize < bufsize); - ret = piperead(fd, msg + msgsize, 1); - if (ret <= 0) - break; /* error or connection closure */ - - Assert(ret == 1); - - if (msg[msgsize] == '\0') - return msg; /* collected whole message */ - - msgsize++; - if (msgsize == bufsize) /* enlarge buffer if needed */ - { - bufsize += 16; /* could be any number */ - msg = (char *) pg_realloc(msg, bufsize); - } - } - - /* Other end has closed the connection */ - pg_free(msg); - return NULL; -} - -#ifdef WIN32 - -/* - * This is a replacement version of pipe(2) for Windows which allows the pipe - * handles to be used in select(). - * - * Reads and writes on the pipe must go through piperead()/pipewrite(). - * - * For consistency with Unix we declare the returned handles as "int". - * This is okay even on WIN64 because system handles are not more than - * 32 bits wide, but we do have to do some casting. - */ -static int -pgpipe(int handles[2]) -{ - pgsocket s, - tmp_sock; - struct sockaddr_in serv_addr; - int len = sizeof(serv_addr); - - /* We have to use the Unix socket invalid file descriptor value here. */ - handles[0] = handles[1] = -1; - - /* - * setup listen socket - */ - if ((s = socket(AF_INET, SOCK_STREAM, 0)) == PGINVALID_SOCKET) - { - pg_log_error("pgpipe: could not create socket: error code %d", - WSAGetLastError()); - return -1; - } - - memset(&serv_addr, 0, sizeof(serv_addr)); - serv_addr.sin_family = AF_INET; - serv_addr.sin_port = pg_hton16(0); - serv_addr.sin_addr.s_addr = pg_hton32(INADDR_LOOPBACK); - if (bind(s, (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR) - { - pg_log_error("pgpipe: could not bind: error code %d", - WSAGetLastError()); - closesocket(s); - return -1; - } - if (listen(s, 1) == SOCKET_ERROR) - { - pg_log_error("pgpipe: could not listen: error code %d", - WSAGetLastError()); - closesocket(s); - return -1; - } - if (getsockname(s, (SOCKADDR *) &serv_addr, &len) == SOCKET_ERROR) - { - pg_log_error("pgpipe: %s() failed: error code %d", "getsockname", - WSAGetLastError()); - closesocket(s); - return -1; - } - - /* - * setup pipe handles - */ - if ((tmp_sock = socket(AF_INET, SOCK_STREAM, 0)) == PGINVALID_SOCKET) - { - pg_log_error("pgpipe: could not create second socket: error code %d", - WSAGetLastError()); - closesocket(s); - return -1; - } - handles[1] = (int) tmp_sock; - - if (connect(handles[1], (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR) - { - pg_log_error("pgpipe: could not connect socket: error code %d", - WSAGetLastError()); - closesocket(handles[1]); - handles[1] = -1; - closesocket(s); - return -1; - } - if ((tmp_sock = accept(s, (SOCKADDR *) &serv_addr, &len)) == PGINVALID_SOCKET) - { - pg_log_error("pgpipe: could not accept connection: error code %d", - WSAGetLastError()); - closesocket(handles[1]); - handles[1] = -1; - closesocket(s); - return -1; - } - handles[0] = (int) tmp_sock; - - closesocket(s); - return 0; } - -#endif /* WIN32 */ diff --git a/src/bin/pg_dump/parallel.h b/src/bin/pg_dump/parallel.h index f7557cd089..06764eefca 100644 --- a/src/bin/pg_dump/parallel.h +++ b/src/bin/pg_dump/parallel.h @@ -19,6 +19,7 @@ #include <limits.h> #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_utils.c b/src/bin/pg_dump/pg_backup_utils.c index 6d7ae9afc5..ce0e059554 100644 --- a/src/bin/pg_dump/pg_backup_utils.c +++ b/src/bin/pg_dump/pg_backup_utils.c @@ -13,25 +13,22 @@ */ #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; -#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 + * The cancel handler 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 memory barrier, but it's good enough here. A lot of things - * happen between set_cancel_in_progress() in the cancel thread and the other + * happen between set_cancel_in_progress() in the cancel handler and the other * threads calling is_cancel_in_progress(), including network operations, * which implicitly act as memory barriers. Furthermore, the flag is only * ever flipped one way (false to true) and a worker briefly observing the @@ -40,7 +37,7 @@ const char *progname = NULL; * cancelled" errors during the shutdown process. * * XXX: This should be swapped out for a proper atomic when we have those in - * the frontend code, so that we wouldn't need to rationalizee all of the + * the frontend code, so that we wouldn't need to rationalize all of the * above. */ static volatile bool cancelInProgress = false; @@ -57,8 +54,6 @@ is_cancel_in_progress(void) return cancelInProgress; } -#endif /* WIN32 */ - #define MAX_ON_EXIT_NICELY 20 static struct @@ -113,18 +108,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) @@ -135,10 +125,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); } diff --git a/src/bin/pg_dump/pg_backup_utils.h b/src/bin/pg_dump/pg_backup_utils.h index 8e0dd478cb..82e8c66048 100644 --- a/src/bin/pg_dump/pg_backup_utils.h +++ b/src/bin/pg_dump/pg_backup_utils.h @@ -32,25 +32,16 @@ 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' + * The parallel workers are threads inside the leader process on all platforms. + * When a cancel is processed, 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 + * handler sets this flag before sending any cancel, and worker threads check * it before reporting a query failure. - * - * 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 underlying 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 -- 2.54.0.windows.1 [text/plain] v2-0005-pg_dump-pass-worker-commands-as-structs-instead-o.patch (16.3K, ../../[email protected]/6-v2-0005-pg_dump-pass-worker-commands-as-structs-instead-o.patch) download | inline diff: From fc5463c0a1e0aa3c5b00b1e4a54d9b4e39f7efd2 Mon Sep 17 00:00:00 2001 From: Bryan Green <[email protected]> Date: Thu, 9 Jul 2026 15:26:47 -0500 Subject: [PATCH v2 5/5] pg_dump: pass worker commands as structs instead of strings Now that the leader and workers share an address space, there's no reason to serialize each command and response to a string and parse it back. Send a WorkerCommand and WorkerResponse by value through the channel instead, removing buildWorkerCommand()/parseWorkerCommand() and their response counterparts along with the "DUMP %d" / "OK %d %d %d" formatting. This drops the never-used provision for format modules to add format-specific data to a command or response, but a struct can gain a field just as easily should that ever be wanted. --- src/bin/pg_dump/parallel.c | 310 ++++++++++--------------------- src/tools/pgindent/typedefs.list | 2 + 2 files changed, 105 insertions(+), 207 deletions(-) diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index 0c98be6f4a..29c5a372c4 100644 --- a/src/bin/pg_dump/parallel.c +++ b/src/bin/pg_dump/parallel.c @@ -20,12 +20,11 @@ * the desired number of worker threads, which each enter WaitForCommands(). * * 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 receives and decodes the command and passes it to the - * routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr, - * which are routines of the current archive format. That routine performs - * the required action (dump or restore) and returns an integer status code. + * threads in DispatchJobForTocEntry(). We hand over a WorkerCommand naming + * the action to take and the TocEntry to act on. The worker runs the routine + * pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr, which are + * routines of the current archive format. That routine performs the required + * action (dump or restore) and returns an integer status code. * This is passed back to the leader where we pass it to the * ParallelCompletionPtr callback function that was passed to * DispatchJobForTocEntry(). The callback function does state updating @@ -78,6 +77,25 @@ typedef enum #define WORKER_IS_RUNNING(workerStatus) \ ((workerStatus) == WRKR_IDLE || (workerStatus) == WRKR_WORKING) +/* + * A command handed from the leader to a worker: the action to take and the + * TocEntry to take it on. + */ +typedef struct +{ + T_Action act; /* ACT_DUMP or ACT_RESTORE */ + DumpId dumpId; /* TocEntry to act on */ +} WorkerCommand; + +/* + * A worker's response back to the leader. + */ +typedef struct +{ + int status; /* status code from the worker's job */ + int n_errors; /* errors to fold into the leader's count */ +} WorkerResponse; + /* * Private per-parallel-worker state (typedef for this is in parallel.h). * @@ -96,11 +114,14 @@ struct ParallelSlot /* * In-process channel used to exchange messages between the leader and - * this worker. A message is a malloc'd string; ownership passes to the - * receiver. All fields are protected by msg_lock. + * this worker. Since the two share an address space the messages are + * passed by value rather than serialized. All fields are protected by + * msg_lock. */ - char *cmdMsg; /* command pending for the worker, or NULL */ - char *respMsg; /* response pending for the leader, or NULL */ + bool cmdPending; /* a command is waiting for the worker */ + WorkerCommand cmd; /* the pending command */ + bool respPending; /* a response is waiting for the leader */ + WorkerResponse resp; /* the pending response */ bool chanClosed; /* leader closed the command channel (EOF) */ bool workerDied; /* worker exited without sending a response */ @@ -149,7 +170,7 @@ static pg_mtx_t signal_info_lock = PG_MTX_INIT; /* * Synchronization for the in-process channels (see struct ParallelSlot). - * msg_lock protects the per-slot cmdMsg/respMsg/chanClosed/workerDied fields; + * msg_lock protects the per-slot cmd/resp and their pending/closed/died flags; * worker_cv wakes a worker, leader_cv wakes the leader. */ static pg_mtx_t msg_lock = PG_MTX_INIT; @@ -190,15 +211,12 @@ static void lockTableForWorker(ArchiveHandle *AH, TocEntry *te); static void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot); static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait); -static char *getMessageFromLeader(ParallelSlot *slot); -static void sendMessageToLeader(ParallelSlot *slot, const char *str); -static char *getMessageFromWorker(ParallelState *pstate, - bool do_wait, int *worker); -static void sendMessageToWorker(ParallelState *pstate, - int worker, const char *str); - -#define messageStartsWith(msg, prefix) \ - (strncmp(msg, prefix, strlen(prefix)) == 0) +static bool getMessageFromLeader(ParallelSlot *slot, WorkerCommand *cmd); +static void sendMessageToLeader(ParallelSlot *slot, WorkerResponse resp); +static bool getMessageFromWorker(ParallelState *pstate, bool do_wait, + int *worker, WorkerResponse *resp); +static void sendMessageToWorker(ParallelState *pstate, int worker, + WorkerCommand cmd); /* @@ -799,108 +817,6 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate) free(pstate); } -/* - * These next four functions handle construction and parsing of the command - * strings and response strings for parallel workers. - * - * Currently, these can be the same regardless of which archive format we are - * processing. In future, we might want to let format modules override these - * functions to add format-specific data to a command or response. - */ - -/* - * buildWorkerCommand: format a command string to send to a worker. - * - * The string is built in the caller-supplied buffer of size buflen. - */ -static void -buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act, - char *buf, int buflen) -{ - if (act == ACT_DUMP) - snprintf(buf, buflen, "DUMP %d", te->dumpId); - else if (act == ACT_RESTORE) - snprintf(buf, buflen, "RESTORE %d", te->dumpId); - else - Assert(false); -} - -/* - * parseWorkerCommand: interpret a command string in a worker. - */ -static void -parseWorkerCommand(ArchiveHandle *AH, TocEntry **te, T_Action *act, - const char *msg) -{ - DumpId dumpId; - int nBytes; - - if (messageStartsWith(msg, "DUMP ")) - { - *act = ACT_DUMP; - sscanf(msg, "DUMP %d%n", &dumpId, &nBytes); - Assert(nBytes == strlen(msg)); - *te = getTocEntryByDumpId(AH, dumpId); - Assert(*te != NULL); - } - else if (messageStartsWith(msg, "RESTORE ")) - { - *act = ACT_RESTORE; - sscanf(msg, "RESTORE %d%n", &dumpId, &nBytes); - Assert(nBytes == strlen(msg)); - *te = getTocEntryByDumpId(AH, dumpId); - Assert(*te != NULL); - } - else - pg_fatal("unrecognized command received from leader: \"%s\"", - msg); -} - -/* - * buildWorkerResponse: format a response string to send to the leader. - * - * The string is built in the caller-supplied buffer of size buflen. - */ -static void -buildWorkerResponse(ArchiveHandle *AH, TocEntry *te, T_Action act, int status, - char *buf, int buflen) -{ - snprintf(buf, buflen, "OK %d %d %d", - te->dumpId, - status, - status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0); -} - -/* - * parseWorkerResponse: parse the status message returned by a worker. - * - * Returns the integer status code, and may update fields of AH and/or te. - */ -static int -parseWorkerResponse(ArchiveHandle *AH, TocEntry *te, - const char *msg) -{ - DumpId dumpId; - int nBytes, - n_errors; - int status = 0; - - if (messageStartsWith(msg, "OK ")) - { - sscanf(msg, "OK %d %d %d%n", &dumpId, &status, &n_errors, &nBytes); - - Assert(dumpId == te->dumpId); - Assert(nBytes == strlen(msg)); - - AH->public.n_errors += n_errors; - } - else - pg_fatal("invalid message received from worker: \"%s\"", - msg); - - return status; -} - /* * Dispatch a job to some free worker. * @@ -919,16 +835,16 @@ DispatchJobForTocEntry(ArchiveHandle *AH, void *callback_data) { int worker; - char buf[256]; + WorkerCommand cmd; /* Get a worker, waiting if none are idle */ while ((worker = GetIdleWorker(pstate)) == NO_SLOT) WaitForWorkers(AH, pstate, WFW_ONE_IDLE); - /* Construct and send command string */ - buildWorkerCommand(AH, te, act, buf, sizeof(buf)); - - sendMessageToWorker(pstate, worker, buf); + /* Hand the worker its command */ + cmd.act = act; + cmd.dumpId = te->dumpId; + sendMessageToWorker(pstate, worker, cmd); /* Remember worker is busy, and which TocEntry it's working on */ pstate->parallelSlot[worker].workerStatus = WRKR_WORKING; @@ -1029,24 +945,17 @@ lockTableForWorker(ArchiveHandle *AH, TocEntry *te) static void WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) { - char *command; + WorkerCommand cmd; TocEntry *te; - T_Action act; int status = 0; - char buf[256]; + WorkerResponse resp; - for (;;) + while (getMessageFromLeader(slot, &cmd)) { - if (!(command = getMessageFromLeader(slot))) - { - /* EOF, so done */ - return; - } + te = getTocEntryByDumpId(AH, cmd.dumpId); + Assert(te != NULL); - /* Decode the command */ - parseWorkerCommand(AH, &te, &act, command); - - if (act == ACT_DUMP) + if (cmd.act == ACT_DUMP) { /* Acquire lock on this table within the worker's session */ lockTableForWorker(AH, te); @@ -1054,7 +963,7 @@ WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) /* Perform the dump command */ status = (AH->WorkerJobDumpPtr) (AH, te); } - else if (act == ACT_RESTORE) + else if (cmd.act == ACT_RESTORE) { /* Perform the restore command */ status = (AH->WorkerJobRestorePtr) (AH, te); @@ -1063,12 +972,9 @@ WaitForCommands(ArchiveHandle *AH, ParallelSlot *slot) Assert(false); /* Return status to leader */ - buildWorkerResponse(AH, te, act, status, buf, sizeof(buf)); - - sendMessageToLeader(slot, buf); - - /* command was pg_malloc'd and we are responsible for free()ing it. */ - free(command); + resp.status = status; + resp.n_errors = (status == WORKER_IGNORED_ERRORS) ? AH->public.n_errors : 0; + sendMessageToLeader(slot, resp); } } @@ -1092,12 +998,12 @@ static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait) { int worker; - char *msg; + WorkerResponse resp; + ParallelSlot *slot; + TocEntry *te; /* Try to collect a status message */ - msg = getMessageFromWorker(pstate, do_wait, &worker); - - if (!msg) + if (!getMessageFromWorker(pstate, do_wait, &worker, &resp)) { /* If do_wait is true, a worker must have died without responding */ if (do_wait) @@ -1106,23 +1012,13 @@ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait) } /* Process it and update our idea of the worker's status */ - if (messageStartsWith(msg, "OK ")) - { - ParallelSlot *slot = &pstate->parallelSlot[worker]; - TocEntry *te = pstate->te[worker]; - int status; - - status = parseWorkerResponse(AH, te, msg); - slot->callback(AH, te, status, slot->callback_data); - slot->workerStatus = WRKR_IDLE; - pstate->te[worker] = NULL; - } - else - pg_fatal("invalid message received from worker: \"%s\"", - msg); + slot = &pstate->parallelSlot[worker]; + te = pstate->te[worker]; - /* Free the string returned from getMessageFromWorker */ - free(msg); + AH->public.n_errors += resp.n_errors; + slot->callback(AH, te, resp.status, slot->callback_data); + slot->workerStatus = WRKR_IDLE; + pstate->te[worker] = NULL; return true; } @@ -1200,57 +1096,63 @@ WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode) } /* - * Read one command message from the leader, blocking if necessary - * until one is available, and return it as a malloc'd string. - * On channel close, return NULL. + * Wait for the next command from the leader. Returns true and fills *cmd when + * one arrives, or false if the leader closed the channel, meaning the worker + * should exit. * * This function is executed in worker threads. */ -static char * -getMessageFromLeader(ParallelSlot *slot) +static bool +getMessageFromLeader(ParallelSlot *slot, WorkerCommand *cmd) { - char *msg; + bool gotCmd; pg_mtx_lock(&msg_lock); - while (slot->cmdMsg == NULL && !slot->chanClosed) + while (!slot->cmdPending && !slot->chanClosed) pg_cnd_wait(&worker_cv, &msg_lock); - msg = slot->cmdMsg; /* NULL here means the channel was closed */ - slot->cmdMsg = NULL; + gotCmd = slot->cmdPending; + if (gotCmd) + { + *cmd = slot->cmd; + slot->cmdPending = false; + } pg_mtx_unlock(&msg_lock); - return msg; + return gotCmd; } /* - * Send a status message to the leader. + * Send a status response to the leader. * * This function is executed in worker threads. */ static void -sendMessageToLeader(ParallelSlot *slot, const char *str) +sendMessageToLeader(ParallelSlot *slot, WorkerResponse resp) { pg_mtx_lock(&msg_lock); - Assert(slot->respMsg == NULL); - slot->respMsg = pg_strdup(str); + Assert(!slot->respPending); + slot->resp = resp; + slot->respPending = true; pg_cnd_broadcast(&leader_cv); pg_mtx_unlock(&msg_lock); } /* - * Check for messages from worker threads. + * Check for a response from a worker thread. * - * If a message is available, return it as a malloc'd string, and put the - * index of the sending worker in *worker. + * If one is available, fill *resp, put the index of the sending worker in + * *worker, and return true. * - * If nothing is available, wait if "do_wait" is true, else return NULL. + * If nothing is available, wait if "do_wait" is true, else return false. * - * If a worker has died without responding, we'll return NULL. It's not great + * If a worker has died without responding, we'll return false. It's not great * that that's hard to distinguish from the no-data-available case, but for now * our one caller is okay with that. * * This function is executed in the leader thread. */ -static char * -getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) +static bool +getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker, + WorkerResponse *resp) { int i; @@ -1265,53 +1167,47 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker) for (i = 0; i < pstate->numWorkers; i++) { - char *msg; - if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus)) continue; - msg = pstate->parallelSlot[i].respMsg; - if (msg != NULL) + if (pstate->parallelSlot[i].respPending) { - pstate->parallelSlot[i].respMsg = NULL; + *resp = pstate->parallelSlot[i].resp; + pstate->parallelSlot[i].respPending = false; pg_mtx_unlock(&msg_lock); *worker = i; - return msg; + return true; } if (pstate->parallelSlot[i].workerDied) anyDied = true; } /* - * A worker died without responding: return NULL so the caller reports - * the failure instead of hanging. + * A worker died without responding, or nothing is pending and the + * caller doesn't want to wait: report that there's no message. */ - if (anyDied) + if (anyDied || !do_wait) { pg_mtx_unlock(&msg_lock); - return NULL; - } - if (!do_wait) - { - pg_mtx_unlock(&msg_lock); - return NULL; + return false; } pg_cnd_wait(&leader_cv, &msg_lock); } } /* - * Send a command message to the specified worker thread. + * Send a command to the specified worker thread. * * This function is executed in the leader thread. */ static void -sendMessageToWorker(ParallelState *pstate, int worker, const char *str) +sendMessageToWorker(ParallelState *pstate, int worker, WorkerCommand cmd) { ParallelSlot *slot = &pstate->parallelSlot[worker]; pg_mtx_lock(&msg_lock); - Assert(slot->cmdMsg == NULL); - slot->cmdMsg = pg_strdup(str); + Assert(!slot->cmdPending); + slot->cmd = cmd; + slot->cmdPending = true; pg_cnd_broadcast(&worker_cv); pg_mtx_unlock(&msg_lock); } diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 59a0f0a0fa..ae5f288444 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3485,11 +3485,13 @@ WordEntryPosVector WordEntryPosVector1 WorkTableScan WorkTableScanState +WorkerCommand WorkerInfo WorkerInfoData WorkerJobDumpPtrType WorkerJobRestorePtrType WorkerNodeInstrumentation +WorkerResponse Working_State WriteBufPtrType WriteBytePtrType -- 2.54.0.windows.1 ^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2026-07-09 22:07 UTC | newest] Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2023-11-28 10:18 Re: WIP: libpq: add a possibility to not send D(escribe) when executing a prepared statement Ivan Trofimov <[email protected]> 2026-07-09 22:07 Re: pg_dump: use threads for parallel workers on all platforms Bryan Green <[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