#include "access/xlogdefs.h"
#include "lib/ilist.h"
#include "postmaster/bgworker.h"
#include "storage/shm_mq.h"
#include "storage/shm_toc.h"

Include dependency graph for parallel.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	ParallelWorkerInfo

struct	ParallelContext

struct	ParallelWorkerContext

Macros
#define	IsParallelWorker() (ParallelWorkerNumber >= 0)

Typedefs
typedef void(*	parallel_worker_main_type) (dsm_segment seg, shm_toc toc)

typedef struct ParallelWorkerInfo	ParallelWorkerInfo

typedef struct ParallelContext	ParallelContext

typedef struct ParallelWorkerContext	ParallelWorkerContext

Functions
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	HandleParallelMessages (void)

void	AtEOXact_Parallel (bool isCommit)

void	AtEOSubXact_Parallel (bool isCommit, SubTransactionId mySubId)

void	ParallelWorkerReportLastRecEnd (XLogRecPtr last_xlog_end)

void	ParallelWorkerMain (Datum main_arg)

Variables
PGDLLIMPORT volatile sig_atomic_t	ParallelMessagePending

PGDLLIMPORT int	ParallelWorkerNumber

PGDLLIMPORT bool	InitializingParallelWorker

Macro Definition Documentation

◆ IsParallelWorker

#define IsParallelWorker ( ) (ParallelWorkerNumber >= 0)

Definition at line 61 of file parallel.h.

Typedef Documentation

◆ parallel_worker_main_type

typedef void(* parallel_worker_main_type) (dsm_segment *seg, shm_toc *toc)

Definition at line 23 of file parallel.h.

◆ ParallelContext

typedef struct ParallelContext ParallelContext

◆ ParallelWorkerContext

typedef struct ParallelWorkerContext ParallelWorkerContext

◆ ParallelWorkerInfo

typedef struct ParallelWorkerInfo ParallelWorkerInfo

Function Documentation

◆ AtEOSubXact_Parallel()

void AtEOSubXact_Parallel	(	bool	isCommit,
		SubTransactionId	mySubId
	)

Definition at line 1244 of file parallel.c.

 {
     while (!dlist_is_empty(&pcxt_list))
     {
         ParallelContext *pcxt;
  
         pcxt = dlist_head_element(ParallelContext, node, &pcxt_list);
         if (pcxt->subid != mySubId)
             break;
         if (isCommit)
             elog(WARNING, "leaked parallel context");
         DestroyParallelContext(pcxt);
     }
 }

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 1263 of file parallel.c.

 {
     while (!dlist_is_empty(&pcxt_list))
     {
         ParallelContext *pcxt;
  
         pcxt = dlist_head_element(ParallelContext, node, &pcxt_list);
         if (isCommit)
             elog(WARNING, "leaked parallel context");
         DestroyParallelContext(pcxt);
     }
 }

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 166 of file parallel.c.

 {
     MemoryContext oldcontext;
     ParallelContext *pcxt;
  
     /* It is unsafe to create a parallel context if not in parallel mode. */
     Assert(IsInParallelMode());
  
     /* Number of workers should be non-negative. */
     Assert(nworkers >= 0);
  
     /* We might be running in a short-lived memory context. */
     oldcontext = MemoryContextSwitchTo(TopTransactionContext);
  
     /* Initialize a new ParallelContext. */
     pcxt = palloc0(sizeof(ParallelContext));
     pcxt->subid = GetCurrentSubTransactionId();
     pcxt->nworkers = nworkers;
     pcxt->nworkers_to_launch = nworkers;
     pcxt->library_name = pstrdup(library_name);
     pcxt->function_name = pstrdup(function_name);
     pcxt->error_context_stack = error_context_stack;
     shm_toc_initialize_estimator(&pcxt->estimator);
     dlist_push_head(&pcxt_list, &pcxt->node);
  
     /* Restore previous memory context. */
     MemoryContextSwitchTo(oldcontext);
  
     return pcxt;
 }

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 _bt_begin_parallel(), ExecInitParallelPlan(), and parallel_vacuum_init().

◆ DestroyParallelContext()

void DestroyParallelContext ( ParallelContext * pcxt )

Definition at line 928 of file parallel.c.

 {
     int         i;
  
     /*
      * Be careful about order of operations here!  We remove the parallel
      * context from the list before we do anything else; otherwise, if an
      * error occurs during a subsequent step, we might try to nuke it again
      * from AtEOXact_Parallel or AtEOSubXact_Parallel.
      */
     dlist_delete(&pcxt->node);
  
     /* Kill each worker in turn, and forget their error queues. */
     if (pcxt->worker != NULL)
     {
         for (i = 0; i < pcxt->nworkers_launched; ++i)
         {
             if (pcxt->worker[i].error_mqh != NULL)
             {
                 TerminateBackgroundWorker(pcxt->worker[i].bgwhandle);
  
                 shm_mq_detach(pcxt->worker[i].error_mqh);
                 pcxt->worker[i].error_mqh = NULL;
             }
         }
     }
  
     /*
      * If we have allocated a shared memory segment, detach it.  This will
      * implicitly detach the error queues, and any other shared memory queues,
      * stored there.
      */
     if (pcxt->seg != NULL)
     {
         dsm_detach(pcxt->seg);
         pcxt->seg = NULL;
     }
  
     /*
      * If this parallel context is actually in backend-private memory rather
      * than shared memory, free that memory instead.
      */
     if (pcxt->private_memory != NULL)
     {
         pfree(pcxt->private_memory);
         pcxt->private_memory = NULL;
     }
  
     /*
      * We can't finish transaction commit or abort until all of the workers
      * have exited.  This means, in particular, that we can't respond to
      * interrupts at this stage.
      */
     HOLD_INTERRUPTS();
     WaitForParallelWorkersToExit(pcxt);
     RESUME_INTERRUPTS();
  
     /* Free the worker array itself. */
     if (pcxt->worker != NULL)
     {
         pfree(pcxt->worker);
         pcxt->worker = NULL;
     }
  
     /* Free memory. */
     pfree(pcxt->library_name);
     pfree(pcxt->function_name);
     pfree(pcxt);
 }

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 _bt_begin_parallel(), _bt_end_parallel(), AtEOSubXact_Parallel(), AtEOXact_Parallel(), ExecParallelCleanup(), and parallel_vacuum_end().

