PostgreSQL Source Code git master
Data Structures | Macros | Typedefs | Functions | Variables
parallel.c File Reference
#include "postgres.h"
#include "access/brin.h"
#include "access/gin.h"
#include "access/nbtree.h"
#include "access/parallel.h"
#include "access/session.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "catalog/index.h"
#include "catalog/namespace.h"
#include "catalog/pg_enum.h"
#include "catalog/storage.h"
#include "commands/async.h"
#include "commands/vacuum.h"
#include "executor/execParallel.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
#include "storage/ipc.h"
#include "storage/predicate.h"
#include "storage/spin.h"
#include "tcop/tcopprot.h"
#include "utils/combocid.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/relmapper.h"
#include "utils/snapmgr.h"
Include dependency graph for parallel.c:

Go to the source code of this file.

Data Structures

struct   FixedParallelState
 

Macros

#define  PARALLEL_ERROR_QUEUE_SIZE   16384
 
#define  PARALLEL_MAGIC   0x50477c7c
 
#define  PARALLEL_KEY_FIXED   UINT64CONST(0xFFFFFFFFFFFF0001)
 
#define  PARALLEL_KEY_ERROR_QUEUE   UINT64CONST(0xFFFFFFFFFFFF0002)
 
#define  PARALLEL_KEY_LIBRARY   UINT64CONST(0xFFFFFFFFFFFF0003)
 
#define  PARALLEL_KEY_GUC   UINT64CONST(0xFFFFFFFFFFFF0004)
 
#define  PARALLEL_KEY_COMBO_CID   UINT64CONST(0xFFFFFFFFFFFF0005)
 
#define  PARALLEL_KEY_TRANSACTION_SNAPSHOT   UINT64CONST(0xFFFFFFFFFFFF0006)
 
#define  PARALLEL_KEY_ACTIVE_SNAPSHOT   UINT64CONST(0xFFFFFFFFFFFF0007)
 
#define  PARALLEL_KEY_TRANSACTION_STATE   UINT64CONST(0xFFFFFFFFFFFF0008)
 
#define  PARALLEL_KEY_ENTRYPOINT   UINT64CONST(0xFFFFFFFFFFFF0009)
 
#define  PARALLEL_KEY_SESSION_DSM   UINT64CONST(0xFFFFFFFFFFFF000A)
 
#define  PARALLEL_KEY_PENDING_SYNCS   UINT64CONST(0xFFFFFFFFFFFF000B)
 
#define  PARALLEL_KEY_REINDEX_STATE   UINT64CONST(0xFFFFFFFFFFFF000C)
 
#define  PARALLEL_KEY_RELMAPPER_STATE   UINT64CONST(0xFFFFFFFFFFFF000D)
 
#define  PARALLEL_KEY_UNCOMMITTEDENUMS   UINT64CONST(0xFFFFFFFFFFFF000E)
 
#define  PARALLEL_KEY_CLIENTCONNINFO   UINT64CONST(0xFFFFFFFFFFFF000F)
 

Typedefs

typedef struct FixedParallelState  FixedParallelState
 

Functions

static void  ProcessParallelMessage (ParallelContext *pcxt, int i, StringInfo msg)
 
static void  WaitForParallelWorkersToExit (ParallelContext *pcxt)
 
static parallel_worker_main_type  LookupParallelWorkerFunction (const char *libraryname, const char *funcname)
 
static void  ParallelWorkerShutdown (int code, Datum arg)
 
ParallelContext *  CreateParallelContext (const char *library_name, const char *function_name, int nworkers)
 
void  InitializeParallelDSM (ParallelContext *pcxt)
 
void  ReinitializeParallelDSM (ParallelContext *pcxt)
 
void  ReinitializeParallelWorkers (ParallelContext *pcxt, int nworkers_to_launch)
 
void  LaunchParallelWorkers (ParallelContext *pcxt)
 
void  WaitForParallelWorkersToAttach (ParallelContext *pcxt)
 
void  WaitForParallelWorkersToFinish (ParallelContext *pcxt)
 
void  DestroyParallelContext (ParallelContext *pcxt)
 
bool  ParallelContextActive (void)
 
void  HandleParallelMessageInterrupt (void)
 
void  ProcessParallelMessages (void)
 
void  AtEOSubXact_Parallel (bool isCommit, SubTransactionId mySubId)
 
void  AtEOXact_Parallel (bool isCommit)
 
void  ParallelWorkerMain (Datum main_arg)
 
void  ParallelWorkerReportLastRecEnd (XLogRecPtr last_xlog_end)
 

Variables

int  ParallelWorkerNumber = -1
 
volatile sig_atomic_t  ParallelMessagePending = false
 
bool  InitializingParallelWorker = false
 
static FixedParallelState *  MyFixedParallelState
 
static dlist_head  pcxt_list = DLIST_STATIC_INIT(pcxt_list)
 
static pid_t  ParallelLeaderPid
 
struct {
   const char *   fn_name
 
   parallel_worker_main_type   fn_addr
 
}  InternalParallelWorkers []
 

Macro Definition Documentation

◆ PARALLEL_ERROR_QUEUE_SIZE

#define PARALLEL_ERROR_QUEUE_SIZE   16384

Definition at line 55 of file parallel.c.

◆ PARALLEL_KEY_ACTIVE_SNAPSHOT

#define PARALLEL_KEY_ACTIVE_SNAPSHOT   UINT64CONST(0xFFFFFFFFFFFF0007)

Definition at line 71 of file parallel.c.

◆ PARALLEL_KEY_CLIENTCONNINFO

#define PARALLEL_KEY_CLIENTCONNINFO   UINT64CONST(0xFFFFFFFFFFFF000F)

Definition at line 79 of file parallel.c.

◆ PARALLEL_KEY_COMBO_CID

#define PARALLEL_KEY_COMBO_CID   UINT64CONST(0xFFFFFFFFFFFF0005)

Definition at line 69 of file parallel.c.

◆ PARALLEL_KEY_ENTRYPOINT

#define PARALLEL_KEY_ENTRYPOINT   UINT64CONST(0xFFFFFFFFFFFF0009)

Definition at line 73 of file parallel.c.

◆ PARALLEL_KEY_ERROR_QUEUE

#define PARALLEL_KEY_ERROR_QUEUE   UINT64CONST(0xFFFFFFFFFFFF0002)

Definition at line 66 of file parallel.c.

◆ PARALLEL_KEY_FIXED

#define PARALLEL_KEY_FIXED   UINT64CONST(0xFFFFFFFFFFFF0001)

Definition at line 65 of file parallel.c.

◆ PARALLEL_KEY_GUC

#define PARALLEL_KEY_GUC   UINT64CONST(0xFFFFFFFFFFFF0004)

Definition at line 68 of file parallel.c.

◆ PARALLEL_KEY_LIBRARY

#define PARALLEL_KEY_LIBRARY   UINT64CONST(0xFFFFFFFFFFFF0003)

Definition at line 67 of file parallel.c.

◆ PARALLEL_KEY_PENDING_SYNCS

#define PARALLEL_KEY_PENDING_SYNCS   UINT64CONST(0xFFFFFFFFFFFF000B)

Definition at line 75 of file parallel.c.

◆ PARALLEL_KEY_REINDEX_STATE

#define PARALLEL_KEY_REINDEX_STATE   UINT64CONST(0xFFFFFFFFFFFF000C)

Definition at line 76 of file parallel.c.

◆ PARALLEL_KEY_RELMAPPER_STATE

#define PARALLEL_KEY_RELMAPPER_STATE   UINT64CONST(0xFFFFFFFFFFFF000D)

Definition at line 77 of file parallel.c.

◆ PARALLEL_KEY_SESSION_DSM

#define PARALLEL_KEY_SESSION_DSM   UINT64CONST(0xFFFFFFFFFFFF000A)

Definition at line 74 of file parallel.c.

◆ PARALLEL_KEY_TRANSACTION_SNAPSHOT

#define PARALLEL_KEY_TRANSACTION_SNAPSHOT   UINT64CONST(0xFFFFFFFFFFFF0006)

Definition at line 70 of file parallel.c.

◆ PARALLEL_KEY_TRANSACTION_STATE

#define PARALLEL_KEY_TRANSACTION_STATE   UINT64CONST(0xFFFFFFFFFFFF0008)

Definition at line 72 of file parallel.c.

◆ PARALLEL_KEY_UNCOMMITTEDENUMS

#define PARALLEL_KEY_UNCOMMITTEDENUMS   UINT64CONST(0xFFFFFFFFFFFF000E)

Definition at line 78 of file parallel.c.

◆ PARALLEL_MAGIC

#define PARALLEL_MAGIC   0x50477c7c

Definition at line 58 of file parallel.c.

Typedef Documentation

◆ FixedParallelState

typedef struct FixedParallelState FixedParallelState

Function Documentation

◆ AtEOSubXact_Parallel()

void AtEOSubXact_Parallel ( bool  isCommit,
SubTransactionId  mySubId 
)

Definition at line 1254 of file parallel.c.

1255{
1256 while (!dlist_is_empty(&pcxt_list))
1257 {
1258 ParallelContext *pcxt;
1259
1261 if (pcxt->subid != mySubId)
1262 break;
1263 if (isCommit)
1264 elog(WARNING, "leaked parallel context");
1266 }
1267}
void DestroyParallelContext(ParallelContext *pcxt)
Definition: parallel.c:950
static dlist_head pcxt_list
Definition: parallel.c:127
#define WARNING
Definition: elog.h:36
#define elog(elevel,...)
Definition: elog.h:226
#define dlist_head_element(type, membername, lhead)
Definition: ilist.h:603
static bool dlist_is_empty(const dlist_head *head)
Definition: ilist.h:336
SubTransactionId subid
Definition: parallel.h:34

References DestroyParallelContext(), dlist_head_element, dlist_is_empty(), elog, pcxt_list, ParallelContext::subid, and WARNING.

Referenced by AbortSubTransaction(), and CommitSubTransaction().

◆ AtEOXact_Parallel()

void AtEOXact_Parallel ( bool  isCommit )

Definition at line 1275 of file parallel.c.

1276{
1277 while (!dlist_is_empty(&pcxt_list))
1278 {
1279 ParallelContext *pcxt;
1280
1282 if (isCommit)
1283 elog(WARNING, "leaked parallel context");
1285 }
1286}

References DestroyParallelContext(), dlist_head_element, dlist_is_empty(), elog, pcxt_list, and WARNING.

Referenced by AbortTransaction(), and CommitTransaction().

◆ CreateParallelContext()

ParallelContext * CreateParallelContext ( const char *  library_name,
const char *  function_name,
int  nworkers 
)

Definition at line 173 of file parallel.c.

