PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
parallel.h File Reference
#include <limits.h>
#include "pg_backup_archiver.h"
Include dependency graph for parallel.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  ParallelState
 

Macros

#define PG_MAX_JOBS   INT_MAX
 

Typedefs

typedef void(* ParallelCompletionPtr) (ArchiveHandle *AH, TocEntry *te, int status, void *callback_data)
 
typedef struct ParallelSlot ParallelSlot
 
typedef struct ParallelState ParallelState
 

Enumerations

enum  WFW_WaitOption { WFW_NO_WAIT , WFW_GOT_STATUS , WFW_ONE_IDLE , WFW_ALL_IDLE }
 

Functions

void init_parallel_dump_utils (void)
 
bool IsEveryWorkerIdle (ParallelState *pstate)
 
void WaitForWorkers (ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
 
ParallelStateParallelBackupStart (ArchiveHandle *AH)
 
void DispatchJobForTocEntry (ArchiveHandle *AH, ParallelState *pstate, TocEntry *te, T_Action act, ParallelCompletionPtr callback, void *callback_data)
 
void ParallelBackupEnd (ArchiveHandle *AH, ParallelState *pstate)
 
void set_archive_cancel_info (ArchiveHandle *AH, PGconn *conn)
 

Macro Definition Documentation

◆ PG_MAX_JOBS

#define PG_MAX_JOBS   INT_MAX

Definition at line 48 of file parallel.h.

Typedef Documentation

◆ ParallelCompletionPtr

typedef void(* ParallelCompletionPtr) (ArchiveHandle *AH, TocEntry *te, int status, void *callback_data)

Definition at line 24 of file parallel.h.

◆ ParallelSlot

typedef struct ParallelSlot ParallelSlot

Definition at line 52 of file parallel.h.

◆ ParallelState

typedef struct ParallelState ParallelState

Enumeration Type Documentation

◆ WFW_WaitOption

Enumerator
WFW_NO_WAIT 
WFW_GOT_STATUS 
WFW_ONE_IDLE 
WFW_ALL_IDLE 

Definition at line 30 of file parallel.h.

31{
WFW_WaitOption
Definition: parallel.h:31
@ WFW_ALL_IDLE
Definition: parallel.h:35
@ WFW_GOT_STATUS
Definition: parallel.h:33
@ WFW_NO_WAIT
Definition: parallel.h:32
@ WFW_ONE_IDLE
Definition: parallel.h:34

Function Documentation

◆ DispatchJobForTocEntry()

void DispatchJobForTocEntry ( ArchiveHandle AH,
ParallelState pstate,
TocEntry te,
T_Action  act,
ParallelCompletionPtr  callback,
void *  callback_data 
)

Definition at line 1217 of file parallel.c.

1223{
1224 int worker;
1225 char buf[256];
1226
1227 /* Get a worker, waiting if none are idle */
1228 while ((worker = GetIdleWorker(pstate)) == NO_SLOT)
1229 WaitForWorkers(AH, pstate, WFW_ONE_IDLE);
1230
1231 /* Construct and send command string */
1232 buildWorkerCommand(AH, te, act, buf, sizeof(buf));
1233
1234 sendMessageToWorker(pstate, worker, buf);
1235
1236 /* Remember worker is busy, and which TocEntry it's working on */
1237 pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
1238 pstate->parallelSlot[worker].callback = callback;
1239 pstate->parallelSlot[worker].callback_data = callback_data;
1240 pstate->te[worker] = te;
1241}
void WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
Definition: parallel.c:1463
@ WRKR_WORKING
Definition: parallel.c:81
static int GetIdleWorker(ParallelState *pstate)
Definition: parallel.c:1248
static void buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act, char *buf, int buflen)
Definition: parallel.c:1120
#define NO_SLOT
Definition: parallel.c:74
static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str)
Definition: parallel.c:1656
static char * buf
Definition: pg_test_fsync.c:72
ParallelCompletionPtr callback
Definition: parallel.c:100
void * callback_data
Definition: parallel.c:101
T_WorkerStatus workerStatus
Definition: parallel.c:97
TocEntry ** te
Definition: parallel.h:59
ParallelSlot * parallelSlot
Definition: parallel.h:60
static void callback(struct sockaddr *addr, struct sockaddr *mask, void *unused)
Definition: test_ifaddrs.c:46

