LCOV - code coverage report
Current view: top level - src/test/modules/test_shm_mq - test.c (source / functions) Hit Total Coverage
Test: PostgreSQL 16devel Lines: 66 78 84.6 %
Date: 2022-08-17 04:10:37 Functions: 6 6 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*--------------------------------------------------------------------------
       2             :  *
       3             :  * test.c
       4             :  *      Test harness code for shared memory message queues.
       5             :  *
       6             :  * Copyright (c) 2013-2022, PostgreSQL Global Development Group
       7             :  *
       8             :  * IDENTIFICATION
       9             :  *      src/test/modules/test_shm_mq/test.c
      10             :  *
      11             :  * -------------------------------------------------------------------------
      12             :  */
      13             : 
      14             : #include "postgres.h"
      15             : 
      16             : #include "fmgr.h"
      17             : #include "miscadmin.h"
      18             : #include "pgstat.h"
      19             : 
      20             : #include "test_shm_mq.h"
      21             : 
      22          16 : PG_MODULE_MAGIC;
      23             : 
      24           4 : PG_FUNCTION_INFO_V1(test_shm_mq);
      25           4 : PG_FUNCTION_INFO_V1(test_shm_mq_pipelined);
      26             : 
      27             : static void verify_message(Size origlen, char *origdata, Size newlen,
      28             :                            char *newdata);
      29             : 
      30             : /*
      31             :  * Simple test of the shared memory message queue infrastructure.
      32             :  *
      33             :  * We set up a ring of message queues passing through 1 or more background
      34             :  * processes and eventually looping back to ourselves.  We then send a message
      35             :  * through the ring a number of times indicated by the loop count.  At the end,
      36             :  * we check whether the final message matches the one we started with.
      37             :  */
      38             : Datum
      39           8 : test_shm_mq(PG_FUNCTION_ARGS)
      40             : {
      41           8 :     int64       queue_size = PG_GETARG_INT64(0);
      42           8 :     text       *message = PG_GETARG_TEXT_PP(1);
      43           8 :     char       *message_contents = VARDATA_ANY(message);
      44           8 :     int         message_size = VARSIZE_ANY_EXHDR(message);
      45           8 :     int32       loop_count = PG_GETARG_INT32(2);
      46           8 :     int32       nworkers = PG_GETARG_INT32(3);
      47             :     dsm_segment *seg;
      48             :     shm_mq_handle *outqh;
      49             :     shm_mq_handle *inqh;
      50             :     shm_mq_result res;
      51             :     Size        len;
      52             :     void       *data;
      53             : 
      54             :     /* A negative loopcount is nonsensical. */
      55           8 :     if (loop_count < 0)
      56           0 :         ereport(ERROR,
      57             :                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      58             :                  errmsg("repeat count size must be an integer value greater than or equal to zero")));
      59             : 
      60             :     /*
      61             :      * Since this test sends data using the blocking interfaces, it cannot
      62             :      * send data to itself.  Therefore, a minimum of 1 worker is required. Of
      63             :      * course, a negative worker count is nonsensical.
      64             :      */
      65           8 :     if (nworkers <= 0)
      66           0 :         ereport(ERROR,
      67             :                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      68             :                  errmsg("number of workers must be an integer value greater than zero")));
      69             : 
      70             :     /* Set up dynamic shared memory segment and background workers. */
      71           8 :     test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
      72             : 
      73             :     /* Send the initial message. */
      74           8 :     res = shm_mq_send(outqh, message_size, message_contents, false, true);
      75           8 :     if (res != SHM_MQ_SUCCESS)
      76           0 :         ereport(ERROR,
      77             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      78             :                  errmsg("could not send message")));
      79             : 
      80             :     /*
      81             :      * Receive a message and send it back out again.  Do this a number of
      82             :      * times equal to the loop count.
      83             :      */
      84             :     for (;;)
      85             :     {
      86             :         /* Receive a message. */
      87       48002 :         res = shm_mq_receive(inqh, &len, &data, false);
      88       48002 :         if (res != SHM_MQ_SUCCESS)
      89           0 :             ereport(ERROR,
      90             :                     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      91             :                      errmsg("could not receive message")));
      92             : 
      93             :         /* If this is supposed to be the last iteration, stop here. */
      94       48002 :         if (--loop_count <= 0)
      95           8 :             break;
      96             : 
      97             :         /* Send it back out. */
      98       47994 :         res = shm_mq_send(outqh, len, data, false, true);
      99       47994 :         if (res != SHM_MQ_SUCCESS)
     100           0 :             ereport(ERROR,
     101             :                     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     102             :                      errmsg("could not send message")));
     103             :     }
     104             : 
     105             :     /*
     106             :      * Finally, check that we got back the same message from the last
     107             :      * iteration that we originally sent.
     108             :      */
     109           8 :     verify_message(message_size, message_contents, len, data);
     110             : 
     111             :     /* Clean up. */
     112           8 :     dsm_detach(seg);
     113             : 
     114           8 :     PG_RETURN_VOID();
     115             : }
     116             : 
     117             : /*
     118             :  * Pipelined test of the shared memory message queue infrastructure.
     119             :  *
     120             :  * As in the basic test, we set up a ring of message queues passing through
     121             :  * 1 or more background processes and eventually looping back to ourselves.
     122             :  * Then, we send N copies of the user-specified message through the ring and
     123             :  * receive them all back.  Since this might fill up all message queues in the
     124             :  * ring and then stall, we must be prepared to begin receiving the messages
     125             :  * back before we've finished sending them.
     126             :  */
     127             : Datum
     128           2 : test_shm_mq_pipelined(PG_FUNCTION_ARGS)
     129             : {
     130           2 :     int64       queue_size = PG_GETARG_INT64(0);
     131           2 :     text       *message = PG_GETARG_TEXT_PP(1);
     132           2 :     char       *message_contents = VARDATA_ANY(message);
     133           2 :     int         message_size = VARSIZE_ANY_EXHDR(message);
     134           2 :     int32       loop_count = PG_GETARG_INT32(2);
     135           2 :     int32       nworkers = PG_GETARG_INT32(3);
     136           2 :     bool        verify = PG_GETARG_BOOL(4);
     137           2 :     int32       send_count = 0;
     138           2 :     int32       receive_count = 0;
     139             :     dsm_segment *seg;
     140             :     shm_mq_handle *outqh;
     141             :     shm_mq_handle *inqh;
     142             :     shm_mq_result res;
     143             :     Size        len;
     144             :     void       *data;
     145             : 
     146             :     /* A negative loopcount is nonsensical. */
     147           2 :     if (loop_count < 0)
     148           0 :         ereport(ERROR,
     149             :                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     150             :                  errmsg("repeat count size must be an integer value greater than or equal to zero")));
     151             : 
     152             :     /*
     153             :      * Using the nonblocking interfaces, we can even send data to ourselves,
     154             :      * so the minimum number of workers for this test is zero.
     155             :      */
     156           2 :     if (nworkers < 0)
     157           0 :         ereport(ERROR,
     158             :                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     159             :                  errmsg("number of workers must be an integer value greater than or equal to zero")));
     160             : 
     161             :     /* Set up dynamic shared memory segment and background workers. */
     162           2 :     test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
     163             : 
     164             :     /* Main loop. */
     165             :     for (;;)
     166       11190 :     {
     167       11192 :         bool        wait = true;
     168             : 
     169             :         /*
     170             :          * If we haven't yet sent the message the requisite number of times,
     171             :          * try again to send it now.  Note that when shm_mq_send() returns
     172             :          * SHM_MQ_WOULD_BLOCK, the next call to that function must pass the
     173             :          * same message size and contents; that's not an issue here because
     174             :          * we're sending the same message every time.
     175             :          */
     176       11192 :         if (send_count < loop_count)
     177             :         {
     178       11146 :             res = shm_mq_send(outqh, message_size, message_contents, true,
     179             :                               true);
     180       11146 :             if (res == SHM_MQ_SUCCESS)
     181             :             {
     182         400 :                 ++send_count;
     183         400 :                 wait = false;
     184             :             }
     185       10746 :             else if (res == SHM_MQ_DETACHED)
     186           0 :                 ereport(ERROR,
     187             :                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     188             :                          errmsg("could not send message")));
     189             :         }
     190             : 
     191             :         /*
     192             :          * If we haven't yet received the message the requisite number of
     193             :          * times, try to receive it again now.
     194             :          */
     195       11192 :         if (receive_count < loop_count)
     196             :         {
     197       11190 :             res = shm_mq_receive(inqh, &len, &data, true);
     198       11190 :             if (res == SHM_MQ_SUCCESS)
     199             :             {
     200         400 :                 ++receive_count;
     201             :                 /* Verifying every time is slow, so it's optional. */
     202         400 :                 if (verify)
     203         400 :                     verify_message(message_size, message_contents, len, data);
     204         400 :                 wait = false;
     205             :             }
     206       10790 :             else if (res == SHM_MQ_DETACHED)
     207           0 :                 ereport(ERROR,
     208             :                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     209             :                          errmsg("could not receive message")));
     210             :         }
     211             :         else
     212             :         {
     213             :             /*
     214             :              * Otherwise, we've received the message enough times.  This
     215             :              * shouldn't happen unless we've also sent it enough times.
     216             :              */
     217           2 :             if (send_count != receive_count)
     218           0 :                 ereport(ERROR,
     219             :                         (errcode(ERRCODE_INTERNAL_ERROR),
     220             :                          errmsg("message sent %d times, but received %d times",
     221             :                                 send_count, receive_count)));
     222           2 :             break;
     223             :         }
     224             : 
     225       11190 :         if (wait)
     226             :         {
     227             :             /*
     228             :              * If we made no progress, wait for one of the other processes to
     229             :              * which we are connected to set our latch, indicating that they
     230             :              * have read or written data and therefore there may now be work
     231             :              * for us to do.
     232             :              */
     233       10410 :             (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
     234             :                              PG_WAIT_EXTENSION);
     235       10410 :             ResetLatch(MyLatch);
     236       10410 :             CHECK_FOR_INTERRUPTS();
     237             :         }
     238             :     }
     239             : 
     240             :     /* Clean up. */
     241           2 :     dsm_detach(seg);
     242             : 
     243           2 :     PG_RETURN_VOID();
     244             : }
     245             : 
     246             : /*
     247             :  * Verify that two messages are the same.
     248             :  */
     249             : static void
     250         408 : verify_message(Size origlen, char *origdata, Size newlen, char *newdata)
     251             : {
     252             :     Size        i;
     253             : 
     254         408 :     if (origlen != newlen)
     255           0 :         ereport(ERROR,
     256             :                 (errmsg("message corrupted"),
     257             :                  errdetail("The original message was %zu bytes but the final message is %zu bytes.",
     258             :                            origlen, newlen)));
     259             : 
     260   108002482 :     for (i = 0; i < origlen; ++i)
     261   108002074 :         if (origdata[i] != newdata[i])
     262           0 :             ereport(ERROR,
     263             :                     (errmsg("message corrupted"),
     264             :                      errdetail("The new and original messages differ at byte %zu of %zu.", i, origlen)));
     265         408 : }

Generated by: LCOV version 1.14