PostgreSQL Source Code  git master
parallel.h File Reference
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
 

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)
 

Typedef Documentation

◆ ParallelCompletionPtr

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

Definition at line 22 of file parallel.h.

◆ ParallelSlot

typedef struct ParallelSlot ParallelSlot

Definition at line 37 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 28 of file parallel.h.

Function Documentation

◆ DispatchJobForTocEntry()

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

Definition at line 1210 of file parallel.c.

References buf, buildWorkerCommand(), ParallelSlot::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().

1216 {
1217  int worker;
1218  char buf[256];
1219 
1220  /* Get a worker, waiting if none are idle */
1221  while ((worker = GetIdleWorker(pstate)) == NO_SLOT)
1222  WaitForWorkers(AH, pstate, WFW_ONE_IDLE);
1223 
1224  /* Construct and send command string */
1225  buildWorkerCommand(AH, te, act, buf, sizeof(buf));
1226 
1227  sendMessageToWorker(pstate, worker, buf);
1228 
1229  /* Remember worker is busy, and which TocEntry it's working on */
1230  pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
1231  pstate->parallelSlot[worker].callback = callback;
1232  pstate->parallelSlot[worker].callback_data = callback_data;
1233  pstate->te[worker] = te;
1234 }
static void buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act, char *buf, int buflen)
Definition: parallel.c:1113
void * callback_data
Definition: parallel.c:101
static int GetIdleWorker(ParallelState *pstate)
Definition: parallel.c:1241
#define NO_SLOT
Definition: parallel.c:74
static void callback(struct sockaddr *addr, struct sockaddr *mask, void *unused)
Definition: test_ifaddrs.c:48
static char * buf
Definition: pg_test_fsync.c:68
ParallelSlot * parallelSlot
Definition: parallel.h:45
static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str)
Definition: parallel.c:1649
void WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
Definition: parallel.c:1456
TocEntry ** te
Definition: parallel.h:44
ParallelCompletionPtr callback
Definition: parallel.c:100
T_WorkerStatus workerStatus
Definition: parallel.c:97

◆ init_parallel_dump_utils()

void init_parallel_dump_utils ( void  )

Definition at line 238 of file parallel.c.

References exit_nicely, and pg_log_error.

Referenced by main().

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  {
254  pg_log_error("%s() failed: error code %d", "WSAStartup", err);
255  exit_nicely(1);
256  }
257 
258  parallel_init_done = true;
259  }
260 #endif
261 }
#define pg_log_error(...)
Definition: logging.h:80
#define exit_nicely(code)
Definition: pg_dumpall.c:96

◆ IsEveryWorkerIdle()

bool IsEveryWorkerIdle ( ParallelState pstate)

Definition at line 1273 of file parallel.c.

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

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

1274 {
1275  int i;
1276 
1277  for (i = 0; i < pstate->numWorkers; i++)
1278  {
1279  if (pstate->parallelSlot[i].workerStatus != WRKR_IDLE)
1280  return false;
1281  }
1282  return true;
1283 }
ParallelSlot * parallelSlot
Definition: parallel.h:45
T_WorkerStatus workerStatus
Definition: parallel.c:97
int i
int numWorkers
Definition: parallel.h:42

◆ ParallelBackupEnd()

void ParallelBackupEnd ( ArchiveHandle AH,
ParallelState pstate 
)

Definition at line 1064 of file parallel.c.

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

Referenced by _CloseArchive(), and RestoreArchive().

1065 {
1066  int i;
1067 
1068  /* No work if non-parallel */
1069  if (pstate->numWorkers == 1)
1070  return;
1071 
1072  /* There should not be any unfinished jobs */
1073  Assert(IsEveryWorkerIdle(pstate));
1074 
1075  /* Close the sockets so that the workers know they can exit */
1076  for (i = 0; i < pstate->numWorkers; i++)
1077  {
1078  closesocket(pstate->parallelSlot[i].pipeRead);
1079  closesocket(pstate->parallelSlot[i].pipeWrite);
1080  }
1081 
1082  /* Wait for them to exit */
1083  WaitForTerminatingWorkers(pstate);
1084 
1085  /*
1086  * Unlink pstate from shutdown_info, so the exit handler will not try to
1087  * use it; and likewise unlink from signal_info.
1088  */
1089  shutdown_info.pstate = NULL;
1090  set_cancel_pstate(NULL);
1091 
1092  /* Release state (mere neatnik-ism, since we're about to terminate) */
1093  free(pstate->te);
1094  free(pstate->parallelSlot);
1095  free(pstate);
1096 }
int pipeRead
Definition: parallel.c:105
static void set_cancel_pstate(ParallelState *pstate)
Definition: parallel.c:794
bool IsEveryWorkerIdle(ParallelState *pstate)
Definition: parallel.c:1273
#define closesocket
Definition: port.h:332
ParallelSlot * parallelSlot
Definition: parallel.h:45
ParallelState * pstate
Definition: parallel.c:150
#define free(a)
Definition: header.h:65
static ShutdownInformation shutdown_info
Definition: parallel.c:154
#define Assert(condition)
Definition: c.h:804
TocEntry ** te
Definition: parallel.h:44
int pipeWrite
Definition: parallel.c:106
int i
static void WaitForTerminatingWorkers(ParallelState *pstate)
Definition: parallel.c:451
int numWorkers
Definition: parallel.h:42

◆ ParallelBackupStart()

ParallelState* ParallelBackupStart ( ArchiveHandle AH)

Definition at line 902 of file parallel.c.

References ParallelSlot::AH, DumpSignalInformation::am_worker, Assert, closesocket, _archiveHandle::connection, fatal, getLocalPQExpBuffer, i, ParallelState::numWorkers, Archive::numWorkers, ParallelState::parallelSlot, 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(), SIG_IGN, SIGPIPE, ParallelState::te, ParallelSlot::workerStatus, and WRKR_IDLE.

Referenced by _CloseArchive(), and RestoreArchive().

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

◆ set_archive_cancel_info()

void set_archive_cancel_info ( ArchiveHandle AH,
PGconn conn 
)

Definition at line 735 of file parallel.c.

References ParallelSlot::AH, _archiveHandle::connCancel, DumpSignalInformation::myAH, PQfreeCancel(), PQgetCancel(), and setup_cancel_handler().

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

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

◆ WaitForWorkers()

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

Definition at line 1456 of file parallel.c.

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

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

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