#include "postmaster/bgworker.h"
#include "storage/dsm.h"
#include "storage/proc.h"

Include dependency graph for shm_mq.h:

This graph shows which files directly or indirectly include this file:

Data Structures
struct	shm_mq_iovec

Typedefs
typedef struct shm_mq	shm_mq

typedef struct shm_mq_handle	shm_mq_handle

Enumerations
enum	shm_mq_result { SHM_MQ_SUCCESS , SHM_MQ_WOULD_BLOCK , SHM_MQ_DETACHED }

Functions
shm_mq *	shm_mq_create (void *address, Size size)

void	shm_mq_set_receiver (shm_mq mq, PGPROC )

void	shm_mq_set_sender (shm_mq mq, PGPROC )

PGPROC *	shm_mq_get_receiver (shm_mq *)

PGPROC *	shm_mq_get_sender (shm_mq *)

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

void	shm_mq_set_handle (shm_mq_handle , BackgroundWorkerHandle )

void	shm_mq_detach (shm_mq_handle *mqh)

shm_mq *	shm_mq_get_queue (shm_mq_handle *mqh)

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)

Variables
PGDLLIMPORT const Size	shm_mq_minimum_size

Typedef Documentation

◆ shm_mq

typedef struct shm_mq shm_mq

Definition at line 1 of file shm_mq.h.

◆ shm_mq_handle

typedef struct shm_mq_handle shm_mq_handle

Definition at line 1 of file shm_mq.h.

Enumeration Type Documentation

◆ shm_mq_result

enum shm_mq_result

Enumerator
SHM_MQ_SUCCESS
SHM_MQ_WOULD_BLOCK
SHM_MQ_DETACHED

Definition at line 36 of file shm_mq.h.

 {
     SHM_MQ_SUCCESS,             /* Sent or received a message. */
     SHM_MQ_WOULD_BLOCK,         /* Not completed; retry later. */
     SHM_MQ_DETACHED             /* Other process has detached queue. */
 } shm_mq_result;

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

Variable Documentation

◆ shm_mq_minimum_size

PGDLLIMPORT const Size shm_mq_minimum_size

extern

Definition at line 169 of file shm_mq.c.

Referenced by setup_dynamic_shared_memory().

Data Structures

Typedefs

Enumerations

Functions

Variables