shm_mq.c File Reference

#include "postgres.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "port/pg_bitutils.h"
#include "postmaster/bgworker.h"
#include "storage/shm_mq.h"
#include "storage/spin.h"
#include "utils/memutils.h"

Include dependency graph for shm_mq.c:

Go to the source code of this file.

Data Structures
struct	shm_mq
struct	shm_mq_handle

Macros
#define	MQH_INITIAL_BUFSIZE   8192

Functions
static void	shm_mq_detach_internal (shm_mq *mq)
static shm_mq_result	shm_mq_send_bytes (shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait, Size *bytes_written)
static shm_mq_result	shm_mq_receive_bytes (shm_mq_handle *mqh, Size bytes_needed, bool nowait, Size *nbytesp, void **datap)
static bool	shm_mq_counterparty_gone (shm_mq *mq, BackgroundWorkerHandle *handle)
static bool	shm_mq_wait_internal (shm_mq *mq, PGPROC **ptr, BackgroundWorkerHandle *handle)
static void	shm_mq_inc_bytes_read (shm_mq *mq, Size n)
static void	shm_mq_inc_bytes_written (shm_mq *mq, Size n)
static void	shm_mq_detach_callback (dsm_segment *seg, Datum arg)
shm_mq *	shm_mq_create (void *address, Size size)
void	shm_mq_set_receiver (shm_mq *mq, PGPROC *proc)
void	shm_mq_set_sender (shm_mq *mq, PGPROC *proc)
PGPROC *	shm_mq_get_receiver (shm_mq *mq)
PGPROC *	shm_mq_get_sender (shm_mq *mq)
shm_mq_handle *	shm_mq_attach (shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
void	shm_mq_set_handle (shm_mq_handle *mqh, BackgroundWorkerHandle *handle)
shm_mq_result	shm_mq_send (shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait, bool force_flush)
shm_mq_result	shm_mq_sendv (shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait, bool force_flush)
shm_mq_result	shm_mq_receive (shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
shm_mq_result	shm_mq_wait_for_attach (shm_mq_handle *mqh)
void	shm_mq_detach (shm_mq_handle *mqh)
shm_mq *	shm_mq_get_queue (shm_mq_handle *mqh)

Variables
const Size	shm_mq_minimum_size

Macro Definition Documentation

MQH_INITIAL_BUFSIZE

#define MQH_INITIAL_BUFSIZE   8192

Definition at line 171 of file shm_mq.c.

Function Documentation

shm_mq_attach()

shm_mq_handle * shm_mq_attach	(	shm_mq *	mq,
		dsm_segment *	seg,
		BackgroundWorkerHandle *	handle
	)

Definition at line 290 of file shm_mq.c.

{
    shm_mq_handle *mqh = palloc(sizeof(shm_mq_handle));
 
    Assert(mq->mq_receiver == MyProc || mq->mq_sender == MyProc);
    mqh->mqh_queue = mq;
    mqh->mqh_segment = seg;
    mqh->mqh_handle = handle;
    mqh->mqh_buffer = NULL;
    mqh->mqh_buflen = 0;
    mqh->mqh_consume_pending = 0;
    mqh->mqh_send_pending = 0;
    mqh->mqh_partial_bytes = 0;
    mqh->mqh_expected_bytes = 0;
    mqh->mqh_length_word_complete = false;
    mqh->mqh_counterparty_attached = false;
    mqh->mqh_context = CurrentMemoryContext;
 
    if (seg != NULL)
        on_dsm_detach(seg, shm_mq_detach_callback, PointerGetDatum(mq));
 
    return mqh;
}

References Assert(), CurrentMemoryContext, shm_mq::mq_receiver, shm_mq::mq_sender, shm_mq_handle::mqh_buffer, shm_mq_handle::mqh_buflen, shm_mq_handle::mqh_consume_pending, shm_mq_handle::mqh_context, shm_mq_handle::mqh_counterparty_attached, shm_mq_handle::mqh_expected_bytes, shm_mq_handle::mqh_handle, shm_mq_handle::mqh_length_word_complete, shm_mq_handle::mqh_partial_bytes, shm_mq_handle::mqh_queue, shm_mq_handle::mqh_segment, shm_mq_handle::mqh_send_pending, MyProc, on_dsm_detach(), palloc(), PointerGetDatum(), and shm_mq_detach_callback().

Referenced by attach_to_queues(), ExecParallelGetReceiver(), ExecParallelSetupTupleQueues(), InitializeParallelDSM(), pa_setup_dsm(), ParallelApplyWorkerMain(), ParallelWorkerMain(), ReinitializeParallelDSM(), and test_shm_mq_setup().

shm_mq_counterparty_gone()

static bool shm_mq_counterparty_gone ( shm_mq *  mq, BackgroundWorkerHandle *  handle  )

static

Definition at line 1179 of file shm_mq.c.

{
    pid_t       pid;
 
    /* If the queue has been detached, counterparty is definitely gone. */
    if (mq->mq_detached)
        return true;
 
    /* If there's a handle, check worker status. */
    if (handle != NULL)
    {
        BgwHandleStatus status;
 
        /* Check for unexpected worker death. */
        status = GetBackgroundWorkerPid(handle, &pid);
        if (status != BGWH_STARTED && status != BGWH_NOT_YET_STARTED)
        {
            /* Mark it detached, just to make it official. */
            mq->mq_detached = true;
            return true;
        }
    }
 
    /* Counterparty is not definitively gone. */
    return false;
}

References BGWH_NOT_YET_STARTED, BGWH_STARTED, GetBackgroundWorkerPid(), and shm_mq::mq_detached.

Referenced by shm_mq_receive(), and shm_mq_send_bytes().

shm_mq_create()

shm_mq * shm_mq_create	(	void *	address,
		Size	size
	)

Definition at line 177 of file shm_mq.c.

{
    shm_mq     *mq = address;
    Size        data_offset = MAXALIGN(offsetof(shm_mq, mq_ring));
 
    /* If the size isn't MAXALIGN'd, just discard the odd bytes. */
    size = MAXALIGN_DOWN(size);
 
    /* Queue size must be large enough to hold some data. */
    Assert(size > data_offset);
 
    /* Initialize queue header. */
    SpinLockInit(&mq->mq_mutex);
    mq->mq_receiver = NULL;
    mq->mq_sender = NULL;
    pg_atomic_init_u64(&mq->mq_bytes_read, 0);
    pg_atomic_init_u64(&mq->mq_bytes_written, 0);
    mq->mq_ring_size = size - data_offset;
    mq->mq_detached = false;
    mq->mq_ring_offset = data_offset - offsetof(shm_mq, mq_ring);
 
    return mq;
}

References Assert(), MAXALIGN, MAXALIGN_DOWN, shm_mq::mq_bytes_read, shm_mq::mq_bytes_written, shm_mq::mq_detached, shm_mq::mq_mutex, shm_mq::mq_receiver, shm_mq::mq_ring_offset, shm_mq::mq_ring_size, shm_mq::mq_sender, pg_atomic_init_u64(), and SpinLockInit.

Referenced by ExecParallelSetupTupleQueues(), InitializeParallelDSM(), pa_setup_dsm(), ReinitializeParallelDSM(), and setup_dynamic_shared_memory().

shm_mq_detach()

void shm_mq_detach

(

shm_mq_handle *

mqh

)

Definition at line 843 of file shm_mq.c.

{
    /* Before detaching, notify the receiver about any already-written data. */
    if (mqh->mqh_send_pending > 0)
    {
        shm_mq_inc_bytes_written(mqh->mqh_queue, mqh->mqh_send_pending);
        mqh->mqh_send_pending = 0;
    }
 
    /* Notify counterparty that we're outta here. */
    shm_mq_detach_internal(mqh->mqh_queue);
 
    /* Cancel on_dsm_detach callback, if any. */
    if (mqh->mqh_segment)
        cancel_on_dsm_detach(mqh->mqh_segment,
                             shm_mq_detach_callback,
                             PointerGetDatum(mqh->mqh_queue));
 
    /* Release local memory associated with handle. */
    if (mqh->mqh_buffer != NULL)
        pfree(mqh->mqh_buffer);
    pfree(mqh);
}

References cancel_on_dsm_detach(), shm_mq_handle::mqh_buffer, shm_mq_handle::mqh_queue, shm_mq_handle::mqh_segment, shm_mq_handle::mqh_send_pending, pfree(), PointerGetDatum(), shm_mq_detach_callback(), shm_mq_detach_internal(), and shm_mq_inc_bytes_written().

Referenced by DestroyParallelContext(), ExecParallelFinish(), LaunchParallelWorkers(), logicalrep_pa_worker_stop(), mq_putmessage(), pa_detach_all_error_mq(), pa_free_worker_info(), ProcessParallelMessage(), tqueueDestroyReceiver(), and tqueueShutdownReceiver().

shm_mq_detach_callback()

static void shm_mq_detach_callback ( dsm_segment *  seg, Datum  arg  )

static

Definition at line 1323 of file shm_mq.c.

{
    shm_mq     *mq = (shm_mq *) DatumGetPointer(arg);
 
    shm_mq_detach_internal(mq);
}

References arg, DatumGetPointer(), and shm_mq_detach_internal().

Referenced by shm_mq_attach(), and shm_mq_detach().

shm_mq_detach_internal()

static void shm_mq_detach_internal ( shm_mq *  mq )

static

Definition at line 882 of file shm_mq.c.

{
    PGPROC     *victim;
 
    SpinLockAcquire(&mq->mq_mutex);
    if (mq->mq_sender == MyProc)
        victim = mq->mq_receiver;
    else
    {
        Assert(mq->mq_receiver == MyProc);
        victim = mq->mq_sender;
    }
    mq->mq_detached = true;
    SpinLockRelease(&mq->mq_mutex);
 
    if (victim != NULL)
        SetLatch(&victim->procLatch);
}

References Assert(), shm_mq::mq_detached, shm_mq::mq_mutex, shm_mq::mq_receiver, shm_mq::mq_sender, MyProc, PGPROC::procLatch, SetLatch(), SpinLockAcquire, and SpinLockRelease.

Referenced by shm_mq_detach(), and shm_mq_detach_callback().

shm_mq_get_queue()

shm_mq * shm_mq_get_queue

(

shm_mq_handle *

mqh

)

Definition at line 905 of file shm_mq.c.

{
    return mqh->mqh_queue;
}

References shm_mq_handle::mqh_queue.

Referenced by WaitForParallelWorkersToAttach(), and WaitForParallelWorkersToFinish().

shm_mq_get_receiver()

PGPROC * shm_mq_get_receiver

(

shm_mq *

mq

)

Definition at line 242 of file shm_mq.c.

{
    PGPROC     *receiver;
 
    SpinLockAcquire(&mq->mq_mutex);
    receiver = mq->mq_receiver;
    SpinLockRelease(&mq->mq_mutex);
 
    return receiver;
}

References shm_mq::mq_mutex, shm_mq::mq_receiver, SpinLockAcquire, and SpinLockRelease.

Referenced by shm_mq_send_bytes(), and shm_mq_wait_for_attach().

shm_mq_get_sender()

PGPROC * shm_mq_get_sender

(

shm_mq *

mq

)

Definition at line 257 of file shm_mq.c.

{
    PGPROC     *sender;
 
    SpinLockAcquire(&mq->mq_mutex);
    sender = mq->mq_sender;
    SpinLockRelease(&mq->mq_mutex);
 
    return sender;
}

References shm_mq::mq_mutex, shm_mq::mq_sender, SpinLockAcquire, and SpinLockRelease.

Referenced by shm_mq_receive(), shm_mq_wait_for_attach(), WaitForParallelWorkersToAttach(), and WaitForParallelWorkersToFinish().

shm_mq_inc_bytes_read()

static void shm_mq_inc_bytes_read ( shm_mq *  mq, Size  n  )

static

Definition at line 1270 of file shm_mq.c.

{
    PGPROC     *sender;
 
    /*
     * Separate prior reads of mq_ring from the increment of mq_bytes_read
     * which follows.  This pairs with the full barrier in
     * shm_mq_send_bytes(). We only need a read barrier here because the
     * increment of mq_bytes_read is actually a read followed by a dependent
     * write.
     */
    pg_read_barrier();
 
    /*
     * There's no need to use pg_atomic_fetch_add_u64 here, because nobody
     * else can be changing this value.  This method should be cheaper.
     */
    pg_atomic_write_u64(&mq->mq_bytes_read,
                        pg_atomic_read_u64(&mq->mq_bytes_read) + n);
 
    /*
     * We shouldn't have any bytes to read without a sender, so we can read
     * mq_sender here without a lock.  Once it's initialized, it can't change.
     */
    sender = mq->mq_sender;
    Assert(sender != NULL);
    SetLatch(&sender->procLatch);
}

References Assert(), shm_mq::mq_bytes_read, shm_mq::mq_sender, pg_atomic_read_u64(), pg_atomic_write_u64(), pg_read_barrier, PGPROC::procLatch, and SetLatch().

Referenced by shm_mq_receive(), and shm_mq_receive_bytes().

shm_mq_inc_bytes_written()

static void shm_mq_inc_bytes_written ( shm_mq *  mq, Size  n  )

static

Definition at line 1303 of file shm_mq.c.

{
    /*
     * Separate prior reads of mq_ring from the write of mq_bytes_written
     * which we're about to do.  Pairs with the read barrier found in
     * shm_mq_receive_bytes.
     */
    pg_write_barrier();
 
    /*
     * There's no need to use pg_atomic_fetch_add_u64 here, because nobody
     * else can be changing this value.  This method avoids taking the bus
     * lock unnecessarily.
     */
    pg_atomic_write_u64(&mq->mq_bytes_written,
                        pg_atomic_read_u64(&mq->mq_bytes_written) + n);
}

References shm_mq::mq_bytes_written, pg_atomic_read_u64(), pg_atomic_write_u64(), and pg_write_barrier.

Referenced by shm_mq_detach(), shm_mq_send_bytes(), and shm_mq_sendv().

shm_mq_receive()

shm_mq_result shm_mq_receive	(	shm_mq_handle *	mqh,
		Size *	nbytesp,
		void **	datap,
		bool	nowait
	)

Definition at line 572 of file shm_mq.c.

{
    shm_mq     *mq = mqh->mqh_queue;
    shm_mq_result res;
    Size        rb = 0;
    Size        nbytes;
    void       *rawdata;
 
    Assert(mq->mq_receiver == MyProc);
 
    /* We can't receive data until the sender has attached. */
    if (!mqh->mqh_counterparty_attached)
    {
        if (nowait)
        {
            int         counterparty_gone;
 
            /*
             * We shouldn't return at this point at all unless the sender
             * hasn't attached yet.  However, the correct return value depends
             * on whether the sender is still attached.  If we first test
             * whether the sender has ever attached and then test whether the
             * sender has detached, there's a race condition: a sender that
             * attaches and detaches very quickly might fool us into thinking
             * the sender never attached at all.  So, test whether our
             * counterparty is definitively gone first, and only afterwards
             * check whether the sender ever attached in the first place.
             */
            counterparty_gone = shm_mq_counterparty_gone(mq, mqh->mqh_handle);
            if (shm_mq_get_sender(mq) == NULL)
            {
                if (counterparty_gone)
                    return SHM_MQ_DETACHED;
                else
                    return SHM_MQ_WOULD_BLOCK;
            }
        }
        else if (!shm_mq_wait_internal(mq, &mq->mq_sender, mqh->mqh_handle)
                 && shm_mq_get_sender(mq) == NULL)
        {
            mq->mq_detached = true;
            return SHM_MQ_DETACHED;
        }
        mqh->mqh_counterparty_attached = true;
    }
 
    /*
     * If we've consumed an amount of data greater than 1/4th of the ring
     * size, mark it consumed in shared memory.  We try to avoid doing this
     * unnecessarily when only a small amount of data has been consumed,
     * because SetLatch() is fairly expensive and we don't want to do it too
     * often.
     */
    if (mqh->mqh_consume_pending > mq->mq_ring_size / 4)
    {
        shm_mq_inc_bytes_read(mq, mqh->mqh_consume_pending);
        mqh->mqh_consume_pending = 0;
    }
 
    /* Try to read, or finish reading, the length word from the buffer. */
    while (!mqh->mqh_length_word_complete)
    {
        /* Try to receive the message length word. */
        Assert(mqh->mqh_partial_bytes < sizeof(Size));
        res = shm_mq_receive_bytes(mqh, sizeof(Size) - mqh->mqh_partial_bytes,
                                   nowait, &rb, &rawdata);
        if (res != SHM_MQ_SUCCESS)
            return res;
 
        /*
         * Hopefully, we'll receive the entire message length word at once.
         * But if sizeof(Size) > MAXIMUM_ALIGNOF, then it might be split over
         * multiple reads.
         */
        if (mqh->mqh_partial_bytes == 0 && rb >= sizeof(Size))
        {
            Size        needed;
 
            nbytes = *(Size *) rawdata;
 
            /* If we've already got the whole message, we're done. */
            needed = MAXALIGN(sizeof(Size)) + MAXALIGN(nbytes);
            if (rb >= needed)
            {
                mqh->mqh_consume_pending += needed;
                *nbytesp = nbytes;
                *datap = ((char *) rawdata) + MAXALIGN(sizeof(Size));
                return SHM_MQ_SUCCESS;
            }
 
            /*
             * We don't have the whole message, but we at least have the whole
             * length word.
             */
            mqh->mqh_expected_bytes = nbytes;
            mqh->mqh_length_word_complete = true;
            mqh->mqh_consume_pending += MAXALIGN(sizeof(Size));
            rb -= MAXALIGN(sizeof(Size));
        }
        else
        {
            Size        lengthbytes;
 
            /* Can't be split unless bigger than required alignment. */
            Assert(sizeof(Size) > MAXIMUM_ALIGNOF);
 
            /* Message word is split; need buffer to reassemble. */
            if (mqh->mqh_buffer == NULL)
            {
                mqh->mqh_buffer = MemoryContextAlloc(mqh->mqh_context,
                                                     MQH_INITIAL_BUFSIZE);
                mqh->mqh_buflen = MQH_INITIAL_BUFSIZE;
            }
            Assert(mqh->mqh_buflen >= sizeof(Size));
 
            /* Copy partial length word; remember to consume it. */
            if (mqh->mqh_partial_bytes + rb > sizeof(Size))
                lengthbytes = sizeof(Size) - mqh->mqh_partial_bytes;
            else
                lengthbytes = rb;
            memcpy(&mqh->mqh_buffer[mqh->mqh_partial_bytes], rawdata,
                   lengthbytes);
            mqh->mqh_partial_bytes += lengthbytes;
            mqh->mqh_consume_pending += MAXALIGN(lengthbytes);
            rb -= lengthbytes;
 
            /* If we now have the whole word, we're ready to read payload. */
            if (mqh->mqh_partial_bytes >= sizeof(Size))
            {
                Assert(mqh->mqh_partial_bytes == sizeof(Size));
                mqh->mqh_expected_bytes = *(Size *) mqh->mqh_buffer;
                mqh->mqh_length_word_complete = true;
                mqh->mqh_partial_bytes = 0;
            }
        }
    }
    nbytes = mqh->mqh_expected_bytes;
 
    /*
     * Should be disallowed on the sending side already, but better check and
     * error out on the receiver side as well rather than trying to read a
     * prohibitively large message.
     */
    if (nbytes > MaxAllocSize)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("invalid message size %zu in shared memory queue",
                        nbytes)));
 
    if (mqh->mqh_partial_bytes == 0)
    {
        /*
         * Try to obtain the whole message in a single chunk.  If this works,
         * we need not copy the data and can return a pointer directly into
         * shared memory.
         */
        res = shm_mq_receive_bytes(mqh, nbytes, nowait, &rb, &rawdata);
        if (res != SHM_MQ_SUCCESS)
            return res;
        if (rb >= nbytes)
        {
            mqh->mqh_length_word_complete = false;
            mqh->mqh_consume_pending += MAXALIGN(nbytes);
            *nbytesp = nbytes;
            *datap = rawdata;
            return SHM_MQ_SUCCESS;
        }
 
        /*
         * The message has wrapped the buffer.  We'll need to copy it in order
         * to return it to the client in one chunk.  First, make sure we have
         * a large enough buffer available.
         */
        if (mqh->mqh_buflen < nbytes)
        {
            Size        newbuflen;
 
            /*
             * Increase size to the next power of 2 that's >= nbytes, but
             * limit to MaxAllocSize.
             */
            newbuflen = pg_nextpower2_size_t(nbytes);
            newbuflen = Min(newbuflen, MaxAllocSize);
 
            if (mqh->mqh_buffer != NULL)
            {
                pfree(mqh->mqh_buffer);
                mqh->mqh_buffer = NULL;
                mqh->mqh_buflen = 0;
            }
            mqh->mqh_buffer = MemoryContextAlloc(mqh->mqh_context, newbuflen);
            mqh->mqh_buflen = newbuflen;
        }
    }
 
    /* Loop until we've copied the entire message. */
    for (;;)
    {
        Size        still_needed;
 
        /* Copy as much as we can. */
        Assert(mqh->mqh_partial_bytes + rb <= nbytes);
        if (rb > 0)
        {
            memcpy(&mqh->mqh_buffer[mqh->mqh_partial_bytes], rawdata, rb);
            mqh->mqh_partial_bytes += rb;
        }
 
        /*
         * Update count of bytes that can be consumed, accounting for
         * alignment padding.  Note that this will never actually insert any
         * padding except at the end of a message, because the buffer size is
         * a multiple of MAXIMUM_ALIGNOF, and each read and write is as well.
         */
        Assert(mqh->mqh_partial_bytes == nbytes || rb == MAXALIGN(rb));
        mqh->mqh_consume_pending += MAXALIGN(rb);
 
        /* If we got all the data, exit the loop. */
        if (mqh->mqh_partial_bytes >= nbytes)
            break;
 
        /* Wait for some more data. */
        still_needed = nbytes - mqh->mqh_partial_bytes;
        res = shm_mq_receive_bytes(mqh, still_needed, nowait, &rb, &rawdata);
        if (res != SHM_MQ_SUCCESS)
            return res;
        if (rb > still_needed)
            rb = still_needed;
    }
 
    /* Return the complete message, and reset for next message. */
    *nbytesp = nbytes;
    *datap = mqh->mqh_buffer;
    mqh->mqh_length_word_complete = false;
    mqh->mqh_partial_bytes = 0;
    return SHM_MQ_SUCCESS;
}

