PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
parallel.c File Reference
#include "postgres_fe.h"
#include <sys/select.h>
#include <sys/wait.h>
#include <signal.h>
#include <unistd.h>
#include <fcntl.h>
#include "fe_utils/string_utils.h"
#include "parallel.h"
#include "pg_backup_utils.h"
Include dependency graph for parallel.c:

Go to the source code of this file.

Data Structures

struct  ParallelSlot
 
struct  ShutdownInformation
 
struct  DumpSignalInformation
 

Macros

#define PIPE_READ   0
 
#define PIPE_WRITE   1
 
#define NO_SLOT   (-1) /* Failure result for GetIdleWorker() */
 
#define WORKER_IS_RUNNING(workerStatus)    ((workerStatus) == WRKR_IDLE || (workerStatus) == WRKR_WORKING)
 
#define pgpipe(a)   pipe(a)
 
#define piperead(a, b, c)   read(a,b,c)
 
#define pipewrite(a, b, c)   write(a,b,c)
 
#define write_stderr(str)
 
#define messageStartsWith(msg, prefix)    (strncmp(msg, prefix, strlen(prefix)) == 0)
 

Typedefs

typedef struct ShutdownInformation ShutdownInformation
 
typedef struct DumpSignalInformation DumpSignalInformation
 

Enumerations

enum  T_WorkerStatus { WRKR_NOT_STARTED = 0 , WRKR_IDLE , WRKR_WORKING , WRKR_TERMINATED }
 

Functions

static ParallelSlotGetMyPSlot (ParallelState *pstate)
 
static void archive_close_connection (int code, void *arg)
 
static void ShutdownWorkersHard (ParallelState *pstate)
 
static void WaitForTerminatingWorkers (ParallelState *pstate)
 
static void set_cancel_handler (void)
 
static void set_cancel_pstate (ParallelState *pstate)
 
static void set_cancel_slot_archive (ParallelSlot *slot, ArchiveHandle *AH)
 
static void RunWorker (ArchiveHandle *AH, ParallelSlot *slot)
 
static int GetIdleWorker (ParallelState *pstate)
 
static bool HasEveryWorkerTerminated (ParallelState *pstate)
 
static void lockTableForWorker (ArchiveHandle *AH, TocEntry *te)
 
static void WaitForCommands (ArchiveHandle *AH, int pipefd[2])
 
static bool ListenToWorkers (ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
 
static char * getMessageFromLeader (int pipefd[2])
 
static void sendMessageToLeader (int pipefd[2], const char *str)
 
static int select_loop (int maxFd, fd_set *workerset)
 
static char * getMessageFromWorker (ParallelState *pstate, bool do_wait, int *worker)
 
static void sendMessageToWorker (ParallelState *pstate, int worker, const char *str)
 
static char * readMessageFromPipe (int fd)
 
void init_parallel_dump_utils (void)
 
void on_exit_close_archive (Archive *AHX)
 
void replace_on_exit_close_archive (Archive *AHX)
 
static void sigTermHandler (SIGNAL_ARGS)
 
void set_archive_cancel_info (ArchiveHandle *AH, PGconn *conn)
 
ParallelStateParallelBackupStart (ArchiveHandle *AH)
 
void ParallelBackupEnd (ArchiveHandle *AH, ParallelState *pstate)
 
static void buildWorkerCommand (ArchiveHandle *AH, TocEntry *te, T_Action act, char *buf, int buflen)
 
static void parseWorkerCommand (ArchiveHandle *AH, TocEntry **te, T_Action *act, const char *msg)
 
static void buildWorkerResponse (ArchiveHandle *AH, TocEntry *te, T_Action act, int status, char *buf, int buflen)
 
static int parseWorkerResponse (ArchiveHandle *AH, TocEntry *te, const char *msg)
 
void DispatchJobForTocEntry (ArchiveHandle *AH, ParallelState *pstate, TocEntry *te, T_Action act, ParallelCompletionPtr callback, void *callback_data)
 
bool IsEveryWorkerIdle (ParallelState *pstate)
 
void WaitForWorkers (ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
 

Variables

static ShutdownInformation shutdown_info
 
static volatile DumpSignalInformation signal_info
 

Macro Definition Documentation

◆ messageStartsWith

#define messageStartsWith (   msg,
  prefix 
)     (strncmp(msg, prefix, strlen(prefix)) == 0)

Definition at line 228 of file parallel.c.

◆ NO_SLOT

#define NO_SLOT   (-1) /* Failure result for GetIdleWorker() */

Definition at line 74 of file parallel.c.

◆ pgpipe

#define pgpipe (   a)    pipe(a)

Definition at line 139 of file parallel.c.

◆ PIPE_READ

#define PIPE_READ   0

Definition at line 71 of file parallel.c.

◆ PIPE_WRITE

#define PIPE_WRITE   1

Definition at line 72 of file parallel.c.

◆ piperead

#define piperead (   a,
  b,
  c 
)    read(a,b,c)

Definition at line 140 of file parallel.c.

◆ pipewrite

