pqmq.c File Reference

#include "postgres.h"
#include "access/parallel.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/logicalworker.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"

Include dependency graph for pqmq.c:

Go to the source code of this file.

Functions
static void	pq_cleanup_redirect_to_shm_mq (dsm_segment *seg, Datum arg)
static void	mq_comm_reset (void)
static int	mq_flush (void)
static int	mq_flush_if_writable (void)
static bool	mq_is_send_pending (void)
static int	mq_putmessage (char msgtype, const char *s, size_t len)
static void	mq_putmessage_noblock (char msgtype, const char *s, size_t len)
void	pq_redirect_to_shm_mq (dsm_segment *seg, shm_mq_handle *mqh)
void	pq_set_parallel_leader (pid_t pid, ProcNumber procNumber)
void	pq_parse_errornotice (StringInfo msg, ErrorData *edata)

Variables
static shm_mq_handle *	pq_mq_handle = NULL
static bool	pq_mq_busy = false
static pid_t	pq_mq_parallel_leader_pid = 0
static ProcNumber	pq_mq_parallel_leader_proc_number = INVALID_PROC_NUMBER
static const PQcommMethods	PqCommMqMethods

Function Documentation

mq_comm_reset()

static void mq_comm_reset ( void  )

static

Definition at line 90 of file pqmq.c.

{
    /* Nothing to do. */
}

mq_flush()

static int mq_flush ( void  )

static

Definition at line 96 of file pqmq.c.

{
    /* Nothing to do. */
    return 0;
}

mq_flush_if_writable()

static int mq_flush_if_writable ( void  )

static

Definition at line 103 of file pqmq.c.

{
    /* Nothing to do. */
    return 0;
}

mq_is_send_pending()

static bool mq_is_send_pending ( void  )

static

Definition at line 110 of file pqmq.c.

{
    /* There's never anything pending. */
    return 0;
}

mq_putmessage()

static int mq_putmessage ( char  msgtype, const char *  s, size_t  len  )

static

Definition at line 122 of file pqmq.c.

