PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
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

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

Definition at line 22 of file parallel.h.

Definition at line 37 of file parallel.h.

Enumeration Type Documentation

Enumerator
WFW_NO_WAIT 
WFW_GOT_STATUS 
WFW_ONE_IDLE 
WFW_ALL_IDLE 

Definition at line 28 of file parallel.h.

Function Documentation

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

Definition at line 1227 of file parallel.c.

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

1233 {
1234  int worker;
1235  char buf[256];
1236 
1237  /* Get a worker, waiting if none are idle */
1238  while ((worker = GetIdleWorker(pstate)) == NO_SLOT)
1239  WaitForWorkers(AH, pstate, WFW_ONE_IDLE);
1240 
1241  /* Construct and send command string */
1242  buildWorkerCommand(AH, te, act, buf, sizeof(buf));
1243 
1244  sendMessageToWorker(pstate, worker, buf);
1245 
1246  /* Remember worker is busy, and which TocEntry it's working on */
1247  pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
1248  pstate->parallelSlot[worker].callback = callback;
1249  pstate->parallelSlot[worker].callback_data = callback_data;
1250  pstate->te[worker] = te;
1251 }
static void buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act, char *buf, int buflen)
Definition: parallel.c:1128
void * callback_data
Definition: parallel.c:95
static int GetIdleWorker(ParallelState *pstate)
Definition: parallel.c:1258
#define NO_SLOT
Definition: parallel.c:72
static void callback(struct sockaddr *addr, struct sockaddr *mask, void *unused)
Definition: test_ifaddrs.c:48
static char * buf
Definition: pg_test_fsync.c:65
ParallelSlot * parallelSlot
Definition: parallel.h:45
static void sendMessageToWorker(ParallelState *pstate, int worker, const char *str)
Definition: parallel.c:1668
void WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
Definition: parallel.c:1475
TocEntry ** te
Definition: parallel.h:44
ParallelCompletionPtr callback
Definition: parallel.c:94
T_WorkerStatus workerStatus
Definition: parallel.c:91
void init_parallel_dump_utils ( void  )

Definition at line 249 of file parallel.c.

References _, exit_nicely, NULL, on_exit_nicely(), and progname.

Referenced by main().

250 {
251 #ifdef WIN32
252  if (!parallel_init_done)
253  {
254  WSADATA wsaData;
255  int err;
256 
257  /* Prepare for threaded operation */
258  tls_index = TlsAlloc();
259  mainThreadId = GetCurrentThreadId();
260 
261  /* Initialize socket access */
262  err = WSAStartup(MAKEWORD(2, 2), &wsaData);
263  if (err != 0)
264  {
265  fprintf(stderr, _("%s: WSAStartup failed: %d\n"), progname, err);
266  exit_nicely(1);
267  }
268  /* ... and arrange to shut it down at exit */
269  on_exit_nicely(shutdown_parallel_dump_utils, NULL);
270  parallel_init_done = true;
271  }
272 #endif
273 }
void on_exit_nicely(on_exit_nicely_callback function, void *arg)
const char * progname
Definition: pg_standby.c:37
#define exit_nicely(code)
Definition: pg_dumpall.c:89
#define NULL
Definition: c.h:229
#define _(x)
Definition: elog.c:84
bool IsEveryWorkerIdle ( ParallelState pstate)

Definition at line 1290 of file parallel.c.

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

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

1291 {
1292  int i;
1293 
1294  for (i = 0; i < pstate->numWorkers; i++)
1295  {
1296  if (pstate->parallelSlot[i].workerStatus != WRKR_IDLE)
1297  return false;
1298  }
1299  return true;
1300 }
ParallelSlot * parallelSlot
Definition: parallel.h:45
T_WorkerStatus workerStatus
Definition: parallel.c:91
int i
int numWorkers
Definition: parallel.h:42
void ParallelBackupEnd ( ArchiveHandle AH,
ParallelState pstate 
)

Definition at line 1079 of file parallel.c.

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

Referenced by _CloseArchive(), and RestoreArchive().