#define pipewrite (   a,
  b,
  c 
)    write(a,b,c)

Definition at line 141 of file parallel.c.

◆ WORKER_IS_RUNNING

#define WORKER_IS_RUNNING (   workerStatus)     ((workerStatus) == WRKR_IDLE || (workerStatus) == WRKR_WORKING)

Definition at line 85 of file parallel.c.

◆ write_stderr

#define write_stderr (   str)
Value:
do { \
const char *str_ = (str); \
int rc_; \
rc_ = write(fileno(stderr), str_, strlen(str_)); \
(void) rc_; \
} while (0)
const char * str
#define write(a, b, c)
Definition: win32.h:14

Definition at line 186 of file parallel.c.

Typedef Documentation

◆ DumpSignalInformation

◆ ShutdownInformation

Enumeration Type Documentation

◆ T_WorkerStatus

Enumerator
WRKR_NOT_STARTED 
WRKR_IDLE 
WRKR_WORKING 
WRKR_TERMINATED 

Definition at line 77 of file parallel.c.

78{
T_WorkerStatus
Definition: parallel.c:78
@ WRKR_WORKING
Definition: parallel.c:81
@ WRKR_IDLE
Definition: parallel.c:80
@ WRKR_TERMINATED
Definition: parallel.c:82
@ WRKR_NOT_STARTED
Definition: parallel.c:79

Function Documentation

◆ archive_close_connection()

static void archive_close_connection ( int  code,
void *  arg 
)
static

Definition at line 351 of file parallel.c.

352{
354
355 if (si->pstate)
356 {
357 /* In parallel mode, must figure out who we are */
358 ParallelSlot *slot = GetMyPSlot(si->pstate);
359
360 if (!slot)
361 {
362 /*
363 * We're the leader. Forcibly shut down workers, then close our
364 * own database connection, if any.
365 */
367
368 if (si->AHX)
370 }
371 else
372 {
373 /*
374 * We're a worker. Shut down our own DB connection if any. On
375 * Windows, we also have to close our communication sockets, to
376 * emulate what will happen on Unix when the worker process exits.
377 * (Without this, if this is a premature exit, the leader would
378 * fail to detect it because there would be no EOF condition on
379 * the other end of the pipe.)
380 */
381 if (slot->AH)
382 DisconnectDatabase(&(slot->AH->public));
383
384#ifdef WIN32
387#endif
388 }
389 }
390 else
391 {
392 /* Non-parallel operation: just kill the leader DB connection */
393 if (si->AHX)
395 }
396}
static ParallelSlot * GetMyPSlot(ParallelState *pstate)
Definition: parallel.c:266
static void ShutdownWorkersHard(ParallelState *pstate)
Definition: parallel.c:407
void DisconnectDatabase(Archive *AHX)
Definition: pg_backup_db.c:164
void * arg
#define closesocket
Definition: port.h:377
ArchiveHandle * AH
Definition: parallel.c:103
int pipeRevRead
Definition: parallel.c:107
int pipeRevWrite
Definition: parallel.c:108
ParallelState * pstate
Definition: parallel.c:150

References ParallelSlot::AH, ShutdownInformation::AHX, arg, closesocket, DisconnectDatabase(), GetMyPSlot(), ParallelSlot::pipeRevRead, ParallelSlot::pipeRevWrite, ShutdownInformation::pstate, _archiveHandle::public, and ShutdownWorkersHard().

Referenced by on_exit_close_archive().

◆ buildWorkerCommand()

static void buildWorkerCommand ( ArchiveHandle AH,
TocEntry te,
T_Action  act,
char *  buf,
int  buflen 
)
static

Definition at line 1120 of file parallel.c.

1122{
1123 if (act == ACT_DUMP)
1124 snprintf(buf, buflen, "DUMP %d", te->dumpId);
1125 else if (act == ACT_RESTORE)
1126 snprintf(buf, buflen, "RESTORE %d", te->dumpId);
1127 else
1128 Assert(false);
1129}
Assert(PointerIsAligned(start, uint64))
@ ACT_RESTORE
@ ACT_DUMP
static char * buf
Definition: pg_test_fsync.c:72
#define snprintf
Definition: port.h:239

References ACT_DUMP, ACT_RESTORE, Assert(), buf, _tocEntry::dumpId, and snprintf.

Referenced by DispatchJobForTocEntry().

◆ buildWorkerResponse()

static void buildWorkerResponse ( ArchiveHandle AH,
TocEntry te,
T_Action  act,
int  status,
char *  buf,
int  buflen 
)
static

Definition at line 1168 of file parallel.c.

1170{
1171 snprintf(buf, buflen, "OK %d %d %d",
1172 te->dumpId,
1173 status,
1174 status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0);
1175}
#define WORKER_IGNORED_ERRORS
int n_errors
Definition: pg_backup.h:248

References buf, _tocEntry::dumpId, Archive::n_errors, _archiveHandle::public, snprintf, and WORKER_IGNORED_ERRORS.

Referenced by WaitForCommands().

◆ 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
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
@ WFW_ONE_IDLE
Definition: parallel.h:34
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().

◆ GetIdleWorker()

static int GetIdleWorker ( ParallelState pstate)
static

Definition at line 1248 of file parallel.c.

1249{
1250 int i;
1251
1252 for (i = 0; i < pstate->numWorkers; i++)
1253 {
1254 if (pstate->parallelSlot[i].workerStatus == WRKR_IDLE)
1255 return i;
1256 }
1257 return NO_SLOT;
1258}
int i
Definition: isn.c:77
int numWorkers
Definition: parallel.h:57

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

Referenced by DispatchJobForTocEntry(), and WaitForWorkers().

◆ getMessageFromLeader()

static char * getMessageFromLeader ( int  pipefd[2])
static

Definition at line 1528 of file parallel.c.

1529{
1530 return readMessageFromPipe(pipefd[PIPE_READ]);
1531}
#define PIPE_READ
Definition: parallel.c:71
static char * readMessageFromPipe(int fd)
Definition: parallel.c:1674

References PIPE_READ, and readMessageFromPipe().

Referenced by WaitForCommands().

◆ getMessageFromWorker()

static char * getMessageFromWorker ( ParallelState pstate,
bool  do_wait,
int *  worker 
)
static

Definition at line 1591 of file parallel.c.

1592{
1593 int i;
1594 fd_set workerset;
1595 int maxFd = -1;
1596 struct timeval nowait = {0, 0};
1597
1598 /* construct bitmap of socket descriptors for select() */
1599 FD_ZERO(&workerset);
1600 for (i = 0; i < pstate->numWorkers; i++)
1601 {
1603 continue;
1604 FD_SET(pstate->parallelSlot[i].pipeRead, &workerset);
1605 if (pstate->parallelSlot[i].pipeRead > maxFd)
1606 maxFd = pstate->parallelSlot[i].pipeRead;
1607 }
1608
1609 if (do_wait)
1610 {
1611 i = select_loop(maxFd, &workerset);
1612 Assert(i != 0);
1613 }
1614 else
1615 {
1616 if ((i = select(maxFd + 1, &workerset, NULL, NULL, &nowait)) == 0)
1617 return NULL;
1618 }
1619
1620 if (i < 0)
1621 pg_fatal("%s() failed: %m", "select");
1622
1623 for (i = 0; i < pstate->numWorkers; i++)
1624 {
1625 char *msg;
1626
1628 continue;
1629 if (!FD_ISSET(pstate->parallelSlot[i].pipeRead, &workerset))
1630 continue;
1631
1632 /*
1633 * Read the message if any. If the socket is ready because of EOF,
1634 * we'll return NULL instead (and the socket will stay ready, so the
1635 * condition will persist).
1636 *
1637 * Note: because this is a blocking read, we'll wait if only part of
1638 * the message is available. Waiting a long time would be bad, but
1639 * since worker status messages are short and are always sent in one
1640 * operation, it shouldn't be a problem in practice.
1641 */
1643 *worker = i;
1644 return msg;
1645 }
1646 Assert(false);
1647 return NULL;
1648}
static int select_loop(int maxFd, fd_set *workerset)
Definition: parallel.c:1552
#define WORKER_IS_RUNNING(workerStatus)
Definition: parallel.c:85
#define pg_fatal(...)
static bool do_wait
Definition: pg_ctl.c:75
int pipeRead
Definition: parallel.c:105
#define select(n, r, w, e, timeout)
Definition: win32_port.h:503

References Assert(), do_wait, i, ParallelState::numWorkers, ParallelState::parallelSlot, pg_fatal, ParallelSlot::pipeRead, readMessageFromPipe(), select, select_loop(), WORKER_IS_RUNNING, and ParallelSlot::workerStatus.

Referenced by ListenToWorkers().

◆ GetMyPSlot()

static ParallelSlot * GetMyPSlot ( ParallelState pstate)
static

Definition at line 266 of file parallel.c.

267{
268 int i;
269
270 for (i = 0; i < pstate->numWorkers; i++)
271 {
272#ifdef WIN32
273 if (pstate->parallelSlot[i].threadId == GetCurrentThreadId())
274#else
275 if (pstate->parallelSlot[i].pid == getpid())
276#endif
277 return &(pstate->parallelSlot[i]);
278 }
279
280 return NULL;
281}
pid_t pid
Definition: parallel.c:115

References i, ParallelState::numWorkers, ParallelState::parallelSlot, and ParallelSlot::pid.

Referenced by archive_close_connection().

◆ HasEveryWorkerTerminated()

static bool HasEveryWorkerTerminated ( ParallelState pstate)
static

Definition at line 1264 of file parallel.c.

1265{
1266 int i;
1267
1268 for (i = 0; i < pstate->numWorkers; i++)
1269 {
1271 return false;
1272 }
1273 return true;
1274}

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

Referenced by WaitForTerminatingWorkers().

◆ 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

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}

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

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