175{
176 MemoryContext oldcontext;
177 ParallelContext *pcxt;
178
179 /* It is unsafe to create a parallel context if not in parallel mode. */
181
182 /* Number of workers should be non-negative. */
183 Assert(nworkers >= 0);
184
185 /* We might be running in a short-lived memory context. */
187
188 /* Initialize a new ParallelContext. */
189 pcxt = palloc0(sizeof(ParallelContext));
191 pcxt->nworkers = nworkers;
192 pcxt->nworkers_to_launch = nworkers;
193 pcxt->library_name = pstrdup(library_name);
194 pcxt->function_name = pstrdup(function_name);
198
199 /* Restore previous memory context. */
200 MemoryContextSwitchTo(oldcontext);
201
202 return pcxt;
203}
ErrorContextCallback * error_context_stack
Definition: elog.c:95
Assert(PointerIsAligned(start, uint64))
static void dlist_push_head(dlist_head *head, dlist_node *node)
Definition: ilist.h:347
MemoryContext TopTransactionContext
Definition: mcxt.c:171
char * pstrdup(const char *in)
Definition: mcxt.c:1759
void * palloc0(Size size)
Definition: mcxt.c:1395
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
#define shm_toc_initialize_estimator(e)
Definition: shm_toc.h:49
char * library_name
Definition: parallel.h:38
ErrorContextCallback * error_context_stack
Definition: parallel.h:40
shm_toc_estimator estimator
Definition: parallel.h:41
dlist_node node
Definition: parallel.h:33
int nworkers
Definition: parallel.h:35
int nworkers_to_launch
Definition: parallel.h:36
char * function_name
Definition: parallel.h:39
SubTransactionId GetCurrentSubTransactionId(void)
Definition: xact.c:791
bool IsInParallelMode(void)
Definition: xact.c:1089

References Assert(), dlist_push_head(), error_context_stack, ParallelContext::error_context_stack, ParallelContext::estimator, ParallelContext::function_name, GetCurrentSubTransactionId(), IsInParallelMode(), ParallelContext::library_name, MemoryContextSwitchTo(), ParallelContext::node, ParallelContext::nworkers, ParallelContext::nworkers_to_launch, palloc0(), pcxt_list, pstrdup(), shm_toc_initialize_estimator, ParallelContext::subid, and TopTransactionContext.

Referenced by _brin_begin_parallel(), _bt_begin_parallel(), _gin_begin_parallel(), ExecInitParallelPlan(), and parallel_vacuum_init().

◆ DestroyParallelContext()

void DestroyParallelContext ( ParallelContext *  pcxt )

Definition at line 950 of file parallel.c.

951{
952 int i;
953
954 /*
955 * Be careful about order of operations here! We remove the parallel
956 * context from the list before we do anything else; otherwise, if an
957 * error occurs during a subsequent step, we might try to nuke it again
958 * from AtEOXact_Parallel or AtEOSubXact_Parallel.
959 */
960 dlist_delete(&pcxt->node);
961
962 /* Kill each worker in turn, and forget their error queues. */
963 if (pcxt->worker != NULL)
964 {
965 for (i = 0; i < pcxt->nworkers_launched; ++i)
966 {
967 if (pcxt->worker[i].error_mqh != NULL)
968 {
970
972 pcxt->worker[i].error_mqh = NULL;
973 }
974 }
975 }
976
977 /*
978 * If we have allocated a shared memory segment, detach it. This will
979 * implicitly detach the error queues, and any other shared memory queues,
980 * stored there.
981 */
982 if (pcxt->seg != NULL)
983 {
984 dsm_detach(pcxt->seg);
985 pcxt->seg = NULL;
986 }
987
988 /*
989 * If this parallel context is actually in backend-private memory rather
990 * than shared memory, free that memory instead.
991 */
992 if (pcxt->private_memory != NULL)
993 {
994 pfree(pcxt->private_memory);
995 pcxt->private_memory = NULL;
996 }
997
998 /*
999 * We can't finish transaction commit or abort until all of the workers
1000 * have exited. This means, in particular, that we can't respond to
1001 * interrupts at this stage.
1002 */
1006
1007 /* Free the worker array itself. */
1008 if (pcxt->worker != NULL)
1009 {
1010 pfree(pcxt->worker);
1011 pcxt->worker = NULL;
1012 }
1013
1014 /* Free memory. */
1015 pfree(pcxt->library_name);
1016 pfree(pcxt->function_name);
1017 pfree(pcxt);
1018}
static void WaitForParallelWorkersToExit(ParallelContext *pcxt)
Definition: parallel.c:910
void TerminateBackgroundWorker(BackgroundWorkerHandle *handle)
Definition: bgworker.c:1297
void dsm_detach(dsm_segment *seg)
Definition: dsm.c:803
static void dlist_delete(dlist_node *node)
Definition: ilist.h:405
i
int i
Definition: isn.c:77
void pfree(void *pointer)
Definition: mcxt.c:1594
#define RESUME_INTERRUPTS()
Definition: miscadmin.h:135
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:133
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:843
dsm_segment * seg
Definition: parallel.h:42
ParallelWorkerInfo * worker
Definition: parallel.h:45
void * private_memory
Definition: parallel.h:43
int nworkers_launched
Definition: parallel.h:37
BackgroundWorkerHandle * bgwhandle
Definition: parallel.h:27
shm_mq_handle * error_mqh
Definition: parallel.h:28

References ParallelWorkerInfo::bgwhandle, dlist_delete(), dsm_detach(), ParallelWorkerInfo::error_mqh, ParallelContext::function_name, HOLD_INTERRUPTS, i, ParallelContext::library_name, ParallelContext::node, ParallelContext::nworkers_launched, pfree(), ParallelContext::private_memory, RESUME_INTERRUPTS, ParallelContext::seg, shm_mq_detach(), TerminateBackgroundWorker(), WaitForParallelWorkersToExit(), and ParallelContext::worker.

Referenced by _brin_begin_parallel(), _brin_end_parallel(), _bt_begin_parallel(), _bt_end_parallel(), _gin_begin_parallel(), _gin_end_parallel(), AtEOSubXact_Parallel(), AtEOXact_Parallel(), ExecParallelCleanup(), and parallel_vacuum_end().

◆ HandleParallelMessageInterrupt()

void HandleParallelMessageInterrupt ( void  )

Definition at line 1037 of file parallel.c.

1038{
1039 InterruptPending = true;
1042}
volatile sig_atomic_t ParallelMessagePending
Definition: parallel.c:118
volatile sig_atomic_t InterruptPending
Definition: globals.c:32
struct Latch * MyLatch
Definition: globals.c:63
void SetLatch(Latch *latch)
Definition: latch.c:290

References InterruptPending, MyLatch, ParallelMessagePending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

◆ InitializeParallelDSM()

void InitializeParallelDSM ( ParallelContext *  pcxt )

Definition at line 211 of file parallel.c.