1080 {
1081  int i;
1082 
1083  /* No work if non-parallel */
1084  if (pstate->numWorkers == 1)
1085  return;
1086 
1087  /* There should not be any unfinished jobs */
1088  Assert(IsEveryWorkerIdle(pstate));
1089 
1090  /* Close the sockets so that the workers know they can exit */
1091  for (i = 0; i < pstate->numWorkers; i++)
1092  {
1093  closesocket(pstate->parallelSlot[i].pipeRead);
1094  closesocket(pstate->parallelSlot[i].pipeWrite);
1095  }
1096 
1097  /* Wait for them to exit */
1098  WaitForTerminatingWorkers(pstate);
1099 
1100  /*
1101  * Unlink pstate from shutdown_info, so the exit handler will not try to
1102  * use it; and likewise unlink from signal_info.
1103  */
1106 
1107  /* Release state (mere neatnik-ism, since we're about to terminate) */
1108  free(pstate->te);
1109  free(pstate->parallelSlot);
1110  free(pstate);
1111 }
int pipeRead
Definition: parallel.c:99
static void set_cancel_pstate(ParallelState *pstate)
Definition: parallel.c:801
bool IsEveryWorkerIdle(ParallelState *pstate)
Definition: parallel.c:1290
#define closesocket
Definition: port.h:328
ParallelSlot * parallelSlot
Definition: parallel.h:45
ParallelState * pstate
Definition: parallel.c:144
#define free(a)
Definition: header.h:65
static ShutdownInformation shutdown_info
Definition: parallel.c:148
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
TocEntry ** te
Definition: parallel.h:44
int pipeWrite
Definition: parallel.c:100
int i
static void WaitForTerminatingWorkers(ParallelState *pstate)
Definition: parallel.c:461
int numWorkers
Definition: parallel.h:42
ParallelState* ParallelBackupStart ( ArchiveHandle AH)

Definition at line 909 of file parallel.c.

References ParallelSlot::AH, DumpSignalInformation::am_worker, Assert, ParallelSlot::callback, ParallelSlot::callback_data, closesocket, _archiveHandle::connection, exit_horribly(), getLocalPQExpBuffer, i, modulename, NULL, 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, strerror(), ParallelState::te, ParallelSlot::workerStatus, and WRKR_IDLE.

