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

Include dependency graph for shm_mq.c:

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 172 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 291 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;
 }

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 1180 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 178 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 844 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(), HandleParallelMessage(), LaunchParallelWorkers(), logicalrep_pa_worker_stop(), mq_putmessage(), pa_detach_all_error_mq(), pa_free_worker_info(), tqueueDestroyReceiver(), and tqueueShutdownReceiver().

◆ shm_mq_detach_callback()

static void shm_mq_detach_callback	(	dsm_segment *	seg,
		Datum	arg
	)

static

Definition at line 1324 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 883 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 906 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 243 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 258 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 1271 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 1304 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 573 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;
 }

Referenced by copy_messages(), HandleParallelApplyMessages(), HandleParallelMessages(), LogicalParallelApplyLoop(), 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 1080 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 330 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 915 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 362 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, res, 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 320 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 207 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 225 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 821 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 1219 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 169 of file shm_mq.c.

Referenced by setup_dynamic_shared_memory().

Data Structures

Macros

Functions

Variables

Macro Definition Documentation

◆ MQH_INITIAL_BUFSIZE

Function Documentation

◆ shm_mq_attach()

◆ shm_mq_counterparty_gone()

◆ shm_mq_create()

◆ shm_mq_detach()

◆ shm_mq_detach_callback()

◆ shm_mq_detach_internal()

◆ shm_mq_get_queue()

◆ shm_mq_get_receiver()

◆ shm_mq_get_sender()

◆ shm_mq_inc_bytes_read()

◆ shm_mq_inc_bytes_written()

◆ shm_mq_receive()

◆ shm_mq_receive_bytes()

◆ shm_mq_send()

◆ shm_mq_send_bytes()

◆ shm_mq_sendv()

◆ shm_mq_set_handle()

◆ shm_mq_set_receiver()

◆ shm_mq_set_sender()

◆ shm_mq_wait_for_attach()

◆ shm_mq_wait_internal()

Variable Documentation

◆ shm_mq_minimum_size