PostgreSQL Source Code  git master
parallel.c File Reference
#include "postgres.h"
#include "access/brin.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 HandleParallelMessage (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)
 
ParallelContextCreateParallelContext (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 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 FixedParallelStateMyFixedParallelState
 
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 54 of file parallel.c.

◆ PARALLEL_KEY_ACTIVE_SNAPSHOT

#define PARALLEL_KEY_ACTIVE_SNAPSHOT   UINT64CONST(0xFFFFFFFFFFFF0007)

Definition at line 70 of file parallel.c.

◆ PARALLEL_KEY_CLIENTCONNINFO

#define PARALLEL_KEY_CLIENTCONNINFO   UINT64CONST(0xFFFFFFFFFFFF000F)

Definition at line 78 of file parallel.c.

◆ PARALLEL_KEY_COMBO_CID

#define PARALLEL_KEY_COMBO_CID   UINT64CONST(0xFFFFFFFFFFFF0005)

Definition at line 68 of file parallel.c.

◆ PARALLEL_KEY_ENTRYPOINT

#define PARALLEL_KEY_ENTRYPOINT   UINT64CONST(0xFFFFFFFFFFFF0009)

Definition at line 72 of file parallel.c.

◆ PARALLEL_KEY_ERROR_QUEUE

#define PARALLEL_KEY_ERROR_QUEUE   UINT64CONST(0xFFFFFFFFFFFF0002)

Definition at line 65 of file parallel.c.

◆ PARALLEL_KEY_FIXED

#define PARALLEL_KEY_FIXED   UINT64CONST(0xFFFFFFFFFFFF0001)

Definition at line 64 of file parallel.c.

◆ PARALLEL_KEY_GUC

#define PARALLEL_KEY_GUC   UINT64CONST(0xFFFFFFFFFFFF0004)

Definition at line 67 of file parallel.c.

◆ PARALLEL_KEY_LIBRARY

#define PARALLEL_KEY_LIBRARY   UINT64CONST(0xFFFFFFFFFFFF0003)

Definition at line 66 of file parallel.c.

◆ PARALLEL_KEY_PENDING_SYNCS

#define PARALLEL_KEY_PENDING_SYNCS   UINT64CONST(0xFFFFFFFFFFFF000B)

Definition at line 74 of file parallel.c.

◆ PARALLEL_KEY_REINDEX_STATE

#define PARALLEL_KEY_REINDEX_STATE   UINT64CONST(0xFFFFFFFFFFFF000C)

Definition at line 75 of file parallel.c.

◆ PARALLEL_KEY_RELMAPPER_STATE

#define PARALLEL_KEY_RELMAPPER_STATE   UINT64CONST(0xFFFFFFFFFFFF000D)

Definition at line 76 of file parallel.c.

◆ PARALLEL_KEY_SESSION_DSM

#define PARALLEL_KEY_SESSION_DSM   UINT64CONST(0xFFFFFFFFFFFF000A)

Definition at line 73 of file parallel.c.

◆ PARALLEL_KEY_TRANSACTION_SNAPSHOT

#define PARALLEL_KEY_TRANSACTION_SNAPSHOT   UINT64CONST(0xFFFFFFFFFFFF0006)

Definition at line 69 of file parallel.c.

◆ PARALLEL_KEY_TRANSACTION_STATE

#define PARALLEL_KEY_TRANSACTION_STATE   UINT64CONST(0xFFFFFFFFFFFF0008)

Definition at line 71 of file parallel.c.

◆ PARALLEL_KEY_UNCOMMITTEDENUMS

#define PARALLEL_KEY_UNCOMMITTEDENUMS   UINT64CONST(0xFFFFFFFFFFFF000E)

Definition at line 77 of file parallel.c.

◆ PARALLEL_MAGIC

#define PARALLEL_MAGIC   0x50477c7c

Definition at line 57 of file parallel.c.

Typedef Documentation

◆ FixedParallelState

Function Documentation

◆ AtEOSubXact_Parallel()

void AtEOSubXact_Parallel ( bool  isCommit,
SubTransactionId  mySubId 
)

Definition at line 1235 of file parallel.c.

1236 {
1237  while (!dlist_is_empty(&pcxt_list))
1238  {
1239  ParallelContext *pcxt;
1240 
1242  if (pcxt->subid != mySubId)
1243  break;
1244  if (isCommit)
1245  elog(WARNING, "leaked parallel context");
1246  DestroyParallelContext(pcxt);
1247  }
1248 }
void DestroyParallelContext(ParallelContext *pcxt)
Definition: parallel.c:929
static dlist_head pcxt_list
Definition: parallel.c:124
#define WARNING
Definition: elog.h:36
#define elog(elevel,...)
Definition: elog.h:224
#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 1254 of file parallel.c.

1255 {
1256  while (!dlist_is_empty(&pcxt_list))
1257  {
1258  ParallelContext *pcxt;
1259 
1261  if (isCommit)
1262  elog(WARNING, "leaked parallel context");
1263  DestroyParallelContext(pcxt);
1264  }
1265 }

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

169 {
170  MemoryContext oldcontext;
171  ParallelContext *pcxt;
172 
173  /* It is unsafe to create a parallel context if not in parallel mode. */
175 
176  /* Number of workers should be non-negative. */
177  Assert(nworkers >= 0);
178 
179  /* We might be running in a short-lived memory context. */
181 
182  /* Initialize a new ParallelContext. */
183  pcxt = palloc0(sizeof(ParallelContext));
185  pcxt->nworkers = nworkers;
186  pcxt->nworkers_to_launch = nworkers;
187  pcxt->library_name = pstrdup(library_name);
188  pcxt->function_name = pstrdup(function_name);
191  dlist_push_head(&pcxt_list, &pcxt->node);
192 
193  /* Restore previous memory context. */
194  MemoryContextSwitchTo(oldcontext);
195 
196  return pcxt;
197 }
ErrorContextCallback * error_context_stack
Definition: elog.c:94
static void dlist_push_head(dlist_head *head, dlist_node *node)
Definition: ilist.h:347
Assert(fmt[strlen(fmt) - 1] !='\n')
MemoryContext TopTransactionContext
Definition: mcxt.c:142
char * pstrdup(const char *in)
Definition: mcxt.c:1683
void * palloc0(Size size)
Definition: mcxt.c:1334
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_to_launch
Definition: parallel.h:36
char * function_name
Definition: parallel.h:39
SubTransactionId GetCurrentSubTransactionId(void)
Definition: xact.c:781
bool IsInParallelMode(void)
Definition: xact.c:1070

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

◆ DestroyParallelContext()

void DestroyParallelContext ( ParallelContext pcxt)

Definition at line 929 of file parallel.c.

930 {
931  int i;
932 
933  /*
934  * Be careful about order of operations here! We remove the parallel
935  * context from the list before we do anything else; otherwise, if an
936  * error occurs during a subsequent step, we might try to nuke it again
937  * from AtEOXact_Parallel or AtEOSubXact_Parallel.
938  */
939  dlist_delete(&pcxt->node);
940 
941  /* Kill each worker in turn, and forget their error queues. */
942  if (pcxt->worker != NULL)
943  {
944  for (i = 0; i < pcxt->nworkers_launched; ++i)
945  {
946  if (pcxt->worker[i].error_mqh != NULL)
947  {
949 
951  pcxt->worker[i].error_mqh = NULL;
952  }
953  }
954  }
955 
956  /*
957  * If we have allocated a shared memory segment, detach it. This will
958  * implicitly detach the error queues, and any other shared memory queues,
959  * stored there.
960  */
961  if (pcxt->seg != NULL)
962  {
963  dsm_detach(pcxt->seg);
964  pcxt->seg = NULL;
965  }
966 
967  /*
968  * If this parallel context is actually in backend-private memory rather
969  * than shared memory, free that memory instead.
970  */
971  if (pcxt->private_memory != NULL)
972  {
973  pfree(pcxt->private_memory);
974  pcxt->private_memory = NULL;
975  }
976 
977  /*
978  * We can't finish transaction commit or abort until all of the workers
979  * have exited. This means, in particular, that we can't respond to
980  * interrupts at this stage.
981  */
982  HOLD_INTERRUPTS();
985 
986  /* Free the worker array itself. */
987  if (pcxt->worker != NULL)
988  {
989  pfree(pcxt->worker);
990  pcxt->worker = NULL;
991  }
992 
993  /* Free memory. */
994  pfree(pcxt->library_name);
995  pfree(pcxt->function_name);
996  pfree(pcxt);
997 }
static void WaitForParallelWorkersToExit(ParallelContext *pcxt)
Definition: parallel.c:889
void TerminateBackgroundWorker(BackgroundWorkerHandle *handle)
Definition: bgworker.c:1221
void dsm_detach(dsm_segment *seg)
Definition: dsm.c:803
static void dlist_delete(dlist_node *node)
Definition: ilist.h:405
int i
Definition: isn.c:73
void pfree(void *pointer)
Definition: mcxt.c:1508
#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(), AtEOSubXact_Parallel(), AtEOXact_Parallel(), ExecParallelCleanup(), and parallel_vacuum_end().

◆ HandleParallelMessage()

static void HandleParallelMessage ( ParallelContext pcxt,
int  i,
StringInfo  msg 
)
static

Definition at line 1116 of file parallel.c.

1117 {
1118  char msgtype;
1119 
1120  if (pcxt->known_attached_workers != NULL &&
1121  !pcxt->known_attached_workers[i])
1122  {
1123  pcxt->known_attached_workers[i] = true;
1124  pcxt->nknown_attached_workers++;
1125  }
1126 
1127  msgtype = pq_getmsgbyte(msg);
1128 
1129  switch (msgtype)
1130  {
1131  case PqMsg_ErrorResponse:
1132  case PqMsg_NoticeResponse:
1133  {
1134  ErrorData edata;
1135  ErrorContextCallback *save_error_context_stack;
1136 
1137  /* Parse ErrorResponse or NoticeResponse. */
1138  pq_parse_errornotice(msg, &edata);
1139 
1140  /* Death of a worker isn't enough justification for suicide. */
1141  edata.elevel = Min(edata.elevel, ERROR);
1142 
1143  /*
1144  * If desired, add a context line to show that this is a
1145  * message propagated from a parallel worker. Otherwise, it
1146  * can sometimes be confusing to understand what actually
1147  * happened. (We don't do this in DEBUG_PARALLEL_REGRESS mode
1148  * because it causes test-result instability depending on
1149  * whether a parallel worker is actually used or not.)
1150  */
1152  {
1153  if (edata.context)
1154  edata.context = psprintf("%s\n%s", edata.context,
1155  _("parallel worker"));
1156  else
1157  edata.context = pstrdup(_("parallel worker"));
1158  }
1159 
1160  /*
1161  * Context beyond that should use the error context callbacks
1162  * that were in effect when the ParallelContext was created,
1163  * not the current ones.
1164  */
1165  save_error_context_stack = error_context_stack;
1167 
1168  /* Rethrow error or print notice. */
1169  ThrowErrorData(&edata);
1170 
1171  /* Not an error, so restore previous context stack. */
1172  error_context_stack = save_error_context_stack;
1173 
1174  break;
1175  }
1176 
1178  {
1179  /* Propagate NotifyResponse. */
1180  int32 pid;
1181  const char *channel;
1182  const char *payload;
1183 
1184  pid = pq_getmsgint(msg, 4);
1185  channel = pq_getmsgrawstring(msg);
1186  payload = pq_getmsgrawstring(msg);
1187  pq_endmessage(msg);
1188 
1189  NotifyMyFrontEnd(channel, payload, pid);
1190 
1191  break;
1192  }
1193 
1194  case 'P': /* Parallel progress reporting */
1195  {
1196  /*
1197  * Only incremental progress reporting is currently supported.
1198  * However, it's possible to add more fields to the message to
1199  * allow for handling of other backend progress APIs.
1200  */
1201  int index = pq_getmsgint(msg, 4);
1202  int64 incr = pq_getmsgint64(msg);
1203 
1204  pq_getmsgend(msg);
1205 
1207 
1208  break;
1209  }
1210 
1211  case PqMsg_Terminate:
1212  {
1213  shm_mq_detach(pcxt->worker[i].error_mqh);
1214  pcxt->worker[i].error_mqh = NULL;
1215  break;
1216  }
1217 
1218  default:
1219  {
1220  elog(ERROR, "unrecognized message type received from parallel worker: %c (message length %d bytes)",
1221  msgtype, msg->len);
1222  }
1223  }
1224 }
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:991
signed int int32
Definition: c.h:481
void ThrowErrorData(ErrorData *edata)
Definition: elog.c:1853
#define _(x)
Definition: elog.c:90
#define ERROR
Definition: elog.h:39
@ DEBUG_PARALLEL_REGRESS
Definition: optimizer.h:108
int debug_parallel_query
Definition: planner.c:66
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
int64 pq_getmsgint64(StringInfo msg)
Definition: pqformat.c:453
const char * pq_getmsgrawstring(StringInfo msg)
Definition: pqformat.c:608
void pq_parse_errornotice(StringInfo msg, ErrorData *edata)
Definition: pqmq.c:216
#define PqMsg_NotificationResponse
Definition: protocol.h:41
#define PqMsg_ErrorResponse
Definition: protocol.h:44
#define PqMsg_NoticeResponse
Definition: protocol.h:49
#define PqMsg_Terminate
Definition: protocol.h:28
char * psprintf(const char *fmt,...)
Definition: psprintf.c:46
char * context
Definition: elog.h:443
int elevel
Definition: elog.h:428
bool * known_attached_workers
Definition: parallel.h:47
int nknown_attached_workers
Definition: parallel.h:46
Definition: type.h:95

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_Terminate, psprintf(), pstrdup(), shm_mq_detach(), ThrowErrorData(), and ParallelContext::worker.

Referenced by HandleParallelMessages().

◆ HandleParallelMessageInterrupt()

void HandleParallelMessageInterrupt ( void  )

Definition at line 1016 of file parallel.c.

1017 {
1018  InterruptPending = true;
1019  ParallelMessagePending = true;
1020  SetLatch(MyLatch);
1021 }
volatile sig_atomic_t ParallelMessagePending
Definition: parallel.c:115
volatile sig_atomic_t InterruptPending
Definition: globals.c:30
struct Latch * MyLatch
Definition: globals.c:60
void SetLatch(Latch *latch)
Definition: latch.c:632

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

Referenced by procsignal_sigusr1_handler().

◆ HandleParallelMessages()

void HandleParallelMessages ( void  )

Definition at line 1027 of file parallel.c.

1028 {
1029  dlist_iter iter;
1030  MemoryContext oldcontext;
1031 
1032  static MemoryContext hpm_context = NULL;
1033 
1034  /*
1035  * This is invoked from ProcessInterrupts(), and since some of the
1036  * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
1037  * for recursive calls if more signals are received while this runs. It's
1038  * unclear that recursive entry would be safe, and it doesn't seem useful
1039  * even if it is safe, so let's block interrupts until done.
1040  */
1041  HOLD_INTERRUPTS();
1042 
1043  /*
1044  * Moreover, CurrentMemoryContext might be pointing almost anywhere. We
1045  * don't want to risk leaking data into long-lived contexts, so let's do
1046  * our work here in a private context that we can reset on each use.
1047  */
1048  if (hpm_context == NULL) /* first time through? */
1050  "HandleParallelMessages",
1052  else
1053  MemoryContextReset(hpm_context);
1054 
1055  oldcontext = MemoryContextSwitchTo(hpm_context);
1056 
1057  /* OK to process messages. Reset the flag saying there are more to do. */
1058  ParallelMessagePending = false;
1059 
1060  dlist_foreach(iter, &pcxt_list)
1061  {
1062  ParallelContext *pcxt;
1063  int i;
1064 
1065  pcxt = dlist_container(ParallelContext, node, iter.cur);
1066  if (pcxt->worker == NULL)
1067  continue;
1068 
1069  for (i = 0; i < pcxt->nworkers_launched; ++i)
1070  {
1071  /*
1072  * Read as many messages as we can from each worker, but stop when
1073  * either (1) the worker's error queue goes away, which can happen
1074  * if we receive a Terminate message from the worker; or (2) no
1075  * more messages can be read from the worker without blocking.
1076  */
1077  while (pcxt->worker[i].error_mqh != NULL)
1078  {
1080  Size nbytes;
1081  void *data;
1082 
1083  res = shm_mq_receive(pcxt->worker[i].error_mqh, &nbytes,
1084  &data, true);
1085  if (res == SHM_MQ_WOULD_BLOCK)
1086  break;
1087  else if (res == SHM_MQ_SUCCESS)
1088  {
1089  StringInfoData msg;
1090 
1091  initStringInfo(&msg);
1092  appendBinaryStringInfo(&msg, data, nbytes);
1093  HandleParallelMessage(pcxt, i, &msg);
1094  pfree(msg.data);
1095  }
1096  else
1097  ereport(ERROR,
1098  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1099  errmsg("lost connection to parallel worker")));
1100  }
1101  }
1102  }
1103 
1104  MemoryContextSwitchTo(oldcontext);
1105 
1106  /* Might as well clear the context on our way out */
1107  MemoryContextReset(hpm_context);
1108 
1110 }
static void HandleParallelMessage(ParallelContext *pcxt, int i, StringInfo msg)
Definition: parallel.c:1116
size_t Size
Definition: c.h:592
int errcode(int sqlerrcode)
Definition: elog.c:859
int errmsg(const char *fmt,...)
Definition: elog.c:1072
#define ereport(elevel,...)
Definition: elog.h:149
#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:371
MemoryContext TopMemoryContext
Definition: mcxt.c:137
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:153
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:233
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
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, 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 205 of file parallel.c.

