test.c File Reference

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

Include dependency graph for test.c:

Go to the source code of this file.

Functions
	PG_FUNCTION_INFO_V1 (test_shm_mq)
	PG_FUNCTION_INFO_V1 (test_shm_mq_pipelined)
void	_PG_init (void)
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

void _PG_init

(

void

)

Definition at line 56 of file auth_delay.c.

References add_int_reloption(), add_reloption_kind(), apw_start_master_worker(), AssertVariableIsOfType, auth_delay_checks(), auth_delay_milliseconds, auto_explain_log_analyze, auto_explain_log_buffers, auto_explain_log_format, auto_explain_log_min_duration, auto_explain_log_nested_statements, auto_explain_log_timing, auto_explain_log_triggers, auto_explain_log_verbose, auto_explain_sample_rate, autoprewarm, autoprewarm_interval, BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BackgroundWorker::bgw_name, BGW_NEVER_RESTART, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, bl_relopt_kind, buf, check_password(), check_password_hook, ClientAuthentication_hook, DEFAULT_BLOOM_BITS, DEFAULT_BLOOM_LENGTH, DefineCustomBoolVariable(), DefineCustomEnumVariable(), DefineCustomIntVariable(), DefineCustomRealVariable(), DefineCustomStringVariable(), dummy_object_relabel(), EAN13_index, EAN13_range, elog, EmitWarningsOnPlaceholders(), HASHCTL::entrysize, ereport, errcode(), errmsg(), ERROR, ExecutorCheckPerms_hook, ExecutorEnd_hook, ExecutorFinish_hook, ExecutorRun_hook, ExecutorStart_hook, explain_ExecutorEnd(), explain_ExecutorFinish(), explain_ExecutorRun(), explain_ExecutorStart(), EXPLAIN_FORMAT_TEXT, find_rendezvous_variable(), gettext_noop, GUC_LIST_INPUT, GUC_NOT_IN_SAMPLE, GUC_UNIT_MS, GUC_UNIT_S, HASH_BLOBS, hash_create(), HASH_ELEM, hstoreCheckKeyLen, hstoreCheckKeyLen_p, hstoreCheckValLen, hstoreCheckValLen_p, hstorePairs, hstorePairs_p, hstoreUniquePairs, hstoreUniquePairs_p, hstoreUpgrade, hstoreUpgrade_p, i, INDEX_MAX_KEYS, Int32GetDatum, ISBN_index, ISBN_range, ISMN_index, ISMN_range, ISN_DEBUG, ISSN_index, ISSN_range, IsUnderPostmaster, HASHCTL::keysize, load_external_function(), MAX_BLOOM_BITS, MAX_BLOOM_LENGTH, MAXALIGN, MemoryContextStrdup(), next_exec_check_perms_hook, next_object_access_hook, next_ProcessUtility_hook, object_access_hook, relopt_parse_elt::offset, offsetof, relopt_parse_elt::optname, relopt_parse_elt::opttype, original_client_auth_hook, pg_bindtextdomain(), PGC_POSTMASTER, PGC_SIGHUP, PGC_SUSET, PGC_USERSET, pgss_ExecutorEnd(), pgss_ExecutorFinish(), pgss_ExecutorRun(), pgss_ExecutorStart(), pgss_max, pgss_memsize(), pgss_post_parse_analyze(), pgss_ProcessUtility(), pgss_save, pgss_shmem_startup(), pgss_track, PGSS_TRACK_TOP, pgss_track_utility, plperl_held_interp, plperl_init_interp(), plperl_on_init, plperl_on_plperl_init, plperl_on_plperlu_init, plperl_opmask, plperl_use_strict, plpgsql_check_asserts, plpgsql_extra_checks_check_hook(), plpgsql_extra_errors_assign_hook(), plpgsql_extra_errors_string, plpgsql_extra_warnings_assign_hook(), plpgsql_extra_warnings_string, plpgsql_HashTableInit(), plpgsql_print_strict_params, PLPGSQL_RESOLVE_ERROR, plpgsql_subxact_cb(), plpgsql_variable_conflict, plpgsql_xact_cb(), plpython_version_bitmask, plpython_version_bitmask_ptr, pltcl_AlertNotifier(), pltcl_CreateFileHandler(), pltcl_DeleteFileHandler(), pltcl_FinalizeNotifier(), pltcl_hold_interp, pltcl_InitNotifier(), pltcl_pm_init_done, pltcl_ServiceModeHook(), pltcl_SetTimer(), pltcl_start_proc, pltcl_WaitForEvent(), pltclu_start_proc, PLyObject_AsString, PLyObject_AsString_p, PLyUnicode_FromStringAndSize, post_parse_analyze_hook, PQWalReceiverFunctions, prev_ExecutorEnd, prev_ExecutorFinish, prev_ExecutorRun, prev_ExecutorStart, prev_post_parse_analyze_hook, prev_ProcessUtility, prev_row_security_policy_hook_permissive, prev_row_security_policy_hook_restrictive, prev_shmem_startup_hook, process_shared_preload_libraries_in_progress, ProcessUtility_hook, register_label_provider(), RegisterBackgroundWorker(), RegisterSubXactCallback(), RegisterXactCallback(), RELOPT_TYPE_INT, RequestAddinShmemSpace(), RequestNamedLWLockTranche(), row_security_policy_hook_permissive, row_security_policy_hook_restrictive, sepgsql_avc_init(), sepgsql_debug_audit, sepgsql_exec_check_perms(), sepgsql_init_client_label(), SEPGSQL_LABEL_TAG, SEPGSQL_MODE_DISABLED, sepgsql_object_access(), sepgsql_object_relabel(), sepgsql_permissive, sepgsql_set_mode(), sepgsql_utility_command(), shmem_startup_hook, similarity_threshold, snprintf(), test_rls_hooks_permissive(), test_rls_hooks_restrictive(), TEXTDOMAIN, TopMemoryContext, UPC_index, UPC_range, WalReceiverFunctions, and word_similarity_threshold.