References buf, buildWorkerCommand(), ParallelSlot::callback, callback(), ParallelSlot::callback_data, GetIdleWorker(), NO_SLOT, ParallelState::parallelSlot, sendMessageToWorker(), ParallelState::te, WaitForWorkers(), WFW_ONE_IDLE, ParallelSlot::workerStatus, and WRKR_WORKING.

Referenced by restore_toc_entries_parallel(), and WriteDataChunks().

◆ init_parallel_dump_utils()

void init_parallel_dump_utils ( void  )

Definition at line 238 of file parallel.c.

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}
void err(int eval, const char *fmt,...)
Definition: err.c:43
#define pg_fatal(...)

References err(), and pg_fatal.

Referenced by main().

◆ IsEveryWorkerIdle()

bool IsEveryWorkerIdle ( ParallelState pstate)

Definition at line 1280 of file parallel.c.

1281{
1282 int i;
1283
1284 for (i = 0; i < pstate->numWorkers; i++)
1285 {
1286 if (pstate->parallelSlot[i].workerStatus != WRKR_IDLE)
1287 return false;
1288 }
1289 return true;
1290}
@ WRKR_IDLE
Definition: parallel.c:80
int i
Definition: isn.c:77
int numWorkers
Definition: parallel.h:57

References i, ParallelState::numWorkers, ParallelState::parallelSlot, ParallelSlot::workerStatus, and WRKR_IDLE.

Referenced by ParallelBackupEnd(), restore_toc_entries_parallel(), and WaitForWorkers().

◆ ParallelBackupEnd()

void ParallelBackupEnd ( ArchiveHandle AH,
ParallelState pstate 
)

Definition at line 1071 of file parallel.c.

1072{
1073 int i;
1074
1075 /* No work if non-parallel */
1076 if (pstate->numWorkers == 1)
1077 return;
1078
1079 /* There should not be any unfinished jobs */
1080 Assert(IsEveryWorkerIdle(pstate));
1081
1082 /* Close the sockets so that the workers know they can exit */
1083 for (i = 0; i < pstate->numWorkers; i++)
1084 {
1087 }
1088
1089 /* Wait for them to exit */
1091
1092 /*
1093 * Unlink pstate from shutdown_info, so the exit handler will not try to
1094 * use it; and likewise unlink from signal_info.
1095 */
1096 shutdown_info.pstate = NULL;
1097 set_cancel_pstate(NULL);
1098
1099 /* Release state (mere neatnik-ism, since we're about to terminate) */
1100 free(pstate->te);
1101 free(pstate->parallelSlot);
1102 free(pstate);
1103}
static void set_cancel_pstate(ParallelState *pstate)
Definition: parallel.c:801
static ShutdownInformation shutdown_info
Definition: parallel.c:154
bool IsEveryWorkerIdle(ParallelState *pstate)
Definition: parallel.c:1280
static void WaitForTerminatingWorkers(ParallelState *pstate)
Definition: parallel.c:458
Assert(PointerIsAligned(start, uint64))
#define free(a)
Definition: header.h:65
#define closesocket
Definition: port.h:377
int pipeRead
Definition: parallel.c:105
int pipeWrite
Definition: parallel.c:106
ParallelState * pstate
Definition: parallel.c:150

References Assert(), closesocket, free, i, IsEveryWorkerIdle(), ParallelState::numWorkers, ParallelState::parallelSlot, ParallelSlot::pipeRead, ParallelSlot::pipeWrite, ShutdownInformation::pstate, set_cancel_pstate(), shutdown_info, ParallelState::te, and WaitForTerminatingWorkers().

Referenced by _CloseArchive(), and RestoreArchive().

◆ ParallelBackupStart()

