/*-------------------------------------------------------------------------

 *

 * parallel.c

 *    Infrastructure for launching parallel workers

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 * IDENTIFICATION

 *    src/backend/access/transam/parallel.c

 *

 *-------------------------------------------------------------------------

 */

#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/proc.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 "utils/wait_event.h"

/*

 * We don't want to waste a lot of memory on an error queue which, most of

 * the time, will process only a handful of small messages.  However, it is

 * desirable to make it large enough that a typical ErrorResponse can be sent

 * without blocking.  That way, a worker that errors out can write the whole

 * message into the queue and terminate without waiting for the user backend.

 */

#define PARALLEL_ERROR_QUEUE_SIZE     16384

/* Magic number for parallel context TOC. */

#define PARALLEL_MAGIC            0x50477c7c

/*

 * Magic numbers for per-context parallel state sharing.  Higher-level code

 * should use smaller values, leaving these very large ones for use by this

 * module.

 */

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

/* Fixed-size parallel state. */

typedef struct FixedParallelState

{

  /* Fixed-size state that workers must restore. */

  Oid     database_id;

  Oid     authenticated_user_id;

  Oid     session_user_id;

  Oid     outer_user_id;

  Oid     current_user_id;

  Oid     temp_namespace_id;

  Oid     temp_toast_namespace_id;

  int     sec_context;

  bool    session_user_is_superuser;

  bool    role_is_superuser;

  PGPROC     *parallel_leader_pgproc;

  pid_t   parallel_leader_pid;

  ProcNumber  parallel_leader_proc_number;

  TimestampTz xact_ts;

  TimestampTz stmt_ts;

  SerializableXactHandle serializable_xact_handle;

  /* Maximum XactLastRecEnd of any worker. */

  pg_atomic_uint64 last_xlog_end;

} FixedParallelState;

/*

 * Our parallel worker number.  We initialize this to -1, meaning that we are

 * not a parallel worker.  In parallel workers, it will be set to a value >= 0

 * and < the number of workers before any user code is invoked; each parallel

 * worker will get a different parallel worker number.

 */

int     ParallelWorkerNumber = -1;

/* Is there a parallel message pending which we need to receive? */

volatile sig_atomic_t ParallelMessagePending = false;

/* Are we initializing a parallel worker? */

bool    InitializingParallelWorker = false;

/* Pointer to our fixed parallel state. */

static FixedParallelState *MyFixedParallelState;

/* List of active parallel contexts. */

static dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list);

/* Backend-local copy of data from FixedParallelState. */

static pid_t ParallelLeaderPid;

/*

 * List of internal parallel worker entry points.  We need this for

 * reasons explained in LookupParallelWorkerFunction(), below.

 */

static const struct

{

  const char *fn_name;

  parallel_worker_main_type fn_addr;

}     InternalParallelWorkers[] =

{

  {

    "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

  }

};

/* Private 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);

/*

 * Establish a new parallel context.  This should be done after entering

 * parallel mode, and (unless there is an error) the context should be

 * destroyed before exiting the current subtransaction.

 */

ParallelContext *

CreateParallelContext(const char *library_name, const char *function_name,

            int nworkers)

{

  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_object(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;

}

/*

 * Establish the dynamic shared memory segment for a parallel context and

 * copy state and other bookkeeping information that will be needed by

 * parallel workers into it.

 */

void

InitializeParallelDSM(ParallelContext *pcxt)

{

  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);

  /*

   * If we manage to reach here while non-interruptible, it's unsafe to

   * launch any workers: we would fail to process interrupts sent by them.

   * We can deal with that edge case by pretending no workers were

   * requested.

   */

  if (!INTERRUPTS_CAN_BE_PROCESSED())

    pcxt->nworkers = 0;

  /*

   * 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)

  {

    StaticAssertDecl(BUFFERALIGN(PARALLEL_ERROR_QUEUE_SIZE) ==

             PARALLEL_ERROR_QUEUE_SIZE,

             "parallel error queue size not buffer-aligned");

    /* 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. */

    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->session_user_id = GetSessionUserId();

  fps->outer_user_id = GetCurrentRoleId();

  GetUserIdAndSecContext(&fps->current_user_id, &fps->sec_context);

  fps->session_user_is_superuser = GetSessionUserIsSuperuser();

  fps->role_is_superuser = current_role_is_superuser;

  GetTempNamespaceState(&fps->temp_namespace_id,

              &fps->temp_toast_namespace_id);

  fps->parallel_leader_pgproc = MyProc;

  fps->parallel_leader_pid = MyProcPid;

  fps->parallel_leader_proc_number = MyProcNumber;

  fps->xact_ts = GetCurrentTransactionStartTimestamp();

  fps->stmt_ts = GetCurrentStatementStartTimestamp();

  fps->serializable_xact_handle = ShareSerializableXact();

  pg_atomic_init_u64(&fps->last_xlog_end, InvalidXLogRecPtr);

  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_array(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);

  }

  /* Update nworkers_to_launch, in case we changed nworkers above. */

  pcxt->nworkers_to_launch = pcxt->nworkers;

  /* Restore previous memory context. */

  MemoryContextSwitchTo(oldcontext);

}

/*

 * Reinitialize the dynamic shared memory segment for a parallel context such

 * that we could launch workers for it again.

 */

void

ReinitializeParallelDSM(ParallelContext *pcxt)