◆ ListenToWorkers()

static bool ListenToWorkers ( ArchiveHandle AH,
ParallelState pstate,
bool  do_wait 
)
static

Definition at line 1410 of file parallel.c.

1411{
1412 int worker;
1413 char *msg;
1414
1415 /* Try to collect a status message */
1416 msg = getMessageFromWorker(pstate, do_wait, &worker);
1417
1418 if (!msg)
1419 {
1420 /* If do_wait is true, we must have detected EOF on some socket */
1421 if (do_wait)
1422 pg_fatal("a worker process died unexpectedly");
1423 return false;
1424 }
1425
1426 /* Process it and update our idea of the worker's status */
1427 if (messageStartsWith(msg, "OK "))
1428 {
1429 ParallelSlot *slot = &pstate->parallelSlot[worker];
1430 TocEntry *te = pstate->te[worker];
1431 int status;
1432
1433 status = parseWorkerResponse(AH, te, msg);
1434 slot->callback(AH, te, status, slot->callback_data);
1435 slot->workerStatus = WRKR_IDLE;
1436 pstate->te[worker] = NULL;
1437 }
1438 else
1439 pg_fatal("invalid message received from worker: \"%s\"",
1440 msg);
1441
1442 /* Free the string returned from getMessageFromWorker */
1443 free(msg);
1444
1445 return true;
1446}
static int parseWorkerResponse(ArchiveHandle *AH, TocEntry *te, const char *msg)
Definition: parallel.c:1183
static char * getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
Definition: parallel.c:1591
#define messageStartsWith(msg, prefix)
Definition: parallel.c:228
#define free(a)
Definition: header.h:65