ParallelState * ParallelBackupStart ( ArchiveHandle AH)

Definition at line 909 of file parallel.c.

910{
911 ParallelState *pstate;
912 int i;
913
914 Assert(AH->public.numWorkers > 0);
915
916 pstate = (ParallelState *) pg_malloc(sizeof(ParallelState));
917
918 pstate->numWorkers = AH->public.numWorkers;
919 pstate->te = NULL;
920 pstate->parallelSlot = NULL;
921
922 if (AH->public.numWorkers == 1)
923 return pstate;
924
925 /* Create status arrays, being sure to initialize all fields to 0 */
926 pstate->te = (TocEntry **)
927 pg_malloc0(pstate->numWorkers * sizeof(TocEntry *));
928 pstate->parallelSlot = (ParallelSlot *)
929 pg_malloc0(pstate->numWorkers * sizeof(ParallelSlot));
930
931#ifdef WIN32
932 /* Make fmtId() and fmtQualifiedId() use thread-local storage */
933 getLocalPQExpBuffer = getThreadLocalPQExpBuffer;
934#endif
935
936 /*
937 * Set the pstate in shutdown_info, to tell the exit handler that it must
938 * clean up workers as well as the main database connection. But we don't
939 * set this in signal_info yet, because we don't want child processes to
940 * inherit non-NULL signal_info.pstate.
941 */
942 shutdown_info.pstate = pstate;
943
944 /*
945 * Temporarily disable query cancellation on the leader connection. This
946 * ensures that child processes won't inherit valid AH->connCancel
947 * settings and thus won't try to issue cancels against the leader's
948 * connection. No harm is done if we fail while it's disabled, because
949 * the leader connection is idle at this point anyway.
950 */
951 set_archive_cancel_info(AH, NULL);
952
953 /* Ensure stdio state is quiesced before forking */
954 fflush(NULL);
955
956 /* Create desired number of workers */
957 for (i = 0; i < pstate->numWorkers; i++)
958 {
959#ifdef WIN32
960 WorkerInfo *wi;
961 uintptr_t handle;
962#else
963 pid_t pid;
964#endif
965 ParallelSlot *slot = &(pstate->parallelSlot[i]);
966 int pipeMW[2],
967 pipeWM[2];
968
969 /* Create communication pipes for this worker */
970 if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0)
971 pg_fatal("could not create communication channels: %m");
972
973 /* leader's ends of the pipes */
974 slot->pipeRead = pipeWM[PIPE_READ];
975 slot->pipeWrite = pipeMW[PIPE_WRITE];
976 /* child's ends of the pipes */
977 slot->pipeRevRead = pipeMW[PIPE_READ];
978 slot->pipeRevWrite = pipeWM[PIPE_WRITE];
979
980#ifdef WIN32
981 /* Create transient structure to pass args to worker function */
982 wi = (WorkerInfo *) pg_malloc(sizeof(WorkerInfo));
983
984 wi->AH = AH;
985 wi->slot = slot;
986
987 handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32,
988 wi, 0, &(slot->threadId));
989 slot->hThread = handle;
990 slot->workerStatus = WRKR_IDLE;
991#else /* !WIN32 */
992 pid = fork();
993 if (pid == 0)
994 {
995 /* we are the worker */
996 int j;
997
998 /* this is needed for GetMyPSlot() */
999 slot->pid = getpid();
1000
1001 /* instruct signal handler that we're in a worker now */
1002 signal_info.am_worker = true;
1003
1004 /* close read end of Worker -> Leader */
1005 closesocket(pipeWM[PIPE_READ]);
1006 /* close write end of Leader -> Worker */
1007 closesocket(pipeMW[PIPE_WRITE]);
1008
1009 /*
1010 * Close all inherited fds for communication of the leader with
1011 * previously-forked workers.
1012 */
1013 for (j = 0; j < i; j++)
1014 {
1017 }
1018
1019 /* Run the worker ... */
1020 RunWorker(AH, slot);
1021
1022 /* We can just exit(0) when done */
1023 exit(0);
1024 }
1025 else if (pid < 0)
1026 {
1027 /* fork failed */
1028 pg_fatal("could not create worker process: %m");
1029 }
1030
1031 /* In Leader after successful fork */
1032 slot->pid = pid;
1033 slot->workerStatus = WRKR_IDLE;
1034
1035 /* close read end of Leader -> Worker */
1036 closesocket(pipeMW[PIPE_READ]);
1037 /* close write end of Worker -> Leader */
1038 closesocket(pipeWM[PIPE_WRITE]);
1039#endif /* WIN32 */
1040 }
1041
1042 /*
1043 * Having forked off the workers, disable SIGPIPE so that leader isn't
1044 * killed if it tries to send a command to a dead worker. We don't want
1045 * the workers to inherit this setting, though.
1046 */
1047#ifndef WIN32
1048 pqsignal(SIGPIPE, SIG_IGN);
1049#endif
1050
1051 /*
1052 * Re-establish query cancellation on the leader connection.
1053 */
1055
1056 /*
1057 * Tell the cancel signal handler to forward signals to worker processes,
1058 * too. (As with query cancel, we did not need this earlier because the
1059 * workers have not yet been given anything to do; if we die before this
1060 * point, any already-started workers will see EOF and quit promptly.)
1061 */
1062 set_cancel_pstate(pstate);
1063
1064 return pstate;
1065}
#define pgpipe(a)
Definition: parallel.c:139
#define PIPE_READ
Definition: parallel.c:71
static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot)
Definition: parallel.c:841
#define PIPE_WRITE
Definition: parallel.c:72
static volatile DumpSignalInformation signal_info
Definition: parallel.c:175
void set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn)
Definition: parallel.c:742
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
void * pg_malloc0(size_t size)
Definition: fe_memutils.c:53
int j
Definition: isn.c:78
#define pqsignal
Definition: port.h:531
PQExpBuffer(* getLocalPQExpBuffer)(void)
Definition: string_utils.c:28
int numWorkers
Definition: pg_backup.h:235
int pipeRevRead
Definition: parallel.c:107
int pipeRevWrite
Definition: parallel.c:108
pid_t pid
Definition: parallel.c:115
#define SIGPIPE
Definition: win32_port.h:163