206 {
207  MemoryContext oldcontext;
208  Size library_len = 0;
209  Size guc_len = 0;
210  Size combocidlen = 0;
211  Size tsnaplen = 0;
212  Size asnaplen = 0;
213  Size tstatelen = 0;
214  Size pendingsyncslen = 0;
215  Size reindexlen = 0;
216  Size relmapperlen = 0;
217  Size uncommittedenumslen = 0;
218  Size clientconninfolen = 0;
219  Size segsize = 0;
220  int i;
221  FixedParallelState *fps;
222  dsm_handle session_dsm_handle = DSM_HANDLE_INVALID;
223  Snapshot transaction_snapshot = GetTransactionSnapshot();
224  Snapshot active_snapshot = GetActiveSnapshot();
225 
226  /* We might be running in a very short-lived memory context. */
228 
229  /* Allow space to store the fixed-size parallel state. */
231  shm_toc_estimate_keys(&pcxt->estimator, 1);
232 
233  /*
234  * Normally, the user will have requested at least one worker process, but
235  * if by chance they have not, we can skip a bunch of things here.
236  */
237  if (pcxt->nworkers > 0)
238  {
239  /* Get (or create) the per-session DSM segment's handle. */
240  session_dsm_handle = GetSessionDsmHandle();
241 
242  /*
243  * If we weren't able to create a per-session DSM segment, then we can
244  * continue but we can't safely launch any workers because their
245  * record typmods would be incompatible so they couldn't exchange
246  * tuples.
247  */
248  if (session_dsm_handle == DSM_HANDLE_INVALID)
249  pcxt->nworkers = 0;
250  }
251 
252  if (pcxt->nworkers > 0)
253  {
254  /* Estimate space for various kinds of state sharing. */
255  library_len = EstimateLibraryStateSpace();
256  shm_toc_estimate_chunk(&pcxt->estimator, library_len);
257  guc_len = EstimateGUCStateSpace();
258  shm_toc_estimate_chunk(&pcxt->estimator, guc_len);
259  combocidlen = EstimateComboCIDStateSpace();
260  shm_toc_estimate_chunk(&pcxt->estimator, combocidlen);
262  {
263  tsnaplen = EstimateSnapshotSpace(transaction_snapshot);
264  shm_toc_estimate_chunk(&pcxt->estimator, tsnaplen);
265  }
266  asnaplen = EstimateSnapshotSpace(active_snapshot);
267  shm_toc_estimate_chunk(&pcxt->estimator, asnaplen);
268  tstatelen = EstimateTransactionStateSpace();
269  shm_toc_estimate_chunk(&pcxt->estimator, tstatelen);
271  pendingsyncslen = EstimatePendingSyncsSpace();
272  shm_toc_estimate_chunk(&pcxt->estimator, pendingsyncslen);
273  reindexlen = EstimateReindexStateSpace();
274  shm_toc_estimate_chunk(&pcxt->estimator, reindexlen);
275  relmapperlen = EstimateRelationMapSpace();
276  shm_toc_estimate_chunk(&pcxt->estimator, relmapperlen);
277  uncommittedenumslen = EstimateUncommittedEnumsSpace();
278  shm_toc_estimate_chunk(&pcxt->estimator, uncommittedenumslen);
279  clientconninfolen = EstimateClientConnectionInfoSpace();
280  shm_toc_estimate_chunk(&pcxt->estimator, clientconninfolen);
281  /* If you add more chunks here, you probably need to add keys. */
282  shm_toc_estimate_keys(&pcxt->estimator, 12);
283 
284  /* Estimate space need for error queues. */
287  "parallel error queue size not buffer-aligned");
290  pcxt->nworkers));
291  shm_toc_estimate_keys(&pcxt->estimator, 1);
292 
293  /* Estimate how much we'll need for the entrypoint info. */
294  shm_toc_estimate_chunk(&pcxt->estimator, strlen(pcxt->library_name) +
295  strlen(pcxt->function_name) + 2);
296  shm_toc_estimate_keys(&pcxt->estimator, 1);
297  }
298 
299  /*
300  * Create DSM and initialize with new table of contents. But if the user
301  * didn't request any workers, then don't bother creating a dynamic shared
302  * memory segment; instead, just use backend-private memory.
303  *
304  * Also, if we can't create a dynamic shared memory segment because the
305  * maximum number of segments have already been created, then fall back to
306  * backend-private memory, and plan not to use any workers. We hope this
307  * won't happen very often, but it's better to abandon the use of
308  * parallelism than to fail outright.
309  */
310  segsize = shm_toc_estimate(&pcxt->estimator);
311  if (pcxt->nworkers > 0)
313  if (pcxt->seg != NULL)
315  dsm_segment_address(pcxt->seg),
316  segsize);
317  else
318  {
319  pcxt->nworkers = 0;
322  segsize);
323  }
324 
325  /* Initialize fixed-size state in shared memory. */
326  fps = (FixedParallelState *)
327  shm_toc_allocate(pcxt->toc, sizeof(FixedParallelState));
328  fps->database_id = MyDatabaseId;
341  SpinLockInit(&fps->mutex);
342  fps->last_xlog_end = 0;
344 
345  /* We can skip the rest of this if we're not budgeting for any workers. */
346  if (pcxt->nworkers > 0)
347  {
348  char *libraryspace;
349  char *gucspace;
350  char *combocidspace;
351  char *tsnapspace;
352  char *asnapspace;
353  char *tstatespace;
354  char *pendingsyncsspace;
355  char *reindexspace;
356  char *relmapperspace;
357  char *error_queue_space;
358  char *session_dsm_handle_space;
359  char *entrypointstate;
360  char *uncommittedenumsspace;
361  char *clientconninfospace;
362  Size lnamelen;
363 
364  /* Serialize shared libraries we have loaded. */
365  libraryspace = shm_toc_allocate(pcxt->toc, library_len);
366  SerializeLibraryState(library_len, libraryspace);
367  shm_toc_insert(pcxt->toc, PARALLEL_KEY_LIBRARY, libraryspace);
368 
369  /* Serialize GUC settings. */
370  gucspace = shm_toc_allocate(pcxt->toc, guc_len);
371  SerializeGUCState(guc_len, gucspace);
372  shm_toc_insert(pcxt->toc, PARALLEL_KEY_GUC, gucspace);
373 
374  /* Serialize combo CID state. */
375  combocidspace = shm_toc_allocate(pcxt->toc, combocidlen);
376  SerializeComboCIDState(combocidlen, combocidspace);
377  shm_toc_insert(pcxt->toc, PARALLEL_KEY_COMBO_CID, combocidspace);
378 
379  /*
380  * Serialize the transaction snapshot if the transaction isolation
381  * level uses a transaction snapshot.
382  */
384  {
385  tsnapspace = shm_toc_allocate(pcxt->toc, tsnaplen);
386  SerializeSnapshot(transaction_snapshot, tsnapspace);
388  tsnapspace);
389  }
390 
391  /* Serialize the active snapshot. */
392  asnapspace = shm_toc_allocate(pcxt->toc, asnaplen);
393  SerializeSnapshot(active_snapshot, asnapspace);
394  shm_toc_insert(pcxt->toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, asnapspace);
395 
396  /* Provide the handle for per-session segment. */
397  session_dsm_handle_space = shm_toc_allocate(pcxt->toc,
398  sizeof(dsm_handle));
399  *(dsm_handle *) session_dsm_handle_space = session_dsm_handle;
401  session_dsm_handle_space);
402 
403  /* Serialize transaction state. */
404  tstatespace = shm_toc_allocate(pcxt->toc, tstatelen);
405  SerializeTransactionState(tstatelen, tstatespace);
407 
408  /* Serialize pending syncs. */
409  pendingsyncsspace = shm_toc_allocate(pcxt->toc, pendingsyncslen);
410  SerializePendingSyncs(pendingsyncslen, pendingsyncsspace);
412  pendingsyncsspace);
413 
414  /* Serialize reindex state. */
415  reindexspace = shm_toc_allocate(pcxt->toc, reindexlen);
416  SerializeReindexState(reindexlen, reindexspace);
417  shm_toc_insert(pcxt->toc, PARALLEL_KEY_REINDEX_STATE, reindexspace);
418 
419  /* Serialize relmapper state. */
420  relmapperspace = shm_toc_allocate(pcxt->toc, relmapperlen);
421  SerializeRelationMap(relmapperlen, relmapperspace);
423  relmapperspace);
424 
425  /* Serialize uncommitted enum state. */
426  uncommittedenumsspace = shm_toc_allocate(pcxt->toc,
427  uncommittedenumslen);
428  SerializeUncommittedEnums(uncommittedenumsspace, uncommittedenumslen);
430  uncommittedenumsspace);
431 
432  /* Serialize our ClientConnectionInfo. */
433  clientconninfospace = shm_toc_allocate(pcxt->toc, clientconninfolen);
434  SerializeClientConnectionInfo(clientconninfolen, clientconninfospace);
436  clientconninfospace);
437 
438  /* Allocate space for worker information. */
439  pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * pcxt->nworkers);
440 
441  /*
442  * Establish error queues in dynamic shared memory.
443  *
444  * These queues should be used only for transmitting ErrorResponse,
445  * NoticeResponse, and NotifyResponse protocol messages. Tuple data
446  * should be transmitted via separate (possibly larger?) queues.
447  */
448  error_queue_space =
449  shm_toc_allocate(pcxt->toc,
451  pcxt->nworkers));
452  for (i = 0; i < pcxt->nworkers; ++i)
453  {
454  char *start;
455  shm_mq *mq;
456 
457  start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
460  pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
461  }
462  shm_toc_insert(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, error_queue_space);
463 
464  /*
465  * Serialize entrypoint information. It's unsafe to pass function
466  * pointers across processes, as the function pointer may be different
467  * in each process in EXEC_BACKEND builds, so we always pass library
468  * and function name. (We use library name "postgres" for functions
469  * in the core backend.)
470  */
471  lnamelen = strlen(pcxt->library_name);
472  entrypointstate = shm_toc_allocate(pcxt->toc, lnamelen +
473  strlen(pcxt->function_name) + 2);
474  strcpy(entrypointstate, pcxt->library_name);
475  strcpy(entrypointstate + lnamelen + 1, pcxt->function_name);
476  shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENTRYPOINT, entrypointstate);
477  }
478 
479  /* Restore previous memory context. */
480  MemoryContextSwitchTo(oldcontext);
481 }
#define PARALLEL_KEY_TRANSACTION_STATE
Definition: parallel.c:71
#define PARALLEL_KEY_GUC
Definition: parallel.c:67
#define PARALLEL_KEY_UNCOMMITTEDENUMS
Definition: parallel.c:77
#define PARALLEL_KEY_TRANSACTION_SNAPSHOT
Definition: parallel.c:69
#define PARALLEL_KEY_CLIENTCONNINFO
Definition: parallel.c:78
#define PARALLEL_KEY_PENDING_SYNCS
Definition: parallel.c:74
#define PARALLEL_KEY_ACTIVE_SNAPSHOT
Definition: parallel.c:70
#define PARALLEL_KEY_ERROR_QUEUE
Definition: parallel.c:65
#define PARALLEL_KEY_SESSION_DSM
Definition: parallel.c:73
#define PARALLEL_MAGIC
Definition: parallel.c:57
#define PARALLEL_KEY_REINDEX_STATE
Definition: parallel.c:75
#define PARALLEL_KEY_LIBRARY
Definition: parallel.c:66
#define PARALLEL_KEY_FIXED
Definition: parallel.c:64
#define PARALLEL_KEY_ENTRYPOINT
Definition: parallel.c:72
#define PARALLEL_KEY_COMBO_CID
Definition: parallel.c:68
#define PARALLEL_ERROR_QUEUE_SIZE
Definition: parallel.c:54
#define PARALLEL_KEY_RELMAPPER_STATE
Definition: parallel.c:76
#define BUFFERALIGN(LEN)
Definition: c.h:800
#define StaticAssertStmt(condition, errmessage)
Definition: c.h:925
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:654
Size EstimateLibraryStateSpace(void)
Definition: dfmgr.c:637
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:45
ProcNumber MyProcNumber
Definition: globals.c:87
Oid MyDatabaseId
Definition: globals.c:91
void SerializeGUCState(Size maxsize, char *start_address)
Definition: guc.c:6045
Size EstimateGUCStateSpace(void)
Definition: guc.c:5892
bool current_role_is_superuser
Definition: guc_tables.c:519
void SerializeReindexState(Size maxsize, char *start_address)
Definition: index.c:4182
Size EstimateReindexStateSpace(void)
Definition: index.c:4171
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:1168
void SerializeClientConnectionInfo(Size maxsize, char *start_address)
Definition: miscinit.c:1043
void GetUserIdAndSecContext(Oid *userid, int *sec_context)
Definition: miscinit.c:635
Size EstimateClientConnectionInfoSpace(void)
Definition: miscinit.c:1027
Oid GetAuthenticatedUserId(void)
Definition: miscinit.c:583
Oid GetCurrentRoleId(void)
Definition: miscinit.c:924
void GetTempNamespaceState(Oid *tempNamespaceId, Oid *tempToastNamespaceId)
Definition: namespace.c:3790
Size EstimateUncommittedEnumsSpace(void)
Definition: pg_enum.c:724
void SerializeUncommittedEnums(void *space, Size size)
Definition: pg_enum.c:738
SerializableXactHandle ShareSerializableXact(void)
Definition: predicate.c:5026
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_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:290
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_toc * shm_toc_create(uint64 magic, void *address, Size nbytes)
Definition: shm_toc.c:40
Size shm_toc_estimate(shm_toc_estimator *e)
Definition: shm_toc.c:263
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
#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:1716
Snapshot GetTransactionSnapshot(void)
Definition: snapmgr.c:216
Size EstimateSnapshotSpace(Snapshot snapshot)
Definition: snapmgr.c:1692
Snapshot GetActiveSnapshot(void)
Definition: snapmgr.c:770
#define SpinLockInit(lock)
Definition: spin.h:60
PGPROC * MyProc
Definition: proc.c:66
void SerializePendingSyncs(Size maxSize, char *startAddress)
Definition: storage.c:559
Size EstimatePendingSyncsSpace(void)
Definition: storage.c:546
Oid temp_toast_namespace_id
Definition: parallel.c:89
XLogRecPtr last_xlog_end
Definition: parallel.c:103
TimestampTz stmt_ts
Definition: parallel.c:96
SerializableXactHandle serializable_xact_handle
Definition: parallel.c:97
TimestampTz xact_ts
Definition: parallel.c:95
PGPROC * parallel_leader_pgproc
Definition: parallel.c:92
pid_t parallel_leader_pid
Definition: parallel.c:93
Oid authenticated_user_id
Definition: parallel.c:85
ProcNumber parallel_leader_proc_number
Definition: parallel.c:94
shm_toc * toc
Definition: parallel.h:44
Definition: shm_mq.c:72
void SerializeTransactionState(Size maxsize, char *start_address)
Definition: xact.c:5462
Size EstimateTransactionStateSpace(void)
Definition: xact.c:5434
TimestampTz GetCurrentStatementStartTimestamp(void)
Definition: xact.c:865
TimestampTz GetCurrentTransactionStartTimestamp(void)
Definition: xact.c:856
#define IsolationUsesXactSnapshot()
Definition: xact.h:51

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(), GetTempNamespaceState(), GetTransactionSnapshot(), GetUserIdAndSecContext(), i, FixedParallelState::is_superuser, IsolationUsesXactSnapshot, FixedParallelState::last_xlog_end, ParallelContext::library_name, MemoryContextAlloc(), MemoryContextSwitchTo(), mul_size(), FixedParallelState::mutex, MyDatabaseId, MyProc, MyProcNumber, MyProcPid, ParallelContext::nworkers, 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::sec_context, ParallelContext::seg, FixedParallelState::serializable_xact_handle, SerializeClientConnectionInfo(), SerializeComboCIDState(), SerializeGUCState(), SerializeLibraryState(), SerializePendingSyncs(), SerializeReindexState(), SerializeRelationMap(), SerializeSnapshot(), SerializeTransactionState(), SerializeUncommittedEnums(), 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, 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(), ExecInitParallelPlan(), and parallel_vacuum_init().

