#include "postgres.h"
#include "fmgr.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "varatt.h"
#include "test_shm_mq.h"

Include dependency graph for test.c:

Functions
	PG_FUNCTION_INFO_V1 (test_shm_mq)

	PG_FUNCTION_INFO_V1 (test_shm_mq_pipelined)

static void	verify_message (Size origlen, char origdata, Size newlen, char newdata)

Datum	test_shm_mq (PG_FUNCTION_ARGS)

Datum	test_shm_mq_pipelined (PG_FUNCTION_ARGS)

Variables
	PG_MODULE_MAGIC

Function Documentation

◆ PG_FUNCTION_INFO_V1() [1/2]

PG_FUNCTION_INFO_V1 ( test_shm_mq )

◆ PG_FUNCTION_INFO_V1() [2/2]

PG_FUNCTION_INFO_V1 ( test_shm_mq_pipelined )

◆ test_shm_mq()

Datum test_shm_mq ( PG_FUNCTION_ARGS )

Definition at line 40 of file test.c.

 {
     int64       queue_size = PG_GETARG_INT64(0);
     text       *message = PG_GETARG_TEXT_PP(1);
     char       *message_contents = VARDATA_ANY(message);
     int         message_size = VARSIZE_ANY_EXHDR(message);
     int32       loop_count = PG_GETARG_INT32(2);
     int32       nworkers = PG_GETARG_INT32(3);
     dsm_segment *seg;
     shm_mq_handle *outqh;
     shm_mq_handle *inqh;
     shm_mq_result res;
     Size        len;
     void       *data;
  
     /* A negative loopcount is nonsensical. */
     if (loop_count < 0)
         ereport(ERROR,
                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                  errmsg("repeat count size must be an integer value greater than or equal to zero")));
  
     /*
      * Since this test sends data using the blocking interfaces, it cannot
      * send data to itself.  Therefore, a minimum of 1 worker is required. Of
      * course, a negative worker count is nonsensical.
      */
     if (nworkers <= 0)
         ereport(ERROR,
                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                  errmsg("number of workers must be an integer value greater than zero")));
  
     /* Set up dynamic shared memory segment and background workers. */
     test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
  
     /* Send the initial message. */
     res = shm_mq_send(outqh, message_size, message_contents, false, true);
     if (res != SHM_MQ_SUCCESS)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("could not send message")));
  
     /*
      * Receive a message and send it back out again.  Do this a number of
      * times equal to the loop count.
      */
     for (;;)
     {
         /* Receive a message. */
         res = shm_mq_receive(inqh, &len, &data, false);
         if (res != SHM_MQ_SUCCESS)
             ereport(ERROR,
                     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                      errmsg("could not receive message")));
  
         /* If this is supposed to be the last iteration, stop here. */
         if (--loop_count <= 0)
             break;
  
         /* Send it back out. */
         res = shm_mq_send(outqh, len, data, false, true);
         if (res != SHM_MQ_SUCCESS)
             ereport(ERROR,
                     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                      errmsg("could not send message")));
     }
  
     /*
      * Finally, check that we got back the same message from the last
      * iteration that we originally sent.
      */
     verify_message(message_size, message_contents, len, data);
  
     /* Clean up. */
     dsm_detach(seg);
  
     PG_RETURN_VOID();
 }

References data, dsm_detach(), ereport, errcode(), errmsg(), ERROR, len, PG_GETARG_INT32, PG_GETARG_INT64, PG_GETARG_TEXT_PP, PG_RETURN_VOID, res, shm_mq_receive(), shm_mq_send(), SHM_MQ_SUCCESS, test_shm_mq_setup(), VARDATA_ANY, VARSIZE_ANY_EXHDR, and verify_message().

◆ test_shm_mq_pipelined()

Datum test_shm_mq_pipelined ( PG_FUNCTION_ARGS )