◆ HandleParallelMessageInterrupt()

void HandleParallelMessageInterrupt ( void )

Definition at line 1015 of file parallel.c.

 {
     InterruptPending = true;
     ParallelMessagePending = true;
     SetLatch(MyLatch);
 }

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

Referenced by procsignal_sigusr1_handler().

◆ HandleParallelMessages()

void HandleParallelMessages ( void )

Definition at line 1026 of file parallel.c.

 {
     dlist_iter  iter;
     MemoryContext oldcontext;
  
     static MemoryContext hpm_context = NULL;
  
     /*
      * This is invoked from ProcessInterrupts(), and since some of the
      * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
      * for recursive calls if more signals are received while this runs.  It's
      * unclear that recursive entry would be safe, and it doesn't seem useful
      * even if it is safe, so let's block interrupts until done.
      */
     HOLD_INTERRUPTS();
  
     /*
      * Moreover, CurrentMemoryContext might be pointing almost anywhere.  We
      * don't want to risk leaking data into long-lived contexts, so let's do
      * our work here in a private context that we can reset on each use.
      */
     if (hpm_context == NULL)    /* first time through? */
         hpm_context = AllocSetContextCreate(TopMemoryContext,
                                             "HandleParallelMessages",
                                             ALLOCSET_DEFAULT_SIZES);
     else
         MemoryContextReset(hpm_context);
  
     oldcontext = MemoryContextSwitchTo(hpm_context);
  
     /* OK to process messages.  Reset the flag saying there are more to do. */
     ParallelMessagePending = false;
  
     dlist_foreach(iter, &pcxt_list)
     {
         ParallelContext *pcxt;
         int         i;
  
         pcxt = dlist_container(ParallelContext, node, iter.cur);
         if (pcxt->worker == NULL)
             continue;
  
         for (i = 0; i < pcxt->nworkers_launched; ++i)
         {
             /*
              * Read as many messages as we can from each worker, but stop when
              * either (1) the worker's error queue goes away, which can happen
              * if we receive a Terminate message from the worker; or (2) no
              * more messages can be read from the worker without blocking.
              */
             while (pcxt->worker[i].error_mqh != NULL)
             {
                 shm_mq_result res;
                 Size        nbytes;
                 void       *data;
  
                 res = shm_mq_receive(pcxt->worker[i].error_mqh, &nbytes,
                                      &data, true);
                 if (res == SHM_MQ_WOULD_BLOCK)
                     break;
                 else if (res == SHM_MQ_SUCCESS)
                 {
                     StringInfoData msg;
  
                     initStringInfo(&msg);
                     appendBinaryStringInfo(&msg, data, nbytes);
                     HandleParallelMessage(pcxt, i, &msg);
                     pfree(msg.data);
                 }
                 else
                     ereport(ERROR,
                             (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                              errmsg("lost connection to parallel worker")));
             }
         }
     }
  
     MemoryContextSwitchTo(oldcontext);
  
     /* Might as well clear the context on our way out */
     MemoryContextReset(hpm_context);
  
     RESUME_INTERRUPTS();
 }

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

Referenced by ProcessInterrupts().

◆ InitializeParallelDSM()

void InitializeParallelDSM ( ParallelContext * pcxt )