212{
213 MemoryContext oldcontext;
214 Size library_len = 0;
215 Size guc_len = 0;
216 Size combocidlen = 0;
217 Size tsnaplen = 0;
218 Size asnaplen = 0;
219 Size tstatelen = 0;
220 Size pendingsyncslen = 0;
221 Size reindexlen = 0;
222 Size relmapperlen = 0;
223 Size uncommittedenumslen = 0;
224 Size clientconninfolen = 0;
225 Size segsize = 0;
226 int i;
228 dsm_handle session_dsm_handle = DSM_HANDLE_INVALID;
229 Snapshot transaction_snapshot = GetTransactionSnapshot();
230 Snapshot active_snapshot = GetActiveSnapshot();
231
232 /* We might be running in a very short-lived memory context. */
234
235 /* Allow space to store the fixed-size parallel state. */
238
239 /*
240 * If we manage to reach here while non-interruptible, it's unsafe to
241 * launch any workers: we would fail to process interrupts sent by them.
242 * We can deal with that edge case by pretending no workers were
243 * requested.
244 */
246 pcxt->nworkers = 0;
247
248 /*
249 * Normally, the user will have requested at least one worker process, but
250 * if by chance they have not, we can skip a bunch of things here.
251 */
252 if (pcxt->nworkers > 0)
253 {
254 /* Get (or create) the per-session DSM segment's handle. */
255 session_dsm_handle = GetSessionDsmHandle();
256
257 /*
258 * If we weren't able to create a per-session DSM segment, then we can
259 * continue but we can't safely launch any workers because their
260 * record typmods would be incompatible so they couldn't exchange
261 * tuples.
262 */
263 if (session_dsm_handle == DSM_HANDLE_INVALID)
264 pcxt->nworkers = 0;
265 }
266
267 if (pcxt->nworkers > 0)
268 {
269 /* Estimate space for various kinds of state sharing. */
270 library_len = EstimateLibraryStateSpace();
271 shm_toc_estimate_chunk(&pcxt->estimator, library_len);
272 guc_len = EstimateGUCStateSpace();
273 shm_toc_estimate_chunk(&pcxt->estimator, guc_len);
274 combocidlen = EstimateComboCIDStateSpace();
275 shm_toc_estimate_chunk(&pcxt->estimator, combocidlen);
277 {
278 tsnaplen = EstimateSnapshotSpace(transaction_snapshot);
279 shm_toc_estimate_chunk(&pcxt->estimator, tsnaplen);
280 }
281 asnaplen = EstimateSnapshotSpace(active_snapshot);
282 shm_toc_estimate_chunk(&pcxt->estimator, asnaplen);
283 tstatelen = EstimateTransactionStateSpace();
284 shm_toc_estimate_chunk(&pcxt->estimator, tstatelen);
286 pendingsyncslen = EstimatePendingSyncsSpace();
287 shm_toc_estimate_chunk(&pcxt->estimator, pendingsyncslen);
288 reindexlen = EstimateReindexStateSpace();
289 shm_toc_estimate_chunk(&pcxt->estimator, reindexlen);
290 relmapperlen = EstimateRelationMapSpace();
291 shm_toc_estimate_chunk(&pcxt->estimator, relmapperlen);
292 uncommittedenumslen = EstimateUncommittedEnumsSpace();
293 shm_toc_estimate_chunk(&pcxt->estimator, uncommittedenumslen);
294 clientconninfolen = EstimateClientConnectionInfoSpace();
295 shm_toc_estimate_chunk(&pcxt->estimator, clientconninfolen);
296 /* If you add more chunks here, you probably need to add keys. */
298
299 /* Estimate space need for error queues. */
302 "parallel error queue size not buffer-aligned");
305 pcxt->nworkers));
307
308 /* Estimate how much we'll need for the entrypoint info. */
309 shm_toc_estimate_chunk(&pcxt->estimator, strlen(pcxt->library_name) +
310 strlen(pcxt->function_name) + 2);
312 }
313
314 /*
315 * Create DSM and initialize with new table of contents. But if the user
316 * didn't request any workers, then don't bother creating a dynamic shared
317 * memory segment; instead, just use backend-private memory.
318 *
319 * Also, if we can't create a dynamic shared memory segment because the
320 * maximum number of segments have already been created, then fall back to
321 * backend-private memory, and plan not to use any workers. We hope this
322 * won't happen very often, but it's better to abandon the use of
323 * parallelism than to fail outright.
324 */
325 segsize = shm_toc_estimate(&pcxt->estimator);
326 if (pcxt->nworkers > 0)
328 if (pcxt->seg != NULL)
331 segsize);
332 else
333 {
334 pcxt->nworkers = 0;
337 segsize);
338 }
339
340 /* Initialize fixed-size state in shared memory. */
341 fps = (FixedParallelState *)
358 SpinLockInit(&fps->mutex);
359 fps->last_xlog_end = 0;
361
362 /* We can skip the rest of this if we're not budgeting for any workers. */
363 if (pcxt->nworkers > 0)
364 {
365 char *libraryspace;
366 char *gucspace;
367 char *combocidspace;
368 char *tsnapspace;
369 char *asnapspace;
370 char *tstatespace;
371 char *pendingsyncsspace;
372 char *reindexspace;
373 char *relmapperspace;
374 char *error_queue_space;
375 char *session_dsm_handle_space;
376 char *entrypointstate;
377 char *uncommittedenumsspace;
378 char *clientconninfospace;
379 Size lnamelen;
380
381 /* Serialize shared libraries we have loaded. */
382 libraryspace = shm_toc_allocate(pcxt->toc, library_len);
383 SerializeLibraryState(library_len, libraryspace);
384 shm_toc_insert(pcxt->toc, PARALLEL_KEY_LIBRARY, libraryspace);
385
386 /* Serialize GUC settings. */
387 gucspace = shm_toc_allocate(pcxt->toc, guc_len);
388 SerializeGUCState(guc_len, gucspace);
389 shm_toc_insert(pcxt->toc, PARALLEL_KEY_GUC, gucspace);
390
391 /* Serialize combo CID state. */
392 combocidspace = shm_toc_allocate(pcxt->toc, combocidlen);
393 SerializeComboCIDState(combocidlen, combocidspace);
394 shm_toc_insert(pcxt->toc, PARALLEL_KEY_COMBO_CID, combocidspace);
395
396 /*
397 * Serialize the transaction snapshot if the transaction isolation
398 * level uses a transaction snapshot.
399 */
401 {
402 tsnapspace = shm_toc_allocate(pcxt->toc, tsnaplen);
403 SerializeSnapshot(transaction_snapshot, tsnapspace);
405 tsnapspace);
406 }
407
408 /* Serialize the active snapshot. */
409 asnapspace = shm_toc_allocate(pcxt->toc, asnaplen);
410 SerializeSnapshot(active_snapshot, asnapspace);
412
413 /* Provide the handle for per-session segment. */
414 session_dsm_handle_space = shm_toc_allocate(pcxt->toc,
415 sizeof(dsm_handle));
416 *(dsm_handle *) session_dsm_handle_space = session_dsm_handle;
418 session_dsm_handle_space);
419
420 /* Serialize transaction state. */
421 tstatespace = shm_toc_allocate(pcxt->toc, tstatelen);
422 SerializeTransactionState(tstatelen, tstatespace);
424
425 /* Serialize pending syncs. */
426 pendingsyncsspace = shm_toc_allocate(pcxt->toc, pendingsyncslen);
427 SerializePendingSyncs(pendingsyncslen, pendingsyncsspace);
429 pendingsyncsspace);
430
431 /* Serialize reindex state. */
432 reindexspace = shm_toc_allocate(pcxt->toc, reindexlen);
433 SerializeReindexState(reindexlen, reindexspace);
434 shm_toc_insert(pcxt->toc, PARALLEL_KEY_REINDEX_STATE, reindexspace);
435
436 /* Serialize relmapper state. */
437 relmapperspace = shm_toc_allocate(pcxt->toc, relmapperlen);
438 SerializeRelationMap(relmapperlen, relmapperspace);
440 relmapperspace);
441
442 /* Serialize uncommitted enum state. */
443 uncommittedenumsspace = shm_toc_allocate(pcxt->toc,
444 uncommittedenumslen);
445 SerializeUncommittedEnums(uncommittedenumsspace, uncommittedenumslen);
447 uncommittedenumsspace);
448
449 /* Serialize our ClientConnectionInfo. */
450 clientconninfospace = shm_toc_allocate(pcxt->toc, clientconninfolen);
451 SerializeClientConnectionInfo(clientconninfolen, clientconninfospace);
453 clientconninfospace);
454
455 /* Allocate space for worker information. */
456 pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * pcxt->nworkers);
457
458 /*
459 * Establish error queues in dynamic shared memory.
460 *
461 * These queues should be used only for transmitting ErrorResponse,
462 * NoticeResponse, and NotifyResponse protocol messages. Tuple data
463 * should be transmitted via separate (possibly larger?) queues.
464 */
465 error_queue_space =
466 shm_toc_allocate(pcxt->toc,
468 pcxt->nworkers));
469 for (i = 0; i < pcxt->nworkers; ++i)
470 {
471 char *start;
472 shm_mq *mq;
473
474 start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
477 pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
478 }
479 shm_toc_insert(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, error_queue_space);
480
481 /*
482 * Serialize entrypoint information. It's unsafe to pass function
483 * pointers across processes, as the function pointer may be different
484 * in each process in EXEC_BACKEND builds, so we always pass library
485 * and function name. (We use library name "postgres" for functions
486 * in the core backend.)
487 */
488 lnamelen = strlen(pcxt->library_name);
489 entrypointstate = shm_toc_allocate(pcxt->toc, lnamelen +
490 strlen(pcxt->function_name) + 2);
491 strcpy(entrypointstate, pcxt->library_name);
492 strcpy(entrypointstate + lnamelen + 1, pcxt->function_name);
493 shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENTRYPOINT, entrypointstate);
494 }
495
496 /* Update nworkers_to_launch, in case we changed nworkers above. */
497 pcxt->nworkers_to_launch = pcxt->nworkers;
498
499 /* Restore previous memory context. */
500 MemoryContextSwitchTo(oldcontext);
501}
#define PARALLEL_KEY_TRANSACTION_STATE
Definition: parallel.c:72
#define PARALLEL_KEY_GUC
Definition: parallel.c:68
#define PARALLEL_KEY_UNCOMMITTEDENUMS
Definition: parallel.c:78
#define PARALLEL_KEY_TRANSACTION_SNAPSHOT
Definition: parallel.c:70
#define PARALLEL_KEY_CLIENTCONNINFO
Definition: parallel.c:79
#define PARALLEL_KEY_PENDING_SYNCS
Definition: parallel.c:75
#define PARALLEL_KEY_ACTIVE_SNAPSHOT
Definition: parallel.c:71
#define PARALLEL_KEY_ERROR_QUEUE
Definition: parallel.c:66
#define PARALLEL_KEY_SESSION_DSM
Definition: parallel.c:74
#define PARALLEL_MAGIC
Definition: parallel.c:58
#define PARALLEL_KEY_REINDEX_STATE
Definition: parallel.c:76
#define PARALLEL_KEY_LIBRARY
Definition: parallel.c:67
#define PARALLEL_KEY_FIXED
Definition: parallel.c:65
#define PARALLEL_KEY_ENTRYPOINT
Definition: parallel.c:73
#define PARALLEL_KEY_COMBO_CID
Definition: parallel.c:69
#define PARALLEL_ERROR_QUEUE_SIZE
Definition: parallel.c:55
#define PARALLEL_KEY_RELMAPPER_STATE
Definition: parallel.c:77
#define BUFFERALIGN(LEN)
Definition: c.h:812
#define StaticAssertStmt(condition, errmessage)
Definition: c.h:937
size_t Size
Definition: c.h:610
void SerializeComboCIDState(Size maxsize, char *start_address)
Definition: combocid.c:316
Size EstimateComboCIDStateSpace(void)
Definition: combocid.c:297
void SerializeLibraryState(Size maxsize, char *start_address)
Definition: dfmgr.c:719
Size EstimateLibraryStateSpace(void)
Definition: dfmgr.c:702
void * dsm_segment_address(dsm_segment *seg)
Definition: dsm.c:1095
dsm_segment * dsm_create(Size size, int flags)
Definition: dsm.c:516
#define DSM_CREATE_NULL_IF_MAXSEGMENTS
Definition: dsm.h:20
uint32 dsm_handle
Definition: dsm_impl.h:55
#define DSM_HANDLE_INVALID
Definition: dsm_impl.h:58
int MyProcPid
Definition: globals.c:47
ProcNumber MyProcNumber
Definition: globals.c:90
Oid MyDatabaseId
Definition: globals.c:94
void SerializeGUCState(Size maxsize, char *start_address)
Definition: guc.c:6124
Size EstimateGUCStateSpace(void)
Definition: guc.c:5971
bool current_role_is_superuser
Definition: guc_tables.c:537
return str start
void SerializeReindexState(Size maxsize, char *start_address)
Definition: index.c:4253
Size EstimateReindexStateSpace(void)
Definition: index.c:4242
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:1229
MemoryContext TopMemoryContext
Definition: mcxt.c:166
#define INTERRUPTS_CAN_BE_PROCESSED()
Definition: miscadmin.h:129
void GetUserIdAndSecContext(Oid *userid, int *sec_context)
Definition: miscinit.c:612
bool GetSessionUserIsSuperuser(void)
Definition: miscinit.c:515
Size EstimateClientConnectionInfoSpace(void)
Definition: miscinit.c:1035
Oid GetSessionUserId(void)
Definition: miscinit.c:508
Oid GetAuthenticatedUserId(void)
Definition: miscinit.c:545
Oid GetCurrentRoleId(void)
Definition: miscinit.c:935
void SerializeClientConnectionInfo(Size maxsize PG_USED_FOR_ASSERTS_ONLY, char *start_address)
Definition: miscinit.c:1051
void GetTempNamespaceState(Oid *tempNamespaceId, Oid *tempToastNamespaceId)
Definition: namespace.c:3875
Size EstimateUncommittedEnumsSpace(void)
Definition: pg_enum.c:826
void SerializeUncommittedEnums(void *space, Size size)
Definition: pg_enum.c:840
SerializableXactHandle ShareSerializableXact(void)
Definition: predicate.c:5047
Size EstimateRelationMapSpace(void)
Definition: relmapper.c:713
void SerializeRelationMap(Size maxSize, char *startAddress)
Definition: relmapper.c:724
dsm_handle GetSessionDsmHandle(void)
Definition: session.c:70
shm_mq * shm_mq_create(void *address, Size size)
Definition: shm_mq.c:177
void shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:206
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:290
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
Size shm_toc_estimate(shm_toc_estimator *e)
Definition: shm_toc.c:263
shm_toc * shm_toc_create(uint64 magic, void *address, Size nbytes)
Definition: shm_toc.c:40
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
Size mul_size(Size s1, Size s2)
Definition: shmem.c:510
void SerializeSnapshot(Snapshot snapshot, char *start_address)
Definition: snapmgr.c:1734
Snapshot GetTransactionSnapshot(void)
Definition: snapmgr.c:271
Size EstimateSnapshotSpace(Snapshot snapshot)
Definition: snapmgr.c:1710
Snapshot GetActiveSnapshot(void)
Definition: snapmgr.c:798
#define SpinLockInit(lock)
Definition: spin.h:57
PGPROC * MyProc
Definition: proc.c:66
void SerializePendingSyncs(Size maxSize, char *startAddress)
Definition: storage.c:600
Size EstimatePendingSyncsSpace(void)
Definition: storage.c:587
Oid temp_namespace_id
Definition: parallel.c:90
Oid temp_toast_namespace_id
Definition: parallel.c:91
XLogRecPtr last_xlog_end
Definition: parallel.c:106
bool role_is_superuser
Definition: parallel.c:94
Oid current_user_id
Definition: parallel.c:89
TimestampTz stmt_ts
Definition: parallel.c:99
slock_t mutex
Definition: parallel.c:103
Oid outer_user_id
Definition: parallel.c:88
SerializableXactHandle serializable_xact_handle
Definition: parallel.c:100
TimestampTz xact_ts
Definition: parallel.c:98
PGPROC * parallel_leader_pgproc
Definition: parallel.c:95
bool session_user_is_superuser
Definition: parallel.c:93
pid_t parallel_leader_pid
Definition: parallel.c:96
Oid session_user_id
Definition: parallel.c:87
Oid authenticated_user_id
Definition: parallel.c:86
ProcNumber parallel_leader_proc_number
Definition: parallel.c:97
shm_toc * toc
Definition: parallel.h:44
Definition: shm_mq.c:72
void SerializeTransactionState(Size maxsize, char *start_address)
Definition: xact.c:5552
Size EstimateTransactionStateSpace(void)
Definition: xact.c:5524
TimestampTz GetCurrentStatementStartTimestamp(void)
Definition: xact.c:879
TimestampTz GetCurrentTransactionStartTimestamp(void)
Definition: xact.c:870
#define IsolationUsesXactSnapshot()
Definition: xact.h:52

