worker.c File Reference

#include "postgres.h"
#include "miscadmin.h"
#include "storage/ipc.h"
#include "storage/procarray.h"
#include "storage/shm_mq.h"
#include "storage/shm_toc.h"
#include "tcop/tcopprot.h"
#include "test_shm_mq.h"

Include dependency graph for worker.c:

Go to the source code of this file.

Functions
static void	attach_to_queues (dsm_segment *seg, shm_toc *toc, int myworkernumber, shm_mq_handle **inqhp, shm_mq_handle **outqhp)
static void	copy_messages (shm_mq_handle *inqh, shm_mq_handle *outqh)
void	test_shm_mq_main (Datum main_arg)

Function Documentation

attach_to_queues()

static void attach_to_queues ( dsm_segment *  seg, shm_toc *  toc, int  myworkernumber, shm_mq_handle **  inqhp, shm_mq_handle **  outqhp  )

static

Definition at line 154 of file worker.c.

{
    shm_mq     *inq;
    shm_mq     *outq;
 
    inq = shm_toc_lookup(toc, myworkernumber, false);
    shm_mq_set_receiver(inq, MyProc);
    *inqhp = shm_mq_attach(inq, seg, NULL);
    outq = shm_toc_lookup(toc, myworkernumber + 1, false);
    shm_mq_set_sender(outq, MyProc);
    *outqhp = shm_mq_attach(outq, seg, NULL);
}

References MyProc, shm_mq_attach(), shm_mq_set_receiver(), shm_mq_set_sender(), and shm_toc_lookup().

Referenced by test_shm_mq_main().

copy_messages()

static void copy_messages ( shm_mq_handle *  inqh, shm_mq_handle *  outqh  )

static

Definition at line 176 of file worker.c.

{
    Size        len;
    void       *data;
    shm_mq_result res;
 
    for (;;)
    {
        /* Notice any interrupts that have occurred. */
        CHECK_FOR_INTERRUPTS();
 
        /* Receive a message. */
        res = shm_mq_receive(inqh, &len, &data, false);
        if (res != SHM_MQ_SUCCESS)
            break;
 
        /* Send it back out. */
        res = shm_mq_send(outqh, len, data, false, true);
        if (res != SHM_MQ_SUCCESS)
            break;
    }
}

References CHECK_FOR_INTERRUPTS, data, len, res, shm_mq_receive(), shm_mq_send(), and SHM_MQ_SUCCESS.

Referenced by test_shm_mq_main().

test_shm_mq_main()

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