Definition at line 204 of file parallel.c.

 {
     MemoryContext oldcontext;
     Size        library_len = 0;
     Size        guc_len = 0;
     Size        combocidlen = 0;
     Size        tsnaplen = 0;
     Size        asnaplen = 0;
     Size        tstatelen = 0;
     Size        pendingsyncslen = 0;
     Size        reindexlen = 0;
     Size        relmapperlen = 0;
     Size        uncommittedenumslen = 0;
     Size        clientconninfolen = 0;
     Size        segsize = 0;
     int         i;
     FixedParallelState *fps;
     dsm_handle  session_dsm_handle = DSM_HANDLE_INVALID;
     Snapshot    transaction_snapshot = GetTransactionSnapshot();
     Snapshot    active_snapshot = GetActiveSnapshot();
  
     /* We might be running in a very short-lived memory context. */
     oldcontext = MemoryContextSwitchTo(TopTransactionContext);
  
     /* Allow space to store the fixed-size parallel state. */
     shm_toc_estimate_chunk(&pcxt->estimator, sizeof(FixedParallelState));
     shm_toc_estimate_keys(&pcxt->estimator, 1);
  
     /*
      * Normally, the user will have requested at least one worker process, but
      * if by chance they have not, we can skip a bunch of things here.
      */
     if (pcxt->nworkers > 0)
     {
         /* Get (or create) the per-session DSM segment's handle. */
         session_dsm_handle = GetSessionDsmHandle();
  
         /*
          * If we weren't able to create a per-session DSM segment, then we can
          * continue but we can't safely launch any workers because their
          * record typmods would be incompatible so they couldn't exchange
          * tuples.
          */
         if (session_dsm_handle == DSM_HANDLE_INVALID)
             pcxt->nworkers = 0;
     }
  
     if (pcxt->nworkers > 0)
     {
         /* Estimate space for various kinds of state sharing. */
         library_len = EstimateLibraryStateSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, library_len);
         guc_len = EstimateGUCStateSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, guc_len);
         combocidlen = EstimateComboCIDStateSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, combocidlen);
         if (IsolationUsesXactSnapshot())
         {
             tsnaplen = EstimateSnapshotSpace(transaction_snapshot);
             shm_toc_estimate_chunk(&pcxt->estimator, tsnaplen);
         }
         asnaplen = EstimateSnapshotSpace(active_snapshot);
         shm_toc_estimate_chunk(&pcxt->estimator, asnaplen);
         tstatelen = EstimateTransactionStateSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, tstatelen);
         shm_toc_estimate_chunk(&pcxt->estimator, sizeof(dsm_handle));
         pendingsyncslen = EstimatePendingSyncsSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, pendingsyncslen);
         reindexlen = EstimateReindexStateSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, reindexlen);
         relmapperlen = EstimateRelationMapSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, relmapperlen);
         uncommittedenumslen = EstimateUncommittedEnumsSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, uncommittedenumslen);
         clientconninfolen = EstimateClientConnectionInfoSpace();
         shm_toc_estimate_chunk(&pcxt->estimator, clientconninfolen);
         /* If you add more chunks here, you probably need to add keys. */
         shm_toc_estimate_keys(&pcxt->estimator, 12);
  
         /* Estimate space need for error queues. */
         StaticAssertStmt(BUFFERALIGN(PARALLEL_ERROR_QUEUE_SIZE) ==
                          PARALLEL_ERROR_QUEUE_SIZE,
                          "parallel error queue size not buffer-aligned");
         shm_toc_estimate_chunk(&pcxt->estimator,
                                mul_size(PARALLEL_ERROR_QUEUE_SIZE,
                                         pcxt->nworkers));
         shm_toc_estimate_keys(&pcxt->estimator, 1);
  
         /* Estimate how much we'll need for the entrypoint info. */
         shm_toc_estimate_chunk(&pcxt->estimator, strlen(pcxt->library_name) +
                                strlen(pcxt->function_name) + 2);
         shm_toc_estimate_keys(&pcxt->estimator, 1);
     }
  
     /*
      * Create DSM and initialize with new table of contents.  But if the user
      * didn't request any workers, then don't bother creating a dynamic shared
      * memory segment; instead, just use backend-private memory.
      *
      * Also, if we can't create a dynamic shared memory segment because the
      * maximum number of segments have already been created, then fall back to
      * backend-private memory, and plan not to use any workers.  We hope this
      * won't happen very often, but it's better to abandon the use of
      * parallelism than to fail outright.
      */
     segsize = shm_toc_estimate(&pcxt->estimator);
     if (pcxt->nworkers > 0)
         pcxt->seg = dsm_create(segsize, DSM_CREATE_NULL_IF_MAXSEGMENTS);
     if (pcxt->seg != NULL)
         pcxt->toc = shm_toc_create(PARALLEL_MAGIC,
                                    dsm_segment_address(pcxt->seg),
                                    segsize);
     else
     {
         pcxt->nworkers = 0;
         pcxt->private_memory = MemoryContextAlloc(TopMemoryContext, segsize);
         pcxt->toc = shm_toc_create(PARALLEL_MAGIC, pcxt->private_memory,
                                    segsize);
     }
  
     /* Initialize fixed-size state in shared memory. */
     fps = (FixedParallelState *)
         shm_toc_allocate(pcxt->toc, sizeof(FixedParallelState));
     fps->database_id = MyDatabaseId;
     fps->authenticated_user_id = GetAuthenticatedUserId();
     fps->outer_user_id = GetCurrentRoleId();
     fps->is_superuser = current_role_is_superuser;
     GetUserIdAndSecContext(&fps->current_user_id, &fps->sec_context);
     GetTempNamespaceState(&fps->temp_namespace_id,
                           &fps->temp_toast_namespace_id);
     fps->parallel_leader_pgproc = MyProc;
     fps->parallel_leader_pid = MyProcPid;
     fps->parallel_leader_backend_id = MyBackendId;
     fps->xact_ts = GetCurrentTransactionStartTimestamp();
     fps->stmt_ts = GetCurrentStatementStartTimestamp();
     fps->serializable_xact_handle = ShareSerializableXact();
     SpinLockInit(&fps->mutex);
     fps->last_xlog_end = 0;
     shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps);
  
     /* We can skip the rest of this if we're not budgeting for any workers. */
     if (pcxt->nworkers > 0)
     {
         char       *libraryspace;
         char       *gucspace;
         char       *combocidspace;
         char       *tsnapspace;
         char       *asnapspace;
         char       *tstatespace;
         char       *pendingsyncsspace;
         char       *reindexspace;
         char       *relmapperspace;
         char       *error_queue_space;
         char       *session_dsm_handle_space;
         char       *entrypointstate;
         char       *uncommittedenumsspace;
         char       *clientconninfospace;
         Size        lnamelen;
  
         /* Serialize shared libraries we have loaded. */
         libraryspace = shm_toc_allocate(pcxt->toc, library_len);
         SerializeLibraryState(library_len, libraryspace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_LIBRARY, libraryspace);
  
         /* Serialize GUC settings. */
         gucspace = shm_toc_allocate(pcxt->toc, guc_len);
         SerializeGUCState(guc_len, gucspace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_GUC, gucspace);
  
         /* Serialize combo CID state. */
         combocidspace = shm_toc_allocate(pcxt->toc, combocidlen);
         SerializeComboCIDState(combocidlen, combocidspace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_COMBO_CID, combocidspace);
  
         /*
          * Serialize the transaction snapshot if the transaction isolation
          * level uses a transaction snapshot.
          */
         if (IsolationUsesXactSnapshot())
         {
             tsnapspace = shm_toc_allocate(pcxt->toc, tsnaplen);
             SerializeSnapshot(transaction_snapshot, tsnapspace);
             shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT,
                            tsnapspace);
         }
  
         /* Serialize the active snapshot. */
         asnapspace = shm_toc_allocate(pcxt->toc, asnaplen);
         SerializeSnapshot(active_snapshot, asnapspace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, asnapspace);
  
         /* Provide the handle for per-session segment. */
         session_dsm_handle_space = shm_toc_allocate(pcxt->toc,
                                                     sizeof(dsm_handle));
         *(dsm_handle *) session_dsm_handle_space = session_dsm_handle;
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_SESSION_DSM,
                        session_dsm_handle_space);
  
         /* Serialize transaction state. */
         tstatespace = shm_toc_allocate(pcxt->toc, tstatelen);
         SerializeTransactionState(tstatelen, tstatespace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_STATE, tstatespace);
  
         /* Serialize pending syncs. */
         pendingsyncsspace = shm_toc_allocate(pcxt->toc, pendingsyncslen);
         SerializePendingSyncs(pendingsyncslen, pendingsyncsspace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_PENDING_SYNCS,
                        pendingsyncsspace);
  
         /* Serialize reindex state. */
         reindexspace = shm_toc_allocate(pcxt->toc, reindexlen);
         SerializeReindexState(reindexlen, reindexspace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_REINDEX_STATE, reindexspace);
  
         /* Serialize relmapper state. */
         relmapperspace = shm_toc_allocate(pcxt->toc, relmapperlen);
         SerializeRelationMap(relmapperlen, relmapperspace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_RELMAPPER_STATE,
                        relmapperspace);
  
         /* Serialize uncommitted enum state. */
         uncommittedenumsspace = shm_toc_allocate(pcxt->toc,
                                                  uncommittedenumslen);
         SerializeUncommittedEnums(uncommittedenumsspace, uncommittedenumslen);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_UNCOMMITTEDENUMS,
                        uncommittedenumsspace);
  
         /* Serialize our ClientConnectionInfo. */
         clientconninfospace = shm_toc_allocate(pcxt->toc, clientconninfolen);
         SerializeClientConnectionInfo(clientconninfolen, clientconninfospace);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_CLIENTCONNINFO,
                        clientconninfospace);
  
         /* Allocate space for worker information. */
         pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * pcxt->nworkers);
  
         /*
          * Establish error queues in dynamic shared memory.
          *
          * These queues should be used only for transmitting ErrorResponse,
          * NoticeResponse, and NotifyResponse protocol messages.  Tuple data
          * should be transmitted via separate (possibly larger?) queues.
          */
         error_queue_space =
             shm_toc_allocate(pcxt->toc,
                              mul_size(PARALLEL_ERROR_QUEUE_SIZE,
                                       pcxt->nworkers));
         for (i = 0; i < pcxt->nworkers; ++i)
         {
             char       *start;
             shm_mq     *mq;
  
             start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
             mq = shm_mq_create(start, PARALLEL_ERROR_QUEUE_SIZE);
             shm_mq_set_receiver(mq, MyProc);
             pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
         }
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, error_queue_space);
  
         /*
          * Serialize entrypoint information.  It's unsafe to pass function
          * pointers across processes, as the function pointer may be different
          * in each process in EXEC_BACKEND builds, so we always pass library
          * and function name.  (We use library name "postgres" for functions
          * in the core backend.)
          */
         lnamelen = strlen(pcxt->library_name);
         entrypointstate = shm_toc_allocate(pcxt->toc, lnamelen +
                                            strlen(pcxt->function_name) + 2);
         strcpy(entrypointstate, pcxt->library_name);
         strcpy(entrypointstate + lnamelen + 1, pcxt->function_name);
         shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENTRYPOINT, entrypointstate);
     }
  
     /* Restore previous memory context. */
     MemoryContextSwitchTo(oldcontext);
 }

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