References Assert(), ereport, errcode(), errmsg(), ERROR, MAXALIGN, MaxAllocSize, MemoryContextAlloc(), Min, shm_mq::mq_detached, shm_mq::mq_receiver, shm_mq::mq_ring_size, shm_mq::mq_sender, shm_mq_handle::mqh_buffer, shm_mq_handle::mqh_buflen, shm_mq_handle::mqh_consume_pending, shm_mq_handle::mqh_context, shm_mq_handle::mqh_counterparty_attached, shm_mq_handle::mqh_expected_bytes, shm_mq_handle::mqh_handle, MQH_INITIAL_BUFSIZE, shm_mq_handle::mqh_length_word_complete, shm_mq_handle::mqh_partial_bytes, shm_mq_handle::mqh_queue, MyProc, pfree(), pg_nextpower2_size_t, shm_mq_counterparty_gone(), SHM_MQ_DETACHED, shm_mq_get_sender(), shm_mq_inc_bytes_read(), shm_mq_receive_bytes(), SHM_MQ_SUCCESS, shm_mq_wait_internal(), and SHM_MQ_WOULD_BLOCK.

Referenced by copy_messages(), LogicalParallelApplyLoop(), ProcessParallelApplyMessages(), ProcessParallelMessages(), test_shm_mq(), test_shm_mq_pipelined(), and TupleQueueReaderNext().

