PostgreSQL Source Code  git master
shm_mq.h File Reference
#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:

Go to the source code of 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_mqshm_mq_create (void *address, Size size)
 
void shm_mq_set_receiver (shm_mq *mq, PGPROC *)
 
void shm_mq_set_sender (shm_mq *mq, PGPROC *)
 
PGPROCshm_mq_get_receiver (shm_mq *)
 
PGPROCshm_mq_get_sender (shm_mq *)
 
shm_mq_handleshm_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_mqshm_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

Enumerator
SHM_MQ_SUCCESS 
SHM_MQ_WOULD_BLOCK 
SHM_MQ_DETACHED 

Definition at line 36 of file shm_mq.h.

37 {
38  SHM_MQ_SUCCESS, /* Sent or received a message. */
39  SHM_MQ_WOULD_BLOCK, /* Not completed; retry later. */
40  SHM_MQ_DETACHED /* Other process has detached queue. */
shm_mq_result
Definition: shm_mq.h:37
@ SHM_MQ_SUCCESS
Definition: shm_mq.h:38
@ SHM_MQ_WOULD_BLOCK
Definition: shm_mq.h:39
@ SHM_MQ_DETACHED
Definition: shm_mq.h:40

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.

292 {
293  shm_mq_handle *mqh = palloc(sizeof(shm_mq_handle));
294 
295  Assert(mq->mq_receiver == MyProc || mq->mq_sender == MyProc);
296  mqh->mqh_queue = mq;
297  mqh->mqh_segment = seg;
298  mqh->mqh_handle = handle;
299  mqh->mqh_buffer = NULL;
300  mqh->mqh_buflen = 0;
301  mqh->mqh_consume_pending = 0;
302  mqh->mqh_send_pending = 0;
303  mqh->mqh_partial_bytes = 0;
304  mqh->mqh_expected_bytes = 0;
305  mqh->mqh_length_word_complete = false;
306  mqh->mqh_counterparty_attached = false;
308 
309  if (seg != NULL)
311 
312  return mqh;
313 }
void on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1103
Assert(fmt[strlen(fmt) - 1] !='\n')
MemoryContext CurrentMemoryContext
Definition: mcxt.c:135
void * palloc(Size size)
Definition: mcxt.c:1226
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
static void shm_mq_detach_callback(dsm_segment *seg, Datum arg)
Definition: shm_mq.c:1324
PGPROC * MyProc
Definition: proc.c:66
Size mqh_consume_pending
Definition: shm_mq.c:145
BackgroundWorkerHandle * mqh_handle
Definition: shm_mq.c:142
char * mqh_buffer
Definition: shm_mq.c:143
Size mqh_send_pending
Definition: shm_mq.c:146
Size mqh_expected_bytes
Definition: shm_mq.c:148
bool mqh_counterparty_attached
Definition: shm_mq.c:150
shm_mq * mqh_queue
Definition: shm_mq.c:140
dsm_segment * mqh_segment
Definition: shm_mq.c:141
bool mqh_length_word_complete
Definition: shm_mq.c:149
Size mqh_buflen
Definition: shm_mq.c:144
MemoryContext mqh_context
Definition: shm_mq.c:151
Size mqh_partial_bytes
Definition: shm_mq.c:147
PGPROC * mq_sender
Definition: shm_mq.c:76
PGPROC * mq_receiver
Definition: shm_mq.c:75

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

shm_mq* shm_mq_create ( void *  address,
Size  size 
)

Definition at line 178 of file shm_mq.c.

179 {
180  shm_mq *mq = address;
181  Size data_offset = MAXALIGN(offsetof(shm_mq, mq_ring));
182 
183  /* If the size isn't MAXALIGN'd, just discard the odd bytes. */
184  size = MAXALIGN_DOWN(size);
185 
186  /* Queue size must be large enough to hold some data. */
187  Assert(size > data_offset);
188 
189  /* Initialize queue header. */
190  SpinLockInit(&mq->mq_mutex);
191  mq->mq_receiver = NULL;
192  mq->mq_sender = NULL;
195  mq->mq_ring_size = size - data_offset;
196  mq->mq_detached = false;
197  mq->mq_ring_offset = data_offset - offsetof(shm_mq, mq_ring);
198 
199  return mq;
200 }
static void pg_atomic_init_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:410
#define MAXALIGN_DOWN(LEN)
Definition: c.h:812
#define MAXALIGN(LEN)
Definition: c.h:800
size_t Size
Definition: c.h:594
#define SpinLockInit(lock)
Definition: spin.h:60
Definition: shm_mq.c:73
pg_atomic_uint64 mq_bytes_written
Definition: shm_mq.c:78
pg_atomic_uint64 mq_bytes_read
Definition: shm_mq.c:77
bool mq_detached
Definition: shm_mq.c:80
uint8 mq_ring_offset
Definition: shm_mq.c:81
slock_t mq_mutex
Definition: shm_mq.c:74
Size mq_ring_size
Definition: shm_mq.c:79

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.

845 {
846  /* Before detaching, notify the receiver about any already-written data. */
847  if (mqh->mqh_send_pending > 0)
848  {
850  mqh->mqh_send_pending = 0;
851  }
852 
853  /* Notify counterparty that we're outta here. */
855 
856  /* Cancel on_dsm_detach callback, if any. */
857  if (mqh->mqh_segment)
860  PointerGetDatum(mqh->mqh_queue));
861 
862  /* Release local memory associated with handle. */
863  if (mqh->mqh_buffer != NULL)
864  pfree(mqh->mqh_buffer);
865  pfree(mqh);
866 }
void cancel_on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1118
void pfree(void *pointer)
Definition: mcxt.c:1456
static void shm_mq_inc_bytes_written(shm_mq *mq, Size n)
Definition: shm_mq.c:1304
static void shm_mq_detach_internal(shm_mq *mq)
Definition: shm_mq.c:883

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.

