PostgreSQL Source Code git master
Loading...
Searching...
No Matches
Data Structures | Macros | Functions
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 **segp, shm_mq_handle **output, shm_mq_handle **input)
 
pg_noreturn PGDLLEXPORT void test_shm_mq_main (Datum)
 

Macro Definition Documentation

◆ PG_TEST_SHM_MQ_MAGIC

#define PG_TEST_SHM_MQ_MAGIC   0x79fb2447

Definition at line 22 of file test_shm_mq.h.

Function Documentation

◆ test_shm_mq_main()

pg_noreturn PGDLLEXPORT void test_shm_mq_main ( Datum  main_arg)
extern

Definition at line 49 of file worker.c.

50{
51 dsm_segment *seg;
52 shm_toc *toc;
53 shm_mq_handle *inqh;
54 shm_mq_handle *outqh;
55 volatile test_shm_mq_header *hdr;
56 int myworkernumber;
57 PGPROC *registrant;
58
59 /* Unblock signals. The standard signal handlers are OK for us. */
60 BackgroundWorkerUnblockSignals();
61
62 /*
63 * Connect to the dynamic shared memory segment.
64 *
65 * The backend that registered this worker passed us the ID of a shared
66 * memory segment to which we must attach for further instructions. Once
67 * we've mapped the segment in our address space, attach to the table of
68 * contents so we can locate the various data structures we'll need to
69 * find within the segment.
70 *
71 * Note: at this point, we have not created any ResourceOwner in this
72 * process. This will result in our DSM mapping surviving until process
73 * exit, which is fine. If there were a ResourceOwner, it would acquire
74 * ownership of the mapping, but we have no need for that.
75 */
76 seg = dsm_attach(DatumGetUInt32(main_arg));
77 if (seg == NULL)
78 ereport(ERROR,
79 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
80 errmsg("unable to map dynamic shared memory segment")));
81 toc = shm_toc_attach(PG_TEST_SHM_MQ_MAGIC, dsm_segment_address(seg));
82 if (toc == NULL)
83 ereport(ERROR,
84 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
85 errmsg("bad magic number in dynamic shared memory segment")));
86
87 /*
88 * Acquire a worker number.
89 *
90 * By convention, the process registering this background worker should
91 * have stored the control structure at key 0. We look up that key to
92 * find it. Our worker number gives our identity: there may be just one
93 * worker involved in this parallel operation, or there may be many.
94 */
95 hdr = shm_toc_lookup(toc, 0, false);
96 SpinLockAcquire(&hdr->mutex);
97 myworkernumber = ++hdr->workers_attached;
98 SpinLockRelease(&hdr->mutex);
99 if (myworkernumber > hdr->workers_total)
100 ereport(ERROR,
101 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
102 errmsg("too many message queue testing workers already")));
103
104 /*
105 * Attach to the appropriate message queues.
106 */
107 attach_to_queues(seg, toc, myworkernumber, &inqh, &outqh);
108
109 /*
110 * Indicate that we're fully initialized and ready to begin the main part
111 * of the parallel operation.
112 *
113 * Once we signal that we're ready, the user backend is entitled to assume
114 * that our on_dsm_detach callbacks will fire before we disconnect from
115 * the shared memory segment and exit. Generally, that means we must have
116 * attached to all relevant dynamic shared memory data structures by now.
117 */
118 SpinLockAcquire(&hdr->mutex);
119 ++hdr->workers_ready;
120 SpinLockRelease(&hdr->mutex);
121 registrant = BackendPidGetProc(MyBgworkerEntry->bgw_notify_pid);
122 if (registrant == NULL)
123 {
124 elog(DEBUG1, "registrant backend has exited prematurely");
125 proc_exit(1);
126 }
127 SetLatch(&registrant->procLatch);
128
129 /* Do the work. */
130 copy_messages(inqh, outqh);
131
132 /*
133 * We're done. For cleanliness, explicitly detach from the shared memory
134 * segment (that would happen anyway during process exit, though).
135 */
136 dsm_detach(seg);
137 proc_exit(1);
138}
BackgroundWorkerUnblockSignals
void BackgroundWorkerUnblockSignals(void)
Definition bgworker.c:933
dsm_detach
void dsm_detach(dsm_segment *seg)
Definition dsm.c:803
dsm_segment_address
void * dsm_segment_address(dsm_segment *seg)
Definition dsm.c:1095
dsm_attach
dsm_segment * dsm_attach(dsm_handle h)
Definition dsm.c:665
errcode
int errcode(int sqlerrcode)
Definition elog.c:874
errmsg
int errmsg(const char *fmt,...)
Definition elog.c:1093
DEBUG1
#define DEBUG1
Definition elog.h:30
ERROR
#define ERROR
Definition elog.h:39
elog
#define elog(elevel,...)
Definition elog.h:226
ereport
#define ereport(elevel,...)
Definition elog.h:150
proc_exit
void proc_exit(int code)
Definition ipc.c:105
SetLatch
void SetLatch(Latch *latch)
Definition latch.c:290
DatumGetUInt32
static uint32 DatumGetUInt32(Datum X)
Definition postgres.h:232
MyBgworkerEntry
BackgroundWorker * MyBgworkerEntry
Definition postmaster.c:200
fb
static int fb(int x)
Definition preproc-init.c:92
BackendPidGetProc
PGPROC * BackendPidGetProc(int pid)
Definition procarray.c:3160
shm_toc_lookup
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition shm_toc.c:232
shm_toc_attach
shm_toc * shm_toc_attach(uint64 magic, void *address)
Definition shm_toc.c:64
SpinLockRelease
static void SpinLockRelease(volatile slock_t *lock)
Definition spin.h:62
SpinLockAcquire
static void SpinLockAcquire(volatile slock_t *lock)
Definition spin.h:56
BackgroundWorker::bgw_notify_pid
pid_t bgw_notify_pid
Definition bgworker.h:107
PGPROC
Definition proc.h:176
dsm_segment
Definition dsm.c:67
shm_mq_handle
Definition shm_mq.c:139
shm_toc
Definition shm_toc.c:27
test_shm_mq_header
Definition test_shm_mq.h:30
test_shm_mq_header::workers_total
int workers_total
Definition test_shm_mq.h:32
test_shm_mq_header::mutex
slock_t mutex
Definition test_shm_mq.h:31
test_shm_mq_header::workers_attached
int workers_attached
Definition test_shm_mq.h:33
test_shm_mq_header::workers_ready
int workers_ready
Definition test_shm_mq.h:34
attach_to_queues
static void attach_to_queues(dsm_segment *seg, shm_toc *toc, int myworkernumber, shm_mq_handle **inqhp, shm_mq_handle **outqhp)
Definition worker.c:150
copy_messages
static void copy_messages(shm_mq_handle *inqh, shm_mq_handle *outqh)
Definition worker.c:172
PG_TEST_SHM_MQ_MAGIC
#define PG_TEST_SHM_MQ_MAGIC
Definition test_shm_mq.h:22