shm_mq_receive_bytes()

static shm_mq_result shm_mq_receive_bytes ( shm_mq_handle *  mqh, Size  bytes_needed, bool  nowait, Size *  nbytesp, void **  datap  )

static

Definition at line 1079 of file shm_mq.c.

{
    shm_mq     *mq = mqh->mqh_queue;
    Size        ringsize = mq->mq_ring_size;
    uint64      used;
    uint64      written;
 
    for (;;)
    {
        Size        offset;
        uint64      read;
 
        /* Get bytes written, so we can compute what's available to read. */
        written = pg_atomic_read_u64(&mq->mq_bytes_written);
 
        /*
         * Get bytes read.  Include bytes we could consume but have not yet
         * consumed.
         */
        read = pg_atomic_read_u64(&mq->mq_bytes_read) +
            mqh->mqh_consume_pending;
        used = written - read;
        Assert(used <= ringsize);
        offset = read % (uint64) ringsize;
 
        /* If we have enough data or buffer has wrapped, we're done. */
        if (used >= bytes_needed || offset + used >= ringsize)
        {
            *nbytesp = Min(used, ringsize - offset);
            *datap = &mq->mq_ring[mq->mq_ring_offset + offset];
 
            /*
             * Separate the read of mq_bytes_written, above, from caller's
             * attempt to read the data itself.  Pairs with the barrier in
             * shm_mq_inc_bytes_written.
             */
            pg_read_barrier();
            return SHM_MQ_SUCCESS;
        }
 
        /*
         * Fall out before waiting if the queue has been detached.
         *
         * Note that we don't check for this until *after* considering whether
         * the data already available is enough, since the receiver can finish
         * receiving a message stored in the buffer even after the sender has
         * detached.
         */
        if (mq->mq_detached)
        {
            /*
             * If the writer advanced mq_bytes_written and then set
             * mq_detached, we might not have read the final value of
             * mq_bytes_written above.  Insert a read barrier and then check
             * again if mq_bytes_written has advanced.
             */
            pg_read_barrier();
            if (written != pg_atomic_read_u64(&mq->mq_bytes_written))
                continue;
 
            return SHM_MQ_DETACHED;
        }
 
        /*
         * We didn't get enough data to satisfy the request, so mark any data
         * previously-consumed as read to make more buffer space.
         */
        if (mqh->mqh_consume_pending > 0)
        {
            shm_mq_inc_bytes_read(mq, mqh->mqh_consume_pending);
            mqh->mqh_consume_pending = 0;
        }
 
        /* Skip manipulation of our latch if nowait = true. */
        if (nowait)
            return SHM_MQ_WOULD_BLOCK;
 
        /*
         * Wait for our latch to be set.  It might already be set for some
         * unrelated reason, but that'll just result in one extra trip through
         * the loop.  It's worth it to avoid resetting the latch at top of
         * loop, because setting an already-set latch is much cheaper than
         * setting one that has been reset.
         */
        (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
                         WAIT_EVENT_MESSAGE_QUEUE_RECEIVE);
 
        /* Reset the latch so we don't spin. */
        ResetLatch(MyLatch);
 
        /* An interrupt may have occurred while we were waiting. */
        CHECK_FOR_INTERRUPTS();
    }
}