References ParallelSlot::callback, ParallelSlot::callback_data, do_wait, free, getMessageFromWorker(), messageStartsWith, ParallelState::parallelSlot, parseWorkerResponse(), pg_fatal, ParallelState::te, ParallelSlot::workerStatus, and WRKR_IDLE.

Referenced by WaitForWorkers().

◆ lockTableForWorker()

static void lockTableForWorker ( ArchiveHandle AH,
TocEntry te 
)
static

Definition at line 1313 of file parallel.c.

1314{
1315 const char *qualId;
1316 PQExpBuffer query;
1317 PGresult *res;
1318
1319 /* Nothing to do for BLOBS */
1320 if (strcmp(te->desc, "BLOBS") == 0)
1321 return;
1322
1323 query = createPQExpBuffer();
1324
1325 qualId = fmtQualifiedId(te->namespace, te->tag);
1326
1327 appendPQExpBuffer(query, "LOCK TABLE %s IN ACCESS SHARE MODE NOWAIT",
1328 qualId);
1329
1330 res = PQexec(AH->connection, query->data);
1331
1332 if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
1333 pg_fatal("could not obtain lock on relation \"%s\"\n"
1334 "This usually means that someone requested an ACCESS EXCLUSIVE lock "
1335 "on the table after the pg_dump parent process had gotten the "
1336 "initial ACCESS SHARE lock on the table.", qualId);
1337
1338 PQclear(res);
1339 destroyPQExpBuffer(query);
1340}
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3411
void PQclear(PGresult *res)
Definition: fe-exec.c:721
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:2262
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:125
PQExpBuffer createPQExpBuffer(void)
Definition: pqexpbuffer.c:72
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
Definition: pqexpbuffer.c:265
void destroyPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:114
const char * fmtQualifiedId(const char *schema, const char *id)
Definition: string_utils.c:296

References appendPQExpBuffer(), _archiveHandle::connection, createPQExpBuffer(), PQExpBufferData::data, _tocEntry::desc, destroyPQExpBuffer(), fmtQualifiedId(), pg_fatal, PGRES_COMMAND_OK, PQclear(), PQexec(), PQresultStatus(), and _tocEntry::tag.

Referenced by WaitForCommands().

◆ on_exit_close_archive()

void on_exit_close_archive ( Archive AHX)

Definition at line 330 of file parallel.c.

331{
332 shutdown_info.AHX = AHX;
334}
static void archive_close_connection(int code, void *arg)
Definition: parallel.c:351
static ShutdownInformation shutdown_info
Definition: parallel.c:154
void on_exit_nicely(on_exit_nicely_callback function, void *arg)

References ShutdownInformation::AHX, archive_close_connection(), on_exit_nicely(), and shutdown_info.

Referenced by main(), and restore_one_database().

◆ 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
bool IsEveryWorkerIdle(ParallelState *pstate)
Definition: parallel.c:1280
static void WaitForTerminatingWorkers(ParallelState *pstate)
Definition: parallel.c:458
int pipeWrite
Definition: parallel.c:106

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

◆ parseWorkerCommand()

static void parseWorkerCommand ( ArchiveHandle AH,
TocEntry **  te,
T_Action act,
const char *  msg 
)
static

Definition at line 1135 of file parallel.c.