{
    /* Define custom GUC variables */
    DefineCustomIntVariable("auth_delay.milliseconds",
                            "Milliseconds to delay before reporting authentication failure",
                            NULL,
                            &auth_delay_milliseconds,
                            0,
                            0, INT_MAX / 1000,
                            PGC_SIGHUP,
                            GUC_UNIT_MS,
                            NULL,
                            NULL,
                            NULL);
    /* Install Hooks */
    original_client_auth_hook = ClientAuthentication_hook;
    ClientAuthentication_hook = auth_delay_checks;
}

PG_FUNCTION_INFO_V1

(

test_shm_mq

)

PG_FUNCTION_INFO_V1

(

test_shm_mq_pipelined

)

Datum test_shm_mq

(

PG_FUNCTION_ARGS

)

Definition at line 41 of file test.c.

References dsm_detach(), ereport, errcode(), errmsg(), ERROR, 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().

{
    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 a non-negative integer")));

    /*
     * 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 < 1)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("number of workers must be a positive integer")));

    /* 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);
    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);
        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();
}

Datum test_shm_mq_pipelined

(

PG_FUNCTION_ARGS

)

Definition at line 130 of file test.c.

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

{
    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 a non-negative integer")));

    /*
     * 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 a non-negative integer")));

    /* 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);
            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.
             */
            WaitLatch(MyLatch, WL_LATCH_SET, 0, PG_WAIT_EXTENSION);
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }
    }

    /* Clean up. */
    dsm_detach(seg);

    PG_RETURN_VOID();
}

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

static

Definition at line 250 of file test.c.

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

Referenced by test_shm_mq(), and test_shm_mq_pipelined().

{
    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)));
}

Variable Documentation

PG_MODULE_MAGIC

Definition at line 22 of file test.c.