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

)

Definition at line 47 of file worker.c.

{
    dsm_segment *seg;
    shm_toc    *toc;
    shm_mq_handle *inqh;
    shm_mq_handle *outqh;
    volatile test_shm_mq_header *hdr;
    int         myworkernumber;
    PGPROC     *registrant;
 
    /*
     * Establish signal handlers.
     *
     * We want CHECK_FOR_INTERRUPTS() to kill off this worker process just as
     * it would a normal user backend.  To make that happen, we use die().
     */
    pqsignal(SIGTERM, die);
    BackgroundWorkerUnblockSignals();
 
    /*
     * Connect to the dynamic shared memory segment.
     *
     * The backend that registered this worker passed us the ID of a shared
     * memory segment to which we must attach for further instructions.  Once
     * we've mapped the segment in our address space, attach to the table of
     * contents so we can locate the various data structures we'll need to
     * find within the segment.
     *
     * Note: at this point, we have not created any ResourceOwner in this
     * process.  This will result in our DSM mapping surviving until process
     * exit, which is fine.  If there were a ResourceOwner, it would acquire
     * ownership of the mapping, but we have no need for that.
     */
    seg = dsm_attach(DatumGetUInt32(main_arg));
    if (seg == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("unable to map dynamic shared memory segment")));
    toc = shm_toc_attach(PG_TEST_SHM_MQ_MAGIC, dsm_segment_address(seg));
    if (toc == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("bad magic number in dynamic shared memory segment")));
 
    /*
     * Acquire a worker number.
     *
     * By convention, the process registering this background worker should
     * have stored the control structure at key 0.  We look up that key to
     * find it.  Our worker number gives our identity: there may be just one
     * worker involved in this parallel operation, or there may be many.
     */
    hdr = shm_toc_lookup(toc, 0, false);
    SpinLockAcquire(&hdr->mutex);
    myworkernumber = ++hdr->workers_attached;
    SpinLockRelease(&hdr->mutex);
    if (myworkernumber > hdr->workers_total)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("too many message queue testing workers already")));
 
    /*
     * Attach to the appropriate message queues.
     */
    attach_to_queues(seg, toc, myworkernumber, &inqh, &outqh);
 
    /*
     * Indicate that we're fully initialized and ready to begin the main part
     * of the parallel operation.
     *
     * Once we signal that we're ready, the user backend is entitled to assume
     * that our on_dsm_detach callbacks will fire before we disconnect from
     * the shared memory segment and exit.  Generally, that means we must have
     * attached to all relevant dynamic shared memory data structures by now.
     */
    SpinLockAcquire(&hdr->mutex);
    ++hdr->workers_ready;
    SpinLockRelease(&hdr->mutex);
    registrant = BackendPidGetProc(MyBgworkerEntry->bgw_notify_pid);
    if (registrant == NULL)
    {
        elog(DEBUG1, "registrant backend has exited prematurely");
        proc_exit(1);
    }
    SetLatch(&registrant->procLatch);
 
    /* Do the work. */
    copy_messages(inqh, outqh);
 
    /*
     * We're done.  For cleanliness, explicitly detach from the shared memory
     * segment (that would happen anyway during process exit, though).
     */
    dsm_detach(seg);
    proc_exit(1);
}

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

Definition at line 50 of file setup.c.

{
    dsm_segment *seg;
    test_shm_mq_header *hdr;
    shm_mq     *outq = NULL;    /* placate compiler */
    shm_mq     *inq = NULL;     /* placate compiler */
    worker_state *wstate;
 
    /* Set up a dynamic shared memory segment. */
    setup_dynamic_shared_memory(queue_size, nworkers, &seg, &hdr, &outq, &inq);
    *segp = seg;
 
    /* Register background workers. */
    wstate = setup_background_workers(nworkers, seg);
 
    /* Attach the queues. */
    *output = shm_mq_attach(outq, seg, wstate->handle[0]);
    *input = shm_mq_attach(inq, seg, wstate->handle[nworkers - 1]);
 
    /* Wait for workers to become ready. */
    wait_for_workers_to_become_ready(wstate, hdr);
 
    /*
     * Once we reach this point, all workers are ready.  We no longer need to
     * kill them if we die; they'll die on their own as the message queues
     * shut down.
     */
    cancel_on_dsm_detach(seg, cleanup_background_workers,
                         PointerGetDatum(wstate));
    pfree(wstate);
}

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