◆ LaunchParallelWorkers()

void LaunchParallelWorkers ( ParallelContext * pcxt )

Definition at line 551 of file parallel.c.

 {
     MemoryContext oldcontext;
     BackgroundWorker worker;
     int         i;
     bool        any_registrations_failed = false;
  
     /* Skip this if we have no workers. */
     if (pcxt->nworkers == 0 || pcxt->nworkers_to_launch == 0)
         return;
  
     /* We need to be a lock group leader. */
     BecomeLockGroupLeader();
  
     /* If we do have workers, we'd better have a DSM segment. */
     Assert(pcxt->seg != NULL);
  
     /* We might be running in a short-lived memory context. */
     oldcontext = MemoryContextSwitchTo(TopTransactionContext);
  
     /* Configure a worker. */
     memset(&worker, 0, sizeof(worker));
     snprintf(worker.bgw_name, BGW_MAXLEN, "parallel worker for PID %d",
              MyProcPid);
     snprintf(worker.bgw_type, BGW_MAXLEN, "parallel worker");
     worker.bgw_flags =
         BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION
         | BGWORKER_CLASS_PARALLEL;
     worker.bgw_start_time = BgWorkerStart_ConsistentState;
     worker.bgw_restart_time = BGW_NEVER_RESTART;
     sprintf(worker.bgw_library_name, "postgres");
     sprintf(worker.bgw_function_name, "ParallelWorkerMain");
     worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(pcxt->seg));
     worker.bgw_notify_pid = MyProcPid;
  
     /*
      * Start workers.
      *
      * The caller must be able to tolerate ending up with fewer workers than
      * expected, so there is no need to throw an error here if registration
      * fails.  It wouldn't help much anyway, because registering the worker in
      * no way guarantees that it will start up and initialize successfully.
      */
     for (i = 0; i < pcxt->nworkers_to_launch; ++i)
     {
         memcpy(worker.bgw_extra, &i, sizeof(int));
         if (!any_registrations_failed &&
             RegisterDynamicBackgroundWorker(&worker,
                                             &pcxt->worker[i].bgwhandle))
         {
             shm_mq_set_handle(pcxt->worker[i].error_mqh,
                               pcxt->worker[i].bgwhandle);
             pcxt->nworkers_launched++;
         }
         else
         {
             /*
              * If we weren't able to register the worker, then we've bumped up
              * against the max_worker_processes limit, and future
              * registrations will probably fail too, so arrange to skip them.
              * But we still have to execute this code for the remaining slots
              * to make sure that we forget about the error queues we budgeted
              * for those workers.  Otherwise, we'll wait for them to start,
              * but they never will.
              */
             any_registrations_failed = true;
             pcxt->worker[i].bgwhandle = NULL;
             shm_mq_detach(pcxt->worker[i].error_mqh);
             pcxt->worker[i].error_mqh = NULL;
         }
     }
  
     /*
      * Now that nworkers_launched has taken its final value, we can initialize
      * known_attached_workers.
      */
     if (pcxt->nworkers_launched > 0)
     {
         pcxt->known_attached_workers =
             palloc0(sizeof(bool) * pcxt->nworkers_launched);
         pcxt->nknown_attached_workers = 0;
     }
  
     /* Restore previous memory context. */
     MemoryContextSwitchTo(oldcontext);
 }

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