{
    shm_mq_iovec iov[2];
    shm_mq_result result;
 
    /*
     * If we're sending a message, and we have to wait because the queue is
     * full, and then we get interrupted, and that interrupt results in trying
     * to send another message, we respond by detaching the queue.  There's no
     * way to return to the original context, but even if there were, just
     * queueing the message would amount to indefinitely postponing the
     * response to the interrupt.  So we do this instead.
     */
    if (pq_mq_busy)
    {
        if (pq_mq_handle != NULL)
        {
            shm_mq_detach(pq_mq_handle);
            pfree(pq_mq_handle);
            pq_mq_handle = NULL;
        }
        return EOF;
    }
 
    /*
     * If the message queue is already gone, just ignore the message. This
     * doesn't necessarily indicate a problem; for example, DEBUG messages can
     * be generated late in the shutdown sequence, after all DSMs have already
     * been detached.
     */
    if (pq_mq_handle == NULL)
        return 0;
 
    pq_mq_busy = true;
 
    iov[0].data = &msgtype;
    iov[0].len = 1;
    iov[1].data = s;
    iov[1].len = len;
 
    for (;;)
    {
        /*
         * Immediately notify the receiver by passing force_flush as true so
         * that the shared memory value is updated before we send the parallel
         * message signal right after this.
         */
        Assert(pq_mq_handle != NULL);
        result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
 
        if (pq_mq_parallel_leader_pid != 0)
        {
            if (IsLogicalParallelApplyWorker())
                SendProcSignal(pq_mq_parallel_leader_pid,
                               PROCSIG_PARALLEL_APPLY_MESSAGE,
                               pq_mq_parallel_leader_proc_number);
            else
            {
                Assert(IsParallelWorker());
                SendProcSignal(pq_mq_parallel_leader_pid,
                               PROCSIG_PARALLEL_MESSAGE,
                               pq_mq_parallel_leader_proc_number);
            }
        }
 
        if (result != SHM_MQ_WOULD_BLOCK)
            break;
 
        (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
                         WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
        ResetLatch(MyLatch);
        CHECK_FOR_INTERRUPTS();
    }
 
    pq_mq_busy = false;
 
    Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
    if (result != SHM_MQ_SUCCESS)
        return EOF;
    return 0;
}

References Assert, CHECK_FOR_INTERRUPTS, fb(), IsLogicalParallelApplyWorker(), IsParallelWorker, len, MyLatch, pfree(), pq_mq_busy, pq_mq_handle, pq_mq_parallel_leader_pid, pq_mq_parallel_leader_proc_number, PROCSIG_PARALLEL_APPLY_MESSAGE, PROCSIG_PARALLEL_MESSAGE, ResetLatch(), SendProcSignal(), shm_mq_detach(), SHM_MQ_DETACHED, shm_mq_sendv(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, WaitLatch(), WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

mq_putmessage_noblock()

static void mq_putmessage_noblock ( char  msgtype, const char *  s, size_t  len  )

static

Definition at line 205 of file pqmq.c.

{
    /*
     * While the shm_mq machinery does support sending a message in
     * non-blocking mode, there's currently no way to try sending beginning to
     * send the message that doesn't also commit us to completing the
     * transmission.  This could be improved in the future, but for now we
     * don't need it.
     */
    elog(ERROR, "not currently supported");
}

References elog, and ERROR.

pq_cleanup_redirect_to_shm_mq()

static void pq_cleanup_redirect_to_shm_mq ( dsm_segment *  seg, Datum  arg  )

static

Definition at line 67 of file pqmq.c.

{
    if (pq_mq_handle != NULL)
    {
        pfree(pq_mq_handle);
        pq_mq_handle = NULL;
    }
    whereToSendOutput = DestNone;
}

References DestNone, fb(), pfree(), pq_mq_handle, and whereToSendOutput.

Referenced by pq_redirect_to_shm_mq().

pq_parse_errornotice()

void pq_parse_errornotice	(	StringInfo	msg,
		ErrorData *	edata
	)

Definition at line 222 of file pqmq.c.

{
    /* Initialize edata with reasonable defaults. */
    MemSet(edata, 0, sizeof(ErrorData));
    edata->elevel = ERROR;
    edata->assoc_context = CurrentMemoryContext;
 
    /* Loop over fields and extract each one. */
    for (;;)
    {
        char        code = pq_getmsgbyte(msg);
        const char *value;
 
        if (code == '\0')
        {
            pq_getmsgend(msg);
            break;
        }
        value = pq_getmsgrawstring(msg);
 
        switch (code)
        {
            case PG_DIAG_SEVERITY:
                /* ignore, trusting we'll get a nonlocalized version */
                break;
            case PG_DIAG_SEVERITY_NONLOCALIZED:
                if (strcmp(value, "DEBUG") == 0)
                {
                    /*
                     * We can't reconstruct the exact DEBUG level, but
                     * presumably it was >= client_min_messages, so select
                     * DEBUG1 to ensure we'll pass it on to the client.
                     */
                    edata->elevel = DEBUG1;
                }
                else if (strcmp(value, "LOG") == 0)
                {
                    /*
                     * It can't be LOG_SERVER_ONLY, or the worker wouldn't
                     * have sent it to us; so LOG is the correct value.
                     */
                    edata->elevel = LOG;
                }
                else if (strcmp(value, "INFO") == 0)
                    edata->elevel = INFO;
                else if (strcmp(value, "NOTICE") == 0)
                    edata->elevel = NOTICE;
                else if (strcmp(value, "WARNING") == 0)
                    edata->elevel = WARNING;
                else if (strcmp(value, "ERROR") == 0)
                    edata->elevel = ERROR;
                else if (strcmp(value, "FATAL") == 0)
                    edata->elevel = FATAL;
                else if (strcmp(value, "PANIC") == 0)
                    edata->elevel = PANIC;
                else
                    elog(ERROR, "unrecognized error severity: \"%s\"", value);
                break;
            case PG_DIAG_SQLSTATE:
                if (strlen(value) != 5)
                    elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
                edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
                                                  value[3], value[4]);
                break;
            case PG_DIAG_MESSAGE_PRIMARY:
                edata->message = pstrdup(value);
                break;
            case PG_DIAG_MESSAGE_DETAIL:
                edata->detail = pstrdup(value);
                break;
            case PG_DIAG_MESSAGE_HINT:
                edata->hint = pstrdup(value);
                break;
            case PG_DIAG_STATEMENT_POSITION:
                edata->cursorpos = pg_strtoint32(value);
                break;
            case PG_DIAG_INTERNAL_POSITION:
                edata->internalpos = pg_strtoint32(value);
                break;
            case PG_DIAG_INTERNAL_QUERY:
                edata->internalquery = pstrdup(value);
                break;
            case PG_DIAG_CONTEXT:
                edata->context = pstrdup(value);
                break;
            case PG_DIAG_SCHEMA_NAME:
                edata->schema_name = pstrdup(value);
                break;
            case PG_DIAG_TABLE_NAME:
                edata->table_name = pstrdup(value);
                break;
            case PG_DIAG_COLUMN_NAME:
                edata->column_name = pstrdup(value);
                break;
            case PG_DIAG_DATATYPE_NAME:
                edata->datatype_name = pstrdup(value);
                break;
            case PG_DIAG_CONSTRAINT_NAME:
                edata->constraint_name = pstrdup(value);
                break;
            case PG_DIAG_SOURCE_FILE:
                edata->filename = pstrdup(value);
                break;
            case PG_DIAG_SOURCE_LINE:
                edata->lineno = pg_strtoint32(value);
                break;
            case PG_DIAG_SOURCE_FUNCTION:
                edata->funcname = pstrdup(value);
                break;
            default:
                elog(ERROR, "unrecognized error field code: %d", code);
                break;
        }
    }
}

References CurrentMemoryContext, DEBUG1, elog, ERROR, FATAL, fb(), INFO, LOG, MAKE_SQLSTATE, MemSet, NOTICE, PANIC, PG_DIAG_COLUMN_NAME, PG_DIAG_CONSTRAINT_NAME, PG_DIAG_CONTEXT, PG_DIAG_DATATYPE_NAME, PG_DIAG_INTERNAL_POSITION, PG_DIAG_INTERNAL_QUERY, PG_DIAG_MESSAGE_DETAIL, PG_DIAG_MESSAGE_HINT, PG_DIAG_MESSAGE_PRIMARY, PG_DIAG_SCHEMA_NAME, PG_DIAG_SEVERITY, PG_DIAG_SEVERITY_NONLOCALIZED, PG_DIAG_SOURCE_FILE, PG_DIAG_SOURCE_FUNCTION, PG_DIAG_SOURCE_LINE, PG_DIAG_SQLSTATE, PG_DIAG_STATEMENT_POSITION, PG_DIAG_TABLE_NAME, pg_strtoint32(), pq_getmsgbyte(), pq_getmsgend(), pq_getmsgrawstring(), pstrdup(), value, and WARNING.

Referenced by ProcessParallelApplyMessage(), and ProcessParallelMessage().

pq_redirect_to_shm_mq()

void pq_redirect_to_shm_mq	(	dsm_segment *	seg,
		shm_mq_handle *	mqh
	)

Definition at line 53 of file pqmq.c.

{
    PqCommMethods = &PqCommMqMethods;
    pq_mq_handle = mqh;
    whereToSendOutput = DestRemote;
    FrontendProtocol = PG_PROTOCOL_LATEST;
    on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0);
}

