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

Generated by: LCOV version 1.13