◆ ParallelContextActive()

bool ParallelContextActive ( void )

Definition at line 1002 of file parallel.c.

 {
     return !dlist_is_empty(&pcxt_list);
 }

References dlist_is_empty(), and pcxt_list.

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

◆ ParallelWorkerMain()

void ParallelWorkerMain ( Datum main_arg )

Definition at line 1280 of file parallel.c.

 {
     dsm_segment *seg;
     shm_toc    *toc;
     FixedParallelState *fps;
     char       *error_queue_space;
     shm_mq     *mq;
     shm_mq_handle *mqh;
     char       *libraryspace;
     char       *entrypointstate;
     char       *library_name;
     char       *function_name;
     parallel_worker_main_type entrypt;
     char       *gucspace;
     char       *combocidspace;
     char       *tsnapspace;
     char       *asnapspace;
     char       *tstatespace;
     char       *pendingsyncsspace;
     char       *reindexspace;
     char       *relmapperspace;
     char       *uncommittedenumsspace;
     char       *clientconninfospace;
     StringInfoData msgbuf;
     char       *session_dsm_handle_space;
     Snapshot    tsnapshot;
     Snapshot    asnapshot;
  
     /* Set flag to indicate that we're initializing a parallel worker. */
     InitializingParallelWorker = true;
  
     /* Establish signal handlers. */
     pqsignal(SIGTERM, die);
     BackgroundWorkerUnblockSignals();
  
     /* Determine and set our parallel worker number. */
     Assert(ParallelWorkerNumber == -1);
     memcpy(&ParallelWorkerNumber, MyBgworkerEntry->bgw_extra, sizeof(int));
  
     /* Set up a memory context to work in, just for cleanliness. */
     CurrentMemoryContext = AllocSetContextCreate(TopMemoryContext,
                                                  "Parallel worker",
                                                  ALLOCSET_DEFAULT_SIZES);
  
     /*
      * Attach to the dynamic shared memory segment for the parallel query, and
      * find its table of contents.
      *
      * Note: at this point, we have not created any ResourceOwner in this
      * process.  This will result in our DSM mapping surviving until process
      * exit, which is fine.  If there were a ResourceOwner, it would acquire
      * ownership of the mapping, but we have no need for that.
      */
     seg = dsm_attach(DatumGetUInt32(main_arg));
     if (seg == NULL)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("could not map dynamic shared memory segment")));
     toc = shm_toc_attach(PARALLEL_MAGIC, dsm_segment_address(seg));
     if (toc == NULL)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("invalid magic number in dynamic shared memory segment")));
  
     /* Look up fixed parallel state. */
     fps = shm_toc_lookup(toc, PARALLEL_KEY_FIXED, false);
     MyFixedParallelState = fps;
  
     /* Arrange to signal the leader if we exit. */
     ParallelLeaderPid = fps->parallel_leader_pid;
     ParallelLeaderBackendId = fps->parallel_leader_backend_id;
     before_shmem_exit(ParallelWorkerShutdown, PointerGetDatum(seg));
  
     /*
      * Now we can find and attach to the error queue provided for us.  That's
      * good, because until we do that, any errors that happen here will not be
      * reported back to the process that requested that this worker be
      * launched.
      */
     error_queue_space = shm_toc_lookup(toc, PARALLEL_KEY_ERROR_QUEUE, false);
     mq = (shm_mq *) (error_queue_space +
                      ParallelWorkerNumber * PARALLEL_ERROR_QUEUE_SIZE);
     shm_mq_set_sender(mq, MyProc);
     mqh = shm_mq_attach(mq, seg, NULL);
     pq_redirect_to_shm_mq(seg, mqh);
     pq_set_parallel_leader(fps->parallel_leader_pid,
                            fps->parallel_leader_backend_id);
  
     /*
      * Send a BackendKeyData message to the process that initiated parallelism
      * so that it has access to our PID before it receives any other messages
      * from us.  Our cancel key is sent, too, since that's the way the
      * protocol message is defined, but it won't actually be used for anything
      * in this case.
      */
     pq_beginmessage(&msgbuf, PqMsg_BackendKeyData);
     pq_sendint32(&msgbuf, (int32) MyProcPid);
     pq_sendint32(&msgbuf, (int32) MyCancelKey);
     pq_endmessage(&msgbuf);
  
     /*
      * Hooray! Primary initialization is complete.  Now, we need to set up our
      * backend-local state to match the original backend.
      */
  
     /*
      * Join locking group.  We must do this before anything that could try to
      * acquire a heavyweight lock, because any heavyweight locks acquired to
      * this point could block either directly against the parallel group
      * leader or against some process which in turn waits for a lock that
      * conflicts with the parallel group leader, causing an undetected
      * deadlock.  (If we can't join the lock group, the leader has gone away,
      * so just exit quietly.)
      */
     if (!BecomeLockGroupMember(fps->parallel_leader_pgproc,
                                fps->parallel_leader_pid))
         return;
  
     /*
      * Restore transaction and statement start-time timestamps.  This must
      * happen before anything that would start a transaction, else asserts in
      * xact.c will fire.
      */
     SetParallelStartTimestamps(fps->xact_ts, fps->stmt_ts);
  
     /*
      * Identify the entry point to be called.  In theory this could result in
      * loading an additional library, though most likely the entry point is in
      * the core backend or in a library we just loaded.
      */
     entrypointstate = shm_toc_lookup(toc, PARALLEL_KEY_ENTRYPOINT, false);
     library_name = entrypointstate;
     function_name = entrypointstate + strlen(library_name) + 1;
  
     entrypt = LookupParallelWorkerFunction(library_name, function_name);
  
     /* Restore database connection. */
     BackgroundWorkerInitializeConnectionByOid(fps->database_id,
                                               fps->authenticated_user_id,
                                               0);
  
     /*
      * Set the client encoding to the database encoding, since that is what
      * the leader will expect.
      */
     SetClientEncoding(GetDatabaseEncoding());
  
     /*
      * Load libraries that were loaded by original backend.  We want to do
      * this before restoring GUCs, because the libraries might define custom
      * variables.
      */
     libraryspace = shm_toc_lookup(toc, PARALLEL_KEY_LIBRARY, false);
     StartTransactionCommand();
     RestoreLibraryState(libraryspace);
  
     /* Restore GUC values from launching backend. */
     gucspace = shm_toc_lookup(toc, PARALLEL_KEY_GUC, false);
     RestoreGUCState(gucspace);
     CommitTransactionCommand();
  
     /* Crank up a transaction state appropriate to a parallel worker. */
     tstatespace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_STATE, false);
     StartParallelWorkerTransaction(tstatespace);
  
     /* Restore combo CID state. */
     combocidspace = shm_toc_lookup(toc, PARALLEL_KEY_COMBO_CID, false);
     RestoreComboCIDState(combocidspace);
  
     /* Attach to the per-session DSM segment and contained objects. */
     session_dsm_handle_space =
         shm_toc_lookup(toc, PARALLEL_KEY_SESSION_DSM, false);
     AttachSession(*(dsm_handle *) session_dsm_handle_space);
  
     /*
      * If the transaction isolation level is REPEATABLE READ or SERIALIZABLE,
      * the leader has serialized the transaction snapshot and we must restore
      * it. At lower isolation levels, there is no transaction-lifetime
      * snapshot, but we need TransactionXmin to get set to a value which is
      * less than or equal to the xmin of every snapshot that will be used by
      * this worker. The easiest way to accomplish that is to install the
      * active snapshot as the transaction snapshot. Code running in this
      * parallel worker might take new snapshots via GetTransactionSnapshot()
      * or GetLatestSnapshot(), but it shouldn't have any way of acquiring a
      * snapshot older than the active snapshot.
      */
     asnapspace = shm_toc_lookup(toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, false);
     tsnapspace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT, true);
     asnapshot = RestoreSnapshot(asnapspace);
     tsnapshot = tsnapspace ? RestoreSnapshot(tsnapspace) : asnapshot;
     RestoreTransactionSnapshot(tsnapshot,
                                fps->parallel_leader_pgproc);
     PushActiveSnapshot(asnapshot);
  
     /*
      * We've changed which tuples we can see, and must therefore invalidate
      * system caches.
      */
     InvalidateSystemCaches();
  
     /*
      * Restore current role id.  Skip verifying whether session user is
      * allowed to become this role and blindly restore the leader's state for
      * current role.
      */
     SetCurrentRoleId(fps->outer_user_id, fps->is_superuser);
  
     /* Restore user ID and security context. */
     SetUserIdAndSecContext(fps->current_user_id, fps->sec_context);
  
     /* Restore temp-namespace state to ensure search path matches leader's. */
     SetTempNamespaceState(fps->temp_namespace_id,
                           fps->temp_toast_namespace_id);
  
     /* Restore pending syncs. */
     pendingsyncsspace = shm_toc_lookup(toc, PARALLEL_KEY_PENDING_SYNCS,
                                        false);
     RestorePendingSyncs(pendingsyncsspace);
  
     /* Restore reindex state. */
     reindexspace = shm_toc_lookup(toc, PARALLEL_KEY_REINDEX_STATE, false);
     RestoreReindexState(reindexspace);
  
     /* Restore relmapper state. */
     relmapperspace = shm_toc_lookup(toc, PARALLEL_KEY_RELMAPPER_STATE, false);
     RestoreRelationMap(relmapperspace);
  
     /* Restore uncommitted enums. */
     uncommittedenumsspace = shm_toc_lookup(toc, PARALLEL_KEY_UNCOMMITTEDENUMS,
                                            false);
     RestoreUncommittedEnums(uncommittedenumsspace);
  
     /* Restore the ClientConnectionInfo. */
     clientconninfospace = shm_toc_lookup(toc, PARALLEL_KEY_CLIENTCONNINFO,
                                          false);
     RestoreClientConnectionInfo(clientconninfospace);
  
     /*
      * Initialize SystemUser now that MyClientConnectionInfo is restored. Also
      * ensure that auth_method is actually valid, aka authn_id is not NULL.
      */
     if (MyClientConnectionInfo.authn_id)
         InitializeSystemUser(MyClientConnectionInfo.authn_id,
                              hba_authname(MyClientConnectionInfo.auth_method));
  
     /* Attach to the leader's serializable transaction, if SERIALIZABLE. */
     AttachSerializableXact(fps->serializable_xact_handle);
  
     /*
      * We've initialized all of our state now; nothing should change
      * hereafter.
      */
     InitializingParallelWorker = false;
     EnterParallelMode();
  
     /*
      * Time to do the real work: invoke the caller-supplied code.
      */
     entrypt(seg, toc);
  
     /* Must exit parallel mode to pop active snapshot. */
     ExitParallelMode();
  
     /* Must pop active snapshot so snapmgr.c doesn't complain. */
     PopActiveSnapshot();
  
     /* Shut down the parallel-worker transaction. */
     EndParallelWorkerTransaction();
  
     /* Detach from the per-session DSM segment. */
     DetachSession();
  
     /* Report success. */
     pq_putmessage(PqMsg_Terminate, NULL, 0);
 }