◆ LaunchParallelWorkers()

void LaunchParallelWorkers ( ParallelContext pcxt)

Definition at line 552 of file parallel.c.

553 {
554  MemoryContext oldcontext;
555  BackgroundWorker worker;
556  int i;
557  bool any_registrations_failed = false;
558 
559  /* Skip this if we have no workers. */
560  if (pcxt->nworkers == 0 || pcxt->nworkers_to_launch == 0)
561  return;
562 
563  /* We need to be a lock group leader. */
565 
566  /* If we do have workers, we'd better have a DSM segment. */
567  Assert(pcxt->seg != NULL);
568 
569  /* We might be running in a short-lived memory context. */
571 
572  /* Configure a worker. */
573  memset(&worker, 0, sizeof(worker));
574  snprintf(worker.bgw_name, BGW_MAXLEN, "parallel worker for PID %d",
575  MyProcPid);
576  snprintf(worker.bgw_type, BGW_MAXLEN, "parallel worker");
577  worker.bgw_flags =
582  sprintf(worker.bgw_library_name, "postgres");
583  sprintf(worker.bgw_function_name, "ParallelWorkerMain");
585  worker.bgw_notify_pid = MyProcPid;
586 
587  /*
588  * Start workers.
589  *
590  * The caller must be able to tolerate ending up with fewer workers than
591  * expected, so there is no need to throw an error here if registration
592  * fails. It wouldn't help much anyway, because registering the worker in
593  * no way guarantees that it will start up and initialize successfully.
594  */
595  for (i = 0; i < pcxt->nworkers_to_launch; ++i)
596  {
597  memcpy(worker.bgw_extra, &i, sizeof(int));
598  if (!any_registrations_failed &&
600  &pcxt->worker[i].bgwhandle))
601  {
603  pcxt->worker[i].bgwhandle);
604  pcxt->nworkers_launched++;
605  }
606  else
607  {
608  /*
609  * If we weren't able to register the worker, then we've bumped up
610  * against the max_worker_processes limit, and future
611  * registrations will probably fail too, so arrange to skip them.
612  * But we still have to execute this code for the remaining slots
613  * to make sure that we forget about the error queues we budgeted
614  * for those workers. Otherwise, we'll wait for them to start,
615  * but they never will.
616  */
617  any_registrations_failed = true;
618  pcxt->worker[i].bgwhandle = NULL;
620  pcxt->worker[i].error_mqh = NULL;
621  }
622  }
623 
624  /*
625  * Now that nworkers_launched has taken its final value, we can initialize
626  * known_attached_workers.
627  */
628  if (pcxt->nworkers_launched > 0)
629  {
630  pcxt->known_attached_workers =
631  palloc0(sizeof(bool) * pcxt->nworkers_launched);
632  pcxt->nknown_attached_workers = 0;
633  }
634 
635  /* Restore previous memory context. */
636  MemoryContextSwitchTo(oldcontext);
637 }
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:970
#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:240
#define snprintf
Definition: port.h:238
static Datum UInt32GetDatum(uint32 X)
Definition: postgres.h:232
void shm_mq_set_handle(shm_mq_handle *mqh, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:319
void BecomeLockGroupLeader(void)
Definition: proc.c:1893
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
pid_t bgw_notify_pid
Definition: bgworker.h:100
char bgw_library_name[MAXPGPATH]
Definition: bgworker.h:96

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

1596 {
1597  /*
1598  * If the function is to be loaded from postgres itself, search the
1599  * InternalParallelWorkers array.
1600  */
1601  if (strcmp(libraryname, "postgres") == 0)
1602  {
1603  int i;
1604 
1605  for (i = 0; i < lengthof(InternalParallelWorkers); i++)
1606  {
1607  if (strcmp(InternalParallelWorkers[i].fn_name, funcname) == 0)
1609  }
1610 
1611  /* We can only reach this by programming error. */
1612  elog(ERROR, "internal function \"%s\" not found", funcname);
1613  }
1614 
1615  /* Otherwise load from external library. */
1616  return (parallel_worker_main_type)
1617  load_external_function(libraryname, funcname, true, NULL);
1618 }
static const struct @13 InternalParallelWorkers[]
parallel_worker_main_type fn_addr
Definition: parallel.c:136
const char * fn_name
Definition: parallel.c:135
#define lengthof(array)
Definition: c.h:775
void * load_external_function(const char *filename, const char *funcname, bool signalNotFound, void **filehandle)
Definition: dfmgr.c:105
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 1003 of file parallel.c.

1004 {
1005  return !dlist_is_empty(&pcxt_list);
1006 }

References dlist_is_empty(), and pcxt_list.

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

◆ ParallelWorkerMain()

void ParallelWorkerMain ( Datum  main_arg)

Definition at line 1271 of file parallel.c.

1272 {
1273  dsm_segment *seg;
1274  shm_toc *toc;
1275  FixedParallelState *fps;
1276  char *error_queue_space;
1277  shm_mq *mq;
1278  shm_mq_handle *mqh;
1279  char *libraryspace;
1280  char *entrypointstate;
1281  char *library_name;
1282  char *function_name;
1283  parallel_worker_main_type entrypt;
1284  char *gucspace;
1285  char *combocidspace;
1286  char *tsnapspace;
1287  char *asnapspace;
1288  char *tstatespace;
1289  char *pendingsyncsspace;
1290  char *reindexspace;
1291  char *relmapperspace;
1292  char *uncommittedenumsspace;
1293  char *clientconninfospace;
1294  char *session_dsm_handle_space;
1295  Snapshot tsnapshot;
1296  Snapshot asnapshot;
1297 
1298  /* Set flag to indicate that we're initializing a parallel worker. */
1300 
1301  /* Establish signal handlers. */
1302  pqsignal(SIGTERM, die);
1304 
1305  /* Determine and set our parallel worker number. */
1307  memcpy(&ParallelWorkerNumber, MyBgworkerEntry->bgw_extra, sizeof(int));
1308 
1309  /* Set up a memory context to work in, just for cleanliness. */
1311  "Parallel worker",
1313 
1314  /*
1315  * Attach to the dynamic shared memory segment for the parallel query, and
1316  * find its table of contents.
1317  *
1318  * Note: at this point, we have not created any ResourceOwner in this
1319  * process. This will result in our DSM mapping surviving until process
1320  * exit, which is fine. If there were a ResourceOwner, it would acquire
1321  * ownership of the mapping, but we have no need for that.
1322  */
1323  seg = dsm_attach(DatumGetUInt32(main_arg));
1324  if (seg == NULL)
1325  ereport(ERROR,
1326  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1327  errmsg("could not map dynamic shared memory segment")));
1329  if (toc == NULL)
1330  ereport(ERROR,
1331  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1332  errmsg("invalid magic number in dynamic shared memory segment")));
1333 
1334  /* Look up fixed parallel state. */
1335  fps = shm_toc_lookup(toc, PARALLEL_KEY_FIXED, false);
1336  MyFixedParallelState = fps;
1337 
1338  /* Arrange to signal the leader if we exit. */
1342 
1343  /*
1344  * Now we can find and attach to the error queue provided for us. That's
1345  * good, because until we do that, any errors that happen here will not be
1346  * reported back to the process that requested that this worker be
1347  * launched.
1348  */
1349  error_queue_space = shm_toc_lookup(toc, PARALLEL_KEY_ERROR_QUEUE, false);
1350  mq = (shm_mq *) (error_queue_space +
1353  mqh = shm_mq_attach(mq, seg, NULL);
1354  pq_redirect_to_shm_mq(seg, mqh);
1357 
1358  /*
1359  * Hooray! Primary initialization is complete. Now, we need to set up our
1360  * backend-local state to match the original backend.
1361  */
1362 
1363  /*
1364  * Join locking group. We must do this before anything that could try to
1365  * acquire a heavyweight lock, because any heavyweight locks acquired to
1366  * this point could block either directly against the parallel group
1367  * leader or against some process which in turn waits for a lock that
1368  * conflicts with the parallel group leader, causing an undetected
1369  * deadlock. (If we can't join the lock group, the leader has gone away,
1370  * so just exit quietly.)
1371  */
1373  fps->parallel_leader_pid))
1374  return;
1375 
1376  /*
1377  * Restore transaction and statement start-time timestamps. This must
1378  * happen before anything that would start a transaction, else asserts in
1379  * xact.c will fire.
1380  */
1382 
1383  /*
1384  * Identify the entry point to be called. In theory this could result in
1385  * loading an additional library, though most likely the entry point is in
1386  * the core backend or in a library we just loaded.
1387  */
1388  entrypointstate = shm_toc_lookup(toc, PARALLEL_KEY_ENTRYPOINT, false);
1389  library_name = entrypointstate;
1390  function_name = entrypointstate + strlen(library_name) + 1;
1391 
1392  entrypt = LookupParallelWorkerFunction(library_name, function_name);
1393 
1394  /* Restore database connection. */
1396  fps->authenticated_user_id,
1397  0);
1398 
1399  /*
1400  * Set the client encoding to the database encoding, since that is what
1401  * the leader will expect.
1402  */
1404 
1405  /*
1406  * Load libraries that were loaded by original backend. We want to do
1407  * this before restoring GUCs, because the libraries might define custom
1408  * variables.
1409  */
1410  libraryspace = shm_toc_lookup(toc, PARALLEL_KEY_LIBRARY, false);
1412  RestoreLibraryState(libraryspace);
1413 
1414  /* Restore GUC values from launching backend. */
1415  gucspace = shm_toc_lookup(toc, PARALLEL_KEY_GUC, false);
1416  RestoreGUCState(gucspace);
1418 
1419  /* Crank up a transaction state appropriate to a parallel worker. */
1420  tstatespace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_STATE, false);
1421  StartParallelWorkerTransaction(tstatespace);
1422 
1423  /* Restore combo CID state. */
1424  combocidspace = shm_toc_lookup(toc, PARALLEL_KEY_COMBO_CID, false);
1425  RestoreComboCIDState(combocidspace);
1426 
1427  /* Attach to the per-session DSM segment and contained objects. */
1428  session_dsm_handle_space =
1430  AttachSession(*(dsm_handle *) session_dsm_handle_space);
1431 
1432  /*
1433  * If the transaction isolation level is REPEATABLE READ or SERIALIZABLE,
1434  * the leader has serialized the transaction snapshot and we must restore
1435  * it. At lower isolation levels, there is no transaction-lifetime
1436  * snapshot, but we need TransactionXmin to get set to a value which is
1437  * less than or equal to the xmin of every snapshot that will be used by
1438  * this worker. The easiest way to accomplish that is to install the
1439  * active snapshot as the transaction snapshot. Code running in this
1440  * parallel worker might take new snapshots via GetTransactionSnapshot()
1441  * or GetLatestSnapshot(), but it shouldn't have any way of acquiring a
1442  * snapshot older than the active snapshot.
1443  */
1444  asnapspace = shm_toc_lookup(toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, false);
1445  tsnapspace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT, true);
1446  asnapshot = RestoreSnapshot(asnapspace);
1447  tsnapshot = tsnapspace ? RestoreSnapshot(tsnapspace) : asnapshot;
1448  RestoreTransactionSnapshot(tsnapshot,
1449  fps->parallel_leader_pgproc);
1450  PushActiveSnapshot(asnapshot);
1451 
1452  /*
1453  * We've changed which tuples we can see, and must therefore invalidate
1454  * system caches.
1455  */
1457 
1458  /*
1459  * Restore current role id. Skip verifying whether session user is
1460  * allowed to become this role and blindly restore the leader's state for
1461  * current role.
1462  */
1464 
1465  /* Restore user ID and security context. */
1467 
1468  /* Restore temp-namespace state to ensure search path matches leader's. */
1471 
1472  /* Restore pending syncs. */
1473  pendingsyncsspace = shm_toc_lookup(toc, PARALLEL_KEY_PENDING_SYNCS,
1474  false);
1475  RestorePendingSyncs(pendingsyncsspace);
1476 
1477  /* Restore reindex state. */
1478  reindexspace = shm_toc_lookup(toc, PARALLEL_KEY_REINDEX_STATE, false);
1479  RestoreReindexState(reindexspace);
1480 
1481  /* Restore relmapper state. */
1482  relmapperspace = shm_toc_lookup(toc, PARALLEL_KEY_RELMAPPER_STATE, false);
1483  RestoreRelationMap(relmapperspace);
1484 
1485  /* Restore uncommitted enums. */
1486  uncommittedenumsspace = shm_toc_lookup(toc, PARALLEL_KEY_UNCOMMITTEDENUMS,
1487  false);
1488  RestoreUncommittedEnums(uncommittedenumsspace);
1489 
1490  /* Restore the ClientConnectionInfo. */
1491  clientconninfospace = shm_toc_lookup(toc, PARALLEL_KEY_CLIENTCONNINFO,
1492  false);
1493  RestoreClientConnectionInfo(clientconninfospace);
1494 
1495  /*
1496  * Initialize SystemUser now that MyClientConnectionInfo is restored. Also
1497  * ensure that auth_method is actually valid, aka authn_id is not NULL.
1498  */
1502 
1503  /* Attach to the leader's serializable transaction, if SERIALIZABLE. */
1505 
1506  /*
1507  * We've initialized all of our state now; nothing should change
1508  * hereafter.
1509  */
1512 
1513  /*
1514  * Time to do the real work: invoke the caller-supplied code.
1515  */
1516  entrypt(seg, toc);
1517 
1518  /* Must exit parallel mode to pop active snapshot. */
1519  ExitParallelMode();
1520 
1521  /* Must pop active snapshot so snapmgr.c doesn't complain. */
1523 
1524  /* Shut down the parallel-worker transaction. */
1526 
1527  /* Detach from the per-session DSM segment. */
1528  DetachSession();
1529 
1530  /* Report success. */
1531  pq_putmessage(PqMsg_Terminate, NULL, 0);
1532 }
static parallel_worker_main_type LookupParallelWorkerFunction(const char *libraryname, const char *funcname)
Definition: parallel.c:1595
int ParallelWorkerNumber
Definition: parallel.c:112
bool InitializingParallelWorker
Definition: parallel.c:118
static FixedParallelState * MyFixedParallelState
Definition: parallel.c:121
static pid_t ParallelLeaderPid
Definition: parallel.c:127
static void ParallelWorkerShutdown(int code, Datum arg)
Definition: parallel.c:1567
void RestoreComboCIDState(char *comboCIDstate)
Definition: combocid.c:342
void RestoreLibraryState(char *start_address)
Definition: dfmgr.c:676
dsm_segment * dsm_attach(dsm_handle h)
Definition: dsm.c:665
ProcNumber ParallelLeaderProcNumber
Definition: globals.c:89
void RestoreGUCState(void *gucstate)
Definition: guc.c:6137
const char * hba_authname(UserAuth auth_method)
Definition: hba.c:3061
void RestoreReindexState(const void *reindexstate)
Definition: index.c:4200
void InvalidateSystemCaches(void)
Definition: inval.c:792
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:1261
int SetClientEncoding(int encoding)
Definition: mbutils.c:208
MemoryContext CurrentMemoryContext
Definition: mcxt.c:131
void InitializeSystemUser(const char *authn_id, const char *auth_method)
Definition: miscinit.c:867
void SetCurrentRoleId(Oid roleid, bool is_superuser)
Definition: miscinit.c:945
ClientConnectionInfo MyClientConnectionInfo
Definition: miscinit.c:1010
void RestoreClientConnectionInfo(char *conninfo)
Definition: miscinit.c:1075
void SetUserIdAndSecContext(Oid userid, int sec_context)
Definition: miscinit.c:642
void SetTempNamespaceState(Oid tempNamespaceId, Oid tempToastNamespaceId)
Definition: namespace.c:3806
void RestoreUncommittedEnums(void *space)
Definition: pg_enum.c:770
#define die(msg)
pqsigfunc pqsignal(int signo, pqsigfunc func)
static uint32 DatumGetUInt32(Datum X)
Definition: postgres.h:222
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
void BackgroundWorkerUnblockSignals(void)
Definition: postmaster.c:4229
void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags)
Definition: postmaster.c:4189
BackgroundWorker * MyBgworkerEntry
Definition: postmaster.c:185
void pq_set_parallel_leader(pid_t pid, ProcNumber procNumber)
Definition: pqmq.c:78
void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
Definition: pqmq.c:53
void AttachSerializableXact(SerializableXactHandle handle)
Definition: predicate.c:5035
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
shm_toc * shm_toc_attach(uint64 magic, void *address)
Definition: shm_toc.c:64
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
void PushActiveSnapshot(Snapshot snapshot)
Definition: snapmgr.c:648
Snapshot RestoreSnapshot(char *start_address)
Definition: snapmgr.c:1775
void RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc)
Definition: snapmgr.c:1840
void PopActiveSnapshot(void)
Definition: snapmgr.c:743
bool BecomeLockGroupMember(PGPROC *leader, int pid)
Definition: proc.c:1923
void RestorePendingSyncs(char *startAddress)
Definition: storage.c:610
const char * authn_id
Definition: libpq-be.h:103
UserAuth auth_method
Definition: libpq-be.h:109
void ExitParallelMode(void)
Definition: xact.c:1050
void EnterParallelMode(void)
Definition: xact.c:1037
void StartTransactionCommand(void)
Definition: xact.c:2955
void StartParallelWorkerTransaction(char *tstatespace)
Definition: xact.c:5533
void SetParallelStartTimestamps(TimestampTz xact_ts, TimestampTz stmt_ts)
Definition: xact.c:845
void EndParallelWorkerTransaction(void)
Definition: xact.c:5558
void CommitTransactionCommand(void)
Definition: xact.c:3053

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, CommitTransactionCommand(), FixedParallelState::current_user_id, CurrentMemoryContext, FixedParallelState::database_id, DatumGetUInt32(), DetachSession(), die, dsm_attach(), dsm_segment_address(), EndParallelWorkerTransaction(), EnterParallelMode(), ereport, errcode(), errmsg(), ERROR, ExitParallelMode(), GetDatabaseEncoding(), hba_authname(), InitializeSystemUser(), InitializingParallelWorker, InvalidateSystemCaches(), FixedParallelState::is_superuser, 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::sec_context, FixedParallelState::serializable_xact_handle, SetClientEncoding(), SetCurrentRoleId(), SetParallelStartTimestamps(), 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 1539 of file parallel.c.