References Assert(), CHECK_FOR_INTERRUPTS, Min, shm_mq::mq_bytes_read, shm_mq::mq_bytes_written, shm_mq::mq_detached, shm_mq::mq_ring, shm_mq::mq_ring_offset, shm_mq::mq_ring_size, shm_mq_handle::mqh_consume_pending, shm_mq_handle::mqh_queue, MyLatch, pg_atomic_read_u64(), pg_read_barrier, read, ResetLatch(), SHM_MQ_DETACHED, shm_mq_inc_bytes_read(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, WaitLatch(), WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

Referenced by shm_mq_receive().

shm_mq_send()

shm_mq_result shm_mq_send	(	shm_mq_handle *	mqh,
		Size	nbytes,
		const void *	data,
		bool	nowait,
		bool	force_flush
	)

Definition at line 329 of file shm_mq.c.

{
    shm_mq_iovec iov;
 
    iov.data = data;
    iov.len = nbytes;
 
    return shm_mq_sendv(mqh, &iov, 1, nowait, force_flush);
}

References shm_mq_iovec::data, data, shm_mq_iovec::len, and shm_mq_sendv().

Referenced by copy_messages(), pa_send_data(), test_shm_mq(), test_shm_mq_pipelined(), and tqueueReceiveSlot().

shm_mq_send_bytes()

static shm_mq_result shm_mq_send_bytes ( shm_mq_handle *  mqh, Size  nbytes, const void *  data, bool  nowait, Size *  bytes_written  )

static

Definition at line 914 of file shm_mq.c.

{
    shm_mq     *mq = mqh->mqh_queue;
    Size        sent = 0;
    uint64      used;
    Size        ringsize = mq->mq_ring_size;
    Size        available;
 
    while (sent < nbytes)
    {
        uint64      rb;
        uint64      wb;
 
        /* Compute number of ring buffer bytes used and available. */
        rb = pg_atomic_read_u64(&mq->mq_bytes_read);
        wb = pg_atomic_read_u64(&mq->mq_bytes_written) + mqh->mqh_send_pending;
        Assert(wb >= rb);
        used = wb - rb;
        Assert(used <= ringsize);
        available = Min(ringsize - used, nbytes - sent);
 
        /*
         * Bail out if the queue has been detached.  Note that we would be in
         * trouble if the compiler decided to cache the value of
         * mq->mq_detached in a register or on the stack across loop
         * iterations.  It probably shouldn't do that anyway since we'll
         * always return, call an external function that performs a system
         * call, or reach a memory barrier at some point later in the loop,
         * but just to be sure, insert a compiler barrier here.
         */
        pg_compiler_barrier();
        if (mq->mq_detached)
        {
            *bytes_written = sent;
            return SHM_MQ_DETACHED;
        }
 
        if (available == 0 && !mqh->mqh_counterparty_attached)
        {
            /*
             * The queue is full, so if the receiver isn't yet known to be
             * attached, we must wait for that to happen.
             */
            if (nowait)
            {
                if (shm_mq_counterparty_gone(mq, mqh->mqh_handle))
                {
                    *bytes_written = sent;
                    return SHM_MQ_DETACHED;
                }
                if (shm_mq_get_receiver(mq) == NULL)
                {
                    *bytes_written = sent;
                    return SHM_MQ_WOULD_BLOCK;
                }
            }
            else if (!shm_mq_wait_internal(mq, &mq->mq_receiver,
                                           mqh->mqh_handle))
            {
                mq->mq_detached = true;
                *bytes_written = sent;
                return SHM_MQ_DETACHED;
            }
            mqh->mqh_counterparty_attached = true;
 
            /*
             * The receiver may have read some data after attaching, so we
             * must not wait without rechecking the queue state.
             */
        }
        else if (available == 0)
        {
            /* Update the pending send bytes in the shared memory. */
            shm_mq_inc_bytes_written(mq, mqh->mqh_send_pending);
 
            /*
             * Since mq->mqh_counterparty_attached is known to be true at this
             * point, mq_receiver has been set, and it can't change once set.
             * Therefore, we can read it without acquiring the spinlock.
             */
            Assert(mqh->mqh_counterparty_attached);
            SetLatch(&mq->mq_receiver->procLatch);
 
            /*
             * We have just updated the mqh_send_pending bytes in the shared
             * memory so reset it.
             */
            mqh->mqh_send_pending = 0;
 
            /* Skip manipulation of our latch if nowait = true. */
            if (nowait)
            {
                *bytes_written = sent;
                return SHM_MQ_WOULD_BLOCK;
            }
 
            /*
             * Wait for our latch to be set.  It might already be set for some
             * unrelated reason, but that'll just result in one extra trip
             * through the loop.  It's worth it to avoid resetting the latch
             * at top of loop, because setting an already-set latch is much
             * cheaper than setting one that has been reset.
             */
            (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
                             WAIT_EVENT_MESSAGE_QUEUE_SEND);
 
            /* Reset the latch so we don't spin. */
            ResetLatch(MyLatch);
 
            /* An interrupt may have occurred while we were waiting. */
            CHECK_FOR_INTERRUPTS();
        }
        else
        {
            Size        offset;
            Size        sendnow;
 
            offset = wb % (uint64) ringsize;
            sendnow = Min(available, ringsize - offset);
 
            /*
             * Write as much data as we can via a single memcpy(). Make sure
             * these writes happen after the read of mq_bytes_read, above.
             * This barrier pairs with the one in shm_mq_inc_bytes_read.
             * (Since we're separating the read of mq_bytes_read from a
             * subsequent write to mq_ring, we need a full barrier here.)
             */
            pg_memory_barrier();
            memcpy(&mq->mq_ring[mq->mq_ring_offset + offset],
                   (char *) data + sent, sendnow);
            sent += sendnow;
 
            /*
             * Update count of bytes written, with alignment padding.  Note
             * that this will never actually insert any padding except at the
             * end of a run of bytes, because the buffer size is a multiple of
             * MAXIMUM_ALIGNOF, and each read is as well.
             */
            Assert(sent == nbytes || sendnow == MAXALIGN(sendnow));
 
            /*
             * For efficiency, we don't update the bytes written in the shared
             * memory and also don't set the reader's latch here.  Refer to
             * the comments atop the shm_mq_handle structure for more
             * information.
             */
            mqh->mqh_send_pending += MAXALIGN(sendnow);
        }
    }
 
    *bytes_written = sent;
    return SHM_MQ_SUCCESS;
}

References Assert(), CHECK_FOR_INTERRUPTS, data, MAXALIGN, Min, shm_mq::mq_bytes_read, shm_mq::mq_bytes_written, shm_mq::mq_detached, shm_mq::mq_receiver, shm_mq::mq_ring, shm_mq::mq_ring_offset, shm_mq::mq_ring_size, shm_mq_handle::mqh_counterparty_attached, shm_mq_handle::mqh_handle, shm_mq_handle::mqh_queue, shm_mq_handle::mqh_send_pending, MyLatch, pg_atomic_read_u64(), pg_compiler_barrier, pg_memory_barrier, PGPROC::procLatch, ResetLatch(), SetLatch(), shm_mq_counterparty_gone(), SHM_MQ_DETACHED, shm_mq_get_receiver(), shm_mq_inc_bytes_written(), SHM_MQ_SUCCESS, shm_mq_wait_internal(), SHM_MQ_WOULD_BLOCK, WaitLatch(), WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

Referenced by shm_mq_sendv().

shm_mq_sendv()

shm_mq_result shm_mq_sendv	(	shm_mq_handle *	mqh,
		shm_mq_iovec *	iov,
		int	iovcnt,
		bool	nowait,
		bool	force_flush
	)

Definition at line 361 of file shm_mq.c.

{
    shm_mq_result res;
    shm_mq     *mq = mqh->mqh_queue;
    PGPROC     *receiver;
    Size        nbytes = 0;
    Size        bytes_written;
    int         i;
    int         which_iov = 0;
    Size        offset;
 
    Assert(mq->mq_sender == MyProc);
 
    /* Compute total size of write. */
    for (i = 0; i < iovcnt; ++i)
        nbytes += iov[i].len;
 
    /* Prevent writing messages overwhelming the receiver. */
    if (nbytes > MaxAllocSize)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("cannot send a message of size %zu via shared memory queue",
                        nbytes)));
 
    /* Try to write, or finish writing, the length word into the buffer. */
    while (!mqh->mqh_length_word_complete)
    {
        Assert(mqh->mqh_partial_bytes < sizeof(Size));
        res = shm_mq_send_bytes(mqh, sizeof(Size) - mqh->mqh_partial_bytes,
                                ((char *) &nbytes) + mqh->mqh_partial_bytes,
                                nowait, &bytes_written);
 
        if (res == SHM_MQ_DETACHED)
        {
            /* Reset state in case caller tries to send another message. */
            mqh->mqh_partial_bytes = 0;
            mqh->mqh_length_word_complete = false;
            return res;
        }
        mqh->mqh_partial_bytes += bytes_written;
 
        if (mqh->mqh_partial_bytes >= sizeof(Size))
        {
            Assert(mqh->mqh_partial_bytes == sizeof(Size));
 
            mqh->mqh_partial_bytes = 0;
            mqh->mqh_length_word_complete = true;
        }
 
        if (res != SHM_MQ_SUCCESS)
            return res;
 
        /* Length word can't be split unless bigger than required alignment. */
        Assert(mqh->mqh_length_word_complete || sizeof(Size) > MAXIMUM_ALIGNOF);
    }
 
    /* Write the actual data bytes into the buffer. */
    Assert(mqh->mqh_partial_bytes <= nbytes);
    offset = mqh->mqh_partial_bytes;
    do
    {
        Size        chunksize;
 
        /* Figure out which bytes need to be sent next. */
        if (offset >= iov[which_iov].len)
        {
            offset -= iov[which_iov].len;
            ++which_iov;
            if (which_iov >= iovcnt)
                break;
            continue;
        }
 
        /*
         * We want to avoid copying the data if at all possible, but every
         * chunk of bytes we write into the queue has to be MAXALIGN'd, except
         * the last.  Thus, if a chunk other than the last one ends on a
         * non-MAXALIGN'd boundary, we have to combine the tail end of its
         * data with data from one or more following chunks until we either
         * reach the last chunk or accumulate a number of bytes which is
         * MAXALIGN'd.
         */
        if (which_iov + 1 < iovcnt &&
            offset + MAXIMUM_ALIGNOF > iov[which_iov].len)
        {
            char        tmpbuf[MAXIMUM_ALIGNOF];
            int         j = 0;
 
            for (;;)
            {
                if (offset < iov[which_iov].len)
                {
                    tmpbuf[j] = iov[which_iov].data[offset];
                    j++;
                    offset++;
                    if (j == MAXIMUM_ALIGNOF)
                        break;
                }
                else
                {
                    offset -= iov[which_iov].len;
                    which_iov++;
                    if (which_iov >= iovcnt)
                        break;
                }
            }
 
            res = shm_mq_send_bytes(mqh, j, tmpbuf, nowait, &bytes_written);
 
            if (res == SHM_MQ_DETACHED)
            {
                /* Reset state in case caller tries to send another message. */
                mqh->mqh_partial_bytes = 0;
                mqh->mqh_length_word_complete = false;
                return res;
            }
 
            mqh->mqh_partial_bytes += bytes_written;
            if (res != SHM_MQ_SUCCESS)
                return res;
            continue;
        }
 
        /*
         * If this is the last chunk, we can write all the data, even if it
         * isn't a multiple of MAXIMUM_ALIGNOF.  Otherwise, we need to
         * MAXALIGN_DOWN the write size.
         */
        chunksize = iov[which_iov].len - offset;
        if (which_iov + 1 < iovcnt)
            chunksize = MAXALIGN_DOWN(chunksize);
        res = shm_mq_send_bytes(mqh, chunksize, &iov[which_iov].data[offset],
                                nowait, &bytes_written);
 
        if (res == SHM_MQ_DETACHED)
        {
            /* Reset state in case caller tries to send another message. */
            mqh->mqh_length_word_complete = false;
            mqh->mqh_partial_bytes = 0;
            return res;
        }
 
        mqh->mqh_partial_bytes += bytes_written;
        offset += bytes_written;
        if (res != SHM_MQ_SUCCESS)
            return res;
    } while (mqh->mqh_partial_bytes < nbytes);
 
    /* Reset for next message. */
    mqh->mqh_partial_bytes = 0;
    mqh->mqh_length_word_complete = false;
 
    /* If queue has been detached, let caller know. */
    if (mq->mq_detached)
        return SHM_MQ_DETACHED;
 
    /*
     * If the counterparty is known to have attached, we can read mq_receiver
     * without acquiring the spinlock.  Otherwise, more caution is needed.
     */
    if (mqh->mqh_counterparty_attached)
        receiver = mq->mq_receiver;
    else
    {
        SpinLockAcquire(&mq->mq_mutex);
        receiver = mq->mq_receiver;
        SpinLockRelease(&mq->mq_mutex);
        if (receiver != NULL)
            mqh->mqh_counterparty_attached = true;
    }
 
    /*
     * If the caller has requested force flush or we have written more than
     * 1/4 of the ring size, mark it as written in shared memory and notify
     * the receiver.
     */
    if (force_flush || mqh->mqh_send_pending > (mq->mq_ring_size >> 2))
    {
        shm_mq_inc_bytes_written(mq, mqh->mqh_send_pending);
        if (receiver != NULL)
            SetLatch(&receiver->procLatch);
        mqh->mqh_send_pending = 0;
    }
 
    return SHM_MQ_SUCCESS;
}