1137{
1138 DumpId dumpId;
1139 int nBytes;
1140
1141 if (messageStartsWith(msg, "DUMP "))
1142 {
1143 *act = ACT_DUMP;
1144 sscanf(msg, "DUMP %d%n", &dumpId, &nBytes);
1145 Assert(nBytes == strlen(msg));
1146 *te = getTocEntryByDumpId(AH, dumpId);
1147 Assert(*te != NULL);
1148 }
1149 else if (messageStartsWith(msg, "RESTORE "))
1150 {
1151 *act = ACT_RESTORE;
1152 sscanf(msg, "RESTORE %d%n", &dumpId, &nBytes);
1153 Assert(nBytes == strlen(msg));
1154 *te = getTocEntryByDumpId(AH, dumpId);
1155 Assert(*te != NULL);
1156 }
1157 else
1158 pg_fatal("unrecognized command received from leader: \"%s\"",
1159 msg);
1160}
int DumpId
Definition: pg_backup.h:280
TocEntry * getTocEntryByDumpId(ArchiveHandle *AH, DumpId id)

References ACT_DUMP, ACT_RESTORE, Assert(), getTocEntryByDumpId(), messageStartsWith, and pg_fatal.

Referenced by WaitForCommands().

◆ parseWorkerResponse()

static int parseWorkerResponse ( ArchiveHandle AH,
TocEntry te,
const char *  msg 
)
static

Definition at line 1183 of file parallel.c.

1185{
1186 DumpId dumpId;
1187 int nBytes,
1188 n_errors;
1189 int status = 0;
1190
1191 if (messageStartsWith(msg, "OK "))
1192 {
1193 sscanf(msg, "OK %d %d %d%n", &dumpId, &status, &n_errors, &nBytes);
1194
1195 Assert(dumpId == te->dumpId);
1196 Assert(nBytes == strlen(msg));
1197
1198 AH->public.n_errors += n_errors;
1199 }
1200 else
1201 pg_fatal("invalid message received from worker: \"%s\"",
1202 msg);
1203
1204 return status;
1205}

References Assert(), _tocEntry::dumpId, messageStartsWith, Archive::n_errors, pg_fatal, and _archiveHandle::public.

Referenced by ListenToWorkers().

◆ readMessageFromPipe()

static char * readMessageFromPipe ( int  fd)
static

Definition at line 1674 of file parallel.c.

1675{
1676 char *msg;
1677 int msgsize,
1678 bufsize;
1679 int ret;
1680
1681 /*
1682 * In theory, if we let piperead() read multiple bytes, it might give us
1683 * back fragments of multiple messages. (That can't actually occur, since
1684 * neither leader nor workers send more than one message without waiting
1685 * for a reply, but we don't wish to assume that here.) For simplicity,
1686 * read a byte at a time until we get the terminating '\0'. This method
1687 * is a bit inefficient, but since this is only used for relatively short
1688 * command and status strings, it shouldn't matter.
1689 */
1690 bufsize = 64; /* could be any number */
1691 msg = (char *) pg_malloc(bufsize);
1692 msgsize = 0;
1693 for (;;)
1694 {
1695 Assert(msgsize < bufsize);
1696 ret = piperead(fd, msg + msgsize, 1);
1697 if (ret <= 0)
1698 break; /* error or connection closure */
1699
1700 Assert(ret == 1);
1701
1702 if (msg[msgsize] == '\0')
1703 return msg; /* collected whole message */
1704
1705 msgsize++;
1706 if (msgsize == bufsize) /* enlarge buffer if needed */
1707 {
1708 bufsize += 16; /* could be any number */
1709 msg = (char *) pg_realloc(msg, bufsize);
1710 }
1711 }
1712
1713 /* Other end has closed the connection */
1714 pg_free(msg);
1715 return NULL;
1716}
#define piperead(a, b, c)
Definition: parallel.c:140
void pg_free(void *ptr)
Definition: fe_memutils.c:105
void * pg_realloc(void *ptr, size_t size)
Definition: fe_memutils.c:65
#define bufsize
Definition: indent_globs.h:36
static int fd(const char *x, int i)
Definition: preproc-init.c:105

References Assert(), bufsize, fd(), pg_free(), pg_malloc(), pg_realloc(), and piperead.

Referenced by getMessageFromLeader(), and getMessageFromWorker().

◆ replace_on_exit_close_archive()

void replace_on_exit_close_archive ( Archive AHX)

Definition at line 341 of file parallel.c.

342{
343 shutdown_info.AHX = AHX;
344}

References ShutdownInformation::AHX, and shutdown_info.

Referenced by restore_one_database().

◆ RunWorker()

static void RunWorker ( ArchiveHandle AH,
ParallelSlot slot 
)
static

Definition at line 841 of file parallel.c.