1540 {
1542 
1543  Assert(fps != NULL);
1544  SpinLockAcquire(&fps->mutex);
1545  if (fps->last_xlog_end < last_xlog_end)
1546  fps->last_xlog_end = last_xlog_end;
1547  SpinLockRelease(&fps->mutex);
1548 }
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62

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

1568 {
1572 
1574 }
void * arg
static Pointer DatumGetPointer(Datum X)
Definition: postgres.h:312
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
Definition: procsignal.c:257
@ PROCSIG_PARALLEL_MESSAGE
Definition: procsignal.h:34

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

Referenced by ParallelWorkerMain().

◆ ReinitializeParallelDSM()

void ReinitializeParallelDSM ( ParallelContext pcxt)

Definition at line 488 of file parallel.c.

489 {
490  FixedParallelState *fps;
491 
492  /* Wait for any old workers to exit. */
493  if (pcxt->nworkers_launched > 0)
494  {
497  pcxt->nworkers_launched = 0;
498  if (pcxt->known_attached_workers)
499  {
501  pcxt->known_attached_workers = NULL;
502  pcxt->nknown_attached_workers = 0;
503  }
504  }
505 
506  /* Reset a few bits of fixed parallel state to a clean state. */
507  fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);
508  fps->last_xlog_end = 0;
509 
510  /* Recreate error queues (if they exist). */
511  if (pcxt->nworkers > 0)
512  {
513  char *error_queue_space;
514  int i;
515 
516  error_queue_space =
518  for (i = 0; i < pcxt->nworkers; ++i)
519  {
520  char *start;
521  shm_mq *mq;
522 
523  start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
526  pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
527  }
528  }
529 }
void WaitForParallelWorkersToFinish(ParallelContext *pcxt)
Definition: parallel.c:775

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

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

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