References attach_to_queues(), BackendPidGetProc(), BackgroundWorkerUnblockSignals(), BackgroundWorker::bgw_notify_pid, copy_messages(), DatumGetUInt32(), DEBUG1, dsm_attach(), dsm_detach(), dsm_segment_address(), elog, ereport, errcode(), errmsg(), ERROR, fb(), test_shm_mq_header::mutex, MyBgworkerEntry, PG_TEST_SHM_MQ_MAGIC, proc_exit(), 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.

◆ 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 
)
extern

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. */
72 wait_for_workers_to_become_ready(wstate, hdr);
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 */
79 cancel_on_dsm_detach(seg, cleanup_background_workers,
80 PointerGetDatum(wstate));
81 pfree(wstate);
82}
cancel_on_dsm_detach
void cancel_on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition dsm.c:1147
input
FILE * input
output
FILE * output
Definition pg_test_timing.c:235
pfree
void pfree(void *pointer)
Definition mcxt.c:1616
PointerGetDatum
static Datum PointerGetDatum(const void *X)
Definition postgres.h:352
setup_background_workers
static worker_state * setup_background_workers(int nworkers, dsm_segment *seg)
Definition setup.c:175
cleanup_background_workers
static void cleanup_background_workers(dsm_segment *seg, Datum arg)
Definition setup.c:247
wait_for_workers_to_become_ready
static void wait_for_workers_to_become_ready(worker_state *wstate, volatile test_shm_mq_header *hdr)
Definition setup.c:259
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)
Definition setup.c:92
shm_mq_attach
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
Definition shm_mq.c:291
shm_mq
Definition shm_mq.c:73
worker_state
Definition setup.c:27

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