{

  MemoryContext oldcontext;

  FixedParallelState *fps;

  /* We might be running in a very short-lived memory context. */

  oldcontext = MemoryContextSwitchTo(TopTransactionContext);

  /* 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);

  pg_atomic_write_u64(&fps->last_xlog_end, InvalidXLogRecPtr);

  /* 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);

    }

  }

  /* Restore previous memory context. */

  MemoryContextSwitchTo(oldcontext);

}

/*

 * Reinitialize parallel workers for a parallel context such that we could

 * launch a different number of workers.  This is required for cases where

 * we need to reuse the same DSM segment, but the number of workers can

 * vary from run-to-run.

 */

void

ReinitializeParallelWorkers(ParallelContext *pcxt, int nworkers_to_launch)

{

  /*

   * The number of workers that need to be launched must be less than the

   * number of workers with which the parallel context is initialized.  But

   * the caller might not know that InitializeParallelDSM reduced nworkers,

   * so just silently trim the request.

   */

  pcxt->nworkers_to_launch = Min(pcxt->nworkers, nworkers_to_launch);

}

/*

 * Launch parallel workers.

 */

void

LaunchParallelWorkers(ParallelContext *pcxt)

{

  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_array(bool, pcxt->nworkers_launched);

    pcxt->nknown_attached_workers = 0;

  }

  /* Restore previous memory context. */

  MemoryContextSwitchTo(oldcontext);

}

/*

 * Wait for all workers to attach to their error queues, and throw an error if

 * any worker fails to do this.

 *

 * Callers can assume that if this function returns successfully, then the

 * number of workers given by pcxt->nworkers_launched have initialized and

 * attached to their error queues.  Whether or not these workers are guaranteed

 * to still be running depends on what code the caller asked them to run;

 * this function does not guarantee that they have not exited.  However, it

 * does guarantee that any workers which exited must have done so cleanly and

 * after successfully performing the work with which they were tasked.

 *

 * If this function is not called, then some of the workers that were launched

 * may not have been started due to a fork() failure, or may have exited during

 * early startup prior to attaching to the error queue, so nworkers_launched

 * cannot be viewed as completely reliable.  It will never be less than the

 * number of workers which actually started, but it might be more.  Any workers

 * that failed to start will still be discovered by

 * WaitForParallelWorkersToFinish and an error will be thrown at that time,

 * provided that function is eventually reached.

 *

 * In general, the leader process should do as much work as possible before

 * calling this function.  fork() failures and other early-startup failures

 * are very uncommon, and having the leader sit idle when it could be doing

 * useful work is undesirable.  However, if the leader needs to wait for

 * all of its workers or for a specific worker, it may want to call this

 * function before doing so.  If not, it must make some other provision for

 * the failure-to-start case, lest it wait forever.  On the other hand, a

 * leader which never waits for a worker that might not be started yet, or

 * at least never does so prior to WaitForParallelWorkersToFinish(), need not

 * call this function at all.

 */

void

WaitForParallelWorkersToAttach(ParallelContext *pcxt)

{

  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;

    }

  }

}

/*

 * Wait for all workers to finish computing.

 *

 * Even if the parallel operation seems to have completed successfully, it's

 * important to call this function afterwards.  We must not miss any errors

 * the workers may have thrown during the parallel operation, or any that they

 * may yet throw while shutting down.

 *

 * Also, we want to update our notion of XactLastRecEnd based on worker

 * feedback.

 */

void

WaitForParallelWorkersToFinish(ParallelContext *pcxt)

{

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

    /*

     * An autovacuum worker running a parallel vacuum (leader) propagates

     * changes to the cost-based delay parameters to its parallel workers

     * at its own cost delay points, which it no longer reaches while

     * waiting here. Do it here instead, on every wakeup, so that a config

     * reload or a change in the number of autovacuum workers sharing the

     * cost limit reaches the parallel workers before they finish.

     */

    if (AmAutoVacuumWorkerProcess())

      parallel_vacuum_refresh_cost_params();

    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,

         * our 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;

    XLogRecPtr  last_xlog_end;

    fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);

    last_xlog_end = pg_atomic_read_u64(&fps->last_xlog_end);

    if (last_xlog_end > XactLastRecEnd)

      XactLastRecEnd = last_xlog_end;

  }

}

/*

 * Wait for all workers to exit.

 *

 * This function ensures that workers have been completely shutdown.  The

 * difference between WaitForParallelWorkersToFinish and this function is

 * that the former just ensures that last message sent by a worker backend is

 * received by the leader backend whereas this ensures the complete shutdown.

 */

static void

WaitForParallelWorkersToExit(ParallelContext *pcxt)

{

  int     i;

  /* Wait until the workers actually die. */

  for (i = 0; i < pcxt->nworkers_launched; ++i)

  {

    BgwHandleStatus status;

    if (pcxt->worker == NULL || pcxt->worker[i].bgwhandle == NULL)

      continue;

    status = WaitForBackgroundWorkerShutdown(pcxt->worker[i].bgwhandle);

    /*

     * If the postmaster kicked the bucket, we have no chance of cleaning

     * up safely -- we won't be able to tell when our workers are actually

     * dead.  This doesn't necessitate a PANIC since they will all abort

     * eventually, but we can't safely continue this session.

     */

    if (status == BGWH_POSTMASTER_DIED)

      ereport(FATAL,

          (errcode(ERRCODE_ADMIN_SHUTDOWN),

           errmsg("postmaster exited during a parallel transaction")));