References Assert(), shm_mq_iovec::data, data, ereport, errcode(), errmsg(), ERROR, i, j, shm_mq_iovec::len, len, MAXALIGN_DOWN, MaxAllocSize, shm_mq::mq_detached, shm_mq::mq_mutex, shm_mq::mq_receiver, shm_mq::mq_ring_size, shm_mq::mq_sender, shm_mq_handle::mqh_counterparty_attached, shm_mq_handle::mqh_length_word_complete, shm_mq_handle::mqh_partial_bytes, shm_mq_handle::mqh_queue, shm_mq_handle::mqh_send_pending, MyProc, PGPROC::procLatch, SetLatch(), SHM_MQ_DETACHED, shm_mq_inc_bytes_written(), shm_mq_send_bytes(), SHM_MQ_SUCCESS, SpinLockAcquire, SpinLockRelease, and tmpbuf.

Referenced by mq_putmessage(), and shm_mq_send().

shm_mq_set_handle()

void shm_mq_set_handle	(	shm_mq_handle *	mqh,
		BackgroundWorkerHandle *	handle
	)

Definition at line 319 of file shm_mq.c.

{
    Assert(mqh->mqh_handle == NULL);
    mqh->mqh_handle = handle;
}

References Assert(), and shm_mq_handle::mqh_handle.