References DestRemote, fb(), FrontendProtocol, on_dsm_detach(), PG_PROTOCOL_LATEST, pq_cleanup_redirect_to_shm_mq(), pq_mq_handle, PqCommMethods, PqCommMqMethods, and whereToSendOutput.

Referenced by ParallelApplyWorkerMain(), and ParallelWorkerMain().

pq_set_parallel_leader()

void pq_set_parallel_leader	(	pid_t	pid,
		ProcNumber	procNumber
	)

Definition at line 82 of file pqmq.c.

{
    Assert(PqCommMethods == &PqCommMqMethods);
    pq_mq_parallel_leader_pid = pid;
    pq_mq_parallel_leader_proc_number = procNumber;
}

References Assert, pq_mq_parallel_leader_pid, pq_mq_parallel_leader_proc_number, PqCommMethods, and PqCommMqMethods.

Referenced by ParallelApplyWorkerMain(), and ParallelWorkerMain().

Variable Documentation

pq_mq_busy

bool pq_mq_busy = false

static

Definition at line 27 of file pqmq.c.

Referenced by mq_putmessage().

pq_mq_handle

shm_mq_handle* pq_mq_handle = NULL

static

Definition at line 26 of file pqmq.c.

Referenced by mq_putmessage(), pq_cleanup_redirect_to_shm_mq(), and pq_redirect_to_shm_mq().

pq_mq_parallel_leader_pid

pid_t pq_mq_parallel_leader_pid = 0

static

Definition at line 28 of file pqmq.c.

Referenced by mq_putmessage(), and pq_set_parallel_leader().

pq_mq_parallel_leader_proc_number

ProcNumber pq_mq_parallel_leader_proc_number = INVALID_PROC_NUMBER

static

Definition at line 29 of file pqmq.c.

Referenced by mq_putmessage(), and pq_set_parallel_leader().

PqCommMqMethods

const PQcommMethods PqCommMqMethods

static

Initial value:

= {
    .comm_reset = mq_comm_reset,
    .flush = mq_flush,
    .flush_if_writable = mq_flush_if_writable,
    .is_send_pending = mq_is_send_pending,
    .putmessage = mq_putmessage,
    .putmessage_noblock = mq_putmessage_noblock
}

Definition at line 39 of file pqmq.c.

                                             {
    .comm_reset = mq_comm_reset,
    .flush = mq_flush,
    .flush_if_writable = mq_flush_if_writable,
    .is_send_pending = mq_is_send_pending,
    .putmessage = mq_putmessage,
    .putmessage_noblock = mq_putmessage_noblock
};

Referenced by pq_redirect_to_shm_mq(), and pq_set_parallel_leader().