code: replace 'master' with 'leader' where appropriate.
Leader already is the more widely used terminology, but a few places didn't get the message. Author: Andres Freund Reviewed-By: David Steele Discussion: https://postgr.es/m/20200615182235.x7lch5n6kcjq4aue@alap3.anarazel.de
This commit is contained in:
parent
5e7bbb5286
commit
e07633646a
|
@ -11,7 +11,7 @@
|
||||||
* pages from a relation that is in the process of being dropped.
|
* pages from a relation that is in the process of being dropped.
|
||||||
*
|
*
|
||||||
* While prewarming, autoprewarm will use two workers. There's a
|
* While prewarming, autoprewarm will use two workers. There's a
|
||||||
* master worker that reads and sorts the list of blocks to be
|
* leader worker that reads and sorts the list of blocks to be
|
||||||
* prewarmed and then launches a per-database worker for each
|
* prewarmed and then launches a per-database worker for each
|
||||||
* relevant database in turn. The former keeps running after the
|
* relevant database in turn. The former keeps running after the
|
||||||
* initial prewarm is complete to update the dump file periodically.
|
* initial prewarm is complete to update the dump file periodically.
|
||||||
|
@ -88,7 +88,7 @@ PG_FUNCTION_INFO_V1(autoprewarm_dump_now);
|
||||||
|
|
||||||
static void apw_load_buffers(void);
|
static void apw_load_buffers(void);
|
||||||
static int apw_dump_now(bool is_bgworker, bool dump_unlogged);
|
static int apw_dump_now(bool is_bgworker, bool dump_unlogged);
|
||||||
static void apw_start_master_worker(void);
|
static void apw_start_leader_worker(void);
|
||||||
static void apw_start_database_worker(void);
|
static void apw_start_database_worker(void);
|
||||||
static bool apw_init_shmem(void);
|
static bool apw_init_shmem(void);
|
||||||
static void apw_detach_shmem(int code, Datum arg);
|
static void apw_detach_shmem(int code, Datum arg);
|
||||||
|
@ -146,11 +146,11 @@ _PG_init(void)
|
||||||
|
|
||||||
/* Register autoprewarm worker, if enabled. */
|
/* Register autoprewarm worker, if enabled. */
|
||||||
if (autoprewarm)
|
if (autoprewarm)
|
||||||
apw_start_master_worker();
|
apw_start_leader_worker();
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Main entry point for the master autoprewarm process. Per-database workers
|
* Main entry point for the leader autoprewarm process. Per-database workers
|
||||||
* have a separate entry point.
|
* have a separate entry point.
|
||||||
*/
|
*/
|
||||||
void
|
void
|
||||||
|
@ -716,7 +716,7 @@ autoprewarm_start_worker(PG_FUNCTION_ARGS)
|
||||||
errmsg("autoprewarm worker is already running under PID %lu",
|
errmsg("autoprewarm worker is already running under PID %lu",
|
||||||
(unsigned long) pid)));
|
(unsigned long) pid)));
|
||||||
|
|
||||||
apw_start_master_worker();
|
apw_start_leader_worker();
|
||||||
|
|
||||||
PG_RETURN_VOID();
|
PG_RETURN_VOID();
|
||||||
}
|
}
|
||||||
|
@ -786,10 +786,10 @@ apw_detach_shmem(int code, Datum arg)
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Start autoprewarm master worker process.
|
* Start autoprewarm leader worker process.
|
||||||
*/
|
*/
|
||||||
static void
|
static void
|
||||||
apw_start_master_worker(void)
|
apw_start_leader_worker(void)
|
||||||
{
|
{
|
||||||
BackgroundWorker worker;
|
BackgroundWorker worker;
|
||||||
BackgroundWorkerHandle *handle;
|
BackgroundWorkerHandle *handle;
|
||||||
|
@ -801,8 +801,8 @@ apw_start_master_worker(void)
|
||||||
worker.bgw_start_time = BgWorkerStart_ConsistentState;
|
worker.bgw_start_time = BgWorkerStart_ConsistentState;
|
||||||
strcpy(worker.bgw_library_name, "pg_prewarm");
|
strcpy(worker.bgw_library_name, "pg_prewarm");
|
||||||
strcpy(worker.bgw_function_name, "autoprewarm_main");
|
strcpy(worker.bgw_function_name, "autoprewarm_main");
|
||||||
strcpy(worker.bgw_name, "autoprewarm master");
|
strcpy(worker.bgw_name, "autoprewarm leader");
|
||||||
strcpy(worker.bgw_type, "autoprewarm master");
|
strcpy(worker.bgw_type, "autoprewarm leader");
|
||||||
|
|
||||||
if (process_shared_preload_libraries_in_progress)
|
if (process_shared_preload_libraries_in_progress)
|
||||||
{
|
{
|
||||||
|
|
|
@ -332,12 +332,12 @@ PostgreSQL documentation
|
||||||
</para>
|
</para>
|
||||||
<para>
|
<para>
|
||||||
Requesting exclusive locks on database objects while running a parallel dump could
|
Requesting exclusive locks on database objects while running a parallel dump could
|
||||||
cause the dump to fail. The reason is that the <application>pg_dump</application> master process
|
cause the dump to fail. The reason is that the <application>pg_dump</application> leader process
|
||||||
requests shared locks on the objects that the worker processes are going to dump later
|
requests shared locks on the objects that the worker processes are going to dump later
|
||||||
in order to
|
in order to
|
||||||
make sure that nobody deletes them and makes them go away while the dump is running.
|
make sure that nobody deletes them and makes them go away while the dump is running.
|
||||||
If another client then requests an exclusive lock on a table, that lock will not be
|
If another client then requests an exclusive lock on a table, that lock will not be
|
||||||
granted but will be queued waiting for the shared lock of the master process to be
|
granted but will be queued waiting for the shared lock of the leader process to be
|
||||||
released. Consequently any other access to the table will not be granted either and
|
released. Consequently any other access to the table will not be granted either and
|
||||||
will queue after the exclusive lock request. This includes the worker process trying
|
will queue after the exclusive lock request. This includes the worker process trying
|
||||||
to dump the table. Without any precautions this would be a classic deadlock situation.
|
to dump the table. Without any precautions this would be a classic deadlock situation.
|
||||||
|
@ -354,14 +354,14 @@ PostgreSQL documentation
|
||||||
for standbys. With this feature, database clients can ensure they see
|
for standbys. With this feature, database clients can ensure they see
|
||||||
the same data set even though they use different connections.
|
the same data set even though they use different connections.
|
||||||
<command>pg_dump -j</command> uses multiple database connections; it
|
<command>pg_dump -j</command> uses multiple database connections; it
|
||||||
connects to the database once with the master process and once again
|
connects to the database once with the leader process and once again
|
||||||
for each worker job. Without the synchronized snapshot feature, the
|
for each worker job. Without the synchronized snapshot feature, the
|
||||||
different worker jobs wouldn't be guaranteed to see the same data in
|
different worker jobs wouldn't be guaranteed to see the same data in
|
||||||
each connection, which could lead to an inconsistent backup.
|
each connection, which could lead to an inconsistent backup.
|
||||||
</para>
|
</para>
|
||||||
<para>
|
<para>
|
||||||
If you want to run a parallel dump of a pre-9.2 server, you need to make sure that the
|
If you want to run a parallel dump of a pre-9.2 server, you need to make sure that the
|
||||||
database content doesn't change from between the time the master connects to the
|
database content doesn't change from between the time the leader connects to the
|
||||||
database until the last worker job has connected to the database. The easiest way to
|
database until the last worker job has connected to the database. The easiest way to
|
||||||
do this is to halt any data modifying processes (DDL and DML) accessing the database
|
do this is to halt any data modifying processes (DDL and DML) accessing the database
|
||||||
before starting the backup. You also need to specify the
|
before starting the backup. You also need to specify the
|
||||||
|
|
|
@ -89,9 +89,9 @@ typedef struct FixedParallelState
|
||||||
Oid temp_toast_namespace_id;
|
Oid temp_toast_namespace_id;
|
||||||
int sec_context;
|
int sec_context;
|
||||||
bool is_superuser;
|
bool is_superuser;
|
||||||
PGPROC *parallel_master_pgproc;
|
PGPROC *parallel_leader_pgproc;
|
||||||
pid_t parallel_master_pid;
|
pid_t parallel_leader_pid;
|
||||||
BackendId parallel_master_backend_id;
|
BackendId parallel_leader_backend_id;
|
||||||
TimestampTz xact_ts;
|
TimestampTz xact_ts;
|
||||||
TimestampTz stmt_ts;
|
TimestampTz stmt_ts;
|
||||||
SerializableXactHandle serializable_xact_handle;
|
SerializableXactHandle serializable_xact_handle;
|
||||||
|
@ -124,7 +124,7 @@ static FixedParallelState *MyFixedParallelState;
|
||||||
static dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list);
|
static dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list);
|
||||||
|
|
||||||
/* Backend-local copy of data from FixedParallelState. */
|
/* Backend-local copy of data from FixedParallelState. */
|
||||||
static pid_t ParallelMasterPid;
|
static pid_t ParallelLeaderPid;
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* List of internal parallel worker entry points. We need this for
|
* List of internal parallel worker entry points. We need this for
|
||||||
|
@ -323,9 +323,9 @@ InitializeParallelDSM(ParallelContext *pcxt)
|
||||||
GetUserIdAndSecContext(&fps->current_user_id, &fps->sec_context);
|
GetUserIdAndSecContext(&fps->current_user_id, &fps->sec_context);
|
||||||
GetTempNamespaceState(&fps->temp_namespace_id,
|
GetTempNamespaceState(&fps->temp_namespace_id,
|
||||||
&fps->temp_toast_namespace_id);
|
&fps->temp_toast_namespace_id);
|
||||||
fps->parallel_master_pgproc = MyProc;
|
fps->parallel_leader_pgproc = MyProc;
|
||||||
fps->parallel_master_pid = MyProcPid;
|
fps->parallel_leader_pid = MyProcPid;
|
||||||
fps->parallel_master_backend_id = MyBackendId;
|
fps->parallel_leader_backend_id = MyBackendId;
|
||||||
fps->xact_ts = GetCurrentTransactionStartTimestamp();
|
fps->xact_ts = GetCurrentTransactionStartTimestamp();
|
||||||
fps->stmt_ts = GetCurrentStatementStartTimestamp();
|
fps->stmt_ts = GetCurrentStatementStartTimestamp();
|
||||||
fps->serializable_xact_handle = ShareSerializableXact();
|
fps->serializable_xact_handle = ShareSerializableXact();
|
||||||
|
@ -857,8 +857,8 @@ WaitForParallelWorkersToFinish(ParallelContext *pcxt)
|
||||||
*
|
*
|
||||||
* This function ensures that workers have been completely shutdown. The
|
* This function ensures that workers have been completely shutdown. The
|
||||||
* difference between WaitForParallelWorkersToFinish and this function is
|
* difference between WaitForParallelWorkersToFinish and this function is
|
||||||
* that former just ensures that last message sent by worker backend is
|
* that the former just ensures that last message sent by a worker backend is
|
||||||
* received by master backend whereas this ensures the complete shutdown.
|
* received by the leader backend whereas this ensures the complete shutdown.
|
||||||
*/
|
*/
|
||||||
static void
|
static void
|
||||||
WaitForParallelWorkersToExit(ParallelContext *pcxt)
|
WaitForParallelWorkersToExit(ParallelContext *pcxt)
|
||||||
|
@ -1302,8 +1302,8 @@ ParallelWorkerMain(Datum main_arg)
|
||||||
MyFixedParallelState = fps;
|
MyFixedParallelState = fps;
|
||||||
|
|
||||||
/* Arrange to signal the leader if we exit. */
|
/* Arrange to signal the leader if we exit. */
|
||||||
ParallelMasterPid = fps->parallel_master_pid;
|
ParallelLeaderPid = fps->parallel_leader_pid;
|
||||||
ParallelMasterBackendId = fps->parallel_master_backend_id;
|
ParallelLeaderBackendId = fps->parallel_leader_backend_id;
|
||||||
on_shmem_exit(ParallelWorkerShutdown, (Datum) 0);
|
on_shmem_exit(ParallelWorkerShutdown, (Datum) 0);
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
@ -1318,8 +1318,8 @@ ParallelWorkerMain(Datum main_arg)
|
||||||
shm_mq_set_sender(mq, MyProc);
|
shm_mq_set_sender(mq, MyProc);
|
||||||
mqh = shm_mq_attach(mq, seg, NULL);
|
mqh = shm_mq_attach(mq, seg, NULL);
|
||||||
pq_redirect_to_shm_mq(seg, mqh);
|
pq_redirect_to_shm_mq(seg, mqh);
|
||||||
pq_set_parallel_master(fps->parallel_master_pid,
|
pq_set_parallel_leader(fps->parallel_leader_pid,
|
||||||
fps->parallel_master_backend_id);
|
fps->parallel_leader_backend_id);
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Send a BackendKeyData message to the process that initiated parallelism
|
* Send a BackendKeyData message to the process that initiated parallelism
|
||||||
|
@ -1347,8 +1347,8 @@ ParallelWorkerMain(Datum main_arg)
|
||||||
* deadlock. (If we can't join the lock group, the leader has gone away,
|
* deadlock. (If we can't join the lock group, the leader has gone away,
|
||||||
* so just exit quietly.)
|
* so just exit quietly.)
|
||||||
*/
|
*/
|
||||||
if (!BecomeLockGroupMember(fps->parallel_master_pgproc,
|
if (!BecomeLockGroupMember(fps->parallel_leader_pgproc,
|
||||||
fps->parallel_master_pid))
|
fps->parallel_leader_pid))
|
||||||
return;
|
return;
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
@ -1410,7 +1410,7 @@ ParallelWorkerMain(Datum main_arg)
|
||||||
/* Restore transaction snapshot. */
|
/* Restore transaction snapshot. */
|
||||||
tsnapspace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT, false);
|
tsnapspace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT, false);
|
||||||
RestoreTransactionSnapshot(RestoreSnapshot(tsnapspace),
|
RestoreTransactionSnapshot(RestoreSnapshot(tsnapspace),
|
||||||
fps->parallel_master_pgproc);
|
fps->parallel_leader_pgproc);
|
||||||
|
|
||||||
/* Restore active snapshot. */
|
/* Restore active snapshot. */
|
||||||
asnapspace = shm_toc_lookup(toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, false);
|
asnapspace = shm_toc_lookup(toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, false);
|
||||||
|
@ -1510,9 +1510,9 @@ ParallelWorkerReportLastRecEnd(XLogRecPtr last_xlog_end)
|
||||||
static void
|
static void
|
||||||
ParallelWorkerShutdown(int code, Datum arg)
|
ParallelWorkerShutdown(int code, Datum arg)
|
||||||
{
|
{
|
||||||
SendProcSignal(ParallelMasterPid,
|
SendProcSignal(ParallelLeaderPid,
|
||||||
PROCSIG_PARALLEL_MESSAGE,
|
PROCSIG_PARALLEL_MESSAGE,
|
||||||
ParallelMasterBackendId);
|
ParallelLeaderBackendId);
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
|
|
@ -750,7 +750,7 @@ GetCurrentCommandId(bool used)
|
||||||
{
|
{
|
||||||
/*
|
/*
|
||||||
* Forbid setting currentCommandIdUsed in a parallel worker, because
|
* Forbid setting currentCommandIdUsed in a parallel worker, because
|
||||||
* we have no provision for communicating this back to the master. We
|
* we have no provision for communicating this back to the leader. We
|
||||||
* could relax this restriction when currentCommandIdUsed was already
|
* could relax this restriction when currentCommandIdUsed was already
|
||||||
* true at the start of the parallel operation.
|
* true at the start of the parallel operation.
|
||||||
*/
|
*/
|
||||||
|
@ -987,7 +987,7 @@ ExitParallelMode(void)
|
||||||
/*
|
/*
|
||||||
* IsInParallelMode
|
* IsInParallelMode
|
||||||
*
|
*
|
||||||
* Are we in a parallel operation, as either the master or a worker? Check
|
* Are we in a parallel operation, as either the leader or a worker? Check
|
||||||
* this to prohibit operations that change backend-local state expected to
|
* this to prohibit operations that change backend-local state expected to
|
||||||
* match across all workers. Mere caches usually don't require such a
|
* match across all workers. Mere caches usually don't require such a
|
||||||
* restriction. State modified in a strict push/pop fashion, such as the
|
* restriction. State modified in a strict push/pop fashion, such as the
|
||||||
|
@ -2164,13 +2164,13 @@ CommitTransaction(void)
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
/*
|
/*
|
||||||
* We must not mark our XID committed; the parallel master is
|
* We must not mark our XID committed; the parallel leader is
|
||||||
* responsible for that.
|
* responsible for that.
|
||||||
*/
|
*/
|
||||||
latestXid = InvalidTransactionId;
|
latestXid = InvalidTransactionId;
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Make sure the master will know about any WAL we wrote before it
|
* Make sure the leader will know about any WAL we wrote before it
|
||||||
* commits.
|
* commits.
|
||||||
*/
|
*/
|
||||||
ParallelWorkerReportLastRecEnd(XactLastRecEnd);
|
ParallelWorkerReportLastRecEnd(XactLastRecEnd);
|
||||||
|
@ -2699,7 +2699,7 @@ AbortTransaction(void)
|
||||||
latestXid = InvalidTransactionId;
|
latestXid = InvalidTransactionId;
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Since the parallel master won't get our value of XactLastRecEnd in
|
* Since the parallel leader won't get our value of XactLastRecEnd in
|
||||||
* this case, we nudge WAL-writer ourselves in this case. See related
|
* this case, we nudge WAL-writer ourselves in this case. See related
|
||||||
* comments in RecordTransactionAbort for why this matters.
|
* comments in RecordTransactionAbort for why this matters.
|
||||||
*/
|
*/
|
||||||
|
@ -4488,7 +4488,7 @@ RollbackAndReleaseCurrentSubTransaction(void)
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Unlike ReleaseCurrentSubTransaction(), this is nominally permitted
|
* Unlike ReleaseCurrentSubTransaction(), this is nominally permitted
|
||||||
* during parallel operations. That's because we may be in the master,
|
* during parallel operations. That's because we may be in the leader,
|
||||||
* recovering from an error thrown while we were in parallel mode. We
|
* recovering from an error thrown while we were in parallel mode. We
|
||||||
* won't reach here in a worker, because BeginInternalSubTransaction()
|
* won't reach here in a worker, because BeginInternalSubTransaction()
|
||||||
* will have failed.
|
* will have failed.
|
||||||
|
|
|
@ -190,7 +190,7 @@ BuildTupleHashTableExt(PlanState *parent,
|
||||||
hashtable->cur_eq_func = NULL;
|
hashtable->cur_eq_func = NULL;
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* If parallelism is in use, even if the master backend is performing the
|
* If parallelism is in use, even if the leader backend is performing the
|
||||||
* scan itself, we don't want to create the hashtable exactly the same way
|
* scan itself, we don't want to create the hashtable exactly the same way
|
||||||
* in all workers. As hashtables are iterated over in keyspace-order,
|
* in all workers. As hashtables are iterated over in keyspace-order,
|
||||||
* doing so in all processes in the same way is likely to lead to
|
* doing so in all processes in the same way is likely to lead to
|
||||||
|
|
|
@ -23,8 +23,8 @@
|
||||||
|
|
||||||
static shm_mq_handle *pq_mq_handle;
|
static shm_mq_handle *pq_mq_handle;
|
||||||
static bool pq_mq_busy = false;
|
static bool pq_mq_busy = false;
|
||||||
static pid_t pq_mq_parallel_master_pid = 0;
|
static pid_t pq_mq_parallel_leader_pid = 0;
|
||||||
static pid_t pq_mq_parallel_master_backend_id = InvalidBackendId;
|
static pid_t pq_mq_parallel_leader_backend_id = InvalidBackendId;
|
||||||
|
|
||||||
static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg);
|
static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg);
|
||||||
static void mq_comm_reset(void);
|
static void mq_comm_reset(void);
|
||||||
|
@ -73,15 +73,15 @@ pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Arrange to SendProcSignal() to the parallel master each time we transmit
|
* Arrange to SendProcSignal() to the parallel leader each time we transmit
|
||||||
* message data via the shm_mq.
|
* message data via the shm_mq.
|
||||||
*/
|
*/
|
||||||
void
|
void
|
||||||
pq_set_parallel_master(pid_t pid, BackendId backend_id)
|
pq_set_parallel_leader(pid_t pid, BackendId backend_id)
|
||||||
{
|
{
|
||||||
Assert(PqCommMethods == &PqCommMqMethods);
|
Assert(PqCommMethods == &PqCommMqMethods);
|
||||||
pq_mq_parallel_master_pid = pid;
|
pq_mq_parallel_leader_pid = pid;
|
||||||
pq_mq_parallel_master_backend_id = backend_id;
|
pq_mq_parallel_leader_backend_id = backend_id;
|
||||||
}
|
}
|
||||||
|
|
||||||
static void
|
static void
|
||||||
|
@ -160,10 +160,10 @@ mq_putmessage(char msgtype, const char *s, size_t len)
|
||||||
{
|
{
|
||||||
result = shm_mq_sendv(pq_mq_handle, iov, 2, true);
|
result = shm_mq_sendv(pq_mq_handle, iov, 2, true);
|
||||||
|
|
||||||
if (pq_mq_parallel_master_pid != 0)
|
if (pq_mq_parallel_leader_pid != 0)
|
||||||
SendProcSignal(pq_mq_parallel_master_pid,
|
SendProcSignal(pq_mq_parallel_leader_pid,
|
||||||
PROCSIG_PARALLEL_MESSAGE,
|
PROCSIG_PARALLEL_MESSAGE,
|
||||||
pq_mq_parallel_master_backend_id);
|
pq_mq_parallel_leader_backend_id);
|
||||||
|
|
||||||
if (result != SHM_MQ_WOULD_BLOCK)
|
if (result != SHM_MQ_WOULD_BLOCK)
|
||||||
break;
|
break;
|
||||||
|
|
|
@ -11,7 +11,7 @@
|
||||||
* cpu_tuple_cost Cost of typical CPU time to process a tuple
|
* cpu_tuple_cost Cost of typical CPU time to process a tuple
|
||||||
* cpu_index_tuple_cost Cost of typical CPU time to process an index tuple
|
* cpu_index_tuple_cost Cost of typical CPU time to process an index tuple
|
||||||
* cpu_operator_cost Cost of CPU time to execute an operator or function
|
* cpu_operator_cost Cost of CPU time to execute an operator or function
|
||||||
* parallel_tuple_cost Cost of CPU time to pass a tuple from worker to master backend
|
* parallel_tuple_cost Cost of CPU time to pass a tuple from worker to leader backend
|
||||||
* parallel_setup_cost Cost of setting up shared memory for parallelism
|
* parallel_setup_cost Cost of setting up shared memory for parallelism
|
||||||
*
|
*
|
||||||
* We expect that the kernel will typically do some amount of read-ahead
|
* We expect that the kernel will typically do some amount of read-ahead
|
||||||
|
|
|
@ -1028,8 +1028,8 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
|
||||||
* We can't pass Params to workers at the moment either, so they are also
|
* We can't pass Params to workers at the moment either, so they are also
|
||||||
* parallel-restricted, unless they are PARAM_EXTERN Params or are
|
* parallel-restricted, unless they are PARAM_EXTERN Params or are
|
||||||
* PARAM_EXEC Params listed in safe_param_ids, meaning they could be
|
* PARAM_EXEC Params listed in safe_param_ids, meaning they could be
|
||||||
* either generated within the worker or can be computed in master and
|
* either generated within workers or can be computed by the leader and
|
||||||
* then their value can be passed to the worker.
|
* then their value can be passed to workers.
|
||||||
*/
|
*/
|
||||||
else if (IsA(node, Param))
|
else if (IsA(node, Param))
|
||||||
{
|
{
|
||||||
|
|
|
@ -80,7 +80,7 @@ char postgres_exec_path[MAXPGPATH]; /* full path to backend */
|
||||||
|
|
||||||
BackendId MyBackendId = InvalidBackendId;
|
BackendId MyBackendId = InvalidBackendId;
|
||||||
|
|
||||||
BackendId ParallelMasterBackendId = InvalidBackendId;
|
BackendId ParallelLeaderBackendId = InvalidBackendId;
|
||||||
|
|
||||||
Oid MyDatabaseId = InvalidOid;
|
Oid MyDatabaseId = InvalidOid;
|
||||||
|
|
||||||
|
|
|
@ -3448,7 +3448,7 @@ static struct config_real ConfigureNamesReal[] =
|
||||||
{
|
{
|
||||||
{"parallel_tuple_cost", PGC_USERSET, QUERY_TUNING_COST,
|
{"parallel_tuple_cost", PGC_USERSET, QUERY_TUNING_COST,
|
||||||
gettext_noop("Sets the planner's estimate of the cost of "
|
gettext_noop("Sets the planner's estimate of the cost of "
|
||||||
"passing each tuple (row) from worker to master backend."),
|
"passing each tuple (row) from worker to leader backend."),
|
||||||
NULL,
|
NULL,
|
||||||
GUC_EXPLAIN
|
GUC_EXPLAIN
|
||||||
},
|
},
|
||||||
|
|
|
@ -16,20 +16,20 @@
|
||||||
/*
|
/*
|
||||||
* Parallel operation works like this:
|
* Parallel operation works like this:
|
||||||
*
|
*
|
||||||
* The original, master process calls ParallelBackupStart(), which forks off
|
* The original, leader process calls ParallelBackupStart(), which forks off
|
||||||
* the desired number of worker processes, which each enter WaitForCommands().
|
* the desired number of worker processes, which each enter WaitForCommands().
|
||||||
*
|
*
|
||||||
* The master process dispatches an individual work item to one of the worker
|
* The leader process dispatches an individual work item to one of the worker
|
||||||
* processes in DispatchJobForTocEntry(). We send a command string such as
|
* processes in DispatchJobForTocEntry(). We send a command string such as
|
||||||
* "DUMP 1234" or "RESTORE 1234", where 1234 is the TocEntry ID.
|
* "DUMP 1234" or "RESTORE 1234", where 1234 is the TocEntry ID.
|
||||||
* The worker process receives and decodes the command and passes it to the
|
* The worker process receives and decodes the command and passes it to the
|
||||||
* routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr,
|
* routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr,
|
||||||
* which are routines of the current archive format. That routine performs
|
* which are routines of the current archive format. That routine performs
|
||||||
* the required action (dump or restore) and returns an integer status code.
|
* the required action (dump or restore) and returns an integer status code.
|
||||||
* This is passed back to the master where we pass it to the
|
* This is passed back to the leader where we pass it to the
|
||||||
* ParallelCompletionPtr callback function that was passed to
|
* ParallelCompletionPtr callback function that was passed to
|
||||||
* DispatchJobForTocEntry(). The callback function does state updating
|
* DispatchJobForTocEntry(). The callback function does state updating
|
||||||
* for the master control logic in pg_backup_archiver.c.
|
* for the leader control logic in pg_backup_archiver.c.
|
||||||
*
|
*
|
||||||
* In principle additional archive-format-specific information might be needed
|
* In principle additional archive-format-specific information might be needed
|
||||||
* in commands or worker status responses, but so far that hasn't proved
|
* in commands or worker status responses, but so far that hasn't proved
|
||||||
|
@ -40,7 +40,7 @@
|
||||||
* threads in the same process. To avoid problems, they work with cloned
|
* threads in the same process. To avoid problems, they work with cloned
|
||||||
* copies of the Archive data structure; see RunWorker().)
|
* copies of the Archive data structure; see RunWorker().)
|
||||||
*
|
*
|
||||||
* In the master process, the workerStatus field for each worker has one of
|
* In the leader process, the workerStatus field for each worker has one of
|
||||||
* the following values:
|
* the following values:
|
||||||
* WRKR_NOT_STARTED: we've not yet forked this worker
|
* WRKR_NOT_STARTED: we've not yet forked this worker
|
||||||
* WRKR_IDLE: it's waiting for a command
|
* WRKR_IDLE: it's waiting for a command
|
||||||
|
@ -88,8 +88,8 @@ typedef enum
|
||||||
/*
|
/*
|
||||||
* Private per-parallel-worker state (typedef for this is in parallel.h).
|
* Private per-parallel-worker state (typedef for this is in parallel.h).
|
||||||
*
|
*
|
||||||
* Much of this is valid only in the master process (or, on Windows, should
|
* Much of this is valid only in the leader process (or, on Windows, should
|
||||||
* be touched only by the master thread). But the AH field should be touched
|
* be touched only by the leader thread). But the AH field should be touched
|
||||||
* only by workers. The pipe descriptors are valid everywhere.
|
* only by workers. The pipe descriptors are valid everywhere.
|
||||||
*/
|
*/
|
||||||
struct ParallelSlot
|
struct ParallelSlot
|
||||||
|
@ -102,7 +102,7 @@ struct ParallelSlot
|
||||||
|
|
||||||
ArchiveHandle *AH; /* Archive data worker is using */
|
ArchiveHandle *AH; /* Archive data worker is using */
|
||||||
|
|
||||||
int pipeRead; /* master's end of the pipes */
|
int pipeRead; /* leader's end of the pipes */
|
||||||
int pipeWrite;
|
int pipeWrite;
|
||||||
int pipeRevRead; /* child's end of the pipes */
|
int pipeRevRead; /* child's end of the pipes */
|
||||||
int pipeRevWrite;
|
int pipeRevWrite;
|
||||||
|
@ -124,7 +124,7 @@ struct ParallelSlot
|
||||||
*/
|
*/
|
||||||
typedef struct
|
typedef struct
|
||||||
{
|
{
|
||||||
ArchiveHandle *AH; /* master database connection */
|
ArchiveHandle *AH; /* leader database connection */
|
||||||
ParallelSlot *slot; /* this worker's parallel slot */
|
ParallelSlot *slot; /* this worker's parallel slot */
|
||||||
} WorkerInfo;
|
} WorkerInfo;
|
||||||
|
|
||||||
|
@ -157,9 +157,9 @@ static ShutdownInformation shutdown_info;
|
||||||
* State info for signal handling.
|
* State info for signal handling.
|
||||||
* We assume signal_info initializes to zeroes.
|
* We assume signal_info initializes to zeroes.
|
||||||
*
|
*
|
||||||
* On Unix, myAH is the master DB connection in the master process, and the
|
* On Unix, myAH is the leader DB connection in the leader process, and the
|
||||||
* worker's own connection in worker processes. On Windows, we have only one
|
* worker's own connection in worker processes. On Windows, we have only one
|
||||||
* instance of signal_info, so myAH is the master connection and the worker
|
* instance of signal_info, so myAH is the leader connection and the worker
|
||||||
* connections must be dug out of pstate->parallelSlot[].
|
* connections must be dug out of pstate->parallelSlot[].
|
||||||
*/
|
*/
|
||||||
typedef struct DumpSignalInformation
|
typedef struct DumpSignalInformation
|
||||||
|
@ -216,8 +216,8 @@ static void lockTableForWorker(ArchiveHandle *AH, TocEntry *te);
|
||||||
static void WaitForCommands(ArchiveHandle *AH, int pipefd[2]);
|
static void WaitForCommands(ArchiveHandle *AH, int pipefd[2]);
|
||||||
static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate,
|
static bool ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate,
|
||||||
bool do_wait);
|
bool do_wait);
|
||||||
static char *getMessageFromMaster(int pipefd[2]);
|
static char *getMessageFromLeader(int pipefd[2]);
|
||||||
static void sendMessageToMaster(int pipefd[2], const char *str);
|
static void sendMessageToLeader(int pipefd[2], const char *str);
|
||||||
static int select_loop(int maxFd, fd_set *workerset);
|
static int select_loop(int maxFd, fd_set *workerset);
|
||||||
static char *getMessageFromWorker(ParallelState *pstate,
|
static char *getMessageFromWorker(ParallelState *pstate,
|
||||||
bool do_wait, int *worker);
|
bool do_wait, int *worker);
|
||||||
|
@ -277,7 +277,7 @@ init_parallel_dump_utils(void)
|
||||||
/*
|
/*
|
||||||
* Find the ParallelSlot for the current worker process or thread.
|
* Find the ParallelSlot for the current worker process or thread.
|
||||||
*
|
*
|
||||||
* Returns NULL if no matching slot is found (this implies we're the master).
|
* Returns NULL if no matching slot is found (this implies we're the leader).
|
||||||
*/
|
*/
|
||||||
static ParallelSlot *
|
static ParallelSlot *
|
||||||
GetMyPSlot(ParallelState *pstate)
|
GetMyPSlot(ParallelState *pstate)
|
||||||
|
@ -367,7 +367,7 @@ archive_close_connection(int code, void *arg)
|
||||||
if (!slot)
|
if (!slot)
|
||||||
{
|
{
|
||||||
/*
|
/*
|
||||||
* We're the master. Forcibly shut down workers, then close our
|
* We're the leader. Forcibly shut down workers, then close our
|
||||||
* own database connection, if any.
|
* own database connection, if any.
|
||||||
*/
|
*/
|
||||||
ShutdownWorkersHard(si->pstate);
|
ShutdownWorkersHard(si->pstate);
|
||||||
|
@ -381,7 +381,7 @@ archive_close_connection(int code, void *arg)
|
||||||
* We're a worker. Shut down our own DB connection if any. On
|
* We're a worker. Shut down our own DB connection if any. On
|
||||||
* Windows, we also have to close our communication sockets, to
|
* Windows, we also have to close our communication sockets, to
|
||||||
* emulate what will happen on Unix when the worker process exits.
|
* emulate what will happen on Unix when the worker process exits.
|
||||||
* (Without this, if this is a premature exit, the master would
|
* (Without this, if this is a premature exit, the leader would
|
||||||
* fail to detect it because there would be no EOF condition on
|
* fail to detect it because there would be no EOF condition on
|
||||||
* the other end of the pipe.)
|
* the other end of the pipe.)
|
||||||
*/
|
*/
|
||||||
|
@ -396,7 +396,7 @@ archive_close_connection(int code, void *arg)
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
/* Non-parallel operation: just kill the master DB connection */
|
/* Non-parallel operation: just kill the leader DB connection */
|
||||||
if (si->AHX)
|
if (si->AHX)
|
||||||
DisconnectDatabase(si->AHX);
|
DisconnectDatabase(si->AHX);
|
||||||
}
|
}
|
||||||
|
@ -541,11 +541,11 @@ WaitForTerminatingWorkers(ParallelState *pstate)
|
||||||
*
|
*
|
||||||
* In parallel operation on Unix, each process is responsible for canceling
|
* In parallel operation on Unix, each process is responsible for canceling
|
||||||
* its own connection (this must be so because nobody else has access to it).
|
* its own connection (this must be so because nobody else has access to it).
|
||||||
* Furthermore, the master process should attempt to forward its signal to
|
* Furthermore, the leader process should attempt to forward its signal to
|
||||||
* each child. In simple manual use of pg_dump/pg_restore, forwarding isn't
|
* each child. In simple manual use of pg_dump/pg_restore, forwarding isn't
|
||||||
* needed because typing control-C at the console would deliver SIGINT to
|
* needed because typing control-C at the console would deliver SIGINT to
|
||||||
* every member of the terminal process group --- but in other scenarios it
|
* every member of the terminal process group --- but in other scenarios it
|
||||||
* might be that only the master gets signaled.
|
* might be that only the leader gets signaled.
|
||||||
*
|
*
|
||||||
* On Windows, the cancel handler runs in a separate thread, because that's
|
* On Windows, the cancel handler runs in a separate thread, because that's
|
||||||
* how SetConsoleCtrlHandler works. We make it stop worker threads, send
|
* how SetConsoleCtrlHandler works. We make it stop worker threads, send
|
||||||
|
@ -576,8 +576,8 @@ sigTermHandler(SIGNAL_ARGS)
|
||||||
pqsignal(SIGQUIT, SIG_IGN);
|
pqsignal(SIGQUIT, SIG_IGN);
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* If we're in the master, forward signal to all workers. (It seems best
|
* If we're in the leader, forward signal to all workers. (It seems best
|
||||||
* to do this before PQcancel; killing the master transaction will result
|
* to do this before PQcancel; killing the leader transaction will result
|
||||||
* in invalid-snapshot errors from active workers, which maybe we can
|
* in invalid-snapshot errors from active workers, which maybe we can
|
||||||
* quiet by killing workers first.) Ignore any errors.
|
* quiet by killing workers first.) Ignore any errors.
|
||||||
*/
|
*/
|
||||||
|
@ -601,7 +601,7 @@ sigTermHandler(SIGNAL_ARGS)
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Report we're quitting, using nothing more complicated than write(2).
|
* Report we're quitting, using nothing more complicated than write(2).
|
||||||
* When in parallel operation, only the master process should do this.
|
* When in parallel operation, only the leader process should do this.
|
||||||
*/
|
*/
|
||||||
if (!signal_info.am_worker)
|
if (!signal_info.am_worker)
|
||||||
{
|
{
|
||||||
|
@ -665,7 +665,7 @@ consoleHandler(DWORD dwCtrlType)
|
||||||
* If in parallel mode, stop worker threads and send QueryCancel to
|
* If in parallel mode, stop worker threads and send QueryCancel to
|
||||||
* their connected backends. The main point of stopping the worker
|
* their connected backends. The main point of stopping the worker
|
||||||
* threads is to keep them from reporting the query cancels as errors,
|
* threads is to keep them from reporting the query cancels as errors,
|
||||||
* which would clutter the user's screen. We needn't stop the master
|
* which would clutter the user's screen. We needn't stop the leader
|
||||||
* thread since it won't be doing much anyway. Do this before
|
* thread since it won't be doing much anyway. Do this before
|
||||||
* canceling the main transaction, else we might get invalid-snapshot
|
* canceling the main transaction, else we might get invalid-snapshot
|
||||||
* errors reported before we can stop the workers. Ignore errors,
|
* errors reported before we can stop the workers. Ignore errors,
|
||||||
|
@ -693,7 +693,7 @@ consoleHandler(DWORD dwCtrlType)
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Send QueryCancel to master connection, if enabled. Ignore errors,
|
* Send QueryCancel to leader connection, if enabled. Ignore errors,
|
||||||
* there's not much we can do about them anyway.
|
* there's not much we can do about them anyway.
|
||||||
*/
|
*/
|
||||||
if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL)
|
if (signal_info.myAH != NULL && signal_info.myAH->connCancel != NULL)
|
||||||
|
@ -949,11 +949,11 @@ ParallelBackupStart(ArchiveHandle *AH)
|
||||||
shutdown_info.pstate = pstate;
|
shutdown_info.pstate = pstate;
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Temporarily disable query cancellation on the master connection. This
|
* Temporarily disable query cancellation on the leader connection. This
|
||||||
* ensures that child processes won't inherit valid AH->connCancel
|
* ensures that child processes won't inherit valid AH->connCancel
|
||||||
* settings and thus won't try to issue cancels against the master's
|
* settings and thus won't try to issue cancels against the leader's
|
||||||
* connection. No harm is done if we fail while it's disabled, because
|
* connection. No harm is done if we fail while it's disabled, because
|
||||||
* the master connection is idle at this point anyway.
|
* the leader connection is idle at this point anyway.
|
||||||
*/
|
*/
|
||||||
set_archive_cancel_info(AH, NULL);
|
set_archive_cancel_info(AH, NULL);
|
||||||
|
|
||||||
|
@ -977,7 +977,7 @@ ParallelBackupStart(ArchiveHandle *AH)
|
||||||
if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0)
|
if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0)
|
||||||
fatal("could not create communication channels: %m");
|
fatal("could not create communication channels: %m");
|
||||||
|
|
||||||
/* master's ends of the pipes */
|
/* leader's ends of the pipes */
|
||||||
slot->pipeRead = pipeWM[PIPE_READ];
|
slot->pipeRead = pipeWM[PIPE_READ];
|
||||||
slot->pipeWrite = pipeMW[PIPE_WRITE];
|
slot->pipeWrite = pipeMW[PIPE_WRITE];
|
||||||
/* child's ends of the pipes */
|
/* child's ends of the pipes */
|
||||||
|
@ -1008,13 +1008,13 @@ ParallelBackupStart(ArchiveHandle *AH)
|
||||||
/* instruct signal handler that we're in a worker now */
|
/* instruct signal handler that we're in a worker now */
|
||||||
signal_info.am_worker = true;
|
signal_info.am_worker = true;
|
||||||
|
|
||||||
/* close read end of Worker -> Master */
|
/* close read end of Worker -> Leader */
|
||||||
closesocket(pipeWM[PIPE_READ]);
|
closesocket(pipeWM[PIPE_READ]);
|
||||||
/* close write end of Master -> Worker */
|
/* close write end of Leader -> Worker */
|
||||||
closesocket(pipeMW[PIPE_WRITE]);
|
closesocket(pipeMW[PIPE_WRITE]);
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Close all inherited fds for communication of the master with
|
* Close all inherited fds for communication of the leader with
|
||||||
* previously-forked workers.
|
* previously-forked workers.
|
||||||
*/
|
*/
|
||||||
for (j = 0; j < i; j++)
|
for (j = 0; j < i; j++)
|
||||||
|
@ -1035,19 +1035,19 @@ ParallelBackupStart(ArchiveHandle *AH)
|
||||||
fatal("could not create worker process: %m");
|
fatal("could not create worker process: %m");
|
||||||
}
|
}
|
||||||
|
|
||||||
/* In Master after successful fork */
|
/* In Leader after successful fork */
|
||||||
slot->pid = pid;
|
slot->pid = pid;
|
||||||
slot->workerStatus = WRKR_IDLE;
|
slot->workerStatus = WRKR_IDLE;
|
||||||
|
|
||||||
/* close read end of Master -> Worker */
|
/* close read end of Leader -> Worker */
|
||||||
closesocket(pipeMW[PIPE_READ]);
|
closesocket(pipeMW[PIPE_READ]);
|
||||||
/* close write end of Worker -> Master */
|
/* close write end of Worker -> Leader */
|
||||||
closesocket(pipeWM[PIPE_WRITE]);
|
closesocket(pipeWM[PIPE_WRITE]);
|
||||||
#endif /* WIN32 */
|
#endif /* WIN32 */
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Having forked off the workers, disable SIGPIPE so that master isn't
|
* Having forked off the workers, disable SIGPIPE so that leader isn't
|
||||||
* killed if it tries to send a command to a dead worker. We don't want
|
* killed if it tries to send a command to a dead worker. We don't want
|
||||||
* the workers to inherit this setting, though.
|
* the workers to inherit this setting, though.
|
||||||
*/
|
*/
|
||||||
|
@ -1056,7 +1056,7 @@ ParallelBackupStart(ArchiveHandle *AH)
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Re-establish query cancellation on the master connection.
|
* Re-establish query cancellation on the leader connection.
|
||||||
*/
|
*/
|
||||||
set_archive_cancel_info(AH, AH->connection);
|
set_archive_cancel_info(AH, AH->connection);
|
||||||
|
|
||||||
|
@ -1162,12 +1162,12 @@ parseWorkerCommand(ArchiveHandle *AH, TocEntry **te, T_Action *act,
|
||||||
Assert(*te != NULL);
|
Assert(*te != NULL);
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
fatal("unrecognized command received from master: \"%s\"",
|
fatal("unrecognized command received from leader: \"%s\"",
|
||||||
msg);
|
msg);
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* buildWorkerResponse: format a response string to send to the master.
|
* buildWorkerResponse: format a response string to send to the leader.
|
||||||
*
|
*
|
||||||
* The string is built in the caller-supplied buffer of size buflen.
|
* The string is built in the caller-supplied buffer of size buflen.
|
||||||
*/
|
*/
|
||||||
|
@ -1299,16 +1299,16 @@ IsEveryWorkerIdle(ParallelState *pstate)
|
||||||
/*
|
/*
|
||||||
* Acquire lock on a table to be dumped by a worker process.
|
* Acquire lock on a table to be dumped by a worker process.
|
||||||
*
|
*
|
||||||
* The master process is already holding an ACCESS SHARE lock. Ordinarily
|
* The leader process is already holding an ACCESS SHARE lock. Ordinarily
|
||||||
* it's no problem for a worker to get one too, but if anything else besides
|
* it's no problem for a worker to get one too, but if anything else besides
|
||||||
* pg_dump is running, there's a possible deadlock:
|
* pg_dump is running, there's a possible deadlock:
|
||||||
*
|
*
|
||||||
* 1) Master dumps the schema and locks all tables in ACCESS SHARE mode.
|
* 1) Leader dumps the schema and locks all tables in ACCESS SHARE mode.
|
||||||
* 2) Another process requests an ACCESS EXCLUSIVE lock (which is not granted
|
* 2) Another process requests an ACCESS EXCLUSIVE lock (which is not granted
|
||||||
* because the master holds a conflicting ACCESS SHARE lock).
|
* because the leader holds a conflicting ACCESS SHARE lock).
|
||||||
* 3) A worker process also requests an ACCESS SHARE lock to read the table.
|
* 3) A worker process also requests an ACCESS SHARE lock to read the table.
|
||||||
* The worker is enqueued behind the ACCESS EXCLUSIVE lock request.
|
* The worker is enqueued behind the ACCESS EXCLUSIVE lock request.
|
||||||
* 4) Now we have a deadlock, since the master is effectively waiting for
|
* 4) Now we have a deadlock, since the leader is effectively waiting for
|
||||||
* the worker. The server cannot detect that, however.
|
* the worker. The server cannot detect that, however.
|
||||||
*
|
*
|
||||||
* To prevent an infinite wait, prior to touching a table in a worker, request
|
* To prevent an infinite wait, prior to touching a table in a worker, request
|
||||||
|
@ -1349,7 +1349,7 @@ lockTableForWorker(ArchiveHandle *AH, TocEntry *te)
|
||||||
/*
|
/*
|
||||||
* WaitForCommands: main routine for a worker process.
|
* WaitForCommands: main routine for a worker process.
|
||||||
*
|
*
|
||||||
* Read and execute commands from the master until we see EOF on the pipe.
|
* Read and execute commands from the leader until we see EOF on the pipe.
|
||||||
*/
|
*/
|
||||||
static void
|
static void
|
||||||
WaitForCommands(ArchiveHandle *AH, int pipefd[2])
|
WaitForCommands(ArchiveHandle *AH, int pipefd[2])
|
||||||
|
@ -1362,7 +1362,7 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2])
|
||||||
|
|
||||||
for (;;)
|
for (;;)
|
||||||
{
|
{
|
||||||
if (!(command = getMessageFromMaster(pipefd)))
|
if (!(command = getMessageFromLeader(pipefd)))
|
||||||
{
|
{
|
||||||
/* EOF, so done */
|
/* EOF, so done */
|
||||||
return;
|
return;
|
||||||
|
@ -1387,10 +1387,10 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2])
|
||||||
else
|
else
|
||||||
Assert(false);
|
Assert(false);
|
||||||
|
|
||||||
/* Return status to master */
|
/* Return status to leader */
|
||||||
buildWorkerResponse(AH, te, act, status, buf, sizeof(buf));
|
buildWorkerResponse(AH, te, act, status, buf, sizeof(buf));
|
||||||
|
|
||||||
sendMessageToMaster(pipefd, buf);
|
sendMessageToLeader(pipefd, buf);
|
||||||
|
|
||||||
/* command was pg_malloc'd and we are responsible for free()ing it. */
|
/* command was pg_malloc'd and we are responsible for free()ing it. */
|
||||||
free(command);
|
free(command);
|
||||||
|
@ -1464,7 +1464,7 @@ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
|
||||||
* Any received results are passed to the callback specified to
|
* Any received results are passed to the callback specified to
|
||||||
* DispatchJobForTocEntry.
|
* DispatchJobForTocEntry.
|
||||||
*
|
*
|
||||||
* This function is executed in the master process.
|
* This function is executed in the leader process.
|
||||||
*/
|
*/
|
||||||
void
|
void
|
||||||
WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
|
WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
|
||||||
|
@ -1525,25 +1525,25 @@ WaitForWorkers(ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Read one command message from the master, blocking if necessary
|
* Read one command message from the leader, blocking if necessary
|
||||||
* until one is available, and return it as a malloc'd string.
|
* until one is available, and return it as a malloc'd string.
|
||||||
* On EOF, return NULL.
|
* On EOF, return NULL.
|
||||||
*
|
*
|
||||||
* This function is executed in worker processes.
|
* This function is executed in worker processes.
|
||||||
*/
|
*/
|
||||||
static char *
|
static char *
|
||||||
getMessageFromMaster(int pipefd[2])
|
getMessageFromLeader(int pipefd[2])
|
||||||
{
|
{
|
||||||
return readMessageFromPipe(pipefd[PIPE_READ]);
|
return readMessageFromPipe(pipefd[PIPE_READ]);
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Send a status message to the master.
|
* Send a status message to the leader.
|
||||||
*
|
*
|
||||||
* This function is executed in worker processes.
|
* This function is executed in worker processes.
|
||||||
*/
|
*/
|
||||||
static void
|
static void
|
||||||
sendMessageToMaster(int pipefd[2], const char *str)
|
sendMessageToLeader(int pipefd[2], const char *str)
|
||||||
{
|
{
|
||||||
int len = strlen(str) + 1;
|
int len = strlen(str) + 1;
|
||||||
|
|
||||||
|
@ -1592,7 +1592,7 @@ select_loop(int maxFd, fd_set *workerset)
|
||||||
* that's hard to distinguish from the no-data-available case, but for now
|
* that's hard to distinguish from the no-data-available case, but for now
|
||||||
* our one caller is okay with that.
|
* our one caller is okay with that.
|
||||||
*
|
*
|
||||||
* This function is executed in the master process.
|
* This function is executed in the leader process.
|
||||||
*/
|
*/
|
||||||
static char *
|
static char *
|
||||||
getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
|
getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
|
||||||
|
@ -1657,7 +1657,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
|
||||||
/*
|
/*
|
||||||
* Send a command message to the specified worker process.
|
* Send a command message to the specified worker process.
|
||||||
*
|
*
|
||||||
* This function is executed in the master process.
|
* This function is executed in the leader process.
|
||||||
*/
|
*/
|
||||||
static void
|
static void
|
||||||
sendMessageToWorker(ParallelState *pstate, int worker, const char *str)
|
sendMessageToWorker(ParallelState *pstate, int worker, const char *str)
|
||||||
|
@ -1688,7 +1688,7 @@ readMessageFromPipe(int fd)
|
||||||
/*
|
/*
|
||||||
* In theory, if we let piperead() read multiple bytes, it might give us
|
* In theory, if we let piperead() read multiple bytes, it might give us
|
||||||
* back fragments of multiple messages. (That can't actually occur, since
|
* back fragments of multiple messages. (That can't actually occur, since
|
||||||
* neither master nor workers send more than one message without waiting
|
* neither leader nor workers send more than one message without waiting
|
||||||
* for a reply, but we don't wish to assume that here.) For simplicity,
|
* for a reply, but we don't wish to assume that here.) For simplicity,
|
||||||
* read a byte at a time until we get the terminating '\0'. This method
|
* read a byte at a time until we get the terminating '\0'. This method
|
||||||
* is a bit inefficient, but since this is only used for relatively short
|
* is a bit inefficient, but since this is only used for relatively short
|
||||||
|
|
|
@ -18,7 +18,7 @@
|
||||||
|
|
||||||
#include "pg_backup_archiver.h"
|
#include "pg_backup_archiver.h"
|
||||||
|
|
||||||
/* Function to call in master process on completion of a worker task */
|
/* Function to call in leader process on completion of a worker task */
|
||||||
typedef void (*ParallelCompletionPtr) (ArchiveHandle *AH,
|
typedef void (*ParallelCompletionPtr) (ArchiveHandle *AH,
|
||||||
TocEntry *te,
|
TocEntry *te,
|
||||||
int status,
|
int status,
|
||||||
|
|
|
@ -662,7 +662,7 @@ RestoreArchive(Archive *AHX)
|
||||||
restore_toc_entries_parallel(AH, pstate, &pending_list);
|
restore_toc_entries_parallel(AH, pstate, &pending_list);
|
||||||
ParallelBackupEnd(AH, pstate);
|
ParallelBackupEnd(AH, pstate);
|
||||||
|
|
||||||
/* reconnect the master and see if we missed something */
|
/* reconnect the leader and see if we missed something */
|
||||||
restore_toc_entries_postfork(AH, &pending_list);
|
restore_toc_entries_postfork(AH, &pending_list);
|
||||||
Assert(AH->connection != NULL);
|
Assert(AH->connection != NULL);
|
||||||
}
|
}
|
||||||
|
@ -2393,7 +2393,7 @@ WriteDataChunks(ArchiveHandle *AH, ParallelState *pstate)
|
||||||
if (pstate && pstate->numWorkers > 1)
|
if (pstate && pstate->numWorkers > 1)
|
||||||
{
|
{
|
||||||
/*
|
/*
|
||||||
* In parallel mode, this code runs in the master process. We
|
* In parallel mode, this code runs in the leader process. We
|
||||||
* construct an array of candidate TEs, then sort it into decreasing
|
* construct an array of candidate TEs, then sort it into decreasing
|
||||||
* size order, then dispatch each TE to a data-transfer worker. By
|
* size order, then dispatch each TE to a data-transfer worker. By
|
||||||
* dumping larger tables first, we avoid getting into a situation
|
* dumping larger tables first, we avoid getting into a situation
|
||||||
|
@ -2447,7 +2447,7 @@ WriteDataChunks(ArchiveHandle *AH, ParallelState *pstate)
|
||||||
|
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Callback function that's invoked in the master process after a step has
|
* Callback function that's invoked in the leader process after a step has
|
||||||
* been parallel dumped.
|
* been parallel dumped.
|
||||||
*
|
*
|
||||||
* We don't need to do anything except check for worker failure.
|
* We don't need to do anything except check for worker failure.
|
||||||
|
@ -4437,7 +4437,7 @@ pop_next_work_item(ArchiveHandle *AH, ParallelReadyList *ready_list,
|
||||||
* this is run in the worker, i.e. in a thread (Windows) or a separate process
|
* this is run in the worker, i.e. in a thread (Windows) or a separate process
|
||||||
* (everything else). A worker process executes several such work items during
|
* (everything else). A worker process executes several such work items during
|
||||||
* a parallel backup or restore. Once we terminate here and report back that
|
* a parallel backup or restore. Once we terminate here and report back that
|
||||||
* our work is finished, the master process will assign us a new work item.
|
* our work is finished, the leader process will assign us a new work item.
|
||||||
*/
|
*/
|
||||||
int
|
int
|
||||||
parallel_restore(ArchiveHandle *AH, TocEntry *te)
|
parallel_restore(ArchiveHandle *AH, TocEntry *te)
|
||||||
|
@ -4457,7 +4457,7 @@ parallel_restore(ArchiveHandle *AH, TocEntry *te)
|
||||||
|
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Callback function that's invoked in the master process after a step has
|
* Callback function that's invoked in the leader process after a step has
|
||||||
* been parallel restored.
|
* been parallel restored.
|
||||||
*
|
*
|
||||||
* Update status and reduce the dependency count of any dependent items.
|
* Update status and reduce the dependency count of any dependent items.
|
||||||
|
|
|
@ -807,7 +807,7 @@ _Clone(ArchiveHandle *AH)
|
||||||
*/
|
*/
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* We also don't copy the ParallelState pointer (pstate), only the master
|
* We also don't copy the ParallelState pointer (pstate), only the leader
|
||||||
* process ever writes to it.
|
* process ever writes to it.
|
||||||
*/
|
*/
|
||||||
}
|
}
|
||||||
|
|
|
@ -1238,7 +1238,7 @@ static void
|
||||||
setupDumpWorker(Archive *AH)
|
setupDumpWorker(Archive *AH)
|
||||||
{
|
{
|
||||||
/*
|
/*
|
||||||
* We want to re-select all the same values the master connection is
|
* We want to re-select all the same values the leader connection is
|
||||||
* using. We'll have inherited directly-usable values in
|
* using. We'll have inherited directly-usable values in
|
||||||
* AH->sync_snapshot_id and AH->use_role, but we need to translate the
|
* AH->sync_snapshot_id and AH->use_role, but we need to translate the
|
||||||
* inherited encoding value back to a string to pass to setup_connection.
|
* inherited encoding value back to a string to pass to setup_connection.
|
||||||
|
|
|
@ -157,10 +157,10 @@ typedef FormData_pg_proc *Form_pg_proc;
|
||||||
/*
|
/*
|
||||||
* Symbolic values for proparallel column: these indicate whether a function
|
* Symbolic values for proparallel column: these indicate whether a function
|
||||||
* can be safely be run in a parallel backend, during parallelism but
|
* can be safely be run in a parallel backend, during parallelism but
|
||||||
* necessarily in the master, or only in non-parallel mode.
|
* necessarily in the leader, or only in non-parallel mode.
|
||||||
*/
|
*/
|
||||||
#define PROPARALLEL_SAFE 's' /* can run in worker or master */
|
#define PROPARALLEL_SAFE 's' /* can run in worker or leader */
|
||||||
#define PROPARALLEL_RESTRICTED 'r' /* can run in parallel master only */
|
#define PROPARALLEL_RESTRICTED 'r' /* can run in parallel leader only */
|
||||||
#define PROPARALLEL_UNSAFE 'u' /* banned while in parallel mode */
|
#define PROPARALLEL_UNSAFE 'u' /* banned while in parallel mode */
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
|
|
@ -17,7 +17,7 @@
|
||||||
#include "storage/shm_mq.h"
|
#include "storage/shm_mq.h"
|
||||||
|
|
||||||
extern void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh);
|
extern void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh);
|
||||||
extern void pq_set_parallel_master(pid_t pid, BackendId backend_id);
|
extern void pq_set_parallel_leader(pid_t pid, BackendId backend_id);
|
||||||
|
|
||||||
extern void pq_parse_errornotice(StringInfo str, ErrorData *edata);
|
extern void pq_parse_errornotice(StringInfo str, ErrorData *edata);
|
||||||
|
|
||||||
|
|
|
@ -25,13 +25,13 @@ typedef int BackendId; /* unique currently active backend identifier */
|
||||||
extern PGDLLIMPORT BackendId MyBackendId; /* backend id of this backend */
|
extern PGDLLIMPORT BackendId MyBackendId; /* backend id of this backend */
|
||||||
|
|
||||||
/* backend id of our parallel session leader, or InvalidBackendId if none */
|
/* backend id of our parallel session leader, or InvalidBackendId if none */
|
||||||
extern PGDLLIMPORT BackendId ParallelMasterBackendId;
|
extern PGDLLIMPORT BackendId ParallelLeaderBackendId;
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* The BackendId to use for our session's temp relations is normally our own,
|
* The BackendId to use for our session's temp relations is normally our own,
|
||||||
* but parallel workers should use their leader's ID.
|
* but parallel workers should use their leader's ID.
|
||||||
*/
|
*/
|
||||||
#define BackendIdForTempRelations() \
|
#define BackendIdForTempRelations() \
|
||||||
(ParallelMasterBackendId == InvalidBackendId ? MyBackendId : ParallelMasterBackendId)
|
(ParallelLeaderBackendId == InvalidBackendId ? MyBackendId : ParallelLeaderBackendId)
|
||||||
|
|
||||||
#endif /* BACKENDID_H */
|
#endif /* BACKENDID_H */
|
||||||
|
|
Loading…
Reference in New Issue