PostgreSQL Source Code  git master
setup.c File Reference
#include "postgres.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "storage/procsignal.h"
#include "storage/shm_toc.h"
#include "test_shm_mq.h"
#include "utils/memutils.h"
Include dependency graph for setup.c:

Go to the source code of this file.

Data Structures

struct  worker_state
 

Functions

static void setup_dynamic_shared_memory (int64 queue_size, int nworkers, dsm_segment **segp, test_shm_mq_header **hdrp, shm_mq **outp, shm_mq **inp)
 
static worker_statesetup_background_workers (int nworkers, dsm_segment *seg)
 
static void cleanup_background_workers (dsm_segment *seg, Datum arg)
 
static void wait_for_workers_to_become_ready (worker_state *wstate, volatile test_shm_mq_header *hdr)
 
static bool check_worker_status (worker_state *wstate)
 
void test_shm_mq_setup (int64 queue_size, int32 nworkers, dsm_segment **segp, shm_mq_handle **output, shm_mq_handle **input)
 

Variables

static uint32 we_bgworker_startup = 0
 

Function Documentation

◆ check_worker_status()

static bool check_worker_status ( worker_state wstate)
static

Definition at line 306 of file setup.c.

307 {
308  int n;
309 
310  /* If any workers (or the postmaster) have died, we have failed. */
311  for (n = 0; n < wstate->nworkers; ++n)
312  {
313  BgwHandleStatus status;
314  pid_t pid;
315 
316  status = GetBackgroundWorkerPid(wstate->handle[n], &pid);
317  if (status == BGWH_STOPPED || status == BGWH_POSTMASTER_DIED)
318  return false;
319  }
320 
321  /* Otherwise, things still look OK. */
322  return true;
323 }
BgwHandleStatus GetBackgroundWorkerPid(BackgroundWorkerHandle *handle, pid_t *pidp)
Definition: bgworker.c:1072
BgwHandleStatus
Definition: bgworker.h:104
@ BGWH_POSTMASTER_DIED
Definition: bgworker.h:108
@ BGWH_STOPPED
Definition: bgworker.h:107
int nworkers
Definition: setup.c:28
BackgroundWorkerHandle * handle[FLEXIBLE_ARRAY_MEMBER]
Definition: setup.c:29

References BGWH_POSTMASTER_DIED, BGWH_STOPPED, GetBackgroundWorkerPid(), worker_state::handle, and worker_state::nworkers.

Referenced by wait_for_workers_to_become_ready().

◆ cleanup_background_workers()

static void cleanup_background_workers ( dsm_segment seg,
Datum  arg 
)
static

Definition at line 246 of file setup.c.

247 {
249 
250  while (wstate->nworkers > 0)
251  {
252  --wstate->nworkers;
253  TerminateBackgroundWorker(wstate->handle[wstate->nworkers]);
254  }
255 }
void TerminateBackgroundWorker(BackgroundWorkerHandle *handle)
Definition: bgworker.c:1211
void * arg
static Pointer DatumGetPointer(Datum X)
Definition: postgres.h:312

References arg, DatumGetPointer(), worker_state::handle, worker_state::nworkers, and TerminateBackgroundWorker().

Referenced by setup_background_workers(), and test_shm_mq_setup().

◆ setup_background_workers()

static worker_state * setup_background_workers ( int  nworkers,
dsm_segment seg 
)
static

Definition at line 175 of file setup.c.

