PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
test_shm_mq.h File Reference
#include "storage/dsm.h"
#include "storage/shm_mq.h"
#include "storage/spin.h"
Include dependency graph for test_shm_mq.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  test_shm_mq_header
 

Macros

#define PG_TEST_SHM_MQ_MAGIC   0x79fb2447
 

Functions

void test_shm_mq_setup (int64 queue_size, int32 nworkers, dsm_segment **seg, shm_mq_handle **output, shm_mq_handle **input)
 
void test_shm_mq_main (Datum) pg_attribute_noreturn()
 

Macro Definition Documentation

#define PG_TEST_SHM_MQ_MAGIC   0x79fb2447

Definition at line 22 of file test_shm_mq.h.

Referenced by setup_dynamic_shared_memory(), and test_shm_mq_main().

Function Documentation

void test_shm_mq_main ( Datum  )

Definition at line 48 of file worker.c.

References attach_to_queues(), BackendPidGetProc(), BackgroundWorkerUnblockSignals(), BackgroundWorker::bgw_notify_pid, copy_messages(), CurrentResourceOwner, DatumGetInt32, DEBUG1, dsm_attach(), dsm_detach(), dsm_segment_address(), elog, ereport, errcode(), errmsg(), ERROR, handle_sigterm(), test_shm_mq_header::mutex, MyBgworkerEntry, PG_TEST_SHM_MQ_MAGIC, pqsignal(), proc_exit(), PGPROC::procLatch, ResourceOwnerCreate(), SetLatch(), shm_toc_attach(), shm_toc_lookup(), SpinLockAcquire, SpinLockRelease, test_shm_mq_header::workers_attached, test_shm_mq_header::workers_ready, and test_shm_mq_header::workers_total.

49 {
50  dsm_segment *seg;
51  shm_toc *toc;
52  shm_mq_handle *inqh;
53  shm_mq_handle *outqh;
54  volatile test_shm_mq_header *hdr;
55  int myworkernumber;
56  PGPROC *registrant;
57 
58  /*
59  * Establish signal handlers.
60  *
61  * We want CHECK_FOR_INTERRUPTS() to kill off this worker process just as
62  * it would a normal user backend. To make that happen, we establish a
63  * signal handler that is a stripped-down version of die().
64  */
65  pqsignal(SIGTERM, handle_sigterm);
67 
68  /*
69  * Connect to the dynamic shared memory segment.
70  *
71  * The backend that registered this worker passed us the ID of a shared
72  * memory segment to which we must attach for further instructions. In
73  * order to attach to dynamic shared memory, we need a resource owner.
74  * Once we've mapped the segment in our address space, attach to the table
75  * of contents so we can locate the various data structures we'll need to
76  * find within the segment.
77  */
78  CurrentResourceOwner = ResourceOwnerCreate(NULL, "test_shm_mq worker");
79  seg = dsm_attach(DatumGetInt32(main_arg));
80  if (seg == NULL)
81  ereport(ERROR,
82  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
83  errmsg("unable to map dynamic shared memory segment")));
85  if (toc == NULL)
86  ereport(ERROR,
87  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
88  errmsg("bad magic number in dynamic shared memory segment")));
89 
90  /*
91  * Acquire a worker number.
92  *
93  * By convention, the process registering this background worker should
94  * have stored the control structure at key 0. We look up that key to
95  * find it. Our worker number gives our identity: there may be just one
96  * worker involved in this parallel operation, or there may be many.
97  */
98  hdr = shm_toc_lookup(toc, 0, false);
99  SpinLockAcquire(&hdr->mutex);
100  myworkernumber = ++hdr->workers_attached;
101  SpinLockRelease(&hdr->mutex);
102  if (myworkernumber > hdr->workers_total)
103  ereport(ERROR,
104  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
105  errmsg("too many message queue testing workers already")));
106 
107  /*
108  * Attach to the appropriate message queues.
109  */
110  attach_to_queues(seg, toc, myworkernumber, &inqh, &outqh);
111 
112  /*
113  * Indicate that we're fully initialized and ready to begin the main part
114  * of the parallel operation.
115  *
116  * Once we signal that we're ready, the user backend is entitled to assume
117  * that our on_dsm_detach callbacks will fire before we disconnect from
118  * the shared memory segment and exit. Generally, that means we must have
119  * attached to all relevant dynamic shared memory data structures by now.
120  */
121  SpinLockAcquire(&hdr->mutex);
122  ++hdr->workers_ready;
123  SpinLockRelease(&hdr->mutex);
125  if (registrant == NULL)
126  {
127  elog(DEBUG1, "registrant backend has exited prematurely");
128  proc_exit(1);
129  }
130  SetLatch(&registrant->procLatch);
131 
132  /* Do the work. */
133  copy_messages(inqh, outqh);
134 
135  /*
136  * We're done. Explicitly detach the shared memory segment so that we
137  * don't get a resource leak warning at commit time. This will fire any
138  * on_dsm_detach callbacks we've registered, as well. Once that's done,
139  * we can go ahead and exit.
140  */
141  dsm_detach(seg);
142  proc_exit(1);
143 }
#define DEBUG1
Definition: elog.h:25
#define DatumGetInt32(X)
Definition: postgres.h:478
PGPROC * BackendPidGetProc(int pid)
Definition: procarray.c:2346
ResourceOwner CurrentResourceOwner
Definition: resowner.c:138
dsm_segment * dsm_attach(dsm_handle h)
Definition: dsm.c:561
void proc_exit(int code)
Definition: ipc.c:99
int errcode(int sqlerrcode)
Definition: elog.c:575
BackgroundWorker * MyBgworkerEntry
Definition: postmaster.c:189
Latch procLatch
Definition: proc.h:104
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERROR
Definition: elog.h:43
static void copy_messages(shm_mq_handle *inqh, shm_mq_handle *outqh)
Definition: worker.c:177
#define ereport(elevel, rest)
Definition: elog.h:122
#define SpinLockRelease(lock)
Definition: spin.h:64
shm_toc * shm_toc_attach(uint64 magic, void *address)
Definition: shm_toc.c:64
pqsigfunc pqsignal(int signum, pqsigfunc handler)
Definition: signal.c:168
void SetLatch(volatile Latch *latch)
Definition: latch.c:414
void * dsm_segment_address(dsm_segment *seg)
Definition: dsm.c:1000
static void handle_sigterm(SIGNAL_ARGS)
Definition: worker.c:206
void dsm_detach(dsm_segment *seg)
Definition: dsm.c:726
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t bgw_notify_pid
Definition: bgworker.h:99
#define elog
Definition: elog.h:219
Definition: proc.h:95
static void attach_to_queues(dsm_segment *seg, shm_toc *toc, int myworkernumber, shm_mq_handle **inqhp, shm_mq_handle **outqhp)
Definition: worker.c:155
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
#define PG_TEST_SHM_MQ_MAGIC
Definition: test_shm_mq.h:22
ResourceOwner ResourceOwnerCreate(ResourceOwner parent, const char *name)
Definition: resowner.c:416
void BackgroundWorkerUnblockSignals(void)
Definition: postmaster.c:5592
void test_shm_mq_setup ( int64  queue_size,
int32  nworkers,
dsm_segment **  seg,
shm_mq_handle **  output,
shm_mq_handle **  input 
)