Referenced by parallel_vacuum_process_all_indexes().

◆ WaitForParallelWorkersToAttach()

void WaitForParallelWorkersToAttach ( ParallelContext pcxt)

Definition at line 672 of file parallel.c.

673 {
674  int i;
675 
676  /* Skip this if we have no launched workers. */
677  if (pcxt->nworkers_launched == 0)
678  return;
679 
680  for (;;)
681  {
682  /*
683  * This will process any parallel messages that are pending and it may
684  * also throw an error propagated from a worker.
685  */
687 
688  for (i = 0; i < pcxt->nworkers_launched; ++i)
689  {
690  BgwHandleStatus status;
691  shm_mq *mq;
692  int rc;
693  pid_t pid;
694 
695  if (pcxt->known_attached_workers[i])
696  continue;
697 
698  /*
699  * If error_mqh is NULL, then the worker has already exited
700  * cleanly.
701  */
702  if (pcxt->worker[i].error_mqh == NULL)
703  {
704  pcxt->known_attached_workers[i] = true;
705  ++pcxt->nknown_attached_workers;
706  continue;
707  }
708 
709  status = GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle, &pid);
710  if (status == BGWH_STARTED)
711  {
712  /* Has the worker attached to the error queue? */
713  mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
714  if (shm_mq_get_sender(mq) != NULL)
715  {
716  /* Yes, so it is known to be attached. */
717  pcxt->known_attached_workers[i] = true;
718  ++pcxt->nknown_attached_workers;
719  }
720  }
721  else if (status == BGWH_STOPPED)
722  {
723  /*
724  * If the worker stopped without attaching to the error queue,
725  * throw an error.
726  */
727  mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
728  if (shm_mq_get_sender(mq) == NULL)
729  ereport(ERROR,
730  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
731  errmsg("parallel worker failed to initialize"),
732  errhint("More details may be available in the server log.")));
733 
734  pcxt->known_attached_workers[i] = true;
735  ++pcxt->nknown_attached_workers;
736  }
737  else
738  {
739  /*
740  * Worker not yet started, so we must wait. The postmaster
741  * will notify us if the worker's state changes. Our latch
742  * might also get set for some other reason, but if so we'll
743  * just end up waiting for the same worker again.
744  */
745  rc = WaitLatch(MyLatch,
747  -1, WAIT_EVENT_BGWORKER_STARTUP);
748 
749  if (rc & WL_LATCH_SET)
751  }
752  }
753 
754  /* If all workers are known to have started, we're done. */
755  if (pcxt->nknown_attached_workers >= pcxt->nworkers_launched)
756  {
758  break;
759  }
760  }
761 }
BgwHandleStatus GetBackgroundWorkerPid(BackgroundWorkerHandle *handle, pid_t *pidp)
Definition: bgworker.c:1082
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:1319
void ResetLatch(Latch *latch)
Definition: latch.c:724
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:517
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:132
#define WL_LATCH_SET
Definition: latch.h:127
#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

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