◆ ParallelWorkerReportLastRecEnd()

void ParallelWorkerReportLastRecEnd ( XLogRecPtr last_xlog_end )

Definition at line 1561 of file parallel.c.

 {
     FixedParallelState *fps = MyFixedParallelState;
  
     Assert(fps != NULL);
     SpinLockAcquire(&fps->mutex);
     if (fps->last_xlog_end < last_xlog_end)
         fps->last_xlog_end = last_xlog_end;
     SpinLockRelease(&fps->mutex);
 }

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

Referenced by CommitTransaction().

◆ ReinitializeParallelDSM()

void ReinitializeParallelDSM ( ParallelContext * pcxt )

Definition at line 487 of file parallel.c.

 {
     FixedParallelState *fps;
  
     /* Wait for any old workers to exit. */
     if (pcxt->nworkers_launched > 0)
     {
         WaitForParallelWorkersToFinish(pcxt);
         WaitForParallelWorkersToExit(pcxt);
         pcxt->nworkers_launched = 0;
         if (pcxt->known_attached_workers)
         {
             pfree(pcxt->known_attached_workers);
             pcxt->known_attached_workers = NULL;
             pcxt->nknown_attached_workers = 0;
         }
     }
  
     /* Reset a few bits of fixed parallel state to a clean state. */
     fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);
     fps->last_xlog_end = 0;
  
     /* Recreate error queues (if they exist). */
     if (pcxt->nworkers > 0)
     {
         char       *error_queue_space;
         int         i;
  
         error_queue_space =
             shm_toc_lookup(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, false);
         for (i = 0; i < pcxt->nworkers; ++i)
         {
             char       *start;
             shm_mq     *mq;
  
             start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
             mq = shm_mq_create(start, PARALLEL_ERROR_QUEUE_SIZE);
             shm_mq_set_receiver(mq, MyProc);
             pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
         }
     }
 }

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(), 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 537 of file parallel.c.

 {
     /*
      * The number of workers that need to be launched must be less than the
      * number of workers with which the parallel context is initialized.
      */
     Assert(pcxt->nworkers >= nworkers_to_launch);
     pcxt->nworkers_to_launch = nworkers_to_launch;
 }