907 {
908  return mqh->mqh_queue;
909 }

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.

244 {
245  PGPROC *receiver;
246 
248  receiver = mq->mq_receiver;
250 
251  return receiver;
252 }
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
Definition: proc.h:162

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.

259 {
260  PGPROC *sender;
261 
263  sender = mq->mq_sender;
265 
266  return sender;
267 }

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.

574 {
575  shm_mq *mq = mqh->mqh_queue;
577  Size rb = 0;
578  Size nbytes;
579  void *rawdata;
580 
581  Assert(mq->mq_receiver == MyProc);
582 
583  /* We can't receive data until the sender has attached. */
584  if (!mqh->mqh_counterparty_attached)
585  {
586  if (nowait)
587  {
588  int counterparty_gone;
589 
590  /*
591  * We shouldn't return at this point at all unless the sender
592  * hasn't attached yet. However, the correct return value depends
593  * on whether the sender is still attached. If we first test
594  * whether the sender has ever attached and then test whether the
595  * sender has detached, there's a race condition: a sender that
596  * attaches and detaches very quickly might fool us into thinking
597  * the sender never attached at all. So, test whether our
598  * counterparty is definitively gone first, and only afterwards
599  * check whether the sender ever attached in the first place.
600  */
601  counterparty_gone = shm_mq_counterparty_gone(mq, mqh->mqh_handle);
602  if (shm_mq_get_sender(mq) == NULL)
603  {
604  if (counterparty_gone)
605  return SHM_MQ_DETACHED;
606  else
607  return SHM_MQ_WOULD_BLOCK;
608  }
609  }
610  else if (!shm_mq_wait_internal(mq, &mq->mq_sender, mqh->mqh_handle)
611  && shm_mq_get_sender(mq) == NULL)
612  {
613  mq->mq_detached = true;
614  return SHM_MQ_DETACHED;
615  }
616  mqh->mqh_counterparty_attached = true;
617  }
618 
619  /*
620  * If we've consumed an amount of data greater than 1/4th of the ring
621  * size, mark it consumed in shared memory. We try to avoid doing this
622  * unnecessarily when only a small amount of data has been consumed,
623  * because SetLatch() is fairly expensive and we don't want to do it too
624  * often.
625  */
626  if (mqh->mqh_consume_pending > mq->mq_ring_size / 4)
627  {
629  mqh->mqh_consume_pending = 0;
630  }
631 
632  /* Try to read, or finish reading, the length word from the buffer. */
633  while (!mqh->mqh_length_word_complete)
634  {
635  /* Try to receive the message length word. */
636  Assert(mqh->mqh_partial_bytes < sizeof(Size));
637  res = shm_mq_receive_bytes(mqh, sizeof(Size) - mqh->mqh_partial_bytes,
638  nowait, &rb, &rawdata);
639  if (res != SHM_MQ_SUCCESS)
640  return res;
641 
642  /*
643  * Hopefully, we'll receive the entire message length word at once.
644  * But if sizeof(Size) > MAXIMUM_ALIGNOF, then it might be split over
645  * multiple reads.
646  */
647  if (mqh->mqh_partial_bytes == 0 && rb >= sizeof(Size))
648  {
649  Size needed;
650 
651  nbytes = *(Size *) rawdata;
652 
653  /* If we've already got the whole message, we're done. */
654  needed = MAXALIGN(sizeof(Size)) + MAXALIGN(nbytes);
655  if (rb >= needed)
656  {
657  mqh->mqh_consume_pending += needed;
658  *nbytesp = nbytes;
659  *datap = ((char *) rawdata) + MAXALIGN(sizeof(Size));
660  return SHM_MQ_SUCCESS;
661  }
662 
663  /*
664  * We don't have the whole message, but we at least have the whole
665  * length word.
666  */
667  mqh->mqh_expected_bytes = nbytes;
668  mqh->mqh_length_word_complete = true;
669  mqh->mqh_consume_pending += MAXALIGN(sizeof(Size));
670  rb -= MAXALIGN(sizeof(Size));
671  }
672  else
673  {
674  Size lengthbytes;
675 
676  /* Can't be split unless bigger than required alignment. */
677  Assert(sizeof(Size) > MAXIMUM_ALIGNOF);
678 
679  /* Message word is split; need buffer to reassemble. */
680  if (mqh->mqh_buffer == NULL)
681  {
685  }
686  Assert(mqh->mqh_buflen >= sizeof(Size));
687 
688  /* Copy partial length word; remember to consume it. */
689  if (mqh->mqh_partial_bytes + rb > sizeof(Size))
690  lengthbytes = sizeof(Size) - mqh->mqh_partial_bytes;
691  else
692  lengthbytes = rb;
693  memcpy(&mqh->mqh_buffer[mqh->mqh_partial_bytes], rawdata,
694  lengthbytes);
695  mqh->mqh_partial_bytes += lengthbytes;
696  mqh->mqh_consume_pending += MAXALIGN(lengthbytes);
697  rb -= lengthbytes;
698 
699  /* If we now have the whole word, we're ready to read payload. */
700  if (mqh->mqh_partial_bytes >= sizeof(Size))
701  {
702  Assert(mqh->mqh_partial_bytes == sizeof(Size));
703  mqh->mqh_expected_bytes = *(Size *) mqh->mqh_buffer;
704  mqh->mqh_length_word_complete = true;
705  mqh->mqh_partial_bytes = 0;
706  }
707  }
708  }
709  nbytes = mqh->mqh_expected_bytes;
710 
711  /*
712  * Should be disallowed on the sending side already, but better check and
713  * error out on the receiver side as well rather than trying to read a
714  * prohibitively large message.
715  */
716  if (nbytes > MaxAllocSize)
717  ereport(ERROR,
718  (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
719  errmsg("invalid message size %zu in shared memory queue",
720  nbytes)));
721 
722  if (mqh->mqh_partial_bytes == 0)
723  {
724  /*
725  * Try to obtain the whole message in a single chunk. If this works,
726  * we need not copy the data and can return a pointer directly into
727  * shared memory.
728  */
729  res = shm_mq_receive_bytes(mqh, nbytes, nowait, &rb, &rawdata);
730  if (res != SHM_MQ_SUCCESS)
731  return res;
732  if (rb >= nbytes)
733  {
734  mqh->mqh_length_word_complete = false;
735  mqh->mqh_consume_pending += MAXALIGN(nbytes);
736  *nbytesp = nbytes;
737  *datap = rawdata;
738  return SHM_MQ_SUCCESS;
739  }
740 
741  /*
742  * The message has wrapped the buffer. We'll need to copy it in order
743  * to return it to the client in one chunk. First, make sure we have
744  * a large enough buffer available.
745  */
746  if (mqh->mqh_buflen < nbytes)
747  {
748  Size newbuflen;
749 
750  /*
751  * Increase size to the next power of 2 that's >= nbytes, but
752  * limit to MaxAllocSize.
753  */
754  newbuflen = pg_nextpower2_size_t(nbytes);
755  newbuflen = Min(newbuflen, MaxAllocSize);
756 
757  if (mqh->mqh_buffer != NULL)
758  {
759  pfree(mqh->mqh_buffer);
760  mqh->mqh_buffer = NULL;
761  mqh->mqh_buflen = 0;
762  }
763  mqh->mqh_buffer = MemoryContextAlloc(mqh->mqh_context, newbuflen);
764  mqh->mqh_buflen = newbuflen;
765  }
766  }
767 
768  /* Loop until we've copied the entire message. */
769  for (;;)
770  {
771  Size still_needed;
772 
773  /* Copy as much as we can. */
774  Assert(mqh->mqh_partial_bytes + rb <= nbytes);
775  if (rb > 0)
776  {
777  memcpy(&mqh->mqh_buffer[mqh->mqh_partial_bytes], rawdata, rb);
778  mqh->mqh_partial_bytes += rb;
779  }
780 
781  /*
782  * Update count of bytes that can be consumed, accounting for
783  * alignment padding. Note that this will never actually insert any
784  * padding except at the end of a message, because the buffer size is
785  * a multiple of MAXIMUM_ALIGNOF, and each read and write is as well.
786  */
787  Assert(mqh->mqh_partial_bytes == nbytes || rb == MAXALIGN(rb));
788  mqh->mqh_consume_pending += MAXALIGN(rb);
789 
790  /* If we got all the data, exit the loop. */
791  if (mqh->mqh_partial_bytes >= nbytes)
792  break;
793 
794  /* Wait for some more data. */
795  still_needed = nbytes - mqh->mqh_partial_bytes;
796  res = shm_mq_receive_bytes(mqh, still_needed, nowait, &rb, &rawdata);
797  if (res != SHM_MQ_SUCCESS)
798  return res;
799  if (rb > still_needed)
800  rb = still_needed;
801  }
802 
803  /* Return the complete message, and reset for next message. */
804  *nbytesp = nbytes;
805  *datap = mqh->mqh_buffer;
806  mqh->mqh_length_word_complete = false;
807  mqh->mqh_partial_bytes = 0;
808  return SHM_MQ_SUCCESS;
809 }
#define Min(x, y)
Definition: c.h:993
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:1021
#define MaxAllocSize
Definition: memutils.h:40
#define pg_nextpower2_size_t
Definition: pg_bitutils.h:339
#define MQH_INITIAL_BUFSIZE
Definition: shm_mq.c:172
static bool shm_mq_counterparty_gone(shm_mq *mq, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:1180
static bool shm_mq_wait_internal(shm_mq *mq, PGPROC **ptr, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:1219
PGPROC * shm_mq_get_sender(shm_mq *mq)
Definition: shm_mq.c:258
static void shm_mq_inc_bytes_read(shm_mq *mq, Size n)
Definition: shm_mq.c:1271
static shm_mq_result shm_mq_receive_bytes(shm_mq_handle *mqh, Size bytes_needed, bool nowait, Size *nbytesp, void **datap)
Definition: shm_mq.c:1080

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

332 {
333  shm_mq_iovec iov;
334 
335  iov.data = data;
336  iov.len = nbytes;
337 
338  return shm_mq_sendv(mqh, &iov, 1, nowait, force_flush);
339 }
const void * data
shm_mq_result shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait, bool force_flush)
Definition: shm_mq.c:362
const char * data
Definition: shm_mq.h:31
Size len
Definition: shm_mq.h:32

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.

364 {
366  shm_mq *mq = mqh->mqh_queue;
367  PGPROC *receiver;
368  Size nbytes = 0;
369  Size bytes_written;
370  int i;
371  int which_iov = 0;
372  Size offset;
373 
374  Assert(mq->mq_sender == MyProc);
375 
376  /* Compute total size of write. */
377  for (i = 0; i < iovcnt; ++i)
378  nbytes += iov[i].len;
379 
380  /* Prevent writing messages overwhelming the receiver. */
381  if (nbytes > MaxAllocSize)
382  ereport(ERROR,
383  (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
384  errmsg("cannot send a message of size %zu via shared memory queue",
385  nbytes)));
386 
387  /* Try to write, or finish writing, the length word into the buffer. */
388  while (!mqh->mqh_length_word_complete)
389  {
390  Assert(mqh->mqh_partial_bytes < sizeof(Size));
391  res = shm_mq_send_bytes(mqh, sizeof(Size) - mqh->mqh_partial_bytes,
392  ((char *) &nbytes) + mqh->mqh_partial_bytes,
393  nowait, &bytes_written);
394 
395  if (res == SHM_MQ_DETACHED)
396  {
397  /* Reset state in case caller tries to send another message. */
398  mqh->mqh_partial_bytes = 0;
399  mqh->mqh_length_word_complete = false;
400  return res;
401  }
402  mqh->mqh_partial_bytes += bytes_written;
403 
404  if (mqh->mqh_partial_bytes >= sizeof(Size))
405  {
406  Assert(mqh->mqh_partial_bytes == sizeof(Size));
407 
408  mqh->mqh_partial_bytes = 0;
409  mqh->mqh_length_word_complete = true;
410  }
411 
412  if (res != SHM_MQ_SUCCESS)
413  return res;
414 
415  /* Length word can't be split unless bigger than required alignment. */
416  Assert(mqh->mqh_length_word_complete || sizeof(Size) > MAXIMUM_ALIGNOF);
417  }
418 
419  /* Write the actual data bytes into the buffer. */
420  Assert(mqh->mqh_partial_bytes <= nbytes);
421  offset = mqh->mqh_partial_bytes;
422  do
423  {
424  Size chunksize;
425 
426  /* Figure out which bytes need to be sent next. */
427  if (offset >= iov[which_iov].len)
428  {
429  offset -= iov[which_iov].len;
430  ++which_iov;
431  if (which_iov >= iovcnt)
432  break;
433  continue;
434  }
435 
436  /*
437  * We want to avoid copying the data if at all possible, but every
438  * chunk of bytes we write into the queue has to be MAXALIGN'd, except
439  * the last. Thus, if a chunk other than the last one ends on a
440  * non-MAXALIGN'd boundary, we have to combine the tail end of its
441  * data with data from one or more following chunks until we either
442  * reach the last chunk or accumulate a number of bytes which is
443  * MAXALIGN'd.
444  */
445  if (which_iov + 1 < iovcnt &&
446  offset + MAXIMUM_ALIGNOF > iov[which_iov].len)
447  {
448  char tmpbuf[MAXIMUM_ALIGNOF];
449  int j = 0;
450 
451  for (;;)
452  {
453  if (offset < iov[which_iov].len)
454  {
455  tmpbuf[j] = iov[which_iov].data[offset];
456  j++;
457  offset++;
458  if (j == MAXIMUM_ALIGNOF)
459  break;
460  }
461  else
462  {
463  offset -= iov[which_iov].len;
464  which_iov++;
465  if (which_iov >= iovcnt)
466  break;
467  }
468  }
469 
470  res = shm_mq_send_bytes(mqh, j, tmpbuf, nowait, &bytes_written);
471 
472  if (res == SHM_MQ_DETACHED)
473  {
474  /* Reset state in case caller tries to send another message. */
475  mqh->mqh_partial_bytes = 0;
476  mqh->mqh_length_word_complete = false;
477  return res;
478  }
479 
480  mqh->mqh_partial_bytes += bytes_written;
481  if (res != SHM_MQ_SUCCESS)
482  return res;
483  continue;
484  }
485 
486  /*
487  * If this is the last chunk, we can write all the data, even if it
488  * isn't a multiple of MAXIMUM_ALIGNOF. Otherwise, we need to
489  * MAXALIGN_DOWN the write size.
490  */
491  chunksize = iov[which_iov].len - offset;
492  if (which_iov + 1 < iovcnt)
493  chunksize = MAXALIGN_DOWN(chunksize);
494  res = shm_mq_send_bytes(mqh, chunksize, &iov[which_iov].data[offset],
495  nowait, &bytes_written);
496 
497  if (res == SHM_MQ_DETACHED)
498  {
499  /* Reset state in case caller tries to send another message. */
500  mqh->mqh_length_word_complete = false;
501  mqh->mqh_partial_bytes = 0;
502  return res;
503  }
504 
505  mqh->mqh_partial_bytes += bytes_written;
506  offset += bytes_written;
507  if (res != SHM_MQ_SUCCESS)
508  return res;
509  } while (mqh->mqh_partial_bytes < nbytes);
510 
511  /* Reset for next message. */
512  mqh->mqh_partial_bytes = 0;
513  mqh->mqh_length_word_complete = false;
514 
515  /* If queue has been detached, let caller know. */
516  if (mq->mq_detached)
517  return SHM_MQ_DETACHED;
518 
519  /*
520  * If the counterparty is known to have attached, we can read mq_receiver
521  * without acquiring the spinlock. Otherwise, more caution is needed.
522  */
523  if (mqh->mqh_counterparty_attached)
524  receiver = mq->mq_receiver;
525  else
526  {
528  receiver = mq->mq_receiver;
530  if (receiver != NULL)
531  mqh->mqh_counterparty_attached = true;
532  }
533 
534  /*
535  * If the caller has requested force flush or we have written more than
536  * 1/4 of the ring size, mark it as written in shared memory and notify
537  * the receiver.
538  */
539  if (force_flush || mqh->mqh_send_pending > (mq->mq_ring_size >> 2))
540  {
542  if (receiver != NULL)
543  SetLatch(&receiver->procLatch);
544  mqh->mqh_send_pending = 0;
545  }
546 
547  return SHM_MQ_SUCCESS;
548 }
int j
Definition: isn.c:74
int i
Definition: isn.c:73
void SetLatch(Latch *latch)
Definition: latch.c:605
const void size_t len
static shm_mq_result shm_mq_send_bytes(shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait, Size *bytes_written)
Definition: shm_mq.c:915
Latch procLatch
Definition: proc.h:170
static StringInfoData tmpbuf
Definition: walsender.c:160

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.

321 {
322  Assert(mqh->mqh_handle == NULL);
323  mqh->mqh_handle = handle;
324 }

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.

208 {
209  PGPROC *sender;
210 
212  Assert(mq->mq_receiver == NULL);
213  mq->mq_receiver = proc;
214  sender = mq->mq_sender;
216 
217  if (sender != NULL)
218  SetLatch(&sender->procLatch);
219 }

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.

226 {
227  PGPROC *receiver;
228 
230  Assert(mq->mq_sender == NULL);
231  mq->mq_sender = proc;
232  receiver = mq->mq_receiver;
234 
235  if (receiver != NULL)
236  SetLatch(&receiver->procLatch);
237 }

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.

822 {
823  shm_mq *mq = mqh->mqh_queue;
824  PGPROC **victim;
825 
826  if (shm_mq_get_receiver(mq) == MyProc)
827  victim = &mq->mq_sender;
828  else
829  {
831  victim = &mq->mq_receiver;
832  }
833 
834  if (shm_mq_wait_internal(mq, victim, mqh->mqh_handle))
835  return SHM_MQ_SUCCESS;
836  else
837  return SHM_MQ_DETACHED;
838 }
PGPROC * shm_mq_get_receiver(shm_mq *mq)
Definition: shm_mq.c:243

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