References FixedParallelState::authenticated_user_id, BUFFERALIGN, current_role_is_superuser, FixedParallelState::current_user_id, FixedParallelState::database_id, dsm_create(), DSM_CREATE_NULL_IF_MAXSEGMENTS, DSM_HANDLE_INVALID, dsm_segment_address(), ParallelWorkerInfo::error_mqh, EstimateClientConnectionInfoSpace(), EstimateComboCIDStateSpace(), EstimateGUCStateSpace(), EstimateLibraryStateSpace(), EstimatePendingSyncsSpace(), EstimateReindexStateSpace(), EstimateRelationMapSpace(), EstimateSnapshotSpace(), EstimateTransactionStateSpace(), EstimateUncommittedEnumsSpace(), ParallelContext::estimator, ParallelContext::function_name, GetActiveSnapshot(), GetAuthenticatedUserId(), GetCurrentRoleId(), GetCurrentStatementStartTimestamp(), GetCurrentTransactionStartTimestamp(), GetSessionDsmHandle(), GetSessionUserId(), GetSessionUserIsSuperuser(), GetTempNamespaceState(), GetTransactionSnapshot(), GetUserIdAndSecContext(), i, INTERRUPTS_CAN_BE_PROCESSED, IsolationUsesXactSnapshot, FixedParallelState::last_xlog_end, ParallelContext::library_name, MemoryContextAlloc(), MemoryContextSwitchTo(), mul_size(), FixedParallelState::mutex, MyDatabaseId, MyProc, MyProcNumber, MyProcPid, ParallelContext::nworkers, ParallelContext::nworkers_to_launch, FixedParallelState::outer_user_id, palloc0(), PARALLEL_ERROR_QUEUE_SIZE, PARALLEL_KEY_ACTIVE_SNAPSHOT, PARALLEL_KEY_CLIENTCONNINFO, PARALLEL_KEY_COMBO_CID, PARALLEL_KEY_ENTRYPOINT, PARALLEL_KEY_ERROR_QUEUE, PARALLEL_KEY_FIXED, PARALLEL_KEY_GUC, PARALLEL_KEY_LIBRARY, PARALLEL_KEY_PENDING_SYNCS, PARALLEL_KEY_REINDEX_STATE, PARALLEL_KEY_RELMAPPER_STATE, PARALLEL_KEY_SESSION_DSM, PARALLEL_KEY_TRANSACTION_SNAPSHOT, PARALLEL_KEY_TRANSACTION_STATE, PARALLEL_KEY_UNCOMMITTEDENUMS, FixedParallelState::parallel_leader_pgproc, FixedParallelState::parallel_leader_pid, FixedParallelState::parallel_leader_proc_number, PARALLEL_MAGIC, ParallelContext::private_memory, FixedParallelState::role_is_superuser, FixedParallelState::sec_context, ParallelContext::seg, FixedParallelState::serializable_xact_handle, SerializeClientConnectionInfo(), SerializeComboCIDState(), SerializeGUCState(), SerializeLibraryState(), SerializePendingSyncs(), SerializeReindexState(), SerializeRelationMap(), SerializeSnapshot(), SerializeTransactionState(), SerializeUncommittedEnums(), FixedParallelState::session_user_id, FixedParallelState::session_user_is_superuser, ShareSerializableXact(), shm_mq_attach(), shm_mq_create(), shm_mq_set_receiver(), shm_toc_allocate(), shm_toc_create(), shm_toc_estimate(), shm_toc_estimate_chunk, shm_toc_estimate_keys, shm_toc_insert(), SpinLockInit, start, StaticAssertStmt, FixedParallelState::stmt_ts, FixedParallelState::temp_namespace_id, FixedParallelState::temp_toast_namespace_id, ParallelContext::toc, TopMemoryContext, TopTransactionContext, ParallelContext::worker, and FixedParallelState::xact_ts.

Referenced by _brin_begin_parallel(), _bt_begin_parallel(), _gin_begin_parallel(), ExecInitParallelPlan(), and parallel_vacuum_init().

◆ LaunchParallelWorkers()

void LaunchParallelWorkers ( ParallelContext *  pcxt )

Definition at line 573 of file parallel.c.

