PostgreSQL Source Code  git master
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 24 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 1207 of file parallel.c.

1213 {
1214  int worker;
1215  char buf[256];
1216 
1217  /* Get a worker, waiting if none are idle */
1218  while ((worker = GetIdleWorker(pstate)) == NO_SLOT)
1219  WaitForWorkers(AH, pstate, WFW_ONE_IDLE);
1220 
1221  /* Construct and send command string */
1222  buildWorkerCommand(AH, te, act, buf, sizeof(buf));
1223 
1224  sendMessageToWorker(pstate, worker, buf);
1225 
1226  /* Remember worker is busy, and which TocEntry it's working on */
1227  pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
1228  pstate->parallelSlot[worker].callback = callback;
1229  pstate->parallelSlot[worker].callback_data = callback_data;
1230  pstate->te[worker] = te;
1231 }
void WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
Definition: parallel.c:1453
@ WRKR_WORKING
Definition: parallel.c:81
static int GetIdleWorker(ParallelState *pstate)
Definition: parallel.c:1238
static void buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act, char *buf, int buflen)
Definition: parallel.c:1110
#define NO_SLOT
Definition: parallel.c:74
static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str)
Definition: parallel.c:1646
static char * buf
Definition: pg_test_fsync.c:67
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:48

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 }
#define pg_fatal(...)

References pg_fatal.

Referenced by main().

◆ IsEveryWorkerIdle()

bool IsEveryWorkerIdle ( ParallelState pstate)

Definition at line 1270 of file parallel.c.

1271 {
1272  int i;
1273 
1274  for (i = 0; i < pstate->numWorkers; i++)
1275  {
1276  if (pstate->parallelSlot[i].workerStatus != WRKR_IDLE)
1277  return false;
1278  }
1279  return true;
1280 }
@ WRKR_IDLE
Definition: parallel.c:80
int i
Definition: isn.c:73
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 1061 of file parallel.c.

1062 {
1063  int i;
1064 
1065  /* No work if non-parallel */
1066  if (pstate->numWorkers == 1)
1067  return;
1068 
1069  /* There should not be any unfinished jobs */
1070  Assert(IsEveryWorkerIdle(pstate));
1071 
1072  /* Close the sockets so that the workers know they can exit */
1073  for (i = 0; i < pstate->numWorkers; i++)
1074  {
1075  closesocket(pstate->parallelSlot[i].pipeRead);
1076  closesocket(pstate->parallelSlot[i].pipeWrite);
1077  }
1078 
1079  /* Wait for them to exit */
1080  WaitForTerminatingWorkers(pstate);
1081 
1082  /*
1083  * Unlink pstate from shutdown_info, so the exit handler will not try to
1084  * use it; and likewise unlink from signal_info.
1085  */
1086  shutdown_info.pstate = NULL;
1087  set_cancel_pstate(NULL);
1088 
1089  /* Release state (mere neatnik-ism, since we're about to terminate) */
1090  free(pstate->te);
1091  free(pstate->parallelSlot);
1092  free(pstate);
1093 }
static void set_cancel_pstate(ParallelState *pstate)
Definition: parallel.c:791
static ShutdownInformation shutdown_info
Definition: parallel.c:154
bool IsEveryWorkerIdle(ParallelState *pstate)
Definition: parallel.c:1270
static void WaitForTerminatingWorkers(ParallelState *pstate)
Definition: parallel.c:448
#define free(a)
Definition: header.h:65
Assert(fmt[strlen(fmt) - 1] !='\n')
#define closesocket
Definition: port.h:342
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 899 of file parallel.c.

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

References DumpSignalInformation::am_worker, Assert(), closesocket, _archiveHandle::connection, exit(), fflush(), 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, SIG_IGN, 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 732 of file parallel.c.

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

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

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

◆ WaitForWorkers()

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

Definition at line 1453 of file parallel.c.

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

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().