References DumpSignalInformation::am_worker, Assert(), closesocket, _archiveHandle::connection, getLocalPQExpBuffer, i, j, ParallelState::numWorkers, Archive::numWorkers, ParallelState::parallelSlot, pg_fatal, pg_malloc(), pg_malloc0(), pgpipe, ParallelSlot::pid, PIPE_READ, PIPE_WRITE, ParallelSlot::pipeRead, ParallelSlot::pipeRevRead, ParallelSlot::pipeRevWrite, ParallelSlot::pipeWrite, pqsignal, ShutdownInformation::pstate, _archiveHandle::public, RunWorker(), set_archive_cancel_info(), set_cancel_pstate(), shutdown_info, signal_info, SIGPIPE, ParallelState::te, ParallelSlot::workerStatus, and WRKR_IDLE.

Referenced by _CloseArchive(), and RestoreArchive().

◆ set_archive_cancel_info()

void set_archive_cancel_info ( ArchiveHandle AH,
PGconn conn 
)

Definition at line 742 of file parallel.c.

743{
744 PGcancel *oldConnCancel;
745
746 /*
747 * Activate the interrupt handler if we didn't yet in this process. On
748 * Windows, this also initializes signal_info_lock; therefore it's
749 * important that this happen at least once before we fork off any
750 * threads.
751 */
753
754 /*
755 * On Unix, we assume that storing a pointer value is atomic with respect
756 * to any possible signal interrupt. On Windows, use a critical section.
757 */
758
759#ifdef WIN32
760 EnterCriticalSection(&signal_info_lock);
761#endif
762
763 /* Free the old one if we have one */
764 oldConnCancel = AH->connCancel;
765 /* be sure interrupt handler doesn't use pointer while freeing */
766 AH->connCancel = NULL;
767
768 if (oldConnCancel != NULL)
769 PQfreeCancel(oldConnCancel);
770
771 /* Set the new one if specified */
772 if (conn)
774
775 /*
776 * On Unix, there's only ever one active ArchiveHandle per process, so we
777 * can just set signal_info.myAH unconditionally. On Windows, do that
778 * only in the main thread; worker threads have to make sure their
779 * ArchiveHandle appears in the pstate data, which is dealt with in
780 * RunWorker().
781 */
782#ifndef WIN32
783 signal_info.myAH = AH;
784#else
785 if (mainThreadId == GetCurrentThreadId())
786 signal_info.myAH = AH;
787#endif
788
789#ifdef WIN32
790 LeaveCriticalSection(&signal_info_lock);
791#endif
792}
static void set_cancel_handler(void)
Definition: parallel.c:620
PGcancel * PQgetCancel(PGconn *conn)
Definition: fe-cancel.c:367
void PQfreeCancel(PGcancel *cancel)
Definition: fe-cancel.c:484
PGconn * conn
Definition: streamutil.c:52
ArchiveHandle * myAH
Definition: parallel.c:167
PGcancel *volatile connCancel