176 {
177  MemoryContext oldcontext;
178  BackgroundWorker worker;
179  worker_state *wstate;
180  int i;
181 
182  /*
183  * We need the worker_state object and the background worker handles to
184  * which it points to be allocated in CurTransactionContext rather than
185  * ExprContext; otherwise, they'll be destroyed before the on_dsm_detach
186  * hooks run.
187  */
189 
190  /* Create worker state object. */
192  offsetof(worker_state, handle) +
193  sizeof(BackgroundWorkerHandle *) * nworkers);
194  wstate->nworkers = 0;
195 
196  /*
197  * Arrange to kill all the workers if we abort before all workers are
198  * finished hooking themselves up to the dynamic shared memory segment.
199  *
200  * If we die after all the workers have finished hooking themselves up to
201  * the dynamic shared memory segment, we'll mark the two queues to which
202  * we're directly connected as detached, and the worker(s) connected to
203  * those queues will exit, marking any other queues to which they are
204  * connected as detached. This will cause any as-yet-unaware workers
205  * connected to those queues to exit in their turn, and so on, until
206  * everybody exits.
207  *
208  * But suppose the workers which are supposed to connect to the queues to
209  * which we're directly attached exit due to some error before they
210  * actually attach the queues. The remaining workers will have no way of
211  * knowing this. From their perspective, they're still waiting for those
212  * workers to start, when in fact they've already died.
213  */
215  PointerGetDatum(wstate));
216 
217  /* Configure a worker. */
218  memset(&worker, 0, sizeof(worker));
222  sprintf(worker.bgw_library_name, "test_shm_mq");
223  sprintf(worker.bgw_function_name, "test_shm_mq_main");
224  snprintf(worker.bgw_type, BGW_MAXLEN, "test_shm_mq");
226  /* set bgw_notify_pid, so we can detect if the worker stops */
227  worker.bgw_notify_pid = MyProcPid;
228 
229  /* Register the workers. */
230  for (i = 0; i < nworkers; ++i)
231  {
232  if (!RegisterDynamicBackgroundWorker(&worker, &wstate->handle[i]))
233  ereport(ERROR,
234  (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
235  errmsg("could not register background process"),
236  errhint("You may need to increase max_worker_processes.")));
237  ++wstate->nworkers;
238  }
239 
240  /* All done. */
241  MemoryContextSwitchTo(oldcontext);
242  return wstate;
243 }
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:960
#define BGW_NEVER_RESTART
Definition: bgworker.h:85
@ BgWorkerStart_ConsistentState
Definition: bgworker.h:80
#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:1124
void on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1133
int errhint(const char *fmt,...)
Definition: elog.c:1322
int errcode(int sqlerrcode)
Definition: elog.c:860
int errmsg(const char *fmt,...)
Definition: elog.c:1075
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
int MyProcPid
Definition: globals.c:45
int i
Definition: isn.c:73
MemoryContext TopTransactionContext
Definition: mcxt.c:146
MemoryContext CurTransactionContext
Definition: mcxt.c:147
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:1034
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
#define sprintf
Definition: port.h:240
#define snprintf
Definition: port.h:238
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
static Datum UInt32GetDatum(uint32 X)
Definition: postgres.h:232
static void cleanup_background_workers(dsm_segment *seg, Datum arg)
Definition: setup.c:246
char bgw_function_name[BGW_MAXLEN]
Definition: bgworker.h:97
Datum bgw_main_arg
Definition: bgworker.h:98
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
pid_t bgw_notify_pid
Definition: bgworker.h:100
char bgw_library_name[MAXPGPATH]
Definition: bgworker.h:96

References BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BGW_NEVER_RESTART, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, BGWORKER_SHMEM_ACCESS, BgWorkerStart_ConsistentState, cleanup_background_workers(), CurTransactionContext, dsm_segment_handle(), ereport, errcode(), errhint(), errmsg(), ERROR, worker_state::handle, i, MemoryContextAlloc(), MemoryContextSwitchTo(), MyProcPid, worker_state::nworkers, on_dsm_detach(), PointerGetDatum(), RegisterDynamicBackgroundWorker(), snprintf, sprintf, TopTransactionContext, and UInt32GetDatum().

Referenced by test_shm_mq_setup().

◆ setup_dynamic_shared_memory()

static void setup_dynamic_shared_memory ( int64  queue_size,
int  nworkers,
dsm_segment **  segp,
test_shm_mq_header **  hdrp,
shm_mq **  outp,
shm_mq **  inp 
)
static

Definition at line 92 of file setup.c.