References Assert(), ParallelContext::nworkers, and ParallelContext::nworkers_to_launch.

Referenced by parallel_vacuum_process_all_indexes().

◆ WaitForParallelWorkersToAttach()

void WaitForParallelWorkersToAttach ( ParallelContext * pcxt )

Definition at line 671 of file parallel.c.

 {
     int         i;
  
     /* Skip this if we have no launched workers. */
     if (pcxt->nworkers_launched == 0)
         return;
  
     for (;;)
     {
         /*
          * This will process any parallel messages that are pending and it may
          * also throw an error propagated from a worker.
          */
         CHECK_FOR_INTERRUPTS();
  
         for (i = 0; i < pcxt->nworkers_launched; ++i)
         {
             BgwHandleStatus status;
             shm_mq     *mq;
             int         rc;
             pid_t       pid;
  
             if (pcxt->known_attached_workers[i])
                 continue;
  
             /*
              * If error_mqh is NULL, then the worker has already exited
              * cleanly.
              */
             if (pcxt->worker[i].error_mqh == NULL)
             {
                 pcxt->known_attached_workers[i] = true;
                 ++pcxt->nknown_attached_workers;
                 continue;
             }
  
             status = GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle, &pid);
             if (status == BGWH_STARTED)
             {
                 /* Has the worker attached to the error queue? */
                 mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
                 if (shm_mq_get_sender(mq) != NULL)
                 {
                     /* Yes, so it is known to be attached. */
                     pcxt->known_attached_workers[i] = true;
                     ++pcxt->nknown_attached_workers;
                 }
             }
             else if (status == BGWH_STOPPED)
             {
                 /*
                  * If the worker stopped without attaching to the error queue,
                  * throw an error.
                  */
                 mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
                 if (shm_mq_get_sender(mq) == NULL)
                     ereport(ERROR,
                             (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                              errmsg("parallel worker failed to initialize"),
                              errhint("More details may be available in the server log.")));
  
                 pcxt->known_attached_workers[i] = true;
                 ++pcxt->nknown_attached_workers;
             }
             else
             {
                 /*
                  * Worker not yet started, so we must wait.  The postmaster
                  * will notify us if the worker's state changes.  Our latch
                  * might also get set for some other reason, but if so we'll
                  * just end up waiting for the same worker again.
                  */
                 rc = WaitLatch(MyLatch,
                                WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
                                -1, WAIT_EVENT_BGWORKER_STARTUP);
  
                 if (rc & WL_LATCH_SET)
                     ResetLatch(MyLatch);
             }
         }
  
         /* If all workers are known to have started, we're done. */
         if (pcxt->nknown_attached_workers >= pcxt->nworkers_launched)
         {
             Assert(pcxt->nknown_attached_workers == pcxt->nworkers_launched);
             break;
         }
     }
 }

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