842{
843 int pipefd[2];
844
845 /* fetch child ends of pipes */
846 pipefd[PIPE_READ] = slot->pipeRevRead;
847 pipefd[PIPE_WRITE] = slot->pipeRevWrite;
848
849 /*
850 * Clone the archive so that we have our own state to work with, and in
851 * particular our own database connection.
852 *
853 * We clone on Unix as well as Windows, even though technically we don't
854 * need to because fork() gives us a copy in our own address space
855 * already. But CloneArchive resets the state information and also clones
856 * the database connection which both seem kinda helpful.
857 */
858 AH = CloneArchive(AH);
859
860 /* Remember cloned archive where signal handler can find it */
861 set_cancel_slot_archive(slot, AH);
862
863 /*
864 * Call the setup worker function that's defined in the ArchiveHandle.
865 */
866 (AH->SetupWorkerPtr) ((Archive *) AH);
867
868 /*
869 * Execute commands until done.
870 */
871 WaitForCommands(AH, pipefd);
872
873 /*
874 * Disconnect from database and clean up.
875 */
876 set_cancel_slot_archive(slot, NULL);
878 DeCloneArchive(AH);
879}
static void WaitForCommands(ArchiveHandle *AH, int pipefd[2])
Definition: parallel.c:1348
static void set_cancel_slot_archive(ParallelSlot *slot, ArchiveHandle *AH)
Definition: parallel.c:821
void DeCloneArchive(ArchiveHandle *AH)
ArchiveHandle * CloneArchive(ArchiveHandle *AH)
SetupWorkerPtrType SetupWorkerPtr

References CloneArchive(), DeCloneArchive(), DisconnectDatabase(), PIPE_READ, PIPE_WRITE, ParallelSlot::pipeRevRead, ParallelSlot::pipeRevWrite, _archiveHandle::public, set_cancel_slot_archive(), _archiveHandle::SetupWorkerPtr, and WaitForCommands().

Referenced by ParallelBackupStart().

◆ select_loop()

static int select_loop ( int  maxFd,
fd_set *  workerset 
)
static

Definition at line 1552 of file parallel.c.

1553{
1554 int i;
1555 fd_set saveSet = *workerset;
1556
1557 for (;;)
1558 {
1559 *workerset = saveSet;
1560 i = select(maxFd + 1, workerset, NULL, NULL, NULL);
1561
1562#ifndef WIN32
1563 if (i < 0 && errno == EINTR)
1564 continue;
1565#else
1566 if (i == SOCKET_ERROR && WSAGetLastError() == WSAEINTR)
1567 continue;
1568#endif
1569 break;
1570 }
1571
1572 return i;
1573}
#define EINTR
Definition: win32_port.h:364

References EINTR, i, and select.

Referenced by getMessageFromWorker().

◆ sendMessageToLeader()

static void sendMessageToLeader ( int  pipefd[2],
const char *  str 
)
static

Definition at line 1539 of file parallel.c.

1540{
1541 int len = strlen(str) + 1;
1542
1543 if (pipewrite(pipefd[PIPE_WRITE], str, len) != len)
1544 pg_fatal("could not write to the communication channel: %m");
1545}
#define pipewrite(a, b, c)
Definition: parallel.c:141
const void size_t len

References len, pg_fatal, PIPE_WRITE, pipewrite, and str.

Referenced by WaitForCommands().

◆ sendMessageToWorker()

static void sendMessageToWorker ( ParallelState pstate,
int  worker,
const char *  str 
)
static

Definition at line 1656 of file parallel.c.

1657{
1658 int len = strlen(str) + 1;
1659
1660 if (pipewrite(pstate->parallelSlot[worker].pipeWrite, str, len) != len)
1661 {
1662 pg_fatal("could not write to the communication channel: %m");
1663 }
1664}

References len, ParallelState::parallelSlot, pg_fatal, ParallelSlot::pipeWrite, pipewrite, and str.

Referenced by DispatchJobForTocEntry().

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

◆ set_cancel_handler()

static void set_cancel_handler ( void  )
static

Definition at line 620 of file parallel.c.

621{
622 /*
623 * When forking, signal_info.handler_set will propagate into the new
624 * process, but that's fine because the signal handler state does too.
625 */
627 {
629
630 pqsignal(SIGINT, sigTermHandler);
631 pqsignal(SIGTERM, sigTermHandler);
633 }
634}
static void sigTermHandler(SIGNAL_ARGS)
Definition: parallel.c:557
#define SIGQUIT
Definition: win32_port.h:159

References DumpSignalInformation::handler_set, pqsignal, signal_info, SIGQUIT, and sigTermHandler().

Referenced by set_archive_cancel_info().

◆ set_cancel_pstate()

static void set_cancel_pstate ( ParallelState pstate)
static

Definition at line 801 of file parallel.c.

802{
803#ifdef WIN32
804 EnterCriticalSection(&signal_info_lock);
805#endif
806
807 signal_info.pstate = pstate;
808
809#ifdef WIN32
810 LeaveCriticalSection(&signal_info_lock);
811#endif
812}
ParallelState * pstate
Definition: parallel.c:168

References DumpSignalInformation::pstate, and signal_info.

Referenced by ParallelBackupEnd(), and ParallelBackupStart().

◆ set_cancel_slot_archive()

static void set_cancel_slot_archive ( ParallelSlot slot,
ArchiveHandle AH 
)
static

Definition at line 821 of file parallel.c.

822{
823#ifdef WIN32
824 EnterCriticalSection(&signal_info_lock);
825#endif
826
827 slot->AH = AH;
828
829#ifdef WIN32
830 LeaveCriticalSection(&signal_info_lock);
831#endif
832}