Referenced by ExecParallelCreateReaders(), and LaunchParallelWorkers().

shm_mq_set_receiver()

void shm_mq_set_receiver	(	shm_mq *	mq,
		PGPROC *	proc
	)

Definition at line 206 of file shm_mq.c.

{
    PGPROC     *sender;
 
    SpinLockAcquire(&mq->mq_mutex);
    Assert(mq->mq_receiver == NULL);
    mq->mq_receiver = proc;
    sender = mq->mq_sender;
    SpinLockRelease(&mq->mq_mutex);
 
    if (sender != NULL)
        SetLatch(&sender->procLatch);
}

References Assert(), shm_mq::mq_mutex, shm_mq::mq_receiver, shm_mq::mq_sender, PGPROC::procLatch, SetLatch(), SpinLockAcquire, and SpinLockRelease.

Referenced by attach_to_queues(), ExecParallelSetupTupleQueues(), InitializeParallelDSM(), pa_setup_dsm(), ParallelApplyWorkerMain(), ReinitializeParallelDSM(), and setup_dynamic_shared_memory().

shm_mq_set_sender()

void shm_mq_set_sender	(	shm_mq *	mq,
		PGPROC *	proc
	)

Definition at line 224 of file shm_mq.c.

{
    PGPROC     *receiver;
 
    SpinLockAcquire(&mq->mq_mutex);
    Assert(mq->mq_sender == NULL);
    mq->mq_sender = proc;
    receiver = mq->mq_receiver;
    SpinLockRelease(&mq->mq_mutex);
 
    if (receiver != NULL)
        SetLatch(&receiver->procLatch);
}