◆ WaitForParallelWorkersToFinish()

void WaitForParallelWorkersToFinish ( ParallelContext * pcxt )

Definition at line 774 of file parallel.c.

 {
     for (;;)
     {
         bool        anyone_alive = false;
         int         nfinished = 0;
         int         i;
  
         /*
          * This will process any parallel messages that are pending, which may
          * change the outcome of the loop that follows.  It may also throw an
          * error propagated from a worker.
          */
         CHECK_FOR_INTERRUPTS();
  
         for (i = 0; i < pcxt->nworkers_launched; ++i)
         {
             /*
              * If error_mqh is NULL, then the worker has already exited
              * cleanly.  If we have received a message through error_mqh from
              * the worker, we know it started up cleanly, and therefore we're
              * certain to be notified when it exits.
              */
             if (pcxt->worker[i].error_mqh == NULL)
                 ++nfinished;
             else if (pcxt->known_attached_workers[i])
             {
                 anyone_alive = true;
                 break;
             }
         }
  
         if (!anyone_alive)
         {
             /* If all workers are known to have finished, we're done. */
             if (nfinished >= pcxt->nworkers_launched)
             {
                 Assert(nfinished == pcxt->nworkers_launched);
                 break;
             }
  
             /*
              * We didn't detect any living workers, but not all workers are
              * known to have exited cleanly.  Either not all workers have
              * launched yet, or maybe some of them failed to start or
              * terminated abnormally.
              */
             for (i = 0; i < pcxt->nworkers_launched; ++i)
             {
                 pid_t       pid;
                 shm_mq     *mq;
  
                 /*
                  * If the worker is BGWH_NOT_YET_STARTED or BGWH_STARTED, we
                  * should just keep waiting.  If it is BGWH_STOPPED, then
                  * further investigation is needed.
                  */
                 if (pcxt->worker[i].error_mqh == NULL ||
                     pcxt->worker[i].bgwhandle == NULL ||
                     GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle,
                                            &pid) != BGWH_STOPPED)
                     continue;
  
                 /*
                  * Check whether the worker ended up stopped without ever
                  * attaching to the error queue.  If so, the postmaster was
                  * unable to fork the worker or it exited without initializing
                  * properly.  We must throw an error, since the caller may
                  * have been expecting the worker to do some work before
                  * exiting.
                  */
                 mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
                 if (shm_mq_get_sender(mq) == NULL)
                     ereport(ERROR,
                             (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                              errmsg("parallel worker failed to initialize"),
                              errhint("More details may be available in the server log.")));
  
                 /*
                  * The worker is stopped, but is attached to the error queue.
                  * Unless there's a bug somewhere, this will only happen when
                  * the worker writes messages and terminates after the
                  * CHECK_FOR_INTERRUPTS() near the top of this function and
                  * before the call to GetBackgroundWorkerPid().  In that case,
                  * or latch should have been set as well and the right things
                  * will happen on the next pass through the loop.
                  */
             }
         }
  
         (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, -1,
                          WAIT_EVENT_PARALLEL_FINISH);
         ResetLatch(MyLatch);
     }
  
     if (pcxt->toc != NULL)
     {
         FixedParallelState *fps;
  
         fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);
         if (fps->last_xlog_end > XactLastRecEnd)
             XactLastRecEnd = fps->last_xlog_end;
     }
 }

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 _bt_end_parallel(), ExecParallelFinish(), parallel_vacuum_process_all_indexes(), and ReinitializeParallelDSM().

Variable Documentation

◆ InitializingParallelWorker

PGDLLIMPORT bool InitializingParallelWorker

extern

Definition at line 120 of file parallel.c.

Referenced by assign_client_encoding(), check_role(), check_transaction_read_only(), and ParallelWorkerMain().

◆ ParallelMessagePending

PGDLLIMPORT volatile sig_atomic_t ParallelMessagePending

extern

Definition at line 117 of file parallel.c.

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

◆ ParallelWorkerNumber

PGDLLIMPORT int ParallelWorkerNumber

extern

Definition at line 114 of file parallel.c.

Referenced by _bt_parallel_build_main(), BuildTupleHashTableExt(), ExecEndAgg(), ExecEndMemoize(), ExecHashInitializeWorker(), ExecParallelGetReceiver(), ExecParallelHashEnsureBatchAccessors(), ExecParallelHashJoinSetUpBatches(), ExecParallelHashRepartitionRest(), ExecParallelReportInstrumentation(), ExecSort(), parallel_vacuum_main(), ParallelQueryMain(), and ParallelWorkerMain().

Data Structures

Macros

Typedefs

Functions

Variables

Macro Definition Documentation

◆ IsParallelWorker

Typedef Documentation

◆ parallel_worker_main_type

◆ ParallelContext

◆ ParallelWorkerContext

◆ ParallelWorkerInfo

Function Documentation

◆ AtEOSubXact_Parallel()

◆ AtEOXact_Parallel()

◆ CreateParallelContext()

◆ DestroyParallelContext()

◆ HandleParallelMessageInterrupt()

◆ HandleParallelMessages()

◆ InitializeParallelDSM()

◆ LaunchParallelWorkers()

◆ ParallelContextActive()

◆ ParallelWorkerMain()

◆ ParallelWorkerReportLastRecEnd()

◆ ReinitializeParallelDSM()

◆ ReinitializeParallelWorkers()

◆ WaitForParallelWorkersToAttach()

◆ WaitForParallelWorkersToFinish()

Variable Documentation

◆ InitializingParallelWorker

◆ ParallelMessagePending

◆ ParallelWorkerNumber