◆ WaitForParallelWorkersToExit()

static void WaitForParallelWorkersToExit ( ParallelContext pcxt)
static

Definition at line 889 of file parallel.c.

890 {
891  int i;
892 
893  /* Wait until the workers actually die. */
894  for (i = 0; i < pcxt->nworkers_launched; ++i)
895  {
896  BgwHandleStatus status;
897 
898  if (pcxt->worker == NULL || pcxt->worker[i].bgwhandle == NULL)
899  continue;
900 
902 
903  /*
904  * If the postmaster kicked the bucket, we have no chance of cleaning
905  * up safely -- we won't be able to tell when our workers are actually
906  * dead. This doesn't necessitate a PANIC since they will all abort
907  * eventually, but we can't safely continue this session.
908  */
909  if (status == BGWH_POSTMASTER_DIED)
910  ereport(FATAL,
911  (errcode(ERRCODE_ADMIN_SHUTDOWN),
912  errmsg("postmaster exited during a parallel transaction")));
913 
914  /* Release memory. */
915  pfree(pcxt->worker[i].bgwhandle);
916  pcxt->worker[i].bgwhandle = NULL;
917  }
918 }
BgwHandleStatus WaitForBackgroundWorkerShutdown(BackgroundWorkerHandle *handle)
Definition: bgworker.c:1182
@ 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 775 of file parallel.c.

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

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