95 {
97  int i;
98  Size segsize;
99  dsm_segment *seg;
100  shm_toc *toc;
101  test_shm_mq_header *hdr;
102 
103  /* Ensure a valid queue size. */
104  if (queue_size < 0 || ((uint64) queue_size) < shm_mq_minimum_size)
105  ereport(ERROR,
106  (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
107  errmsg("queue size must be at least %zu bytes",
109  if (queue_size != ((Size) queue_size))
110  ereport(ERROR,
111  (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
112  errmsg("queue size overflows size_t")));
113 
114  /*
115  * Estimate how much shared memory we need.
116  *
117  * Because the TOC machinery may choose to insert padding of oddly-sized
118  * requests, we must estimate each chunk separately.
119  *
120  * We need one key to register the location of the header, and we need
121  * nworkers + 1 keys to track the locations of the message queues.
122  */
125  for (i = 0; i <= nworkers; ++i)
126  shm_toc_estimate_chunk(&e, (Size) queue_size);
127  shm_toc_estimate_keys(&e, 2 + nworkers);
128  segsize = shm_toc_estimate(&e);
129 
130  /* Create the shared memory segment and establish a table of contents. */
131  seg = dsm_create(shm_toc_estimate(&e), 0);
133  segsize);
134 
135  /* Set up the header region. */
136  hdr = shm_toc_allocate(toc, sizeof(test_shm_mq_header));
137  SpinLockInit(&hdr->mutex);
138  hdr->workers_total = nworkers;
139  hdr->workers_attached = 0;
140  hdr->workers_ready = 0;
141  shm_toc_insert(toc, 0, hdr);
142 
143  /* Set up one message queue per worker, plus one. */
144  for (i = 0; i <= nworkers; ++i)
145  {
146  shm_mq *mq;
147 
148  mq = shm_mq_create(shm_toc_allocate(toc, (Size) queue_size),
149  (Size) queue_size);
150  shm_toc_insert(toc, i + 1, mq);
151 
152  if (i == 0)
153  {
154  /* We send messages to the first queue. */
156  *outp = mq;
157  }
158  if (i == nworkers)
159  {
160  /* We receive messages from the last queue. */
162  *inp = mq;
163  }
164  }
165 
166  /* Return results to caller. */
167  *segp = seg;
168  *hdrp = hdr;
169 }
size_t Size
Definition: c.h:594
void * dsm_segment_address(dsm_segment *seg)
Definition: dsm.c:1096
dsm_segment * dsm_create(Size size, int flags)
Definition: dsm.c:517
e
Definition: preproc-init.c:82
void shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:225
shm_mq * shm_mq_create(void *address, Size size)
Definition: shm_mq.c:178
void shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:207
const Size shm_mq_minimum_size
Definition: shm_mq.c:169
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_initialize_estimator(e)
Definition: shm_toc.h:49
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
#define SpinLockInit(lock)
Definition: spin.h:60
PGPROC * MyProc
Definition: proc.c:68
Definition: shm_mq.c:73
#define PG_TEST_SHM_MQ_MAGIC
Definition: test_shm_mq.h:22

References dsm_create(), dsm_segment_address(), ereport, errcode(), errmsg(), ERROR, i, test_shm_mq_header::mutex, MyProc, PG_TEST_SHM_MQ_MAGIC, shm_mq_create(), shm_mq_minimum_size, shm_mq_set_receiver(), shm_mq_set_sender(), shm_toc_allocate(), shm_toc_create(), shm_toc_estimate(), shm_toc_estimate_chunk, shm_toc_estimate_keys, shm_toc_initialize_estimator, shm_toc_insert(), SpinLockInit, test_shm_mq_header::workers_attached, test_shm_mq_header::workers_ready, and test_shm_mq_header::workers_total.

Referenced by test_shm_mq_setup().

◆ test_shm_mq_setup()

void test_shm_mq_setup ( int64  queue_size,
int32  nworkers,
dsm_segment **  segp,
shm_mq_handle **  output,
shm_mq_handle **  input 
)

Definition at line 51 of file setup.c.

53 {
54  dsm_segment *seg;
55  test_shm_mq_header *hdr;
56  shm_mq *outq = NULL; /* placate compiler */
57  shm_mq *inq = NULL; /* placate compiler */
58  worker_state *wstate;
59 
60  /* Set up a dynamic shared memory segment. */
61  setup_dynamic_shared_memory(queue_size, nworkers, &seg, &hdr, &outq, &inq);
62  *segp = seg;
63 
64  /* Register background workers. */
65  wstate = setup_background_workers(nworkers, seg);
66 
67  /* Attach the queues. */
68  *output = shm_mq_attach(outq, seg, wstate->handle[0]);
69  *input = shm_mq_attach(inq, seg, wstate->handle[nworkers - 1]);
70 
71  /* Wait for workers to become ready. */
73 
74  /*
75  * Once we reach this point, all workers are ready. We no longer need to
76  * kill them if we die; they'll die on their own as the message queues
77  * shut down.
78  */
80  PointerGetDatum(wstate));
81  pfree(wstate);
82 }
void cancel_on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1148
FILE * input
FILE * output
void pfree(void *pointer)
Definition: mcxt.c:1431
static worker_state * setup_background_workers(int nworkers, dsm_segment *seg)
Definition: setup.c:175
static void wait_for_workers_to_become_ready(worker_state *wstate, volatile test_shm_mq_header *hdr)
Definition: setup.c:258
static void setup_dynamic_shared_memory(int64 queue_size, int nworkers, dsm_segment **segp, test_shm_mq_header **hdrp, shm_mq **outp, shm_mq **inp)
Definition: setup.c:92
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:291

References cancel_on_dsm_detach(), cleanup_background_workers(), worker_state::handle, input, output, pfree(), PointerGetDatum(), setup_background_workers(), setup_dynamic_shared_memory(), shm_mq_attach(), and wait_for_workers_to_become_ready().

Referenced by test_shm_mq(), and test_shm_mq_pipelined().

◆ wait_for_workers_to_become_ready()

static void wait_for_workers_to_become_ready ( worker_state wstate,
volatile test_shm_mq_header hdr 
)
static

Definition at line 258 of file setup.c.

260 {
261  bool result = false;
262 
263  for (;;)
264  {
265  int workers_ready;
266 
267  /* If all the workers are ready, we have succeeded. */
268  SpinLockAcquire(&hdr->mutex);
269  workers_ready = hdr->workers_ready;
270  SpinLockRelease(&hdr->mutex);
271  if (workers_ready >= wstate->nworkers)
272  {
273  result = true;
274  break;
275  }
276 
277  /* If any workers (or the postmaster) have died, we have failed. */
278  if (!check_worker_status(wstate))
279  {
280  result = false;
281  break;
282  }
283 
284  /* first time, allocate or get the custom wait event */
285  if (we_bgworker_startup == 0)
286  we_bgworker_startup = WaitEventExtensionNew("TestShmMqBgWorkerStartup");
287 
288  /* Wait to be signaled. */
291 
292  /* Reset the latch so we don't spin. */
294 
295  /* An interrupt may have occurred while we were waiting. */
297  }
298 
299  if (!result)
300  ereport(ERROR,
301  (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
302  errmsg("one or more background workers failed to start")));
303 }
struct Latch * MyLatch
Definition: globals.c:59
void ResetLatch(Latch *latch)
Definition: latch.c:725
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:518
#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
static uint32 we_bgworker_startup
Definition: setup.c:44
static bool check_worker_status(worker_state *wstate)
Definition: setup.c:306
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
uint32 WaitEventExtensionNew(const char *wait_event_name)
Definition: wait_event.c:164

References CHECK_FOR_INTERRUPTS, check_worker_status(), ereport, errcode(), errmsg(), ERROR, test_shm_mq_header::mutex, MyLatch, worker_state::nworkers, ResetLatch(), SpinLockAcquire, SpinLockRelease, WaitEventExtensionNew(), WaitLatch(), we_bgworker_startup, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and test_shm_mq_header::workers_ready.

Referenced by test_shm_mq_setup().

Variable Documentation

◆ we_bgworker_startup

uint32 we_bgworker_startup = 0
static

Definition at line 44 of file setup.c.

Referenced by wait_for_workers_to_become_ready().