References Assert(), shm_mq::mq_mutex, shm_mq::mq_receiver, shm_mq::mq_sender, PGPROC::procLatch, SetLatch(), SpinLockAcquire, and SpinLockRelease.

Referenced by attach_to_queues(), ExecParallelGetReceiver(), pa_setup_dsm(), ParallelApplyWorkerMain(), ParallelWorkerMain(), and setup_dynamic_shared_memory().

shm_mq_wait_for_attach()

shm_mq_result shm_mq_wait_for_attach

(

shm_mq_handle *

mqh

)

Definition at line 820 of file shm_mq.c.

{
    shm_mq     *mq = mqh->mqh_queue;
    PGPROC    **victim;
 
    if (shm_mq_get_receiver(mq) == MyProc)
        victim = &mq->mq_sender;
    else
    {
        Assert(shm_mq_get_sender(mq) == MyProc);
        victim = &mq->mq_receiver;
    }
 
    if (shm_mq_wait_internal(mq, victim, mqh->mqh_handle))
        return SHM_MQ_SUCCESS;
    else
        return SHM_MQ_DETACHED;
}

References Assert(), shm_mq::mq_receiver, shm_mq::mq_sender, shm_mq_handle::mqh_handle, shm_mq_handle::mqh_queue, MyProc, SHM_MQ_DETACHED, shm_mq_get_receiver(), shm_mq_get_sender(), SHM_MQ_SUCCESS, and shm_mq_wait_internal().

