#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

static uint32	we_message_queue = 0

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 43 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, 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 132 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)
        {
            /* first time, allocate or get the custom wait event */
            if (we_message_queue == 0)
                we_message_queue = WaitEventExtensionNew("TestShmMqMessageQueue");
 
            /*
             * 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,
                             we_message_queue);
            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, ResetLatch(), SHM_MQ_DETACHED, shm_mq_receive(), shm_mq_send(), SHM_MQ_SUCCESS, test_shm_mq_setup(), VARDATA_ANY, VARSIZE_ANY_EXHDR, verify_message(), WaitEventExtensionNew(), WaitLatch(), we_message_queue, 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 258 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.

◆ we_message_queue

uint32 we_message_queue = 0