Definition at line 49 of file setup.c.

References cancel_on_dsm_detach(), cleanup_background_workers(), worker_state::handle, 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().

51 {
52  dsm_segment *seg;
53  test_shm_mq_header *hdr;
54  shm_mq *outq = NULL; /* placate compiler */
55  shm_mq *inq = NULL; /* placate compiler */
56  worker_state *wstate;
57 
58  /* Set up a dynamic shared memory segment. */
59  setup_dynamic_shared_memory(queue_size, nworkers, &seg, &hdr, &outq, &inq);
60  *segp = seg;
61 
62  /* Register background workers. */
63  wstate = setup_background_workers(nworkers, seg);
64 
65  /* Attach the queues. */
66  *output = shm_mq_attach(outq, seg, wstate->handle[0]);
67  *input = shm_mq_attach(inq, seg, wstate->handle[nworkers - 1]);
68 
69  /* Wait for workers to become ready. */
71 
72  /*
73  * Once we reach this point, all workers are ready. We no longer need to
74  * kill them if we die; they'll die on their own as the message queues
75  * shut down.
76  */
78  PointerGetDatum(wstate));
79  pfree(wstate);
80 }
#define PointerGetDatum(X)
Definition: postgres.h:562
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:90
static void wait_for_workers_to_become_ready(worker_state *wstate, volatile test_shm_mq_header *hdr)
Definition: setup.c:256
void pfree(void *pointer)
Definition: mcxt.c:949
static worker_state * setup_background_workers(int nworkers, dsm_segment *seg)
Definition: setup.c:173
BackgroundWorkerHandle * handle[FLEXIBLE_ARRAY_MEMBER]
Definition: setup.c:30
static void cleanup_background_workers(dsm_segment *seg, Datum arg)
Definition: setup.c:244
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:287
Definition: shm_mq.c:69
void cancel_on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1052