shm_mq_wait_internal()

static bool shm_mq_wait_internal ( shm_mq *  mq, PGPROC **  ptr, BackgroundWorkerHandle *  handle  )

static

Definition at line 1218 of file shm_mq.c.

{
    bool        result = false;
 
    for (;;)
    {
        BgwHandleStatus status;
        pid_t       pid;
 
        /* Acquire the lock just long enough to check the pointer. */
        SpinLockAcquire(&mq->mq_mutex);
        result = (*ptr != NULL);
        SpinLockRelease(&mq->mq_mutex);
 
        /* Fail if detached; else succeed if initialized. */
        if (mq->mq_detached)
        {
            result = false;
            break;
        }
        if (result)
            break;
 
        if (handle != NULL)
        {
            /* Check for unexpected worker death. */
            status = GetBackgroundWorkerPid(handle, &pid);
            if (status != BGWH_STARTED && status != BGWH_NOT_YET_STARTED)
            {
                result = false;
                break;
            }
        }
 
        /* Wait to be signaled. */
        (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
                         WAIT_EVENT_MESSAGE_QUEUE_INTERNAL);
 
        /* Reset the latch so we don't spin. */
        ResetLatch(MyLatch);
 
        /* An interrupt may have occurred while we were waiting. */
        CHECK_FOR_INTERRUPTS();
    }
 
    return result;
}

References BGWH_NOT_YET_STARTED, BGWH_STARTED, CHECK_FOR_INTERRUPTS, GetBackgroundWorkerPid(), shm_mq::mq_detached, shm_mq::mq_mutex, MyLatch, ResetLatch(), SpinLockAcquire, SpinLockRelease, WaitLatch(), WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

Referenced by shm_mq_receive(), shm_mq_send_bytes(), and shm_mq_wait_for_attach().

Variable Documentation

shm_mq_minimum_size

const Size shm_mq_minimum_size

Initial value:

=
MAXALIGN(offsetof(shm_mq, mq_ring)) + MAXIMUM_ALIGNOF

Definition at line 168 of file shm_mq.c.

Referenced by setup_dynamic_shared_memory().