References conn, _archiveHandle::connCancel, DumpSignalInformation::myAH, PQfreeCancel(), PQgetCancel(), set_cancel_handler(), and signal_info.

Referenced by ConnectDatabaseAhx(), DisconnectDatabase(), and ParallelBackupStart().

◆ WaitForWorkers()

void WaitForWorkers ( ArchiveHandle AH,
ParallelState pstate,
WFW_WaitOption  mode 
)

Definition at line 1463 of file parallel.c.

1464{
1465 bool do_wait = false;
1466
1467 /*
1468 * In GOT_STATUS mode, always block waiting for a message, since we can't
1469 * return till we get something. In other modes, we don't block the first
1470 * time through the loop.
1471 */
1472 if (mode == WFW_GOT_STATUS)
1473 {
1474 /* Assert that caller knows what it's doing */
1475 Assert(!IsEveryWorkerIdle(pstate));
1476 do_wait = true;
1477 }
1478
1479 for (;;)
1480 {
1481 /*
1482 * Check for status messages, even if we don't need to block. We do
1483 * not try very hard to reap all available messages, though, since
1484 * there's unlikely to be more than one.
1485 */
1486 if (ListenToWorkers(AH, pstate, do_wait))
1487 {
1488 /*
1489 * If we got a message, we are done by definition for GOT_STATUS
1490 * mode, and we can also be certain that there's at least one idle
1491 * worker. So we're done in all but ALL_IDLE mode.
1492 */
1493 if (mode != WFW_ALL_IDLE)
1494 return;
1495 }
1496
1497 /* Check whether we must wait for new status messages */
1498 switch (mode)
1499 {
1500 case WFW_NO_WAIT:
1501 return; /* never wait */
1502 case WFW_GOT_STATUS:
1503 Assert(false); /* can't get here, because we waited */
1504 break;
1505 case WFW_ONE_IDLE:
1506 if (GetIdleWorker(pstate) != NO_SLOT)
1507 return;
1508 break;
1509 case WFW_ALL_IDLE:
1510 if (IsEveryWorkerIdle(pstate))
1511 return;
1512 break;
1513 }
1514
1515 /* Loop back, and this time wait for something to happen */
1516 do_wait = true;
1517 }
1518}
static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
Definition: parallel.c:1410
static PgChecksumMode mode
Definition: pg_checksums.c:55
static bool do_wait
Definition: pg_ctl.c:75

References Assert(), do_wait, GetIdleWorker(), IsEveryWorkerIdle(), ListenToWorkers(), mode, NO_SLOT, WFW_ALL_IDLE, WFW_GOT_STATUS, WFW_NO_WAIT, and WFW_ONE_IDLE.

Referenced by DispatchJobForTocEntry(), restore_toc_entries_parallel(), and WriteDataChunks().