References ParallelSlot::AH.

Referenced by RunWorker().

◆ ShutdownWorkersHard()

static void ShutdownWorkersHard ( ParallelState pstate)
static

Definition at line 407 of file parallel.c.

408{
409 int i;
410
411 /*
412 * Close our write end of the sockets so that any workers waiting for
413 * commands know they can exit. (Note: some of the pipeWrite fields might
414 * still be zero, if we failed to initialize all the workers. Hence, just
415 * ignore errors here.)
416 */
417 for (i = 0; i < pstate->numWorkers; i++)
419
420 /*
421 * Force early termination of any commands currently in progress.
422 */
423#ifndef WIN32
424 /* On non-Windows, send SIGTERM to each worker process. */
425 for (i = 0; i < pstate->numWorkers; i++)
426 {
427 pid_t pid = pstate->parallelSlot[i].pid;
428
429 if (pid != 0)
430 kill(pid, SIGTERM);
431 }
432#else
433
434 /*
435 * On Windows, send query cancels directly to the workers' backends. Use
436 * a critical section to ensure worker threads don't change state.
437 */
438 EnterCriticalSection(&signal_info_lock);
439 for (i = 0; i < pstate->numWorkers; i++)
440 {
441 ArchiveHandle *AH = pstate->parallelSlot[i].AH;
442 char errbuf[1];
443
444 if (AH != NULL && AH->connCancel != NULL)
445 (void) PQcancel(AH->connCancel, errbuf, sizeof(errbuf));
446 }
447 LeaveCriticalSection(&signal_info_lock);
448#endif
449
450 /* Now wait for them to terminate. */
452}
int PQcancel(PGcancel *cancel, char *errbuf, int errbufsize)
Definition: fe-cancel.c:530
#define kill(pid, sig)
Definition: win32_port.h:493

References ParallelSlot::AH, closesocket, _archiveHandle::connCancel, i, kill, ParallelState::numWorkers, ParallelState::parallelSlot, ParallelSlot::pid, ParallelSlot::pipeWrite, PQcancel(), and WaitForTerminatingWorkers().

Referenced by archive_close_connection().

◆ sigTermHandler()

static void sigTermHandler ( SIGNAL_ARGS  )
static

Definition at line 557 of file parallel.c.

558{
559 int i;
560 char errbuf[1];
561
562 /*
563 * Some platforms allow delivery of new signals to interrupt an active
564 * signal handler. That could muck up our attempt to send PQcancel, so
565 * disable the signals that set_cancel_handler enabled.
566 */
567 pqsignal(SIGINT, SIG_IGN);
568 pqsignal(SIGTERM, SIG_IGN);
569 pqsignal(SIGQUIT, SIG_IGN);
570
571 /*
572 * If we're in the leader, forward signal to all workers. (It seems best
573 * to do this before PQcancel; killing the leader transaction will result
574 * in invalid-snapshot errors from active workers, which maybe we can
575 * quiet by killing workers first.) Ignore any errors.
576 */
577 if (signal_info.pstate != NULL)
578 {
579 for (i = 0; i < signal_info.pstate->numWorkers; i++)
580 {
581 pid_t pid = signal_info.pstate->parallelSlot[i].pid;
582
583 if (pid != 0)
584 kill(pid, SIGTERM);
585 }
586 }
587
588 /*
589 * Send QueryCancel if we have a connection to send to. Ignore errors,
590 * there's not much we can do about them anyway.
591 */
592 if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL)
593 (void) PQcancel(signal_info.myAH->connCancel, errbuf, sizeof(errbuf));
594
595 /*
596 * Report we're quitting, using nothing more complicated than write(2).
597 * When in parallel operation, only the leader process should do this.
598 */
600 {
601 if (progname)
602 {
604 write_stderr(": ");
605 }
606 write_stderr("terminated by user\n");
607 }
608
609 /*
610 * And die, using _exit() not exit() because the latter will invoke atexit
611 * handlers that can fail if we interrupted related code.
612 */
613 _exit(1);
614}
#define write_stderr(str)
Definition: parallel.c:186
const char * progname
Definition: main.c:44

References DumpSignalInformation::am_worker, _archiveHandle::connCancel, i, kill, DumpSignalInformation::myAH, ParallelState::numWorkers, ParallelState::parallelSlot, ParallelSlot::pid, PQcancel(), pqsignal, progname, DumpSignalInformation::pstate, signal_info, SIGQUIT, and write_stderr.

Referenced by set_cancel_handler().

◆ WaitForCommands()

static void WaitForCommands ( ArchiveHandle AH,
int  pipefd[2] 
)
static

Definition at line 1348 of file parallel.c.