574{
575 MemoryContext oldcontext;
576 BackgroundWorker worker;
577 int i;
578 bool any_registrations_failed = false;
579
580 /* Skip this if we have no workers. */
581 if (pcxt->nworkers == 0 || pcxt->nworkers_to_launch == 0)
582 return;
583
584 /* We need to be a lock group leader. */
586
587 /* If we do have workers, we'd better have a DSM segment. */
588 Assert(pcxt->seg != NULL);
589
590 /* We might be running in a short-lived memory context. */
592
593 /* Configure a worker. */
594 memset(&worker, 0, sizeof(worker));
595 snprintf(worker.bgw_name, BGW_MAXLEN, "parallel worker for PID %d",
596 MyProcPid);
597 snprintf(worker.bgw_type, BGW_MAXLEN, "parallel worker");
598 worker.bgw_flags =
603 sprintf(worker.bgw_library_name, "postgres");
604 sprintf(worker.bgw_function_name, "ParallelWorkerMain");
606 worker.bgw_notify_pid = MyProcPid;
607
608 /*
609 * Start workers.
610 *
611 * The caller must be able to tolerate ending up with fewer workers than
612 * expected, so there is no need to throw an error here if registration
613 * fails. It wouldn't help much anyway, because registering the worker in
614 * no way guarantees that it will start up and initialize successfully.
615 */
616 for (i = 0; i < pcxt->nworkers_to_launch; ++i)
617 {
618 memcpy(worker.bgw_extra, &i, sizeof(int));
619 if (!any_registrations_failed &&
621 &pcxt->worker[i].bgwhandle))
622 {
624 pcxt->worker[i].bgwhandle);
625 pcxt->nworkers_launched++;
626 }
627 else
628 {
629 /*
630 * If we weren't able to register the worker, then we've bumped up
631 * against the max_worker_processes limit, and future
632 * registrations will probably fail too, so arrange to skip them.
633 * But we still have to execute this code for the remaining slots
634 * to make sure that we forget about the error queues we budgeted
635 * for those workers. Otherwise, we'll wait for them to start,
636 * but they never will.
637 */
638 any_registrations_failed = true;
639 pcxt->worker[i].bgwhandle = NULL;
641 pcxt->worker[i].error_mqh = NULL;
642 }
643 }
644
645 /*
646 * Now that nworkers_launched has taken its final value, we can initialize
647 * known_attached_workers.
648 */
649 if (pcxt->nworkers_launched > 0)
650 {
652 palloc0(sizeof(bool) * pcxt->nworkers_launched);
653 pcxt->nknown_attached_workers = 0;
654 }
655
656 /* Restore previous memory context. */
657 MemoryContextSwitchTo(oldcontext);
658}
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:1046
#define BGW_NEVER_RESTART
Definition: bgworker.h:85
#define BGWORKER_CLASS_PARALLEL
Definition: bgworker.h:68
@ BgWorkerStart_ConsistentState
Definition: bgworker.h:80
#define BGWORKER_BACKEND_DATABASE_CONNECTION
Definition: bgworker.h:60
#define BGWORKER_SHMEM_ACCESS
Definition: bgworker.h:53
#define BGW_MAXLEN
Definition: bgworker.h:86
dsm_handle dsm_segment_handle(dsm_segment *seg)
Definition: dsm.c:1123
#define sprintf
Definition: port.h:241
#define snprintf
Definition: port.h:239
static Datum UInt32GetDatum(uint32 X)
Definition: postgres.h:242
void shm_mq_set_handle(shm_mq_handle *mqh, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:319
void BecomeLockGroupLeader(void)
Definition: proc.c:2001
char bgw_function_name[BGW_MAXLEN]
Definition: bgworker.h:97
Datum bgw_main_arg
Definition: bgworker.h:98
char bgw_name[BGW_MAXLEN]
Definition: bgworker.h:91
int bgw_restart_time
Definition: bgworker.h:95
char bgw_type[BGW_MAXLEN]
Definition: bgworker.h:92
BgWorkerStartTime bgw_start_time
Definition: bgworker.h:94
char bgw_extra[BGW_EXTRALEN]
Definition: bgworker.h:99
int bgw_flags
Definition: bgworker.h:93
pid_t bgw_notify_pid
Definition: bgworker.h:100
char bgw_library_name[MAXPGPATH]
Definition: bgworker.h:96
bool * known_attached_workers
Definition: parallel.h:47
int nknown_attached_workers
Definition: parallel.h:46

References Assert(), BecomeLockGroupLeader(), BackgroundWorker::bgw_extra, BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BackgroundWorker::bgw_name, BGW_NEVER_RESTART, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, ParallelWorkerInfo::bgwhandle, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_CLASS_PARALLEL, BGWORKER_SHMEM_ACCESS, BgWorkerStart_ConsistentState, dsm_segment_handle(), ParallelWorkerInfo::error_mqh, i, ParallelContext::known_attached_workers, MemoryContextSwitchTo(), MyProcPid, ParallelContext::nknown_attached_workers, ParallelContext::nworkers, ParallelContext::nworkers_launched, ParallelContext::nworkers_to_launch, palloc0(), RegisterDynamicBackgroundWorker(), ParallelContext::seg, shm_mq_detach(), shm_mq_set_handle(), snprintf, sprintf, TopTransactionContext, UInt32GetDatum(), and ParallelContext::worker.

Referenced by _brin_begin_parallel(), _bt_begin_parallel(), _gin_begin_parallel(), ExecGather(), ExecGatherMerge(), and parallel_vacuum_process_all_indexes().

◆ LookupParallelWorkerFunction()

static parallel_worker_main_type LookupParallelWorkerFunction ( const char *  libraryname,
const char *  funcname 
)
static

Definition at line 1642 of file parallel.c.

1643{
1644 /*
1645 * If the function is to be loaded from postgres itself, search the
1646 * InternalParallelWorkers array.
1647 */
1648 if (strcmp(libraryname, "postgres") == 0)
1649 {
1650 int i;
1651
1652 for (i = 0; i < lengthof(InternalParallelWorkers); i++)
1653 {
1654 if (strcmp(InternalParallelWorkers[i].fn_name, funcname) == 0)
1656 }
1657
1658 /* We can only reach this by programming error. */
1659 elog(ERROR, "internal function \"%s\" not found", funcname);
1660 }
1661
1662 /* Otherwise load from external library. */
1664 load_external_function(libraryname, funcname, true, NULL);
1665}
parallel_worker_main_type fn_addr
Definition: parallel.c:139
const char * fn_name
Definition: parallel.c:138
static const struct @16 InternalParallelWorkers[]
#define lengthof(array)
Definition: c.h:787
void * load_external_function(const char *filename, const char *funcname, bool signalNotFound, void **filehandle)
Definition: dfmgr.c:95
#define ERROR
Definition: elog.h:39
void(* parallel_worker_main_type)(dsm_segment *seg, shm_toc *toc)
Definition: parallel.h:23
#define funcname
Definition: indent_codes.h:69

References elog, ERROR, fn_addr, fn_name, funcname, i, InternalParallelWorkers, lengthof, and load_external_function().

Referenced by ParallelWorkerMain().

◆ ParallelContextActive()

bool ParallelContextActive ( void  )

Definition at line 1024 of file parallel.c.

1025{
1026 return !dlist_is_empty(&pcxt_list);
1027}

References dlist_is_empty(), and pcxt_list.

Referenced by AtPrepare_PredicateLocks(), ExitParallelMode(), and ReleasePredicateLocks().

◆ ParallelWorkerMain()

void ParallelWorkerMain ( Datum  main_arg )

Definition at line 1292 of file parallel.c.

1293{
1294 dsm_segment *seg;
1295 shm_toc *toc;
1296 FixedParallelState *fps;
1297 char *error_queue_space;
1298 shm_mq *mq;
1299 shm_mq_handle *mqh;
1300 char *libraryspace;
1301 char *entrypointstate;
1302 char *library_name;
1303 char *function_name;
1305 char *gucspace;
1306 char *combocidspace;
1307 char *tsnapspace;
1308 char *asnapspace;
1309 char *tstatespace;
1310 char *pendingsyncsspace;
1311 char *reindexspace;
1312 char *relmapperspace;
1313 char *uncommittedenumsspace;
1314 char *clientconninfospace;
1315 char *session_dsm_handle_space;
1316 Snapshot tsnapshot;
1317 Snapshot asnapshot;
1318
1319 /* Set flag to indicate that we're initializing a parallel worker. */
1321
1322 /* Establish signal handlers. */
1323 pqsignal(SIGTERM, die);
1325
1326 /* Determine and set our parallel worker number. */
1328 memcpy(&ParallelWorkerNumber, MyBgworkerEntry->bgw_extra, sizeof(int));
1329
1330 /* Set up a memory context to work in, just for cleanliness. */
1332 "Parallel worker",
1334
1335 /*
1336 * Attach to the dynamic shared memory segment for the parallel query, and
1337 * find its table of contents.
1338 *
1339 * Note: at this point, we have not created any ResourceOwner in this
1340 * process. This will result in our DSM mapping surviving until process
1341 * exit, which is fine. If there were a ResourceOwner, it would acquire
1342 * ownership of the mapping, but we have no need for that.
1343 */
1344 seg = dsm_attach(DatumGetUInt32(main_arg));
1345 if (seg == NULL)
1346 ereport(ERROR,
1347 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1348 errmsg("could not map dynamic shared memory segment")));
1350 if (toc == NULL)
1351 ereport(ERROR,
1352 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1353 errmsg("invalid magic number in dynamic shared memory segment")));
1354
1355 /* Look up fixed parallel state. */
1356 fps = shm_toc_lookup(toc, PARALLEL_KEY_FIXED, false);
1358
1359 /* Arrange to signal the leader if we exit. */
1363
1364 /*
1365 * Now we can find and attach to the error queue provided for us. That's
1366 * good, because until we do that, any errors that happen here will not be
1367 * reported back to the process that requested that this worker be
1368 * launched.
1369 */
1370 error_queue_space = shm_toc_lookup(toc, PARALLEL_KEY_ERROR_QUEUE, false);
1371 mq = (shm_mq *) (error_queue_space +
1374 mqh = shm_mq_attach(mq, seg, NULL);
1375 pq_redirect_to_shm_mq(seg, mqh);
1378
1379 /*
1380 * Hooray! Primary initialization is complete. Now, we need to set up our
1381 * backend-local state to match the original backend.
1382 */
1383
1384 /*
1385 * Join locking group. We must do this before anything that could try to
1386 * acquire a heavyweight lock, because any heavyweight locks acquired to
1387 * this point could block either directly against the parallel group
1388 * leader or against some process which in turn waits for a lock that
1389 * conflicts with the parallel group leader, causing an undetected
1390 * deadlock. (If we can't join the lock group, the leader has gone away,
1391 * so just exit quietly.)
1392 */
1394 fps->parallel_leader_pid))
1395 return;
1396
1397 /*
1398 * Restore transaction and statement start-time timestamps. This must
1399 * happen before anything that would start a transaction, else asserts in
1400 * xact.c will fire.
1401 */
1403
1404 /*
1405 * Identify the entry point to be called. In theory this could result in
1406 * loading an additional library, though most likely the entry point is in
1407 * the core backend or in a library we just loaded.
1408 */
1409 entrypointstate = shm_toc_lookup(toc, PARALLEL_KEY_ENTRYPOINT, false);
1410 library_name = entrypointstate;
1411 function_name = entrypointstate + strlen(library_name) + 1;
1412
1413 entrypt = LookupParallelWorkerFunction(library_name, function_name);
1414
1415 /*
1416 * Restore current session authorization and role id. No verification
1417 * happens here, we just blindly adopt the leader's state. Note that this
1418 * has to happen before InitPostgres, since InitializeSessionUserId will
1419 * not set these variables.
1420 */
1425
1426 /*
1427 * Restore database connection. We skip connection authorization checks,
1428 * reasoning that (a) the leader checked these things when it started, and
1429 * (b) we do not want parallel mode to cause these failures, because that
1430 * would make use of parallel query plans not transparent to applications.
1431 */
1436
1437 /*
1438 * Set the client encoding to the database encoding, since that is what
1439 * the leader will expect. (We're cheating a bit by not calling
1440 * PrepareClientEncoding first. It's okay because this call will always
1441 * result in installing a no-op conversion. No error should be possible,
1442 * but check anyway.)
1443 */
1445 elog(ERROR, "SetClientEncoding(%d) failed", GetDatabaseEncoding());
1446
1447 /*
1448 * Load libraries that were loaded by original backend. We want to do
1449 * this before restoring GUCs, because the libraries might define custom
1450 * variables.
1451 */
1452 libraryspace = shm_toc_lookup(toc, PARALLEL_KEY_LIBRARY, false);
1454 RestoreLibraryState(libraryspace);
1456
1457 /* Crank up a transaction state appropriate to a parallel worker. */
1458 tstatespace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_STATE, false);
1459 StartParallelWorkerTransaction(tstatespace);
1460
1461 /*
1462 * Restore state that affects catalog access. Ideally we'd do this even
1463 * before calling InitPostgres, but that has order-of-initialization
1464 * problems, and also the relmapper would get confused during the
1465 * CommitTransactionCommand call above.
1466 */
1467 pendingsyncsspace = shm_toc_lookup(toc, PARALLEL_KEY_PENDING_SYNCS,
1468 false);
1469 RestorePendingSyncs(pendingsyncsspace);
1470 relmapperspace = shm_toc_lookup(toc, PARALLEL_KEY_RELMAPPER_STATE, false);
1471 RestoreRelationMap(relmapperspace);
1472 reindexspace = shm_toc_lookup(toc, PARALLEL_KEY_REINDEX_STATE, false);
1473 RestoreReindexState(reindexspace);
1474 combocidspace = shm_toc_lookup(toc, PARALLEL_KEY_COMBO_CID, false);
1475 RestoreComboCIDState(combocidspace);
1476
1477 /* Attach to the per-session DSM segment and contained objects. */
1478 session_dsm_handle_space =
1480 AttachSession(*(dsm_handle *) session_dsm_handle_space);
1481
1482 /*
1483 * If the transaction isolation level is REPEATABLE READ or SERIALIZABLE,
1484 * the leader has serialized the transaction snapshot and we must restore
1485 * it. At lower isolation levels, there is no transaction-lifetime
1486 * snapshot, but we need TransactionXmin to get set to a value which is
1487 * less than or equal to the xmin of every snapshot that will be used by
1488 * this worker. The easiest way to accomplish that is to install the
1489 * active snapshot as the transaction snapshot. Code running in this
1490 * parallel worker might take new snapshots via GetTransactionSnapshot()
1491 * or GetLatestSnapshot(), but it shouldn't have any way of acquiring a
1492 * snapshot older than the active snapshot.
1493 */
1494 asnapspace = shm_toc_lookup(toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, false);
1495 tsnapspace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT, true);
1496 asnapshot = RestoreSnapshot(asnapspace);
1497 tsnapshot = tsnapspace ? RestoreSnapshot(tsnapspace) : asnapshot;
1500 PushActiveSnapshot(asnapshot);
1501
1502 /*
1503 * We've changed which tuples we can see, and must therefore invalidate
1504 * system caches.
1505 */
1507
1508 /*
1509 * Restore GUC values from launching backend. We can't do this earlier,
1510 * because GUC check hooks that do catalog lookups need to see the same
1511 * database state as the leader. Also, the check hooks for
1512 * session_authorization and role assume we already set the correct role
1513 * OIDs.
1514 */
1515 gucspace = shm_toc_lookup(toc, PARALLEL_KEY_GUC, false);
1516 RestoreGUCState(gucspace);
1517
1518 /*
1519 * Restore current user ID and security context. No verification happens
1520 * here, we just blindly adopt the leader's state. We can't do this till
1521 * after restoring GUCs, else we'll get complaints about restoring
1522 * session_authorization and role. (In effect, we're assuming that all
1523 * the restored values are okay to set, even if we are now inside a
1524 * restricted context.)
1525 */
1527
1528 /* Restore temp-namespace state to ensure search path matches leader's. */
1531
1532 /* Restore uncommitted enums. */
1533 uncommittedenumsspace = shm_toc_lookup(toc, PARALLEL_KEY_UNCOMMITTEDENUMS,
1534 false);
1535 RestoreUncommittedEnums(uncommittedenumsspace);
1536
1537 /* Restore the ClientConnectionInfo. */
1538 clientconninfospace = shm_toc_lookup(toc, PARALLEL_KEY_CLIENTCONNINFO,
1539 false);
1540 RestoreClientConnectionInfo(clientconninfospace);
1541
1542 /*
1543 * Initialize SystemUser now that MyClientConnectionInfo is restored. Also
1544 * ensure that auth_method is actually valid, aka authn_id is not NULL.
1545 */
1549
1550 /* Attach to the leader's serializable transaction, if SERIALIZABLE. */
1552
1553 /*
1554 * We've initialized all of our state now; nothing should change
1555 * hereafter.
1556 */
1559
1560 /*
1561 * Time to do the real work: invoke the caller-supplied code.
1562 */
1563 entrypt(seg, toc);
1564
1565 /* Must exit parallel mode to pop active snapshot. */
1567
1568 /* Must pop active snapshot so snapmgr.c doesn't complain. */
1570
1571 /* Shut down the parallel-worker transaction. */
1573
1574 /* Detach from the per-session DSM segment. */
1575 DetachSession();
1576
1577 /* Report success. */
1579}
static parallel_worker_main_type LookupParallelWorkerFunction(const char *libraryname, const char *funcname)
Definition: parallel.c:1642
int ParallelWorkerNumber
Definition: parallel.c:115
bool InitializingParallelWorker
Definition: parallel.c:121
static FixedParallelState * MyFixedParallelState
Definition: parallel.c:124
static pid_t ParallelLeaderPid
Definition: parallel.c:130
static void ParallelWorkerShutdown(int code, Datum arg)
Definition: parallel.c:1614
void BackgroundWorkerUnblockSignals(void)
Definition: bgworker.c:927
void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags)
Definition: bgworker.c:887
#define BGWORKER_BYPASS_ROLELOGINCHECK
Definition: bgworker.h:157
#define BGWORKER_BYPASS_ALLOWCONN
Definition: bgworker.h:156
void RestoreComboCIDState(char *comboCIDstate)
Definition: combocid.c:342
void RestoreLibraryState(char *start_address)
Definition: dfmgr.c:741
dsm_segment * dsm_attach(dsm_handle h)
Definition: dsm.c:665
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define ereport(elevel,...)
Definition: elog.h:150
ProcNumber ParallelLeaderProcNumber
Definition: globals.c:92
void RestoreGUCState(void *gucstate)
Definition: guc.c:6216
const char * hba_authname(UserAuth auth_method)
Definition: hba.c:3141
void RestoreReindexState(const void *reindexstate)
Definition: index.c:4271
void InvalidateSystemCaches(void)
Definition: inval.c:916
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
#define pq_putmessage(msgtype, s, len)
Definition: libpq.h:49
int GetDatabaseEncoding(void)
Definition: mbutils.c:1262
int SetClientEncoding(int encoding)
Definition: mbutils.c:209
MemoryContext CurrentMemoryContext
Definition: mcxt.c:160
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:160
void InitializeSystemUser(const char *authn_id, const char *auth_method)
Definition: miscinit.c:874
void SetSessionAuthorization(Oid userid, bool is_superuser)
Definition: miscinit.c:920
void SetCurrentRoleId(Oid roleid, bool is_superuser)
Definition: miscinit.c:956
ClientConnectionInfo MyClientConnectionInfo
Definition: miscinit.c:1018
void RestoreClientConnectionInfo(char *conninfo)
Definition: miscinit.c:1084
void SetAuthenticatedUserId(Oid userid)
Definition: miscinit.c:552
void SetUserIdAndSecContext(Oid userid, int sec_context)
Definition: miscinit.c:619
void SetTempNamespaceState(Oid tempNamespaceId, Oid tempToastNamespaceId)
Definition: namespace.c:3891
void RestoreUncommittedEnums(void *space)
Definition: pg_enum.c:886
#define die(msg)
Definition: pg_test_fsync.c:100
#define pqsignal
Definition: port.h:531
static uint32 DatumGetUInt32(Datum X)
Definition: postgres.h:232
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:332
BackgroundWorker * MyBgworkerEntry
Definition: postmaster.c:200
void pq_set_parallel_leader(pid_t pid, ProcNumber procNumber)
Definition: pqmq.c:82
void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
Definition: pqmq.c:53
void AttachSerializableXact(SerializableXactHandle handle)
Definition: predicate.c:5056
#define PqMsg_Terminate
Definition: protocol.h:28
void RestoreRelationMap(char *startAddress)
Definition: relmapper.c:741
void DetachSession(void)
Definition: session.c:201
void AttachSession(dsm_handle handle)
Definition: session.c:155
void shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:224
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
shm_toc * shm_toc_attach(uint64 magic, void *address)
Definition: shm_toc.c:64
void PushActiveSnapshot(Snapshot snapshot)
Definition: snapmgr.c:680
Snapshot RestoreSnapshot(char *start_address)
Definition: snapmgr.c:1791
void RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc)
Definition: snapmgr.c:1854
void PopActiveSnapshot(void)
Definition: snapmgr.c:773
bool BecomeLockGroupMember(PGPROC *leader, int pid)
Definition: proc.c:2031
void RestorePendingSyncs(char *startAddress)
Definition: storage.c:651
const char * authn_id
Definition: libpq-be.h:99
UserAuth auth_method
Definition: libpq-be.h:105
Definition: dsm.c:67
Definition: shm_mq.c:138
Definition: shm_toc.c:27
void ExitParallelMode(void)
Definition: xact.c:1064
void EnterParallelMode(void)
Definition: xact.c:1051
void StartTransactionCommand(void)
Definition: xact.c:3071
void StartParallelWorkerTransaction(char *tstatespace)
Definition: xact.c:5623
void SetParallelStartTimestamps(TimestampTz xact_ts, TimestampTz stmt_ts)
Definition: xact.c:859
void EndParallelWorkerTransaction(void)
Definition: xact.c:5648
void CommitTransactionCommand(void)
Definition: xact.c:3169

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert(), AttachSerializableXact(), AttachSession(), ClientConnectionInfo::auth_method, FixedParallelState::authenticated_user_id, ClientConnectionInfo::authn_id, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), BecomeLockGroupMember(), before_shmem_exit(), BackgroundWorker::bgw_extra, BGWORKER_BYPASS_ALLOWCONN, BGWORKER_BYPASS_ROLELOGINCHECK, CommitTransactionCommand(), FixedParallelState::current_user_id, CurrentMemoryContext, FixedParallelState::database_id, DatumGetUInt32(), DetachSession(), die, dsm_attach(), dsm_segment_address(), elog, EndParallelWorkerTransaction(), EnterParallelMode(), ereport, errcode(), errmsg(), ERROR, ExitParallelMode(), GetDatabaseEncoding(), hba_authname(), InitializeSystemUser(), InitializingParallelWorker, InvalidateSystemCaches(), LookupParallelWorkerFunction(), MyBgworkerEntry, MyClientConnectionInfo, MyFixedParallelState, MyProc, FixedParallelState::outer_user_id, PARALLEL_ERROR_QUEUE_SIZE, PARALLEL_KEY_ACTIVE_SNAPSHOT, PARALLEL_KEY_CLIENTCONNINFO, PARALLEL_KEY_COMBO_CID, PARALLEL_KEY_ENTRYPOINT, PARALLEL_KEY_ERROR_QUEUE, PARALLEL_KEY_FIXED, PARALLEL_KEY_GUC, PARALLEL_KEY_LIBRARY, PARALLEL_KEY_PENDING_SYNCS, PARALLEL_KEY_REINDEX_STATE, PARALLEL_KEY_RELMAPPER_STATE, PARALLEL_KEY_SESSION_DSM, PARALLEL_KEY_TRANSACTION_SNAPSHOT, PARALLEL_KEY_TRANSACTION_STATE, PARALLEL_KEY_UNCOMMITTEDENUMS, FixedParallelState::parallel_leader_pgproc, FixedParallelState::parallel_leader_pid, FixedParallelState::parallel_leader_proc_number, PARALLEL_MAGIC, ParallelLeaderPid, ParallelLeaderProcNumber, ParallelWorkerNumber, ParallelWorkerShutdown(), PointerGetDatum(), PopActiveSnapshot(), pq_putmessage, pq_redirect_to_shm_mq(), pq_set_parallel_leader(), PqMsg_Terminate, pqsignal, PushActiveSnapshot(), RestoreClientConnectionInfo(), RestoreComboCIDState(), RestoreGUCState(), RestoreLibraryState(), RestorePendingSyncs(), RestoreReindexState(), RestoreRelationMap(), RestoreSnapshot(), RestoreTransactionSnapshot(), RestoreUncommittedEnums(), FixedParallelState::role_is_superuser, FixedParallelState::sec_context, FixedParallelState::serializable_xact_handle, FixedParallelState::session_user_id, FixedParallelState::session_user_is_superuser, SetAuthenticatedUserId(), SetClientEncoding(), SetCurrentRoleId(), SetParallelStartTimestamps(), SetSessionAuthorization(), SetTempNamespaceState(), SetUserIdAndSecContext(), shm_mq_attach(), shm_mq_set_sender(), shm_toc_attach(), shm_toc_lookup(), StartParallelWorkerTransaction(), StartTransactionCommand(), FixedParallelState::stmt_ts, FixedParallelState::temp_namespace_id, FixedParallelState::temp_toast_namespace_id, TopMemoryContext, and FixedParallelState::xact_ts.