Referenced by _CloseArchive(), and RestoreArchive().

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  pstate->te = (TocEntry **)
926  pg_malloc0(pstate->numWorkers * sizeof(TocEntry *));
927  pstate->parallelSlot = (ParallelSlot *)
928  pg_malloc0(pstate->numWorkers * sizeof(ParallelSlot));
929 
930 #ifdef WIN32
931  /* Make fmtId() and fmtQualifiedId() use thread-local storage */
932  getLocalPQExpBuffer = getThreadLocalPQExpBuffer;
933 #endif
934 
935  /*
936  * Set the pstate in shutdown_info, to tell the exit handler that it must
937  * clean up workers as well as the main database connection. But we don't
938  * set this in signal_info yet, because we don't want child processes to
939  * inherit non-NULL signal_info.pstate.
940  */
941  shutdown_info.pstate = pstate;
942 
943  /*
944  * Temporarily disable query cancellation on the master connection. This
945  * ensures that child processes won't inherit valid AH->connCancel
946  * settings and thus won't try to issue cancels against the master's
947  * connection. No harm is done if we fail while it's disabled, because
948  * the master connection is idle at this point anyway.
949  */
951 
952  /* Ensure stdio state is quiesced before forking */
953  fflush(NULL);
954 
955  /* Create desired number of workers */
956  for (i = 0; i < pstate->numWorkers; i++)
957  {
958 #ifdef WIN32
959  WorkerInfo *wi;
960  uintptr_t handle;
961 #else
962  pid_t pid;
963 #endif
964  ParallelSlot *slot = &(pstate->parallelSlot[i]);
965  int pipeMW[2],
966  pipeWM[2];
967 
968  /* Create communication pipes for this worker */
969  if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0)
971  "could not create communication channels: %s\n",
972  strerror(errno));
973 
974  pstate->te[i] = NULL; /* just for safety */
975 
976  slot->workerStatus = WRKR_IDLE;
977  slot->AH = NULL;
978  slot->callback = NULL;
979  slot->callback_data = NULL;
980 
981  /* master's ends of the pipes */
982  slot->pipeRead = pipeWM[PIPE_READ];
983  slot->pipeWrite = pipeMW[PIPE_WRITE];
984  /* child's ends of the pipes */
985  slot->pipeRevRead = pipeMW[PIPE_READ];
986  slot->pipeRevWrite = pipeWM[PIPE_WRITE];
987 
988 #ifdef WIN32
989  /* Create transient structure to pass args to worker function */
990  wi = (WorkerInfo *) pg_malloc(sizeof(WorkerInfo));
991 
992  wi->AH = AH;
993  wi->slot = slot;
994 
995  handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32,
996  wi, 0, &(slot->threadId));
997  slot->hThread = handle;
998 #else /* !WIN32 */
999  pid = fork();
1000  if (pid == 0)
1001  {
1002  /* we are the worker */
1003  int j;
1004 
1005  /* this is needed for GetMyPSlot() */
1006  slot->pid = getpid();
1007 
1008  /* instruct signal handler that we're in a worker now */
1009  signal_info.am_worker = true;
1010 
1011  /* close read end of Worker -> Master */
1012  closesocket(pipeWM[PIPE_READ]);
1013  /* close write end of Master -> Worker */
1014  closesocket(pipeMW[PIPE_WRITE]);
1015 
1016  /*
1017  * Close all inherited fds for communication of the master with
1018  * previously-forked workers.
1019  */
1020  for (j = 0; j < i; j++)
1021  {
1022  closesocket(pstate->parallelSlot[j].pipeRead);
1023  closesocket(pstate->parallelSlot[j].pipeWrite);
1024  }
1025 
1026  /* Run the worker ... */
1027  RunWorker(AH, slot);
1028 
1029  /* We can just exit(0) when done */
1030  exit(0);
1031  }
1032  else if (pid < 0)
1033  {
1034  /* fork failed */
1036  "could not create worker process: %s\n",
1037  strerror(errno));
1038  }
1039 
1040  /* In Master after successful fork */
1041  slot->pid = pid;
1042 
1043  /* close read end of Master -> Worker */
1044  closesocket(pipeMW[PIPE_READ]);
1045  /* close write end of Worker -> Master */
1046  closesocket(pipeWM[PIPE_WRITE]);
1047 #endif /* WIN32 */
1048  }
1049 
1050  /*
1051  * Having forked off the workers, disable SIGPIPE so that master isn't
1052  * killed if it tries to send a command to a dead worker. We don't want
1053  * the workers to inherit this setting, though.
1054  */
1055 #ifndef WIN32
1057 #endif
1058 
1059  /*
1060  * Re-establish query cancellation on the master connection.
1061  */
1063 
1064  /*
1065  * Tell the cancel signal handler to forward signals to worker processes,
1066  * too. (As with query cancel, we did not need this earlier because the
1067  * workers have not yet been given anything to do; if we die before this
1068  * point, any already-started workers will see EOF and quit promptly.)
1069  */
1070  set_cancel_pstate(pstate);
1071 
1072  return pstate;
1073 }
int pipeRead
Definition: parallel.c:99
static void set_cancel_pstate(ParallelState *pstate)
Definition: parallel.c:801
ArchiveHandle * AH
Definition: parallel.c:97
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
void * callback_data
Definition: parallel.c:95
#define closesocket
Definition: port.h:328
#define PIPE_READ
Definition: parallel.c:69
static const char * modulename
Definition: parallel.c:198
#define SIG_IGN
Definition: win32.h:193
void * pg_malloc0(size_t size)
Definition: fe_memutils.c:53
pid_t pid
Definition: parallel.c:109
static void RunWorker(ArchiveHandle *AH, ParallelSlot *slot)
Definition: parallel.c:841
int pipeRevRead
Definition: parallel.c:101
#define pgpipe(a)
Definition: parallel.c:133
ParallelSlot * parallelSlot
Definition: parallel.h:45
ParallelState * pstate
Definition: parallel.c:144
#define SIGPIPE
Definition: win32.h:201
pqsigfunc pqsignal(int signum, pqsigfunc handler)
Definition: signal.c:168
static ShutdownInformation shutdown_info
Definition: parallel.c:148
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
TocEntry ** te
Definition: parallel.h:44
ParallelCompletionPtr callback
Definition: parallel.c:94
int pipeRevWrite
Definition: parallel.c:102
int numWorkers
Definition: pg_backup.h:190
static volatile DumpSignalInformation signal_info
Definition: parallel.c:169
T_WorkerStatus workerStatus
Definition: parallel.c:91
void exit_horribly(const char *modulename, const char *fmt,...)
void set_archive_cancel_info(ArchiveHandle *AH, PGconn *conn)
Definition: parallel.c:742
int pipeWrite
Definition: parallel.c:100
int i
const char * strerror(int errnum)
Definition: strerror.c:19
#define PIPE_WRITE
Definition: parallel.c:70
PQExpBuffer(* getLocalPQExpBuffer)(void)
Definition: string_utils.c:29
int numWorkers
Definition: parallel.h:42
void set_archive_cancel_info ( ArchiveHandle AH,
PGconn conn 
)

Definition at line 742 of file parallel.c.

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

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

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)
773  AH->connCancel = PQgetCancel(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 }
void PQfreeCancel(PGcancel *cancel)
Definition: fe-connect.c:3712
ArchiveHandle * myAH
Definition: parallel.c:161
static void setup_cancel_handler(void)
Definition: parallel.c:620
PGcancel *volatile connCancel
PGcancel * PQgetCancel(PGconn *conn)
Definition: fe-connect.c:3689
#define NULL
Definition: c.h:229
static volatile DumpSignalInformation signal_info
Definition: parallel.c:169
void WaitForWorkers ( ArchiveHandle AH,
ParallelState pstate,
WFW_WaitOption  mode 
)

Definition at line 1475 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().

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