Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * parallel.c
4 : : *
5 : : * Parallel support for pg_dump and pg_restore
6 : : *
7 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
8 : : * Portions Copyright (c) 1994, Regents of the University of California
9 : : *
10 : : * IDENTIFICATION
11 : : * src/bin/pg_dump/parallel.c
12 : : *
13 : : *-------------------------------------------------------------------------
14 : : */
15 : :
16 : : /*
17 : : * Parallel operation works like this:
18 : : *
19 : : * The original, leader process calls ParallelBackupStart(), which forks off
20 : : * the desired number of worker processes, which each enter WaitForCommands().
21 : : *
22 : : * The leader process dispatches an individual work item to one of the worker
23 : : * processes in DispatchJobForTocEntry(). We send a command string such as
24 : : * "DUMP 1234" or "RESTORE 1234", where 1234 is the TocEntry ID.
25 : : * The worker process receives and decodes the command and passes it to the
26 : : * routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr,
27 : : * which are routines of the current archive format. That routine performs
28 : : * the required action (dump or restore) and returns an integer status code.
29 : : * This is passed back to the leader where we pass it to the
30 : : * ParallelCompletionPtr callback function that was passed to
31 : : * DispatchJobForTocEntry(). The callback function does state updating
32 : : * for the leader control logic in pg_backup_archiver.c.
33 : : *
34 : : * In principle additional archive-format-specific information might be needed
35 : : * in commands or worker status responses, but so far that hasn't proved
36 : : * necessary, since workers have full copies of the ArchiveHandle/TocEntry
37 : : * data structures. Remember that we have forked off the workers only after
38 : : * we have read in the catalog. That's why our worker processes can also
39 : : * access the catalog information. (In the Windows case, the workers are
40 : : * threads in the same process. To avoid problems, they work with cloned
41 : : * copies of the Archive data structure; see RunWorker().)
42 : : *
43 : : * In the leader process, the workerStatus field for each worker has one of
44 : : * the following values:
45 : : * WRKR_NOT_STARTED: we've not yet forked this worker
46 : : * WRKR_IDLE: it's waiting for a command
47 : : * WRKR_WORKING: it's working on a command
48 : : * WRKR_TERMINATED: process ended
49 : : * The pstate->te[] entry for each worker is valid when it's in WRKR_WORKING
50 : : * state, and must be NULL in other states.
51 : : */
52 : :
53 : : #include "postgres_fe.h"
54 : :
55 : : #ifndef WIN32
56 : : #include <sys/select.h>
57 : : #include <sys/wait.h>
58 : : #include <signal.h>
59 : : #include <unistd.h>
60 : : #include <fcntl.h>
61 : : #endif
62 : :
63 : : #include "fe_utils/string_utils.h"
64 : : #include "parallel.h"
65 : : #include "pg_backup_utils.h"
66 : : #ifdef WIN32
67 : : #include "port/pg_bswap.h"
68 : : #endif
69 : :
70 : : /* Mnemonic macros for indexing the fd array returned by pipe(2) */
71 : : #define PIPE_READ 0
72 : : #define PIPE_WRITE 1
73 : :
74 : : #define NO_SLOT (-1) /* Failure result for GetIdleWorker() */
75 : :
76 : : /* Worker process statuses */
77 : : typedef enum
78 : : {
79 : : WRKR_NOT_STARTED = 0,
80 : : WRKR_IDLE,
81 : : WRKR_WORKING,
82 : : WRKR_TERMINATED,
83 : : } T_WorkerStatus;
84 : :
85 : : #define WORKER_IS_RUNNING(workerStatus) \
86 : : ((workerStatus) == WRKR_IDLE || (workerStatus) == WRKR_WORKING)
87 : :
88 : : /*
89 : : * Private per-parallel-worker state (typedef for this is in parallel.h).
90 : : *
91 : : * Much of this is valid only in the leader process (or, on Windows, should
92 : : * be touched only by the leader thread). But the AH field should be touched
93 : : * only by workers. The pipe descriptors are valid everywhere.
94 : : */
95 : : struct ParallelSlot
96 : : {
97 : : T_WorkerStatus workerStatus; /* see enum above */
98 : :
99 : : /* These fields are valid if workerStatus == WRKR_WORKING: */
100 : : ParallelCompletionPtr callback; /* function to call on completion */
101 : : void *callback_data; /* passthrough data for it */
102 : :
103 : : ArchiveHandle *AH; /* Archive data worker is using */
104 : :
105 : : int pipeRead; /* leader's end of the pipes */
106 : : int pipeWrite;
107 : : int pipeRevRead; /* child's end of the pipes */
108 : : int pipeRevWrite;
109 : :
110 : : /* Child process/thread identity info: */
111 : : #ifdef WIN32
112 : : uintptr_t hThread;
113 : : unsigned int threadId;
114 : : #else
115 : : pid_t pid;
116 : : #endif
117 : : };
118 : :
119 : : #ifdef WIN32
120 : :
121 : : /*
122 : : * Structure to hold info passed by _beginthreadex() to the function it calls
123 : : * via its single allowed argument.
124 : : */
125 : : typedef struct
126 : : {
127 : : ArchiveHandle *AH; /* leader database connection */
128 : : ParallelSlot *slot; /* this worker's parallel slot */
129 : : } WorkerInfo;
130 : :
131 : : /* Windows implementation of pipe access */
132 : : static int pgpipe(int handles[2]);
133 : : #define piperead(a,b,c) recv(a,b,c,0)
134 : : #define pipewrite(a,b,c) send(a,b,c,0)
135 : :
136 : : #else /* !WIN32 */
137 : :
138 : : /* Non-Windows implementation of pipe access */
139 : : #define pgpipe(a) pipe(a)
140 : : #define piperead(a,b,c) read(a,b,c)
141 : : #define pipewrite(a,b,c) write(a,b,c)
142 : :
143 : : #endif /* WIN32 */
144 : :
145 : : /*
146 : : * State info for archive_close_connection() shutdown callback.
147 : : */
148 : : typedef struct ShutdownInformation
149 : : {
150 : : ParallelState *pstate;
151 : : Archive *AHX;
152 : : } ShutdownInformation;
153 : :
154 : : static ShutdownInformation shutdown_info;
155 : :
156 : : /*
157 : : * State info for signal handling.
158 : : * We assume signal_info initializes to zeroes.
159 : : *
160 : : * On Unix, myAH is the leader DB connection in the leader process, and the
161 : : * worker's own connection in worker processes. On Windows, we have only one
162 : : * instance of signal_info, so myAH is the leader connection and the worker
163 : : * connections must be dug out of pstate->parallelSlot[].
164 : : */
165 : : typedef struct DumpSignalInformation
166 : : {
167 : : ArchiveHandle *myAH; /* database connection to issue cancel for */
168 : : ParallelState *pstate; /* parallel state, if any */
169 : : bool handler_set; /* signal handler set up in this process? */
170 : : #ifndef WIN32
171 : : bool am_worker; /* am I a worker process? */
172 : : #endif
173 : : } DumpSignalInformation;
174 : :
175 : : static volatile DumpSignalInformation signal_info;
176 : :
177 : : #ifdef WIN32
178 : : static CRITICAL_SECTION signal_info_lock;
179 : : #endif
180 : :
181 : : /*
182 : : * Write a simple string to stderr --- must be safe in a signal handler.
183 : : * We ignore the write() result since there's not much we could do about it.
184 : : * Certain compilers make that harder than it ought to be.
185 : : */
186 : : #define write_stderr(str) \
187 : : do { \
188 : : const char *str_ = (str); \
189 : : int rc_; \
190 : : rc_ = write(fileno(stderr), str_, strlen(str_)); \
191 : : (void) rc_; \
192 : : } while (0)
193 : :
194 : :
195 : : #ifdef WIN32
196 : : /* file-scope variables */
197 : : static DWORD tls_index;
198 : :
199 : : /* globally visible variables (needed by exit_nicely) */
200 : : bool parallel_init_done = false;
201 : : DWORD mainThreadId;
202 : : #endif /* WIN32 */
203 : :
204 : : /* Local function prototypes */
205 : : static ParallelSlot *GetMyPSlot(ParallelState *pstate);
206 : : static void archive_close_connection(int code, void *arg);
207 : : static void ShutdownWorkersHard(ParallelState *pstate);
208 : : static void WaitForTerminatingWorkers(ParallelState *pstate);
209 : : static void set_cancel_handler(void);
210 : : static void set_cancel_pstate(ParallelState *pstate);
211 : : static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH);
212 : : static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot);
213 : : static int GetIdleWorker(ParallelState *pstate);
214 : : static bool HasEveryWorkerTerminated(ParallelState *pstate);
215 : : static void lockTableForWorker(ArchiveHandle *AH, TocEntry *te);
216 : : static void WaitForCommands(ArchiveHandle *AH, int pipefd[2]);
217 : : static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate,
218 : : bool do_wait);
219 : : static char *getMessageFromLeader(int pipefd[2]);
220 : : static void sendMessageToLeader(int pipefd[2], const char *str);
221 : : static int select_loop(int maxFd, fd_set *workerset);
222 : : static char *getMessageFromWorker(ParallelState *pstate,
223 : : bool do_wait, int *worker);
224 : : static void sendMessageToWorker(ParallelState *pstate,
225 : : int worker, const char *str);
226 : : static char *readMessageFromPipe(int fd);
227 : :
228 : : #define messageStartsWith(msg, prefix) \
229 : : (strncmp(msg, prefix, strlen(prefix)) == 0)
230 : :
231 : :
232 : : /*
233 : : * Initialize parallel dump support --- should be called early in process
234 : : * startup. (Currently, this is called whether or not we intend parallel
235 : : * activity.)
236 : : */
237 : : void
238 : 403 : init_parallel_dump_utils(void)
239 : : {
240 : : #ifdef WIN32
241 : : if (!parallel_init_done)
242 : : {
243 : : WSADATA wsaData;
244 : : int err;
245 : :
246 : : /* Prepare for threaded operation */
247 : : tls_index = TlsAlloc();
248 : : mainThreadId = GetCurrentThreadId();
249 : :
250 : : /* Initialize socket access */
251 : : err = WSAStartup(MAKEWORD(2, 2), &wsaData);
252 : : if (err != 0)
253 : : pg_fatal("%s() failed: error code %d", "WSAStartup", err);
254 : :
255 : : parallel_init_done = true;
256 : : }
257 : : #endif
258 : 403 : }
259 : :
260 : : /*
261 : : * Find the ParallelSlot for the current worker process or thread.
262 : : *
263 : : * Returns NULL if no matching slot is found (this implies we're the leader).
264 : : */
265 : : static ParallelSlot *
266 : 0 : GetMyPSlot(ParallelState *pstate)
267 : : {
268 : : int i;
269 : :
270 [ # # ]: 0 : for (i = 0; i < pstate->numWorkers; i++)
271 : : {
272 : : #ifdef WIN32
273 : : if (pstate->parallelSlot[i].threadId == GetCurrentThreadId())
274 : : #else
275 [ # # ]: 0 : if (pstate->parallelSlot[i].pid == getpid())
276 : : #endif
277 : 0 : return &(pstate->parallelSlot[i]);
278 : : }
279 : :
280 : 0 : return NULL;
281 : : }
282 : :
283 : : /*
284 : : * A thread-local version of getLocalPQExpBuffer().
285 : : *
286 : : * Non-reentrant but reduces memory leakage: we'll consume one buffer per
287 : : * thread, which is much better than one per fmtId/fmtQualifiedId call.
288 : : */
289 : : #ifdef WIN32
290 : : static PQExpBuffer
291 : : getThreadLocalPQExpBuffer(void)
292 : : {
293 : : /*
294 : : * The Tls code goes awry if we use a static var, so we provide for both
295 : : * static and auto, and omit any use of the static var when using Tls. We
296 : : * rely on TlsGetValue() to return 0 if the value is not yet set.
297 : : */
298 : : static PQExpBuffer s_id_return = NULL;
299 : : PQExpBuffer id_return;
300 : :
301 : : if (parallel_init_done)
302 : : id_return = (PQExpBuffer) TlsGetValue(tls_index);
303 : : else
304 : : id_return = s_id_return;
305 : :
306 : : if (id_return) /* first time through? */
307 : : {
308 : : /* same buffer, just wipe contents */
309 : : resetPQExpBuffer(id_return);
310 : : }
311 : : else
312 : : {
313 : : /* new buffer */
314 : : id_return = createPQExpBuffer();
315 : : if (parallel_init_done)
316 : : TlsSetValue(tls_index, id_return);
317 : : else
318 : : s_id_return = id_return;
319 : : }
320 : :
321 : : return id_return;
322 : : }
323 : : #endif /* WIN32 */
324 : :
325 : : /*
326 : : * pg_dump and pg_restore call this to register the cleanup handler
327 : : * as soon as they've created the ArchiveHandle.
328 : : */
329 : : void
330 : 275 : on_exit_close_archive(Archive *AHX)
331 : : {
332 : 275 : shutdown_info.AHX = AHX;
333 : 275 : on_exit_nicely(archive_close_connection, &shutdown_info);
334 : 275 : }
335 : :
336 : : /*
337 : : * on_exit_nicely handler for shutting down database connections and
338 : : * worker processes cleanly.
339 : : */
340 : : static void
341 : 210 : archive_close_connection(int code, void *arg)
342 : : {
343 : 210 : ShutdownInformation *si = (ShutdownInformation *) arg;
344 : :
345 [ - + ]: 210 : if (si->pstate)
346 : : {
347 : : /* In parallel mode, must figure out who we are */
348 : 0 : ParallelSlot *slot = GetMyPSlot(si->pstate);
349 : :
350 [ # # ]: 0 : if (!slot)
351 : : {
352 : : /*
353 : : * We're the leader. Forcibly shut down workers, then close our
354 : : * own database connection, if any.
355 : : */
356 : 0 : ShutdownWorkersHard(si->pstate);
357 : :
358 [ # # ]: 0 : if (si->AHX)
359 : 0 : DisconnectDatabase(si->AHX);
360 : : }
361 : : else
362 : : {
363 : : /*
364 : : * We're a worker. Shut down our own DB connection if any. On
365 : : * Windows, we also have to close our communication sockets, to
366 : : * emulate what will happen on Unix when the worker process exits.
367 : : * (Without this, if this is a premature exit, the leader would
368 : : * fail to detect it because there would be no EOF condition on
369 : : * the other end of the pipe.)
370 : : */
371 [ # # ]: 0 : if (slot->AH)
372 : 0 : DisconnectDatabase(&(slot->AH->public));
373 : :
374 : : #ifdef WIN32
375 : : closesocket(slot->pipeRevRead);
376 : : closesocket(slot->pipeRevWrite);
377 : : #endif
378 : : }
379 : : }
380 : : else
381 : : {
382 : : /* Non-parallel operation: just kill the leader DB connection */
383 [ + - ]: 210 : if (si->AHX)
384 : 210 : DisconnectDatabase(si->AHX);
385 : : }
386 : 210 : }
387 : :
388 : : /*
389 : : * Forcibly shut down any remaining workers, waiting for them to finish.
390 : : *
391 : : * Note that we don't expect to come here during normal exit (the workers
392 : : * should be long gone, and the ParallelState too). We're only here in a
393 : : * pg_fatal() situation, so intervening to cancel active commands is
394 : : * appropriate.
395 : : */
396 : : static void
397 : 0 : ShutdownWorkersHard(ParallelState *pstate)
398 : : {
399 : : int i;
400 : :
401 : : /*
402 : : * Close our write end of the sockets so that any workers waiting for
403 : : * commands know they can exit. (Note: some of the pipeWrite fields might
404 : : * still be zero, if we failed to initialize all the workers. Hence, just
405 : : * ignore errors here.)
406 : : */
407 [ # # ]: 0 : for (i = 0; i < pstate->numWorkers; i++)
408 : 0 : closesocket(pstate->parallelSlot[i].pipeWrite);
409 : :
410 : : /*
411 : : * Force early termination of any commands currently in progress.
412 : : */
413 : : #ifndef WIN32
414 : : /* On non-Windows, send SIGTERM to each worker process. */
415 [ # # ]: 0 : for (i = 0; i < pstate->numWorkers; i++)
416 : : {
417 : 0 : pid_t pid = pstate->parallelSlot[i].pid;
418 : :
419 [ # # ]: 0 : if (pid != 0)
420 : 0 : kill(pid, SIGTERM);
421 : : }
422 : : #else
423 : :
424 : : /*
425 : : * On Windows, send query cancels directly to the workers' backends. Use
426 : : * a critical section to ensure worker threads don't change state.
427 : : */
428 : : EnterCriticalSection(&signal_info_lock);
429 : : for (i = 0; i < pstate->numWorkers; i++)
430 : : {
431 : : ArchiveHandle *AH = pstate->parallelSlot[i].AH;
432 : : char errbuf[1];
433 : :
434 : : if (AH != NULL && AH->connCancel != NULL)
435 : : (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf));
436 : : }
437 : : LeaveCriticalSection(&signal_info_lock);
438 : : #endif
439 : :
440 : : /* Now wait for them to terminate. */
441 : 0 : WaitForTerminatingWorkers(pstate);
442 : 0 : }
443 : :
444 : : /*
445 : : * Wait for all workers to terminate.
446 : : */
447 : : static void
448 : 12 : WaitForTerminatingWorkers(ParallelState *pstate)
449 : : {
450 [ + + ]: 38 : while (!HasEveryWorkerTerminated(pstate))
451 : : {
452 : 26 : ParallelSlot *slot = NULL;
453 : : int j;
454 : :
455 : : #ifndef WIN32
456 : : /* On non-Windows, use wait() to wait for next worker to end */
457 : : int status;
458 : 26 : pid_t pid = wait(&status);
459 : :
460 : : /* Find dead worker's slot, and clear the PID field */
461 [ + - ]: 42 : for (j = 0; j < pstate->numWorkers; j++)
462 : : {
463 : 42 : slot = &(pstate->parallelSlot[j]);
464 [ + + ]: 42 : if (slot->pid == pid)
465 : : {
466 : 26 : slot->pid = 0;
467 : 26 : break;
468 : : }
469 : : }
470 : : #else /* WIN32 */
471 : : /* On Windows, we must use WaitForMultipleObjects() */
472 : : HANDLE *lpHandles = pg_malloc_array(HANDLE, pstate->numWorkers);
473 : : int nrun = 0;
474 : : DWORD ret;
475 : : uintptr_t hThread;
476 : :
477 : : for (j = 0; j < pstate->numWorkers; j++)
478 : : {
479 : : if (WORKER_IS_RUNNING(pstate->parallelSlot[j].workerStatus))
480 : : {
481 : : lpHandles[nrun] = (HANDLE) pstate->parallelSlot[j].hThread;
482 : : nrun++;
483 : : }
484 : : }
485 : : ret = WaitForMultipleObjects(nrun, lpHandles, false, INFINITE);
486 : : Assert(ret != WAIT_FAILED);
487 : : hThread = (uintptr_t) lpHandles[ret - WAIT_OBJECT_0];
488 : : pg_free(lpHandles);
489 : :
490 : : /* Find dead worker's slot, and clear the hThread field */
491 : : for (j = 0; j < pstate->numWorkers; j++)
492 : : {
493 : : slot = &(pstate->parallelSlot[j]);
494 : : if (slot->hThread == hThread)
495 : : {
496 : : /* For cleanliness, close handles for dead threads */
497 : : CloseHandle((HANDLE) slot->hThread);
498 : : slot->hThread = (uintptr_t) INVALID_HANDLE_VALUE;
499 : : break;
500 : : }
501 : : }
502 : : #endif /* WIN32 */
503 : :
504 : : /* On all platforms, update workerStatus and te[] as well */
505 : : Assert(j < pstate->numWorkers);
506 : 26 : slot->workerStatus = WRKR_TERMINATED;
507 : 26 : pstate->te[j] = NULL;
508 : : }
509 : 12 : }
510 : :
511 : :
512 : : /*
513 : : * Code for responding to cancel interrupts (SIGINT, control-C, etc)
514 : : *
515 : : * This doesn't quite belong in this module, but it needs access to the
516 : : * ParallelState data, so there's not really a better place either.
517 : : *
518 : : * When we get a cancel interrupt, we could just die, but in pg_restore that
519 : : * could leave a SQL command (e.g., CREATE INDEX on a large table) running
520 : : * for a long time. Instead, we try to send a cancel request and then die.
521 : : * pg_dump probably doesn't really need this, but we might as well use it
522 : : * there too. Note that sending the cancel directly from the signal handler
523 : : * is safe because PQcancel() is written to make it so.
524 : : *
525 : : * In parallel operation on Unix, each process is responsible for canceling
526 : : * its own connection (this must be so because nobody else has access to it).
527 : : * Furthermore, the leader process should attempt to forward its signal to
528 : : * each child. In simple manual use of pg_dump/pg_restore, forwarding isn't
529 : : * needed because typing control-C at the console would deliver SIGINT to
530 : : * every member of the terminal process group --- but in other scenarios it
531 : : * might be that only the leader gets signaled.
532 : : *
533 : : * On Windows, the cancel handler runs in a separate thread, because that's
534 : : * how SetConsoleCtrlHandler works. Because the workers are threads in this
535 : : * same process, we set a flag (is_cancel_in_progress()) so they stay quiet
536 : : * about the query cancellations instead of cluttering the screen, then send
537 : : * cancels on all active connections and return FALSE, which will allow the
538 : : * process to die. For safety's sake, we use a critical section to protect
539 : : * the PGcancel structures against being changed while the signal thread runs.
540 : : */
541 : :
542 : : #ifndef WIN32
543 : :
544 : : /*
545 : : * Signal handler (Unix only)
546 : : */
547 : : static void
548 : 0 : sigTermHandler(SIGNAL_ARGS)
549 : : {
550 : : int i;
551 : : char errbuf[1];
552 : :
553 : : /*
554 : : * Some platforms allow delivery of new signals to interrupt an active
555 : : * signal handler. That could muck up our attempt to send PQcancel, so
556 : : * disable the signals that set_cancel_handler enabled.
557 : : */
558 : 0 : pqsignal(SIGINT, PG_SIG_IGN);
559 : 0 : pqsignal(SIGTERM, PG_SIG_IGN);
560 : 0 : pqsignal(SIGQUIT, PG_SIG_IGN);
561 : :
562 : : /*
563 : : * If we're in the leader, forward signal to all workers. (It seems best
564 : : * to do this before PQcancel; killing the leader transaction will result
565 : : * in invalid-snapshot errors from active workers, which maybe we can
566 : : * quiet by killing workers first.) Ignore any errors.
567 : : */
568 [ # # ]: 0 : if (signal_info.pstate != NULL)
569 : : {
570 [ # # ]: 0 : for (i = 0; i < signal_info.pstate->numWorkers; i++)
571 : : {
572 : 0 : pid_t pid = signal_info.pstate->parallelSlot[i].pid;
573 : :
574 [ # # ]: 0 : if (pid != 0)
575 : 0 : kill(pid, SIGTERM);
576 : : }
577 : : }
578 : :
579 : : /*
580 : : * Send QueryCancel if we have a connection to send to. Ignore errors,
581 : : * there's not much we can do about them anyway.
582 : : */
583 [ # # # # ]: 0 : if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL)
584 : 0 : (void) PQcancel(signal_info.myAH->connCancel, errbuf, sizeof(errbuf));
585 : :
586 : : /*
587 : : * Report we're quitting, using nothing more complicated than write(2).
588 : : * When in parallel operation, only the leader process should do this.
589 : : */
590 [ # # ]: 0 : if (!signal_info.am_worker)
591 : : {
592 [ # # ]: 0 : if (progname)
593 : : {
594 : 0 : write_stderr(progname);
595 : 0 : write_stderr(": ");
596 : : }
597 : 0 : write_stderr("terminated by user\n");
598 : : }
599 : :
600 : : /*
601 : : * And die, using _exit() not exit() because the latter will invoke atexit
602 : : * handlers that can fail if we interrupted related code.
603 : : */
604 : 0 : _exit(1);
605 : : }
606 : :
607 : : /*
608 : : * Enable cancel interrupt handler, if not already done.
609 : : */
610 : : static void
611 : 637 : set_cancel_handler(void)
612 : : {
613 : : /*
614 : : * When forking, signal_info.handler_set will propagate into the new
615 : : * process, but that's fine because the signal handler state does too.
616 : : */
617 [ + + ]: 637 : if (!signal_info.handler_set)
618 : : {
619 : 244 : signal_info.handler_set = true;
620 : :
621 : 244 : pqsignal(SIGINT, sigTermHandler);
622 : 244 : pqsignal(SIGTERM, sigTermHandler);
623 : 244 : pqsignal(SIGQUIT, sigTermHandler);
624 : : }
625 : 637 : }
626 : :
627 : : #else /* WIN32 */
628 : :
629 : : /*
630 : : * Console interrupt handler --- runs in a newly-started thread.
631 : : *
632 : : * After stopping other threads and sending cancel requests on all open
633 : : * connections, we return FALSE which will allow the default ExitProcess()
634 : : * action to be taken.
635 : : */
636 : : static BOOL WINAPI
637 : : consoleHandler(DWORD dwCtrlType)
638 : : {
639 : : int i;
640 : : char errbuf[1];
641 : :
642 : : if (dwCtrlType == CTRL_C_EVENT ||
643 : : dwCtrlType == CTRL_BREAK_EVENT)
644 : : {
645 : : /*
646 : : * Tell worker threads to stay quiet about the query cancellations
647 : : * we're about to send them; otherwise they'd report them as errors
648 : : * and clutter the user's screen. This must be set before we send any
649 : : * cancel, so that a worker is guaranteed to see it by the time its
650 : : * query fails as a result.
651 : : */
652 : : set_cancel_in_progress();
653 : :
654 : : /* Critical section prevents changing data we look at here */
655 : : EnterCriticalSection(&signal_info_lock);
656 : :
657 : : /*
658 : : * If in parallel mode, send QueryCancel to each worker's connected
659 : : * backend. Do this before canceling the main transaction, else we
660 : : * might get invalid-snapshot errors reported before we can stop the
661 : : * workers. Ignore errors, there's not much we can do about them
662 : : * anyway.
663 : : */
664 : : if (signal_info.pstate != NULL)
665 : : {
666 : : for (i = 0; i < signal_info.pstate->numWorkers; i++)
667 : : {
668 : : ArchiveHandle *AH = signal_info.pstate->parallelSlot[i].AH;
669 : :
670 : : if (AH != NULL && AH->connCancel != NULL)
671 : : (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf));
672 : : }
673 : : }
674 : :
675 : : /*
676 : : * Send QueryCancel to leader connection, if enabled. Ignore errors,
677 : : * there's not much we can do about them anyway.
678 : : */
679 : : if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL)
680 : : (void) PQcancel(signal_info.myAH->connCancel,
681 : : errbuf, sizeof(errbuf));
682 : :
683 : : LeaveCriticalSection(&signal_info_lock);
684 : :
685 : : /*
686 : : * Report we're quitting, using nothing more complicated than
687 : : * write(2). We should be able to use pg_log_*() here, but for now we
688 : : * stay aligned with the sigTermHandler behavior.
689 : : */
690 : : if (progname)
691 : : {
692 : : write_stderr(progname);
693 : : write_stderr(": ");
694 : : }
695 : : write_stderr("terminated by user\n");
696 : : }
697 : :
698 : : /* Always return FALSE to allow signal handling to continue */
699 : : return FALSE;
700 : : }
701 : :
702 : : /*
703 : : * Enable cancel interrupt handler, if not already done.
704 : : */
705 : : static void
706 : : set_cancel_handler(void)
707 : : {
708 : : if (!signal_info.handler_set)
709 : : {
710 : : signal_info.handler_set = true;
711 : :
712 : : InitializeCriticalSection(&signal_info_lock);
713 : :
714 : : SetConsoleCtrlHandler(consoleHandler, TRUE);
715 : : }
716 : : }
717 : :
718 : : #endif /* WIN32 */
719 : :
720 : :
721 : : /*
722 : : * set_archive_cancel_info
723 : : *
724 : : * Fill AH->connCancel with cancellation info for the specified database
725 : : * connection; or clear it if conn is NULL.
726 : : */
727 : : void
728 : 637 : set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn)
729 : : {
730 : : PGcancel *oldConnCancel;
731 : :
732 : : /*
733 : : * Activate the interrupt handler if we didn't yet in this process. On
734 : : * Windows, this also initializes signal_info_lock; therefore it's
735 : : * important that this happen at least once before we fork off any
736 : : * threads.
737 : : */
738 : 637 : set_cancel_handler();
739 : :
740 : : /*
741 : : * On Unix, we assume that storing a pointer value is atomic with respect
742 : : * to any possible signal interrupt. On Windows, use a critical section.
743 : : */
744 : :
745 : : #ifdef WIN32
746 : : EnterCriticalSection(&signal_info_lock);
747 : : #endif
748 : :
749 : : /* Free the old one if we have one */
750 : 637 : oldConnCancel = AH->connCancel;
751 : : /* be sure interrupt handler doesn't use pointer while freeing */
752 : 637 : AH->connCancel = NULL;
753 : :
754 [ + + ]: 637 : if (oldConnCancel != NULL)
755 : 347 : PQfreeCancel(oldConnCancel);
756 : :
757 : : /* Set the new one if specified */
758 [ + + ]: 637 : if (conn)
759 : 347 : AH->connCancel = PQgetCancel(conn);
760 : :
761 : : /*
762 : : * On Unix, there's only ever one active ArchiveHandle per process, so we
763 : : * can just set signal_info.myAH unconditionally. On Windows, do that
764 : : * only in the main thread; worker threads have to make sure their
765 : : * ArchiveHandle appears in the pstate data, which is dealt with in
766 : : * RunWorker().
767 : : */
768 : : #ifndef WIN32
769 : 637 : signal_info.myAH = AH;
770 : : #else
771 : : if (mainThreadId == GetCurrentThreadId())
772 : : signal_info.myAH = AH;
773 : : #endif
774 : :
775 : : #ifdef WIN32
776 : : LeaveCriticalSection(&signal_info_lock);
777 : : #endif
778 : 637 : }
779 : :
780 : : /*
781 : : * set_cancel_pstate
782 : : *
783 : : * Set signal_info.pstate to point to the specified ParallelState, if any.
784 : : * We need this mainly to have an interlock against Windows signal thread.
785 : : */
786 : : static void
787 : 24 : set_cancel_pstate(ParallelState *pstate)
788 : : {
789 : : #ifdef WIN32
790 : : EnterCriticalSection(&signal_info_lock);
791 : : #endif
792 : :
793 : 24 : signal_info.pstate = pstate;
794 : :
795 : : #ifdef WIN32
796 : : LeaveCriticalSection(&signal_info_lock);
797 : : #endif
798 : 24 : }
799 : :
800 : : /*
801 : : * set_cancel_slot_archive
802 : : *
803 : : * Set ParallelSlot's AH field to point to the specified archive, if any.
804 : : * We need this mainly to have an interlock against Windows signal thread.
805 : : */
806 : : static void
807 : 52 : set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH)
808 : : {
809 : : #ifdef WIN32
810 : : EnterCriticalSection(&signal_info_lock);
811 : : #endif
812 : :
813 : 52 : slot->AH = AH;
814 : :
815 : : #ifdef WIN32
816 : : LeaveCriticalSection(&signal_info_lock);
817 : : #endif
818 : 52 : }
819 : :
820 : :
821 : : /*
822 : : * This function is called by both Unix and Windows variants to set up
823 : : * and run a worker process. Caller should exit the process (or thread)
824 : : * upon return.
825 : : */
826 : : static void
827 : 26 : RunWorker(ArchiveHandle *AH, ParallelSlot *slot)
828 : : {
829 : : int pipefd[2];
830 : :
831 : : /* fetch child ends of pipes */
832 : 26 : pipefd[PIPE_READ] = slot->pipeRevRead;
833 : 26 : pipefd[PIPE_WRITE] = slot->pipeRevWrite;
834 : :
835 : : /*
836 : : * Clone the archive so that we have our own state to work with, and in
837 : : * particular our own database connection.
838 : : *
839 : : * We clone on Unix as well as Windows, even though technically we don't
840 : : * need to because fork() gives us a copy in our own address space
841 : : * already. But CloneArchive resets the state information and also clones
842 : : * the database connection which both seem kinda helpful.
843 : : */
844 : 26 : AH = CloneArchive(AH);
845 : :
846 : : /* Remember cloned archive where signal handler can find it */
847 : 26 : set_cancel_slot_archive(slot, AH);
848 : :
849 : : /*
850 : : * Call the setup worker function that's defined in the ArchiveHandle.
851 : : */
852 : 26 : (AH->SetupWorkerPtr) ((Archive *) AH);
853 : :
854 : : /*
855 : : * Execute commands until done.
856 : : */
857 : 26 : WaitForCommands(AH, pipefd);
858 : :
859 : : /*
860 : : * Disconnect from database and clean up.
861 : : */
862 : 26 : set_cancel_slot_archive(slot, NULL);
863 : 26 : DisconnectDatabase(&(AH->public));
864 : 26 : DeCloneArchive(AH);
865 : 26 : }
866 : :
867 : : /*
868 : : * Thread base function for Windows
869 : : */
870 : : #ifdef WIN32
871 : : static unsigned __stdcall
872 : : init_spawned_worker_win32(WorkerInfo *wi)
873 : : {
874 : : ArchiveHandle *AH = wi->AH;
875 : : ParallelSlot *slot = wi->slot;
876 : :
877 : : /* Don't need WorkerInfo anymore */
878 : : free(wi);
879 : :
880 : : /* Run the worker ... */
881 : : RunWorker(AH, slot);
882 : :
883 : : /* Exit the thread */
884 : : _endthreadex(0);
885 : : return 0;
886 : : }
887 : : #endif /* WIN32 */
888 : :
889 : : /*
890 : : * This function starts a parallel dump or restore by spawning off the worker
891 : : * processes. For Windows, it creates a number of threads; on Unix the
892 : : * workers are created with fork().
893 : : */
894 : : ParallelState *
895 : 14 : ParallelBackupStart(ArchiveHandle *AH)
896 : : {
897 : : ParallelState *pstate;
898 : : int i;
899 : :
900 : : Assert(AH->public.numWorkers > 0);
901 : :
902 : 14 : pstate = pg_malloc_object(ParallelState);
903 : :
904 : 14 : pstate->numWorkers = AH->public.numWorkers;
905 : 14 : pstate->te = NULL;
906 : 14 : pstate->parallelSlot = NULL;
907 : :
908 [ + + ]: 14 : if (AH->public.numWorkers == 1)
909 : 2 : return pstate;
910 : :
911 : : /* Create status arrays, being sure to initialize all fields to 0 */
912 : 12 : pstate->te =
913 : 12 : pg_malloc0_array(TocEntry *, pstate->numWorkers);
914 : 12 : pstate->parallelSlot =
915 : 12 : pg_malloc0_array(ParallelSlot, pstate->numWorkers);
916 : :
917 : : #ifdef WIN32
918 : : /* Make fmtId() and fmtQualifiedId() use thread-local storage */
919 : : getLocalPQExpBuffer = getThreadLocalPQExpBuffer;
920 : : #endif
921 : :
922 : : /*
923 : : * Set the pstate in shutdown_info, to tell the exit handler that it must
924 : : * clean up workers as well as the main database connection. But we don't
925 : : * set this in signal_info yet, because we don't want child processes to
926 : : * inherit non-NULL signal_info.pstate.
927 : : */
928 : 12 : shutdown_info.pstate = pstate;
929 : :
930 : : /*
931 : : * Temporarily disable query cancellation on the leader connection. This
932 : : * ensures that child processes won't inherit valid AH->connCancel
933 : : * settings and thus won't try to issue cancels against the leader's
934 : : * connection. No harm is done if we fail while it's disabled, because
935 : : * the leader connection is idle at this point anyway.
936 : : */
937 : 12 : set_archive_cancel_info(AH, NULL);
938 : :
939 : : /* Ensure stdio state is quiesced before forking */
940 : 12 : fflush(NULL);
941 : :
942 : : /* Create desired number of workers */
943 [ + + ]: 38 : for (i = 0; i < pstate->numWorkers; i++)
944 : : {
945 : : #ifdef WIN32
946 : : WorkerInfo *wi;
947 : : uintptr_t handle;
948 : : #else
949 : : pid_t pid;
950 : : #endif
951 : 26 : ParallelSlot *slot = &(pstate->parallelSlot[i]);
952 : : int pipeMW[2],
953 : : pipeWM[2];
954 : :
955 : : /* Create communication pipes for this worker */
956 [ + - - + ]: 26 : if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0)
957 : 0 : pg_fatal("could not create communication channels: %m");
958 : :
959 : : /* leader's ends of the pipes */
960 : 26 : slot->pipeRead = pipeWM[PIPE_READ];
961 : 26 : slot->pipeWrite = pipeMW[PIPE_WRITE];
962 : : /* child's ends of the pipes */
963 : 26 : slot->pipeRevRead = pipeMW[PIPE_READ];
964 : 26 : slot->pipeRevWrite = pipeWM[PIPE_WRITE];
965 : :
966 : : #ifdef WIN32
967 : : /* Create transient structure to pass args to worker function */
968 : : wi = pg_malloc_object(WorkerInfo);
969 : :
970 : : wi->AH = AH;
971 : : wi->slot = slot;
972 : :
973 : : handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32,
974 : : wi, 0, &(slot->threadId));
975 : : if (handle == 0)
976 : : pg_fatal("could not create worker thread: %m");
977 : : slot->hThread = handle;
978 : : slot->workerStatus = WRKR_IDLE;
979 : : #else /* !WIN32 */
980 : 26 : pid = fork();
981 [ + + ]: 52 : if (pid == 0)
982 : : {
983 : : /* we are the worker */
984 : : int j;
985 : :
986 : : /* this is needed for GetMyPSlot() */
987 : 26 : slot->pid = getpid();
988 : :
989 : : /* instruct signal handler that we're in a worker now */
990 : 26 : signal_info.am_worker = true;
991 : :
992 : : /* close read end of Worker -> Leader */
993 : 26 : closesocket(pipeWM[PIPE_READ]);
994 : : /* close write end of Leader -> Worker */
995 : 26 : closesocket(pipeMW[PIPE_WRITE]);
996 : :
997 : : /*
998 : : * Close all inherited fds for communication of the leader with
999 : : * previously-forked workers.
1000 : : */
1001 [ + + ]: 42 : for (j = 0; j < i; j++)
1002 : : {
1003 : 16 : closesocket(pstate->parallelSlot[j].pipeRead);
1004 : 16 : closesocket(pstate->parallelSlot[j].pipeWrite);
1005 : : }
1006 : :
1007 : : /* Run the worker ... */
1008 : 26 : RunWorker(AH, slot);
1009 : :
1010 : : /* We can just exit(0) when done */
1011 : 26 : exit(0);
1012 : : }
1013 [ - + ]: 26 : else if (pid < 0)
1014 : : {
1015 : : /* fork failed */
1016 : 0 : pg_fatal("could not create worker process: %m");
1017 : : }
1018 : :
1019 : : /* In Leader after successful fork */
1020 : 26 : slot->pid = pid;
1021 : 26 : slot->workerStatus = WRKR_IDLE;
1022 : :
1023 : : /* close read end of Leader -> Worker */
1024 : 26 : closesocket(pipeMW[PIPE_READ]);
1025 : : /* close write end of Worker -> Leader */
1026 : 26 : closesocket(pipeWM[PIPE_WRITE]);
1027 : : #endif /* WIN32 */
1028 : : }
1029 : :
1030 : : /*
1031 : : * Having forked off the workers, disable SIGPIPE so that leader isn't
1032 : : * killed if it tries to send a command to a dead worker. We don't want
1033 : : * the workers to inherit this setting, though.
1034 : : */
1035 : : #ifndef WIN32
1036 : 12 : pqsignal(SIGPIPE, PG_SIG_IGN);
1037 : : #endif
1038 : :
1039 : : /*
1040 : : * Re-establish query cancellation on the leader connection.
1041 : : */
1042 : 12 : set_archive_cancel_info(AH, AH->connection);
1043 : :
1044 : : /*
1045 : : * Tell the cancel signal handler to forward signals to worker processes,
1046 : : * too. (As with query cancel, we did not need this earlier because the
1047 : : * workers have not yet been given anything to do; if we die before this
1048 : : * point, any already-started workers will see EOF and quit promptly.)
1049 : : */
1050 : 12 : set_cancel_pstate(pstate);
1051 : :
1052 : 12 : return pstate;
1053 : : }
1054 : :
1055 : : /*
1056 : : * Close down a parallel dump or restore.
1057 : : */
1058 : : void
1059 : 14 : ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate)
1060 : : {
1061 : : int i;
1062 : :
1063 : : /* No work if non-parallel */
1064 [ + + ]: 14 : if (pstate->numWorkers == 1)
1065 : 2 : return;
1066 : :
1067 : : /* There should not be any unfinished jobs */
1068 : : Assert(IsEveryWorkerIdle(pstate));
1069 : :
1070 : : /* Close the sockets so that the workers know they can exit */
1071 [ + + ]: 38 : for (i = 0; i < pstate->numWorkers; i++)
1072 : : {
1073 : 26 : closesocket(pstate->parallelSlot[i].pipeRead);
1074 : 26 : closesocket(pstate->parallelSlot[i].pipeWrite);
1075 : : }
1076 : :
1077 : : /* Wait for them to exit */
1078 : 12 : WaitForTerminatingWorkers(pstate);
1079 : :
1080 : : /*
1081 : : * Unlink pstate from shutdown_info, so the exit handler will not try to
1082 : : * use it; and likewise unlink from signal_info.
1083 : : */
1084 : 12 : shutdown_info.pstate = NULL;
1085 : 12 : set_cancel_pstate(NULL);
1086 : :
1087 : : /* Release state (mere neatnik-ism, since we're about to terminate) */
1088 : 12 : free(pstate->te);
1089 : 12 : free(pstate->parallelSlot);
1090 : 12 : free(pstate);
1091 : : }
1092 : :
1093 : : /*
1094 : : * These next four functions handle construction and parsing of the command
1095 : : * strings and response strings for parallel workers.
1096 : : *
1097 : : * Currently, these can be the same regardless of which archive format we are
1098 : : * processing. In future, we might want to let format modules override these
1099 : : * functions to add format-specific data to a command or response.
1100 : : */
1101 : :
1102 : : /*
1103 : : * buildWorkerCommand: format a command string to send to a worker.
1104 : : *
1105 : : * The string is built in the caller-supplied buffer of size buflen.
1106 : : */
1107 : : static void
1108 : 115 : buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act,
1109 : : char *buf, int buflen)
1110 : : {
1111 [ + + ]: 115 : if (act == ACT_DUMP)
1112 : 69 : snprintf(buf, buflen, "DUMP %d", te->dumpId);
1113 [ + - ]: 46 : else if (act == ACT_RESTORE)
1114 : 46 : snprintf(buf, buflen, "RESTORE %d", te->dumpId);
1115 : : else
1116 : : Assert(false);
1117 : 115 : }
1118 : :
1119 : : /*
1120 : : * parseWorkerCommand: interpret a command string in a worker.
1121 : : */
1122 : : static void
1123 : 115 : parseWorkerCommand(ArchiveHandle *AH, TocEntry **te, T_Action *act,
1124 : : const char *msg)
1125 : : {
1126 : : DumpId dumpId;
1127 : : int nBytes;
1128 : :
1129 [ + + ]: 115 : if (messageStartsWith(msg, "DUMP "))
1130 : : {
1131 : 69 : *act = ACT_DUMP;
1132 : 69 : sscanf(msg, "DUMP %d%n", &dumpId, &nBytes);
1133 : : Assert(nBytes == strlen(msg));
1134 : 69 : *te = getTocEntryByDumpId(AH, dumpId);
1135 : : Assert(*te != NULL);
1136 : : }
1137 [ + - ]: 46 : else if (messageStartsWith(msg, "RESTORE "))
1138 : : {
1139 : 46 : *act = ACT_RESTORE;
1140 : 46 : sscanf(msg, "RESTORE %d%n", &dumpId, &nBytes);
1141 : : Assert(nBytes == strlen(msg));
1142 : 46 : *te = getTocEntryByDumpId(AH, dumpId);
1143 : : Assert(*te != NULL);
1144 : : }
1145 : : else
1146 : 0 : pg_fatal("unrecognized command received from leader: \"%s\"",
1147 : : msg);
1148 : 115 : }
1149 : :
1150 : : /*
1151 : : * buildWorkerResponse: format a response string to send to the leader.
1152 : : *
1153 : : * The string is built in the caller-supplied buffer of size buflen.
1154 : : */
1155 : : static void
1156 : 115 : buildWorkerResponse(ArchiveHandle *AH, TocEntry *te, T_Action act, int status,
1157 : : char *buf, int buflen)
1158 : : {
1159 [ - + ]: 115 : snprintf(buf, buflen, "OK %d %d %d",
1160 : : te->dumpId,
1161 : : status,
1162 : : status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0);
1163 : 115 : }
1164 : :
1165 : : /*
1166 : : * parseWorkerResponse: parse the status message returned by a worker.
1167 : : *
1168 : : * Returns the integer status code, and may update fields of AH and/or te.
1169 : : */
1170 : : static int
1171 : 115 : parseWorkerResponse(ArchiveHandle *AH, TocEntry *te,
1172 : : const char *msg)
1173 : : {
1174 : : DumpId dumpId;
1175 : : int nBytes,
1176 : : n_errors;
1177 : 115 : int status = 0;
1178 : :
1179 [ + - ]: 115 : if (messageStartsWith(msg, "OK "))
1180 : : {
1181 : 115 : sscanf(msg, "OK %d %d %d%n", &dumpId, &status, &n_errors, &nBytes);
1182 : :
1183 : : Assert(dumpId == te->dumpId);
1184 : : Assert(nBytes == strlen(msg));
1185 : :
1186 : 115 : AH->public.n_errors += n_errors;
1187 : : }
1188 : : else
1189 : 0 : pg_fatal("invalid message received from worker: \"%s\"",
1190 : : msg);
1191 : :
1192 : 115 : return status;
1193 : : }
1194 : :
1195 : : /*
1196 : : * Dispatch a job to some free worker.
1197 : : *
1198 : : * te is the TocEntry to be processed, act is the action to be taken on it.
1199 : : * callback is the function to call on completion of the job.
1200 : : *
1201 : : * If no worker is currently available, this will block, and previously
1202 : : * registered callback functions may be called.
1203 : : */
1204 : : void
1205 : 115 : DispatchJobForTocEntry(ArchiveHandle *AH,
1206 : : ParallelState *pstate,
1207 : : TocEntry *te,
1208 : : T_Action act,
1209 : : ParallelCompletionPtr callback,
1210 : : void *callback_data)
1211 : : {
1212 : : int worker;
1213 : : char buf[256];
1214 : :
1215 : : /* Get a worker, waiting if none are idle */
1216 [ + + ]: 169 : while ((worker = GetIdleWorker(pstate)) == NO_SLOT)
1217 : 54 : WaitForWorkers(AH, pstate, WFW_ONE_IDLE);
1218 : :
1219 : : /* Construct and send command string */
1220 : 115 : buildWorkerCommand(AH, te, act, buf, sizeof(buf));
1221 : :
1222 : 115 : sendMessageToWorker(pstate, worker, buf);
1223 : :
1224 : : /* Remember worker is busy, and which TocEntry it's working on */
1225 : 115 : pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
1226 : 115 : pstate->parallelSlot[worker].callback = callback;
1227 : 115 : pstate->parallelSlot[worker].callback_data = callback_data;
1228 : 115 : pstate->te[worker] = te;
1229 : 115 : }
1230 : :
1231 : : /*
1232 : : * Find an idle worker and return its slot number.
1233 : : * Return NO_SLOT if none are idle.
1234 : : */
1235 : : static int
1236 : 244 : GetIdleWorker(ParallelState *pstate)
1237 : : {
1238 : : int i;
1239 : :
1240 [ + + ]: 609 : for (i = 0; i < pstate->numWorkers; i++)
1241 : : {
1242 [ + + ]: 489 : if (pstate->parallelSlot[i].workerStatus == WRKR_IDLE)
1243 : 124 : return i;
1244 : : }
1245 : 120 : return NO_SLOT;
1246 : : }
1247 : :
1248 : : /*
1249 : : * Return true iff no worker is running.
1250 : : */
1251 : : static bool
1252 : 38 : HasEveryWorkerTerminated(ParallelState *pstate)
1253 : : {
1254 : : int i;
1255 : :
1256 [ + + ]: 74 : for (i = 0; i < pstate->numWorkers; i++)
1257 : : {
1258 [ + + - + ]: 62 : if (WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
1259 : 26 : return false;
1260 : : }
1261 : 12 : return true;
1262 : : }
1263 : :
1264 : : /*
1265 : : * Return true iff every worker is in the WRKR_IDLE state.
1266 : : */
1267 : : bool
1268 : 43 : IsEveryWorkerIdle(ParallelState *pstate)
1269 : : {
1270 : : int i;
1271 : :
1272 [ + + ]: 96 : for (i = 0; i < pstate->numWorkers; i++)
1273 : : {
1274 [ + + ]: 76 : if (pstate->parallelSlot[i].workerStatus != WRKR_IDLE)
1275 : 23 : return false;
1276 : : }
1277 : 20 : return true;
1278 : : }
1279 : :
1280 : : /*
1281 : : * Acquire lock on a table to be dumped by a worker process.
1282 : : *
1283 : : * The leader process is already holding an ACCESS SHARE lock. Ordinarily
1284 : : * it's no problem for a worker to get one too, but if anything else besides
1285 : : * pg_dump is running, there's a possible deadlock:
1286 : : *
1287 : : * 1) Leader dumps the schema and locks all tables in ACCESS SHARE mode.
1288 : : * 2) Another process requests an ACCESS EXCLUSIVE lock (which is not granted
1289 : : * because the leader holds a conflicting ACCESS SHARE lock).
1290 : : * 3) A worker process also requests an ACCESS SHARE lock to read the table.
1291 : : * The worker is enqueued behind the ACCESS EXCLUSIVE lock request.
1292 : : * 4) Now we have a deadlock, since the leader is effectively waiting for
1293 : : * the worker. The server cannot detect that, however.
1294 : : *
1295 : : * To prevent an infinite wait, prior to touching a table in a worker, request
1296 : : * a lock in ACCESS SHARE mode but with NOWAIT. If we don't get the lock,
1297 : : * then we know that somebody else has requested an ACCESS EXCLUSIVE lock and
1298 : : * so we have a deadlock. We must fail the backup in that case.
1299 : : */
1300 : : static void
1301 : 69 : lockTableForWorker(ArchiveHandle *AH, TocEntry *te)
1302 : : {
1303 : : const char *qualId;
1304 : : PQExpBuffer query;
1305 : : PGresult *res;
1306 : :
1307 : : /* Nothing to do for BLOBS */
1308 [ + + ]: 69 : if (strcmp(te->desc, "BLOBS") == 0)
1309 : 4 : return;
1310 : :
1311 : 65 : query = createPQExpBuffer();
1312 : :
1313 : 65 : qualId = fmtQualifiedId(te->namespace, te->tag);
1314 : :
1315 : 65 : appendPQExpBuffer(query, "LOCK TABLE %s IN ACCESS SHARE MODE NOWAIT",
1316 : : qualId);
1317 : :
1318 : 65 : res = PQexec(AH->connection, query->data);
1319 : :
1320 [ + - - + ]: 65 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
1321 : 0 : pg_fatal("could not obtain lock on relation \"%s\"\n"
1322 : : "This usually means that someone requested an ACCESS EXCLUSIVE lock "
1323 : : "on the table after the pg_dump parent process had gotten the "
1324 : : "initial ACCESS SHARE lock on the table.", qualId);
1325 : :
1326 : 65 : PQclear(res);
1327 : 65 : destroyPQExpBuffer(query);
1328 : : }
1329 : :
1330 : : /*
1331 : : * WaitForCommands: main routine for a worker process.
1332 : : *
1333 : : * Read and execute commands from the leader until we see EOF on the pipe.
1334 : : */
1335 : : static void
1336 : 26 : WaitForCommands(ArchiveHandle *AH, int pipefd[2])
1337 : : {
1338 : : char *command;
1339 : : TocEntry *te;
1340 : : T_Action act;
1341 : 26 : int status = 0;
1342 : : char buf[256];
1343 : :
1344 : : for (;;)
1345 : : {
1346 [ + + ]: 141 : if (!(command = getMessageFromLeader(pipefd)))
1347 : : {
1348 : : /* EOF, so done */
1349 : 26 : return;
1350 : : }
1351 : :
1352 : : /* Decode the command */
1353 : 115 : parseWorkerCommand(AH, &te, &act, command);
1354 : :
1355 [ + + ]: 115 : if (act == ACT_DUMP)
1356 : : {
1357 : : /* Acquire lock on this table within the worker's session */
1358 : 69 : lockTableForWorker(AH, te);
1359 : :
1360 : : /* Perform the dump command */
1361 : 69 : status = (AH->WorkerJobDumpPtr) (AH, te);
1362 : : }
1363 [ + - ]: 46 : else if (act == ACT_RESTORE)
1364 : : {
1365 : : /* Perform the restore command */
1366 : 46 : status = (AH->WorkerJobRestorePtr) (AH, te);
1367 : : }
1368 : : else
1369 : : Assert(false);
1370 : :
1371 : : /* Return status to leader */
1372 : 115 : buildWorkerResponse(AH, te, act, status, buf, sizeof(buf));
1373 : :
1374 : 115 : sendMessageToLeader(pipefd, buf);
1375 : :
1376 : : /* command was pg_malloc'd and we are responsible for free()ing it. */
1377 : 115 : free(command);
1378 : : }
1379 : : }
1380 : :
1381 : : /*
1382 : : * Check for status messages from workers.
1383 : : *
1384 : : * If do_wait is true, wait to get a status message; otherwise, just return
1385 : : * immediately if there is none available.
1386 : : *
1387 : : * When we get a status message, we pass the status code to the callback
1388 : : * function that was specified to DispatchJobForTocEntry, then reset the
1389 : : * worker status to IDLE.
1390 : : *
1391 : : * Returns true if we collected a status message, else false.
1392 : : *
1393 : : * XXX is it worth checking for more than one status message per call?
1394 : : * It seems somewhat unlikely that multiple workers would finish at exactly
1395 : : * the same time.
1396 : : */
1397 : : static bool
1398 : 197 : ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
1399 : : {
1400 : : int worker;
1401 : : char *msg;
1402 : :
1403 : : /* Try to collect a status message */
1404 : 197 : msg = getMessageFromWorker(pstate, do_wait, &worker);
1405 : :
1406 [ + + ]: 197 : if (!msg)
1407 : : {
1408 : : /* If do_wait is true, we must have detected EOF on some socket */
1409 [ - + ]: 82 : if (do_wait)
1410 : 0 : pg_fatal("a worker process died unexpectedly");
1411 : 82 : return false;
1412 : : }
1413 : :
1414 : : /* Process it and update our idea of the worker's status */
1415 [ + - ]: 115 : if (messageStartsWith(msg, "OK "))
1416 : : {
1417 : 115 : ParallelSlot *slot = &pstate->parallelSlot[worker];
1418 : 115 : TocEntry *te = pstate->te[worker];
1419 : : int status;
1420 : :
1421 : 115 : status = parseWorkerResponse(AH, te, msg);
1422 : 115 : slot->callback(AH, te, status, slot->callback_data);
1423 : 115 : slot->workerStatus = WRKR_IDLE;
1424 : 115 : pstate->te[worker] = NULL;
1425 : : }
1426 : : else
1427 : 0 : pg_fatal("invalid message received from worker: \"%s\"",
1428 : : msg);
1429 : :
1430 : : /* Free the string returned from getMessageFromWorker */
1431 : 115 : free(msg);
1432 : :
1433 : 115 : return true;
1434 : : }
1435 : :
1436 : : /*
1437 : : * Check for status results from workers, waiting if necessary.
1438 : : *
1439 : : * Available wait modes are:
1440 : : * WFW_NO_WAIT: reap any available status, but don't block
1441 : : * WFW_GOT_STATUS: wait for at least one more worker to finish
1442 : : * WFW_ONE_IDLE: wait for at least one worker to be idle
1443 : : * WFW_ALL_IDLE: wait for all workers to be idle
1444 : : *
1445 : : * Any received results are passed to the callback specified to
1446 : : * DispatchJobForTocEntry.
1447 : : *
1448 : : * This function is executed in the leader process.
1449 : : */
1450 : : void
1451 : 117 : WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
1452 : : {
1453 : 117 : bool do_wait = false;
1454 : :
1455 : : /*
1456 : : * In GOT_STATUS mode, always block waiting for a message, since we can't
1457 : : * return till we get something. In other modes, we don't block the first
1458 : : * time through the loop.
1459 : : */
1460 [ + + ]: 117 : if (mode == WFW_GOT_STATUS)
1461 : : {
1462 : : /* Assert that caller knows what it's doing */
1463 : : Assert(!IsEveryWorkerIdle(pstate));
1464 : 9 : do_wait = true;
1465 : : }
1466 : :
1467 : : for (;;)
1468 : : {
1469 : : /*
1470 : : * Check for status messages, even if we don't need to block. We do
1471 : : * not try very hard to reap all available messages, though, since
1472 : : * there's unlikely to be more than one.
1473 : : */
1474 [ + + ]: 197 : if (ListenToWorkers(AH, pstate, do_wait))
1475 : : {
1476 : : /*
1477 : : * If we got a message, we are done by definition for GOT_STATUS
1478 : : * mode, and we can also be certain that there's at least one idle
1479 : : * worker. So we're done in all but ALL_IDLE mode.
1480 : : */
1481 [ + + ]: 115 : if (mode != WFW_ALL_IDLE)
1482 : 100 : return;
1483 : : }
1484 : :
1485 : : /* Check whether we must wait for new status messages */
1486 [ - - + + : 97 : switch (mode)
- ]
1487 : : {
1488 : 0 : case WFW_NO_WAIT:
1489 : 0 : return; /* never wait */
1490 : 0 : case WFW_GOT_STATUS:
1491 : : Assert(false); /* can't get here, because we waited */
1492 : 0 : break;
1493 : 75 : case WFW_ONE_IDLE:
1494 [ + + ]: 75 : if (GetIdleWorker(pstate) != NO_SLOT)
1495 : 9 : return;
1496 : 66 : break;
1497 : 22 : case WFW_ALL_IDLE:
1498 [ + + ]: 22 : if (IsEveryWorkerIdle(pstate))
1499 : 8 : return;
1500 : 14 : break;
1501 : : }
1502 : :
1503 : : /* Loop back, and this time wait for something to happen */
1504 : 80 : do_wait = true;
1505 : : }
1506 : : }
1507 : :
1508 : : /*
1509 : : * Read one command message from the leader, blocking if necessary
1510 : : * until one is available, and return it as a malloc'd string.
1511 : : * On EOF, return NULL.
1512 : : *
1513 : : * This function is executed in worker processes.
1514 : : */
1515 : : static char *
1516 : 141 : getMessageFromLeader(int pipefd[2])
1517 : : {
1518 : 141 : return readMessageFromPipe(pipefd[PIPE_READ]);
1519 : : }
1520 : :
1521 : : /*
1522 : : * Send a status message to the leader.
1523 : : *
1524 : : * This function is executed in worker processes.
1525 : : */
1526 : : static void
1527 : 115 : sendMessageToLeader(int pipefd[2], const char *str)
1528 : : {
1529 : 115 : int len = strlen(str) + 1;
1530 : :
1531 [ - + ]: 115 : if (pipewrite(pipefd[PIPE_WRITE], str, len) != len)
1532 : 0 : pg_fatal("could not write to the communication channel: %m");
1533 : 115 : }
1534 : :
1535 : : /*
1536 : : * Wait until some descriptor in "workerset" becomes readable.
1537 : : * Returns -1 on error, else the number of readable descriptors.
1538 : : */
1539 : : static int
1540 : 89 : select_loop(int maxFd, fd_set *workerset)
1541 : : {
1542 : : int i;
1543 : 89 : fd_set saveSet = *workerset;
1544 : :
1545 : : for (;;)
1546 : : {
1547 : 89 : *workerset = saveSet;
1548 : 89 : i = select(maxFd + 1, workerset, NULL, NULL, NULL);
1549 : :
1550 : : #ifndef WIN32
1551 [ - + - - ]: 89 : if (i < 0 && errno == EINTR)
1552 : 0 : continue;
1553 : : #else
1554 : : if (i == SOCKET_ERROR && WSAGetLastError() == WSAEINTR)
1555 : : continue;
1556 : : #endif
1557 : 89 : break;
1558 : : }
1559 : :
1560 : 89 : return i;
1561 : : }
1562 : :
1563 : :
1564 : : /*
1565 : : * Check for messages from worker processes.
1566 : : *
1567 : : * If a message is available, return it as a malloc'd string, and put the
1568 : : * index of the sending worker in *worker.
1569 : : *
1570 : : * If nothing is available, wait if "do_wait" is true, else return NULL.
1571 : : *
1572 : : * If we detect EOF on any socket, we'll return NULL. It's not great that
1573 : : * that's hard to distinguish from the no-data-available case, but for now
1574 : : * our one caller is okay with that.
1575 : : *
1576 : : * This function is executed in the leader process.
1577 : : */
1578 : : static char *
1579 : 197 : getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
1580 : : {
1581 : : int i;
1582 : : fd_set workerset;
1583 : 197 : int maxFd = -1;
1584 : 197 : struct timeval nowait = {0, 0};
1585 : :
1586 : : /* construct bitmap of socket descriptors for select() */
1587 [ + + ]: 3349 : FD_ZERO(&workerset);
1588 [ + + ]: 669 : for (i = 0; i < pstate->numWorkers; i++)
1589 : : {
1590 [ + + - + ]: 472 : if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
1591 : 0 : continue;
1592 : 472 : FD_SET(pstate->parallelSlot[i].pipeRead, &workerset);
1593 [ + - ]: 472 : if (pstate->parallelSlot[i].pipeRead > maxFd)
1594 : 472 : maxFd = pstate->parallelSlot[i].pipeRead;
1595 : : }
1596 : :
1597 [ + + ]: 197 : if (do_wait)
1598 : : {
1599 : 89 : i = select_loop(maxFd, &workerset);
1600 : : Assert(i != 0);
1601 : : }
1602 : : else
1603 : : {
1604 [ + + ]: 108 : if ((i = select(maxFd + 1, &workerset, NULL, NULL, &nowait)) == 0)
1605 : 82 : return NULL;
1606 : : }
1607 : :
1608 [ - + ]: 115 : if (i < 0)
1609 : 0 : pg_fatal("%s() failed: %m", "select");
1610 : :
1611 [ + - ]: 200 : for (i = 0; i < pstate->numWorkers; i++)
1612 : : {
1613 : : char *msg;
1614 : :
1615 [ + + - + ]: 200 : if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
1616 : 0 : continue;
1617 [ + + ]: 200 : if (!FD_ISSET(pstate->parallelSlot[i].pipeRead, &workerset))
1618 : 85 : continue;
1619 : :
1620 : : /*
1621 : : * Read the message if any. If the socket is ready because of EOF,
1622 : : * we'll return NULL instead (and the socket will stay ready, so the
1623 : : * condition will persist).
1624 : : *
1625 : : * Note: because this is a blocking read, we'll wait if only part of
1626 : : * the message is available. Waiting a long time would be bad, but
1627 : : * since worker status messages are short and are always sent in one
1628 : : * operation, it shouldn't be a problem in practice.
1629 : : */
1630 : 115 : msg = readMessageFromPipe(pstate->parallelSlot[i].pipeRead);
1631 : 115 : *worker = i;
1632 : 115 : return msg;
1633 : : }
1634 : : Assert(false);
1635 : 0 : return NULL;
1636 : : }
1637 : :
1638 : : /*
1639 : : * Send a command message to the specified worker process.
1640 : : *
1641 : : * This function is executed in the leader process.
1642 : : */
1643 : : static void
1644 : 115 : sendMessageToWorker(ParallelState *pstate, int worker, const char *str)
1645 : : {
1646 : 115 : int len = strlen(str) + 1;
1647 : :
1648 [ - + ]: 115 : if (pipewrite(pstate->parallelSlot[worker].pipeWrite, str, len) != len)
1649 : : {
1650 : 0 : pg_fatal("could not write to the communication channel: %m");
1651 : : }
1652 : 115 : }
1653 : :
1654 : : /*
1655 : : * Read one message from the specified pipe (fd), blocking if necessary
1656 : : * until one is available, and return it as a malloc'd string.
1657 : : * On EOF, return NULL.
1658 : : *
1659 : : * A "message" on the channel is just a null-terminated string.
1660 : : */
1661 : : static char *
1662 : 256 : readMessageFromPipe(int fd)
1663 : : {
1664 : : char *msg;
1665 : : int msgsize,
1666 : : bufsize;
1667 : : int ret;
1668 : :
1669 : : /*
1670 : : * In theory, if we let piperead() read multiple bytes, it might give us
1671 : : * back fragments of multiple messages. (That can't actually occur, since
1672 : : * neither leader nor workers send more than one message without waiting
1673 : : * for a reply, but we don't wish to assume that here.) For simplicity,
1674 : : * read a byte at a time until we get the terminating '\0'. This method
1675 : : * is a bit inefficient, but since this is only used for relatively short
1676 : : * command and status strings, it shouldn't matter.
1677 : : */
1678 : 256 : bufsize = 64; /* could be any number */
1679 : 256 : msg = (char *) pg_malloc(bufsize);
1680 : 256 : msgsize = 0;
1681 : : for (;;)
1682 : : {
1683 : 2438 : Assert(msgsize < bufsize);
1684 : 2694 : ret = piperead(fd, msg + msgsize, 1);
1685 [ + + ]: 2694 : if (ret <= 0)
1686 : 26 : break; /* error or connection closure */
1687 : :
1688 : : Assert(ret == 1);
1689 : :
1690 [ + + ]: 2668 : if (msg[msgsize] == '\0')
1691 : 230 : return msg; /* collected whole message */
1692 : :
1693 : 2438 : msgsize++;
1694 [ - + ]: 2438 : if (msgsize == bufsize) /* enlarge buffer if needed */
1695 : : {
1696 : 0 : bufsize += 16; /* could be any number */
1697 : 0 : msg = (char *) pg_realloc(msg, bufsize);
1698 : : }
1699 : : }
1700 : :
1701 : : /* Other end has closed the connection */
1702 : 26 : pg_free(msg);
1703 : 26 : return NULL;
1704 : : }
1705 : :
1706 : : #ifdef WIN32
1707 : :
1708 : : /*
1709 : : * This is a replacement version of pipe(2) for Windows which allows the pipe
1710 : : * handles to be used in select().
1711 : : *
1712 : : * Reads and writes on the pipe must go through piperead()/pipewrite().
1713 : : *
1714 : : * For consistency with Unix we declare the returned handles as "int".
1715 : : * This is okay even on WIN64 because system handles are not more than
1716 : : * 32 bits wide, but we do have to do some casting.
1717 : : */
1718 : : static int
1719 : : pgpipe(int handles[2])
1720 : : {
1721 : : pgsocket s,
1722 : : tmp_sock;
1723 : : struct sockaddr_in serv_addr;
1724 : : int len = sizeof(serv_addr);
1725 : :
1726 : : /* We have to use the Unix socket invalid file descriptor value here. */
1727 : : handles[0] = handles[1] = -1;
1728 : :
1729 : : /*
1730 : : * setup listen socket
1731 : : */
1732 : : if ((s = socket(AF_INET, SOCK_STREAM, 0)) == PGINVALID_SOCKET)
1733 : : {
1734 : : pg_log_error("pgpipe: could not create socket: error code %d",
1735 : : WSAGetLastError());
1736 : : return -1;
1737 : : }
1738 : :
1739 : : memset(&serv_addr, 0, sizeof(serv_addr));
1740 : : serv_addr.sin_family = AF_INET;
1741 : : serv_addr.sin_port = pg_hton16(0);
1742 : : serv_addr.sin_addr.s_addr = pg_hton32(INADDR_LOOPBACK);
1743 : : if (bind(s, (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR)
1744 : : {
1745 : : pg_log_error("pgpipe: could not bind: error code %d",
1746 : : WSAGetLastError());
1747 : : closesocket(s);
1748 : : return -1;
1749 : : }
1750 : : if (listen(s, 1) == SOCKET_ERROR)
1751 : : {
1752 : : pg_log_error("pgpipe: could not listen: error code %d",
1753 : : WSAGetLastError());
1754 : : closesocket(s);
1755 : : return -1;
1756 : : }
1757 : : if (getsockname(s, (SOCKADDR *) &serv_addr, &len) == SOCKET_ERROR)
1758 : : {
1759 : : pg_log_error("pgpipe: %s() failed: error code %d", "getsockname",
1760 : : WSAGetLastError());
1761 : : closesocket(s);
1762 : : return -1;
1763 : : }
1764 : :
1765 : : /*
1766 : : * setup pipe handles
1767 : : */
1768 : : if ((tmp_sock = socket(AF_INET, SOCK_STREAM, 0)) == PGINVALID_SOCKET)
1769 : : {
1770 : : pg_log_error("pgpipe: could not create second socket: error code %d",
1771 : : WSAGetLastError());
1772 : : closesocket(s);
1773 : : return -1;
1774 : : }
1775 : : handles[1] = (int) tmp_sock;
1776 : :
1777 : : if (connect(handles[1], (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR)
1778 : : {
1779 : : pg_log_error("pgpipe: could not connect socket: error code %d",
1780 : : WSAGetLastError());
1781 : : closesocket(handles[1]);
1782 : : handles[1] = -1;
1783 : : closesocket(s);
1784 : : return -1;
1785 : : }
1786 : : if ((tmp_sock = accept(s, (SOCKADDR *) &serv_addr, &len)) == PGINVALID_SOCKET)
1787 : : {
1788 : : pg_log_error("pgpipe: could not accept connection: error code %d",
1789 : : WSAGetLastError());
1790 : : closesocket(handles[1]);
1791 : : handles[1] = -1;
1792 : : closesocket(s);
1793 : : return -1;
1794 : : }
1795 : : handles[0] = (int) tmp_sock;
1796 : :
1797 : : closesocket(s);
1798 : : return 0;
1799 : : }
1800 : :
1801 : : #endif /* WIN32 */
|