◆ ParallelWorkerReportLastRecEnd()

void ParallelWorkerReportLastRecEnd ( XLogRecPtr  last_xlog_end )

Definition at line 1586 of file parallel.c.

1587{
1589
1590 Assert(fps != NULL);
1591 SpinLockAcquire(&fps->mutex);
1592 if (fps->last_xlog_end < last_xlog_end)
1593 fps->last_xlog_end = last_xlog_end;
1594 SpinLockRelease(&fps->mutex);
1595}
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59

References Assert(), FixedParallelState::last_xlog_end, FixedParallelState::mutex, MyFixedParallelState, SpinLockAcquire, and SpinLockRelease.

Referenced by CommitTransaction().

◆ ParallelWorkerShutdown()

static void ParallelWorkerShutdown ( int  code,
Datum  arg 
)
static

Definition at line 1614 of file parallel.c.

1615{
1619
1621}
void * arg
static Pointer DatumGetPointer(Datum X)
Definition: postgres.h:322
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
Definition: procsignal.c:284
@ PROCSIG_PARALLEL_MESSAGE
Definition: procsignal.h:34

References arg, DatumGetPointer(), dsm_detach(), ParallelLeaderPid, ParallelLeaderProcNumber, PROCSIG_PARALLEL_MESSAGE, and SendProcSignal().

Referenced by ParallelWorkerMain().

◆ ProcessParallelMessage()

static void ProcessParallelMessage ( ParallelContext *  pcxt,
int  i,
StringInfo  msg 
)
static

Definition at line 1137 of file parallel.c.