Definition at line 129 of file test.c.

 {
     int64       queue_size = PG_GETARG_INT64(0);
     text       *message = PG_GETARG_TEXT_PP(1);
     char       *message_contents = VARDATA_ANY(message);
     int         message_size = VARSIZE_ANY_EXHDR(message);
     int32       loop_count = PG_GETARG_INT32(2);
     int32       nworkers = PG_GETARG_INT32(3);
     bool        verify = PG_GETARG_BOOL(4);
     int32       send_count = 0;
     int32       receive_count = 0;
     dsm_segment *seg;
     shm_mq_handle *outqh;
     shm_mq_handle *inqh;
     shm_mq_result res;
     Size        len;
     void       *data;
  
     /* A negative loopcount is nonsensical. */
     if (loop_count < 0)
         ereport(ERROR,
                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                  errmsg("repeat count size must be an integer value greater than or equal to zero")));
  
     /*
      * Using the nonblocking interfaces, we can even send data to ourselves,
      * so the minimum number of workers for this test is zero.
      */
     if (nworkers < 0)
         ereport(ERROR,
                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                  errmsg("number of workers must be an integer value greater than or equal to zero")));
  
     /* Set up dynamic shared memory segment and background workers. */
     test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
  
     /* Main loop. */
     for (;;)
     {
         bool        wait = true;
  
         /*
          * If we haven't yet sent the message the requisite number of times,
          * try again to send it now.  Note that when shm_mq_send() returns
          * SHM_MQ_WOULD_BLOCK, the next call to that function must pass the
          * same message size and contents; that's not an issue here because
          * we're sending the same message every time.
          */
         if (send_count < loop_count)
         {
             res = shm_mq_send(outqh, message_size, message_contents, true,
                               true);
             if (res == SHM_MQ_SUCCESS)
             {
                 ++send_count;
                 wait = false;
             }
             else if (res == SHM_MQ_DETACHED)
                 ereport(ERROR,
                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                          errmsg("could not send message")));
         }
  
         /*
          * If we haven't yet received the message the requisite number of
          * times, try to receive it again now.
          */
         if (receive_count < loop_count)
         {
             res = shm_mq_receive(inqh, &len, &data, true);
             if (res == SHM_MQ_SUCCESS)
             {
                 ++receive_count;
                 /* Verifying every time is slow, so it's optional. */
                 if (verify)
                     verify_message(message_size, message_contents, len, data);
                 wait = false;
             }
             else if (res == SHM_MQ_DETACHED)
                 ereport(ERROR,
                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                          errmsg("could not receive message")));
         }
         else
         {
             /*
              * Otherwise, we've received the message enough times.  This
              * shouldn't happen unless we've also sent it enough times.
              */
             if (send_count != receive_count)
                 ereport(ERROR,
                         (errcode(ERRCODE_INTERNAL_ERROR),
                          errmsg("message sent %d times, but received %d times",
                                 send_count, receive_count)));
             break;
         }
  
         if (wait)
         {
             /*
              * If we made no progress, wait for one of the other processes to
              * which we are connected to set our latch, indicating that they
              * have read or written data and therefore there may now be work
              * for us to do.
              */
             (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
                              WAIT_EVENT_EXTENSION);
             ResetLatch(MyLatch);
             CHECK_FOR_INTERRUPTS();
         }
     }
  
     /* Clean up. */
     dsm_detach(seg);
  
     PG_RETURN_VOID();
 }

References CHECK_FOR_INTERRUPTS, data, dsm_detach(), ereport, errcode(), errmsg(), ERROR, len, MyLatch, PG_GETARG_BOOL, PG_GETARG_INT32, PG_GETARG_INT64, PG_GETARG_TEXT_PP, PG_RETURN_VOID, res, ResetLatch(), SHM_MQ_DETACHED, shm_mq_receive(), shm_mq_send(), SHM_MQ_SUCCESS, test_shm_mq_setup(), VARDATA_ANY, VARSIZE_ANY_EXHDR, verify_message(), WAIT_EVENT_EXTENSION, WaitLatch(), WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

◆ verify_message()

static void verify_message	(	Size	origlen,
		char *	origdata,
		Size	newlen,
		char *	newdata
	)

static

Definition at line 251 of file test.c.

 {
     Size        i;
  
     if (origlen != newlen)
         ereport(ERROR,
                 (errmsg("message corrupted"),
                  errdetail("The original message was %zu bytes but the final message is %zu bytes.",
                            origlen, newlen)));
  
     for (i = 0; i < origlen; ++i)
         if (origdata[i] != newdata[i])
             ereport(ERROR,
                     (errmsg("message corrupted"),
                      errdetail("The new and original messages differ at byte %zu of %zu.", i, origlen)));
 }

References ereport, errdetail(), errmsg(), ERROR, and i.

Referenced by test_shm_mq(), and test_shm_mq_pipelined().

Variable Documentation

◆ PG_MODULE_MAGIC

PG_MODULE_MAGIC

Definition at line 23 of file test.c.

Functions

Variables