1349{
1350 char *command;
1351 TocEntry *te;
1352 T_Action act;
1353 int status = 0;
1354 char buf[256];
1355
1356 for (;;)
1357 {
1358 if (!(command = getMessageFromLeader(pipefd)))
1359 {
1360 /* EOF, so done */
1361 return;
1362 }
1363
1364 /* Decode the command */
1365 parseWorkerCommand(AH, &te, &act, command);
1366
1367 if (act == ACT_DUMP)
1368 {
1369 /* Acquire lock on this table within the worker's session */
1370 lockTableForWorker(AH, te);
1371
1372 /* Perform the dump command */
1373 status = (AH->WorkerJobDumpPtr) (AH, te);
1374 }
1375 else if (act == ACT_RESTORE)
1376 {
1377 /* Perform the restore command */
1378 status = (AH->WorkerJobRestorePtr) (AH, te);
1379 }
1380 else
1381 Assert(false);
1382
1383 /* Return status to leader */
1384 buildWorkerResponse(AH, te, act, status, buf, sizeof(buf));
1385
1386 sendMessageToLeader(pipefd, buf);
1387
1388 /* command was pg_malloc'd and we are responsible for free()ing it. */
1389 free(command);
1390 }
1391}
static void sendMessageToLeader(int pipefd[2], const char *str)
Definition: parallel.c:1539
static char * getMessageFromLeader(int pipefd[2])
Definition: parallel.c:1528
static void lockTableForWorker(ArchiveHandle *AH, TocEntry *te)
Definition: parallel.c:1313
static void buildWorkerResponse(ArchiveHandle *AH, TocEntry *te, T_Action act, int status, char *buf, int buflen)
Definition: parallel.c:1168
static void parseWorkerCommand(ArchiveHandle *AH, TocEntry **te, T_Action *act, const char *msg)
Definition: parallel.c:1135
WorkerJobDumpPtrType WorkerJobDumpPtr
WorkerJobRestorePtrType WorkerJobRestorePtr

References ACT_DUMP, ACT_RESTORE, Assert(), buf, buildWorkerResponse(), free, getMessageFromLeader(), lockTableForWorker(), parseWorkerCommand(), sendMessageToLeader(), _archiveHandle::WorkerJobDumpPtr, and _archiveHandle::WorkerJobRestorePtr.

Referenced by RunWorker().

◆ WaitForTerminatingWorkers()

static void WaitForTerminatingWorkers ( ParallelState pstate)
static

Definition at line 458 of file parallel.c.

459{
460 while (!HasEveryWorkerTerminated(pstate))
461 {
462 ParallelSlot *slot = NULL;
463 int j;
464
465#ifndef WIN32
466 /* On non-Windows, use wait() to wait for next worker to end */
467 int status;
468 pid_t pid = wait(&status);
469
470 /* Find dead worker's slot, and clear the PID field */
471 for (j = 0; j < pstate->numWorkers; j++)
472 {
473 slot = &(pstate->parallelSlot[j]);
474 if (slot->pid == pid)
475 {
476 slot->pid = 0;
477 break;
478 }
479 }
480#else /* WIN32 */
481 /* On Windows, we must use WaitForMultipleObjects() */
482 HANDLE *lpHandles = pg_malloc(sizeof(HANDLE) * pstate->numWorkers);
483 int nrun = 0;
484 DWORD ret;
485 uintptr_t hThread;
486
487 for (j = 0; j < pstate->numWorkers; j++)
488 {
490 {
491 lpHandles[nrun] = (HANDLE) pstate->parallelSlot[j].hThread;
492 nrun++;
493 }
494 }
495 ret = WaitForMultipleObjects(nrun, lpHandles, false, INFINITE);
496 Assert(ret != WAIT_FAILED);
497 hThread = (uintptr_t) lpHandles[ret - WAIT_OBJECT_0];
498 free(lpHandles);
499
500 /* Find dead worker's slot, and clear the hThread field */
501 for (j = 0; j < pstate->numWorkers; j++)
502 {
503 slot = &(pstate->parallelSlot[j]);
504 if (slot->hThread == hThread)
505 {
506 /* For cleanliness, close handles for dead threads */
507 CloseHandle((HANDLE) slot->hThread);
508 slot->hThread = (uintptr_t) INVALID_HANDLE_VALUE;
509 break;
510 }
511 }
512#endif /* WIN32 */
513
514 /* On all platforms, update workerStatus and te[] as well */
515 Assert(j < pstate->numWorkers);
517 pstate->te[j] = NULL;
518 }
519}
static bool HasEveryWorkerTerminated(ParallelState *pstate)
Definition: parallel.c:1264

References Assert(), free, HasEveryWorkerTerminated(), j, ParallelState::numWorkers, ParallelState::parallelSlot, pg_malloc(), ParallelSlot::pid, ParallelState::te, WORKER_IS_RUNNING, ParallelSlot::workerStatus, and WRKR_TERMINATED.

Referenced by ParallelBackupEnd(), and ShutdownWorkersHard().

◆ 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
@ WFW_ALL_IDLE
Definition: parallel.h:35
@ WFW_GOT_STATUS
Definition: parallel.h:33
@ WFW_NO_WAIT
Definition: parallel.h:32
static PgChecksumMode mode
Definition: pg_checksums.c:55

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

Variable Documentation

◆ shutdown_info

◆ signal_info