1138{
1139 char msgtype;
1140
1141 if (pcxt->known_attached_workers != NULL &&
1142 !pcxt->known_attached_workers[i])
1143 {
1144 pcxt->known_attached_workers[i] = true;
1146 }
1147
1148 msgtype = pq_getmsgbyte(msg);
1149
1150 switch (msgtype)
1151 {
1154 {
1155 ErrorData edata;
1156 ErrorContextCallback *save_error_context_stack;
1157
1158 /* Parse ErrorResponse or NoticeResponse. */
1159 pq_parse_errornotice(msg, &edata);
1160
1161 /* Death of a worker isn't enough justification for suicide. */
1162 edata.elevel = Min(edata.elevel, ERROR);
1163
1164 /*
1165 * If desired, add a context line to show that this is a
1166 * message propagated from a parallel worker. Otherwise, it
1167 * can sometimes be confusing to understand what actually
1168 * happened. (We don't do this in DEBUG_PARALLEL_REGRESS mode
1169 * because it causes test-result instability depending on
1170 * whether a parallel worker is actually used or not.)
1171 */
1173 {
1174 if (edata.context)
1175 edata.context = psprintf("%s\n%s", edata.context,
1176 _("parallel worker"));
1177 else
1178 edata.context = pstrdup(_("parallel worker"));
1179 }
1180
1181 /*
1182 * Context beyond that should use the error context callbacks
1183 * that were in effect when the ParallelContext was created,
1184 * not the current ones.
1185 */
1186 save_error_context_stack = error_context_stack;
1188
1189 /* Rethrow error or print notice. */
1190 ThrowErrorData(&edata);
1191
1192 /* Not an error, so restore previous context stack. */
1193 error_context_stack = save_error_context_stack;
1194
1195 break;
1196 }
1197
1199 {
1200 /* Propagate NotifyResponse. */
1201 int32 pid;
1202 const char *channel;
1203 const char *payload;
1204
1205 pid = pq_getmsgint(msg, 4);
1206 channel = pq_getmsgrawstring(msg);
1207 payload = pq_getmsgrawstring(msg);
1208 pq_endmessage(msg);
1209
1210 NotifyMyFrontEnd(channel, payload, pid);
1211
1212 break;
1213 }
1214
1215 case PqMsg_Progress:
1216 {
1217 /*
1218 * Only incremental progress reporting is currently supported.
1219 * However, it's possible to add more fields to the message to
1220 * allow for handling of other backend progress APIs.
1221 */
1222 int index = pq_getmsgint(msg, 4);
1223 int64 incr = pq_getmsgint64(msg);
1224
1225 pq_getmsgend(msg);
1226
1228
1229 break;
1230 }
1231
1232 case PqMsg_Terminate:
1233 {
1235 pcxt->worker[i].error_mqh = NULL;
1236 break;
1237 }
1238
1239 default:
1240 {
1241 elog(ERROR, "unrecognized message type received from parallel worker: %c (message length %d bytes)",
1242 msgtype, msg->len);
1243 }
1244 }
1245}
void NotifyMyFrontEnd(const char *channel, const char *payload, int32 srcPid)
Definition: async.c:2224
void pgstat_progress_incr_param(int index, int64 incr)
#define Min(x, y)
Definition: c.h:1003
int64_t int64
Definition: c.h:535
int32_t int32
Definition: c.h:534
void ThrowErrorData(ErrorData *edata)
Definition: elog.c:1903
_
#define _(x)
Definition: elog.c:91
@ DEBUG_PARALLEL_REGRESS
Definition: optimizer.h:97
int debug_parallel_query
Definition: planner.c:69
unsigned int pq_getmsgint(StringInfo msg, int b)
Definition: pqformat.c:415
void pq_getmsgend(StringInfo msg)
Definition: pqformat.c:635
void pq_endmessage(StringInfo buf)
Definition: pqformat.c:296
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:399
const char * pq_getmsgrawstring(StringInfo msg)
Definition: pqformat.c:608
int64 pq_getmsgint64(StringInfo msg)
Definition: pqformat.c:453
void pq_parse_errornotice(StringInfo msg, ErrorData *edata)
Definition: pqmq.c:222
#define PqMsg_NotificationResponse
Definition: protocol.h:41
#define PqMsg_Progress
Definition: protocol.h:70
#define PqMsg_ErrorResponse
Definition: protocol.h:44
#define PqMsg_NoticeResponse
Definition: protocol.h:49
char * psprintf(const char *fmt,...)
Definition: psprintf.c:43
Definition: elog.h:420
char * context
Definition: elog.h:436
int elevel
Definition: elog.h:421
Definition: type.h:96

References _, ErrorData::context, debug_parallel_query, DEBUG_PARALLEL_REGRESS, ErrorData::elevel, elog, ERROR, error_context_stack, ParallelContext::error_context_stack, ParallelWorkerInfo::error_mqh, i, ParallelContext::known_attached_workers, StringInfoData::len, Min, ParallelContext::nknown_attached_workers, NotifyMyFrontEnd(), pgstat_progress_incr_param(), pq_endmessage(), pq_getmsgbyte(), pq_getmsgend(), pq_getmsgint(), pq_getmsgint64(), pq_getmsgrawstring(), pq_parse_errornotice(), PqMsg_ErrorResponse, PqMsg_NoticeResponse, PqMsg_NotificationResponse, PqMsg_Progress, PqMsg_Terminate, psprintf(), pstrdup(), shm_mq_detach(), ThrowErrorData(), and ParallelContext::worker.

Referenced by ProcessParallelMessages().

◆ ProcessParallelMessages()

void ProcessParallelMessages ( void  )

Definition at line 1048 of file parallel.c.

1049{
1050 dlist_iter iter;
1051 MemoryContext oldcontext;
1052
1053 static MemoryContext hpm_context = NULL;
1054
1055 /*
1056 * This is invoked from ProcessInterrupts(), and since some of the
1057 * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
1058 * for recursive calls if more signals are received while this runs. It's
1059 * unclear that recursive entry would be safe, and it doesn't seem useful
1060 * even if it is safe, so let's block interrupts until done.
1061 */
1063
1064 /*
1065 * Moreover, CurrentMemoryContext might be pointing almost anywhere. We
1066 * don't want to risk leaking data into long-lived contexts, so let's do
1067 * our work here in a private context that we can reset on each use.
1068 */
1069 if (hpm_context == NULL) /* first time through? */
1071 "ProcessParallelMessages",
1073 else
1074 MemoryContextReset(hpm_context);
1075
1076 oldcontext = MemoryContextSwitchTo(hpm_context);
1077
1078 /* OK to process messages. Reset the flag saying there are more to do. */
1079 ParallelMessagePending = false;
1080
1081 dlist_foreach(iter, &pcxt_list)
1082 {
1083 ParallelContext *pcxt;
1084 int i;
1085
1086 pcxt = dlist_container(ParallelContext, node, iter.cur);
1087 if (pcxt->worker == NULL)
1088 continue;
1089
1090 for (i = 0; i < pcxt->nworkers_launched; ++i)
1091 {
1092 /*
1093 * Read as many messages as we can from each worker, but stop when
1094 * either (1) the worker's error queue goes away, which can happen
1095 * if we receive a Terminate message from the worker; or (2) no
1096 * more messages can be read from the worker without blocking.
1097 */
1098 while (pcxt->worker[i].error_mqh != NULL)
1099 {
1100 shm_mq_result res;
1101 Size nbytes;
1102 void *data;
1103
1104 res = shm_mq_receive(pcxt->worker[i].error_mqh, &nbytes,
1105 &data, true);
1106 if (res == SHM_MQ_WOULD_BLOCK)
1107 break;
1108 else if (res == SHM_MQ_SUCCESS)
1109 {
1110 StringInfoData msg;
1111
1112 initStringInfo(&msg);
1113 appendBinaryStringInfo(&msg, data, nbytes);
1114 ProcessParallelMessage(pcxt, i, &msg);
1115 pfree(msg.data);
1116 }
1117 else
1118 ereport(ERROR,
1119 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1120 errmsg("lost connection to parallel worker")));
1121 }
1122 }
1123 }
1124
1125 MemoryContextSwitchTo(oldcontext);
1126
1127 /* Might as well clear the context on our way out */
1128 MemoryContextReset(hpm_context);
1129
1131}
static void ProcessParallelMessage(ParallelContext *pcxt, int i, StringInfo msg)
Definition: parallel.c:1137
#define dlist_foreach(iter, lhead)
Definition: ilist.h:623
#define dlist_container(type, membername, ptr)
Definition: ilist.h:593
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:400
const void * data
shm_mq_result shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
Definition: shm_mq.c:572
shm_mq_result
Definition: shm_mq.h:37
@ SHM_MQ_SUCCESS
Definition: shm_mq.h:38
@ SHM_MQ_WOULD_BLOCK
Definition: shm_mq.h:39
void appendBinaryStringInfo(StringInfo str, const void *data, int datalen)
Definition: stringinfo.c:281
void initStringInfo(StringInfo str)
Definition: stringinfo.c:97
char * data
Definition: stringinfo.h:48
Definition: ilist.h:178
dlist_node * cur
Definition: ilist.h:179

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, appendBinaryStringInfo(), dlist_iter::cur, StringInfoData::data, data, dlist_container, dlist_foreach, ereport, errcode(), errmsg(), ERROR, ParallelWorkerInfo::error_mqh, HOLD_INTERRUPTS, i, initStringInfo(), MemoryContextReset(), MemoryContextSwitchTo(), ParallelContext::nworkers_launched, ParallelMessagePending, pcxt_list, pfree(), ProcessParallelMessage(), RESUME_INTERRUPTS, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, TopMemoryContext, and ParallelContext::worker.

Referenced by ProcessInterrupts().

◆ ReinitializeParallelDSM()

void ReinitializeParallelDSM ( ParallelContext *  pcxt )

Definition at line 508 of file parallel.c.

509{
511
512 /* Wait for any old workers to exit. */
513 if (pcxt->nworkers_launched > 0)
514 {
517 pcxt->nworkers_launched = 0;
518 if (pcxt->known_attached_workers)
519 {
521 pcxt->known_attached_workers = NULL;
522 pcxt->nknown_attached_workers = 0;
523 }
524 }
525
526 /* Reset a few bits of fixed parallel state to a clean state. */
527 fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);
528 fps->last_xlog_end = 0;
529
530 /* Recreate error queues (if they exist). */
531 if (pcxt->nworkers > 0)
532 {
533 char *error_queue_space;
534 int i;
535
536 error_queue_space =
538 for (i = 0; i < pcxt->nworkers; ++i)
539 {
540 char *start;
541 shm_mq *mq;
542
543 start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
546 pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
547 }
548 }
549}
void WaitForParallelWorkersToFinish(ParallelContext *pcxt)
Definition: parallel.c:796

References ParallelWorkerInfo::error_mqh, i, ParallelContext::known_attached_workers, FixedParallelState::last_xlog_end, MyProc, ParallelContext::nknown_attached_workers, ParallelContext::nworkers, ParallelContext::nworkers_launched, PARALLEL_ERROR_QUEUE_SIZE, PARALLEL_KEY_ERROR_QUEUE, PARALLEL_KEY_FIXED, pfree(), ParallelContext::seg, shm_mq_attach(), shm_mq_create(), shm_mq_set_receiver(), shm_toc_lookup(), start, ParallelContext::toc, WaitForParallelWorkersToExit(), WaitForParallelWorkersToFinish(), and ParallelContext::worker.

Referenced by ExecParallelReinitialize(), and parallel_vacuum_process_all_indexes().

◆ ReinitializeParallelWorkers()

void ReinitializeParallelWorkers ( ParallelContext *  pcxt,
int  nworkers_to_launch 
)

Definition at line 558 of file parallel.c.

559{
560 /*
561 * The number of workers that need to be launched must be less than the
562 * number of workers with which the parallel context is initialized. But
563 * the caller might not know that InitializeParallelDSM reduced nworkers,
564 * so just silently trim the request.
565 */
566 pcxt->nworkers_to_launch = Min(pcxt->nworkers, nworkers_to_launch);
567}

References Min, ParallelContext::nworkers, and ParallelContext::nworkers_to_launch.

Referenced by parallel_vacuum_process_all_indexes().

◆ WaitForParallelWorkersToAttach()

void WaitForParallelWorkersToAttach ( ParallelContext *  pcxt )

Definition at line 693 of file parallel.c.