Variable Documentation

◆ fn_addr

Definition at line 136 of file parallel.c.

Referenced by LookupParallelWorkerFunction().

◆ fn_name

const char* fn_name

Definition at line 135 of file parallel.c.

Referenced by LookupParallelWorkerFunction().

◆ InitializingParallelWorker

bool InitializingParallelWorker = false

◆ 

const { ... } InternalParallelWorkers[]
Initial value:
=
{
{
"ParallelQueryMain", ParallelQueryMain
},
{
"_bt_parallel_build_main", _bt_parallel_build_main
},
{
"_brin_parallel_build_main", _brin_parallel_build_main
},
{
"parallel_vacuum_main", parallel_vacuum_main
}
}
void _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
Definition: brin.c:2785
void ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
void _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
Definition: nbtsort.c:1740
void parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)

Referenced by LookupParallelWorkerFunction().

◆ MyFixedParallelState

FixedParallelState* MyFixedParallelState
static

Definition at line 121 of file parallel.c.

Referenced by ParallelWorkerMain(), and ParallelWorkerReportLastRecEnd().

◆ ParallelLeaderPid

pid_t ParallelLeaderPid
static

Definition at line 127 of file parallel.c.

Referenced by ParallelWorkerMain(), and ParallelWorkerShutdown().

◆ ParallelMessagePending

volatile sig_atomic_t ParallelMessagePending = false

◆ ParallelWorkerNumber

◆ pcxt_list