694{
695 int i;
696
697 /* Skip this if we have no launched workers. */
698 if (pcxt->nworkers_launched == 0)
699 return;
700
701 for (;;)
702 {
703 /*
704 * This will process any parallel messages that are pending and it may
705 * also throw an error propagated from a worker.
706 */
708
709 for (i = 0; i < pcxt->nworkers_launched; ++i)
710 {
711 BgwHandleStatus status;
712 shm_mq *mq;
713 int rc;
714 pid_t pid;
715
716 if (pcxt->known_attached_workers[i])
717 continue;
718
719 /*
720 * If error_mqh is NULL, then the worker has already exited
721 * cleanly.
722 */
723 if (pcxt->worker[i].error_mqh == NULL)
724 {
725 pcxt->known_attached_workers[i] = true;
727 continue;
728 }
729
730 status = GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle, &pid);
731 if (status == BGWH_STARTED)
732 {
733 /* Has the worker attached to the error queue? */
734 mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
735 if (shm_mq_get_sender(mq) != NULL)
736 {
737 /* Yes, so it is known to be attached. */
738 pcxt->known_attached_workers[i] = true;
740 }
741 }
742 else if (status == BGWH_STOPPED)
743 {
744 /*
745 * If the worker stopped without attaching to the error queue,
746 * throw an error.
747 */
748 mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
749 if (shm_mq_get_sender(mq) == NULL)
751 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
752 errmsg("parallel worker failed to initialize"),
753 errhint("More details may be available in the server log.")));
754
755 pcxt->known_attached_workers[i] = true;
757 }
758 else
759 {
760 /*
761 * Worker not yet started, so we must wait. The postmaster
762 * will notify us if the worker's state changes. Our latch
763 * might also get set for some other reason, but if so we'll
764 * just end up waiting for the same worker again.
765 */
766 rc = WaitLatch(MyLatch,
768 -1, WAIT_EVENT_BGWORKER_STARTUP);
769
770 if (rc & WL_LATCH_SET)
772 }
773 }
774
775 /* If all workers are known to have started, we're done. */
777 {
779 break;
780 }
781 }
782}
BgwHandleStatus GetBackgroundWorkerPid(BackgroundWorkerHandle *handle, pid_t *pidp)
Definition: bgworker.c:1158
BgwHandleStatus
Definition: bgworker.h:104
@ BGWH_STARTED
Definition: bgworker.h:105
@ BGWH_STOPPED
Definition: bgworker.h:107
int errhint(const char *fmt,...)
Definition: elog.c:1321
void ResetLatch(Latch *latch)
Definition: latch.c:374
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:172
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
shm_mq * shm_mq_get_queue(shm_mq_handle *mqh)
Definition: shm_mq.c:905
PGPROC * shm_mq_get_sender(shm_mq *mq)
Definition: shm_mq.c:257
#define WL_EXIT_ON_PM_DEATH
Definition: waiteventset.h:39
#define WL_LATCH_SET
Definition: waiteventset.h:34

References Assert(), BGWH_STARTED, BGWH_STOPPED, ParallelWorkerInfo::bgwhandle, CHECK_FOR_INTERRUPTS, ereport, errcode(), errhint(), errmsg(), ERROR, ParallelWorkerInfo::error_mqh, GetBackgroundWorkerPid(), i, ParallelContext::known_attached_workers, MyLatch, ParallelContext::nknown_attached_workers, ParallelContext::nworkers_launched, ResetLatch(), shm_mq_get_queue(), shm_mq_get_sender(), WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and ParallelContext::worker.

Referenced by _brin_begin_parallel(), _bt_begin_parallel(), and _gin_begin_parallel().

◆ WaitForParallelWorkersToExit()

static void WaitForParallelWorkersToExit ( ParallelContext *  pcxt )
static

Definition at line 910 of file parallel.c.

911{
912 int i;
913
914 /* Wait until the workers actually die. */
915 for (i = 0; i < pcxt->nworkers_launched; ++i)
916 {
917 BgwHandleStatus status;
918
919 if (pcxt->worker == NULL || pcxt->worker[i].bgwhandle == NULL)
920 continue;
921
923
924 /*
925 * If the postmaster kicked the bucket, we have no chance of cleaning
926 * up safely -- we won't be able to tell when our workers are actually
927 * dead. This doesn't necessitate a PANIC since they will all abort
928 * eventually, but we can't safely continue this session.
929 */
930 if (status == BGWH_POSTMASTER_DIED)
932 (errcode(ERRCODE_ADMIN_SHUTDOWN),
933 errmsg("postmaster exited during a parallel transaction")));
934
935 /* Release memory. */
936 pfree(pcxt->worker[i].bgwhandle);
937 pcxt->worker[i].bgwhandle = NULL;
938 }
939}
BgwHandleStatus WaitForBackgroundWorkerShutdown(BackgroundWorkerHandle *handle)
Definition: bgworker.c:1258
@ BGWH_POSTMASTER_DIED
Definition: bgworker.h:108
#define FATAL
Definition: elog.h:41

References BGWH_POSTMASTER_DIED, ParallelWorkerInfo::bgwhandle, ereport, errcode(), errmsg(), FATAL, i, ParallelContext::nworkers_launched, pfree(), WaitForBackgroundWorkerShutdown(), and ParallelContext::worker.

Referenced by DestroyParallelContext(), and ReinitializeParallelDSM().

◆ WaitForParallelWorkersToFinish()

void WaitForParallelWorkersToFinish ( ParallelContext *  pcxt )

Definition at line 796 of file parallel.c.

797{
798 for (;;)
799 {
800 bool anyone_alive = false;
801 int nfinished = 0;
802 int i;
803
804 /*
805 * This will process any parallel messages that are pending, which may
806 * change the outcome of the loop that follows. It may also throw an
807 * error propagated from a worker.
808 */
810
811 for (i = 0; i < pcxt->nworkers_launched; ++i)
812 {
813 /*
814 * If error_mqh is NULL, then the worker has already exited
815 * cleanly. If we have received a message through error_mqh from
816 * the worker, we know it started up cleanly, and therefore we're
817 * certain to be notified when it exits.
818 */
819 if (pcxt->worker[i].error_mqh == NULL)
820 ++nfinished;
821 else if (pcxt->known_attached_workers[i])
822 {
823 anyone_alive = true;
824 break;
825 }
826 }
827
828 if (!anyone_alive)
829 {
830 /* If all workers are known to have finished, we're done. */
831 if (nfinished >= pcxt->nworkers_launched)
832 {
833 Assert(nfinished == pcxt->nworkers_launched);
834 break;
835 }
836
837 /*
838 * We didn't detect any living workers, but not all workers are
839 * known to have exited cleanly. Either not all workers have
840 * launched yet, or maybe some of them failed to start or
841 * terminated abnormally.
842 */
843 for (i = 0; i < pcxt->nworkers_launched; ++i)
844 {
845 pid_t pid;
846 shm_mq *mq;
847
848 /*
849 * If the worker is BGWH_NOT_YET_STARTED or BGWH_STARTED, we
850 * should just keep waiting. If it is BGWH_STOPPED, then
851 * further investigation is needed.
852 */
853 if (pcxt->worker[i].error_mqh == NULL ||
854 pcxt->worker[i].bgwhandle == NULL ||
856 &pid) != BGWH_STOPPED)
857 continue;
858
859 /*
860 * Check whether the worker ended up stopped without ever
861 * attaching to the error queue. If so, the postmaster was
862 * unable to fork the worker or it exited without initializing
863 * properly. We must throw an error, since the caller may
864 * have been expecting the worker to do some work before
865 * exiting.
866 */
867 mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
868 if (shm_mq_get_sender(mq) == NULL)
870 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
871 errmsg("parallel worker failed to initialize"),
872 errhint("More details may be available in the server log.")));
873
874 /*
875 * The worker is stopped, but is attached to the error queue.
876 * Unless there's a bug somewhere, this will only happen when
877 * the worker writes messages and terminates after the
878 * CHECK_FOR_INTERRUPTS() near the top of this function and
879 * before the call to GetBackgroundWorkerPid(). In that case,
880 * or latch should have been set as well and the right things
881 * will happen on the next pass through the loop.
882 */
883 }
884 }
885
887 WAIT_EVENT_PARALLEL_FINISH);
889 }
890
891 if (pcxt->toc != NULL)
892 {
894
895 fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);
896 if (fps->last_xlog_end > XactLastRecEnd)
898 }
899}
XLogRecPtr XactLastRecEnd
Definition: xlog.c:255

References Assert(), BGWH_STOPPED, ParallelWorkerInfo::bgwhandle, CHECK_FOR_INTERRUPTS, ereport, errcode(), errhint(), errmsg(), ERROR, ParallelWorkerInfo::error_mqh, GetBackgroundWorkerPid(), i, ParallelContext::known_attached_workers, FixedParallelState::last_xlog_end, MyLatch, ParallelContext::nworkers_launched, PARALLEL_KEY_FIXED, ResetLatch(), shm_mq_get_queue(), shm_mq_get_sender(), shm_toc_lookup(), ParallelContext::toc, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, ParallelContext::worker, and XactLastRecEnd.

Referenced by _brin_end_parallel(), _bt_end_parallel(), _gin_end_parallel(), ExecParallelFinish(), parallel_vacuum_process_all_indexes(), and ReinitializeParallelDSM().

Variable Documentation

◆ fn_addr

parallel_worker_main_type fn_addr

Definition at line 139 of file parallel.c.

Referenced by LookupParallelWorkerFunction().

◆ fn_name

const char* fn_name

Definition at line 138 of file parallel.c.

Referenced by LookupParallelWorkerFunction().

◆ InitializingParallelWorker

bool InitializingParallelWorker = false

Definition at line 121 of file parallel.c.

Referenced by check_client_encoding(), check_role(), check_session_authorization(), check_transaction_deferrable(), check_transaction_isolation(), check_transaction_read_only(), InitializeSessionUserId(), and ParallelWorkerMain().

◆

const struct { ... } InternalParallelWorkers[]
Initial value:
=
{
{
"ParallelQueryMain", ParallelQueryMain
},
{
"_bt_parallel_build_main", _bt_parallel_build_main
},
{
"_brin_parallel_build_main", _brin_parallel_build_main
},
{
"_gin_parallel_build_main", _gin_parallel_build_main
},
{
"parallel_vacuum_main", parallel_vacuum_main
}
}
void _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
Definition: brin.c:2863
void ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
Definition: execParallel.c:1430
void _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
Definition: gininsert.c:2076
void _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
Definition: nbtsort.c:1743
void parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)

Referenced by LookupParallelWorkerFunction().

◆ MyFixedParallelState

FixedParallelState* MyFixedParallelState
static

Definition at line 124 of file parallel.c.

Referenced by ParallelWorkerMain(), and ParallelWorkerReportLastRecEnd().

◆ ParallelLeaderPid

pid_t ParallelLeaderPid
static

Definition at line 130 of file parallel.c.

Referenced by ParallelWorkerMain(), and ParallelWorkerShutdown().

◆ ParallelMessagePending

volatile sig_atomic_t ParallelMessagePending = false

Definition at line 118 of file parallel.c.

Referenced by HandleParallelMessageInterrupt(), ProcessInterrupts(), and ProcessParallelMessages().

◆ ParallelWorkerNumber

int ParallelWorkerNumber = -1

Definition at line 115 of file parallel.c.

Referenced by _brin_parallel_build_main(), _bt_parallel_build_main(), _gin_parallel_build_main(), BuildTupleHashTable(), ExecEndAgg(), ExecEndBitmapHeapScan(), ExecEndBitmapIndexScan(), ExecEndIndexOnlyScan(), ExecEndIndexScan(), ExecEndMemoize(), ExecHashInitializeWorker(), ExecParallelGetReceiver(), ExecParallelHashEnsureBatchAccessors(), ExecParallelHashJoinSetUpBatches(), ExecParallelHashRepartitionRest(), ExecParallelReportInstrumentation(), ExecSort(), parallel_vacuum_main(), ParallelQueryMain(), and ParallelWorkerMain().

◆ pcxt_list

dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list)
static

Definition at line 127 of file parallel.c.

Referenced by AtEOSubXact_Parallel(), AtEOXact_Parallel(), CreateParallelContext(), ParallelContextActive(), and ProcessParallelMessages().

AltStyle によって変換されたページ (->オリジナル) /